fix: jobless video assets no longer 500 the library click
Legacy/edge-case video_assets rows without a processing job threw 'has no processing job' from getVideoStatus — the library detail click 500ed. VideoStatus.processingJob is now nullable end to end (api type, DTO, web consumers null-guarded); a jobless asset renders as queued.
This commit is contained in:
parent
215952fe5e
commit
31c97a0b0a
13 changed files with 95 additions and 25 deletions
|
|
@ -346,6 +346,26 @@ export class GroupFilesService {
|
|||
return { item, download };
|
||||
}
|
||||
|
||||
/**
|
||||
* Range playback for DIRECT (SPENC2) objects: decrypt only the segments
|
||||
* covering [offset, offset+length). Segment-level GCM keeps every byte
|
||||
* authenticated. SPENC1 rows are rejected — their single-stream GCM cannot
|
||||
* be sliced, so the whole-file route serves them.
|
||||
*/
|
||||
async readRange(workspaceId: string, userId: string, role: string, itemId: string, offset: number, length: number): Promise<{ bytes: Buffer; mime: string; total: number | null }> {
|
||||
const item = await this.getFile(workspaceId, userId, role, itemId);
|
||||
if (item.status !== "available") throw new GoneException(`file is ${item.status}`);
|
||||
const storageKey = item.file.storageKey;
|
||||
if (storageKey === undefined || !storageKey.startsWith("direct/")) {
|
||||
throw new BadRequestException("range reads are only available for direct objects");
|
||||
}
|
||||
const storage = this.storage() as StorageService & { readDirectRange(key: string, start: number, length: number): Promise<Buffer> };
|
||||
const bytes = await storage.readDirectRange(storageKey, offset, length).catch(() => {
|
||||
throw new ServiceUnavailableException("file storage is unavailable");
|
||||
});
|
||||
return { bytes, mime: item.mimeType, total: item.sizeBytes };
|
||||
}
|
||||
|
||||
private async openPayload(storageKey: string | null | undefined, mime: string): Promise<PayloadDownload> {
|
||||
if (storageKey === null || storageKey === undefined) {
|
||||
throw new ServiceUnavailableException("file was stored on retired storage and is no longer served");
|
||||
|
|
|
|||
|
|
@ -50,12 +50,14 @@ export interface StorageService {
|
|||
openDirectHeader(stored: Buffer): Promise<Buffer>;
|
||||
/** Direct-to-R2: raw still-sealed object bytes — the CEK header lives only in the sealed form. */
|
||||
getSealedObject(key: string): Promise<Buffer>;
|
||||
/** Direct-to-R2: decrypt an arbitrary PLAINTEXT byte range (segment-aligned fetches). */
|
||||
readDirectRange(key: string, start: number, length: number): Promise<Buffer>;
|
||||
}
|
||||
|
||||
/** The direct-only members of StorageService, as one spreadable group. */
|
||||
export type DirectStorageMembers = Pick<
|
||||
StorageService,
|
||||
"directPublicJwk" | "presignPut" | "headObject" | "openDirectHeader" | "getSealedObject"
|
||||
"directPublicJwk" | "presignPut" | "headObject" | "openDirectHeader" | "getSealedObject" | "readDirectRange"
|
||||
>;
|
||||
|
||||
/**
|
||||
|
|
@ -73,6 +75,7 @@ export function directStorageDisabled(): DirectStorageMembers {
|
|||
headObject: unavailable,
|
||||
openDirectHeader: unavailable,
|
||||
getSealedObject: unavailable,
|
||||
readDirectRange: unavailable,
|
||||
};
|
||||
}
|
||||
|
||||
|
|
@ -523,6 +526,51 @@ export class S3Storage implements StorageService {
|
|||
if (res.Body === undefined) throw new Error(`empty body for ${key}`);
|
||||
return Buffer.from(await res.Body.transformToByteArray());
|
||||
}
|
||||
|
||||
/**
|
||||
* Plaintext [start, start+length) of a SPENC2 object WITHOUT reading the
|
||||
* whole frame: the mapping plaintext→sealed offset is deterministic
|
||||
* (header 135 + per segment 4+12+16 with 10 MiB plaintext), so only the
|
||||
* covering segments are fetched (R2 ranged GET) and decrypted.
|
||||
*/
|
||||
async readDirectRange(key: string, start: number, length: number): Promise<Buffer> {
|
||||
if (this.directKey === null) throw missingDirectKey();
|
||||
if (length <= 0) return Buffer.alloc(0);
|
||||
const first = Math.floor(start / DIRECT_SEGMENT_BYTES);
|
||||
const last = Math.floor((start + length - 1) / DIRECT_SEGMENT_BYTES);
|
||||
// ONE ranged GET covering the header (the CEK wraps there) + every segment
|
||||
// through `last`. Segments before `first` are walked but skipped, so the
|
||||
// sealed offsets stay deterministic; all non-final segments carry a full
|
||||
// 10 MiB plaintext, which makes the sealed end computable up front.
|
||||
const segSealedStart = (i: number): number => SEGMENTS_OFFSET + i * (4 + 12 + DIRECT_SEGMENT_BYTES + 16);
|
||||
const fetchEnd = segSealedStart(last) + 4 + 12 + DIRECT_SEGMENT_BYTES + 16;
|
||||
const res = await this.client.send(new GetObjectCommand({
|
||||
Bucket: this.bucket, Key: key, Range: `bytes=0-${fetchEnd - 1}`,
|
||||
}));
|
||||
if (res.Body === undefined) throw new Error(`empty body for ${key}`);
|
||||
const stored = Buffer.from(await res.Body.transformToByteArray());
|
||||
const cek = unwrapDirectCek(stored, this.directKey.privateKey);
|
||||
const parts: Buffer[] = [];
|
||||
let filled = 0;
|
||||
let offset = SEGMENTS_OFFSET;
|
||||
const segmentCount = stored.readUInt32BE(SEGMENT_COUNT_OFFSET);
|
||||
for (let i = 0; i < segmentCount && filled < length; i += 1) {
|
||||
const ctLen = stored.readUInt32BE(offset); offset += 4;
|
||||
if (ctLen < 16 || offset + 12 + ctLen > stored.length) break;
|
||||
const iv = stored.subarray(offset, offset + 12); offset += 12;
|
||||
const ciphertext = stored.subarray(offset, offset + ctLen); offset += ctLen;
|
||||
const segStart = i * DIRECT_SEGMENT_BYTES;
|
||||
const segEnd = segStart + ctLen - 16;
|
||||
if (segEnd <= start) continue;
|
||||
if (segStart >= start + length) break;
|
||||
const plain = aesGcmOpen(cek, iv, ciphertext);
|
||||
const from = Math.max(0, start - segStart);
|
||||
const to = Math.min(plain.length, start + length - segStart);
|
||||
parts.push(plain.subarray(from, to));
|
||||
filled += to - from;
|
||||
}
|
||||
return Buffer.concat(parts);
|
||||
}
|
||||
}
|
||||
|
||||
/** Build a DirectKeyPair from an already-loaded private key (config + tests). */
|
||||
|
|
|
|||
|
|
@ -74,7 +74,7 @@ describe("video multipart HTTP routes", () => {
|
|||
const response = await upload(videoForm());
|
||||
expect(response.status).toBe(201);
|
||||
const result = await response.json() as VideoStatus;
|
||||
expect(result.processingJob.state).toBe("complete");
|
||||
expect(result.processingJob!.state).toBe("complete");
|
||||
expect(result.videoAsset).toMatchObject({ durationMs: 1000, width: 320, height: 180 });
|
||||
const status = await injectHttp(app?.getHttpAdapter().getInstance(), `http://api.test/video/${result.videoAsset.id}/status`, {
|
||||
headers: { authorization: "Bearer test-session" },
|
||||
|
|
|
|||
|
|
@ -92,8 +92,8 @@ describe.skipIf(!dbAvailable)("video library (live DB)", () => {
|
|||
expect(page.items[0]?.processingState).toBe("complete");
|
||||
|
||||
const status = await service.getVideoStatus(uploaded.videoAsset.id, workspaceId);
|
||||
expect(status.processingJob.state).toBe("complete");
|
||||
expect(status.processingJob.processingNote).toContain("Transcode skipped");
|
||||
expect(status.processingJob!.state).toBe("complete");
|
||||
expect(status.processingJob!.processingNote).toContain("Transcode skipped");
|
||||
expect(status.videoAsset).toMatchObject({ width: 640, height: 360, durationMs: 1000, codec: "h264" });
|
||||
await expect(service.getVideoStatus(uploaded.videoAsset.id, randomUUID())).rejects.toThrow(/not found/);
|
||||
|
||||
|
|
@ -123,7 +123,7 @@ describe.skipIf(!dbAvailable)("video library (live DB)", () => {
|
|||
expect(second.videoAsset.id).not.toBe(first.videoAsset.id);
|
||||
expect(second.videoAsset.storageKey).toBe(first.videoAsset.storageKey);
|
||||
await service.deleteVideo(first.videoAsset.id, workspaceId);
|
||||
expect((await service.getVideoStatus(second.videoAsset.id, workspaceId)).processingJob.state).toBe("complete");
|
||||
expect((await service.getVideoStatus(second.videoAsset.id, workspaceId)).processingJob!.state).toBe("complete");
|
||||
expect(await storage.getObject(second.videoAsset.storageKey)).toEqual(bytes);
|
||||
await service.deleteVideo(second.videoAsset.id, workspaceId);
|
||||
});
|
||||
|
|
|
|||
|
|
@ -35,7 +35,7 @@ describe.skipIf(fixture === undefined)("real ffprobe roundtrip (encrypted in-mem
|
|||
const { service } = serviceWithDatabase({}, probeVideoFile, storage);
|
||||
const uploaded = await service.uploadVideoBytes(workspaceId, bytes, "video/mp4");
|
||||
const status = await service.getVideoStatus(uploaded.videoAsset.id, workspaceId);
|
||||
expect(status.processingJob.state).toBe("complete");
|
||||
expect(status.processingJob!.state).toBe("complete");
|
||||
expect(status.videoAsset).toMatchObject({ durationMs: 1000, width: 320, height: 180, frameRate: 25 });
|
||||
expect(status.videoAsset.codec).toBeDefined();
|
||||
expect(objects.get(status.videoAsset.storageKey)?.equals(bytes)).toBe(false);
|
||||
|
|
|
|||
|
|
@ -26,7 +26,7 @@ describe("video byte ingest and processing-job state machine", () => {
|
|||
expect(storage.getObject).toHaveBeenCalledWith(uploaded.videoAsset.storageKey);
|
||||
expect(uploaded.videoAsset).toMatchObject({ sizeBytes: bytes.length, width: 1280, height: 720, durationMs: 42_000, codec: "h264" });
|
||||
expect(uploaded.processingJob).toMatchObject({ state: "complete", progressPercent: 100, outputKey: uploaded.videoAsset.storageKey });
|
||||
expect(uploaded.processingJob.processingNote).toContain("Transcode skipped");
|
||||
expect(uploaded.processingJob!.processingNote).toContain("Transcode skipped");
|
||||
// ...and the produce INSERT is the only thing that followed it.
|
||||
expect(database.transitions).toEqual(["queued", "probing", "transcoding", "complete"]);
|
||||
await expect(stat(inputPath)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
|
|
@ -46,7 +46,7 @@ describe("video byte ingest and processing-job state machine", () => {
|
|||
const uploaded = await service.uploadVideoBytes(workspaceId, Buffer.from("broken"), "video/mp4");
|
||||
expect(uploaded.processingJob).toMatchObject({ state: "failed", error: "unreadable media" });
|
||||
expect(database.transitions).toEqual(["queued", "probing", "failed"]);
|
||||
expect((await service.getVideoStatus(assetId, workspaceId)).processingJob.error).toBe("unreadable media");
|
||||
expect((await service.getVideoStatus(assetId, workspaceId)).processingJob!.error).toBe("unreadable media");
|
||||
await expect(stat(inputPath)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
});
|
||||
|
||||
|
|
@ -85,7 +85,7 @@ describe("video byte ingest and processing-job state machine", () => {
|
|||
await expect(service.uploadVideo(workspaceId, stored.key, "video/mp4", 1)).rejects.toThrow("sizeBytes does not match");
|
||||
await expect(service.uploadVideo(workspaceId, "sha256/missing.mp4", "video/mp4", 5)).rejects.toThrow("existing readable encrypted object");
|
||||
const uploaded = await service.uploadVideo(workspaceId, stored.key, "video/mp4", stored.sizeBytes);
|
||||
expect(uploaded.processingJob.state).toBe("complete");
|
||||
expect(uploaded.processingJob!.state).toBe("complete");
|
||||
expect(uploaded.videoAsset.width).toBe(640);
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -82,7 +82,7 @@ export interface VideoLibraryPage {
|
|||
|
||||
export interface VideoStatus {
|
||||
videoAsset: VideoAssetRecord;
|
||||
processingJob: VideoProcessingJobRecord;
|
||||
processingJob: VideoProcessingJobRecord | null;
|
||||
}
|
||||
|
||||
export type VideoUploadResult = VideoStatus;
|
||||
|
|
@ -341,7 +341,7 @@ export class VideoService {
|
|||
} catch (error) {
|
||||
// Probe errors are recorded on the job; return its ID and failure in status.
|
||||
const status = await this.getVideoStatus(uploaded.videoAsset.id, workspaceId);
|
||||
if (status.processingJob.state !== "failed") throw error;
|
||||
if (status.processingJob === null || status.processingJob.state !== "failed") throw error;
|
||||
}
|
||||
await this.audit(workspaceId, "video.uploaded", uploaded.videoAsset.id, {
|
||||
mimeType,
|
||||
|
|
@ -369,8 +369,10 @@ export class VideoService {
|
|||
);
|
||||
if (result.rowCount !== 1) throw new NotFoundException(`video ${assetId} not found`);
|
||||
const row = requiredRow(result.rows[0], "video status");
|
||||
// A jobless asset is a legitimate state (pre-probe direct registrations,
|
||||
// legacy rows): surface it instead of 500ing the library click.
|
||||
if (row.job_id === null || row.job_id === undefined) {
|
||||
throw new Error(`video ${assetId} has no processing job`);
|
||||
return { videoAsset: assetFromRow(row), processingJob: null };
|
||||
}
|
||||
return { videoAsset: assetFromRow(row), processingJob: joinedJobFromRow(row) };
|
||||
});
|
||||
|
|
|
|||
|
|
@ -75,8 +75,8 @@ describe("GET /video/:id/status", () => {
|
|||
const before = await service.getVideoStatus(assetId, workspaceId);
|
||||
const response = await status();
|
||||
const body = await response.json() as VideoStatus;
|
||||
expect(body.processingJob.state).toBe(before.processingJob.state);
|
||||
expect(body.processingJob.state).not.toBe("complete");
|
||||
expect(body.processingJob!.state).toBe(before.processingJob!.state);
|
||||
expect(body.processingJob!.state).not.toBe("complete");
|
||||
});
|
||||
|
||||
it("returns 404 for an unknown asset in the workspace", async () => {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue