Skip to content

Commit b564f6c

Browse files
richiemcilroyrichiemcilroy
committed
fix: fence video edit dispatch and completion by operation
1 parent edb4734 commit b564f6c

9 files changed

Lines changed: 1029 additions & 368 deletions

File tree

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

Lines changed: 75 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@ vi.mock("@cap/database/schema", () => ({
2424
rawFileKey: "rawFileKey",
2525
},
2626
videoEdits: { table: "edits" },
27+
videoProcessingJobs: { table: "jobs", videoId: "videoId" },
2728
comments: { table: "comments" },
2829
}));
2930
vi.mock("drizzle-orm", () => ({ and: vi.fn(), eq: vi.fn() }));
@@ -76,6 +77,7 @@ import {
7677
import { saveMetadataAndComplete } from "@/workflows/admin-reprocess-video";
7778
import {
7879
saveEditResultAndComplete,
80+
startMediaServerEditJob,
7981
verifyRenderedEditOutput,
8082
} from "@/workflows/edit-video";
8183

@@ -94,6 +96,12 @@ let video: {
9496
};
9597
let events: string[];
9698
let updates: Record<string, unknown>[];
99+
const operation = {
100+
token: "11111111-1111-4111-8111-111111111111",
101+
startedAt: "2026-09-08T12:00:00.000Z",
102+
};
103+
const sourceKey = "user/video/edit-original.mp4";
104+
let upload: Record<string, unknown>;
97105
const metadata = { duration: 5, width: 320, height: 180, fps: 30 };
98106
const editSpec: VideoEditSpec = {
99107
version: 1,
@@ -107,10 +115,15 @@ function createClient() {
107115
return {
108116
from: (table: { table: string }) => ({
109117
where: () => {
110-
const rows = table.table === "comments" ? [] : [video];
118+
const rows =
119+
table.table === "comments"
120+
? []
121+
: table.table === "uploads"
122+
? [upload]
123+
: [video];
111124
return Object.assign(Promise.resolve(rows), {
112125
for: async () => {
113-
events.push("lock-video");
126+
events.push(table.table === "jobs" ? "lock-job" : "lock-video");
114127
return rows;
115128
},
116129
});
@@ -175,6 +188,11 @@ beforeEach(() => {
175188
summary: "old summary",
176189
},
177190
};
191+
upload = {
192+
phase: "complete",
193+
startedAt: new Date(operation.startedAt),
194+
rawFileKey: sourceKey,
195+
};
178196
events = [];
179197
updates = [];
180198
mocks.db.mockReturnValue(createClient());
@@ -235,22 +253,42 @@ describe("edited recording publication", () => {
235253
});
236254

237255
it("switches a completed edit to canonical output and retires old upload proof atomically", async () => {
256+
video.metadata.editProcessing = {
257+
...operation,
258+
ownerId: video.ownerId,
259+
bucket: video.bucket,
260+
storageIntegrationId: video.storageIntegrationId,
261+
sourceKey,
262+
source: JSON.stringify(video.source),
263+
dispatch: "accepted",
264+
};
238265
await saveEditResultAndComplete(
239266
"video",
240267
"user/video/edit-original.mp4",
241268
editSpec,
242269
editSpec,
243270
metadata,
271+
operation,
244272
);
245273
expect(video.source).toEqual({ type: "desktopMP4" });
246274
expect(video.metadata).not.toHaveProperty("desktopRecordingUpload");
247275
expect(video.metadata.customCreatedAt).toBe("2020-01-01T00:00:00Z");
248-
expect(events.indexOf("retire-job")).toBeLessThan(
276+
expect(events.indexOf("lock-job")).toBeLessThan(
249277
events.indexOf("lock-video"),
250278
);
251279
expect(events.indexOf("lock-video")).toBeLessThan(
252280
events.indexOf("update-videos"),
253281
);
282+
const before = events.length;
283+
await saveEditResultAndComplete(
284+
"video",
285+
sourceKey,
286+
editSpec,
287+
editSpec,
288+
metadata,
289+
operation,
290+
);
291+
expect(events).toHaveLength(before);
254292
});
255293

256294
it("checks a reprocessed canonical object before clearing its previous immutable publication", async () => {
@@ -326,3 +364,37 @@ describe("intentional administrator replacements", () => {
326364
expect(mocks.retire).not.toHaveBeenCalled();
327365
});
328366
});
367+
368+
describe("edit dispatch acceptance", () => {
369+
it("retries only an explicit capacity rejection", async () => {
370+
mocks.fetch.mockResolvedValue(
371+
Response.json({ code: "SERVER_BUSY" }, { status: 503 }),
372+
);
373+
await expect(
374+
startMediaServerEditJob("https://media.test", { videoId: "video" }),
375+
).resolves.toEqual({ status: "capacity" });
376+
expect(mocks.fetch).toHaveBeenCalledOnce();
377+
});
378+
it("treats proxy failures as uncertain acceptance", async () => {
379+
mocks.fetch.mockResolvedValue(
380+
Response.json({ error: "Gateway timeout" }, { status: 504 }),
381+
);
382+
await expect(
383+
startMediaServerEditJob("https://media.test", { videoId: "video" }),
384+
).rejects.toThrow("uncertain");
385+
expect(mocks.fetch).toHaveBeenCalledOnce();
386+
});
387+
it("does not resend a request after a lost response", async () => {
388+
mocks.fetch.mockRejectedValue(new Error("Connection reset"));
389+
await expect(
390+
startMediaServerEditJob("https://media.test", { videoId: "video" }),
391+
).rejects.toThrow("Connection reset");
392+
expect(mocks.fetch).toHaveBeenCalledOnce();
393+
});
394+
it("accepts only responses containing the worker identity", async () => {
395+
mocks.fetch.mockResolvedValue(Response.json({ jobId: "worker-1" }));
396+
await expect(
397+
startMediaServerEditJob("https://media.test", { videoId: "video" }),
398+
).resolves.toEqual({ status: "accepted", jobId: "worker-1" });
399+
});
400+
});

‎apps/web/__tests__/unit/media-server-progress.test.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -179,7 +179,7 @@ function databaseFixture(initial: DesktopRecordingJob | null = fixture().job) {
179179
};
180180
const mutations: Mutation[] = [];
181181
const rows = (table: unknown) => {
182-
if (table === mocks.tables.videos) return [structuredClone(video)];
182+
if (table === mocks.tables.videos) return [{ ...video }];
183183
if (table === mocks.tables.jobs)
184184
return current ? [structuredClone(current)] : [];
185185
if (table === mocks.tables.uploads) return [{ rawFileKey }];

‎apps/web/__tests__/unit/video-edit-operation.test.ts‎

Lines changed: 242 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,242 @@
1+
import { User } from "@cap/web-domain";
2+
import { beforeEach, describe, expect, it, vi } from "vitest";
3+
4+
const mocks = vi.hoisted(() => ({
5+
video: undefined as Record<string, unknown> | undefined,
6+
upload: undefined as Record<string, unknown> | undefined,
7+
writes: [] as { table: string; data: Record<string, unknown> }[],
8+
}));
9+
vi.mock("@cap/database/schema", () => ({
10+
videos: { id: "video.id", name: "video" },
11+
videoUploads: { videoId: "upload.videoId", name: "upload" },
12+
}));
13+
vi.mock("@cap/database", () => {
14+
const tx = {
15+
select: () => ({
16+
from: (table: { name: "video" | "upload" }) => ({
17+
where: () => {
18+
const rows = mocks[table.name] ? [mocks[table.name]] : [];
19+
return Object.assign(Promise.resolve(rows), {
20+
for: async () => rows,
21+
});
22+
},
23+
}),
24+
}),
25+
update: (table: { name: "video" | "upload" }) => ({
26+
set: (data: Record<string, unknown>) => ({
27+
where: async () => {
28+
mocks.writes.push({ table: table.name, data });
29+
Object.assign(mocks[table.name] ?? {}, data);
30+
},
31+
}),
32+
}),
33+
delete: (table: { name: "video" | "upload" }) => ({
34+
where: async () => {
35+
mocks.writes.push({ table: table.name, data: { deleted: true } });
36+
mocks[table.name] = undefined;
37+
},
38+
}),
39+
};
40+
return {
41+
db: () => ({
42+
...tx,
43+
transaction: (run: (value: typeof tx) => unknown) => run(tx),
44+
}),
45+
};
46+
});
47+
48+
import {
49+
clearPendingEdit,
50+
getEditProcessingState,
51+
matchesEditOperation,
52+
} from "@/lib/video-edit-operation";
53+
import { applyEditProgress } from "@/lib/video-edit-progress";
54+
55+
const operation = {
56+
token: "11111111-1111-4111-8111-111111111111",
57+
startedAt: "2026-09-08T12:00:00.000Z",
58+
};
59+
const sourceKey = "owner/video/source/original.mp4";
60+
const storage = {
61+
ownerId: User.UserId.make("owner"),
62+
bucket: null,
63+
storageIntegrationId: null,
64+
};
65+
const source = { type: "webMP4" as const };
66+
const state = {
67+
...operation,
68+
...storage,
69+
sourceKey,
70+
source: JSON.stringify(source),
71+
dispatch: "dispatching" as const,
72+
};
73+
const progress = {
74+
videoId: "video",
75+
jobId: "job-1",
76+
phase: "processing",
77+
progress: 25,
78+
};
79+
80+
beforeEach(() => {
81+
mocks.video = {
82+
id: "video",
83+
...storage,
84+
source,
85+
metadata: { editProcessing: { ...state } },
86+
duration: 10,
87+
};
88+
mocks.upload = {
89+
rawFileKey: sourceKey,
90+
startedAt: new Date(operation.startedAt),
91+
phase: "processing",
92+
};
93+
mocks.writes = [];
94+
});
95+
96+
describe("edit operation ownership", () => {
97+
it("matches both the original source and the precise operation", () => {
98+
const video = { ...storage, source, metadata: { editProcessing: state } };
99+
const upload = {
100+
rawFileKey: sourceKey,
101+
startedAt: new Date(operation.startedAt),
102+
};
103+
expect(matchesEditOperation(video, upload, sourceKey, operation)).toBe(
104+
true,
105+
);
106+
expect(
107+
matchesEditOperation(video, upload, sourceKey, {
108+
...operation,
109+
token: "stale",
110+
}),
111+
).toBe(false);
112+
expect(
113+
matchesEditOperation(
114+
video,
115+
{ ...upload, startedAt: new Date("2026-09-08T12:00:01Z") },
116+
sourceKey,
117+
operation,
118+
),
119+
).toBe(false);
120+
expect(
121+
matchesEditOperation(
122+
{ ...video, source: { type: "desktopMP4" } },
123+
upload,
124+
sourceKey,
125+
operation,
126+
),
127+
).toBe(false);
128+
});
129+
it("accepts the first callback even when the dispatch response was lost", async () => {
130+
expect(
131+
await applyEditProgress(progress, operation.token, operation.startedAt),
132+
).toBe(true);
133+
expect(mocks.video?.metadata).toMatchObject({
134+
editProcessing: { dispatch: "accepted", jobId: "job-1" },
135+
});
136+
expect(mocks.upload?.processingProgress).toBe(25);
137+
});
138+
it.each(["stale-token", null])(
139+
"ignores callbacks without the current token (%s)",
140+
async (token) => {
141+
expect(
142+
await applyEditProgress(progress, token, operation.startedAt),
143+
).toBe(true);
144+
expect(mocks.writes).toEqual([]);
145+
},
146+
);
147+
it("ignores callbacks after recording storage is moved", async () => {
148+
if (!mocks.video) throw new Error("Missing fixture");
149+
mocks.video.bucket = "new-bucket";
150+
await applyEditProgress(progress, operation.token, operation.startedAt);
151+
expect(mocks.writes).toEqual([]);
152+
});
153+
it("rejects callbacks from a second worker", async () => {
154+
await applyEditProgress(progress, operation.token, operation.startedAt);
155+
mocks.writes = [];
156+
await applyEditProgress(
157+
{ ...progress, jobId: "job-2", phase: "error" },
158+
operation.token,
159+
operation.startedAt,
160+
);
161+
expect(mocks.writes).toEqual([]);
162+
});
163+
it("keeps completion terminal when delayed progress or errors arrive", async () => {
164+
await applyEditProgress(
165+
{
166+
...progress,
167+
phase: "complete",
168+
metadata: { duration: 5, width: 1920, height: 1080, fps: 30 },
169+
},
170+
operation.token,
171+
operation.startedAt,
172+
);
173+
mocks.writes = [];
174+
await applyEditProgress(progress, operation.token, operation.startedAt);
175+
await applyEditProgress(
176+
{ ...progress, phase: "error" },
177+
operation.token,
178+
operation.startedAt,
179+
);
180+
expect(mocks.writes).toEqual([]);
181+
expect(mocks.upload?.phase).toBe("complete");
182+
expect(mocks.video?.duration).toBe(5);
183+
});
184+
it("rejects invalid completion metadata without changing state", async () => {
185+
await expect(
186+
applyEditProgress(
187+
{
188+
...progress,
189+
phase: "complete",
190+
metadata: {
191+
duration: Number.NaN,
192+
width: 1920,
193+
height: 1080,
194+
fps: 30,
195+
},
196+
},
197+
operation.token,
198+
operation.startedAt,
199+
),
200+
).rejects.toThrow("valid media metadata");
201+
expect(mocks.writes).toEqual([]);
202+
});
203+
it("preserves ambiguous dispatches during cleanup", async () => {
204+
await clearPendingEdit("video", sourceKey, operation);
205+
expect(mocks.writes).toEqual([]);
206+
});
207+
it("clears only an operation that has not dispatched", async () => {
208+
if (!mocks.video) throw new Error("Missing fixture");
209+
mocks.video.metadata = {
210+
editProcessing: { ...state, dispatch: "pending" },
211+
};
212+
await clearPendingEdit("video", sourceKey, operation);
213+
expect(mocks.upload).toBeUndefined();
214+
expect(mocks.video.metadata).toEqual({});
215+
});
216+
it("never clears a newer operation", async () => {
217+
await clearPendingEdit("video", sourceKey, {
218+
...operation,
219+
token: "stale",
220+
});
221+
expect(mocks.writes).toEqual([]);
222+
});
223+
it("leaves legacy callbacks to the existing processor", async () => {
224+
if (!mocks.video) throw new Error("Missing fixture");
225+
mocks.video.metadata = {};
226+
expect(await applyEditProgress(progress, null, null)).toBe(false);
227+
expect(mocks.writes).toEqual([]);
228+
});
229+
it("ignores callbacks from an edit after its lock was released", async () => {
230+
if (!mocks.video) throw new Error("Missing fixture");
231+
mocks.video.metadata = {};
232+
expect(
233+
await applyEditProgress(progress, operation.token, operation.startedAt),
234+
).toBe(true);
235+
expect(mocks.writes).toEqual([]);
236+
});
237+
it("rejects malformed persisted identities", () => {
238+
expect(
239+
getEditProcessingState({ editProcessing: { ...state, token: "bad" } }),
240+
).toBeUndefined();
241+
});
242+
});

0 commit comments

Comments
 (0)