will.anderson
aae7f883cd
- EventBus: dumb wire, publish/subscribe, failures surface - Envelope: four schema planes, inject-only headers, content-addressed - TopicLog: owned SQLite ledger, replay + by-type/time/causality - Seed facade: durable=ledger+wire, ephemeral=wire only, fold-for-state - Rung 1 kernel: OELBR loop, DAG plans, durable cancellation tokens - Neuron v0 executable spec source; specs/event-bus.md citizen zero - Standing orders: design-before-code doctrine installed
120 lines
4.0 KiB
TypeScript
120 lines
4.0 KiB
TypeScript
export * as Seed from "./seed"
|
|
|
|
/**
|
|
* RUNG 0 FACADE - the bus and the ledger, joined.
|
|
*
|
|
* Durable topics: publish = append to the log, then fan out on the wire.
|
|
* Ephemeral topics: wire only - nothing persisted, exactly per contract.
|
|
* Replay: fold the ledger back into live subscribers (crash recovery,
|
|
* late joiners).
|
|
*
|
|
* One owned SQLite file. Zero coupling to any other storage.
|
|
*/
|
|
|
|
import { EventBus } from "./event-bus"
|
|
import { TopicLog } from "./topic-log"
|
|
import { EventBusSchema } from "./event-bus-schema"
|
|
|
|
type Envelope = EventBusSchema.Envelope
|
|
type EnvelopeInput = Omit<Envelope, "id" | "seq">
|
|
|
|
export interface Seed {
|
|
/** Declare a topic with its persistence contract. */
|
|
readonly topic: (
|
|
input: {
|
|
readonly id: string
|
|
readonly durability: "durable" | "ephemeral"
|
|
readonly author: string
|
|
readonly retainLast?: boolean
|
|
},
|
|
) => { readonly id: string; readonly durability: "durable" | "ephemeral" }
|
|
|
|
/**
|
|
* Publish an envelope. Durable topic -> appended to the ledger first
|
|
* (seq assigned, identity stamped), then fanned out. Ephemeral ->
|
|
* straight to the wire.
|
|
*/
|
|
readonly publish: (input: EnvelopeInput) => Envelope
|
|
|
|
/** Subscribe to a topic. Returns unsubscribe. */
|
|
readonly subscribe: (topic: string, handler: (envelope: Envelope) => void) => () => void
|
|
|
|
/** THE LEDGER: replay a durable topic in order. */
|
|
readonly replay: (topic: string, fromSeq?: number) => Array<Envelope>
|
|
|
|
/** Fold a durable topic through a reducer - state from history. */
|
|
readonly fold: <S>(topic: string, initial: S, step: (state: S, envelope: Envelope) => S) => S
|
|
|
|
/** Late joiner / crash recovery: replay everything since `fromSeq` into a subscriber. */
|
|
readonly catchUp: (topic: string, handler: (envelope: Envelope) => void, fromSeq?: number) => number
|
|
|
|
readonly query: {
|
|
byType(topic: string, type: string): Array<Envelope>
|
|
byTimeRange(opts: { from?: number; to?: number; topic?: string }): Array<Envelope>
|
|
walkCausality(topic: string, envelope: Envelope): Array<Envelope>
|
|
}
|
|
}
|
|
|
|
export function open(path: string): Seed {
|
|
const log = TopicLog.open(path)
|
|
const durability = new Map<string, "durable" | "ephemeral">()
|
|
const lastValue = new Map<string, Envelope>()
|
|
|
|
const api: Seed = {
|
|
topic(input) {
|
|
durability.set(input.id, input.durability)
|
|
return { id: input.id, durability: input.durability }
|
|
},
|
|
|
|
publish(input) {
|
|
const durabilityOfTopic =
|
|
durability.get(input.topic) ?? ("durable" as const)
|
|
if (durabilityOfTopic === "ephemeral") {
|
|
const provisional: Envelope = {
|
|
...input,
|
|
seq: -1,
|
|
id: "sha256:ephemeral",
|
|
} as unknown as Envelope
|
|
lastValue.set(input.topic, provisional)
|
|
EventBus.publish(input.topic, input.body.type, provisional)
|
|
return provisional
|
|
}
|
|
const stored = TopicLog.append(log, input)
|
|
EventBus.publish(input.topic, stored.body.type, stored)
|
|
return stored
|
|
},
|
|
|
|
subscribe(topic, handler) {
|
|
// The generic bus wraps payloads as {topic,type,payload}; subscribers
|
|
// here always want THE Envelope itself.
|
|
return EventBus.subscribe(topic, (raw) => {
|
|
const wrapped = raw as unknown as { payload?: unknown }
|
|
handler((wrapped.payload ?? raw) as Envelope)
|
|
})
|
|
},
|
|
|
|
replay(topic, fromSeq) {
|
|
return TopicLog.replay(log, topic, fromSeq)
|
|
},
|
|
|
|
fold(topic, initial, step) {
|
|
let state = initial
|
|
for (const envelope of TopicLog.replay(log, topic)) state = step(state, envelope)
|
|
return state
|
|
},
|
|
|
|
catchUp(topic, handler, fromSeq) {
|
|
const past = TopicLog.replay(log, topic, fromSeq)
|
|
for (const envelope of past) handler(envelope)
|
|
return past.length
|
|
},
|
|
|
|
query: {
|
|
byType: (topic, type) => TopicLog.byType(log, topic, type),
|
|
byTimeRange: (opts) => TopicLog.byTimeRange(log, opts),
|
|
walkCausality: (topic, envelope) => TopicLog.walkCausality(log, topic, envelope),
|
|
},
|
|
}
|
|
return api
|
|
}
|