/**
* std/session-store — typed Harn access to the canonical durable session store.
*
* Import: import "std/session-store"
*
* Streams share `<root>/.harn/session-store.sqlite`. The canonical Rust store
* owns event ids, hashes, signatures, replay, and lineage. Retired per-stream
* JSONL files are imported once on first access; all later writes go to SQLite.
* Import is atomic and rejects malformed or hash-chain-invalid sources rather
* than silently dropping rows and laundering the remainder into a new chain.
*
* A `payload` is treated as a mutation: `{operation: "upsert", record}`,
* `{operation: "delete", id}`, `{operation: "replace", records}`, or
* `{operation: "clear"}`. `session_store_project` folds these to the latest
* surviving record per id (first-seen order); `session_store_project_value`
* folds `replace`/`clear` to a single value.
*/
pub type SessionStoreOptions = {
root?: string,
kind?: string,
kind_type?: string,
tenant_id?: string,
actor?: string,
tags?: list<string>,
headers?: dict<string, string>,
}
pub type SessionStoreEventKind = {kind: string, type?: string}
pub type SessionStoreEventSignature = {algorithm: string, key_id: string, signature: string}
pub type SessionStoreEvent = {
event_id: int,
session_id: string,
tenant_id: string?,
parent_event_id: int?,
actor: string?,
kind: SessionStoreEventKind,
payload: any,
tags: list<string>,
headers: dict<string, string>,
ts_ms: int,
ts: string,
record_hash: string,
prev_hash: string?,
signed_by: SessionStoreEventSignature?,
}
pub type SessionStoreVerifyReport = {
_type: "session_store_verify",
session_id: string,
ok: bool,
count: int,
signed_event_count: int,
chain_root_hash?: string?,
broken_at?: int?,
reason?: string?,
}
fn __session_store_event(raw: dict) -> SessionStoreEvent {
return {
event_id: raw.event_id,
session_id: raw.session_id,
tenant_id: raw.tenant_id,
parent_event_id: raw.parent_event_id,
actor: raw.actor,
kind: raw.kind,
payload: raw.payload,
tags: raw.tags,
headers: raw.headers,
ts_ms: raw.ts_ms,
ts: raw.ts,
record_hash: raw.record_hash,
prev_hash: raw.prev_hash,
signed_by: raw.signed_by,
}
}
fn __session_store_verify_report(raw: dict) -> SessionStoreVerifyReport {
return {
_type: "session_store_verify",
session_id: raw.session_id,
ok: raw.ok,
count: raw.count,
signed_event_count: raw.signed_event_count,
chain_root_hash: raw.chain_root_hash,
broken_at: raw.broken_at,
reason: raw.reason,
}
}
/**
* Append an event to `session_id`'s stream and return the stored event dict
* (including `event_id`, `record_hash`, and `prev_hash`).
*
* `options.root` overrides the storage root (default: project root, or
* `HARN_SESSION_STORE_ROOT`). `options.kind` = "hypothesis" tags the event as
* `{kind: "hypothesis"}`; otherwise the event is `{kind: "custom", type:
* options.kind_type}` (default type "LibrarianEntry"). `options.tenant_id`,
* `options.actor` (default "harn"), `options.tags`, and `options.headers` are
* optional. Timestamps are assigned by the canonical store; the retired
* `options.now` override fails loudly instead of being silently ignored. A
* stream's non-nil `tenant_id` is immutable after creation. String options,
* tag lists, and string-valued header dicts are validated at the boundary.
*
* @effects: [store.read, store.write, fs.read, fs.write, env.read, time]
* @errors: [runtime]
*/
pub fn session_store_append(session_id: string, payload, options: SessionStoreOptions? = nil) -> SessionStoreEvent {
return __session_store_event(__session_store_append(session_id, payload, options))
}
/**
* Return every event dict in `session_id`'s stream, in append order.
*
* @effects: [store.read, store.write, fs.read, fs.write, env.read, time]
* @errors: [runtime]
*/
pub fn session_store_events(session_id: string, options: SessionStoreOptions? = nil) -> list<SessionStoreEvent> {
return __session_store_events(session_id, options)
.map({ event -> __session_store_event(event) })
.to_list()
}
/**
* Return the non-nil `payload` of every event in append order.
*
* @effects: [store.read, store.write, fs.read, fs.write, env.read, time]
* @errors: [runtime]
*/
pub fn session_store_payloads(session_id: string, options: SessionStoreOptions? = nil) -> list<any> {
let payloads = []
for event in session_store_events(session_id, options) {
if event.payload != nil {
payloads = payloads.push(event.payload)
}
}
return payloads
}
/**
* Project the stream's mutation payloads to the latest surviving record per
* id, in first-seen order (`upsert`/`delete`/`replace`/`clear`).
*
* @effects: [store.read, store.write, fs.read, fs.write, env.read, time]
* @errors: [runtime]
*/
pub fn session_store_project(session_id: string, options: SessionStoreOptions? = nil) -> list<any> {
let records = {}
let order = []
for mutation in session_store_payloads(session_id, options) {
const operation = to_string(mutation?.operation ?? "")
if operation == "upsert" {
const record = mutation?.record
const id = to_string(record?.id ?? "")
if id != "" {
if records[id] == nil {
order = order + [id]
}
records = records + {[id]: record}
}
} else if operation == "delete" {
const id = to_string(mutation?.id ?? "")
records = records + {[id]: nil}
order = order.filter({ existing -> existing != id }).to_list()
} else if operation == "replace" {
records = {}
order = []
for record in mutation?.records ?? [] {
const id = to_string(record?.id ?? "")
if id != "" {
if records[id] == nil {
order = order + [id]
}
records = records + {[id]: record}
}
}
} else if operation == "clear" {
records = {}
order = []
}
}
return order.map({ id -> records[id] }).filter({ record -> record != nil }).to_list()
}
/**
* Project the stream's `replace`/`clear` mutations to a single value, falling
* back to `default_value`.
*
* @effects: [store.read, store.write, fs.read, fs.write, env.read, time]
* @errors: [runtime]
*/
pub fn session_store_project_value(
session_id: string,
default_value = nil,
options: SessionStoreOptions? = nil,
) -> any {
let value = default_value
for mutation in session_store_payloads(session_id, options) {
const operation = to_string(mutation?.operation ?? "")
if operation == "replace" && mutation?.value != nil {
value = mutation.value
} else if operation == "clear" {
value = default_value
}
}
return value
}
/**
* Verify the hash chain of `session_id`. Returns `{ok, session_id, count}`;
* on the first break also `broken_at` (event index) and `reason`.
*
* @effects: [store.read, store.write, fs.read, fs.write, env.read, time]
* @errors: [runtime]
*/
pub fn session_store_verify(session_id: string, options: SessionStoreOptions? = nil) -> SessionStoreVerifyReport {
return __session_store_verify_report(__session_store_verify(session_id, options))
}
/**
* Return the retired per-stream JSONL path for compatibility.
*
* Canonical writes no longer target this path. Use
* `session_store_database_path` when the shared SQLite database is intended.
*
* @effects: [env.read]
* @errors: [runtime]
*/
pub fn session_store_path(session_id: string, options: SessionStoreOptions? = nil) -> string {
return __session_store_path(session_id, options)
}
/**
* Return the absolute path of the shared canonical SQLite database.
*
* The returned database contains every stream under the selected root.
* Callers must not treat it as a per-session artifact.
*
* @effects: [env.read]
* @errors: [runtime]
*/
pub fn session_store_database_path(options: SessionStoreOptions? = nil) -> string {
return __session_store_database_path(options)
}