public class KVSExportRequest {
    private String streamName;
    @JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd'T'HH:mm:ss'Z'")
    private LocalDateTime startTime;
    @JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd'T'HH:mm:ss'Z'")
    private LocalDateTime endTime;
    private String s3Path;

    // Builder pattern implementation
    public static class Builder {
        private String streamName;
        private LocalDateTime startTime;
        private LocalDateTime endTime;
        private String s3Path;

        public Builder streamName(String streamName) {
            this.streamName = streamName;
            return this;
        }

        public Builder startTime(LocalDateTime startTime) {
            this.startTime = startTime;
            return this;
        }

        public Builder endTime(LocalDateTime endTime) {
            this.endTime = endTime;
            return this;
        }

        public Builder s3Path(String s3Path) {
            this.s3Path = s3Path;
            return this;
        }

        public KVSExportRequest build() {
            validateFields();
            KVSExportRequest request = new KVSExportRequest();
            request.streamName = this.streamName;
            request.startTime = this.startTime;
            request.endTime = this.endTime;
            request.s3Path = this.s3Path;
            return request;
        }

        private void validateFields() {
            if (streamName == null || streamName.isEmpty()) {
                throw new IllegalArgumentException("streamName cannot be null or empty");
            }
            if (startTime == null) {
                throw new IllegalArgumentException("startTime cannot be null");
            }
            if (endTime == null) {
                throw new IllegalArgumentException("endTime cannot be null");
            }
            if (endTime.isBefore(startTime)) {
                throw new IllegalArgumentException("endTime cannot be before startTime");
            }
            if (s3Path == null || s3Path.isEmpty()) {
                throw new IllegalArgumentException("s3Path cannot be null or empty");
            }
            if (!s3Path.startsWith("s3://")) {
                throw new IllegalArgumentException("s3Path must start with 's3://'");
            }
            if (!s3Path.endsWith(".mp4")) {
                throw new IllegalArgumentException("s3Path must end with '.mp4'");
            }
        }
    }

    public static Builder builder() {
        return new Builder();
    }

    // Getters
    public String getStreamName() { return streamName; }
    public LocalDateTime getStartTime() { return startTime; }
    public LocalDateTime getEndTime() { return endTime; }
    public String getS3Path() { return s3Path; }
}

public class KVSExportResponse {
    private String s3Path;
    private LocalDateTime lastExportedTimestamp;
    private boolean complete;

    // Builder pattern implementation
    public static class Builder {
        private String s3Path;
        private LocalDateTime lastExportedTimestamp;
        private boolean complete;

        public Builder s3Path(String s3Path) {
            this.s3Path = s3Path;
            return this;
        }

        public Builder lastExportedTimestamp(LocalDateTime lastExportedTimestamp) {
            this.lastExportedTimestamp = lastExportedTimestamp;
            return this;
        }

        public Builder complete(boolean complete) {
            this.complete = complete;
            return this;
        }

        public KVSExportResponse build() {
            KVSExportResponse response = new KVSExportResponse();
            response.s3Path = this.s3Path;
            response.lastExportedTimestamp = this.lastExportedTimestamp;
            response.complete = this.complete;
            return response;
        }
    }

    public static Builder builder() {
        return new Builder();
    }

    // Getters
    public String getS3Path() { return s3Path; }
    public LocalDateTime getLastExportedTimestamp() { return lastExportedTimestamp; }
    public boolean isComplete() { return complete; }
}

public class VideoExportBatch {
    private static final DateTimeFormatter TIMESTAMP_FORMAT = 
        DateTimeFormatter.ofPattern("HH:mm:ss,SSS");

    private final List<Fragment> fragments;
    private final LocalDateTime startTime;
    private final LocalDateTime endTime;
    private final String s3Path;
    private final int batchNumber;

    public VideoExportBatch(List<Fragment> fragments, LocalDateTime startTime, 
                          LocalDateTime endTime, String s3Path, int batchNumber) {
        this.fragments = fragments;
        this.startTime = startTime;
        this.endTime = endTime;
        this.s3Path = s3Path;
        this.batchNumber = batchNumber;
        validateInput(startTime, endTime, s3Path);
    }

    private void validateInput(LocalDateTime startTime, LocalDateTime endTime, String s3Path) {
        Objects.requireNonNull(startTime, "startTime cannot be null");
        Objects.requireNonNull(endTime, "endTime cannot be null");
        Objects.requireNonNull(s3Path, "s3Path cannot be null");

        if (endTime.isBefore(startTime)) {
            throw new IllegalArgumentException("endTime cannot be before startTime");
        }

        if (!s3Path.startsWith("s3://")) {
            throw new IllegalArgumentException("s3Path must start with 's3://'");
        }
    }

    public String generateSrtContent() {
        if (fragments == null || fragments.isEmpty()) {
            return "";
        }

        StringBuilder srt = new StringBuilder();
        int subtitleNumber = 1;

        for (Fragment fragment : fragments) {
            LocalDateTime fragmentStart = LocalDateTime.ofInstant(
                fragment.getProducerTimestamp().toInstant(), ZoneOffset.UTC);
            LocalDateTime fragmentEnd = fragmentStart.plusNanos(
                fragment.getFragmentLengthInMilliseconds() * 1_000_000L);
            
            if (fragmentEnd.isAfter(endTime)) {
                fragmentEnd = endTime;
            }
            
            if (fragmentStart.isBefore(startTime)) {
                fragmentStart = startTime;
            }

            if (fragmentStart.isBefore(fragmentEnd)) {
                srt.append(subtitleNumber).append("\n");
                srt.append(formatTimestamp(fragmentStart))
                   .append(" --> ")
                   .append(formatTimestamp(fragmentEnd))
                   .append("\n");
                srt.append("Fragment: ")
                   .append(fragment.getFragmentNumber())
                   .append("\n\n");
                subtitleNumber++;
            }
        }

        return srt.toString();
    }

    private String formatTimestamp(LocalDateTime time) {
        return time.format(TIMESTAMP_FORMAT);
    }

    // Getters
    public LocalDateTime getStartTime() { return startTime; }
    public LocalDateTime getEndTime() { return endTime; }
    public String getS3Path() { return s3Path; }
    public int getBatchNumber() { return batchNumber; }
}

@Override
public KVSExportResponse handleRequest(KVSExportRequest request, Context context) {
    logger.info("Processing export request for stream: {}", request.getStreamName());

    try {
        // Get KVS archive media client
        String endpointUrl = getArchiveMediaEndpoint(request.getStreamName());
        AmazonKinesisVideoArchivedMedia archiveMediaClient = createArchiveMediaClient(endpointUrl);

        // List fragments
        ListFragmentsResult fragments = archiveMediaClient.listFragments(new ListFragmentsRequest()
            .withStreamName(request.getStreamName())
            .withFragmentSelector(new FragmentSelector()
                .withFragmentSelectorType(FragmentSelectorType.PRODUCER_TIMESTAMP)
                .withTimestampRange(new TimestampRange()
                    .withStartTimestamp(Date.from(request.getStartTime().toInstant(ZoneOffset.UTC)))
                    .withEndTimestamp(Date.from(request.getEndTime().toInstant(ZoneOffset.UTC))))));

        if (fragments.getFragments().isEmpty()) {
            logger.info("No fragments found in the specified time range");
            return KVSExportResponse.builder()
                .s3Path(request.getS3Path())
                .complete(true)
                .lastExportedTimestamp(request.getStartTime())
                .build();
        }

        // Process fragments in batches
        List<Fragment> currentBatch = new ArrayList<>();
        LocalDateTime lastExportedTimestamp = request.getStartTime();
        boolean isComplete = true;
        int batchNumber = 1;

        for (Fragment fragment : fragments.getFragments()) {
            // Check remaining time
            if (context.getRemainingTimeInMillis() < MIN_REMAINING_TIME_MS) {
                logger.warn("Insufficient remaining time ({}ms). Stopping early.", 
                    context.getRemainingTimeInMillis());
                isComplete = false;
                break;
            }

            currentBatch.add(fragment);

            // Process batch if it's full or it's the last fragment
            if (currentBatch.size() >= MAX_FRAGMENTS_PER_BATCH || 
                fragment == fragments.getFragments().get(fragments.getFragments().size() - 1)) {
                
                LocalDateTime batchStartTime = LocalDateTime.ofInstant(
                    currentBatch.get(0).getProducerTimestamp().toInstant(), 
                    ZoneOffset.UTC
                );
                LocalDateTime batchEndTime = LocalDateTime.ofInstant(
                    currentBatch.get(currentBatch.size() - 1).getProducerTimestamp().toInstant(), 
                    ZoneOffset.UTC
                );

                VideoExportBatch batch = new VideoExportBatch(
                    new ArrayList<>(currentBatch),
                    batchStartTime,
                    batchEndTime,
                    request.getS3Path(),
                    batchNumber++
                );

                VideoExportUtil.processBatch(batch, archiveMediaClient, request.getStreamName(), s3Client);
                lastExportedTimestamp = batchEndTime;
                currentBatch.clear();
            }
        }

        return KVSExportResponse.builder()
            .s3Path(request.getS3Path())
            .complete(isComplete)
            .lastExportedTimestamp(lastExportedTimestamp)
            .build();

    } catch (Exception e) {
        logger.error("Error processing video export", e);
        throw new RuntimeException("Failed to process video export", e);
    }
}

public static void processBatch(VideoExportBatch batch, AmazonKinesisVideoArchivedMedia archiveMediaClient, 
                              String streamName, AmazonS3 s3Client) {
    Path tempDir = null;
    try {
        // Create temporary directory
        tempDir = Files.createTempDirectory(TEMP_DIR + File.separator + "kvs_export_");
        logger.info("Created temp directory: {}", tempDir);

        // Generate SRT file
        Path srtPath = tempDir.resolve("subtitles.srt");
        Files.writeString(srtPath, batch.generateSrtContent());
        logger.info("Generated SRT file at: {}", srtPath);

        // Download video clip
        Path videoPath = tempDir.resolve("clip.mp4");
        downloadClip(batch, archiveMediaClient, streamName, videoPath);
        logger.info("Downloaded video clip to: {}", videoPath);

        // Merge video and subtitles
        Path outputPath = tempDir.resolve("output.mp4");
        mergeVideoAndSubtitles(videoPath, srtPath, outputPath);
        logger.info("Merged video and subtitles to: {}", outputPath);

        // Upload to S3
        uploadToS3(outputPath.toFile(), batch.getS3Path(), s3Client);
        logger.info("Uploaded processed video to S3: {}", batch.getS3Path());

    } catch (Exception e) {
        logger.error("Error processing batch {}", batch.getBatchNumber(), e);
        throw new RuntimeException("Error processing video batch", e);
    } finally {
        // Cleanup temporary files
        if (tempDir != null) {
            try {
                Files.walk(tempDir)
                    .map(Path::toFile)
                    .forEach(File::delete);
                Files.delete(tempDir);
                logger.info("Cleaned up temporary directory: {}", tempDir);
            } catch (IOException e) {
                logger.warn("Failed to cleanup temporary directory: {}", tempDir, e);
            }
        }
    }
}