diff --git a/CHANGELOG.md b/CHANGELOG.md index 4d767f1..9328929 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,7 @@ All notable changes to Primora. Format follows [Keep a Changelog](https://keepac - **Vault page in the dashboard** — project secrets table with create/update, `.env` import, delete, and copy-to-clipboard reveal; plus a "This machine" card when running inside the desktop shell that drives the local vault through the same Tauri IPC commands as the vault window. - **CLI remote mode** — `primora secrets set|get|list|rm|import --remote` operates on the project vault via the existing API auth (`--project` overrides). `get --remote` is the explicit reveal path; `list --remote` never returns values. - **Project-vault MCP tools** — `primora_secrets_list` (metadata only) and `primora_secrets_set` (write-only). No reveal or delete tools — agents consume secrets through `secret://NAME` references in job payloads without holding plaintext. +- **Realtime channel subscriptions** (`@primora/client`) — `createRealtimeClient({ endpoint, key, projectId })` opens the project SSE stream (`/realtime/stream`) via fetch + ReadableStream (EventSource can't send `X-API-Key`). `channel("documents").on("created", cb)` subscribes by event namespace (`document`/`object`/`issue`/`deploy`, `"*"` for all); auto-reconnects with capped exponential backoff; `onStateChange` reports `connecting|open|reconnecting|closed`. - **Document filtering and ordering** — `GET /collections/:id/documents` accepts `filter` (comma-separated `field.op.value`; ops `eq neq gt gte lt lte like in is`; dotted paths index into document data, `id`/`created_at`/`updated_at` filterable) and `order` (`field.asc|field.desc` list). Whitelisted parsing + parameterized values; plain lists unchanged. CLI: `primora documents list --collection --filter … --order …`. ## [0.7.1] - 2026-09-23 diff --git a/ROADMAP.md b/ROADMAP.md index 65e33f5..42c309c 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -295,14 +295,22 @@ PostgREST-style filtering and ordering on collection documents. - `primora documents list --collection --filter … --order …` in the CLI; `filter`/`order` params in the generated client. +## Phase 8 — Realtime client SDK (unreleased) + +Channel subscriptions over the Phase 7 SSE stream, in `@primora/client`. + +- `createRealtimeClient({ endpoint, key, projectId })` — fetch + + ReadableStream reader (EventSource can't send `X-API-Key`), auto-reconnect + with capped backoff, `onStateChange` lifecycle. +- `channel("documents").on("created", cb)` — namespaces `document`, `object`, + `issue`, `deploy`, `"*"` for everything; trailing `s` optional. +- Presence and client-broadcast deliberately deferred — protocol question, + no consumer yet. + ## Future phases — candidates Ordered loosely by leverage. None committed; each gets scoped when picked. -- **Phase 8 — Client realtime SDK** — `@primora/client` channel - subscriptions (`channel('documents').on('created', …)`), presence, - and broadcast over the Phase 7 stream. Supabase-realtime parity is a - protocol question, not a transport one. - **Phase 9 — Functions** — user code on schedule/event/HTTP triggers. Deno or Bun isolates in a sidecar; secrets via env, payload via stdin. The job runner already supplies the scheduling half. diff --git a/packages/client/src/index.ts b/packages/client/src/index.ts index 5b22013..2e613d6 100644 --- a/packages/client/src/index.ts +++ b/packages/client/src/index.ts @@ -1,4 +1,14 @@ export type PrimoraEventType = "error" | "metric" | "log" | "heartbeat" | "event"; + +export { + createRealtimeClient, + type RealtimeChannel, + type RealtimeClient, + type RealtimeClientOptions, + type RealtimeEvent, + type RealtimeHandler, + type RealtimeState, +} from "./realtime.js"; export type PrimoraComponentKind = | "frontend" | "backend" | "database" | "android" | "desktop" | "web" | "other"; diff --git a/packages/client/src/realtime.ts b/packages/client/src/realtime.ts new file mode 100644 index 0000000..3e9f105 --- /dev/null +++ b/packages/client/src/realtime.ts @@ -0,0 +1,150 @@ +// Realtime subscriptions over the project's SSE domain-event stream +// (GET /api/v1/projects/:id/realtime/stream). EventSource can't send auth +// headers, so the stream is consumed via fetch + ReadableStream. +// +// jarvis: ceiling — subscriptions only. Presence and client-broadcast are a +// protocol question; add when a concrete consumer needs them. + +export interface RealtimeEvent { + /** e.g. "document.created", "object.deleted", "issue.created", "deploy.marker" */ + type: string; + occurred_at: string; + data: Record; +} + +export type RealtimeState = "connecting" | "open" | "reconnecting" | "closed"; + +export interface RealtimeClientOptions { + /** Primora deployment origin, e.g. http://localhost:8090 */ + endpoint: string; + /** project API key (prm_…) — sent as X-API-Key */ + key: string; + projectId: string; + onStateChange?: (state: RealtimeState) => void; +} + +export type RealtimeHandler = (event: RealtimeEvent) => void; + +export interface RealtimeChannel { + /** + * `on("created", cb)` matches `.created`; `on("*")` every event in + * the channel; a full type like `"document.created"` also works. + * Returns an unsubscribe function. + */ + on(event: string, handler: RealtimeHandler): () => void; +} + +export interface RealtimeClient { + /** + * Channel names are event namespaces — "document", "object", "issue", + * "deploy", or "*" for everything. A trailing "s" is dropped so + * `channel("documents")` and `channel("document")` are equivalent. + */ + channel(name: string): RealtimeChannel; + close(): void; + readonly state: RealtimeState; +} + +interface Listener { + prefix: string; + pattern: string; + handler: RealtimeHandler; +} + +export function createRealtimeClient(opts: RealtimeClientOptions): RealtimeClient { + const url = + opts.endpoint.replace(/\/$/, "") + + `/api/v1/projects/${encodeURIComponent(opts.projectId)}/realtime/stream`; + + const listeners = new Set(); + const abort = new AbortController(); + let state: RealtimeState = "connecting"; + + const setState = (s: RealtimeState) => { + state = s; + opts.onStateChange?.(s); + }; + + const dispatch = (evt: RealtimeEvent) => { + for (const l of listeners) { + const match = + l.pattern === "*" + ? evt.type.startsWith(l.prefix) + : l.pattern.includes(".") + ? evt.type === l.pattern + : evt.type === l.prefix + l.pattern; + if (match) { + try { + l.handler(evt); + } catch { + // a throwing listener must not kill the stream + } + } + } + }; + + // Minimal SSE frame parser — we only need `data:` lines. + async function readStream(res: Response): Promise { + if (!res.body) throw new Error("no response body"); + const reader = res.body.getReader(); + const dec = new TextDecoder(); + let buf = ""; + for (;;) { + const { done, value } = await reader.read(); + if (done) return; + buf += dec.decode(value, { stream: true }); + let idx; + while ((idx = buf.indexOf("\n\n")) >= 0) { + const frame = buf.slice(0, idx); + buf = buf.slice(idx + 2); + for (const line of frame.split("\n")) { + if (!line.startsWith("data:")) continue; + try { + dispatch(JSON.parse(line.slice(5).trim()) as RealtimeEvent); + } catch { + // keepalive pings and malformed frames are ignored + } + } + } + } + } + + (async () => { + let attempt = 0; + while (!abort.signal.aborted) { + try { + setState(attempt === 0 ? "connecting" : "reconnecting"); + const res = await fetch(url, { + headers: { "X-API-Key": opts.key, Accept: "text/event-stream" }, + signal: abort.signal, + }); + if (!res.ok) throw new Error(`realtime stream: HTTP ${res.status}`); + setState("open"); + attempt = 0; + await readStream(res); + } catch (err) { + if (abort.signal.aborted) break; + } + // stream ended or failed — back off, capped at 30 s + const delay = Math.min(1000 * 2 ** attempt++, 30000); + await new Promise((r) => setTimeout(r, delay)); + } + })().finally(() => setState("closed")); + + return { + channel(name) { + const prefix = name === "*" ? "" : name.replace(/s$/, "") + "."; + return { + on(event, handler) { + const l: Listener = { prefix, pattern: event, handler }; + listeners.add(l); + return () => listeners.delete(l); + }, + }; + }, + close: () => abort.abort(), + get state() { + return state; + }, + }; +}