package main import ( "crypto/ed25519" "crypto/rand" "crypto/sha256" "crypto/subtle" "database/sql" "encoding/base64" "encoding/hex" "errors" "log" "path/filepath" "strconv" "strings" "time" _ "modernc.org/sqlite" ) // defaultGroupName is used until an admin sets one. "lchat" is the // implementation name, never a public one — an unnamed space presents as the // product, Eleven (PWA install name, page title, native add-chat screens). const defaultGroupName = "Eleven" // getGroupName returns the chat's display name (used as the PWA/app name, page // title, and chat header), falling back to a default until an admin sets one. func (db *DB) getGroupName() string { v, ok, _ := db.getConfig("group_name") if !ok || strings.TrimSpace(v) == "" { return defaultGroupName } return v } func (db *DB) setGroupName(name string) error { return db.setConfig("group_name", name) } // groupNameSet reports whether an admin explicitly named the instance (vs the // compiled-in default) — the setup conversation asks fresh instead of offering // a default to keep. A sealed instance is always named: the name lives in the // branding blob and the plaintext column is deliberately empty. func (db *DB) groupNameSet() bool { if db.brandingSealed() { return true } v, ok, _ := db.getConfig("group_name") return ok && strings.TrimSpace(v) != "" } // getAdminName / getAdminEmail return the optional administrator identity an // admin may publish to members (who runs this space + how to reach them). Shown // by Eleven's guided tour and answers. Plaintext branding like the group name — // admin-authored, never user content — and empty until set. Served only to // authenticated members (see getGroup), not on the public onboarding read. func (db *DB) getAdminName() string { v, _, _ := db.getConfig("admin_name"); return v } func (db *DB) getAdminEmail() string { v, _, _ := db.getConfig("admin_email"); return v } func (db *DB) setAdminName(v string) error { return db.setConfig("admin_name", v) } func (db *DB) setAdminEmail(v string) error { return db.setConfig("admin_email", v) } // getGroupDescription / setGroupDescription hold the space's free-text // description. Plaintext like the admin name/email above — admin-authored // operator metadata, never user content — so the keyless admin console can // edit it and the join card can show an invitee what they're joining. Served // to members and to valid invite holders, not on the public onboarding read. func (db *DB) getGroupDescription() string { v, _, _ := db.getConfig("group_description") return v } func (db *DB) setGroupDescription(v string) error { return db.setConfig("group_description", v) } // setupComplete gates the first-run Eleven setup conversation: false until the // admin finishes (or dismisses) it. See docs/conversational-onboarding.md. func (db *DB) setupComplete() bool { v, _, _ := db.getConfig("setup_complete") return v == "1" } func (db *DB) setSetupComplete(on bool) error { return db.setConfig("setup_complete", boolStr(on)) } // getInstanceAvatar returns the admin-set instance branding photo (bytes + MIME + // an updated-at stamp for cache validation), or ok=false when none is set. Like the // group name, this is plaintext branding the server serves in the clear. func (db *DB) getInstanceAvatar() (data []byte, mime string, updatedAt int64, ok bool) { err := db.sql.QueryRow("SELECT data, mime, updated_at FROM instance_avatar WHERE id = 1"). Scan(&data, &mime, &updatedAt) if err == sql.ErrNoRows { return nil, "", 0, false } if err != nil { return nil, "", 0, false } return data, mime, updatedAt, true } // setInstanceAvatar stores (or replaces) the instance branding photo. func (db *DB) setInstanceAvatar(data []byte, mime string, updatedAt int64) error { _, err := db.sql.Exec( `INSERT INTO instance_avatar(id, data, mime, updated_at) VALUES(1, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET data = excluded.data, mime = excluded.mime, updated_at = excluded.updated_at`, data, mime, updatedAt, ) return err } // clearInstanceAvatar removes the branding photo (reverting to the default icon). func (db *DB) clearInstanceAvatar() error { _, err := db.sql.Exec("DELETE FROM instance_avatar WHERE id = 1") return err } // usersCanInvite reports whether ordinary members (not just the admin) may invite // new people. Off by default — the safest posture — and loosened only as an // explicit admin opt-in. func (db *DB) usersCanInvite() bool { v, ok, _ := db.getConfig("user_invites_enabled") return ok && v == "1" } func (db *DB) setUsersCanInvite(on bool) error { v := "0" if on { v = "1" } return db.setConfig("user_invites_enabled", v) } // Per-instance feature gates: whether this space offers AI replies, bot // creation, and (member/mini) apps. All default ON — an absent key reads // true — so existing spaces keep today's behavior and turning one off is an // explicit admin opt-out. The server enforces each gate at its endpoints; // clients only hide the entry points. func (db *DB) aiEnabled() bool { v, ok, _ := db.getConfig("ai_enabled") return !ok || v == "1" } // aiEnabledExplicitly distinguishes "on because nothing was ever written" // (the compat default above) from "an admin explicitly said yes". The AI // opt-in signal (#858) keys on the latter: a fresh space's first "Bring // Eleven in" must count as consent even though the implicit default already // read true — and the default-on reading itself is harmless there, because // no sidecar exists until that explicit consent asks for one. func (db *DB) aiEnabledExplicitly() bool { v, ok, _ := db.getConfig("ai_enabled") return ok && v == "1" } func (db *DB) setAIEnabled(on bool) error { return db.setConfig("ai_enabled", boolStr(on)) } func (db *DB) botsEnabled() bool { v, ok, _ := db.getConfig("bots_enabled") return !ok || v == "1" } func (db *DB) setBotsEnabled(on bool) error { return db.setConfig("bots_enabled", boolStr(on)) } func (db *DB) appsEnabled() bool { v, ok, _ := db.getConfig("apps_enabled") return !ok || v == "1" } func (db *DB) setAppsEnabled(on bool) error { return db.setConfig("apps_enabled", boolStr(on)) } // memberAppsEnabled gates one slice of the apps feature: members building and // installing their OWN apps (App Studio deploys + bot builds). Narrower than // appsEnabled — a space can keep the built-in apps while switching off member // authoring. Like the others it only stops new use; apps already deployed keep // working. func (db *DB) memberAppsEnabled() bool { v, ok, _ := db.getConfig("member_apps_enabled") return !ok || v == "1" } func (db *DB) setMemberAppsEnabled(on bool) error { return db.setConfig("member_apps_enabled", boolStr(on)) } // spaceInstallsAdmins: whether adding an app FOR THE WHOLE SPACE is an // admin-only act (docs/space-apps.md § controls). Default OFF — any member, // attributed, mirroring group apps. Unlike the enable-* gates this is policy // the CLIENT enforces (the install registry is E2EE fold state the server // can't referee): served as page meta, honored by the sheet, @everyone, and // spaceAppsFor's read-time filter. func (db *DB) spaceInstallsAdmins() bool { v, _, _ := db.getConfig("space_installs_admins") return v == "1" } func (db *DB) setSpaceInstallsAdmins(on bool) error { return db.setConfig("space_installs_admins", boolStr(on)) } // Update settings: whether this instance follows its release channel (see // docs/server-releases.md). Unlike usersCanInvite, the booleans default ON — // an absent key reads true — so a fresh instance keeps itself current unless // the admin opts out. The box updater reads these via GET /api/update-prefs. func (db *DB) updateAuto() bool { v, ok, _ := db.getConfig("update_auto") return !ok || v == "1" } func (db *DB) setUpdateAuto(on bool) error { return db.setConfig("update_auto", boolStr(on)) } func (db *DB) updateDownload() bool { v, ok, _ := db.getConfig("update_download") return !ok || v == "1" } func (db *DB) setUpdateDownload(on bool) error { return db.setConfig("update_download", boolStr(on)) } // updateChannel is the release channel this instance tracks; anything unset or // unrecognised (e.g. a channel later renamed) falls back to stable. func (db *DB) updateChannel() string { v, _, _ := db.getConfig("update_channel") if !validChannel(v) { return "stable" } return v } func (db *DB) setUpdateChannel(c string) error { return db.setConfig("update_channel", c) } // updatePin, when set, freezes this instance on one exact version — the // auto-updater leaves a pinned instance alone entirely (pin beats channel). // This is how a canary declares "I'm deliberately on this build": without it, // an hourly pass would stomp any manually-deployed version back to the channel // head. Empty = not pinned; an unparseable stored value reads as unpinned. func (db *DB) updatePin() string { v, _, _ := db.getConfig("update_pin") if !validVersion(v) { return "" } return v } func (db *DB) setUpdatePin(v string) error { return db.setConfig("update_pin", v) } // updateMajor, when set, keeps this instance on one major version: it follows // its channel for minors and patches but a release whose semver crosses the // major is held, never auto-installed (majorHold, release.go). Empty = follow // everything; a garbled stored value reads as unset — like updatePin, it must // not freeze anything silently. func (db *DB) updateMajor() string { v, _, _ := db.getConfig("update_major") if !validMajor(v) { return "" } return v } func (db *DB) setUpdateMajor(v string) error { return db.setConfig("update_major", v) } // charterNextDue is the sidecar-reported earliest next run across this space's // member-bot charters (unix seconds; 0 or absent = none). Charters themselves // are instance-key-encrypted blobs the server can't read (memberbots.go), so // the charter host reports this one timestamp so a hibernating instance can be // woken in time to run them (see getNextDue / postCharterNextDue). A config // key, not a table: it's a single space-wide value with last-write-wins // semantics, exactly the update_pin idiom — and no schema change at all. func (db *DB) charterNextDue() int64 { v, _, _ := db.getConfig("charter_next_due") n, _ := strconv.ParseInt(v, 10, 64) if n < 0 { return 0 } return n } func (db *DB) setCharterNextDue(due int64) error { return db.setConfig("charter_next_due", strconv.FormatInt(due, 10)) } // DB wraps the SQLite connection. Note: the server never stores message // plaintext or any encryption key. Messages live here only as opaque // ciphertext blobs produced by the browser. type DB struct { sql *sql.DB // blobs holds attachment/photo bytes (see putBlob / the upload handler). // It's the blobStore seam: the default is on-disk files next to the DB // (-blobs/), but nothing outside blobstore.go may assume that — the // blobs *table* keeps the metadata (size, thread tag) either way. blobs blobStore // vapidPath is the sidecar keyfile (.vapid) holding the Web Push VAPID // pair — outside the DB so hibernation's S3 replica never carries a private // key (see ensureVAPID in push.go). vapidPath string } func openDB(path string) (*DB, error) { d, err := sql.Open("sqlite", path) if err != nil { return nil, err } // One writer at a time keeps WAL happy and avoids "database is locked". d.SetMaxOpenConns(1) for _, p := range []string{ "PRAGMA journal_mode=WAL", "PRAGMA busy_timeout=5000", "PRAGMA synchronous=NORMAL", "PRAGMA foreign_keys=ON", } { if _, err := d.Exec(p); err != nil { return nil, err } } disk, err := newDiskBlobStore(strings.TrimSuffix(path, filepath.Ext(path)) + "-blobs") if err != nil { return nil, err } // The blobStore seam's switch (#550 Phase 1): LCHAT_BLOB_S3_BUCKET set // puts NEW attachment bytes in the object store, with the disk dir as the // fallback for pre-flip blobs and the staging spool. Unset = today's // on-disk layout, byte-identical behavior. LCHAT_BLOB_S3_MIRROR_* beside // it adds the second copy on a second provider (federated blobs Phase A, // #992 — the mirror sweep in blobmirror.go keeps it converged). var blobs blobStore = disk if cfg := s3BlobConfigFromEnv(); cfg != nil { host := strings.TrimSuffix(filepath.Base(path), filepath.Ext(path)) s3s, err := newS3BlobStore(cfg, disk.dir, disk, host) if err != nil { return nil, err } blobs = s3s log.Printf("blobs: s3 store on (bucket %s, prefix %s)", cfg.Bucket, s3s.prefix) if mcfg := s3BlobMirrorConfigFromEnv(); mcfg != nil { mirror, err := newS3BlobStore(mcfg, disk.dir, nil, host) if err != nil { return nil, err } blobs = &mirroredBlobStore{primary: s3s, mirror: mirror} log.Printf("blobs: mirror store on (bucket %s, prefix %s)", mcfg.Bucket, mirror.prefix) } } else if s3BlobMirrorConfigFromEnv() != nil { // A mirror with no primary is a config mistake, not a mode — the // disk layout's second copy is the fleet backup's job. return nil, errors.New("LCHAT_BLOB_S3_MIRROR_* set without LCHAT_BLOB_S3_BUCKET") } db := &DB{sql: d, blobs: blobs, vapidPath: strings.TrimSuffix(path, filepath.Ext(path)) + ".vapid"} if err := db.migrate(); err != nil { return nil, err } return db, nil } func (db *DB) migrate() error { _, err := db.sql.Exec(` CREATE TABLE IF NOT EXISTS config ( key TEXT PRIMARY KEY, value TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at INTEGER NOT NULL, blob TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS push_subscriptions ( endpoint TEXT PRIMARY KEY, token TEXT NOT NULL, p256dh TEXT NOT NULL, auth TEXT NOT NULL, created_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS apns_tokens ( apns_token TEXT PRIMARY KEY, token TEXT NOT NULL, sandbox INTEGER NOT NULL DEFAULT 0, created_at INTEGER NOT NULL ); -- Attachment/photo bytes. Two storage modes coexist so older DBs keep working: -- * disk-backed (new): bytes live at blobDir/id; the data column is empty, size>0. -- * legacy (pre-disk): bytes live in the data column; no file. getBlob falls back. -- thread_id ties a blob to its thread for GC + quota (NULL for legacy/branding). CREATE TABLE IF NOT EXISTS blobs ( id TEXT PRIMARY KEY, data BLOB NOT NULL, size INTEGER NOT NULL, thread_id TEXT, created_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS users ( id TEXT PRIMARY KEY, label TEXT NOT NULL DEFAULT '', created_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS key_invites ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL, wrapped TEXT NOT NULL, created_at INTEGER NOT NULL, expires_at INTEGER, used_at INTEGER ); CREATE TABLE IF NOT EXISTS device_tokens ( token_hash TEXT PRIMARY KEY, user_id TEXT NOT NULL, label TEXT NOT NULL DEFAULT '', created_at INTEGER NOT NULL ); -- Link-preview cache, sealed at rest (docs/no-plaintext.md): the row key is a -- salted hash of the URL and the card is AES-GCM ciphertext under a key derived -- from the URL itself (convergent — only someone who knows the URL can find or -- open a row; the server derives it transiently at request time, when it sees -- the URL anyway). fetched_at stays plaintext for TTL eviction — a timestamp, -- like messages.created_at. CREATE TABLE IF NOT EXISTS link_previews_sealed ( url_hash TEXT PRIMARY KEY, blob TEXT NOT NULL, fetched_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS threads ( id TEXT PRIMARY KEY, kind TEXT NOT NULL, -- self | dm | group name_blob TEXT, -- client-encrypted group name (null for dm/self) created_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS thread_members ( thread_id TEXT NOT NULL, user_id TEXT NOT NULL, created_at INTEGER NOT NULL, PRIMARY KEY (thread_id, user_id) ); CREATE TABLE IF NOT EXISTS thread_keys ( thread_id TEXT NOT NULL, -- the thread's AES key, wrapped to each user_id TEXT NOT NULL, -- member's user public key (server-blind) wrapped TEXT NOT NULL, PRIMARY KEY (thread_id, user_id) ); CREATE TABLE IF NOT EXISTS dm_pairs ( user_a TEXT NOT NULL, -- canonical (user_a < user_b) so there is user_b TEXT NOT NULL, -- exactly one DM thread per pair thread_id TEXT NOT NULL, PRIMARY KEY (user_a, user_b) ); -- one shared instance key, wrapped to each member's pubkey, for all-members -- content like profile photos. The server stays blind. CREATE TABLE IF NOT EXISTS instance_keys ( user_id TEXT PRIMARY KEY, wrapped TEXT NOT NULL ); -- Tiers (docs/community-and-tiers.md): admin-curated member sets that CARRY -- KEY MATERIAL — the opposite of users.role, which by invariant carries none. -- A member holds a set of tiers; each tier has its own AES key (tier_keys), -- and tier-gated open groups wrap their thread key under it. The name is -- plaintext BY DESIGN, like branding: admin-set space policy, never member -- content, and it must render where no key exists yet (invite claim screens, -- later checkout). CREATE TABLE IF NOT EXISTS tiers ( id TEXT PRIMARY KEY, name TEXT NOT NULL, created_at INTEGER NOT NULL ); -- who's in a tier, and which assigner put them there ('admin' today; -- 'invite' when cohorts land, billing sources later — the assignment axis of -- the design doc; the machinery below this line never cares which). CREATE TABLE IF NOT EXISTS user_tiers ( tier_id TEXT NOT NULL, user_id TEXT NOT NULL, source TEXT NOT NULL DEFAULT 'admin', created_at INTEGER NOT NULL, PRIMARY KEY (tier_id, user_id) ); -- each tier's key wrapped to each member's pubkey — instance_keys, per tier. -- Client-side wraps only; the server stores blobs it cannot open. CREATE TABLE IF NOT EXISTS tier_keys ( tier_id TEXT NOT NULL, user_id TEXT NOT NULL, wrapped TEXT NOT NULL, PRIMARY KEY (tier_id, user_id) ); -- refunded/disputed lifetime purchases (payment intent ids). A refund deletes -- the membership row — and with it the per-sub ordering watermark — so this -- tombstone is what stops a REDELIVERED grant (Stripe is at-least-once) from -- silently re-buying the level. Subscriptions don't need it: their redelivered -- receipts resolve against provider truth (SubState); a payment intent has no -- equivalent living state to ask about. CREATE TABLE IF NOT EXISTS revoked_payments ( payment_intent TEXT PRIMARY KEY, revoked_at INTEGER NOT NULL ); -- per-user "closed" chats: a thread hidden from that user's list (synced across -- their devices). Just a marker — no content. Cleared for everyone on new -- activity (see insertMessage) so a closed chat returns when it's messaged. CREATE TABLE IF NOT EXISTS hidden_threads ( user_id TEXT NOT NULL, thread_id TEXT NOT NULL, PRIMARY KEY (user_id, thread_id) ); -- per-user pinned chats: threads that user keeps at the top of their list -- (synced across their devices). Just a marker — no content. CREATE TABLE IF NOT EXISTS pinned_threads ( user_id TEXT NOT NULL, thread_id TEXT NOT NULL, PRIMARY KEY (user_id, thread_id) ); -- per-user muted chats: threads that user gets no notifications for (synced -- across their devices). The server consults it only to SKIP push sends — the -- same routing-preference class as hidden/pinned, never content. Other members -- never see it. CREATE TABLE IF NOT EXISTS muted_threads ( user_id TEXT NOT NULL, thread_id TEXT NOT NULL, PRIMARY KEY (user_id, thread_id) ); -- a closed DM (issue #578): either party ended the conversation. The thread -- and its history stay (the words belong to both people), but the server -- refuses new messages/edits/typing into it until someone reopens it — a -- client-side-only close would be bypassable. Connection metadata only: who -- closed and when, never why. CREATE TABLE IF NOT EXISTS closed_dms ( thread_id TEXT PRIMARY KEY, closed_by TEXT NOT NULL, created_at INTEGER NOT NULL ); -- blocks (issue #578): "blocker cut contact with blocked" — the server-side -- half of a block (the client side is a mute). Consulted at DM creation and -- reopen, group adds, and push fan-out; blocking also closes any existing DM. -- Connection metadata only (the dm_pairs class): the server learns that two -- members are no longer connectable, never why, and the blocked party gets -- the same generic refusal a closed DM door gives. CREATE TABLE IF NOT EXISTS blocks ( blocker_id TEXT NOT NULL, blocked_id TEXT NOT NULL, created_at INTEGER NOT NULL, PRIMARY KEY (blocker_id, blocked_id) ); -- bans (issue #578): removal that stays removed. Bans key on the member's -- PUBLIC KEY — the one durable handle in an E2EE roster — so the same key -- can't re-register through any invite (admin, member, or cohort). Honest -- limit, documented in docs/moderation.md: a determined person with a fresh -- invite and a fresh keypair is a new stranger; a ban stops re-entry, not -- reincarnation — pair it with pausing member invites during trouble. CREATE TABLE IF NOT EXISTS bans ( public_key TEXT PRIMARY KEY, user_id TEXT NOT NULL, banned_by TEXT NOT NULL, banned_at INTEGER NOT NULL ); -- reports (issue #578): the moderation queue's METADATA — who reported whom, -- where, and what became of it. The report's CONTENT (category, words, quoted -- evidence) never lands here: it travels in report_blobs, sealed to each -- admin's public key, unreadable by the server. Reports are testimony, not -- proof — the queue UI says so out loud — and the row here is the same -- connection-level class as thread membership. CREATE TABLE IF NOT EXISTS reports ( id TEXT PRIMARY KEY, created_at INTEGER NOT NULL, reporter_id TEXT NOT NULL, reported_id TEXT NOT NULL, thread_id TEXT NOT NULL DEFAULT '', status TEXT NOT NULL DEFAULT 'open', action TEXT NOT NULL DEFAULT '', resolved_by TEXT NOT NULL DEFAULT '', resolved_at INTEGER NOT NULL DEFAULT 0 ); CREATE TABLE IF NOT EXISTS report_blobs ( report_id TEXT NOT NULL, admin_id TEXT NOT NULL, blob TEXT NOT NULL, PRIMARY KEY (report_id, admin_id) ); -- audit_log (issue #578): every moderation action — who did what to whom, -- when, off which report. Append-only action metadata (the roster's class), -- served to admins only; any quoted content stays sealed in report_blobs, -- never here. CREATE TABLE IF NOT EXISTS audit_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts INTEGER NOT NULL, actor_id TEXT NOT NULL, action TEXT NOT NULL, subject_id TEXT NOT NULL DEFAULT '', thread_id TEXT NOT NULL DEFAULT '', report_id TEXT NOT NULL DEFAULT '', detail TEXT NOT NULL DEFAULT '' ); -- per-user read watermark: the time of the newest message that user has read in -- a thread, so "read here" clears the unread dot on their other devices too. -- A timestamp only — no content, and it never affects other members. CREATE TABLE IF NOT EXISTS read_marks ( user_id TEXT NOT NULL, thread_id TEXT NOT NULL, last_read_at INTEGER NOT NULL, PRIMARY KEY (user_id, thread_id) ); -- Console-managed config for an agent (bot), keyed by its user id: the model, -- system prompt (personality), and reference docs the sidecar applies. This is -- operator config, not message content; the sidecar pulls it with its own token. CREATE TABLE IF NOT EXISTS bot_configs ( user_id TEXT PRIMARY KEY, model TEXT NOT NULL DEFAULT '', system_prompt TEXT NOT NULL DEFAULT '', docs TEXT NOT NULL DEFAULT '', updated_at INTEGER NOT NULL ); -- Mini-app scheduling (docs: web/static/apps/README.md): a "future message" — -- client-encrypted ciphertext the server holds until deliver_at, then inserts -- into the thread's message log and fans out exactly like a posted message. -- The server learns only (thread, time); the payload stays opaque. CREATE TABLE IF NOT EXISTS scheduled_messages ( id TEXT PRIMARY KEY, thread_id TEXT NOT NULL, deliver_at INTEGER NOT NULL, blob TEXT NOT NULL, author_id TEXT NOT NULL, -- creator's user id (delivery attribution) created_at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS idx_scheduled_deliver ON scheduled_messages(deliver_at); -- Mini-app state snapshots: one client-encrypted blob per (thread, slot), with a -- version counter for optimistic concurrency (two devices saving concurrently — -- the loser gets a conflict and refolds). The server can't read it; it's the same -- trust boundary as the thread's messages. CREATE TABLE IF NOT EXISTS thread_state ( thread_id TEXT NOT NULL, slot TEXT NOT NULL, blob TEXT NOT NULL, version INTEGER NOT NULL, updated_at INTEGER NOT NULL, PRIMARY KEY (thread_id, slot) ); -- One optional per-instance avatar: an admin-set branding photo shown where a -- person first meets the instance (onboarding, native add-chat, the home -- Save-passkey card). Like the group name, this is plaintext instance branding -- the server holds and serves in the clear — never user content, so it doesn't -- weaken E2E. Single row (id = 1); absent until an admin uploads one. CREATE TABLE IF NOT EXISTS instance_avatar ( id INTEGER PRIMARY KEY CHECK (id = 1), data BLOB NOT NULL, mime TEXT NOT NULL, updated_at INTEGER NOT NULL ); -- Connected third-party services (docs/connected-services.md): one opaque, -- client-encrypted credential blob per (member, service) — e.g. a scoped -- DNSimple token the member sealed to their OWN key. The server keys it by -- member + service and never reads it (same trust boundary as instance_keys). -- Actually *using* a credential is a separate, time-boxed grant the member seals -- to a bot; nothing usable rests here. CREATE TABLE IF NOT EXISTS connected_services ( user_id TEXT NOT NULL, service TEXT NOT NULL, blob TEXT NOT NULL, updated_at INTEGER NOT NULL, PRIMARY KEY (user_id, service) ); -- Custom mini-apps installed via the admin console (customapps.go): plain JS -- module source served to every member at /static/apps-custom/.js. This -- is admin-chosen CODE, not user content — the compiled-in apps ship in the -- clear the same way — so, like branding, it lives unencrypted. version bumps -- on every update so the admin list can tell revisions apart. CREATE TABLE IF NOT EXISTS custom_apps ( name TEXT PRIMARY KEY, source TEXT NOT NULL, enabled INTEGER NOT NULL DEFAULT 1, version INTEGER NOT NULL DEFAULT 1, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ); -- Member-built mini-apps (memberapps.go): unlike custom_apps, this IS user -- content, so the blob is client-encrypted under the instance key — -- { source, manifest } sealed before it reaches us. The id is a random slug, -- so the server never learns the app's name, code, or author-chosen anything. -- author_id records which member deployed it; owner_id records the member it -- belongs to (the same person for a hand-deployed app; the requester for one a -- bot built on their behalf). Either may update/delete it — that's the gate. CREATE TABLE IF NOT EXISTS member_apps ( id TEXT PRIMARY KEY, blob TEXT NOT NULL, author_id TEXT NOT NULL, owner_id TEXT, version INTEGER NOT NULL DEFAULT 1, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ); -- Version history: every deploy of an app records its blob here, keyed by -- (app_id, version). member_apps holds the CURRENT blob + version; this is the -- trail, so a member can restore an earlier one (which deploys as a new version). -- Opaque like the current blob — the server never opens it. Pruned to the last N. CREATE TABLE IF NOT EXISTS member_app_versions ( app_id TEXT NOT NULL, version INTEGER NOT NULL, blob TEXT NOT NULL, author_id TEXT NOT NULL, created_at INTEGER NOT NULL, based_on INTEGER, -- the version this one was reverted from ("based on vN"), or NULL PRIMARY KEY (app_id, version) ); -- Member bots (docs/member-bots.md): a member's autonomous bot as an opaque -- charter blob — persona + schedule + task, sealed under the instance key -- before it reaches us, exactly like a member app's { source, manifest }. -- The sidecar (an ordinary E2EE member) decrypts charters and runs them; the -- server learns only (random id, author, owner, size, times). CREATE TABLE IF NOT EXISTS member_bots ( id TEXT PRIMARY KEY, blob TEXT NOT NULL, author_id TEXT NOT NULL, owner_id TEXT, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ); -- Incoming webhooks (docs/webhooks.md, #610): each row is one webhook — an -- agent identity that turns inbound HTTP into encrypted typed events. Both -- blobs are opaque here: def_blob is sealed to the hooks sidecar's service -- identity (it alone can read the webhook's private key, member token, and -- URL secret); meta_blob is instance-key-encrypted (the admin app's list -- view). The server learns only (id, which agent user, sizes, times) — never -- a secret, a key, or which external service feeds it. deployed_at is the -- sidecar's plaintext "route is live" ack, 0 until the first poll picks it up. CREATE TABLE IF NOT EXISTS webhooks ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL, def_blob TEXT NOT NULL DEFAULT '', meta_blob TEXT NOT NULL DEFAULT '', deployed_at INTEGER NOT NULL DEFAULT 0, created_at INTEGER NOT NULL ); -- Tamper evidence (docs/tamper-evidence.md, #573): the append-only event -- journal. The messages table mutates on purpose (edits UPDATE, unsend hard- -- DELETEs), so checkpoints commit to this journal instead: every message -- insert/edit/delete appends one row IN THE SAME TRANSACTION. blob_hash is -- sha256 hex of the ciphertext — never the ciphertext itself, so unsend still -- destroys the blob; the journal retains only that an event happened. Rows are -- never updated or deleted. Adding a plaintext column here would break the one -- rule — don't. CREATE TABLE IF NOT EXISTS event_journal ( seq INTEGER PRIMARY KEY AUTOINCREMENT, ts INTEGER NOT NULL, kind TEXT NOT NULL, -- 'msg' | 'edit' | 'delete' msg_id INTEGER NOT NULL, thread_id TEXT NOT NULL, blob_hash TEXT NOT NULL -- '' for delete ); -- Signed checkpoint chain over the journal (checkpoint.go). "signed" is the -- full signed wrapper clients verify; the bare columns exist so the server -- can chain and page without re-parsing it. CREATE TABLE IF NOT EXISTS checkpoints ( id INTEGER PRIMARY KEY AUTOINCREMENT, seq_from INTEGER NOT NULL, seq_to INTEGER NOT NULL, root TEXT NOT NULL, chain TEXT NOT NULL, ts INTEGER NOT NULL, signed TEXT NOT NULL ); -- Two checkpoints claiming the same journal range would fork the chain — a -- second writer (e.g. two processes briefly overlapping on one DB during an -- activator wake race) must fail loudly, not fork silently. CREATE UNIQUE INDEX IF NOT EXISTS idx_checkpoints_seq_from ON checkpoints(seq_from); -- There is at most ONE space thread per instance (spacethread.go): the -- create-if-absent transaction relies on this to make a racing second create -- fail loudly instead of forking the space log into two threads. CREATE UNIQUE INDEX IF NOT EXISTS idx_one_space_thread ON threads(kind) WHERE kind = 'space';`) if err != nil { return err } // Additive column migrations (older DBs created `messages` without these). // author_hash is hashToken(senderToken) — lets the server enforce that only // the original author may edit, without ever seeing the (encrypted) author // name. edited_at is set when a message is edited in place. for _, c := range []string{ "ALTER TABLE messages ADD COLUMN author_hash TEXT", "ALTER TABLE messages ADD COLUMN edited_at INTEGER", // Multi-thread (docs/threads-and-dms.md): which thread a message belongs to, // and each user's ECDH public key (the roster). Both additive over old DBs. "ALTER TABLE messages ADD COLUMN thread_id TEXT", "ALTER TABLE users ADD COLUMN public_key TEXT", // Key transparency (#573, docs/tamper-evidence.md): the joiner's HMAC of // its public key under bind_key = HKDF(invite secret T). A T-holder (the // inviter) verifies it before trusting/wrapping to the member, catching a // server that substituted its own key. Opaque to the server (no T). "ALTER TABLE users ADD COLUMN bind_mac TEXT", // Profile photo: an encrypted blob id (encrypted with the instance key). "ALTER TABLE users ADD COLUMN avatar_blob TEXT", // A person's display name encrypted under the instance key (server-blind). // Bots keep their plaintext `label` (admin-authored); people move here so the // server never holds a readable name. Set by the member who holds the key. "ALTER TABLE users ADD COLUMN label_blob TEXT", // The profile photo's alt text, encrypted under the instance key like the // label — unlike the server photo's plaintext alt (branding), a member's // photo description is user content and never readable by the server. "ALTER TABLE users ADD COLUMN avatar_alt_blob TEXT", // Group photo: an encrypted blob id (encrypted with the thread key). "ALTER TABLE threads ADD COLUMN avatar_blob TEXT", // User invites: the pending DM thread an invite bootstraps the joiner into // (null for admin auth-only / add-device invites). See createUserInvite. "ALTER TABLE key_invites ADD COLUMN thread_id TEXT", // Proof-of-secret for claiming: hashToken of a value derived from the invite // secret T (which never reaches the server). Claiming requires presenting the // preimage, so reading the DB (which stores only the id + this hash) is not // enough to claim a pending invite. See deriveInvite in crypto.js. "ALTER TABLE key_invites ADD COLUMN claim_hash TEXT", // Legacy invites are removed (superseded by key-invites). Their tokens were // stored in clear and are no longer accepted; drop the table so a stolen DB // retains no plaintext credentials. Members re-authenticate via device tokens. "DROP TABLE IF EXISTS invites", // The reactions table is vestigial: reactions became encrypted message // events (the chat.react fold), so nothing writes or reads it anymore. // Legacy rows stored the reactor's RAW session token in the clear — drop // the table so a stolen DB retains no plaintext credentials (same // rationale as the invites drop above). "DROP TABLE IF EXISTS reactions", // The old link-preview cache stored plaintext URLs + OpenGraph cards // derived from message content, unevicted. Superseded by // link_previews_sealed (ciphertext at rest); it's a cache, so old rows // are simply dropped and re-fetched on demand. "DROP TABLE IF EXISTS link_previews", // Which user a push subscription belongs to, so removing a user can drop their // subscriptions (sign-out clears them by token; removal has only the user id). "ALTER TABLE push_subscriptions ADD COLUMN user_id TEXT", // Store only a HASH of the session token in the push/APNs tables — the raw // token is a live credential (cf. device_tokens.token_hash). Backfilled and // scrubbed below for rows created before this migration. "ALTER TABLE push_subscriptions ADD COLUMN token_hash TEXT", "ALTER TABLE apns_tokens ADD COLUMN token_hash TEXT", // Bot flag: a user driven by an agent (see agent/), not a person. Only // used to badge them in the UI — it grants no privilege. "ALTER TABLE users ADD COLUMN is_agent INTEGER NOT NULL DEFAULT 0", // Helpers catalog (docs/bots.md): admin-authored one-line description shown // to members, and the "show in members' Helpers list" flag. Catalog metadata // like the roster label — in clear, never message content. "ALTER TABLE bot_configs ADD COLUMN description TEXT NOT NULL DEFAULT ''", "ALTER TABLE bot_configs ADD COLUMN available_to_members INTEGER NOT NULL DEFAULT 0", // Listen mode (docs/bots.md): when a bot replies in a group — ''/'mention' // = only on an @mention (the default), 'name' = also when spoken to by // bare name (the sidecar still gates that through a classifier). "ALTER TABLE bot_configs ADD COLUMN listen TEXT NOT NULL DEFAULT ''", // Helper kind (docs/bots.md): 'ai' (LLM-backed — messages it sees are sent // to a model provider) vs 'bot' (deterministic — nothing leaves the server). // The sidecar SELF-DECLARES this on connect, so Eleven reads BOT while it // runs commands-only and AI once its model brain is live. Badge only; no // privilege. Default 'ai' so a pre-declare agent row reads as an assistant. "ALTER TABLE bot_configs ADD COLUMN kind TEXT NOT NULL DEFAULT 'ai'", // Admin-tweakable welcome DM text used by the guide (Eleven's welcomer). // '' = the shipped default; a literal '{name}' is replaced with the new // member's name. Operator copy, like the persona — never message content. "ALTER TABLE bot_configs ADD COLUMN welcome TEXT NOT NULL DEFAULT ''", // Blobs gain a thread tag (for thread-scoped GC + quota accounting). Older // blobs (branding, pre-disk photos) stay NULL and are never thread-GC'd. "ALTER TABLE blobs ADD COLUMN thread_id TEXT", // Federated blobs Phase A (#992): the ciphertext's sha256 (hex), // recorded at upload — the integrity anchor mirror and audit sweeps // verify against. '' = pre-hash row; the mirror sweep backfills it // (it reads the full bytes anyway). "ALTER TABLE blobs ADD COLUMN sha256 TEXT NOT NULL DEFAULT ''", // When the mirror store last received this blob (unix seconds; 0 = // unmirrored — what the mirror sweep selects on). Reset to 0 fleet-wide // when the mirror TARGET changes (documented beside the env vars); // Phase B's store registry replaces this with per-store placement. "ALTER TABLE blobs ADD COLUMN mirrored_at INTEGER NOT NULL DEFAULT 0", // Roles (docs/conversational-onboarding.md): 'admin' | 'moderator' | // 'member'. Authorisation only — a role grants no key material. The admin // bearer token remains as break-glass; role-admins reach the same admin API // with their own device token. Moderator is reserved (nothing enforces it // yet). Existing members default to 'member'. "ALTER TABLE users ADD COLUMN role TEXT NOT NULL DEFAULT 'member'", // A setup invite (minted by add-instance for a fresh instance) makes its // claimer the first admin. Ordinary invites never carry this. "ALTER TABLE key_invites ADD COLUMN grants_admin INTEGER NOT NULL DEFAULT 0", // Member app ownership: who the app belongs to (may differ from the // deployer — a bot deploys under its own author_id but sets owner_id to the // requester). Nullable: apps deployed before this stay owner-less and remain // managed by their author, as before. "ALTER TABLE member_apps ADD COLUMN owner_id TEXT", // The version a reverted deploy was based on ("based on vN"), or NULL. "ALTER TABLE member_app_versions ADD COLUMN based_on INTEGER", // Pinned presence: '' (automatic) or 'away' (the member set themselves // Away, so activity doesn't flip them back online). Connection metadata // like role — the server already holds the sockets — never content. "ALTER TABLE users ADD COLUMN presence TEXT NOT NULL DEFAULT ''", // A person's free-text "about me", encrypted under the instance key like // their name and photo alt — user content, never readable by the server. "ALTER TABLE users ADD COLUMN description_blob TEXT", // A member's private per-viewer name overrides ("I call them Dermo", // issue #577), sealed to their OWN public key — unlike the instance-key // blobs above, not even other members can open it; the server holds // ciphertext either way. "ALTER TABLE users ADD COLUMN nicknames_blob TEXT", // Compare-and-swap counter for the blob above (same optimistic // versioning as the mini-app state slots): a save from a stale read // 409s instead of clobbering another device's write. "ALTER TABLE users ADD COLUMN nicknames_version INTEGER NOT NULL DEFAULT 0", // Open "community" groups (docs/community-and-tiers.md): a group any // member can discover and self-join. visibility is 'invite' (default) // or 'open'; open_key is the thread's AES key encrypted under the // shared instance key — every member can unwrap it client-side, while // the server (which never holds the instance key) cannot. "ALTER TABLE threads ADD COLUMN visibility TEXT NOT NULL DEFAULT 'invite'", "ALTER TABLE threads ADD COLUMN open_key TEXT", // Default groups (Slack's #general): open groups every NEW member is // auto-joined to on their first device claim (existing members are // never force-joined). Admin-set, meaningful only while visibility is // 'open' — joining still needs no key material beyond open_key. "ALTER TABLE threads ADD COLUMN default_join INTEGER NOT NULL DEFAULT 0", // Cohort invites (docs/community-and-tiers.md): an invite may carry an // access level — its claimer is assigned the level on first-device // claim ('invite' source). Admin-set only. "ALTER TABLE key_invites ADD COLUMN tier_id TEXT", // Level visibility policy (admin-set, per level): 'shown' (on everyone's // profile — the default, = today's behaviour), 'hidden' (only the // level's own members see who's in it — non-members never do), or // 'user' (each member chooses via user_tiers.visible). NB: "hidden" // can only mean hidden-from-NON-members — a level's members share its // key, so they inherently know each other (E2E, not a choice). "ALTER TABLE tiers ADD COLUMN visibility TEXT NOT NULL DEFAULT 'shown'", // Per-member show/hide, consulted only when the level's policy is // 'user'. Default 1 = shown. "ALTER TABLE user_tiers ADD COLUMN visible INTEGER NOT NULL DEFAULT 1", // One-way mint marker: set (transactionally, with the first grants) // when a level's key is first created. "Zero key rows" must NOT mean // "never minted" — holders can all be removed later, and silently // minting a SECOND key would seal every group gated under the first. "ALTER TABLE tiers ADD COLUMN key_minted INTEGER NOT NULL DEFAULT 0", // Level-gated open groups: the tier whose members may discover/join // this open group ('' / NULL = every member; the open_key is then // wrapped under that TIER's key instead of the instance key). Fixed // once open_key exists — changing the gate would need a rekey the // never-overwrite rule forbids (close the group and make a new one). "ALTER TABLE threads ADD COLUMN open_tier TEXT", // Paid access levels (docs/community-and-tiers.md layer 3): an access // level can be sellable — members subscribe to it. Price is plaintext // commerce config (admin-set, like branding), in minor units (cents). // provider_ref is the payment provider's price id (Stripe price). All // unset → the level is admin/invite-assigned only (no subscribe path). "ALTER TABLE tiers ADD COLUMN price_cents INTEGER NOT NULL DEFAULT 0", "ALTER TABLE tiers ADD COLUMN currency TEXT NOT NULL DEFAULT ''", "ALTER TABLE tiers ADD COLUMN interval TEXT NOT NULL DEFAULT ''", // 'month' | 'year' "ALTER TABLE tiers ADD COLUMN provider_ref TEXT", // Subscription entitlement on a membership (source='stripe'): its // lifecycle. status '' for admin/invite assignments; 'active' | // 'past_due' | 'canceled' for paid ones. paid_until is the paid-through // time (the current period end). grace_deadline is when access is // finally revoked after lapse/cancel (paid_until + the grace window); // 0 = not lapsing. ext_sub/ext_customer are the provider's ids, so a // webhook can find the membership again. "ALTER TABLE user_tiers ADD COLUMN status TEXT NOT NULL DEFAULT ''", "ALTER TABLE user_tiers ADD COLUMN paid_until INTEGER NOT NULL DEFAULT 0", "ALTER TABLE user_tiers ADD COLUMN grace_deadline INTEGER NOT NULL DEFAULT 0", "ALTER TABLE user_tiers ADD COLUMN ext_sub TEXT", "ALTER TABLE user_tiers ADD COLUMN ext_customer TEXT", // The provider event `created` time last folded for this subscription — // so a redelivered or out-of-order webhook (Stripe is at-least-once and // can reorder) is ignored rather than resurrecting a revoked membership. "ALTER TABLE user_tiers ADD COLUMN ext_event_ts INTEGER NOT NULL DEFAULT 0", // Lifetime (one-time) purchases: a paid membership with no lifecycle — // pay once, keep access. The flag is what exempts the row from every // renewal-shaped sweep (grace, silent-lapse backstop, repair); the only // way out is a refund/chargeback event or an admin removal. "ALTER TABLE user_tiers ADD COLUMN lifetime INTEGER NOT NULL DEFAULT 0", // Delete tombstones (#581): unsending hard-deletes the row (the // ciphertext really goes) and inserts a fresh empty row whose // tombstone_of names the deleted id. The fresh autoincrement id lands // above every client's forward-sync watermark, so offline clients // replay the delete from ordinary catch-up — the live broadcast stays // the instant path. Stores only (thread, target id, deleted-at): // metadata the server transiently knew when it performed the delete. "ALTER TABLE messages ADD COLUMN tombstone_of INTEGER", // A member's private moderation preferences (muted people, for now), // sealed to their OWN public key like nicknames_blob — the server can't // read who muted whom, it just syncs the ciphertext across their devices. "ALTER TABLE users ADD COLUMN prefs_blob TEXT", "ALTER TABLE users ADD COLUMN prefs_version INTEGER NOT NULL DEFAULT 0", // DM privacy (issue #578): who may open a NEW direct message with this // member. '' = everyone in the space (the default, today's behaviour); // 'nobody' = no one — with a disclosed carve-out for admins and bots // (the operator must stay reachable; onboarding speaks through Eleven's // welcome DM). Enforced server-side at DM creation, where a client-only // check would be bypassable. A routing preference, never content. "ALTER TABLE users ADD COLUMN dm_privacy TEXT NOT NULL DEFAULT ''", // Invite lineage (issue #578): which member's invite brought this // person in ('' for admin/setup invites and everyone pre-dating the // column). Written at claim time; the report queue shows it so a // problem member's way in is visible. "ALTER TABLE users ADD COLUMN invited_by TEXT NOT NULL DEFAULT ''", } { if _, err := db.sql.Exec(c); err != nil && !strings.Contains(err.Error(), "duplicate column") { return err } } // Backfill each app's CURRENT version into its history, for apps deployed // before version tracking existed, so History always shows at least the // current version (never empty for an app that hasn't been updated since). if err := db.backfillMemberAppVersions(); err != nil { return err } // Baseline pre-journal messages into the event journal (tamper evidence) — // without this, editing or unsending a message that predates the journal // would append an 'edit'/'delete' row for a message the journal never saw, // which reads as tampering to a verifier. if err := db.backfillEventJournal(); err != nil { return err } // Established instances never ran the setup conversation — mark it complete // ONCE at upgrade time, so promoting someone to admin later doesn't drop them // into first-run onboarding on a live space. Guarded by its own marker: a // fresh instance (no devices yet) records the marker but leaves setup pending // for its first claimer, and a restart mid-setup doesn't re-run the check. if v, _, _ := db.getConfig("setup_backfilled"); v != "1" { var devices int db.sql.QueryRow("SELECT COUNT(*) FROM device_tokens").Scan(&devices) if devices > 0 { if err := db.setSetupComplete(true); err != nil { return err } } if err := db.setConfig("setup_backfilled", "1"); err != nil { return err } } // Older push/APNs rows stored the session token in clear; replace it with its // hash and scrub the plaintext so a stolen DB yields no live credentials. return db.backfillPushTokenHashes() } // backfillPushTokenHashes migrates any legacy plaintext push/APNs token to a hash // and clears the plaintext column, so the tables never retain a usable credential. func (db *DB) backfillPushTokenHashes() error { if err := db.backfillTokenHash("push_subscriptions", "endpoint"); err != nil { return err } return db.backfillTokenHash("apns_tokens", "apns_token") } // backfillTokenHash rewrites rows that still hold a plaintext token. table/keyCol // are in-code constants, never user input, so the string-built SQL is safe. func (db *DB) backfillTokenHash(table, keyCol string) error { rows, err := db.sql.Query( "SELECT " + keyCol + ", token FROM " + table + " WHERE (token_hash IS NULL OR token_hash = '') AND token != ''") if err != nil { return err } type legacy struct{ key, token string } var pending []legacy for rows.Next() { var r legacy if err := rows.Scan(&r.key, &r.token); err != nil { rows.Close() return err } pending = append(pending, r) } rows.Close() if err := rows.Err(); err != nil { return err } // Update after the read completes (single-connection SQLite dislikes writing // mid-iteration). for _, r := range pending { if _, err := db.sql.Exec( "UPDATE "+table+" SET token_hash = ?, token = '' WHERE "+keyCol+" = ?", hashToken(r.token), r.key, ); err != nil { return err } } return nil } // ---- link previews (cache, sealed at rest) ------------------------------- // // Rows are found by a salted hash of the URL and opened with a key derived // from the URL (see sealPreview/openPreview in unfurl.go). The per-instance // random salt only defeats precomputation across instances — it is not a // secret; the protection at rest is the URL's own entropy, which is exactly // the honest claim: bulk reading of the cache is gone, guessable URLs are // guessable (docs/no-plaintext.md § the plan). // previewSalt returns the instance's preview-cache salt, minting it on first // use (INSERT OR IGNORE + read-back keeps concurrent first calls agreeing). func (db *DB) previewSalt() (string, error) { salt, err := randToken() if err != nil { return "", err } if _, err := db.sql.Exec( "INSERT OR IGNORE INTO config(key, value) VALUES('preview_salt', ?)", salt); err != nil { return "", err } v, _, err := db.getConfig("preview_salt") return v, err } func (db *DB) getLinkPreview(u, lang string) (LinkPreview, bool, error) { salt, err := db.previewSalt() if err != nil { return LinkPreview{}, false, err } key := previewCacheKey(lang, u) var blob string var fetchedAt int64 err = db.sql.QueryRow( "SELECT blob, fetched_at FROM link_previews_sealed WHERE url_hash = ?", previewURLHash(salt, key), ).Scan(&blob, &fetchedAt) if err == sql.ErrNoRows { return LinkPreview{}, false, nil } if err != nil { return LinkPreview{}, false, err } pv, ok := openPreview(salt, key, blob) if !ok { // Undecryptable row (corruption): treat as a miss; the next put // overwrites it under the same hash. return LinkPreview{}, false, nil } pv.URL = u pv.FetchedAt = fetchedAt return pv, true, nil } func (db *DB) putLinkPreview(pv LinkPreview, lang string) error { salt, err := db.previewSalt() if err != nil { return err } key := previewCacheKey(lang, pv.URL) blob, err := sealPreview(salt, key, pv) if err != nil { return err } // Put-time TTL sweep: the cache self-prunes with activity, so previews of // long-forgotten URLs don't sit at rest (or in hibernation replicas) forever. db.sql.Exec("DELETE FROM link_previews_sealed WHERE fetched_at < ?", time.Now().Unix()-previewMaxAge) _, err = db.sql.Exec( `INSERT INTO link_previews_sealed(url_hash, blob, fetched_at) VALUES(?, ?, ?) ON CONFLICT(url_hash) DO UPDATE SET blob=excluded.blob, fetched_at=excluded.fetched_at`, previewURLHash(salt, key), blob, pv.FetchedAt, ) return err } // ---- config ------------------------------------------------------------- func (db *DB) getConfig(key string) (string, bool, error) { var v string err := db.sql.QueryRow("SELECT value FROM config WHERE key = ?", key).Scan(&v) if err == sql.ErrNoRows { return "", false, nil } if err != nil { return "", false, err } return v, true, nil } func (db *DB) setConfig(key, value string) error { _, err := db.sql.Exec( "INSERT INTO config(key, value) VALUES(?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value", key, value, ) return err } // setConfigTx sets a config value within a transaction. func setConfigTx(tx *sql.Tx, key, value string) error { _, err := tx.Exec( "INSERT INTO config(key, value) VALUES(?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value", key, value, ) return err } // ---- admin token -------------------------------------------------------- // // The admin token only authorises invite management. It is NOT an encryption // key — even with it, an attacker cannot read messages. We store only its hash. func hashToken(t string) string { h := sha256.Sum256([]byte(t)) return hex.EncodeToString(h[:]) } // validClaimSecret reports whether claimSecret is the preimage of the stored // invite claim_hash, compared in constant time. An invite with no claim_hash // (created before this migration) is unclaimable — we deliberately do not fall // back to unauthenticated claiming. func validClaimSecret(claimHash, claimSecret string) bool { if claimHash == "" || claimSecret == "" { return false } return subtle.ConstantTimeCompare([]byte(claimHash), []byte(hashToken(claimSecret))) == 1 } func randToken() (string, error) { b := make([]byte, 32) if _, err := rand.Read(b); err != nil { return "", err } return base64.RawURLEncoding.EncodeToString(b), nil } // ensureAdminToken sets up the admin token. If env is non-empty it is used. // Otherwise, if none exists yet, a fresh one is generated. Returns the // plaintext token only when a new one was generated (so it can be printed once). func (db *DB) ensureAdminToken(env string) (string, error) { if env != "" { return "", db.setConfig("admin_token_hash", hashToken(env)) } if _, ok, err := db.getConfig("admin_token_hash"); err != nil || ok { return "", err } tok, err := randToken() if err != nil { return "", err } return tok, db.setConfig("admin_token_hash", hashToken(tok)) } func (db *DB) resetAdminToken() (string, error) { tok, err := randToken() if err != nil { return "", err } return tok, db.setConfig("admin_token_hash", hashToken(tok)) } func (db *DB) checkAdmin(token string) bool { want, ok, err := db.getConfig("admin_token_hash") if err != nil || !ok { return false } return subtle.ConstantTimeCompare([]byte(hashToken(token)), []byte(want)) == 1 } // ---- users, key-invites, device tokens (invite redesign) ---------------- // // The server stores a client-WRAPPED group key per invite, keyed by an id the // client derives from a secret it never sends (see docs/invite-redesign.md). The // server can't unwrap. Claiming an invite mints a device token for ongoing auth. type User struct { ID string `json:"id"` Label string `json:"label"` CreatedAt int64 `json:"created_at"` // Devices is how many devices have claimed in (0 = invited but not joined). Devices int `json:"devices"` // JoinedAt is when they first claimed a device (nil until they join). JoinedAt *int64 `json:"joined_at,omitempty"` // Pending is true while an unused, unexpired invite link is still out there. Pending bool `json:"pending"` // PendingDM is true when that pending invite drops them into a DM (a member's // in-app invite), vs a bare admin invite. The admin doesn't re-send DM invites — // the inviting member does, in-app. PendingDM bool `json:"pending_dm"` // IsAgent marks a bot, so the console can badge it and offer "Edit bot". IsAgent bool `json:"is_agent,omitempty"` // Kind is "ai" or "bot" for agents (drives the admin AI/BOT badge); "" for people. Kind string `json:"kind,omitempty"` // Role is 'admin' | 'moderator' | 'member' (authorisation only, no keys). Role string `json:"role"` } func (db *DB) createUser(label string) (string, error) { id, err := randToken() if err != nil { return "", err } _, err = db.sql.Exec("INSERT INTO users(id, label, created_at) VALUES(?, ?, ?)", id, label, time.Now().Unix()) return id, err } // createDeviceToken mints a device token for an existing user outside the // invite-claim path — used for webhook identities (hooks.go), whose creator // receives the token once and seals it into the sidecar's definition blob. // Same storage discipline as claimKeyInvite: only the hash is kept. func (db *DB) createDeviceToken(userID, label string) (string, error) { tok, err := randToken() if err != nil { return "", err } _, err = db.sql.Exec( "INSERT INTO device_tokens(token_hash, user_id, label, created_at) VALUES(?, ?, ?, ?)", hashToken(tok), userID, label, time.Now().Unix()) if err != nil { return "", err } return tok, nil } func (db *DB) listUsers() ([]User, error) { now := time.Now().Unix() rows, err := db.sql.Query(` SELECT u.id, u.label, u.created_at, COALESCE(u.is_agent, 0), COALESCE(u.role, 'member'), (SELECT COUNT(*) FROM device_tokens d WHERE d.user_id = u.id), (SELECT MIN(d.created_at) FROM device_tokens d WHERE d.user_id = u.id), (SELECT COUNT(*) FROM key_invites k WHERE k.user_id = u.id AND k.used_at IS NULL AND (k.expires_at IS NULL OR k.expires_at > ?)), (SELECT COUNT(*) FROM key_invites k WHERE k.user_id = u.id AND k.thread_id IS NOT NULL AND k.used_at IS NULL), COALESCE((SELECT kind FROM bot_configs b WHERE b.user_id = u.id), 'ai') FROM users u ORDER BY u.created_at DESC`, now) if err != nil { return nil, err } defer rows.Close() out := []User{} for rows.Next() { var u User var joinedAt sql.NullInt64 var pending, pendingDM, isAgent int var kind string if err := rows.Scan(&u.ID, &u.Label, &u.CreatedAt, &isAgent, &u.Role, &u.Devices, &joinedAt, &pending, &pendingDM, &kind); err != nil { return nil, err } if joinedAt.Valid { j := joinedAt.Int64 u.JoinedAt = &j } u.Pending = pending > 0 u.PendingDM = pendingDM > 0 u.IsAgent = isAgent != 0 if u.IsAgent { u.Kind = kind } out = append(out, u) } return out, rows.Err() } // setUserLabelBlob stores a person's display name encrypted under the instance // key and clears any plaintext label, so the server never holds a readable name // for them. Called by the member themselves (they hold the key); bots are never // routed here — they keep their admin-authored plaintext label. func (db *DB) setUserLabelBlob(userID, blob string) error { _, err := db.sql.Exec("UPDATE users SET label_blob = ?, label = '' WHERE id = ? AND COALESCE(is_agent, 0) = 0", blob, userID) return err } // backfillUserLabelBlob seals a LEGACY plaintext label (docs/no-plaintext.md): // members whose name predates label_blob keep a readable label at rest until // they next load the app — which an inactive member never does. Any // instance-key holder may seal it for them, but strictly fill-once: people // only, only while the plaintext is still there, and only if no blob exists — // so it can never overwrite a name someone set themselves, and racing or // repeated calls are harmless no-ops. Returns whether this call did the fill. func (db *DB) backfillUserLabelBlob(userID, blob string) (bool, error) { res, err := db.sql.Exec( `UPDATE users SET label_blob = ?, label = '' WHERE id = ? AND COALESCE(is_agent, 0) = 0 AND COALESCE(label_blob, '') = '' AND label != ''`, blob, userID) if err != nil { return false, err } n, err := res.RowsAffected() return n > 0, err } // setUserDescriptionBlob stores a person's "about me" text encrypted under the // instance key ("" clears it). Like the label, it's set by the member themselves; // an agent's blurb is the plaintext bot_configs.description instead. func (db *DB) setUserDescriptionBlob(userID, blob string) error { _, err := db.sql.Exec("UPDATE users SET description_blob = ? WHERE id = ? AND COALESCE(is_agent, 0) = 0", blob, userID) return err } // setUserNicknamesBlob stores the member's per-viewer name overrides (issue // #577), sealed to their OWN public key — readable only by their devices, // which share the keypair ("" clears it). Optimistic versioning like the // mini-app state slots: baseVersion is what the writer last read; a stale base // writes nothing (ok=false, current version returned) so the caller re-reads, // replays its journaled ops, and retries. baseVersion < 0 skips the check — // the pre-versioning client's unconditional write, kept working on purpose. func (db *DB) setUserNicknamesBlob(userID, blob string, baseVersion int64) (int64, bool, error) { q := "UPDATE users SET nicknames_blob = ?, nicknames_version = COALESCE(nicknames_version, 0) + 1 WHERE id = ? AND COALESCE(is_agent, 0) = 0" args := []any{blob, userID} if baseVersion >= 0 { q += " AND COALESCE(nicknames_version, 0) = ?" args = append(args, baseVersion) } res, err := db.sql.Exec(q, args...) if err != nil { return 0, false, err } n, err := res.RowsAffected() if err != nil { return 0, false, err } _, v, verr := db.userNicknamesBlob(userID) // where the row (or the winner) is now return v, n > 0, verr } // userNicknamesBlob returns the member's own sealed nickname overrides ("" when // never set) and the blob's version (0 until first write). func (db *DB) userNicknamesBlob(userID string) (string, int64, error) { var blob string var version int64 err := db.sql.QueryRow("SELECT COALESCE(nicknames_blob, ''), COALESCE(nicknames_version, 0) FROM users WHERE id = ?", userID).Scan(&blob, &version) return blob, version, err } // setUserPrefsBlob stores the member's private moderation preferences (muted // people, for now), sealed to their OWN public key like the nicknames blob — // only their devices can open it ("" clears it). Same optimistic versioning: // a stale base writes nothing (ok=false, current version returned); < 0 skips // the check. func (db *DB) setUserPrefsBlob(userID, blob string, baseVersion int64) (int64, bool, error) { q := "UPDATE users SET prefs_blob = ?, prefs_version = COALESCE(prefs_version, 0) + 1 WHERE id = ? AND COALESCE(is_agent, 0) = 0" args := []any{blob, userID} if baseVersion >= 0 { q += " AND COALESCE(prefs_version, 0) = ?" args = append(args, baseVersion) } res, err := db.sql.Exec(q, args...) if err != nil { return 0, false, err } n, err := res.RowsAffected() if err != nil { return 0, false, err } _, v, verr := db.userPrefsBlob(userID) // where the row (or the winner) is now return v, n > 0, verr } // userPrefsBlob returns the member's own sealed moderation preferences ("" // when never set) and the blob's version (0 until first write). func (db *DB) userPrefsBlob(userID string) (string, int64, error) { var blob string var version int64 err := db.sql.QueryRow("SELECT COALESCE(prefs_blob, ''), COALESCE(prefs_version, 0) FROM users WHERE id = ?", userID).Scan(&blob, &version) return blob, version, err } // userCreatedAt returns when a member joined (0 when unknown) — the // moderation brakes use it for new-member probation. func (db *DB) userCreatedAt(id string) int64 { var t int64 db.sql.QueryRow("SELECT COALESCE(created_at, 0) FROM users WHERE id = ?", id).Scan(&t) return t } func (db *DB) userExists(id string) bool { var x int return db.sql.QueryRow("SELECT 1 FROM users WHERE id = ?", id).Scan(&x) == nil } // createKeyInvite stores a client-wrapped key under a client-derived id. func (db *DB) createKeyInvite(id, userID, wrapped, claimHash string, expiresAt int64) error { return db.createKeyInviteWithThread(id, userID, wrapped, "", claimHash, expiresAt) } // createKeyInviteWithThread is createKeyInvite that also links the invite to a // pending DM thread — used to re-send a user invite (a fresh link for the same // pending invitee), so a lost link can be replaced without spawning a new chat. func (db *DB) createKeyInviteWithThread(id, userID, wrapped, threadID, claimHash string, expiresAt int64) error { var exp, tid any if expiresAt > 0 { exp = expiresAt } if threadID != "" { tid = threadID } _, err := db.sql.Exec( "INSERT INTO key_invites(id, user_id, wrapped, thread_id, claim_hash, created_at, expires_at) VALUES(?, ?, ?, ?, ?, ?, ?)", id, userID, wrapped, tid, claimHash, time.Now().Unix(), exp, ) return err } // createUserInvite sets up an invite that drops the joiner straight into a DM with // the inviter. In one transaction it creates a placeholder user for the invitee // (label = the inviter's suggested name, no public key until they join), a `dm` // thread with the inviter (keyed) and the placeholder (keyless), records the DM // pair, and stores the key-invite whose wrapped payload carries the thread key // encrypted to the invite secret. inviterWrapped is the thread key wrapped to the // inviter's own public key; bootstrapWrapped is it encrypted to the invite secret // (so the joiner can recover it). The server sees only blobs it can't open. // Returns the new thread id and the placeholder user id. func (db *DB) createUserInvite(inviterID, inviteID, labelBlob, bootstrapWrapped, inviterWrapped, claimHash string, expiresAt int64) (string, string, error) { inviteeID, err := randToken() if err != nil { return "", "", err } threadID, err := randToken() if err != nil { return "", "", err } tx, err := db.sql.Begin() if err != nil { return "", "", err } defer tx.Rollback() now := time.Now().Unix() lo, hi := inviterID, inviteeID if lo > hi { lo, hi = hi, lo } var exp any if expiresAt > 0 { exp = expiresAt } stmts := []struct { q string args []any }{ {"INSERT INTO users(id, label_blob, created_at) VALUES(?, ?, ?)", []any{inviteeID, labelBlob, now}}, {"INSERT INTO threads(id, kind, name_blob, created_at) VALUES(?, 'dm', NULL, ?)", []any{threadID, now}}, {"INSERT INTO thread_members(thread_id, user_id, created_at) VALUES(?, ?, ?)", []any{threadID, inviterID, now}}, {"INSERT INTO thread_members(thread_id, user_id, created_at) VALUES(?, ?, ?)", []any{threadID, inviteeID, now}}, {"INSERT INTO thread_keys(thread_id, user_id, wrapped) VALUES(?, ?, ?)", []any{threadID, inviterID, inviterWrapped}}, {"INSERT INTO dm_pairs(user_a, user_b, thread_id) VALUES(?, ?, ?)", []any{lo, hi, threadID}}, {"INSERT INTO key_invites(id, user_id, wrapped, thread_id, claim_hash, created_at, expires_at) VALUES(?, ?, ?, ?, ?, ?, ?)", []any{inviteID, inviteeID, bootstrapWrapped, threadID, claimHash, now, exp}}, } for _, s := range stmts { if _, err := tx.Exec(s.q, s.args...); err != nil { return "", "", err } } return threadID, inviteeID, tx.Commit() } type KeyInvite struct { ID string UserID string Wrapped string ThreadID string // the pending DM this invite drops the joiner into ("" if none) ClaimHash string // hashToken of the claim secret; "" for pre-migration invites GrantsAdmin bool // a setup invite: claiming makes the user the (first) admin TierID string // cohort invite: the access level its claimer joins ("" = none) Used bool Expired bool } func (db *DB) getKeyInvite(id string) (KeyInvite, bool, error) { var ki KeyInvite var used, exp sql.NullInt64 var threadID, claimHash sql.NullString var grantsAdmin int err := db.sql.QueryRow( "SELECT id, user_id, wrapped, thread_id, claim_hash, COALESCE(grants_admin, 0), COALESCE(tier_id, ''), used_at, expires_at FROM key_invites WHERE id = ?", id, ).Scan(&ki.ID, &ki.UserID, &ki.Wrapped, &threadID, &claimHash, &grantsAdmin, &ki.TierID, &used, &exp) if err == sql.ErrNoRows { return KeyInvite{}, false, nil } if err != nil { return KeyInvite{}, false, err } ki.ThreadID = threadID.String ki.ClaimHash = claimHash.String ki.GrantsAdmin = grantsAdmin != 0 ki.Used = used.Valid ki.Expired = exp.Valid && time.Now().Unix() > exp.Int64 return ki, true, nil } // setInviteGrantsAdmin marks an unclaimed invite as a setup invite: whoever // claims it becomes an admin. Only reachable via admin-authed invite creation. func (db *DB) setInviteGrantsAdmin(id string) error { _, err := db.sql.Exec("UPDATE key_invites SET grants_admin = 1 WHERE id = ? AND used_at IS NULL", id) return err } // adoptAdminInvite claims an admin-granting invite on an existing member's // account, promoting them to admin without creating a new user. Also marks // setup_complete=true since the adopter is not a fresh first-time user. // Returns ok=false (no error) if the invite is unknown, used, expired, or has invalid claim secret. func (db *DB) adoptAdminInvite(inviteID, existingUserID, claimSecret string) (bool, error) { ki, found, err := db.getKeyInvite(inviteID) if err != nil || !found || ki.Used || ki.Expired || !ki.GrantsAdmin { return false, err } if !validClaimSecret(ki.ClaimHash, claimSecret) { return false, nil } // Mark the invite as used, promote the user to admin, and mark setup complete // (since existing members should not see the setup flow). tx, err := db.sql.Begin() if err != nil { return false, err } defer tx.Rollback() res, err := tx.Exec("UPDATE key_invites SET used_at = ? WHERE id = ? AND used_at IS NULL", time.Now().Unix(), inviteID) if err != nil { return false, err } if n, _ := res.RowsAffected(); n == 0 { return false, nil // already claimed } if err := setUserRoleTx(tx, existingUserID, "admin"); err != nil { return false, err } // Mark setup as complete so existing members don't see the setup flow. if err := setConfigTx(tx, "setup_complete", "1"); err != nil { return false, err } return true, tx.Commit() } // claimKeyInvite marks an unused, unexpired invite as used (single-use) and mints // a device token for its user. Returns the plaintext token (shown once). ok=false // (no error) if the invite is unknown, used, expired, or lost a claim race. // // For a user invite (one that bootstraps a pending DM), wrappedSelf is the thread // key wrapped to the joiner's own public key — we store it as their thread_keys // row so they can decrypt the DM immediately. The joiner IS the invite's user // (the placeholder created at invite time now becomes a real, joined person). func (db *DB) claimKeyInvite(id, deviceLabel, wrappedSelf, claimSecret string) (string, bool, error) { ki, found, err := db.getKeyInvite(id) if err != nil || !found || ki.Used || ki.Expired { return "", false, err } // Proof of the invite secret: knowing the id (which the server stores) is not // enough — the claimer must present the preimage of the stored claim_hash. if !validClaimSecret(ki.ClaimHash, claimSecret) { return "", false, nil } // A user invite (one that bootstraps a pending DM) is for a not-yet-joined // placeholder. Refuse once that user has joined (has a device token): otherwise a // second, still-unused invite for them — e.g. a re-send whose earlier link was // already used — could mint a token AS them and overwrite the DM key wrapped to // their real keypair, taking over the account and bricking their chat. if ki.ThreadID != "" && db.deviceCount(ki.UserID) > 0 { return "", false, nil } res, err := db.sql.Exec("UPDATE key_invites SET used_at = ? WHERE id = ? AND used_at IS NULL", time.Now().Unix(), id) if err != nil { return "", false, err } if n, _ := res.RowsAffected(); n == 0 { return "", false, nil // already claimed } // First device = this person is entering the space (an add-device claim // has devices already). Checked before the token insert below creates one. firstDevice := db.deviceCount(ki.UserID) == 0 tok, err := randToken() if err != nil { return "", false, err } _, err = db.sql.Exec( "INSERT INTO device_tokens(token_hash, user_id, label, created_at) VALUES(?, ?, ?, ?)", hashToken(tok), ki.UserID, deviceLabel, time.Now().Unix(), ) if err != nil { return "", false, err } if firstDevice { // Land the newcomer in every default open group. Best-effort ON // PURPOSE: used_at is already burned above, so failing the claim here // would brick the invite — and a missed default join is recoverable // (the group is one Discover tap away). _ = db.joinDefaultThreads(ki.UserID) // A cohort invite also carries an access level — assign it now // ('invite' source), same best-effort rule. Their level key follows // once they publish a pubkey (setPubkey prods the holders). if ki.TierID != "" && db.tierExists(ki.TierID) { _ = db.assignTier(ki.TierID, ki.UserID, "invite") } } if ki.ThreadID != "" && wrappedSelf != "" { // They're already a member of the pending DM (added at invite time); this // records the key wrapped to the fresh keypair they just generated. if err := db.addThreadMember(ki.ThreadID, ki.UserID, wrappedSelf); err != nil { return "", false, err } // Invite lineage (issue #578): the pending DM's other member is who // invited them — recorded so moderation can see a member's way in. if inviter, ok := db.otherThreadMember(ki.ThreadID, ki.UserID); ok { _, _ = db.sql.Exec("UPDATE users SET invited_by = ? WHERE id = ? AND COALESCE(invited_by, '') = ''", inviter, ki.UserID) } } if ki.GrantsAdmin { // A setup invite: its claimer is the (first) admin. Role is authorisation // only — no key material rides on it. if err := db.setUserRole(ki.UserID, "admin"); err != nil { return "", false, err } } return tok, true, nil } // userRole returns a user's role ('member' when unset or unknown). func (db *DB) userRole(id string) string { var r string if db.sql.QueryRow("SELECT COALESCE(role, 'member') FROM users WHERE id = ?", id).Scan(&r) != nil || r == "" { return "member" } return r } // adminUserIDs returns the ids of every role-admin — the recipients of // changelog announcements (announce.go). func (db *DB) adminUserIDs() ([]string, error) { rows, err := db.sql.Query("SELECT id FROM users WHERE COALESCE(role, 'member') = 'admin'") if err != nil { return nil, err } defer rows.Close() var out []string for rows.Next() { var id string if err := rows.Scan(&id); err != nil { return nil, err } out = append(out, id) } return out, rows.Err() } // adminCount reports how many members hold the admin role. Removed users are // hard-deleted rows, so this is the live count — 1 means the caller (if admin) // holds the only admin key on the instance. func (db *DB) adminCount() int { var n int db.sql.QueryRow("SELECT COUNT(*) FROM users WHERE role = 'admin'").Scan(&n) return n } func (db *DB) setUserRole(id, role string) error { _, err := db.sql.Exec("UPDATE users SET role = ? WHERE id = ?", role, id) return err } // setUserRoleTx sets a user's role within a transaction (for atomic operations). func setUserRoleTx(tx *sql.Tx, id, role string) error { _, err := tx.Exec("UPDATE users SET role = ? WHERE id = ?", role, id) return err } // deviceCount returns how many devices (tokens) a user currently has signed in — so // the client can warn before signing out the LAST one (which loses their keys). func (db *DB) deviceCount(userID string) int { var n int db.sql.QueryRow("SELECT COUNT(*) FROM device_tokens WHERE user_id = ?", userID).Scan(&n) return n } // deleteDeviceToken revokes a single device token (signing out on that device). func (db *DB) deleteDeviceToken(token string) error { _, err := db.sql.Exec("DELETE FROM device_tokens WHERE token_hash = ?", hashToken(token)) return err } func (db *DB) deviceTokenValid(token string) bool { var x int return db.sql.QueryRow("SELECT 1 FROM device_tokens WHERE token_hash = ?", hashToken(token)).Scan(&x) == nil } // deviceTokenUser returns the user a device token belongs to (for self-service // "add a device"). ok=false for unknown or legacy (non-device) tokens. func (db *DB) deviceTokenUser(token string) (string, bool) { return db.deviceTokenUserByHash(hashToken(token)) } // deviceTokenUserByHash resolves the owning user from a token hash, so callers // that only hold the hash (push/APNs subscriptions) never need the raw token. func (db *DB) deviceTokenUserByHash(tokenHash string) (string, bool) { var uid string if db.sql.QueryRow("SELECT user_id FROM device_tokens WHERE token_hash = ?", tokenHash).Scan(&uid) != nil { return "", false } return uid, true } // ---- roster + threads (see docs/threads-and-dms.md) --------------------- // RosterEntry is one discoverable person: their label + ECDH public key (used to // wrap thread/instance keys to them) + their profile avatar blob id (if set). type RosterEntry struct { UserID string `json:"user_id"` // Label is a plaintext name — only ever set for bots (admin-authored). For a // person it's empty; their name is in LabelBlob, encrypted under the instance // key, and the client decrypts it. (Un-migrated people may still have a // plaintext Label until their client republishes it encrypted — see setUserLabelBlob.) Label string `json:"label"` LabelBlob string `json:"label_blob,omitempty"` PublicKey string `json:"public_key"` // Key transparency (#573): the joiner's HMAC of PublicKey under HKDF(invite // secret T). A verifier holding T recomputes it (crypto.js verifyBindMac) to // confirm the server didn't substitute its own key. "" for pre-KT members. BindMAC string `json:"bind_mac,omitempty"` Avatar string `json:"avatar,omitempty"` // The photo's alt text, encrypted under the instance key (client decrypts). AvatarAltBlob string `json:"avatar_alt_blob,omitempty"` // A person's free-text "about me", encrypted under the instance key (client // decrypts). Served only to their connections, like the rest of the entry. DescriptionBlob string `json:"description_blob,omitempty"` // admin | moderator | member — lets clients say who runs the space (the // welcome conversation's "run by "). Server-known metadata, not // content; it already gates the admin API. Role string `json:"role,omitempty"` IsAgent bool `json:"is_agent,omitempty"` // an agent/bot, not a person // One-line admin-authored blurb for agents (the Helpers catalog entry — // doubles as the consent hint). Always "" for people. Description string `json:"description,omitempty"` // Kind is "ai" (LLM-backed) or "bot" (deterministic) for agents; "" for // people. Drives the AI/BOT badge — the honest "where do my words go?" cue. Kind string `json:"kind,omitempty"` // Presence: "online" | "away" | "offline". Scanned as the raw pinned flag // ('' or 'away'); the roster handler overwrites it with the effective state // (stampPresence, presence.go) before it leaves the server. Presence string `json:"presence,omitempty"` } func (db *DB) setUserPublicKey(userID, pub, bindMac string) error { // Transactional so the key transparency 'join' event can't be missing for a // member who's now keyed — a gap would read as roster tampering to a // verifier, exactly like a missing message journal row. tx, err := db.sql.Begin() if err != nil { return err } defer tx.Rollback() var cur string tx.QueryRow("SELECT COALESCE(public_key, '') FROM users WHERE id = ?", userID).Scan(&cur) // First key for this member (empty → set): store the pubkey AND its // key-transparency bind-MAC, and journal the 'join'. The bind-MAC is written // ONLY here and never overwritten, so an add-device re-register of the same // key can't drop (or replace) the original binding (#573). if cur == "" && pub != "" { if _, err := tx.Exec("UPDATE users SET public_key = ?, bind_mac = ? WHERE id = ?", pub, bindMac, userID); err != nil { return err } if err := appendRosterEventTx(tx, time.Now().Unix(), "join", userID, ktPubHash(pub)); err != nil { return err } } else { // Re-registering the same key (add-device) — keep the existing bind_mac. if _, err := tx.Exec("UPDATE users SET public_key = ? WHERE id = ?", pub, userID); err != nil { return err } } return tx.Commit() } // appendRosterEventTx journals a key-transparency roster event (docs/tamper- // evidence.md slice 4): membership changes are committed to the signed // checkpoint chain like message events, so a monitor catches a server that // rewrites the roster. The journal's generic columns carry the kind-scoped // detail: thread_id = the member's user id, blob_hash = sha256 of their public // key ('join') or "" ('remove'); msg_id is unused (0). func appendRosterEventTx(tx *sql.Tx, ts int64, kind, userID, pubHash string) error { return appendJournalTx(tx, ts, kind, 0, userID, pubHash) } // ktPubHash commits to a public key without storing it twice — sha256 hex of // the key string, the same shape ciphertextHash uses for message blobs. func ktPubHash(pub string) string { sum := sha256.Sum256([]byte(pub)) return hex.EncodeToString(sum[:]) } // userPublicKey returns a user's registered ECDH public key ("" if not set yet). func (db *DB) userPublicKey(id string) string { var pk sql.NullString db.sql.QueryRow("SELECT public_key FROM users WHERE id = ?", id).Scan(&pk) return pk.String } // setUserPresence pins ('away') or unpins (”) a user's presence. func (db *DB) setUserPresence(userID, presence string) error { _, err := db.sql.Exec("UPDATE users SET presence = ? WHERE id = ?", presence, userID) return err } // userPresenceInfo returns the pinned presence flag plus whether the user is // an agent — the two per-user inputs to presenceState (presence.go). func (db *DB) userPresenceInfo(userID string) (pinned string, isAgent bool) { var agent int db.sql.QueryRow("SELECT COALESCE(presence, ''), COALESCE(is_agent, 0) FROM users WHERE id = ?", userID).Scan(&pinned, &agent) return pinned, agent != 0 } // removeUser deletes a user and all their traces (admin roster management): device // tokens, thread membership + wrapped keys, DM-pair records, pending invites, their // instance-key copy, and their own messages — then cleans up any thread // left with no members. Threads they merely shared (a DM/group with others) survive; // the other members keep them. func (db *DB) removeUser(id string) error { tx, err := db.sql.Begin() if err != nil { return err } defer tx.Rollback() // The threads the user is in — used to scope the orphan sweep and to find any // pending DMs they set up. (Collect fully before issuing more statements: one // SQLite connection can't interleave an open cursor with writes.) rows, err := tx.Query("SELECT thread_id FROM thread_members WHERE user_id = ?", id) if err != nil { return err } var threadIDs []string for rows.Next() { var t string if err := rows.Scan(&t); err != nil { rows.Close() return err } threadIDs = append(threadIDs, t) } rows.Close() if err := rows.Err(); err != nil { return err } teardown := func(tid string) error { // Journal the messages this teardown removes (tamper evidence) before // they're gone — same transaction, so the trail can't drift. if err := journalMessageDeletesTx(tx, "thread_id = ?", tid); err != nil { return err } for _, q := range []string{ "DELETE FROM key_invites WHERE thread_id = ?", "DELETE FROM thread_keys WHERE thread_id = ?", "DELETE FROM thread_members WHERE thread_id = ?", "DELETE FROM dm_pairs WHERE thread_id = ?", "DELETE FROM messages WHERE thread_id = ?", "DELETE FROM threads WHERE id = ?", } { if _, err := tx.Exec(q, tid); err != nil { return err } } return nil } // Pending DMs this user set up whose invitee never joined: tear the whole thing // down and delete the keyless placeholder, so the invitee's still-live link can't // drop them into a dead, unreadable one-person chat. for _, tid := range threadIDs { var invitee string err := tx.QueryRow("SELECT user_id FROM key_invites WHERE thread_id = ? LIMIT 1", tid).Scan(&invitee) if err == sql.ErrNoRows { continue } if err != nil { return err } var joined int if err := tx.QueryRow("SELECT COUNT(*) FROM device_tokens WHERE user_id = ?", invitee).Scan(&joined); err != nil { return err } if joined > 0 { continue // the invitee has joined; leave their real chat alone } if err := teardown(tid); err != nil { return err } if invitee != id { if _, err := tx.Exec("DELETE FROM users WHERE id = ? AND (public_key IS NULL OR public_key = '')", invitee); err != nil { return err } } } // The user's own rows. Their messages get journaled first (tamper // evidence) — removal deletes them below, and the trail must say so. if err := journalMessageDeletesTx(tx, "author_hash = ?", id); err != nil { return err } // A 'remove' key-transparency event, so the roster change is committed to // the chain like the member's join was (docs/tamper-evidence.md slice 4). if err := appendRosterEventTx(tx, time.Now().Unix(), "remove", id, ""); err != nil { return err } // Tombstone each doomed message (#765, same mechanism as deleteMessage's // #581): members' devices that already hold these messages learn to drop // them from ordinary forward catch-up — a purge is exactly when that // propagation matters most. Skipped for legacy NULL-thread rows // (unreachable by thread reads); the fresh tombstones have a NULL // author_hash, so the cascade's DELETE below can't touch them. Threads // torn down whole (above) need none — the thread's disappearance already // clears every client's copy. purgeNow := time.Now().Unix() if _, err := tx.Exec( "INSERT INTO messages(created_at, blob, thread_id, tombstone_of) SELECT ?, '', thread_id, id FROM messages WHERE author_hash = ? AND thread_id IS NOT NULL", purgeNow, id, ); err != nil { return err } // Each tombstone is a real message row (clients replay it from catch-up), // so each gets a journal row like any other insert — an unexplained row // would read as tampering to a verifier. Set-based like the insert above: // the fresh tombstones are exactly those pointing at a still-present row // of the purged author (deleteMessage's older tombstones point at ids // already gone, so the join can't double-journal them). if _, err := tx.Exec( "INSERT INTO event_journal(ts, kind, msg_id, thread_id, blob_hash) "+ "SELECT ?, 'msg', t.id, COALESCE(t.thread_id, ''), ? FROM messages t "+ "JOIN messages d ON d.id = t.tombstone_of AND d.author_hash = ?", purgeNow, ciphertextHash(""), id, ); err != nil { return err } for _, s := range []struct { q string arg []any }{ {"DELETE FROM thread_keys WHERE user_id = ?", []any{id}}, {"DELETE FROM thread_members WHERE user_id = ?", []any{id}}, // APNs tokens hang off device-session tokens (matched by token hash) — // resolve them BEFORE the device_tokens rows below are gone. {"DELETE FROM apns_tokens WHERE token_hash IN (SELECT token_hash FROM device_tokens WHERE user_id = ?)", []any{id}}, {"DELETE FROM device_tokens WHERE user_id = ?", []any{id}}, {"DELETE FROM key_invites WHERE user_id = ?", []any{id}}, {"DELETE FROM instance_keys WHERE user_id = ?", []any{id}}, // Access levels: their memberships and key copies go too — a removed // member must not linger as a ghost holder (which would block the // "key lost" detection) or a ghost member (wrong counts, a forever // "keys on their way" line in the Admin app). {"DELETE FROM user_tiers WHERE user_id = ?", []any{id}}, {"DELETE FROM tier_keys WHERE user_id = ?", []any{id}}, {"DELETE FROM dm_pairs WHERE user_a = ? OR user_b = ?", []any{id, id}}, {"DELETE FROM messages WHERE author_hash = ?", []any{id}}, {"DELETE FROM scheduled_messages WHERE author_id = ?", []any{id}}, {"DELETE FROM push_subscriptions WHERE user_id = ?", []any{id}}, {"DELETE FROM users WHERE id = ?", []any{id}}, } { if _, err := tx.Exec(s.q, s.arg...); err != nil { return err } } // Sweep only the user's former threads that now have no members left (e.g. their // "Notes" thread, or a DM whose other member was already gone). for _, tid := range threadIDs { var n int if err := tx.QueryRow("SELECT COUNT(*) FROM thread_members WHERE thread_id = ?", tid).Scan(&n); err != nil { return err } if n == 0 { if err := teardown(tid); err != nil { return err } } } return tx.Commit() } // setUserLabel renames a user in the roster (admin roster management — e.g. // renaming a bot). Labels are display names, not identity: keys, tokens, and // message history are untouched. func (db *DB) setUserLabel(userID, label string) error { res, err := db.sql.Exec("UPDATE users SET label = ? WHERE id = ?", label, userID) if err != nil { return err } if n, _ := res.RowsAffected(); n == 0 { return sql.ErrNoRows } return nil } // setUserAgent flags a user as an agent (bot) so the UI can badge it. On its // own it confers no privilege; combined with available_to_members it makes the // bot a vetted helper (catalog-visible, full-roster reads — see isVettedAgent). func (db *DB) setUserAgent(userID string) error { _, err := db.sql.Exec("UPDATE users SET is_agent = 1 WHERE id = ?", userID) return err } // isAgent reports whether a user is a bot (the admin bot-config endpoints // refuse to attach config to a person). func (db *DB) isAgent(userID string) bool { var n int db.sql.QueryRow("SELECT COALESCE(is_agent, 0) FROM users WHERE id = ?", userID).Scan(&n) return n != 0 } // isVettedAgent reports whether a user is an agent the admin marked available // to members. Vetted bots are instance-wide infrastructure: they appear in // everyone's Helpers catalog and read the full joined roster (so e.g. Eleven // can welcome a brand-new member it shares no thread with). func (db *DB) isVettedAgent(userID string) bool { var n int db.sql.QueryRow(` SELECT COUNT(*) FROM users u JOIN bot_configs b ON b.user_id = u.id WHERE u.id = ? AND COALESCE(u.is_agent, 0) = 1 AND COALESCE(b.available_to_members, 0) = 1`, userID).Scan(&n) return n > 0 } // BotConfig is an agent's console-managed settings (see bot_configs). type BotConfig struct { Model string `json:"model"` SystemPrompt string `json:"system_prompt"` Docs string `json:"docs"` Listen string `json:"listen"` // ''/'mention' = @mention only, 'name' = also bare name Welcome string `json:"welcome"` // admin-tweakable welcome DM (the guide uses it); "" = shipped default // The instance-wide AI gate, not a bot_configs column — getBotConfig fills // it from config so the sidecar learns a flip on its next (re)connect. AIEnabled bool `json:"ai_enabled"` } // setBotKind records an agent's kind ("ai" | "bot"), self-declared by the sidecar // on connect (LLM-backed vs deterministic). A narrow upsert: it touches only the // kind column, so it never clobbers admin-edited config — and setBotConfig never // clobbers kind. Order-independent. func (db *DB) setBotKind(userID, kind string) error { _, err := db.sql.Exec( `INSERT INTO bot_configs(user_id, kind, updated_at) VALUES(?, ?, ?) ON CONFLICT(user_id) DO UPDATE SET kind = excluded.kind, updated_at = excluded.updated_at`, userID, kind, time.Now().Unix()) return err } // setBotWelcome stores the admin-tweakable welcome DM text (same narrow-upsert // discipline as setBotKind, so it composes with setBotConfig without a signature // change or clobbering other fields). func (db *DB) setBotWelcome(userID, welcome string) error { _, err := db.sql.Exec( `INSERT INTO bot_configs(user_id, welcome, updated_at) VALUES(?, ?, ?) ON CONFLICT(user_id) DO UPDATE SET welcome = excluded.welcome, updated_at = excluded.updated_at`, userID, welcome, time.Now().Unix()) return err } func (db *DB) setBotConfig(userID, model, systemPrompt, docs, listen, description string, available bool) error { av := 0 if available { av = 1 } _, err := db.sql.Exec( `INSERT INTO bot_configs(user_id, model, system_prompt, docs, listen, description, available_to_members, updated_at) VALUES(?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(user_id) DO UPDATE SET model = excluded.model, system_prompt = excluded.system_prompt, docs = excluded.docs, listen = excluded.listen, description = excluded.description, available_to_members = excluded.available_to_members, updated_at = excluded.updated_at`, userID, model, systemPrompt, docs, listen, description, av, time.Now().Unix(), ) return err } // botConfig returns an agent's config (ok=false if none has been set). func (db *DB) botConfig(userID string) (BotConfig, bool, error) { var c BotConfig err := db.sql.QueryRow("SELECT model, system_prompt, docs, listen, COALESCE(welcome, '') FROM bot_configs WHERE user_id = ?", userID). Scan(&c.Model, &c.SystemPrompt, &c.Docs, &c.Listen, &c.Welcome) if err == sql.ErrNoRows { return c, false, nil } return c, err == nil, err } // FullBotConfig is BotConfig plus the Helpers-catalog metadata — what the // admin console edits (the sidecar's own view stays BotConfig). type FullBotConfig struct { BotConfig Description string `json:"description"` Available bool `json:"available_to_members"` } // fullBotConfig returns everything the console shows for a bot; ok=false when // no config row exists yet (a pre-console bot — fields default to empty). func (db *DB) fullBotConfig(userID string) (FullBotConfig, bool, error) { var c FullBotConfig var av int err := db.sql.QueryRow( `SELECT model, system_prompt, docs, COALESCE(listen, ''), COALESCE(welcome, ''), COALESCE(description, ''), COALESCE(available_to_members, 0) FROM bot_configs WHERE user_id = ?`, userID). Scan(&c.Model, &c.SystemPrompt, &c.Docs, &c.Listen, &c.Welcome, &c.Description, &av) if err == sql.ErrNoRows { return c, false, nil } c.Available = av != 0 return c, err == nil, err } // setAvatar sets the photo and its (encrypted) alt text together — a photo // change always re-states the alt, so a swap never keeps a stale description. func (db *DB) setAvatar(userID, blobID, altBlob string) error { var b, a any if blobID != "" { b = blobID if altBlob != "" { a = altBlob } } _, err := db.sql.Exec("UPDATE users SET avatar_blob = ?, avatar_alt_blob = ? WHERE id = ?", b, a, userID) return err } // roster lists everyone who has joined (registered a public key) — the full // directory. Members no longer see this; they get connections() (self + people // they share a thread with). Kept for admin/back-end use and tests. func (db *DB) roster() ([]RosterEntry, error) { rows, err := db.sql.Query( `SELECT u.id, u.label, COALESCE(u.label_blob, ''), u.public_key, COALESCE(u.avatar_blob, ''), COALESCE(u.avatar_alt_blob, ''), COALESCE(u.description_blob, ''), COALESCE(u.role, 'member'), COALESCE(u.is_agent, 0), COALESCE(b.description, ''), COALESCE(b.kind, 'ai'), COALESCE(u.presence, ''), COALESCE(u.bind_mac, '') FROM users u LEFT JOIN bot_configs b ON b.user_id = u.id WHERE u.public_key IS NOT NULL AND u.public_key <> '' ORDER BY u.label`, ) if err != nil { return nil, err } return scanRoster(rows) } // connections is the roster a member actually sees: themselves plus everyone they // already share a thread with (a DM or a group). It intentionally does NOT expose // the whole instance — you can only start a new chat with someone you're already // connected to, or by inviting someone new. Keyless people (a pending invitee who // hasn't joined yet) are included so a pending DM can still show their name; the // client omits keyless entries from the "start a new chat" picker. // // The one exemption is the Helpers catalog (docs/bots.md): agents the admin has // marked available_to_members are shown to everyone — bots share no thread when // new, so they can't ride the connections rule for discovery. People stay // connections-only; only vetted, clearly-badged helpers are instance-visible. func (db *DB) connections(userID string) ([]RosterEntry, error) { rows, err := db.sql.Query(` SELECT u.id, u.label, COALESCE(u.label_blob, ''), COALESCE(u.public_key, ''), COALESCE(u.avatar_blob, ''), COALESCE(u.avatar_alt_blob, ''), COALESCE(u.description_blob, ''), COALESCE(u.role, 'member'), COALESCE(u.is_agent, 0), COALESCE(b.description, ''), COALESCE(b.kind, 'ai'), COALESCE(u.presence, ''), COALESCE(u.bind_mac, '') FROM users u LEFT JOIN bot_configs b ON b.user_id = u.id WHERE u.id = ? OR u.id IN ( SELECT m2.user_id FROM thread_members m1 JOIN thread_members m2 ON m2.thread_id = m1.thread_id WHERE m1.user_id = ? AND m2.user_id <> ? ) OR (COALESCE(u.is_agent, 0) = 1 AND COALESCE(b.available_to_members, 0) = 1 AND u.public_key IS NOT NULL AND u.public_key <> '') ORDER BY u.label`, userID, userID, userID) if err != nil { return nil, err } return scanRoster(rows) } func scanRoster(rows *sql.Rows) ([]RosterEntry, error) { defer rows.Close() out := []RosterEntry{} for rows.Next() { var e RosterEntry var isAgent int var kind string if err := rows.Scan(&e.UserID, &e.Label, &e.LabelBlob, &e.PublicKey, &e.Avatar, &e.AvatarAltBlob, &e.DescriptionBlob, &e.Role, &isAgent, &e.Description, &kind, &e.Presence, &e.BindMAC); err != nil { return nil, err } e.IsAgent = isAgent != 0 if e.IsAgent { e.Kind = kind // "ai" | "bot" — badge cue; only meaningful for agents } else { e.Description = "" // description is Helpers-catalog metadata, agents only } out = append(out, e) } return out, rows.Err() } // ---- instance key (shared key for all-members content, e.g. avatars) ---- // instanceKeyFor returns the caller's wrapped copy of the instance key (if any). func (db *DB) instanceKeyFor(userID string) (string, bool, error) { var w string err := db.sql.QueryRow("SELECT wrapped FROM instance_keys WHERE user_id = ?", userID).Scan(&w) if err == sql.ErrNoRows { return "", false, nil } return w, err == nil, err } // instanceKeyHolders lists user ids that already hold a wrapped instance key, so // a member who has it only re-wraps to those who don't (bootstrap + reconcile). func (db *DB) instanceKeyHolders() ([]string, error) { rows, err := db.sql.Query("SELECT user_id FROM instance_keys") if err != nil { return nil, err } defer rows.Close() out := []string{} for rows.Next() { var u string if err := rows.Scan(&u); err != nil { return nil, err } out = append(out, u) } return out, rows.Err() } // lastReadActivity is the newest read watermark across a set of users — the // durable "when were they last around" proxy (#785): read marks persist and // advance with any real use, unlike the in-memory presence map that resets on // restart. 0 when none of them ever read anything. func (db *DB) lastReadActivity(userIDs []string) (int64, error) { if len(userIDs) == 0 { return 0, nil } args := make([]any, len(userIDs)) marks := make([]string, len(userIDs)) for i, u := range userIDs { args[i] = u marks[i] = "?" } var at sql.NullInt64 err := db.sql.QueryRow( "SELECT MAX(last_read_at) FROM read_marks WHERE user_id IN ("+strings.Join(marks, ",")+")", args..., ).Scan(&at) if err != nil { return 0, err } return at.Int64, nil } // resetInstanceKey is the admin break-glass for an ORPHANED instance key // (#785): every holder gone for good, so no one can ever be granted it and // every newer member stays keyless. Dropping all holder rows lets the next // member to boot mint a fresh key (setupInstanceKey's bootstrap branch) and // grant it outward as people come online. The old key's sealed blobs are // unreadable forever, so this also clears what can regrow and demotes what // can't: // - members' sealed name/photo/about columns → NULL (each republishes their // name on next load; photos/abouts are re-entered) // - open groups gated under the instance key → invite-only, open_key // dropped (members keep their thread keys; an admin can re-open the // group afterwards, which wraps its key under the NEW instance key) // - member apps/bots stay as rows but skip as undecryptable — deleting // user content outright is worse than leaving it inert // - the space thread (spacethread.go) is deleted outright: unlike an open // group, its members hold NO per-member key wraps — open_key under the // dead instance key was the ONLY path to its key, so its whole log is // unreadable forever. The next space-wide install mints a fresh one. // // The server never sees any key either way — this deletes wrapped blobs it // could never open. func (db *DB) resetInstanceKey() error { tx, err := db.sql.Begin() if err != nil { return err } defer tx.Rollback() // Resolve the space thread INSIDE the transaction: a concurrent // POST /api/space-thread landing between an outside read and the commit // would leave a thread wrapped under the key we're about to orphan. spaceID, err := deleteSpaceThreadTx(tx) if err != nil { return err } for _, q := range []string{ "DELETE FROM instance_keys", "UPDATE users SET label_blob = NULL, avatar_blob = NULL, avatar_alt_blob = NULL, description_blob = NULL", "UPDATE threads SET visibility = 'invite', open_key = NULL, default_join = 0 WHERE visibility = 'open' AND (open_tier IS NULL OR open_tier = '')", } { if _, err := tx.Exec(q); err != nil { return err } } if err := tx.Commit(); err != nil { return err } if spaceID != "" { // The thread's attachment blobs (bytes on disk) go with it. return db.deleteThreadBlobs(spaceID) } return nil } // deleteSpaceThreadTx removes the space thread and its rows inside tx, // returning its id ("" when none existed) so the caller can reclaim its blob // bytes after commit. Shared by resetInstanceKey (whose reset orphans the // thread's only key path) and the scoped admin space-thread reset — the lever // for a poisoned open_key that shouldn't cost every member their name/photo. func deleteSpaceThreadTx(tx *sql.Tx) (string, error) { var spaceID string err := tx.QueryRow("SELECT id FROM threads WHERE kind = 'space'").Scan(&spaceID) if err == sql.ErrNoRows { return "", nil } if err != nil { return "", err } // Journal the message deletes like every other hard delete (tamper // evidence), then drop the thread's rows — including thread_keys, which an // explicitly-added bot may have (mirroring deleteThread's teardown). if err := journalMessageDeletesTx(tx, "thread_id = ?", spaceID); err != nil { return "", err } for _, q := range []string{ "DELETE FROM messages WHERE thread_id = ?", "DELETE FROM thread_state WHERE thread_id = ?", "DELETE FROM scheduled_messages WHERE thread_id = ?", "DELETE FROM thread_keys WHERE thread_id = ?", "DELETE FROM thread_members WHERE thread_id = ?", "DELETE FROM threads WHERE id = ?", } { if _, err := tx.Exec(q, spaceID); err != nil { return "", err } } return spaceID, nil } // resetSpaceThread is the scoped teardown behind the admin space-thread // reset: just the space thread, nothing else — for recovering from a // poisoned/undecryptable open_key without the full instance-key break-glass. func (db *DB) resetSpaceThread() error { tx, err := db.sql.Begin() if err != nil { return err } defer tx.Rollback() spaceID, err := deleteSpaceThreadTx(tx) if err != nil { return err } if err := tx.Commit(); err != nil { return err } if spaceID != "" { return db.deleteThreadBlobs(spaceID) } return nil } // grantInstanceKey wraps the instance key to a member who doesn't yet hold it. // DO NOTHING (not overwrite): any member may call this for any user id, so // overwriting would let one member replace another's good wrapped key with junk // and brick their avatar decryption. Grants only ever fill a missing holder; // there is no re-key path through here. func (db *DB) grantInstanceKey(userID, wrapped string) error { _, err := db.sql.Exec( "INSERT INTO instance_keys(user_id, wrapped) VALUES(?, ?) ON CONFLICT(user_id) DO NOTHING", userID, wrapped, ) return err } // Tier is one admin-curated member set with key material (see the schema // comment). Members/Mine/Wrapped/Have are filled per requesting user by // listTiers, so one fetch renders the whole management + key-grant picture. type Tier struct { ID string `json:"id"` Name string `json:"name"` CreatedAt int64 `json:"created_at"` Members []string `json:"members,omitempty"` // Mine: the requesting user belongs to this tier. Wrapped: their copy of // the tier key (unwrap client-side). Have: who already holds a wrapped // copy, so a holder only wraps to those missing (the grantInstanceKeyTo // rule, per tier). Mine bool `json:"mine,omitempty"` Wrapped string `json:"wrapped,omitempty"` Have []string `json:"have,omitempty"` // KeyLost: the key was minted but every holder is gone — unrecoverable // (grants are refused with 410); the level should be deleted and remade. KeyLost bool `json:"key_lost,omitempty"` // Visibility policy: 'shown' | 'hidden' | 'user' (see the schema comment). Visibility string `json:"visibility,omitempty"` // Hidden: MY membership of this level is currently hidden from non-members // — meaningful only when I'm a member and Visibility=='user'. Hidden bool `json:"hidden,omitempty"` // Paid access (layer 3): when PriceCents>0 the level is sellable — members // can subscribe. Plaintext commerce config, like branding. PriceCents int `json:"price_cents,omitempty"` Currency string `json:"currency,omitempty"` Interval string `json:"interval,omitempty"` // 'month' | 'year' | 'once' (lifetime) ProviderRef string `json:"-"` // provider price id; never sent to clients // The requesting user's subscription on this level (source='stripe'), // filled by listTiers for the member's own view: status + when access // runs out. Status string `json:"status,omitempty"` PaidUntil int64 `json:"paid_until,omitempty"` GraceDeadline int64 `json:"grace_deadline,omitempty"` // Lifetime: MY membership here is a one-time purchase (never renews, // nothing to cancel) — the UI drops every billing control off it. Lifetime bool `json:"lifetime,omitempty"` } func (db *DB) createTier(name string) (string, error) { id, err := randToken() if err != nil { return "", err } _, err = db.sql.Exec("INSERT INTO tiers(id, name, created_at) VALUES(?, ?, ?)", id, name, time.Now().Unix()) return id, err } // deleteTier removes the tier, its memberships and its key rows. Callers must // refuse deletion while any thread is gated on this level (tierGatesThreads): // a gated group's open_key is wrapped under this key, and for self-serve // joiners that wrap is their ONLY key source — deleting it would seal the // group's name and history on their devices for good. func (db *DB) deleteTier(id string) error { tx, err := db.sql.Begin() if err != nil { return err } defer tx.Rollback() for _, q := range []string{ "DELETE FROM tier_keys WHERE tier_id = ?", "DELETE FROM user_tiers WHERE tier_id = ?", "DELETE FROM tiers WHERE id = ?", } { if _, err := tx.Exec(q, id); err != nil { return err } } return tx.Commit() } func (db *DB) tierExists(id string) bool { var x int return db.sql.QueryRow("SELECT 1 FROM tiers WHERE id = ?", id).Scan(&x) == nil } func (db *DB) tierMember(tierID, userID string) bool { var x int return db.sql.QueryRow("SELECT 1 FROM user_tiers WHERE tier_id = ? AND user_id = ?", tierID, userID).Scan(&x) == nil } // tierHolder reports whether a member holds a wrapped copy of the tier key — // the gate for granting to others once the key exists (see postTierKeys). func (db *DB) tierHolder(tierID, userID string) bool { var x int return db.sql.QueryRow("SELECT 1 FROM tier_keys WHERE tier_id = ? AND user_id = ?", tierID, userID).Scan(&x) == nil } func (db *DB) tierKeyCount(tierID string) int { var n int db.sql.QueryRow("SELECT COUNT(*) FROM tier_keys WHERE tier_id = ?", tierID).Scan(&n) return n } func (db *DB) assignTier(tierID, userID, source string) error { _, err := db.sql.Exec( "INSERT OR IGNORE INTO user_tiers(tier_id, user_id, source, created_at) VALUES(?, ?, ?, ?)", tierID, userID, source, time.Now().Unix()) return err } // unassignTier drops a member from a tier: membership + their key row. Like // removeThreadMember, this is service-level revocation — the tier key isn't // rotated, so it stops the server serving them tier-gated content, but a // device that already holds the key keeps what it has (rekey is deferred; // docs/community-and-tiers.md § revocation). func (db *DB) unassignTier(tierID, userID string) error { tx, err := db.sql.Begin() if err != nil { return err } defer tx.Rollback() if _, err := tx.Exec("DELETE FROM user_tiers WHERE tier_id = ? AND user_id = ?", tierID, userID); err != nil { return err } if _, err := tx.Exec("DELETE FROM tier_keys WHERE tier_id = ? AND user_id = ?", tierID, userID); err != nil { return err } return tx.Commit() } // setTierPrice makes a level sellable (or, with cents=0, not). Plaintext // commerce config; provider_ref is the payment provider's price id. func (db *DB) setTierPrice(tierID string, cents int, currency, interval, providerRef string) error { _, err := db.sql.Exec( "UPDATE tiers SET price_cents = ?, currency = ?, interval = ?, provider_ref = ? WHERE id = ?", cents, currency, interval, nullable(providerRef), tierID) return err } // tierPrice returns a level's price config (for building a checkout). func (db *DB) tierPrice(tierID string) (cents int, currency, interval, providerRef string, ok bool) { var pr sql.NullString err := db.sql.QueryRow( "SELECT COALESCE(price_cents,0), COALESCE(currency,''), COALESCE(interval,''), provider_ref FROM tiers WHERE id = ?", tierID).Scan(¢s, ¤cy, &interval, &pr) if err != nil { return 0, "", "", "", false } return cents, currency, interval, pr.String, cents > 0 } // Entitlement is a member's paid membership of a level (source='stripe'). type Entitlement struct { TierID string UserID string Status string // active | past_due | canceled PaidUntil int64 GraceDeadline int64 ExtSub string ExtCustomer string EventTS int64 // provider event `created` last folded, for ordering Source string // how the membership was assigned: admin | invite | stripe Lifetime bool // a one-time purchase: no lifecycle, never swept } // markPaymentRevoked tombstones a refunded/disputed lifetime purchase; a // redelivered grant for it is dead on arrival (applyWebhookEvent). func (db *DB) markPaymentRevoked(paymentIntent string) error { _, err := db.sql.Exec( "INSERT INTO revoked_payments(payment_intent, revoked_at) VALUES(?, ?) ON CONFLICT DO NOTHING", paymentIntent, time.Now().Unix()) return err } func (db *DB) isPaymentRevoked(paymentIntent string) bool { var n int _ = db.sql.QueryRow("SELECT COUNT(*) FROM revoked_payments WHERE payment_intent = ?", paymentIntent).Scan(&n) return n > 0 } // setEntitlement upserts a member's paid membership. It ensures the user_tiers // row exists, then stamps the subscription lifecycle — so a webhook can // create-or-update in one call. It NEVER changes an existing row's source: a // paid subscription tracks its lifecycle in the ext_* columns, but an // admin/invite grant underneath stays sticky (so a lapse can't strip an // admin's permanent comp — the revoke path checks source). Assigning the level // to the member (key delivery) is the caller's job on activation, via // assignTier + a tier-key prod, exactly like an admin assignment. func (db *DB) setEntitlement(e Entitlement) error { tx, err := db.sql.Begin() if err != nil { return err } defer tx.Rollback() if _, err := tx.Exec( "INSERT INTO user_tiers(tier_id, user_id, source, created_at) VALUES(?, ?, 'stripe', ?) ON CONFLICT(tier_id, user_id) DO NOTHING", e.TierID, e.UserID, time.Now().Unix()); err != nil { return err } // No source= here: preserve an existing admin/invite grant (see doc above). // The ordering watermark's domain is ONE subscription: within the same sub, // MAX() so a checkout receipt (EventTS 0 — see applySubEvent) folded after a // real lifecycle event can't wind it backwards; when the row's subscription // CHANGES (a level switch moved a different sub onto this row), the old // sub's watermark is meaningless — keep the incoming event's stamp, or the // moved sub's later, lower-stamped events would be dropped as stale. if _, err := tx.Exec( `UPDATE user_tiers SET status=?, paid_until=?, grace_deadline=?, lifetime=?, ext_event_ts = CASE WHEN COALESCE(ext_sub,'') = COALESCE(?,'') THEN MAX(ext_event_ts, ?) ELSE ? END, ext_sub=?, ext_customer=? WHERE tier_id=? AND user_id=?`, e.Status, e.PaidUntil, e.GraceDeadline, e.Lifetime, nullable(e.ExtSub), e.EventTS, e.EventTS, nullable(e.ExtSub), nullable(e.ExtCustomer), e.TierID, e.UserID); err != nil { return err } return tx.Commit() } // clearEntitlement drops only the subscription tracking (status + provider ids) // from a membership, KEEPING the row, its source, and the tier key. Used when a // stripe subscription lapses on a membership whose sticky source is admin/invite // — the paid lifecycle ends, but the admin-granted access must not. func (db *DB) clearEntitlement(tierID, userID string) error { _, err := db.sql.Exec( "UPDATE user_tiers SET status='', paid_until=0, grace_deadline=0, ext_sub=NULL, ext_customer=NULL, ext_event_ts=0 WHERE tier_id=? AND user_id=?", tierID, userID) return err } // revokeEntitlement ends a lapsed/cancelled subscription. If the membership // exists ONLY because of the subscription (source='stripe'), it fully unassigns // the tier (key + membership); if an admin/invite grant sits underneath, it // keeps that access and just clears the subscription tracking. removed reports // whether access was actually taken away. func (db *DB) revokeEntitlement(tierID, userID string) (removed bool, err error) { src, _ := db.tierSource(tierID, userID) if src == "stripe" { return true, db.unassignTier(tierID, userID) } return false, db.clearEntitlement(tierID, userID) } // tierSource reads how a membership was assigned (admin | invite | stripe). func (db *DB) tierSource(tierID, userID string) (string, bool) { var src sql.NullString err := db.sql.QueryRow( "SELECT source FROM user_tiers WHERE tier_id=? AND user_id=?", tierID, userID).Scan(&src) if err != nil { return "", false } return src.String, true } // entitlementBySub finds the paid membership a provider subscription id maps to // — how a webhook (which knows only the subscription) locates the member. func (db *DB) entitlementBySub(extSub string) (Entitlement, bool) { var e Entitlement var status sql.NullString err := db.sql.QueryRow( "SELECT tier_id, user_id, COALESCE(status,''), COALESCE(paid_until,0), COALESCE(grace_deadline,0), COALESCE(ext_event_ts,0), COALESCE(lifetime,0) FROM user_tiers WHERE ext_sub = ?", extSub).Scan(&e.TierID, &e.UserID, &status, &e.PaidUntil, &e.GraceDeadline, &e.EventTS, &e.Lifetime) if err != nil { return Entitlement{}, false } e.ExtSub = extSub e.Status = status.String return e, true } // memberEntitlement returns a member's subscription on one level (for the // subscription-management UI). Identified by holding a provider subscription id // (any source — an admin-granted member can still self-subscribe), or ok=false. func (db *DB) memberEntitlement(tierID, userID string) (Entitlement, bool) { var e Entitlement var status, sub, cust sql.NullString err := db.sql.QueryRow( "SELECT COALESCE(status,''), COALESCE(paid_until,0), COALESCE(grace_deadline,0), ext_sub, ext_customer, COALESCE(ext_event_ts,0), COALESCE(lifetime,0) FROM user_tiers WHERE tier_id=? AND user_id=? AND ext_sub IS NOT NULL AND ext_sub != ''", tierID, userID).Scan(&status, &e.PaidUntil, &e.GraceDeadline, &sub, &cust, &e.EventTS, &e.Lifetime) if err != nil { return Entitlement{}, false } e.TierID, e.UserID = tierID, userID e.Status, e.ExtSub, e.ExtCustomer = status.String, sub.String, cust.String return e, true } // entitlementsDueForRevoke lists subscriptions to revoke: those whose explicit // grace deadline has elapsed, AND — the true backstop — active/past-due subs // whose paid period ran out more than the grace window ago with no renewal // webhook ever arriving (endpoint down, instance hibernated). source rides along // so the sweep can tell a pure-stripe membership (fully unassign) from one with // an admin/invite grant underneath (keep access, just drop the sub). func (db *DB) entitlementsDueForRevoke(now int64) ([]Entitlement, error) { graceSecs := int64(billingGrace.Seconds()) rows, err := db.sql.Query( `SELECT tier_id, user_id, source FROM user_tiers WHERE ext_sub IS NOT NULL AND ext_sub != '' AND ( (grace_deadline > 0 AND grace_deadline <= ?) OR (paid_until > 0 AND paid_until + ? <= ?) )`, now, graceSecs, now) if err != nil { return nil, err } defer rows.Close() out := []Entitlement{} for rows.Next() { var e Entitlement var src sql.NullString if err := rows.Scan(&e.TierID, &e.UserID, &src); err != nil { return nil, err } e.Source = src.String out = append(out, e) } return out, rows.Err() } // entitlementsNeedingRepair finds paid memberships stuck ACTIVE with no // period end — rows whose lifecycle events were lost, which neither // entitlementsDueForRevoke clause can ever match. The sweep re-fetches // provider truth for these (repairEntitlements). func (db *DB) entitlementsNeedingRepair(limit int) ([]Entitlement, error) { rows, err := db.sql.Query( `SELECT tier_id, user_id, ext_sub FROM user_tiers WHERE ext_sub IS NOT NULL AND ext_sub != '' AND status = 'active' AND paid_until = 0 AND COALESCE(lifetime, 0) = 0 LIMIT ?`, limit) if err != nil { return nil, err } defer rows.Close() out := []Entitlement{} for rows.Next() { var e Entitlement if err := rows.Scan(&e.TierID, &e.UserID, &e.ExtSub); err != nil { return nil, err } out = append(out, e) } return out, rows.Err() } // grantTierKey stores one member's wrapped copy of a tier key. DO NOTHING, // never overwrite — the grantInstanceKey anti-brick rule, per tier. func (db *DB) grantTierKey(tierID, userID, wrapped string) error { _, err := db.sql.Exec( "INSERT INTO tier_keys(tier_id, user_id, wrapped) VALUES(?, ?, ?) ON CONFLICT(tier_id, user_id) DO NOTHING", tierID, userID, wrapped) return err } var errAlreadyMinted = errors.New("tier key already minted") // mintTierKeys is the ONE-TIME key creation for a level: it flips key_minted // and stores the initial grants in a single transaction. The guarded UPDATE // makes concurrent minters serialize — exactly one wins; the loser's key // never enters the tier (it reloads and proceeds as a nudger/holder instead). func (db *DB) mintTierKeys(tierID string, grants []ThreadMemberKey) error { tx, err := db.sql.Begin() if err != nil { return err } defer tx.Rollback() res, err := tx.Exec("UPDATE tiers SET key_minted = 1 WHERE id = ? AND key_minted = 0", tierID) if err != nil { return err } if n, _ := res.RowsAffected(); n == 0 { return errAlreadyMinted } for _, g := range grants { if _, err := tx.Exec( "INSERT INTO tier_keys(tier_id, user_id, wrapped) VALUES(?, ?, ?) ON CONFLICT(tier_id, user_id) DO NOTHING", tierID, g.UserID, g.Wrapped); err != nil { return err } } return tx.Commit() } func (db *DB) tierKeyMinted(tierID string) bool { var x int if db.sql.QueryRow("SELECT COALESCE(key_minted, 0) FROM tiers WHERE id = ?", tierID).Scan(&x) != nil { return false } return x == 1 } // setInviteTier attaches an access level to an invite (a cohort invite): the // claimer is assigned the level on first-device claim (claimKeyInvite). func (db *DB) setInviteTier(inviteID, tierID string) error { _, err := db.sql.Exec("UPDATE key_invites SET tier_id = ? WHERE id = ?", tierID, inviteID) return err } // tierName returns a level's plaintext name ("" if unknown). Names are // branding-class by design — they render on claim screens where no key exists. func (db *DB) tierName(id string) string { var n string if db.sql.QueryRow("SELECT name FROM tiers WHERE id = ?", id).Scan(&n) != nil { return "" } return n } // userTiersNeedingKeys lists the levels a member belongs to but holds no key // copy for — the setPubkey prod targets (a cohort joiner becomes wrappable-to // the moment their pubkey lands). func (db *DB) userTiersNeedingKeys(userID string) []string { rows, err := db.sql.Query(` SELECT ut.tier_id FROM user_tiers ut LEFT JOIN tier_keys k ON k.tier_id = ut.tier_id AND k.user_id = ut.user_id WHERE ut.user_id = ? AND k.user_id IS NULL`, userID) if err != nil { return nil } defer rows.Close() out := []string{} for rows.Next() { var id string if rows.Scan(&id) == nil { out = append(out, id) } } return out } // tierGatesThreads counts threads (any visibility) whose gate is this level — // the deletion guard: those groups' open_keys are wrapped under this level's // key forever, so deleting the level would seal them for self-serve joiners. func (db *DB) tierGatesThreads(tierID string) int { var n int db.sql.QueryRow("SELECT COUNT(*) FROM threads WHERE open_tier = ?", tierID).Scan(&n) return n } // tierMemberIDs / tierHolderIDs feed the targeted WS prods: a key nudge goes // only to members who can actually help (holders), a grant prod only to the // members who just received a copy. func (db *DB) tierMemberIDs(tierID string) ([]string, error) { return db.idColumn("SELECT user_id FROM user_tiers WHERE tier_id = ?", tierID) } func (db *DB) tierHolderIDs(tierID string) ([]string, error) { return db.idColumn("SELECT user_id FROM tier_keys WHERE tier_id = ?", tierID) } func (db *DB) idColumn(query, arg string) ([]string, error) { rows, err := db.sql.Query(query, arg) if err != nil { return nil, err } defer rows.Close() out := []string{} for rows.Next() { var id string if err := rows.Scan(&id); err != nil { return nil, err } out = append(out, id) } return out, rows.Err() } // listTiers returns every tier (names are plaintext space policy, visible to // all members like the roster), personalised with the caller's membership, // their wrapped key copy, and who already holds one. // listTiers returns every level, personalised to userID. isAdmin callers (and // the level's own members) get the FULL member/holder lists — an admin manages // them, and a member must be able to wrap the key to every co-member. To an // OUTSIDER (not a member, not admin) each level's members are filtered by its // visibility policy: 'shown' → all, 'hidden' → none, 'user' → only members who // opted visible. So a hidden level never leaks its roster to non-members, while // the crypto's own truth — members know each other — is preserved. func (db *DB) listTiers(userID string, isAdmin bool) ([]Tier, error) { rows, err := db.sql.Query(` SELECT t.id, t.name, t.created_at, COALESCE(k.wrapped, ''), COALESCE(t.key_minted, 0), COALESCE(t.visibility, 'shown'), CASE WHEN m.user_id IS NULL THEN 0 ELSE 1 END, COALESCE(m.visible, 1), COALESCE(t.price_cents, 0), COALESCE(t.currency, ''), COALESCE(t.interval, ''), COALESCE(m.status, ''), COALESCE(m.paid_until, 0), COALESCE(m.grace_deadline, 0), COALESCE(m.lifetime, 0) FROM tiers t LEFT JOIN tier_keys k ON k.tier_id = t.id AND k.user_id = ? LEFT JOIN user_tiers m ON m.tier_id = t.id AND m.user_id = ? ORDER BY t.created_at, t.id`, userID, userID) if err != nil { return nil, err } defer rows.Close() out := []Tier{} minted := []bool{} myVisible := []bool{} for rows.Next() { var t Tier var m, mine, vis int if err := rows.Scan(&t.ID, &t.Name, &t.CreatedAt, &t.Wrapped, &m, &t.Visibility, &mine, &vis, &t.PriceCents, &t.Currency, &t.Interval, &t.Status, &t.PaidUntil, &t.GraceDeadline, &t.Lifetime); err != nil { return nil, err } t.Mine = mine == 1 minted = append(minted, m == 1) myVisible = append(myVisible, vis == 1) out = append(out, t) } if err := rows.Err(); err != nil { return nil, err } for i := range out { insider := isAdmin || out[i].Mine if insider { if out[i].Members, err = db.tierMemberIDs(out[i].ID); err != nil { return nil, err } if out[i].Have, err = db.tierHolderIDs(out[i].ID); err != nil { return nil, err } out[i].KeyLost = minted[i] && len(out[i].Have) == 0 } else if out[i].Members, err = db.publicTierMemberIDs(out[i].ID, out[i].Visibility); err != nil { return nil, err } // My own show/hide state — only a control when I'm a member of a // 'user'-policy level; otherwise the policy decides, no personal say. out[i].Hidden = out[i].Mine && out[i].Visibility == "user" && !myVisible[i] } return out, nil } // publicTierMemberIDs is the member list an OUTSIDER may see: all for 'shown', // none for 'hidden', the opted-in subset for 'user'. func (db *DB) publicTierMemberIDs(tierID, visibility string) ([]string, error) { switch visibility { case "hidden": return []string{}, nil case "user": return db.idColumn("SELECT user_id FROM user_tiers WHERE tier_id = ? AND COALESCE(visible, 1) = 1", tierID) default: // shown return db.tierMemberIDs(tierID) } } // setTierVisibility sets a level's admin policy ('shown'|'hidden'|'user'). func (db *DB) setTierVisibility(tierID, visibility string) error { _, err := db.sql.Exec("UPDATE tiers SET visibility = ? WHERE id = ?", visibility, tierID) return err } // setUserTierVisible sets one member's show/hide for a level (their own choice, // honoured only while the level's policy is 'user'). func (db *DB) setUserTierVisible(tierID, userID string, visible bool) error { v := 0 if visible { v = 1 } _, err := db.sql.Exec("UPDATE user_tiers SET visible = ? WHERE tier_id = ? AND user_id = ?", v, tierID, userID) return err } type Thread struct { ID string `json:"id"` Kind string `json:"kind"` // self | dm | group NameBlob string `json:"name_blob,omitempty"` CreatedAt int64 `json:"created_at"` // Wrapped is this thread's key, wrapped to the requesting user (set only by // threadsForUser, so the client can unwrap and decrypt in one round trip). Wrapped string `json:"wrapped,omitempty"` // Members lists the thread's user ids (so the client can label a DM by the // other person and render group membership). Members []string `json:"members,omitempty"` // LastBlob/LastAt are the most recent message's ciphertext + time, so the // chat list can show a (client-decrypted) preview without opening the thread. LastBlob string `json:"last_blob,omitempty"` LastAt int64 `json:"last_at,omitempty"` // Avatar is the group photo's blob id (encrypted with the thread key). Avatar string `json:"avatar,omitempty"` // Hidden is true when the requesting user has closed (hidden) this chat. Hidden bool `json:"hidden,omitempty"` // LastReadAt is the requesting user's read watermark — the time of the // newest message they've read here, on any of their devices (0 = never). // The client compares it to LastAt to decide the unread dot. LastReadAt int64 `json:"last_read_at,omitempty"` // Pinned is true when the requesting user keeps this chat at the top of // their list. Pinned bool `json:"pinned,omitempty"` // Muted is true when the requesting user has muted this chat — no pushes, // no unread-badge escalation on their devices. Muted bool `json:"muted,omitempty"` // DMClosed is true when either party ended this DM (issue #578): the // server refuses new messages until someone reopens it, and clients show // the closed banner instead of the composer. DMClosed bool `json:"dm_closed,omitempty"` // Visibility is "open" for a community group any member can discover and // join (see openThreads); omitted for ordinary invite-only threads. Visibility string `json:"visibility,omitempty"` // OpenKey is the thread key encrypted under the instance key — or, when // OpenTier is set, under that tier's key (open groups only). It's the key // source for members who joined self-serve with no per-member wrapped copy. OpenKey string `json:"open_key,omitempty"` // OpenTier gates an open group to one access level ("" = every member). OpenTier string `json:"open_tier,omitempty"` // DefaultJoin marks an open group new members are auto-joined to. DefaultJoin bool `json:"default_join,omitempty"` } // ThreadMemberKey pairs a member with the thread key wrapped to their pubkey. type ThreadMemberKey struct { UserID string `json:"user_id"` Wrapped string `json:"wrapped"` } // createThread creates a thread with its members and their wrapped key blobs in // one transaction. nameBlob is the client-encrypted group name ("" for dm/self). func (db *DB) createThread(kind, nameBlob string, members []ThreadMemberKey) (string, error) { return db.createThreadVis(kind, nameBlob, members, "", "", "") } // createThreadVis also stamps visibility + open_key for open community groups // (see openThreads). "" visibility means the default 'invite'. Members with an // empty Wrapped get no thread_keys row — an open group's key reaches them via // open_key instead. func (db *DB) createThreadVis(kind, nameBlob string, members []ThreadMemberKey, visibility, openKey, openTier string) (string, error) { id, err := randToken() if err != nil { return "", err } tx, err := db.sql.Begin() if err != nil { return "", err } defer tx.Rollback() now := time.Now().Unix() var nb any if nameBlob != "" { nb = nameBlob } if visibility == "" { visibility = "invite" } var ok, ot any if openKey != "" { ok = openKey } if openTier != "" { ot = openTier } if _, err := tx.Exec("INSERT INTO threads(id, kind, name_blob, created_at, visibility, open_key, open_tier) VALUES(?, ?, ?, ?, ?, ?, ?)", id, kind, nb, now, visibility, ok, ot); err != nil { return "", err } for _, m := range members { if _, err := tx.Exec("INSERT INTO thread_members(thread_id, user_id, created_at) VALUES(?, ?, ?)", id, m.UserID, now); err != nil { return "", err } if m.Wrapped == "" { continue } if _, err := tx.Exec("INSERT INTO thread_keys(thread_id, user_id, wrapped) VALUES(?, ?, ?)", id, m.UserID, m.Wrapped); err != nil { return "", err } } return id, tx.Commit() } // addThreadMember adds one member of an existing thread. The thread_keys insert // is DO NOTHING, never REPLACE: any member may add another, so replacing would // let one member overwrite an existing member's good wrapped key with junk and // lock them out of the thread. Legitimate re-keying (invite adoption) is a // separate, explicit path (see adoptUserInvite). func (db *DB) addThreadMember(threadID, userID, wrapped string) error { tx, err := db.sql.Begin() if err != nil { return err } defer tx.Rollback() now := time.Now().Unix() if _, err := tx.Exec("INSERT OR IGNORE INTO thread_members(thread_id, user_id, created_at) VALUES(?, ?, ?)", threadID, userID, now); err != nil { return err } if _, err := tx.Exec("INSERT INTO thread_keys(thread_id, user_id, wrapped) VALUES(?, ?, ?) ON CONFLICT(thread_id, user_id) DO NOTHING", threadID, userID, wrapped); err != nil { return err } return tx.Commit() } func (db *DB) threadMember(threadID, userID string) bool { var x int if db.sql.QueryRow("SELECT 1 FROM thread_members WHERE thread_id = ? AND user_id = ?", threadID, userID).Scan(&x) == nil { return true } // The space thread (spacethread.go) has IMPLICIT membership: every member // of the space is a member by construction, so there are no rows to keep in // sync and no join path that can drift. Agents are excluded (mirroring // joinDefaultThreads) — a bot reading the space log must be a deliberate, // disclosed choice, made by adding an explicit membership row above. return db.threadKind(threadID) == "space" && db.userExists(userID) && !db.isAgent(userID) } // messageInThread reports whether message id lives in threadID. Binds a // reaction/edit to the thread the client names (membership is checked // separately), so a member of one thread can't react to or edit messages in a // thread they don't belong to. func (db *DB) messageInThread(id int64, threadID string) bool { var x int return db.sql.QueryRow("SELECT 1 FROM messages WHERE id = ? AND thread_id = ?", id, threadID).Scan(&x) == nil } // threadKind returns a thread's kind ("self"/"dm"/"group"), or "" if unknown. func (db *DB) threadKind(threadID string) string { var k string if db.sql.QueryRow("SELECT kind FROM threads WHERE id = ?", threadID).Scan(&k) != nil { return "" } return k } // deleteThread removes a thread and its membership/key rows. Used to roll back a // DM thread that lost the dedupe race (see postThread). func (db *DB) deleteThread(id string) error { tx, err := db.sql.Begin() if err != nil { return err } defer tx.Rollback() for _, q := range []string{ "DELETE FROM thread_keys WHERE thread_id = ?", "DELETE FROM thread_members WHERE thread_id = ?", "DELETE FROM closed_dms WHERE thread_id = ?", "DELETE FROM threads WHERE id = ?", } { if _, err := tx.Exec(q, id); err != nil { return err } } if err := tx.Commit(); err != nil { return err } // The thread's messages referenced these blobs; drop their bytes too. return db.deleteThreadBlobs(id) } // wrappedKeyFor returns a member's wrapped thread key (to unwrap client-side). func (db *DB) wrappedKeyFor(threadID, userID string) (string, bool, error) { var w string err := db.sql.QueryRow("SELECT wrapped FROM thread_keys WHERE thread_id = ? AND user_id = ?", threadID, userID).Scan(&w) if err == sql.ErrNoRows { return "", false, nil } if err != nil { return "", false, err } return w, true, nil } // threadsForUser lists the threads a user belongs to (their chat list), each // with the thread key wrapped to that user and the member id list. func (db *DB) threadsForUser(userID string) ([]Thread, error) { rows, err := db.sql.Query(` SELECT t.id, t.kind, COALESCE(t.name_blob, ''), t.created_at, COALESCE(k.wrapped, ''), -- tombstone rows (#581) are sync bookkeeping, not activity: they must -- not become the preview blob or bump the thread's recency below. COALESCE((SELECT blob FROM messages WHERE thread_id = t.id AND tombstone_of IS NULL ORDER BY id DESC LIMIT 1), ''), COALESCE((SELECT created_at FROM messages WHERE thread_id = t.id AND tombstone_of IS NULL ORDER BY id DESC LIMIT 1), 0), COALESCE(t.avatar_blob, ''), COALESCE(t.visibility, 'invite'), COALESCE(t.open_key, ''), COALESCE(t.open_tier, ''), COALESCE(t.default_join, 0), EXISTS(SELECT 1 FROM hidden_threads h WHERE h.thread_id = t.id AND h.user_id = m.user_id), EXISTS(SELECT 1 FROM pinned_threads p WHERE p.thread_id = t.id AND p.user_id = m.user_id) AS pinned, EXISTS(SELECT 1 FROM muted_threads mu WHERE mu.thread_id = t.id AND mu.user_id = m.user_id), EXISTS(SELECT 1 FROM closed_dms c WHERE c.thread_id = t.id), COALESCE((SELECT last_read_at FROM read_marks r WHERE r.thread_id = t.id AND r.user_id = m.user_id), 0) FROM threads t JOIN thread_members m ON m.thread_id = t.id LEFT JOIN thread_keys k ON k.thread_id = t.id AND k.user_id = m.user_id WHERE m.user_id = ? ORDER BY pinned DESC, COALESCE((SELECT MAX(created_at) FROM messages WHERE thread_id = t.id AND tombstone_of IS NULL), t.created_at) DESC, -- created_at is second-resolution, so threads active in the same -- second tie; message ids are monotonic, making the order stable. COALESCE((SELECT MAX(id) FROM messages WHERE thread_id = t.id AND tombstone_of IS NULL), 0) DESC, t.created_at DESC`, userID) if err != nil { return nil, err } defer rows.Close() out := []Thread{} for rows.Next() { var t Thread if err := rows.Scan(&t.ID, &t.Kind, &t.NameBlob, &t.CreatedAt, &t.Wrapped, &t.LastBlob, &t.LastAt, &t.Avatar, &t.Visibility, &t.OpenKey, &t.OpenTier, &t.DefaultJoin, &t.Hidden, &t.Pinned, &t.Muted, &t.DMClosed, &t.LastReadAt); err != nil { return nil, err } if t.Visibility == "invite" { t.Visibility = "" // the default; keep the JSON row lean } out = append(out, t) } if err := rows.Err(); err != nil { return nil, err } for i := range out { if out[i].Members, err = db.memberUserIDs(out[i].ID); err != nil { return nil, err } } return out, nil } // OpenThread is a community-directory row: enough for any member to render // and self-join an open group. The name/avatar blobs decrypt with the thread // key, which every member can reach via open_key + the instance key — so the // whole directory is readable by any member and by nobody else. type OpenThread struct { ID string `json:"id"` NameBlob string `json:"name_blob,omitempty"` CreatedAt int64 `json:"created_at"` Avatar string `json:"avatar,omitempty"` OpenKey string `json:"open_key"` MemberCount int `json:"member_count"` Joined bool `json:"joined"` DefaultJoin bool `json:"default_join,omitempty"` // Level gate: the tier that may join ("" = everyone), its plaintext name // (so ineligible members see an honest "For Coaches" lock, not a blank), // and whether the requesting user is eligible. OpenTier string `json:"open_tier,omitempty"` TierName string `json:"tier_name,omitempty"` Eligible bool `json:"eligible"` } // openThreads lists every open group (joined or not) for the community // directory, busiest first. func (db *DB) openThreads(userID string) ([]OpenThread, error) { rows, err := db.sql.Query(` SELECT t.id, COALESCE(t.name_blob, ''), t.created_at, COALESCE(t.avatar_blob, ''), COALESCE(t.open_key, ''), (SELECT COUNT(*) FROM thread_members m WHERE m.thread_id = t.id) AS n, EXISTS(SELECT 1 FROM thread_members m WHERE m.thread_id = t.id AND m.user_id = ?), COALESCE(t.default_join, 0), COALESCE(t.open_tier, ''), COALESCE(tr.name, ''), (COALESCE(t.open_tier, '') = '' OR EXISTS( SELECT 1 FROM user_tiers ut WHERE ut.tier_id = t.open_tier AND ut.user_id = ?)) FROM threads t LEFT JOIN tiers tr ON tr.id = t.open_tier WHERE t.visibility = 'open' ORDER BY n DESC, t.created_at DESC`, userID, userID) if err != nil { return nil, err } defer rows.Close() out := []OpenThread{} for rows.Next() { var t OpenThread if err := rows.Scan(&t.ID, &t.NameBlob, &t.CreatedAt, &t.Avatar, &t.OpenKey, &t.MemberCount, &t.Joined, &t.DefaultJoin, &t.OpenTier, &t.TierName, &t.Eligible); err != nil { return nil, err } out = append(out, t) } return out, rows.Err() } // threadIsOpen reports whether a thread is an open (discoverable) group. func (db *DB) threadIsOpen(threadID string) bool { var v string if db.sql.QueryRow("SELECT COALESCE(visibility, 'invite') FROM threads WHERE id = ?", threadID).Scan(&v) != nil { return false } return v == "open" } // threadOpenInfo reports whether a thread already has an open_key stored and // which access level (if any) gates it — the immutability check for // postThreadVisibility. func (db *DB) threadOpenInfo(threadID string) (hasKey bool, tier string) { var k sql.NullString var t sql.NullString if db.sql.QueryRow("SELECT open_key, open_tier FROM threads WHERE id = ?", threadID).Scan(&k, &t) != nil { return false, "" } return k.Valid && k.String != "", t.String } // nullable maps "" to SQL NULL, for optional TEXT columns. func nullable(s string) any { if s == "" { return nil } return s } // setThreadDefault flags (or unflags) an open group as joined-by-default for // new members. Policy, not key material — flipping it moves no keys. func (db *DB) setThreadDefault(threadID string, def bool) error { v := 0 if def { v = 1 } _, err := db.sql.Exec("UPDATE threads SET default_join = ? WHERE id = ?", v, threadID) return err } // joinDefaultThreads adds a brand-new member to every default open group — // membership rows only, like any self-serve join (open_key carries the key). // Closed groups stop defaulting even if their flag is still set. Agents are // skipped: auto-adding a bot would silently widen what its sidecar can read, // and bot membership must stay a deliberate, disclosed choice (docs/bots.md). func (db *DB) joinDefaultThreads(userID string) error { _, err := db.sql.Exec(` INSERT OR IGNORE INTO thread_members(thread_id, user_id, created_at) SELECT t.id, ?, ? FROM threads t WHERE t.visibility = 'open' AND COALESCE(t.default_join, 0) = 1 AND COALESCE(t.open_tier, '') = '' AND (SELECT COALESCE(is_agent, 0) FROM users WHERE id = ?) = 0`, userID, time.Now().Unix(), userID) return err } var errNotOpen = errors.New("not an open thread") // joinOpenThread self-serve-joins a member to an open group: a membership row // only, no thread_keys row — the group's key reaches them via open_key, which // they can already unwrap with the instance key. Errors errNotOpen for // anything that isn't an open group (including unknown threads). func (db *DB) joinOpenThread(threadID, userID string) error { tx, err := db.sql.Begin() if err != nil { return err } defer tx.Rollback() var v, gate string if err := tx.QueryRow("SELECT COALESCE(visibility, 'invite'), COALESCE(open_tier, '') FROM threads WHERE id = ?", threadID).Scan(&v, &gate); err != nil { if err == sql.ErrNoRows { return errNotOpen } return err } if v != "open" { return errNotOpen } // A level-gated group is open only to that access level — the same // errNotOpen to outsiders, so the refusal leaks nothing they can't // already see in the directory's honest lock. if gate != "" { var x int if err := tx.QueryRow("SELECT 1 FROM user_tiers WHERE tier_id = ? AND user_id = ?", gate, userID).Scan(&x); err != nil { if err == sql.ErrNoRows { return errNotOpen } return err } } if _, err := tx.Exec("INSERT OR IGNORE INTO thread_members(thread_id, user_id, created_at) VALUES(?, ?, ?)", threadID, userID, time.Now().Unix()); err != nil { return err } return tx.Commit() } // setThreadVisibility opens or closes a group for discovery. Opening stores // open_key only when none exists yet (COALESCE, the grantInstanceKey // anti-brick rule: any member may call this, so an overwrite could replace a // good key blob with junk and cut off every self-serve joiner). Closing keeps // open_key — members who joined self-serve still decrypt through it; 'invite' // only stops discovery and new joins. func (db *DB) setThreadVisibility(threadID, visibility, openKey, openTier string) error { if visibility == "open" { // open_tier is stamped in the same breath as open_key and only then — // SET expressions read the pre-update row, so both CASEs test the OLD // open_key. A later gate change would need a rekey the never-overwrite // rule forbids. _, err := db.sql.Exec(`UPDATE threads SET visibility = 'open', open_tier = CASE WHEN open_key IS NULL THEN ? ELSE open_tier END, open_key = COALESCE(open_key, ?) WHERE id = ?`, nullable(openTier), openKey, threadID) return err } _, err := db.sql.Exec("UPDATE threads SET visibility = ? WHERE id = ?", visibility, threadID) return err } // spaceThread returns the instance's one space thread (spacethread.go), if it // has been created: its id and open_key (the space key wrapped under the // instance key — the same open-group mechanism, opaque to the server). func (db *DB) spaceThread() (id, openKey string, ok bool, err error) { var k sql.NullString err = db.sql.QueryRow("SELECT id, open_key FROM threads WHERE kind = 'space'").Scan(&id, &k) if err == sql.ErrNoRows { return "", "", false, nil } if err != nil { return "", "", false, err } return id, k.String, true, nil } // createSpaceThread creates THE space thread if none exists, storing openKey. // Idempotent under races: the partial unique index (idx_one_space_thread) // makes a losing concurrent insert fail, and the loser re-reads the winner's // row — so there is exactly one space thread and its open_key, once stored, is // never overwritten (the grantInstanceKey anti-brick rule). func (db *DB) createSpaceThread(openKey string) (id, storedKey string, created bool, err error) { if id, storedKey, ok, err := db.spaceThread(); err != nil || ok { return id, storedKey, false, err } id, err = randToken() if err != nil { return "", "", false, err } _, err = db.sql.Exec( "INSERT INTO threads(id, kind, created_at, visibility, open_key) VALUES(?, 'space', ?, 'invite', ?)", id, time.Now().Unix(), openKey) if err != nil { // Lost the create race: the unique index rejected a second space thread. // The winner's row is the space thread — return it. if id2, key2, ok, err2 := db.spaceThread(); err2 == nil && ok { return id2, key2, false, nil } return "", "", false, err } return id, openKey, true, nil } // removeThreadMember drops a user from a thread (their membership + key copy). // Note: the thread key isn't rotated, so this stops the server delivering future // messages to them, but isn't forward-secret (rekey-on-leave is deferred). func (db *DB) removeThreadMember(threadID, userID string) error { tx, err := db.sql.Begin() if err != nil { return err } defer tx.Rollback() if _, err := tx.Exec("DELETE FROM thread_members WHERE thread_id = ? AND user_id = ?", threadID, userID); err != nil { return err } if _, err := tx.Exec("DELETE FROM thread_keys WHERE thread_id = ? AND user_id = ?", threadID, userID); err != nil { return err } return tx.Commit() } // setThreadHidden closes (hidden=true) or reopens (hidden=false) a chat for one // user — a per-user marker, synced across their devices via getThreads. func (db *DB) setThreadHidden(userID, threadID string, hidden bool) error { if hidden { _, err := db.sql.Exec( "INSERT OR IGNORE INTO hidden_threads(user_id, thread_id) VALUES(?, ?)", userID, threadID, ) return err } _, err := db.sql.Exec("DELETE FROM hidden_threads WHERE user_id = ? AND thread_id = ?", userID, threadID) return err } // setThreadPinned pins (pinned=true) or unpins (pinned=false) a chat for one // user — a per-user marker, synced across their devices via getThreads. func (db *DB) setThreadPinned(userID, threadID string, pinned bool) error { if pinned { _, err := db.sql.Exec( "INSERT OR IGNORE INTO pinned_threads(user_id, thread_id) VALUES(?, ?)", userID, threadID, ) return err } _, err := db.sql.Exec("DELETE FROM pinned_threads WHERE user_id = ? AND thread_id = ?", userID, threadID) return err } // userDMPrivacy returns a member's DM-privacy setting: "" (everyone, the // default) or "nobody". func (db *DB) userDMPrivacy(userID string) string { var v string db.sql.QueryRow("SELECT COALESCE(dm_privacy, '') FROM users WHERE id = ?", userID).Scan(&v) return v } func (db *DB) setUserDMPrivacy(userID, v string) error { _, err := db.sql.Exec("UPDATE users SET dm_privacy = ? WHERE id = ?", v, userID) return err } // closeDM marks a DM closed: the server stops accepting new traffic into it // until reopenDM. Idempotent — the first closer wins the closed_by credit. func (db *DB) closeDM(threadID, userID string) error { _, err := db.sql.Exec( "INSERT OR IGNORE INTO closed_dms(thread_id, closed_by, created_at) VALUES(?, ?, ?)", threadID, userID, time.Now().Unix(), ) return err } func (db *DB) reopenDM(threadID string) error { _, err := db.sql.Exec("DELETE FROM closed_dms WHERE thread_id = ?", threadID) return err } // dmClosed reports whether a DM is closed, and by whom. func (db *DB) dmClosed(threadID string) (bool, string) { var by string err := db.sql.QueryRow("SELECT closed_by FROM closed_dms WHERE thread_id = ?", threadID).Scan(&by) return err == nil, by } // ---- reports + audit log (issue #578) --------------------------------------- // Report is a queue row: metadata + (when the caller is an admin member) the // content blob sealed to THAT admin's key. ReporterCount/ReportedCount are the // repeat signals the queue shows — how often the reporter files, how often the // reported member is named — because testimony is judged on patterns. type Report struct { ID string `json:"id"` CreatedAt int64 `json:"created_at"` ReporterID string `json:"reporter_id"` ReportedID string `json:"reported_id"` ThreadID string `json:"thread_id,omitempty"` Status string `json:"status"` Action string `json:"action,omitempty"` ResolvedBy string `json:"resolved_by,omitempty"` ResolvedAt int64 `json:"resolved_at,omitempty"` Blob string `json:"blob,omitempty"` ReporterCount int `json:"reporter_count"` ReportedCount int `json:"reported_count"` // InvitedBy is who brought the reported member in ('' when unknown) — // invite hygiene for the queue (issue #578). InvitedBy string `json:"invited_by,omitempty"` } // createReport stores the queue row + the per-admin sealed content blobs in // one transaction. func (db *DB) createReport(reporterID, reportedID, threadID string, blobs map[string]string) (string, error) { id, err := randToken() if err != nil { return "", err } tx, err := db.sql.Begin() if err != nil { return "", err } defer tx.Rollback() if _, err := tx.Exec( "INSERT INTO reports(id, created_at, reporter_id, reported_id, thread_id) VALUES(?, ?, ?, ?, ?)", id, time.Now().Unix(), reporterID, reportedID, threadID, ); err != nil { return "", err } for adminID, blob := range blobs { if _, err := tx.Exec( "INSERT INTO report_blobs(report_id, admin_id, blob) VALUES(?, ?, ?)", id, adminID, blob, ); err != nil { return "", err } } return id, tx.Commit() } // reportsSince counts a reporter's filings in a window — the content-blind // brake on harassment-by-reporting. func (db *DB) reportsSince(reporterID string, since int64) int { var n int db.sql.QueryRow("SELECT COUNT(*) FROM reports WHERE reporter_id = ? AND created_at >= ?", reporterID, since).Scan(&n) return n } // listReports returns the queue (open first, newest first), each row carrying // the caller-admin's own sealed blob and the repeat-signal counts. func (db *DB) listReports(forAdminID string) ([]Report, error) { rows, err := db.sql.Query(` SELECT r.id, r.created_at, r.reporter_id, r.reported_id, r.thread_id, r.status, r.action, r.resolved_by, r.resolved_at, COALESCE((SELECT blob FROM report_blobs b WHERE b.report_id = r.id AND b.admin_id = ?), ''), (SELECT COUNT(*) FROM reports x WHERE x.reporter_id = r.reporter_id), (SELECT COUNT(*) FROM reports x WHERE x.reported_id = r.reported_id), COALESCE((SELECT invited_by FROM users u WHERE u.id = r.reported_id), '') FROM reports r ORDER BY (r.status = 'open') DESC, r.created_at DESC LIMIT 500`, forAdminID) if err != nil { return nil, err } defer rows.Close() out := []Report{} for rows.Next() { var r Report if err := rows.Scan(&r.ID, &r.CreatedAt, &r.ReporterID, &r.ReportedID, &r.ThreadID, &r.Status, &r.Action, &r.ResolvedBy, &r.ResolvedAt, &r.Blob, &r.ReporterCount, &r.ReportedCount, &r.InvitedBy); err != nil { return nil, err } out = append(out, r) } return out, rows.Err() } // openReportCount is what the report notification says — "N waiting" is // queue metadata the server already holds, never content. func (db *DB) openReportCount() int { var n int db.sql.QueryRow("SELECT COUNT(*) FROM reports WHERE status = 'open'").Scan(&n) return n } // resolveReport records the outcome. Reports whether the row was still open // (a second resolver loses, keeping one outcome per report). func (db *DB) resolveReport(id, action, resolvedBy string) (bool, error) { res, err := db.sql.Exec( "UPDATE reports SET status = 'resolved', action = ?, resolved_by = ?, resolved_at = ? WHERE id = ? AND status = 'open'", action, resolvedBy, time.Now().Unix(), id, ) if err != nil { return false, err } n, err := res.RowsAffected() return n > 0, err } // AuditEntry is one appended moderation action. type AuditEntry struct { ID int64 `json:"id"` Ts int64 `json:"ts"` ActorID string `json:"actor_id"` Action string `json:"action"` SubjectID string `json:"subject_id,omitempty"` ThreadID string `json:"thread_id,omitempty"` ReportID string `json:"report_id,omitempty"` Detail string `json:"detail,omitempty"` } // addAudit appends one moderation action. Best-effort at call sites (an audit // write must never fail the action it records), so it returns no error. func (db *DB) addAudit(actorID, action, subjectID, threadID, reportID, detail string) { _, err := db.sql.Exec( "INSERT INTO audit_log(ts, actor_id, action, subject_id, thread_id, report_id, detail) VALUES(?, ?, ?, ?, ?, ?, ?)", time.Now().Unix(), actorID, action, subjectID, threadID, reportID, detail, ) if err != nil { log.Println("audit:", err) } } // listAudit pages the log newest-first; before=0 starts at the top. func (db *DB) listAudit(before int64, limit int) ([]AuditEntry, error) { if limit <= 0 || limit > 200 { limit = 100 } q := "SELECT id, ts, actor_id, action, subject_id, thread_id, report_id, detail FROM audit_log" args := []any{} if before > 0 { q += " WHERE id < ?" args = append(args, before) } q += " ORDER BY id DESC LIMIT ?" args = append(args, limit) rows, err := db.sql.Query(q, args...) if err != nil { return nil, err } defer rows.Close() out := []AuditEntry{} for rows.Next() { var e AuditEntry if err := rows.Scan(&e.ID, &e.Ts, &e.ActorID, &e.Action, &e.SubjectID, &e.ThreadID, &e.ReportID, &e.Detail); err != nil { return nil, err } out = append(out, e) } return out, rows.Err() } // setBlocked records (or with on=false clears) blocker→blocked. Blocking is // one-directional in the record, mutual in effect (blockedEither). func (db *DB) setBlocked(blockerID, blockedID string, on bool) error { if on { _, err := db.sql.Exec( "INSERT OR IGNORE INTO blocks(blocker_id, blocked_id, created_at) VALUES(?, ?, ?)", blockerID, blockedID, time.Now().Unix(), ) return err } _, err := db.sql.Exec("DELETE FROM blocks WHERE blocker_id = ? AND blocked_id = ?", blockerID, blockedID) return err } // blockedEither reports whether either member has blocked the other — the // question every contact gate asks (new DM, reopen, group add). func (db *DB) blockedEither(a, b string) bool { var x int return db.sql.QueryRow( "SELECT 1 FROM blocks WHERE (blocker_id = ? AND blocked_id = ?) OR (blocker_id = ? AND blocked_id = ?)", a, b, b, a, ).Scan(&x) == nil } // blocksOf lists who this member has blocked (their own list; nobody else's). func (db *DB) blocksOf(userID string) ([]string, error) { rows, err := db.sql.Query("SELECT blocked_id FROM blocks WHERE blocker_id = ? ORDER BY created_at", userID) if err != nil { return nil, err } defer rows.Close() out := []string{} for rows.Next() { var u string if err := rows.Scan(&u); err != nil { return nil, err } out = append(out, u) } return out, rows.Err() } // blockersOf returns the members who block this user — the push fan-out drops // them so a blocked member's group messages never buzz the blocker. func (db *DB) blockersOf(userID string) (map[string]bool, error) { rows, err := db.sql.Query("SELECT blocker_id FROM blocks WHERE blocked_id = ?", userID) if err != nil { return nil, err } defer rows.Close() out := map[string]bool{} for rows.Next() { var u string if err := rows.Scan(&u); err != nil { return nil, err } out[u] = true } return out, rows.Err() } // addBan records a banned public key. Idempotent. func (db *DB) addBan(publicKey, userID, bannedBy string) error { _, err := db.sql.Exec( "INSERT OR IGNORE INTO bans(public_key, user_id, banned_by, banned_at) VALUES(?, ?, ?, ?)", publicKey, userID, bannedBy, time.Now().Unix(), ) return err } // keyBanned reports whether a public key is banned — checked wherever a key // registers (setPubkey covers every invite path, since a member exists only // once their key does). func (db *DB) keyBanned(publicKey string) bool { var x int return db.sql.QueryRow("SELECT 1 FROM bans WHERE public_key = ?", publicKey).Scan(&x) == nil } // setThreadMuted mutes (muted=true) or unmutes (muted=false) a chat for one // user — a per-user marker, synced across their devices via getThreads. The // push fan-out consults it so a muted chat never buzzes any of their devices. func (db *DB) setThreadMuted(userID, threadID string, muted bool) error { if muted { _, err := db.sql.Exec( "INSERT OR IGNORE INTO muted_threads(user_id, thread_id) VALUES(?, ?)", userID, threadID, ) return err } _, err := db.sql.Exec("DELETE FROM muted_threads WHERE user_id = ? AND thread_id = ?", userID, threadID) return err } // mutedUserIDs lists the users who have muted a thread, so push fan-out can // leave them out. func (db *DB) mutedUserIDs(threadID string) (map[string]bool, error) { rows, err := db.sql.Query("SELECT user_id FROM muted_threads WHERE thread_id = ?", threadID) if err != nil { return nil, err } defer rows.Close() out := map[string]bool{} for rows.Next() { var u string if err := rows.Scan(&u); err != nil { return nil, err } out[u] = true } return out, rows.Err() } // setThreadRead advances one user's read watermark for a thread. Forward-only: // a stale write from a lagging device can never wind the mark back and re-light // unread dots the user's other devices already cleared. Reports whether the // mark actually advanced (callers skip the fan-out when it didn't). func (db *DB) setThreadRead(userID, threadID string, at int64) (bool, error) { res, err := db.sql.Exec(` INSERT INTO read_marks(user_id, thread_id, last_read_at) VALUES(?, ?, ?) ON CONFLICT(user_id, thread_id) DO UPDATE SET last_read_at = excluded.last_read_at WHERE excluded.last_read_at > last_read_at`, userID, threadID, at) if err != nil { return false, err } n, err := res.RowsAffected() return n > 0, err } // clearThreadHidden reopens a thread for every user who had closed it — called on // new activity so a closed chat returns to the list when it's messaged. func (db *DB) clearThreadHidden(threadID string) error { _, err := db.sql.Exec("DELETE FROM hidden_threads WHERE thread_id = ?", threadID) return err } // setThreadName updates a group's (client-encrypted) name. Server-blind. func (db *DB) setThreadName(threadID, nameBlob string) error { _, err := db.sql.Exec("UPDATE threads SET name_blob = ? WHERE id = ?", nameBlob, threadID) return err } // setThreadAvatar sets a group's photo (an encrypted blob id, or "" to clear). func (db *DB) setThreadAvatar(threadID, blobID string) error { var b any if blobID != "" { b = blobID } _, err := db.sql.Exec("UPDATE threads SET avatar_blob = ? WHERE id = ?", b, threadID) return err } // memberUserIDs returns the user ids in a thread (for fan-out / rekey). func (db *DB) memberUserIDs(threadID string) ([]string, error) { q, args := "SELECT user_id FROM thread_members WHERE thread_id = ?", []any{threadID} // The space thread's membership is implicit — every non-agent member, plus // any explicit rows (a deliberately-added bot). Mirrors threadMember. if db.threadKind(threadID) == "space" { q = "SELECT id FROM users WHERE COALESCE(is_agent, 0) = 0 UNION " + q } rows, err := db.sql.Query(q, args...) if err != nil { return nil, err } defer rows.Close() out := []string{} for rows.Next() { var u string if err := rows.Scan(&u); err != nil { return nil, err } out = append(out, u) } return out, rows.Err() } // canonPair orders a user pair so there's exactly one dm_pairs row per pair. func canonPair(a, b string) (string, string) { if a > b { return b, a } return a, b } // dmThreadFor returns the existing DM thread between two users, if any (so DMs // are deduped — exactly one per pair). func (db *DB) dmThreadFor(a, b string) (string, bool, error) { lo, hi := canonPair(a, b) var id string err := db.sql.QueryRow("SELECT thread_id FROM dm_pairs WHERE user_a = ? AND user_b = ?", lo, hi).Scan(&id) if err == sql.ErrNoRows { return "", false, nil } if err != nil { return "", false, err } return id, true, nil } func (db *DB) recordDMPair(a, b, threadID string) error { lo, hi := canonPair(a, b) _, err := db.sql.Exec("INSERT INTO dm_pairs(user_a, user_b, thread_id) VALUES(?, ?, ?)", lo, hi, threadID) return err } // otherThreadMember returns the other member of a 2-person thread — the member // whose id isn't notUserID. Used to find the inviter of a pending DM. func (db *DB) otherThreadMember(threadID, notUserID string) (string, bool) { var uid string err := db.sql.QueryRow( "SELECT user_id FROM thread_members WHERE thread_id = ? AND user_id <> ? LIMIT 1", threadID, notUserID, ).Scan(&uid) return uid, err == nil } // adoptUserInvite handles the case where the person opening a user invite is // ALREADY a member on this device: rather than the placeholder becoming them, we // wire their existing account into the DM. If they already share a DM with the // inviter we reuse it (and drop the pending one); otherwise we swap the placeholder // out of the pending thread for the real user. The placeholder — keyless, with no // messages of its own — is deleted either way. wrappedSelf is the thread key // wrapped to the existing user's public key. Returns the thread to open. ok=false // (no error) if the invite is unusable or lost the single-use race. func (db *DB) adoptUserInvite(inviteID, existingUserID, wrappedSelf, claimSecret string) (string, bool, error) { ki, found, err := db.getKeyInvite(inviteID) if err != nil || !found || ki.Used || ki.Expired || ki.ThreadID == "" { return "", false, err } if !validClaimSecret(ki.ClaimHash, claimSecret) { return "", false, nil } placeholder := ki.UserID inviter, ok := db.otherThreadMember(ki.ThreadID, placeholder) if !ok || inviter == existingUserID { return "", false, nil // malformed, or you can't adopt your own invite } existing, haveDM, err := db.dmThreadFor(inviter, existingUserID) if err != nil { return "", false, err } tx, err := db.sql.Begin() if err != nil { return "", false, err } defer tx.Rollback() // Single-use guard: claim the invite; bail if someone beat us to it. res, err := tx.Exec("UPDATE key_invites SET used_at = ? WHERE id = ? AND used_at IS NULL", time.Now().Unix(), inviteID) if err != nil { return "", false, err } if n, _ := res.RowsAffected(); n == 0 { return "", false, nil } oldLo, oldHi := canonPair(inviter, placeholder) target := ki.ThreadID if haveDM && existing != ki.ThreadID { // Already have a DM together — discard the pending thread entirely. target = existing for _, q := range []string{ "DELETE FROM thread_keys WHERE thread_id = ?", "DELETE FROM thread_members WHERE thread_id = ?", "DELETE FROM threads WHERE id = ?", } { if _, err := tx.Exec(q, ki.ThreadID); err != nil { return "", false, err } } if _, err := tx.Exec("DELETE FROM dm_pairs WHERE user_a = ? AND user_b = ?", oldLo, oldHi); err != nil { return "", false, err } } else { // Swap the placeholder out of the pending thread for the real user. now := time.Now().Unix() if _, err := tx.Exec("DELETE FROM thread_members WHERE thread_id = ? AND user_id = ?", ki.ThreadID, placeholder); err != nil { return "", false, err } if _, err := tx.Exec("DELETE FROM thread_keys WHERE thread_id = ? AND user_id = ?", ki.ThreadID, placeholder); err != nil { return "", false, err } if _, err := tx.Exec("INSERT OR IGNORE INTO thread_members(thread_id, user_id, created_at) VALUES(?, ?, ?)", ki.ThreadID, existingUserID, now); err != nil { return "", false, err } if _, err := tx.Exec("INSERT OR REPLACE INTO thread_keys(thread_id, user_id, wrapped) VALUES(?, ?, ?)", ki.ThreadID, existingUserID, wrappedSelf); err != nil { return "", false, err } // Re-key the dm_pair from (inviter, placeholder) to (inviter, existing). newLo, newHi := canonPair(inviter, existingUserID) if _, err := tx.Exec("DELETE FROM dm_pairs WHERE user_a = ? AND user_b = ?", oldLo, oldHi); err != nil { return "", false, err } if _, err := tx.Exec("INSERT OR IGNORE INTO dm_pairs(user_a, user_b, thread_id) VALUES(?, ?, ?)", newLo, newHi, ki.ThreadID); err != nil { return "", false, err } } // The placeholder never joined (no key, no messages) — remove it. if _, err := tx.Exec("DELETE FROM users WHERE id = ?", placeholder); err != nil { return "", false, err } return target, true, tx.Commit() } // cancelUserInvite tears down an unclaimed pending DM: the thread, its keys and // members, the placeholder user, the DM pair, and the invite itself. Only the // inviter (a real, keyed member) may cancel, and only while the placeholder hasn't // joined (still keyless). ok=false (no error) if it doesn't qualify. func (db *DB) cancelUserInvite(threadID, inviterID string) (bool, error) { // The invite for this pending thread, plus the placeholder it belongs to. var inviteID, placeholder string err := db.sql.QueryRow( "SELECT id, user_id FROM key_invites WHERE thread_id = ? AND used_at IS NULL", threadID, ).Scan(&inviteID, &placeholder) if err == sql.ErrNoRows { return false, nil } if err != nil { return false, err } // Guard: caller must be the (keyed) inviter, and the placeholder must not have // joined (a joined member has a public key — cancelling then would delete a real // person and their DM). if !db.threadMember(threadID, inviterID) { return false, nil } var pub sql.NullString if err := db.sql.QueryRow("SELECT public_key FROM users WHERE id = ?", placeholder).Scan(&pub); err != nil { return false, err } if pub.Valid && pub.String != "" { return false, nil // they joined — not a pending invite anymore } tx, err := db.sql.Begin() if err != nil { return false, err } defer tx.Rollback() for _, s := range []struct { q string arg string }{ {"DELETE FROM key_invites WHERE thread_id = ?", threadID}, {"DELETE FROM thread_keys WHERE thread_id = ?", threadID}, {"DELETE FROM thread_members WHERE thread_id = ?", threadID}, {"DELETE FROM dm_pairs WHERE thread_id = ?", threadID}, {"DELETE FROM threads WHERE id = ?", threadID}, {"DELETE FROM users WHERE id = ?", placeholder}, } { if _, err := tx.Exec(s.q, s.arg); err != nil { return false, err } } return true, tx.Commit() } // ---- messages ----------------------------------------------------------- type Message struct { ID int64 `json:"id"` CreatedAt int64 `json:"created_at"` Blob string `json:"blob"` EditedAt int64 `json:"edited_at,omitempty"` // Non-zero marks this row a delete tombstone: the id of the message it // unsent (#581). Tombstone rows have an empty blob; created_at is the // deletion time. Old clients drop empty-blob rows, so this is additive. TombstoneOf int64 `json:"tombstone_of,omitempty"` } func (db *DB) insertMessage(threadID, blob, authorHash string) (Message, error) { now := time.Now().Unix() // A transaction so the journal row (tamper evidence, checkpoint.go) can // never be missing for a message that exists — a gap would read as // tampering to a verifier. tx, err := db.sql.Begin() if err != nil { return Message{}, err } defer tx.Rollback() res, err := tx.Exec( "INSERT INTO messages(created_at, blob, author_hash, thread_id) VALUES(?, ?, ?, ?)", now, blob, authorHash, threadID, ) if err != nil { return Message{}, err } id, err := res.LastInsertId() if err != nil { return Message{}, err } if err := appendJournalTx(tx, now, "msg", id, threadID, ciphertextHash(blob)); err != nil { return Message{}, err } if err := tx.Commit(); err != nil { return Message{}, err } // New activity reopens the thread for anyone who'd closed it (best-effort). _ = db.clearThreadHidden(threadID) return Message{ID: id, CreatedAt: now, Blob: blob}, nil } // editMessage replaces a message's ciphertext in place, but only if authorHash // matches the original sender and the message is still within the edit window. // Both checks live in the WHERE clause so they're enforced atomically by the DB, // never by client-supplied claims. Returns ok=false (no error) when the edit is // not permitted — wrong author, too old, or unknown id. func (db *DB) editMessage(id int64, blob, authorHash string, window time.Duration) (Message, bool, error) { now := time.Now() cutoff := now.Add(-window).Unix() tx, err := db.sql.Begin() if err != nil { return Message{}, false, err } defer tx.Rollback() res, err := tx.Exec( "UPDATE messages SET blob = ?, edited_at = ? WHERE id = ? AND author_hash = ? AND created_at >= ?", blob, now.Unix(), id, authorHash, cutoff, ) if err != nil { return Message{}, false, err } n, err := res.RowsAffected() if err != nil { return Message{}, false, err } if n == 0 { return Message{}, false, tx.Commit() } // Journal the edit (tamper evidence): the new ciphertext's hash — the // pre-edit hash is already the message's previous journal row. var threadID string if err := tx.QueryRow("SELECT COALESCE(thread_id, '') FROM messages WHERE id = ?", id).Scan(&threadID); err != nil { return Message{}, false, err } if err := appendJournalTx(tx, now.Unix(), "edit", id, threadID, ciphertextHash(blob)); err != nil { return Message{}, false, err } if err := tx.Commit(); err != nil { return Message{}, false, err } return Message{ID: id, Blob: blob, EditedAt: now.Unix()}, true, nil } // deleteMessage hard-deletes a message if authorHash is the author — // unsend-style, so the ciphertext is actually gone, not just hidden. // Returns whether a row was removed (false = not the author / unknown id). // Reactions live inside encrypted message events, so there is nothing else // to clean up here. // // In the same transaction it inserts a tombstone row (#581): fresh id, empty // blob, tombstone_of = the deleted id. The live "delete" broadcast reaches // only connected clients; the tombstone's fresh id lands above every // watermark, so clients that were offline replay the delete from the // ordinary `after:` catch-up with no new sync machinery. (Tombstones // themselves can't be deleted or edited: their author_hash is NULL, which // matches no caller.) func (db *DB) deleteMessage(id int64, authorHash string) (bool, error) { tx, err := db.sql.Begin() if err != nil { return false, err } defer tx.Rollback() var threadID sql.NullString err = tx.QueryRow("SELECT thread_id FROM messages WHERE id = ? AND author_hash = ?", id, authorHash).Scan(&threadID) if err == sql.ErrNoRows { return false, nil } if err != nil { return false, err } // Journal before the row is gone (tamper evidence) — stays atomic with the // DELETE below inside this transaction. if err := journalMessageDeletesTx(tx, "id = ? AND author_hash = ?", id, authorHash); err != nil { return false, err } // The DELETE re-checks author_hash and gates on RowsAffected, preserving the // old single-statement invariant: true means THIS call removed the row. A // raced duplicate (double-tap, retried frame) loses here and mints no // second tombstone. res, err := tx.Exec("DELETE FROM messages WHERE id = ? AND author_hash = ?", id, authorHash) if err != nil { return false, err } if n, err := res.RowsAffected(); err != nil || n == 0 { return false, err } // A legacy pre-thread_id row (NULL) is unreachable by thread reads, so no // client ever rendered it — skip the tombstone rather than insert one no // catch-up query could deliver. if threadID.Valid { now := time.Now().Unix() res, err := tx.Exec( "INSERT INTO messages(created_at, blob, thread_id, tombstone_of) VALUES(?, '', ?, ?)", now, threadID.String, id, ) if err != nil { return false, err } // The tombstone is a real message row (clients replay it from catch-up), // so it gets a journal row like any other insert — an unexplained row // would read as tampering to a verifier. tombID, err := res.LastInsertId() if err != nil { return false, err } if err := appendJournalTx(tx, now, "msg", tombID, threadID.String, ciphertextHash("")); err != nil { return false, err } } return true, tx.Commit() } // messagesBefore returns the `limit` messages immediately older than `before` // (id < before), ascending — used to paginate history upward. func (db *DB) messagesBefore(threadID string, before int64, limit int) ([]Message, error) { rows, err := db.sql.Query( "SELECT id, created_at, blob, COALESCE(edited_at, 0), COALESCE(tombstone_of, 0) FROM (SELECT id, created_at, blob, edited_at, tombstone_of FROM messages WHERE thread_id = ? AND id < ? ORDER BY id DESC LIMIT ?) ORDER BY id ASC", threadID, before, limit, ) if err != nil { return nil, err } defer rows.Close() out := []Message{} for rows.Next() { var m Message if err := rows.Scan(&m.ID, &m.CreatedAt, &m.Blob, &m.EditedAt, &m.TombstoneOf); err != nil { return nil, err } out = append(out, m) } return out, rows.Err() } // messagesAfter returns messages with id > after (ascending). When after is 0, // it returns the most recent `limit` messages instead of the entire history. func (db *DB) messagesAfter(threadID string, after int64, limit int) ([]Message, error) { var rows *sql.Rows var err error if after == 0 { rows, err = db.sql.Query( "SELECT id, created_at, blob, COALESCE(edited_at, 0), COALESCE(tombstone_of, 0) FROM (SELECT id, created_at, blob, edited_at, tombstone_of FROM messages WHERE thread_id = ? ORDER BY id DESC LIMIT ?) ORDER BY id ASC", threadID, limit, ) } else { rows, err = db.sql.Query( "SELECT id, created_at, blob, COALESCE(edited_at, 0), COALESCE(tombstone_of, 0) FROM messages WHERE thread_id = ? AND id > ? ORDER BY id ASC LIMIT ?", threadID, after, limit, ) } if err != nil { return nil, err } defer rows.Close() out := []Message{} for rows.Next() { var m Message if err := rows.Scan(&m.ID, &m.CreatedAt, &m.Blob, &m.EditedAt, &m.TombstoneOf); err != nil { return nil, err } out = append(out, m) } return out, rows.Err() } // ---- blobs (encrypted attachments) -------------------------------------- // // An uploaded file is encrypted in the browser before it ever reaches here, so // the server stores only opaque bytes — no filename, type, or content. The // referencing message (also encrypted) carries the metadata. // recordBlob inserts the metadata row for a store-backed blob whose bytes the // caller has already committed to db.blobs under id. `data` is stored empty — // the row exists only for quota accounting and thread-scoped GC. threadID ties // the blob to its thread (empty → NULL, e.g. profile-avatar uploads that aren't // thread-scoped and are replaced rather than message-GC'd). func (db *DB) recordBlob(id, threadID string, size int64, sha256Hex string) error { var tid any if threadID != "" { tid = threadID } _, err := db.sql.Exec( "INSERT INTO blobs(id, data, size, thread_id, created_at, sha256) VALUES(?, X'', ?, ?, ?, ?)", id, size, tid, time.Now().Unix(), sha256Hex, ) return err } // putBlob stores small in-memory blob bytes in the store + records the row. // Kept for callers that already hold the bytes (tests, any non-streaming path); // the HTTP upload path streams through a staged write instead (see uploadBlob). func (db *DB) putBlob(id, threadID string, data []byte) error { w, err := db.blobs.create() if err != nil { return err } if _, err := w.Write(data); err != nil { w.abort() return err } if err := w.commit(id); err != nil { return err } sum := sha256.Sum256(data) if err := db.recordBlob(id, threadID, int64(len(data)), hex.EncodeToString(sum[:])); err != nil { _ = db.blobs.remove(id) return err } return nil } // legacyBlob returns the bytes of a pre-disk (column-stored) blob, or nil if the // blob is store-backed (empty column) or unknown. Only the HTTP handler's // fallback path calls this, after the store reports no such blob. func (db *DB) legacyBlob(id string) ([]byte, error) { var data []byte err := db.sql.QueryRow("SELECT data FROM blobs WHERE id = ?", id).Scan(&data) if err == sql.ErrNoRows { return nil, nil } if err != nil { return nil, err } if len(data) == 0 { return nil, nil // disk-backed row — bytes live on disk, not here } return data, nil } // blobBytesTotal is the sum of all stored blob sizes — the figure the // per-instance storage quota is checked against. func (db *DB) blobBytesTotal() (int64, error) { var n sql.NullInt64 err := db.sql.QueryRow("SELECT COALESCE(SUM(size), 0) FROM blobs").Scan(&n) return n.Int64, err } // deleteBlobs removes the given blobs *scoped to one thread*: a row only goes if // its thread_id matches, so a client reporting ids to GC when it deletes its own // message can never drop another thread's (or another member's) blobs. Stored // bytes are removed best-effort for each row actually removed. func (db *DB) deleteBlobs(threadID string, ids []string) error { for _, id := range ids { if !validBlobID(id) { continue } res, err := db.sql.Exec("DELETE FROM blobs WHERE id = ? AND thread_id = ?", id, threadID) if err != nil { return err } if n, _ := res.RowsAffected(); n > 0 { _ = db.blobs.remove(id) } } return nil } // deleteThreadBlobs unlinks + removes every blob tagged to a thread, called when // the whole thread is torn down (the messages referencing them are going too). func (db *DB) deleteThreadBlobs(threadID string) error { rows, err := db.sql.Query("SELECT id FROM blobs WHERE thread_id = ?", threadID) if err != nil { return err } var ids []string for rows.Next() { var id string if err := rows.Scan(&id); err != nil { rows.Close() return err } ids = append(ids, id) } rows.Close() if len(ids) == 0 { return nil } if _, err := db.sql.Exec("DELETE FROM blobs WHERE thread_id = ?", threadID); err != nil { return err } for _, id := range ids { _ = db.blobs.remove(id) } return nil } // ---- scheduled messages (mini-app "future messages") --------------------- // // A scheduled message is client-encrypted ciphertext the server holds until // deliver_at, then moves into the thread's message log (one transaction, so a // crash can't deliver twice) and fans out like any posted message. The server // only ever learns (thread, time) — never what is scheduled. type ScheduledMessage struct { ID string `json:"id"` ThreadID string `json:"thread_id"` DeliverAt int64 `json:"deliver_at"` Blob string `json:"blob"` AuthorID string `json:"-"` } func (db *DB) createScheduledMessage(threadID, blob, authorID string, deliverAt int64) (string, error) { id, err := randToken() if err != nil { return "", err } _, err = db.sql.Exec( "INSERT INTO scheduled_messages(id, thread_id, deliver_at, blob, author_id, created_at) VALUES(?, ?, ?, ?, ?, ?)", id, threadID, deliverAt, blob, authorID, time.Now().Unix(), ) return id, err } // scheduledMessageThread returns which thread a pending schedule belongs to, so // the cancel handler can check the caller's membership of THAT thread. A missing // row reports (found=false, nil); a real DB failure is propagated as an error so // the caller can't mistake an outage for "already cancelled" and skip the delete. func (db *DB) scheduledMessageThread(id string) (string, bool, error) { var tid string err := db.sql.QueryRow("SELECT thread_id FROM scheduled_messages WHERE id = ?", id).Scan(&tid) if err == sql.ErrNoRows { return "", false, nil } if err != nil { return "", false, err } return tid, true, nil } func (db *DB) deleteScheduledMessage(id string) error { _, err := db.sql.Exec("DELETE FROM scheduled_messages WHERE id = ?", id) return err } // listScheduledMessages returns a thread's pending schedules (soonest first) so a // client can reconcile — e.g. a fresh device matching schedule ids to app items. func (db *DB) listScheduledMessages(threadID string) ([]ScheduledMessage, error) { rows, err := db.sql.Query( "SELECT id, thread_id, deliver_at, blob FROM scheduled_messages WHERE thread_id = ? ORDER BY deliver_at ASC", threadID, ) if err != nil { return nil, err } defer rows.Close() out := []ScheduledMessage{} for rows.Next() { var sm ScheduledMessage if err := rows.Scan(&sm.ID, &sm.ThreadID, &sm.DeliverAt, &sm.Blob); err != nil { return nil, err } out = append(out, sm) } return out, rows.Err() } // countScheduledMessages caps runaway scheduling per thread (see postSchedule). func (db *DB) countScheduledMessages(threadID string) int { var n int db.sql.QueryRow("SELECT COUNT(*) FROM scheduled_messages WHERE thread_id = ?", threadID).Scan(&n) return n } // nextScheduledDeliverAt returns the earliest pending deliver_at (unix // seconds), or 0 when nothing is scheduled. Delivered schedules are deleted // (deliverScheduledMessage), so MIN over the whole table IS the next due time — // this feeds the router's due-time index so a hibernating instance is woken in // time to fire it (see getNextDue). func (db *DB) nextScheduledDeliverAt() (int64, error) { var due int64 err := db.sql.QueryRow("SELECT COALESCE(MIN(deliver_at), 0) FROM scheduled_messages").Scan(&due) return due, err } // dueScheduledMessages returns schedules whose time has come, oldest first. func (db *DB) dueScheduledMessages(now int64, limit int) ([]ScheduledMessage, error) { rows, err := db.sql.Query( "SELECT id, thread_id, deliver_at, blob, author_id FROM scheduled_messages WHERE deliver_at <= ? ORDER BY deliver_at ASC LIMIT ?", now, limit, ) if err != nil { return nil, err } defer rows.Close() out := []ScheduledMessage{} for rows.Next() { var sm ScheduledMessage if err := rows.Scan(&sm.ID, &sm.ThreadID, &sm.DeliverAt, &sm.Blob, &sm.AuthorID); err != nil { return nil, err } out = append(out, sm) } return out, rows.Err() } // deliverScheduledMessage moves one due schedule into the message log: insert the // message and delete the schedule row in a single transaction, so a crash between // the two can't deliver the same schedule twice on restart. The WHERE-guarded // delete also makes concurrent sweeps safe: only the sweep that actually removes // the row (ok=true) broadcasts. func (db *DB) deliverScheduledMessage(sm ScheduledMessage) (Message, bool, error) { tx, err := db.sql.Begin() if err != nil { return Message{}, false, err } defer tx.Rollback() res, err := tx.Exec("DELETE FROM scheduled_messages WHERE id = ?", sm.ID) if err != nil { return Message{}, false, err } if n, _ := res.RowsAffected(); n == 0 { return Message{}, false, tx.Commit() // another sweep won the race } now := time.Now().Unix() ins, err := tx.Exec( "INSERT INTO messages(created_at, blob, author_hash, thread_id) VALUES(?, ?, ?, ?)", now, sm.Blob, sm.AuthorID, sm.ThreadID, ) if err != nil { return Message{}, false, err } id, err := ins.LastInsertId() if err != nil { return Message{}, false, err } // A delivered schedule is an ordinary message insert to the journal // (tamper evidence) — verifiers don't care how a message arrived. if err := appendJournalTx(tx, now, "msg", id, sm.ThreadID, ciphertextHash(sm.Blob)); err != nil { return Message{}, false, err } if err := tx.Commit(); err != nil { return Message{}, false, err } // New activity reopens the thread for anyone who'd closed it (best-effort), // mirroring insertMessage. _ = db.clearThreadHidden(sm.ThreadID) return Message{ID: id, CreatedAt: now, Blob: sm.Blob}, true, nil } // ---- thread state (mini-app snapshots) ------------------------------------ // getThreadState returns a slot's blob + version (ok=false when none saved yet). func (db *DB) getThreadState(threadID, slot string) (string, int64, bool, error) { var blob string var version int64 err := db.sql.QueryRow( "SELECT blob, version FROM thread_state WHERE thread_id = ? AND slot = ?", threadID, slot, ).Scan(&blob, &version) if err == sql.ErrNoRows { return "", 0, false, nil } if err != nil { return "", 0, false, err } return blob, version, true, nil } // putThreadState writes a slot iff the caller's expected version matches what's // stored (0 = create). Returns the new version, or ok=false on a version // conflict — the caller reloads, refolds, and retries. Both branches enforce the // check in the statement itself, so two racing writers can't both win. func (db *DB) putThreadState(threadID, slot, blob string, expect int64) (int64, bool, error) { now := time.Now().Unix() if expect == 0 { res, err := db.sql.Exec( "INSERT INTO thread_state(thread_id, slot, blob, version, updated_at) VALUES(?, ?, ?, 1, ?) ON CONFLICT(thread_id, slot) DO NOTHING", threadID, slot, blob, now, ) if err != nil { return 0, false, err } if n, _ := res.RowsAffected(); n == 0 { return 0, false, nil // someone created it first } return 1, true, nil } res, err := db.sql.Exec( "UPDATE thread_state SET blob = ?, version = version + 1, updated_at = ? WHERE thread_id = ? AND slot = ? AND version = ?", blob, now, threadID, slot, expect, ) if err != nil { return 0, false, err } if n, _ := res.RowsAffected(); n == 0 { return 0, false, nil // stale version — conflict } return expect + 1, true, nil } // ---- connected services (docs/connected-services.md) -------------------- // // One opaque, client-encrypted credential blob per (member, service). The server // keys/indexes it but never reads it. Keyed by user_id like instance_keys. // putConnectedService upserts a member's blob for a service. func (db *DB) putConnectedService(userID, service, blob string) error { _, err := db.sql.Exec( "INSERT INTO connected_services(user_id, service, blob, updated_at) VALUES(?, ?, ?, ?) "+ "ON CONFLICT(user_id, service) DO UPDATE SET blob = excluded.blob, updated_at = excluded.updated_at", userID, service, blob, time.Now().Unix(), ) return err } // getConnectedService returns a member's blob for a service (ok=false if none). func (db *DB) getConnectedService(userID, service string) (string, bool, error) { var blob string err := db.sql.QueryRow( "SELECT blob FROM connected_services WHERE user_id = ? AND service = ?", userID, service, ).Scan(&blob) if err == sql.ErrNoRows { return "", false, nil } if err != nil { return "", false, err } return blob, true, nil } // deleteConnectedService removes a member's connection to a service (disconnect). func (db *DB) deleteConnectedService(userID, service string) error { _, err := db.sql.Exec("DELETE FROM connected_services WHERE user_id = ? AND service = ?", userID, service) return err } // listConnectedServices returns the service names a member has connected. func (db *DB) listConnectedServices(userID string) ([]string, error) { rows, err := db.sql.Query("SELECT service FROM connected_services WHERE user_id = ? ORDER BY service", userID) if err != nil { return nil, err } defer rows.Close() var out []string for rows.Next() { var svc string if err := rows.Scan(&svc); err != nil { return nil, err } out = append(out, svc) } return out, rows.Err() } // ---- custom apps (customapps.go) ----------------------------------------- // // Admin-installed mini-app module source, served to members like the compiled-in // apps. The DB is the admin console's store and shadows a same-named file in // LCHAT_APPS_DIR — the console is authoritative; the disk dir stays a dev tool. // CustomApp is one installed app's metadata for the admin list (the source is // read separately — it can be large and the list never needs it). type CustomApp struct { Name string `json:"name"` Version int `json:"version"` Size int `json:"size"` UpdatedAt int64 `json:"updated_at"` // OnDisk marks an LCHAT_APPS_DIR sideload in the admin list — never a DB // row; the console shows it read-only (see listCustomAppsHandler). OnDisk bool `json:"on_disk,omitempty"` } // putCustomApp installs an app or replaces its source, bumping version on update // so the admin list shows that a revision landed. func (db *DB) putCustomApp(name, source string) error { now := time.Now().Unix() _, err := db.sql.Exec( "INSERT INTO custom_apps(name, source, enabled, version, created_at, updated_at) VALUES(?, ?, 1, 1, ?, ?) "+ "ON CONFLICT(name) DO UPDATE SET source = excluded.source, enabled = 1, "+ "version = custom_apps.version + 1, updated_at = excluded.updated_at", name, source, now, now, ) return err } // getCustomApp returns an installed, enabled app's source (ok=false if none). func (db *DB) getCustomApp(name string) (string, bool, error) { var source string err := db.sql.QueryRow( "SELECT source FROM custom_apps WHERE name = ? AND enabled = 1", name, ).Scan(&source) if err == sql.ErrNoRows { return "", false, nil } if err != nil { return "", false, err } return source, true, nil } // listCustomApps returns the installed apps' metadata, sorted by name. func (db *DB) listCustomApps() ([]CustomApp, error) { rows, err := db.sql.Query( "SELECT name, version, length(source), updated_at FROM custom_apps ORDER BY name") if err != nil { return nil, err } defer rows.Close() out := []CustomApp{} for rows.Next() { var a CustomApp if err := rows.Scan(&a.Name, &a.Version, &a.Size, &a.UpdatedAt); err != nil { return nil, err } out = append(out, a) } return out, rows.Err() } // deleteCustomApp uninstalls an app; reports whether it existed. func (db *DB) deleteCustomApp(name string) (bool, error) { res, err := db.sql.Exec("DELETE FROM custom_apps WHERE name = ?", name) if err != nil { return false, err } n, _ := res.RowsAffected() return n > 0, nil } // ---- member apps (memberapps.go) ----------------------------------------- // // Member-built apps: one instance-key-encrypted { source, manifest } blob per // app, keyed by a random id. The server can't open the blob; author_id is the // deploying member (bots deploy on a requester's behalf under their own id). // MemberApp is one deployed member app as the member API returns it — the blob // rides along because only clients can decrypt it; there is no metadata to // list separately. type MemberApp struct { ID string `json:"id"` Blob string `json:"blob,omitempty"` AuthorID string `json:"author_id"` OwnerID string `json:"owner_id"` // who it belongs to ("" for pre-ownership apps) Version int `json:"version"` UpdatedAt int64 `json:"updated_at"` } // maxMemberAppVersions bounds the stored history per app — enough to revert a // bad change, not so many that big blobs pile up unbounded. const maxMemberAppVersions = 25 // putMemberApp deploys or updates a member app and returns the resulting // version. A fresh insert is author-owned (owner = author); an update keeps the // original author_id AND owner_id (ownership never transfers via an overwrite) // and bumps version. Each deploy records the blob in member_app_versions (the // history the member can restore). basedOn > 0 marks a revert — the version // this deploy's code came from. func (db *DB) putMemberApp(id, blob, authorID string, basedOn int) (int, error) { tx, err := db.sql.Begin() if err != nil { return 0, err } defer tx.Rollback() now := time.Now().Unix() if _, err := tx.Exec( "INSERT INTO member_apps(id, blob, author_id, owner_id, version, created_at, updated_at) VALUES(?, ?, ?, ?, 1, ?, ?) "+ "ON CONFLICT(id) DO UPDATE SET blob = excluded.blob, "+ "version = member_apps.version + 1, updated_at = excluded.updated_at", id, blob, authorID, authorID, now, now, ); err != nil { return 0, err } var version int if err := tx.QueryRow("SELECT version FROM member_apps WHERE id = ?", id).Scan(&version); err != nil { return 0, err } if err := recordMemberAppVersion(tx, id, version, blob, authorID, now, basedOn); err != nil { return 0, err } return version, tx.Commit() } // recordMemberAppVersion writes one version row (basedOn > 0 records the version // a revert came from) and prunes the app's history to the most recent // maxMemberAppVersions. Runs inside the deploy's transaction. func recordMemberAppVersion(tx *sql.Tx, appID string, version int, blob, authorID string, now int64, basedOn int) error { var based any if basedOn > 0 { based = basedOn } if _, err := tx.Exec( "INSERT OR REPLACE INTO member_app_versions(app_id, version, blob, author_id, created_at, based_on) VALUES(?, ?, ?, ?, ?, ?)", appID, version, blob, authorID, now, based, ); err != nil { return err } _, err := tx.Exec("DELETE FROM member_app_versions WHERE app_id = ? AND version <= ?", appID, version-maxMemberAppVersions) return err } // getMemberApp returns one member app (ok=false if none). func (db *DB) getMemberApp(id string) (MemberApp, bool, error) { var a MemberApp err := db.sql.QueryRow( "SELECT id, blob, author_id, COALESCE(owner_id, ''), version, updated_at FROM member_apps WHERE id = ?", id, ).Scan(&a.ID, &a.Blob, &a.AuthorID, &a.OwnerID, &a.Version, &a.UpdatedAt) if err == sql.ErrNoRows { return MemberApp{}, false, nil } if err != nil { return MemberApp{}, false, err } return a, true, nil } // listMemberApps returns every deployed member app, blobs included, oldest // first so client registration order is stable across devices. func (db *DB) listMemberApps() ([]MemberApp, error) { rows, err := db.sql.Query( "SELECT id, blob, author_id, COALESCE(owner_id, ''), version, updated_at FROM member_apps ORDER BY created_at, id") if err != nil { return nil, err } defer rows.Close() out := []MemberApp{} for rows.Next() { var a MemberApp if err := rows.Scan(&a.ID, &a.Blob, &a.AuthorID, &a.OwnerID, &a.Version, &a.UpdatedAt); err != nil { return nil, err } out = append(out, a) } return out, rows.Err() } // ErrMemberAppLimit / ErrMemberAppAuthorLimit are the sentinel results of an // over-cap createMemberApp — distinct so the handler can word each. var ( ErrMemberAppLimit = errors.New("member app limit reached") ErrMemberAppAuthorLimit = errors.New("per-author member app limit reached") ) // createMemberApp inserts a NEW member app, enforcing both the global and the // per-owner caps INSIDE one transaction so concurrent deploys can't race past // them (the count-then-insert was a TOCTOU hole otherwise). Returns // ErrMemberAppLimit / ErrMemberAppAuthorLimit when a cap is hit. The cap is // per-OWNER, so a member's apps count against them whether they deployed one // themselves or a bot built it for them (rather than piling every member's // bot-built apps onto the bot). Updates to an existing app go through // putMemberApp, which is uncapped by design. func (db *DB) createMemberApp(id, blob, authorID, ownerID string, globalCap, ownerCap int) error { tx, err := db.sql.Begin() if err != nil { return err } defer tx.Rollback() var total, mine int if err := tx.QueryRow("SELECT COUNT(*) FROM member_apps").Scan(&total); err != nil { return err } if total >= globalCap { return ErrMemberAppLimit } if err := tx.QueryRow("SELECT COUNT(*) FROM member_apps WHERE owner_id = ?", ownerID).Scan(&mine); err != nil { return err } if mine >= ownerCap { return ErrMemberAppAuthorLimit } now := time.Now().Unix() if _, err := tx.Exec( "INSERT INTO member_apps(id, blob, author_id, owner_id, version, created_at, updated_at) VALUES(?, ?, ?, ?, 1, ?, ?)", id, blob, authorID, ownerID, now, now, ); err != nil { return err } if err := recordMemberAppVersion(tx, id, 1, blob, authorID, now, 0); err != nil { return err } return tx.Commit() } // countMemberApps returns the total deployed member apps (for tests + the // cap; the enforcing count lives inside createMemberApp's transaction). func (db *DB) countMemberApps() (int, error) { var n int err := db.sql.QueryRow("SELECT COUNT(*) FROM member_apps").Scan(&n) return n, err } // deleteMemberApp removes a member app and its version history; reports whether // it existed. func (db *DB) deleteMemberApp(id string) (bool, error) { res, err := db.sql.Exec("DELETE FROM member_apps WHERE id = ?", id) if err != nil { return false, err } // Reclaim the history too (best-effort; the app row is the source of truth). _, _ = db.sql.Exec("DELETE FROM member_app_versions WHERE app_id = ?", id) n, _ := res.RowsAffected() return n > 0, nil } // MemberAppVersion is one entry in an app's history, blob included (only clients // can open it). Newest first. BasedOn is the version a revert came from (0 = not // a revert). type MemberAppVersion struct { Version int `json:"version"` Blob string `json:"blob"` AuthorID string `json:"author_id"` CreatedAt int64 `json:"created_at"` BasedOn int `json:"based_on,omitempty"` } // backfillMemberAppVersions records each app's current version into the history // for apps deployed before version tracking existed. Idempotent (INSERT OR IGNORE // skips any already recorded); pre-feature intermediate versions are gone, but // the current one is recovered from member_apps so History is never empty. func (db *DB) backfillMemberAppVersions() error { _, err := db.sql.Exec( "INSERT OR IGNORE INTO member_app_versions(app_id, version, blob, author_id, created_at) " + "SELECT id, version, blob, author_id, updated_at FROM member_apps") return err } // listMemberAppVersions returns an app's stored versions, newest first. func (db *DB) listMemberAppVersions(appID string) ([]MemberAppVersion, error) { rows, err := db.sql.Query( "SELECT version, blob, author_id, created_at, COALESCE(based_on, 0) FROM member_app_versions WHERE app_id = ? ORDER BY version DESC", appID) if err != nil { return nil, err } defer rows.Close() out := []MemberAppVersion{} for rows.Next() { var v MemberAppVersion if err := rows.Scan(&v.Version, &v.Blob, &v.AuthorID, &v.CreatedAt, &v.BasedOn); err != nil { return nil, err } out = append(out, v) } return out, rows.Err() } // ---- member bots --------------------------------------------------------- // // The charter store mirrors member_apps (same trust model, same ownership // gate) minus version history: a charter is small and editable in // conversation, so "ask the bot to change it back" is the history. // MemberBot is one stored charter as the member API returns it — the blob // rides along because only clients (and the sidecar) can decrypt it. type MemberBot struct { ID string `json:"id"` Blob string `json:"blob,omitempty"` AuthorID string `json:"author_id"` OwnerID string `json:"owner_id"` UpdatedAt int64 `json:"updated_at"` } // ErrMemberBotLimit / ErrMemberBotOwnerLimit are the sentinel results of an // over-cap createMemberBot — distinct so the handler can word each. var ( ErrMemberBotLimit = errors.New("member bot limit reached") ErrMemberBotOwnerLimit = errors.New("per-owner member bot limit reached") ) // createMemberBot inserts a NEW charter, enforcing the global and per-owner // caps inside one transaction (the same TOCTOU-proof shape as createMemberApp). func (db *DB) createMemberBot(id, blob, authorID, ownerID string, globalCap, ownerCap int) error { tx, err := db.sql.Begin() if err != nil { return err } defer tx.Rollback() var total, mine int if err := tx.QueryRow("SELECT COUNT(*) FROM member_bots").Scan(&total); err != nil { return err } if total >= globalCap { return ErrMemberBotLimit } if err := tx.QueryRow("SELECT COUNT(*) FROM member_bots WHERE owner_id = ?", ownerID).Scan(&mine); err != nil { return err } if mine >= ownerCap { return ErrMemberBotOwnerLimit } now := time.Now().Unix() if _, err := tx.Exec( "INSERT INTO member_bots(id, blob, author_id, owner_id, created_at, updated_at) VALUES(?, ?, ?, ?, ?, ?)", id, blob, authorID, ownerID, now, now, ); err != nil { return err } return tx.Commit() } // putMemberBot overwrites an existing charter's blob. Author and owner stay // as created (ownership never transfers via an overwrite). func (db *DB) putMemberBot(id, blob string) error { _, err := db.sql.Exec( "UPDATE member_bots SET blob = ?, updated_at = ? WHERE id = ?", blob, time.Now().Unix(), id) return err } // getMemberBot returns one stored charter (ok=false if none). func (db *DB) getMemberBot(id string) (MemberBot, bool, error) { var b MemberBot err := db.sql.QueryRow( "SELECT id, blob, author_id, COALESCE(owner_id, ''), updated_at FROM member_bots WHERE id = ?", id, ).Scan(&b.ID, &b.Blob, &b.AuthorID, &b.OwnerID, &b.UpdatedAt) if err == sql.ErrNoRows { return MemberBot{}, false, nil } if err != nil { return MemberBot{}, false, err } return b, true, nil } // listMemberBots returns every stored charter, blobs included, oldest first. func (db *DB) listMemberBots() ([]MemberBot, error) { rows, err := db.sql.Query( "SELECT id, blob, author_id, COALESCE(owner_id, ''), updated_at FROM member_bots ORDER BY created_at, id") if err != nil { return nil, err } defer rows.Close() out := []MemberBot{} for rows.Next() { var b MemberBot if err := rows.Scan(&b.ID, &b.Blob, &b.AuthorID, &b.OwnerID, &b.UpdatedAt); err != nil { return nil, err } out = append(out, b) } return out, rows.Err() } // deleteMemberBot removes a charter; reports whether it existed. func (db *DB) deleteMemberBot(id string) (bool, error) { res, err := db.sql.Exec("DELETE FROM member_bots WHERE id = ?", id) if err != nil { return false, err } n, _ := res.RowsAffected() return n > 0, nil } // ---- incoming webhooks (hooks.go, docs/webhooks.md) ---------------------- // Webhook is one incoming-webhook row. The server can open neither blob: def // is sealed to the hooks sidecar's service identity, meta to the instance key. type Webhook struct { ID string `json:"id"` UserID string `json:"user_id"` DefBlob string `json:"def_blob,omitempty"` MetaBlob string `json:"meta_blob,omitempty"` DeployedAt int64 `json:"deployed_at"` CreatedAt int64 `json:"created_at"` } func (db *DB) createWebhook(id, userID string) error { _, err := db.sql.Exec( "INSERT INTO webhooks(id, user_id, created_at) VALUES(?, ?, ?)", id, userID, time.Now().Unix()) return err } // setWebhookBlobs stores both encrypted blobs (the create flow's second step — // the definition can only be sealed once the create response has minted the // webhook's device token). func (db *DB) setWebhookBlobs(id, defBlob, metaBlob string) (bool, error) { res, err := db.sql.Exec("UPDATE webhooks SET def_blob = ?, meta_blob = ? WHERE id = ?", defBlob, metaBlob, id) if err != nil { return false, err } n, _ := res.RowsAffected() return n > 0, nil } func (db *DB) getWebhook(id string) (Webhook, bool, error) { var h Webhook err := db.sql.QueryRow( "SELECT id, user_id, def_blob, meta_blob, deployed_at, created_at FROM webhooks WHERE id = ?", id, ).Scan(&h.ID, &h.UserID, &h.DefBlob, &h.MetaBlob, &h.DeployedAt, &h.CreatedAt) if err == sql.ErrNoRows { return Webhook{}, false, nil } if err != nil { return Webhook{}, false, err } return h, true, nil } // listWebhooks returns every webhook, oldest first, meta blobs included (the // admin app's list view; def blobs stay out — the console never needs them). func (db *DB) listWebhooks() ([]Webhook, error) { rows, err := db.sql.Query( "SELECT id, user_id, meta_blob, deployed_at, created_at FROM webhooks ORDER BY created_at, id") if err != nil { return nil, err } defer rows.Close() out := []Webhook{} for rows.Next() { var h Webhook if err := rows.Scan(&h.ID, &h.UserID, &h.MetaBlob, &h.DeployedAt, &h.CreatedAt); err != nil { return nil, err } out = append(out, h) } return out, rows.Err() } // listWebhookDefs returns the rows the hooks sidecar polls: only webhooks whose // definition has landed (the create flow's second step may still be in flight). func (db *DB) listWebhookDefs() ([]Webhook, error) { rows, err := db.sql.Query( "SELECT id, user_id, def_blob, deployed_at, created_at FROM webhooks WHERE def_blob != '' ORDER BY created_at, id") if err != nil { return nil, err } defer rows.Close() out := []Webhook{} for rows.Next() { var h Webhook if err := rows.Scan(&h.ID, &h.UserID, &h.DefBlob, &h.DeployedAt, &h.CreatedAt); err != nil { return nil, err } out = append(out, h) } return out, rows.Err() } // setWebhookDeployed records the sidecar's "route is live" ack (first ack wins; // re-acks on later polls are no-ops so the timestamp stays the first go-live). func (db *DB) setWebhookDeployed(id string) error { _, err := db.sql.Exec("UPDATE webhooks SET deployed_at = ? WHERE id = ? AND deployed_at = 0", time.Now().Unix(), id) return err } func (db *DB) deleteWebhook(id string) (bool, error) { res, err := db.sql.Exec("DELETE FROM webhooks WHERE id = ?", id) if err != nil { return false, err } n, _ := res.RowsAffected() return n > 0, nil } func (db *DB) countWebhooks() (int, error) { var n int err := db.sql.QueryRow("SELECT COUNT(*) FROM webhooks").Scan(&n) return n, err } // ---- event journal + checkpoints (tamper evidence, checkpoint.go) -------- // // The journal is append-only: rows are written inside the same transaction as // the message write they describe, and nothing ever updates or deletes them. // It stores ciphertext HASHES, never ciphertext — unsend still destroys the // blob; what survives is evidence that an event happened, which is the whole // point (docs/tamper-evidence.md). // JournalRow is one appended event. Field order and encoding are pinned by // journalLeafHash (checkpoint.go) — they are wire format. type JournalRow struct { Seq int64 `json:"seq"` TS int64 `json:"ts"` Kind string `json:"kind"` MsgID int64 `json:"msg_id"` ThreadID string `json:"thread_id"` BlobHash string `json:"blob_hash"` } // ciphertextHash is the journal's commitment to a message blob: sha256 hex of // the stored ciphertext text. Hashing the base64url string as stored (rather // than decoded bytes) keeps verification a straight string comparison against // what /api/messages serves. It happens to match hashToken today but is // deliberately NOT shared with it: this is journal wire format — recorded // hashes must stay verifiable forever — and must never follow a change to how // tokens are hashed. func ciphertextHash(blob string) string { sum := sha256.Sum256([]byte(blob)) return hex.EncodeToString(sum[:]) } func appendJournalTx(tx *sql.Tx, ts int64, kind string, msgID int64, threadID, blobHash string) error { _, err := tx.Exec( "INSERT INTO event_journal(ts, kind, msg_id, thread_id, blob_hash) VALUES(?, ?, ?, ?, ?)", ts, kind, msgID, threadID, blobHash, ) return err } // journalMessageDeletesTx appends a 'delete' row for every message matching // where — called immediately before the DELETE with the same predicate, in the // same transaction, so the journal and the table can't disagree about what was // removed. Bulk removals (thread teardown, member removal) journal one row per // message on purpose: every vanished message row has exactly one explaining // journal row, so a verifier never has to guess. One set-based statement, so // purging a prolific member doesn't hold the single connection through N // round-trips. func journalMessageDeletesTx(tx *sql.Tx, where string, args ...any) error { _, err := tx.Exec( "INSERT INTO event_journal(ts, kind, msg_id, thread_id, blob_hash) "+ "SELECT ?, 'delete', id, COALESCE(thread_id, ''), '' FROM messages WHERE "+where, append([]any{time.Now().Unix()}, args...)..., ) return err } // ensureCheckpointSeed loads (or mints) the checkpoint signing seed from the // config table. In the DB on purpose: the seed must survive everything that // moves instance state — hibernation wake, VACUUM-INTO migration, escrow // restore — all of which carry only the DB. A key that silently changed on // routine ops would make every legitimate restore read as forgery. The trade // (a DB copy can sign as the instance) is accepted: the signature is // attribution, not the security boundary (docs/tamper-evidence.md). func (db *DB) ensureCheckpointSeed() (ed25519.PrivateKey, error) { enc, ok, err := db.getConfig("ckpt_seed") if err != nil { return nil, err } if ok && enc != "" { return checkpointSeedKey(enc, "config ckpt_seed") } _, priv, err := ed25519.GenerateKey(rand.Reader) if err != nil { return nil, err } if err := db.setConfig("ckpt_seed", base64.StdEncoding.EncodeToString(priv.Seed())); err != nil { return nil, err } return priv, nil } // backfillEventJournal baselines messages that predate the journal: one 'msg' // row per existing message, once (config-marked), in a single transaction. // A backfilled row commits to the ciphertext as it stood at upgrade — evidence // can't be retroactive (a pre-upgrade rewrite is invisible), but from this // baseline forward every change to old messages is accounted for. func (db *DB) backfillEventJournal() error { if v, _, _ := db.getConfig("journal_backfilled"); v == "1" { return nil } tx, err := db.sql.Begin() if err != nil { return err } defer tx.Rollback() // Buffer then insert: blob hashes are computed in Go (no sha256 in SQL), // and interleaving inserts with an open cursor on one connection is asking // for trouble. Fleet instances hold a few thousand messages at most. rows, err := tx.Query("SELECT id, created_at, COALESCE(thread_id, ''), blob FROM messages ORDER BY id") if err != nil { return err } var rs []JournalRow for rows.Next() { var r JournalRow var blob string if err := rows.Scan(&r.MsgID, &r.TS, &r.ThreadID, &blob); err != nil { rows.Close() return err } r.BlobHash = ciphertextHash(blob) rs = append(rs, r) } rows.Close() if err := rows.Err(); err != nil { return err } for _, r := range rs { if err := appendJournalTx(tx, r.TS, "msg", r.MsgID, r.ThreadID, r.BlobHash); err != nil { return err } } if err := setConfigTx(tx, "journal_backfilled", "1"); err != nil { return err } return tx.Commit() } // journalAfter returns up to limit rows with seq > after, in order — the // checkpoint sweep's read, and a verifier's replay feed. func (db *DB) journalAfter(after int64, limit int) ([]JournalRow, error) { rows, err := db.sql.Query( "SELECT seq, ts, kind, msg_id, thread_id, blob_hash FROM event_journal WHERE seq > ? ORDER BY seq LIMIT ?", after, limit, ) if err != nil { return nil, err } defer rows.Close() var out []JournalRow for rows.Next() { var r JournalRow if err := rows.Scan(&r.Seq, &r.TS, &r.Kind, &r.MsgID, &r.ThreadID, &r.BlobHash); err != nil { return nil, err } out = append(out, r) } return out, rows.Err() } // Checkpoint is one signed link of the chain as served to verifiers. type Checkpoint struct { ID int64 `json:"id"` SeqFrom int64 `json:"seq_from"` SeqTo int64 `json:"seq_to"` Root string `json:"root"` Chain string `json:"chain"` TS int64 `json:"ts"` Signed string `json:"signed"` } func (db *DB) insertCheckpoint(p ckptPayload, signed string) error { _, err := db.sql.Exec( "INSERT INTO checkpoints(seq_from, seq_to, root, chain, ts, signed) VALUES(?, ?, ?, ?, ?, ?)", p.SeqFrom, p.SeqTo, p.Root, p.Chain, p.TS, signed, ) return err } func (db *DB) latestCheckpoint() (Checkpoint, bool, error) { var c Checkpoint err := db.sql.QueryRow( "SELECT id, seq_from, seq_to, root, chain, ts, signed FROM checkpoints ORDER BY id DESC LIMIT 1", ).Scan(&c.ID, &c.SeqFrom, &c.SeqTo, &c.Root, &c.Chain, &c.TS, &c.Signed) if err == sql.ErrNoRows { return Checkpoint{}, false, nil } if err != nil { return Checkpoint{}, false, err } return c, true, nil } func (db *DB) checkpointsAfter(after int64, limit int) ([]Checkpoint, error) { rows, err := db.sql.Query( "SELECT id, seq_from, seq_to, root, chain, ts, signed FROM checkpoints WHERE id > ? ORDER BY id LIMIT ?", after, limit, ) if err != nil { return nil, err } defer rows.Close() out := []Checkpoint{} for rows.Next() { var c Checkpoint if err := rows.Scan(&c.ID, &c.SeqFrom, &c.SeqTo, &c.Root, &c.Chain, &c.TS, &c.Signed); err != nil { return nil, err } out = append(out, c) } return out, rows.Err() } // checkpointCovering returns the checkpoint whose range includes journal seq // (seq_from <= seq <= seq_to) — the one an inclusion proof for that seq builds // against. ok=false when seq is beyond the last checkpoint (not yet committed, // which the client reads as "too recent", not tampering). func (db *DB) checkpointCovering(seq int64) (Checkpoint, bool, error) { var c Checkpoint err := db.sql.QueryRow( "SELECT id, seq_from, seq_to, root, chain, ts, signed FROM checkpoints WHERE seq_from <= ? AND seq_to >= ? LIMIT 1", seq, seq, ).Scan(&c.ID, &c.SeqFrom, &c.SeqTo, &c.Root, &c.Chain, &c.TS, &c.Signed) if err == sql.ErrNoRows { return Checkpoint{}, false, nil } if err != nil { return Checkpoint{}, false, err } return c, true, nil } // journalRange returns the journal rows with seq in [from, to] in order — the // leaves of a checkpoint's Merkle tree, for building an audit path. func (db *DB) journalRange(from, to int64) ([]JournalRow, error) { rows, err := db.sql.Query( "SELECT seq, ts, kind, msg_id, thread_id, blob_hash FROM event_journal WHERE seq >= ? AND seq <= ? ORDER BY seq", from, to, ) if err != nil { return nil, err } defer rows.Close() var out []JournalRow for rows.Next() { var r JournalRow if err := rows.Scan(&r.Seq, &r.TS, &r.Kind, &r.MsgID, &r.ThreadID, &r.BlobHash); err != nil { return nil, err } out = append(out, r) } return out, rows.Err() } // journalRowsForMsg returns every journal row for a message id (its insert, // any edits, a delete) in order — a client sampling that message asks for its // proofs and matches the one whose blob_hash equals its held ciphertext. func (db *DB) journalRowsForMsg(msgID int64) ([]JournalRow, error) { rows, err := db.sql.Query( "SELECT seq, ts, kind, msg_id, thread_id, blob_hash FROM event_journal WHERE msg_id = ? ORDER BY seq", msgID, ) if err != nil { return nil, err } defer rows.Close() var out []JournalRow for rows.Next() { var r JournalRow if err := rows.Scan(&r.Seq, &r.TS, &r.Kind, &r.MsgID, &r.ThreadID, &r.BlobHash); err != nil { return nil, err } out = append(out, r) } return out, rows.Err() }