feat: route Instant uploads to the nearest configured AWS region

apps/web/__tests__/unit/content-transfer.test.ts

@@ -175,6 +175,24 @@ describe("content transfer planning", () => {
 		};
 
 		expect(getContentTransferStorageBlockReason(input)).toBeNull();
+		expect(
+			getContentTransferStorageBlockReason({
+				...input,
+				bucketId: "cap-tokyo",
+				bucketOwnerId: null,
+				bucketOrganizationId: null,
+			}),
+		).toBeNull();
+		expect(
+			getContentTransferStorageBlockReason({ ...input, bucketOwnerId: null }),
+		).toBe("The storage bucket is missing");
+		expect(
+			getContentTransferStorageBlockReason({
+				...input,
+				bucketId: "cap-tokyo",
+				storageIntegrationId: "drive",
+			}),
+		).toBe("The Cap has conflicting storage assignments");
 		expect(
 			getContentTransferStorageBlockReason({
 				...input,

apps/web/__tests__/unit/desktop-video-create.test.ts

@@ -5,8 +5,10 @@ import {
 	Storage as StorageDomain,
 	Video,
 } from "@cap/web-domain";
-import { Effect, Option } from "effect";
-import { beforeEach, describe, expect, it, vi } from "vitest";
+import { Effect, Layer, Option } from "effect";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+const regionalStorage = vi.hoisted(() => ({ select: vi.fn(), s3: vi.fn() }));
 
 const deletion = vi.hoisted(() => ({
 	deleteVideo: vi.fn(),
@@ -86,19 +88,23 @@ vi.mock("@cap/web-backend", async () => {
 	class Videos extends Effect.Service<Videos>()("Videos", {
 		sync: () => ({ delete: deletion.deleteVideo }),
 	}) {}
+	class S3Buckets extends Effect.Service<S3Buckets>()("S3Buckets", {
+		sync: () => ({ getRegionalUploadBucketId: regionalStorage.select }),
+	}) {}
 	return {
 		makeCurrentUserLayer,
 		Videos,
+		S3Buckets,
 		Storage: {
 			getOrganizationWritableAccess: vi.fn(),
-			getS3WritableAccessForUser: vi.fn(),
+			getS3WritableAccessForUser: regionalStorage.s3,
 		},
 	};
 });
 
 vi.mock("@/lib/server", async () => {
 	const { Effect } = await import("effect");
-	const { Videos } = await import("@cap/web-backend");
+	const { Videos, S3Buckets } = await import("@cap/web-backend");
 	return {
 		runPromise: vi.fn(async (value: unknown) =>
 			Effect.isEffect(value)
@@ -107,9 +113,11 @@ vi.mock("@/lib/server", async () => {
 							value as Effect.Effect<
 								unknown,
 								unknown,
-								InstanceType<typeof Videos>
+								InstanceType<typeof Videos> | InstanceType<typeof S3Buckets>
 							>
-						).pipe(Effect.provide(Videos.Default)),
+						).pipe(
+							Effect.provide(Layer.mergeAll(Videos.Default, S3Buckets.Default)),
+						),
 					)
 				: value,
 		),
@@ -126,6 +134,8 @@ vi.mock("@/lib/google-drive-storage-quota", () => ({
 
 // The live-transcription stack drags in the whole workflow graph
 // (server-only modules included); these tests only care that create works.
+vi.mock("next/server", () => ({ after: vi.fn() }));
+
 vi.mock("@/lib/live-transcribe", () => ({
 	maybeStartLiveTranscription: vi.fn(async () => "skipped"),
 }));
@@ -667,3 +677,156 @@ describe("GET /create", () => {
 		expect(mockDb.insert).not.toHaveBeenCalled();
 	});
 });
+
+describe("new Instant recording regions", () => {
+	let app: typeof import("@/app/api/desktop/[...route]/video")["app"];
+	beforeEach(async () => {
+		vi.clearAllMocks();
+		vi.stubEnv("VERCEL", "1");
+		resetMockDb();
+		stubStorage();
+		regionalStorage.s3.mockReturnValue(
+			Effect.succeed({
+				bucketId: Option.none(),
+				storageIntegrationId: Option.none(),
+			}),
+		);
+		regionalStorage.select.mockImplementation(
+			(latitude: string | undefined, longitude: string | undefined) =>
+				latitude === "35.68" && longitude === "139.69"
+					? Option.some("cap-tokyo")
+					: Option.none(),
+		);
+		defaultSharing.getNewVideoPublic.mockResolvedValue(true);
+		mockGetCurrentUser.mockResolvedValue({
+			id: "user-1",
+			defaultOrgId: "org-1",
+			activeOrganizationId: "org-1",
+		});
+		mockDb.where
+			.mockResolvedValueOnce([
+				{ id: "org-1", name: "Org", createdAt: new Date() },
+			])
+			.mockResolvedValueOnce([])
+			.mockResolvedValueOnce([{ count: 5 }]);
+		app = (await import("@/app/api/desktop/[...route]/video")).app;
+	});
+	afterEach(() => vi.unstubAllEnvs());
+
+	it.each(["desktopMP4", "desktopSegments"])(
+		"persists the selected bucket for %s",
+		async (mode) => {
+			const response = await app.request(
+				`https://cap.test/create?recordingMode=${mode}`,
+				{
+					headers: {
+						"x-vercel-ip-latitude": "35.68",
+						"x-vercel-ip-longitude": "139.69",
+					},
+				},
+			);
+			expect(response.status).toBe(200);
+			expect(insertedValues(schema.videos)?.bucket).toBe("cap-tokyo");
+			expect(regionalStorage.select).toHaveBeenCalledWith("35.68", "139.69");
+		},
+	);
+
+	it.each<Record<string, string>>([
+		{ "x-vercel-ip-latitude": "40.71", "x-vercel-ip-longitude": "-74.01" },
+		{},
+		{ "x-vercel-ip-latitude": "bad", "x-vercel-ip-longitude": "139.69" },
+	])(
+		"keeps Virginia when no regional bucket is selected (%#)",
+		async (headers) => {
+			const response = await app.request(
+				"https://cap.test/create?recordingMode=desktopMP4",
+				{ headers },
+			);
+			expect(response.status).toBe(200);
+			expect(insertedValues(schema.videos)?.bucket).toBeNull();
+		},
+	);
+
+	it.each([
+		{
+			bucket: "custom-bucket",
+			integration: null,
+			vercel: "1",
+			query: "recordingMode=desktopMP4",
+		},
+		{
+			bucket: null,
+			integration: "drive-id",
+			vercel: "1",
+			query: "recordingMode=desktopMP4",
+		},
+		{
+			bucket: null,
+			integration: null,
+			vercel: "",
+			query: "recordingMode=desktopMP4",
+		},
+		{
+			bucket: null,
+			integration: null,
+			vercel: "1",
+			query: "isScreenshot=true&recordingMode=desktopMP4",
+		},
+		{
+			bucket: null,
+			integration: null,
+			vercel: "1",
+			query: "recordingMode=hls",
+		},
+	])(
+		"preserves storage outside eligible new Instant recordings (%#)",
+		async ({ bucket, integration, vercel, query }) => {
+			vi.stubEnv("VERCEL", vercel);
+			regionalStorage.s3.mockReturnValue(
+				Effect.succeed({
+					bucketId: Option.fromNullable(bucket),
+					storageIntegrationId: Option.fromNullable(integration),
+				}),
+			);
+			const response = await app.request(`https://cap.test/create?${query}`, {
+				headers: {
+					"x-vercel-ip-latitude": "35.68",
+					"x-vercel-ip-longitude": "139.69",
+				},
+			});
+			expect(response.status).toBe(200);
+			expect(insertedValues(schema.videos)).toMatchObject({
+				bucket,
+				storageIntegrationId: integration,
+			});
+			expect(regionalStorage.select).not.toHaveBeenCalled();
+		},
+	);
+
+	it.each([null, "cap-tokyo", "custom-bucket"])(
+		"does not reroute an existing recording after travel: %s",
+		async (bucket) => {
+			mockDb.where.mockReset().mockResolvedValue([
+				{
+					id: "existing",
+					ownerId: "user-1",
+					bucket,
+					source: { type: "desktopMP4" },
+				},
+			]);
+			const response = await app.request(
+				"https://cap.test/create?videoId=existing&recordingMode=desktopMP4",
+				{
+					headers: {
+						"x-vercel-ip-latitude": "35.68",
+						"x-vercel-ip-longitude": "139.69",
+					},
+				},
+			);
+			expect(response.status).toBe(200);
+			expect(mockDb.insert).not.toHaveBeenCalled();
+			expect(mockDb.update).not.toHaveBeenCalled();
+			expect(regionalStorage.select).not.toHaveBeenCalled();
+		},
+	);
+});

apps/web/__tests__/unit/regional-organization-cleanup.test.ts

@@ -0,0 +1,152 @@
+import { CurrentUser, Organisation, S3Bucket, User } from "@cap/web-domain";
+import { Effect, Option } from "effect";
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const mocks = vi.hoisted(() => ({
+	database: vi.fn(),
+	bucket: vi.fn(),
+	deleted: vi.fn(),
+}));
+vi.mock("@cap/web-backend/src/Database", async () => {
+	const { Effect } = await import("effect");
+	class Database extends Effect.Service<Database>()("Database", {
+		sync: () => ({ use: mocks.database }),
+	}) {}
+	return { Database };
+});
+vi.mock("@cap/web-backend/src/S3Buckets", async () => {
+	const { Effect } = await import("effect");
+	class S3Buckets extends Effect.Service<S3Buckets>()("S3Buckets", {
+		sync: () => ({ getBucketAccess: mocks.bucket }),
+	}) {}
+	return { S3Buckets };
+});
+vi.mock("@cap/web-backend/src/ImageUploads", async () => {
+	const { Effect } = await import("effect");
+	class ImageUploads extends Effect.Service<ImageUploads>()("ImageUploads", {
+		sync: () => ({}),
+	}) {}
+	return { ImageUploads };
+});
+vi.mock("@cap/web-backend/src/Tinybird", async () => {
+	const { Effect } = await import("effect");
+	class Tinybird extends Effect.Service<Tinybird>()("Tinybird", {
+		sync: () => ({ deleteData: () => Effect.void }),
+	}) {}
+	return { Tinybird };
+});
+vi.mock("@cap/web-backend/src/Organisations/OrganisationsPolicy", async () => {
+	const { Effect } = await import("effect");
+	const { Policy } = await import("@cap/web-domain");
+	class OrganisationsPolicy extends Effect.Service<OrganisationsPolicy>()(
+		"OrganisationsPolicy",
+		{
+			sync: () => ({
+				isOwner: () => Policy.policy(() => Effect.succeed(true)),
+			}),
+		},
+	) {}
+	return { OrganisationsPolicy };
+});
+
+import { Organisations } from "@cap/web-backend/src/Organisations";
+
+const cleanup = () =>
+	Effect.runPromise(
+		Effect.flatMap(Organisations, (organizations) =>
+			organizations.softDelete(Organisation.OrganisationId.make("org")),
+		).pipe(
+			Effect.provide(Organisations.Default),
+			Effect.provideService(CurrentUser, {
+				id: User.UserId.make("owner"),
+				email: "owner@cap.test",
+				activeOrganizationId: Organisation.OrganisationId.make("org"),
+				iconUrlOrKey: Option.none(),
+			}),
+		),
+	);
+
+beforeEach(() => {
+	vi.resetAllMocks();
+	mocks.database
+		.mockReturnValueOnce(Effect.succeed([{ id: "org", ownerId: "owner" }]))
+		.mockReturnValueOnce(
+			Effect.succeed([
+				{
+					id: "virginia",
+					ownerId: "owner",
+					bucket: null,
+					storageIntegrationId: null,
+				},
+				{
+					id: "tokyo",
+					ownerId: "owner",
+					bucket: S3Bucket.S3BucketId.make("cap-tokyo"),
+					storageIntegrationId: null,
+				},
+				{
+					id: "custom",
+					ownerId: "owner",
+					bucket: "custom-bucket",
+					storageIntegrationId: null,
+				},
+				{
+					id: "drive",
+					ownerId: "owner",
+					bucket: null,
+					storageIntegrationId: "drive-id",
+				},
+			]),
+		)
+		.mockReturnValue(Effect.void);
+	mocks.bucket.mockImplementation((bucket: Option.Option<string>) =>
+		Effect.succeed([
+			{
+				listObjects: ({
+					prefix,
+					continuationToken,
+				}: {
+					prefix: string;
+					continuationToken?: string;
+				}) =>
+					Effect.succeed({
+						Contents: [{ Key: `${prefix}${continuationToken ?? "first"}` }],
+						IsTruncated: !continuationToken,
+						NextContinuationToken: "second",
+					}),
+				deleteObjects: (objects: Array<{ Key: string }>) =>
+					Effect.sync(() => mocks.deleted(Option.getOrNull(bucket), objects)),
+			},
+			Option.none(),
+		]),
+	);
+});
+
+describe("organization regional media cleanup", () => {
+	it("deletes both pages in each managed bucket and leaves customer storage alone", async () => {
+		await cleanup();
+		expect(mocks.deleted.mock.calls).toEqual(
+			expect.arrayContaining([
+				[null, [{ Key: "owner/virginia/first" }]],
+				[null, [{ Key: "owner/virginia/second" }]],
+				["cap-tokyo", [{ Key: "owner/tokyo/first" }]],
+				["cap-tokyo", [{ Key: "owner/tokyo/second" }]],
+				[null, [{ Key: "organizations/org/first" }]],
+				[null, [{ Key: "organizations/org/second" }]],
+			]),
+		);
+		expect(mocks.deleted).toHaveBeenCalledTimes(6);
+		expect(mocks.database).toHaveBeenCalledTimes(3);
+	});
+
+	it("keeps the database records when the regional bucket cannot be opened", async () => {
+		const original = mocks.bucket.getMockImplementation();
+		mocks.bucket.mockImplementation((bucket: Option.Option<string>) =>
+			Option.getOrNull(bucket) === S3Bucket.S3BucketId.make("cap-tokyo")
+				? Effect.fail(new Error("Tokyo unavailable"))
+				: original?.(bucket),
+		);
+		await expect(cleanup()).rejects.toThrow("Tokyo unavailable");
+		expect(mocks.database).toHaveBeenCalledTimes(2);
+	});
+});