fix: publish recording replacements without overwriting originals

Author: richiemcilroy

apps/desktop-gpui/src/upload.rs

@@ -1347,7 +1347,8 @@ async fn upload_video(
     cancel: &AtomicBool,
     replace_existing: bool,
 ) -> Result<(String, Option<String>), AuthApiError> {
-    let initiate = checked_upload_step(cancel, || multipart_initiate(video_id)).await?;
+    let initiate =
+        checked_upload_step(cancel, || multipart_initiate(video_id, replace_existing)).await?;
     let is_drive = is_google_drive_upload(initiate.provider.as_deref(), &initiate.upload_id);
     let parts = upload_parts(
         video_id,
@@ -1394,13 +1395,17 @@ struct InitiateResponse {
     provider: Option<String>,
 }
 
-async fn multipart_initiate(video_id: &str) -> Result<InitiateResponse, AuthApiError> {
+async fn multipart_initiate(
+    video_id: &str,
+    replace_existing: bool,
+) -> Result<InitiateResponse, AuthApiError> {
     let response = auth::authed_request(
         reqwest::Method::POST,
         "/api/upload/multipart/initiate",
         Some(json!({
             "videoId": video_id,
-            "contentType": "video/mp4"
+            "contentType": "video/mp4",
+            "replaceExisting": replace_existing
         })),
     )
     .await

apps/desktop/src-tauri/src/api.rs

@@ -21,14 +21,16 @@ pub struct MultipartUploadInitiateResponse {
 pub async fn upload_multipart_initiate(
     app: &AppHandle,
     video_id: &str,
+    replace_existing: bool,
 ) -> Result<MultipartUploadInitiateResponse, AuthedApiError> {
     let resp = app
         .authed_api_request("/api/upload/multipart/initiate", |c, url| {
             c.post(url)
                 .header("Content-Type", "application/json")
                 .json(&serde_json::json!({
                     "videoId": video_id,
-                    "contentType": "video/mp4"
+                    "contentType": "video/mp4",
+                    "replaceExisting": replace_existing
                 }))
         })
         .await

apps/desktop/src-tauri/src/upload.rs

@@ -255,7 +255,7 @@ pub async fn upload_video(
     info!("Uploading video {video_id}...");
 
     let start = Instant::now();
-    let upload = api::upload_multipart_initiate(app, &video_id).await?;
+    let upload = api::upload_multipart_initiate(app, &video_id, replace_existing).await?;
     let is_drive_upload = is_google_drive_upload(upload.provider.as_deref(), &upload.upload_id);
     let upload_id = upload.upload_id;
 
@@ -726,7 +726,7 @@ impl InstantMultipartUpload {
             recording_dir: recording_dir.clone(),
         })?;
 
-        let upload = api::upload_multipart_initiate(&app, &video_id).await?;
+        let upload = api::upload_multipart_initiate(&app, &video_id, false).await?;
         let is_drive_upload = is_google_drive_upload(upload.provider.as_deref(), &upload.upload_id);
         let upload_id = upload.upload_id;
 
@@ -4273,7 +4273,7 @@ pub(crate) mod strict_instant {
         required_audio: bool,
     ) -> Result<(), AuthedApiError> {
         let upload = control
-            .step(|| api::upload_multipart_initiate(app, &video.id))
+            .step(|| api::upload_multipart_initiate(app, &video.id, false))
             .await?;
         let concurrency = if is_google_drive_upload(upload.provider.as_deref(), &upload.upload_id) {
             1

apps/web/__tests__/unit/desktop-recording-output-replacement.test.ts

@@ -1,6 +1,6 @@
 import type { VideoEditSpec } from "@cap/database/types";
 import { User, Video } from "@cap/web-domain";
-import { Effect, Option } from "effect";
+import { Effect, Option, Schema } from "effect";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 
 const mocks = vi.hoisted(() => ({
@@ -76,6 +76,11 @@ import {
 	invalidateVideoCache,
 } from "@/actions/admin/replace-video";
 import { prepareDesktopReupload } from "@/lib/desktop-reupload";
+import {
+	createDesktopReuploadKey,
+	createDesktopReuploadToken,
+	decodeDesktopReuploadToken,
+} from "@/lib/desktop-reupload-token";
 import { saveMetadataAndComplete } from "@/workflows/admin-reprocess-video";
 import {
 	saveEditResultAndComplete,
@@ -128,7 +133,13 @@ function createClient() {
 									: [video];
 						return Object.assign(Promise.resolve(rows), {
 							for: async () => {
-								events.push(table.table === "jobs" ? "lock-job" : "lock-video");
+								events.push(
+									table.table === "jobs"
+										? "lock-job"
+										: table.table === "uploads"
+											? "lock-upload"
+											: "lock-video",
+								);
 								return rows;
 							},
 						});
@@ -145,9 +156,10 @@ function createClient() {
 							updates.push(values);
 							Object.assign(video, values);
 							if (values.source && reorderSourceKeys) {
-								video.source = Object.fromEntries(
-									Object.entries(video.source).reverse(),
-								) as typeof video.source;
+								video.source = Object.assign(
+									{ type: video.source.type },
+									Object.fromEntries(Object.entries(video.source).reverse()),
+								);
 							}
 						}
 						return [{ affectedRows: 1 }];
@@ -236,6 +248,7 @@ beforeEach(() => {
 		events.push("retire-job");
 	});
 	mocks.env.mockReturnValue({
+		NEXTAUTH_SECRET: "test-only-desktop-reupload-signing-key",
 		MEDIA_SERVER_URL: "https://media.test",
 		MEDIA_SERVER_WEBHOOK_SECRET: "secret",
 		WEB_URL: "https://cap.test",
@@ -253,35 +266,234 @@ describe("desktop reupload publication", () => {
 		ownerId: User.UserId.make(video.ownerId),
 		bucketId: Option.none(),
 		storageIntegrationId: Option.none(),
+		source: Schema.decodeUnknownSync(Video.Video.fields.source)(video.source),
 	});
+	const tokenFor = (snapshot = original()) => {
+		const token = decodeDesktopReuploadToken(
+			createDesktopReuploadToken(snapshot, {
+				uploadId: "provider-upload-id",
+				provider: "s3",
+				outputKey: createDesktopReuploadKey(snapshot),
+			}),
+		);
+		if (!token) throw new Error("Expected a replacement token");
+		upload.rawFileKey = token.outputKey;
+		return token;
+	};
+	const transaction = () =>
+		createClient() as unknown as Parameters<typeof prepareDesktopReupload>[0];
 
 	it("replaces processed playback and stale AI while preserving the link's other metadata", async () => {
 		video.source.audioLevelOutputKey =
 			"user/video/.recording/outputs/old-audio.mp4";
 		video.metadata.editProcessing = { token: "old-edit" };
-		const tx = createClient() as unknown as Parameters<
-			typeof prepareDesktopReupload
-		>[0];
-		const replacement = await prepareDesktopReupload(tx, original());
+		video.metadata.completedVideoEdit = { token: "completed-edit" };
+		video.metadata.chapters = [{ title: "old chapter" }];
+		video.metadata.aiGenerationStatus = "complete";
+		const before = structuredClone(video);
+		const snapshot = original();
+		const token = tokenFor(snapshot);
+		const tx = transaction();
+		const replacement = await prepareDesktopReupload(tx, snapshot, token);
 		expect(replacement).toEqual({
-			source: { type: "desktopMP4" },
+			source: { type: "desktopMP4", outputKey: token.outputKey },
 			metadata: { customCreatedAt: "2020-01-01T00:00:00Z" },
 			transcriptionStatus: null,
 		});
-		expect(video.source.outputKey).toBeDefined();
-		expect(events).toEqual(["retire-job", "lock-video"]);
+		expect(video).toEqual(before);
+		expect(events).toEqual([
+			"lock-job",
+			"lock-video",
+			"lock-upload",
+			"retire-job",
+		]);
+		expect(mocks.retire).toHaveBeenCalledWith(tx, {
+			videoId: "video",
+			userId: "user",
+		});
+		expect(mocks.head).not.toHaveBeenCalled();
+		expect(mocks.access).not.toHaveBeenCalled();
 	});
 
-	it("rejects publication if storage changed during upload", async () => {
+	it.each(["desktopSegments", "webMP4"])(
+		"publishes a %s replacement with its immutable MP4 key",
+		async (type) => {
+			video.source = { type };
+			const replacement = await prepareDesktopReupload(
+				transaction(),
+				original(),
+				tokenFor(),
+			);
+			expect(replacement?.source.type).toBe(
+				type === "webMP4" ? "webMP4" : "desktopMP4",
+			);
+			expect(replacement?.source.outputKey).toContain(
+				"user/video/.recording/outputs/reupload-",
+			);
+		},
+	);
+
+	it.each(["bucket", "storageIntegrationId", "ownerId"] as const)(
+		"rejects publication if %s changed during upload before retiring any job",
+		async (field) => {
+			const snapshot = original();
+			const token = tokenFor(snapshot);
+			video[field] = "different-identity";
+			const before = structuredClone(video);
+			await expect(
+				prepareDesktopReupload(transaction(), snapshot, token),
+			).rejects.toThrow("storage changed");
+			expect(video).toEqual(before);
+			expect(updates).toEqual([]);
+			expect(events).toEqual(["lock-job", "lock-video"]);
+			expect(mocks.retire).not.toHaveBeenCalled();
+		},
+	);
+
+	it.each([null, "newer-upload"])(
+		"rejects canceled or superseded upload %s under the lock",
+		async (rawFileKey) => {
+			const snapshot = original();
+			const token = tokenFor(snapshot);
+			upload.rawFileKey = rawFileKey;
+			const before = structuredClone(video);
+			await expect(
+				prepareDesktopReupload(transaction(), snapshot, token),
+			).rejects.toThrow("canceled or superseded");
+			expect(video).toEqual(before);
+			expect(mocks.retire).not.toHaveBeenCalled();
+			expect(events).toEqual(["lock-job", "lock-video", "lock-upload"]);
+		},
+	);
+	it("rejects an older upload completion after another publication", async () => {
 		const snapshot = original();
-		video.bucket = "different-bucket";
-		const tx = createClient() as unknown as Parameters<
-			typeof prepareDesktopReupload
-		>[0];
-		await expect(prepareDesktopReupload(tx, snapshot)).rejects.toThrow(
-			"storage changed",
+		const token = tokenFor(snapshot);
+		video.source.outputKey = "user/video/.recording/outputs/newer/result.mp4";
+		const before = structuredClone(video);
+		await expect(
+			prepareDesktopReupload(transaction(), snapshot, token),
+		).rejects.toThrow("source changed");
+		expect(video).toEqual(before);
+		expect(mocks.retire).not.toHaveBeenCalled();
+	});
+
+	it("compares normalized source fields under the lock", async () => {
+		const snapshot = original();
+		const token = tokenFor(snapshot);
+		video.source = Object.assign(
+			{ type: video.source.type },
+			Object.fromEntries(Object.entries(video.source).reverse()),
+			{ legacyExtra: "ignored by Video.source" },
 		);
+		await expect(
+			prepareDesktopReupload(transaction(), snapshot, token),
+		).resolves.toMatchObject({ source: { outputKey: token.outputKey } });
+		expect(events).toEqual([
+			"lock-job",
+			"lock-video",
+			"lock-upload",
+			"retire-job",
+		]);
+	});
+
+	it("returns no update for the exact already-published attempt without retiring a new job", async () => {
+		const snapshot = original();
+		const token = tokenFor(snapshot);
+		video.source = { type: "desktopMP4", outputKey: token.outputKey };
+		video.metadata.summary = "new summary";
+		const before = structuredClone(video);
+		await expect(
+			prepareDesktopReupload(transaction(), snapshot, token),
+		).resolves.toBeNull();
+		expect(video).toEqual(before);
+		expect(events).toEqual(["lock-job", "lock-video"]);
+		expect(mocks.retire).not.toHaveBeenCalled();
+		expect(updates).toEqual([]);
+	});
+
+	it("still validates storage on an already-published retry", async () => {
+		const snapshot = original();
+		const token = tokenFor(snapshot);
+		video.source = { type: "desktopMP4", outputKey: token.outputKey };
+		video.bucket = "changed-bucket";
+		await expect(
+			prepareDesktopReupload(transaction(), snapshot, token),
+		).rejects.toThrow("storage changed");
+		expect(mocks.retire).not.toHaveBeenCalled();
+	});
+
+	it("leaves the publication and metadata intact if retiring the job fails", async () => {
+		const before = structuredClone(video);
+		mocks.retire.mockImplementationOnce(async () => {
+			events.push("retire-job");
+			throw new Error("Job retirement failed");
+		});
+		await expect(
+			prepareDesktopReupload(transaction(), original(), tokenFor()),
+		).rejects.toThrow("Job retirement failed");
+		expect(video).toEqual(before);
 		expect(updates).toEqual([]);
+		expect(events).toEqual([
+			"lock-job",
+			"lock-video",
+			"lock-upload",
+			"retire-job",
+		]);
+	});
+
+	it("keeps job retirement inside the caller's publication transaction", async () => {
+		const before = structuredClone(video);
+		const retainedJob = {
+			generation: "original-generation",
+			state: "complete",
+			sourceKey: "user/video/original-source.mp4",
+			outputKey: video.source.outputKey,
+		};
+		let job = { ...retainedJob };
+		const tx = transaction();
+		mocks.retire.mockImplementationOnce(async (owner) => {
+			expect(owner).toBe(tx);
+			events.push("retire-job");
+			job = {
+				...job,
+				generation: "retired-generation",
+				state: "source-blocked",
+			};
+		});
+		const publishTransaction = async () => {
+			events.push("transaction");
+			const savedVideo = structuredClone(video);
+			const savedJob = { ...job };
+			try {
+				const replacement = await prepareDesktopReupload(
+					tx,
+					original(),
+					tokenFor(),
+				);
+				expect(replacement).not.toBeNull();
+				expect(job.state).toBe("source-blocked");
+				Object.assign(video, replacement);
+				events.push("publication-failed");
+				throw new Error("Publication failed");
+			} catch (error) {
+				video = savedVideo;
+				job = savedJob;
+				events.push("rollback");
+				throw error;
+			}
+		};
+		await expect(publishTransaction()).rejects.toThrow("Publication failed");
+		expect(video).toEqual(before);
+		expect(job).toEqual(retainedJob);
+		expect(events).toEqual([
+			"transaction",
+			"lock-job",
+			"lock-video",
+			"lock-upload",
+			"retire-job",
+			"publication-failed",
+			"rollback",
+		]);
 	});
 });