# @celilo/event-bus

SQLite-backed pub/sub for single-host systems. Persistent event log,
subprocess or in-process handlers, query/reply, typed event registry.
Bun-first.

See [`infra/openspec/specs/event-bus/spec.md`](../../openspec/specs/event-bus/spec.md)
for the full design.

## Quick start

```ts
import { z } from 'zod';
import { defineEvents, openBus, runDispatcher } from '@celilo/event-bus';

const events = defineEvents({
  'deploy.completed.lunacycle': z.object({ version: z.string() }),
  'cert.rotated': z.object({ domain: z.string() }),
  'domain.handler': {
    payload: z.object({ domain: z.string() }),
    replies: z.object({ module: z.string() }),
  },
});

const bus = openBus({ dbPath: '/var/lib/celilo/events.db', events });

// Producer side
bus.emit('deploy.completed.lunacycle', { version: '1.2.0' });

// Persistent subscriber (rule, not a process)
bus.subscribe({
  name: 'lunacycle.smoke-after-deploy',
  pattern: 'deploy.completed.lunacycle',
  handler: 'bun /opt/lunacycle/scripts/smoke-handler.ts',
  timeoutMs: 120_000,
});

// Long-running dispatcher
runDispatcher(bus, { pollIntervalMs: 1000, concurrency: 4 });
```

## Linked handler

```ts
// smoke-handler.ts
import { defineHandler } from '@celilo/event-bus/handler';
import { events } from '../events.ts';

const handler = defineHandler(events, 'deploy.completed.lunacycle',
  async (event) => {
    await runSmoke(event.payload.version);
  });

if (import.meta.main) await handler.run(Number(process.argv[2]));
```

## Unlinked handler (bash, foreign langs)

```bash
#!/bin/bash
EVENT_JSON=$(event-bus get "$1")
DOMAIN=$(jq -r .payload.domain <<< "$EVENT_JSON")

if reload-caddy "$DOMAIN"; then
  event-bus ack "$1"
else
  event-bus fail "$1" --error "caddy reload failed"
fi
```

Set `EVENT_BUS_DB` in env or pass `--db <path>` to the CLI.

## Transient watch (CLI live progress)

```ts
const watch = bus.watch('deploy.*.lunacycle', (event) => ui.update(event));
// ... later
watch.close();
```

## Query / reply

```ts
const replies = await bus.query('domain.handler', { domain: 'foo.com' }, {
  timeoutMs: 5000,
  expect: 'first',
});
```

## CLI

```
event-bus get <event_id>
event-bus ack <event_id> [--subscriber <id>]
event-bus fail <event_id> --error <msg> [--no-retry]
event-bus emit <type> [<payload-json>] [--dedup-key <k>]
event-bus query <type> <payload-json> [--timeout 5000] [--expect first|all]
event-bus reply <query_event_id> <payload-json>
event-bus subscribe --name --pattern --handler [--max-attempts] [--timeout-ms]
event-bus unsubscribe --name
event-bus list-subscribers
event-bus list-pending [--subscriber <name>]
event-bus tail [--type <t>] [--limit N]
event-bus drain [--concurrency N]
event-bus run [--poll-ms N] [--concurrency N] [--halt-on-recovery]
event-bus status
```

## Health & crash recovery

The dispatcher writes a heartbeat every poll tick. On startup, if it
detects a stale heartbeat with stuck `running` deliveries, it resets
them to `pending` and emits `_internal.dispatcher.recovered_from_crash`.
Pass `haltOnRecovery: true` to `runDispatcher` to require operator
acknowledgment instead of auto-recovering.

`bus.health()` returns `{ status, lastHeartbeatAgeMs, stuckRunningCount, oldestPendingAgeMs }`
where status is `'healthy' | 'recovering' | 'stuck' | 'wedged' | 'no_dispatcher'`. `wedged` is a registered dispatcher whose process is alive but whose heartbeat has gone stale with nothing running — restart it, do not start a second.

## Retention

Events whose deliveries are all terminal AND emitted past
`eventsRetentionDays` (default 30) are purged in the dispatcher's
idle loop. Abandoned deliveries past `failedRetentionDays` (default
90) are purged separately.

## Status

v0.1.0 — single-dispatcher, single-host. Multi-dispatcher / multi-host,
schema-typed glob patterns at compile time, and a Prometheus exporter
are deferred.