diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1fe8ed8..1384575 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -236,3 +236,6 @@ jobs: with: version: latest working-directory: apps/backend + # Prebuilt binaries lag the repo's Go version — build the linter + # with the toolchain above so it understands go 1.26. + install-mode: goinstall diff --git a/CHANGELOG.md b/CHANGELOG.md index 1e6d15a..26c930b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,21 @@ All notable changes to Primora. Format follows [Keep a Changelog](https://keepac ## [Unreleased] +### Added + +- **Scheduled jobs** — `core.scheduled_jobs` + `core.scheduled_job_runs` (migration 00009). Cron expressions and descriptors (`@hourly`, `@daily`, `@every 30m`, `*/15 * * * *`) parsed via robfig/cron; an in-process `JobScheduler` scans every 15 s, queues due jobs through a buffered channel, and collapses missed schedules into a single run. Endpoints under `/projects/:id/jobs`: list/create/update/delete, `GET …/jobs/:id/runs`, `POST …/jobs/:id/run` (manual execution). Jobs deliver a JSON envelope (event type, project/job/run IDs, trigger source, timestamp, static payload) via POST with `X-Primora-Signature: sha256=`, `X-Primora-Event`, `X-Primora-Job`, `X-Primora-Run` headers and a 10 s timeout. Secrets are AES-256-GCM sealed at rest; a generated secret is returned exactly once on create/rotate (`has_secret` thereafter). Run records carry status, trigger, HTTP status, duration, error text, and timestamps; history is pruned by retention. Reads require viewer+, writes admin/developer. +- **Domain events + realtime stream** — `PlatformService.publishEvent` fans project mutations out to realtime subscribers and matching webhooks through one path. New webhook event types: `job.run`, `document.created`/`updated`/`deleted`, `object.created`/`updated`/`deleted`. `GET …/projects/:projectID/realtime/stream` serves an SSE feed with 25 s keepalives and `?token=`/`?api_key=` query auth for `EventSource` clients. +- **Automation page** — SolidJS view with a Schedules tab (job table, create modal with schedule helper text, expandable run history, manual Run, enable/disable, delete, secret-shown-once banner) and a Live events tab (live/offline badge, pause, clear, auto-reconnect). Sidebar + command palette entries; demo-mode seeds and stubs included. +- **`primora jobs:*` CLI group** — `jobs:list`, `jobs:create --schedule --url [--payload --secret --disabled]`, `jobs:run `, `jobs:runs --limit`, `jobs:rm --yes`. Both `jobs list` and `jobs:list` spellings. +- **New `primora_*` MCP tools** — `primora_list_jobs`, `primora_create_job`, `primora_update_job`, `primora_delete_job`, `primora_list_job_runs`, `primora_run_job`. Webhook event schema extended to all new event types. + +### Fixed + +- **SSE broken by gzip middleware** — `Compression()` wrapped every `Accept-Encoding: gzip` response, and the gzip buffer swallowed each `Flush()`, so `EventSource` received headers then silence and closed. Paths ending in `/stream` now bypass compression; this repaired the pre-existing telemetry stream too. +- **Event payloads serialized as base64** — `document`/`object` fields inside domain-event envelopes were `json.RawMessage` placed into `map[string]any`, which marshals as base64. They're now wrapped so the raw document embeds in the event JSON. +- **Auth trusted origins ignored the configured public URL** — `docker-compose.yml` hardcoded `http://localhost` for `BETTER_AUTH_URL`, `AUTH_BASE_URL`, and `VITE_APP_URL`, so any non-default `NGINX_PORT` or domain broke sign-in with a 403. All three now read `.env` with localhost defaults. +- **Frontend auth base URL on non-default ports** — the bundled default baked `http://localhost`; `auth-client` now resolves `/auth` against `window.location.origin` (absolute, as better-auth requires) when `VITE_AUTH_BASE_URL` is unset. + ### Security - **Managed platform connections removed** — `platform-postgres`/`platform-dragonfly` seeds carried the backend's own `DATABASE_URL`/`DRAGONFLY_URL` into every project, where any member could run arbitrary queries through DBX against `core` tables. The seeds are gone and migration 00008 deletes existing managed rows. Operators keep direct `psql`/`redis-cli` access. diff --git a/ROADMAP.md b/ROADMAP.md index ace8a88..410fbfb 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -219,12 +219,82 @@ release pipeline with images + binaries + desktop bundles. smoke green in CI, and the upgrade runbook followed once on a real upgrade. Docs site revisited here, not before. +## Phase 7 — Automation & realtime (v0.7.0) ✅ + +The last Supabase/Appwrite-parity gaps: scheduled work and a live view of +what the platform is doing. One event fan-out drives webhooks and the +dashboard's live feed off the same stream. + +- **Scheduled jobs** — `core.scheduled_jobs` + `core.scheduled_job_runs` + (migration 00009). Cron expressions and descriptors (`@hourly`, + `@daily`, `@every 30m`, `*/15 * * * *`) via robfig/cron. In-process + `JobScheduler`: 15 s due-scan, buffered execution queue, missed + schedules collapse into one run, run history pruned by retention. +- **Signed delivery** — jobs POST a JSON envelope (event, project/job/run + IDs, trigger, timestamp, static payload) with + `X-Primora-Signature: sha256=`, 10 s timeout. Secrets + are AES-256-GCM sealed at rest and the generated value is returned + exactly once (`has_secret` thereafter). +- **Run history** — status (running/success/failed), trigger source, + HTTP status, duration, error text, timestamps; `POST …/jobs/:id/run` + for manual execution. Role-gated: viewers read, admin/developer write. +- **Domain events** — `publishEvent` fans document/object/issue/deploy/ + job mutations out to realtime subscribers and matching webhooks. + Webhook event filter extended to `job.run`, `document.*`, `object.*`. +- **Realtime stream** — `GET …/projects/:projectID/realtime/stream` SSE, + 25 s keepalives, `?token=`/`?api_key=` query auth for `EventSource`. +- **Automation page** — Schedules tab (table, create modal, run history + expansion, manual run, enable/disable, delete, secret-shown-once) and + Live events tab (live badge, pause, clear, reconnect). Demo-mode seeds + and stubs included; command palette entry added. +- **Surfaces** — `primora jobs:*` CLI group (list/create/run/runs/rm), + six new `primora_*_job*` MCP tools, `AutomationService` in the + generated client. + +### Fixes folded in + +- **gzip swallowed SSE** — the compression middleware buffered every + flush, so `EventSource` saw headers then nothing and died. `*/stream` + paths now bypass compression (telemetry stream was equally broken). +- **Trusted origins ignored `PUBLIC_URL`** — compose hardcoded + `http://localhost` for `BETTER_AUTH_URL`/`AUTH_BASE_URL`/`VITE_APP_URL`; + they now read env so a non-default port/domain actually works. +- **Relative auth base URL** — better-auth rejects relative URLs; + `auth-client` now resolves `/auth` against `window.location.origin`, + preserving same-origin defaults without breaking split-origin dev. + +## 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. +- **Phase 10 — Storage backends** — S3-compatible object storage behind + the existing bucket abstraction (deferred from Phase 6), plus + presigned-URL uploads for large files. +- **Phase 11 — Generated REST per collection** — PostgREST-style + auto-CRUD + filtering over Collections documents; a clean read layer + on data that already exists. +- **Notifications** — alert rules on telemetry (error-rate thresholds, + heartbeat silence) fanning out through the webhook dispatcher. +- **Inbound webhooks** — project-scoped ingest endpoints that turn + external events into domain events/jobs. +- **HA mode** — leader-elected scheduler + distributed locks so two + backend replicas don't double-fire jobs; Postgres advisory locks are + enough at this scale. +- **Email surface** — transactional templates + send log in the + dashboard, backed by the auth service's existing SMTP/Resend wiring. + --- ## What is deliberately out of scope - Managed/hosted tier — the product is the self-hosted thing. -- Realtime subscriptions, edge functions, generated REST/GraphQL per-table - (Supabase PostgREST-style). Revisit after Phase 3 if wanted — Collections - already cover the common case. +- Per-request compute billing / hosted function runtime — if the + Functions phase lands it stays user-managed infrastructure. - Embedding DBX's desktop UI — we use its MCP server as a library, not its app. diff --git a/apps/backend/db/migrations/00009_scheduled_jobs.sql b/apps/backend/db/migrations/00009_scheduled_jobs.sql new file mode 100644 index 0000000..5ab684c --- /dev/null +++ b/apps/backend/db/migrations/00009_scheduled_jobs.sql @@ -0,0 +1,39 @@ +-- +goose Up +CREATE TABLE core.scheduled_jobs ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + project_id UUID NOT NULL REFERENCES core.projects(id) ON DELETE CASCADE, + name TEXT NOT NULL, + schedule TEXT NOT NULL, + url TEXT NOT NULL, + secret BYTEA NOT NULL DEFAULT ''::bytea, + payload JSONB NOT NULL DEFAULT '{}'::jsonb, + enabled BOOLEAN NOT NULL DEFAULT TRUE, + last_run_at TIMESTAMPTZ, + last_status TEXT NOT NULL DEFAULT '' CHECK (last_status IN ('', 'success', 'failed')), + next_run_at TIMESTAMPTZ, + created_by_user_id UUID REFERENCES core.users(id) ON DELETE SET NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + UNIQUE (project_id, name) +); + +CREATE INDEX idx_scheduled_jobs_due ON core.scheduled_jobs(next_run_at) + WHERE enabled AND next_run_at IS NOT NULL; + +CREATE TABLE core.scheduled_job_runs ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + job_id UUID NOT NULL REFERENCES core.scheduled_jobs(id) ON DELETE CASCADE, + status TEXT NOT NULL DEFAULT 'running' CHECK (status IN ('running', 'success', 'failed')), + triggered_by TEXT NOT NULL DEFAULT 'schedule' CHECK (triggered_by IN ('schedule', 'manual')), + status_code INT, + error TEXT NOT NULL DEFAULT '', + duration_ms INT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + finished_at TIMESTAMPTZ +); + +CREATE INDEX idx_scheduled_job_runs_job ON core.scheduled_job_runs(job_id, created_at DESC); + +-- +goose Down +DROP TABLE IF EXISTS core.scheduled_job_runs; +DROP TABLE IF EXISTS core.scheduled_jobs; diff --git a/apps/backend/db/queries/jobs.sql b/apps/backend/db/queries/jobs.sql new file mode 100644 index 0000000..98391c7 --- /dev/null +++ b/apps/backend/db/queries/jobs.sql @@ -0,0 +1,88 @@ +-- name: ListScheduledJobs :many +SELECT * FROM core.scheduled_jobs +WHERE project_id = $1 +ORDER BY created_at ASC; + +-- name: GetScheduledJobByID :one +SELECT * FROM core.scheduled_jobs +WHERE id = $1; + +-- name: CreateScheduledJob :one +INSERT INTO core.scheduled_jobs ( + project_id, + name, + schedule, + url, + secret, + payload, + enabled, + next_run_at, + created_by_user_id +) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) +RETURNING *; + +-- name: UpdateScheduledJob :one +UPDATE core.scheduled_jobs +SET name = $3, + schedule = $4, + url = $5, + enabled = $6, + next_run_at = $7, + secret = CASE WHEN $8::boolean THEN $9 ELSE secret END, + payload = CASE WHEN $10::boolean THEN $11 ELSE payload END, + updated_at = NOW() +WHERE id = $1 + AND project_id = $2 +RETURNING *; + +-- name: DeleteScheduledJob :one +DELETE FROM core.scheduled_jobs +WHERE id = $1 + AND project_id = $2 +RETURNING id; + +-- name: ListDueScheduledJobs :many +SELECT * FROM core.scheduled_jobs +WHERE enabled + AND next_run_at IS NOT NULL + AND next_run_at <= $1; + +-- name: MarkScheduledJobRan :one +UPDATE core.scheduled_jobs +SET last_run_at = $2, + last_status = $3, + next_run_at = $4, + updated_at = NOW() +WHERE id = $1 +RETURNING *; + +-- name: InsertScheduledJobRun :one +INSERT INTO core.scheduled_job_runs (job_id, triggered_by) +VALUES ($1, $2) +RETURNING *; + +-- name: FinishScheduledJobRun :one +UPDATE core.scheduled_job_runs +SET status = $2, + status_code = $3, + error = $4, + duration_ms = $5, + finished_at = NOW() +WHERE id = $1 +RETURNING *; + +-- name: ListScheduledJobRuns :many +SELECT * FROM core.scheduled_job_runs +WHERE job_id = $1 +ORDER BY created_at DESC +LIMIT $2; + +-- name: PruneScheduledJobRuns :exec +DELETE FROM core.scheduled_job_runs r +WHERE r.job_id = $1 + AND r.id NOT IN ( + SELECT k.id FROM core.scheduled_job_runs k + WHERE k.job_id = $1 + ORDER BY k.created_at DESC + LIMIT $2 + ); diff --git a/apps/backend/go.mod b/apps/backend/go.mod index 2cb441f..b33513f 100644 --- a/apps/backend/go.mod +++ b/apps/backend/go.mod @@ -14,6 +14,7 @@ require ( github.com/pressly/goose/v3 v3.26.0 github.com/redis/go-redis/v9 v9.16.0 github.com/resend/resend-go/v2 v2.15.0 + github.com/robfig/cron/v3 v3.0.1 github.com/rs/zerolog v1.34.0 github.com/stretchr/testify v1.11.1 ) diff --git a/apps/backend/go.sum b/apps/backend/go.sum index dda1919..eb94ecc 100644 --- a/apps/backend/go.sum +++ b/apps/backend/go.sum @@ -104,6 +104,8 @@ github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94 github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/resend/resend-go/v2 v2.15.0 h1:B6oMEPf8IEQwn2Ovx/9yymkESLDSeNfLFaNMw+mzHhE= github.com/resend/resend-go/v2 v2.15.0/go.mod h1:3YCb8c8+pLiqhtRFXTyFwlLvfjQtluxOr9HEh2BwCkQ= +github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs= +github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/rs/xid v1.6.0/go.mod h1:7XoLgs4eV+QndskICGsho+ADou8ySMSjJKDIan90Nz0= diff --git a/apps/backend/internal/database/db/jobs.sql.go b/apps/backend/internal/database/db/jobs.sql.go new file mode 100644 index 0000000..299c86c --- /dev/null +++ b/apps/backend/internal/database/db/jobs.sql.go @@ -0,0 +1,441 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.29.0 +// source: jobs.sql + +package db + +import ( + "context" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" +) + +const createScheduledJob = `-- name: CreateScheduledJob :one +INSERT INTO core.scheduled_jobs ( + project_id, + name, + schedule, + url, + secret, + payload, + enabled, + next_run_at, + created_by_user_id +) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) +RETURNING id, project_id, name, schedule, url, secret, payload, enabled, last_run_at, last_status, next_run_at, created_by_user_id, created_at, updated_at +` + +type CreateScheduledJobParams struct { + ProjectID uuid.UUID `json:"project_id"` + Name string `json:"name"` + Schedule string `json:"schedule"` + Url string `json:"url"` + Secret []byte `json:"secret"` + Payload []byte `json:"payload"` + Enabled bool `json:"enabled"` + NextRunAt pgtype.Timestamptz `json:"next_run_at"` + CreatedByUserID pgtype.UUID `json:"created_by_user_id"` +} + +func (q *Queries) CreateScheduledJob(ctx context.Context, arg CreateScheduledJobParams) (CoreScheduledJob, error) { + row := q.db.QueryRow(ctx, createScheduledJob, + arg.ProjectID, + arg.Name, + arg.Schedule, + arg.Url, + arg.Secret, + arg.Payload, + arg.Enabled, + arg.NextRunAt, + arg.CreatedByUserID, + ) + var i CoreScheduledJob + err := row.Scan( + &i.ID, + &i.ProjectID, + &i.Name, + &i.Schedule, + &i.Url, + &i.Secret, + &i.Payload, + &i.Enabled, + &i.LastRunAt, + &i.LastStatus, + &i.NextRunAt, + &i.CreatedByUserID, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} + +const deleteScheduledJob = `-- name: DeleteScheduledJob :one +DELETE FROM core.scheduled_jobs +WHERE id = $1 + AND project_id = $2 +RETURNING id +` + +type DeleteScheduledJobParams struct { + ID uuid.UUID `json:"id"` + ProjectID uuid.UUID `json:"project_id"` +} + +func (q *Queries) DeleteScheduledJob(ctx context.Context, arg DeleteScheduledJobParams) (uuid.UUID, error) { + row := q.db.QueryRow(ctx, deleteScheduledJob, arg.ID, arg.ProjectID) + var id uuid.UUID + err := row.Scan(&id) + return id, err +} + +const finishScheduledJobRun = `-- name: FinishScheduledJobRun :one +UPDATE core.scheduled_job_runs +SET status = $2, + status_code = $3, + error = $4, + duration_ms = $5, + finished_at = NOW() +WHERE id = $1 +RETURNING id, job_id, status, triggered_by, status_code, error, duration_ms, created_at, finished_at +` + +type FinishScheduledJobRunParams struct { + ID uuid.UUID `json:"id"` + Status string `json:"status"` + StatusCode *int32 `json:"status_code"` + Error string `json:"error"` + DurationMs *int32 `json:"duration_ms"` +} + +func (q *Queries) FinishScheduledJobRun(ctx context.Context, arg FinishScheduledJobRunParams) (CoreScheduledJobRun, error) { + row := q.db.QueryRow(ctx, finishScheduledJobRun, + arg.ID, + arg.Status, + arg.StatusCode, + arg.Error, + arg.DurationMs, + ) + var i CoreScheduledJobRun + err := row.Scan( + &i.ID, + &i.JobID, + &i.Status, + &i.TriggeredBy, + &i.StatusCode, + &i.Error, + &i.DurationMs, + &i.CreatedAt, + &i.FinishedAt, + ) + return i, err +} + +const getScheduledJobByID = `-- name: GetScheduledJobByID :one +SELECT id, project_id, name, schedule, url, secret, payload, enabled, last_run_at, last_status, next_run_at, created_by_user_id, created_at, updated_at FROM core.scheduled_jobs +WHERE id = $1 +` + +func (q *Queries) GetScheduledJobByID(ctx context.Context, id uuid.UUID) (CoreScheduledJob, error) { + row := q.db.QueryRow(ctx, getScheduledJobByID, id) + var i CoreScheduledJob + err := row.Scan( + &i.ID, + &i.ProjectID, + &i.Name, + &i.Schedule, + &i.Url, + &i.Secret, + &i.Payload, + &i.Enabled, + &i.LastRunAt, + &i.LastStatus, + &i.NextRunAt, + &i.CreatedByUserID, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} + +const insertScheduledJobRun = `-- name: InsertScheduledJobRun :one +INSERT INTO core.scheduled_job_runs (job_id, triggered_by) +VALUES ($1, $2) +RETURNING id, job_id, status, triggered_by, status_code, error, duration_ms, created_at, finished_at +` + +type InsertScheduledJobRunParams struct { + JobID uuid.UUID `json:"job_id"` + TriggeredBy string `json:"triggered_by"` +} + +func (q *Queries) InsertScheduledJobRun(ctx context.Context, arg InsertScheduledJobRunParams) (CoreScheduledJobRun, error) { + row := q.db.QueryRow(ctx, insertScheduledJobRun, arg.JobID, arg.TriggeredBy) + var i CoreScheduledJobRun + err := row.Scan( + &i.ID, + &i.JobID, + &i.Status, + &i.TriggeredBy, + &i.StatusCode, + &i.Error, + &i.DurationMs, + &i.CreatedAt, + &i.FinishedAt, + ) + return i, err +} + +const listDueScheduledJobs = `-- name: ListDueScheduledJobs :many +SELECT id, project_id, name, schedule, url, secret, payload, enabled, last_run_at, last_status, next_run_at, created_by_user_id, created_at, updated_at FROM core.scheduled_jobs +WHERE enabled + AND next_run_at IS NOT NULL + AND next_run_at <= $1 +` + +func (q *Queries) ListDueScheduledJobs(ctx context.Context, nextRunAt pgtype.Timestamptz) ([]CoreScheduledJob, error) { + rows, err := q.db.Query(ctx, listDueScheduledJobs, nextRunAt) + if err != nil { + return nil, err + } + defer rows.Close() + var items []CoreScheduledJob + for rows.Next() { + var i CoreScheduledJob + if err := rows.Scan( + &i.ID, + &i.ProjectID, + &i.Name, + &i.Schedule, + &i.Url, + &i.Secret, + &i.Payload, + &i.Enabled, + &i.LastRunAt, + &i.LastStatus, + &i.NextRunAt, + &i.CreatedByUserID, + &i.CreatedAt, + &i.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listScheduledJobRuns = `-- name: ListScheduledJobRuns :many +SELECT id, job_id, status, triggered_by, status_code, error, duration_ms, created_at, finished_at FROM core.scheduled_job_runs +WHERE job_id = $1 +ORDER BY created_at DESC +LIMIT $2 +` + +type ListScheduledJobRunsParams struct { + JobID uuid.UUID `json:"job_id"` + Limit int32 `json:"limit"` +} + +func (q *Queries) ListScheduledJobRuns(ctx context.Context, arg ListScheduledJobRunsParams) ([]CoreScheduledJobRun, error) { + rows, err := q.db.Query(ctx, listScheduledJobRuns, arg.JobID, arg.Limit) + if err != nil { + return nil, err + } + defer rows.Close() + var items []CoreScheduledJobRun + for rows.Next() { + var i CoreScheduledJobRun + if err := rows.Scan( + &i.ID, + &i.JobID, + &i.Status, + &i.TriggeredBy, + &i.StatusCode, + &i.Error, + &i.DurationMs, + &i.CreatedAt, + &i.FinishedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listScheduledJobs = `-- name: ListScheduledJobs :many +SELECT id, project_id, name, schedule, url, secret, payload, enabled, last_run_at, last_status, next_run_at, created_by_user_id, created_at, updated_at FROM core.scheduled_jobs +WHERE project_id = $1 +ORDER BY created_at ASC +` + +func (q *Queries) ListScheduledJobs(ctx context.Context, projectID uuid.UUID) ([]CoreScheduledJob, error) { + rows, err := q.db.Query(ctx, listScheduledJobs, projectID) + if err != nil { + return nil, err + } + defer rows.Close() + var items []CoreScheduledJob + for rows.Next() { + var i CoreScheduledJob + if err := rows.Scan( + &i.ID, + &i.ProjectID, + &i.Name, + &i.Schedule, + &i.Url, + &i.Secret, + &i.Payload, + &i.Enabled, + &i.LastRunAt, + &i.LastStatus, + &i.NextRunAt, + &i.CreatedByUserID, + &i.CreatedAt, + &i.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const markScheduledJobRan = `-- name: MarkScheduledJobRan :one +UPDATE core.scheduled_jobs +SET last_run_at = $2, + last_status = $3, + next_run_at = $4, + updated_at = NOW() +WHERE id = $1 +RETURNING id, project_id, name, schedule, url, secret, payload, enabled, last_run_at, last_status, next_run_at, created_by_user_id, created_at, updated_at +` + +type MarkScheduledJobRanParams struct { + ID uuid.UUID `json:"id"` + LastRunAt pgtype.Timestamptz `json:"last_run_at"` + LastStatus string `json:"last_status"` + NextRunAt pgtype.Timestamptz `json:"next_run_at"` +} + +func (q *Queries) MarkScheduledJobRan(ctx context.Context, arg MarkScheduledJobRanParams) (CoreScheduledJob, error) { + row := q.db.QueryRow(ctx, markScheduledJobRan, + arg.ID, + arg.LastRunAt, + arg.LastStatus, + arg.NextRunAt, + ) + var i CoreScheduledJob + err := row.Scan( + &i.ID, + &i.ProjectID, + &i.Name, + &i.Schedule, + &i.Url, + &i.Secret, + &i.Payload, + &i.Enabled, + &i.LastRunAt, + &i.LastStatus, + &i.NextRunAt, + &i.CreatedByUserID, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} + +const pruneScheduledJobRuns = `-- name: PruneScheduledJobRuns :exec +DELETE FROM core.scheduled_job_runs r +WHERE r.job_id = $1 + AND r.id NOT IN ( + SELECT k.id FROM core.scheduled_job_runs k + WHERE k.job_id = $1 + ORDER BY k.created_at DESC + LIMIT $2 + ) +` + +type PruneScheduledJobRunsParams struct { + JobID uuid.UUID `json:"job_id"` + Limit int32 `json:"limit"` +} + +func (q *Queries) PruneScheduledJobRuns(ctx context.Context, arg PruneScheduledJobRunsParams) error { + _, err := q.db.Exec(ctx, pruneScheduledJobRuns, arg.JobID, arg.Limit) + return err +} + +const updateScheduledJob = `-- name: UpdateScheduledJob :one +UPDATE core.scheduled_jobs +SET name = $3, + schedule = $4, + url = $5, + enabled = $6, + next_run_at = $7, + secret = CASE WHEN $8::boolean THEN $9 ELSE secret END, + payload = CASE WHEN $10::boolean THEN $11 ELSE payload END, + updated_at = NOW() +WHERE id = $1 + AND project_id = $2 +RETURNING id, project_id, name, schedule, url, secret, payload, enabled, last_run_at, last_status, next_run_at, created_by_user_id, created_at, updated_at +` + +type UpdateScheduledJobParams struct { + ID uuid.UUID `json:"id"` + ProjectID uuid.UUID `json:"project_id"` + Name string `json:"name"` + Schedule string `json:"schedule"` + Url string `json:"url"` + Enabled bool `json:"enabled"` + NextRunAt pgtype.Timestamptz `json:"next_run_at"` + Column8 bool `json:"column_8"` + Secret []byte `json:"secret"` + Column10 bool `json:"column_10"` + Payload []byte `json:"payload"` +} + +func (q *Queries) UpdateScheduledJob(ctx context.Context, arg UpdateScheduledJobParams) (CoreScheduledJob, error) { + row := q.db.QueryRow(ctx, updateScheduledJob, + arg.ID, + arg.ProjectID, + arg.Name, + arg.Schedule, + arg.Url, + arg.Enabled, + arg.NextRunAt, + arg.Column8, + arg.Secret, + arg.Column10, + arg.Payload, + ) + var i CoreScheduledJob + err := row.Scan( + &i.ID, + &i.ProjectID, + &i.Name, + &i.Schedule, + &i.Url, + &i.Secret, + &i.Payload, + &i.Enabled, + &i.LastRunAt, + &i.LastStatus, + &i.NextRunAt, + &i.CreatedByUserID, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} diff --git a/apps/backend/internal/database/db/models.go b/apps/backend/internal/database/db/models.go index ea6c0cb..0f93039 100644 --- a/apps/backend/internal/database/db/models.go +++ b/apps/backend/internal/database/db/models.go @@ -307,6 +307,35 @@ type CoreProjectMember struct { CreatedAt pgtype.Timestamptz `json:"created_at"` } +type CoreScheduledJob struct { + ID uuid.UUID `json:"id"` + ProjectID uuid.UUID `json:"project_id"` + Name string `json:"name"` + Schedule string `json:"schedule"` + Url string `json:"url"` + Secret []byte `json:"secret"` + Payload []byte `json:"payload"` + Enabled bool `json:"enabled"` + LastRunAt pgtype.Timestamptz `json:"last_run_at"` + LastStatus string `json:"last_status"` + NextRunAt pgtype.Timestamptz `json:"next_run_at"` + CreatedByUserID pgtype.UUID `json:"created_by_user_id"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type CoreScheduledJobRun struct { + ID uuid.UUID `json:"id"` + JobID uuid.UUID `json:"job_id"` + Status string `json:"status"` + TriggeredBy string `json:"triggered_by"` + StatusCode *int32 `json:"status_code"` + Error string `json:"error"` + DurationMs *int32 `json:"duration_ms"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + FinishedAt pgtype.Timestamptz `json:"finished_at"` +} + type CoreUser struct { ID uuid.UUID `json:"id"` AuthSubject string `json:"auth_subject"` diff --git a/apps/backend/internal/database/db/querier.go b/apps/backend/internal/database/db/querier.go index 843b5e9..5fe0a8a 100644 --- a/apps/backend/internal/database/db/querier.go +++ b/apps/backend/internal/database/db/querier.go @@ -8,6 +8,7 @@ import ( "context" "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" ) type Querier interface { @@ -32,6 +33,7 @@ type Querier interface { CreateInvitation(ctx context.Context, arg CreateInvitationParams) (CoreProjectInvitation, error) CreateOrganization(ctx context.Context, arg CreateOrganizationParams) (CoreOrganization, error) CreateProject(ctx context.Context, arg CreateProjectParams) (CoreProject, error) + CreateScheduledJob(ctx context.Context, arg CreateScheduledJobParams) (CoreScheduledJob, error) CreateWebhook(ctx context.Context, arg CreateWebhookParams) (CoreWebhook, error) DeleteBucketByID(ctx context.Context, id uuid.UUID) (CoreBucket, error) DeleteBucketObjectByKey(ctx context.Context, arg DeleteBucketObjectByKeyParams) (CoreBucketObject, error) @@ -43,9 +45,11 @@ type Querier interface { DeleteOrganizationByID(ctx context.Context, id uuid.UUID) (CoreOrganization, error) DeletePendingInvitationByIDForOrganization(ctx context.Context, arg DeletePendingInvitationByIDForOrganizationParams) (CoreProjectInvitation, error) DeleteProjectByID(ctx context.Context, id uuid.UUID) (CoreProject, error) + DeleteScheduledJob(ctx context.Context, arg DeleteScheduledJobParams) (uuid.UUID, error) DeleteWebhook(ctx context.Context, arg DeleteWebhookParams) (CoreWebhook, error) EventTotals(ctx context.Context, arg EventTotalsParams) (EventTotalsRow, error) EventTypeSeries(ctx context.Context, arg EventTypeSeriesParams) ([]EventTypeSeriesRow, error) + FinishScheduledJobRun(ctx context.Context, arg FinishScheduledJobRunParams) (CoreScheduledJobRun, error) GetAPIKeyByIDForProject(ctx context.Context, arg GetAPIKeyByIDForProjectParams) (CoreApiKey, error) GetAPIKeyByPrefix(ctx context.Context, prefix string) (GetAPIKeyByPrefixRow, error) GetBucketByID(ctx context.Context, id uuid.UUID) (GetBucketByIDRow, error) @@ -63,12 +67,14 @@ type Querier interface { GetProjectByID(ctx context.Context, id uuid.UUID) (CoreProject, error) GetProjectMembership(ctx context.Context, arg GetProjectMembershipParams) (GetProjectMembershipRow, error) GetProjectOverview(ctx context.Context, id uuid.UUID) (GetProjectOverviewRow, error) + GetScheduledJobByID(ctx context.Context, id uuid.UUID) (CoreScheduledJob, error) GetUserByAuthSubject(ctx context.Context, authSubject string) (CoreUser, error) GetUserByID(ctx context.Context, id uuid.UUID) (CoreUser, error) GetWebhookByID(ctx context.Context, id uuid.UUID) (CoreWebhook, error) GetWebhookDeliveryByID(ctx context.Context, id uuid.UUID) (CoreWebhookDelivery, error) HasFingerprintSeen(ctx context.Context, arg HasFingerprintSeenParams) (bool, error) InsertEvent(ctx context.Context, arg InsertEventParams) (CoreEvent, error) + InsertScheduledJobRun(ctx context.Context, arg InsertScheduledJobRunParams) (CoreScheduledJobRun, error) InsertWebhookDelivery(ctx context.Context, arg InsertWebhookDeliveryParams) (CoreWebhookDelivery, error) ListAPIKeysForProject(ctx context.Context, projectID uuid.UUID) ([]CoreApiKey, error) ListAuditLogsForProject(ctx context.Context, arg ListAuditLogsForProjectParams) ([]CoreAuditLog, error) @@ -79,6 +85,7 @@ type Querier interface { ListComponents(ctx context.Context, projectID uuid.UUID) ([]CoreComponent, error) ListDBConnections(ctx context.Context, projectID uuid.UUID) ([]CoreDbConnection, error) ListDocuments(ctx context.Context, arg ListDocumentsParams) ([]CoreDocument, error) + ListDueScheduledJobs(ctx context.Context, nextRunAt pgtype.Timestamptz) ([]CoreScheduledJob, error) ListErrorGroups(ctx context.Context, arg ListErrorGroupsParams) ([]ListErrorGroupsRow, error) ListEvents(ctx context.Context, arg ListEventsParams) ([]ListEventsRow, error) ListIntegrations(ctx context.Context, projectID uuid.UUID) ([]CoreIntegration, error) @@ -88,14 +95,18 @@ type Querier interface { ListPendingWebhookDeliveries(ctx context.Context, limit int32) ([]CoreWebhookDelivery, error) ListProjectMembers(ctx context.Context, projectID uuid.UUID) ([]ListProjectMembersRow, error) ListProjectsForOrganization(ctx context.Context, arg ListProjectsForOrganizationParams) ([]ListProjectsForOrganizationRow, error) + ListScheduledJobRuns(ctx context.Context, arg ListScheduledJobRunsParams) ([]CoreScheduledJobRun, error) + ListScheduledJobs(ctx context.Context, projectID uuid.UUID) ([]CoreScheduledJob, error) ListWebhookDeliveries(ctx context.Context, arg ListWebhookDeliveriesParams) ([]CoreWebhookDelivery, error) ListWebhooks(ctx context.Context, projectID uuid.UUID) ([]CoreWebhook, error) ListWebhooksForEvent(ctx context.Context, arg ListWebhooksForEventParams) ([]CoreWebhook, error) MarkInvitationAccepted(ctx context.Context, id uuid.UUID) (CoreProjectInvitation, error) + MarkScheduledJobRan(ctx context.Context, arg MarkScheduledJobRanParams) (CoreScheduledJob, error) MarkWebhookDelivery(ctx context.Context, arg MarkWebhookDeliveryParams) error MetricNames(ctx context.Context, projectID uuid.UUID) ([]string, error) MetricSeries(ctx context.Context, arg MetricSeriesParams) ([]MetricSeriesRow, error) MoveBucketObject(ctx context.Context, arg MoveBucketObjectParams) (CoreBucketObject, error) + PruneScheduledJobRuns(ctx context.Context, arg PruneScheduledJobRunsParams) error RemoveOrganizationMember(ctx context.Context, arg RemoveOrganizationMemberParams) (CoreOrganizationMember, error) RemoveProjectMember(ctx context.Context, arg RemoveProjectMemberParams) (CoreProjectMember, error) RemoveProjectMembershipsForOrganizationUser(ctx context.Context, arg RemoveProjectMembershipsForOrganizationUserParams) error @@ -118,6 +129,7 @@ type Querier interface { UpdateOrganizationMemberRole(ctx context.Context, arg UpdateOrganizationMemberRoleParams) (CoreOrganizationMember, error) UpdateProjectByID(ctx context.Context, arg UpdateProjectByIDParams) (CoreProject, error) UpdateProjectMemberRole(ctx context.Context, arg UpdateProjectMemberRoleParams) (CoreProjectMember, error) + UpdateScheduledJob(ctx context.Context, arg UpdateScheduledJobParams) (CoreScheduledJob, error) UpdateWebhook(ctx context.Context, arg UpdateWebhookParams) (CoreWebhook, error) UpsertComponent(ctx context.Context, arg UpsertComponentParams) (CoreComponent, error) UpsertUser(ctx context.Context, arg UpsertUserParams) (CoreUser, error) diff --git a/apps/backend/internal/dbx/dbx.go b/apps/backend/internal/dbx/dbx.go index 5dd67f3..160bdef 100644 --- a/apps/backend/internal/dbx/dbx.go +++ b/apps/backend/internal/dbx/dbx.go @@ -86,7 +86,7 @@ func (d *Client) reset() { d.session = nil } if d.cmd != nil && d.cmd.Process != nil { - d.cmd.Process.Kill() + _ = d.cmd.Process.Kill() } d.cmd = nil d.registered = map[string]bool{} diff --git a/apps/backend/internal/handlers/http.go b/apps/backend/internal/handlers/http.go index cb692e4..d2050c4 100644 --- a/apps/backend/internal/handlers/http.go +++ b/apps/backend/internal/handlers/http.go @@ -105,6 +105,13 @@ func (h *HTTPHandler) Register(router *gin.Engine) { api.GET("/projects/:projectID/webhooks/:webhookID/deliveries", h.listWebhookDeliveries) api.POST("/projects/:projectID/webhooks/:webhookID/test", h.testWebhook) api.POST("/projects/:projectID/deploy-markers", h.createDeployMarker) + api.GET("/projects/:projectID/jobs", h.listJobs) + api.POST("/projects/:projectID/jobs", h.createJob) + api.PATCH("/projects/:projectID/jobs/:jobID", h.updateJob) + api.DELETE("/projects/:projectID/jobs/:jobID", h.deleteJob) + api.GET("/projects/:projectID/jobs/:jobID/runs", h.listJobRuns) + api.POST("/projects/:projectID/jobs/:jobID/run", h.runJob) + api.GET("/projects/:projectID/realtime/stream", h.realtimeStream) 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/handlers/jobs.go b/apps/backend/internal/handlers/jobs.go new file mode 100644 index 0000000..158e5b1 --- /dev/null +++ b/apps/backend/internal/handlers/jobs.go @@ -0,0 +1,180 @@ +package handlers + +import ( + "encoding/json" + "errors" + "io" + "net/http" + "strings" + "time" + + "github.com/gin-gonic/gin" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + + "github.com/tdvorak/primora/apps/backend/internal/middleware" + "github.com/tdvorak/primora/apps/backend/internal/models" + apperrors "github.com/tdvorak/primora/apps/backend/internal/response" + "github.com/tdvorak/primora/apps/backend/internal/services" +) + +// actorProjectJob resolves the actor, :projectID and :jobID. +func (h *HTTPHandler) actorProjectJob(c *gin.Context) (*models.Actor, uuid.UUID, uuid.UUID, bool) { + actor, projectID, ok := h.actorAndProject(c) + if !ok { + return nil, uuid.Nil, uuid.Nil, false + } + jobID, ok := parseUUIDParam(c, "jobID") + if !ok { + return nil, uuid.Nil, uuid.Nil, false + } + return actor, projectID, jobID, true +} + +func (h *HTTPHandler) jobError(c *gin.Context, err error) { + if errors.Is(err, pgx.ErrNoRows) { + apperrors.Abort(c, http.StatusNotFound, "not_found", "job not found") + return + } + msg := err.Error() + if strings.Contains(msg, "already exists") { + apperrors.Abort(c, http.StatusConflict, "name_conflict", msg) + return + } + if strings.Contains(msg, "invalid schedule") || strings.Contains(msg, "schedule is required") || strings.Contains(msg, "https") { + apperrors.Abort(c, http.StatusBadRequest, "bad_request", msg) + return + } + h.handleError(c, err) +} + +func (h *HTTPHandler) listJobs(c *gin.Context) { + actor, projectID, ok := h.actorAndProject(c) + if !ok { + return + } + result, err := h.Platform.ListScheduledJobs(c.Request.Context(), actor, projectID) + if err != nil { + h.jobError(c, err) + return + } + c.JSON(http.StatusOK, gin.H{"items": result}) +} + +func (h *HTTPHandler) createJob(c *gin.Context) { + actor, projectID, ok := h.actorAndProject(c) + if !ok { + return + } + var body services.CreateScheduledJobInput + if !h.bindAndValidate(c, &body) { + return + } + result, err := h.Platform.CreateScheduledJob(c.Request.Context(), actor, projectID, body, middleware.RequestIDFromContext(c)) + if err != nil { + h.jobError(c, err) + return + } + c.JSON(http.StatusCreated, result) +} + +func (h *HTTPHandler) updateJob(c *gin.Context) { + actor, projectID, jobID, ok := h.actorProjectJob(c) + if !ok { + return + } + var body services.UpdateScheduledJobInput + if !h.bindAndValidate(c, &body) { + return + } + result, err := h.Platform.UpdateScheduledJob(c.Request.Context(), actor, projectID, jobID, body, middleware.RequestIDFromContext(c)) + if err != nil { + h.jobError(c, err) + return + } + c.JSON(http.StatusOK, result) +} + +func (h *HTTPHandler) deleteJob(c *gin.Context) { + actor, projectID, jobID, ok := h.actorProjectJob(c) + if !ok { + return + } + if err := h.Platform.DeleteScheduledJob(c.Request.Context(), actor, projectID, jobID, middleware.RequestIDFromContext(c)); err != nil { + h.jobError(c, err) + return + } + c.Status(http.StatusNoContent) +} + +func (h *HTTPHandler) listJobRuns(c *gin.Context) { + actor, projectID, jobID, ok := h.actorProjectJob(c) + if !ok { + return + } + limit, ok := parsePaginationQuery(c, "limit", 50, 1, 200) + if !ok { + return + } + result, err := h.Platform.ListScheduledJobRuns(c.Request.Context(), actor, projectID, jobID, limit) + if err != nil { + h.jobError(c, err) + return + } + c.JSON(http.StatusOK, gin.H{"items": result}) +} + +func (h *HTTPHandler) runJob(c *gin.Context) { + actor, projectID, jobID, ok := h.actorProjectJob(c) + if !ok { + return + } + result, err := h.Platform.RunScheduledJob(c.Request.Context(), actor, projectID, jobID, middleware.RequestIDFromContext(c)) + if err != nil { + h.jobError(c, err) + return + } + c.JSON(http.StatusAccepted, result) +} + +// realtimeStream pushes project domain events (document/object/issue/job) to +// the dashboard over SSE — the same pattern as telemetryStream. +func (h *HTTPHandler) realtimeStream(c *gin.Context) { + actor, projectID, ok := h.actorAndProject(c) + if !ok { + return + } + ch, err := h.Platform.SubscribeRealtime(c.Request.Context(), actor, projectID) + if err != nil { + h.handleError(c, err) + return + } + defer h.Platform.UnsubscribeRealtime(projectID, ch) + + c.Writer.Header().Set("Content-Type", "text/event-stream") + c.Writer.Header().Set("Cache-Control", "no-cache") + c.Writer.Header().Set("X-Accel-Buffering", "no") + c.Status(http.StatusOK) + if _, err := c.Writer.WriteString(": connected\n\n"); err == nil { + c.Writer.Flush() + } + + keepalive := time.NewTicker(25 * time.Second) + defer keepalive.Stop() + + c.Stream(func(w io.Writer) bool { + select { + case msg, ok := <-ch: + if !ok { + return false + } + c.SSEvent("event", json.RawMessage(msg)) + return true + case <-keepalive.C: + c.SSEvent("ping", "keepalive") + return true + case <-c.Request.Context().Done(): + return false + } + }) +} diff --git a/apps/backend/internal/middleware/compression.go b/apps/backend/internal/middleware/compression.go index 3a91d73..30b3473 100644 --- a/apps/backend/internal/middleware/compression.go +++ b/apps/backend/internal/middleware/compression.go @@ -35,8 +35,11 @@ func Compression() gin.HandlerFunc { return } - // Skip compression for small responses or streaming - if c.Request.Method == "HEAD" || c.Request.URL.Path == "/api/v1/health/liveness" { + // Skip compression for small responses or streaming — gzip buffers + // writes, which breaks SSE flush semantics. + if c.Request.Method == "HEAD" || + c.Request.URL.Path == "/api/v1/health/liveness" || + strings.HasSuffix(c.Request.URL.Path, "/stream") { c.Next() return } diff --git a/apps/backend/internal/middleware/context.go b/apps/backend/internal/middleware/context.go index c9a6589..0a5111a 100644 --- a/apps/backend/internal/middleware/context.go +++ b/apps/backend/internal/middleware/context.go @@ -78,9 +78,11 @@ func (m AuthMiddleware) ResolveActor() gin.HandlerFunc { apiKey = strings.TrimSpace(c.GetHeader("X-Primora-Key")) } authz := strings.TrimSpace(c.GetHeader("Authorization")) - // EventSource cannot set headers — the SSE stream accepts credentials + // EventSource cannot set headers — the SSE streams accept credentials // as query params instead. - if apiKey == "" && authz == "" && strings.HasSuffix(c.Request.URL.Path, "/telemetry/stream") { + sseStream := strings.HasSuffix(c.Request.URL.Path, "/telemetry/stream") || + strings.HasSuffix(c.Request.URL.Path, "/realtime/stream") + if apiKey == "" && authz == "" && sseStream { apiKey = strings.TrimSpace(c.Query("api_key")) if apiKey == "" { if t := strings.TrimSpace(c.Query("token")); t != "" { @@ -94,7 +96,6 @@ func (m AuthMiddleware) ResolveActor() gin.HandlerFunc { token = strings.TrimSpace(strings.TrimPrefix(authz, "Bearer ")) } if strings.HasPrefix(strings.ToLower(authz), "bearer ") { - apiKey = "" actor, err := m.resolveJWTActor(c.Request.Context(), token) if err != nil { apperrors.Abort(c, http.StatusUnauthorized, "invalid_token", err.Error()) diff --git a/apps/backend/internal/services/jobs.go b/apps/backend/internal/services/jobs.go new file mode 100644 index 0000000..3c84ac3 --- /dev/null +++ b/apps/backend/internal/services/jobs.go @@ -0,0 +1,591 @@ +package services + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + "strings" + "sync" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" + "github.com/robfig/cron/v3" + + db "github.com/tdvorak/primora/apps/backend/internal/database/db" + "github.com/tdvorak/primora/apps/backend/internal/models" + "github.com/tdvorak/primora/apps/backend/internal/secrets" +) + +// WebhookEventJobRun fires on every scheduled-job execution, successful or not. +const WebhookEventJobRun = "job.run" + +// maxJobRunsPerJob bounds run history — older rows are pruned on each finish. +const maxJobRunsPerJob = 200 + +type ScheduledJobSummary struct { + ID uuid.UUID `json:"id"` + ProjectID uuid.UUID `json:"project_id"` + Name string `json:"name"` + Schedule string `json:"schedule"` + URL string `json:"url"` + Payload json.RawMessage `json:"payload"` + Enabled bool `json:"enabled"` + HasSecret bool `json:"has_secret"` + LastRunAt *time.Time `json:"last_run_at"` + LastStatus *string `json:"last_status"` + NextRunAt *time.Time `json:"next_run_at"` + CreatedAt time.Time `json:"created_at"` + // Secret is populated only when Primora generated it — one chance to see it. + Secret string `json:"secret,omitempty"` +} + +type ScheduledJobRunSummary struct { + ID uuid.UUID `json:"id"` + JobID uuid.UUID `json:"job_id"` + Status string `json:"status"` + TriggeredBy string `json:"triggered_by"` + StatusCode *int32 `json:"status_code"` + Error string `json:"error"` + DurationMs *int32 `json:"duration_ms"` + StartedAt time.Time `json:"started_at"` + FinishedAt *time.Time `json:"finished_at"` +} + +type CreateScheduledJobInput struct { + Name string `json:"name" validate:"required,min=2"` + Schedule string `json:"schedule" validate:"required"` + URL string `json:"url" validate:"required,url"` + Secret string `json:"secret"` + Payload map[string]any `json:"payload"` + Enabled *bool `json:"enabled"` +} + +type UpdateScheduledJobInput struct { + Name *string `json:"name" validate:"omitempty,min=2"` + Schedule *string `json:"schedule"` + URL *string `json:"url" validate:"omitempty,url"` + Secret *string `json:"secret"` + Payload map[string]any `json:"payload"` + Enabled *bool `json:"enabled"` +} + +func toJobSummary(row db.CoreScheduledJob) ScheduledJobSummary { + out := ScheduledJobSummary{ + ID: row.ID, + ProjectID: row.ProjectID, + Name: row.Name, + Schedule: row.Schedule, + URL: row.Url, + Payload: row.Payload, + Enabled: row.Enabled, + HasSecret: len(row.Secret) > 0, + CreatedAt: row.CreatedAt.Time, + } + if row.LastRunAt.Valid { + t := row.LastRunAt.Time + out.LastRunAt = &t + } + if row.LastStatus != "" { + s := row.LastStatus + out.LastStatus = &s + } + if row.NextRunAt.Valid { + t := row.NextRunAt.Time + out.NextRunAt = &t + } + return out +} + +func toJobRunSummary(row db.CoreScheduledJobRun) ScheduledJobRunSummary { + out := ScheduledJobRunSummary{ + ID: row.ID, + JobID: row.JobID, + Status: row.Status, + TriggeredBy: row.TriggeredBy, + StatusCode: row.StatusCode, + Error: row.Error, + DurationMs: row.DurationMs, + StartedAt: row.CreatedAt.Time, + } + if row.FinishedAt.Valid { + t := row.FinishedAt.Time + out.FinishedAt = &t + } + return out +} + +// parseSchedule accepts standard 5-field cron expressions plus robfig +// descriptors (@hourly, @every 30m, …). +func parseSchedule(expr string) (cron.Schedule, error) { + expr = strings.TrimSpace(expr) + if expr == "" { + return nil, errors.New("schedule is required") + } + sched, err := cron.ParseStandard(expr) + if err != nil { + return nil, fmt.Errorf("invalid schedule %q — expected a cron expression like \"*/15 * * * *\" or \"@every 1h\"", expr) + } + return sched, nil +} + +func nextRunAt(sched cron.Schedule, from time.Time) pgtype.Timestamptz { + return pgtype.Timestamptz{Time: sched.Next(from), Valid: true} +} + +func (s *PlatformService) jobFor(ctx context.Context, actor *models.Actor, projectID, jobID uuid.UUID, write bool) (db.CoreScheduledJob, error) { + roles := []string{"admin", "developer", "viewer"} + if write { + roles = []string{"admin", "developer"} + } + if err := s.requireProjectRole(ctx, actor, projectID, roles...); err != nil { + return db.CoreScheduledJob{}, err + } + row, err := s.repo.Queries().GetScheduledJobByID(ctx, jobID) + if err != nil { + return db.CoreScheduledJob{}, err + } + if row.ProjectID != projectID { + return db.CoreScheduledJob{}, fmt.Errorf("job access denied") + } + return row, nil +} + +func (s *PlatformService) ListScheduledJobs(ctx context.Context, actor *models.Actor, projectID uuid.UUID) ([]ScheduledJobSummary, error) { + if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer", "viewer"); err != nil { + return nil, err + } + rows, err := s.repo.Queries().ListScheduledJobs(ctx, projectID) + if err != nil { + return nil, err + } + out := make([]ScheduledJobSummary, 0, len(rows)) + for _, row := range rows { + out = append(out, toJobSummary(row)) + } + return out, nil +} + +func (s *PlatformService) CreateScheduledJob(ctx context.Context, actor *models.Actor, projectID uuid.UUID, input CreateScheduledJobInput, requestID string) (ScheduledJobSummary, error) { + if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer"); err != nil { + return ScheduledJobSummary{}, err + } + jobURL, err := normalizeWebhookURL(input.URL) + if err != nil { + return ScheduledJobSummary{}, err + } + sched, err := parseSchedule(input.Schedule) + if err != nil { + return ScheduledJobSummary{}, err + } + secret := strings.TrimSpace(input.Secret) + generated := false + if secret == "" { + if secret, err = randomToken(16); err != nil { + return ScheduledJobSummary{}, err + } + generated = true + } + secretBytes, err := s.enc.Encrypt([]byte(secret)) + if err != nil { + return ScheduledJobSummary{}, err + } + payload := []byte("{}") + if input.Payload != nil { + payload, _ = json.Marshal(input.Payload) + } + enabled := true + if input.Enabled != nil { + enabled = *input.Enabled + } + var createdBy pgtype.UUID + if actor.UserID != nil { + createdBy = pgtype.UUID{Bytes: *actor.UserID, Valid: true} + } + var next pgtype.Timestamptz + if enabled { + next = nextRunAt(sched, time.Now()) + } + row, err := s.repo.Queries().CreateScheduledJob(ctx, db.CreateScheduledJobParams{ + ProjectID: projectID, + Name: strings.TrimSpace(input.Name), + Schedule: strings.TrimSpace(input.Schedule), + Url: jobURL, + Secret: secretBytes, + Payload: payload, + Enabled: enabled, + NextRunAt: next, + CreatedByUserID: createdBy, + }) + if err != nil { + if strings.Contains(err.Error(), "scheduled_jobs_project_id_name_key") { + return ScheduledJobSummary{}, fmt.Errorf("a job named %q already exists", input.Name) + } + return ScheduledJobSummary{}, err + } + if project, err := s.repo.Queries().GetProjectByID(ctx, projectID); err == nil { + _, _ = s.repo.Queries().CreateAuditLog(ctx, newAuditParams(project.OrganizationID, projectID, actor, requestID, "job.created", "scheduled_job", row.ID.String(), map[string]any{ + "name": row.Name, "schedule": row.Schedule, "url": row.Url, + })) + } + summary := toJobSummary(row) + if generated { + summary.Secret = secret + } + return summary, nil +} + +func (s *PlatformService) UpdateScheduledJob(ctx context.Context, actor *models.Actor, projectID, jobID uuid.UUID, input UpdateScheduledJobInput, requestID string) (ScheduledJobSummary, error) { + row, err := s.jobFor(ctx, actor, projectID, jobID, true) + if err != nil { + return ScheduledJobSummary{}, err + } + name := row.Name + if input.Name != nil { + name = strings.TrimSpace(*input.Name) + } + schedule := row.Schedule + if input.Schedule != nil { + schedule = strings.TrimSpace(*input.Schedule) + } + jobURL := row.Url + if input.URL != nil { + if jobURL, err = normalizeWebhookURL(*input.URL); err != nil { + return ScheduledJobSummary{}, err + } + } + enabled := row.Enabled + if input.Enabled != nil { + enabled = *input.Enabled + } + sched, err := parseSchedule(schedule) + if err != nil { + return ScheduledJobSummary{}, err + } + var next pgtype.Timestamptz + if enabled { + next = nextRunAt(sched, time.Now()) + } + rotateSecret := input.Secret != nil + var secretBytes []byte + if rotateSecret { + secret := strings.TrimSpace(*input.Secret) + if secret == "" { + if secret, err = randomToken(16); err != nil { + return ScheduledJobSummary{}, err + } + } + if secretBytes, err = s.enc.Encrypt([]byte(secret)); err != nil { + return ScheduledJobSummary{}, err + } + } + rotatePayload := input.Payload != nil + var payload []byte + if rotatePayload { + payload, _ = json.Marshal(input.Payload) + } + updated, err := s.repo.Queries().UpdateScheduledJob(ctx, db.UpdateScheduledJobParams{ + ID: jobID, + ProjectID: projectID, + Name: name, + Schedule: schedule, + Url: jobURL, + Enabled: enabled, + NextRunAt: next, + Column8: rotateSecret, + Secret: secretBytes, + Column10: rotatePayload, + Payload: payload, + }) + if err != nil { + if strings.Contains(err.Error(), "scheduled_jobs_project_id_name_key") { + return ScheduledJobSummary{}, fmt.Errorf("a job named %q already exists", name) + } + return ScheduledJobSummary{}, err + } + project, _ := s.repo.Queries().GetProjectByID(ctx, projectID) + _, _ = s.repo.Queries().CreateAuditLog(ctx, newAuditParams(project.OrganizationID, projectID, actor, requestID, "job.updated", "scheduled_job", jobID.String(), map[string]any{ + "name": updated.Name, "schedule": updated.Schedule, "url": updated.Url, "enabled": updated.Enabled, + })) + summary := toJobSummary(updated) + if rotateSecret && strings.TrimSpace(*input.Secret) == "" { + if plain, decErr := s.enc.Decrypt(updated.Secret); decErr == nil { + summary.Secret = string(plain) + } + } + return summary, nil +} + +func (s *PlatformService) DeleteScheduledJob(ctx context.Context, actor *models.Actor, projectID, jobID uuid.UUID, requestID string) error { + row, err := s.jobFor(ctx, actor, projectID, jobID, true) + if err != nil { + return err + } + if _, err := s.repo.Queries().DeleteScheduledJob(ctx, db.DeleteScheduledJobParams{ + ID: jobID, + ProjectID: projectID, + }); err != nil { + return err + } + project, _ := s.repo.Queries().GetProjectByID(ctx, projectID) + _, _ = s.repo.Queries().CreateAuditLog(ctx, newAuditParams(project.OrganizationID, projectID, actor, requestID, "job.deleted", "scheduled_job", jobID.String(), map[string]any{ + "name": row.Name, + })) + return nil +} + +func (s *PlatformService) ListScheduledJobRuns(ctx context.Context, actor *models.Actor, projectID, jobID uuid.UUID, limit int32) ([]ScheduledJobRunSummary, error) { + if _, err := s.jobFor(ctx, actor, projectID, jobID, false); err != nil { + return nil, err + } + if limit <= 0 || limit > 200 { + limit = 50 + } + rows, err := s.repo.Queries().ListScheduledJobRuns(ctx, db.ListScheduledJobRunsParams{ + JobID: jobID, + Limit: limit, + }) + if err != nil { + return nil, err + } + out := make([]ScheduledJobRunSummary, 0, len(rows)) + for _, row := range rows { + out = append(out, toJobRunSummary(row)) + } + return out, nil +} + +// RunScheduledJob queues an out-of-band manual run. The run row is returned +// in "running" state; the result lands asynchronously. +func (s *PlatformService) RunScheduledJob(ctx context.Context, actor *models.Actor, projectID, jobID uuid.UUID, requestID string) (ScheduledJobRunSummary, error) { + job, err := s.jobFor(ctx, actor, projectID, jobID, true) + if err != nil { + return ScheduledJobRunSummary{}, err + } + if s.scheduler == nil { + return ScheduledJobRunSummary{}, errors.New("scheduler unavailable") + } + run, err := s.scheduler.enqueue(ctx, job, "manual") + if err != nil { + return ScheduledJobRunSummary{}, err + } + project, _ := s.repo.Queries().GetProjectByID(ctx, projectID) + _, _ = s.repo.Queries().CreateAuditLog(ctx, newAuditParams(project.OrganizationID, projectID, actor, requestID, "job.run", "scheduled_job", jobID.String(), map[string]any{ + "name": job.Name, "run_id": run.ID.String(), + })) + return toJobRunSummary(run), nil +} + +// --------------------------------------------------------------------------- +// JobScheduler — in-process ticker, signed POST per due job, run history. +// --------------------------------------------------------------------------- + +// jobQueries is the slice of db.Queries the scheduler needs — narrow so tests +// can stub it. +type jobQueries interface { + ListDueScheduledJobs(ctx context.Context, now pgtype.Timestamptz) ([]db.CoreScheduledJob, error) + InsertScheduledJobRun(ctx context.Context, params db.InsertScheduledJobRunParams) (db.CoreScheduledJobRun, error) + FinishScheduledJobRun(ctx context.Context, params db.FinishScheduledJobRunParams) (db.CoreScheduledJobRun, error) + MarkScheduledJobRan(ctx context.Context, params db.MarkScheduledJobRanParams) (db.CoreScheduledJob, error) + PruneScheduledJobRuns(ctx context.Context, params db.PruneScheduledJobRunsParams) error +} + +type JobScheduler struct { + repo jobQueries + enc *secrets.Encryptor + logger *slog.Logger + http *http.Client + interval time.Duration + queue chan jobExec + quit chan struct{} + wg sync.WaitGroup + started sync.Once + // onEvent fans a finished run out to webhooks and the realtime stream. + onEvent func(ctx context.Context, projectID uuid.UUID, eventType string, data map[string]any) +} + +type jobExec struct { + job db.CoreScheduledJob + run db.CoreScheduledJobRun +} + +func NewJobScheduler(q jobQueries, enc *secrets.Encryptor, logger *slog.Logger, onEvent func(ctx context.Context, projectID uuid.UUID, eventType string, data map[string]any)) *JobScheduler { + if logger == nil { + logger = slog.Default() + } + return &JobScheduler{ + repo: q, + enc: enc, + logger: logger, + http: &http.Client{Timeout: 10 * time.Second}, + interval: 15 * time.Second, + queue: make(chan jobExec, 256), + quit: make(chan struct{}), + onEvent: onEvent, + } +} + +func (j *JobScheduler) Start(ctx context.Context) { + j.started.Do(func() { + j.wg.Add(1) + go j.run(ctx) + }) +} + +func (j *JobScheduler) Stop() { + close(j.quit) + j.wg.Wait() +} + +func (j *JobScheduler) run(ctx context.Context) { + defer j.wg.Done() + ticker := time.NewTicker(j.interval) + defer ticker.Stop() + for { + select { + case exec := <-j.queue: + j.execute(exec.job, exec.run) + case <-ticker.C: + j.scan() + case <-j.quit: + return + case <-ctx.Done(): + return + } + } +} + +// scan picks up enabled jobs whose next_run_at has passed. Missed occurrences +// collapse into a single run — Primora fires once, not once per missed tick. +func (j *JobScheduler) scan() { + ctx := context.Background() + jobs, err := j.repo.ListDueScheduledJobs(ctx, pgtype.Timestamptz{Time: time.Now(), Valid: true}) + if err != nil { + j.logger.Warn("scheduled job scan failed", "error", err) + return + } + for _, job := range jobs { + run, err := j.repo.InsertScheduledJobRun(ctx, db.InsertScheduledJobRunParams{JobID: job.ID, TriggeredBy: "schedule"}) + if err != nil { + j.logger.Warn("job run insert failed", "job_id", job.ID, "error", err) + j.markRan(ctx, job, "failed") + continue + } + j.execute(job, run) + } +} + +// enqueue inserts the run row synchronously so callers get a run ID, then +// hands the execution to the worker. +func (j *JobScheduler) enqueue(ctx context.Context, job db.CoreScheduledJob, triggeredBy string) (db.CoreScheduledJobRun, error) { + run, err := j.repo.InsertScheduledJobRun(ctx, db.InsertScheduledJobRunParams{ + JobID: job.ID, + TriggeredBy: triggeredBy, + }) + if err != nil { + return db.CoreScheduledJobRun{}, err + } + select { + case j.queue <- jobExec{job: job, run: run}: + case <-j.quit: + default: + j.logger.Warn("job queue full, run dropped", "job_id", job.ID, "run_id", run.ID) + } + return run, nil +} + +func (j *JobScheduler) execute(job db.CoreScheduledJob, run db.CoreScheduledJobRun) { + ctx := context.Background() + start := time.Now() + statusCode, sendErr := j.deliver(ctx, job, run) + status := "success" + errText := "" + if sendErr != nil { + status = "failed" + errText = sendErr.Error() + } + duration := int32(time.Since(start).Milliseconds()) + _, _ = j.repo.FinishScheduledJobRun(ctx, db.FinishScheduledJobRunParams{ + ID: run.ID, + Status: status, + StatusCode: statusCode, + Error: errText, + DurationMs: &duration, + }) + j.markRan(ctx, job, status) + _ = j.repo.PruneScheduledJobRuns(ctx, db.PruneScheduledJobRunsParams{JobID: job.ID, Limit: maxJobRunsPerJob}) + if j.onEvent != nil { + j.onEvent(ctx, job.ProjectID, WebhookEventJobRun, map[string]any{ + "job_id": job.ID.String(), + "job_name": job.Name, + "run_id": run.ID.String(), + "triggered_by": run.TriggeredBy, + "status": status, + "status_code": statusCode, + "error": errText, + "duration_ms": duration, + }) + } +} + +func (j *JobScheduler) markRan(ctx context.Context, job db.CoreScheduledJob, status string) { + var next pgtype.Timestamptz + if job.Enabled { + if sched, err := parseSchedule(job.Schedule); err == nil { + next = nextRunAt(sched, time.Now()) + } + } + if _, err := j.repo.MarkScheduledJobRan(ctx, db.MarkScheduledJobRanParams{ + ID: job.ID, + LastRunAt: pgtype.Timestamptz{Time: time.Now(), Valid: true}, + LastStatus: status, + NextRunAt: next, + }); err != nil { + j.logger.Warn("job mark-ran failed", "job_id", job.ID, "error", err) + } +} + +// deliver POSTs the job payload to its URL, signed like a webhook delivery. +func (j *JobScheduler) deliver(ctx context.Context, job db.CoreScheduledJob, run db.CoreScheduledJobRun) (*int32, error) { + payload, _ := json.Marshal(map[string]any{ + "event": WebhookEventJobRun, + "project_id": job.ProjectID.String(), + "job_id": job.ID.String(), + "job_name": job.Name, + "run_id": run.ID.String(), + "triggered_by": run.TriggeredBy, + "occurred_at": time.Now().UTC().Format(time.RFC3339), + "data": json.RawMessage(job.Payload), + }) + secret := "" + if len(job.Secret) > 0 && j.enc != nil { + if plain, err := j.enc.Decrypt(job.Secret); err == nil { + secret = string(plain) + } + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, job.Url, bytes.NewReader(payload)) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Primora-Event", WebhookEventJobRun) + req.Header.Set("X-Primora-Job", job.ID.String()) + req.Header.Set("X-Primora-Run", run.ID.String()) + req.Header.Set("X-Primora-Signature", "sha256="+signBody(secret, payload)) + resp, err := j.http.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 1<<16)) + code := int32(resp.StatusCode) + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return &code, fmt.Errorf("endpoint returned %d", resp.StatusCode) + } + return &code, nil +} diff --git a/apps/backend/internal/services/platform_service.go b/apps/backend/internal/services/platform_service.go index fb9a6d0..b1273bd 100644 --- a/apps/backend/internal/services/platform_service.go +++ b/apps/backend/internal/services/platform_service.go @@ -34,8 +34,10 @@ type PlatformService struct { publicURL string dbx *dbx.Client hub *EventHub + realtime *EventHub enc *secrets.Encryptor dispatcher *WebhookDispatcher + scheduler *JobScheduler } type BootstrapInput struct { @@ -222,19 +224,50 @@ type InvitationSummary struct { } func NewPlatformService(repo *repositories.CoreRepository, store *storage.LocalStore, mailer *Mailer, publicURL string, dbxClient *dbx.Client, enc *secrets.Encryptor, logger *slog.Logger) *PlatformService { - s := &PlatformService{repo: repo, store: store, mailer: mailer, publicURL: publicURL, dbx: dbxClient, hub: NewEventHub(), enc: enc} + s := &PlatformService{repo: repo, store: store, mailer: mailer, publicURL: publicURL, dbx: dbxClient, hub: NewEventHub(), realtime: NewEventHub(), enc: enc} if enc != nil { s.dispatcher = NewWebhookDispatcher(repo.Queries(), enc, logger) s.dispatcher.Start(context.Background()) + s.scheduler = NewJobScheduler(repo.Queries(), enc, logger, s.publishEvent) + s.scheduler.Start(context.Background()) } return s } -// Close stops the webhook dispatcher. Idempotent. +// Close stops the webhook dispatcher and job scheduler. Idempotent. func (s *PlatformService) Close() { if s.dispatcher != nil { s.dispatcher.Stop() } + if s.scheduler != nil { + s.scheduler.Stop() + } +} + +// publishEvent fans a domain event out to realtime SSE subscribers and to +// matching webhooks. Failures are logged by the callees — events never block +// the request that produced them. +func (s *PlatformService) publishEvent(ctx context.Context, projectID uuid.UUID, eventType string, data map[string]any) { + if s.realtime != nil { + s.realtime.Broadcast(projectID.String(), map[string]any{ + "type": eventType, + "occurred_at": time.Now().UTC().Format(time.RFC3339), + "data": data, + }) + } + s.dispatchEvent(ctx, projectID, eventType, data) +} + +// SubscribeRealtime registers an SSE consumer for project domain events. +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 +} + +func (s *PlatformService) UnsubscribeRealtime(projectID uuid.UUID, ch chan []byte) { + s.realtime.Unsubscribe(projectID.String(), ch) } func (s *PlatformService) Me(ctx context.Context, actor *models.Actor) (PlatformSummary, error) { @@ -1242,6 +1275,12 @@ func (s *PlatformService) UploadObject(ctx context.Context, actor *models.Actor, "bucketId": bucket.ID.String(), "key": objectKey, })) + s.publishEvent(ctx, bucket.ProjectID, WebhookEventObjectCreated, map[string]any{ + "bucket_id": bucket.ID.String(), + "object_key": objectKey, + "content_type": contentType, + "size_bytes": stored.SizeBytes, + }) return row, nil } @@ -1363,6 +1402,12 @@ func (s *PlatformService) UpdateObject(ctx context.Context, actor *models.Actor, "oldObjectKey": oldObject.ObjectKey, "newObjectKey": updated.ObjectKey, })) + s.publishEvent(ctx, sourceBucket.ProjectID, WebhookEventObjectUpdated, map[string]any{ + "bucket_id": destinationBucket.ID.String(), + "object_key": updated.ObjectKey, + "old_bucket_id": sourceBucket.ID.String(), + "old_object_key": oldObject.ObjectKey, + }) return updated, nil } @@ -1444,6 +1489,14 @@ func (s *PlatformService) CopyObject(ctx context.Context, actor *models.Actor, b "sourceObjectKey": sourceObjectKey, "newObjectKey": created.ObjectKey, })) + s.publishEvent(ctx, sourceBucket.ProjectID, WebhookEventObjectCreated, map[string]any{ + "bucket_id": destinationBucket.ID.String(), + "object_key": created.ObjectKey, + "content_type": created.ContentType, + "size_bytes": created.SizeBytes, + "source_bucket_id": sourceBucket.ID.String(), + "source_object_key": sourceObjectKey, + }) return created, nil } @@ -1469,6 +1522,10 @@ func (s *PlatformService) DeleteObject(ctx context.Context, actor *models.Actor, "bucketId": bucketID.String(), "key": objectKey, })) + s.publishEvent(ctx, bucket.ProjectID, WebhookEventObjectDeleted, map[string]any{ + "bucket_id": bucketID.String(), + "object_key": objectKey, + }) return nil } @@ -1692,6 +1749,11 @@ func (s *PlatformService) CreateDocument(ctx context.Context, actor *models.Acto _, _ = s.repo.Queries().CreateAuditLog(ctx, newAuditParams(project.OrganizationID, collection.ProjectID, actor, requestID, "document.created", "document", row.ID.String(), map[string]any{ "collectionId": collectionID.String(), })) + s.publishEvent(ctx, collection.ProjectID, WebhookEventDocumentCreated, map[string]any{ + "collection_id": collectionID.String(), + "document_id": row.ID.String(), + "document": json.RawMessage(row.Data), + }) return row, nil } @@ -1716,6 +1778,11 @@ func (s *PlatformService) UpdateDocument(ctx context.Context, actor *models.Acto _, _ = s.repo.Queries().CreateAuditLog(ctx, newAuditParams(project.OrganizationID, collection.ProjectID, actor, requestID, "document.updated", "document", documentID.String(), map[string]any{ "collectionId": collectionID.String(), })) + s.publishEvent(ctx, collection.ProjectID, WebhookEventDocumentUpdated, map[string]any{ + "collection_id": collectionID.String(), + "document_id": documentID.String(), + "document": json.RawMessage(row.Data), + }) return row, nil } @@ -1737,6 +1804,10 @@ func (s *PlatformService) DeleteDocument(ctx context.Context, actor *models.Acto _, _ = s.repo.Queries().CreateAuditLog(ctx, newAuditParams(project.OrganizationID, collection.ProjectID, actor, requestID, "document.deleted", "document", documentID.String(), map[string]any{ "collectionId": collectionID.String(), })) + s.publishEvent(ctx, collection.ProjectID, WebhookEventDocumentDeleted, map[string]any{ + "collection_id": collectionID.String(), + "document_id": documentID.String(), + }) return nil } diff --git a/apps/backend/internal/services/telemetry.go b/apps/backend/internal/services/telemetry.go index f4271cb..a2dd8f5 100644 --- a/apps/backend/internal/services/telemetry.go +++ b/apps/backend/internal/services/telemetry.go @@ -192,9 +192,9 @@ func (s *PlatformService) IngestEvents(ctx context.Context, actor *models.Actor, for _, ev := range inserted { s.hub.Broadcast(ev.ProjectID.String(), ev) } - // A fresh error fingerprint is a new issue group — notify webhooks. + // A fresh error fingerprint is a new issue group — notify webhooks + SSE. for _, issue := range newIssues { - s.dispatchEvent(ctx, projectID, WebhookEventIssueCreated, map[string]any{ + s.publishEvent(ctx, projectID, WebhookEventIssueCreated, map[string]any{ "fingerprint": issue.Fingerprint, "message": issue.Message, "severity": issue.Severity, diff --git a/apps/backend/internal/services/webhooks.go b/apps/backend/internal/services/webhooks.go index 475ea55..f9f6925 100644 --- a/apps/backend/internal/services/webhooks.go +++ b/apps/backend/internal/services/webhooks.go @@ -28,15 +28,28 @@ import ( ) const ( - WebhookEventIssueCreated = "issue.created" - WebhookEventDeployMarker = "deploy.marker" - WebhookEventTest = "webhook.test" + WebhookEventIssueCreated = "issue.created" + WebhookEventDeployMarker = "deploy.marker" + WebhookEventTest = "webhook.test" + WebhookEventDocumentCreated = "document.created" + WebhookEventDocumentUpdated = "document.updated" + WebhookEventDocumentDeleted = "document.deleted" + WebhookEventObjectCreated = "object.created" + WebhookEventObjectUpdated = "object.updated" + WebhookEventObjectDeleted = "object.deleted" ) var webhookEventTypes = map[string]bool{ - WebhookEventIssueCreated: true, - WebhookEventDeployMarker: true, - WebhookEventTest: true, + WebhookEventIssueCreated: true, + WebhookEventDeployMarker: true, + WebhookEventTest: true, + WebhookEventJobRun: true, + WebhookEventDocumentCreated: true, + WebhookEventDocumentUpdated: true, + WebhookEventDocumentDeleted: true, + WebhookEventObjectCreated: true, + WebhookEventObjectUpdated: true, + WebhookEventObjectDeleted: true, } const webhookMaxAttempts = 3 @@ -436,7 +449,7 @@ func (s *PlatformService) CreateDeployMarker(ctx context.Context, actor *models. if _, err := s.repo.Queries().CreateAuditLog(ctx, newAuditParams(project.OrganizationID, projectID, actor, requestID, "deploy.marker", "deploy_marker", marker.Ref+marker.Version, meta)); err != nil { return DeployMarker{}, err } - s.dispatchEvent(ctx, projectID, WebhookEventDeployMarker, map[string]any{ + s.publishEvent(ctx, projectID, WebhookEventDeployMarker, map[string]any{ "version": marker.Version, "ref": marker.Ref, "environment": marker.Environment, diff --git a/apps/backend/openapi/openapi.yaml b/apps/backend/openapi/openapi.yaml index 2e9724b..ed0ae66 100644 --- a/apps/backend/openapi/openapi.yaml +++ b/apps/backend/openapi/openapi.yaml @@ -14,6 +14,9 @@ tags: - name: Collections - name: Databases - name: Telemetry + - name: Integrations + - name: Webhooks + - name: Automation paths: /health/liveness: get: @@ -1597,6 +1600,162 @@ paths: $ref: "#/components/schemas/DeployMarker" "401": $ref: "#/components/responses/ErrorResponse" + /projects/{projectID}/jobs: + get: + tags: [Automation] + operationId: listScheduledJobs + security: + - bearerAuth: [] + - apiKeyAuth: [] + parameters: + - $ref: "#/components/parameters/ProjectID" + responses: + "200": + description: Scheduled jobs for the project + content: + application/json: + schema: + $ref: "#/components/schemas/ScheduledJobListResponse" + "401": + $ref: "#/components/responses/ErrorResponse" + post: + tags: [Automation] + operationId: createScheduledJob + summary: Runs a signed POST to the job URL on a cron schedule (standard 5-field cron or @every/@hourly descriptors). Payload is signed X-Primora-Signature like webhook deliveries; an empty secret generates one returned once. HTTPS required for public targets, HTTP allowed for private/self-hosted. + security: + - bearerAuth: [] + - apiKeyAuth: [] + parameters: + - $ref: "#/components/parameters/ProjectID" + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/CreateScheduledJobRequest" + responses: + "201": + description: Job created + content: + application/json: + schema: + $ref: "#/components/schemas/ScheduledJobCreateResponse" + "400": + $ref: "#/components/responses/ErrorResponse" + "401": + $ref: "#/components/responses/ErrorResponse" + "409": + $ref: "#/components/responses/ErrorResponse" + /projects/{projectID}/jobs/{jobID}: + patch: + tags: [Automation] + operationId: updateScheduledJob + security: + - bearerAuth: [] + - apiKeyAuth: [] + parameters: + - $ref: "#/components/parameters/ProjectID" + - $ref: "#/components/parameters/JobID" + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/UpdateScheduledJobRequest" + responses: + "200": + description: Job updated + content: + application/json: + schema: + $ref: "#/components/schemas/ScheduledJobCreateResponse" + "404": + $ref: "#/components/responses/ErrorResponse" + delete: + tags: [Automation] + operationId: deleteScheduledJob + security: + - bearerAuth: [] + - apiKeyAuth: [] + parameters: + - $ref: "#/components/parameters/ProjectID" + - $ref: "#/components/parameters/JobID" + responses: + "204": + description: Job deleted + "404": + $ref: "#/components/responses/ErrorResponse" + /projects/{projectID}/jobs/{jobID}/runs: + get: + tags: [Automation] + operationId: listScheduledJobRuns + summary: Recent executions, newest first. History is capped at 200 runs per job. + security: + - bearerAuth: [] + - apiKeyAuth: [] + parameters: + - $ref: "#/components/parameters/ProjectID" + - $ref: "#/components/parameters/JobID" + - name: limit + in: query + required: false + schema: + type: integer + minimum: 1 + maximum: 200 + default: 50 + responses: + "200": + description: Job run history + content: + application/json: + schema: + $ref: "#/components/schemas/ScheduledJobRunListResponse" + "404": + $ref: "#/components/responses/ErrorResponse" + /projects/{projectID}/jobs/{jobID}/run: + post: + tags: [Automation] + operationId: runScheduledJob + summary: Triggers an out-of-schedule manual run. Returns the run row in "running" state. + security: + - bearerAuth: [] + - apiKeyAuth: [] + parameters: + - $ref: "#/components/parameters/ProjectID" + - $ref: "#/components/parameters/JobID" + responses: + "202": + description: Manual run queued + content: + application/json: + schema: + $ref: "#/components/schemas/ScheduledJobRun" + "404": + $ref: "#/components/responses/ErrorResponse" + /projects/{projectID}/realtime/stream: + get: + tags: [Automation] + operationId: streamRealtimeEvents + summary: Server-sent events stream of project domain events — document/object mutations, new issues, deploy markers, job runs. EventSource clients pass `?api_key=` or `?token=` since headers cannot be set. + security: + - bearerAuth: [] + - apiKeyAuth: [] + parameters: + - $ref: "#/components/parameters/ProjectID" + - name: api_key + in: query + schema: + type: string + - name: token + in: query + schema: + type: string + responses: + "200": + description: text/event-stream of {type, occurred_at, data} messages + "401": + $ref: "#/components/responses/ErrorResponse" components: securitySchemes: bearerAuth: @@ -1730,6 +1889,13 @@ components: schema: type: string format: uuid + JobID: + name: jobID + in: path + required: true + schema: + type: string + format: uuid DatabaseQuery: name: database in: query @@ -3106,7 +3272,7 @@ components: type: array items: type: string - description: Event filter (issue.created, deploy.marker, webhook.test). Empty means all events. + description: Event filter (issue.created, deploy.marker, webhook.test, job.run, document.*, object.*). Empty means all events. enabled: type: boolean has_secret: @@ -3128,7 +3294,7 @@ components: type: array items: type: string - enum: [issue.created, deploy.marker, webhook.test] + enum: [issue.created, deploy.marker, webhook.test, job.run, document.created, document.updated, document.deleted, object.created, object.updated, object.deleted] enabled: type: boolean required: [url] @@ -3145,7 +3311,7 @@ components: type: array items: type: string - enum: [issue.created, deploy.marker, webhook.test] + enum: [issue.created, deploy.marker, webhook.test, job.run, document.created, document.updated, document.deleted, object.created, object.updated, object.deleted] enabled: type: boolean WebhookCreateResponse: @@ -3234,3 +3400,137 @@ components: type: string format: date-time required: [recorded_at] + ScheduledJob: + type: object + properties: + id: + type: string + format: uuid + project_id: + type: string + format: uuid + name: + type: string + schedule: + type: string + description: Cron expression (5-field) or descriptor (@hourly, @every 30m). + url: + type: string + payload: + type: object + description: Static JSON payload merged into every delivery's data field. + enabled: + type: boolean + has_secret: + type: boolean + last_run_at: + type: string + format: date-time + nullable: true + last_status: + type: string + enum: [success, failed] + nullable: true + next_run_at: + type: string + format: date-time + nullable: true + created_at: + type: string + format: date-time + required: [id, project_id, name, schedule, url, payload, enabled, has_secret] + CreateScheduledJobRequest: + type: object + properties: + name: + type: string + minLength: 2 + schedule: + type: string + description: 'Cron expression, e.g. "*/15 * * * *", or descriptor like "@every 1h".' + url: + type: string + format: uri + secret: + type: string + description: HMAC key for X-Primora-Signature. Generated when omitted. + payload: + type: object + description: Static JSON payload delivered as data on every run. + enabled: + type: boolean + default: true + required: [name, schedule, url] + UpdateScheduledJobRequest: + type: object + properties: + name: + type: string + minLength: 2 + schedule: + type: string + url: + type: string + format: uri + secret: + type: string + description: Replace the signing secret. Omit to keep the current one. + payload: + type: object + enabled: + type: boolean + ScheduledJobCreateResponse: + allOf: + - $ref: "#/components/schemas/ScheduledJob" + - type: object + properties: + secret: + type: string + description: Plaintext secret — present only when Primora generated it. Store it; it cannot be recovered. + ScheduledJobListResponse: + type: object + properties: + items: + type: array + items: + $ref: "#/components/schemas/ScheduledJob" + required: [items] + ScheduledJobRun: + type: object + properties: + id: + type: string + format: uuid + job_id: + type: string + format: uuid + status: + type: string + enum: [running, success, failed] + triggered_by: + type: string + enum: [schedule, manual] + status_code: + type: integer + nullable: true + error: + type: string + duration_ms: + type: integer + nullable: true + started_at: + type: string + format: date-time + finished_at: + type: string + format: date-time + nullable: true + required: [id, job_id, status, triggered_by] + ScheduledJobRunListResponse: + type: object + properties: + items: + type: array + items: + $ref: "#/components/schemas/ScheduledJobRun" + required: [items] diff --git a/apps/cli/src/cli.ts b/apps/cli/src/cli.ts index 6195599..6ffb6e0 100644 --- a/apps/cli/src/cli.ts +++ b/apps/cli/src/cli.ts @@ -4,6 +4,7 @@ import { cmdLogin, cmdLogout, cmdWhoami } from "./commands/auth.js"; import { cmdAuditList } from "./commands/audit.js"; import { cmdBucketsCreate, cmdBucketsList } from "./commands/buckets.js"; import { cmdContext, cmdUse } from "./commands/context.js"; +import { cmdJobsCreate, cmdJobsList, cmdJobsRemove, cmdJobsRun, cmdJobsRuns } from "./commands/jobs.js"; import { cmdKeysCreate, cmdKeysList, cmdKeysRevoke } from "./commands/keys.js"; import { cmdObjectsDownload, @@ -104,6 +105,34 @@ cli .option("--yes", "Skip the confirmation prompt") .action(cmdKeysRevoke); +cli + .command("jobs:list", "List scheduled jobs in the current project") + .option("--project ", "Project override") + .action(cmdJobsList); +cli + .command("jobs:create ", "Create a scheduled job (secret shown once if generated)") + .option("--project ", "Project override") + .option("--schedule ", "Cron or descriptor, e.g. '*/15 * * * *' or '@every 1h'") + .option("--url ", "Target endpoint (HTTPS public, HTTP private)") + .option("--payload ", "Static JSON payload merged into every delivery") + .option("--secret ", "HMAC signing secret (generated if omitted)") + .option("--disabled", "Create paused — enable via the dashboard or API") + .action(cmdJobsCreate); +cli + .command("jobs:run ", "Trigger a manual run now") + .option("--project ", "Project override") + .action(cmdJobsRun); +cli + .command("jobs:runs ", "List run history for a job") + .option("--project ", "Project override") + .option("--limit ", "Page size (default 50)") + .action(cmdJobsRuns); +cli + .command("jobs:rm ", "Delete a scheduled job") + .option("--project ", "Project override") + .option("--yes", "Skip the confirmation prompt") + .action(cmdJobsRemove); + cli .command("audit:list", "List audit events for the current project") .option("--project ", "Project override") @@ -121,7 +150,7 @@ cli.version("0.4.0"); // cac only matches the first positional token, so "orgs list" cannot be a // command name. Fold "group verb" pairs into "group:verb" before parsing — // `primora orgs list` and `primora orgs:list` both work. -const GROUPS = new Set(["orgs", "projects", "buckets", "objects", "keys", "audit"]); +const GROUPS = new Set(["orgs", "projects", "buckets", "objects", "keys", "jobs", "audit"]); const argv = process.argv.slice(2); if (GROUPS.has(argv[0]) && argv[1] && !argv[1].startsWith("-")) { argv.splice(0, 2, `${argv[0]}:${argv[1]}`); diff --git a/apps/cli/src/commands/jobs.ts b/apps/cli/src/commands/jobs.ts new file mode 100644 index 0000000..0faa6b0 --- /dev/null +++ b/apps/cli/src/commands/jobs.ts @@ -0,0 +1,149 @@ +import type { ScheduledJob, ScheduledJobRun } from "@primora/api-client"; +import { AutomationService } from "@primora/api-client"; + +import { loadConfig } from "../config.js"; +import { configureClient, requireAuth, requireProject } from "../http.js"; +import { formatDate, isJson, printJson, printTable, success } from "../out.js"; + +interface JobOptions { + project?: string; + json?: boolean; +} + +function jobRow(job: ScheduledJob): string[] { + return [ + job.name, + job.schedule, + job.url, + job.enabled ? "on" : "off", + job.last_status ?? "—", + job.last_run_at ? formatDate(job.last_run_at) : "never", + job.enabled && job.next_run_at ? formatDate(job.next_run_at) : "—", + job.id, + ]; +} + +export async function cmdJobsList(options: JobOptions): Promise { + const cfg = loadConfig(); + requireAuth(cfg); + configureClient(cfg); + + const projectId = requireProject(cfg, options.project); + const page = await AutomationService.listScheduledJobs({ projectId }); + if (isJson(options)) { + printJson(page); + } else { + printTable( + ["NAME", "SCHEDULE", "URL", "ENABLED", "LAST", "LAST RUN", "NEXT RUN", "ID"], + page.items.map(jobRow), + ); + } +} + +export async function cmdJobsCreate( + name: string, + options: JobOptions & { + schedule?: string; + url?: string; + payload?: string; + secret?: string; + disabled?: boolean; + }, +): Promise { + const cfg = loadConfig(); + requireAuth(cfg); + configureClient(cfg); + + if (!options.schedule) throw new Error("pass --schedule, e.g. '*/15 * * * *' or '@every 1h'"); + if (!options.url) throw new Error("pass --url "); + + let payload: Record | undefined; + if (options.payload) { + try { + payload = JSON.parse(options.payload) as Record; + } catch { + throw new Error("--payload must be valid JSON"); + } + } + + const projectId = requireProject(cfg, options.project); + const job = await AutomationService.createScheduledJob({ + projectId, + requestBody: { + name, + schedule: options.schedule, + url: options.url, + secret: options.secret || undefined, + payload, + enabled: !options.disabled, + }, + }); + if (isJson(options)) { + printJson(job); + return; + } + success(`created ${job.name} (${job.id})`); + if (job.secret) { + process.stderr.write(`signing secret (shown once): ${job.secret}\n`); + } +} + +export async function cmdJobsRemove(jobId: string, options: JobOptions & { yes?: boolean }): Promise { + const cfg = loadConfig(); + requireAuth(cfg); + configureClient(cfg); + + const projectId = requireProject(cfg, options.project); + await AutomationService.deleteScheduledJob({ projectId, jobId }); + if (isJson(options)) { + printJson({ deleted: jobId }); + } else { + success(`deleted ${jobId}`); + } +} + +export async function cmdJobsRun(jobId: string, options: JobOptions): Promise { + const cfg = loadConfig(); + requireAuth(cfg); + configureClient(cfg); + + const projectId = requireProject(cfg, options.project); + const run = await AutomationService.runScheduledJob({ projectId, jobId }); + if (isJson(options)) { + printJson(run); + } else { + success(`run queued (${run.id}) — status ${run.status}`); + } +} + +function runRow(run: ScheduledJobRun): string[] { + return [ + formatDate(run.started_at), + run.status, + run.triggered_by, + run.status_code != null ? String(run.status_code) : "—", + run.duration_ms != null ? `${run.duration_ms}ms` : "—", + run.error || "—", + ]; +} + +export async function cmdJobsRuns( + jobId: string, + options: JobOptions & { limit?: string }, +): Promise { + const cfg = loadConfig(); + requireAuth(cfg); + configureClient(cfg); + + const projectId = requireProject(cfg, options.project); + const page = await AutomationService.listScheduledJobRuns({ + projectId, + jobId, + limit: options.limit ? Number(options.limit) : 50, + }); + if (isJson(options)) { + printJson(page); + } else { + printTable(["STARTED", "STATUS", "TRIGGER", "HTTP", "DURATION", "ERROR"], page.items.map(runRow)); + } +} diff --git a/apps/frontend/src/App.tsx b/apps/frontend/src/App.tsx index 1b3ab3a..5938f8f 100644 --- a/apps/frontend/src/App.tsx +++ b/apps/frontend/src/App.tsx @@ -39,6 +39,7 @@ import { import { AuditPage, AuthPage, + AutomationPage, type AuthUser, CollectionsPage, DatabasesPage, @@ -58,6 +59,7 @@ import { IconStorage, IconDatabases, IconTelemetry, + IconZap, IconCollections, IconAudit, IconAuth, @@ -1457,6 +1459,7 @@ export default function App() { { id: "nav-members", label: "Go to Members", category: "Navigate", icon: , keywords: ["team", "users", "invite"], action: () => setActiveView("members") }, { id: "nav-databases", label: "Go to Databases", category: "Navigate", icon: , keywords: ["db", "sql", "query", "tables", "redis"], action: () => setActiveView("databases") }, { id: "nav-telemetry", label: "Go to Telemetry", category: "Navigate", icon: , keywords: ["errors", "metrics", "logs", "observability", "issues"], action: () => setActiveView("telemetry") }, + { id: "nav-automation", label: "Go to Automation", category: "Navigate", icon: , keywords: ["cron", "jobs", "schedule", "events", "realtime"], action: () => setActiveView("automation") }, { id: "nav-storage", label: "Go to Storage", category: "Navigate", icon: , keywords: ["files", "buckets", "upload"], action: () => setActiveView("storage") }, { id: "nav-collections", label: "Go to Collections", category: "Navigate", icon: , keywords: ["documents", "database"], action: () => setActiveView("collections") }, { id: "nav-auth", label: "Go to Authentication", category: "Navigate", icon: , keywords: ["users", "sign in", "ban"], action: () => setActiveView("auth") }, @@ -1687,6 +1690,14 @@ export default function App() { /> + + + + !["collections", "telemetry", "integrations"].includes(e.id)); +const bottomNav: NavEntry[] = projectNav.filter((e) => !["collections", "telemetry", "integrations", "automation"].includes(e.id)); const workspaceNav: NavEntry[] = [ { id: "projects", label: "Projects", icon: IconProjects }, @@ -92,6 +95,7 @@ const viewTitle: Record = { databases: "Databases", telemetry: "Telemetry", integrations: "Integrations", + automation: "Automation", auth: "Authentication", audit: "Audit log", settings: "Settings", diff --git a/apps/frontend/src/lib/auth-client.ts b/apps/frontend/src/lib/auth-client.ts index 0e8bc0d..f2d3e37 100644 --- a/apps/frontend/src/lib/auth-client.ts +++ b/apps/frontend/src/lib/auth-client.ts @@ -1,8 +1,12 @@ import { createAuthClient } from "better-auth/solid"; import { adminClient } from "better-auth/client/plugins"; -// Use full URL for dev mode, relative path for production -const baseURL = import.meta.env.VITE_AUTH_BASE_URL ?? "http://localhost/auth"; +// Same-origin default — nginx routes /auth to the auth service at whatever +// port or domain serves the app. better-auth requires an absolute baseURL, so +// resolve it against the runtime origin; VITE_AUTH_BASE_URL overrides for +// split-origin setups. +const baseURL = + import.meta.env.VITE_AUTH_BASE_URL ?? new URL("/auth", window.location.origin).href; export const authClient = createAuthClient({ baseURL, diff --git a/apps/frontend/src/lib/demo-mode.ts b/apps/frontend/src/lib/demo-mode.ts index 5e94c7f..c2a5044 100644 --- a/apps/frontend/src/lib/demo-mode.ts +++ b/apps/frontend/src/lib/demo-mode.ts @@ -8,6 +8,8 @@ import { OrganizationInvitation, OrganizationMember, ProjectMember, + ScheduledJob, + ScheduledJobRun, WebhookDelivery, } from "@primora/api-client"; import type { @@ -32,6 +34,7 @@ import type { TelemetryStats, Webhook, WebhookCreateResponse, + ScheduledJobCreateResponse, } from "@primora/api-client"; const STORAGE_KEY = "primora_demo_mode"; @@ -485,6 +488,74 @@ const demoDeliveries: WebhookDelivery[] = [ }, ]; +const demoJobs: ScheduledJob[] = [ + { + id: "demo-job-1", + project_id: "demo-project-1", + name: "nightly-rollup", + schedule: "0 3 * * *", + url: "https://api.example.com/jobs/rollup", + payload: { task: "rollup", window: "24h" }, + enabled: true, + has_secret: true, + last_run_at: new Date(Date.now() - 7 * 3_600_000).toISOString(), + last_status: ScheduledJob.last_status.SUCCESS, + next_run_at: new Date(Date.now() + 17 * 3_600_000).toISOString(), + created_at: new Date(Date.now() - 14 * 86_400_000).toISOString(), + }, + { + id: "demo-job-2", + project_id: "demo-project-1", + name: "healthcheck-ping", + schedule: "@every 15m", + url: "https://status.example.com/ping", + payload: {}, + enabled: false, + has_secret: false, + last_run_at: new Date(Date.now() - 3 * 86_400_000).toISOString(), + last_status: ScheduledJob.last_status.FAILED, + created_at: new Date(Date.now() - 30 * 86_400_000).toISOString(), + }, +]; + +const demoJobRuns: Record = { + "demo-job-1": [ + { + id: "demo-run-1", + job_id: "demo-job-1", + status: ScheduledJobRun.status.SUCCESS, + triggered_by: ScheduledJobRun.triggered_by.SCHEDULE, + status_code: 200, + duration_ms: 214, + started_at: new Date(Date.now() - 7 * 3_600_000).toISOString(), + finished_at: new Date(Date.now() - 7 * 3_600_000 + 214).toISOString(), + }, + { + id: "demo-run-2", + job_id: "demo-job-1", + status: ScheduledJobRun.status.FAILED, + triggered_by: ScheduledJobRun.triggered_by.MANUAL, + status_code: 500, + error: "endpoint returned 500", + duration_ms: 1024, + started_at: new Date(Date.now() - 31 * 3_600_000).toISOString(), + finished_at: new Date(Date.now() - 31 * 3_600_000 + 1024).toISOString(), + }, + ], + "demo-job-2": [ + { + id: "demo-run-3", + job_id: "demo-job-2", + status: ScheduledJobRun.status.FAILED, + triggered_by: ScheduledJobRun.triggered_by.SCHEDULE, + error: "connection refused", + duration_ms: 12, + started_at: new Date(Date.now() - 3 * 86_400_000).toISOString(), + finished_at: new Date(Date.now() - 3 * 86_400_000 + 12).toISOString(), + }, + ], +}; + class DemoService { private delay(ms = 250) { return new Promise((resolve) => setTimeout(resolve, ms)); @@ -1233,6 +1304,76 @@ class DemoService { return delivery; } + async listScheduledJobs(): Promise<{ items: ScheduledJob[] }> { + await this.delay(); + return { items: demoJobs }; + } + + async createScheduledJob(data: { + requestBody?: { name?: string; schedule?: string; url?: string; secret?: string; payload?: Record; enabled?: boolean }; + }): Promise { + await this.delay(); + const generated = !data.requestBody?.secret; + const job: ScheduledJob = { + id: `demo-job-${demoJobs.length + 1}`, + project_id: "demo-project-1", + name: data.requestBody?.name ?? "untitled", + schedule: data.requestBody?.schedule ?? "0 * * * *", + url: data.requestBody?.url ?? "https://example.com", + payload: (data.requestBody?.payload ?? {}) as Record, + enabled: data.requestBody?.enabled ?? true, + has_secret: true, + last_status: null, + next_run_at: new Date(Date.now() + 3_600_000).toISOString(), + created_at: new Date().toISOString(), + }; + demoJobs.push(job); + return { ...job, secret: generated ? "demo-secret-" + Math.random().toString(36).slice(2, 18) : undefined }; + } + + async updateScheduledJob(data: { jobId?: string; requestBody?: Partial }): Promise { + await this.delay(); + const job = demoJobs.find((j) => j.id === data.jobId); + if (job && data.requestBody) { + Object.assign(job, data.requestBody); + } + return job as ScheduledJob; + } + + async deleteScheduledJob(data: { jobId?: string }) { + await this.delay(); + const idx = demoJobs.findIndex((j) => j.id === data.jobId); + if (idx >= 0) demoJobs.splice(idx, 1); + return {}; + } + + async listScheduledJobRuns(data: { jobId?: string; limit?: number }): Promise<{ items: ScheduledJobRun[] }> { + await this.delay(); + return { items: (demoJobRuns[data.jobId ?? ""] ?? []).slice(0, data.limit ?? 50) }; + } + + async runScheduledJob(data: { jobId?: string }): Promise { + await this.delay(); + const jobID = data.jobId ?? "demo-job-1"; + const run: ScheduledJobRun = { + id: `demo-run-${Date.now()}`, + job_id: jobID, + status: ScheduledJobRun.status.SUCCESS, + triggered_by: ScheduledJobRun.triggered_by.MANUAL, + status_code: 200, + duration_ms: Math.floor(Math.random() * 800) + 50, + started_at: new Date().toISOString(), + finished_at: new Date().toISOString(), + }; + (demoJobRuns[jobID] ??= []).unshift(run); + const job = demoJobs.find((j) => j.id === jobID); + if (job) { + job.last_run_at = run.started_at; + job.last_status = ScheduledJob.last_status.SUCCESS; + } + return run; + } + async createDeployMarker(data: { requestBody?: { version?: string; ref?: string; environment?: string; note?: string }; }): Promise { diff --git a/apps/frontend/src/pages/AutomationPage.tsx b/apps/frontend/src/pages/AutomationPage.tsx new file mode 100644 index 0000000..781f7fc --- /dev/null +++ b/apps/frontend/src/pages/AutomationPage.tsx @@ -0,0 +1,594 @@ +import { For, Show, createEffect, createSignal, onCleanup } from "solid-js"; +import { + AutomationService, + OpenAPI, + type ScheduledJob, + type ScheduledJobRun, +} from "@primora/api-client"; +import { fetchApiToken } from "../lib/auth-client"; +import { demoService } from "../lib/demo-mode"; +import { Badge } from "../components/Badge"; +import { Modal } from "../components/Modal"; +import { Input, Textarea } from "../components/Input"; +import { + IconPlus, + IconRefresh, + IconTrash, + IconPlay, + IconChevronRight, + IconCheck, + IconCopy, + IconZap, +} from "../components/Icons"; + +interface AutomationPageProps { + projectID?: string; + canManage: boolean; + demoMode: boolean; +} + +type Tab = "schedules" | "events"; + +interface RealtimeEvent { + type: string; + occurred_at: string; + data: Record; +} + +const err = (e: unknown) => (e instanceof Error ? e.message : String(e)); +const svc = (demo: boolean) => + (demo ? (demoService as unknown as typeof AutomationService) : AutomationService); + +const fmtTime = (ts?: string | null) => (ts ? new Date(ts).toLocaleString() : "—"); + +const statusVariant = (s?: string | null) => + s === "success" ? "success" : s === "failed" ? "error" : s === "running" ? "warning" : "neutral"; + +const eventVariant = (t?: string) => + t?.startsWith("document.") + ? "primary" + : t?.startsWith("object.") + ? "success" + : t === "job.run" + ? "warning" + : t === "issue.created" + ? "error" + : "neutral"; + +const DEMO_EVENT_TYPES = ["document.created", "object.created", "job.run", "deploy.marker", "issue.created"]; + +export function AutomationPage(props: AutomationPageProps) { + const [tab, setTab] = createSignal("schedules"); + const [jobs, setJobs] = createSignal([]); + const [runs, setRuns] = createSignal>({}); + const [expandedJob, setExpandedJob] = createSignal(); + const [showCreate, setShowCreate] = createSignal(false); + const [jobForm, setJobForm] = createSignal({ name: "", schedule: "", url: "", payload: "", secret: "", enabled: true }); + const [createdSecret, setCreatedSecret] = createSignal(""); + const [loading, setLoading] = createSignal(false); + const [busy, setBusy] = createSignal(); + const [message, setMessage] = createSignal(""); + const [error, setError] = createSignal(""); + + const [events, setEvents] = createSignal([]); + const [live, setLive] = createSignal(false); + const [paused, setPaused] = createSignal(false); + const [typeFilter, setTypeFilter] = createSignal(""); + + const refresh = async () => { + if (!props.projectID) return; + setLoading(true); + setError(""); + try { + const res = await svc(props.demoMode).listScheduledJobs({ projectId: props.projectID }); + setJobs(res.items ?? []); + } catch (e) { + setError(err(e)); + } finally { + setLoading(false); + } + }; + + const loadRuns = async (jobID: string) => { + if (!props.projectID) return; + try { + const res = await svc(props.demoMode).listScheduledJobRuns({ projectId: props.projectID, jobId: jobID, limit: 50 }); + setRuns((c) => ({ ...c, [jobID]: res.items ?? [] })); + } catch { + /* non-fatal */ + } + }; + + const toggleRuns = (jobID: string) => { + if (expandedJob() === jobID) { + setExpandedJob(undefined); + return; + } + setExpandedJob(jobID); + void loadRuns(jobID); + }; + + createEffect(() => { + if (props.projectID) void refresh(); + }); + + /* Live events — SSE against /realtime/stream; demo mode fakes a feed. */ + createEffect(() => { + if (!props.projectID || tab() !== "events") return; + if (props.demoMode) { + let i = 0; + const timer = setInterval(() => { + if (paused()) return; + const type = DEMO_EVENT_TYPES[i++ % DEMO_EVENT_TYPES.length]; + const data: Record = + type === "job.run" + ? { job_name: "nightly-rollup", status: "success", duration_ms: 42 + i } + : type.startsWith("document.") + ? { collection_id: "demo-col-1", document_id: `doc-${i}`, document: { title: `Note ${i}` } } + : type.startsWith("object.") + ? { bucket_id: "demo-bucket-1", object_key: `uploads/file-${i}.png`, size_bytes: 4096 * i } + : type === "issue.created" + ? { fingerprint: `fp-${i}`, message: "TypeError: demo failure", severity: "error" } + : { version: `0.7.${i}`, environment: "production" }; + setEvents((prev) => [{ type, occurred_at: new Date().toISOString(), data }, ...prev].slice(0, 200)); + }, 2500); + setLive(true); + onCleanup(() => { + clearInterval(timer); + setLive(false); + }); + return; + } + + let es: EventSource | null = null; + let cancelled = false; + void fetchApiToken().then((token) => { + if (cancelled || !token || !props.projectID) return; + const url = `${OpenAPI.BASE}/projects/${props.projectID}/realtime/stream?token=${encodeURIComponent(token)}`; + es = new EventSource(url); + es.addEventListener("event", (m) => { + setLive(true); + if (paused()) return; + try { + const ev = JSON.parse((m as MessageEvent).data) as RealtimeEvent; + setEvents((prev) => [ev, ...prev].slice(0, 200)); + } catch { + /* ignore malformed frames */ + } + }); + es.onerror = () => setLive(false); + es.onopen = () => setLive(true); + }); + onCleanup(() => { + cancelled = true; + es?.close(); + setLive(false); + }); + }); + + const run = async (key: string, task: () => Promise, done?: string) => { + setBusy(key); + setMessage(""); + setError(""); + try { + await task(); + if (done) setMessage(done); + await refresh(); + } catch (e) { + setError(err(e)); + } finally { + setBusy(undefined); + } + }; + + const createJob = (e: Event) => { + e.preventDefault(); + const form = jobForm(); + let payload: Record | undefined; + if (form.payload.trim()) { + try { + payload = JSON.parse(form.payload) as Record; + } catch { + setError("Payload must be valid JSON"); + return; + } + } + void run("create-job", async () => { + const created = await svc(props.demoMode).createScheduledJob({ + projectId: props.projectID!, + requestBody: { + name: form.name.trim(), + schedule: form.schedule.trim(), + url: form.url.trim(), + secret: form.secret.trim() || undefined, + payload, + enabled: form.enabled, + }, + }); + if (created.secret) setCreatedSecret(created.secret); + setJobForm({ name: "", schedule: "", url: "", payload: "", secret: "", enabled: true }); + setShowCreate(false); + }, "Schedule created"); + }; + + const toggleEnabled = (job: ScheduledJob) => + void run(`toggle-${job.id}`, () => + svc(props.demoMode).updateScheduledJob({ + projectId: props.projectID!, + jobId: job.id, + requestBody: { enabled: !job.enabled }, + }), + ); + + const runNow = (job: ScheduledJob) => + void run( + `run-${job.id}`, + () => + svc(props.demoMode).runScheduledJob({ projectId: props.projectID!, jobId: job.id }), + `Run queued for ${job.name}`, + ); + + const deleteJob = (job: ScheduledJob) => + void run( + `del-${job.id}`, + () => svc(props.demoMode).deleteScheduledJob({ projectId: props.projectID!, jobId: job.id }), + `Deleted ${job.name}`, + ); + + const filteredEvents = () => { + const f = typeFilter(); + return f ? events().filter((e) => e.type === f) : events(); + }; + + const eventTypes = () => [...new Set(events().map((e) => e.type))]; + + return ( +
+
+
+

Automation

+

Scheduled jobs and the live event stream for this project.

+
+ +
+ + +
{message()}
+ + +
{error()}
+ + +
+
+ Job signing secret — shown once. + {createdSecret()} +
+ + +
+ + +
+ + +
+ + +
+
+
+
Scheduled jobs
+
+ Cron-driven signed POSTs to your endpoints. Deliveries carry{" "} + X-Primora-Signature (HMAC-SHA256) — same scheme as webhooks. Missed + occurrences collapse into a single run. +
+
+ + + +
+ 0} + fallback={ +

+ No schedules yet. Add one to ping an endpoint on a cron — backups, rollups, report + generation. +

+ } + > +
+ + + + + + + + + + + + + + + {(job) => ( + <> + + + + + + + + + + + + + + + + )} + + +
NameScheduleTargetEnabledActions
{job.name} + {job.schedule} + + {job.url} + + {job.enabled ? "on" : "off"}}> + + + + + + + + +
+
+
+ Run history — last {(runs()[job.id] ?? []).length} of 200 kept +
+ 0} + fallback={

No runs yet.

} + > + + + + + + + + + + + + + + {(r) => ( + + + + + + + + + )} + + +
StatusTriggerStartedError
+ {r.status} + {r.triggered_by}{fmtTime(r.started_at)} + {r.error || "—"} +
+ +
+
+
+ +
+ + + +
+
+
+
+ Live events{" "} + {live() ? "live" : "offline"} +
+
+ Domain events as they happen — document and object mutations, new issues, deploy + markers, job runs. Webhooks subscribed to the same types fire in parallel. +
+
+
+ + +
+
+ 1}> +
+ + + {(t) => ( + + )} + +
+ + 0} + fallback={ +

+ {live() + ? "Listening — create a document, upload a file, or run a job and it lands here." + : "Connecting to the event stream…"} +

+ } + > +
+ + + + {(ev) => ( + + + + + + )} + + +
+ {ev.type} + + {fmtTime(ev.occurred_at)} + + + {JSON.stringify(ev.data)} + +
+
+ +
+ + + setShowCreate(false)} title="New schedule" size="md"> +
+ setJobForm((c) => ({ ...c, name: e.currentTarget.value }))} + placeholder="nightly-rollup" + required + /> +
+ setJobForm((c) => ({ ...c, schedule: e.currentTarget.value }))} + placeholder="*/15 * * * * or @every 1h" + required + /> +

+ Standard 5-field cron, or a descriptor: @hourly, @daily,{" "} + @every 30m. +

+
+
+ setJobForm((c) => ({ ...c, url: e.currentTarget.value }))} + placeholder="https://api.example.com/jobs/rollup" + required + /> +

+ HTTPS for public endpoints; plain HTTP allowed for private/self-hosted targets. +

+
+