Merge pull request #18 from Dvorinka/feat/realtime-client

feat: realtime channel subscriptions in @primora/client

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 <slug> --filter … --order …`.
 
 ## [0.7.1] - 2026-09-23

ROADMAP.md

@@ -295,14 +295,22 @@ PostgREST-style filtering and ordering on collection documents.
 - `primora documents list --collection <slug> --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.

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";
 

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<string, unknown>;
+}
+
+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 `<channel>.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<Listener>();
+  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<void> {
+    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;
+    },
+  };
+}