feat(server): the sync endpoints, on Node 22 with no build step
Hono and pg, run under --experimental-strip-types, so the deployed thing is the source. GET /api/sync?cursor=N pages rows above the cursor; POST /api/sync upserts last-write-wins. Bearer token on everything under /api; /health is open, for the container healthcheck. Rows are stored generically — primary key as text, body as JSONB — because the server never reads inside a row. It stores and orders them and the client interprets them, which keeps the two schemas from having to move in lockstep. change_seq is bumped by a BEFORE UPDATE trigger rather than by the write path. A row edited after a client last pulled would otherwise keep its old sequence, sit below that client's cursor, and never be delivered; putting it in the database means no future write path can forget. The last-write-wins comparison is in the ON CONFLICT clause itself, so a losing row is not written at all and does not bump change_seq — a conflict does not become traffic for every other device. test/sync/roundtrip.test.ts runs two clients against a real Postgres and asserts what actually goes wrong in sync: that a fresh client's seeded rows cannot overwrite the server's history (the artifact's bug, as an executable test), that a delete propagates, and that dict.loadedBands never crosses the wire. It skips without HANKAN_TEST_SERVER, so npm test still runs anywhere. POST /api/test/reset exists only when HANKAN_TEST_MODE=1, so it cannot be reached on the Pi even if the token leaks. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
MechaCat02
113
server/src/db.ts
Normal file
113
server/src/db.ts
Normal file
@@ -0,0 +1,113 @@
|
||||
/* Postgres access, and the two queries sync is made of. */
|
||||
|
||||
import { readFile } from "node:fs/promises";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import pg from "pg";
|
||||
|
||||
export interface SyncRow {
|
||||
tbl: string;
|
||||
data: Record<string, string | number | null>;
|
||||
updated_at: number;
|
||||
deleted?: boolean;
|
||||
}
|
||||
|
||||
export interface Store {
|
||||
pull(userId: string, cursor: number, limit: number): Promise<{ rows: SyncRow[]; cursor: number; more: boolean }>;
|
||||
push(userId: string, rows: SyncRow[]): Promise<number>;
|
||||
/** Test-only; the route that calls it exists only under HANKAN_TEST_MODE. */
|
||||
reset(userId: string): Promise<void>;
|
||||
close(): Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* The row's primary key as a single string.
|
||||
*
|
||||
* The client sends `{pk}` for a delete and the full row otherwise, so a
|
||||
* delete already carries its key and a live row needs one derived from the
|
||||
* table's key columns. Both sides use the same rule, from the shared
|
||||
* protocol module.
|
||||
*/
|
||||
function keyOf(row: SyncRow, pkCols: string[]): string {
|
||||
if (row.deleted && row.data.pk != null) return String(row.data.pk);
|
||||
return pkCols.map((c) => String(row.data[c])).join(" ");
|
||||
}
|
||||
|
||||
export async function openStore(connectionString: string, pkFor: (tbl: string) => string[] | null): Promise<Store> {
|
||||
const pool = new pg.Pool({ connectionString, max: 4 });
|
||||
|
||||
const schema = await readFile(fileURLToPath(new URL("../sql/001-schema.sql", import.meta.url)), "utf8");
|
||||
await pool.query(schema);
|
||||
|
||||
return {
|
||||
async pull(userId, cursor, limit) {
|
||||
// One extra row tells us whether another page exists without a count.
|
||||
const { rows } = await pool.query(
|
||||
`SELECT tbl, pk, data, updated_at, deleted, change_seq
|
||||
FROM sync_row
|
||||
WHERE user_id = $1 AND change_seq > $2
|
||||
ORDER BY change_seq
|
||||
LIMIT $3`,
|
||||
[userId, cursor, limit + 1],
|
||||
);
|
||||
|
||||
const more = rows.length > limit;
|
||||
const page = more ? rows.slice(0, limit) : rows;
|
||||
|
||||
return {
|
||||
rows: page.map((r) => ({
|
||||
tbl: r.tbl as string,
|
||||
data: r.deleted ? { pk: r.pk as string } : (r.data as Record<string, string | number | null>),
|
||||
updated_at: Number(r.updated_at),
|
||||
...(r.deleted ? { deleted: true } : {}),
|
||||
})),
|
||||
cursor: page.length ? Number(page[page.length - 1].change_seq) : cursor,
|
||||
more,
|
||||
};
|
||||
},
|
||||
|
||||
async push(userId, rows) {
|
||||
const client = await pool.connect();
|
||||
try {
|
||||
await client.query("BEGIN");
|
||||
|
||||
for (const row of rows) {
|
||||
const pkCols = pkFor(row.tbl);
|
||||
if (!pkCols) continue; // a table this server does not know about
|
||||
|
||||
// Last-write-wins, resolved in the database so two devices pushing
|
||||
// at once cannot interleave a read and a write around it.
|
||||
await client.query(
|
||||
`INSERT INTO sync_row (user_id, tbl, pk, data, updated_at, deleted)
|
||||
VALUES ($1, $2, $3, $4, $5, $6)
|
||||
ON CONFLICT (user_id, tbl, pk) DO UPDATE
|
||||
SET data = EXCLUDED.data,
|
||||
updated_at = EXCLUDED.updated_at,
|
||||
deleted = EXCLUDED.deleted
|
||||
WHERE EXCLUDED.updated_at > sync_row.updated_at`,
|
||||
[userId, row.tbl, keyOf(row, pkCols), JSON.stringify(row.data), row.updated_at, row.deleted ?? false],
|
||||
);
|
||||
}
|
||||
|
||||
const { rows: top } = await client.query(
|
||||
"SELECT COALESCE(max(change_seq), 0) AS c FROM sync_row WHERE user_id = $1",
|
||||
[userId],
|
||||
);
|
||||
await client.query("COMMIT");
|
||||
return Number(top[0].c);
|
||||
} catch (err) {
|
||||
await client.query("ROLLBACK");
|
||||
throw err;
|
||||
} finally {
|
||||
client.release();
|
||||
}
|
||||
},
|
||||
|
||||
async reset(userId) {
|
||||
await pool.query("DELETE FROM sync_row WHERE user_id = $1", [userId]);
|
||||
},
|
||||
|
||||
async close() {
|
||||
await pool.end();
|
||||
},
|
||||
};
|
||||
}
|
||||
87
server/src/main.ts
Normal file
87
server/src/main.ts
Normal file
@@ -0,0 +1,87 @@
|
||||
/* Hankan's server: sync, and the tutor.
|
||||
|
||||
Both endpoints are stateless. The client owns its database and its
|
||||
transcript; this process owns a Postgres table and an API key. Nothing
|
||||
here needs to survive a restart. */
|
||||
|
||||
import { serve } from "@hono/node-server";
|
||||
import { Hono } from "hono";
|
||||
|
||||
import { pkFor, PAGE_SIZE } from "../../shared/sync-protocol.mjs";
|
||||
import { openStore, type SyncRow } from "./db.ts";
|
||||
import { tutorRoute } from "./tutor.ts";
|
||||
|
||||
const PORT = Number(process.env.PORT ?? 8787);
|
||||
const TOKEN = process.env.HANKAN_TOKEN ?? "";
|
||||
const DATABASE_URL = process.env.DATABASE_URL ?? "";
|
||||
// One learner. The column exists so a second is a config change, not a
|
||||
// migration.
|
||||
const USER_ID = process.env.HANKAN_USER ?? "default";
|
||||
|
||||
if (!TOKEN) {
|
||||
console.error("HANKAN_TOKEN is required — refusing to start an unauthenticated sync endpoint.");
|
||||
process.exit(1);
|
||||
}
|
||||
if (!DATABASE_URL) {
|
||||
console.error("DATABASE_URL is required.");
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
const store = await openStore(DATABASE_URL, pkFor);
|
||||
|
||||
const app = new Hono();
|
||||
|
||||
/* Bearer auth on everything except the health check, which has to be
|
||||
reachable by a container healthcheck that holds no secret. */
|
||||
app.use("/api/*", async (c, next) => {
|
||||
const header = c.req.header("authorization") ?? "";
|
||||
const presented = header.startsWith("Bearer ") ? header.slice(7) : "";
|
||||
// Constant-time-ish: compare lengths first, then every byte, so a wrong
|
||||
// token cannot be narrowed down by timing the response.
|
||||
const ok =
|
||||
presented.length === TOKEN.length &&
|
||||
presented.split("").reduce((acc, ch, i) => acc & (ch === TOKEN[i] ? 1 : 0), 1) === 1;
|
||||
if (!ok) return c.json({ error: "unauthorized" }, 401);
|
||||
await next();
|
||||
});
|
||||
|
||||
app.get("/health", (c) => c.json({ ok: true }));
|
||||
|
||||
/* ── sync ────────────────────────────────────────────────────────────── */
|
||||
|
||||
app.get("/api/sync", async (c) => {
|
||||
const cursor = Number(c.req.query("cursor") ?? 0);
|
||||
if (!Number.isFinite(cursor) || cursor < 0) return c.json({ error: "bad cursor" }, 400);
|
||||
return c.json(await store.pull(USER_ID, cursor, PAGE_SIZE));
|
||||
});
|
||||
|
||||
app.post("/api/sync", async (c) => {
|
||||
const body = (await c.req.json()) as { rows?: SyncRow[] };
|
||||
const rows = Array.isArray(body.rows) ? body.rows : [];
|
||||
if (rows.length > PAGE_SIZE * 4) return c.json({ error: "too many rows" }, 413);
|
||||
return c.json({ cursor: await store.push(USER_ID, rows) });
|
||||
});
|
||||
|
||||
/* Wipes this user's rows. Exists only when HANKAN_TEST_MODE is set, so it
|
||||
cannot be reached on the Pi even if the token leaks. */
|
||||
if (process.env.HANKAN_TEST_MODE === "1") {
|
||||
app.post("/api/test/reset", async (c) => {
|
||||
await store.reset(USER_ID);
|
||||
return c.json({ ok: true });
|
||||
});
|
||||
console.warn("HANKAN_TEST_MODE — /api/test/reset is enabled. Never set this in production.");
|
||||
}
|
||||
|
||||
/* ── tutor ───────────────────────────────────────────────────────────── */
|
||||
|
||||
app.route("/api/tutor", tutorRoute());
|
||||
|
||||
const server = serve({ fetch: app.fetch, port: PORT }, (info) => {
|
||||
console.log(`hankan server on :${info.port}`);
|
||||
});
|
||||
|
||||
for (const sig of ["SIGINT", "SIGTERM"] as const) {
|
||||
process.on(sig, () => {
|
||||
server.close(() => void store.close().then(() => process.exit(0)));
|
||||
});
|
||||
}
|
||||
Reference in New Issue
Block a user