<dependencies>
    <!-- Spring Boot Web -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
 
    <!-- Spring Data JPA (for upload session persistence) -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
 
    <!-- Redis (for distributed upload session lock + fast offset lookup) -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>
 
    <!-- AWS SDK v2 for S3 multipart upload -->
    <dependency>
        <groupId>software.amazon.awssdk</groupId>
        <artifactId>s3</artifactId>
        <version>2.25.0</version>
    </dependency>
 
    <!-- Micrometer for upload metrics -->
    <dependency>
        <groupId>io.micrometer</groupId>
        <artifactId>micrometer-registry-prometheus</artifactId>
    </dependency>
</dependencies>

package com.example.upload.domain;
 
@Entity
@Table(name = "upload_sessions",
       indexes = {
           @Index(name = "idx_upload_sessions_status", columnList = "status"),
           @Index(name = "idx_upload_sessions_expires", columnList = "expires_at")
       })
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class UploadSession {
 
    @Id
    @Column(name = "upload_id", length = 36)
    private String uploadId;  // UUID
 
    @Column(name = "user_id", nullable = false, length = 36)
    private String userId;
 
    @Column(name = "filename", nullable = false, length = 255)
    private String filename;
 
    @Column(name = "content_type", length = 128)
    private String contentType;
 
    @Column(name = "total_length", nullable = false)
    private long totalLength;  // Total file size in bytes
 
    @Column(name = "received_offset", nullable = false)
    private long receivedOffset;  // Bytes confirmed received by server
 
    @Enumerated(EnumType.STRING)
    @Column(name = "status", nullable = false, length = 20)
    private UploadStatus status;
 
    @Column(name = "storage_path", length = 500)
    private String storagePath;  // Path in S3 or local FS
 
    @Column(name = "storage_upload_id", length = 500)
    private String storageUploadId;  // S3 UploadId for multipart
 
    @Column(name = "checksum_md5", length = 32)
    private String checksumMd5;  // MD5 of the completed file
 
    @Column(name = "created_at", nullable = false)
    private Instant createdAt;
 
    @Column(name = "expires_at", nullable = false)
    private Instant expiresAt;  // TUS: session valid for 7 days
 
    @Column(name = "completed_at")
    private Instant completedAt;
 
    @Version
    private Long version;  // Optimistic locking to prevent concurrent writes
 
    // ---- Computed methods ----
 
    public boolean isExpired() {
        return expiresAt.isBefore(Instant.now());
    }
 
    public boolean isComplete() {
        return receivedOffset >= totalLength;
    }
 
    public double progressPercent() {
        if (totalLength == 0) return 100.0;
        return (double) receivedOffset / totalLength * 100.0;
    }
}

public enum UploadStatus {
    INITIATED,    // Session created, no bytes received yet
    IN_PROGRESS,  // Chunks are being received
    PAUSED,       // Upload paused by client (optional, informational)
    COMPLETE,     // All bytes received and verified
    PROCESSING,   // Post-upload processing (transcoding, virus scan, indexing)
    FAILED,       // Upload failed and cannot be resumed
    EXPIRED,      // Session TTL exceeded before completion
    ABORTED       // Explicitly aborted by client or admin
}

@Entity
@Table(name = "upload_parts",
       uniqueConstraints = @UniqueConstraint(
           columnNames = {"upload_session_id", "part_number"}
       ))
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class UploadPart {
 
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
 
    @Column(name = "upload_session_id", nullable = false, length = 36)
    private String uploadSessionId;
 
    @Column(name = "part_number", nullable = false)
    private int partNumber;       // 1-based, matching S3 part numbers
 
    @Column(name = "etag", length = 64)
    private String etag;          // S3 ETag for this part (required for CompleteMultipartUpload)
 
    @Column(name = "part_size")
    private long partSize;        // Size of this part in bytes
 
    @Column(name = "uploaded_at")
    private Instant uploadedAt;
}

@Configuration
public class S3Config {
 
    @Value("${aws.region:us-east-1}")
    private String region;
 
    @Bean
    public S3Client s3Client() {
        return S3Client.builder()
            .region(Region.of(region))
            // Credentials from environment, IAM role, or ~/.aws/credentials
            .credentialsProvider(DefaultCredentialsProvider.create())
            // Transfer acceleration for global users
            // .serviceConfiguration(S3Configuration.builder().accelerateModeEnabled(true).build())
            .build();
    }
 
    @Bean
    public S3Presigner s3Presigner() {
        return S3Presigner.builder()
            .region(Region.of(region))
            .credentialsProvider(DefaultCredentialsProvider.create())
            .build();
    }
 
    @Bean
    public TransferManager transferManager(S3AsyncClient s3AsyncClient) {
        // AWS Transfer Manager handles multipart automatically for large files
        return TransferManager.builder()
            .s3Client(s3AsyncClient)
            .uploadDirectoryFollowSymbolicLinks(false)
            .minimumPartSizeInBytes(8L * 1024 * 1024)  // 8 MB minimum part
            .multipartUploadThreshold(16L * 1024 * 1024)  // Use multipart for files > 16 MB
            .build();
    }
}

@Service
@Slf4j
@RequiredArgsConstructor
public class S3MultipartUploadService {
 
    private final S3Client s3Client;
    private final S3Presigner s3Presigner;
    private final UploadSessionRepository sessionRepository;
    private final UploadPartRepository partRepository;
 
    @Value("${s3.bucket.uploads}")
    private String uploadsBucket;
 
    /**
     * Step 1: Initiate a multipart upload on S3.
     * Returns an UploadId that must be saved and used for all subsequent calls.
     */
    public String initiateS3MultipartUpload(String s3Key, String contentType, Map<String, String> metadata) {
        CreateMultipartUploadRequest request = CreateMultipartUploadRequest.builder()
            .bucket(uploadsBucket)
            .key(s3Key)
            .contentType(contentType)
            .metadata(metadata)
            // Server-side encryption
            .serverSideEncryption(ServerSideEncryption.AES256)
            // Storage class: STANDARD for frequent access, INTELLIGENT_TIERING for uncertain access
            .storageClass(StorageClass.STANDARD)
            .build();
 
        CreateMultipartUploadResponse response = s3Client.createMultipartUpload(request);
        log.info("Initiated S3 multipart upload for key={}, uploadId={}", s3Key, response.uploadId());
        return response.uploadId();
    }
 
    /**
     * Generate presigned URLs for each part.
     * The browser will PUT directly to these URLs, bypassing the Spring Boot server.
     * Your server never handles the file bytes — massive bandwidth and memory saving.
     *
     * @param s3Key       The S3 object key
     * @param s3UploadId  The multipart upload ID from initiateS3MultipartUpload
     * @param partCount   How many parts (ceil(fileSize / partSize))
     * @param urlValidity How long each URL is valid
     */
    public List<PresignedPartInfo> generatePresignedPartUrls(
            String s3Key,
            String s3UploadId,
            int partCount,
            Duration urlValidity) {
 
        List<PresignedPartInfo> urls = new ArrayList<>(partCount);
 
        for (int partNumber = 1; partNumber <= partCount; partNumber++) {
            UploadPartRequest uploadPartRequest = UploadPartRequest.builder()
                .bucket(uploadsBucket)
                .key(s3Key)
                .uploadId(s3UploadId)
                .partNumber(partNumber)
                .build();
 
            UploadPartPresignRequest presignRequest = UploadPartPresignRequest.builder()
                .signatureDuration(urlValidity)
                .uploadPartRequest(uploadPartRequest)
                .build();
 
            PresignedUploadPartRequest presignedRequest = s3Presigner.presignUploadPart(presignRequest);
 
            urls.add(PresignedPartInfo.builder()
                .partNumber(partNumber)
                .presignedUrl(presignedRequest.url().toString())
                .expiresAt(presignedRequest.expiration())
                .build());
        }
 
        log.info("Generated {} presigned URLs for s3Key={}", partCount, s3Key);
        return urls;
    }
 
    /**
     * Step 3: Complete the S3 multipart upload.
     * Called AFTER all parts have been uploaded to S3 via presigned URLs.
     * Parts are assembled by S3 into a single object.
     *
     * @param parts List of (partNumber, etag) pairs returned by S3 when each part was uploaded
     */
    @Transactional
    public String completeMultipartUpload(
            String s3Key,
            String s3UploadId,
            List<PartInfo> parts) {
 
        List<CompletedPart> completedParts = parts.stream()
            .sorted(Comparator.comparingInt(PartInfo::getPartNumber))
            .map(p -> CompletedPart.builder()
                .partNumber(p.getPartNumber())
                .eTag(p.getEtag())  // ETag must match exactly (including quotes if S3 returns them)
                .build())
            .toList();
 
        CompleteMultipartUploadRequest request = CompleteMultipartUploadRequest.builder()
            .bucket(uploadsBucket)
            .key(s3Key)
            .uploadId(s3UploadId)
            .multipartUpload(mp -> mp.parts(completedParts))
            .build();
 
        CompleteMultipartUploadResponse response = s3Client.completeMultipartUpload(request);
        log.info("Completed S3 multipart upload. key={}, etag={}", s3Key, response.eTag());
 
        // Return the final S3 object URL
        return "s3://" + uploadsBucket + "/" + s3Key;
    }
 
    /**
     * Abort an incomplete multipart upload.
     * CRITICAL: Call this on upload failure/expiry to prevent orphaned S3 parts
     * accumulating storage costs. Also configure a lifecycle rule as a safety net.
     */
    public void abortMultipartUpload(String s3Key, String s3UploadId) {
        try {
            s3Client.abortMultipartUpload(r -> r
                .bucket(uploadsBucket)
                .key(s3Key)
                .uploadId(s3UploadId)
            );
            log.info("Aborted S3 multipart upload. key={}, uploadId={}", s3Key, s3UploadId);
        } catch (Exception e) {
            // Log but do not throw: the S3 lifecycle rule will clean this up
            log.error("Failed to abort S3 multipart upload key={}: {}", s3Key, e.getMessage());
        }
    }
 
    /**
     * List parts that S3 has already received.
     * Used for server-side resume: determine which parts are already uploaded.
     */
    public List<S3UploadedPart> listUploadedParts(String s3Key, String s3UploadId) {
        ListPartsRequest request = ListPartsRequest.builder()
            .bucket(uploadsBucket)
            .key(s3Key)
            .uploadId(s3UploadId)
            .build();
 
        return s3Client.listParts(request).parts().stream()
            .map(part -> S3UploadedPart.builder()
                .partNumber(part.partNumber())
                .etag(part.eTag())
                .sizeBytes(part.size())
                .build())
            .toList();
    }
 
    /**
     * Configure S3 lifecycle rule to auto-abort stale multipart uploads.
     * Run this once during application startup or via AWS Console.
     * Without this, incomplete parts continue to consume storage and incur charges.
     */
    @PostConstruct
    public void ensureMultipartCleanupLifecycleRule() {
        try {
            LifecycleRule cleanupRule = LifecycleRule.builder()
                .id("abort-incomplete-multipart-uploads")
                .status(ExpirationStatus.ENABLED)
                .filter(f -> f.prefix("uploads/"))
                .abortIncompleteMultipartUpload(
                    a -> a.daysAfterInitiation(7)  // Abort after 7 days if not completed
                )
                .build();
 
            s3Client.putBucketLifecycleConfiguration(r -> r
                .bucket(uploadsBucket)
                .lifecycleConfiguration(c -> c.rules(cleanupRule))
            );
 
            log.info("S3 lifecycle rule for multipart cleanup is configured.");
        } catch (Exception e) {
            log.warn("Could not configure S3 lifecycle rule: {}", e.getMessage());
        }
    }
}

@RestController
@RequestMapping("/api/upload/s3")
@Slf4j
@RequiredArgsConstructor
public class S3UploadController {
 
    private final S3MultipartUploadService s3Service;
    private final UploadSessionService sessionService;
 
    /**
     * Client calls this to get a list of presigned URLs.
     * Client then PUTs each part directly to S3.
     * No file bytes pass through the Spring Boot server.
     */
    @PostMapping("/initiate")
    @PreAuthorize("isAuthenticated()")
    public ResponseEntity<InitiateS3UploadResponse> initiateUpload(
            @RequestBody InitiateS3UploadRequest request,
            @AuthenticationPrincipal UserDetails currentUser) {
 
        // Calculate number of parts
        long partSize = 10L * 1024 * 1024;  // 10 MB per part
        int partCount = (int) Math.ceil((double) request.getFileSizeBytes() / partSize);
 
        if (partCount > 10_000) {  // S3 max: 10,000 parts
            return ResponseEntity.status(HttpStatus.PAYLOAD_TOO_LARGE).build();
        }
 
        String s3Key = "uploads/" + currentUser.getUsername() + "/"
                + UUID.randomUUID() + "/" + sanitizeFilename(request.getFilename());
 
        // 1. Create multipart upload on S3
        String s3UploadId = s3Service.initiateS3MultipartUpload(
            s3Key, request.getContentType(),
            Map.of("original-filename", request.getFilename(), "user-id", currentUser.getUsername())
        );
 
        // 2. Create local session record for tracking
        UploadSession session = sessionService.createSession(UploadSession.builder()
            .uploadId(UUID.randomUUID().toString())
            .userId(currentUser.getUsername())
            .filename(request.getFilename())
            .contentType(request.getContentType())
            .totalLength(request.getFileSizeBytes())
            .receivedOffset(0)
            .status(UploadStatus.INITIATED)
            .storagePath(s3Key)
            .storageUploadId(s3UploadId)
            .createdAt(Instant.now())
            .expiresAt(Instant.now().plus(7, ChronoUnit.DAYS))
            .build());
 
        // 3. Generate presigned URLs for each part
        List<PresignedPartInfo> presignedUrls = s3Service.generatePresignedPartUrls(
            s3Key, s3UploadId, partCount, Duration.ofHours(1)
        );
 
        return ResponseEntity.ok(InitiateS3UploadResponse.builder()
            .uploadId(session.getUploadId())
            .s3Key(s3Key)
            .partSize(partSize)
            .parts(presignedUrls)
            .expiresAt(session.getExpiresAt())
            .build());
    }
 
    /**
     * Client calls this after successfully PUTting all parts to S3.
     * Triggers S3 to assemble the parts into a single object.
     */
    @PostMapping("/{uploadId}/complete")
    @PreAuthorize("isAuthenticated()")
    public ResponseEntity<CompleteUploadResponse> completeUpload(
            @PathVariable String uploadId,
            @RequestBody CompleteS3UploadRequest request,
            @AuthenticationPrincipal UserDetails currentUser) {
 
        UploadSession session = sessionService.findById(uploadId)
            .orElseThrow(() -> new UploadNotFoundException(uploadId));
 
        // Security: ensure user owns this upload
        if (!session.getUserId().equals(currentUser.getUsername())) {
            return ResponseEntity.status(HttpStatus.FORBIDDEN).build();
        }
 
        // Tell S3 to assemble the parts
        String finalS3Url = s3Service.completeMultipartUpload(
            session.getStoragePath(),
            session.getStorageUploadId(),
            request.getParts()
        );
 
        session.setStatus(UploadStatus.COMPLETE);
        session.setCompletedAt(Instant.now());
        sessionService.save(session);
 
        // Publish event for post-upload processing (virus scan, thumbnail generation, etc.)
        eventPublisher.publishEvent(new UploadCompletedEvent(uploadId, finalS3Url));
 
        return ResponseEntity.ok(CompleteUploadResponse.builder()
            .uploadId(uploadId)
            .fileUrl(finalS3Url)
            .completedAt(session.getCompletedAt())
            .build());
    }
 
    /**
     * Resume a previous upload: return presigned URLs for parts not yet uploaded.
     */
    @GetMapping("/{uploadId}/resume")
    @PreAuthorize("isAuthenticated()")
    public ResponseEntity<ResumeUploadResponse> resumeUpload(
            @PathVariable String uploadId,
            @AuthenticationPrincipal UserDetails currentUser) {
 
        UploadSession session = sessionService.findById(uploadId)
            .orElseThrow(() -> new UploadNotFoundException(uploadId));
 
        if (!session.getUserId().equals(currentUser.getUsername())) {
            return ResponseEntity.status(HttpStatus.FORBIDDEN).build();
        }
 
        if (session.isExpired()) {
            return ResponseEntity.status(HttpStatus.GONE).build();
        }
 
        // Find which parts S3 already has
        List<S3UploadedPart> uploadedParts = s3Service.listUploadedParts(
            session.getStoragePath(), session.getStorageUploadId()
        );
        Set<Integer> uploadedPartNumbers = uploadedParts.stream()
            .map(S3UploadedPart::getPartNumber)
            .collect(Collectors.toSet());
 
        // Calculate which parts still need uploading
        long partSize = 10L * 1024 * 1024;
        int totalParts = (int) Math.ceil((double) session.getTotalLength() / partSize);
 
        List<PresignedPartInfo> remainingParts = new ArrayList<>();
        for (int part = 1; part <= totalParts; part++) {
            if (!uploadedPartNumbers.contains(part)) {
                // Generate presigned URL for this remaining part
                remainingParts.addAll(s3Service.generatePresignedPartUrls(
                    session.getStoragePath(), session.getStorageUploadId(),
                    1, Duration.ofHours(1)  // One URL at a time for remaining parts
                ));
            }
        }
 
        return ResponseEntity.ok(ResumeUploadResponse.builder()
            .uploadId(uploadId)
            .uploadedParts(uploadedParts)       // Already done
            .remainingParts(remainingParts)     // Still need to upload
            .partSize(partSize)
            .expiresAt(session.getExpiresAt())
            .build());
    }
}

@Service
@Transactional
@Slf4j
@RequiredArgsConstructor
public class UploadSessionService {
 
    private final UploadSessionRepository sessionRepository;
    private final RedisTemplate<String, Long> redisTemplate;  // Fast offset cache
 
    private static final String REDIS_OFFSET_KEY = "upload:offset:";
    private static final Duration OFFSET_CACHE_TTL = Duration.ofDays(8);
 
    public UploadSession createSession(UploadSession session) {
        session = sessionRepository.save(session);
 
        // Cache offset in Redis for fast HEAD queries (avoids DB hit on every chunk)
        redisTemplate.opsForValue().set(
            REDIS_OFFSET_KEY + session.getUploadId(),
            0L,
            OFFSET_CACHE_TTL
        );
 
        log.info("Created upload session: id={}, size={}, user={}",
            session.getUploadId(), session.getTotalLength(), session.getUserId());
        return session;
    }
 
    /**
     * Append a chunk to the session.
     * Uses optimistic locking (@Version) to prevent concurrent writes to the same session.
     * If two clients try to write simultaneously, one will get an OptimisticLockException.
     */
    public long appendChunk(String uploadId, long expectedOffset, long bytesWritten) {
        UploadSession session = sessionRepository.findById(uploadId)
            .orElseThrow(() -> new UploadNotFoundException(uploadId));
 
        if (session.getReceivedOffset() != expectedOffset) {
            throw new OffsetConflictException(
                "Expected offset " + expectedOffset + " but session has " + session.getReceivedOffset()
            );
        }
 
        long newOffset = expectedOffset + bytesWritten;
        session.setReceivedOffset(newOffset);
        session.setStatus(UploadStatus.IN_PROGRESS);
 
        if (newOffset >= session.getTotalLength()) {
            session.setStatus(UploadStatus.COMPLETE);
            session.setCompletedAt(Instant.now());
        }
 
        sessionRepository.save(session);  // May throw OptimisticLockException on version conflict
 
        // Update Redis offset cache
        redisTemplate.opsForValue().set(
            REDIS_OFFSET_KEY + uploadId, newOffset, OFFSET_CACHE_TTL
        );
 
        return newOffset;
    }
 
    /**
     * Get current offset from Redis (fast) or fall back to DB.
     * This is called on every TUS HEAD request to check resume position.
     */
    public long getOffset(String uploadId) {
        Long cachedOffset = redisTemplate.opsForValue().get(REDIS_OFFSET_KEY + uploadId);
        if (cachedOffset != null) {
            return cachedOffset;
        }
 
        // Cache miss: read from DB and repopulate cache
        UploadSession session = sessionRepository.findById(uploadId)
            .orElseThrow(() -> new UploadNotFoundException(uploadId));
 
        redisTemplate.opsForValue().set(
            REDIS_OFFSET_KEY + uploadId, session.getReceivedOffset(), OFFSET_CACHE_TTL
        );
 
        return session.getReceivedOffset();
    }
 
    /**
     * Scheduled job: expire sessions older than 7 days.
     * Runs every hour. Marks them EXPIRED and triggers cleanup (abort S3 multipart).
     */
    @Scheduled(cron = "0 0 * * * *")  // Every hour on the hour
    public void expireOldSessions() {
        List<UploadSession> expiredSessions = sessionRepository
            .findByStatusInAndExpiresAtBefore(
                List.of(UploadStatus.INITIATED, UploadStatus.IN_PROGRESS, UploadStatus.PAUSED),
                Instant.now()
            );
 
        for (UploadSession session : expiredSessions) {
            session.setStatus(UploadStatus.EXPIRED);
            sessionRepository.save(session);
 
            // Clean up Redis cache
            redisTemplate.delete(REDIS_OFFSET_KEY + session.getUploadId());
 
            // If S3 multipart was initiated, abort it
            if (session.getStorageUploadId() != null) {
                eventPublisher.publishEvent(new UploadExpiredEvent(
                    session.getUploadId(), session.getStoragePath(), session.getStorageUploadId()
                ));
            }
 
            log.info("Expired upload session: id={}, user={}, progress={}%",
                session.getUploadId(), session.getUserId(), session.progressPercent());
        }
 
        if (!expiredSessions.isEmpty()) {
            log.info("Expired {} stale upload sessions", expiredSessions.size());
        }
    }
}

@Repository
public interface UploadSessionRepository extends JpaRepository<UploadSession, String> {
 
    List<UploadSession> findByUserIdAndStatusIn(String userId, List<UploadStatus> statuses);
 
    List<UploadSession> findByStatusInAndExpiresAtBefore(
        List<UploadStatus> statuses, Instant before
    );
 
    // Used for metrics
    long countByStatus(UploadStatus status);
 
    @Query("SELECT SUM(u.receivedOffset) FROM UploadSession u WHERE u.userId = :userId AND u.status = 'COMPLETE'")
    Long sumCompletedBytesForUser(@Param("userId") String userId);
}

-- V1__create_upload_tables.sql
 
CREATE TABLE upload_sessions (
    upload_id          VARCHAR(36)     PRIMARY KEY,
    user_id            VARCHAR(36)     NOT NULL,
    filename           VARCHAR(255)    NOT NULL,
    content_type       VARCHAR(128),
    total_length       BIGINT          NOT NULL,
    received_offset    BIGINT          NOT NULL DEFAULT 0,
    status             VARCHAR(20)     NOT NULL,
    storage_path       VARCHAR(500),
    storage_upload_id  VARCHAR(500),
    checksum_md5       VARCHAR(32),
    created_at         TIMESTAMPTZ     NOT NULL,
    expires_at         TIMESTAMPTZ     NOT NULL,
    completed_at       TIMESTAMPTZ,
    version            BIGINT          NOT NULL DEFAULT 0
);
 
CREATE INDEX idx_upload_sessions_user_status ON upload_sessions(user_id, status);
CREATE INDEX idx_upload_sessions_status ON upload_sessions(status);
CREATE INDEX idx_upload_sessions_expires ON upload_sessions(expires_at)
    WHERE status IN ('INITIATED', 'IN_PROGRESS', 'PAUSED');
 
CREATE TABLE upload_parts (
    id                 BIGSERIAL       PRIMARY KEY,
    upload_session_id  VARCHAR(36)     NOT NULL REFERENCES upload_sessions(upload_id) ON DELETE CASCADE,
    part_number        INTEGER         NOT NULL,
    etag               VARCHAR(64),
    part_size          BIGINT,
    uploaded_at        TIMESTAMPTZ,
    UNIQUE (upload_session_id, part_number)
);
 
CREATE INDEX idx_upload_parts_session ON upload_parts(upload_session_id);

@RestController
@RequestMapping("/api/files")
@Slf4j
@RequiredArgsConstructor
public class ResumableDownloadController {
 
    private final FileMetadataService fileMetadataService;
    private final S3DownloadService s3DownloadService;
    private final PresignedUrlService presignedUrlService;
    private final MeterRegistry meterRegistry;
 
    /**
     * Download endpoint with full HTTP Range request support.
     *
     * Supports:
     * - Full download (no Range header): 200 OK, full file
     * - Partial download (Range header): 206 Partial Content
     * - Resume with validation (If-Range header): honors range only if file hasn't changed
     * - Invalid range: 416 Range Not Satisfiable
     */
    @GetMapping("/{fileId}/content")
    public ResponseEntity<StreamingResponseBody> downloadFile(
            @PathVariable String fileId,
            @RequestHeader(value = "Range",    required = false) String rangeHeader,
            @RequestHeader(value = "If-Range", required = false) String ifRangeHeader,
            @RequestHeader(value = "If-None-Match", required = false) String ifNoneMatch,
            HttpServletRequest request) {
 
        FileMetadata file = fileMetadataService.findById(fileId)
            .orElseThrow(() -> new FileNotFoundException(fileId));
 
        // Authorization check (implement based on your security model)
        validateDownloadAccess(file, request);
 
        // Handle ETag cache hit: return 304 Not Modified
        if (ifNoneMatch != null && ifNoneMatch.equals("\"" + file.getEtag() + "\"")) {
            return ResponseEntity.status(HttpStatus.NOT_MODIFIED).build();
        }
 
        long fileSize = file.getSizeBytes();
 
        // If-Range validation: if the file changed since the client started downloading,
        // ignore the Range header and serve the full new file.
        boolean fileChanged = false;
        if (ifRangeHeader != null) {
            // If-Range can be an ETag or a date
            if (ifRangeHeader.startsWith("\"")) {
                // ETag comparison
                String requestedEtag = ifRangeHeader.replaceAll("\"", "");
                fileChanged = !requestedEtag.equals(file.getEtag());
            } else {
                // Date comparison
                try {
                    Instant ifRangeDate = Instant.parse(ifRangeHeader);
                    fileChanged = file.getLastModified().isAfter(ifRangeDate);
                } catch (DateTimeParseException e) {
                    fileChanged = true;  // Invalid date: treat as changed
                }
            }
        }
 
        // Determine effective range
        ParsedRange parsedRange = null;
        if (rangeHeader != null && !fileChanged) {
            parsedRange = parseRangeHeader(rangeHeader, fileSize);
            if (parsedRange.isInvalid()) {
                return ResponseEntity.status(HttpStatus.REQUESTED_RANGE_NOT_SATISFIABLE)
                    .header("Content-Range", "bytes */" + fileSize)
                    .build();
            }
        }
 
        long start  = parsedRange != null ? parsedRange.start() : 0L;
        long end    = parsedRange != null ? parsedRange.end()   : fileSize - 1;
        long length = end - start + 1;
        boolean isPartial = parsedRange != null;
 
        // Build streaming response body
        StreamingResponseBody streamBody = buildStreamingBody(file, start, length, fileId);
 
        // Track download metrics
        Counter.builder("file.downloads.total")
            .tag("partial", String.valueOf(isPartial))
            .tag("file_type", file.getContentType().split("/")[0])
            .register(meterRegistry)
            .increment();
 
        ResponseEntity.BodyBuilder responseBuilder = ResponseEntity
            .status(isPartial ? HttpStatus.PARTIAL_CONTENT : HttpStatus.OK)
            .header("Content-Type",    file.getContentType())
            .header("Content-Length",  String.valueOf(length))
            .header("Accept-Ranges",   "bytes")
            .header("ETag",            "\"" + file.getEtag() + "\"")
            .header("Last-Modified",   DateTimeFormatter.RFC_1123_DATE_TIME.format(
                                           file.getLastModified().atZone(ZoneOffset.UTC)))
            // Allows CDN (CloudFront/Cloudflare) to cache public files
            .header("Cache-Control",   file.isPublic() ? "public, max-age=86400" : "private, no-store")
            .header("Content-Disposition", "attachment; filename=\"" + file.getFilename() + "\"");
 
        if (isPartial) {
            responseBuilder.header("Content-Range",
                "bytes " + start + "-" + end + "/" + fileSize);
        }
 
        return responseBuilder.body(streamBody);
    }
 
    private StreamingResponseBody buildStreamingBody(
            FileMetadata file, long startByte, long length, String fileId) {
 
        return outputStream -> {
            long startTime = System.currentTimeMillis();
            long bytesStreamed = 0L;
 
            try (InputStream s3Stream = s3DownloadService.openStreamAtOffset(
                    file.getS3Key(), startByte, length)) {
 
                byte[] buffer = new byte[64 * 1024];  // 64 KB streaming buffer
                long remaining = length;
 
                int bytesRead;
                while (remaining > 0 &&
                       (bytesRead = s3Stream.read(buffer, 0, (int) Math.min(buffer.length, remaining))) != -1) {
 
                    outputStream.write(buffer, 0, bytesRead);
                    bytesStreamed += bytesRead;
                    remaining    -= bytesRead;
 
                    // Periodic flush - sends data to client without waiting for full buffer
                    // This is what enables smooth video playback and progress visibility
                    if (bytesStreamed % (1024 * 1024) == 0) {  // Flush every 1 MB
                        outputStream.flush();
                    }
                }
 
                outputStream.flush();  // Final flush
 
                long durationMs = System.currentTimeMillis() - startTime;
                double mbps = (bytesStreamed / 1024.0 / 1024.0) / (durationMs / 1000.0);
 
                log.debug("Streamed {} bytes for file {} in {}ms ({:.1f} MB/s)",
                    bytesStreamed, fileId, durationMs, mbps);
 
            } catch (IOException e) {
                if (isClientDisconnect(e)) {
                    log.debug("Client disconnected mid-download. fileId={}, bytesStreamed={}",
                        fileId, bytesStreamed);
                } else {
                    log.error("Error streaming file {}: {}", fileId, e.getMessage());
                    throw e;
                }
            }
        };
    }
 
    private boolean isClientDisconnect(IOException e) {
        String msg = e.getMessage();
        return msg != null && (
            msg.contains("Broken pipe") ||
            msg.contains("Connection reset by peer") ||
            msg.contains("Connection reset")
        );
    }
 
    private ParsedRange parseRangeHeader(String rangeHeader, long fileSize) {
        if (!rangeHeader.startsWith("bytes=")) {
            return ParsedRange.invalid();
        }
 
        String rangeSpec = rangeHeader.substring(6);
 
        // Only support first range in multi-range requests (for simplicity)
        // Full implementation would handle multipart/byteranges responses
        String firstRange = rangeSpec.split(",")[0].trim();
        String[] parts = firstRange.split("-", -1);
 
        try {
            long start, end;
 
            if (parts[0].isEmpty()) {
                // Suffix: bytes=-500 (last 500 bytes)
                long suffixLength = Long.parseLong(parts[1]);
                start = Math.max(0, fileSize - suffixLength);
                end = fileSize - 1;
            } else {
                start = Long.parseLong(parts[0]);
                end = parts[1].isEmpty() ? fileSize - 1 : Long.parseLong(parts[1]);
            }
 
            if (start > end || end >= fileSize || start < 0) {
                return ParsedRange.invalid();
            }
 
            return new ParsedRange(start, end, false);
 
        } catch (NumberFormatException e) {
            return ParsedRange.invalid();
        }
    }
 
    record ParsedRange(long start, long end, boolean invalid) {
        static ParsedRange invalid() { return new ParsedRange(0, 0, true); }
        boolean isInvalid() { return invalid; }
    }
}

@Service
@RequiredArgsConstructor
@Slf4j
public class S3DownloadService {
 
    private final S3Client s3Client;
 
    @Value("${s3.bucket.uploads}")
    private String bucket;
 
    /**
     * Open an InputStream to S3 starting at the given byte offset.
     * Uses S3 GetObject with Range header for efficient partial reads.
     * S3 serves only the requested bytes — no unnecessary data transfer.
     */
    public InputStream openStreamAtOffset(String s3Key, long startByte, long length) {
        String rangeHeader = "bytes=" + startByte + "-" + (startByte + length - 1);
 
        GetObjectRequest request = GetObjectRequest.builder()
            .bucket(bucket)
            .key(s3Key)
            .range(rangeHeader)  // S3 supports Range: bytes=X-Y natively
            .build();
 
        ResponseInputStream<GetObjectResponse> response = s3Client.getObject(request);
 
        log.debug("Opened S3 range stream: key={}, range={}", s3Key, rangeHeader);
        return response;
    }
 
    /**
     * Generate a presigned download URL (time-limited, no auth required).
     * Use for content that should be accessible without Spring Boot authentication.
     * Useful for: CDN distribution, sharing, browser-native video playback.
     */
    public String generatePresignedDownloadUrl(String s3Key, Duration validity, String filename) {
        GetObjectPresignRequest presignRequest = GetObjectPresignRequest.builder()
            .signatureDuration(validity)
            .getObjectRequest(r -> r
                .bucket(bucket)
                .key(s3Key)
                .responseContentDisposition("attachment; filename=\"" + filename + "\"")
            )
            .build();
 
        return s3Presigner.presignGetObject(presignRequest).url().toString();
    }
}

@GetMapping("/{fileId}/stream")
public ResponseEntity<StreamingResponseBody> streamVideo(
        @PathVariable String fileId,
        @RequestHeader(value = "Range", required = false) String rangeHeader) {
 
    FileMetadata file = fileMetadataService.findById(fileId).orElseThrow();
 
    if (!file.getContentType().startsWith("video/")) {
        return ResponseEntity.status(HttpStatus.UNSUPPORTED_MEDIA_TYPE).build();
    }
 
    long fileSize = file.getSizeBytes();
 
    // For video: default to streaming the first 1 MB if no Range specified
    // (browsers will request more as needed)
    long start  = 0;
    long end    = fileSize - 1;
    boolean isPartial = rangeHeader != null;
 
    if (rangeHeader != null) {
        ParsedRange parsed = parseRangeHeader(rangeHeader, fileSize);
        if (parsed.isInvalid()) {
            return ResponseEntity.status(416)
                .header("Content-Range", "bytes */" + fileSize)
                .build();
        }
        start = parsed.start();
        end   = parsed.end();
    } else {
        // First request with no Range: serve first 1 MB and let browser request more
        end = Math.min(fileSize - 1, 1024 * 1024 - 1);
        isPartial = end < fileSize - 1;
    }
 
    long length = end - start + 1;
    final long finalStart = start;
    final long finalLength = length;
 
    StreamingResponseBody body = outputStream -> {
        try (InputStream stream = s3DownloadService.openStreamAtOffset(
                file.getS3Key(), finalStart, finalLength)) {
            stream.transferTo(outputStream);
            outputStream.flush();
        }
    };
 
    return ResponseEntity.status(isPartial ? 206 : 200)
        // Note: NO Content-Disposition for streaming video (allows browser to play inline)
        .header("Content-Type",   file.getContentType())   // Must be "video/mp4" etc.
        .header("Content-Length", String.valueOf(length))
        .header("Accept-Ranges",  "bytes")                  // Critical for video seek to work
        .header("Content-Range",  isPartial
            ? "bytes " + start + "-" + end + "/" + fileSize
            : null)
        .header("Cache-Control",  "no-cache")               // Don't cache video segments in CDN
        .body(body);
}

@Service
@Slf4j
@RequiredArgsConstructor
public class GcsResumableUploadService {
 
    private final RestTemplate restTemplate;
    private final GoogleCredentials credentials;
 
    @Value("${gcs.bucket}")
    private String gcsBucket;
 
    private static final String GCS_UPLOAD_URL =
        "https://storage.googleapis.com/upload/storage/v1/b/{bucket}/o?uploadType=resumable";
    private static final int CHUNK_SIZE = 8 * 256 * 1024;  // 2 MB (multiple of 256KB)
 
    /**
     * Initiate a GCS resumable upload session.
     * Returns the session URI to use for all subsequent PATCH requests.
     */
    public String initiateResumableUpload(String objectName, String contentType, long totalSize) throws IOException {
        String accessToken = credentials.refreshIfExpired().getAccessToken().getTokenValue();
 
        HttpHeaders headers = new HttpHeaders();
        headers.setBearerAuth(accessToken);
        headers.setContentType(MediaType.APPLICATION_JSON);
        headers.set("X-Upload-Content-Type", contentType);
        headers.set("X-Upload-Content-Length", String.valueOf(totalSize));
 
        Map<String, Object> body = Map.of("name", objectName);
 
        ResponseEntity<Void> response = restTemplate.exchange(
            GCS_UPLOAD_URL, HttpMethod.POST,
            new HttpEntity<>(body, headers),
            Void.class,
            Map.of("bucket", gcsBucket)
        );
 
        String sessionUri = Objects.requireNonNull(response.getHeaders().getLocation()).toString();
        log.info("GCS resumable upload session created: {}", sessionUri.substring(0, 60) + "...");
        return sessionUri;
    }
 
    /**
     * Upload a chunk to GCS.
     * Returns the new offset confirmed by GCS.
     * Status 308 = more chunks needed.
     * Status 200/201 = upload complete.
     */
    public GcsUploadResult uploadChunk(
            String sessionUri,
            byte[] chunkData,
            long chunkOffset,
            long totalSize) throws IOException {
 
        long chunkEnd = chunkOffset + chunkData.length - 1;
        String contentRange = "bytes " + chunkOffset + "-" + chunkEnd + "/" + totalSize;
 
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_OCTET_STREAM);
        headers.set("Content-Range", contentRange);
 
        try {
            ResponseEntity<Void> response = restTemplate.exchange(
                sessionUri, HttpMethod.PUT,
                new HttpEntity<>(chunkData, headers),
                Void.class
            );
 
            if (response.getStatusCode().is2xxSuccessful()) {
                // Upload complete!
                return GcsUploadResult.complete();
            }
 
        } catch (HttpClientErrorException e) {
            if (e.getStatusCode().value() == 308) {
                // 308 Resume Incomplete = chunk received, more needed
                String rangeHeader = e.getResponseHeaders() != null
                    ? e.getResponseHeaders().getFirst("Range")
                    : null;
 
                long confirmedOffset = 0;
                if (rangeHeader != null && rangeHeader.startsWith("0-")) {
                    confirmedOffset = Long.parseLong(rangeHeader.substring(2)) + 1;
                }
                return GcsUploadResult.incomplete(confirmedOffset);
            }
            throw e;
        }
 
        return GcsUploadResult.incomplete(chunkOffset + chunkData.length);
    }
 
    /**
     * Query GCS for the current upload offset (for resume after interruption).
     */
    public long queryOffset(String sessionUri, long totalSize) {
        HttpHeaders headers = new HttpHeaders();
        headers.set("Content-Range", "bytes */" + totalSize);
 
        try {
            restTemplate.exchange(sessionUri, HttpMethod.PUT,
                new HttpEntity<>(new byte[0], headers), Void.class);
        } catch (HttpClientErrorException e) {
            if (e.getStatusCode().value() == 308) {
                String rangeHeader = e.getResponseHeaders() != null
                    ? e.getResponseHeaders().getFirst("Range")
                    : null;
                if (rangeHeader != null && rangeHeader.startsWith("0-")) {
                    return Long.parseLong(rangeHeader.substring(2)) + 1;
                }
                return 0;  // Nothing uploaded yet
            }
        }
        return totalSize;  // 200/201 = complete
    }
}

@RestController
@RequestMapping("/api/upload/progress")
@RequiredArgsConstructor
public class UploadProgressController {
 
    private final UploadSessionService sessionService;
 
    /**
     * Server-Sent Events endpoint for real-time upload progress.
     * Client subscribes once. Server pushes updates as chunks are received.
     * Automatically reconnects if connection drops.
     *
     * Usage:
     * const es = new EventSource('/api/upload/progress/abc123');
     * es.onmessage = event => updateProgressBar(JSON.parse(event.data));
     */
    @GetMapping(value = "/{uploadId}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter subscribeToProgress(@PathVariable String uploadId) {
        SseEmitter emitter = new SseEmitter(300_000L);  // 5 minute max SSE connection
 
        // Schedule periodic progress updates
        ScheduledFuture<?> task = scheduler.scheduleAtFixedRate(() -> {
            try {
                UploadSession session = sessionService.findById(uploadId).orElse(null);
                if (session == null) {
                    emitter.send(SseEmitter.event()
                        .name("error")
                        .data("{\"error\": \"Upload not found\"}"));
                    emitter.complete();
                    return;
                }
 
                ProgressUpdate update = ProgressUpdate.builder()
                    .uploadId(uploadId)
                    .bytesReceived(session.getReceivedOffset())
                    .totalBytes(session.getTotalLength())
                    .percentComplete(session.progressPercent())
                    .status(session.getStatus().name())
                    .build();
 
                emitter.send(SseEmitter.event()
                    .name("progress")
                    .data(update));
 
                if (session.getStatus() <mark class="obsidian-highlight"> UploadStatus.COMPLETE
                        || session.getStatus() </mark> UploadStatus.FAILED) {
                    emitter.complete();
                }
 
            } catch (IOException e) {
                emitter.completeWithError(e);
            }
        }, 0, 2, TimeUnit.SECONDS);  // Push update every 2 seconds
 
        emitter.onCompletion(() -> task.cancel(true));
        emitter.onTimeout(() -> task.cancel(true));
        emitter.onError(e -> task.cancel(true));
 
        return emitter;
    }
}

@Service
@RequiredArgsConstructor
public class PresignedUrlService {
 
    private final S3Presigner s3Presigner;
    private final HmacTokenService hmacTokenService;
 
    @Value("${s3.bucket.uploads}")
    private String bucket;
 
    /**
     * Generate a time-limited download URL that bypasses Spring Boot authentication.
     * Useful for: sharing files, CDN integration, browser-native video playback.
     */
    public String generateS3PresignedDownloadUrl(String s3Key, Duration validity) {
        PresignedGetObjectRequest presigned = s3Presigner.presignGetObject(r -> r
            .signatureDuration(validity)
            .getObjectRequest(req -> req
                .bucket(bucket)
                .key(s3Key)
            )
        );
        return presigned.url().toString();
    }
 
    /**
     * Generate an HMAC-signed download URL served by your Spring Boot app.
     * Use when you need: custom access control, download logging, rate limiting.
     * The URL contains a signed token. Spring Boot validates the signature on each request.
     *
     * URL format: /api/files/{fileId}/download?token={signedToken}&expires={timestamp}
     */
    public String generateSignedDownloadUrl(String fileId, String userId, Duration validity) {
        long expiresAt = Instant.now().plus(validity).getEpochSecond();
        String payload = fileId + ":" + userId + ":" + expiresAt;
        String signature = hmacTokenService.sign(payload);
 
        return String.format("/api/files/%s/download?token=%s&expires=%d",
            fileId, signature, expiresAt);
    }
 
    /**
     * Validate a signed download URL token.
     */
    public boolean validateSignedToken(String fileId, String userId, long expiresAt, String token) {
        if (Instant.now().getEpochSecond() > expiresAt) {
            return false;  // Expired
        }
 
        String payload = fileId + ":" + userId + ":" + expiresAt;
        String expectedToken = hmacTokenService.sign(payload);
        return MessageDigest.isEqual(
            token.getBytes(StandardCharsets.UTF_8),
            expectedToken.getBytes(StandardCharsets.UTF_8)
        );  // Constant-time comparison to prevent timing attacks
    }
}

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Testcontainers
class ChunkedUploadIntegrationTest {
 
    @Container
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:16")
        .withDatabaseName("testdb")
        .withUsername("test")
        .withPassword("test");
 
    @Container
    static GenericContainer<?> redis = new GenericContainer<>("redis:7-alpine")
        .withExposedPorts(6379);
 
    @Autowired
    private TestRestTemplate restTemplate;
 
    @Autowired
    private UploadSessionRepository sessionRepository;
 
    @Test
    void tusUploadShouldCompleteInThreeChunks() throws IOException {
        byte[] fileContent = "Hello World! ".repeat(1000).getBytes();
        int chunkSize = fileContent.length / 3 + 1;
 
        // Step 1: Create upload session
        HttpHeaders createHeaders = new HttpHeaders();
        createHeaders.set("Upload-Length", String.valueOf(fileContent.length));
        createHeaders.set("Upload-Metadata", "filename " + Base64.getEncoder().encodeToString("test.txt".getBytes()));
        createHeaders.set("Tus-Resumable", "1.0.0");
        createHeaders.setContentLength(0);
 
        ResponseEntity<Void> createResponse = restTemplate.exchange(
            "/files", HttpMethod.POST,
            new HttpEntity<>(null, createHeaders), Void.class
        );
 
        assertThat(createResponse.getStatusCode()).isEqualTo(HttpStatus.CREATED);
        String location = createResponse.getHeaders().getFirst("Location");
        assertThat(location).isNotNull();
 
        String uploadId = location.substring(location.lastIndexOf('/') + 1);
 
        // Step 2: Upload chunks
        long offset = 0;
        while (offset < fileContent.length) {
            byte[] chunk = Arrays.copyOfRange(fileContent, (int) offset,
                (int) Math.min(offset + chunkSize, fileContent.length));
 
            HttpHeaders chunkHeaders = new HttpHeaders();
            chunkHeaders.setContentType(MediaType.parseMediaType("application/offset+octet-stream"));
            chunkHeaders.setContentLength(chunk.length);
            chunkHeaders.set("Upload-Offset", String.valueOf(offset));
            chunkHeaders.set("Tus-Resumable", "1.0.0");
 
            ResponseEntity<Void> chunkResponse = restTemplate.exchange(
                "/files/" + uploadId, HttpMethod.PATCH,
                new HttpEntity<>(chunk, chunkHeaders), Void.class
            );
 
            assertThat(chunkResponse.getStatusCode()).isEqualTo(HttpStatus.NO_CONTENT);
 
            String newOffset = chunkResponse.getHeaders().getFirst("Upload-Offset");
            assertThat(newOffset).isNotNull();
            offset = Long.parseLong(newOffset);
        }
 
        // Verify upload is complete
        UploadSession session = sessionRepository.findById(uploadId).orElseThrow();
        assertThat(session.getStatus()).isEqualTo(UploadStatus.COMPLETE);
        assertThat(session.getReceivedOffset()).isEqualTo(fileContent.length);
    }
 
    @Test
    void resumeAfterInterruptedUpload() throws IOException {
        byte[] fileContent = new byte[10 * 1024];  // 10 KB file
        new Random().nextBytes(fileContent);
 
        // Create session
        String uploadId = createUploadSession(fileContent.length);
 
        // Upload first half only (simulating interruption)
        uploadChunk(uploadId, fileContent, 0, fileContent.length / 2);
 
        // Simulate reconnect: query offset
        ResponseEntity<Void> headResponse = restTemplate.exchange(
            "/files/" + uploadId, HttpMethod.HEAD,
            new HttpEntity<>(headersWithTus()), Void.class
        );
        long resumeOffset = Long.parseLong(headResponse.getHeaders().getFirst("Upload-Offset"));
        assertThat(resumeOffset).isEqualTo(fileContent.length / 2);
 
        // Resume from where we left off
        uploadChunk(uploadId, fileContent, resumeOffset, fileContent.length - (int)resumeOffset);
 
        // Verify complete
        UploadSession session = sessionRepository.findById(uploadId).orElseThrow();
        assertThat(session.getStatus()).isEqualTo(UploadStatus.COMPLETE);
    }
 
    @Test
    void rangeDownloadShouldReturn206WithCorrectBytes() throws IOException {
        // Given: a file with known content
        String fileId = createTestFileWithContent("ABCDEFGHIJ".repeat(100));
 
        // Request bytes 10-19 (10 bytes: "ABCDEFGHIJ" starting at position 10)
        HttpHeaders rangeHeaders = new HttpHeaders();
        rangeHeaders.set("Range", "bytes=10-19");
 
        ResponseEntity<byte[]> response = restTemplate.exchange(
            "/api/files/" + fileId + "/content", HttpMethod.GET,
            new HttpEntity<>(rangeHeaders), byte[].class
        );
 
        assertThat(response.getStatusCode()).isEqualTo(HttpStatus.PARTIAL_CONTENT);
        assertThat(response.getHeaders().getFirst("Content-Range")).isEqualTo("bytes 10-19/1000");
        assertThat(response.getHeaders().getFirst("Accept-Ranges")).isEqualTo("bytes");
        assertThat(new String(response.getBody())).isEqualTo("ABCDEFGHIJ");
    }
 
    @Test
    void invalidRangeShouldReturn416() {
        String fileId = createTestFileWithContent("Hello");  // 5 bytes
 
        HttpHeaders rangeHeaders = new HttpHeaders();
        rangeHeaders.set("Range", "bytes=10-20");  // Beyond end of file
 
        ResponseEntity<Void> response = restTemplate.exchange(
            "/api/files/" + fileId + "/content", HttpMethod.GET,
            new HttpEntity<>(rangeHeaders), Void.class
        );
 
        assertThat(response.getStatusCode()).isEqualTo(HttpStatus.REQUESTED_RANGE_NOT_SATISFIABLE);
        assertThat(response.getHeaders().getFirst("Content-Range")).isEqualTo("bytes */5");
    }
}

@Component
@RequiredArgsConstructor
@Slf4j
public class UploadMetrics {
 
    private final MeterRegistry meterRegistry;
    private final UploadSessionRepository sessionRepository;
 
    // Gauges updated every minute to reflect current state
    @Scheduled(fixedDelay = 60_000)
    public void updateGauges() {
        // Active uploads (currently receiving chunks)
        long activeCount = sessionRepository.countByStatus(UploadStatus.IN_PROGRESS);
        Gauge.builder("uploads.active", () -> activeCount)
            .description("Number of uploads currently in progress")
            .register(meterRegistry);
 
        // Stale uploads (initiated but no activity in 1 hour)
        long staleCount = sessionRepository.countByStatus(UploadStatus.INITIATED);
        Gauge.builder("uploads.stale.initiated", () -> staleCount)
            .description("Upload sessions created but not yet receiving data")
            .register(meterRegistry);
    }
 
    // Called from TusUploadController on each successful chunk
    public void recordChunkReceived(String uploadId, long bytes, long durationMs) {
        // Counter: total bytes received
        Counter.builder("uploads.bytes.received")
            .register(meterRegistry)
            .increment(bytes);
 
        // Timer: chunk processing duration
        Timer.builder("uploads.chunk.duration")
            .publishPercentiles(0.5, 0.95, 0.99)
            .register(meterRegistry)
            .record(Duration.ofMillis(durationMs));
 
        // Throughput: MB/s per chunk
        if (durationMs > 0) {
            double mbps = (bytes / 1024.0 / 1024.0) / (durationMs / 1000.0);
            DistributionSummary.builder("uploads.throughput.mbps")
                .register(meterRegistry)
                .record(mbps);
        }
    }
 
    // Called when upload completes
    public void recordUploadComplete(String uploadId, long totalBytes, long totalDurationMs) {
        Counter.builder("uploads.completed.total").register(meterRegistry).increment();
 
        Timer.builder("uploads.total.duration")
            .publishPercentiles(0.5, 0.95, 0.99)
            .register(meterRegistry)
            .record(Duration.ofMillis(totalDurationMs));
 
        log.info("Upload complete: id={}, size={}MB, duration={}s",
            uploadId, totalBytes / 1024 / 1024, totalDurationMs / 1000);
    }
 
    // Called on any upload failure
    public void recordUploadFailed(String uploadId, String reason) {
        Counter.builder("uploads.failed.total")
            .tag("reason", reason)
            .register(meterRegistry)
            .increment();
    }
}

@Service
@Slf4j
@RequiredArgsConstructor
public class ConcurrentUploadGuard {
 
    private final RedisTemplate<String, String> redisTemplate;
 
    private static final String LOCK_KEY_PREFIX = "upload:lock:";
    private static final Duration LOCK_TTL = Duration.ofSeconds(60);  // Lock expires if holder crashes
 
    /**
     * Acquire a distributed lock for a specific upload session.
     * Only one client can upload a chunk at a time.
     * Uses Redis SET NX (set if not exists) for atomicity.
     *
     * Returns a lock token if acquired, null if not (another client holds the lock).
     */
    public String tryAcquireLock(String uploadId) {
        String lockKey = LOCK_KEY_PREFIX + uploadId;
        String lockToken = UUID.randomUUID().toString();
 
        Boolean acquired = redisTemplate.opsForValue()
            .setIfAbsent(lockKey, lockToken, LOCK_TTL);
 
        return Boolean.TRUE.equals(acquired) ? lockToken : null;
    }
 
    /**
     * Release the lock. Only the holder (identified by lockToken) can release it.
     * Uses a Lua script for atomic check-and-delete.
     */
    public void releaseLock(String uploadId, String lockToken) {
        String lockKey = LOCK_KEY_PREFIX + uploadId;
        String luaScript =
            "if redis.call('get', KEYS[1]) == ARGV[1] then " +
            "  return redis.call('del', KEYS[1]) " +
            "else " +
            "  return 0 " +
            "end";
 
        redisTemplate.execute(
            RedisScript.of(luaScript, Long.class),
            List.of(lockKey),
            lockToken
        );
    }
}

// Usage in TUS controller:
@PatchMapping(value = "/{uploadId}", consumes = "application/offset+octet-stream")
public ResponseEntity<Void> uploadChunk(@PathVariable String uploadId, ...) {
 
    // Try to acquire lock
    String lockToken = uploadGuard.tryAcquireLock(uploadId);
    if (lockToken == null) {
        // Another request is currently uploading for this session
        return ResponseEntity.status(HttpStatus.LOCKED)
            .header("Tus-Resumable", "1.0.0")
            .header("Retry-After", "2")  // Try again in 2 seconds
            .build();
    }
 
    try {
        // ... normal chunk upload logic ...
        return ResponseEntity.noContent().header("Tus-Resumable", "1.0.0").build();
    } finally {
        uploadGuard.releaseLock(uploadId, lockToken);
    }
}