feat(web): webhooks for Cap Infra videos (video.ready, video.errored)

apps/web/__tests__/unit/agent-docs.test.ts

@@ -145,7 +145,8 @@ describe("Cap for Agents docs", () => {
 		);
 
 		const webhooks = getDocBySlug("api/webhooks")?.content ?? "";
-		expect(webhooks).toContain(
+		expect(webhooks).toContain("POST /api/developer/v1/webhooks");
+		expect(webhooks).not.toContain(
 			"does not currently expose a documented public webhook subscription API",
 		);
 	});

apps/web/__tests__/unit/deliver-developer-webhook-workflow.test.ts

@@ -0,0 +1,95 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const mocks = vi.hoisted(() => ({
+	sleep: vi.fn(),
+	attempt: vi.fn(),
+}));
+
+vi.mock("workflow", () => ({ sleep: mocks.sleep }));
+vi.mock("@/lib/developer-webhook-delivery", () => ({
+	deliverDeveloperWebhookAttempt: mocks.attempt,
+}));
+
+const { deliverDeveloperWebhookWorkflow } = await import(
+	"@/workflows/deliver-developer-webhook"
+);
+
+const job = {
+	endpointId: "wh-1",
+	eventId: "evt_1",
+	eventType: "video.ready",
+	body: "{}",
+};
+
+beforeEach(() => {
+	vi.clearAllMocks();
+	mocks.sleep.mockResolvedValue(undefined);
+	vi.spyOn(console, "error").mockImplementation(() => undefined);
+});
+
+describe("deliver developer webhook workflow", () => {
+	it("stops after the first attempt when it is delivered", async () => {
+		mocks.attempt.mockResolvedValue("delivered");
+		expect(await deliverDeveloperWebhookWorkflow(job)).toEqual({
+			outcome: "delivered",
+			attempts: 1,
+		});
+		expect(mocks.attempt).toHaveBeenCalledWith(job);
+		expect(mocks.sleep).not.toHaveBeenCalled();
+	});
+
+	it("retries a failing endpoint on the schedule and then delivers", async () => {
+		mocks.attempt
+			.mockResolvedValueOnce("retry")
+			.mockResolvedValueOnce("retry")
+			.mockResolvedValueOnce("delivered");
+		expect(await deliverDeveloperWebhookWorkflow(job)).toEqual({
+			outcome: "delivered",
+			attempts: 3,
+		});
+		expect(mocks.sleep.mock.calls.map(([wait]) => wait)).toEqual([
+			"60s",
+			"300s",
+		]);
+	});
+
+	it("gives up after eight attempts spread over about a day", async () => {
+		mocks.attempt.mockResolvedValue("retry");
+		expect(await deliverDeveloperWebhookWorkflow(job)).toEqual({
+			outcome: "gave_up",
+			attempts: 8,
+		});
+		expect(mocks.attempt).toHaveBeenCalledTimes(8);
+		expect(mocks.sleep.mock.calls.map(([wait]) => wait)).toEqual([
+			"60s",
+			"300s",
+			"1800s",
+			"7200s",
+			"18000s",
+			"28800s",
+			"28800s",
+		]);
+	});
+
+	it("stops retrying once the endpoint is deleted or disabled", async () => {
+		mocks.attempt
+			.mockResolvedValueOnce("retry")
+			.mockResolvedValueOnce("dropped");
+		expect(await deliverDeveloperWebhookWorkflow(job)).toEqual({
+			outcome: "dropped",
+			attempts: 2,
+		});
+		expect(mocks.sleep).toHaveBeenCalledTimes(1);
+	});
+
+	it("keeps an attempt that could not run on the same schedule", async () => {
+		mocks.attempt
+			.mockRejectedValueOnce(new Error("database is away"))
+			.mockResolvedValueOnce("delivered");
+		expect(await deliverDeveloperWebhookWorkflow(job)).toEqual({
+			outcome: "delivered",
+			attempts: 2,
+		});
+		expect(mocks.sleep.mock.calls.map(([wait]) => wait)).toEqual(["60s"]);
+	});
+});

apps/web/__tests__/unit/developer-transcribe-download.test.ts

@@ -80,6 +80,25 @@ vi.mock("@/lib/developer-transcripts", async (importOriginal) => ({
 	TRANSCRIPT_URL_DOWNLOAD_TIMEOUT_MS: 2_000,
 }));
 
+// requestPublicOnly refuses a loopback address, by name or as a literal, so
+// the local server is reached through a name this lookup pins to it.
+vi.mock("@/lib/s3-endpoint-policy", async (importOriginal) => {
+	const actual =
+		await importOriginal<typeof import("@/lib/s3-endpoint-policy")>();
+	const publicOnlyLookup: typeof actual.publicOnlyLookup = (
+		_hostname,
+		options,
+		callback,
+	) => {
+		if (typeof options === "object" && options.all) {
+			callback(null, [{ address: "127.0.0.1", family: 4 }]);
+			return;
+		}
+		callback(null, "127.0.0.1", 4);
+	};
+	return { ...actual, publicOnlyLookup };
+});
+
 vi.mock("@/lib/public-media-fetch", async (importOriginal) => {
 	const actual =
 		await importOriginal<typeof import("@/lib/public-media-fetch")>();
@@ -263,7 +282,7 @@ const serve = async (options: {
 	await new Promise<void>((resolve) =>
 		state.server.listen(0, "127.0.0.1", resolve),
 	);
-	context.origin = `http://127.0.0.1:${(state.server.address() as AddressInfo).port}`;
+	context.origin = `http://media.test:${(state.server.address() as AddressInfo).port}`;
 	origin = state;
 	return state;
 };