Expose sync status API

This commit is contained in:
2026-05-08 19:02:21 -04:00
parent 5752eefea9
commit e9f8b1eb43
5 changed files with 653 additions and 135 deletions
+5 -134
View File
@@ -1,9 +1,5 @@
import type { Env } from "./bindings.js";
import {
withApprovedDeviceApiControls,
withAuthenticatedApiControls,
withPublicApiControls,
} from "./api_controls.js";
import { withAuthenticatedApiControls, withPublicApiControls } from "./api_controls.js";
import {
DevicePermissionError,
DevicePersistenceError,
@@ -35,21 +31,7 @@ import {
parsePublicSigningKeysDocument,
publicSigningKeysKvKey,
} from "./signing_keys.js";
import {
SyncPushConflictError,
SyncPushPersistenceError,
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 { handleSyncRoute } from "./sync_routes.js";
import { jsonResponse } from "./responses.js";
export default {
@@ -188,120 +170,9 @@ export async function handleRequest(request: Request, env: Env): Promise<Respons
},
);
}
if (url.pathname === "/api/sync/pull") {
return withApprovedDeviceApiControls(
request,
env,
"sync.pull",
["GET"],
async (context) => {
try {
return jsonResponse(await syncPullDocument(url, env, context), 200, {
"Cache-Control": "no-store",
});
} catch (error) {
if (error instanceof SyncRequestError) {
return jsonResponse(
{ error: "invalid_sync_pull" },
400,
{ "Cache-Control": "no-store" },
);
}
if (error instanceof SyncSchemaError) {
return jsonResponse(
{ error: "sync_pull_invalid" },
500,
{ "Cache-Control": "no-store" },
);
}
throw error;
}
},
);
}
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/sync/snapshot") {
return withApprovedDeviceApiControls(
request,
env,
"sync.snapshot",
["GET", "POST"],
async (context) => {
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;
}
},
);
const syncResponse = await handleSyncRoute(request, env, url);
if (syncResponse !== null) {
return syncResponse;
}
if (url.pathname === "/api/plugins/signing-keys") {
return withPublicApiControls(request, env, "plugins.signing_keys", ["GET"], () =>
+182
View File
@@ -0,0 +1,182 @@
import type { Env } from "./bindings.js";
import { withApprovedDeviceApiControls } from "./api_controls.js";
import { jsonResponse } from "./responses.js";
import { SyncRequestError, SyncSchemaError, syncPullDocument } from "./sync_pull.js";
import {
SyncPushConflictError,
SyncPushPersistenceError,
SyncPushRequestError,
syncPushDocument,
} from "./sync_push.js";
import {
SyncSnapshotConflictError,
SyncSnapshotNotFoundError,
SyncSnapshotPersistenceError,
SyncSnapshotRequestError,
syncSnapshotDownloadDocument,
syncSnapshotUploadDocument,
} from "./sync_snapshot.js";
import { SyncStatusSchemaError, syncStatusDocument } from "./sync_status.js";
export async function handleSyncRoute(
request: Request,
env: Env,
url: URL,
): Promise<Response | null> {
if (url.pathname === "/api/sync/pull") {
return handleSyncPull(request, env, url);
}
if (url.pathname === "/api/sync/push") {
return handleSyncPush(request, env);
}
if (url.pathname === "/api/sync/snapshot") {
return handleSyncSnapshot(request, env, url);
}
if (url.pathname === "/api/sync/status") {
return handleSyncStatus(request, env);
}
return null;
}
function handleSyncPull(request: Request, env: Env, url: URL): Promise<Response> {
return withApprovedDeviceApiControls(
request,
env,
"sync.pull",
["GET"],
async (context) => {
try {
return jsonResponse(await syncPullDocument(url, env, context), 200, {
"Cache-Control": "no-store",
});
} catch (error) {
if (error instanceof SyncRequestError) {
return jsonResponse(
{ error: "invalid_sync_pull" },
400,
{ "Cache-Control": "no-store" },
);
}
if (error instanceof SyncSchemaError) {
return jsonResponse(
{ error: "sync_pull_invalid" },
500,
{ "Cache-Control": "no-store" },
);
}
throw error;
}
},
);
}
function handleSyncPush(request: Request, env: Env): Promise<Response> {
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;
}
},
);
}
function handleSyncSnapshot(request: Request, env: Env, url: URL): Promise<Response> {
return withApprovedDeviceApiControls(
request,
env,
"sync.snapshot",
["GET", "POST"],
async (context) => {
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;
}
},
);
}
function handleSyncStatus(request: Request, env: Env): Promise<Response> {
return withApprovedDeviceApiControls(
request,
env,
"sync.status",
["GET"],
async (context) => {
try {
return jsonResponse(await syncStatusDocument(env, context), 200, {
"Cache-Control": "no-store",
});
} catch (error) {
if (error instanceof SyncStatusSchemaError) {
return jsonResponse(
{ error: "sync_status_invalid" },
500,
{ "Cache-Control": "no-store" },
);
}
throw error;
}
},
);
}
+270
View File
@@ -0,0 +1,270 @@
import type { AuthContext } from "./auth.js";
import type { Env } from "./bindings.js";
import { StorageObjectError, assertSyncObjectType } from "./storage.js";
const CHANGE_CURSOR_QUERY = `
SELECT
COALESCE(MAX(change_id), 0) AS latest_change_id,
COUNT(*) AS total_changes
FROM sync_change_log
WHERE user_id = ?
`;
const OBJECT_STATUS_QUERY = `
SELECT
object_type,
SUM(CASE WHEN deleted_at IS NULL THEN 1 ELSE 0 END) AS active_count,
SUM(CASE WHEN deleted_at IS NOT NULL THEN 1 ELSE 0 END) AS deleted_count,
COALESCE(MAX(logical_clock), 0) AS latest_logical_clock,
COALESCE(MAX(updated_at), 0) AS latest_updated_at
FROM sync_objects
WHERE user_id = ?
GROUP BY object_type
ORDER BY object_type ASC
`;
const SNAPSHOT_COUNT_QUERY = `
SELECT COUNT(*) AS total_snapshots
FROM sync_snapshots
WHERE user_id = ?
`;
const LATEST_SNAPSHOT_QUERY = `
SELECT
snapshot_id,
payload_hash,
logical_clock,
device_id,
size_bytes,
created_at
FROM sync_snapshots
WHERE user_id = ?
ORDER BY created_at DESC, snapshot_id ASC
LIMIT 1
`;
const APPROVED_DEVICE_COUNT_QUERY = `
SELECT COUNT(*) AS approved_devices
FROM user_devices
WHERE user_id = ? AND approval_status = 'approved' AND revoked_at IS NULL
`;
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}$/;
export interface SyncStatusDocument {
version: 1;
user_id: string;
device_id: string;
cursor: SyncCursorStatusDocument;
objects: SyncObjectStatusDocument[];
snapshots: SyncSnapshotStatusDocument;
devices: SyncDeviceStatusDocument;
}
export interface SyncCursorStatusDocument {
latest_change_id: number;
total_changes: number;
}
export interface SyncObjectStatusDocument {
object_type: string;
active_count: number;
deleted_count: number;
latest_logical_clock: number;
latest_updated_at: number;
}
export interface SyncSnapshotStatusDocument {
total_snapshots: number;
latest: SyncLatestSnapshotDocument | null;
}
export interface SyncLatestSnapshotDocument {
snapshot_id: string;
payload_hash: string;
logical_clock: number;
device_id: string;
size_bytes: number;
created_at: number;
}
export interface SyncDeviceStatusDocument {
approved_count: number;
current_device_id: string;
current_device_approved: true;
}
interface ChangeCursorRow {
latest_change_id: unknown;
total_changes: unknown;
}
interface ObjectStatusRow {
object_type: unknown;
active_count: unknown;
deleted_count: unknown;
latest_logical_clock: unknown;
latest_updated_at: unknown;
}
interface SnapshotCountRow {
total_snapshots: unknown;
}
interface LatestSnapshotRow {
snapshot_id: unknown;
payload_hash: unknown;
logical_clock: unknown;
device_id: unknown;
size_bytes: unknown;
created_at: unknown;
}
interface DeviceStatusRow {
approved_devices: unknown;
}
export class SyncStatusSchemaError extends Error {
constructor(message: string) {
super(message);
this.name = "SyncStatusSchemaError";
}
}
export async function syncStatusDocument(
env: Env,
context: AuthContext,
): Promise<SyncStatusDocument> {
const deviceId = currentDeviceId(context);
const cursorRow = await env.ELY_DB.prepare(CHANGE_CURSOR_QUERY)
.bind(context.userId)
.first<ChangeCursorRow>();
const objectRows = await env.ELY_DB.prepare(OBJECT_STATUS_QUERY)
.bind(context.userId)
.all<ObjectStatusRow>();
const snapshotCountRow = await env.ELY_DB.prepare(SNAPSHOT_COUNT_QUERY)
.bind(context.userId)
.first<SnapshotCountRow>();
const latestSnapshotRow = await env.ELY_DB.prepare(LATEST_SNAPSHOT_QUERY)
.bind(context.userId)
.first<LatestSnapshotRow>();
const deviceStatusRow = await env.ELY_DB.prepare(APPROVED_DEVICE_COUNT_QUERY)
.bind(context.userId)
.first<DeviceStatusRow>();
return {
version: 1,
user_id: context.userId,
device_id: deviceId,
cursor: cursorStatus(cursorRow),
objects: objectRows.results.map(objectStatus),
snapshots: snapshotStatus(snapshotCountRow, latestSnapshotRow),
devices: deviceStatus(deviceStatusRow, deviceId),
};
}
function cursorStatus(row: ChangeCursorRow | null): SyncCursorStatusDocument {
if (row === null) {
throw new SyncStatusSchemaError("sync_cursor_status_missing");
}
return {
latest_change_id: integer(row.latest_change_id, "latest_change_id"),
total_changes: integer(row.total_changes, "total_changes"),
};
}
function objectStatus(row: ObjectStatusRow): SyncObjectStatusDocument {
return {
object_type: objectType(row.object_type),
active_count: integer(row.active_count, "active_count"),
deleted_count: integer(row.deleted_count, "deleted_count"),
latest_logical_clock: integer(row.latest_logical_clock, "latest_logical_clock"),
latest_updated_at: integer(row.latest_updated_at, "latest_updated_at"),
};
}
function snapshotStatus(
countRow: SnapshotCountRow | null,
latestRow: LatestSnapshotRow | null,
): SyncSnapshotStatusDocument {
if (countRow === null) {
throw new SyncStatusSchemaError("sync_snapshot_status_missing");
}
return {
total_snapshots: integer(countRow.total_snapshots, "total_snapshots"),
latest: latestRow === null ? null : latestSnapshot(latestRow),
};
}
function latestSnapshot(row: LatestSnapshotRow): SyncLatestSnapshotDocument {
return {
snapshot_id: snapshotId(row.snapshot_id),
payload_hash: payloadHash(row.payload_hash),
logical_clock: integer(row.logical_clock, "logical_clock"),
device_id: deviceId(row.device_id),
size_bytes: integer(row.size_bytes, "size_bytes"),
created_at: integer(row.created_at, "created_at"),
};
}
function deviceStatus(
row: DeviceStatusRow | null,
currentDeviceId: string,
): SyncDeviceStatusDocument {
if (row === null) {
throw new SyncStatusSchemaError("sync_device_status_missing");
}
return {
approved_count: integer(row.approved_devices, "approved_devices"),
current_device_id: currentDeviceId,
current_device_approved: true,
};
}
function currentDeviceId(context: AuthContext): string {
if (context.deviceId === undefined) {
throw new SyncStatusSchemaError("device_context_required");
}
return context.deviceId;
}
function objectType(value: unknown): string {
if (typeof value !== "string") {
throw new SyncStatusSchemaError("object_type_invalid");
}
try {
assertSyncObjectType(value);
} catch (error) {
if (error instanceof StorageObjectError) {
throw new SyncStatusSchemaError(error.message);
}
throw error;
}
return value;
}
function snapshotId(value: unknown): string {
if (typeof value !== "string" || !SNAPSHOT_ID_PATTERN.test(value)) {
throw new SyncStatusSchemaError("snapshot_id_invalid");
}
return value;
}
function deviceId(value: unknown): string {
if (typeof value !== "string" || !DEVICE_ID_PATTERN.test(value)) {
throw new SyncStatusSchemaError("device_id_invalid");
}
return value;
}
function payloadHash(value: unknown): string {
if (typeof value !== "string" || !SHA256_HEX.test(value)) {
throw new SyncStatusSchemaError("payload_hash_invalid");
}
return value;
}
function integer(value: unknown, label: string): number {
if (typeof value !== "number" || !Number.isSafeInteger(value) || value < 0) {
throw new SyncStatusSchemaError(`${label}_invalid`);
}
return value;
}
+1 -1
View File
@@ -74,7 +74,7 @@ describe("worker routes", () => {
it("returns JSON not found for unknown routes", async () => {
const response = await handleRequest(
new Request("https://elydora.test/api/sync/status"),
new Request("https://elydora.test/api/unknown"),
testEnv(null),
);
+195
View File
@@ -0,0 +1,195 @@
import assert from "node:assert/strict";
import { describe, it } from "node:test";
import { authSessionCacheKvKey, authTokenHash } from "../src/auth.js";
import { handleRequest } from "../src/index.js";
import { ACCESS_TOKEN, sessionDocument, testD1Database, testEnv } from "./devices_test_support.js";
const USER_ID = "user-01";
const DEVICE_ID = "device-01";
describe("sync status routes", () => {
it("returns cloud sync cursor, object, snapshot, and device status", async () => {
const tokenHash = await authTokenHash(ACCESS_TOKEN);
const d1 = testD1Database({
firstRows: [
{ device_id: DEVICE_ID },
{ latest_change_id: 51, total_changes: 7 },
{ total_snapshots: 2 },
latestSnapshotRow(),
{ approved_devices: 3 },
],
allRows: [
objectStatusRow({ object_type: "bookmarks", active_count: 4, deleted_count: 1 }),
objectStatusRow({ object_type: "tabs", active_count: 9, latest_logical_clock: 44 }),
],
});
const response = await handleRequest(
syncStatusRequest(),
testEnv({
d1,
kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument(DEVICE_ID)]],
}),
);
assert.equal(response.status, 200);
assert.equal(response.headers.get("cache-control"), "no-store");
assert.deepEqual(await response.json(), {
version: 1,
user_id: USER_ID,
device_id: DEVICE_ID,
cursor: { latest_change_id: 51, total_changes: 7 },
objects: [
objectStatusDocument({ object_type: "bookmarks", active_count: 4, deleted_count: 1 }),
objectStatusDocument({ object_type: "tabs", active_count: 9, latest_logical_clock: 44 }),
],
snapshots: {
total_snapshots: 2,
latest: latestSnapshotRow(),
},
devices: {
approved_count: 3,
current_device_id: DEVICE_ID,
current_device_approved: true,
},
});
assert.ok(d1.queries[0]?.includes("approval_status = 'approved'"));
assert.ok(d1.queries[1]?.includes("FROM sync_change_log"));
assert.ok(d1.queries[2]?.includes("FROM sync_objects"));
assert.ok(d1.queries[3]?.includes("FROM sync_snapshots"));
assert.ok(d1.queries[4]?.includes("FROM sync_snapshots"));
assert.ok(d1.queries[5]?.includes("FROM user_devices"));
assert.deepEqual(d1.binds, [
[USER_ID, DEVICE_ID],
[USER_ID],
[USER_ID],
[USER_ID],
[USER_ID],
[USER_ID],
]);
});
it("returns empty status when the account has no sync facts", async () => {
const tokenHash = await authTokenHash(ACCESS_TOKEN);
const d1 = testD1Database({
firstRows: [
{ device_id: DEVICE_ID },
{ latest_change_id: 0, total_changes: 0 },
{ total_snapshots: 0 },
null,
{ approved_devices: 1 },
],
});
const response = await handleRequest(
syncStatusRequest(),
testEnv({
d1,
kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument(DEVICE_ID)]],
}),
);
assert.equal(response.status, 200);
const body = (await response.json()) as {
cursor: { latest_change_id: number; total_changes: number };
objects: [];
snapshots: { total_snapshots: number; latest: null };
};
assert.deepEqual(body.cursor, { latest_change_id: 0, total_changes: 0 });
assert.deepEqual(body.objects, []);
assert.deepEqual(body.snapshots, { total_snapshots: 0, latest: null });
});
it("rejects revoked devices before reading sync status", async () => {
const tokenHash = await authTokenHash(ACCESS_TOKEN);
const d1 = testD1Database({
firstRows: [null, { latest_change_id: 51, total_changes: 7 }],
allRows: [objectStatusRow()],
});
const response = await handleRequest(
syncStatusRequest(),
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);
});
it("rejects unsupported methods before session reads", async () => {
const response = await handleRequest(
new Request("https://elydora.test/api/sync/status", {
method: "POST",
headers: { authorization: `Bearer ${ACCESS_TOKEN}` },
}),
testEnv({ d1: testD1Database([]) }),
);
assert.equal(response.status, 405);
assert.equal(response.headers.get("allow"), "GET");
assert.deepEqual(await response.json(), { error: "method_not_allowed" });
});
it("returns a server error for malformed status rows", async () => {
const tokenHash = await authTokenHash(ACCESS_TOKEN);
const d1 = testD1Database({
firstRows: [
{ device_id: DEVICE_ID },
{ latest_change_id: 51, total_changes: 7 },
{ total_snapshots: 1 },
latestSnapshotRow(),
{ approved_devices: 1 },
],
allRows: [objectStatusRow({ object_type: "passwords" })],
});
const response = await handleRequest(
syncStatusRequest(),
testEnv({
d1,
kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument(DEVICE_ID)]],
}),
);
assert.equal(response.status, 500);
assert.deepEqual(await response.json(), { error: "sync_status_invalid" });
});
});
function syncStatusRequest(): Request {
return new Request("https://elydora.test/api/sync/status", {
headers: { authorization: `Bearer ${ACCESS_TOKEN}` },
});
}
function objectStatusRow(overrides: Record<string, unknown> = {}): Record<string, unknown> {
return {
object_type: "tabs",
active_count: 3,
deleted_count: 0,
latest_logical_clock: 42,
latest_updated_at: 1_780_000_700,
...overrides,
};
}
function objectStatusDocument(overrides: Record<string, unknown> = {}): Record<string, unknown> {
return objectStatusRow(overrides);
}
function latestSnapshotRow(overrides: Record<string, unknown> = {}): Record<string, unknown> {
return {
snapshot_id: "snapshot-01",
payload_hash: "a".repeat(64),
logical_clock: 42,
device_id: DEVICE_ID,
size_bytes: 26,
created_at: 1_780_000_900,
...overrides,
};
}