forked from jsl/video_editing_poc
uploads:complete now goes through the storage port instead of being a pure metadata flip. I added provider upload metadata to UploadSession, wired it through src/main/java/org/example/videoclips/
application/VideoAssetService.java, src/main/java/org/example/videoclips/storage/ObjectStoragePort.java, and the JPA upload-session mapping, and added a migration at src/main/resources/db/ migration/V3__upload_session_provider_metadata.sql. The storage adapters now expose completeMultipartUpload(...), with the in-memory adapter as a no-op and the S3 adapter implementing the AWS multipart completion call. I also added an FFmpeg-backed clipper adapter in src/main/java/org/example/videoclips/processing/FfmpegVideoClipperAdapter.java, controlled by video-clipping.processing=ffmpeg. The stub clipper remains the default via video-clipping.processing=stub, so the app still has a safe fallback. The new FFmpeg configuration lives under src/main/java/org/example/videoclips/config/ VideoClippingProperties.java and src/main/resources/application.properties. I also extended the test expectations so the upload-session response now verifies providerUploadId in src/test/java/org/example/videoclips/api/VideoAssetControllerTest.java.
This commit is contained in:
JSLMPR
parent
67c7b65639
commit
7e8a214c3e
|
|
@ -80,21 +80,22 @@ public class VideoAssetService {
|
|||
long partSizeBytes = properties.getMultipartPartSizeBytes();
|
||||
int partCount = Math.max(1, (int) Math.ceil((double) request.contentLengthBytes() / partSizeBytes));
|
||||
Instant uploadExpiresAt = now.plus(properties.getUploadUrlTtlMinutes(), ChronoUnit.MINUTES);
|
||||
repository.saveUploadSession(new UploadSession(
|
||||
uploadId,
|
||||
assetId,
|
||||
UploadSession.Status.OPEN,
|
||||
partSizeBytes,
|
||||
partCount,
|
||||
uploadExpiresAt,
|
||||
null
|
||||
));
|
||||
ObjectStoragePort.UploadSessionDescriptor upload = objectStoragePort.createMultipartUpload(
|
||||
uploadId,
|
||||
partCount,
|
||||
partSizeBytes,
|
||||
uploadExpiresAt
|
||||
);
|
||||
repository.saveUploadSession(new UploadSession(
|
||||
uploadId,
|
||||
assetId,
|
||||
upload.providerUploadId(),
|
||||
UploadSession.Status.OPEN,
|
||||
partSizeBytes,
|
||||
partCount,
|
||||
uploadExpiresAt,
|
||||
null
|
||||
));
|
||||
appendJobEvent(assetId, "ASSET_CREATED", "Upload session created");
|
||||
|
||||
Map<String, Object> response = Map.of(
|
||||
|
|
@ -128,9 +129,17 @@ public class VideoAssetService {
|
|||
VideoAsset updated = asset.withUploadCompleted(request.sourceDurationSeconds(), Instant.now());
|
||||
repository.saveAsset(updated);
|
||||
UploadSession existingSession = requireUploadSession(request.uploadId());
|
||||
objectStoragePort.completeMultipartUpload(
|
||||
existingSession.providerUploadId(),
|
||||
existingSession.id(),
|
||||
request.parts().stream()
|
||||
.map(part -> new ObjectStoragePort.CompletedUploadPart(part.partNumber(), part.etag()))
|
||||
.toList()
|
||||
);
|
||||
repository.saveUploadSession(new UploadSession(
|
||||
existingSession.id(),
|
||||
existingSession.assetId(),
|
||||
existingSession.providerUploadId(),
|
||||
UploadSession.Status.COMPLETED,
|
||||
existingSession.partSizeBytes(),
|
||||
existingSession.partCount(),
|
||||
|
|
@ -205,6 +214,7 @@ public class VideoAssetService {
|
|||
Map<String, Object> response = new LinkedHashMap<>();
|
||||
response.put("assetId", assetId);
|
||||
response.put("uploadId", uploadSession.id());
|
||||
response.put("providerUploadId", uploadSession.providerUploadId());
|
||||
response.put("status", uploadSession.status().name());
|
||||
response.put("partSizeBytes", uploadSession.partSizeBytes());
|
||||
response.put("partCount", uploadSession.partCount());
|
||||
|
|
@ -287,6 +297,7 @@ public class VideoAssetService {
|
|||
repository.saveUploadSession(new UploadSession(
|
||||
uploadSession.id(),
|
||||
uploadSession.assetId(),
|
||||
uploadSession.providerUploadId(),
|
||||
UploadSession.Status.ABORTED,
|
||||
uploadSession.partSizeBytes(),
|
||||
uploadSession.partCount(),
|
||||
|
|
|
|||
|
|
@ -28,10 +28,14 @@ public class VideoClippingProperties {
|
|||
|
||||
private String queue = "memory";
|
||||
|
||||
private String processing = "stub";
|
||||
|
||||
private final S3 s3 = new S3();
|
||||
|
||||
private final DatabaseQueue databaseQueue = new DatabaseQueue();
|
||||
|
||||
private final Ffmpeg ffmpeg = new Ffmpeg();
|
||||
|
||||
public long getMaxFileSizeBytes() {
|
||||
return maxFileSizeBytes;
|
||||
}
|
||||
|
|
@ -88,6 +92,14 @@ public class VideoClippingProperties {
|
|||
this.queue = queue;
|
||||
}
|
||||
|
||||
public String getProcessing() {
|
||||
return processing;
|
||||
}
|
||||
|
||||
public void setProcessing(String processing) {
|
||||
this.processing = processing;
|
||||
}
|
||||
|
||||
public S3 getS3() {
|
||||
return s3;
|
||||
}
|
||||
|
|
@ -96,6 +108,10 @@ public class VideoClippingProperties {
|
|||
return databaseQueue;
|
||||
}
|
||||
|
||||
public Ffmpeg getFfmpeg() {
|
||||
return ffmpeg;
|
||||
}
|
||||
|
||||
public static class S3 {
|
||||
private String region = "us-east-1";
|
||||
private String bucket = "video-clipping-dev";
|
||||
|
|
@ -158,4 +174,34 @@ public class VideoClippingProperties {
|
|||
this.batchSize = batchSize;
|
||||
}
|
||||
}
|
||||
|
||||
public static class Ffmpeg {
|
||||
private String ffmpegBinary = "ffmpeg";
|
||||
private String inputDirectory = "./tmp/ffmpeg-input";
|
||||
private String outputDirectory = "./tmp/ffmpeg-output";
|
||||
|
||||
public String getFfmpegBinary() {
|
||||
return ffmpegBinary;
|
||||
}
|
||||
|
||||
public void setFfmpegBinary(String ffmpegBinary) {
|
||||
this.ffmpegBinary = ffmpegBinary;
|
||||
}
|
||||
|
||||
public String getInputDirectory() {
|
||||
return inputDirectory;
|
||||
}
|
||||
|
||||
public void setInputDirectory(String inputDirectory) {
|
||||
this.inputDirectory = inputDirectory;
|
||||
}
|
||||
|
||||
public String getOutputDirectory() {
|
||||
return outputDirectory;
|
||||
}
|
||||
|
||||
public void setOutputDirectory(String outputDirectory) {
|
||||
this.outputDirectory = outputDirectory;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ import java.time.Instant;
|
|||
public record UploadSession(
|
||||
String id,
|
||||
String assetId,
|
||||
String providerUploadId,
|
||||
Status status,
|
||||
long partSizeBytes,
|
||||
int partCount,
|
||||
|
|
|
|||
|
|
@ -53,6 +53,7 @@ public class JpaVideoClippingMapper {
|
|||
UploadSessionEntity entity = new UploadSessionEntity();
|
||||
entity.setId(session.id());
|
||||
entity.setAssetId(session.assetId());
|
||||
entity.setProviderUploadId(session.providerUploadId());
|
||||
entity.setStatus(session.status().name());
|
||||
entity.setPartSizeBytes(session.partSizeBytes());
|
||||
entity.setPartCount(session.partCount());
|
||||
|
|
@ -65,6 +66,7 @@ public class JpaVideoClippingMapper {
|
|||
return new UploadSession(
|
||||
entity.getId(),
|
||||
entity.getAssetId(),
|
||||
entity.getProviderUploadId(),
|
||||
UploadSession.Status.valueOf(entity.getStatus()),
|
||||
entity.getPartSizeBytes(),
|
||||
entity.getPartCount(),
|
||||
|
|
|
|||
|
|
@ -17,6 +17,9 @@ public class UploadSessionEntity {
|
|||
@Column(nullable = false, length = 64)
|
||||
private String assetId;
|
||||
|
||||
@Column(nullable = false, length = 128)
|
||||
private String providerUploadId;
|
||||
|
||||
@Column(nullable = false, length = 32)
|
||||
private String status;
|
||||
|
||||
|
|
@ -47,6 +50,14 @@ public class UploadSessionEntity {
|
|||
this.assetId = assetId;
|
||||
}
|
||||
|
||||
public String getProviderUploadId() {
|
||||
return providerUploadId;
|
||||
}
|
||||
|
||||
public void setProviderUploadId(String providerUploadId) {
|
||||
this.providerUploadId = providerUploadId;
|
||||
}
|
||||
|
||||
public String getStatus() {
|
||||
return status;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,129 @@
|
|||
package org.example.videoclips.processing;
|
||||
|
||||
import org.example.videoclips.config.VideoClippingProperties;
|
||||
import org.example.videoclips.domain.AccuracyMode;
|
||||
import org.example.videoclips.domain.Clip;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.time.Instant;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
|
||||
@Component
|
||||
@ConditionalOnProperty(name = "video-clipping.processing", havingValue = "ffmpeg")
|
||||
public class FfmpegVideoClipperAdapter implements VideoClipperPort {
|
||||
|
||||
private final VideoClippingProperties properties;
|
||||
|
||||
public FfmpegVideoClipperAdapter(VideoClippingProperties properties) {
|
||||
this.properties = properties;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<Clip> generateClips(
|
||||
String jobId,
|
||||
String assetId,
|
||||
long sourceDurationSeconds,
|
||||
int segmentDurationSeconds,
|
||||
AccuracyMode accuracyMode
|
||||
) {
|
||||
Path inputFile = Path.of(properties.getFfmpeg().getInputDirectory(), assetId + ".mp4");
|
||||
Path outputDir = Path.of(properties.getFfmpeg().getOutputDirectory(), jobId);
|
||||
if (!Files.exists(inputFile)) {
|
||||
throw new IllegalStateException("FFmpeg input file not found: " + inputFile);
|
||||
}
|
||||
try {
|
||||
Files.createDirectories(outputDir);
|
||||
} catch (IOException ex) {
|
||||
throw new IllegalStateException("Unable to create FFmpeg output directory", ex);
|
||||
}
|
||||
|
||||
ProcessBuilder processBuilder = new ProcessBuilder(buildCommand(inputFile, outputDir, segmentDurationSeconds, accuracyMode));
|
||||
processBuilder.redirectErrorStream(true);
|
||||
try {
|
||||
Process process = processBuilder.start();
|
||||
int exitCode = process.waitFor();
|
||||
if (exitCode != 0) {
|
||||
throw new IllegalStateException("FFmpeg exited with code " + exitCode);
|
||||
}
|
||||
} catch (IOException ex) {
|
||||
throw new IllegalStateException("Unable to start FFmpeg process", ex);
|
||||
} catch (InterruptedException ex) {
|
||||
Thread.currentThread().interrupt();
|
||||
throw new IllegalStateException("FFmpeg process interrupted", ex);
|
||||
}
|
||||
|
||||
return buildClipManifest(jobId, assetId, sourceDurationSeconds, segmentDurationSeconds, accuracyMode);
|
||||
}
|
||||
|
||||
private List<String> buildCommand(Path inputFile, Path outputDir, int segmentDurationSeconds, AccuracyMode accuracyMode) {
|
||||
List<String> command = new ArrayList<>();
|
||||
command.add(properties.getFfmpeg().getFfmpegBinary());
|
||||
command.add("-hide_banner");
|
||||
command.add("-y");
|
||||
command.add("-i");
|
||||
command.add(inputFile.toString());
|
||||
command.add("-map");
|
||||
command.add("0");
|
||||
if (accuracyMode == AccuracyMode.EXACT) {
|
||||
command.add("-c:v");
|
||||
command.add("libx264");
|
||||
command.add("-preset");
|
||||
command.add("veryfast");
|
||||
command.add("-crf");
|
||||
command.add("20");
|
||||
command.add("-force_key_frames");
|
||||
command.add("expr:gte(t,n_forced*" + segmentDurationSeconds + ")");
|
||||
command.add("-c:a");
|
||||
command.add("aac");
|
||||
command.add("-b:a");
|
||||
command.add("128k");
|
||||
} else {
|
||||
command.add("-c");
|
||||
command.add("copy");
|
||||
}
|
||||
command.add("-f");
|
||||
command.add("segment");
|
||||
command.add("-segment_time");
|
||||
command.add(String.valueOf(segmentDurationSeconds));
|
||||
command.add("-reset_timestamps");
|
||||
command.add("1");
|
||||
command.add(outputDir.resolve("clip_%05d.mp4").toString());
|
||||
return command;
|
||||
}
|
||||
|
||||
private List<Clip> buildClipManifest(
|
||||
String jobId,
|
||||
String assetId,
|
||||
long sourceDurationSeconds,
|
||||
int segmentDurationSeconds,
|
||||
AccuracyMode accuracyMode
|
||||
) {
|
||||
List<Clip> clips = new ArrayList<>();
|
||||
long start = 0;
|
||||
int index = 0;
|
||||
long bytesPerSecond = accuracyMode == AccuracyMode.EXACT ? 1_280_000L : 1_024_000L;
|
||||
while (start < sourceDurationSeconds) {
|
||||
long duration = Math.min(segmentDurationSeconds, sourceDurationSeconds - start);
|
||||
clips.add(new Clip(
|
||||
"clip_" + UUID.randomUUID().toString().replace("-", "").substring(0, 12),
|
||||
jobId,
|
||||
assetId,
|
||||
index,
|
||||
start,
|
||||
duration,
|
||||
"video/mp4",
|
||||
duration * bytesPerSecond,
|
||||
Instant.now()
|
||||
));
|
||||
start += segmentDurationSeconds;
|
||||
index++;
|
||||
}
|
||||
return clips;
|
||||
}
|
||||
}
|
||||
|
|
@ -2,6 +2,7 @@ package org.example.videoclips.processing;
|
|||
|
||||
import org.example.videoclips.domain.AccuracyMode;
|
||||
import org.example.videoclips.domain.Clip;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.time.Instant;
|
||||
|
|
@ -10,6 +11,7 @@ import java.util.List;
|
|||
import java.util.UUID;
|
||||
|
||||
@Component
|
||||
@ConditionalOnProperty(name = "video-clipping.processing", havingValue = "stub", matchIfMissing = true)
|
||||
public class StubVideoClipperAdapter implements VideoClipperPort {
|
||||
|
||||
@Override
|
||||
|
|
|
|||
|
|
@ -18,7 +18,12 @@ public class InMemoryObjectStorageAdapter implements ObjectStoragePort {
|
|||
"https://storage.example/uploads/" + uploadId + "/parts/" + partNumber
|
||||
))
|
||||
.toList();
|
||||
return new UploadSessionDescriptor(uploadId, "multipart", partSizeBytes, expiresAt, parts);
|
||||
return new UploadSessionDescriptor(uploadId, uploadId, "multipart", partSizeBytes, expiresAt, parts);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void completeMultipartUpload(String providerUploadId, String uploadId, List<CompletedUploadPart> parts) {
|
||||
// No-op for the in-memory adapter.
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
|
|||
|
|
@ -7,6 +7,8 @@ public interface ObjectStoragePort {
|
|||
|
||||
UploadSessionDescriptor createMultipartUpload(String uploadId, int partCount, long partSizeBytes, Instant expiresAt);
|
||||
|
||||
void completeMultipartUpload(String providerUploadId, String uploadId, List<CompletedUploadPart> parts);
|
||||
|
||||
String createClipDownloadUrl(String clipId, Instant expiresAt);
|
||||
|
||||
record UploadPartDescriptor(int partNumber, String url) {
|
||||
|
|
@ -14,10 +16,14 @@ public interface ObjectStoragePort {
|
|||
|
||||
record UploadSessionDescriptor(
|
||||
String uploadId,
|
||||
String providerUploadId,
|
||||
String mode,
|
||||
long partSizeBytes,
|
||||
Instant expiresAt,
|
||||
List<UploadPartDescriptor> parts
|
||||
) {
|
||||
}
|
||||
|
||||
record CompletedUploadPart(int partNumber, String etag) {
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -6,6 +6,9 @@ import org.springframework.stereotype.Component;
|
|||
import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
|
||||
import software.amazon.awssdk.regions.Region;
|
||||
import software.amazon.awssdk.services.s3.S3Configuration;
|
||||
import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
|
||||
import software.amazon.awssdk.services.s3.model.CompletedMultipartUpload;
|
||||
import software.amazon.awssdk.services.s3.model.CompletedPart;
|
||||
import software.amazon.awssdk.services.s3.model.GetObjectRequest;
|
||||
import software.amazon.awssdk.services.s3.model.UploadPartRequest;
|
||||
import software.amazon.awssdk.services.s3.presigner.S3Presigner;
|
||||
|
|
@ -25,19 +28,26 @@ import java.util.stream.IntStream;
|
|||
public class S3ObjectStorageAdapter implements ObjectStoragePort {
|
||||
|
||||
private final VideoClippingProperties properties;
|
||||
private final software.amazon.awssdk.services.s3.S3Client s3Client;
|
||||
private final S3Presigner s3Presigner;
|
||||
|
||||
public S3ObjectStorageAdapter(VideoClippingProperties properties) {
|
||||
this.properties = properties;
|
||||
Region region = Region.of(properties.getS3().getRegion());
|
||||
software.amazon.awssdk.services.s3.S3ClientBuilder clientBuilder = software.amazon.awssdk.services.s3.S3Client.builder()
|
||||
.region(region)
|
||||
.serviceConfiguration(S3Configuration.builder().pathStyleAccessEnabled(properties.getS3().isPathStyle()).build())
|
||||
.credentialsProvider(DefaultCredentialsProvider.create());
|
||||
S3Presigner.Builder presignerBuilder = S3Presigner.builder()
|
||||
.region(region)
|
||||
.serviceConfiguration(S3Configuration.builder().pathStyleAccessEnabled(properties.getS3().isPathStyle()).build())
|
||||
.credentialsProvider(DefaultCredentialsProvider.create());
|
||||
if (properties.getS3().getEndpoint() != null && !properties.getS3().getEndpoint().isBlank()) {
|
||||
URI endpoint = URI.create(properties.getS3().getEndpoint());
|
||||
clientBuilder = clientBuilder.endpointOverride(endpoint);
|
||||
presignerBuilder = presignerBuilder.endpointOverride(endpoint);
|
||||
}
|
||||
this.s3Client = clientBuilder.build();
|
||||
this.s3Presigner = presignerBuilder.build();
|
||||
}
|
||||
|
||||
|
|
@ -60,7 +70,21 @@ public class S3ObjectStorageAdapter implements ObjectStoragePort {
|
|||
return new UploadPartDescriptor(partNumber, presignedRequest.url().toString());
|
||||
})
|
||||
.toList();
|
||||
return new UploadSessionDescriptor(uploadId, "multipart", partSizeBytes, expiresAt, parts);
|
||||
return new UploadSessionDescriptor(uploadId, uploadId, "multipart", partSizeBytes, expiresAt, parts);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void completeMultipartUpload(String providerUploadId, String uploadId, List<CompletedUploadPart> parts) {
|
||||
String objectKey = "uploads/" + uploadId + "/source";
|
||||
List<CompletedPart> completedParts = parts.stream()
|
||||
.map(part -> CompletedPart.builder().partNumber(part.partNumber()).eTag(part.etag()).build())
|
||||
.toList();
|
||||
s3Client.completeMultipartUpload(CompleteMultipartUploadRequest.builder()
|
||||
.bucket(properties.getS3().getBucket())
|
||||
.key(objectKey)
|
||||
.uploadId(providerUploadId)
|
||||
.multipartUpload(CompletedMultipartUpload.builder().parts(completedParts).build())
|
||||
.build());
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
video-clipping.repository=jpa
|
||||
video-clipping.queue=db
|
||||
video-clipping.processing=stub
|
||||
|
||||
spring.datasource.url=jdbc:h2:mem:videoclips;MODE=PostgreSQL;DB_CLOSE_DELAY=-1;DB_CLOSE_ON_EXIT=FALSE
|
||||
spring.datasource.driver-class-name=org.h2.Driver
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ spring.jpa.open-in-view=false
|
|||
video-clipping.repository=memory
|
||||
video-clipping.storage=memory
|
||||
video-clipping.queue=memory
|
||||
video-clipping.processing=stub
|
||||
video-clipping.max-file-size-bytes=53687091200
|
||||
video-clipping.multipart-part-size-bytes=104857600
|
||||
video-clipping.upload-url-ttl-minutes=60
|
||||
|
|
@ -17,3 +18,6 @@ video-clipping.s3.bucket=video-clipping-dev
|
|||
video-clipping.s3.path-style=true
|
||||
video-clipping.database-queue.poll-interval-ms=1000
|
||||
video-clipping.database-queue.batch-size=5
|
||||
video-clipping.ffmpeg.ffmpeg-binary=ffmpeg
|
||||
video-clipping.ffmpeg.input-directory=./tmp/ffmpeg-input
|
||||
video-clipping.ffmpeg.output-directory=./tmp/ffmpeg-output
|
||||
|
|
|
|||
|
|
@ -0,0 +1,9 @@
|
|||
alter table upload_sessions
|
||||
add column if not exists provider_upload_id varchar(128);
|
||||
|
||||
update upload_sessions
|
||||
set provider_upload_id = id
|
||||
where provider_upload_id is null;
|
||||
|
||||
alter table upload_sessions
|
||||
alter column provider_upload_id set not null;
|
||||
|
|
@ -70,6 +70,7 @@ class VideoAssetControllerTest {
|
|||
mockMvc.perform(get("/v1/video-assets/{assetId}/upload-session", assetId))
|
||||
.andExpect(status().isOk())
|
||||
.andExpect(jsonPath("$.uploadId").value(uploadId))
|
||||
.andExpect(jsonPath("$.providerUploadId").value(uploadId))
|
||||
.andExpect(jsonPath("$.status").value("COMPLETED"));
|
||||
|
||||
mockMvc.perform(get("/v1/video-assets/{assetId}/events", assetId))
|
||||
|
|
|
|||
Loading…
Reference in New Issue