diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ca1b504..6b6bbd5 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -32,9 +32,10 @@ jobs: - name: Typecheck run: npm run typecheck - # The sync round-trip needs a real Postgres and a running server, so a - # fake would not exercise the change_seq trigger, the last-write-wins - # upsert, or cursor paging — which is where sync actually goes wrong. + # The sync round-trip needs a real Postgres and a running server: a fake + # would not exercise the compare-and-swap push, the advisory locks that + # keep a pull from skipping a change_seq, or cursor paging — which is + # where sync actually goes wrong. - name: Start Postgres run: | docker run -d --name hankan-pg \ diff --git a/app/src/db/migrations.ts b/app/src/db/migrations.ts index 2a83bd8..331c0d9 100644 --- a/app/src/db/migrations.ts +++ b/app/src/db/migrations.ts @@ -260,8 +260,125 @@ export const MIGRATIONS: Migration[] = [ `, run: remodelRoadAndChat, }, + { + id: 9, + name: "sync protocol 2 — base_seq and dirty on every row, counters per device", + sql: /* sql */ ` + -- base_seq: the server change_seq this row last agreed with. + -- dirty: a local edit not yet on the server. + -- rev: bumped by every local edit, so a push can tell whether + -- the row changed again while it was in flight. + -- All default to 0: a seeded row is never dirty and never travels. + ALTER TABLE card ADD COLUMN base_seq INTEGER NOT NULL DEFAULT 0; + ALTER TABLE card ADD COLUMN dirty INTEGER NOT NULL DEFAULT 0; + ALTER TABLE card ADD COLUMN rev INTEGER NOT NULL DEFAULT 0; + ALTER TABLE progress ADD COLUMN base_seq INTEGER NOT NULL DEFAULT 0; + ALTER TABLE progress ADD COLUMN dirty INTEGER NOT NULL DEFAULT 0; + ALTER TABLE progress ADD COLUMN rev INTEGER NOT NULL DEFAULT 0; + ALTER TABLE chat ADD COLUMN base_seq INTEGER NOT NULL DEFAULT 0; + ALTER TABLE chat ADD COLUMN dirty INTEGER NOT NULL DEFAULT 0; + ALTER TABLE chat ADD COLUMN rev INTEGER NOT NULL DEFAULT 0; + ALTER TABLE meta ADD COLUMN base_seq INTEGER NOT NULL DEFAULT 0; + ALTER TABLE meta ADD COLUMN dirty INTEGER NOT NULL DEFAULT 0; + ALTER TABLE meta ADD COLUMN rev INTEGER NOT NULL DEFAULT 0; + ALTER TABLE custom_word ADD COLUMN base_seq INTEGER NOT NULL DEFAULT 0; + ALTER TABLE custom_word ADD COLUMN dirty INTEGER NOT NULL DEFAULT 0; + ALTER TABLE custom_word ADD COLUMN rev INTEGER NOT NULL DEFAULT 0; + ALTER TABLE evidence ADD COLUMN base_seq INTEGER NOT NULL DEFAULT 0; + ALTER TABLE evidence ADD COLUMN dirty INTEGER NOT NULL DEFAULT 0; + ALTER TABLE evidence ADD COLUMN rev INTEGER NOT NULL DEFAULT 0; + ALTER TABLE confusion ADD COLUMN base_seq INTEGER NOT NULL DEFAULT 0; + ALTER TABLE confusion ADD COLUMN dirty INTEGER NOT NULL DEFAULT 0; + ALTER TABLE confusion ADD COLUMN rev INTEGER NOT NULL DEFAULT 0; + ALTER TABLE phase_ledger ADD COLUMN base_seq INTEGER NOT NULL DEFAULT 0; + ALTER TABLE phase_ledger ADD COLUMN dirty INTEGER NOT NULL DEFAULT 0; + ALTER TABLE phase_ledger ADD COLUMN rev INTEGER NOT NULL DEFAULT 0; + ALTER TABLE tombstone ADD COLUMN base_seq INTEGER NOT NULL DEFAULT 0; + ALTER TABLE tombstone ADD COLUMN dirty INTEGER NOT NULL DEFAULT 0; + ALTER TABLE tombstone ADD COLUMN rev INTEGER NOT NULL DEFAULT 0; + + -- Counters per device, summed on read — two devices adding to the same + -- day each own their row, so neither can erase the other's reviews. + CREATE TABLE study_log_v9 ( + day INTEGER NOT NULL, + device TEXT NOT NULL, + reviews INTEGER NOT NULL DEFAULT 0, + correct INTEGER NOT NULL DEFAULT 0, + drills INTEGER NOT NULL DEFAULT 0, + updated_at INTEGER NOT NULL DEFAULT 0, + base_seq INTEGER NOT NULL DEFAULT 0, + dirty INTEGER NOT NULL DEFAULT 0, + rev INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (day, device) + ) WITHOUT ROWID; + INSERT INTO study_log_v9 (day, device, reviews, correct, drills, updated_at) + SELECT day, (SELECT v FROM meta WHERE k = 'sync.device'), reviews, correct, drills, updated_at + FROM study_log; + DROP TABLE study_log; + ALTER TABLE study_log_v9 RENAME TO study_log; + + CREATE TABLE peek_v9 ( + form TEXT NOT NULL, + device TEXT NOT NULL, + count INTEGER NOT NULL DEFAULT 0, + updated_at INTEGER NOT NULL DEFAULT 0, + base_seq INTEGER NOT NULL DEFAULT 0, + dirty INTEGER NOT NULL DEFAULT 0, + rev INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (form, device) + ) WITHOUT ROWID; + INSERT INTO peek_v9 (form, device, count, updated_at) + SELECT form, (SELECT v FROM meta WHERE k = 'sync.device'), count, updated_at FROM peek; + DROP TABLE peek; + ALTER TABLE peek_v9 RENAME TO peek; + + CREATE INDEX IF NOT EXISTS card_dirty ON card(dirty) WHERE dirty = 1; + CREATE INDEX IF NOT EXISTS progress_dirty ON progress(dirty) WHERE dirty = 1; + CREATE INDEX IF NOT EXISTS chat_dirty ON chat(dirty) WHERE dirty = 1; + CREATE INDEX IF NOT EXISTS meta_dirty ON meta(dirty) WHERE dirty = 1; + CREATE INDEX IF NOT EXISTS custom_word_dirty ON custom_word(dirty) WHERE dirty = 1; + CREATE INDEX IF NOT EXISTS evidence_dirty ON evidence(dirty) WHERE dirty = 1; + CREATE INDEX IF NOT EXISTS confusion_dirty ON confusion(dirty) WHERE dirty = 1; + CREATE INDEX IF NOT EXISTS phase_ledger_dirty ON phase_ledger(dirty) WHERE dirty = 1; + CREATE INDEX IF NOT EXISTS tombstone_dirty ON tombstone(dirty) WHERE dirty = 1; + CREATE INDEX IF NOT EXISTS study_log_dirty ON study_log(dirty) WHERE dirty = 1; + CREATE INDEX IF NOT EXISTS peek_dirty ON peek(dirty) WHERE dirty = 1; + + -- Protocol 1's position means nothing to protocol 2. With no epoch + -- recorded, the next sync re-hydrates: it pulls everything, then + -- offers every stamped row back to the server. + DELETE FROM meta WHERE k IN ('sync.cursor', 'sync.pushedAt'); + `, + run: jsonTombstoneKeys, + }, ]; +/** + * Migration 9's data step: tombstone keys in protocol 2's form, a JSON array + * of the key's values. Protocol 1 stored the bare value, or values joined by + * a space — which a key containing a space could not survive. + */ +async function jsonTombstoneKeys(db: Db): Promise { + const device = (await db.get<{ v: string }>("SELECT v FROM meta WHERE k = 'sync.device'"))?.v ?? ""; + const graves = await db.all<{ tbl: string; pk: string }>("SELECT tbl, pk FROM tombstone"); + for (const g of graves) { + if (g.pk.startsWith("[")) continue; // written in the new form already + const key = + g.tbl === "card" + ? [Number(g.pk)] + : g.tbl === "study_log" + ? [Number(g.pk), device] + : g.tbl === "peek" + ? [g.pk, device] + : [g.pk]; + await db.run("UPDATE OR REPLACE tombstone SET pk = ? WHERE tbl = ? AND pk = ?", [ + JSON.stringify(key), + g.tbl, + g.pk, + ]); + } +} + /** * Migration 8's data step. * diff --git a/app/src/db/writes.ts b/app/src/db/writes.ts index 99ae3be..41f36d7 100644 --- a/app/src/db/writes.ts +++ b/app/src/db/writes.ts @@ -7,27 +7,36 @@ real history, and last-write-wins duly clobbered months of progress with an empty seed. - The rule that prevents it: SEEDED OR DEFAULTED STATE NEVER CARRIES A WRITE - TIMESTAMP. Only a genuine user edit stamps the clock. + The rule that prevents it: SEEDED OR DEFAULTED STATE IS NEVER A WRITE. + Only a genuine user edit stamps the clock and marks its row dirty. - Three layers enforce it: - - 1. the DDL — `updated_at INTEGER NOT NULL DEFAULT 0` on every syncable - table, so forgetting the column is the safe failure, not the unsafe - one (see migrations.ts); - 2. this file — seedX() helpers never mention updated_at; editX() helpers - always set it from now(); + 1. the DDL — `updated_at`, `dirty` and `rev` all default to 0 on every + syncable table, so forgetting them is the safe failure: an INSERT + that omits them is invisible to sync (see migrations.ts); + 2. this file — seedX() helpers never mention them; editX() helpers + always stamp updated_at, set dirty = 1 and bump rev; 3. lint + test — Date.now() is banned everywhere under src/db/ except here, and test/db/conformance.ts asserts a freshly seeded database - has no non-zero updated_at anywhere. + has nothing stamped and nothing dirty. - Sync is out of scope for this pass. The point is that when it lands, the - schema already cannot express the bug. */ + ════ WHAT SYNC NEEDS FROM A WRITE ════ + + dirty = 1 queues the row for the server; rev tells the sync client + whether the row changed again while its push was in flight. base_seq is + the sync client's alone — no write here touches it, except that a delete + carries the deleted row's base_seq into its tombstone, which is what lets + the server compare-and-swap the delete like any other write. */ import type { Db, Params } from "./types.js"; import type { Card, Grade } from "@lib/srs.js"; import { lemmaId } from "@shared/lemma-id.mjs"; -import { uuidv7 } from "./ids.js"; +import { + RESET_SCOPES, + SYNC_TABLES, + encodePk, + type ResetScope as MarkerScope, +} from "@shared/sync-protocol.mjs"; +import { randomId, uuidv7 } from "./ids.js"; /* ───────────────────────────────────────────────────────────────────── The clock. This is the ONLY place in src/db/ that reads it. @@ -36,25 +45,54 @@ import { uuidv7 } from "./ids.js"; /** Wall-clock milliseconds, for stamping a genuine user edit. */ export const now = (): number => Date.now(); +/* ───────────────────────────────────────────────────────────────────── + Shared helpers + ───────────────────────────────────────────────────────────────────── */ + +/** This install's id — the owner of its per-device counter rows. */ +const devices = new WeakMap(); +export async function deviceId(db: Db): Promise { + const known = devices.get(db); + if (known) return known; + let id = (await db.get<{ v: string }>("SELECT v FROM meta WHERE k = 'sync.device'"))?.v; + if (!id) { + id = randomId(); + // Describes this install, not the learner: never stamped, never synced. + await db.run("INSERT OR IGNORE INTO meta (k, v, updated_at) VALUES ('sync.device', ?, 0)", [id]); + } + devices.set(db, id); + return id; +} + /** - * Record that a row was deleted. + * Delete rows AND record that they were deleted, in the caller's transaction. * - * A delete leaves nothing for last-write-wins to compare against, so without - * this the other device would push its still-live copy back and the row - * would silently return. Always written inside the same transaction as the - * delete it describes. + * A delete leaves nothing behind to compare, so without a tombstone another + * device would push its live copy straight back. The tombstone carries the + * row's base_seq: the server applies the delete only if nothing has changed + * the row since this device last agreed with it. */ -async function tombstone(db: Db, tbl: string, pk: string | number): Promise { - await db.run( - `INSERT INTO tombstone (tbl, pk, updated_at) VALUES (?, ?, ?) - ON CONFLICT(tbl, pk) DO UPDATE SET updated_at = excluded.updated_at`, - [tbl, String(pk), now()], +async function remove(db: Db, tbl: string, where: string, params: Params = []): Promise { + const spec = SYNC_TABLES[tbl]!; + const rows = await db.all>( + `SELECT ${spec.pk.join(", ")}, base_seq FROM ${tbl} WHERE ${where}`, + params, ); + const t = now(); + for (const r of rows) { + await db.run( + `INSERT INTO tombstone (tbl, pk, updated_at, base_seq, dirty, rev) VALUES (?, ?, ?, ?, 1, 1) + ON CONFLICT(tbl, pk) DO UPDATE SET updated_at = excluded.updated_at, + base_seq = excluded.base_seq, dirty = 1, rev = tombstone.rev + 1`, + [tbl, encodePk(tbl, r), t, Number(r.base_seq ?? 0)], + ); + } + await db.run(`DELETE FROM ${tbl} WHERE ${where}`, params); } /* ═════════════════════════════════════════════════════════════════════ SEED WRITES — defaults, first-run state, anything the user did not do. - None of these may set updated_at; the column default of 0 is the point. + None of these may set updated_at, dirty or rev; the defaults are the point. ═════════════════════════════════════════════════════════════════════ */ /** First-run roadmap position. Unit 1.1 is where everyone starts. */ @@ -90,28 +128,26 @@ export async function seedChatTurn(db: Db, role: string, body: string, at: numbe } /* ═════════════════════════════════════════════════════════════════════ - EDIT WRITES — a real action by the learner. Every one stamps the clock. + EDIT WRITES — a real action by the learner. Every one stamps the clock + and marks its row for the server. ═════════════════════════════════════════════════════════════════════ */ /** Answering a card in the review overlay. */ export async function editCard(db: Db, lemmaId: number, card: Card): Promise { await db.run( - `INSERT INTO card (lemma_id, state, ease, interval, due, reps, lapses, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) + `INSERT INTO card (lemma_id, state, ease, interval, due, reps, lapses, updated_at, dirty, rev) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1, 1) ON CONFLICT(lemma_id) DO UPDATE SET state = excluded.state, ease = excluded.ease, interval = excluded.interval, due = excluded.due, reps = excluded.reps, lapses = excluded.lapses, - updated_at = excluded.updated_at`, + updated_at = excluded.updated_at, dirty = 1, rev = card.rev + 1`, [lemmaId, card.state, card.ease, card.interval, card.due, card.reps, card.lapses, now()], ); } /** Forgetting a card back to new, from the vocabulary tab. */ export async function editCardReset(db: Db, lemmaId: number): Promise { - await db.tx(async (tx) => { - await tx.run("DELETE FROM card WHERE lemma_id = ?", [lemmaId]); - await tombstone(tx, "card", lemmaId); - }); + await db.tx((tx) => remove(tx, "card", "lemma_id = ?", [lemmaId])); } /** The tutor's ::progress read, or the learner's "not yet". */ @@ -121,10 +157,9 @@ export async function editUnitConfidence( confidence: number, ): Promise { await db.run( - `INSERT INTO progress (unit_id, confidence, updated_at) - VALUES (?, ?, ?) + `INSERT INTO progress (unit_id, confidence, updated_at, dirty, rev) VALUES (?, ?, ?, 1, 1) ON CONFLICT(unit_id) DO UPDATE SET confidence = excluded.confidence, - updated_at = excluded.updated_at`, + updated_at = excluded.updated_at, dirty = 1, rev = progress.rev + 1`, [unitId, Math.max(0, Math.min(100, Math.round(confidence))), now()], ); } @@ -132,8 +167,9 @@ export async function editUnitConfidence( /** A unit finished. Nothing un-finishes one short of a reset. */ export async function editUnitDone(db: Db, unitId: string): Promise { await db.run( - `INSERT INTO progress (unit_id, done, updated_at) VALUES (?, 1, ?) - ON CONFLICT(unit_id) DO UPDATE SET done = 1, updated_at = excluded.updated_at`, + `INSERT INTO progress (unit_id, done, updated_at, dirty, rev) VALUES (?, 1, ?, 1, 1) + ON CONFLICT(unit_id) DO UPDATE SET done = 1, + updated_at = excluded.updated_at, dirty = 1, rev = progress.rev + 1`, [unitId, now()], ); } @@ -146,8 +182,9 @@ export async function editCurrentUnit(db: Db, unitId: string): Promise { /** A preference the learner changed. */ export async function editMeta(db: Db, k: string, v: string): Promise { await db.run( - `INSERT INTO meta (k, v, updated_at) VALUES (?, ?, ?) - ON CONFLICT(k) DO UPDATE SET v = excluded.v, updated_at = excluded.updated_at`, + `INSERT INTO meta (k, v, updated_at, dirty, rev) VALUES (?, ?, ?, 1, 1) + ON CONFLICT(k) DO UPDATE SET v = excluded.v, + updated_at = excluded.updated_at, dirty = 1, rev = meta.rev + 1`, [k, v, now()], ); } @@ -156,29 +193,47 @@ export async function editMeta(db: Db, k: string, v: string): Promise { export async function editChatTurn(db: Db, role: string, body: string): Promise { const t = now(); await db.run( - "INSERT INTO chat (id, role, body, created_at, updated_at) VALUES (?, ?, ?, ?, ?)", + "INSERT INTO chat (id, role, body, created_at, updated_at, dirty, rev) VALUES (?, ?, ?, ?, ?, 1, 1)", [uuidv7(t), role, body, t, t], ); } +/** + * Raise a reset marker: `reset.` counts deliberate shrinks, and every + * device that sees it rise empties the same things. The device raising it + * records that it has already done so, so its own marker coming back from + * the server does not wipe what it has written since. + */ +async function raiseMarker(db: Db, scope: MarkerScope): Promise { + const key = `reset.${scope}`; + const at = Number((await db.get<{ v: string }>("SELECT v FROM meta WHERE k = ?", [key]))?.v ?? 0); + const next = String((Number.isFinite(at) ? at : 0) + 1); + await editMeta(db, key, next); + await db.run("INSERT OR REPLACE INTO meta (k, v, updated_at) VALUES (?, ?, 0)", [`sync.applied.${key}`, next]); +} + /** Wipe the transcript. Progress and cards are untouched. */ export async function editChatClear(db: Db): Promise { await db.tx(async (tx) => { - const rows = await tx.all<{ id: string }>("SELECT id FROM chat"); - await tx.run("DELETE FROM chat"); - for (const r of rows) await tombstone(tx, "chat", r.id); + await remove(tx, "chat", "1 = 1"); + await raiseMarker(tx, "chat"); }); } -/** Trim the transcript. The artifact kept the last 26 turns. */ -export async function editChatTrim(db: Db, keep: number): Promise { - await db.tx(async (tx) => { - const kept = "SELECT id FROM chat ORDER BY created_at DESC, id DESC LIMIT ?"; - const doomed = await tx.all<{ id: string }>(`SELECT id FROM chat WHERE id NOT IN (${kept})`, [keep]); - if (!doomed.length) return; - await tx.run(`DELETE FROM chat WHERE id NOT IN (${kept})`, [keep]); - for (const r of doomed) await tombstone(tx, "chat", r.id); - }); +/** + * Keep the transcript to its last `keep` turns — on THIS device. + * + * Not a deletion: nothing is tombstoned, and the server keeps the history. + * A trim used to tombstone every turn outside this device's newest 26, which + * on a device that had just pulled another's turns deleted those turns + * everywhere. A turn not yet pushed is never trimmed. + */ +export async function pruneChat(db: Db, keep: number): Promise { + await db.run( + `DELETE FROM chat WHERE dirty = 0 AND id NOT IN + (SELECT id FROM chat ORDER BY created_at DESC, id DESC LIMIT ?)`, + [keep], + ); } /** Study happened today: reviews, correct answers, drill answers. */ @@ -189,24 +244,24 @@ export async function editStudyLog( ): Promise { const { reviews = 0, correct = 0, drills = 0 } = delta; await db.run( - `INSERT INTO study_log (day, reviews, correct, drills, updated_at) - VALUES (?, ?, ?, ?, ?) - ON CONFLICT(day) DO UPDATE SET + `INSERT INTO study_log (day, device, reviews, correct, drills, updated_at, dirty, rev) + VALUES (?, ?, ?, ?, ?, ?, 1, 1) + ON CONFLICT(day, device) DO UPDATE SET reviews = study_log.reviews + excluded.reviews, correct = study_log.correct + excluded.correct, drills = study_log.drills + excluded.drills, - updated_at = excluded.updated_at`, - [day, reviews, correct, drills, now()], + updated_at = excluded.updated_at, dirty = 1, rev = study_log.rev + 1`, + [day, await deviceId(db), reviews, correct, drills, now()], ); } /** He looked a word up in the rail. Feeds the persistent underline. */ export async function editPeek(db: Db, form: string): Promise { await db.run( - `INSERT INTO peek (form, count, updated_at) VALUES (?, 1, ?) - ON CONFLICT(form) DO UPDATE SET count = peek.count + 1, - updated_at = excluded.updated_at`, - [form, now()], + `INSERT INTO peek (form, device, count, updated_at, dirty, rev) VALUES (?, ?, 1, ?, 1, 1) + ON CONFLICT(form, device) DO UPDATE SET count = peek.count + 1, + updated_at = excluded.updated_at, dirty = 1, rev = peek.rev + 1`, + [form, await deviceId(db), now()], ); } @@ -235,6 +290,23 @@ export interface AddedWord { created: boolean; } +/** The derived rows that make a custom word look like any other word. */ +export async function deriveCustomLemma(db: Db, word: CustomWord): Promise { + const id = lemmaId(word.headword, word.pos); + await db.run( + `INSERT OR REPLACE INTO lemma (id, headword, pos, freq_rank, level, gloss_en, gloss_ko, + unit_band, source) + VALUES (?, ?, ?, NULL, NULL, ?, '', 0, 'custom')`, + [id, word.headword, word.pos, word.gloss], + ); + await db.run("INSERT OR REPLACE INTO surface (form, lemma_id, analysis) VALUES (?, ?, ?)", [ + word.headword, + id, + "headword, custom", + ]); + return id; +} + /** * Add a word of the learner's own. * @@ -254,44 +326,31 @@ export async function editAddCustomWord(db: Db, word: CustomWord): Promise + tx.run( + `INSERT INTO card (lemma_id, state, ease, interval, due, reps, lapses, updated_at, dirty, rev) + VALUES (?, 0, 2.5, 0, 0, 0, 0, ?, 1, 1) + ON CONFLICT(lemma_id) DO NOTHING`, + [id, t], + ); + if (existing) { // Already known — just make sure it is studiable. Its gloss stays the // dictionary's; overwriting curated content from a text field would be // a poor trade. - await tx.run( - `INSERT INTO card (lemma_id, state, ease, interval, due, reps, lapses, updated_at) - VALUES (?, 0, 2.5, 0, 0, 0, 0, ?) - ON CONFLICT(lemma_id) DO NOTHING`, - [existing.id, now()], - ); + await newCardFor(existing.id); return { lemmaId: existing.id, created: false }; } - const id = lemmaId(headword, word.pos); - const t = now(); - await tx.run( - `INSERT INTO custom_word (headword, pos, gloss, updated_at) VALUES (?, ?, ?, ?) - ON CONFLICT(headword, pos) DO UPDATE SET gloss = excluded.gloss, updated_at = excluded.updated_at`, + `INSERT INTO custom_word (headword, pos, gloss, updated_at, dirty, rev) VALUES (?, ?, ?, ?, 1, 1) + ON CONFLICT(headword, pos) DO UPDATE SET gloss = excluded.gloss, + updated_at = excluded.updated_at, dirty = 1, rev = custom_word.rev + 1`, [headword, word.pos, gloss, t], ); - await tx.run( - `INSERT OR REPLACE INTO lemma (id, headword, pos, freq_rank, level, gloss_en, gloss_ko, - unit_band, source) - VALUES (?, ?, ?, NULL, NULL, ?, '', 0, 'custom')`, - [id, headword, word.pos, gloss], - ); - await tx.run("INSERT OR REPLACE INTO surface (form, lemma_id, analysis) VALUES (?, ?, ?)", [ - headword, - id, - "headword, custom", - ]); - await tx.run( - `INSERT INTO card (lemma_id, state, ease, interval, due, reps, lapses, updated_at) - VALUES (?, 0, 2.5, 0, 0, 0, 0, ?) - ON CONFLICT(lemma_id) DO NOTHING`, - [id, t], - ); + const id = await deriveCustomLemma(tx, { headword, gloss, pos: word.pos }); + await newCardFor(id); return { lemmaId: id, created: true }; }); } @@ -304,12 +363,10 @@ export async function editRemoveCustomWord(db: Db, lemmaId: number): Promise + RESET_SCOPES[scope].meta.map((p) => `k LIKE '${p}%'`).join(" OR ") || "0"; export async function editReset(db: Db, scope: ResetScope): Promise { await db.tx(async (tx) => { - // Tombstone before deleting, while the keys are still readable — a reset - // must propagate, or the next pull restores everything it just erased. - for (const r of await tx.all<{ unit_id: string }>("SELECT unit_id FROM progress")) - await tombstone(tx, "progress", r.unit_id); - for (const r of await tx.all<{ id: string }>("SELECT id FROM chat")) - await tombstone(tx, "chat", r.id); - for (const r of await tx.all<{ phase: number; kind: string; item: string }>( - "SELECT phase, kind, item FROM phase_ledger", - )) - await tombstone(tx, "phase_ledger", JSON.stringify([r.phase, r.kind, r.item])); - for (const r of await tx.all<{ k: string }>(`SELECT k FROM meta WHERE ${ROAD_META}`)) - await tombstone(tx, "meta", r.k); - - await tx.run("DELETE FROM progress"); - await tx.run("DELETE FROM chat"); - await tx.run("DELETE FROM phase_ledger"); - await tx.run(`DELETE FROM meta WHERE ${ROAD_META}`); + for (const tbl of RESET_SCOPES[scope].tables) await remove(tx, tbl, "1 = 1"); + await remove(tx, "meta", metaWhere(scope)); if (scope === "everything") { - for (const r of await tx.all<{ lemma_id: number }>("SELECT lemma_id FROM card")) - await tombstone(tx, "card", r.lemma_id); - for (const r of await tx.all<{ day: number }>("SELECT day FROM study_log")) - await tombstone(tx, "study_log", r.day); - for (const r of await tx.all<{ form: string }>("SELECT form FROM peek")) - await tombstone(tx, "peek", r.form); - - for (const r of await tx.all<{ headword: string; pos: string }>( - "SELECT headword, pos FROM custom_word", - )) - await tombstone(tx, "custom_word", JSON.stringify([r.headword, r.pos])); - for (const r of await tx.all<{ word: string }>("SELECT word FROM evidence")) - await tombstone(tx, "evidence", r.word); - for (const r of await tx.all<{ word: string }>("SELECT word FROM confusion")) - await tombstone(tx, "confusion", r.word); - for (const r of await tx.all<{ k: string }>(`SELECT k FROM meta WHERE ${LEARNER_META}`)) - await tombstone(tx, "meta", r.k); - - await tx.run("DELETE FROM card"); - await tx.run("DELETE FROM study_log"); - await tx.run("DELETE FROM peek"); - await tx.run("DELETE FROM custom_word"); - await tx.run("DELETE FROM evidence"); - await tx.run("DELETE FROM confusion"); await tx.run("DELETE FROM surface WHERE lemma_id IN (SELECT id FROM lemma WHERE source = 'custom')"); await tx.run("DELETE FROM lemma WHERE source = 'custom'"); - // Preferences, grammar flags and notes, the trainer score and the - // round counter. Device bookkeeping (schema_version, dict.*, sync.*) - // is left alone — it describes this install, not the learner. - await tx.run(`DELETE FROM meta WHERE ${LEARNER_META}`); } + await raiseMarker(tx, scope); }); } diff --git a/app/src/domain/cards.ts b/app/src/domain/cards.ts index 5d607fa..c5517d6 100644 --- a/app/src/domain/cards.ts +++ b/app/src/domain/cards.ts @@ -177,7 +177,13 @@ export interface DayRow { } export async function studyLog(db: Db, sinceDay: number): Promise { - return db.all("SELECT * FROM study_log WHERE day >= ? ORDER BY day", [sinceDay]); + // One row per device per day — summed, so a phone's reviews and a + // laptop's both count and neither can overwrite the other's. + return db.all( + `SELECT day, SUM(reviews) AS reviews, SUM(correct) AS correct, SUM(drills) AS drills + FROM study_log WHERE day >= ? GROUP BY day ORDER BY day`, + [sinceDay], + ); } /** Consecutive days with any activity, counting back from today. */ diff --git a/app/src/state/store.tsx b/app/src/state/store.tsx index 107add5..507087d 100644 --- a/app/src/state/store.tsx +++ b/app/src/state/store.tsx @@ -30,7 +30,8 @@ import { writeServerConfig, type ServerConfig, } from "../domain/server-config.js"; -import { SYNC_PAUSED, trySync, type SyncResult } from "../sync/client.js"; +import { forgetServerPosition, trySync, type SyncResult } from "../sync/client.js"; +import { unitIndex } from "../domain/gate.js"; import type { ProgressState } from "@lib/gate.js"; import type { FocusMode } from "../domain/gate.js"; @@ -232,6 +233,9 @@ export function StoreProvider({ if (!store) return; if (cfg) await writeServerConfig(store.db, cfg); else await clearServerConfig(store.db); + // A different server is a different conversation: hydrate from its + // first page before anything here may be pushed to it. + await forgetServerPosition(store.db); setServerState(cfg); }, [store], @@ -243,13 +247,9 @@ export function StoreProvider({ const running = useRef(false); const syncNow = useCallback(async () => { if (!store || !server || running.current) return; - if (SYNC_PAUSED) { - setSyncState({ at: Date.now(), result: null, error: SYNC_PAUSED, running: false }); - return; - } running.current = true; setSyncState((s) => ({ ...s, running: true })); - const result = await trySync(store.db, server); + const result = await trySync(store.db, server, { unitIndex }); running.current = false; setSyncState({ at: Date.now(), diff --git a/app/src/sync/client.ts b/app/src/sync/client.ts index 8adb423..0f546d1 100644 --- a/app/src/sync/client.ts +++ b/app/src/sync/client.ts @@ -1,39 +1,34 @@ -/* The sync loop: push what changed here, pull what changed there, apply. +/* The sync loop, protocol 2: pull everything, then push what changed here. Offline-first is not negotiable. Sync runs only when a server is configured, and every failure is non-fatal — the app must stay completely - usable with the Pi unreachable, which is the acceptance criterion the - whole port was built around. Nothing in here may throw into the UI. */ + usable with the Pi unreachable. Nothing in here may throw into the UI. + + The order is the first gate. A device may not push until it has pulled + every page the server holds for this epoch: the week-old laptop that + pushed first is exactly how the artifact lost a week of the phone's work. + See shared/sync-protocol.mjs for the other two gates. */ import type { Db } from "../db/types.js"; +import { deriveCustomLemma } from "../db/writes.js"; import { PAGE_SIZE, + PROTOCOL, + PROTOCOL_HEADER, + RESET_SCOPES, SYNC_TABLES, SYNC_TABLE_NAMES, - incomingWins, + decodePk, + encodePk, isSyncableMetaKey, + resetCovers, + type PullResponse, + type PushResponse, + type ServerRow, + type WireRow, } from "@shared/sync-protocol.mjs"; - -/** A row on the wire: its table, its columns, and when it last changed. */ -export interface SyncRow { - tbl: string; - data: Record; - updated_at: number; - /** Present and true when this row is a delete. */ - deleted?: boolean; -} - -export interface PushRequest { - rows: SyncRow[]; -} -export interface PushResponse { - cursor: number; -} -export interface PullResponse { - rows: SyncRow[]; - cursor: number; - more: boolean; -} +import { lemmaId } from "@shared/lemma-id.mjs"; +import { resolve, type Data, type ResolveContext } from "./resolve.js"; export interface SyncConfig { baseUrl: string; @@ -43,209 +38,405 @@ export interface SyncConfig { export interface SyncResult { pushed: number; pulled: number; + conflicts: number; cursor: number; + /** The server was new to this device, or had been reset: everything was + pulled from the start and every stamped row offered back. */ + rehydrated: boolean; } -/** - * Why sync is not running, or null when it is. - * - * Cards are keyed by stable lemma ids now, while the server still holds rows - * written under the old positional ids. Exchanging them would plant cards - * under ids that name no word, on both sides. Everything stays local — which - * this app is built to do — until the sync protocol that replaces this one - * ships with its own server schema. - */ -export const SYNC_PAUSED: string | null = - "Sync is paused while the storage format changes. Everything is kept on this device"; +const EPOCH = "sync.epoch"; +const CURSOR = "sync.cursor"; +const HYDRATED = "sync.hydrated"; -const CURSOR_KEY = "sync.cursor"; -const PUSHED_KEY = "sync.pushedAt"; +/** Rounds of push-and-settle before giving up until the next sync. */ +const MAX_PUSH_ROUNDS = 5; /* ── local bookkeeping ──────────────────────────────────────────────── - These are device state, not user data. They are written with - updated_at = 0 and excluded from the syncable-meta allowlist, so they - never travel. */ + Device state, not user data: written unstamped and never dirty, and + `sync.*` is outside the syncable-meta allowlist besides. */ -async function readNum(db: Db, key: string): Promise { - const row = await db.get<{ v: string }>("SELECT v FROM meta WHERE k = ?", [key]); - const n = row ? Number(row.v) : 0; - return Number.isFinite(n) ? n : 0; +async function readMeta(db: Db, k: string): Promise { + return (await db.get<{ v: string }>("SELECT v FROM meta WHERE k = ?", [k]))?.v; } -async function writeNum(db: Db, key: string, value: number): Promise { - await db.run("INSERT OR REPLACE INTO meta (k, v, updated_at) VALUES (?, ?, 0)", [ - key, - String(value), - ]); +async function writeMeta(db: Db, k: string, v: string | number): Promise { + await db.run("INSERT OR REPLACE INTO meta (k, v, updated_at) VALUES (?, ?, 0)", [k, String(v)]); } -/* ── collect ─────────────────────────────────────────────────────────── */ +/** Forget this device's place with the server, so the next sync starts over. */ +export async function forgetServerPosition(db: Db): Promise { + await db.run("DELETE FROM meta WHERE k IN (?, ?, ?)", [EPOCH, CURSOR, HYDRATED]); +} -/** Everything edited on this device since the last successful push. */ -export async function collectDirty(db: Db, pushedAt: number): Promise { - const out: SyncRow[] = []; +/** Has this device pulled everything its server holds? */ +export async function isHydrated(db: Db): Promise { + return (await readMeta(db, HYDRATED)) === "1"; +} - for (const tbl of SYNC_TABLE_NAMES) { - const spec = SYNC_TABLES[tbl]!; - const cols = [...spec.cols, "updated_at"].join(", "); - const rows = await db.all>( - `SELECT ${cols} FROM ${tbl} WHERE updated_at > ? ORDER BY updated_at LIMIT ?`, - [pushedAt, PAGE_SIZE], +/* ── rows ─────────────────────────────────────────────────────────────── */ + +type LocalRow = Data & { base_seq: number; dirty: number; rev: number; updated_at: number }; + +interface Local { + live: LocalRow | undefined; + tomb: { base_seq: number; dirty: number; rev: number } | undefined; +} + +const whereOf = (tbl: string) => SYNC_TABLES[tbl]!.pk.map((c) => `${c} = ?`).join(" AND "); + +async function readLocal(db: Db, tbl: string, pk: string): Promise { + const live = await db.get(`SELECT * FROM ${tbl} WHERE ${whereOf(tbl)}`, decodePk(pk)); + const tomb = await db.get<{ base_seq: number; dirty: number; rev: number }>( + "SELECT base_seq, dirty, rev FROM tombstone WHERE tbl = ? AND pk = ?", + [tbl, pk], + ); + return { live, tomb }; +} + +const pick = (tbl: string, row: Data): Data => + Object.fromEntries(SYNC_TABLES[tbl]!.cols.map((c) => [c, row[c] ?? null])); + +/** Write a row as the server holds it: clean, and agreed at `seq`. */ +async function adopt(db: Db, row: ServerRow): Promise { + const spec = SYNC_TABLES[row.tbl]!; + const key = decodePk(row.pk); + + if (row.deleted) { + const before = await db.get(`SELECT * FROM ${row.tbl} WHERE ${whereOf(row.tbl)}`, key); + await db.run(`DELETE FROM ${row.tbl} WHERE ${whereOf(row.tbl)}`, key); + // A clean tombstone remembers the server's seq for this key, so a later + // re-creation here pushes against the right base. + await db.run( + `INSERT INTO tombstone (tbl, pk, updated_at, base_seq, dirty, rev) VALUES (?, ?, ?, ?, 0, 0) + ON CONFLICT(tbl, pk) DO UPDATE SET base_seq = excluded.base_seq, dirty = 0`, + [row.tbl, row.pk, row.updated_at ?? 0, row.seq], ); - - for (const row of rows) { - // The allowlist is applied at the point of collection, so a - // device-local meta key cannot leave this machine even by accident. - if (tbl === "meta" && !isSyncableMetaKey(String(row.k))) continue; - const { updated_at, ...data } = row; - out.push({ tbl, data, updated_at: Number(updated_at) }); + if (row.tbl === "custom_word" && before) { + const id = lemmaId(String(before.headword), String(before.pos)); + await db.run("DELETE FROM surface WHERE lemma_id = ? AND lemma_id IN (SELECT id FROM lemma WHERE source = 'custom')", [id]); + await db.run("DELETE FROM lemma WHERE id = ? AND source = 'custom'", [id]); } + return; } - const graves = await db.all<{ tbl: string; pk: string; updated_at: number }>( - "SELECT tbl, pk, updated_at FROM tombstone WHERE updated_at > ? ORDER BY updated_at LIMIT ?", - [pushedAt, PAGE_SIZE], + const data = pick(row.tbl, row.data ?? {}); + const cols = spec.cols; + await db.run( + `INSERT INTO ${row.tbl} (${cols.join(", ")}, updated_at, base_seq, dirty) + VALUES (${cols.map(() => "?").join(", ")}, ?, ?, 0) + ON CONFLICT(${spec.pk.join(", ")}) DO UPDATE SET + ${cols.map((c) => `${c} = excluded.${c}`).join(", ")}, + updated_at = excluded.updated_at, base_seq = excluded.base_seq, dirty = 0`, + [...cols.map((c) => data[c] ?? null), row.updated_at ?? 0, row.seq], + ); + await db.run("DELETE FROM tombstone WHERE tbl = ? AND pk = ?", [row.tbl, row.pk]); + + if (row.tbl === "custom_word") { + await deriveCustomLemma(db, { + headword: String(data.headword), + pos: String(data.pos), + gloss: String(data.gloss ?? ""), + }); + } + if (row.tbl === "meta" && String(data.k).startsWith("reset.")) { + await obeyMarker(db, String(data.k), String(data.v)); + } +} + +/** Ours stands: agree with the server's seq, so the next push overwrites it. */ +async function keep(db: Db, tbl: string, pk: string, local: Local, seq: number): Promise { + if (local.live) { + await db.run(`UPDATE ${tbl} SET base_seq = ?, dirty = 1 WHERE ${whereOf(tbl)}`, [seq, ...decodePk(pk)]); + } else { + await db.run("UPDATE tombstone SET base_seq = ?, dirty = 1 WHERE tbl = ? AND pk = ?", [seq, tbl, pk]); + } +} + +/** A combination of both: written here, agreed at `seq`, and pushed. */ +async function merge(db: Db, tbl: string, pk: string, seq: number, data: Data): Promise { + const cols = SYNC_TABLES[tbl]!.cols; + await db.run( + `UPDATE ${tbl} SET ${cols.map((c) => `${c} = ?`).join(", ")}, + base_seq = ?, dirty = 1, rev = rev + 1 WHERE ${whereOf(tbl)}`, + [...cols.map((c) => data[c] ?? null), seq, ...decodePk(pk)], + ); +} + +/** Settle one disagreement between this device and the server. */ +async function settle(db: Db, remote: ServerRow, local: Local, ctx: ResolveContext): Promise { + // The server has no row at all: nothing to disagree with. + if (remote.seq === 0) { + if (local.live) await keep(db, remote.tbl, remote.pk, local, 0); + else await db.run("DELETE FROM tombstone WHERE tbl = ? AND pk = ?", [remote.tbl, remote.pk]); + return; + } + + const localSide = local.live + ? { deleted: false, data: local.live as Data } + : { deleted: true, data: null }; + const outcome = resolve(remote.tbl, localSide, { deleted: remote.deleted, data: remote.data }, ctx); + if (outcome.kind === "adopt") await adopt(db, remote); + else if (outcome.kind === "keep") await keep(db, remote.tbl, remote.pk, local, remote.seq); + else await merge(db, remote.tbl, remote.pk, remote.seq, outcome.data); +} + +/* ── deliberate shrinks ───────────────────────────────────────────────── */ + +/** + * Empty what a reset marker covers, once per rise of its counter. + * + * Unsynced edits here go too: a reset this device had not heard of still + * wins over work done on the data it reset. Rows written after the reset + * arrive later in the pull — higher change_seq — and survive it. + */ +async function obeyMarker(db: Db, key: string, value: string): Promise { + const scope = key.slice("reset.".length) as keyof typeof RESET_SCOPES; + const spec = RESET_SCOPES[scope]; + if (!spec) return; + const seen = Number((await readMeta(db, `sync.applied.${key}`)) ?? 0); + const n = Number(value); + if (!(n > seen)) return; + + for (const tbl of spec.tables) { + await db.run(`DELETE FROM ${tbl}`); + await db.run("DELETE FROM tombstone WHERE tbl = ?", [tbl]); + } + for (const prefix of spec.meta) { + await db.run("DELETE FROM meta WHERE k LIKE ?", [`${prefix}%`]); + } + if (spec.tables.includes("custom_word")) { + await db.run("DELETE FROM surface WHERE lemma_id IN (SELECT id FROM lemma WHERE source = 'custom')"); + await db.run("DELETE FROM lemma WHERE source = 'custom'"); + } + await writeMeta(db, `sync.applied.${key}`, n); +} + +/** Resets raised here that the server has not heard of yet. */ +async function pendingResets(db: Db): Promise { + const rows = await db.all<{ k: string }>("SELECT k FROM meta WHERE k LIKE 'reset.%' AND dirty = 1"); + return rows.map((r) => r.k.slice("reset.".length)); +} + +/* ── pull ─────────────────────────────────────────────────────────────── */ + +async function applyIncoming( + db: Db, + row: ServerRow, + pending: string[], + ctx: ResolveContext, +): Promise<"applied" | "settled" | "skipped"> { + const spec = SYNC_TABLES[row.tbl]; + if (!spec) return "skipped"; // a newer server; ignore rather than fail + const key = decodePk(row.pk); + const probe: Data = row.data ?? Object.fromEntries(spec.pk.map((c, i) => [c, key[i] ?? null])); + if (row.tbl === "meta" && !isSyncableMetaKey(String(probe.k ?? key[0]))) return "skipped"; + + // A reset raised here, not yet pushed, covers this row: the server's copy + // predates it and is about to be emptied everywhere anyway. + if (pending.some((scope) => resetCovers(scope, row.tbl, probe))) return "skipped"; + + const local = await readLocal(db, row.tbl, row.pk); + const mine = local.live ?? local.tomb; + if (mine && mine.base_seq === row.seq) return "skipped"; // our own write, echoed back + + if (!mine || !mine.dirty) { + await adopt(db, row); + return "applied"; + } + await settle(db, row, local, ctx); + return "settled"; +} + +/* ── push ─────────────────────────────────────────────────────────────── */ + +interface Outgoing { + wire: WireRow; + rev: number; + /** A live row re-created over a tombstone this device had applied. */ + overTomb: boolean; +} + +const keyOf = (tbl: string, pk: string) => JSON.stringify([tbl, pk]); + +/** Up to one page of dirty rows: deletes and reset markers first, so the + server orders a reset before anything written after it. */ +async function collectDirty(db: Db): Promise { + const out: Outgoing[] = []; + + const graves = await db.all<{ tbl: string; pk: string; base_seq: number; rev: number; updated_at: number }>( + "SELECT tbl, pk, base_seq, rev, updated_at FROM tombstone WHERE dirty = 1 LIMIT ?", + [PAGE_SIZE], ); for (const g of graves) { - out.push({ tbl: g.tbl, data: { pk: g.pk }, updated_at: Number(g.updated_at), deleted: true }); + if (!SYNC_TABLES[g.tbl]) continue; + out.push({ + wire: { tbl: g.tbl, pk: g.pk, base: g.base_seq, deleted: true, data: null, updated_at: g.updated_at }, + rev: g.rev, + overTomb: false, + }); } + const ordered = ["meta", ...SYNC_TABLE_NAMES.filter((t) => t !== "meta")]; + for (const tbl of ordered) { + if (out.length >= PAGE_SIZE) break; + const rows = await db.all( + `SELECT * FROM ${tbl} WHERE dirty = 1 ${tbl === "meta" ? "ORDER BY k LIKE 'reset.%' DESC" : ""} LIMIT ?`, + [PAGE_SIZE - out.length], + ); + for (const row of rows) { + if (tbl === "meta" && !isSyncableMetaKey(String(row.k))) { + // Device bookkeeping that was edited: it never travels, so it is + // never really dirty. + await db.run("UPDATE meta SET dirty = 0 WHERE k = ?", [String(row.k)]); + continue; + } + const pk = encodePk(tbl, row); + let base = row.base_seq; + let overTomb = false; + if (base === 0) { + const tomb = await db.get<{ base_seq: number }>( + "SELECT base_seq FROM tombstone WHERE tbl = ? AND pk = ? AND dirty = 0", + [tbl, pk], + ); + if (tomb) { + base = tomb.base_seq; + overTomb = true; + } + } + out.push({ + wire: { tbl, pk, base, deleted: false, data: pick(tbl, row), updated_at: row.updated_at }, + rev: row.rev, + overTomb, + }); + } + } return out; } -/* ── apply ───────────────────────────────────────────────────────────── */ - -/** Apply one incoming row, last-write-wins against what is already here. */ -async function applyRow(db: Db, row: SyncRow): Promise { - const spec = SYNC_TABLES[row.tbl]; - if (!spec) return false; // unknown table — a newer server; ignore rather than fail - - if (row.deleted) { - const pk = String(row.data.pk); - const where = spec.pk.map((c) => `${c} = ?`).join(" AND "); - const parts = pk.split(" "); - const local = await db.get<{ updated_at: number }>( - `SELECT updated_at FROM ${row.tbl} WHERE ${where}`, - parts, - ); - if (local && !incomingWins(row.updated_at, local.updated_at)) return false; - - await db.tx(async (tx) => { - await tx.run(`DELETE FROM ${row.tbl} WHERE ${where}`, parts); - await tx.run( - `INSERT INTO tombstone (tbl, pk, updated_at) VALUES (?, ?, ?) - ON CONFLICT(tbl, pk) DO UPDATE SET updated_at = excluded.updated_at`, - [row.tbl, pk, row.updated_at], +async function acknowledge(db: Db, sent: Outgoing[], applied: PushResponse["applied"]): Promise { + const byKey = new Map(sent.map((s) => [keyOf(s.wire.tbl, s.wire.pk), s])); + for (const a of applied) { + const s = byKey.get(keyOf(a.tbl, a.pk)); + if (!s) continue; + if (s.wire.deleted) { + await db.run( + `UPDATE tombstone SET base_seq = ?, dirty = CASE WHEN rev = ? THEN 0 ELSE dirty END + WHERE tbl = ? AND pk = ?`, + [a.seq, s.rev, a.tbl, a.pk], ); - }); - return true; + } else { + // Clean only if nothing changed the row while the push was in flight. + await db.run( + `UPDATE ${a.tbl} SET base_seq = ?, dirty = CASE WHEN rev = ? THEN 0 ELSE dirty END + WHERE ${whereOf(a.tbl)}`, + [a.seq, s.rev, ...decodePk(a.pk)], + ); + if (s.overTomb) await db.run("DELETE FROM tombstone WHERE tbl = ? AND pk = ?", [a.tbl, a.pk]); + } } - - if (row.tbl === "meta" && !isSyncableMetaKey(String(row.data.k))) return false; - - const where = spec.pk.map((c) => `${c} = ?`).join(" AND "); - const keys = spec.pk.map((c) => row.data[c] as string | number); - const local = await db.get<{ updated_at: number }>( - `SELECT updated_at FROM ${row.tbl} WHERE ${where}`, - keys, - ); - if (local && !incomingWins(row.updated_at, local.updated_at)) return false; - - const cols = [...spec.cols, "updated_at"]; - const holes = cols.map(() => "?").join(", "); - const values = [...spec.cols.map((c) => row.data[c] ?? null), row.updated_at]; - await db.run(`INSERT OR REPLACE INTO ${row.tbl} (${cols.join(", ")}) VALUES (${holes})`, values); - return true; } -/* ── transport ───────────────────────────────────────────────────────── */ +/* ── transport ────────────────────────────────────────────────────────── */ -async function call( - cfg: SyncConfig, - path: string, - init: RequestInit, - signal?: AbortSignal, -): Promise { +async function call(cfg: SyncConfig, path: string, init: RequestInit, signal?: AbortSignal): Promise { const res = await fetch(`${cfg.baseUrl.replace(/\/$/, "")}${path}`, { ...init, signal, headers: { "content-type": "application/json", authorization: `Bearer ${cfg.token}`, + [PROTOCOL_HEADER]: String(PROTOCOL), ...(init.headers ?? {}), }, }); + if (res.status === 426) throw new Error("the server speaks a different sync protocol — update it"); if (!res.ok) throw new Error(`sync ${path}: HTTP ${res.status}`); return (await res.json()) as T; } -/* ── the loop ────────────────────────────────────────────────────────── */ +/* ── the loop ─────────────────────────────────────────────────────────── */ + +/** Re-agree with a server from nothing: every stamped row is offered again. */ +async function rehydrate(db: Db, epoch: string): Promise { + await db.tx(async (tx) => { + for (const tbl of [...SYNC_TABLE_NAMES, "tombstone"]) { + await tx.run( + `UPDATE ${tbl} SET base_seq = 0, dirty = CASE WHEN updated_at > 0 THEN 1 ELSE 0 END, rev = rev + 1`, + ); + } + await writeMeta(tx, EPOCH, epoch); + await writeMeta(tx, CURSOR, 0); + await writeMeta(tx, HYDRATED, 0); + }); +} /** - * One full exchange. Push first so local edits are never lost to an - * incoming row that would have overwritten them; then pull, paging until - * the server has nothing newer. + * One full exchange: pull every page, then push until nothing is dirty. + * + * `ctx` lets a conflict ask the roadmap which of two units is further on. */ export async function syncOnce( db: Db, cfg: SyncConfig, + ctx: ResolveContext, signal?: AbortSignal, ): Promise { - // Anything edited after this instant stays dirty, whatever the pull - // brings back. That is what keeps an edit made *during* a sync from being - // marked clean by a remote row that happens to carry a later timestamp. - const startedAt = Date.now(); - - const pushedAt = await readNum(db, PUSHED_KEY); - const dirty = await collectDirty(db, pushedAt); - let watermark = pushedAt; - - if (dirty.length) { - const body: PushRequest = { rows: dirty }; - await call( - cfg, - "/api/sync", - { method: "POST", body: JSON.stringify(body) }, - signal, - ); - // Only advance once the server has them. A failed push throws before - // this line and the rows stay dirty, which is the safe direction. - watermark = dirty.reduce((m, r) => Math.max(m, r.updated_at), watermark); - await writeNum(db, PUSHED_KEY, watermark); - } - - /* The pull cursor advances ONLY as rows are applied. - It must never be taken from the push response: the server's newest - change_seq includes rows this device has not seen, and adopting it - would skip them permanently — a silent, unrecoverable data loss, since - nothing would ever ask for that range again. */ - let cursor = await readNum(db, CURSOR_KEY); + let epoch = (await readMeta(db, EPOCH)) ?? ""; + let cursor = Number((await readMeta(db, CURSOR)) ?? 0) || 0; + let rehydrated = false; let pulled = 0; + let conflicts = 0; + /* Gate 1: hydrate. Nothing leaves this device until every page is in. */ for (;;) { - const page = await call( - cfg, - `/api/sync?cursor=${cursor}`, - { method: "GET" }, - signal, - ); - - for (const row of page.rows) { - if (await applyRow(db, row)) pulled++; - // A row we just received is not a local edit. Marking it clean stops - // the next sync pushing it straight back. Bounded by startedAt so a - // concurrent local edit is never swept up with it. - if (row.updated_at <= startedAt) watermark = Math.max(watermark, row.updated_at); + const page = await call(cfg, `/api/sync?cursor=${cursor}`, { method: "GET" }, signal); + if (page.epoch !== epoch) { + if (rehydrated) throw new Error("the server was reset during this sync"); + await rehydrate(db, page.epoch); + epoch = page.epoch; + cursor = 0; + rehydrated = true; + continue; + } + const pending = await pendingResets(db); + for (const row of page.rows) { + const r = await db.tx((tx) => applyIncoming(tx, row, pending, ctx)); + if (r === "applied") pulled++; + if (r === "settled") conflicts++; } - cursor = page.cursor; - await writeNum(db, CURSOR_KEY, cursor); + await writeMeta(db, CURSOR, cursor); if (!page.more) break; } + await writeMeta(db, HYDRATED, 1); - if (watermark > pushedAt) await writeNum(db, PUSHED_KEY, watermark); + /* Then push. A conflict is settled locally and, if ours stands, pushed + again against the server's current seq. */ + let pushed = 0; + for (let round = 0; round < MAX_PUSH_ROUNDS; round++) { + const batch = await collectDirty(db); + if (!batch.length) break; + const res = await call( + cfg, + "/api/sync", + { method: "POST", body: JSON.stringify({ rows: batch.map((b) => b.wire) }) }, + signal, + ); + if (res.epoch !== epoch) { + // Reset between our pull and our push: start over next time. + await forgetServerPosition(db); + break; + } + await acknowledge(db, batch, res.applied); + pushed += res.applied.length; + for (const c of res.conflicts) { + await db.tx(async (tx) => settle(tx, c, await readLocal(tx, c.tbl, c.pk), ctx)); + conflicts++; + } + } - return { pushed: dirty.length, pulled, cursor }; + return { pushed, pulled, conflicts, cursor, rehydrated }; } /** @@ -258,11 +449,12 @@ export async function syncOnce( export async function trySync( db: Db, cfg: SyncConfig | null, + ctx: ResolveContext, signal?: AbortSignal, ): Promise { - if (!cfg?.baseUrl || !cfg.token || SYNC_PAUSED) return null; + if (!cfg?.baseUrl || !cfg.token) return null; try { - return await syncOnce(db, cfg, signal); + return await syncOnce(db, cfg, ctx, signal); } catch (err) { console.warn("[sync] skipped:", err instanceof Error ? err.message : err); return null; diff --git a/app/src/sync/resolve.ts b/app/src/sync/resolve.ts new file mode 100644 index 0000000..0d1d2d1 --- /dev/null +++ b/app/src/sync/resolve.ts @@ -0,0 +1,131 @@ +/* Gate 3: what to do when two copies of one row disagree. + + A conflict means the server's copy changed after this device last agreed + with it, while this device changed its own copy too. The counter says + that much and no more: it cannot say which change is "right". + + The artifact's answer, which PORT.md adopts: THE COPY THAT HOLDS MORE + WINS. For one learner whose worst conflict is a duplicated grade, "more + reviews, more evidence, more finished units" is a better tiebreak than + "newer" — it converges, and it never trades real work for an empty + default. The one exception is a deliberate delete, which is obeyed. + + Three outcomes: + adopt — take the server's copy; + keep — keep ours, and push it back over the server's; + merge — write a combination, and push that. */ + +export type Value = string | number | null; +export type Data = Record; + +export interface Side { + deleted: boolean; + data: Data | null; +} + +export type Resolution = { kind: "adopt" } | { kind: "keep" } | { kind: "merge"; data: Data }; + +export interface ResolveContext { + /** A unit's position on the roadmap, or -1. */ + unitIndex: (unitId: string) => number; +} + +const ADOPT: Resolution = { kind: "adopt" }; +const KEEP: Resolution = { kind: "keep" }; + +const num = (v: Value | undefined): number => { + const n = Number(v ?? 0); + return Number.isFinite(n) ? n : 0; +}; + +/** More wins; a tie goes to the server, so every device lands on one copy. */ +const heavier = (local: number, remote: number): Resolution => (local > remote ? KEEP : ADOPT); + +/** Pick adopt/keep when a merge came out equal to one side. */ +function settle(local: Data, remote: Data, merged: Data): Resolution { + const same = (a: Data) => Object.keys(merged).every((k) => String(merged[k]) === String(a[k])); + if (same(remote)) return ADOPT; + if (same(local)) return KEEP; + return { kind: "merge", data: merged }; +} + +function parseObject(v: Value | undefined): Record | null { + try { + const o = JSON.parse(String(v)); + return o && typeof o === "object" && !Array.isArray(o) ? (o as Record) : null; + } catch { + return null; + } +} + +function resolveMeta(local: Data, remote: Data, ctx: ResolveContext): Resolution { + const k = String(remote.k ?? local.k ?? ""); + + if (k === "road.unit") { + return heavier(ctx.unitIndex(String(local.v)), ctx.unitIndex(String(remote.v))); + } + if (k === "learner.round" || k.startsWith("reset.")) { + return heavier(num(local.v), num(remote.v)); + } + if (k === "trainer.conjugation") { + return heavier(num(parseObject(local.v)?.n as Value), num(parseObject(remote.v)?.n as Value)); + } + if (k === "grammar.learned" || k === "grammar.notes") { + const a = parseObject(local.v); + const b = parseObject(remote.v); + if (!a || !b) return ADOPT; + // Union. Where both have a key, the server's value for a flag; for a + // note, the longer one — a note is only ever grown by its owner. + const out: Record = { ...a, ...b }; + if (k === "grammar.notes") { + for (const key of Object.keys(a)) { + if (key in b && String(a[key] ?? "").length > String(b[key] ?? "").length) out[key] = a[key]; + } + } + return settle(local, remote, { ...remote, v: JSON.stringify(out) }); + } + // Preferences, recent task types: the server's copy. + return ADOPT; +} + +export function resolve(tbl: string, local: Side, remote: Side, ctx: ResolveContext): Resolution { + if (local.deleted && remote.deleted) return ADOPT; + // A deliberate delete is obeyed, whichever side made it. + if (remote.deleted) return ADOPT; + if (local.deleted) return KEEP; + + const a = local.data ?? {}; + const b = remote.data ?? {}; + + switch (tbl) { + case "card": + return heavier(num(a.reps) + num(a.lapses), num(b.reps) + num(b.lapses)); + + case "evidence": + return heavier( + num(a.ok) + num(a.wrong) + num(a.lookups), + num(b.ok) + num(b.wrong) + num(b.lookups), + ); + + case "progress": { + const more = num(a.answers) > num(b.answers) ? a : b; + return settle(a, b, { + ...b, + done: Math.max(num(a.done), num(b.done)), + answers: Math.max(num(a.answers), num(b.answers)), + confidence: more.confidence ?? 0, + note: more.note ?? "", + }); + } + + case "meta": + return resolveMeta(a, b, ctx); + + // Keyed so that the same key means the same thing on every device: + // chat turns and per-device counters never genuinely conflict, and for + // a custom word's gloss, a confusion or a ledger tick the server's copy + // is as good as ours. + default: + return ADOPT; + } +} diff --git a/app/src/ui/tutor/TutorTab.tsx b/app/src/ui/tutor/TutorTab.tsx index 128d3f0..085139c 100644 --- a/app/src/ui/tutor/TutorTab.tsx +++ b/app/src/ui/tutor/TutorTab.tsx @@ -15,14 +15,16 @@ import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { useStore } from "../../state/store.js"; -import { editChatClear, editChatTrim, editChatTurn, seedChatTurn } from "../../db/writes.js"; +import { editChatClear, editChatTurn, pruneChat, seedChatTurn } from "../../db/writes.js"; import { parseMessage } from "../../domain/gloss.js"; import { assemblePrompt, gateFor, FOCUS_MODES, + UNITS, VOCAB_CAP, type BandQuery, + type FlatUnit, } from "../../domain/gate.js"; import type { FocusMode } from "../../domain/gate.js"; import { applyProgressReport, currentUnit } from "../../domain/progress.js"; @@ -88,6 +90,24 @@ async function readBandWords(db: Db, band: number, ceiling: number): Promise r.headword); } +/* ── the opening ─────────────────────────────────────────────────── */ + +/** The app's own first message — the artifact's wording. */ +function openingTurn(unit: FlatUnit): string { + const first = unit.id === UNITS[0]!.id; + return [ + "반가워요. 저는 선생님이에요 — your reading tutor.", + "", + first + ? "We're starting at the beginning, and we'll go one small step at a time. Nothing will turn up in an exercise that I haven't taught you first — if it does, tell me and I'll drop it." + : "Picking up where you left off. Nothing will turn up in an exercise that I haven't taught you first — if it does, tell me and I'll drop it.", + "", + `The plan is six phases, ${UNITS.length} units, ending with you reading a manhwa page at speed. **Roadmap** above shows the whole thing; you can jump anywhere in it whenever you like.`, + "", + `Hit **시작 · Start unit** below, and we'll pick up ${unit.id} ${unit.ko} — ${unit.name}.`, + ].join("\n"); +} + /* ── the tab ─────────────────────────────────────────────────────── */ export function TutorTab() { @@ -216,6 +236,11 @@ export function TutorTab() { focus: prefs.focus, }); const history = turns.map((t) => ({ role: t.role, content: t.body })); + // The transcript opens with the app's own assistant turn, and a + // conversation has to open with the learner — the artifact's fix. + if (history[0]?.role === "assistant") { + history.unshift({ role: "user", content: "Let's continue the lesson." }); + } try { const result = await sample( @@ -232,7 +257,7 @@ export function TutorTab() { remounted it once the read returned — the reply visibly disappeared and the log jumped by its height each time. */ await editChatTurn(db, "assistant", result.text); - await editChatTrim(db, KEEP_TURNS); + await pruneChat(db, KEEP_TURNS); const committed = await readTurns(); /* All three in one batch. Leaving `busy` set until the finally block meant one render where the reply was committed but the @@ -288,7 +313,14 @@ export function TutorTab() { /* Open the lesson if there is no transcript yet — EXACTLY ONCE. The guard is claimed synchronously, before the first await: setting it after one would let every concurrent run past it, which is precisely how - this managed to seed the opening turn three times over. */ + this managed to seed the opening turn three times over. + + The opening is written by the app, not asked of the model, and it is a + seed: unstamped, never dirty, never synced. Booting used to send "Start + unit" at once, so a fresh install wrote a stamped reply and a progress + edit before anyone could configure a server — boot-time writes that + looked like the newest work in the world. Now nothing is written until + the learner presses Start. */ const opened = useRef(false); useEffect(() => { if (opened.current) return; @@ -296,12 +328,10 @@ export function TutorTab() { void (async () => { if ((await loadTurns()) > 0) return; // a transcript already exists - // The opening turn is app-supplied, not a user edit: unstamped. - await seedChatTurn(db, "user", `Start unit ${unit.id}.`, 0); + await seedChatTurn(db, "assistant", openingTurn(unit), 0); await loadTurns(); - await sendRef.current(`Start unit ${unit.id}.`, { record: false }); })(); - // Mount only: `send` and `unit.id` are read through the ref / at mount. + // Mount only: `unit` is read at mount. // eslint-disable-next-line react-hooks/exhaustive-deps }, []); @@ -313,9 +343,12 @@ export function TutorTab() { await editChatClear(db); setRevealed(new Set()); setRecent([]); + await seedChatTurn(db, "assistant", openingTurn(unit), 0); await loadTurns(); - await sendRef.current(`Start unit ${unit.id}.`, { record: true }); - }, [db, loadTurns, unit.id]); + }, [db, loadTurns, unit]); + + /** Nothing asked yet: the opening is all there is, and Start begins it. */ + const notStarted = !turns.some((t) => t.role === "user"); /* ── the rail ── */ @@ -536,6 +569,21 @@ export function TutorTab() { + {notStarted && !busy && ( +
+ +
+ )} + {error &&
{error}
}
diff --git a/app/src/ui/tutor/tutor.css b/app/src/ui/tutor/tutor.css index 401109e..1154940 100644 --- a/app/src/ui/tutor/tutor.css +++ b/app/src/ui/tutor/tutor.css @@ -189,3 +189,10 @@ border-top: 1px solid var(--line); margin: 10px 0; } + +/* Start: the only action on a transcript that is just the app's opening. */ +.chat-start { + display: flex; + justify-content: center; + padding: 10px 0 4px; +} diff --git a/server/sql/002-protocol-2.sql b/server/sql/002-protocol-2.sql new file mode 100644 index 0000000..0b76763 --- /dev/null +++ b/server/sql/002-protocol-2.sql @@ -0,0 +1,32 @@ +-- Sync protocol 2. +-- +-- Rows are now written only by compare-and-swap on change_seq (see +-- server/src/db.ts), keyed by a JSON array of the key's values, and every +-- user's data belongs to an epoch: a random id that changes whenever the +-- server's copy is thrown away, so a client can tell it is talking to a +-- server that no longer holds what it pushed and must hydrate from scratch. + +CREATE TABLE IF NOT EXISTS sync_meta ( + k TEXT PRIMARY KEY, + v TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS sync_epoch ( + user_id TEXT PRIMARY KEY, + epoch TEXT NOT NULL +); + +-- Protocol 1 rows cannot be read as protocol 2: their keys were joined with +-- a space and every one was ordered by a client's clock. They are dropped +-- once, and each user's epoch with them — so every client re-hydrates and +-- offers its own database back, which is the copy that was always the +-- source of truth. No deployment held protocol 1 data when this shipped. +DO $$ +BEGIN + IF NOT EXISTS (SELECT 1 FROM sync_meta WHERE k = 'protocol' AND v = '2') THEN + DELETE FROM sync_row; + DELETE FROM sync_epoch; + INSERT INTO sync_meta (k, v) VALUES ('protocol', '2') + ON CONFLICT (k) DO UPDATE SET v = EXCLUDED.v; + END IF; +END $$; diff --git a/server/src/db.ts b/server/src/db.ts index 3b353bb..2975845 100644 --- a/server/src/db.ts +++ b/server/src/db.ts @@ -1,99 +1,163 @@ -/* Postgres access, and the two queries sync is made of. */ +/* Postgres access, and the two operations sync is made of. + + Protocol 2 (shared/sync-protocol.mjs). The server's whole job is order: + it assigns change_seq, it refuses a write made against a stale copy, and + it never lets a pull see half of a push. It does not read inside a row — + it stores and orders them; the client interprets them. */ import { readFile } from "node:fs/promises"; import { fileURLToPath } from "node:url"; import pg from "pg"; -export interface SyncRow { +/* The wire shapes, as in types/shared/sync-protocol.mjs.d.ts. Restated here + because the server's type check reads the .mjs itself, not its + declarations. */ + +type Value = string | number | null; + +/** A row a client pushes. `base` is the change_seq it last agreed with. */ +export interface WireRow { tbl: string; - data: Record; - updated_at: number; + pk: string; + base: number; deleted?: boolean; + data?: Record | null; + updated_at?: number; +} + +/** A row as the server holds it. */ +export interface ServerRow { + tbl: string; + pk: string; + seq: number; + deleted: boolean; + data: Record | null; + updated_at?: number; } export interface Store { - pull(userId: string, cursor: number, limit: number): Promise<{ rows: SyncRow[]; cursor: number; more: boolean }>; - push(userId: string, rows: SyncRow[]): Promise; + pull(userId: string, cursor: number, limit: number): Promise<{ epoch: string; rows: ServerRow[]; cursor: number; more: boolean }>; + push(userId: string, rows: WireRow[]): Promise<{ epoch: string; applied: { tbl: string; pk: string; seq: number }[]; conflicts: ServerRow[] }>; /** Test-only; the route that calls it exists only under HANKAN_TEST_MODE. */ reset(userId: string): Promise; close(): Promise; } -/** - * 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(" "); +type Query = { query: pg.Pool["query"] }; + +/** This user's epoch, created on first contact. */ +async function epochOf(db: Query, userId: string): Promise { + await db.query( + "INSERT INTO sync_epoch (user_id, epoch) VALUES ($1, gen_random_uuid()::text) ON CONFLICT (user_id) DO NOTHING", + [userId], + ); + const { rows } = await db.query("SELECT epoch FROM sync_epoch WHERE user_id = $1", [userId]); + return String(rows[0].epoch); } +const asServerRow = (r: Record): ServerRow => ({ + tbl: r.tbl as string, + pk: r.pk as string, + seq: Number(r.change_seq), + deleted: Boolean(r.deleted), + data: r.deleted ? null : (r.data as ServerRow["data"]), + updated_at: Number(r.updated_at), +}); + export async function openStore(connectionString: string, pkFor: (tbl: string) => string[] | null): Promise { 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); + for (const file of ["001-schema.sql", "002-protocol-2.sql"]) { + await pool.query(await readFile(fileURLToPath(new URL(`../sql/${file}`, import.meta.url)), "utf8")); + } 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 client = await pool.connect(); + try { + await client.query("BEGIN"); + /* Shared, against push's exclusive lock on the same key. Without it a + push that took seq 10 but committed after another took seq 11 could + be passed over: a pull that saw 11 moves the cursor beyond 10, and + nothing would ever ask for 10 again. */ + await client.query("SELECT pg_advisory_xact_lock_shared(hashtext($1))", [userId]); + const epoch = await epochOf(client, userId); + // One extra row tells us whether another page exists without a count. + const { rows } = await client.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], + ); + await client.query("COMMIT"); - 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), - updated_at: Number(r.updated_at), - ...(r.deleted ? { deleted: true } : {}), - })), - cursor: page.length ? Number(page[page.length - 1].change_seq) : cursor, - more, - }; + const more = rows.length > limit; + const page = (more ? rows.slice(0, limit) : rows).map(asServerRow); + return { epoch, rows: page, cursor: page.length ? page[page.length - 1]!.seq : cursor, more }; + } catch (err) { + await client.query("ROLLBACK"); + throw err; + } finally { + client.release(); + } }, async push(userId, rows) { const client = await pool.connect(); try { await client.query("BEGIN"); + // One push at a time per user: compare-and-swap needs a stable read. + await client.query("SELECT pg_advisory_xact_lock(hashtext($1))", [userId]); + const epoch = await epochOf(client, userId); + + const applied: { tbl: string; pk: string; seq: number }[] = []; + const conflicts: ServerRow[] = []; for (const row of rows) { - const pkCols = pkFor(row.tbl); - if (!pkCols) continue; // a table this server does not know about + if (!pkFor(row.tbl) || typeof row.pk !== "string") continue; // a table this server does not know + const base = Number(row.base) || 0; - // 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( + const { rows: current } = await client.query( + `SELECT tbl, pk, data, updated_at, deleted, change_seq FROM sync_row + WHERE user_id = $1 AND tbl = $2 AND pk = $3 FOR UPDATE`, + [userId, row.tbl, row.pk], + ); + const cur = current[0] as Record | undefined; + + /* Gate 2. The write stands only if the sender last agreed with the + row the server holds now — or neither has ever seen it. Anything + else goes back as a conflict, with the server's copy, and the + client settles it. No clock is consulted anywhere. */ + const agreed = cur ? Number(cur.change_seq) === base : base === 0; + if (!agreed) { + conflicts.push( + cur ? asServerRow(cur) : { tbl: row.tbl, pk: row.pk, seq: 0, deleted: true, data: null }, + ); + continue; + } + + const { rows: written } = 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], + SET data = EXCLUDED.data, updated_at = EXCLUDED.updated_at, deleted = EXCLUDED.deleted + RETURNING change_seq`, + [ + userId, + row.tbl, + row.pk, + JSON.stringify(row.deleted ? {} : (row.data ?? {})), + Number(row.updated_at) || 0, + Boolean(row.deleted), + ], ); + applied.push({ tbl: row.tbl, pk: row.pk, seq: Number(written[0].change_seq) }); } - 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); + return { epoch, applied, conflicts }; } catch (err) { await client.query("ROLLBACK"); throw err; @@ -104,6 +168,7 @@ export async function openStore(connectionString: string, pkFor: (tbl: string) = async reset(userId) { await pool.query("DELETE FROM sync_row WHERE user_id = $1", [userId]); + await pool.query("DELETE FROM sync_epoch WHERE user_id = $1", [userId]); }, async close() { diff --git a/server/src/main.ts b/server/src/main.ts index 3b61657..d91a519 100644 --- a/server/src/main.ts +++ b/server/src/main.ts @@ -7,8 +7,8 @@ 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 { pkFor, PAGE_SIZE, PROTOCOL, PROTOCOL_HEADER } from "../../shared/sync-protocol.mjs"; +import { openStore, type WireRow } from "./db.ts"; import { tutorRoute } from "./tutor.ts"; const PORT = Number(process.env.PORT ?? 8787); @@ -71,7 +71,7 @@ app.use("*", async (c, next) => { return c.body(null, 204, { "Access-Control-Allow-Origin": allow, "Access-Control-Allow-Methods": "GET, POST, OPTIONS", - "Access-Control-Allow-Headers": "authorization, content-type", + "Access-Control-Allow-Headers": `authorization, content-type, ${PROTOCOL_HEADER}`, "Access-Control-Max-Age": "86400", Vary: "Origin", }); @@ -102,6 +102,17 @@ app.get("/health", (c) => c.json({ ok: true })); /* ── sync ────────────────────────────────────────────────────────────── */ +/* Refuse any client not speaking this protocol. A client on protocol 1 is + exactly the device the artifact lost data to — an old build that pushes + before it pulls and trusts its own clock — so it gets a clear 426 rather + than a half-understood write. */ +app.use("/api/sync", async (c, next) => { + if (c.req.header(PROTOCOL_HEADER) !== String(PROTOCOL)) { + return c.json({ error: "this server speaks sync protocol 2 — update the app", protocol: PROTOCOL }, 426); + } + await next(); +}); + 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); @@ -109,10 +120,10 @@ app.get("/api/sync", async (c) => { }); app.post("/api/sync", async (c) => { - const body = (await c.req.json()) as { rows?: SyncRow[] }; + const body = (await c.req.json()) as { rows?: WireRow[] }; 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) }); + if (rows.length > PAGE_SIZE) return c.json({ error: "too many rows" }, 413); + return c.json(await store.push(USER_ID, rows)); }); /* Wipes this user's rows. Exists only when HANKAN_TEST_MODE is set, so it diff --git a/shared/sync-protocol.mjs b/shared/sync-protocol.mjs index 954f92c..e0ba926 100644 --- a/shared/sync-protocol.mjs +++ b/shared/sync-protocol.mjs @@ -1,18 +1,41 @@ /* The sync wire format, defined once and imported by both sides. - Row-level, last-write-wins on `updated_at`, cursor-based on a - server-assigned `change_seq`. One user, so the loser of a conflict is at - worst one SRS grade — LWW is adequate, and a merge policy would be - over-engineering. + Protocol 2. Row-level and cursor-based on a server-assigned change_seq, + as PORT.md specifies — and ordered by that counter, never by a clock. - THE RULE THAT MATTERS: seeded and defaulted rows carry updated_at = 0. - The artifact's sync bug was a fresh device stamping its own empty state - as newer than the server's real history and clobbering it. Here a seed - row can never be dirty (0 is never greater than any watermark) and can - never win a conflict (0 is never greater than any timestamp). The bug is - not avoided, it is unrepresentable. */ + ════ THE THREE GATES ════════════════════════════════════════════════ -/** Tables that sync, and the columns forming each primary key. */ + The artifact lost a week of a real user's work without these, and the + port's first sync broke all three. In short (lib/sync.js has the long + version, applied there to whole documents): + + 1. HYDRATION. A device pulls everything the server holds before it may + push anything. A laptop with a week-old copy used to boot, re-stamp + that copy through a migration, and push it over the phone's week. + + 2. A COUNTER, NOT A CLOCK. Every row remembers the change_seq it last + agreed with the server on (base_seq). The server applies a write only + if that is still the row's current change_seq — compare-and-swap — + and otherwise hands the current row back as a conflict. A phone clock + four minutes fast used to decide which of two grades survived. + + 3. NO SILENT SHRINKING. A conflict is settled by what each copy HOLDS, + not by which is newer (see app/src/sync/resolve.ts). Emptying things + on purpose is explicit: a reset or a cleared lesson writes a marker + every other device obeys, so the shrink is declared, never raced. + + Seeded and defaulted rows are never dirty, so they never leave the + device: a fresh install cannot claim its empty defaults are news. */ + +export const PROTOCOL = 2; + +/** Header the client sends; the server refuses any other protocol. */ +export const PROTOCOL_HEADER = "x-hankan-protocol"; + +/** Maximum rows in one push or pull page. */ +export const PAGE_SIZE = 500; + +/** Tables that sync: their primary-key columns and the columns that travel. */ export const SYNC_TABLES = { card: { pk: ["lemma_id"], @@ -21,8 +44,31 @@ export const SYNC_TABLES = { progress: { pk: ["unit_id"], cols: ["unit_id", "done", "confidence", "answers", "note"] }, chat: { pk: ["id"], cols: ["id", "role", "body", "created_at"] }, meta: { pk: ["k"], cols: ["k", "v"] }, - study_log: { pk: ["day"], cols: ["day", "reviews", "correct", "drills"] }, - peek: { pk: ["form"], cols: ["form", "count"] }, + /* Counters, kept per device and summed on read. Two devices each adding + a review to the same day would otherwise both write "reviews = n+1", + and whichever landed second would erase the other's review. A device + only ever writes its own row, so these never conflict at all. */ + study_log: { pk: ["day", "device"], cols: ["day", "device", "reviews", "correct", "drills"] }, + peek: { pk: ["form", "device"], cols: ["form", "device", "count"] }, + custom_word: { pk: ["headword", "pos"], cols: ["headword", "pos", "gloss"] }, + evidence: { + pk: ["word"], + cols: [ + "word", + "ok", + "wrong", + "lookups", + "streak", + "first_round", + "last_round", + "last_seen", + "rounds", + "first_ok_round", + "last_ok_round", + ], + }, + confusion: { pk: ["word"], cols: ["word", "mistook"] }, + phase_ledger: { pk: ["phase", "kind", "item"], cols: ["phase", "kind", "item", "confirmed"] }, }; export const SYNC_TABLE_NAMES = Object.keys(SYNC_TABLES); @@ -31,39 +77,67 @@ export const SYNC_TABLE_NAMES = Object.keys(SYNC_TABLES); export const pkFor = (tbl) => Object.prototype.hasOwnProperty.call(SYNC_TABLES, tbl) ? SYNC_TABLES[tbl].pk : null; +/** + * A row's key on the wire: its primary-key values as a JSON array. + * + * Protocol 1 joined them with a space and split them back on one, so a key + * containing a space — the curriculum's 몇 명 and 신경 쓰다 can be peek + * forms — came back as two values and the statement binding them threw. + * The pull stopped at that row, on every device, permanently. + */ +export const encodePk = (tbl, row) => JSON.stringify(SYNC_TABLES[tbl].pk.map((c) => row[c])); +export const decodePk = (pk) => JSON.parse(pk); + /** * `meta` mixes the learner's data with bookkeeping that describes one * install. Only the former may cross the wire. * - * `dict.loadedBands` is the dangerous one: replicating it would tell a - * phone that had loaded bands 0-2 that it holds every row the desktop has, - * and the word rail would then fail to find words it believes are present. - * `schema_version` would be worse — a device could be told it has run a - * migration it has not. `server.*` holds this device's endpoint and bearer - * token: syncing a token through the endpoint it authenticates would be - * circular, and a base URL is network-specific. + * `dict.*` is the dangerous one: replicating the loaded-band record would + * tell a phone it holds rows it never downloaded. `schema_version` would be + * worse — a device could be told it has run a migration it has not. + * `server.*` holds this device's endpoint and bearer token, and `sync.*` + * is this device's own position in the conversation with the server. */ const SYNCABLE_META_EXACT = new Set(["grammar.learned", "grammar.notes", "trainer.conjugation"]); -const SYNCABLE_META_PREFIX = ["prefs."]; +const SYNCABLE_META_PREFIX = ["prefs.", "road.", "learner.", "reset."]; export function isSyncableMetaKey(key) { if (SYNCABLE_META_EXACT.has(key)) return true; return SYNCABLE_META_PREFIX.some((p) => key.startsWith(p)); } -/** Stable string key for a row, used to match it across devices. */ -export const rowKey = (table, row) => SYNC_TABLES[table].pk.map((c) => String(row[c])).join(" "); - -/** Rows a device may send: dirty since its last successful push. */ -export const isDirty = (row, pushedAt) => Number(row.updated_at) > Number(pushedAt); - /** - * Last-write-wins. Strictly greater, so equal timestamps leave the local row - * alone — a tie means both sides already agree, or the clocks are close - * enough that flapping would be worse than either outcome. + * What each deliberate shrink empties. A marker meta key, `reset.`, + * holds a counter; a device that sees it rise empties the same things + * locally — including its own unsynced edits, because a reset it had not + * yet heard of still wins over work done on the old data. */ -export const incomingWins = (incomingUpdatedAt, localUpdatedAt) => - Number(incomingUpdatedAt) > Number(localUpdatedAt ?? -1); +export const RESET_SCOPES = { + chat: { tables: ["chat"], meta: [] }, + roadmap: { tables: ["progress", "chat", "phase_ledger"], meta: ["road."] }, + everything: { + tables: [ + "progress", + "chat", + "phase_ledger", + "card", + "study_log", + "peek", + "custom_word", + "evidence", + "confusion", + ], + meta: ["road.", "prefs.", "grammar.", "trainer.", "learner."], + }, +}; -/** Maximum rows in one push or pull page. */ -export const PAGE_SIZE = 500; +/** Does a reset of this scope empty this row? */ +export function resetCovers(scope, tbl, row) { + const s = RESET_SCOPES[scope]; + if (!s) return false; + if (tbl === "meta") { + const k = String(row.k ?? ""); + return !k.startsWith("reset.") && s.meta.some((p) => k.startsWith(p)); + } + return s.tables.includes(tbl); +} diff --git a/test/db/conformance.ts b/test/db/conformance.ts index 749adaa..16d04ad 100644 --- a/test/db/conformance.ts +++ b/test/db/conformance.ts @@ -14,6 +14,7 @@ import { seedMeta, seedProgress, editCard, + editCardReset, editCurrentUnit, editMeta, editPeek, @@ -201,6 +202,44 @@ export function conformanceSuite(name: string, open: () => Promise): void { newer than the server's real history. ═══════════════════════════════════════════════════════════════ */ describe("seeded state carries no write timestamp", () => { + it("leaves nothing dirty after a full seed, so nothing seeded ever travels", async () => { + await seedProgress(db, "1.1"); + await seedMeta(db, "prefs.newPerDay", "10"); + await db.run( + "INSERT INTO lemma (id, headword, pos, gloss_en, source) VALUES (1,'밥','noun','rice','curated')", + ); + await seedCard(db, 1, newCard()); + await seedChatTurn(db, "assistant", "안녕! 시작하자.", 0); + + for (const table of SYNCABLE) { + const row = await db.get<{ n: number }>(`SELECT count(*) AS n FROM ${table} WHERE dirty != 0`); + expect(row?.n, `${table} has a dirty seed row`).toBe(0); + } + }); + + it("marks every genuine edit dirty and bumps its revision each time", async () => { + await db.run( + "INSERT INTO lemma (id, headword, pos, gloss_en, source) VALUES (3,'책','noun','book','curated')", + ); + await editCard(db, 3, newCard()); + await editCard(db, 3, grade(newCard(), GOOD, 20_000)); + expect(await db.get("SELECT dirty, rev FROM card WHERE lemma_id = 3")).toEqual({ dirty: 1, rev: 2 }); + }); + + it("carries a deleted row's base_seq into its tombstone", async () => { + await db.run( + "INSERT INTO lemma (id, headword, pos, gloss_en, source) VALUES (4,'꽃','noun','flower','curated')", + ); + await editCard(db, 4, newCard()); + await db.run("UPDATE card SET base_seq = 77, dirty = 0 WHERE lemma_id = 4"); + await editCardReset(db, 4); + expect(await db.get("SELECT pk, base_seq, dirty FROM tombstone WHERE tbl = 'card'")).toEqual({ + pk: "[4]", + base_seq: 77, + dirty: 1, + }); + }); + it("leaves updated_at at 0 across every syncable table after a full seed", async () => { await seedProgress(db, "1.1"); await seedMeta(db, "focus", "auto"); diff --git a/test/db/migration-6.test.ts b/test/db/migration-6.test.ts index 3ba8205..2ed991a 100644 --- a/test/db/migration-6.test.ts +++ b/test/db/migration-6.test.ts @@ -57,7 +57,8 @@ describe("migration 6", () => { const graves = await db.all<{ pk: string; updated_at: number }>( "SELECT pk, updated_at FROM tombstone WHERE tbl = 'card'", ); - expect(graves).toEqual([{ pk: String(lemmaId("책", "noun")), updated_at: 333 }]); + // As JSON, once migration 9 has written keys in protocol 2's form. + expect(graves).toEqual([{ pk: JSON.stringify([lemmaId("책", "noun")]), updated_at: 333 }]); }); it("turns a custom lemma into a custom_word, stamped when its card was made", async () => { diff --git a/test/db/migration-8.test.ts b/test/db/migration-8.test.ts index e28ffa1..1dcad62 100644 --- a/test/db/migration-8.test.ts +++ b/test/db/migration-8.test.ts @@ -60,7 +60,8 @@ describe("migration 8", () => { it("renames a chat tombstone the same way", async () => { const device = (await db.get<{ v: string }>("SELECT v FROM meta WHERE k = 'sync.device'"))!.v; expect(await db.all("SELECT pk, updated_at FROM tombstone WHERE tbl = 'chat'")).toEqual([ - { pk: `legacy:${device}:0000000009`, updated_at: 400 }, + // As JSON, once migration 9 has written keys in protocol 2's form. + { pk: JSON.stringify([`legacy:${device}:0000000009`]), updated_at: 400 }, ]); }); diff --git a/test/db/migration-9.test.ts b/test/db/migration-9.test.ts new file mode 100644 index 0000000..5fc3ae1 --- /dev/null +++ b/test/db/migration-9.test.ts @@ -0,0 +1,71 @@ +/* Migration 9 — sync protocol 2's columns, per-device counters, JSON keys. + + A database as migration 8 left it, carrying protocol 1's bookkeeping and + a tombstone of every shape protocol 1 wrote. */ + +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import { SqliteWasmDb } from "@app/db/sqlite-wasm-core.js"; +import { migrate } from "@app/db/migrate.js"; +import type { Db } from "@app/db/types.js"; + +let db: Db; +let device: string; +beforeEach(async () => { + db = await SqliteWasmDb.open({ memory: true }); + await migrate(db, 8); + device = (await db.get<{ v: string }>("SELECT v FROM meta WHERE k = 'sync.device'"))!.v; + await db.exec(` + INSERT INTO study_log (day, reviews, correct, drills, updated_at) VALUES (20000, 5, 4, 1, 700); + INSERT INTO peek (form, count, updated_at) VALUES ('몇 명', 3, 800); + INSERT INTO card (lemma_id, state, reps, updated_at) VALUES (55855054575946, 1, 2, 900); + INSERT INTO meta (k, v, updated_at) VALUES ('sync.cursor', '41', 0), ('sync.pushedAt', '900', 0); + INSERT INTO tombstone (tbl, pk, updated_at) VALUES + ('card', '1434356355562999', 10), + ('study_log', '19999', 11), + ('peek', '몇 명', 12), + ('meta', 'prefs.goal', 13), + ('custom_word', '["던전","noun"]', 14); + `); + await migrate(db); +}); +afterEach(async () => { + await db.close(); +}); + +describe("migration 9", () => { + it("gives every syncable row base_seq, dirty and rev, all zero", async () => { + expect(await db.get("SELECT base_seq, dirty, rev, updated_at FROM card")).toEqual({ + base_seq: 0, + dirty: 0, + rev: 0, + updated_at: 900, + }); + }); + + it("hands this install's counters to this install", async () => { + expect(await db.get("SELECT day, device, reviews, correct, drills, updated_at FROM study_log")).toEqual({ + day: 20000, + device, + reviews: 5, + correct: 4, + drills: 1, + updated_at: 700, + }); + expect(await db.get("SELECT form, device, count FROM peek")).toEqual({ form: "몇 명", device, count: 3 }); + }); + + it("writes every tombstone key as a JSON array, a space included", async () => { + const graves = await db.all<{ tbl: string; pk: string }>("SELECT tbl, pk FROM tombstone ORDER BY updated_at"); + expect(graves).toEqual([ + { tbl: "card", pk: "[1434356355562999]" }, + { tbl: "study_log", pk: JSON.stringify([19999, device]) }, + { tbl: "peek", pk: JSON.stringify(["몇 명", device]) }, + { tbl: "meta", pk: '["prefs.goal"]' }, + { tbl: "custom_word", pk: '["던전","noun"]' }, + ]); + }); + + it("drops protocol 1's position, so the next sync hydrates from scratch", async () => { + expect(await db.all("SELECT k FROM meta WHERE k IN ('sync.cursor', 'sync.pushedAt')")).toEqual([]); + }); +}); diff --git a/test/server/http.test.ts b/test/server/http.test.ts index 1068598..439a6a6 100644 --- a/test/server/http.test.ts +++ b/test/server/http.test.ts @@ -27,13 +27,16 @@ run("CORS", () => { headers: { origin: PHONE_ORIGIN, "access-control-request-method": "POST", - "access-control-request-headers": "authorization,content-type", + "access-control-request-headers": "authorization,content-type,x-hankan-protocol", }, }); expect(res.status).not.toBe(401); expect(res.status).toBeLessThan(300); expect(res.headers.get("access-control-allow-origin")).toBe(PHONE_ORIGIN); expect(res.headers.get("access-control-allow-headers")).toMatch(/authorization/i); + // Without this the phone's browser refuses to send the protocol header, + // and every sync from it is answered 426. + expect(res.headers.get("access-control-allow-headers")).toMatch(/x-hankan-protocol/i); }); it("allows the Authorization header, without which the token cannot be sent", async () => { @@ -46,7 +49,7 @@ run("CORS", () => { it("puts the allow-origin header on the real response too", async () => { const res = await fetch(`${BASE}/api/sync?cursor=0`, { - headers: { origin: PHONE_ORIGIN, authorization: `Bearer ${TOKEN}` }, + headers: { origin: PHONE_ORIGIN, authorization: `Bearer ${TOKEN}`, "x-hankan-protocol": "2" }, }); expect(res.status).toBe(200); expect(res.headers.get("access-control-allow-origin")).toBe(PHONE_ORIGIN); diff --git a/test/sync/resolve.test.ts b/test/sync/resolve.test.ts new file mode 100644 index 0000000..4fbde14 --- /dev/null +++ b/test/sync/resolve.test.ts @@ -0,0 +1,118 @@ +/* Gate 3 — what happens when two copies of one row disagree. + + The rule is PORT.md's: the copy that holds more wins, a tie goes to the + server so every device lands on one copy, and a deliberate delete is + obeyed. These pin each table's reading of "more". */ + +import { describe, it, expect } from "vitest"; +import { resolve, type Resolution, type Side } from "@app/sync/resolve.js"; + +/** The merged meta value, parsed. */ +const mergedValue = (r: Resolution): unknown => { + if (r.kind !== "merge") throw new Error(`expected a merge, got ${r.kind}`); + return JSON.parse(String(r.data.v)); +}; + +const ctx = { unitIndex: (id: string) => ["1.1", "1.2", "1.3", "1.10", "2.1"].indexOf(id) }; +const live = (data: Record): Side => ({ deleted: false, data }); +const gone: Side = { deleted: true, data: null }; + +describe("deletes", () => { + it("obeys a delete from the server over an edit made here", () => { + expect(resolve("card", live({ reps: 9 }), gone, ctx)).toEqual({ kind: "adopt" }); + }); + + it("keeps a delete made here over an edit on the server", () => { + expect(resolve("card", gone, live({ reps: 9 }), ctx)).toEqual({ kind: "keep" }); + }); +}); + +describe("cards and evidence — more work wins", () => { + it("keeps the card with more reviews behind it", () => { + expect(resolve("card", live({ reps: 6, lapses: 1 }), live({ reps: 5, lapses: 0 }), ctx)).toEqual({ kind: "keep" }); + expect(resolve("card", live({ reps: 2, lapses: 0 }), live({ reps: 5, lapses: 0 }), ctx)).toEqual({ kind: "adopt" }); + }); + + it("gives a tie to the server", () => { + expect(resolve("card", live({ reps: 5, lapses: 0 }), live({ reps: 5, lapses: 0 }), ctx)).toEqual({ kind: "adopt" }); + }); + + it("weighs evidence by everything observed, lookups included", () => { + expect( + resolve("evidence", live({ ok: 2, wrong: 1, lookups: 2 }), live({ ok: 3, wrong: 0, lookups: 0 }), ctx), + ).toEqual({ kind: "keep" }); + }); +}); + +describe("progress — merged, never un-finished", () => { + it("keeps a unit finished if either copy finished it", () => { + const r = resolve( + "progress", + live({ unit_id: "1.1", done: 1, confidence: 60, answers: 4, note: "mine" }), + live({ unit_id: "1.1", done: 0, confidence: 90, answers: 7, note: "theirs" }), + ctx, + ); + expect(r).toEqual({ + kind: "merge", + data: { unit_id: "1.1", done: 1, confidence: 90, answers: 7, note: "theirs" }, + }); + }); + + it("takes confidence from the copy with more answers behind it", () => { + expect( + resolve( + "progress", + live({ unit_id: "1.1", done: 0, confidence: 95, answers: 9, note: "a" }), + live({ unit_id: "1.1", done: 0, confidence: 40, answers: 3, note: "b" }), + ctx, + ), + ).toEqual({ kind: "keep" }); + }); +}); + +describe("meta", () => { + it("puts him on the further unit, by roadmap order and not by string", () => { + // "1.10" sorts before "1.2" as text; on the roadmap it comes after. + expect(resolve("meta", live({ k: "road.unit", v: "1.10" }), live({ k: "road.unit", v: "1.2" }), ctx)).toEqual({ + kind: "keep", + }); + }); + + it("never moves a counter backwards", () => { + expect(resolve("meta", live({ k: "learner.round", v: "12" }), live({ k: "learner.round", v: "9" }), ctx)).toEqual({ + kind: "keep", + }); + expect(resolve("meta", live({ k: "reset.chat", v: "1" }), live({ k: "reset.chat", v: "2" }), ctx)).toEqual({ + kind: "adopt", + }); + }); + + it("unites grammar flags learned on two devices", () => { + const r = resolve( + "meta", + live({ k: "grammar.learned", v: JSON.stringify({ a: true }) }), + live({ k: "grammar.learned", v: JSON.stringify({ b: true }) }), + ctx, + ); + expect(mergedValue(r)).toEqual({ a: true, b: true }); + }); + + it("keeps the longer of two versions of one note", () => { + const r = resolve( + "meta", + live({ k: "grammar.notes", v: JSON.stringify({ p1: "a long careful note" }) }), + live({ k: "grammar.notes", v: JSON.stringify({ p1: "short", p2: "other" }) }), + ctx, + ); + expect(mergedValue(r)).toEqual({ + p1: "a long careful note", + p2: "other", + }); + }); + + it("takes the server's copy of a preference", () => { + expect(resolve("meta", live({ k: "prefs.goal", v: "40" }), live({ k: "prefs.goal", v: "20" }), ctx)).toEqual({ + kind: "adopt", + }); + }); +}); diff --git a/test/sync/roundtrip.test.ts b/test/sync/roundtrip.test.ts index 7f1a0fa..6df6992 100644 --- a/test/sync/roundtrip.test.ts +++ b/test/sync/roundtrip.test.ts @@ -1,21 +1,21 @@ -/* Two devices, one server. +/* Two devices, one server — protocol 2, and the three gates. - Runs against a real Postgres and a real server — a fake would not exercise - the trigger that advances change_seq, the LWW clause in the upsert, or - the cursor paging, which is where sync actually goes wrong. + Runs against a real Postgres and a real server: a fake would not exercise + the compare-and-swap, the advisory locks or the change_seq trigger, which + is where sync actually goes wrong. Skipped unless HANKAN_TEST_SERVER is set, so the suite still runs on a machine without Docker: - docker run -d --name hankan-pg-test -e POSTGRES_PASSWORD=test \ - -e POSTGRES_DB=hankan -p 55432:5432 postgres:16-alpine - DATABASE_URL=postgres://postgres:test@localhost:55432/hankan \ - HANKAN_TOKEN=test-token PORT=8788 \ - node --experimental-strip-types server/src/main.ts + docker run -d --rm --name hankan-pg-test -e POSTGRES_PASSWORD=test \ + -e POSTGRES_DB=hankan -p 127.0.0.1:55433:5432 postgres:16-alpine + DATABASE_URL=postgres://postgres:test@127.0.0.1:55433/hankan \ + HANKAN_TOKEN=test-token HANKAN_TEST_MODE=1 HANKAN_TUTOR_BACKEND=echo \ + PORT=8788 node --experimental-strip-types server/src/main.ts HANKAN_TEST_SERVER=http://localhost:8788 npm test */ -import { describe, it, expect, beforeAll, beforeEach } from "vitest"; +import { describe, it, expect, beforeAll, beforeEach, afterEach, vi } from "vitest"; import { SqliteWasmDb } from "@app/db/sqlite-wasm-core.js"; import { migrate } from "@app/db/migrate.js"; @@ -23,190 +23,326 @@ import type { Db } from "@app/db/types.js"; import { editCard, editCardReset, + editChatClear, + editChatTurn, editMeta, + editPeek, + editReset, + editStudyLog, editUnitConfidence, + editUnitDone, seedCard, seedMeta, seedProgress, } from "@app/db/writes.js"; import { syncOnce, type SyncConfig } from "@app/sync/client.js"; +import { unitIndex } from "@app/domain/gate.js"; import { newCard } from "@lib/srs.js"; +import { lemmaId } from "@shared/lemma-id.mjs"; +import { PROTOCOL_HEADER } from "@shared/sync-protocol.mjs"; const BASE = process.env.HANKAN_TEST_SERVER; const TOKEN = process.env.HANKAN_TEST_TOKEN ?? "test-token"; - const suite = BASE ? describe : describe.skip; -/** A fresh device: its own database, migrated, with the dictionary rows it - needs to hang cards off. */ +const RICE = lemmaId("밥", "noun"); +const SCHOOL = lemmaId("학교", "noun"); +const WATER = lemmaId("물", "noun"); + +/** A fresh device: its own database, migrated, with lemmas to hang cards on. */ async function device(): Promise { const db = await SqliteWasmDb.open({ memory: true }); await migrate(db); - await db.exec(` - INSERT INTO lemma (id, headword, pos, gloss_en, source) VALUES - (1, '밥', 'noun', 'rice', 'curated'), - (2, '학교', 'noun', 'school', 'curated'), - (3, '물', 'noun', 'water', 'curated'); - `); + await db.run( + `INSERT INTO lemma (id, headword, pos, gloss_en, source) VALUES + (?, '밥', 'noun', 'rice', 'curated'), (?, '학교', 'noun', 'school', 'curated'), + (?, '물', 'noun', 'water', 'curated')`, + [RICE, SCHOOL, WATER], + ); return db; } -/** Wipe the server between tests so each starts from an empty history. */ -async function resetServer(cfg: SyncConfig): Promise { - await fetch(`${cfg.baseUrl}/api/test/reset`, { - method: "POST", - headers: { authorization: `Bearer ${cfg.token}` }, - }); -} +const one = async (db: Db, sql: string, params: (string | number)[] = []) => db.get(sql, params); +const turns = async (db: Db) => + (await db.all<{ body: string }>("SELECT body FROM chat ORDER BY created_at, id")).map((t) => t.body); -suite("sync — two devices, one server", () => { +suite("sync — protocol 2, two devices, one server", () => { let cfg: SyncConfig; + const sync = (db: Db) => syncOnce(db, cfg, { unitIndex }); + const open: Db[] = []; + const make = async () => { + const db = await device(); + open.push(db); + return db; + }; beforeAll(() => { cfg = { baseUrl: BASE!, token: TOKEN }; }); - beforeEach(async () => { - await resetServer(cfg); + await fetch(`${cfg.baseUrl}/api/test/reset`, { + method: "POST", + headers: { authorization: `Bearer ${cfg.token}` }, + }); + }); + afterEach(async () => { + vi.restoreAllMocks(); + while (open.length) await open.pop()!.close(); }); it("carries an edit from one device to the other", async () => { - const a = await device(); - const b = await device(); - + const a = await make(); + const b = await make(); await editUnitConfidence(a, "1.1", 64); - await syncOnce(a, cfg); - await syncOnce(b, cfg); - - const row = await b.get<{ confidence: number }>( - "SELECT confidence FROM progress WHERE unit_id = '1.1'", - ); - expect(row?.confidence).toBe(64); - - await a.close(); - await b.close(); + await sync(a); + await sync(b); + expect(await one(b, "SELECT confidence FROM progress WHERE unit_id = '1.1'")).toEqual({ confidence: 64 }); }); - /* THE ARTIFACT'S BUG, as an executable test. - - A fresh device stamped its own empty defaults as newer than the - server's real history and clobbered it. Here the seed rows carry - updated_at = 0, so they are never dirty and never win — the device - converges onto the server's data instead of destroying it. */ - it("a fresh device's seeded state never overwrites real progress", async () => { - const a = await device(); + /* The artifact's first bug: a fresh device's defaults clobbering real + progress. Seeds are never dirty, so they never leave the device. */ + it("never lets a fresh device's seeded state overwrite real progress", async () => { + const a = await make(); await editUnitConfidence(a, "1.1", 88); - await editCard(a, 1, { ...newCard(), state: 2, interval: 21, due: 20_050, reps: 9 }); - await syncOnce(a, cfg); + await editMeta(a, "road.unit", "1.3"); + await editCard(a, RICE, { ...newCard(), state: 2, interval: 21, due: 20_050, reps: 9 }); + await sync(a); - // A brand-new device does exactly what boot does: seeds defaults. - const fresh = await device(); + const fresh = await make(); await seedProgress(fresh, "1.1"); await seedMeta(fresh, "prefs.newPerDay", "10"); - await seedCard(fresh, 1, newCard()); + await seedCard(fresh, RICE, newCard()); + const result = await sync(fresh); + expect(result.pushed, "a seed is not news").toBe(0); - const seeded = await fresh.get<{ n: number }>( - "SELECT count(*) AS n FROM progress WHERE updated_at != 0", - ); - expect(seeded?.n, "seed rows must carry no write timestamp").toBe(0); - - await syncOnce(fresh, cfg); - - // The fresh device adopts the real data… - const onFresh = await fresh.get<{ confidence: number }>( - "SELECT confidence FROM progress WHERE unit_id = '1.1'", - ); - expect(onFresh?.confidence).toBe(88); - - // …and the original device still has it after a round trip. - await syncOnce(a, cfg); - const onA = await a.get<{ confidence: number }>( - "SELECT confidence FROM progress WHERE unit_id = '1.1'", - ); - expect(onA?.confidence, "the server's real history survived").toBe(88); - - const card = await a.get<{ interval: number }>("SELECT interval FROM card WHERE lemma_id = 1"); - expect(card?.interval).toBe(21); - - await a.close(); - await fresh.close(); + expect(await one(fresh, "SELECT v FROM meta WHERE k = 'road.unit'")).toEqual({ v: "1.3" }); + expect(await one(fresh, "SELECT interval FROM card WHERE lemma_id = ?", [RICE])).toEqual({ interval: 21 }); + await sync(a); + expect(await one(a, "SELECT confidence FROM progress WHERE unit_id = '1.1'")).toEqual({ confidence: 88 }); }); - it("last write wins on a genuine conflict", async () => { - const a = await device(); - const b = await device(); + /* 14 Sep. A laptop with a week-old copy booted, re-stamped that copy, and + pushed it over a week of phone work. Gate 1: it pulls first, so the + phone's week arrives before anything leaves the laptop — and the laptop's + stale re-stamp meets the week as a conflict, which the week wins. */ + it("hydrates before it pushes: a week-old laptop cannot overwrite a week of work", async () => { + const laptop = await make(); + const phone = await make(); + await editUnitConfidence(laptop, "1.1", 40); + await sync(laptop); + await sync(phone); - await editUnitConfidence(a, "2.1", 30); - await syncOnce(a, cfg); - await syncOnce(b, cfg); + // A week on the phone. + await editUnitDone(phone, "1.1"); + await editUnitConfidence(phone, "1.1", 91); + await editMeta(phone, "road.unit", "1.2"); + await editChatTurn(phone, "user", "a week of lessons"); + await editCard(phone, WATER, { ...newCard(), state: 2, interval: 8, due: 20_010, reps: 6 }); + await sync(phone); - // b edits later, so b wins. - await new Promise((r) => setTimeout(r, 5)); - await editUnitConfidence(b, "2.1", 70); - await syncOnce(b, cfg); - await syncOnce(a, cfg); + // The laptop boots and a stale write stamps the old data as new. + await editUnitConfidence(laptop, "1.1", 40); + await editMeta(laptop, "road.unit", "1.1"); + await sync(laptop); + await sync(phone); - const onA = await a.get<{ confidence: number }>( - "SELECT confidence FROM progress WHERE unit_id = '2.1'", - ); - expect(onA?.confidence).toBe(70); - - await a.close(); - await b.close(); + for (const d of [phone, laptop]) { + expect(await one(d, "SELECT done, confidence FROM progress WHERE unit_id = '1.1'")).toEqual({ + done: 1, + confidence: 91, + }); + expect(await one(d, "SELECT v FROM meta WHERE k = 'road.unit'")).toEqual({ v: "1.2" }); + expect(await turns(d)).toEqual(["a week of lessons"]); + expect(await one(d, "SELECT reps FROM card WHERE lemma_id = ?", [WATER])).toEqual({ reps: 6 }); + } }); - it("propagates a delete through a tombstone", async () => { - const a = await device(); - const b = await device(); + /* Gate 2. Protocol 1 compared timestamps, so a device whose clock ran an + hour fast won every conflict for an hour. The counter does not care. */ + it("orders edits by the server's counter, not by either device's clock", async () => { + const fast = await make(); + const slow = await make(); + const real = Date.now(); - await editCard(a, 2, { ...newCard(), state: 2, interval: 10 }); - await syncOnce(a, cfg); - await syncOnce(b, cfg); - expect(await b.get("SELECT lemma_id FROM card WHERE lemma_id = 2")).toBeDefined(); + vi.spyOn(Date, "now").mockReturnValue(real + 60 * 60_000); + await editMeta(fast, "prefs.goal", "40"); + vi.restoreAllMocks(); + await sync(fast); + await sync(slow); - // Without a tombstone the delete is invisible and b would push the card - // back on its next turn. - await new Promise((r) => setTimeout(r, 5)); - await editCardReset(a, 2); - await syncOnce(a, cfg); - await syncOnce(b, cfg); + // Later in fact, earlier by its own clock. + vi.spyOn(Date, "now").mockReturnValue(real); + await editMeta(slow, "prefs.goal", "25"); + vi.restoreAllMocks(); + await sync(slow); + await sync(fast); - expect(await b.get("SELECT lemma_id FROM card WHERE lemma_id = 2")).toBeUndefined(); - - // And it stays deleted after b syncs again — no resurrection. - await syncOnce(b, cfg); - await syncOnce(a, cfg); - expect(await a.get("SELECT lemma_id FROM card WHERE lemma_id = 2")).toBeUndefined(); - - await a.close(); - await b.close(); + expect(await one(fast, "SELECT v FROM meta WHERE k = 'prefs.goal'")).toEqual({ v: "25" }); + expect(await one(slow, "SELECT v FROM meta WHERE k = 'prefs.goal'")).toEqual({ v: "25" }); + }); + + it("keeps both devices' offline turns — ids that cannot collide", async () => { + const a = await make(); + const b = await make(); + await sync(a); + await sync(b); + + await editChatTurn(a, "user", "a1"); + await editChatTurn(a, "assistant", "a2"); + await editChatTurn(b, "user", "b1"); + await editChatTurn(b, "assistant", "b2"); + await sync(a); + await sync(b); + await sync(a); + + for (const d of [a, b]) expect((await turns(d)).sort()).toEqual(["a1", "a2", "b1", "b2"]); + }); + + /* Gate 3. A cleared lesson is a deliberate shrink: the other device obeys + it, even holding more turns than the device that cleared. */ + it("keeps a cleared lesson cleared against a device holding a longer transcript", async () => { + const a = await make(); + const b = await make(); + for (const t of ["one", "two", "three"]) await editChatTurn(a, "user", t); + await sync(a); + await sync(b); + await editChatTurn(b, "user", "offline four"); + await editChatTurn(b, "user", "offline five"); + + await editChatClear(a); + await editChatTurn(a, "user", "a new lesson"); + await sync(a); + await sync(b); + await sync(a); + + expect(await turns(b)).toEqual(["a new lesson"]); + expect(await turns(a)).toEqual(["a new lesson"]); + }); + + it("honours a full reset on a device that had not heard of it, its own offline work included", async () => { + const a = await make(); + const b = await make(); + await editCard(a, RICE, { ...newCard(), state: 1, reps: 1 }); + await editUnitConfidence(a, "1.1", 50); + await sync(a); + await sync(b); + + await editCard(b, SCHOOL, { ...newCard(), state: 1, reps: 1 }); // a card a never saw + await editUnitConfidence(b, "1.2", 30); + + await editReset(a, "everything"); + await sync(a); + await sync(b); + await sync(a); + + for (const d of [a, b]) { + expect(await one(d, "SELECT count(*) AS n FROM card")).toEqual({ n: 0 }); + expect(await one(d, "SELECT count(*) AS n FROM progress")).toEqual({ n: 0 }); + } + }); + + it("keeps a forgotten card forgotten against a review made before hearing of it", async () => { + const a = await make(); + const b = await make(); + await editCard(a, WATER, { ...newCard(), state: 2, interval: 10, reps: 3 }); + await sync(a); + await sync(b); + + await editCardReset(a, WATER); + await sync(a); + await editCard(b, WATER, { ...newCard(), state: 2, interval: 20, reps: 4 }); + await sync(b); + await sync(a); + + for (const d of [a, b]) expect(await one(d, "SELECT lemma_id FROM card WHERE lemma_id = ?", [WATER])).toBeUndefined(); + }); + + it("re-creates a card forgotten elsewhere, once this device has heard of the forgetting", async () => { + const a = await make(); + const b = await make(); + await editCard(a, RICE, { ...newCard(), state: 1, reps: 1 }); + await sync(a); + await sync(b); + await editCardReset(a, RICE); + await sync(a); + await sync(b); // b hears of it + + await editCard(b, RICE, { ...newCard(), state: 1, reps: 1 }); // and studies it again + await sync(b); + await sync(a); + expect(await one(a, "SELECT reps FROM card WHERE lemma_id = ?", [RICE])).toEqual({ reps: 1 }); + }); + + /* Protocol 1 joined a key's values with a space and split them back on + one; 몇 명 came back as two values and stopped every pull at that row. */ + it("carries and deletes a row whose key contains a space", async () => { + const a = await make(); + const b = await make(); + await editPeek(a, "몇 명"); + await editPeek(a, "몇 명"); + await sync(a); + await sync(b); + expect(await one(b, "SELECT SUM(count) AS n FROM peek WHERE form = '몇 명'")).toEqual({ n: 2 }); + + await editReset(a, "everything"); + await sync(a); + await sync(b); + expect(await one(b, "SELECT count(*) AS n FROM peek")).toEqual({ n: 0 }); + }); + + it("counts two devices' reviews of the same day, rather than keeping one", async () => { + const a = await make(); + const b = await make(); + await editStudyLog(a, 20_000, { reviews: 2 }); + await editStudyLog(b, 20_000, { reviews: 3 }); + await sync(a); + await sync(b); + await sync(a); + for (const d of [a, b]) { + expect(await one(d, "SELECT SUM(reviews) AS n FROM study_log WHERE day = 20000")).toEqual({ n: 5 }); + } }); - /* Device-local bookkeeping must not travel. dict.loadedBands is the - dangerous one: it would tell a device it holds rows it never - downloaded, and the word rail would then miss words it believes are - present. */ it("never syncs device-local meta", async () => { - const a = await device(); - const b = await device(); - + const a = await make(); + const b = await make(); await editMeta(a, "prefs.goal", "40"); - // These are written the way the app writes them — unstamped bookkeeping. - await a.run("INSERT OR REPLACE INTO meta (k, v, updated_at) VALUES ('dict.loadedBands','[0,1,2,3,4,5]',?)", [ - Date.now(), - ]); - await a.run("INSERT OR REPLACE INTO meta (k, v, updated_at) VALUES ('server.token','secret',?)", [ - Date.now(), - ]); + await editMeta(a, "server.token", "secret"); + await a.run("INSERT OR REPLACE INTO meta (k, v, updated_at, dirty) VALUES ('dict.loadedBands', '[0,1,2]', 1, 1)"); + await sync(a); + await sync(b); - await syncOnce(a, cfg); - await syncOnce(b, cfg); + expect(await one(b, "SELECT v FROM meta WHERE k = 'prefs.goal'")).toEqual({ v: "40" }); + expect(await one(b, "SELECT v FROM meta WHERE k = 'server.token'")).toBeUndefined(); + expect(await one(b, "SELECT v FROM meta WHERE k = 'dict.loadedBands'")).toBeUndefined(); + }); - expect((await b.get<{ v: string }>("SELECT v FROM meta WHERE k='prefs.goal'"))?.v).toBe("40"); - expect(await b.get("SELECT v FROM meta WHERE k='dict.loadedBands'")).toBeUndefined(); - expect(await b.get("SELECT v FROM meta WHERE k='server.token'")).toBeUndefined(); + it("re-hydrates when the server has lost what it held, and offers its data back", async () => { + const a = await make(); + await editUnitConfidence(a, "2.1", 70); + await sync(a); - await a.close(); - await b.close(); + await fetch(`${cfg.baseUrl}/api/test/reset`, { + method: "POST", + headers: { authorization: `Bearer ${cfg.token}` }, + }); + const again = await sync(a); + expect(again.rehydrated).toBe(true); + expect(again.pushed).toBeGreaterThan(0); + + const b = await make(); + await sync(b); + expect(await one(b, "SELECT confidence FROM progress WHERE unit_id = '2.1'")).toEqual({ confidence: 70 }); + }); + + it("refuses a client that does not speak protocol 2", async () => { + const res = await fetch(`${cfg.baseUrl}/api/sync?cursor=0`, { + headers: { authorization: `Bearer ${cfg.token}` }, + }); + expect(res.status).toBe(426); + const ok = await fetch(`${cfg.baseUrl}/api/sync?cursor=0`, { + headers: { authorization: `Bearer ${cfg.token}`, [PROTOCOL_HEADER]: "2" }, + }); + expect(ok.status).toBe(200); }); }); diff --git a/types/shared/sync-protocol.mjs.d.ts b/types/shared/sync-protocol.mjs.d.ts index 3cb6f17..79d2b69 100644 --- a/types/shared/sync-protocol.mjs.d.ts +++ b/types/shared/sync-protocol.mjs.d.ts @@ -1,17 +1,60 @@ -/* Declarations for shared/sync-protocol.mjs. */ +/* Declarations for shared/sync-protocol.mjs — protocol 2. */ export interface SyncTableSpec { pk: string[]; cols: string[]; } +export const PROTOCOL: 2; +export const PROTOCOL_HEADER: string; +export const PAGE_SIZE: number; + export const SYNC_TABLES: Record; export const SYNC_TABLE_NAMES: string[]; -export const PAGE_SIZE: number; export function pkFor(tbl: string): string[] | null; +/** A row's primary-key values as a JSON array — the key on the wire. */ +export function encodePk(tbl: string, row: Record): string; +export function decodePk(pk: string): (string | number)[]; + export function isSyncableMetaKey(key: string): boolean; -export function rowKey(table: string, row: Record): string; -export function isDirty(row: { updated_at: number }, pushedAt: number): boolean; -export function incomingWins(incomingUpdatedAt: number, localUpdatedAt: number | null): boolean; + +export type ResetScope = "chat" | "roadmap" | "everything"; +export const RESET_SCOPES: Record; +export function resetCovers(scope: string, tbl: string, row: Record): boolean; + +/** A row on the wire. `base` is the change_seq the sender last agreed with. */ +export interface WireRow { + tbl: string; + pk: string; + base: number; + deleted?: boolean; + data?: Record | null; + updated_at?: number; +} + +/** A row as the server holds it: `seq` is its current change_seq. */ +export interface ServerRow { + tbl: string; + pk: string; + seq: number; + deleted: boolean; + data: Record | null; + updated_at?: number; +} + +export interface PullResponse { + epoch: string; + rows: ServerRow[]; + cursor: number; + more: boolean; +} + +export interface PushResponse { + epoch: string; + applied: { tbl: string; pk: string; seq: number }[]; + /** Rows the push could not overwrite: the server's current version. A + row absent on the server comes back with seq 0 and deleted. */ + conflicts: ServerRow[]; +}