nx_chat_store.nx source
↩ module page · 1025 lines · 45147 B
1// nx_chat_store.nx -- C1 of the comms lane: durable STORE-AND-FORWARD chat history on the sovereign
2// seg-store plane. Contract symbol cs_store_forward == the /compare/comms C1 watch contract.
3// DONE-RULE (pre-declared in comms.plan before this organ existed): a message sent to an offline
4// member is delivered on reconnect byte-identical and in order; history survives room close; the
5// store is additive-only with soft-delete and REFUSES loudly at its conf-named size budget instead
6// of silently dropping. REFEREE: nx_chat_store_gate (the flip is the receipt, the gate is the proof).
7//
8// WHY: the live room chat is nx_chat_ring -- bounded, in-memory, gone when the room closes. Its own
9// header declares persistence as pluggable (disk / memory / future MLS-encrypted); this organ is the
10// disk layer, pure COMPOSITION of incumbents, no second ruler:
11// nx_store_seed_lib sts_lock/sts_unlock (the shared <prefix>plock domain) + sts_seed (explicit
12// room creation ONLY) + sts_atoi/sts_rowkey/sts_mm
13// nx_seg_store ss_open_cached + ss_hget (the amplification-safe read path, seq356) +
14// ss_begin_cap/ss_add/ss_commit_cas (O(1) append; CAS turns a lost update
15// into a loud CHAT-RETRY -- backpressure, never corruption)
16// nx_chatmsg cm_advance -- the monotonic receipt ladder; a delivery cursor NEVER regresses
17//
18// MODEL: one plane per room at <store_root>chat_<room>- .
19// q:<i> message row m|<seq>|<epoch_ms>|<kind>|<sender>|<hexbody> (seq == i+1, ASSERTED at read)
20// q:n row count (the store's own counter -- two counters of one population, free audit)
21// meta:bytes cumulative DECODED body bytes (the budget ruler: cumulative, not a ring)
22// cur:<member> per-member delivery cursor (store-and-forward: reconnect = fetch since cursor)
23// del:<seq> soft-delete tombstone (additive-only, rule 13; history is never rewritten)
24//
25// BODIES ARE HEX-ARMORED: plane rows are newline-framed text and chat bodies are arbitrary bytes
26// (tomorrow: MLS ciphertext), so rows carry lowercase hex and the BUDGET COUNTS DECODED BYTES -- the
27// armor never distorts the policy. Byte-identity round-trip is a gate tooth, not a promise.
28// KINDS are nx_chat_ring's SEALED wire constants, validated as the range 1..6 BY CITATION: a
29// cross-import is refused by the import graph (the ring pins nx_syscalls_x86_64, the seg-store pins
30// nx_syscalls; importing both duplicates the syscall layer and the duplicate guard fails the build).
31// If the ring grows a kind, NX_CHAT_KIND_N moves and CS_KIND_MAX here must follow.
32//
33// EVERY POLICY NUMBER LIVES IN THE CONF (rule 11): knowledge/comms/chatstore.conf rows
34// store_root| room_bytes_cliff| room_msgs_cliff| body_bytes_max|
35// A missing conf row REFUSES BY NAME -- no silent code default. FAIL-CLOSED: appending to an
36// unopened room is REFUSED (a typo must not conjure a plane -- the nx_plane_append law); open of an
37// existing room is REFUSED (create-only, idempotent-safe, rule 10); every refusal ANNOUNCES its
38// reason and writes NOTHING.
39// license_tier: ORIGINAL No hw writes (Rule 26).
40import "nx_syscalls.nx"
41import "nx_itoa_lib.nx"
42import "nx_store_seed_lib.nx"
43import "nx_seg_store.nx"
44import "nx_chatmsg.nx"
45
46const CS_EXIT_OK: i64 = 0
47const CS_EXIT_USAGE: i64 = 2
48const CS_EXIT_REFUSED: i64 = 3
49const CS_EXIT_RETRY: i64 = 4
50const CS_EXIT_CORRUPT: i64 = 5
51// structural format math, not policy (policy is conf-only):
52const CS_NAME_MAX: i64 = 64 // room/member/sender token cap; keeps cur:<member> inside key caps
53const CS_PFX_CAP: i64 = 256 // store_root + chat_ + room + - + NUL
54const CS_ROW_OVH: i64 = 128 // row framing around the hex body
55const CS_WSLACK: i64 = 65536 // writer slack, same margin the seed lib uses (STS_WSLACK)
56const CS_CONF_CAP: i64 = 192 // one conf value
57const CS_KIND_MIN: i64 = 1 // NX_CHAT_KIND_TEXT (nx_chat_ring sealed kinds, by citation)
58const CS_KIND_MAX: i64 = 6 // NX_CHAT_KIND_HANDRAISE (== NX_CHAT_KIND_N - 1)
59const CS_KIND_SYS: i64 = 4 // NX_CHAT_KIND_SYS -- the room-open system row
60const CS_NL: i64 = 10
61const CS_PIPE: i64 = 124
62const CS_LOW_M: i64 = 109 // ASCII m -- the row marker
63
64func cs_w(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(1, s, n); return 0 }
65func cs_wn(s: *u8, n: i64) -> i64 { sys_write(1, s, n); return 0 }
66func cs_n(v: i64) -> i64 { nxi_out(v); return 0 }
67func cs_len(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
68func cs_eq(a: *u8, b: *u8) -> i64 {
69 var i: i64 = 0
70 while a[i] != (0 as u8) { if a[i] != b[i] { return 0 } i = i + 1 }
71 if b[i] != (0 as u8) { return 0 }
72 return 1
73}
74// tokens that land inside keys and pipe-framed rows: ASCII letters, digits, underscore; 1..CS_NAME_MAX
75func cs_tok_ok(s: *u8) -> i64 {
76 var i: i64 = 0
77 while s[i] != (0 as u8) {
78 let c: i64 = s[i] as i64
79 var ok: i64 = 0
80 if c >= 48 { if c <= 57 { ok = 1 } }
81 if c >= 65 { if c <= 90 { ok = 1 } }
82 if c >= 97 { if c <= 122 { ok = 1 } }
83 if c == 95 { ok = 1 }
84 if ok == 0 { return 0 }
85 i = i + 1
86 }
87 if i < 1 { return 0 }
88 if i > CS_NAME_MAX { return 0 }
89 return 1
90}
91// lowercase hex armor -- encode n bytes into 2n chars; returns 2n
92func cs_hexe(src: *u8, n: i64, dst: *u8) -> i64 {
93 var i: i64 = 0
94 while i < n {
95 let b: i64 = src[i] as i64
96 let hi: i64 = b / 16
97 let lo: i64 = b % 16
98 if hi < 10 { dst[i*2] = (48 + hi) as u8 } else { dst[i*2] = (87 + hi) as u8 }
99 if lo < 10 { dst[i*2+1] = (48 + lo) as u8 } else { dst[i*2+1] = (87 + lo) as u8 }
100 i = i + 1
101 }
102 return n * 2
103}
104func cs_hexv(c: i64) -> i64 {
105 if c >= 48 { if c <= 57 { return c - 48 } }
106 if c >= 97 { if c <= 102 { return c - 87 } }
107 if c >= 65 { if c <= 70 { return c - 55 } }
108 return 0 - 1
109}
110// decode 2n chars into n bytes; -1 on odd length or a non-hex byte (refuse, never guess)
111func cs_hexd(src: *u8, n: i64, dst: *u8) -> i64 {
112 if n % 2 != 0 { return 0 - 1 }
113 var i: i64 = 0
114 while i < n {
115 let h: i64 = cs_hexv(src[i] as i64)
116 let l: i64 = cs_hexv(src[i+1] as i64)
117 if h < 0 { return 0 - 1 }
118 if l < 0 { return 0 - 1 }
119 dst[i/2] = (h * 16 + l) as u8
120 i = i + 2
121 }
122 return n / 2
123}
124// conf reader: find <key>| at a line start, copy the value to out; -1 absent/unreadable. Lines
125// starting with a comment byte are skipped by construction (they cannot match a key).
126func cs_confs(confpath: *u8, key: *u8, out: *u8, cap: i64) -> i64 {
127 let ol: *i64 = sts_mm(16) as *i64
128 let b: *u8 = sys_read_file(confpath, ol)
129 let n: i64 = ol[0]
130 if b as i64 == 0 { return 0 - 1 }
131 if n <= 0 { return 0 - 1 }
132 let kl: i64 = cs_len(key)
133 var i: i64 = 0
134 while i < n {
135 var j: i64 = 0
136 var hit: i64 = 1
137 while j < kl {
138 if i + j >= n { hit = 0; break }
139 if b[i+j] != key[j] { hit = 0; break }
140 j = j + 1
141 }
142 if hit == 1 {
143 if i + kl >= n { hit = 0 } else { if b[i+kl] != (CS_PIPE as u8) { hit = 0 } }
144 }
145 if hit == 1 {
146 var v: i64 = i + kl + 1
147 var o: i64 = 0
148 while v < n {
149 if b[v] == (CS_NL as u8) { break }
150 if o < cap - 1 { out[o] = b[v]; o = o + 1 }
151 v = v + 1
152 }
153 out[o] = 0 as u8
154 return o
155 }
156 while i < n { if b[i] == (CS_NL as u8) { break } i = i + 1 }
157 i = i + 1
158 }
159 return 0 - 1
160}
161func cs_confn(confpath: *u8, key: *u8) -> i64 {
162 let v: *u8 = sts_mm(CS_CONF_CAP)
163 let l: i64 = cs_confs(confpath, key, v, CS_CONF_CAP)
164 if l <= 0 { return 0 - 1 }
165 return sts_atoi(v, l)
166}
167func cs_refuse_conf(key: *u8, confpath: *u8) -> i64 {
168 cs_w("CHAT-REFUSED conf-missing row=" as *u8)
169 cs_w(key)
170 cs_w(" conf=" as *u8)
171 cs_w(confpath)
172 cs_w(" -- every budget lives in the conf (rule 11); add the row deliberately, nothing was written\n" as *u8)
173 return CS_EXIT_REFUSED
174}
175// plane prefix for a room: <store_root>chat_<room>- returns len, -1 bad room token, -2 conf missing
176func cs_prefix(confpath: *u8, room: *u8, out: *u8) -> i64 {
177 if cs_tok_ok(room) == 0 { return 0 - 1 }
178 let rl: i64 = cs_confs(confpath, "store_root" as *u8, out, CS_PFX_CAP - CS_NAME_MAX - 8)
179 if rl <= 0 { return 0 - 2 }
180 var o: i64 = rl
181 o = ss_cat(out, o, "chat_" as *u8)
182 o = ss_cat(out, o, room)
183 o = ss_cat(out, o, "-" as *u8)
184 out[o] = 0 as u8
185 return o
186}
187// one-key write (caller holds the plane lock). CAS-any is safe under the held lock.
188func cs_put_key(prefix: *u8, key: *u8, val: *u8, vlen: i64) -> i64 {
189 let w: *i64 = ss_begin_cap(vlen + CS_WSLACK)
190 if ss_add(w, 1, key, val, vlen) < 0 { return 0 - 1 }
191 if ss_commit_cas(prefix, w, ss_next_segid(prefix), SS_CAS_ANY) != 0 { return 0 - 1 }
192 return 0
193}
194func cs_curkey(member: *u8, out: *u8) -> i64 {
195 var o: i64 = ss_cat(out, 0, "cur:" as *u8)
196 o = ss_cat(out, o, member)
197 out[o] = 0 as u8
198 return o
199}
200func cs_delkey(seq: i64, out: *u8) -> i64 {
201 var o: i64 = ss_cat(out, 0, "del:" as *u8)
202 o = ss_catn(out, o, seq)
203 out[o] = 0 as u8
204 return o
205}
206// parse the embedded seq out of a row value m|<seq>|... ; -1 if the shape is wrong
207func cs_rowseq(p: *u8, l: i64) -> i64 {
208 if l < 4 { return 0 - 1 }
209 if p[0] != (CS_LOW_M as u8) { return 0 - 1 }
210 if p[1] != (CS_PIPE as u8) { return 0 - 1 }
211 var v: i64 = 0
212 var any: i64 = 0
213 var i: i64 = 2
214 while i < l {
215 let c: i64 = p[i] as i64
216 if c < 48 { break }
217 if c > 57 { break }
218 v = v * 10 + (c - 48)
219 any = 1
220 i = i + 1
221 }
222 if any == 0 { return 0 - 1 }
223 return v
224}
225// build a message row m|<seq>|<epoch_ms>|<kind>|<sender>|<hexbody> ; returns length (no trailing NL)
226func cs_rowbuild(seq: i64, kind: i64, sender: *u8, body: *u8, blen: i64, out: *u8) -> i64 {
227 var o: i64 = ss_cat(out, 0, "m|" as *u8)
228 o = ss_catn(out, o, seq)
229 o = ss_cat(out, o, "|" as *u8)
230 o = ss_catn(out, o, sys_now_realtime_ms())
231 o = ss_cat(out, o, "|" as *u8)
232 o = ss_catn(out, o, kind)
233 o = ss_cat(out, o, "|" as *u8)
234 o = ss_cat(out, o, sender)
235 o = ss_cat(out, o, "|" as *u8)
236 o = o + cs_hexe(body, blen, (out as i64 + o) as *u8)
237 out[o] = 0 as u8
238 return o
239}
240
241// open = CREATE-ONLY. Seeds row 1 (a system room-open message) + the meta:bytes counter.
242func cs_open(room: *u8, opener: *u8, confpath: *u8) -> i64 {
243 if cs_tok_ok(opener) == 0 { cs_w("CHAT-REFUSED bad-token opener -- letters digits underscore, 1..64\n" as *u8) return CS_EXIT_REFUSED }
244 let pfx: *u8 = sts_mm(CS_PFX_CAP)
245 let pr: i64 = cs_prefix(confpath, room, pfx)
246 if pr == (0 - 1) { cs_w("CHAT-REFUSED bad-token room -- letters digits underscore, 1..64\n" as *u8) return CS_EXIT_REFUSED }
247 if pr == (0 - 2) { return cs_refuse_conf("store_root" as *u8, confpath) }
248 let lk: i64 = sts_lock(pfx)
249 if lk < 0 { cs_w("CHAT-RETRY cannot-lock -- try again\n" as *u8) return CS_EXIT_RETRY }
250 let pq: *i64 = sts_mm(16) as *i64
251 let lq: *i64 = sts_mm(16) as *i64
252 if ss_get(pfx, "q:n" as *u8, pq, lq) == 1 {
253 sts_unlock(lk)
254 cs_w("CHAT-REFUSED room-exists room=" as *u8)
255 cs_w(room)
256 cs_w(" -- open is create-only (idempotent-safe); append to it instead\n" as *u8)
257 return CS_EXIT_REFUSED
258 }
259 let body: *u8 = "room-open" as *u8
260 let bl: i64 = cs_len(body)
261 let row: *u8 = sts_mm(CS_ROW_OVH + CS_NAME_MAX + bl * 2 + 8)
262 var o: i64 = cs_rowbuild(1, CS_KIND_SYS, opener, body, bl, row)
263 row[o] = CS_NL as u8
264 o = o + 1
265 let rows: i64 = sts_seed(pfx, row, o)
266 if rows != 1 {
267 sts_unlock(lk)
268 cs_w("CHAT-RED open-seed-failed rows=" as *u8)
269 cs_n(rows)
270 cs_w("\n" as *u8)
271 return CS_EXIT_CORRUPT
272 }
273 let nb: *u8 = sts_mm(32)
274 let nl2: i64 = ss_catn(nb, 0, bl)
275 if cs_put_key(pfx, "meta:bytes" as *u8, nb, nl2) != 0 {
276 sts_unlock(lk)
277 cs_w("CHAT-RED open-meta-failed\n" as *u8)
278 return CS_EXIT_CORRUPT
279 }
280 sts_unlock(lk)
281 cs_w("CHAT-OPEN-OK room=" as *u8)
282 cs_w(room)
283 cs_w(" seq=1 bytes=" as *u8)
284 cs_n(bl)
285 cs_w(" store=" as *u8)
286 cs_w(pfx)
287 cs_w("\n" as *u8)
288 return CS_EXIT_OK
289}
290
291// THE C1 CONTRACT SYMBOL. Append one message to a room's durable log under the plane lock, budgets
292// from conf, CAS-committed: the row is on disk before anyone is told it exists, so an offline
293// member's later fetch-since-cursor replays it byte-identically. Returns the assigned seq (> 0) or
294// the negated exit code, already announced.
295func cs_store_forward(room: *u8, sender: *u8, kind: i64, body: *u8, blen: i64, confpath: *u8) -> i64 {
296 if cs_tok_ok(sender) == 0 { cs_w("CHAT-REFUSED bad-token sender -- letters digits underscore, 1..64\n" as *u8) return 0 - CS_EXIT_REFUSED }
297 if kind < CS_KIND_MIN {
298 cs_w("CHAT-REFUSED bad-kind kind=" as *u8)
299 cs_n(kind)
300 cs_w(" allowed=1..6 (nx_chat_ring sealed kinds)\n" as *u8)
301 return 0 - CS_EXIT_REFUSED
302 }
303 if kind > CS_KIND_MAX {
304 cs_w("CHAT-REFUSED bad-kind kind=" as *u8)
305 cs_n(kind)
306 cs_w(" allowed=1..6 (nx_chat_ring sealed kinds)\n" as *u8)
307 return 0 - CS_EXIT_REFUSED
308 }
309 let bmax: i64 = cs_confn(confpath, "body_bytes_max" as *u8)
310 if bmax < 0 { cs_refuse_conf("body_bytes_max" as *u8, confpath) return 0 - CS_EXIT_REFUSED }
311 let bcliff: i64 = cs_confn(confpath, "room_bytes_cliff" as *u8)
312 if bcliff < 0 { cs_refuse_conf("room_bytes_cliff" as *u8, confpath) return 0 - CS_EXIT_REFUSED }
313 let mcliff: i64 = cs_confn(confpath, "room_msgs_cliff" as *u8)
314 if mcliff < 0 { cs_refuse_conf("room_msgs_cliff" as *u8, confpath) return 0 - CS_EXIT_REFUSED }
315 if blen > bmax {
316 cs_w("CHAT-REFUSED body-too-big bytes=" as *u8)
317 cs_n(blen)
318 cs_w(" conf_row=body_bytes_max cliff=" as *u8)
319 cs_n(bmax)
320 cs_w(" -- nothing written\n" as *u8)
321 return 0 - CS_EXIT_REFUSED
322 }
323 let pfx: *u8 = sts_mm(CS_PFX_CAP)
324 let pr: i64 = cs_prefix(confpath, room, pfx)
325 if pr == (0 - 1) { cs_w("CHAT-REFUSED bad-token room -- letters digits underscore, 1..64\n" as *u8) return 0 - CS_EXIT_REFUSED }
326 if pr == (0 - 2) { cs_refuse_conf("store_root" as *u8, confpath) return 0 - CS_EXIT_REFUSED }
327 let lk: i64 = sts_lock(pfx)
328 if lk < 0 { cs_w("CHAT-RETRY cannot-lock -- try again\n" as *u8) return 0 - CS_EXIT_RETRY }
329 let pq: *i64 = sts_mm(16) as *i64
330 let lq: *i64 = sts_mm(16) as *i64
331 if ss_get(pfx, "q:n" as *u8, pq, lq) != 1 {
332 sts_unlock(lk)
333 cs_w("CHAT-REFUSED room-absent room=" as *u8)
334 cs_w(room)
335 cs_w(" -- open it deliberately first: nx_chat_store open <room> <opener> (a typo must not conjure a plane)\n" as *u8)
336 return 0 - CS_EXIT_REFUSED
337 }
338 let n: i64 = sts_atoi(pq[0] as *u8, lq[0])
339 if n >= mcliff {
340 sts_unlock(lk)
341 cs_w("CHAT-REFUSED budget-msgs msgs=" as *u8)
342 cs_n(n)
343 cs_w(" conf_row=room_msgs_cliff cliff=" as *u8)
344 cs_n(mcliff)
345 cs_w(" -- REFUSING loudly instead of silently dropping; raise the conf row deliberately or open a new room\n" as *u8)
346 return 0 - CS_EXIT_REFUSED
347 }
348 var bts: i64 = 0
349 if ss_get(pfx, "meta:bytes" as *u8, pq, lq) == 1 { bts = sts_atoi(pq[0] as *u8, lq[0]) } else { cs_w("CHAT-NOTE meta-bytes-absent -- counting from zero (foreign plane?)\n" as *u8) }
350 if bts + blen > bcliff {
351 sts_unlock(lk)
352 cs_w("CHAT-REFUSED budget-bytes bytes_total=" as *u8)
353 cs_n(bts)
354 cs_w(" body=" as *u8)
355 cs_n(blen)
356 cs_w(" conf_row=room_bytes_cliff cliff=" as *u8)
357 cs_n(bcliff)
358 cs_w(" -- nothing written\n" as *u8)
359 return 0 - CS_EXIT_REFUSED
360 }
361 let seq: i64 = n + 1
362 let row: *u8 = sts_mm(CS_ROW_OVH + CS_NAME_MAX + blen * 2 + 8)
363 let rl: i64 = cs_rowbuild(seq, kind, sender, body, blen, row)
364 let gen0: i64 = ss_max_segid(pfx)
365 let key: *u8 = sts_mm(64)
366 sts_rowkey(n, key)
367 let cb: *u8 = sts_mm(32)
368 let cl: i64 = ss_catn(cb, 0, seq)
369 let mb: *u8 = sts_mm(32)
370 let ml: i64 = ss_catn(mb, 0, bts + blen)
371 let w: *i64 = ss_begin_cap(rl + CS_WSLACK)
372 if ss_add(w, 1, key, row, rl) < 0 { sts_unlock(lk) cs_w("CHAT-RED writer-add-row-failed\n" as *u8) return 0 - CS_EXIT_CORRUPT }
373 if ss_add(w, 1, "q:n" as *u8, cb, cl) < 0 { sts_unlock(lk) cs_w("CHAT-RED writer-add-count-failed\n" as *u8) return 0 - CS_EXIT_CORRUPT }
374 if ss_add(w, 1, "meta:bytes" as *u8, mb, ml) < 0 { sts_unlock(lk) cs_w("CHAT-RED writer-add-meta-failed\n" as *u8) return 0 - CS_EXIT_CORRUPT }
375 let rc: i64 = ss_commit_cas(pfx, w, ss_next_segid(pfx), gen0)
376 sts_unlock(lk)
377 if rc == SS_ERR_STALE {
378 cs_w("CHAT-RETRY stale-generation -- another writer committed inside this append; re-run\n" as *u8)
379 return 0 - CS_EXIT_RETRY
380 }
381 if rc != 0 { cs_w("CHAT-RED commit-failed\n" as *u8) return 0 - CS_EXIT_CORRUPT }
382 cs_w("CHAT-APPEND-OK room=" as *u8)
383 cs_w(room)
384 cs_w(" seq=" as *u8)
385 cs_n(seq)
386 cs_w(" bytes_body=" as *u8)
387 cs_n(blen)
388 cs_w(" bytes_total=" as *u8)
389 cs_n(bts + blen)
390 cs_w(" bytes_cliff=" as *u8)
391 cs_n(bcliff)
392 cs_w(" msgs=" as *u8)
393 cs_n(seq)
394 cs_w(" msgs_cliff=" as *u8)
395 cs_n(mcliff)
396 cs_w("\n" as *u8)
397 return seq
398}
399
400// fetch rows AFTER since (seq > since), capped at maxn returned; tombstoned rows are hidden and
401// counted. Prints the partition and CHECKS it -- a partition that does not sum exits CORRUPT.
402func cs_fetch(room: *u8, since: i64, maxn: i64, confpath: *u8) -> i64 {
403 if since < 0 { cs_w("CHAT-USAGE since must be >= 0\n" as *u8) return CS_EXIT_USAGE }
404 if maxn <= 0 { cs_w("CHAT-USAGE max must be > 0\n" as *u8) return CS_EXIT_USAGE }
405 let pfx: *u8 = sts_mm(CS_PFX_CAP)
406 let pr: i64 = cs_prefix(confpath, room, pfx)
407 if pr == (0 - 1) { cs_w("CHAT-REFUSED bad-token room -- letters digits underscore, 1..64\n" as *u8) return CS_EXIT_REFUSED }
408 if pr == (0 - 2) { return cs_refuse_conf("store_root" as *u8, confpath) }
409 let pq: *i64 = sts_mm(16) as *i64
410 let lq: *i64 = sts_mm(16) as *i64
411 if ss_get(pfx, "q:n" as *u8, pq, lq) != 1 {
412 cs_w("CHAT-REFUSED room-absent room=" as *u8)
413 cs_w(room)
414 cs_w(" -- absent is not empty; open it first\n" as *u8)
415 return CS_EXIT_REFUSED
416 }
417 let n: i64 = sts_atoi(pq[0] as *u8, lq[0])
418 let h: *i64 = ss_open_cached(pfx)
419 if h as i64 == 0 { cs_w("CHAT-RETRY open-failed\n" as *u8) return CS_EXIT_RETRY }
420 let key: *u8 = sts_mm(64)
421 let dk: *u8 = sts_mm(64)
422 let pr2: *i64 = sts_mm(16) as *i64
423 let lr2: *i64 = sts_mm(16) as *i64
424 var returned: i64 = 0
425 var hidden: i64 = 0
426 var i: i64 = since
427 while i < n {
428 if returned >= maxn { break }
429 sts_rowkey(i, key)
430 if ss_hget(h, key, pq, lq) != 1 {
431 cs_w("CHAT-CORRUPT row-gap idx=" as *u8)
432 cs_n(i)
433 cs_w(" -- the count declares a row the store cannot produce\n" as *u8)
434 return CS_EXIT_CORRUPT
435 }
436 let rv: *u8 = pq[0] as *u8
437 let rl: i64 = lq[0]
438 let es: i64 = cs_rowseq(rv, rl)
439 if es != i + 1 {
440 cs_w("CHAT-CORRUPT seq-mismatch idx=" as *u8)
441 cs_n(i)
442 cs_w(" embedded=" as *u8)
443 cs_n(es)
444 cs_w(" -- ordinal and embedded seq disagree\n" as *u8)
445 return CS_EXIT_CORRUPT
446 }
447 cs_delkey(i + 1, dk)
448 if ss_hget(h, dk, pr2, lr2) == 1 { hidden = hidden + 1 } else {
449 cs_wn(rv, rl)
450 cs_w("\n" as *u8)
451 returned = returned + 1
452 }
453 i = i + 1
454 }
455 let scanned: i64 = i - since
456 let remaining: i64 = n - i
457 cs_w("CHAT-FETCH-OK room=" as *u8)
458 cs_w(room)
459 cs_w(" since=" as *u8)
460 cs_n(since)
461 cs_w(" returned=" as *u8)
462 cs_n(returned)
463 cs_w(" deleted_hidden=" as *u8)
464 cs_n(hidden)
465 cs_w(" remaining=" as *u8)
466 cs_n(remaining)
467 cs_w(" total=" as *u8)
468 cs_n(n)
469 cs_w("\n" as *u8)
470 if returned + hidden != scanned {
471 cs_w("CHAT-CORRUPT partition-broken returned+hidden != scanned\n" as *u8)
472 return CS_EXIT_CORRUPT
473 }
474 return CS_EXIT_OK
475}
476
477func cs_cursor(room: *u8, member: *u8, confpath: *u8) -> i64 {
478 if cs_tok_ok(member) == 0 { cs_w("CHAT-REFUSED bad-token member\n" as *u8) return CS_EXIT_REFUSED }
479 let pfx: *u8 = sts_mm(CS_PFX_CAP)
480 let pr: i64 = cs_prefix(confpath, room, pfx)
481 if pr == (0 - 1) { cs_w("CHAT-REFUSED bad-token room\n" as *u8) return CS_EXIT_REFUSED }
482 if pr == (0 - 2) { return cs_refuse_conf("store_root" as *u8, confpath) }
483 let pq: *i64 = sts_mm(16) as *i64
484 let lq: *i64 = sts_mm(16) as *i64
485 if ss_get(pfx, "q:n" as *u8, pq, lq) != 1 { cs_w("CHAT-REFUSED room-absent\n" as *u8) return CS_EXIT_REFUSED }
486 let ck: *u8 = sts_mm(96)
487 cs_curkey(member, ck)
488 var v: i64 = 0
489 var never: i64 = 1
490 if ss_get(pfx, ck, pq, lq) == 1 { v = sts_atoi(pq[0] as *u8, lq[0]) never = 0 }
491 cs_w("CHAT-CURSOR room=" as *u8)
492 cs_w(room)
493 cs_w(" member=" as *u8)
494 cs_w(member)
495 cs_w(" seq=" as *u8)
496 cs_n(v)
497 if never == 1 { cs_w(" note=never-acked" as *u8) }
498 cs_w("\n" as *u8)
499 return CS_EXIT_OK
500}
501
502func cs_ack(room: *u8, member: *u8, seq: i64, confpath: *u8) -> i64 {
503 if cs_tok_ok(member) == 0 { cs_w("CHAT-REFUSED bad-token member\n" as *u8) return CS_EXIT_REFUSED }
504 if seq < 0 { cs_w("CHAT-USAGE seq must be >= 0\n" as *u8) return CS_EXIT_USAGE }
505 let pfx: *u8 = sts_mm(CS_PFX_CAP)
506 let pr: i64 = cs_prefix(confpath, room, pfx)
507 if pr == (0 - 1) { cs_w("CHAT-REFUSED bad-token room\n" as *u8) return CS_EXIT_REFUSED }
508 if pr == (0 - 2) { return cs_refuse_conf("store_root" as *u8, confpath) }
509 let lk: i64 = sts_lock(pfx)
510 if lk < 0 { cs_w("CHAT-RETRY cannot-lock\n" as *u8) return CS_EXIT_RETRY }
511 let pq: *i64 = sts_mm(16) as *i64
512 let lq: *i64 = sts_mm(16) as *i64
513 if ss_get(pfx, "q:n" as *u8, pq, lq) != 1 { sts_unlock(lk) cs_w("CHAT-REFUSED room-absent\n" as *u8) return CS_EXIT_REFUSED }
514 let n: i64 = sts_atoi(pq[0] as *u8, lq[0])
515 if seq > n {
516 sts_unlock(lk)
517 cs_w("CHAT-REFUSED ack-beyond-history seq=" as *u8)
518 cs_n(seq)
519 cs_w(" total=" as *u8)
520 cs_n(n)
521 cs_w("\n" as *u8)
522 return CS_EXIT_REFUSED
523 }
524 let ck: *u8 = sts_mm(96)
525 cs_curkey(member, ck)
526 var old: i64 = 0
527 if ss_get(pfx, ck, pq, lq) == 1 { old = sts_atoi(pq[0] as *u8, lq[0]) }
528 // the nx_chatmsg ladder: a late or duplicate ack can never regress the cursor
529 let newv: i64 = cm_advance(old, seq)
530 if newv != old {
531 let vb: *u8 = sts_mm(32)
532 let vl: i64 = ss_catn(vb, 0, newv)
533 if cs_put_key(pfx, ck, vb, vl) != 0 { sts_unlock(lk) cs_w("CHAT-RED cursor-write-failed\n" as *u8) return CS_EXIT_CORRUPT }
534 }
535 sts_unlock(lk)
536 cs_w("CHAT-ACK-OK room=" as *u8)
537 cs_w(room)
538 cs_w(" member=" as *u8)
539 cs_w(member)
540 cs_w(" old=" as *u8)
541 cs_n(old)
542 cs_w(" new=" as *u8)
543 cs_n(newv)
544 cs_w(" advanced=" as *u8)
545 if newv != old { cs_n(1) } else { cs_n(0) }
546 cs_w("\n" as *u8)
547 return CS_EXIT_OK
548}
549
550func cs_del(room: *u8, seq: i64, actor: *u8, confpath: *u8) -> i64 {
551 if cs_tok_ok(actor) == 0 { cs_w("CHAT-REFUSED bad-token actor\n" as *u8) return CS_EXIT_REFUSED }
552 let pfx: *u8 = sts_mm(CS_PFX_CAP)
553 let pr: i64 = cs_prefix(confpath, room, pfx)
554 if pr == (0 - 1) { cs_w("CHAT-REFUSED bad-token room\n" as *u8) return CS_EXIT_REFUSED }
555 if pr == (0 - 2) { return cs_refuse_conf("store_root" as *u8, confpath) }
556 let lk: i64 = sts_lock(pfx)
557 if lk < 0 { cs_w("CHAT-RETRY cannot-lock\n" as *u8) return CS_EXIT_RETRY }
558 let pq: *i64 = sts_mm(16) as *i64
559 let lq: *i64 = sts_mm(16) as *i64
560 if ss_get(pfx, "q:n" as *u8, pq, lq) != 1 { sts_unlock(lk) cs_w("CHAT-REFUSED room-absent\n" as *u8) return CS_EXIT_REFUSED }
561 let n: i64 = sts_atoi(pq[0] as *u8, lq[0])
562 if seq < 1 { sts_unlock(lk) cs_w("CHAT-REFUSED bad-seq\n" as *u8) return CS_EXIT_REFUSED }
563 if seq > n { sts_unlock(lk) cs_w("CHAT-REFUSED bad-seq beyond total\n" as *u8) return CS_EXIT_REFUSED }
564 let dk: *u8 = sts_mm(64)
565 cs_delkey(seq, dk)
566 if ss_get(pfx, dk, pq, lq) == 1 {
567 sts_unlock(lk)
568 cs_w("CHAT-DEL-OK room=" as *u8)
569 cs_w(room)
570 cs_w(" seq=" as *u8)
571 cs_n(seq)
572 cs_w(" already=1 -- idempotent, nothing changed\n" as *u8)
573 return CS_EXIT_OK
574 }
575 if cs_put_key(pfx, dk, "1" as *u8, 1) != 0 { sts_unlock(lk) cs_w("CHAT-RED tombstone-write-failed\n" as *u8) return CS_EXIT_CORRUPT }
576 sts_unlock(lk)
577 cs_w("CHAT-DEL-OK room=" as *u8)
578 cs_w(room)
579 cs_w(" seq=" as *u8)
580 cs_n(seq)
581 cs_w(" soft=1 by=" as *u8)
582 cs_w(actor)
583 cs_w(" -- bytes retained, additive-only (rule 13)\n" as *u8)
584 return CS_EXIT_OK
585}
586
587func cs_status(room: *u8, confpath: *u8) -> i64 {
588 let pfx: *u8 = sts_mm(CS_PFX_CAP)
589 let pr: i64 = cs_prefix(confpath, room, pfx)
590 if pr == (0 - 1) { cs_w("CHAT-REFUSED bad-token room\n" as *u8) return CS_EXIT_REFUSED }
591 if pr == (0 - 2) { return cs_refuse_conf("store_root" as *u8, confpath) }
592 let bcliff: i64 = cs_confn(confpath, "room_bytes_cliff" as *u8)
593 if bcliff < 0 { return cs_refuse_conf("room_bytes_cliff" as *u8, confpath) }
594 let mcliff: i64 = cs_confn(confpath, "room_msgs_cliff" as *u8)
595 if mcliff < 0 { return cs_refuse_conf("room_msgs_cliff" as *u8, confpath) }
596 let pq: *i64 = sts_mm(16) as *i64
597 let lq: *i64 = sts_mm(16) as *i64
598 if ss_get(pfx, "q:n" as *u8, pq, lq) != 1 { cs_w("CHAT-REFUSED room-absent -- absent is not empty\n" as *u8) return CS_EXIT_REFUSED }
599 let n: i64 = sts_atoi(pq[0] as *u8, lq[0])
600 var bts: i64 = 0
601 var metaok: i64 = 1
602 if ss_get(pfx, "meta:bytes" as *u8, pq, lq) == 1 { bts = sts_atoi(pq[0] as *u8, lq[0]) } else { metaok = 0 }
603 let h: *i64 = ss_open_cached(pfx)
604 if h as i64 == 0 { cs_w("CHAT-RETRY open-failed\n" as *u8) return CS_EXIT_RETRY }
605 let dk: *u8 = sts_mm(64)
606 let pr2: *i64 = sts_mm(16) as *i64
607 let lr2: *i64 = sts_mm(16) as *i64
608 var deleted: i64 = 0
609 var i: i64 = 0
610 while i < n {
611 cs_delkey(i + 1, dk)
612 if ss_hget(h, dk, pr2, lr2) == 1 { deleted = deleted + 1 }
613 i = i + 1
614 }
615 let visible: i64 = n - deleted
616 cs_w("CHAT-STATUS room=" as *u8)
617 cs_w(room)
618 cs_w(" total=" as *u8)
619 cs_n(n)
620 cs_w(" visible=" as *u8)
621 cs_n(visible)
622 cs_w(" deleted=" as *u8)
623 cs_n(deleted)
624 cs_w(" bytes=" as *u8)
625 cs_n(bts)
626 cs_w(" bytes_cliff=" as *u8)
627 cs_n(bcliff)
628 cs_w(" headroom_bytes=" as *u8)
629 cs_n(bcliff - bts)
630 cs_w(" msgs_cliff=" as *u8)
631 cs_n(mcliff)
632 cs_w(" headroom_msgs=" as *u8)
633 cs_n(mcliff - n)
634 if metaok == 0 { cs_w(" note=meta-bytes-absent" as *u8) }
635 cs_w("\n" as *u8)
636 cs_w("partition visible+deleted=total " as *u8)
637 if visible + deleted == n { cs_w("ok\n" as *u8) } else { cs_w("BROKEN\n" as *u8) return CS_EXIT_CORRUPT }
638 return CS_EXIT_OK
639}
640
641// set a room's retention window. ms == 0 is the explicit keep-forever pin; a positive value ages
642// messages out at the next sweep. Policy is PER-ROOM DATA (a plane key), defaulted by the conf row
643// retention_default_ms -- never a literal.
644func cs_setretention(room: *u8, ms: i64, actor: *u8, confpath: *u8) -> i64 {
645 if cs_tok_ok(actor) == 0 { cs_w("CHAT-REFUSED bad-token actor\n" as *u8) return CS_EXIT_REFUSED }
646 if ms < 0 { cs_w("CHAT-USAGE retention ms must be >= 0 (0 = keep forever)\n" as *u8) return CS_EXIT_USAGE }
647 let pfx: *u8 = sts_mm(CS_PFX_CAP)
648 let pr: i64 = cs_prefix(confpath, room, pfx)
649 if pr == (0 - 1) { cs_w("CHAT-REFUSED bad-token room\n" as *u8) return CS_EXIT_REFUSED }
650 if pr == (0 - 2) { return cs_refuse_conf("store_root" as *u8, confpath) }
651 let lk: i64 = sts_lock(pfx)
652 if lk < 0 { cs_w("CHAT-RETRY cannot-lock\n" as *u8) return CS_EXIT_RETRY }
653 let pq: *i64 = sts_mm(16) as *i64
654 let lq: *i64 = sts_mm(16) as *i64
655 if ss_get(pfx, "q:n" as *u8, pq, lq) != 1 { sts_unlock(lk) cs_w("CHAT-REFUSED room-absent\n" as *u8) return CS_EXIT_REFUSED }
656 var old: i64 = 0
657 if ss_get(pfx, "meta:retention_ms" as *u8, pq, lq) == 1 { old = sts_atoi(pq[0] as *u8, lq[0]) }
658 let vb: *u8 = sts_mm(32)
659 let vl: i64 = ss_catn(vb, 0, ms)
660 if cs_put_key(pfx, "meta:retention_ms" as *u8, vb, vl) != 0 { sts_unlock(lk) cs_w("CHAT-RED retention-write-failed\n" as *u8) return CS_EXIT_CORRUPT }
661 sts_unlock(lk)
662 cs_w("CHAT-RETENTION-SET room=" as *u8)
663 cs_w(room)
664 cs_w(" old_ms=" as *u8)
665 cs_n(old)
666 cs_w(" new_ms=" as *u8)
667 cs_n(ms)
668 cs_w(" by=" as *u8)
669 cs_w(actor)
670 cs_w(" (0 means keep-forever)\n" as *u8)
671 return CS_EXIT_OK
672}
673
674// THE C13 CONTRACT SYMBOL. Retention sweep: tombstone every visible message older than the room's
675// window. ADDITIVE-ONLY by design (rule 13) -- expiry APPENDS del:<seq> tombstones in ONE commit,
676// announces the full partition, and touches no bytes. Hiding and destroying are DIFFERENT PROMISES:
677// byte-erasure exists only as the separate explicit-confirm purge rung named in comms.plan, so the
678// disappearing-messages claim can never quietly overstate itself. Policy: the room's
679// meta:retention_ms if set, else the conf row retention_default_ms; 0 = keep forever, announced.
680func cs_retention(room: *u8, confpath: *u8) -> i64 {
681 let pfx: *u8 = sts_mm(CS_PFX_CAP)
682 let pr: i64 = cs_prefix(confpath, room, pfx)
683 if pr == (0 - 1) { cs_w("CHAT-REFUSED bad-token room\n" as *u8) return CS_EXIT_REFUSED }
684 if pr == (0 - 2) { return cs_refuse_conf("store_root" as *u8, confpath) }
685 let lk: i64 = sts_lock(pfx)
686 if lk < 0 { cs_w("CHAT-RETRY cannot-lock\n" as *u8) return CS_EXIT_RETRY }
687 let pq: *i64 = sts_mm(16) as *i64
688 let lq: *i64 = sts_mm(16) as *i64
689 if ss_get(pfx, "q:n" as *u8, pq, lq) != 1 { sts_unlock(lk) cs_w("CHAT-REFUSED room-absent\n" as *u8) return CS_EXIT_REFUSED }
690 let n: i64 = sts_atoi(pq[0] as *u8, lq[0])
691 var ret: i64 = 0 - 1
692 if ss_get(pfx, "meta:retention_ms" as *u8, pq, lq) == 1 { ret = sts_atoi(pq[0] as *u8, lq[0]) }
693 if ret < 0 {
694 ret = cs_confn(confpath, "retention_default_ms" as *u8)
695 if ret < 0 { sts_unlock(lk) return cs_refuse_conf("retention_default_ms" as *u8, confpath) }
696 }
697 if ret == 0 {
698 sts_unlock(lk)
699 cs_w("CHAT-RETAIN-OK room=" as *u8)
700 cs_w(room)
701 cs_w(" policy=keep-forever expired_now=0 already_hidden=0 kept=" as *u8)
702 cs_n(n)
703 cs_w(" scanned=" as *u8)
704 cs_n(n)
705 cs_w(" sums=ok\n" as *u8)
706 return CS_EXIT_OK
707 }
708 let now: i64 = sys_now_realtime_ms()
709 let h: *i64 = ss_open_cached(pfx)
710 if h as i64 == 0 { sts_unlock(lk) cs_w("CHAT-RETRY open-failed\n" as *u8) return CS_EXIT_RETRY }
711 let key: *u8 = sts_mm(64)
712 let dk: *u8 = sts_mm(64)
713 let pr2: *i64 = sts_mm(16) as *i64
714 let lr2: *i64 = sts_mm(16) as *i64
715 // pass 1: measure (full population, no cap -- the writer is sized from the count)
716 var expire: i64 = 0
717 var already: i64 = 0
718 var kept: i64 = 0
719 var i: i64 = 0
720 while i < n {
721 sts_rowkey(i, key)
722 if ss_hget(h, key, pq, lq) != 1 { sts_unlock(lk) cs_w("CHAT-CORRUPT row-gap in retention scan\n" as *u8) return CS_EXIT_CORRUPT }
723 let rv: *u8 = pq[0] as *u8
724 let rl: i64 = lq[0]
725 // epoch is field 2 of m|seq|epoch|kind|sender|hex
726 var f: i64 = 0
727 var p2: i64 = 0
728 while p2 < rl { if f == 2 { break } if rv[p2] == (CS_PIPE as u8) { f = f + 1 } p2 = p2 + 1 }
729 var ep: i64 = 0
730 while p2 < rl { let c: i64 = rv[p2] as i64 if c < 48 { break } if c > 57 { break } ep = ep * 10 + (c - 48) p2 = p2 + 1 }
731 cs_delkey(i + 1, dk)
732 if ss_hget(h, dk, pr2, lr2) == 1 { already = already + 1 } else {
733 if now - ep > ret { expire = expire + 1 } else { kept = kept + 1 }
734 }
735 i = i + 1
736 }
737 // pass 2: append every new tombstone in ONE commit
738 if expire > 0 {
739 let w: *i64 = ss_begin_cap(expire * 96 + CS_WSLACK)
740 i = 0
741 while i < n {
742 sts_rowkey(i, key)
743 if ss_hget(h, key, pq, lq) != 1 { sts_unlock(lk) cs_w("CHAT-CORRUPT row-gap in retention write\n" as *u8) return CS_EXIT_CORRUPT }
744 let rv2: *u8 = pq[0] as *u8
745 let rl2: i64 = lq[0]
746 var f2: i64 = 0
747 var p3: i64 = 0
748 while p3 < rl2 { if f2 == 2 { break } if rv2[p3] == (CS_PIPE as u8) { f2 = f2 + 1 } p3 = p3 + 1 }
749 var ep2: i64 = 0
750 while p3 < rl2 { let c2: i64 = rv2[p3] as i64 if c2 < 48 { break } if c2 > 57 { break } ep2 = ep2 * 10 + (c2 - 48) p3 = p3 + 1 }
751 cs_delkey(i + 1, dk)
752 var hid: i64 = 0
753 if ss_hget(h, dk, pr2, lr2) == 1 { hid = 1 }
754 if hid == 0 { if now - ep2 > ret {
755 if ss_add(w, 1, dk, "1" as *u8, 1) < 0 { sts_unlock(lk) cs_w("CHAT-RED retention-writer-add-failed\n" as *u8) return CS_EXIT_CORRUPT }
756 } }
757 i = i + 1
758 }
759 if ss_commit_cas(pfx, w, ss_next_segid(pfx), SS_CAS_ANY) != 0 { sts_unlock(lk) cs_w("CHAT-RED retention-commit-failed\n" as *u8) return CS_EXIT_CORRUPT }
760 }
761 sts_unlock(lk)
762 cs_w("CHAT-RETAIN-OK room=" as *u8)
763 cs_w(room)
764 cs_w(" retention_ms=" as *u8)
765 cs_n(ret)
766 cs_w(" expired_now=" as *u8)
767 cs_n(expire)
768 cs_w(" already_hidden=" as *u8)
769 cs_n(already)
770 cs_w(" kept=" as *u8)
771 cs_n(kept)
772 cs_w(" scanned=" as *u8)
773 cs_n(n)
774 cs_w(" " as *u8)
775 if expire + already + kept == n { cs_w("sums=ok -- hidden not erased; bytes retained (additive, rule 13); byte-erasure is the separate confirmed purge rung\n" as *u8) } else { cs_w("sums=BROKEN\n" as *u8) return CS_EXIT_CORRUPT }
776 return CS_EXIT_OK
777}
778
779// length of a row up to and including the 5th pipe -- the structural head (m|seq|epoch|kind|sender|)
780func cs_rowhead(p: *u8, l: i64) -> i64 {
781 var pipes: i64 = 0
782 var i: i64 = 0
783 while i < l {
784 if p[i] == (CS_PIPE as u8) { pipes = pipes + 1 if pipes == 5 { return i + 1 } }
785 i = i + 1
786 }
787 return l
788}
789// THE PURGE RUNG -- explicit-confirm BYTE ERASURE of tombstoned bodies (closes the purge-rung debt).
790// Rule 13: destructive operations require explicit confirmation, and this is THE destructive verb --
791// argv must carry the literal confirm=yes or the verb only DESCRIBES what it would do and refuses.
792// REDACT-IN-PLACE: a tombstoned row keeps seq/epoch/kind/sender (structure, cursors and the
793// seq==ordinal invariant all survive; visible history is byte-identical) and loses its body; the
794// plane is re-seeded, then COMPACTED, then the superseded segment FILES are UNLINKED -- because a
795// purge that leaves the secret in an unreachable-but-on-disk file has not purged. ss_compact takes
796// the plane lock ITSELF, so it is called with OUR lock released (the flock-deadlock audit: same
797// <prefix>plock, different fd, would block forever). If compaction fails the verb announces
798// PURGE-PARTIAL and the re-run -- a partial destruction never reports as a complete one.
799func cs_purge(room: *u8, actor: *u8, confirm: *u8, confpath: *u8) -> i64 {
800 if cs_tok_ok(actor) == 0 { cs_w("CHAT-REFUSED bad-token actor\n" as *u8) return CS_EXIT_REFUSED }
801 let pfx: *u8 = sts_mm(CS_PFX_CAP)
802 let pr: i64 = cs_prefix(confpath, room, pfx)
803 if pr == (0 - 1) { cs_w("CHAT-REFUSED bad-token room\n" as *u8) return CS_EXIT_REFUSED }
804 if pr == (0 - 2) { return cs_refuse_conf("store_root" as *u8, confpath) }
805 if cs_eq(confirm, "confirm=yes" as *u8) == 0 {
806 cs_w("CHAT-REFUSED purge-needs-confirm room=" as *u8)
807 cs_w(room)
808 cs_w(" -- this verb DESTROYS the bodies of tombstoned rows on disk (redact, compact, unlink superseded segments). Nothing was changed; re-run with the literal argument confirm=yes\n" as *u8)
809 return CS_EXIT_REFUSED
810 }
811 let lk: i64 = sts_lock(pfx)
812 if lk < 0 { cs_w("CHAT-RETRY cannot-lock\n" as *u8) return CS_EXIT_RETRY }
813 let pq: *i64 = sts_mm(16) as *i64
814 let lq: *i64 = sts_mm(16) as *i64
815 if ss_get(pfx, "q:n" as *u8, pq, lq) != 1 { sts_unlock(lk) cs_w("CHAT-REFUSED room-absent\n" as *u8) return CS_EXIT_REFUSED }
816 let n: i64 = sts_atoi(pq[0] as *u8, lq[0])
817 let h: *i64 = ss_open_cached(pfx)
818 if h as i64 == 0 { sts_unlock(lk) cs_w("CHAT-RETRY open-failed\n" as *u8) return CS_EXIT_RETRY }
819 let key: *u8 = sts_mm(64)
820 let dk: *u8 = sts_mm(64)
821 let pr2: *i64 = sts_mm(16) as *i64
822 let lr2: *i64 = sts_mm(16) as *i64
823 // pass 1: measure (buffer sized from the data, never a guessed ceiling)
824 var total: i64 = 0
825 var erows: i64 = 0
826 var ebytes: i64 = 0
827 var i: i64 = 0
828 while i < n {
829 sts_rowkey(i, key)
830 if ss_hget(h, key, pq, lq) != 1 { sts_unlock(lk) cs_w("CHAT-CORRUPT row-gap in purge scan\n" as *u8) return CS_EXIT_CORRUPT }
831 let rl: i64 = lq[0]
832 cs_delkey(i + 1, dk)
833 if ss_hget(h, dk, pr2, lr2) == 1 {
834 let hl: i64 = cs_rowhead(pq[0] as *u8, rl)
835 if rl > hl { erows = erows + 1 ebytes = ebytes + (rl - hl) / 2 }
836 total = total + hl + 1
837 } else {
838 total = total + rl + 1
839 }
840 i = i + 1
841 }
842 if erows == 0 {
843 sts_unlock(lk)
844 cs_w("CHAT-PURGE-OK room=" as *u8)
845 cs_w(room)
846 cs_w(" erased_rows=0 erased_bytes=0 -- nothing to purge (idempotent, rule 10)\n" as *u8)
847 return CS_EXIT_OK
848 }
849 var bts: i64 = 0
850 if ss_get(pfx, "meta:bytes" as *u8, pq, lq) == 1 { bts = sts_atoi(pq[0] as *u8, lq[0]) }
851 // pass 2: rebuild the whole plane buffer, redacting tombstoned bodies
852 let buf: *u8 = sts_mm(total + CS_WSLACK)
853 var o: i64 = 0
854 i = 0
855 while i < n {
856 sts_rowkey(i, key)
857 if ss_hget(h, key, pq, lq) != 1 { sts_unlock(lk) cs_w("CHAT-CORRUPT row-gap in purge rebuild\n" as *u8) return CS_EXIT_CORRUPT }
858 let rv: *u8 = pq[0] as *u8
859 let rl2: i64 = lq[0]
860 cs_delkey(i + 1, dk)
861 var keep: i64 = rl2
862 if ss_hget(h, dk, pr2, lr2) == 1 { keep = cs_rowhead(rv, rl2) }
863 var k: i64 = 0
864 while k < keep { buf[o] = rv[k] o = o + 1 k = k + 1 }
865 buf[o] = CS_NL as u8
866 o = o + 1
867 i = i + 1
868 }
869 let rows: i64 = sts_seed(pfx, buf, o)
870 if rows != n {
871 sts_unlock(lk)
872 cs_w("CHAT-RED purge-seed-mismatch expected=" as *u8)
873 cs_n(n)
874 cs_w(" got=" as *u8)
875 cs_n(rows)
876 cs_w("\n" as *u8)
877 return CS_EXIT_CORRUPT
878 }
879 var nb: i64 = bts - ebytes
880 if nb < 0 { nb = 0 }
881 let vb: *u8 = sts_mm(32)
882 let vl: i64 = ss_catn(vb, 0, nb)
883 if cs_put_key(pfx, "meta:bytes" as *u8, vb, vl) != 0 { sts_unlock(lk) cs_w("CHAT-RED purge-meta-failed\n" as *u8) return CS_EXIT_CORRUPT }
884 sts_unlock(lk)
885 // compact UNLOCKED -- ss_compact takes the plane lock itself
886 let newseg: i64 = ss_compact(pfx, ss_next_segid(pfx))
887 if newseg <= 0 {
888 cs_w("CHAT-PURGE-PARTIAL room=" as *u8)
889 cs_w(room)
890 cs_w(" erased_rows=" as *u8)
891 cs_n(erows)
892 cs_w(" -- redaction committed but compaction failed rc=" as *u8)
893 cs_n(newseg)
894 cs_w("; superseded segments still hold bytes, RE-RUN purge\n" as *u8)
895 return CS_EXIT_RETRY
896 }
897 // unlink every superseded segment file: after the manifest swap they are unreachable, and a purge
898 // must not leave the bytes in dead files for a janitor to find later
899 var un: i64 = 0
900 var s: i64 = 0
901 while s < newseg {
902 var e: i64 = 0
903 while e < 4 {
904 let path: *u8 = sts_mm(CS_PFX_CAP + 64)
905 var po: i64 = ss_cat(path, 0, pfx)
906 po = ss_cat(path, po, "seg-" as *u8)
907 po = ss_catn(path, po, s)
908 if e == 0 { po = ss_cat(path, po, ".docs" as *u8) }
909 if e == 1 { po = ss_cat(path, po, ".idx" as *u8) }
910 if e == 2 { po = ss_cat(path, po, ".pos" as *u8) }
911 if e == 3 { po = ss_cat(path, po, ".imp" as *u8) }
912 path[po] = 0 as u8
913 if sys_unlinkat(path) == 0 { un = un + 1 }
914 e = e + 1
915 }
916 s = s + 1
917 }
918 cs_w("CHAT-PURGE-OK room=" as *u8)
919 cs_w(room)
920 cs_w(" erased_rows=" as *u8)
921 cs_n(erows)
922 cs_w(" erased_bytes=" as *u8)
923 cs_n(ebytes)
924 cs_w(" compacted_to_seg=" as *u8)
925 cs_n(newseg)
926 cs_w(" files_unlinked=" as *u8)
927 cs_n(un)
928 cs_w(" -- tombstoned bodies DESTROYED on disk; visible history byte-identical\n" as *u8)
929 return CS_EXIT_OK
930}
931
932func cs_usage() -> i64 {
933 cs_w("usage: nx_chat_store open <room> <opener> [conf] | append <room> <sender> <kind> <body> [conf] | appendhex <room> <sender> <kind> <hex> [conf] | fetch <room> <since> <max> [conf] | cursor <room> <member> [conf] | ack <room> <member> <seq> [conf] | del <room> <seq> <actor> [conf] | status <room> [conf] | setretention <room> <ms> <actor> [conf] | retain <room> [conf] | purge <room> <actor> confirm=yes [conf]\n" as *u8)
934 return CS_EXIT_USAGE
935}
936func cs_confpick(argc: i64, argv: *i64, idx: i64) -> *u8 {
937 if argc > idx { return argv[idx] as *u8 }
938 return "knowledge/comms/chatstore.conf" as *u8
939}
940func cs_argn(argv: *i64, idx: i64) -> i64 {
941 let s: *u8 = argv[idx] as *u8
942 return sts_atoi(s, cs_len(s))
943}
944
945func main(argc: i64, argv: *i64) -> i64 {
946 if argc < 2 { let u: i64 = cs_usage() sys_exit(u) return u }
947 let verb: *u8 = argv[1] as *u8
948 if cs_eq(verb, "open" as *u8) == 1 {
949 if argc < 4 { let u: i64 = cs_usage() sys_exit(u) return u }
950 let rc: i64 = cs_open(argv[2] as *u8, argv[3] as *u8, cs_confpick(argc, argv, 4))
951 sys_exit(rc)
952 return rc
953 }
954 if cs_eq(verb, "append" as *u8) == 1 {
955 if argc < 6 { let u: i64 = cs_usage() sys_exit(u) return u }
956 let body: *u8 = argv[5] as *u8
957 let r: i64 = cs_store_forward(argv[2] as *u8, argv[3] as *u8, cs_argn(argv, 4), body, cs_len(body), cs_confpick(argc, argv, 6))
958 if r > 0 { sys_exit(0) return 0 }
959 sys_exit(0 - r)
960 return 0 - r
961 }
962 if cs_eq(verb, "appendhex" as *u8) == 1 {
963 if argc < 6 { let u: i64 = cs_usage() sys_exit(u) return u }
964 let hx: *u8 = argv[5] as *u8
965 let hl: i64 = cs_len(hx)
966 let dec: *u8 = sts_mm(hl / 2 + 16)
967 let dl: i64 = cs_hexd(hx, hl, dec)
968 if dl < 0 { cs_w("CHAT-REFUSED bad-hex -- even-length lowercase hex only\n" as *u8) sys_exit(CS_EXIT_REFUSED) return CS_EXIT_REFUSED }
969 let r: i64 = cs_store_forward(argv[2] as *u8, argv[3] as *u8, cs_argn(argv, 4), dec, dl, cs_confpick(argc, argv, 6))
970 if r > 0 { sys_exit(0) return 0 }
971 sys_exit(0 - r)
972 return 0 - r
973 }
974 if cs_eq(verb, "fetch" as *u8) == 1 {
975 if argc < 5 { let u: i64 = cs_usage() sys_exit(u) return u }
976 let rc: i64 = cs_fetch(argv[2] as *u8, cs_argn(argv, 3), cs_argn(argv, 4), cs_confpick(argc, argv, 5))
977 sys_exit(rc)
978 return rc
979 }
980 if cs_eq(verb, "cursor" as *u8) == 1 {
981 if argc < 4 { let u: i64 = cs_usage() sys_exit(u) return u }
982 let rc: i64 = cs_cursor(argv[2] as *u8, argv[3] as *u8, cs_confpick(argc, argv, 4))
983 sys_exit(rc)
984 return rc
985 }
986 if cs_eq(verb, "ack" as *u8) == 1 {
987 if argc < 5 { let u: i64 = cs_usage() sys_exit(u) return u }
988 let rc: i64 = cs_ack(argv[2] as *u8, argv[3] as *u8, cs_argn(argv, 4), cs_confpick(argc, argv, 5))
989 sys_exit(rc)
990 return rc
991 }
992 if cs_eq(verb, "del" as *u8) == 1 {
993 if argc < 5 { let u: i64 = cs_usage() sys_exit(u) return u }
994 let rc: i64 = cs_del(argv[2] as *u8, cs_argn(argv, 3), argv[4] as *u8, cs_confpick(argc, argv, 5))
995 sys_exit(rc)
996 return rc
997 }
998 if cs_eq(verb, "status" as *u8) == 1 {
999 if argc < 3 { let u: i64 = cs_usage() sys_exit(u) return u }
1000 let rc: i64 = cs_status(argv[2] as *u8, cs_confpick(argc, argv, 3))
1001 sys_exit(rc)
1002 return rc
1003 }
1004 if cs_eq(verb, "setretention" as *u8) == 1 {
1005 if argc < 5 { let u: i64 = cs_usage() sys_exit(u) return u }
1006 let rc: i64 = cs_setretention(argv[2] as *u8, cs_argn(argv, 3), argv[4] as *u8, cs_confpick(argc, argv, 5))
1007 sys_exit(rc)
1008 return rc
1009 }
1010 if cs_eq(verb, "retain" as *u8) == 1 {
1011 if argc < 3 { let u: i64 = cs_usage() sys_exit(u) return u }
1012 let rc: i64 = cs_retention(argv[2] as *u8, cs_confpick(argc, argv, 3))
1013 sys_exit(rc)
1014 return rc
1015 }
1016 if cs_eq(verb, "purge" as *u8) == 1 {
1017 if argc < 5 { let u: i64 = cs_usage() sys_exit(u) return u }
1018 let rc: i64 = cs_purge(argv[2] as *u8, argv[3] as *u8, argv[4] as *u8, cs_confpick(argc, argv, 5))
1019 sys_exit(rc)
1020 return rc
1021 }
1022 let u2: i64 = cs_usage()
1023 sys_exit(u2)
1024 return u2
1025}