diff --git a/cloudflare/src/index.ts b/cloudflare/src/index.ts index 29858b0..f04a28c 100644 --- a/cloudflare/src/index.ts +++ b/cloudflare/src/index.ts @@ -35,6 +35,12 @@ import { parsePublicSigningKeysDocument, publicSigningKeysKvKey, } from "./signing_keys.js"; +import { + SyncPushConflictError, + SyncPushPersistenceError, + SyncPushRequestError, + syncPushDocument, +} from "./sync_push.js"; import { SyncRequestError, SyncSchemaError, syncPullDocument } from "./sync_pull.js"; import { jsonResponse } from "./responses.js"; @@ -205,6 +211,40 @@ export async function handleRequest(request: Request, env: Env): Promise { + try { + return jsonResponse(await syncPushDocument(request, env, context), 201, { + "Cache-Control": "no-store", + }); + } catch (error) { + if (error instanceof SyncPushRequestError) { + return jsonResponse( + { error: "invalid_sync_push" }, + 400, + { "Cache-Control": "no-store" }, + ); + } + if (error instanceof SyncPushConflictError) { + return jsonResponse({ error: "sync_conflict" }, 409, { "Cache-Control": "no-store" }); + } + if (error instanceof SyncPushPersistenceError) { + return jsonResponse( + { error: "sync_push_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/storage.ts b/cloudflare/src/storage.ts index 63ee986..dfeb1d5 100644 --- a/cloudflare/src/storage.ts +++ b/cloudflare/src/storage.ts @@ -195,7 +195,7 @@ function assertSegment(value: string, name: string): void { } } -function assertSyncObjectType(value: string): void { +export function assertSyncObjectType(value: string): void { if (!SYNC_OBJECT_TYPES.has(value)) { throw new StorageObjectError("object_type_invalid"); } diff --git a/cloudflare/src/sync_push.ts b/cloudflare/src/sync_push.ts new file mode 100644 index 0000000..4bc0169 --- /dev/null +++ b/cloudflare/src/sync_push.ts @@ -0,0 +1,255 @@ +import type { AuthContext } from "./auth.js"; +import type { Env } from "./bindings.js"; +import type { ElyD1PreparedStatement } from "./bindings.js"; +import { StorageObjectError, putVerifiedObject } from "./storage.js"; +import { + type SyncObjectRow, + SyncPushConflictError, + type SyncPushDocument, + SyncPushPersistenceError, + type SyncPushRequest, + SyncPushRequestError, + type SyncPushedObjectDocument, + currentDeviceId, + syncObjectDocument, + syncPushRequest, +} from "./sync_push_schema.js"; + +export { + SyncPushConflictError, + SyncPushPersistenceError, + SyncPushRequestError, +} from "./sync_push_schema.js"; + +const SYNC_OBJECT_BY_ID_QUERY = ` + SELECT + object_id, + object_type, + payload_r2_key, + payload_hash, + schema_rev, + logical_clock, + device_id, + created_at, + updated_at, + deleted_at + FROM sync_objects + WHERE user_id = ? AND object_id = ? +`; +const SYNC_OBJECT_UPSERT_QUERY = ` + INSERT INTO sync_objects ( + user_id, + object_id, + object_type, + payload_inline, + payload_r2_key, + payload_hash, + schema_rev, + logical_clock, + device_id, + created_at, + updated_at, + deleted_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(user_id, object_id) DO UPDATE SET + object_type = excluded.object_type, + payload_inline = excluded.payload_inline, + payload_r2_key = excluded.payload_r2_key, + payload_hash = excluded.payload_hash, + schema_rev = excluded.schema_rev, + logical_clock = excluded.logical_clock, + device_id = excluded.device_id, + updated_at = excluded.updated_at, + deleted_at = excluded.deleted_at + WHERE excluded.logical_clock >= sync_objects.logical_clock +`; +const SYNC_CHANGE_INSERT_QUERY = ` + INSERT INTO sync_change_log ( + user_id, + object_id, + object_type, + operation, + payload_hash, + logical_clock, + device_id, + created_at + ) + SELECT ?, ?, ?, ?, ?, ?, ?, ? + WHERE EXISTS ( + SELECT 1 + FROM sync_objects + WHERE user_id = ? + AND object_id = ? + AND object_type = ? + AND payload_hash = ? + AND logical_clock = ? + AND device_id = ? + AND ((? = 1 AND deleted_at IS NOT NULL) OR (? = 0 AND deleted_at IS NULL)) + ) + ON CONFLICT(user_id, object_id, logical_clock, device_id, operation) DO NOTHING +`; +const SYNC_TOMBSTONE_UPSERT_QUERY = ` + INSERT INTO sync_tombstones ( + user_id, + object_id, + object_type, + logical_clock, + device_id, + deleted_at + ) + SELECT user_id, object_id, object_type, logical_clock, device_id, deleted_at + FROM sync_objects + WHERE user_id = ? AND object_id = ? AND logical_clock = ? AND deleted_at IS NOT NULL + ON CONFLICT(user_id, object_id) DO UPDATE SET + object_type = excluded.object_type, + logical_clock = excluded.logical_clock, + device_id = excluded.device_id, + deleted_at = excluded.deleted_at + WHERE excluded.logical_clock >= sync_tombstones.logical_clock +`; + +export async function syncPushDocument( + request: Request, + env: Env, + context: AuthContext, + nowSeconds = Math.floor(Date.now() / 1000), +): Promise { + const deviceId = currentDeviceId(context); + const push = await syncPushRequest(request, context.userId); + const existingRow = await env.ELY_DB.prepare(SYNC_OBJECT_BY_ID_QUERY) + .bind(context.userId, push.objectId) + .first(); + if (existingRow !== null) { + assertPushCanReplaceExisting(push, deviceId, syncObjectDocument(existingRow)); + } + + await persistR2PayloadIfNeeded(env, push); + await env.ELY_DB.batch(syncPushStatements(env, context.userId, deviceId, push, nowSeconds)); + + const savedRow = await env.ELY_DB.prepare(SYNC_OBJECT_BY_ID_QUERY) + .bind(context.userId, push.objectId) + .first(); + if (savedRow === null) { + throw new SyncPushPersistenceError("sync_object_missing"); + } + const object = syncObjectDocument(savedRow); + assertSavedObjectMatchesPush(push, deviceId, object); + + return { version: 1, user_id: context.userId, device_id: deviceId, object }; +} + +async function persistR2PayloadIfNeeded(env: Env, push: SyncPushRequest): Promise { + if (push.payload.kind !== "r2") { + return; + } + try { + await putVerifiedObject( + env.ELY_STORAGE, + push.payload.r2Key, + push.payload.bytes, + push.payloadHash, + "application/octet-stream", + ); + } catch (error) { + if (error instanceof StorageObjectError) { + throw new SyncPushRequestError(error.message); + } + throw error; + } +} + +function syncPushStatements( + env: Env, + userId: string, + deviceId: string, + push: SyncPushRequest, + nowSeconds: number, +): ElyD1PreparedStatement[] { + const deletedAt = push.operation === "delete" ? nowSeconds : null; + const isDelete = push.operation === "delete" ? 1 : 0; + const statements = [ + env.ELY_DB.prepare(SYNC_OBJECT_UPSERT_QUERY).bind( + userId, + push.objectId, + push.objectType, + push.payload.kind === "inline" ? push.payload.bytes : null, + push.payload.r2Key, + push.payloadHash, + push.schemaRev, + push.logicalClock, + deviceId, + nowSeconds, + nowSeconds, + deletedAt, + ), + env.ELY_DB.prepare(SYNC_CHANGE_INSERT_QUERY).bind( + userId, + push.objectId, + push.objectType, + push.operation, + push.payloadHash, + push.logicalClock, + deviceId, + nowSeconds, + userId, + push.objectId, + push.objectType, + push.payloadHash, + push.logicalClock, + deviceId, + isDelete, + isDelete, + ), + ]; + if (push.operation === "delete") { + statements.push( + env.ELY_DB.prepare(SYNC_TOMBSTONE_UPSERT_QUERY).bind( + userId, + push.objectId, + push.logicalClock, + ), + ); + } + return statements; +} + +function assertPushCanReplaceExisting( + push: SyncPushRequest, + deviceId: string, + existing: SyncPushedObjectDocument, +): void { + if (existing.logical_clock > push.logicalClock) { + throw new SyncPushConflictError("logical_clock_stale"); + } + if (existing.logical_clock < push.logicalClock) { + return; + } + if ( + existing.operation !== push.operation || + existing.payload_hash !== push.payloadHash || + existing.device_id !== deviceId + ) { + throw new SyncPushConflictError("logical_clock_conflict"); + } +} + +function assertSavedObjectMatchesPush( + push: SyncPushRequest, + deviceId: string, + object: SyncPushedObjectDocument, +): void { + if (object.logical_clock > push.logicalClock) { + throw new SyncPushConflictError("logical_clock_stale"); + } + if ( + object.object_id !== push.objectId || + object.object_type !== push.objectType || + object.operation !== push.operation || + object.payload_hash !== push.payloadHash || + object.schema_rev !== push.schemaRev || + object.logical_clock !== push.logicalClock || + object.device_id !== deviceId + ) { + throw new SyncPushPersistenceError("sync_object_mismatch"); + } +} diff --git a/cloudflare/src/sync_push_schema.ts b/cloudflare/src/sync_push_schema.ts new file mode 100644 index 0000000..4413091 --- /dev/null +++ b/cloudflare/src/sync_push_schema.ts @@ -0,0 +1,348 @@ +import type { AuthContext } from "./auth.js"; +import { + StorageObjectError, + assertSyncObjectType, + syncPayloadKey, +} from "./storage.js"; + +const MAX_INLINE_PAYLOAD_BYTES = 64 * 1024; +const MAX_R2_PAYLOAD_BYTES = 10 * 1024 * 1024; +const SYNC_OBJECT_ID_PATTERN = /^[a-zA-Z0-9._:-]{1,128}$/; +const SYNC_OBJECT_TYPE_PATTERN = /^[a-z0-9][a-z0-9._:-]{0,127}$/; +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}$/; + +export interface SyncPushDocument { + version: 1; + user_id: string; + device_id: string; + object: SyncPushedObjectDocument; +} + +export interface SyncPushedObjectDocument { + object_id: string; + object_type: string; + operation: "upsert" | "delete"; + payload_hash: string; + schema_rev: number; + logical_clock: number; + device_id: string; + created_at: number; + updated_at: number; + deleted_at: number | null; + payload_storage: "inline" | "r2" | "tombstone"; + payload_r2_key: string | null; +} + +export interface SyncPushRequest { + objectId: string; + objectType: string; + operation: "upsert" | "delete"; + payloadHash: string; + schemaRev: number; + logicalClock: number; + payload: SyncPushPayload; +} + +export type SyncPushPayload = + | { kind: "inline"; bytes: ArrayBuffer; r2Key: null } + | { kind: "r2"; bytes: ArrayBuffer; region: string; r2Key: string } + | { kind: "tombstone"; bytes: null; r2Key: null }; + +export interface SyncObjectRow { + object_id: unknown; + object_type: unknown; + payload_r2_key: unknown; + payload_hash: unknown; + schema_rev: unknown; + logical_clock: unknown; + device_id: unknown; + created_at: unknown; + updated_at: unknown; + deleted_at: unknown; +} + +type RequestBody = Record; + +export class SyncPushRequestError extends Error { + constructor(message: string) { + super(message); + this.name = "SyncPushRequestError"; + } +} + +export class SyncPushConflictError extends Error { + constructor(message: string) { + super(message); + this.name = "SyncPushConflictError"; + } +} + +export class SyncPushPersistenceError extends Error { + constructor(message: string) { + super(message); + this.name = "SyncPushPersistenceError"; + } +} + +export function currentDeviceId(context: AuthContext): string { + if (context.deviceId === undefined) { + throw new SyncPushRequestError("device_context_required"); + } + return context.deviceId; +} + +export async function syncPushRequest( + request: Request, + userId: string, +): Promise { + const body = await requestBody(request); + assertOnlyFields(body, [ + "version", + "object_id", + "object_type", + "operation", + "payload_hash", + "schema_rev", + "logical_clock", + "payload", + ]); + if (body.version !== 1) { + throw new SyncPushRequestError("version_invalid"); + } + + const objectId = syncObjectId(body.object_id); + const operation = syncOperation(body.operation); + const objectType = syncObjectType(body.object_type); + const payloadHash = sha256HexValue(body.payload_hash, "payload_hash"); + const payload = + operation === "delete" + ? tombstonePayload(body.payload) + : await upsertPayload(body.payload, userId, objectType, objectId, payloadHash); + + return { + objectId, + objectType, + operation, + 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), + payload, + }; +} + +export function syncObjectDocument(row: SyncObjectRow): SyncPushedObjectDocument { + const deletedAt = nullableInteger(row.deleted_at, "deleted_at", 0, Number.MAX_SAFE_INTEGER); + const payloadR2Key = nullableText(row.payload_r2_key, "payload_r2_key"); + return { + object_id: syncObjectId(row.object_id), + object_type: syncObjectType(row.object_type), + operation: deletedAt === null ? "upsert" : "delete", + 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: syncObjectId(row.device_id), + created_at: integer(row.created_at, "created_at", 0, Number.MAX_SAFE_INTEGER), + updated_at: integer(row.updated_at, "updated_at", 0, Number.MAX_SAFE_INTEGER), + deleted_at: deletedAt, + payload_storage: deletedAt !== null ? "tombstone" : payloadR2Key === null ? "inline" : "r2", + payload_r2_key: payloadR2Key, + }; +} + +async function upsertPayload( + value: unknown, + userId: string, + objectType: string, + objectId: string, + payloadHash: string, +): Promise { + const payload = record(value, "payload"); + const kind = text(payload.kind, "payload.kind"); + if (kind === "inline") { + assertOnlyFields(payload, ["kind", "data_base64"]); + const bytes = payloadBytes(payload.data_base64, "payload.data_base64", MAX_INLINE_PAYLOAD_BYTES); + await assertPayloadHash(bytes, payloadHash); + return { kind, bytes, r2Key: null }; + } + if (kind === "r2") { + assertOnlyFields(payload, ["kind", "region", "data_base64"]); + const region = regionValue(payload.region); + const bytes = payloadBytes(payload.data_base64, "payload.data_base64", MAX_R2_PAYLOAD_BYTES); + await assertPayloadHash(bytes, payloadHash); + return { + kind, + bytes, + region, + r2Key: await syncPayloadStorageKey(region, userId, objectType, objectId, payloadHash), + }; + } + throw new SyncPushRequestError("payload.kind_invalid"); +} + +async function syncPayloadStorageKey( + region: string, + userId: string, + objectType: string, + objectId: string, + payloadHash: string, +): Promise { + try { + return syncPayloadKey({ + region, + userHash: await sha256Hex(arrayBufferFromBytes(new TextEncoder().encode(userId))), + objectType, + objectId, + payloadHash, + }); + } catch (error) { + if (error instanceof StorageObjectError) { + throw new SyncPushRequestError(error.message); + } + throw error; + } +} + +function tombstonePayload(value: unknown): SyncPushPayload { + if (value !== undefined) { + throw new SyncPushRequestError("payload_forbidden"); + } + return { kind: "tombstone", bytes: null, r2Key: null }; +} + +async function requestBody(request: Request): Promise { + let value: unknown; + try { + value = await request.json(); + } catch { + throw new SyncPushRequestError("json_invalid"); + } + return record(value, "body"); +} + +function record(value: unknown, label: string): RequestBody { + if (typeof value !== "object" || value === null || Array.isArray(value)) { + throw new SyncPushRequestError(`${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 SyncPushRequestError(`unexpected_field:${field}`); + } + } +} + +function syncOperation(value: unknown): SyncPushRequest["operation"] { + if (value !== "upsert" && value !== "delete") { + throw new SyncPushRequestError("operation_invalid"); + } + return value; +} + +function syncObjectId(value: unknown): string { + if (typeof value !== "string" || !SYNC_OBJECT_ID_PATTERN.test(value)) { + throw new SyncPushRequestError("object_id_invalid"); + } + return value; +} + +function syncObjectType(value: unknown): string { + if (typeof value !== "string" || !SYNC_OBJECT_TYPE_PATTERN.test(value)) { + throw new SyncPushRequestError("object_type_invalid"); + } + try { + assertSyncObjectType(value); + } catch (error) { + if (error instanceof StorageObjectError) { + throw new SyncPushRequestError(error.message); + } + throw error; + } + return value; +} + +function sha256HexValue(value: unknown, label: string): string { + if (typeof value !== "string" || !SHA256_HEX.test(value)) { + throw new SyncPushRequestError(`${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 SyncPushRequestError(`${label}_invalid`); + } + return value; +} + +function nullableInteger(value: unknown, label: string, min: number, max: number): number | null { + if (value === null) { + return null; + } + return integer(value, label, min, max); +} + +function text(value: unknown, label: string): string { + if (typeof value !== "string" || value.length === 0) { + throw new SyncPushRequestError(`${label}_invalid`); + } + return value; +} + +function nullableText(value: unknown, label: string): string | null { + if (value === null) { + return null; + } + return text(value, label); +} + +function regionValue(value: unknown): string { + if (typeof value !== "string" || !REGION.test(value)) { + throw new SyncPushRequestError("payload.region_invalid"); + } + return value; +} + +function payloadBytes(value: unknown, label: string, maxBytes: number): ArrayBuffer { + const encoded = text(value, label); + if (!BASE64.test(encoded)) { + throw new SyncPushRequestError(`${label}_invalid`); + } + const bytes = bytesFromBase64(encoded); + if (bytes.byteLength === 0 || bytes.byteLength > maxBytes) { + throw new SyncPushRequestError(`${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 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 SyncPushRequestError("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 6822f45..29c3ad8 100644 --- a/cloudflare/tests/devices_test_support.ts +++ b/cloudflare/tests/devices_test_support.ts @@ -14,6 +14,7 @@ export interface TestEnvOptions { d1?: RecordedD1Database; kvEntries?: [string, string][]; kvReads?: string[]; + r2Puts?: RecordedR2Put[]; } export interface RecordedD1Database extends ElyD1Database { @@ -22,6 +23,12 @@ export interface RecordedD1Database extends ElyD1Database { queries: string[]; } +export interface RecordedR2Put { + key: string; + payload: ArrayBuffer; + options: ElyR2PutOptions; +} + interface TestD1DatabaseOptions { allRows?: unknown[]; firstRows?: unknown[]; @@ -38,7 +45,7 @@ export function testEnv(options: TestEnvOptions): Env { return Promise.resolve(values.get(key) ?? null); }, }, - ELY_STORAGE: testR2Bucket(), + ELY_STORAGE: testR2Bucket(options.r2Puts), ELY_RATE_LIMITER: { limit(): Promise<{ success: boolean }> { return Promise.resolve({ success: true }); @@ -112,12 +119,13 @@ function testD1PreparedStatement( }; } -function testR2Bucket(): Env["ELY_STORAGE"] { +function testR2Bucket(puts: RecordedR2Put[] = []): Env["ELY_STORAGE"] { return { get() { return Promise.resolve(null); }, - put(_key: string, value: ArrayBuffer, _options?: ElyR2PutOptions) { + put(key: string, value: ArrayBuffer, options: ElyR2PutOptions = {}) { + puts.push({ key, payload: value, options }); return Promise.resolve({ arrayBuffer() { return Promise.resolve(value); diff --git a/cloudflare/tests/sync_push_routes.test.ts b/cloudflare/tests/sync_push_routes.test.ts new file mode 100644 index 0000000..55b275c --- /dev/null +++ b/cloudflare/tests/sync_push_routes.test.ts @@ -0,0 +1,313 @@ +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 OBJECT_ID = "tab-01"; +const OBJECT_TYPE = "tabs"; + +describe("sync push routes", () => { + it("pushes an inline encrypted sync object from an approved current device", async () => { + const payload = bytes("encrypted tab payload"); + const payloadHash = sha256(payload); + const tokenHash = await authTokenHash(ACCESS_TOKEN); + const d1 = testD1Database({ + firstRows: [ + { device_id: DEVICE_ID }, + null, + syncObjectRow({ payload_hash: payloadHash }), + ], + }); + + const response = await handleRequest( + syncPushRequest(syncPushBody({ payload_hash: payloadHash, payload: inlinePayload(payload) })), + testEnv({ + d1, + 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, + object: syncObjectDocument({ payload_hash: payloadHash }), + }); + assert.equal(d1.batches[0], 2); + assert.ok(d1.queries[0]?.includes("approval_status = 'approved'")); + assert.ok(d1.queries[1]?.includes("FROM sync_objects")); + assert.ok(d1.queries[2]?.includes("INSERT INTO sync_objects")); + assert.ok(d1.queries[3]?.includes("INSERT INTO sync_change_log")); + assert.deepEqual(d1.binds[0], [USER_ID, DEVICE_ID]); + assert.deepEqual(d1.binds[1], [USER_ID, OBJECT_ID]); + assert.deepEqual(d1.binds[2]?.slice(0, 3), [USER_ID, OBJECT_ID, OBJECT_TYPE]); + assert.deepEqual(new Uint8Array(d1.binds[2]?.[3] as ArrayBuffer), new Uint8Array(payload)); + assert.equal(d1.binds[2]?.[4], null); + assert.equal(d1.binds[3]?.[3], "upsert"); + }); + + it("pushes an R2 encrypted sync object after checksum verification", async () => { + const payload = bytes("large encrypted tab payload"); + const payloadHash = sha256(payload); + const userHash = sha256(bytes(USER_ID)); + const r2Puts: RecordedR2Put[] = []; + const tokenHash = await authTokenHash(ACCESS_TOKEN); + const d1 = testD1Database({ + firstRows: [ + { device_id: DEVICE_ID }, + null, + syncObjectRow({ + payload_hash: payloadHash, + payload_r2_key: `sync-payloads/us-east/${userHash}/tabs/${OBJECT_ID}/${payloadHash}.bin`, + }), + ], + }); + + const response = await handleRequest( + syncPushRequest( + syncPushBody({ payload_hash: payloadHash, payload: r2Payload("us-east", payload) }), + ), + testEnv({ + d1, + r2Puts, + kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument(DEVICE_ID)]], + }), + ); + + assert.equal(response.status, 201); + assert.equal(r2Puts.length, 1); + assert.equal( + r2Puts[0]?.key, + `sync-payloads/us-east/${userHash}/tabs/${OBJECT_ID}/${payloadHash}.bin`, + ); + assert.equal(r2Puts[0]?.options.customMetadata?.sha256, payloadHash); + const body = (await response.json()) as { object: { payload_storage: string } }; + assert.equal(body.object.payload_storage, "r2"); + assert.equal(d1.binds[2]?.[3], null); + assert.equal(d1.binds[2]?.[4], r2Puts[0]?.key); + }); + + it("pushes a delete tombstone and writes the change log", async () => { + const payloadHash = "b".repeat(64); + const tokenHash = await authTokenHash(ACCESS_TOKEN); + const d1 = testD1Database({ + firstRows: [ + { device_id: DEVICE_ID }, + syncObjectRow({ payload_hash: "a".repeat(64), logical_clock: 41 }), + syncObjectRow({ payload_hash: payloadHash, logical_clock: 42, deleted_at: 1_780_000_800 }), + ], + }); + + const response = await handleRequest( + syncPushRequest( + syncPushBody({ operation: "delete", payload_hash: payloadHash, payload: undefined }), + ), + testEnv({ + d1, + kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument(DEVICE_ID)]], + }), + ); + + assert.equal(response.status, 201); + assert.equal(d1.batches[0], 3); + assert.ok(d1.queries[4]?.includes("INSERT INTO sync_tombstones")); + assert.equal(d1.binds[2]?.[3], null); + assert.equal(d1.binds[2]?.[4], null); + assert.equal(d1.binds[3]?.[3], "delete"); + const body = (await response.json()) as { object: { payload_storage: string } }; + assert.equal(body.object.payload_storage, "tombstone"); + }); + + it("rejects payload checksum mismatches before D1 writes", async () => { + const tokenHash = await authTokenHash(ACCESS_TOKEN); + const d1 = testD1Database({ firstRows: [{ device_id: DEVICE_ID }] }); + const response = await handleRequest( + syncPushRequest( + syncPushBody({ + payload_hash: "c".repeat(64), + payload: inlinePayload(bytes("encrypted tab payload")), + }), + ), + testEnv({ + d1, + kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument(DEVICE_ID)]], + }), + ); + + assert.equal(response.status, 400); + assert.deepEqual(await response.json(), { error: "invalid_sync_push" }); + assert.equal(d1.queries.length, 1); + assert.deepEqual(d1.batches, []); + }); + + it("rejects R2 object ids that cannot form storage keys before D1 writes", async () => { + const payload = bytes("large encrypted tab payload"); + const payloadHash = sha256(payload); + const r2Puts: RecordedR2Put[] = []; + const tokenHash = await authTokenHash(ACCESS_TOKEN); + const d1 = testD1Database({ firstRows: [{ device_id: DEVICE_ID }] }); + + const response = await handleRequest( + syncPushRequest( + syncPushBody({ + object_id: "Tab:01", + payload_hash: payloadHash, + payload: r2Payload("us-east", payload), + }), + ), + testEnv({ + d1, + r2Puts, + kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument(DEVICE_ID)]], + }), + ); + + assert.equal(response.status, 400); + assert.deepEqual(await response.json(), { error: "invalid_sync_push" }); + assert.equal(r2Puts.length, 0); + assert.equal(d1.queries.length, 1); + assert.deepEqual(d1.batches, []); + }); + + it("rejects stale logical clocks before persistence writes", async () => { + const payload = bytes("encrypted tab payload"); + const payloadHash = sha256(payload); + const tokenHash = await authTokenHash(ACCESS_TOKEN); + const d1 = testD1Database({ + firstRows: [ + { device_id: DEVICE_ID }, + syncObjectRow({ payload_hash: "d".repeat(64), logical_clock: 43 }), + ], + }); + + const response = await handleRequest( + syncPushRequest( + syncPushBody({ + payload_hash: payloadHash, + logical_clock: 42, + payload: inlinePayload(payload), + }), + ), + testEnv({ + d1, + kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument(DEVICE_ID)]], + }), + ); + + assert.equal(response.status, 409); + assert.deepEqual(await response.json(), { error: "sync_conflict" }); + assert.deepEqual(d1.batches, []); + }); + + it("rejects revoked devices before reading the sync push body", async () => { + const tokenHash = await authTokenHash(ACCESS_TOKEN); + const d1 = testD1Database({ firstRows: [null] }); + const response = await handleRequest( + syncPushRequest(syncPushBody()), + 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, []); + }); +}); + +function syncPushRequest(body: Record): Request { + return new Request("https://elydora.test/api/sync/push", { + method: "POST", + headers: { + authorization: `Bearer ${ACCESS_TOKEN}`, + "content-type": "application/json", + }, + body: JSON.stringify(body), + }); +} + +function syncPushBody(overrides: Record = {}): Record { + const payload = bytes("encrypted tab payload"); + const payloadHash = sha256(payload); + return { + version: 1, + object_id: OBJECT_ID, + object_type: OBJECT_TYPE, + operation: "upsert", + payload_hash: payloadHash, + schema_rev: 1, + logical_clock: 42, + payload: inlinePayload(payload), + ...overrides, + }; +} + +function inlinePayload(payload: ArrayBuffer): Record { + return { kind: "inline", data_base64: base64(payload) }; +} + +function r2Payload(region: string, payload: ArrayBuffer): Record { + return { kind: "r2", region, data_base64: base64(payload) }; +} + +function syncObjectRow(overrides: Record = {}): Record { + return { + object_id: OBJECT_ID, + object_type: OBJECT_TYPE, + payload_r2_key: null, + payload_hash: "a".repeat(64), + schema_rev: 1, + logical_clock: 42, + device_id: DEVICE_ID, + created_at: 1_780_000_700, + updated_at: 1_780_000_700, + deleted_at: null, + ...overrides, + }; +} + +function syncObjectDocument(overrides: Record = {}): Record { + const row = syncObjectRow(overrides); + return { + object_id: row.object_id, + object_type: row.object_type, + operation: row.deleted_at === null ? "upsert" : "delete", + payload_hash: row.payload_hash, + schema_rev: row.schema_rev, + logical_clock: row.logical_clock, + device_id: row.device_id, + created_at: row.created_at, + updated_at: row.updated_at, + deleted_at: row.deleted_at, + payload_storage: + row.deleted_at !== null ? "tombstone" : row.payload_r2_key === null ? "inline" : "r2", + payload_r2_key: row.payload_r2_key, + }; +} + +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"); +}