diff --git a/cloudflare/src/index.ts b/cloudflare/src/index.ts index f04a28c..f79ee73 100644 --- a/cloudflare/src/index.ts +++ b/cloudflare/src/index.ts @@ -41,6 +41,14 @@ import { SyncPushRequestError, syncPushDocument, } from "./sync_push.js"; +import { + SyncSnapshotConflictError, + SyncSnapshotNotFoundError, + SyncSnapshotPersistenceError, + SyncSnapshotRequestError, + syncSnapshotDownloadDocument, + syncSnapshotUploadDocument, +} from "./sync_snapshot.js"; import { SyncRequestError, SyncSchemaError, syncPullDocument } from "./sync_pull.js"; import { jsonResponse } from "./responses.js"; @@ -245,6 +253,56 @@ export async function handleRequest(request: Request, env: Env): Promise { + try { + if (request.method === "POST") { + return jsonResponse(await syncSnapshotUploadDocument(request, env, context), 201, { + "Cache-Control": "no-store", + }); + } + return jsonResponse(await syncSnapshotDownloadDocument(url, env, context), 200, { + "Cache-Control": "no-store", + }); + } catch (error) { + if (error instanceof SyncSnapshotRequestError) { + return jsonResponse( + { error: "invalid_sync_snapshot" }, + 400, + { "Cache-Control": "no-store" }, + ); + } + if (error instanceof SyncSnapshotNotFoundError) { + return jsonResponse( + { error: "sync_snapshot_not_found" }, + 404, + { "Cache-Control": "no-store" }, + ); + } + if (error instanceof SyncSnapshotConflictError) { + return jsonResponse( + { error: "sync_snapshot_conflict" }, + 409, + { "Cache-Control": "no-store" }, + ); + } + if (error instanceof SyncSnapshotPersistenceError) { + return jsonResponse( + { error: "sync_snapshot_failed" }, + 500, + { "Cache-Control": "no-store" }, + ); + } + throw error; + } + }, + ); + } if (url.pathname === "/api/plugins/signing-keys") { return withPublicApiControls(request, env, "plugins.signing_keys", ["GET"], () => handlePublicSigningKeys(env), diff --git a/cloudflare/src/sync_snapshot.ts b/cloudflare/src/sync_snapshot.ts new file mode 100644 index 0000000..84917e3 --- /dev/null +++ b/cloudflare/src/sync_snapshot.ts @@ -0,0 +1,493 @@ +import type { AuthContext } from "./auth.js"; +import type { Env } from "./bindings.js"; +import { StorageObjectError, getVerifiedObject, putVerifiedObject, syncSnapshotKey } from "./storage.js"; + +const MAX_SNAPSHOT_BYTES = 10 * 1024 * 1024; +const SNAPSHOT_ID_PATTERN = /^[a-z0-9][a-z0-9._-]{0,127}$/; +const DEVICE_ID_PATTERN = /^[a-zA-Z0-9._:-]{3,128}$/; +const SHA256_HEX = /^[a-f0-9]{64}$/; +const BASE64 = /^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?$/; +const REGION = /^[a-z0-9][a-z0-9-]{1,31}$/; +const SYNC_SNAPSHOT_BY_ID_QUERY = ` + SELECT + snapshot_id, + r2_key, + payload_hash, + schema_rev, + logical_clock, + device_id, + size_bytes, + created_at + FROM sync_snapshots + WHERE user_id = ? AND snapshot_id = ? +`; +const SYNC_SNAPSHOT_UPSERT_QUERY = ` + INSERT INTO sync_snapshots ( + user_id, + snapshot_id, + r2_key, + payload_hash, + schema_rev, + logical_clock, + device_id, + size_bytes, + created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(user_id, snapshot_id) DO UPDATE SET + r2_key = excluded.r2_key, + payload_hash = excluded.payload_hash, + schema_rev = excluded.schema_rev, + logical_clock = excluded.logical_clock, + device_id = excluded.device_id, + size_bytes = excluded.size_bytes, + created_at = excluded.created_at + WHERE excluded.logical_clock > sync_snapshots.logical_clock + OR ( + excluded.logical_clock = sync_snapshots.logical_clock + AND sync_snapshots.r2_key = excluded.r2_key + AND sync_snapshots.payload_hash = excluded.payload_hash + AND sync_snapshots.schema_rev = excluded.schema_rev + AND sync_snapshots.device_id = excluded.device_id + AND sync_snapshots.size_bytes = excluded.size_bytes + ) +`; + +export interface SyncSnapshotUploadDocument { + version: 1; + user_id: string; + device_id: string; + snapshot: SyncSnapshotDocument; +} + +export interface SyncSnapshotDownloadDocument extends SyncSnapshotUploadDocument { + data_base64: string; +} + +export interface SyncSnapshotDocument { + snapshot_id: string; + r2_key: string; + payload_hash: string; + schema_rev: number; + logical_clock: number; + device_id: string; + size_bytes: number; + created_at: number; +} + +interface SyncSnapshotUploadRequest { + snapshotId: string; + r2Key: string; + payloadHash: string; + schemaRev: number; + logicalClock: number; + bytes: ArrayBuffer; +} + +interface SyncSnapshotRow { + snapshot_id: unknown; + r2_key: unknown; + payload_hash: unknown; + schema_rev: unknown; + logical_clock: unknown; + device_id: unknown; + size_bytes: unknown; + created_at: unknown; +} + +type RequestBody = Record; + +export class SyncSnapshotRequestError extends Error { + constructor(message: string) { + super(message); + this.name = "SyncSnapshotRequestError"; + } +} + +export class SyncSnapshotConflictError extends Error { + constructor(message: string) { + super(message); + this.name = "SyncSnapshotConflictError"; + } +} + +export class SyncSnapshotNotFoundError extends Error { + constructor(message: string) { + super(message); + this.name = "SyncSnapshotNotFoundError"; + } +} + +export class SyncSnapshotPersistenceError extends Error { + constructor(message: string) { + super(message); + this.name = "SyncSnapshotPersistenceError"; + } +} + +export async function syncSnapshotUploadDocument( + request: Request, + env: Env, + context: AuthContext, + nowSeconds = Math.floor(Date.now() / 1000), +): Promise { + const deviceId = currentDeviceId(context); + const snapshot = await syncSnapshotUploadRequest(request, context.userId); + const existingRow = await snapshotRow(env, context.userId, snapshot.snapshotId); + if (existingRow !== null) { + assertSnapshotCanReplaceExisting(snapshot, deviceId, syncSnapshotDocumentFromRow(existingRow)); + } + + await persistSnapshot(env, snapshot); + await env.ELY_DB.batch([ + env.ELY_DB.prepare(SYNC_SNAPSHOT_UPSERT_QUERY).bind( + context.userId, + snapshot.snapshotId, + snapshot.r2Key, + snapshot.payloadHash, + snapshot.schemaRev, + snapshot.logicalClock, + deviceId, + snapshot.bytes.byteLength, + nowSeconds, + ), + ]); + + const savedRow = await snapshotRow(env, context.userId, snapshot.snapshotId); + if (savedRow === null) { + throw new SyncSnapshotPersistenceError("sync_snapshot_missing"); + } + const savedSnapshot = syncSnapshotDocumentFromRow(savedRow); + assertSavedSnapshotMatchesUpload(snapshot, deviceId, savedSnapshot); + + return { + version: 1, + user_id: context.userId, + device_id: deviceId, + snapshot: savedSnapshot, + }; +} + +export async function syncSnapshotDownloadDocument( + url: URL, + env: Env, + context: AuthContext, +): Promise { + const deviceId = currentDeviceId(context); + const snapshotId = syncSnapshotDownloadQuery(url); + const row = await snapshotRow(env, context.userId, snapshotId); + if (row === null) { + throw new SyncSnapshotNotFoundError("sync_snapshot_missing"); + } + + const snapshot = syncSnapshotDocumentFromRow(row); + let payload: ArrayBuffer | null; + try { + payload = await getVerifiedObject(env.ELY_STORAGE, snapshot.r2_key, snapshot.payload_hash); + } catch (error) { + if (error instanceof StorageObjectError) { + throw new SyncSnapshotPersistenceError(error.message); + } + throw error; + } + if (payload === null) { + throw new SyncSnapshotPersistenceError("sync_snapshot_payload_missing"); + } + + return { + version: 1, + user_id: context.userId, + device_id: deviceId, + snapshot, + data_base64: base64FromBytes(payload), + }; +} + +async function syncSnapshotUploadRequest( + request: Request, + userId: string, +): Promise { + const body = await requestBody(request); + assertOnlyFields(body, [ + "version", + "snapshot_id", + "region", + "payload_hash", + "schema_rev", + "logical_clock", + "data_base64", + ]); + if (body.version !== 1) { + throw new SyncSnapshotRequestError("version_invalid"); + } + + const snapshotId = snapshotIdValue(body.snapshot_id); + const region = regionValue(body.region); + const payloadHash = sha256HexValue(body.payload_hash, "payload_hash"); + const bytes = payloadBytes(body.data_base64, "data_base64", MAX_SNAPSHOT_BYTES); + await assertPayloadHash(bytes, payloadHash); + return { + snapshotId, + r2Key: await snapshotStorageKey(region, userId, snapshotId), + payloadHash, + schemaRev: integer(body.schema_rev, "schema_rev", 1, Number.MAX_SAFE_INTEGER), + logicalClock: integer(body.logical_clock, "logical_clock", 0, Number.MAX_SAFE_INTEGER), + bytes, + }; +} + +function syncSnapshotDownloadQuery(url: URL): string { + assertOnlyQueryParams(url, ["snapshot_id"]); + return snapshotIdValue(url.searchParams.get("snapshot_id")); +} + +async function snapshotStorageKey( + region: string, + userId: string, + snapshotId: string, +): Promise { + try { + return syncSnapshotKey({ + region, + userHash: await sha256Hex(arrayBufferFromBytes(new TextEncoder().encode(userId))), + snapshotId, + }); + } catch (error) { + if (error instanceof StorageObjectError) { + throw new SyncSnapshotRequestError(error.message); + } + throw error; + } +} + +async function snapshotRow( + env: Env, + userId: string, + snapshotId: string, +): Promise { + return env.ELY_DB.prepare(SYNC_SNAPSHOT_BY_ID_QUERY).bind(userId, snapshotId).first(); +} + +function syncSnapshotDocumentFromRow(row: SyncSnapshotRow): SyncSnapshotDocument { + try { + return syncSnapshotDocument(row); + } catch (error) { + if (error instanceof SyncSnapshotRequestError) { + throw new SyncSnapshotPersistenceError(error.message); + } + throw error; + } +} + +function syncSnapshotDocument(row: SyncSnapshotRow): SyncSnapshotDocument { + return { + snapshot_id: snapshotIdValue(row.snapshot_id), + r2_key: text(row.r2_key, "r2_key"), + payload_hash: sha256HexValue(row.payload_hash, "payload_hash"), + schema_rev: integer(row.schema_rev, "schema_rev", 1, Number.MAX_SAFE_INTEGER), + logical_clock: integer(row.logical_clock, "logical_clock", 0, Number.MAX_SAFE_INTEGER), + device_id: deviceIdValue(row.device_id), + size_bytes: integer(row.size_bytes, "size_bytes", 1, MAX_SNAPSHOT_BYTES), + created_at: integer(row.created_at, "created_at", 0, Number.MAX_SAFE_INTEGER), + }; +} + +function assertSnapshotCanReplaceExisting( + snapshot: SyncSnapshotUploadRequest, + deviceId: string, + existing: SyncSnapshotDocument, +): void { + if (existing.logical_clock > snapshot.logicalClock) { + throw new SyncSnapshotConflictError("logical_clock_stale"); + } + if (existing.logical_clock < snapshot.logicalClock) { + return; + } + if ( + existing.payload_hash !== snapshot.payloadHash || + existing.schema_rev !== snapshot.schemaRev || + existing.device_id !== deviceId || + existing.size_bytes !== snapshot.bytes.byteLength + ) { + throw new SyncSnapshotConflictError("logical_clock_conflict"); + } +} + +function assertSavedSnapshotMatchesUpload( + upload: SyncSnapshotUploadRequest, + deviceId: string, + snapshot: SyncSnapshotDocument, +): void { + if (snapshot.logical_clock > upload.logicalClock) { + throw new SyncSnapshotConflictError("logical_clock_stale"); + } + if ( + snapshot.logical_clock === upload.logicalClock && + (snapshot.r2_key !== upload.r2Key || + snapshot.payload_hash !== upload.payloadHash || + snapshot.schema_rev !== upload.schemaRev || + snapshot.device_id !== deviceId || + snapshot.size_bytes !== upload.bytes.byteLength) + ) { + throw new SyncSnapshotConflictError("logical_clock_conflict"); + } + if ( + snapshot.snapshot_id !== upload.snapshotId || + snapshot.r2_key !== upload.r2Key || + snapshot.payload_hash !== upload.payloadHash || + snapshot.schema_rev !== upload.schemaRev || + snapshot.logical_clock !== upload.logicalClock || + snapshot.device_id !== deviceId || + snapshot.size_bytes !== upload.bytes.byteLength + ) { + throw new SyncSnapshotPersistenceError("sync_snapshot_mismatch"); + } +} + +async function persistSnapshot(env: Env, snapshot: SyncSnapshotUploadRequest): Promise { + try { + await putVerifiedObject( + env.ELY_STORAGE, + snapshot.r2Key, + snapshot.bytes, + snapshot.payloadHash, + "application/octet-stream", + ); + } catch (error) { + if (error instanceof StorageObjectError) { + throw new SyncSnapshotRequestError(error.message); + } + throw error; + } +} + +function currentDeviceId(context: AuthContext): string { + if (context.deviceId === undefined) { + throw new SyncSnapshotRequestError("device_context_required"); + } + return context.deviceId; +} + +async function requestBody(request: Request): Promise { + let value: unknown; + try { + value = await request.json(); + } catch { + throw new SyncSnapshotRequestError("json_invalid"); + } + return record(value, "body"); +} + +function record(value: unknown, label: string): RequestBody { + if (typeof value !== "object" || value === null || Array.isArray(value)) { + throw new SyncSnapshotRequestError(`${label}_invalid`); + } + return value as RequestBody; +} + +function assertOnlyFields(value: RequestBody, fields: string[]): void { + const allowed = new Set(fields); + for (const field of Object.keys(value)) { + if (!allowed.has(field)) { + throw new SyncSnapshotRequestError(`unexpected_field:${field}`); + } + } +} + +function assertOnlyQueryParams(url: URL, fields: string[]): void { + const allowed = new Set(fields); + for (const field of url.searchParams.keys()) { + if (!allowed.has(field)) { + throw new SyncSnapshotRequestError(`unexpected_query:${field}`); + } + } +} + +function snapshotIdValue(value: unknown): string { + if (typeof value !== "string" || !SNAPSHOT_ID_PATTERN.test(value)) { + throw new SyncSnapshotRequestError("snapshot_id_invalid"); + } + return value; +} + +function deviceIdValue(value: unknown): string { + if (typeof value !== "string" || !DEVICE_ID_PATTERN.test(value)) { + throw new SyncSnapshotRequestError("device_id_invalid"); + } + return value; +} + +function regionValue(value: unknown): string { + if (typeof value !== "string" || !REGION.test(value)) { + throw new SyncSnapshotRequestError("region_invalid"); + } + return value; +} + +function sha256HexValue(value: unknown, label: string): string { + if (typeof value !== "string" || !SHA256_HEX.test(value)) { + throw new SyncSnapshotRequestError(`${label}_invalid`); + } + return value; +} + +function integer(value: unknown, label: string, min: number, max: number): number { + if (typeof value !== "number" || !Number.isSafeInteger(value) || value < min || value > max) { + throw new SyncSnapshotRequestError(`${label}_invalid`); + } + return value; +} + +function text(value: unknown, label: string): string { + if (typeof value !== "string" || value.length === 0) { + throw new SyncSnapshotRequestError(`${label}_invalid`); + } + return value; +} + +function payloadBytes(value: unknown, label: string, maxBytes: number): ArrayBuffer { + const encoded = text(value, label); + if (!BASE64.test(encoded)) { + throw new SyncSnapshotRequestError(`${label}_invalid`); + } + const bytes = bytesFromBase64(encoded); + if (bytes.byteLength === 0 || bytes.byteLength > maxBytes) { + throw new SyncSnapshotRequestError(`${label}_size_invalid`); + } + return bytes; +} + +function bytesFromBase64(value: string): ArrayBuffer { + const binary = atob(value); + const bytes = new Uint8Array(binary.length); + for (let index = 0; index < binary.length; index += 1) { + bytes[index] = binary.charCodeAt(index); + } + return bytes.buffer; +} + +function base64FromBytes(payload: ArrayBuffer): string { + const bytes = new Uint8Array(payload); + const parts: string[] = []; + for (let offset = 0; offset < bytes.length; offset += 0x8000) { + parts.push(String.fromCharCode(...bytes.subarray(offset, offset + 0x8000))); + } + return btoa(parts.join("")); +} + +function arrayBufferFromBytes(bytes: Uint8Array): ArrayBuffer { + const copy = new Uint8Array(bytes.byteLength); + copy.set(bytes); + return copy.buffer; +} + +async function assertPayloadHash(payload: ArrayBuffer, expectedHash: string): Promise { + const actualHash = await sha256Hex(payload); + if (actualHash !== expectedHash) { + throw new SyncSnapshotRequestError("payload_hash_mismatch"); + } +} + +async function sha256Hex(payload: ArrayBuffer): Promise { + const digest = await crypto.subtle.digest("SHA-256", payload); + return [...new Uint8Array(digest)].map((byte) => byte.toString(16).padStart(2, "0")).join(""); +} diff --git a/cloudflare/tests/devices_test_support.ts b/cloudflare/tests/devices_test_support.ts index 29c3ad8..a72b3ce 100644 --- a/cloudflare/tests/devices_test_support.ts +++ b/cloudflare/tests/devices_test_support.ts @@ -14,6 +14,8 @@ export interface TestEnvOptions { d1?: RecordedD1Database; kvEntries?: [string, string][]; kvReads?: string[]; + r2Gets?: string[]; + r2Objects?: [string, ArrayBuffer][]; r2Puts?: RecordedR2Put[]; } @@ -45,7 +47,7 @@ export function testEnv(options: TestEnvOptions): Env { return Promise.resolve(values.get(key) ?? null); }, }, - ELY_STORAGE: testR2Bucket(options.r2Puts), + ELY_STORAGE: testR2Bucket(options.r2Puts, options.r2Objects, options.r2Gets), ELY_RATE_LIMITER: { limit(): Promise<{ success: boolean }> { return Promise.resolve({ success: true }); @@ -119,13 +121,28 @@ function testD1PreparedStatement( }; } -function testR2Bucket(puts: RecordedR2Put[] = []): Env["ELY_STORAGE"] { +function testR2Bucket( + puts: RecordedR2Put[] = [], + objects: [string, ArrayBuffer][] = [], + gets: string[] = [], +): Env["ELY_STORAGE"] { + const values = new Map(objects); return { - get() { - return Promise.resolve(null); + get(key: string) { + gets.push(key); + const value = values.get(key); + if (value === undefined) { + return Promise.resolve(null); + } + return Promise.resolve({ + arrayBuffer() { + return Promise.resolve(value); + }, + }); }, put(key: string, value: ArrayBuffer, options: ElyR2PutOptions = {}) { puts.push({ key, payload: value, options }); + values.set(key, value); return Promise.resolve({ arrayBuffer() { return Promise.resolve(value); diff --git a/cloudflare/tests/sync_snapshot_routes.test.ts b/cloudflare/tests/sync_snapshot_routes.test.ts new file mode 100644 index 0000000..bd8fa48 --- /dev/null +++ b/cloudflare/tests/sync_snapshot_routes.test.ts @@ -0,0 +1,321 @@ +import assert from "node:assert/strict"; +import { createHash } from "node:crypto"; +import { describe, it } from "node:test"; + +import { authSessionCacheKvKey, authTokenHash } from "../src/auth.js"; +import { handleRequest } from "../src/index.js"; +import { + ACCESS_TOKEN, + type RecordedR2Put, + sessionDocument, + testD1Database, + testEnv, +} from "./devices_test_support.js"; + +const USER_ID = "user-01"; +const DEVICE_ID = "device-01"; +const SNAPSHOT_ID = "snapshot-01"; +const REGION = "us-east"; + +describe("sync snapshot routes", () => { + it("uploads an encrypted snapshot from an approved current device", async () => { + const payload = bytes("encrypted snapshot payload"); + const payloadHash = sha256(payload); + const key = snapshotKey(payloadHash); + const r2Puts: RecordedR2Put[] = []; + const tokenHash = await authTokenHash(ACCESS_TOKEN); + const d1 = testD1Database({ + firstRows: [ + { device_id: DEVICE_ID }, + null, + snapshotRow({ r2_key: key, payload_hash: payloadHash, size_bytes: payload.byteLength }), + ], + }); + + const response = await handleRequest( + syncSnapshotPostRequest( + syncSnapshotBody({ payload_hash: payloadHash, data_base64: base64(payload) }), + ), + testEnv({ + d1, + r2Puts, + kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument(DEVICE_ID)]], + }), + ); + + assert.equal(response.status, 201); + assert.equal(response.headers.get("cache-control"), "no-store"); + assert.deepEqual(await response.json(), { + version: 1, + user_id: USER_ID, + device_id: DEVICE_ID, + snapshot: snapshotDocument({ + r2_key: key, + payload_hash: payloadHash, + size_bytes: payload.byteLength, + }), + }); + assert.equal(r2Puts.length, 1); + assert.equal(r2Puts[0]?.key, key); + assert.deepEqual(new Uint8Array(r2Puts[0]?.payload ?? new ArrayBuffer(0)), new Uint8Array(payload)); + assert.equal(r2Puts[0]?.options.customMetadata?.sha256, payloadHash); + assert.equal(d1.batches[0], 1); + assert.ok(d1.queries[1]?.includes("FROM sync_snapshots")); + assert.ok(d1.queries[2]?.includes("INSERT INTO sync_snapshots")); + assert.deepEqual(d1.binds[2]?.slice(0, 4), [USER_ID, SNAPSHOT_ID, key, payloadHash]); + }); + + it("downloads an encrypted snapshot with R2 checksum verification", async () => { + const payload = bytes("encrypted snapshot payload"); + const payloadHash = sha256(payload); + const key = snapshotKey(payloadHash); + const r2Gets: string[] = []; + const tokenHash = await authTokenHash(ACCESS_TOKEN); + const d1 = testD1Database({ + firstRows: [ + { device_id: DEVICE_ID }, + snapshotRow({ r2_key: key, payload_hash: payloadHash, size_bytes: payload.byteLength }), + ], + }); + + const response = await handleRequest( + syncSnapshotGetRequest(), + testEnv({ + d1, + r2Gets, + r2Objects: [[key, payload]], + kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument(DEVICE_ID)]], + }), + ); + + assert.equal(response.status, 200); + assert.deepEqual(await response.json(), { + version: 1, + user_id: USER_ID, + device_id: DEVICE_ID, + snapshot: snapshotDocument({ + r2_key: key, + payload_hash: payloadHash, + size_bytes: payload.byteLength, + }), + data_base64: base64(payload), + }); + assert.deepEqual(r2Gets, [key]); + }); + + it("returns not found for missing snapshot indexes", async () => { + const r2Gets: string[] = []; + const tokenHash = await authTokenHash(ACCESS_TOKEN); + const d1 = testD1Database({ firstRows: [{ device_id: DEVICE_ID }, null] }); + + const response = await handleRequest( + syncSnapshotGetRequest(), + testEnv({ + d1, + r2Gets, + kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument(DEVICE_ID)]], + }), + ); + + assert.equal(response.status, 404); + assert.deepEqual(await response.json(), { error: "sync_snapshot_not_found" }); + assert.deepEqual(r2Gets, []); + }); + + it("rejects snapshot checksum mismatches before D1 writes", async () => { + const payload = bytes("encrypted snapshot payload"); + const r2Puts: RecordedR2Put[] = []; + const tokenHash = await authTokenHash(ACCESS_TOKEN); + const d1 = testD1Database({ firstRows: [{ device_id: DEVICE_ID }] }); + + const response = await handleRequest( + syncSnapshotPostRequest( + syncSnapshotBody({ payload_hash: "c".repeat(64), data_base64: base64(payload) }), + ), + testEnv({ + d1, + r2Puts, + kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument(DEVICE_ID)]], + }), + ); + + assert.equal(response.status, 400); + assert.deepEqual(await response.json(), { error: "invalid_sync_snapshot" }); + assert.equal(r2Puts.length, 0); + assert.equal(d1.queries.length, 1); + assert.deepEqual(d1.batches, []); + }); + + it("rejects stale snapshot clocks before R2 writes", async () => { + const payload = bytes("encrypted snapshot payload"); + const payloadHash = sha256(payload); + const r2Puts: RecordedR2Put[] = []; + const tokenHash = await authTokenHash(ACCESS_TOKEN); + const d1 = testD1Database({ + firstRows: [ + { device_id: DEVICE_ID }, + snapshotRow({ payload_hash: "d".repeat(64), logical_clock: 43 }), + ], + }); + + const response = await handleRequest( + syncSnapshotPostRequest( + syncSnapshotBody({ + payload_hash: payloadHash, + logical_clock: 42, + data_base64: base64(payload), + }), + ), + testEnv({ + d1, + r2Puts, + kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument(DEVICE_ID)]], + }), + ); + + assert.equal(response.status, 409); + assert.deepEqual(await response.json(), { error: "sync_snapshot_conflict" }); + assert.equal(r2Puts.length, 0); + assert.deepEqual(d1.batches, []); + }); + + it("rejects same-clock snapshot write races after D1 persistence", async () => { + const payload = bytes("encrypted snapshot payload"); + const payloadHash = sha256(payload); + const key = snapshotKey(payloadHash); + const r2Puts: RecordedR2Put[] = []; + const tokenHash = await authTokenHash(ACCESS_TOKEN); + const d1 = testD1Database({ + firstRows: [ + { device_id: DEVICE_ID }, + null, + snapshotRow({ r2_key: key, payload_hash: "e".repeat(64), size_bytes: payload.byteLength }), + ], + }); + + const response = await handleRequest( + syncSnapshotPostRequest( + syncSnapshotBody({ payload_hash: payloadHash, data_base64: base64(payload) }), + ), + testEnv({ + d1, + r2Puts, + kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument(DEVICE_ID)]], + }), + ); + + assert.equal(response.status, 409); + assert.deepEqual(await response.json(), { error: "sync_snapshot_conflict" }); + assert.equal(r2Puts.length, 1); + assert.equal(d1.batches[0], 1); + }); + + it("rejects revoked devices before reading snapshot payloads", async () => { + const tokenHash = await authTokenHash(ACCESS_TOKEN); + const d1 = testD1Database({ firstRows: [null] }); + + const response = await handleRequest( + syncSnapshotPostRequest(syncSnapshotBody()), + testEnv({ + d1, + kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument(DEVICE_ID)]], + }), + ); + + assert.equal(response.status, 403); + assert.deepEqual(await response.json(), { error: "device_not_approved" }); + assert.equal(d1.queries.length, 1); + assert.deepEqual(d1.batches, []); + }); + + it("fails closed when a stored snapshot payload fails checksum verification", async () => { + const payload = bytes("encrypted snapshot payload"); + const payloadHash = sha256(payload); + const key = snapshotKey(payloadHash); + const tokenHash = await authTokenHash(ACCESS_TOKEN); + const d1 = testD1Database({ + firstRows: [ + { device_id: DEVICE_ID }, + snapshotRow({ r2_key: key, payload_hash: payloadHash, size_bytes: payload.byteLength }), + ], + }); + + const response = await handleRequest( + syncSnapshotGetRequest(), + testEnv({ + d1, + r2Objects: [[key, bytes("corrupt snapshot payload")]], + kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument(DEVICE_ID)]], + }), + ); + + assert.equal(response.status, 500); + assert.deepEqual(await response.json(), { error: "sync_snapshot_failed" }); + }); +}); + +function syncSnapshotPostRequest(body: Record): Request { + return new Request("https://elydora.test/api/sync/snapshot", { + method: "POST", + headers: { + authorization: `Bearer ${ACCESS_TOKEN}`, + "content-type": "application/json", + }, + body: JSON.stringify(body), + }); +} + +function syncSnapshotGetRequest(): Request { + return new Request(`https://elydora.test/api/sync/snapshot?snapshot_id=${SNAPSHOT_ID}`, { + headers: { authorization: `Bearer ${ACCESS_TOKEN}` }, + }); +} + +function syncSnapshotBody(overrides: Record = {}): Record { + const payload = bytes("encrypted snapshot payload"); + const payloadHash = sha256(payload); + return { + version: 1, + snapshot_id: SNAPSHOT_ID, + region: REGION, + payload_hash: payloadHash, + schema_rev: 1, + logical_clock: 42, + data_base64: base64(payload), + ...overrides, + }; +} + +function snapshotRow(overrides: Record = {}): Record { + return { + snapshot_id: SNAPSHOT_ID, + r2_key: snapshotKey("a".repeat(64)), + payload_hash: "a".repeat(64), + schema_rev: 1, + logical_clock: 42, + device_id: DEVICE_ID, + size_bytes: 26, + created_at: 1_780_000_900, + ...overrides, + }; +} + +function snapshotDocument(overrides: Record = {}): Record { + return snapshotRow(overrides); +} + +function snapshotKey(_payloadHash: string): string { + return `sync-snapshots/${REGION}/${sha256(bytes(USER_ID))}/${SNAPSHOT_ID}.bin`; +} + +function bytes(value: string): ArrayBuffer { + return new TextEncoder().encode(value).buffer; +} + +function base64(payload: ArrayBuffer): string { + return Buffer.from(payload).toString("base64"); +} + +function sha256(payload: ArrayBuffer): string { + return createHash("sha256").update(new Uint8Array(payload)).digest("hex"); +}