feat(canvas): live events + realtime stats for freeform canvases

.claude/skills/canvas-templates/SKILL.md

@@ -103,6 +103,40 @@ iframe hold a token — it posts a request; the host runs the authenticated call
 > anonymous viewers, so publish converts validated query nodes → saved insights +
 > an allowlist and the canvas references them by id. Implement it there, not in edit.
 
+## Live / realtime (SSE)
+
+Two more `ph.*` methods cover **live** data — a streaming counterpart to the
+request/response reads. Use them when a canvas must update as events arrive
+(live event feed, users-online counter) instead of re-polling an insight:
+
+- **`ph.subscribeLiveEvents(params, onEvent, onEnd)`** — a live SSE subscription,
+  NOT request/response. The canvas posts a `live-subscribe` frame; the host
+  renderer asks the host for the brokered connection config
+  (`canvasData.liveConnectionConfig`: `{ eventsUrl, statsUrl, token }`), opens
+  the SSE to the **livestream service** (`live.{region}.posthog.com/events`)
+  itself (`openLiveEventsStream` in `freeformDataBridge.ts`), and fans each
+  event into the iframe as a `live-event` postMessage frame. The scoped JWT
+  lives only in the renderer — it never crosses into the sandbox. Extra
+  postMessage frame types (`live-subscribe` / `live-unsubscribe` from canvas,
+  `live-event` / `live-stream-status` from host) extend the protocol unions in
+  `freeformSchemas.ts`. `params.properties` filters only support the livestream
+  allowlist (`exact|is_not|icontains|not_icontains|regex|not_regex|gt|gte|lt|
+  lte|is_set|is_not_set`, the `LIVE_EVENTS_SUPPORTED_OPERATORS` const) — the
+  prompt tells the agent that; validation lives in `freeformSchemas.ts`.
+- **`await ph.liveStats()`** — a ONE-SHOT poll (rides the generic `data-request`
+  frame with `method: "liveStats"`) → `canvasData.liveStats` →
+  `CanvasDataService.liveStats()` → `GET <live host>/stats` with the JWT →
+  `{ users_on_product, active_recordings }`. Poll on an interval from the
+  canvas; NOT a stream. The derived **sliding-window "users online"** count
+  (60s window over `distinct_id`s from the event stream) is canvas-side JS —
+  no extra host method needed; the prompt spells out the map+prune pattern.
+
+The live host comes from the region: `getLiveEventsUrlFromRegion` in
+`packages/shared/src/urls.ts` (`us`→`live.us.posthog.com`, `eu`→
+`live.eu.posthog.com`, `dev`→`localhost:8666`). The JWT is the project's
+`live_events_token` (Django-minted, `aud=posthog:livestream`, ~24h TTL), fetched
+server-side by `CanvasDataService` and cached per project.
+
 ## Dates
 
 The freeform app owns its date control (the toolbar picker is hidden for freeform —

packages/core/src/canvas/canvasDataService.ts

@@ -5,13 +5,16 @@ import {
   type RootLogger,
   type ScopedLogger,
 } from "@posthog/di/logger";
+import { getLiveEventsUrlFromRegion } from "@posthog/shared";
 import { inject, injectable } from "inversify";
 import type {
   CanvasCaptureConfig,
   CanvasCaptureInput,
   CanvasCaptureResult,
   CanvasDataQueryInput,
   CanvasDataResult,
+  CanvasLiveConnectionConfig,
+  CanvasLiveStats,
   CanvasLoadInsightInput,
 } from "./freeformSchemas";
 import {
@@ -44,6 +47,10 @@ export class CanvasDataService {
   // projects in the same session doesn't reuse the previous project's key (this
   // is a singleton service).
   private readonly projectTokens = new Map<number, string>();
+  // Project-scoped live-events JWTs (Django-minted, aud=posthog:livestream).
+  // Keyed by project id like `projectTokens` — the host holds these only to
+  // open the livestream SSE; they never cross into the iframe.
+  private readonly liveEventsTokens = new Map<number, string>();
   // The signed-in user's distinct_id, the default attribution in edit mode.
   // Per-user (not per-project), so a single cached value is correct.
   private userDistinctId: string | undefined;
@@ -173,6 +180,45 @@ export class CanvasDataService {
     return { ok: true };
   }
 
+  // The brokered connection details for the live-events SSE stream. The token
+  // is the project's `live_events_token` (a scoped JWT minted by Django with
+  // `aud=posthog:livestream`, ~24h TTL) — the host renderer holds it just long
+  // enough to open `/events` and fan events into the canvas iframe; it never
+  // enters the sandbox. The live host is derived from the account's cloud
+  // region (us.eu → live.us.eu.posthog.com; dev → localhost:8666), mirroring
+  // the frontend's `liveEventsHostOrigin()`.
+  async liveConnectionConfig(): Promise<CanvasLiveConnectionConfig> {
+    const { apiHost } = await this.getAuthContext();
+    const origin = getLiveEventsUrlFromRegion(this.getRegion());
+    const token = await this.getLiveEventsToken(apiHost);
+    return {
+      eventsUrl: `${origin}/events`,
+      statsUrl: `${origin}/stats`,
+      token,
+    };
+  }
+
+  // The project's realtime counters (`users_on_product`, `active_recordings`)
+  // from the live-events `/stats` endpoint. The same authenticated fetch the
+  // app's "users online" pill makes — scoped JWT, no project id needed (the
+  // JWT itself identifies the project).
+  async liveStats(): Promise<CanvasLiveStats> {
+    const { apiHost } = await this.getAuthContext();
+    const origin = getLiveEventsUrlFromRegion(this.getRegion());
+    const token = await this.getLiveEventsToken(apiHost);
+    const response = await fetch(`${origin}/stats`, {
+      headers: { Authorization: `Bearer ${token}` },
+    });
+    if (!response.ok) {
+      const err = new Error(`Live stats request failed (${response.status})`);
+      this.log.warn("Canvas live stats failed", {
+        status: response.status,
+      });
+      throw err;
+    }
+    return (await response.json()) as CanvasLiveStats;
+  }
+
   // The project's public capture key. Fetched from the authenticated project
   // endpoint (which the user can already read) and cached; capture itself uses
   // the public key, not the bearer token.
@@ -195,6 +241,18 @@ export class CanvasDataService {
     return data.api_token;
   }
 
+  // Resolve the API host for server-side live calls (token validity is enforced
+  // by `getValidAccessToken`; we never need the raw token itself here since the
+  // livestream service uses the project-scoped JWT).
+  private async getAuthContext(): Promise<{ apiHost: string }> {
+    const { apiHost } = await this.authService.getValidAccessToken();
+    const projectId = this.authService.getState().currentProjectId;
+    if (projectId == null) {
+      throw new Error("No PostHog project selected");
+    }
+    return { apiHost };
+  }
+
   // The signed-in user's distinct_id (so edit-mode captures attribute to "me" in
   // PostHog, not a placeholder). Cached; returns undefined if unavailable.
   private async getUserDistinctId(): Promise<string | undefined> {
@@ -203,4 +261,43 @@ export class CanvasDataService {
     this.userDistinctId = user?.distinctId;
     return this.userDistinctId;
   }
+
+  // The account's cloud region ("us" | "eu" | "dev"), used to derive the
+  // live-events host. Direct session state (`getValidAccessToken` doesn't
+  // expose the region); defaults to "us" when the session is still resolving.
+  private getRegion(): "us" | "eu" | "dev" {
+    return (
+      (
+        this.authService.getState() as {
+          cloudRegion?: "us" | "eu" | "dev" | null;
+        }
+      ).cloudRegion ?? "us"
+    );
+  }
+
+  // The project's scoped `live_events_token` JWT (minted by Django, ~24h TTL).
+  // Fetched from the authenticated project endpoint and cached per project.
+  private async getLiveEventsToken(apiHost: string): Promise<string> {
+    const projectId = this.authService.getState().currentProjectId;
+    if (projectId == null) {
+      throw new Error("No PostHog project selected");
+    }
+    const cached = this.liveEventsTokens.get(projectId);
+    if (cached) return cached;
+    const response = await this.authService.authenticatedFetch(
+      fetch,
+      `${apiHost}/api/environments/${projectId}/`,
+    );
+    if (!response.ok) {
+      throw new Error(`Couldn't read live events token (${response.status})`);
+    }
+    const data = (await response.json()) as {
+      live_events_token?: string | null;
+    };
+    if (!data.live_events_token) {
+      throw new Error("Project has no live events token");
+    }
+    this.liveEventsTokens.set(projectId, data.live_events_token);
+    return data.live_events_token;
+  }
 }

packages/core/src/canvas/canvasTemplates.ts

@@ -51,6 +51,12 @@ const FREEFORM_BASE = [
   '- `ph.openExternal(url)` asks the host to open an absolute `https://posthog.com` (or `*.posthog.com`) URL — anything else is blocked, so do NOT link to other sites. Call it only from a user interaction (e.g. a click handler); the host ignores opens while the canvas is not focused, so calling it on load/in effects does nothing. Sandboxed `target="_blank"` navigation is intentionally blocked.',
   "- Session replay, $session_id, and person attribution are handled automatically by the host's posthog-js running in the sandbox — you do NOT set session ids or initialise recording; just call ph.capture for custom events.",
   "- Load data inside `useEffect` with `useState`; show a loading state first, then render. Handle the empty/error case. Keep result sets small — aggregate in the query, don't fetch raw event dumps.",
+  "",
+  "LIVE / REALTIME — for anything that must update as events arrive (a live event feed, a users-online counter), use the streaming live events API. This is DIFFERENT from a saved insight: insights are historical snapshots, live events stream in real time from the livestream service:",
+  "- `ph.subscribeLiveEvents(params, onEvent, onEnd)` opens a live SSE subscription and calls `onEvent(event)` for every matching event as it lands. `params`: `{ eventType?: string | null, distinctId?: string | null, columns?: string[] | null, properties?: Array<{ key, operator, value? }> | null }`. IMPORTANT: `properties` filters ONLY support these operators — `exact | is_not | icontains | not_icontains | regex | not_regex | gt | gte | lt | lte | is_set | is_not_set` — any other operator fails. `value` may be a scalar or an array (array = OR). The host holds the live JWT; you never see a token.",
+  "- Each event is `{ distinct_id, event, timestamp, properties }`. You MUST keep the returned subscription and CLOSE it: `const sub = ph.subscribeLiveEvents(...); …; sub.close()` — close it in a useEffect cleanup (and before re-subscribing with new params) so stale streams don't pile up. `onEnd(status, message)` fires when the stream closes/errors — resubscribe from there if you want to retry.",
+  "- `await ph.liveStats()` resolves to the project's realtime counters `{ users_on_product, active_recordings }` — the same \"users online\" pill the PostHog Activity page shows. It's a ONE-SHOT poll, NOT a stream: poll it on an interval (e.g. `setInterval` 30s) and clear that interval on unmount.",
+  '- "Users online" pattern (mirror the app): combine BOTH — the server counter from `ph.liveStats()` AND a client-side sliding-window count you derive from the event stream. Keep `useRef(new Map())` of `distinct_id → unix-ts`, on each event `map.set(event.distinct_id, Date.parse(event.timestamp)/1000)`, and every ~5s drop entries older than 60s (`FILTERED_LIVE_USER_WINDOW_SECONDS` = 60) then set the count to `map.size`. That gives a live count that decays within seconds, exactly like PostHog\'s Web Analytics Live tab.',
 ];
 
 const FREEFORM_STYLE = [

packages/core/src/canvas/freeformSchemas.test.ts

@@ -1,5 +1,10 @@
 import { describe, expect, it } from "vitest";
-import { canvasToHostMessageSchema } from "./freeformSchemas";
+import {
+  canvasLiveEventsSubscribeInput,
+  canvasToHostMessageSchema,
+  hostToCanvasMessageSchema,
+  liveEventPropertyFilterSchema,
+} from "./freeformSchemas";
 
 describe("canvasToHostMessageSchema open-external", () => {
   const message = (url: string) => ({
@@ -33,3 +38,70 @@ describe("canvasToHostMessageSchema open-external", () => {
     );
   });
 });
+
+describe("live-event message round-trip", () => {
+  it("accepts a live-subscribe frame with rich filters", () => {
+    const parsed = canvasToHostMessageSchema.safeParse({
+      channel: "posthog-canvas",
+      type: "live-subscribe",
+      subId: "live-1",
+      params: {
+        eventType: "$pageview",
+        columns: ["$current_url", "$browser"],
+        properties: [
+          { key: "$browser", operator: "exact", value: "Chrome" },
+          {
+            key: "$current_url",
+            operator: "icontains",
+            value: ["docs", "blog"],
+          },
+          { key: "$geoip_country_code", operator: "is_set" },
+        ],
+      },
+    });
+    expect(parsed.success).toBe(true);
+  });
+
+  it("rejects a subscribe frame whose operator is outside the live allowlist", () => {
+    const parsed = liveEventPropertyFilterSchema.safeParse({
+      key: "$browser",
+      operator: "contains_everything", // not a livestream-supported operator
+      value: "x",
+    });
+    expect(parsed.success).toBe(false);
+  });
+
+  it("accepts a live-event frame back into the iframe", () => {
+    const parsed = hostToCanvasMessageSchema.safeParse({
+      channel: "posthog-canvas",
+      type: "live-event",
+      subId: "live-1",
+      event: {
+        distinct_id: "user@example.com",
+        event: "$pageview",
+        timestamp: "2026-07-30T10:00:00.000Z",
+        properties: { $current_url: "https://posthog.com/docs" },
+      },
+    });
+    expect(parsed.success).toBe(true);
+  });
+
+  it("accepts a live-stream-status frame", () => {
+    expect(
+      hostToCanvasMessageSchema.safeParse({
+        channel: "posthog-canvas",
+        type: "live-stream-status",
+        subId: "live-1",
+        status: "error",
+        message: "upstream closed",
+      }).success,
+    ).toBe(true);
+  });
+
+  it("rejects subscribe params with extra/garbage fields", () => {
+    const parsed = canvasLiveEventsSubscribeInput.safeParse({
+      eventType: 42, // must be a string
+    });
+    expect(parsed.success).toBe(false);
+  });
+});