Struct Store 

pub struct Store { /* private fields */ }

One eventsdb log, one content-addressed blob directory and one read model, under one root.

Every method is synchronous. eventsdb’s API is async, so the store owns a current-thread runtime and block_ons each call on it: Lua has nothing to suspend into, and a host method that returns a future would be a future nobody polls.

command is the single-writer lock, and it is the reason the decisions below can be trusted. append_if makes one stream’s fold atomic, but a policy that reads one stream and then writes another — an alias bound only to a card that exists, which is what bind_alias is — is two calls, and eventsdb cannot make those one. This process is the only writer of this file, so holding command across the pair is what closes that window. Every write method takes it; query takes it too, because catching the read model up is itself a write.

Implementations

impl Store

pub fn open(root: &Path) -> Result<Store>

Open the store under root, creating <root>/ and <root>/blobs/ if they are not there. The log is <root>/cards.db, and the read model’s tables are in it.

impl Store

Exposed to Teal as require("store"). Its declaration is written to src/store.d.tl by this macro at build time, and by htl dts / htl check without building.

errors = "return": every fallible method comes back Lua-style, value, err. An Option return then has three answers rather than two, which append_if needs — rec, nil wrote, nil, err failed, and nil, nil is the decision declining.

pub fn append( &self, stream: &str, kind: &str, meta: Value, data: Value, ) -> Result<Recorded>

Append {kind, meta, data} to stream.

A null meta or data is left out of the envelope rather than written as JSON null: eventsdb’s contract is that those two keys are optional and scalars-only / any-depth respectively, and null is neither absent nor a value it wants.

pub fn append_if( &self, stream: &str, decision: &str, kind: &str, meta: Value, data: Value, ) -> Result<Option<Recorded>>

Append {kind, meta, data} only if decision, folded over stream inside the write, says so. Returns the event when it wrote and nothing when it declined.

decision names one of a fixed set built here. Teal never passes code: a decision runs while the log holds its write lock, and a callback into Lua from there would put an interpreter this host does not control inside eventsdb’s transaction. Teal chooses; Rust decides.

On the Lua side the three outcomes are rec, nil (written), nil, err (failed) and nil, nil (declined) — so if rec == nil and err == nil then is the test for a decision that found nothing to do.

pub fn bind_alias( &self, name: &str, card_id: &str, note: Option<String>, ) -> Result<Option<Recorded>>

Bind name to card_id, if that card was opened and the name does not already mean it.

Why this is a method and not another decision string. The invariant — an alias points only at a card that exists — spans two streams, and append_if folds one. So this is two calls: read card-<card_id> for its card_opened, then append_if on alias-<name>. What makes the pair atomic is command, held across both, and what makes that enough is that this process is the only writer of this file — the design’s reservation stream, and the BP note that a single local writer is a legitimate answer to a cross-aggregate uniqueness rule rather than a shortcut. Exposing an alias decision through append_if would let Teal make the second call without the first, which is exactly the dangling alias this step is for.

Err when no card was opened under card_id. Ok(None) — the decision declining — when the name already means that card: a rebind to where the alias already points asks for a state that holds, so it is idempotent and writes nothing. Rebinding to a different card appends another alias_bound on the same stream, which is what keeps the history: nothing is overwritten and nothing has to be released first.

pub fn release_alias( &self, name: &str, note: Option<String>, ) -> Result<Option<Recorded>>

Release name, if it currently means anything. Ok(None) when it does not.

The event carries the card_id it released, so the history reads without a join and a rebuild can tell “released from A” from “released from B”.

pub fn read_stream( &self, stream: &str, kinds: Option<Vec<String>>, ) -> Result<Vec<Recorded>>

The whole of stream in seq order, optionally only kinds.

Paged rather than read at once: read_all is one page and a cursor, so this loops until a page comes back short. Nothing is held between pages.

pub fn query(&self, sql: &str, params: Vec<Value>) -> Result<Vec<Value>>

The escape hatch: read-only SQL over the log, the read model’s tables and anything else beside them.

Read-your-writes. The projection is caught up first, under the same lock a write takes, so a Teal find that runs a line after a close sees the closed card. Without that the read model would be eventually consistent, which for a single-process store is a cost with nothing bought by it: the only writer is this process, so “everything written” is a state this call can reach rather than wait for. A caught-up projection costs one empty batch read when there is nothing to do.

params binds by position (?1, ?2, …). Rows come back as JSON objects, one per row, so Teal sees a table per row keyed by column name.

pub fn catch_up(&self) -> Result<u64>

Fold everything the read model has not seen yet, and say how many events that was.

query does this on its own, so nothing needs to call it to read correctly. It is here for the two cases where the number is the point: a batch job that wants the model warm before it starts timing, and a test that wants to prove a read did not need it.

pub fn rebuild(&self) -> Result<u64>

Empty the read model and replay the log into it, returning the events applied.

For a fold that changed without its tables changing shape. When the shape changes incompatibly the move is to rename the projection (cards_v1 → cards_v2), which gives the new model its own cursor and leaves the old one readable until the switch.

pub fn blob_put(&self, bytes: BString) -> Result<Blob>

Store bytes under the hex of their SHA-256 and return the name.

Content-addressed, so it is idempotent by construction: the same bytes are the same file, and a second put of them writes nothing. The write goes to a temporary name in the same directory and is renamed into place, so a reader never sees a half-written blob under a hash that promises the whole of it.

pub fn blob_get(&self, hash: &str) -> Result<Option<BString>>

The bytes stored under hash, or nothing if no blob has that name.

pub fn blob_path(&self, hash: &str) -> String

Where the blob named hash lives, whether or not it is there yet.

pub fn json_encode(&self, v: Value) -> Result<String>

v as JSON text.

Here because Teal has no JSON of its own, and because the policy side needs to weigh a value before deciding where to put it: a batch of sample rows is inlined or blobbed on the length of exactly this text. The conversion is the one every other method on this store uses, so what is measured here is what would be stored.

pub fn digest(&self, v: Value) -> String

text, parsed. The other direction, for reading back what blob_put was handed: a blob is bytes to this store and JSON only to whoever wrote it. A short, stable fingerprint of a JSON value: the first 16 hex digits of the SHA-256 of its canonical text, where canonical means every object’s keys are in sorted order and nothing is pretty-printed.

Here rather than in Teal because Teal has no hash, and canonical because two runs that were given the same params should print the same whatever order their tables happened to be walked in. What goes into the fingerprint is the policy side’s call — cards.open hands it params and nothing else — and this only answers what those bytes are called.

pub fn json_decode(&self, text: &str) -> Result<Value>

pub fn export(&self) -> Result<ExportReport>

Write everything the log holds past the end of the confirmed export chain to one JSON Lines file under <root>/export/, and confirm it.

This is the half of a prune that has to happen first, and the whole log is what it takes: Guard::Exported chains the confirmed unfiltered receipts from position 0 and refuses to remove past the chain’s end, so an export of only the streams being pruned would leave the chain — and therefore the guard — exactly where it was. See transfer for the order and the reasoning; what the directory ends up being is an append-only backup of the log, one file per call, which is what import reads.

Nothing new is not an error and not an empty file: file comes back absent and events is 0.

pub fn import(&self, path: &str) -> Result<ImportReport>

Read a JSON Lines file written by Store::export back into this log, and catch the read models up.

seq and position are this log’s to assign; kind, meta, data, epoch_ms and _schema_version travel unchanged. reproduced_coordinates says whether every event landed back on the position it carried, which is true for a file imported in order into an empty store — the check that a restore really is the same log rather than the same events.

pub fn retain_streams(&self, streams: Vec<String>) -> Result<RetainReport>

Remove every event of streams, if the exports vouch for them and no read model would be left behind, and give the freed pages back to the filesystem.

Guard::Exported, never Force: the one operation here that can make a correct read wrong is the one operation that asks permission. Both refusals come back as errors that say what to do — run an export, or catch the named consumer up.

removed counts events and streams counts the streams they were spread over, which is at most the number asked for: a stream with nothing on it is not an error and is not counted.

pub fn blob_gc(&self) -> Result<BlobGcReport>

Delete every blob nothing points at any more, and the row that counted the pointers.

cb_blobs.refs is the projection’s count — one per samples_appended or checkpoint_saved naming the hash, one back per card the prune journal removed — so a blob two cards share survives the first of them going. cb_blobs is this crate’s table rather than eventsdb’s, which is why the hatch lets the row be deleted at all.

pub fn root(&self) -> String

The directory this store was opened on.

impl Store

pub fn htl_preload(self, h: &Htl) -> Result<()>

Register this instance as the require("MODULE") value.

Trait Implementations

impl HostModule for Store

const MODULE: &'static str = "store"

Module name used in require("...").

const DECL: &'static str = "local record store\n record Recorded\n stream: string\n seq: integer\n position: integer\n epoch_ms: integer\n kind: string\n meta: any\n data: any\n end\n record Blob\n hash: string\n size: integer\n end\n record ExportReport\n file: string\n from: integer\n through: integer\n events: integer\n end\n record ImportReport\n events: integer\n reproduced_coordinates: boolean\n end\n record RetainReport\n removed: integer\n streams: integer\n end\n record BlobGcReport\n deleted: integer\n bytes: integer\n end\n append: function(self: store, stream: string, kind: string, meta: any, data: any): Recorded, string\n append_if: function(self: store, stream: string, decision: string, kind: string, meta: any, data: any): Recorded, string\n bind_alias: function(self: store, name: string, card_id: string, note?: string): Recorded, string\n release_alias: function(self: store, name: string, note?: string): Recorded, string\n read_stream: function(self: store, stream: string, kinds?: {string}): {Recorded}, string\n query: function(self: store, sql: string, params: {any}): {any}, string\n catch_up: function(self: store): integer, string\n rebuild: function(self: store): integer, string\n blob_put: function(self: store, bytes: string): Blob, string\n blob_get: function(self: store, hash: string): string, string\n blob_path: function(self: store, hash: string): string\n json_encode: function(self: store, v: any): string, string\n digest: function(self: store, v: any): string\n json_decode: function(self: store, text: string): any, string\n export: function(self: store): ExportReport, string\n import: function(self: store, path: string): ImportReport, string\n retain_streams: function(self: store, streams: {string}): RetainReport, string\n blob_gc: function(self: store): BlobGcReport, string\n root: function(self: store): string\nend\n\nreturn store\n"

Full .d.tl text for the module.

impl UserData for Store

fn add_methods<M: UserDataMethods<Self>>(m: &mut M)

Adds custom methods and operators specific to this userdata.

fn add_fields<F>(fields: &mut F)

where F: UserDataFields<Self>,

Adds custom fields specific to this userdata.

fn register(registry: &mut UserDataRegistry<Self>)

Registers this type for use in Lua.