Expose sync push API

2026-05-08 18:46:21 -04:00
parent 59b90c60ae
commit 2b3eea861a
6 changed files with 968 additions and 4 deletions
@@ -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<Respons
      },
    );
  }
+  if (url.pathname === "/api/sync/push") {
+    return withApprovedDeviceApiControls(
+      request,
+      env,
+      "sync.push",
+      ["POST"],
+      async (context) => {
+        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),
@@ -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");
  }
@@ -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<SyncPushDocument> {
+  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<SyncObjectRow>();
+  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<SyncObjectRow>();
+  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<void> {
+  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");
+  }
+}
@@ -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<string, unknown>;
+
+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<SyncPushRequest> {
+  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<SyncPushPayload> {
+  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<string> {
+  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<RequestBody> {
+  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<void> {
+  const actualHash = await sha256Hex(payload);
+  if (actualHash !== expectedHash) {
+    throw new SyncPushRequestError("payload_hash_mismatch");
+  }
+}
+
+async function sha256Hex(payload: ArrayBuffer): Promise<string> {
+  const digest = await crypto.subtle.digest("SHA-256", payload);
+  return [...new Uint8Array(digest)].map((byte) => byte.toString(16).padStart(2, "0")).join("");
+}