rickub
nandi/gleanpublic Fork
⑂ 1971b41
Branches
claude/modal-container-min-76527a main View all branches → View all tags →
Commits
⬇ Clone ▾
git clone https://git.rickub.com/nandi/glean.git
git clone ssh://git@rickub.com/nandi/glean.git

Host key fingerprint (ed25519): SHA256:iycHnxEyq0Q7uyVpB7JlznP0G7JrTPXLYRcAU5CSLhc — verify it before your first connect.

Download ZIP

Persist OAuth sessions and the Jetstream cursor

A SQLite wrapper over the library the storage spike proved, plus the three
tables a restart must not lose -- oauth_auth_requests, oauth_sessions,
jetstream_cursor -- matching the Go backend's schema. Rows hold JSON, as
they do there, so adding a session field is not a migration.

Cursor writes are batched: the cursor advances on every event, often
hundreds a second, and writing synchronously would make the database the
bottleneck for a firehose. The stored cursor therefore trails the live one,
which is the gap the supervisor's rewind already covers.

The DPoP key is the check worth having. A missing row is loud; a key that
round-trips almost correctly is not, and would fail every request after a
restart with nothing pointing at storage. ECDSA is randomised, so the
reloaded key is proved by signing with it and verifying against the original
public point, with a second key as a negative control.

The public half is stored rather than re-derived, since deriving it needs a
curve operation and a key that cannot be read back without one is a restart
that drops every session.

These rows are key material: a session holds the DPoP private key and the
refresh token, which together are the account.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
nandithebull committed 2026-09-21T00:14:08-07:00 Browse files
1971b41 parent: 5eef251
modified spike/nim/.gitignore +4 -0
@@ -6,3 +6,7 @@ oauth_probe
66 xrpc_probe
77 jetstream_probe
88 reconnect_probe
9+store_probe
10+*.db
11+*.db-wal
12+*.db-shm
@@ -6,3 +6,7 @@ oauth_probe
6 xrpc_probe6 xrpc_probe
7 jetstream_probe7 jetstream_probe
8 reconnect_probe8 reconnect_probe
9+store_probe
10+*.db
11+*.db-wal
12+*.db-shm
added spike/nim/atproto/cursor_store.nim +59 -0
new file mode 100644
@@ -0,0 +1,59 @@
1+## A Jetstream cursor store backed by SQLite, with batched writes.
2+##
3+## Bridges `store` and `jetstream_run`. Separate from both so the supervisor
4+## does not drag a database dependency behind it, and so an application can
5+## substitute its own persistence.
6+##
7+## Writes are batched because the cursor advances on *every* event -- often
8+## hundreds a second -- and a synchronous write per event would make the
9+## database the bottleneck for a firehose. The cost is that the stored cursor
10+## trails the live one, which is exactly the gap the supervisor's rewind
11+## exists to cover.
12+
13+import std/times
14+import ./jetstream_run
15+import ./sqlite
16+import ./store
17+
18+type DbCursorStore* = ref object of CursorStore
19+ db*: Db
20+ flushEvery*: int64 ## seconds
21+ lastFlush: int64
22+ pending: int64
23+ dirty: bool
24+
25+proc newDbCursorStore*(db: Db, flushEvery = 5'i64): DbCursorStore =
26+ result = DbCursorStore(db: db, flushEvery: flushEvery,
27+ lastFlush: getTime().toUnix)
28+ result.cursor = loadCursor(db)
29+ result.pending = result.cursor
30+
31+method load*(s: DbCursorStore): int64 {.gcsafe, raises: [].} =
32+ s.cursor
33+
34+proc flushNow*(s: DbCursorStore) =
35+ ## Write the newest cursor out. Safe to call at any time; a failure is
36+ ## swallowed, because losing a cursor update costs a replay and raising
37+ ## here would cost the subscription.
38+ if not s.dirty: return
39+ try:
40+ saveCursor(s.db, s.pending)
41+ s.cursor = s.pending
42+ s.dirty = false
43+ s.lastFlush = getTime().toUnix
44+ except CatchableError:
45+ discard
46+
47+method save*(s: DbCursorStore, cursor: int64) {.gcsafe, raises: [].} =
48+ ## Called per event. Records in memory and writes at most once per
49+ ## flushEvery seconds.
50+ if cursor <= s.pending: return
51+ s.pending = cursor
52+ s.dirty = true
53+ let now = try: getTime().toUnix except CatchableError: return
54+ if now - s.lastFlush >= s.flushEvery:
55+ s.flushNow()
56+
57+proc close*(s: DbCursorStore) =
58+ ## Final flush, so a clean shutdown does not throw away the last batch.
59+ s.flushNow()
new file mode 100644
@@ -0,0 +1,59 @@
1+## A Jetstream cursor store backed by SQLite, with batched writes.
2+##
3+## Bridges `store` and `jetstream_run`. Separate from both so the supervisor
4+## does not drag a database dependency behind it, and so an application can
5+## substitute its own persistence.
6+##
7+## Writes are batched because the cursor advances on *every* event -- often
8+## hundreds a second -- and a synchronous write per event would make the
9+## database the bottleneck for a firehose. The cost is that the stored cursor
10+## trails the live one, which is exactly the gap the supervisor's rewind
11+## exists to cover.
12+
13+import std/times
14+import ./jetstream_run
15+import ./sqlite
16+import ./store
17+
18+type DbCursorStore* = ref object of CursorStore
19+ db*: Db
20+ flushEvery*: int64 ## seconds
21+ lastFlush: int64
22+ pending: int64
23+ dirty: bool
24+
25+proc newDbCursorStore*(db: Db, flushEvery = 5'i64): DbCursorStore =
26+ result = DbCursorStore(db: db, flushEvery: flushEvery,
27+ lastFlush: getTime().toUnix)
28+ result.cursor = loadCursor(db)
29+ result.pending = result.cursor
30+
31+method load*(s: DbCursorStore): int64 {.gcsafe, raises: [].} =
32+ s.cursor
33+
34+proc flushNow*(s: DbCursorStore) =
35+ ## Write the newest cursor out. Safe to call at any time; a failure is
36+ ## swallowed, because losing a cursor update costs a replay and raising
37+ ## here would cost the subscription.
38+ if not s.dirty: return
39+ try:
40+ saveCursor(s.db, s.pending)
41+ s.cursor = s.pending
42+ s.dirty = false
43+ s.lastFlush = getTime().toUnix
44+ except CatchableError:
45+ discard
46+
47+method save*(s: DbCursorStore, cursor: int64) {.gcsafe, raises: [].} =
48+ ## Called per event. Records in memory and writes at most once per
49+ ## flushEvery seconds.
50+ if cursor <= s.pending: return
51+ s.pending = cursor
52+ s.dirty = true
53+ let now = try: getTime().toUnix except CatchableError: return
54+ if now - s.lastFlush >= s.flushEvery:
55+ s.flushNow()
56+
57+proc close*(s: DbCursorStore) =
58+ ## Final flush, so a clean shutdown does not throw away the last batch.
59+ s.flushNow()
modified spike/nim/atproto/dpop.nim +25 -0
@@ -99,6 +99,31 @@ proc accessTokenHash*(accessToken: string): string =
9999 ## exists to prevent.
100100 b64u(sha256.digest(accessToken).data)
101101
102+proc verifyEs256*(k: P256Key, signingInput: string,
103+ sig: openArray[byte]): bool =
104+ ## Verify a raw R||S signature against this key's public half.
105+ ##
106+ ## Mostly useful for proving a key survived a round trip: ECDSA is
107+ ## randomised, so a reloaded key cannot be checked by re-signing and
108+ ## comparing bytes. Signing with the reloaded key and verifying against the
109+ ## original public point does establish the private scalar came back intact.
110+ if sig.len != 64: return false
111+ var q: array[65, byte]
112+ q[0] = 0x04
113+ for i in 0 ..< 32:
114+ q[1 + i] = k.pubX[i]
115+ q[33 + i] = k.pubY[i]
116+
117+ var pk: EcPublicKey
118+ pk.curve = EC_secp256r1.cint
119+ pk.q = cast[ptr byte](addr q[0])
120+ pk.qlen = 65
121+
122+ let digest = sha256.digest(signingInput).data
123+ ecdsaVrfyRawGetDefault()(
124+ addr ecPrimeI31, unsafeAddr digest[0], digest.len.uint,
125+ addr pk, unsafeAddr sig[0], sig.len.uint) == 1
126+
102127 proc dpopProof*(k: P256Key, rng: var HmacDrbgContext, htm, htu: string,
103128 nonce = "", accessToken = "", issuer = ""): string =
104129 ## A DPoP proof JWT (RFC 9449).
@@ -99,6 +99,31 @@ proc accessTokenHash*(accessToken: string): string =
99 ## exists to prevent.99 ## exists to prevent.
100 b64u(sha256.digest(accessToken).data)100 b64u(sha256.digest(accessToken).data)
101 101
102+proc verifyEs256*(k: P256Key, signingInput: string,
103+ sig: openArray[byte]): bool =
104+ ## Verify a raw R||S signature against this key's public half.
105+ ##
106+ ## Mostly useful for proving a key survived a round trip: ECDSA is
107+ ## randomised, so a reloaded key cannot be checked by re-signing and
108+ ## comparing bytes. Signing with the reloaded key and verifying against the
109+ ## original public point does establish the private scalar came back intact.
110+ if sig.len != 64: return false
111+ var q: array[65, byte]
112+ q[0] = 0x04
113+ for i in 0 ..< 32:
114+ q[1 + i] = k.pubX[i]
115+ q[33 + i] = k.pubY[i]
116+
117+ var pk: EcPublicKey
118+ pk.curve = EC_secp256r1.cint
119+ pk.q = cast[ptr byte](addr q[0])
120+ pk.qlen = 65
121+
122+ let digest = sha256.digest(signingInput).data
123+ ecdsaVrfyRawGetDefault()(
124+ addr ecPrimeI31, unsafeAddr digest[0], digest.len.uint,
125+ addr pk, unsafeAddr sig[0], sig.len.uint) == 1
126+
102 proc dpopProof*(k: P256Key, rng: var HmacDrbgContext, htm, htu: string,127 proc dpopProof*(k: P256Key, rng: var HmacDrbgContext, htm, htu: string,
103 nonce = "", accessToken = "", issuer = ""): string =128 nonce = "", accessToken = "", issuer = ""): string =
104 ## A DPoP proof JWT (RFC 9449).129 ## A DPoP proof JWT (RFC 9449).
added spike/nim/atproto/sqlite.nim +165 -0
new file mode 100644
@@ -0,0 +1,165 @@
1+## A small SQLite wrapper, with sqlite-vec compiled in.
2+##
3+## Deliberately thin: enough to prepare, bind and step, not an ORM. The
4+## sqlite-vec extension is linked rather than loaded at runtime -- see
5+## nim.cfg, where the SQLITE_CORE define lives, and note that getting that
6+## wrong segfaults rather than erroring.
7+
8+{.compile: "../vendor/sqlite-vec.c".}
9+
10+import std/options
11+
12+type
13+ SqliteError* = object of CatchableError
14+
15+ Db* = object
16+ handle*: pointer
17+
18+ Stmt* = object
19+ handle*: pointer
20+ db*: pointer
21+
22+const
23+ SQLITE_OK* = 0
24+ SQLITE_ROW* = 100
25+ SQLITE_DONE* = 101
26+
27+{.push importc, cdecl.}
28+proc sqlite3_open(filename: cstring, db: var pointer): cint
29+proc sqlite3_close(db: pointer): cint
30+proc sqlite3_errmsg(db: pointer): cstring
31+proc sqlite3_exec(db: pointer, sql: cstring, cb, arg: pointer,
32+ errmsg: ptr cstring): cint
33+proc sqlite3_prepare_v2(db: pointer, sql: cstring, nByte: cint,
34+ stmt: var pointer, tail: ptr cstring): cint
35+proc sqlite3_step(s: pointer): cint
36+proc sqlite3_finalize(s: pointer): cint
37+proc sqlite3_reset(s: pointer): cint
38+proc sqlite3_column_text(s: pointer, col: cint): cstring
39+proc sqlite3_column_int64(s: pointer, col: cint): int64
40+proc sqlite3_column_double(s: pointer, col: cint): cdouble
41+proc sqlite3_column_count(s: pointer): cint
42+proc sqlite3_bind_text(s: pointer, col: cint, data: cstring, n: cint,
43+ destructor: pointer): cint
44+proc sqlite3_bind_int64(s: pointer, col: cint, v: int64): cint
45+proc sqlite3_bind_blob(s: pointer, col: cint, data: pointer, n: cint,
46+ destructor: pointer): cint
47+proc sqlite3_libversion(): cstring
48+{.pop.}
49+
50+proc sqlite3_vec_init(db: pointer, errMsg: ptr cstring,
51+ api: pointer): cint {.importc, cdecl.}
52+
53+const SQLITE_TRANSIENT = cast[pointer](-1)
54+
55+proc libVersion*(): string = $sqlite3_libversion()
56+
57+proc check(db: pointer, rc: cint, what: string) =
58+ if rc notin [SQLITE_OK.cint, SQLITE_ROW.cint, SQLITE_DONE.cint]:
59+ raise newException(SqliteError, what & ": " & $sqlite3_errmsg(db))
60+
61+proc open*(path: string, withVec = true): Db =
62+ ## Open (or create) a database.
63+ ##
64+ ## WAL is set here rather than left to callers: it is what makes a reader
65+ ## and the writer coexist, and a store that forgets it fails only under
66+ ## concurrency, which is the worst time to find out.
67+ var handle: pointer
68+ if sqlite3_open(path.cstring, handle) != SQLITE_OK:
69+ let msg = $sqlite3_errmsg(handle)
70+ discard sqlite3_close(handle)
71+ raise newException(SqliteError, "opening " & path & ": " & msg)
72+ result = Db(handle: handle)
73+
74+ if withVec:
75+ let rc = sqlite3_vec_init(handle, nil, nil)
76+ if rc != SQLITE_OK:
77+ raise newException(SqliteError, "registering sqlite-vec failed")
78+
79+ discard sqlite3_exec(handle, "PRAGMA journal_mode=WAL", nil, nil, nil)
80+ discard sqlite3_exec(handle, "PRAGMA synchronous=NORMAL", nil, nil, nil)
81+ discard sqlite3_exec(handle, "PRAGMA foreign_keys=ON", nil, nil, nil)
82+ discard sqlite3_exec(handle, "PRAGMA busy_timeout=5000", nil, nil, nil)
83+
84+proc close*(db: var Db) =
85+ if db.handle != nil:
86+ discard sqlite3_close(db.handle)
87+ db.handle = nil
88+
89+proc exec*(db: Db, sql: string) =
90+ var err: cstring
91+ let rc = sqlite3_exec(db.handle, sql.cstring, nil, nil, addr err)
92+ if rc != SQLITE_OK:
93+ raise newException(SqliteError, "exec failed: " & $sqlite3_errmsg(db.handle))
94+
95+type Param* = object
96+ case isNull*: bool
97+ of true: discard
98+ of false:
99+ case kind*: range[0 .. 2]
100+ of 0: s*: string
101+ of 1: i*: int64
102+ of 2: b*: seq[byte]
103+
104+proc p*(s: string): Param = Param(isNull: false, kind: 0, s: s)
105+proc p*(i: int64): Param = Param(isNull: false, kind: 1, i: i)
106+proc p*(b: seq[byte]): Param = Param(isNull: false, kind: 2, b: b)
107+
108+proc prepare(db: Db, sql: string, params: openArray[Param]): pointer =
109+ var stmt: pointer
110+ check(db.handle, sqlite3_prepare_v2(db.handle, sql.cstring, -1, stmt, nil),
111+ "prepare")
112+ for i, prm in params:
113+ let col = (i + 1).cint
114+ let rc =
115+ if prm.isNull: SQLITE_OK.cint
116+ else:
117+ case prm.kind
118+ of 0: sqlite3_bind_text(stmt, col, prm.s.cstring, prm.s.len.cint,
119+ SQLITE_TRANSIENT)
120+ of 1: sqlite3_bind_int64(stmt, col, prm.i)
121+ of 2:
122+ if prm.b.len == 0:
123+ sqlite3_bind_blob(stmt, col, nil, 0, SQLITE_TRANSIENT)
124+ else:
125+ sqlite3_bind_blob(stmt, col, prm.b[0].unsafeAddr, prm.b.len.cint,
126+ SQLITE_TRANSIENT)
127+ if rc != SQLITE_OK:
128+ discard sqlite3_finalize(stmt)
129+ raise newException(SqliteError, "binding parameter " & $col)
130+ stmt
131+
132+proc run*(db: Db, sql: string, params: varargs[Param]) =
133+ ## Execute a statement that returns no rows.
134+ let stmt = prepare(db, sql, params)
135+ defer: discard sqlite3_finalize(stmt)
136+ check(db.handle, sqlite3_step(stmt), "step")
137+
138+proc queryText*(db: Db, sql: string, params: varargs[Param]): Option[string] =
139+ ## First column of the first row, as text.
140+ let stmt = prepare(db, sql, params)
141+ defer: discard sqlite3_finalize(stmt)
142+ if sqlite3_step(stmt) != SQLITE_ROW:
143+ return none(string)
144+ let raw = sqlite3_column_text(stmt, 0)
145+ if raw == nil: none(string) else: some($raw)
146+
147+proc queryInt*(db: Db, sql: string, params: varargs[Param]): Option[int64] =
148+ let stmt = prepare(db, sql, params)
149+ defer: discard sqlite3_finalize(stmt)
150+ if sqlite3_step(stmt) != SQLITE_ROW:
151+ return none(int64)
152+ some(sqlite3_column_int64(stmt, 0))
153+
154+iterator rows*(db: Db, sql: string, params: varargs[Param]): seq[string] =
155+ ## Every row, all columns as text. Fine for the small result sets a session
156+ ## store deals in; not for scanning a table of articles.
157+ let stmt = prepare(db, sql, @params)
158+ defer: discard sqlite3_finalize(stmt)
159+ let n = sqlite3_column_count(stmt)
160+ while sqlite3_step(stmt) == SQLITE_ROW:
161+ var row = newSeq[string](n)
162+ for i in 0 ..< n:
163+ let raw = sqlite3_column_text(stmt, i.cint)
164+ row[i] = if raw == nil: "" else: $raw
165+ yield row
new file mode 100644
@@ -0,0 +1,165 @@
1+## A small SQLite wrapper, with sqlite-vec compiled in.
2+##
3+## Deliberately thin: enough to prepare, bind and step, not an ORM. The
4+## sqlite-vec extension is linked rather than loaded at runtime -- see
5+## nim.cfg, where the SQLITE_CORE define lives, and note that getting that
6+## wrong segfaults rather than erroring.
7+
8+{.compile: "../vendor/sqlite-vec.c".}
9+
10+import std/options
11+
12+type
13+ SqliteError* = object of CatchableError
14+
15+ Db* = object
16+ handle*: pointer
17+
18+ Stmt* = object
19+ handle*: pointer
20+ db*: pointer
21+
22+const
23+ SQLITE_OK* = 0
24+ SQLITE_ROW* = 100
25+ SQLITE_DONE* = 101
26+
27+{.push importc, cdecl.}
28+proc sqlite3_open(filename: cstring, db: var pointer): cint
29+proc sqlite3_close(db: pointer): cint
30+proc sqlite3_errmsg(db: pointer): cstring
31+proc sqlite3_exec(db: pointer, sql: cstring, cb, arg: pointer,
32+ errmsg: ptr cstring): cint
33+proc sqlite3_prepare_v2(db: pointer, sql: cstring, nByte: cint,
34+ stmt: var pointer, tail: ptr cstring): cint
35+proc sqlite3_step(s: pointer): cint
36+proc sqlite3_finalize(s: pointer): cint
37+proc sqlite3_reset(s: pointer): cint
38+proc sqlite3_column_text(s: pointer, col: cint): cstring
39+proc sqlite3_column_int64(s: pointer, col: cint): int64
40+proc sqlite3_column_double(s: pointer, col: cint): cdouble
41+proc sqlite3_column_count(s: pointer): cint
42+proc sqlite3_bind_text(s: pointer, col: cint, data: cstring, n: cint,
43+ destructor: pointer): cint
44+proc sqlite3_bind_int64(s: pointer, col: cint, v: int64): cint
45+proc sqlite3_bind_blob(s: pointer, col: cint, data: pointer, n: cint,
46+ destructor: pointer): cint
47+proc sqlite3_libversion(): cstring
48+{.pop.}
49+
50+proc sqlite3_vec_init(db: pointer, errMsg: ptr cstring,
51+ api: pointer): cint {.importc, cdecl.}
52+
53+const SQLITE_TRANSIENT = cast[pointer](-1)
54+
55+proc libVersion*(): string = $sqlite3_libversion()
56+
57+proc check(db: pointer, rc: cint, what: string) =
58+ if rc notin [SQLITE_OK.cint, SQLITE_ROW.cint, SQLITE_DONE.cint]:
59+ raise newException(SqliteError, what & ": " & $sqlite3_errmsg(db))
60+
61+proc open*(path: string, withVec = true): Db =
62+ ## Open (or create) a database.
63+ ##
64+ ## WAL is set here rather than left to callers: it is what makes a reader
65+ ## and the writer coexist, and a store that forgets it fails only under
66+ ## concurrency, which is the worst time to find out.
67+ var handle: pointer
68+ if sqlite3_open(path.cstring, handle) != SQLITE_OK:
69+ let msg = $sqlite3_errmsg(handle)
70+ discard sqlite3_close(handle)
71+ raise newException(SqliteError, "opening " & path & ": " & msg)
72+ result = Db(handle: handle)
73+
74+ if withVec:
75+ let rc = sqlite3_vec_init(handle, nil, nil)
76+ if rc != SQLITE_OK:
77+ raise newException(SqliteError, "registering sqlite-vec failed")
78+
79+ discard sqlite3_exec(handle, "PRAGMA journal_mode=WAL", nil, nil, nil)
80+ discard sqlite3_exec(handle, "PRAGMA synchronous=NORMAL", nil, nil, nil)
81+ discard sqlite3_exec(handle, "PRAGMA foreign_keys=ON", nil, nil, nil)
82+ discard sqlite3_exec(handle, "PRAGMA busy_timeout=5000", nil, nil, nil)
83+
84+proc close*(db: var Db) =
85+ if db.handle != nil:
86+ discard sqlite3_close(db.handle)
87+ db.handle = nil
88+
89+proc exec*(db: Db, sql: string) =
90+ var err: cstring
91+ let rc = sqlite3_exec(db.handle, sql.cstring, nil, nil, addr err)
92+ if rc != SQLITE_OK:
93+ raise newException(SqliteError, "exec failed: " & $sqlite3_errmsg(db.handle))
94+
95+type Param* = object
96+ case isNull*: bool
97+ of true: discard
98+ of false:
99+ case kind*: range[0 .. 2]
100+ of 0: s*: string
101+ of 1: i*: int64
102+ of 2: b*: seq[byte]
103+
104+proc p*(s: string): Param = Param(isNull: false, kind: 0, s: s)
105+proc p*(i: int64): Param = Param(isNull: false, kind: 1, i: i)
106+proc p*(b: seq[byte]): Param = Param(isNull: false, kind: 2, b: b)
107+
108+proc prepare(db: Db, sql: string, params: openArray[Param]): pointer =
109+ var stmt: pointer
110+ check(db.handle, sqlite3_prepare_v2(db.handle, sql.cstring, -1, stmt, nil),
111+ "prepare")
112+ for i, prm in params:
113+ let col = (i + 1).cint
114+ let rc =
115+ if prm.isNull: SQLITE_OK.cint
116+ else:
117+ case prm.kind
118+ of 0: sqlite3_bind_text(stmt, col, prm.s.cstring, prm.s.len.cint,
119+ SQLITE_TRANSIENT)
120+ of 1: sqlite3_bind_int64(stmt, col, prm.i)
121+ of 2:
122+ if prm.b.len == 0:
123+ sqlite3_bind_blob(stmt, col, nil, 0, SQLITE_TRANSIENT)
124+ else:
125+ sqlite3_bind_blob(stmt, col, prm.b[0].unsafeAddr, prm.b.len.cint,
126+ SQLITE_TRANSIENT)
127+ if rc != SQLITE_OK:
128+ discard sqlite3_finalize(stmt)
129+ raise newException(SqliteError, "binding parameter " & $col)
130+ stmt
131+
132+proc run*(db: Db, sql: string, params: varargs[Param]) =
133+ ## Execute a statement that returns no rows.
134+ let stmt = prepare(db, sql, params)
135+ defer: discard sqlite3_finalize(stmt)
136+ check(db.handle, sqlite3_step(stmt), "step")
137+
138+proc queryText*(db: Db, sql: string, params: varargs[Param]): Option[string] =
139+ ## First column of the first row, as text.
140+ let stmt = prepare(db, sql, params)
141+ defer: discard sqlite3_finalize(stmt)
142+ if sqlite3_step(stmt) != SQLITE_ROW:
143+ return none(string)
144+ let raw = sqlite3_column_text(stmt, 0)
145+ if raw == nil: none(string) else: some($raw)
146+
147+proc queryInt*(db: Db, sql: string, params: varargs[Param]): Option[int64] =
148+ let stmt = prepare(db, sql, params)
149+ defer: discard sqlite3_finalize(stmt)
150+ if sqlite3_step(stmt) != SQLITE_ROW:
151+ return none(int64)
152+ some(sqlite3_column_int64(stmt, 0))
153+
154+iterator rows*(db: Db, sql: string, params: varargs[Param]): seq[string] =
155+ ## Every row, all columns as text. Fine for the small result sets a session
156+ ## store deals in; not for scanning a table of articles.
157+ let stmt = prepare(db, sql, @params)
158+ defer: discard sqlite3_finalize(stmt)
159+ let n = sqlite3_column_count(stmt)
160+ while sqlite3_step(stmt) == SQLITE_ROW:
161+ var row = newSeq[string](n)
162+ for i in 0 ..< n:
163+ let raw = sqlite3_column_text(stmt, i.cint)
164+ row[i] = if raw == nil: "" else: $raw
165+ yield row
added spike/nim/atproto/store.nim +217 -0
new file mode 100644
@@ -0,0 +1,217 @@
1+## Persisting what an OAuth session and a firehose subscription need to
2+## survive a restart.
3+##
4+## Three things are stored, matching the tables Glean's Go backend uses:
5+##
6+## oauth_auth_requests a flow in progress, keyed by `state`
7+## oauth_sessions a live session, keyed by (account_did, session_id)
8+## jetstream_cursor a single row: where the firehose got to
9+##
10+## The rows hold JSON, as the Go side does, so the schema does not have to
11+## change every time a field is added to a session.
12+##
13+## **This stores key material.** A session row contains the DPoP private key
14+## and the refresh token; together they are the account. The file wants the
15+## same protection as a password store: not world-readable, and not in a
16+## backup that travels somewhere less careful.
17+
18+import std/[json, options, sequtils, times]
19+import ./dpop
20+import ./oauth
21+import ./sqlite
22+
23+type
24+ StoreError* = object of CatchableError
25+
26+ SessionRecord* = object ## A usable OAuth session.
27+ accountDid*: string
28+ sessionId*: string
29+ pdsUrl*: string
30+ authServerUrl*: string
31+ tokenEndpoint*: string
32+ accessToken*: string
33+ refreshToken*: string
34+ dpopKey*: P256Key
35+ hostNonce*: string
36+ authNonce*: string
37+ scopes*: seq[string]
38+ createdAt*: int64
39+
40+const Schema = [
41+ """CREATE TABLE IF NOT EXISTS oauth_auth_requests (
42+ state TEXT PRIMARY KEY,
43+ data TEXT NOT NULL,
44+ created_at INTEGER NOT NULL
45+ )""",
46+ """CREATE TABLE IF NOT EXISTS oauth_sessions (
47+ account_did TEXT NOT NULL,
48+ session_id TEXT NOT NULL,
49+ data TEXT NOT NULL,
50+ PRIMARY KEY (account_did, session_id)
51+ )""",
52+ """CREATE TABLE IF NOT EXISTS jetstream_cursor (
53+ id INTEGER PRIMARY KEY CHECK(id = 1),
54+ cursor_us INTEGER NOT NULL
55+ )""",
56+]
57+
58+proc migrate*(db: Db) =
59+ for stmt in Schema:
60+ db.exec stmt
61+
62+# --- key serialisation -----------------------------------------------------
63+
64+proc toJson*(k: P256Key): JsonNode =
65+ ## The private scalar plus the public point.
66+ ##
67+ ## The public half is stored rather than re-derived on load: deriving it
68+ ## needs a curve operation, and a stored key that cannot be read back
69+ ## without one is a restart that silently drops every session.
70+ %*{"d": b64u(k.priv), "x": b64u(k.pubX), "y": b64u(k.pubY)}
71+
72+proc keyFromJson*(j: JsonNode): P256Key =
73+ proc part(name: string): seq[byte] =
74+ let raw = b64uDecode(j{name}.getStr)
75+ if raw.len != 32:
76+ raise newException(StoreError, "stored key field " & name & " is not 32 bytes")
77+ result = newSeq[byte](32)
78+ copyMem(result[0].addr, raw[0].unsafeAddr, 32)
79+
80+ let d = part("d")
81+ let x = part("x")
82+ let y = part("y")
83+ for i in 0 ..< 32:
84+ result.priv[i] = d[i]
85+ result.pubX[i] = x[i]
86+ result.pubY[i] = y[i]
87+
88+# --- auth requests ---------------------------------------------------------
89+
90+proc toJson(info: AuthRequestData): JsonNode =
91+ %*{
92+ "state": info.state,
93+ "auth_server_url": info.authServerUrl,
94+ "scopes": info.scopes,
95+ "pkce_verifier": info.pkceVerifier,
96+ "request_uri": info.requestUri,
97+ "token_endpoint": info.tokenEndpoint,
98+ "revocation_endpoint": info.revocationEndpoint,
99+ "dpop_nonce": info.dpopNonce,
100+ "dpop_key": info.dpopKey.toJson,
101+ "account_did": info.accountDid,
102+ }
103+
104+proc authRequestFromJson(j: JsonNode): AuthRequestData =
105+ AuthRequestData(
106+ state: j{"state"}.getStr,
107+ authServerUrl: j{"auth_server_url"}.getStr,
108+ scopes: j{"scopes"}.getElems.mapIt(it.getStr),
109+ pkceVerifier: j{"pkce_verifier"}.getStr,
110+ requestUri: j{"request_uri"}.getStr,
111+ tokenEndpoint: j{"token_endpoint"}.getStr,
112+ revocationEndpoint: j{"revocation_endpoint"}.getStr,
113+ dpopNonce: j{"dpop_nonce"}.getStr,
114+ dpopKey: keyFromJson(j{"dpop_key"}),
115+ accountDid: j{"account_did"}.getStr,
116+ )
117+
118+proc saveAuthRequest*(db: Db, info: AuthRequestData) =
119+ db.run("""INSERT INTO oauth_auth_requests (state, data, created_at)
120+ VALUES (?, ?, ?)
121+ ON CONFLICT(state) DO UPDATE SET data = excluded.data""",
122+ p(info.state), p($info.toJson), p(getTime().toUnix))
123+
124+proc takeAuthRequest*(db: Db, state: string): Option[AuthRequestData] =
125+ ## Read **and delete**: an authorization code is single-use, so leaving the
126+ ## request behind invites a replay of the callback.
127+ let row = db.queryText(
128+ "SELECT data FROM oauth_auth_requests WHERE state = ?", p(state))
129+ if row.isNone:
130+ return none(AuthRequestData)
131+ db.run("DELETE FROM oauth_auth_requests WHERE state = ?", p(state))
132+ try:
133+ some(authRequestFromJson(parseJson(row.get)))
134+ except JsonParsingError:
135+ raise newException(StoreError, "stored auth request is not valid JSON")
136+
137+proc purgeAuthRequests*(db: Db, olderThan: int64) =
138+ ## Flows that were started and never finished. Without this the table grows
139+ ## forever, one row per abandoned sign-in.
140+ db.run("DELETE FROM oauth_auth_requests WHERE created_at < ?",
141+ p(getTime().toUnix - olderThan))
142+
143+# --- sessions --------------------------------------------------------------
144+
145+proc toJson(s: SessionRecord): JsonNode =
146+ %*{
147+ "account_did": s.accountDid,
148+ "session_id": s.sessionId,
149+ "pds_url": s.pdsUrl,
150+ "auth_server_url": s.authServerUrl,
151+ "token_endpoint": s.tokenEndpoint,
152+ "access_token": s.accessToken,
153+ "refresh_token": s.refreshToken,
154+ "dpop_key": s.dpopKey.toJson,
155+ "host_nonce": s.hostNonce,
156+ "auth_nonce": s.authNonce,
157+ "scopes": s.scopes,
158+ "created_at": s.createdAt,
159+ }
160+
161+proc sessionFromJson(j: JsonNode): SessionRecord =
162+ SessionRecord(
163+ accountDid: j{"account_did"}.getStr,
164+ sessionId: j{"session_id"}.getStr,
165+ pdsUrl: j{"pds_url"}.getStr,
166+ authServerUrl: j{"auth_server_url"}.getStr,
167+ tokenEndpoint: j{"token_endpoint"}.getStr,
168+ accessToken: j{"access_token"}.getStr,
169+ refreshToken: j{"refresh_token"}.getStr,
170+ dpopKey: keyFromJson(j{"dpop_key"}),
171+ hostNonce: j{"host_nonce"}.getStr,
172+ authNonce: j{"auth_nonce"}.getStr,
173+ scopes: j{"scopes"}.getElems.mapIt(it.getStr),
174+ createdAt: j{"created_at"}.getBiggestInt.int64,
175+ )
176+
177+proc saveSession*(db: Db, s: SessionRecord) =
178+ if s.accountDid.len == 0 or s.sessionId.len == 0:
179+ raise newException(StoreError, "a session needs both a DID and a session id")
180+ db.run("""INSERT INTO oauth_sessions (account_did, session_id, data)
181+ VALUES (?, ?, ?)
182+ ON CONFLICT(account_did, session_id) DO UPDATE SET data = excluded.data""",
183+ p(s.accountDid), p(s.sessionId), p($s.toJson))
184+
185+proc getSession*(db: Db, did, sessionId: string): Option[SessionRecord] =
186+ let row = db.queryText(
187+ "SELECT data FROM oauth_sessions WHERE account_did = ? AND session_id = ?",
188+ p(did), p(sessionId))
189+ if row.isNone:
190+ return none(SessionRecord)
191+ try:
192+ some(sessionFromJson(parseJson(row.get)))
193+ except JsonParsingError:
194+ raise newException(StoreError, "stored session is not valid JSON")
195+
196+proc deleteSession*(db: Db, did, sessionId: string) =
197+ db.run("DELETE FROM oauth_sessions WHERE account_did = ? AND session_id = ?",
198+ p(did), p(sessionId))
199+
200+proc listSessions*(db: Db, did: string): seq[string] =
201+ for row in db.rows(
202+ "SELECT session_id FROM oauth_sessions WHERE account_did = ?", p(did)):
203+ result.add row[0]
204+
205+proc listAccounts*(db: Db): seq[string] =
206+ for row in db.rows("SELECT DISTINCT account_did FROM oauth_sessions"):
207+ result.add row[0]
208+
209+# --- jetstream cursor ------------------------------------------------------
210+
211+proc saveCursor*(db: Db, cursorUs: int64) =
212+ db.run("""INSERT INTO jetstream_cursor (id, cursor_us) VALUES (1, ?)
213+ ON CONFLICT(id) DO UPDATE SET cursor_us = excluded.cursor_us""",
214+ p(cursorUs))
215+
216+proc loadCursor*(db: Db): int64 =
217+ db.queryInt("SELECT cursor_us FROM jetstream_cursor WHERE id = 1").get(0'i64)
new file mode 100644
@@ -0,0 +1,217 @@
1+## Persisting what an OAuth session and a firehose subscription need to
2+## survive a restart.
3+##
4+## Three things are stored, matching the tables Glean's Go backend uses:
5+##
6+## oauth_auth_requests a flow in progress, keyed by `state`
7+## oauth_sessions a live session, keyed by (account_did, session_id)
8+## jetstream_cursor a single row: where the firehose got to
9+##
10+## The rows hold JSON, as the Go side does, so the schema does not have to
11+## change every time a field is added to a session.
12+##
13+## **This stores key material.** A session row contains the DPoP private key
14+## and the refresh token; together they are the account. The file wants the
15+## same protection as a password store: not world-readable, and not in a
16+## backup that travels somewhere less careful.
17+
18+import std/[json, options, sequtils, times]
19+import ./dpop
20+import ./oauth
21+import ./sqlite
22+
23+type
24+ StoreError* = object of CatchableError
25+
26+ SessionRecord* = object ## A usable OAuth session.
27+ accountDid*: string
28+ sessionId*: string
29+ pdsUrl*: string
30+ authServerUrl*: string
31+ tokenEndpoint*: string
32+ accessToken*: string
33+ refreshToken*: string
34+ dpopKey*: P256Key
35+ hostNonce*: string
36+ authNonce*: string
37+ scopes*: seq[string]
38+ createdAt*: int64
39+
40+const Schema = [
41+ """CREATE TABLE IF NOT EXISTS oauth_auth_requests (
42+ state TEXT PRIMARY KEY,
43+ data TEXT NOT NULL,
44+ created_at INTEGER NOT NULL
45+ )""",
46+ """CREATE TABLE IF NOT EXISTS oauth_sessions (
47+ account_did TEXT NOT NULL,
48+ session_id TEXT NOT NULL,
49+ data TEXT NOT NULL,
50+ PRIMARY KEY (account_did, session_id)
51+ )""",
52+ """CREATE TABLE IF NOT EXISTS jetstream_cursor (
53+ id INTEGER PRIMARY KEY CHECK(id = 1),
54+ cursor_us INTEGER NOT NULL
55+ )""",
56+]
57+
58+proc migrate*(db: Db) =
59+ for stmt in Schema:
60+ db.exec stmt
61+
62+# --- key serialisation -----------------------------------------------------
63+
64+proc toJson*(k: P256Key): JsonNode =
65+ ## The private scalar plus the public point.
66+ ##
67+ ## The public half is stored rather than re-derived on load: deriving it
68+ ## needs a curve operation, and a stored key that cannot be read back
69+ ## without one is a restart that silently drops every session.
70+ %*{"d": b64u(k.priv), "x": b64u(k.pubX), "y": b64u(k.pubY)}
71+
72+proc keyFromJson*(j: JsonNode): P256Key =
73+ proc part(name: string): seq[byte] =
74+ let raw = b64uDecode(j{name}.getStr)
75+ if raw.len != 32:
76+ raise newException(StoreError, "stored key field " & name & " is not 32 bytes")
77+ result = newSeq[byte](32)
78+ copyMem(result[0].addr, raw[0].unsafeAddr, 32)
79+
80+ let d = part("d")
81+ let x = part("x")
82+ let y = part("y")
83+ for i in 0 ..< 32:
84+ result.priv[i] = d[i]
85+ result.pubX[i] = x[i]
86+ result.pubY[i] = y[i]
87+
88+# --- auth requests ---------------------------------------------------------
89+
90+proc toJson(info: AuthRequestData): JsonNode =
91+ %*{
92+ "state": info.state,
93+ "auth_server_url": info.authServerUrl,
94+ "scopes": info.scopes,
95+ "pkce_verifier": info.pkceVerifier,
96+ "request_uri": info.requestUri,
97+ "token_endpoint": info.tokenEndpoint,
98+ "revocation_endpoint": info.revocationEndpoint,
99+ "dpop_nonce": info.dpopNonce,
100+ "dpop_key": info.dpopKey.toJson,
101+ "account_did": info.accountDid,
102+ }
103+
104+proc authRequestFromJson(j: JsonNode): AuthRequestData =
105+ AuthRequestData(
106+ state: j{"state"}.getStr,
107+ authServerUrl: j{"auth_server_url"}.getStr,
108+ scopes: j{"scopes"}.getElems.mapIt(it.getStr),
109+ pkceVerifier: j{"pkce_verifier"}.getStr,
110+ requestUri: j{"request_uri"}.getStr,
111+ tokenEndpoint: j{"token_endpoint"}.getStr,
112+ revocationEndpoint: j{"revocation_endpoint"}.getStr,
113+ dpopNonce: j{"dpop_nonce"}.getStr,
114+ dpopKey: keyFromJson(j{"dpop_key"}),
115+ accountDid: j{"account_did"}.getStr,
116+ )
117+
118+proc saveAuthRequest*(db: Db, info: AuthRequestData) =
119+ db.run("""INSERT INTO oauth_auth_requests (state, data, created_at)
120+ VALUES (?, ?, ?)
121+ ON CONFLICT(state) DO UPDATE SET data = excluded.data""",
122+ p(info.state), p($info.toJson), p(getTime().toUnix))
123+
124+proc takeAuthRequest*(db: Db, state: string): Option[AuthRequestData] =
125+ ## Read **and delete**: an authorization code is single-use, so leaving the
126+ ## request behind invites a replay of the callback.
127+ let row = db.queryText(
128+ "SELECT data FROM oauth_auth_requests WHERE state = ?", p(state))
129+ if row.isNone:
130+ return none(AuthRequestData)
131+ db.run("DELETE FROM oauth_auth_requests WHERE state = ?", p(state))
132+ try:
133+ some(authRequestFromJson(parseJson(row.get)))
134+ except JsonParsingError:
135+ raise newException(StoreError, "stored auth request is not valid JSON")
136+
137+proc purgeAuthRequests*(db: Db, olderThan: int64) =
138+ ## Flows that were started and never finished. Without this the table grows
139+ ## forever, one row per abandoned sign-in.
140+ db.run("DELETE FROM oauth_auth_requests WHERE created_at < ?",
141+ p(getTime().toUnix - olderThan))
142+
143+# --- sessions --------------------------------------------------------------
144+
145+proc toJson(s: SessionRecord): JsonNode =
146+ %*{
147+ "account_did": s.accountDid,
148+ "session_id": s.sessionId,
149+ "pds_url": s.pdsUrl,
150+ "auth_server_url": s.authServerUrl,
151+ "token_endpoint": s.tokenEndpoint,
152+ "access_token": s.accessToken,
153+ "refresh_token": s.refreshToken,
154+ "dpop_key": s.dpopKey.toJson,
155+ "host_nonce": s.hostNonce,
156+ "auth_nonce": s.authNonce,
157+ "scopes": s.scopes,
158+ "created_at": s.createdAt,
159+ }
160+
161+proc sessionFromJson(j: JsonNode): SessionRecord =
162+ SessionRecord(
163+ accountDid: j{"account_did"}.getStr,
164+ sessionId: j{"session_id"}.getStr,
165+ pdsUrl: j{"pds_url"}.getStr,
166+ authServerUrl: j{"auth_server_url"}.getStr,
167+ tokenEndpoint: j{"token_endpoint"}.getStr,
168+ accessToken: j{"access_token"}.getStr,
169+ refreshToken: j{"refresh_token"}.getStr,
170+ dpopKey: keyFromJson(j{"dpop_key"}),
171+ hostNonce: j{"host_nonce"}.getStr,
172+ authNonce: j{"auth_nonce"}.getStr,
173+ scopes: j{"scopes"}.getElems.mapIt(it.getStr),
174+ createdAt: j{"created_at"}.getBiggestInt.int64,
175+ )
176+
177+proc saveSession*(db: Db, s: SessionRecord) =
178+ if s.accountDid.len == 0 or s.sessionId.len == 0:
179+ raise newException(StoreError, "a session needs both a DID and a session id")
180+ db.run("""INSERT INTO oauth_sessions (account_did, session_id, data)
181+ VALUES (?, ?, ?)
182+ ON CONFLICT(account_did, session_id) DO UPDATE SET data = excluded.data""",
183+ p(s.accountDid), p(s.sessionId), p($s.toJson))
184+
185+proc getSession*(db: Db, did, sessionId: string): Option[SessionRecord] =
186+ let row = db.queryText(
187+ "SELECT data FROM oauth_sessions WHERE account_did = ? AND session_id = ?",
188+ p(did), p(sessionId))
189+ if row.isNone:
190+ return none(SessionRecord)
191+ try:
192+ some(sessionFromJson(parseJson(row.get)))
193+ except JsonParsingError:
194+ raise newException(StoreError, "stored session is not valid JSON")
195+
196+proc deleteSession*(db: Db, did, sessionId: string) =
197+ db.run("DELETE FROM oauth_sessions WHERE account_did = ? AND session_id = ?",
198+ p(did), p(sessionId))
199+
200+proc listSessions*(db: Db, did: string): seq[string] =
201+ for row in db.rows(
202+ "SELECT session_id FROM oauth_sessions WHERE account_did = ?", p(did)):
203+ result.add row[0]
204+
205+proc listAccounts*(db: Db): seq[string] =
206+ for row in db.rows("SELECT DISTINCT account_did FROM oauth_sessions"):
207+ result.add row[0]
208+
209+# --- jetstream cursor ------------------------------------------------------
210+
211+proc saveCursor*(db: Db, cursorUs: int64) =
212+ db.run("""INSERT INTO jetstream_cursor (id, cursor_us) VALUES (1, ?)
213+ ON CONFLICT(id) DO UPDATE SET cursor_us = excluded.cursor_us""",
214+ p(cursorUs))
215+
216+proc loadCursor*(db: Db): int64 =
217+ db.queryInt("SELECT cursor_us FROM jetstream_cursor WHERE id = 1").get(0'i64)
modified spike/nim/readme.md +38 -1
@@ -15,6 +15,7 @@ nim c -r -d:ssl oauth_probe.nim [handle] # the flow
1515 nim c -r -d:ssl xrpc_probe.nim # signing
1616 nim c -r -d:ssl jetstream_probe.nim # firehose
1717 nim c -r -d:ssl reconnect_probe.nim # supervisor
18+nim c -r -d:ssl store_probe.nim # persistence
1819 ```
1920
2021 Reusable code lives in `atproto/`; the `*_probe.nim` files are the checks.
@@ -206,6 +207,43 @@ to the initial delay, skipping jitter on the first retry. That is precisely
206207 the moment jitter exists for: one server restart drops every consumer at
207208 once, and un-jittered first retries bring them all back simultaneously.
208209
210+## 7. Session persistence — works
211+
212+`atproto/sqlite.nim` is a thin wrapper over the library the storage spike
213+already proved (sqlite-vec linked in, WAL on). `atproto/store.nim` keeps the
214+three things a restart must not lose, in the same tables the Go backend uses:
215+`oauth_auth_requests`, `oauth_sessions`, `jetstream_cursor`. Rows hold JSON,
216+as on the Go side, so adding a session field does not mean a migration.
217+
218+`atproto/cursor_store.nim` bridges the store and the supervisor, batching
219+writes: the cursor advances on every event, often hundreds a second, and a
220+synchronous write per event would make the database the bottleneck for a
221+firehose. The stored cursor therefore trails the live one — which is exactly
222+the gap the supervisor's rewind exists to cover.
223+
224+The check that earns its keep is the DPoP key. A missing row is loud; a key
225+that round-trips *almost* correctly is not — every request after a restart
226+would fail authentication with nothing pointing at storage. So the reloaded
227+key is checked by **signing with it and verifying against the original public
228+point**, with a second key as a negative control. Comparing fields, or
229+re-signing and comparing bytes, would prove neither thing: ECDSA is
230+randomised.
231+
232+Two other properties worth pinning:
233+
234+- Taking a pending auth request **consumes** it. An authorization code is
235+ single-use, and leaving the row behind invites a replayed callback.
236+- Cursors never move backwards. A late event that rewound the stored cursor
237+ would rewind the subscription on the next reconnect.
238+
239+The public half of the key is stored rather than re-derived on load. Deriving
240+it needs a curve operation, and a stored key that cannot be read back without
241+one is a restart that silently drops every session.
242+
243+**These rows are key material.** A session holds the DPoP private key and the
244+refresh token; together they are the account. The database file wants the
245+protection of a password store, not of a cache.
246+
209247 ## What this does and does not prove
210248
211249 Proven: the cryptography, the storage layer, and the OAuth flow up to user
@@ -217,7 +255,6 @@ Not proven, and still ahead:
217255 exercise. This is the one place a real end-to-end test still has to happen,
218256 and it also covers the `invalid_token` refresh-and-retry branch of
219257 `xrpc.request`, which is currently unexercised.
220-- **Session persistence** across restarts.
221258
222259 ### Not needed: DAG-CBOR, CAR and the MST
223260
@@ -15,6 +15,7 @@ nim c -r -d:ssl oauth_probe.nim [handle] # the flow
15 nim c -r -d:ssl xrpc_probe.nim # signing15 nim c -r -d:ssl xrpc_probe.nim # signing
16 nim c -r -d:ssl jetstream_probe.nim # firehose16 nim c -r -d:ssl jetstream_probe.nim # firehose
17 nim c -r -d:ssl reconnect_probe.nim # supervisor17 nim c -r -d:ssl reconnect_probe.nim # supervisor
18+nim c -r -d:ssl store_probe.nim # persistence
18 ```19 ```
19 20
20 Reusable code lives in `atproto/`; the `*_probe.nim` files are the checks.21 Reusable code lives in `atproto/`; the `*_probe.nim` files are the checks.
@@ -206,6 +207,43 @@ to the initial delay, skipping jitter on the first retry. That is precisely
206 the moment jitter exists for: one server restart drops every consumer at207 the moment jitter exists for: one server restart drops every consumer at
207 once, and un-jittered first retries bring them all back simultaneously.208 once, and un-jittered first retries bring them all back simultaneously.
208 209
210+## 7. Session persistence — works
211+
212+`atproto/sqlite.nim` is a thin wrapper over the library the storage spike
213+already proved (sqlite-vec linked in, WAL on). `atproto/store.nim` keeps the
214+three things a restart must not lose, in the same tables the Go backend uses:
215+`oauth_auth_requests`, `oauth_sessions`, `jetstream_cursor`. Rows hold JSON,
216+as on the Go side, so adding a session field does not mean a migration.
217+
218+`atproto/cursor_store.nim` bridges the store and the supervisor, batching
219+writes: the cursor advances on every event, often hundreds a second, and a
220+synchronous write per event would make the database the bottleneck for a
221+firehose. The stored cursor therefore trails the live one — which is exactly
222+the gap the supervisor's rewind exists to cover.
223+
224+The check that earns its keep is the DPoP key. A missing row is loud; a key
225+that round-trips *almost* correctly is not — every request after a restart
226+would fail authentication with nothing pointing at storage. So the reloaded
227+key is checked by **signing with it and verifying against the original public
228+point**, with a second key as a negative control. Comparing fields, or
229+re-signing and comparing bytes, would prove neither thing: ECDSA is
230+randomised.
231+
232+Two other properties worth pinning:
233+
234+- Taking a pending auth request **consumes** it. An authorization code is
235+ single-use, and leaving the row behind invites a replayed callback.
236+- Cursors never move backwards. A late event that rewound the stored cursor
237+ would rewind the subscription on the next reconnect.
238+
239+The public half of the key is stored rather than re-derived on load. Deriving
240+it needs a curve operation, and a stored key that cannot be read back without
241+one is a restart that silently drops every session.
242+
243+**These rows are key material.** A session holds the DPoP private key and the
244+refresh token; together they are the account. The database file wants the
245+protection of a password store, not of a cache.
246+
209 ## What this does and does not prove247 ## What this does and does not prove
210 248
211 Proven: the cryptography, the storage layer, and the OAuth flow up to user249 Proven: the cryptography, the storage layer, and the OAuth flow up to user
@@ -217,7 +255,6 @@ Not proven, and still ahead:
217 exercise. This is the one place a real end-to-end test still has to happen,255 exercise. This is the one place a real end-to-end test still has to happen,
218 and it also covers the `invalid_token` refresh-and-retry branch of256 and it also covers the `invalid_token` refresh-and-retry branch of
219 `xrpc.request`, which is currently unexercised.257 `xrpc.request`, which is currently unexercised.
220-- **Session persistence** across restarts.
221 258
222 ### Not needed: DAG-CBOR, CAR and the MST259 ### Not needed: DAG-CBOR, CAR and the MST
223 260
added spike/nim/store_probe.nim +185 -0
new file mode 100644
@@ -0,0 +1,185 @@
1+## Checks that a session survives a restart intact.
2+##
3+## nim c -r -d:ssl store_probe.nim
4+##
5+## The interesting failure is not "the row is missing" -- that is loud. It is
6+## a DPoP key that round-trips *almost* correctly: every request after a
7+## restart then fails authentication with no clue that storage is at fault.
8+## So the key is checked by signing with the reloaded copy and verifying
9+## against the original public point, not by comparing fields.
10+
11+import std/[json, options, os, strformat, strutils]
12+import pkg/chronos
13+import bearssl/rand
14+import atproto/[cursor_store, dpop, jetstream_run, oauth, sqlite, store]
15+
16+var failures = 0
17+
18+proc report(name: string, ok: bool, detail = "") =
19+ if not ok: inc failures
20+ let label = if ok: "PASS" else: "FAIL"
21+ if detail.len > 0: echo &" {label} {name} -- {detail}"
22+ else: echo &" {label} {name}"
23+
24+proc main() =
25+ echo "Session persistence probe"
26+ echo ""
27+
28+ let rngRef = HmacDrbgContext.new()
29+ doAssert rngRef != nil
30+ var rng = rngRef[]
31+
32+ let path = getTempDir() / "glean_store_probe.db"
33+ removeFile(path)
34+ removeFile(path & "-wal")
35+ removeFile(path & "-shm")
36+ defer:
37+ removeFile(path)
38+ removeFile(path & "-wal")
39+ removeFile(path & "-shm")
40+
41+ let key = generateKey(rng)
42+ let originalThumb = key.thumbprint
43+
44+ # --- write, then close the database entirely ---------------------------
45+ block:
46+ var db = sqlite.open(path)
47+ migrate(db)
48+ report("schema applies", true, "sqlite " & libVersion())
49+
50+ saveSession(db, SessionRecord(
51+ accountDid: "did:plc:probe",
52+ sessionId: "sess-1",
53+ pdsUrl: "https://pds.example",
54+ authServerUrl: "https://bsky.social",
55+ tokenEndpoint: "https://bsky.social/oauth/token",
56+ accessToken: "access-abc",
57+ refreshToken: "refresh-xyz",
58+ dpopKey: key,
59+ hostNonce: "host-n",
60+ authNonce: "auth-n",
61+ scopes: @["atproto", "transition:generic"],
62+ createdAt: 1_700_000_000,
63+ ))
64+ saveCursor(db, 1_700_000_000_000_000'i64)
65+ saveAuthRequest(db, AuthRequestData(
66+ state: "state-1",
67+ authServerUrl: "https://bsky.social",
68+ scopes: @["atproto"],
69+ pkceVerifier: "verifier-1",
70+ requestUri: "urn:ietf:params:oauth:request_uri:req-1",
71+ tokenEndpoint: "https://bsky.social/oauth/token",
72+ dpopNonce: "n-1",
73+ dpopKey: key,
74+ accountDid: "did:plc:probe",
75+ ))
76+ db.close()
77+
78+ # --- reopen, as a restart would ----------------------------------------
79+ var db = sqlite.open(path)
80+ defer: db.close()
81+ migrate(db) # idempotent, as it is on every boot
82+
83+ block:
84+ let got = getSession(db, "did:plc:probe", "sess-1")
85+ report("session survives a restart", got.isSome)
86+ if got.isSome:
87+ let s = got.get
88+ report("tokens round-trip",
89+ s.accessToken == "access-abc" and s.refreshToken == "refresh-xyz")
90+ report("nonces round-trip",
91+ s.hostNonce == "host-n" and s.authNonce == "auth-n")
92+ report("scopes round-trip", s.scopes == @["atproto", "transition:generic"])
93+
94+ # The check that matters.
95+ report("reloaded key has the same JWK thumbprint",
96+ s.dpopKey.thumbprint == originalThumb, s.dpopKey.thumbprint)
97+
98+ const msg = "eyJhbGciOiJFUzI1NiJ9.eyJodG0iOiJHRVQifQ"
99+ let sig = s.dpopKey.signEs256(rng, msg)
100+ report("reloaded key still signs verifiably",
101+ key.verifyEs256(msg, sig),
102+ "signed with the reloaded key, verified against the original")
103+
104+ # Negative control: a verifier that accepts anything proves nothing.
105+ let other = generateKey(rng)
106+ report("a different key does not verify",
107+ not other.verifyEs256(msg, sig))
108+
109+ block:
110+ report("unknown session reads as none",
111+ getSession(db, "did:plc:probe", "nope").isNone)
112+ report("sessions are listed per account",
113+ listSessions(db, "did:plc:probe") == @["sess-1"])
114+ report("accounts are listed", listAccounts(db) == @["did:plc:probe"])
115+
116+ block:
117+ report("cursor survives a restart",
118+ loadCursor(db) == 1_700_000_000_000_000'i64, $loadCursor(db))
119+
120+ # --- auth requests are single-use --------------------------------------
121+ block:
122+ let first = takeAuthRequest(db, "state-1")
123+ report("pending auth request survives a restart", first.isSome)
124+ if first.isSome:
125+ report("PKCE verifier round-trips", first.get.pkceVerifier == "verifier-1")
126+ report("auth request key round-trips",
127+ first.get.dpopKey.thumbprint == originalThumb)
128+ # An authorization code is single-use; leaving the row behind invites a
129+ # replayed callback.
130+ report("taking an auth request consumes it",
131+ takeAuthRequest(db, "state-1").isNone)
132+
133+ # --- batched cursor store ----------------------------------------------
134+ block:
135+ let cs = newDbCursorStore(db, flushEvery = 3600) # never auto-flush here
136+ report("cursor store loads the persisted value",
137+ cs.load() == 1_700_000_000_000_000'i64)
138+
139+ cs.save(1_700_000_000_500_000'i64)
140+ # Batching means the write is deferred, not lost.
141+ report("a fresh cursor is not written immediately",
142+ loadCursor(db) == 1_700_000_000_000_000'i64)
143+ cs.close()
144+ report("closing flushes the pending cursor",
145+ loadCursor(db) == 1_700_000_000_500_000'i64, $loadCursor(db))
146+
147+ # Cursors must never move backwards, or a late event would rewind the
148+ # subscription on the next reconnect.
149+ let cs2 = newDbCursorStore(db, flushEvery = 0)
150+ cs2.save(1'i64)
151+ cs2.close()
152+ report("an older cursor does not overwrite a newer one",
153+ loadCursor(db) == 1_700_000_000_500_000'i64, $loadCursor(db))
154+
155+ # --- the seam with the supervisor --------------------------------------
156+ # jetstream_run takes a CursorStore base reference, so this only works if
157+ # the overrides dispatch dynamically. A silent fall-through to the base
158+ # method would read the in-memory field instead of the database and look
159+ # entirely healthy while resuming from the wrong place.
160+ block:
161+ let cs = newDbCursorStore(db, flushEvery = 3600)
162+ let asBase: CursorStore = cs
163+ report("DbCursorStore dispatches through the CursorStore base",
164+ asBase.load() == loadCursor(db), $asBase.load())
165+
166+ let rewound = rewoundCursor(asBase, 5.seconds)
167+ report("the supervisor rewinds the persisted cursor",
168+ rewound == loadCursor(db) - 5_000_000, $rewound)
169+
170+ # --- deletion ----------------------------------------------------------
171+ block:
172+ deleteSession(db, "did:plc:probe", "sess-1")
173+ report("sessions can be deleted",
174+ getSession(db, "did:plc:probe", "sess-1").isNone and
175+ listAccounts(db).len == 0)
176+
177+ echo ""
178+ if failures == 0:
179+ echo "RESULT: sessions, keys and cursors survive a restart."
180+ else:
181+ echo &"RESULT: {failures} check(s) failed."
182+ quit 1
183+
184+when isMainModule:
185+ main()
new file mode 100644
@@ -0,0 +1,185 @@
1+## Checks that a session survives a restart intact.
2+##
3+## nim c -r -d:ssl store_probe.nim
4+##
5+## The interesting failure is not "the row is missing" -- that is loud. It is
6+## a DPoP key that round-trips *almost* correctly: every request after a
7+## restart then fails authentication with no clue that storage is at fault.
8+## So the key is checked by signing with the reloaded copy and verifying
9+## against the original public point, not by comparing fields.
10+
11+import std/[json, options, os, strformat, strutils]
12+import pkg/chronos
13+import bearssl/rand
14+import atproto/[cursor_store, dpop, jetstream_run, oauth, sqlite, store]
15+
16+var failures = 0
17+
18+proc report(name: string, ok: bool, detail = "") =
19+ if not ok: inc failures
20+ let label = if ok: "PASS" else: "FAIL"
21+ if detail.len > 0: echo &" {label} {name} -- {detail}"
22+ else: echo &" {label} {name}"
23+
24+proc main() =
25+ echo "Session persistence probe"
26+ echo ""
27+
28+ let rngRef = HmacDrbgContext.new()
29+ doAssert rngRef != nil
30+ var rng = rngRef[]
31+
32+ let path = getTempDir() / "glean_store_probe.db"
33+ removeFile(path)
34+ removeFile(path & "-wal")
35+ removeFile(path & "-shm")
36+ defer:
37+ removeFile(path)
38+ removeFile(path & "-wal")
39+ removeFile(path & "-shm")
40+
41+ let key = generateKey(rng)
42+ let originalThumb = key.thumbprint
43+
44+ # --- write, then close the database entirely ---------------------------
45+ block:
46+ var db = sqlite.open(path)
47+ migrate(db)
48+ report("schema applies", true, "sqlite " & libVersion())
49+
50+ saveSession(db, SessionRecord(
51+ accountDid: "did:plc:probe",
52+ sessionId: "sess-1",
53+ pdsUrl: "https://pds.example",
54+ authServerUrl: "https://bsky.social",
55+ tokenEndpoint: "https://bsky.social/oauth/token",
56+ accessToken: "access-abc",
57+ refreshToken: "refresh-xyz",
58+ dpopKey: key,
59+ hostNonce: "host-n",
60+ authNonce: "auth-n",
61+ scopes: @["atproto", "transition:generic"],
62+ createdAt: 1_700_000_000,
63+ ))
64+ saveCursor(db, 1_700_000_000_000_000'i64)
65+ saveAuthRequest(db, AuthRequestData(
66+ state: "state-1",
67+ authServerUrl: "https://bsky.social",
68+ scopes: @["atproto"],
69+ pkceVerifier: "verifier-1",
70+ requestUri: "urn:ietf:params:oauth:request_uri:req-1",
71+ tokenEndpoint: "https://bsky.social/oauth/token",
72+ dpopNonce: "n-1",
73+ dpopKey: key,
74+ accountDid: "did:plc:probe",
75+ ))
76+ db.close()
77+
78+ # --- reopen, as a restart would ----------------------------------------
79+ var db = sqlite.open(path)
80+ defer: db.close()
81+ migrate(db) # idempotent, as it is on every boot
82+
83+ block:
84+ let got = getSession(db, "did:plc:probe", "sess-1")
85+ report("session survives a restart", got.isSome)
86+ if got.isSome:
87+ let s = got.get
88+ report("tokens round-trip",
89+ s.accessToken == "access-abc" and s.refreshToken == "refresh-xyz")
90+ report("nonces round-trip",
91+ s.hostNonce == "host-n" and s.authNonce == "auth-n")
92+ report("scopes round-trip", s.scopes == @["atproto", "transition:generic"])
93+
94+ # The check that matters.
95+ report("reloaded key has the same JWK thumbprint",
96+ s.dpopKey.thumbprint == originalThumb, s.dpopKey.thumbprint)
97+
98+ const msg = "eyJhbGciOiJFUzI1NiJ9.eyJodG0iOiJHRVQifQ"
99+ let sig = s.dpopKey.signEs256(rng, msg)
100+ report("reloaded key still signs verifiably",
101+ key.verifyEs256(msg, sig),
102+ "signed with the reloaded key, verified against the original")
103+
104+ # Negative control: a verifier that accepts anything proves nothing.
105+ let other = generateKey(rng)
106+ report("a different key does not verify",
107+ not other.verifyEs256(msg, sig))
108+
109+ block:
110+ report("unknown session reads as none",
111+ getSession(db, "did:plc:probe", "nope").isNone)
112+ report("sessions are listed per account",
113+ listSessions(db, "did:plc:probe") == @["sess-1"])
114+ report("accounts are listed", listAccounts(db) == @["did:plc:probe"])
115+
116+ block:
117+ report("cursor survives a restart",
118+ loadCursor(db) == 1_700_000_000_000_000'i64, $loadCursor(db))
119+
120+ # --- auth requests are single-use --------------------------------------
121+ block:
122+ let first = takeAuthRequest(db, "state-1")
123+ report("pending auth request survives a restart", first.isSome)
124+ if first.isSome:
125+ report("PKCE verifier round-trips", first.get.pkceVerifier == "verifier-1")
126+ report("auth request key round-trips",
127+ first.get.dpopKey.thumbprint == originalThumb)
128+ # An authorization code is single-use; leaving the row behind invites a
129+ # replayed callback.
130+ report("taking an auth request consumes it",
131+ takeAuthRequest(db, "state-1").isNone)
132+
133+ # --- batched cursor store ----------------------------------------------
134+ block:
135+ let cs = newDbCursorStore(db, flushEvery = 3600) # never auto-flush here
136+ report("cursor store loads the persisted value",
137+ cs.load() == 1_700_000_000_000_000'i64)
138+
139+ cs.save(1_700_000_000_500_000'i64)
140+ # Batching means the write is deferred, not lost.
141+ report("a fresh cursor is not written immediately",
142+ loadCursor(db) == 1_700_000_000_000_000'i64)
143+ cs.close()
144+ report("closing flushes the pending cursor",
145+ loadCursor(db) == 1_700_000_000_500_000'i64, $loadCursor(db))
146+
147+ # Cursors must never move backwards, or a late event would rewind the
148+ # subscription on the next reconnect.
149+ let cs2 = newDbCursorStore(db, flushEvery = 0)
150+ cs2.save(1'i64)
151+ cs2.close()
152+ report("an older cursor does not overwrite a newer one",
153+ loadCursor(db) == 1_700_000_000_500_000'i64, $loadCursor(db))
154+
155+ # --- the seam with the supervisor --------------------------------------
156+ # jetstream_run takes a CursorStore base reference, so this only works if
157+ # the overrides dispatch dynamically. A silent fall-through to the base
158+ # method would read the in-memory field instead of the database and look
159+ # entirely healthy while resuming from the wrong place.
160+ block:
161+ let cs = newDbCursorStore(db, flushEvery = 3600)
162+ let asBase: CursorStore = cs
163+ report("DbCursorStore dispatches through the CursorStore base",
164+ asBase.load() == loadCursor(db), $asBase.load())
165+
166+ let rewound = rewoundCursor(asBase, 5.seconds)
167+ report("the supervisor rewinds the persisted cursor",
168+ rewound == loadCursor(db) - 5_000_000, $rewound)
169+
170+ # --- deletion ----------------------------------------------------------
171+ block:
172+ deleteSession(db, "did:plc:probe", "sess-1")
173+ report("sessions can be deleted",
174+ getSession(db, "did:plc:probe", "sess-1").isNone and
175+ listAccounts(db).len == 0)
176+
177+ echo ""
178+ if failures == 0:
179+ echo "RESULT: sessions, keys and cursors survive a restart."
180+ else:
181+ echo &"RESULT: {failures} check(s) failed."
182+ quit 1
183+
184+when isMainModule:
185+ main()