From 236357facbbd34088198c245321b1c8c58c02bc6 Mon Sep 17 00:00:00 2001 From: Tomas Dvorak Date: Thu, 24 Sep 2026 20:55:48 +0200 Subject: [PATCH] feat: Client-published events + realtime presence MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - POST /projects/:id/events accepts client-authored events restricted to the custom.* namespace — system types can't be spoofed. They flow through the same publishEvent fan-out: realtime, webhooks, and event_pattern functions all fire. - Presence: subscribe/unsubscribe bumps a per-project count and broadcasts presence.update {online} to the realtime stream only — deliberately not through publishEvent so webhooks and "*" functions don't fire on connect noise. GET …/realtime/presence reads the count. - @primora/client gains publish() and presence(); CLI gains events:send. Verified live: presence 0->1 on subscribe, presence.update frame delivered, custom event delivered to SSE + fired a custom.* function, spoofed document.created rejected 400. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- CHANGELOG.md | 1 + README.md | 4 +- ROADMAP.md | 10 ++- apps/backend/internal/handlers/events.go | 44 +++++++++++++ apps/backend/internal/handlers/http.go | 2 + apps/backend/internal/services/events_test.go | 22 +++++++ .../internal/services/platform_service.go | 66 ++++++++++++++++++- apps/backend/openapi/openapi.yaml | 65 ++++++++++++++++++ apps/cli/README.md | 9 ++- apps/cli/src/cli.ts | 7 ++ apps/cli/src/commands/events.ts | 48 ++++++++++++++ packages/api-client/src/generated/index.ts | 2 + .../generated/models/PublishEventRequest.ts | 15 +++++ .../generated/schemas/$PublishEventRequest.ts | 21 ++++++ .../generated/services/AutomationService.ts | 53 +++++++++++++++ packages/client/src/realtime.ts | 34 +++++++++- 16 files changed, 393 insertions(+), 10 deletions(-) create mode 100644 apps/backend/internal/handlers/events.go create mode 100644 apps/backend/internal/services/events_test.go create mode 100644 apps/cli/src/commands/events.ts create mode 100644 packages/api-client/src/generated/models/PublishEventRequest.ts create mode 100644 packages/api-client/src/generated/schemas/$PublishEventRequest.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index f8240cd..df97d40 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,7 @@ All notable changes to Primora. Format follows [Keep a Changelog](https://keepac - **Desktop auto-updater** — `tauri-plugin-updater` wired end to end: minisign keypair, `createUpdaterArtifacts`, `latest.json` manifest built from signed artifacts on every release, and a silent background check on the connect screen offering "Update & restart" (download → verify → install → relaunch). Unsigned artifacts can never install — missing signature means no manifest, not a bypass. CI signs via `TAURI_SIGNING_PRIVATE_KEY` secrets; OS-level signing (notarization/Authenticode) remains open. - **Release SBOMs** — CycloneDX SBOM for each published GHCR image (backend/auth/frontend) plus one for the source tree, attached to every GitHub Release via `anchore/sbom-action`. - **Sandboxed function driver** — `FUNCTIONS_DRIVER=docker` runs every invoke in an ephemeral locked-down container: no network, read-only rootfs, 128 MiB memory, 0.5 CPU, 64 pids, non-root, all capabilities dropped. Source is injected via base64 env (no bind mount — works on Docker Desktop and remote daemons); timeout force-removes the container so nothing lingers. Verified live: code runs, network is unreachable inside, and deadline kills land. +- **Client events + presence** — `POST /projects/:id/events` publishes client-authored events (`custom.*` namespace enforced — system types can't be spoofed) through the same fan-out as system events: realtime subscribers, matching webhooks, and `event_pattern` functions all fire. Realtime now tracks presence: `presence.update` {online} broadcasts on every subscribe/unsubscribe (stream-only — never reaches webhooks or functions), `GET …/realtime/presence` reads the count. Client SDK gains `publish()` and `presence()`; CLI gains `events:send`. - **Function triggers** (migration 00016) — three new ways to run a function, all through the same executor and run log. Scheduled jobs accept `function_id` instead of `url` (run status mirrors the exit code); inbound hooks gain `mode: "function"` — the received body becomes the function's stdin payload, acked immediately with the run recorded async; and `event_pattern` on a function (`document.*`, `*`, or an exact type) fires it detached whenever a matching domain event publishes — `{event, project_id, occurred_at, data}` as payload. Runs record trigger source: `manual`/`schedule`/`hook`/`event`. Dashboard: function event-trigger field, job target picker (URL or function), and inbound-hook function mode. ## [0.7.1] - 2026-09-23 diff --git a/README.md b/README.md index 60c21e1..a3a44f5 100644 --- a/README.md +++ b/README.md @@ -50,10 +50,10 @@ Supabase, with no hosted tier - your data stays on your hardware. - **Auth** - email/password plus optional GitHub, Google, Discord, and Microsoft OAuth. JWTs verified by the API against JWKS. - **Storage** - S3-style buckets on local disk or any S3-compatible backend (AWS, MinIO, Garage, R2), public or private. With `s3` the API mints presigned upload/download URLs so large objects skip the backend. - **Collections** - schema-flexible JSON documents in Postgres JSONB, with filter/order queries. -- **Functions** - project-scoped JS/TS executed by bun/deno on the host, with per-run logs. Trigger them four ways: manual invoke, cron schedules (`function_id` on a job), inbound hooks (`mode: "function"`), or domain events (`event_pattern` like `document.*`). +- **Functions** - project-scoped JS/TS run by bun/deno — on the host (`exec`) or in ephemeral locked-down containers (`docker` driver) — with per-run logs. Trigger them four ways: manual invoke, cron schedules (`function_id` on a job), inbound hooks (`mode: "function"`), or domain events (`event_pattern` like `document.*`). - **Vault** - AES-256-GCM project secrets plus an Argon2id/XChaCha20 local vault in the CLI and desktop app. - **Automation** - scheduled jobs, inbound webhooks, telemetry alert rules, and `secret://` refs resolved at delivery. -- **Realtime** - SSE event stream with a typed client SDK (`@primora/client`). +- **Realtime** - SSE event stream with a typed client SDK (`@primora/client`) — channel subscriptions, presence counts, and client-published `custom.*` events that fan out to webhooks and functions. - **API keys** - `prm__` credentials, shown once. - **Audit log** - every mutating request recorded, with CSV/JSON export. - **Demo mode** - fully client-side seeded workspace via `?demo=true`. diff --git a/ROADMAP.md b/ROADMAP.md index 14757de..f2cfe9d 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -309,8 +309,14 @@ Channel subscriptions over the Phase 7 SSE stream, in `@primora/client`. 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. +- **Presence** ✅ — server tracks realtime subscribers per project; + `presence.update` {online: n} broadcasts on every join/leave (realtime-only, + never fans out to webhooks/functions), `GET …/realtime/presence` reads the + count, `client.presence()` wraps it. +- **Client events** ✅ — `POST /projects/:id/events` publishes `custom.*` + types through the standard fan-out (realtime + webhooks + functions); + `client.publish("custom.x", data)` and `primora events:send` both reach it. + The `custom.*` prefix is enforced so clients can't spoof system events. ## Notifications — alert rules (unreleased) diff --git a/apps/backend/internal/handlers/events.go b/apps/backend/internal/handlers/events.go new file mode 100644 index 0000000..359e8d0 --- /dev/null +++ b/apps/backend/internal/handlers/events.go @@ -0,0 +1,44 @@ +package handlers + +import ( + "net/http" + + "github.com/gin-gonic/gin" +) + +// publishEvent accepts a client-authored domain event (custom.* namespace) +// and fans it out through the standard publishEvent path — realtime +// subscribers, matching webhooks, and event_pattern functions all fire. +func (h *HTTPHandler) publishEvent(c *gin.Context) { + actor, projectID, ok := h.actorAndProject(c) + if !ok { + return + } + var body struct { + Type string `json:"type" validate:"required"` + Data map[string]any `json:"data"` + } + if !h.bindAndValidate(c, &body) { + return + } + if err := h.Platform.PublishProjectEvent(c.Request.Context(), actor, projectID, body.Type, body.Data); err != nil { + h.handleError(c, err) + return + } + c.JSON(http.StatusAccepted, gin.H{"published": body.Type}) +} + +// realtimePresence returns the count of clients currently subscribed to the +// project's realtime stream. +func (h *HTTPHandler) realtimePresence(c *gin.Context) { + actor, projectID, ok := h.actorAndProject(c) + if !ok { + return + } + n, err := h.Platform.Presence(c.Request.Context(), actor, projectID) + if err != nil { + h.handleError(c, err) + return + } + c.JSON(http.StatusOK, gin.H{"online": n}) +} diff --git a/apps/backend/internal/handlers/http.go b/apps/backend/internal/handlers/http.go index cda314c..4f89616 100644 --- a/apps/backend/internal/handlers/http.go +++ b/apps/backend/internal/handlers/http.go @@ -92,6 +92,7 @@ func (h *HTTPHandler) Register(router *gin.Engine) { api.POST("/projects/:projectID/functions/:functionID/invoke", h.invokeFunction) api.GET("/projects/:projectID/functions/:functionID/runs", h.listFunctionRuns) api.GET("/projects/:projectID/events", h.listEvents) + api.POST("/projects/:projectID/events", h.publishEvent) api.GET("/projects/:projectID/issues", h.listIssues) api.GET("/projects/:projectID/components", h.listComponents) api.DELETE("/projects/:projectID/components/:componentID", h.deleteComponent) @@ -139,6 +140,7 @@ func (h *HTTPHandler) Register(router *gin.Engine) { api.PUT("/projects/:projectID/alerts/:ruleID", h.updateAlertRule) api.DELETE("/projects/:projectID/alerts/:ruleID", h.deleteAlertRule) api.GET("/projects/:projectID/realtime/stream", h.realtimeStream) + api.GET("/projects/:projectID/realtime/presence", h.realtimePresence) api.GET("/buckets/:bucketID/objects", h.listObjects) api.POST("/buckets/:bucketID/objects", h.uploadObject) api.POST("/buckets/:bucketID/object-copies", h.copyObject) diff --git a/apps/backend/internal/services/events_test.go b/apps/backend/internal/services/events_test.go new file mode 100644 index 0000000..d480431 --- /dev/null +++ b/apps/backend/internal/services/events_test.go @@ -0,0 +1,22 @@ +package services + +import "testing" + +func TestCustomEventTypeRe(t *testing.T) { + good := []string{"custom.ping", "custom.deploy.done", "custom.a", "custom.x-y_z.9"} + bad := []string{ + "", "custom", "custom.", "document.created", "CUSTOM.ping", + "custom.Ping", "custom..x", "custom.-x", "custom." + string(rune(0x20)) + "x", + "custom." + string(make([]byte, 64)), + } + for _, s := range good { + if !customEventTypeRe.MatchString(s) { + t.Errorf("expected %q to match", s) + } + } + for _, s := range bad { + if customEventTypeRe.MatchString(s) { + t.Errorf("expected %q to be rejected", s) + } + } +} diff --git a/apps/backend/internal/services/platform_service.go b/apps/backend/internal/services/platform_service.go index e133622..5ee0e0d 100644 --- a/apps/backend/internal/services/platform_service.go +++ b/apps/backend/internal/services/platform_service.go @@ -13,7 +13,9 @@ import ( "net/http" "net/url" "path/filepath" + "regexp" "strings" + "sync" "time" "github.com/google/uuid" @@ -42,6 +44,10 @@ type PlatformService struct { alerts *alertEvaluator functions FunctionRunner logger *slog.Logger + + // presenceMu guards presence: realtime subscriber counts per project. + presenceMu sync.Mutex + presence map[uuid.UUID]int } type BootstrapInput struct { @@ -228,7 +234,7 @@ type InvitationSummary struct { } func NewPlatformService(repo *repositories.CoreRepository, store storage.Store, mailer *Mailer, publicURL string, dbxClient *dbx.Client, enc *secrets.Encryptor, logger *slog.Logger, functions FunctionRunner) *PlatformService { - s := &PlatformService{repo: repo, store: store, mailer: mailer, publicURL: publicURL, dbx: dbxClient, hub: NewEventHub(), realtime: NewEventHub(), enc: enc, functions: functions, logger: logger} + s := &PlatformService{repo: repo, store: store, mailer: mailer, publicURL: publicURL, dbx: dbxClient, hub: NewEventHub(), realtime: NewEventHub(), enc: enc, functions: functions, logger: logger, presence: map[uuid.UUID]int{}} if enc != nil { s.dispatcher = NewWebhookDispatcher(repo.Queries(), enc, logger) s.dispatcher.Start(context.Background()) @@ -333,16 +339,70 @@ func (s *PlatformService) publishEvent(ctx context.Context, projectID uuid.UUID, s.triggerFunctionsForEvent(ctx, projectID, eventType, data) } -// SubscribeRealtime registers an SSE consumer for project domain events. +// SubscribeRealtime registers an SSE consumer for project domain events and +// bumps the project's presence count. func (s *PlatformService) SubscribeRealtime(ctx context.Context, actor *models.Actor, projectID uuid.UUID) (chan []byte, error) { if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer", "viewer"); err != nil { return nil, err } - return s.realtime.Subscribe(projectID.String()), nil + ch := s.realtime.Subscribe(projectID.String()) + s.bumpPresence(projectID, 1) + return ch, nil } func (s *PlatformService) UnsubscribeRealtime(projectID uuid.UUID, ch chan []byte) { s.realtime.Unsubscribe(projectID.String(), ch) + s.bumpPresence(projectID, -1) +} + +// bumpPresence adjusts the online count and broadcasts presence.update to +// realtime subscribers only — deliberately not through publishEvent, so +// webhooks and `event_pattern: "*"` functions don't fire on connect noise. +func (s *PlatformService) bumpPresence(projectID uuid.UUID, delta int) { + s.presenceMu.Lock() + n := s.presence[projectID] + delta + if n <= 0 { + delete(s.presence, projectID) + n = 0 + } else { + s.presence[projectID] = n + } + s.presenceMu.Unlock() + s.realtime.Broadcast(projectID.String(), map[string]any{ + "type": "presence.update", + "occurred_at": time.Now().UTC().Format(time.RFC3339), + "data": map[string]any{"online": n}, + }) +} + +// Presence returns the current realtime subscriber count for a project. +func (s *PlatformService) Presence(ctx context.Context, actor *models.Actor, projectID uuid.UUID) (int, error) { + if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer", "viewer"); err != nil { + return 0, err + } + s.presenceMu.Lock() + defer s.presenceMu.Unlock() + return s.presence[projectID], nil +} + +var customEventTypeRe = regexp.MustCompile(`^custom\.[a-z0-9][a-z0-9_.-]{0,62}$`) + +// PublishProjectEvent lets a member emit a custom domain event — realtime +// subscribers, matching webhooks, and event_pattern functions all fire via +// the standard publishEvent fan-out. The custom.* prefix is enforced so +// clients can't spoof system event types (document.created etc.). +func (s *PlatformService) PublishProjectEvent(ctx context.Context, actor *models.Actor, projectID uuid.UUID, eventType string, data map[string]any) error { + if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer"); err != nil { + return err + } + if !customEventTypeRe.MatchString(eventType) { + return errors.New("invalid event type: must match custom. (lowercase letters, digits, dots, dashes)") + } + if data == nil { + data = map[string]any{} + } + s.publishEvent(ctx, projectID, eventType, data) + return nil } func (s *PlatformService) Me(ctx context.Context, actor *models.Actor) (PlatformSummary, error) { diff --git a/apps/backend/openapi/openapi.yaml b/apps/backend/openapi/openapi.yaml index 740c1de..97756f7 100644 --- a/apps/backend/openapi/openapi.yaml +++ b/apps/backend/openapi/openapi.yaml @@ -1331,6 +1331,37 @@ paths: $ref: "#/components/schemas/TelemetryEventListResponse" "401": $ref: "#/components/responses/ErrorResponse" + post: + tags: [Automation] + operationId: publishProjectEvent + summary: Publishes a client-authored domain event. Type must be in the custom.* namespace; the event fans out to realtime subscribers, matching webhooks, and event_pattern functions. + security: + - bearerAuth: [] + - apiKeyAuth: [] + parameters: + - $ref: "#/components/parameters/ProjectID" + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/PublishEventRequest" + responses: + "202": + description: Event accepted for fan-out + content: + application/json: + schema: + type: object + properties: + published: + type: string + "400": + $ref: "#/components/responses/ErrorResponse" + "401": + $ref: "#/components/responses/ErrorResponse" + "403": + $ref: "#/components/responses/ErrorResponse" /projects/{projectID}/issues: get: tags: [Telemetry] @@ -1887,6 +1918,28 @@ paths: description: text/event-stream of {type, occurred_at, data} messages "401": $ref: "#/components/responses/ErrorResponse" + /projects/{projectID}/realtime/presence: + get: + tags: [Automation] + operationId: getRealtimePresence + summary: Count of clients currently subscribed to the project's realtime stream. The stream also emits presence.update events on every join/leave. + security: + - bearerAuth: [] + - apiKeyAuth: [] + parameters: + - $ref: "#/components/parameters/ProjectID" + responses: + "200": + description: Online subscriber count + content: + application/json: + schema: + type: object + properties: + online: + type: integer + "401": + $ref: "#/components/responses/ErrorResponse" /projects/{projectID}/secrets: get: tags: [Secrets] @@ -3191,6 +3244,18 @@ components: type: string format: date-time required: [url, method, expires_at] + PublishEventRequest: + type: object + properties: + type: + type: string + pattern: '^custom\.[a-z0-9][a-z0-9_.-]{0,62}$' + description: Event type in the custom.* namespace — realtime subscribers, matching webhooks, and event_pattern functions all fire. + data: + type: object + additionalProperties: true + description: Arbitrary JSON payload carried to subscribers. + required: [type] PaginationMetadata: type: object properties: diff --git a/apps/cli/README.md b/apps/cli/README.md index abe9802..fcccea3 100644 --- a/apps/cli/README.md +++ b/apps/cli/README.md @@ -21,7 +21,14 @@ project interactively. ## Commands `orgs`, `projects`, `buckets`, `objects` (list/upload/download/rm), `keys`, -`jobs`, `audit list --follow`. Every command accepts `--json`. +`jobs`, `events:send`, `audit list --follow`. Every command accepts `--json`. + +```sh +primora events:send custom.deploy.done --data '{"sha":"abc123"}' +``` + +`events:send` publishes a `custom.*` event — realtime subscribers, matching +webhooks, and `event_pattern` functions all fire. ```sh primora documents list --collection users \ diff --git a/apps/cli/src/cli.ts b/apps/cli/src/cli.ts index 5075e01..934bda3 100644 --- a/apps/cli/src/cli.ts +++ b/apps/cli/src/cli.ts @@ -3,6 +3,7 @@ import { cac } from "cac"; import { cmdLogin, cmdLogout, cmdWhoami } from "./commands/auth.js"; import { cmdAuditList } from "./commands/audit.js"; import { cmdDocumentsList } from "./commands/documents.js"; +import { cmdEventsSend } from "./commands/events.js"; import { cmdBucketsCreate, cmdBucketsList } from "./commands/buckets.js"; import { cmdContext, cmdUse } from "./commands/context.js"; import { cmdInject } from "./commands/inject.js"; @@ -174,6 +175,12 @@ cli .option("--offset ", "Page offset") .action(cmdDocumentsList); +cli + .command("events:send ", "Publish a custom.* event to realtime subscribers, webhooks, and functions") + .option("--project ", "Project override") + .option("--data ", "JSON object payload") + .action(cmdEventsSend); + cli .command("audit:list", "List audit events for the current project") .option("--project ", "Project override") diff --git a/apps/cli/src/commands/events.ts b/apps/cli/src/commands/events.ts new file mode 100644 index 0000000..e7e5f87 --- /dev/null +++ b/apps/cli/src/commands/events.ts @@ -0,0 +1,48 @@ +import { AutomationService } from "@primora/api-client"; + +import { loadConfig } from "../config.js"; +import { CliError, configureClient, requireAuth, requireProject } from "../http.js"; +import { isJson, printJson } from "../out.js"; + +interface EventsSendOptions { + project?: string; + data?: string; + json?: boolean; +} + +export async function cmdEventsSend(type: string, options: EventsSendOptions): Promise { + if (!type.startsWith("custom.")) { + throw new CliError( + `Invalid event type "${type}".`, + "Client events must live in the custom.* namespace — e.g. custom.deploy.done", + ); + } + let data: Record = {}; + if (options.data) { + try { + const parsed = JSON.parse(options.data); + if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) { + throw new CliError("--data must be a JSON object.", `Got: ${options.data.slice(0, 60)}`); + } + data = parsed; + } catch (e) { + if (e instanceof CliError) throw e; + throw new CliError("--data is not valid JSON.", (e as Error).message); + } + } + + const cfg = loadConfig(); + requireAuth(cfg); + configureClient(cfg); + const projectId = requireProject(cfg, options.project); + + const res = await AutomationService.publishProjectEvent({ + projectId, + requestBody: { type, data }, + }); + if (isJson(options)) { + printJson(res); + return; + } + console.log(`Published ${res.published ?? type} — realtime subscribers, matching webhooks, and event_pattern functions fire.`); +} diff --git a/packages/api-client/src/generated/index.ts b/packages/api-client/src/generated/index.ts index ec3d314..7004bba 100644 --- a/packages/api-client/src/generated/index.ts +++ b/packages/api-client/src/generated/index.ts @@ -117,6 +117,7 @@ export type { ProjectOverview } from './models/ProjectOverview'; export type { ProjectSecret } from './models/ProjectSecret'; export type { ProjectSecretListResponse } from './models/ProjectSecretListResponse'; export type { ProjectSummary } from './models/ProjectSummary'; +export type { PublishEventRequest } from './models/PublishEventRequest'; export type { ReadinessStatus } from './models/ReadinessStatus'; export type { RevealProjectSecretResponse } from './models/RevealProjectSecretResponse'; export type { RybbitMetricItem } from './models/RybbitMetricItem'; @@ -269,6 +270,7 @@ export { $ProjectOverview } from './schemas/$ProjectOverview'; export { $ProjectSecret } from './schemas/$ProjectSecret'; export { $ProjectSecretListResponse } from './schemas/$ProjectSecretListResponse'; export { $ProjectSummary } from './schemas/$ProjectSummary'; +export { $PublishEventRequest } from './schemas/$PublishEventRequest'; export { $ReadinessStatus } from './schemas/$ReadinessStatus'; export { $RevealProjectSecretResponse } from './schemas/$RevealProjectSecretResponse'; export { $RybbitMetricItem } from './schemas/$RybbitMetricItem'; diff --git a/packages/api-client/src/generated/models/PublishEventRequest.ts b/packages/api-client/src/generated/models/PublishEventRequest.ts new file mode 100644 index 0000000..3b33eb0 --- /dev/null +++ b/packages/api-client/src/generated/models/PublishEventRequest.ts @@ -0,0 +1,15 @@ +/* generated using openapi-typescript-codegen -- do not edit */ +/* istanbul ignore file */ +/* tslint:disable */ +/* eslint-disable */ +export type PublishEventRequest = { + /** + * Event type in the custom.* namespace — realtime subscribers, matching webhooks, and event_pattern functions all fire. + */ + type: string; + /** + * Arbitrary JSON payload carried to subscribers. + */ + data?: Record; +}; + diff --git a/packages/api-client/src/generated/schemas/$PublishEventRequest.ts b/packages/api-client/src/generated/schemas/$PublishEventRequest.ts new file mode 100644 index 0000000..35e32b2 --- /dev/null +++ b/packages/api-client/src/generated/schemas/$PublishEventRequest.ts @@ -0,0 +1,21 @@ +/* generated using openapi-typescript-codegen -- do not edit */ +/* istanbul ignore file */ +/* tslint:disable */ +/* eslint-disable */ +export const $PublishEventRequest = { + properties: { + type: { + type: 'string', + description: `Event type in the custom.* namespace — realtime subscribers, matching webhooks, and event_pattern functions all fire.`, + isRequired: true, + pattern: '^custom\\.[a-z0-9][a-z0-9_.-]{0,62}$', + }, + data: { + type: 'dictionary', + contains: { + properties: { + }, + }, + }, + }, +} as const; diff --git a/packages/api-client/src/generated/services/AutomationService.ts b/packages/api-client/src/generated/services/AutomationService.ts index f6663ec..683836f 100644 --- a/packages/api-client/src/generated/services/AutomationService.ts +++ b/packages/api-client/src/generated/services/AutomationService.ts @@ -3,6 +3,7 @@ /* tslint:disable */ /* eslint-disable */ import type { CreateScheduledJobRequest } from '../models/CreateScheduledJobRequest'; +import type { PublishEventRequest } from '../models/PublishEventRequest'; import type { ScheduledJobCreateResponse } from '../models/ScheduledJobCreateResponse'; import type { ScheduledJobListResponse } from '../models/ScheduledJobListResponse'; import type { ScheduledJobRun } from '../models/ScheduledJobRun'; @@ -12,6 +13,35 @@ import type { CancelablePromise } from '../core/CancelablePromise'; import { OpenAPI } from '../core/OpenAPI'; import { request as __request } from '../core/request'; export class AutomationService { + /** + * Publishes a client-authored domain event. Type must be in the custom.* namespace; the event fans out to realtime subscribers, matching webhooks, and event_pattern functions. + * @returns any Event accepted for fan-out + * @throws ApiError + */ + public static publishProjectEvent({ + projectId, + requestBody, + }: { + projectId: string, + requestBody: PublishEventRequest, + }): CancelablePromise<{ + published?: string; + }> { + return __request(OpenAPI, { + method: 'POST', + url: '/projects/{projectID}/events', + path: { + 'projectID': projectId, + }, + body: requestBody, + mediaType: 'application/json', + errors: { + 400: `Error response`, + 401: `Error response`, + 403: `Error response`, + }, + }); + } /** * @returns ScheduledJobListResponse Scheduled jobs for the project * @throws ApiError @@ -191,4 +221,27 @@ export class AutomationService { }, }); } + /** + * Count of clients currently subscribed to the project's realtime stream. The stream also emits presence.update events on every join/leave. + * @returns any Online subscriber count + * @throws ApiError + */ + public static getRealtimePresence({ + projectId, + }: { + projectId: string, + }): CancelablePromise<{ + online?: number; + }> { + return __request(OpenAPI, { + method: 'GET', + url: '/projects/{projectID}/realtime/presence', + path: { + 'projectID': projectId, + }, + errors: { + 401: `Error response`, + }, + }); + } } diff --git a/packages/client/src/realtime.ts b/packages/client/src/realtime.ts index 3e9f105..97646e4 100644 --- a/packages/client/src/realtime.ts +++ b/packages/client/src/realtime.ts @@ -2,8 +2,10 @@ // (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. +// Presence: the server broadcasts `presence.update` {online: n} on every +// subscriber join/leave — subscribe via channel("presence").on("update"). +// Client-broadcast: publish() emits a custom.* event through the server +// fan-out (webhooks and event_pattern functions fire too). export interface RealtimeEvent { /** e.g. "document.created", "object.deleted", "issue.created", "deploy.marker" */ @@ -41,6 +43,13 @@ export interface RealtimeClient { * `channel("documents")` and `channel("document")` are equivalent. */ channel(name: string): RealtimeChannel; + /** + * Publishes a `custom.*` event to the project — realtime subscribers, + * matching webhooks, and event_pattern functions all receive it. + */ + publish(type: string, data?: Record): Promise; + /** Current count of clients subscribed to the project's realtime stream. */ + presence(): Promise; close(): void; readonly state: RealtimeState; } @@ -142,6 +151,27 @@ export function createRealtimeClient(opts: RealtimeClientOptions): RealtimeClien }, }; }, + async publish(type, data = {}) { + const res = await fetch( + opts.endpoint.replace(/\/$/, "") + + `/api/v1/projects/${encodeURIComponent(opts.projectId)}/events`, + { + method: "POST", + headers: { "X-API-Key": opts.key, "Content-Type": "application/json" }, + body: JSON.stringify({ type, data }), + }, + ); + if (!res.ok) throw new Error(`publish event: HTTP ${res.status}`); + }, + async presence() { + const res = await fetch( + opts.endpoint.replace(/\/$/, "") + + `/api/v1/projects/${encodeURIComponent(opts.projectId)}/realtime/presence`, + { headers: { "X-API-Key": opts.key } }, + ); + if (!res.ok) throw new Error(`presence: HTTP ${res.status}`); + return ((await res.json()) as { online: number }).online; + }, close: () => abort.abort(), get state() { return state;