From f12300494cc11aa97f77de2469cfb85e9fddff15 Mon Sep 17 00:00:00 2001 From: Sivert Date: Fri, 25 Sep 2026 11:55:24 +0200 Subject: [PATCH] Tables for the MLS delivery service (GRYT-1500) Stage 1 of docs/mls-design.md in the crypto repo needs somewhere to keep KeyPackages, Welcomes and each DM group's log. Five new tables, all holding MLS bytes the server can't open plus what ordering and routing need: - mls_devices: which devices a member has here, five at most. - mls_key_packages: up to 20 per device plus one last-resort package. Handing one out clears its bytes but keeps the row, so a Welcome that names it by ref can still be routed. The retention sweep drops it later. - mls_groups: one per conversation, with the current epoch and the last seq handed out. head_seq is kept here and not read off the log, because the log gets swept and a seq must never be handed out twice. - mls_log: commits, proposals and application messages, in seq order. - mls_welcomes: one row per receiving device, until that device acks it. appendMlsCommit is where commits get ordered. In one IMMEDIATE transaction it checks the commit's epoch against the group's, writes it and moves the epoch on, or refuses and says what the epoch is now. The same bytes sent again within the last 100 entries get the first seq back rather than a second row, for a client retrying after a lost reply. Two existing paths change. Purging an empty conversation drops its group too: a DM id comes from the pair, so a group left behind would be handed to the next DM those two open. And a guest merged into an account takes its devices, KeyPackages and Welcomes along. Nothing calls these yet. The socket handlers are in the next PR. Co-Authored-By: Claude Opus 5.5 --- src/db/sqlite/connection.ts | 59 ++++ src/db/sqlite/conversations.ts | 2 + src/db/sqlite/index.ts | 1 + src/db/sqlite/mergeGuest.test.ts | 1 + src/db/sqlite/mergeGuest.ts | 4 + src/db/sqlite/mls.test.ts | 331 ++++++++++++++++++++ src/db/sqlite/mls.ts | 517 +++++++++++++++++++++++++++++++ 7 files changed, 915 insertions(+) create mode 100644 src/db/sqlite/mls.test.ts create mode 100644 src/db/sqlite/mls.ts diff --git a/src/db/sqlite/connection.ts b/src/db/sqlite/connection.ts index ad409512..559a736e 100644 --- a/src/db/sqlite/connection.ts +++ b/src/db/sqlite/connection.ts @@ -579,6 +579,65 @@ function createSchema(d: DatabaseSync): void { updated_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_image_jobs_status ON image_jobs(status, created_at); + + -- MLS delivery service (GRYT-1500). Blobs are MLS wire bytes the server can't + -- open. Keyed on server_user_id like conversations, since the groups are theirs. + CREATE TABLE IF NOT EXISTS mls_devices ( + server_user_id TEXT NOT NULL, + device_id TEXT NOT NULL, + created_at TEXT NOT NULL, + last_seen_at TEXT NOT NULL, + PRIMARY KEY (server_user_id, device_id) + ); + + -- data goes to NULL when a package is handed out. The row stays until the + -- retention sweep, so a Welcome naming the ref can still find its device. + CREATE TABLE IF NOT EXISTS mls_key_packages ( + key_package_ref TEXT PRIMARY KEY, + server_user_id TEXT NOT NULL, + device_id TEXT NOT NULL, + data BLOB, + last_resort INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, + claimed_at TEXT + ); + CREATE INDEX IF NOT EXISTS idx_mls_key_packages_device + ON mls_key_packages(server_user_id, device_id, last_resort, claimed_at); + + -- One group per conversation. head_seq is kept here rather than read off the + -- log, because the log gets swept and a cursor must never be handed out twice. + CREATE TABLE IF NOT EXISTS mls_groups ( + group_id TEXT PRIMARY KEY, + conversation_id TEXT NOT NULL UNIQUE, + epoch INTEGER NOT NULL, + head_seq INTEGER NOT NULL, + created_by_server_user_id TEXT NOT NULL, + created_at TEXT NOT NULL + ); + + CREATE TABLE IF NOT EXISTS mls_log ( + group_id TEXT NOT NULL, + seq INTEGER NOT NULL, + kind TEXT NOT NULL, + epoch INTEGER NOT NULL, + sender_server_user_id TEXT NOT NULL, + sender_device_id TEXT NOT NULL, + data BLOB NOT NULL, + created_at TEXT NOT NULL, + PRIMARY KEY (group_id, seq) + ); + CREATE INDEX IF NOT EXISTS idx_mls_log_created ON mls_log(created_at); + + CREATE TABLE IF NOT EXISTS mls_welcomes ( + welcome_id TEXT PRIMARY KEY, + server_user_id TEXT NOT NULL, + device_id TEXT NOT NULL, + group_id TEXT NOT NULL, + data BLOB NOT NULL, + created_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_mls_welcomes_device ON mls_welcomes(server_user_id, device_id); + CREATE INDEX IF NOT EXISTS idx_mls_welcomes_created ON mls_welcomes(created_at); `); } diff --git a/src/db/sqlite/conversations.ts b/src/db/sqlite/conversations.ts index e70264fb..6478e39e 100644 --- a/src/db/sqlite/conversations.ts +++ b/src/db/sqlite/conversations.ts @@ -2,6 +2,7 @@ import { createHash, randomUUID } from "crypto"; import type { ConversationRecord } from "../interfaces"; import { fromIso, fromIsoNullable, getSqliteDb, toIso } from "./connection"; +import { dropMlsGroupForConversation } from "./mls"; /** * Nothing here is cross-server: a DM is filed under `server_user_id`, this @@ -262,6 +263,7 @@ export async function purgeOrphanedConversations(): Promise { const deleteMembers = db.prepare(`DELETE FROM conversation_members WHERE conversation_id = ?`); const deleteConversation = db.prepare(`DELETE FROM conversations WHERE conversation_id = ?`); for (const id of ids) { + dropMlsGroupForConversation(id); deleteMessages.run(id); deleteMembers.run(id); deleteConversation.run(id); diff --git a/src/db/sqlite/index.ts b/src/db/sqlite/index.ts index 061ebe7d..bc50c756 100644 --- a/src/db/sqlite/index.ts +++ b/src/db/sqlite/index.ts @@ -22,3 +22,4 @@ export * from "./contactPrefs"; export * from "./friends"; export * from "./mentions"; export * from "./webhooks"; +export * from "./mls"; diff --git a/src/db/sqlite/mergeGuest.test.ts b/src/db/sqlite/mergeGuest.test.ts index e6301a10..d9907aba 100644 --- a/src/db/sqlite/mergeGuest.test.ts +++ b/src/db/sqlite/mergeGuest.test.ts @@ -216,6 +216,7 @@ describe("merging a guest into an account that is already a member", () => { const handledApart = new Set([ "users.server_user_id", "roles.server_user_id", "conversation_members.server_user_id", "mentions.server_user_id", "refresh_tokens.server_user_id", + "mls_devices.server_user_id", "mls_key_packages.server_user_id", "mls_welcomes.server_user_id", ]); const listed = new Set(AUTHOR_COLUMNS.map(([t, c]) => `${t}.${c}`)); const tables = db.prepare(`SELECT name FROM sqlite_master WHERE type = 'table'`).all() as { name: string }[]; diff --git a/src/db/sqlite/mergeGuest.ts b/src/db/sqlite/mergeGuest.ts index 529f09b2..7db2fc49 100644 --- a/src/db/sqlite/mergeGuest.ts +++ b/src/db/sqlite/mergeGuest.ts @@ -3,6 +3,7 @@ import type { DatabaseSync } from "node:sqlite"; import type { Reaction } from "../interfaces"; import { getSqliteDb, toIso } from "./connection"; import { carryFriendsForward } from "./friends"; +import { carryMlsDevicesForward } from "./mls"; export interface GuestMerge { guestServerUserId: string; @@ -34,6 +35,8 @@ export const AUTHOR_COLUMNS: ReadonlyArray { + dir = mkdtempSync(join(tmpdir(), "gryt-mls-db-")); + process.env.DATA_DIR = dir; + await initSqlite(); +}); + +after(() => { + rmSync(dir, { recursive: true, force: true }); +}); + +const bytes = (...b: number[]) => new Uint8Array(b); +const DAY = 24 * 60 * 60 * 1000; + +let uniq = 0; +async function member(): Promise { + uniq += 1; + return (await upsertUser(`mls-db-${uniq}`, `Member ${uniq}`)).server_user_id; +} + +async function conversation(): Promise { + return (await openDirectConversation(await member(), await member())).conversation_id; +} + +describe("devices", () => { + it("caps each member at five and lets a known one back in", async () => { + const me = await member(); + for (let i = 0; i < MLS_MAX_DEVICES; i++) assert.equal(touchMlsDevice(me, `d${i}`), "ok"); + + assert.equal(touchMlsDevice(me, "one-too-many"), "too_many_devices"); + assert.equal(touchMlsDevice(me, "d0"), "ok", "a device already known is never refused by the cap"); + assert.equal(listMlsDevices([me]).length, MLS_MAX_DEVICES); + }); + + it("frees the slot, its packages and its Welcomes when removed", async () => { + const me = await member(); + touchMlsDevice(me, "phone"); + addMlsKeyPackages(me, "phone", [{ ref: "ref-removed", data: bytes(1), lastResort: false }]); + + assert.equal(removeMlsDevice(me, "phone"), true); + assert.deepEqual(listMlsDevices([me]), []); + assert.equal(mlsKeyPackageOwner("ref-removed"), null); + }); +}); + +describe("KeyPackages", () => { + it("keeps twenty per device and hands them out once each, oldest first", async () => { + const me = await member(); + const packages = Array.from({ length: MLS_MAX_KEY_PACKAGES + 5 }, (_, i) => ({ + ref: `kp-${me}-${i}`, + data: bytes(i), + lastResort: false, + })); + const t0 = new Date("2026-09-25T10:00:00Z"); + const { stored, unclaimed } = addMlsKeyPackages(me, "laptop", packages, t0); + assert.equal(stored, MLS_MAX_KEY_PACKAGES); + assert.equal(unclaimed, MLS_MAX_KEY_PACKAGES); + + const first = claimMlsKeyPackage(me, "laptop"); + const second = claimMlsKeyPackage(me, "laptop"); + assert.equal(first?.ref, `kp-${me}-0`); + assert.equal(second?.ref, `kp-${me}-1`); + assert.notEqual(first?.ref, second?.ref, "a KeyPackage used twice would let two groups share one init key"); + assert.equal(countMlsKeyPackages(me, "laptop").unclaimed, MLS_MAX_KEY_PACKAGES - 2); + + const row = getSqliteDb() + .prepare(`SELECT data FROM mls_key_packages WHERE key_package_ref = ?`) + .get(first!.ref) as { data: Uint8Array | null }; + assert.equal(row.data, null, "a handed-out package's bytes are gone"); + assert.deepEqual(mlsKeyPackageOwner(first!.ref), { serverUserId: me, deviceId: "laptop" }); + }); + + it("falls back to the last-resort package, again and again", async () => { + const me = await member(); + addMlsKeyPackages(me, "tablet", [ + { ref: `once-${me}`, data: bytes(1), lastResort: false }, + { ref: `last-${me}`, data: bytes(9), lastResort: true }, + ]); + + assert.equal(claimMlsKeyPackage(me, "tablet")?.ref, `once-${me}`); + const a = claimMlsKeyPackage(me, "tablet"); + const b = claimMlsKeyPackage(me, "tablet"); + assert.equal(a?.lastResort, true); + assert.equal(b?.ref, `last-${me}`); + assert.deepEqual(b?.data, bytes(9)); + }); + + it("retires the old last-resort package when a new one comes, and still routes to it", async () => { + const me = await member(); + addMlsKeyPackages(me, "desk", [{ ref: `old-${me}`, data: bytes(1), lastResort: true }]); + addMlsKeyPackages(me, "desk", [{ ref: `new-${me}`, data: bytes(2), lastResort: true }]); + + assert.equal(claimMlsKeyPackage(me, "desk")?.ref, `new-${me}`); + assert.deepEqual(mlsKeyPackageOwner(`old-${me}`), { serverUserId: me, deviceId: "desk" }); + }); + + it("gives nothing for a device with nothing", async () => { + assert.equal(claimMlsKeyPackage(await member(), "nothing"), null); + }); +}); + +describe("groups and commit ordering", () => { + it("keeps the first group registered for a conversation", async () => { + const conv = await conversation(); + const first = createMlsGroup("aa01", conv, "u1"); + const second = createMlsGroup("aa02", conv, "u2"); + + assert.equal(first.created, true); + assert.equal(second.created, false); + assert.equal(second.group?.groupId, "aa01", "the loser has to learn which group to wait for"); + }); + + it("refuses a group id already used by another conversation", async () => { + createMlsGroup("aa03", await conversation(), "u1"); + const reuse = createMlsGroup("aa03", await conversation(), "u1"); + assert.deepEqual(reuse, { created: false, group: null }); + }); + + it("accepts one commit per epoch and tells the rest where the group is", async () => { + const conv = await conversation(); + createMlsGroup("bb01", conv, "u1"); + const base = { groupId: "bb01", senderServerUserId: "u1", senderDeviceId: "d1" }; + + const results = [1, 2, 3].map((n) => appendMlsCommit({ ...base, epoch: 0, data: bytes(n) })); + assert.equal(results.filter((r) => r.accepted).length, 1); + for (const r of results.slice(1)) { + assert.deepEqual(r, { accepted: false, reason: "stale_epoch", epoch: 1, headSeq: 1 }); + } + + const next = appendMlsCommit({ ...base, epoch: 1, data: bytes(4) }); + assert.equal(next.accepted && next.epoch, 2); + assert.equal(getMlsGroup("bb01")?.epoch, 2); + assert.deepEqual(listMlsLog("bb01", 0, 10).map((e) => e.data[0]), [1, 4], "a refused commit is never written"); + }); + + it("orders messages by seq without ordering them by epoch", async () => { + const conv = await conversation(); + createMlsGroup("cc01", conv, "u1"); + const base = { groupId: "cc01", senderServerUserId: "u1", senderDeviceId: "d1" }; + + appendMlsCommit({ ...base, epoch: 0, data: bytes(1) }); + appendMlsCommit({ ...base, epoch: 1, data: bytes(2) }); + const late = appendMlsMessage("application", { ...base, epoch: 1, data: bytes(3) }); + const ahead = appendMlsMessage("application", { ...base, epoch: 3, data: bytes(4) }); + + assert.equal(late.accepted, true, "a message sent just before a commit still has to land"); + assert.deepEqual(ahead, { accepted: false, reason: "future_epoch", epoch: 2 }); + assert.deepEqual(listMlsLog("cc01", 0, 10).map((e) => [e.seq, e.kind]), [ + [1, "commit"], + [2, "commit"], + [3, "application"], + ]); + }); + + it("writes a resend once and answers it with the first seq", async () => { + const conv = await conversation(); + createMlsGroup("cc02", conv, "u1"); + const base = { groupId: "cc02", senderServerUserId: "u1", senderDeviceId: "d1" }; + + const commit = appendMlsCommit({ ...base, epoch: 0, data: bytes(1) }); + const commitAgain = appendMlsCommit({ ...base, epoch: 0, data: bytes(1) }); + assert.deepEqual(commitAgain.accepted && [commitAgain.seq, commitAgain.epoch, commitAgain.duplicate], [1, 1, true], + "a commit retried after a lost reply is the one that won, not a stale one"); + assert.ok(commit.accepted && !commit.duplicate); + + const sent = appendMlsMessage("application", { ...base, epoch: 1, data: bytes(2) }); + const again = appendMlsMessage("application", { ...base, epoch: 1, data: bytes(2) }); + assert.deepEqual(again.accepted && [again.seq, again.duplicate], [2, true]); + assert.ok(sent.accepted && !sent.duplicate); + assert.equal(listMlsLog("cc02", 0, 10).length, 2); + }); + + it("pages the log from a cursor", async () => { + const conv = await conversation(); + createMlsGroup("dd01", conv, "u1"); + for (let i = 0; i < 7; i++) { + appendMlsMessage("application", { groupId: "dd01", epoch: 0, senderServerUserId: "u1", senderDeviceId: "d", data: bytes(i) }); + } + assert.deepEqual(listMlsLog("dd01", 0, 3).map((e) => e.seq), [1, 2, 3]); + assert.deepEqual(listMlsLog("dd01", 3, 3).map((e) => e.seq), [4, 5, 6]); + assert.deepEqual(listMlsLog("dd01", 6, 3).map((e) => e.seq), [7]); + assert.deepEqual(listMlsLog("dd01", 7, 3), []); + }); + + it("lists the groups a member is in, and only those", async () => { + const a = await member(); + const b = await member(); + const outsider = await member(); + const conv = (await openDirectConversation(a, b)).conversation_id; + createMlsGroup("ee01", conv, a); + + assert.deepEqual(listMlsGroupsForMember(a).map((g) => g.groupId), ["ee01"]); + assert.deepEqual(listMlsGroupsForMember(outsider), []); + }); +}); + +describe("Welcomes", () => { + it("are written with the commit and only deleted by the device they are for", async () => { + const conv = await conversation(); + createMlsGroup("ff01", conv, "u1"); + const res = appendMlsCommit( + { groupId: "ff01", epoch: 0, senderServerUserId: "u1", senderDeviceId: "d1", data: bytes(1) }, + [{ serverUserId: "u2", deviceId: "phone", data: bytes(7) }], + ); + assert.ok(res.accepted); + const [welcome] = listMlsWelcomes("u2", "phone"); + assert.equal(welcome.conversationId, conv); + assert.deepEqual(welcome.data, bytes(7)); + + assert.equal(deleteMlsWelcomes("u3", "phone", [welcome.welcomeId]), 0); + assert.equal(deleteMlsWelcomes("u2", "laptop", [welcome.welcomeId]), 0); + assert.equal(deleteMlsWelcomes("u2", "phone", [welcome.welcomeId]), 1); + assert.deepEqual(listMlsWelcomes("u2", "phone"), []); + }); + + it("are not written for a refused commit", async () => { + const conv = await conversation(); + createMlsGroup("ff02", conv, "u1"); + const base = { groupId: "ff02", senderServerUserId: "u1", senderDeviceId: "d1" }; + appendMlsCommit({ ...base, epoch: 0, data: bytes(1) }); + appendMlsCommit({ ...base, epoch: 0, data: bytes(2) }, [{ serverUserId: "u9", deviceId: "x", data: bytes(3) }]); + assert.deepEqual(listMlsWelcomes("u9", "x"), []); + }); +}); + +describe("retention", () => { + it("drops what is older than the cutoff and never reuses a seq", async () => { + const conv = await conversation(); + createMlsGroup("aa99", conv, "u1"); + const base = { groupId: "aa99", senderServerUserId: "u1", senderDeviceId: "d1" }; + const old = new Date(Date.now() - 40 * DAY); + appendMlsCommit({ ...base, epoch: 0, data: bytes(1) }, [{ serverUserId: "u2", deviceId: "p", data: bytes(1) }], old); + appendMlsMessage("application", { ...base, epoch: 1, data: bytes(2) }, old); + appendMlsMessage("application", { ...base, epoch: 1, data: bytes(3) }); + + const me = await member(); + addMlsKeyPackages(me, "d", [ + { ref: `swept-${me}`, data: bytes(1), lastResort: false }, + { ref: `waiting-${me}`, data: bytes(2), lastResort: false }, + ], old); + claimMlsKeyPackage(me, "d", old); + + const swept = sweepMls(new Date(Date.now() - 30 * DAY)); + assert.ok(swept.log >= 2); + assert.ok(swept.welcomes >= 1); + assert.ok(swept.keyPackages >= 1); + + assert.deepEqual(listMlsLog("aa99", 0, 10).map((e) => e.seq), [3]); + assert.equal(oldestMlsSeq("aa99"), 3, "a device whose cursor is behind this knows it has a gap"); + assert.equal(mlsKeyPackageOwner(`swept-${me}`), null); + assert.ok(mlsKeyPackageOwner(`waiting-${me}`), "an unclaimed package is not ciphertext, and stays"); + + const again = appendMlsMessage("application", { ...base, epoch: 1, data: bytes(4) }); + assert.equal(again.accepted && again.seq, 4); + assert.equal(getMlsGroup("aa99")?.epoch, 1, "sweeping the log leaves the epoch where it was"); + }); + + it("drops a group whose conversation is gone, with its log", async () => { + const conv = await conversation(); + createMlsGroup("ab01", conv, "u1"); + appendMlsMessage("application", { groupId: "ab01", epoch: 0, senderServerUserId: "u1", senderDeviceId: "d", data: bytes(1) }); + getSqliteDb().prepare(`DELETE FROM conversations WHERE conversation_id = ?`).run(conv); + + const swept = sweepMls(new Date(Date.now() - 30 * DAY)); + assert.ok(swept.groups >= 1); + assert.equal(getMlsGroup("ab01"), null); + assert.deepEqual(listMlsLog("ab01", 0, 10), []); + }); +}); + +describe("lifecycle", () => { + it("drops the group with its conversation, so the pair's next DM starts clean", async () => { + const a = await member(); + const b = await member(); + const conv = (await openDirectConversation(a, b)).conversation_id; + createMlsGroup("ac01", conv, a); + getSqliteDb().prepare(`UPDATE users SET is_active = 0 WHERE server_user_id IN (?, ?)`).run(a, b); + + assert.ok((await purgeOrphanedConversations()).includes(conv)); + assert.equal(getMlsGroup("ac01"), null); + }); + + it("moves a guest's devices, packages and Welcomes to the account it merges into", async () => { + const guest = await upsertUser("mls-guest", "Guest"); + const account = await upsertUser("mls-account", "Account"); + touchMlsDevice(guest.server_user_id, "shared"); + touchMlsDevice(guest.server_user_id, "only-guest"); + touchMlsDevice(account.server_user_id, "shared"); + addMlsKeyPackages(guest.server_user_id, "only-guest", [{ ref: "guest-kp", data: bytes(1), lastResort: false }]); + + assert.ok(mergeGuestIntoAccount("mls-guest", "mls-account")); + assert.deepEqual(listMlsDevices([account.server_user_id]).map((d) => d.deviceId).sort(), ["only-guest", "shared"]); + assert.deepEqual(mlsKeyPackageOwner("guest-kp"), { serverUserId: account.server_user_id, deviceId: "only-guest" }); + }); +}); diff --git a/src/db/sqlite/mls.ts b/src/db/sqlite/mls.ts new file mode 100644 index 00000000..b33b653e --- /dev/null +++ b/src/db/sqlite/mls.ts @@ -0,0 +1,517 @@ +import { randomUUID } from "crypto"; +import type { DatabaseSync } from "node:sqlite"; + +import { getSqliteDb, toIso } from "./connection"; + +/** + * The MLS delivery service's storage (GRYT-1500). Every blob is opaque here; the + * columns beside it are what ordering, routing and retention read. + */ + +/** Five devices per person per server, from docs/mls-design.md in the crypto repo. */ +export const MLS_MAX_DEVICES = 5; + +/** Unclaimed KeyPackages a device may hold, not counting its last-resort one. */ +export const MLS_MAX_KEY_PACKAGES = 20; + +export type MlsLogKind = "commit" | "proposal" | "application"; + +export interface MlsDevice { + serverUserId: string; + deviceId: string; + createdAt: string; + lastSeenAt: string; +} + +export interface MlsGroup { + groupId: string; + conversationId: string; + epoch: number; + headSeq: number; + createdByServerUserId: string; + createdAt: string; +} + +export interface MlsLogEntry { + groupId: string; + seq: number; + kind: MlsLogKind; + epoch: number; + senderServerUserId: string; + senderDeviceId: string; + data: Uint8Array; + createdAt: string; +} + +export interface MlsWelcome { + welcomeId: string; + groupId: string; + conversationId: string | null; + deviceId: string; + data: Uint8Array; + createdAt: string; +} + +/* node:sqlite is synchronous, so a read and its write can't interleave in-process; + IMMEDIATE also shuts out the image worker's process between them. */ +function inTransaction(fn: () => T): T { + const db = getSqliteDb(); + db.exec("BEGIN IMMEDIATE"); + try { + const result = fn(); + db.exec("COMMIT"); + return result; + } catch (err) { + db.exec("ROLLBACK"); + throw err; + } +} + +function rowToGroup(r: Record): MlsGroup { + return { + groupId: r.group_id as string, + conversationId: r.conversation_id as string, + epoch: Number(r.epoch), + headSeq: Number(r.head_seq), + createdByServerUserId: r.created_by_server_user_id as string, + createdAt: r.created_at as string, + }; +} + +function rowToEntry(r: Record): MlsLogEntry { + return { + groupId: r.group_id as string, + seq: Number(r.seq), + kind: r.kind as MlsLogKind, + epoch: Number(r.epoch), + senderServerUserId: r.sender_server_user_id as string, + senderDeviceId: r.sender_device_id as string, + data: r.data as Uint8Array, + createdAt: r.created_at as string, + }; +} + +// ── Devices ───────────────────────────────────────────────────────────── + +/** A new device past the cap is refused; a known one only has last_seen_at moved. */ +export function touchMlsDevice( + serverUserId: string, + deviceId: string, + now = new Date(), +): "ok" | "too_many_devices" { + return inTransaction(() => { + const db = getSqliteDb(); + const at = toIso(now); + const known = db + .prepare(`UPDATE mls_devices SET last_seen_at = ? WHERE server_user_id = ? AND device_id = ?`) + .run(at, serverUserId, deviceId); + if (Number(known.changes) > 0) return "ok"; + + const { n } = db + .prepare(`SELECT COUNT(*) AS n FROM mls_devices WHERE server_user_id = ?`) + .get(serverUserId) as { n: number }; + if (n >= MLS_MAX_DEVICES) return "too_many_devices"; + + db.prepare( + `INSERT INTO mls_devices (server_user_id, device_id, created_at, last_seen_at) VALUES (?, ?, ?, ?)`, + ).run(serverUserId, deviceId, at, at); + return "ok"; + }); +} + +export function isMlsDevice(serverUserId: string, deviceId: string): boolean { + return !!getSqliteDb() + .prepare(`SELECT 1 FROM mls_devices WHERE server_user_id = ? AND device_id = ?`) + .get(serverUserId, deviceId); +} + +export function listMlsDevices(serverUserIds: string[]): MlsDevice[] { + if (serverUserIds.length === 0) return []; + const marks = serverUserIds.map(() => "?").join(", "); + const rows = getSqliteDb() + .prepare( + `SELECT server_user_id, device_id, created_at, last_seen_at FROM mls_devices + WHERE server_user_id IN (${marks}) ORDER BY server_user_id, created_at`, + ) + .all(...serverUserIds) as Record[]; + return rows.map((r) => ({ + serverUserId: r.server_user_id, + deviceId: r.device_id, + createdAt: r.created_at, + lastSeenAt: r.last_seen_at, + })); +} + +/** Its KeyPackages and waiting Welcomes go with it, since nothing can use them now. */ +export function removeMlsDevice(serverUserId: string, deviceId: string): boolean { + return inTransaction(() => { + const db = getSqliteDb(); + const gone = db + .prepare(`DELETE FROM mls_devices WHERE server_user_id = ? AND device_id = ?`) + .run(serverUserId, deviceId); + db.prepare(`DELETE FROM mls_key_packages WHERE server_user_id = ? AND device_id = ?`).run(serverUserId, deviceId); + db.prepare(`DELETE FROM mls_welcomes WHERE server_user_id = ? AND device_id = ?`).run(serverUserId, deviceId); + return Number(gone.changes) > 0; + }); +} + +/** A guest folded into an account brings its devices along. Where both hold one + device id, the account's row is kept. Runs inside the merge's transaction. */ +export function carryMlsDevicesForward(db: DatabaseSync, from: string, to: string): void { + db.prepare( + `DELETE FROM mls_devices WHERE server_user_id = ? + AND device_id IN (SELECT device_id FROM mls_devices WHERE server_user_id = ?)`, + ).run(from, to); + db.prepare(`UPDATE mls_devices SET server_user_id = ? WHERE server_user_id = ?`).run(to, from); + db.prepare(`UPDATE mls_key_packages SET server_user_id = ? WHERE server_user_id = ?`).run(to, from); + db.prepare(`UPDATE mls_welcomes SET server_user_id = ? WHERE server_user_id = ?`).run(to, from); +} + +// ── KeyPackages ───────────────────────────────────────────────────────── + +export interface NewKeyPackage { + /** Hex KeyPackageRef, which is how a Welcome names its recipient. */ + ref: string; + data: Uint8Array; + lastResort: boolean; +} + +/** + * Stores up to the cap and says how many it took. A new last-resort package retires + * the old one, whose row stays so a Welcome naming it can still be routed. + */ +export function addMlsKeyPackages( + serverUserId: string, + deviceId: string, + packages: NewKeyPackage[], + now = new Date(), +): { stored: number; unclaimed: number; lastResort: boolean } { + return inTransaction(() => { + const db = getSqliteDb(); + const at = toIso(now); + const insert = db.prepare( + `INSERT OR IGNORE INTO mls_key_packages + (key_package_ref, server_user_id, device_id, data, last_resort, created_at) + VALUES (?, ?, ?, ?, ?, ?)`, + ); + const retire = db.prepare( + `UPDATE mls_key_packages SET data = NULL, claimed_at = ? + WHERE server_user_id = ? AND device_id = ? AND last_resort = 1 + AND claimed_at IS NULL AND key_package_ref != ?`, + ); + let unclaimed = countMlsKeyPackages(serverUserId, deviceId).unclaimed; + + let stored = 0; + for (const p of packages) { + if (!p.lastResort && unclaimed >= MLS_MAX_KEY_PACKAGES) continue; + const r = insert.run(p.ref, serverUserId, deviceId, p.data, p.lastResort ? 1 : 0, at); + if (Number(r.changes) === 0) continue; + if (p.lastResort) retire.run(at, serverUserId, deviceId, p.ref); + stored += 1; + if (!p.lastResort) unclaimed += 1; + } + return { stored, ...countMlsKeyPackages(serverUserId, deviceId) }; + }); +} + +export function countMlsKeyPackages(serverUserId: string, deviceId: string): { unclaimed: number; lastResort: boolean } { + const db = getSqliteDb(); + const { n } = db + .prepare( + `SELECT COUNT(*) AS n FROM mls_key_packages + WHERE server_user_id = ? AND device_id = ? AND last_resort = 0 AND claimed_at IS NULL`, + ) + .get(serverUserId, deviceId) as { n: number }; + const last = db + .prepare( + `SELECT 1 FROM mls_key_packages + WHERE server_user_id = ? AND device_id = ? AND last_resort = 1 AND claimed_at IS NULL`, + ) + .get(serverUserId, deviceId); + return { unclaimed: n, lastResort: !!last }; +} + +/** + * Hands out one package, oldest first, and forgets its bytes. The last-resort one is + * handed out again and again once the rest are gone, which RFC 9420 allows. + */ +export function claimMlsKeyPackage( + serverUserId: string, + deviceId: string, + now = new Date(), +): { ref: string; data: Uint8Array; lastResort: boolean } | null { + return inTransaction(() => { + const db = getSqliteDb(); + const fresh = db + .prepare( + `SELECT key_package_ref, data FROM mls_key_packages + WHERE server_user_id = ? AND device_id = ? AND last_resort = 0 AND claimed_at IS NULL + ORDER BY created_at, rowid LIMIT 1`, + ) + .get(serverUserId, deviceId) as { key_package_ref: string; data: Uint8Array } | undefined; + if (fresh) { + db.prepare(`UPDATE mls_key_packages SET data = NULL, claimed_at = ? WHERE key_package_ref = ?`).run( + toIso(now), + fresh.key_package_ref, + ); + return { ref: fresh.key_package_ref, data: fresh.data, lastResort: false }; + } + + const last = db + .prepare( + `SELECT key_package_ref, data FROM mls_key_packages + WHERE server_user_id = ? AND device_id = ? AND last_resort = 1 AND claimed_at IS NULL`, + ) + .get(serverUserId, deviceId) as { key_package_ref: string; data: Uint8Array } | undefined; + return last ? { ref: last.key_package_ref, data: last.data, lastResort: true } : null; + }); +} + +/** Whose package a ref was, claimed or not, for as long as the row is kept. */ +export function mlsKeyPackageOwner(ref: string): { serverUserId: string; deviceId: string } | null { + const row = getSqliteDb() + .prepare(`SELECT server_user_id, device_id FROM mls_key_packages WHERE key_package_ref = ?`) + .get(ref) as { server_user_id: string; device_id: string } | undefined; + return row ? { serverUserId: row.server_user_id, deviceId: row.device_id } : null; +} + +// ── Groups ────────────────────────────────────────────────────────────── + +export function getMlsGroup(groupId: string): MlsGroup | null { + const row = getSqliteDb().prepare(`SELECT * FROM mls_groups WHERE group_id = ?`).get(groupId); + return row ? rowToGroup(row as Record) : null; +} + +export function getMlsGroupForConversation(conversationId: string): MlsGroup | null { + const row = getSqliteDb().prepare(`SELECT * FROM mls_groups WHERE conversation_id = ?`).get(conversationId); + return row ? rowToGroup(row as Record) : null; +} + +/** The first group registered for a conversation wins, and the loser is handed the winner. */ +export function createMlsGroup( + groupId: string, + conversationId: string, + createdByServerUserId: string, + now = new Date(), +): { created: true; group: MlsGroup } | { created: false; group: MlsGroup | null } { + return inTransaction(() => { + const existing = getMlsGroupForConversation(conversationId); + if (existing) return { created: false as const, group: existing }; + // A group id already used for another conversation: refused, saying nothing about it. + if (getMlsGroup(groupId)) return { created: false as const, group: null }; + + getSqliteDb() + .prepare( + `INSERT INTO mls_groups (group_id, conversation_id, epoch, head_seq, created_by_server_user_id, created_at) + VALUES (?, ?, 0, 0, ?, ?)`, + ) + .run(groupId, conversationId, createdByServerUserId, toIso(now)); + return { created: true as const, group: getMlsGroup(groupId)! }; + }); +} + +export function listMlsGroupsForMember(serverUserId: string): (MlsGroup & { oldestSeq: number | null })[] { + const rows = getSqliteDb() + .prepare( + `SELECT g.*, (SELECT MIN(seq) FROM mls_log l WHERE l.group_id = g.group_id) AS oldest_seq + FROM mls_groups g + JOIN conversation_members m ON m.conversation_id = g.conversation_id + WHERE m.server_user_id = ?`, + ) + .all(serverUserId) as Record[]; + return rows.map((r) => ({ ...rowToGroup(r), oldestSeq: r.oldest_seq == null ? null : Number(r.oldest_seq) })); +} + +/** With the conversation it belongs to. A DM id comes from the pair, so a group left + behind would be handed to the next conversation those two open. */ +export function dropMlsGroupForConversation(conversationId: string): void { + const group = getMlsGroupForConversation(conversationId); + if (!group) return; + const db = getSqliteDb(); + db.prepare(`DELETE FROM mls_log WHERE group_id = ?`).run(group.groupId); + db.prepare(`DELETE FROM mls_welcomes WHERE group_id = ?`).run(group.groupId); + db.prepare(`DELETE FROM mls_groups WHERE group_id = ?`).run(group.groupId); +} + +// ── The log ───────────────────────────────────────────────────────────── + +export interface MlsAppend { + groupId: string; + epoch: number; + senderServerUserId: string; + senderDeviceId: string; + data: Uint8Array; +} + +export interface MlsWelcomeFor { + serverUserId: string; + deviceId: string; + data: Uint8Array; +} + +/** How far back a resend is recognised. A retry comes seconds after the first try. */ +const DUPLICATE_WINDOW = 100; + +/* The same bytes again, from a client that never heard back. Written once, so a + receiver doesn't fail on a generation it has already used. */ +function recentDuplicate(group: MlsGroup, data: Uint8Array): { seq: number; epoch: number; created_at: string } | undefined { + return getSqliteDb() + .prepare(`SELECT seq, epoch, created_at FROM mls_log WHERE group_id = ? AND seq > ? AND data = ? LIMIT 1`) + .get(group.groupId, group.headSeq - DUPLICATE_WINDOW, data) as + | { seq: number; epoch: number; created_at: string } + | undefined; +} + +export type MlsCommitResult = + | { accepted: true; seq: number; epoch: number; createdAt: string; welcomeIds: string[]; duplicate: boolean } + | { accepted: false; reason: "no_group" } + | { accepted: false; reason: "stale_epoch"; epoch: number; headSeq: number }; + +function insertEntry(kind: MlsLogKind, a: MlsAppend, seq: number, at: string): void { + getSqliteDb() + .prepare( + `INSERT INTO mls_log (group_id, seq, kind, epoch, sender_server_user_id, sender_device_id, data, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .run(a.groupId, seq, kind, a.epoch, a.senderServerUserId, a.senderDeviceId, a.data, at); +} + +/** + * One commit per epoch: the first one built on the current epoch is written and moves + * the group on. Any other is refused with the epoch it should have used. + */ +export function appendMlsCommit(a: MlsAppend, welcomes: MlsWelcomeFor[] = [], now = new Date()): MlsCommitResult { + return inTransaction(() => { + const db = getSqliteDb(); + const group = getMlsGroup(a.groupId); + if (!group) return { accepted: false as const, reason: "no_group" as const }; + const dup = recentDuplicate(group, a.data); + if (dup) { + const epoch = Number(dup.epoch) + 1; + return { accepted: true as const, seq: Number(dup.seq), epoch, createdAt: dup.created_at, welcomeIds: [], duplicate: true }; + } + if (a.epoch !== group.epoch) { + return { accepted: false as const, reason: "stale_epoch" as const, epoch: group.epoch, headSeq: group.headSeq }; + } + + const at = toIso(now); + const seq = group.headSeq + 1; + insertEntry("commit", a, seq, at); + db.prepare(`UPDATE mls_groups SET epoch = ?, head_seq = ? WHERE group_id = ?`).run(group.epoch + 1, seq, a.groupId); + + const insertWelcome = db.prepare( + `INSERT INTO mls_welcomes (welcome_id, server_user_id, device_id, group_id, data, created_at) + VALUES (?, ?, ?, ?, ?, ?)`, + ); + const welcomeIds = welcomes.map((w) => { + const id = randomUUID(); + insertWelcome.run(id, w.serverUserId, w.deviceId, a.groupId, w.data, at); + return id; + }); + return { accepted: true as const, seq, epoch: group.epoch + 1, createdAt: at, welcomeIds, duplicate: false }; + }); +} + +export type MlsMessageResult = + | { accepted: true; seq: number; createdAt: string; duplicate: boolean } + | { accepted: false; reason: "no_group" } + | { accepted: false; reason: "future_epoch"; epoch: number }; + +/** Not ordered by epoch, but none can come from an epoch that hasn't happened yet. */ +export function appendMlsMessage( + kind: Exclude, + a: MlsAppend, + now = new Date(), +): MlsMessageResult { + return inTransaction(() => { + const group = getMlsGroup(a.groupId); + if (!group) return { accepted: false as const, reason: "no_group" as const }; + if (a.epoch > group.epoch) return { accepted: false as const, reason: "future_epoch" as const, epoch: group.epoch }; + const dup = recentDuplicate(group, a.data); + if (dup) return { accepted: true as const, seq: Number(dup.seq), createdAt: dup.created_at, duplicate: true }; + + const at = toIso(now); + const seq = group.headSeq + 1; + insertEntry(kind, a, seq, at); + getSqliteDb().prepare(`UPDATE mls_groups SET head_seq = ? WHERE group_id = ?`).run(seq, a.groupId); + return { accepted: true as const, seq, createdAt: at, duplicate: false }; + }); +} + +/** Oldest first, after the cursor. */ +export function listMlsLog(groupId: string, afterSeq: number, limit: number): MlsLogEntry[] { + const rows = getSqliteDb() + .prepare(`SELECT * FROM mls_log WHERE group_id = ? AND seq > ? ORDER BY seq LIMIT ?`) + .all(groupId, afterSeq, limit) as Record[]; + return rows.map(rowToEntry); +} + +export function oldestMlsSeq(groupId: string): number | null { + const row = getSqliteDb().prepare(`SELECT MIN(seq) AS s FROM mls_log WHERE group_id = ?`).get(groupId) as { + s: number | null; + }; + return row.s == null ? null : Number(row.s); +} + +// ── Welcomes ──────────────────────────────────────────────────────────── + +export function listMlsWelcomes(serverUserId: string, deviceId: string): MlsWelcome[] { + const rows = getSqliteDb() + .prepare( + `SELECT w.welcome_id, w.group_id, w.device_id, w.data, w.created_at, g.conversation_id + FROM mls_welcomes w LEFT JOIN mls_groups g ON g.group_id = w.group_id + WHERE w.server_user_id = ? AND w.device_id = ? + ORDER BY w.created_at, w.rowid`, + ) + .all(serverUserId, deviceId) as Record[]; + return rows.map((r) => ({ + welcomeId: r.welcome_id as string, + groupId: r.group_id as string, + conversationId: (r.conversation_id as string | null) ?? null, + deviceId: r.device_id as string, + data: r.data as Uint8Array, + createdAt: r.created_at as string, + })); +} + +/** Only this device's own: an id belonging to anybody else deletes nothing. */ +export function deleteMlsWelcomes(serverUserId: string, deviceId: string, welcomeIds: string[]): number { + const del = getSqliteDb().prepare( + `DELETE FROM mls_welcomes WHERE welcome_id = ? AND server_user_id = ? AND device_id = ?`, + ); + let n = 0; + for (const id of welcomeIds) n += Number(del.run(id, serverUserId, deviceId).changes); + return n; +} + +// ── Retention ─────────────────────────────────────────────────────────── + +/** + * Drops ciphertext, Welcomes and claimed package refs older than the cutoff, and every + * group whose conversation is gone. A kept group's epoch and head_seq never go back. + */ +export function sweepMls(cutoff: Date): { log: number; welcomes: number; keyPackages: number; groups: number } { + return inTransaction(() => { + const db = getSqliteDb(); + const before = toIso(cutoff); + const orphans = db + .prepare( + `SELECT conversation_id FROM mls_groups g + WHERE NOT EXISTS (SELECT 1 FROM conversations c WHERE c.conversation_id = g.conversation_id)`, + ) + .all() as { conversation_id: string }[]; + for (const { conversation_id } of orphans) dropMlsGroupForConversation(conversation_id); + + const log = db.prepare(`DELETE FROM mls_log WHERE created_at < ?`).run(before); + const welcomes = db.prepare(`DELETE FROM mls_welcomes WHERE created_at < ?`).run(before); + const keyPackages = db + .prepare(`DELETE FROM mls_key_packages WHERE claimed_at IS NOT NULL AND claimed_at < ?`) + .run(before); + return { + log: Number(log.changes), + welcomes: Number(welcomes.changes), + keyPackages: Number(keyPackages.changes), + groups: orphans.length, + }; + }); +}