Merge pull request #2 from Dvorinka/feat/phase-7-automation-realtime
CI / Backend Tests (push) Successful in 10m6s
CI / Frontend Tests (push) Failing after 4m54s
CI / Build All Services (push) Successful in 19m38s
CI / Docker Build Test (push) Successful in 1m45s
CI / Migration Up/Down (push) Failing after 5m46s
CI / Compose E2E Smoke (push) Failing after 3s
CI / Lint (push) Successful in 30m8s

feat: phase 7 automation & realtime — scheduled jobs, domain events, SSE
This commit is contained in:
Tomáš Dvořák authored and GitHub committed 2026-09-18 15:37:34 +02:00
commit 438ee46016
51 files changed
+3553 -28

No files matched your search

+3
View File
@@ -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
+15
View File
@@ -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=<hmac-sha256(body)>`, `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 <name> --schedule --url [--payload --secret --disabled]`, `jobs:run <job>`, `jobs:runs <job> --limit`, `jobs:rm <job> --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.
+73 -3
View File
@@ -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=<hmac-sha256(body)>`, 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.
@@ -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;
+88
View File
@@ -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
);
+1
View File
@@ -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
)
+2
View File
@@ -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=
@@ -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
}
@@ -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"`
@@ -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)
+1 -1
View File
@@ -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{}
+7
View File
@@ -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)
+180
View File
@@ -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
}
})
}
@@ -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
}
+4 -3
View File
@@ -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())
+591
View File
@@ -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
}
@@ -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
}
+2 -2
View File
@@ -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,
+14 -1
View File
@@ -31,12 +31,25 @@ const (
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,
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,
+303 -3
View File
@@ -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]
+30 -1
View File
@@ -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 <id>", "Project override")
.action(cmdJobsList);
cli
.command("jobs:create <name>", "Create a scheduled job (secret shown once if generated)")
.option("--project <id>", "Project override")
.option("--schedule <cron>", "Cron or descriptor, e.g. '*/15 * * * *' or '@every 1h'")
.option("--url <url>", "Target endpoint (HTTPS public, HTTP private)")
.option("--payload <json>", "Static JSON payload merged into every delivery")
.option("--secret <secret>", "HMAC signing secret (generated if omitted)")
.option("--disabled", "Create paused — enable via the dashboard or API")
.action(cmdJobsCreate);
cli
.command("jobs:run <job>", "Trigger a manual run now")
.option("--project <id>", "Project override")
.action(cmdJobsRun);
cli
.command("jobs:runs <job>", "List run history for a job")
.option("--project <id>", "Project override")
.option("--limit <n>", "Page size (default 50)")
.action(cmdJobsRuns);
cli
.command("jobs:rm <job>", "Delete a scheduled job")
.option("--project <id>", "Project override")
.option("--yes", "Skip the confirmation prompt")
.action(cmdJobsRemove);
cli
.command("audit:list", "List audit events for the current project")
.option("--project <id>", "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]}`);
+149
View File
@@ -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<void> {
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<void> {
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 <https endpoint>");
let payload: Record<string, unknown> | undefined;
if (options.payload) {
try {
payload = JSON.parse(options.payload) as Record<string, unknown>;
} 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<void> {
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<void> {
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<void> {
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));
}
}
+11
View File
@@ -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: <IconMembers class="w-4 h-4" />, keywords: ["team", "users", "invite"], action: () => setActiveView("members") },
{ id: "nav-databases", label: "Go to Databases", category: "Navigate", icon: <IconDatabases class="w-4 h-4" />, keywords: ["db", "sql", "query", "tables", "redis"], action: () => setActiveView("databases") },
{ id: "nav-telemetry", label: "Go to Telemetry", category: "Navigate", icon: <IconTelemetry class="w-4 h-4" />, keywords: ["errors", "metrics", "logs", "observability", "issues"], action: () => setActiveView("telemetry") },
{ id: "nav-automation", label: "Go to Automation", category: "Navigate", icon: <IconZap class="w-4 h-4" />, keywords: ["cron", "jobs", "schedule", "events", "realtime"], action: () => setActiveView("automation") },
{ id: "nav-storage", label: "Go to Storage", category: "Navigate", icon: <IconStorage class="w-4 h-4" />, keywords: ["files", "buckets", "upload"], action: () => setActiveView("storage") },
{ id: "nav-collections", label: "Go to Collections", category: "Navigate", icon: <IconCollections class="w-4 h-4" />, keywords: ["documents", "database"], action: () => setActiveView("collections") },
{ id: "nav-auth", label: "Go to Authentication", category: "Navigate", icon: <IconAuth class="w-4 h-4" />, keywords: ["users", "sign in", "ban"], action: () => setActiveView("auth") },
@@ -1687,6 +1690,14 @@ export default function App() {
/>
</Show>
<Show when={activeView() === "automation"}>
<AutomationPage
projectID={activeProject()?.id}
canManage={canUpdateProject()}
demoMode={isDemo}
/>
</Show>
<Show when={activeView() === "storage"}>
<StoragePage
buckets={buckets()}
+5 -1
View File
@@ -10,6 +10,7 @@ import {
IconDatabases,
IconTelemetry,
IconIntegrations,
IconZap,
IconAudit,
IconAuth,
IconSettings,
@@ -30,6 +31,7 @@ export type ViewType =
| "databases"
| "telemetry"
| "integrations"
| "automation"
| "auth"
| "audit"
| "settings";
@@ -46,6 +48,7 @@ const projectNav: NavEntry[] = [
{ id: "databases", label: "Databases", shortLabel: "DBs", icon: IconDatabases },
{ id: "telemetry", label: "Telemetry", shortLabel: "Signals", icon: IconTelemetry },
{ id: "integrations", label: "Integrations", shortLabel: "Integr.", icon: IconIntegrations },
{ id: "automation", label: "Automation", shortLabel: "Auto", icon: IconZap },
{ id: "storage", label: "Storage", icon: IconStorage },
{ id: "collections", label: "Collections", icon: IconCollections },
{ id: "auth", label: "Authentication", shortLabel: "Auth", icon: IconAuth },
@@ -53,7 +56,7 @@ const projectNav: NavEntry[] = [
];
// Instagram-style bottom bar caps at five slots — the rest stay reachable via Menu.
const bottomNav: NavEntry[] = projectNav.filter((e) => !["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<ViewType, string> = {
databases: "Databases",
telemetry: "Telemetry",
integrations: "Integrations",
automation: "Automation",
auth: "Authentication",
audit: "Audit log",
settings: "Settings",
+6 -2
View File
@@ -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,
+141
View File
@@ -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<string, ScheduledJobRun[]> = {
"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<string, unknown>; enabled?: boolean };
}): Promise<ScheduledJobCreateResponse> {
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<string, never>,
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<ScheduledJob> }): Promise<ScheduledJob> {
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<ScheduledJobRun> {
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<DeployMarker> {
+594
View File
@@ -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<string, unknown>;
}
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<Tab>("schedules");
const [jobs, setJobs] = createSignal<ScheduledJob[]>([]);
const [runs, setRuns] = createSignal<Record<string, ScheduledJobRun[]>>({});
const [expandedJob, setExpandedJob] = createSignal<string>();
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<string>();
const [message, setMessage] = createSignal("");
const [error, setError] = createSignal("");
const [events, setEvents] = createSignal<RealtimeEvent[]>([]);
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<string, unknown> =
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<unknown>, 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<string, unknown> | undefined;
if (form.payload.trim()) {
try {
payload = JSON.parse(form.payload) as Record<string, unknown>;
} 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 (
<div class="page">
<div class="page-header">
<div>
<h1 class="page-title">Automation</h1>
<p class="page-description">Scheduled jobs and the live event stream for this project.</p>
</div>
<button class="btn btn-secondary" onClick={() => void refresh()} disabled={loading()}>
<IconRefresh class="w-4 h-4" />
Refresh
</button>
</div>
<Show when={message()}>
<div class="message message-success mb-4">{message()}</div>
</Show>
<Show when={error()}>
<div class="message message-error mb-4">{error()}</div>
</Show>
<Show when={createdSecret()}>
<div class="message message-warning mb-4">
<div class="flex-1 min-w-0">
<strong>Job signing secret — shown once.</strong>
<code class="block mt-1" style="word-break: break-all">{createdSecret()}</code>
</div>
<button class="btn btn-ghost btn-sm" onClick={() => navigator.clipboard?.writeText(createdSecret())} title="Copy secret">
<IconCopy class="w-4 h-4" />
</button>
<button class="btn btn-ghost btn-sm" onClick={() => setCreatedSecret("")} title="Dismiss">
<IconCheck class="w-4 h-4" />
</button>
</div>
</Show>
<div class="tabs mb-4">
<button class={`tab ${tab() === "schedules" ? "active" : ""}`} onClick={() => setTab("schedules")}>
Schedules
</button>
<button class={`tab ${tab() === "events" ? "active" : ""}`} onClick={() => setTab("events")}>
Live events
</button>
</div>
<Show when={tab() === "schedules"}>
<div class="card card-flush">
<div class="card-header">
<div class="flex-1">
<div class="card-header-title">Scheduled jobs</div>
<div class="card-header-description">
Cron-driven signed POSTs to your endpoints. Deliveries carry{" "}
<code>X-Primora-Signature</code> (HMAC-SHA256) — same scheme as webhooks. Missed
occurrences collapse into a single run.
</div>
</div>
<Show when={props.canManage}>
<button class="btn btn-primary btn-sm" onClick={() => setShowCreate(true)}>
<IconPlus class="w-4 h-4" />
New schedule
</button>
</Show>
</div>
<Show
when={jobs().length > 0}
fallback={
<p class="text-text-2 text-sm p-4">
No schedules yet. Add one to ping an endpoint on a cron — backups, rollups, report
generation.
</p>
}
>
<div class="table-container">
<table class="table">
<thead>
<tr>
<th>Name</th>
<th>Schedule</th>
<th>Target</th>
<th class="hidden md:table-cell">Next run</th>
<th class="hidden md:table-cell">Last run</th>
<th>Enabled</th>
<th style="text-align:right">Actions</th>
</tr>
</thead>
<tbody>
<For each={jobs()}>
{(job) => (
<>
<tr>
<td class="font-medium">{job.name}</td>
<td>
<code class="text-xs">{job.schedule}</code>
</td>
<td class="text-text-2" style="max-width:14rem;overflow:hidden;text-overflow:ellipsis">
{job.url}
</td>
<td class="hidden md:table-cell text-text-2">{job.enabled ? fmtTime(job.next_run_at) : "—"}</td>
<td class="hidden md:table-cell">
<Show when={job.last_run_at} fallback={<span class="text-text-2">never</span>}>
<Badge variant={statusVariant(job.last_status)}>{job.last_status}</Badge>{" "}
<span class="text-text-2 text-xs">{fmtTime(job.last_run_at)}</span>
</Show>
</td>
<td>
<Show when={props.canManage} fallback={<Badge variant={job.enabled ? "success" : "neutral"}>{job.enabled ? "on" : "off"}</Badge>}>
<button
class={`badge ${job.enabled ? "badge-success" : ""} cursor-pointer border-0`}
onClick={() => toggleEnabled(job)}
disabled={busy() === `toggle-${job.id}`}
title={job.enabled ? "Disable" : "Enable"}
>
{job.enabled ? "on" : "off"}
</button>
</Show>
</td>
<td style="text-align:right;white-space:nowrap">
<button class="btn btn-ghost btn-sm" onClick={() => toggleRuns(job.id)}>
<IconChevronRight class={`w-4 h-4 ${expandedJob() === job.id ? "rotate-90" : ""}`} />
Runs
</button>
<Show when={props.canManage}>
<button
class="btn btn-ghost btn-sm"
onClick={() => runNow(job)}
disabled={busy() === `run-${job.id}`}
title="Trigger a manual run now"
>
<IconPlay class="w-4 h-4" />
Run
</button>
<button
class="btn btn-ghost btn-sm"
onClick={() => deleteJob(job)}
disabled={busy() === `del-${job.id}`}
>
<IconTrash class="w-4 h-4" />
</button>
</Show>
</td>
</tr>
<Show when={expandedJob() === job.id}>
<tr>
<td colspan={7} style="padding:0">
<div class="p-4" style="background:var(--bg-subtle)">
<div class="text-xs text-text-2 mb-2 font-medium uppercase tracking-wide">
Run history — last {(runs()[job.id] ?? []).length} of 200 kept
</div>
<Show
when={(runs()[job.id] ?? []).length > 0}
fallback={<p class="text-text-2 text-sm">No runs yet.</p>}
>
<table class="table">
<thead>
<tr>
<th>Status</th>
<th>Trigger</th>
<th class="hidden md:table-cell">HTTP</th>
<th class="hidden md:table-cell">Duration</th>
<th>Started</th>
<th>Error</th>
</tr>
</thead>
<tbody>
<For each={runs()[job.id] ?? []}>
{(r) => (
<tr>
<td>
<Badge variant={statusVariant(r.status)}>{r.status}</Badge>
</td>
<td class="text-text-2">{r.triggered_by}</td>
<td class="hidden md:table-cell text-text-2">{r.status_code ?? "—"}</td>
<td class="hidden md:table-cell text-text-2">
{r.duration_ms != null ? `${r.duration_ms} ms` : "—"}
</td>
<td class="text-text-2">{fmtTime(r.started_at)}</td>
<td class="text-text-2" style="max-width:16rem;overflow:hidden;text-overflow:ellipsis">
{r.error || "—"}
</td>
</tr>
)}
</For>
</tbody>
</table>
</Show>
</div>
</td>
</tr>
</Show>
</>
)}
</For>
</tbody>
</table>
</div>
</Show>
</div>
</Show>
<Show when={tab() === "events"}>
<div class="card card-flush">
<div class="card-header">
<div class="flex-1">
<div class="card-header-title">
Live events{" "}
<Badge variant={live() ? "success" : "neutral"}>{live() ? "live" : "offline"}</Badge>
</div>
<div class="card-header-description">
Domain events as they happen — document and object mutations, new issues, deploy
markers, job runs. Webhooks subscribed to the same types fire in parallel.
</div>
</div>
<div class="flex gap-2">
<button class="btn btn-secondary btn-sm" onClick={() => setPaused(!paused())}>
{paused() ? "Resume" : "Pause"}
</button>
<button class="btn btn-secondary btn-sm" onClick={() => setEvents([])}>
Clear
</button>
</div>
</div>
<Show when={eventTypes().length > 1}>
<div class="flex gap-2 p-4 pb-0 flex-wrap">
<button
class={`badge cursor-pointer border-0 ${typeFilter() === "" ? "badge-primary" : ""}`}
onClick={() => setTypeFilter("")}
>
all
</button>
<For each={eventTypes()}>
{(t) => (
<button
class={`badge cursor-pointer border-0 ${typeFilter() === t ? "badge-primary" : ""}`}
onClick={() => setTypeFilter(typeFilter() === t ? "" : t)}
>
{t}
</button>
)}
</For>
</div>
</Show>
<Show
when={filteredEvents().length > 0}
fallback={
<p class="text-text-2 text-sm p-4">
{live()
? "Listening — create a document, upload a file, or run a job and it lands here."
: "Connecting to the event stream…"}
</p>
}
>
<div class="table-container">
<table class="table">
<tbody>
<For each={filteredEvents()}>
{(ev) => (
<tr>
<td style="width:11rem">
<Badge variant={eventVariant(ev.type)}>{ev.type}</Badge>
</td>
<td class="text-text-2 text-xs" style="width:11rem">
{fmtTime(ev.occurred_at)}
</td>
<td>
<code
class="text-xs block"
style="white-space:nowrap;overflow:hidden;text-overflow:ellipsis;max-width:36rem"
title={JSON.stringify(ev.data, null, 2)}
>
{JSON.stringify(ev.data)}
</code>
</td>
</tr>
)}
</For>
</tbody>
</table>
</div>
</Show>
</div>
</Show>
<Modal open={showCreate()} onClose={() => setShowCreate(false)} title="New schedule" size="md">
<form onSubmit={createJob} class="space-y-4">
<Input
label="Name"
value={jobForm().name}
onInput={(e) => setJobForm((c) => ({ ...c, name: e.currentTarget.value }))}
placeholder="nightly-rollup"
required
/>
<div>
<Input
label="Schedule"
value={jobForm().schedule}
onInput={(e) => setJobForm((c) => ({ ...c, schedule: e.currentTarget.value }))}
placeholder="*/15 * * * * or @every 1h"
required
/>
<p class="label-hint">
Standard 5-field cron, or a descriptor: <code>@hourly</code>, <code>@daily</code>,{" "}
<code>@every 30m</code>.
</p>
</div>
<div>
<Input
label="Target URL"
value={jobForm().url}
onInput={(e) => setJobForm((c) => ({ ...c, url: e.currentTarget.value }))}
placeholder="https://api.example.com/jobs/rollup"
required
/>
<p class="label-hint">
HTTPS for public endpoints; plain HTTP allowed for private/self-hosted targets.
</p>
</div>
<Textarea
label="Payload (optional JSON)"
class="font-mono text-xs"
rows={3}
value={jobForm().payload}
onInput={(e) => setJobForm((c) => ({ ...c, payload: e.currentTarget.value }))}
placeholder='{"task": "rollup", "window": "24h"}'
/>
<Input
label="Signing secret (optional)"
value={jobForm().secret}
onInput={(e) => setJobForm((c) => ({ ...c, secret: e.currentTarget.value }))}
placeholder="generated if empty"
/>
<label class="flex items-center gap-2 text-sm">
<input
type="checkbox"
checked={jobForm().enabled}
onChange={(e) => setJobForm((c) => ({ ...c, enabled: e.currentTarget.checked }))}
/>
Enabled — start running on schedule immediately
</label>
<div class="flex gap-2 justify-end">
<button type="button" class="btn btn-secondary" onClick={() => setShowCreate(false)}>
Cancel
</button>
<button type="submit" class="btn btn-primary" disabled={busy() === "create-job"}>
<IconZap class="w-4 h-4" />
Create schedule
</button>
</div>
</form>
</Modal>
</div>
);
}
+1
View File
@@ -7,6 +7,7 @@ export { CollectionsPage } from "./CollectionsPage";
export { DatabasesPage } from "./DatabasesPage";
export { TelemetryPage } from "./TelemetryPage";
export { IntegrationsPage } from "./IntegrationsPage";
export { AutomationPage } from "./AutomationPage";
export { SettingsPage } from "./SettingsPage";
export { AuditPage } from "./AuditPage";
export { AuthPage, type AuthUser } from "./AuthPage";
+149 -1
View File
@@ -3,6 +3,7 @@ import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js"
import { z } from "zod";
import {
AutomationService,
CreateBucketRequest,
IntegrationsService,
OpenAPI,
@@ -574,7 +575,20 @@ server.registerTool(
const webhookRef = z.string().describe("Webhook id (see primora_list_webhooks)");
const webhookEvents = z
.array(z.enum(["issue.created", "deploy.marker", "webhook.test"]))
.array(
z.enum([
"issue.created",
"deploy.marker",
"webhook.test",
"job.run",
"document.created",
"document.updated",
"document.deleted",
"object.created",
"object.updated",
"object.deleted",
]),
)
.describe("Event filter — empty means every event");
server.registerTool(
@@ -732,4 +746,138 @@ server.registerTool(
},
);
/* ---------------------------------------------------------- */
/* automation — scheduled jobs */
/* ---------------------------------------------------------- */
const jobRef = z.string().describe("Scheduled job id (see primora_list_jobs)");
server.registerTool(
"primora_list_jobs",
{
description: "List scheduled jobs for a project — cron schedule, target URL, last/next run",
inputSchema: { projectId: z.string().optional() },
},
async ({ projectId: p }) => {
try {
return ok(await AutomationService.listScheduledJobs({ projectId: projectId(p) }));
} catch (e) {
return fail(e);
}
},
);
server.registerTool(
"primora_create_job",
{
description:
"Create a scheduled job — cron expression or @descriptor, signed POST to url. Omit secret to generate one (returned once)",
inputSchema: {
name: z.string(),
schedule: z
.string()
.describe('Cron (5-field, e.g. "*/15 * * * *") or descriptor: @hourly, @daily, @every 30m'),
url: z.string().describe("HTTPS endpoint (http allowed only for private/self-hosted targets)"),
secret: z.string().optional(),
payload: z.record(z.unknown()).optional().describe("Static JSON payload merged into every delivery"),
enabled: z.boolean().optional(),
projectId: z.string().optional(),
},
},
async ({ name, schedule, url, secret, payload, enabled, projectId: p }) => {
try {
return ok(
await AutomationService.createScheduledJob({
projectId: projectId(p),
requestBody: { name, schedule, url, secret, payload, enabled },
}),
);
} catch (e) {
return fail(e);
}
},
);
server.registerTool(
"primora_update_job",
{
description: "Update a scheduled job — name, schedule, url, payload, enabled flag, or rotate the secret",
inputSchema: {
jobId: jobRef,
name: z.string().optional(),
schedule: z.string().optional(),
url: z.string().optional(),
secret: z.string().optional(),
payload: z.record(z.unknown()).optional(),
enabled: z.boolean().optional(),
projectId: z.string().optional(),
},
},
async ({ jobId, name, schedule, url, secret, payload, enabled, projectId: p }) => {
try {
return ok(
await AutomationService.updateScheduledJob({
projectId: projectId(p),
jobId,
requestBody: { name, schedule, url, secret, payload, enabled },
}),
);
} catch (e) {
return fail(e);
}
},
);
server.registerTool(
"primora_delete_job",
{
description: "Delete a scheduled job and stop future runs",
inputSchema: { jobId: jobRef, projectId: z.string().optional() },
},
async ({ jobId, projectId: p }) => {
try {
await AutomationService.deleteScheduledJob({ projectId: projectId(p), jobId });
return ok({ deleted: jobId });
} catch (e) {
return fail(e);
}
},
);
server.registerTool(
"primora_list_job_runs",
{
description: "Run history for a scheduled job — status, HTTP code, duration, errors",
inputSchema: {
jobId: jobRef,
limit: z.number().int().optional(),
projectId: z.string().optional(),
},
},
async ({ jobId, limit, projectId: p }) => {
try {
return ok(
await AutomationService.listScheduledJobRuns({ projectId: projectId(p), jobId, limit }),
);
} catch (e) {
return fail(e);
}
},
);
server.registerTool(
"primora_run_job",
{
description: "Trigger a manual run of a scheduled job now",
inputSchema: { jobId: jobRef, projectId: z.string().optional() },
},
async ({ jobId, projectId: p }) => {
try {
return ok(await AutomationService.runScheduledJob({ projectId: projectId(p), jobId }));
} catch (e) {
return fail(e);
}
},
);
await server.connect(new StdioServerTransport());
+3 -3
View File
@@ -64,9 +64,9 @@ services:
environment:
DATABASE_URL: postgres://${POSTGRES_USER}:${POSTGRES_PASSWORD}@postgres:5432/${POSTGRES_DB}?sslmode=disable
DRAGONFLY_URL: redis://dragonfly:6379/0
BETTER_AUTH_URL: http://localhost/auth
AUTH_BASE_URL: http://localhost/auth
VITE_APP_URL: http://localhost
BETTER_AUTH_URL: ${BETTER_AUTH_URL:-http://localhost/auth}
AUTH_BASE_URL: ${AUTH_BASE_URL:-http://localhost/auth}
VITE_APP_URL: ${VITE_APP_URL:-http://localhost}
depends_on:
postgres:
condition: service_healthy
@@ -38,6 +38,7 @@ export { CreateIntegrationRequest } from './models/CreateIntegrationRequest';
export { CreateInvitationRequest } from './models/CreateInvitationRequest';
export type { CreateOrganizationRequest } from './models/CreateOrganizationRequest';
export type { CreateProjectRequest } from './models/CreateProjectRequest';
export type { CreateScheduledJobRequest } from './models/CreateScheduledJobRequest';
export type { CreateWebhookRequest } from './models/CreateWebhookRequest';
export type { DatabaseQuery } from './models/DatabaseQuery';
export type { DBConnection } from './models/DBConnection';
@@ -75,6 +76,7 @@ export type { IntegrationListResponse } from './models/IntegrationListResponse';
export type { IntegrationTestResult } from './models/IntegrationTestResult';
export type { Invitation } from './models/Invitation';
export type { InvitationID } from './models/InvitationID';
export type { JobID } from './models/JobID';
export type { MeResponse } from './models/MeResponse';
export type { MetricPoint } from './models/MetricPoint';
export type { MetricSeriesResponse } from './models/MetricSeriesResponse';
@@ -95,6 +97,11 @@ export type { ProjectSummary } from './models/ProjectSummary';
export type { ReadinessStatus } from './models/ReadinessStatus';
export type { RybbitMetricItem } from './models/RybbitMetricItem';
export type { RybbitSite } from './models/RybbitSite';
export { ScheduledJob } from './models/ScheduledJob';
export type { ScheduledJobCreateResponse } from './models/ScheduledJobCreateResponse';
export type { ScheduledJobListResponse } from './models/ScheduledJobListResponse';
export { ScheduledJobRun } from './models/ScheduledJobRun';
export type { ScheduledJobRunListResponse } from './models/ScheduledJobRunListResponse';
export type { SchemaQuery } from './models/SchemaQuery';
export type { TableQuery } from './models/TableQuery';
export type { TelemetryComponent } from './models/TelemetryComponent';
@@ -112,6 +119,7 @@ export { UpdateOrganizationMemberRoleRequest } from './models/UpdateOrganization
export type { UpdateOrganizationRequest } from './models/UpdateOrganizationRequest';
export { UpdateProjectMemberRoleRequest } from './models/UpdateProjectMemberRoleRequest';
export type { UpdateProjectRequest } from './models/UpdateProjectRequest';
export type { UpdateScheduledJobRequest } from './models/UpdateScheduledJobRequest';
export type { UpdateWebhookRequest } from './models/UpdateWebhookRequest';
export type { UserID } from './models/UserID';
export type { UserSummary } from './models/UserSummary';
@@ -153,6 +161,7 @@ export { $CreateIntegrationRequest } from './schemas/$CreateIntegrationRequest';
export { $CreateInvitationRequest } from './schemas/$CreateInvitationRequest';
export { $CreateOrganizationRequest } from './schemas/$CreateOrganizationRequest';
export { $CreateProjectRequest } from './schemas/$CreateProjectRequest';
export { $CreateScheduledJobRequest } from './schemas/$CreateScheduledJobRequest';
export { $CreateWebhookRequest } from './schemas/$CreateWebhookRequest';
export { $DatabaseQuery } from './schemas/$DatabaseQuery';
export { $DBConnection } from './schemas/$DBConnection';
@@ -190,6 +199,7 @@ export { $IntegrationListResponse } from './schemas/$IntegrationListResponse';
export { $IntegrationTestResult } from './schemas/$IntegrationTestResult';
export { $Invitation } from './schemas/$Invitation';
export { $InvitationID } from './schemas/$InvitationID';
export { $JobID } from './schemas/$JobID';
export { $MeResponse } from './schemas/$MeResponse';
export { $MetricPoint } from './schemas/$MetricPoint';
export { $MetricSeriesResponse } from './schemas/$MetricSeriesResponse';
@@ -210,6 +220,11 @@ export { $ProjectSummary } from './schemas/$ProjectSummary';
export { $ReadinessStatus } from './schemas/$ReadinessStatus';
export { $RybbitMetricItem } from './schemas/$RybbitMetricItem';
export { $RybbitSite } from './schemas/$RybbitSite';
export { $ScheduledJob } from './schemas/$ScheduledJob';
export { $ScheduledJobCreateResponse } from './schemas/$ScheduledJobCreateResponse';
export { $ScheduledJobListResponse } from './schemas/$ScheduledJobListResponse';
export { $ScheduledJobRun } from './schemas/$ScheduledJobRun';
export { $ScheduledJobRunListResponse } from './schemas/$ScheduledJobRunListResponse';
export { $SchemaQuery } from './schemas/$SchemaQuery';
export { $TableQuery } from './schemas/$TableQuery';
export { $TelemetryComponent } from './schemas/$TelemetryComponent';
@@ -227,6 +242,7 @@ export { $UpdateOrganizationMemberRoleRequest } from './schemas/$UpdateOrganizat
export { $UpdateOrganizationRequest } from './schemas/$UpdateOrganizationRequest';
export { $UpdateProjectMemberRoleRequest } from './schemas/$UpdateProjectMemberRoleRequest';
export { $UpdateProjectRequest } from './schemas/$UpdateProjectRequest';
export { $UpdateScheduledJobRequest } from './schemas/$UpdateScheduledJobRequest';
export { $UpdateWebhookRequest } from './schemas/$UpdateWebhookRequest';
export { $UserID } from './schemas/$UserID';
export { $UserSummary } from './schemas/$UserSummary';
@@ -237,6 +253,7 @@ export { $WebhookDeliveryListResponse } from './schemas/$WebhookDeliveryListResp
export { $WebhookID } from './schemas/$WebhookID';
export { $WebhookListResponse } from './schemas/$WebhookListResponse';
export { AutomationService } from './services/AutomationService';
export { CollectionsService } from './services/CollectionsService';
export { DatabasesService } from './services/DatabasesService';
export { HealthService } from './services/HealthService';
@@ -0,0 +1,22 @@
/* generated using openapi-typescript-codegen -- do not edit */
/* istanbul ignore file */
/* tslint:disable */
/* eslint-disable */
export type CreateScheduledJobRequest = {
name: string;
/**
* Cron expression, e.g. "*15 * * * *", or descriptor like "@every 1h".
*/
schedule: string;
url: string;
/**
* HMAC key for X-Primora-Signature. Generated when omitted.
*/
secret?: string;
/**
* Static JSON payload delivered as data on every run.
*/
payload?: Record<string, any>;
enabled?: boolean;
};
@@ -8,7 +8,7 @@ export type CreateWebhookRequest = {
* HMAC key for X-Primora-Signature. Generated when omitted.
*/
secret?: string;
events?: Array<'issue.created' | 'deploy.marker' | 'webhook.test'>;
events?: Array<'issue.created' | 'deploy.marker' | 'webhook.test' | 'job.run' | 'document.created' | 'document.updated' | 'document.deleted' | 'object.created' | 'object.updated' | 'object.deleted'>;
enabled?: boolean;
};
@@ -0,0 +1,5 @@
/* generated using openapi-typescript-codegen -- do not edit */
/* istanbul ignore file */
/* tslint:disable */
/* eslint-disable */
export type JobID = string;
@@ -0,0 +1,31 @@
/* generated using openapi-typescript-codegen -- do not edit */
/* istanbul ignore file */
/* tslint:disable */
/* eslint-disable */
export type ScheduledJob = {
id: string;
project_id: string;
name: string;
/**
* Cron expression (5-field) or descriptor (@hourly, @every 30m).
*/
schedule: string;
url: string;
/**
* Static JSON payload merged into every delivery's data field.
*/
payload: Record<string, any>;
enabled: boolean;
has_secret: boolean;
last_run_at?: string | null;
last_status?: ScheduledJob.last_status | null;
next_run_at?: string | null;
created_at?: string;
};
export namespace ScheduledJob {
export enum last_status {
SUCCESS = 'success',
FAILED = 'failed',
}
}
@@ -0,0 +1,12 @@
/* generated using openapi-typescript-codegen -- do not edit */
/* istanbul ignore file */
/* tslint:disable */
/* eslint-disable */
import type { ScheduledJob } from './ScheduledJob';
export type ScheduledJobCreateResponse = (ScheduledJob & {
/**
* Plaintext secret — present only when Primora generated it. Store it; it cannot be recovered.
*/
secret?: string;
});
@@ -0,0 +1,9 @@
/* generated using openapi-typescript-codegen -- do not edit */
/* istanbul ignore file */
/* tslint:disable */
/* eslint-disable */
import type { ScheduledJob } from './ScheduledJob';
export type ScheduledJobListResponse = {
items: Array<ScheduledJob>;
};
@@ -0,0 +1,27 @@
/* generated using openapi-typescript-codegen -- do not edit */
/* istanbul ignore file */
/* tslint:disable */
/* eslint-disable */
export type ScheduledJobRun = {
id: string;
job_id: string;
status: ScheduledJobRun.status;
triggered_by: ScheduledJobRun.triggered_by;
status_code?: number | null;
error?: string;
duration_ms?: number | null;
started_at?: string;
finished_at?: string | null;
};
export namespace ScheduledJobRun {
export enum status {
RUNNING = 'running',
SUCCESS = 'success',
FAILED = 'failed',
}
export enum triggered_by {
SCHEDULE = 'schedule',
MANUAL = 'manual',
}
}
@@ -0,0 +1,9 @@
/* generated using openapi-typescript-codegen -- do not edit */
/* istanbul ignore file */
/* tslint:disable */
/* eslint-disable */
import type { ScheduledJobRun } from './ScheduledJobRun';
export type ScheduledJobRunListResponse = {
items: Array<ScheduledJobRun>;
};
@@ -0,0 +1,16 @@
/* generated using openapi-typescript-codegen -- do not edit */
/* istanbul ignore file */
/* tslint:disable */
/* eslint-disable */
export type UpdateScheduledJobRequest = {
name?: string;
schedule?: string;
url?: string;
/**
* Replace the signing secret. Omit to keep the current one.
*/
secret?: string;
payload?: Record<string, any>;
enabled?: boolean;
};
@@ -8,7 +8,7 @@ export type UpdateWebhookRequest = {
* Replace the signing secret. Omit to keep the current one.
*/
secret?: string;
events?: Array<'issue.created' | 'deploy.marker' | 'webhook.test'>;
events?: Array<'issue.created' | 'deploy.marker' | 'webhook.test' | 'job.run' | 'document.created' | 'document.updated' | 'document.deleted' | 'object.created' | 'object.updated' | 'object.deleted'>;
enabled?: boolean;
};
@@ -7,7 +7,7 @@ export type Webhook = {
project_id: string;
url: string;
/**
* Event filter (issue.created, deploy.marker, webhook.test). Empty means all events.
* Event filter (issue.created, deploy.marker, webhook.test, job.run, document.*, object.*). Empty means all events.
*/
events: Array<string>;
enabled: boolean;
@@ -0,0 +1,37 @@
/* generated using openapi-typescript-codegen -- do not edit */
/* istanbul ignore file */
/* tslint:disable */
/* eslint-disable */
export const $CreateScheduledJobRequest = {
properties: {
name: {
type: 'string',
isRequired: true,
minLength: 2,
},
schedule: {
type: 'string',
description: `Cron expression, e.g. "*/15 * * * *", or descriptor like "@every 1h".`,
isRequired: true,
},
url: {
type: 'string',
isRequired: true,
format: 'uri',
},
secret: {
type: 'string',
description: `HMAC key for X-Primora-Signature. Generated when omitted.`,
},
payload: {
type: 'dictionary',
contains: {
properties: {
},
},
},
enabled: {
type: 'boolean',
},
},
} as const;
@@ -0,0 +1,8 @@
/* generated using openapi-typescript-codegen -- do not edit */
/* istanbul ignore file */
/* tslint:disable */
/* eslint-disable */
export const $JobID = {
type: 'string',
format: 'uuid',
} as const;
@@ -0,0 +1,65 @@
/* generated using openapi-typescript-codegen -- do not edit */
/* istanbul ignore file */
/* tslint:disable */
/* eslint-disable */
export const $ScheduledJob = {
properties: {
id: {
type: 'string',
isRequired: true,
format: 'uuid',
},
project_id: {
type: 'string',
isRequired: true,
format: 'uuid',
},
name: {
type: 'string',
isRequired: true,
},
schedule: {
type: 'string',
description: `Cron expression (5-field) or descriptor (@hourly, @every 30m).`,
isRequired: true,
},
url: {
type: 'string',
isRequired: true,
},
payload: {
type: 'dictionary',
contains: {
properties: {
},
},
isRequired: true,
},
enabled: {
type: 'boolean',
isRequired: true,
},
has_secret: {
type: 'boolean',
isRequired: true,
},
last_run_at: {
type: 'string',
isNullable: true,
format: 'date-time',
},
last_status: {
type: 'Enum',
isNullable: true,
},
next_run_at: {
type: 'string',
isNullable: true,
format: 'date-time',
},
created_at: {
type: 'string',
format: 'date-time',
},
},
} as const;
@@ -0,0 +1,17 @@
/* generated using openapi-typescript-codegen -- do not edit */
/* istanbul ignore file */
/* tslint:disable */
/* eslint-disable */
export const $ScheduledJobCreateResponse = {
type: 'all-of',
contains: [{
type: 'ScheduledJob',
}, {
properties: {
secret: {
type: 'string',
description: `Plaintext secret — present only when Primora generated it. Store it; it cannot be recovered.`,
},
},
}],
} as const;
@@ -0,0 +1,15 @@
/* generated using openapi-typescript-codegen -- do not edit */
/* istanbul ignore file */
/* tslint:disable */
/* eslint-disable */
export const $ScheduledJobListResponse = {
properties: {
items: {
type: 'array',
contains: {
type: 'ScheduledJob',
},
isRequired: true,
},
},
} as const;
@@ -0,0 +1,46 @@
/* generated using openapi-typescript-codegen -- do not edit */
/* istanbul ignore file */
/* tslint:disable */
/* eslint-disable */
export const $ScheduledJobRun = {
properties: {
id: {
type: 'string',
isRequired: true,
format: 'uuid',
},
job_id: {
type: 'string',
isRequired: true,
format: 'uuid',
},
status: {
type: 'Enum',
isRequired: true,
},
triggered_by: {
type: 'Enum',
isRequired: true,
},
status_code: {
type: 'number',
isNullable: true,
},
error: {
type: 'string',
},
duration_ms: {
type: 'number',
isNullable: true,
},
started_at: {
type: 'string',
format: 'date-time',
},
finished_at: {
type: 'string',
isNullable: true,
format: 'date-time',
},
},
} as const;
@@ -0,0 +1,15 @@
/* generated using openapi-typescript-codegen -- do not edit */
/* istanbul ignore file */
/* tslint:disable */
/* eslint-disable */
export const $ScheduledJobRunListResponse = {
properties: {
items: {
type: 'array',
contains: {
type: 'ScheduledJobRun',
},
isRequired: true,
},
},
} as const;
@@ -0,0 +1,33 @@
/* generated using openapi-typescript-codegen -- do not edit */
/* istanbul ignore file */
/* tslint:disable */
/* eslint-disable */
export const $UpdateScheduledJobRequest = {
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: 'dictionary',
contains: {
properties: {
},
},
},
enabled: {
type: 'boolean',
},
},
} as const;
@@ -0,0 +1,194 @@
/* generated using openapi-typescript-codegen -- do not edit */
/* istanbul ignore file */
/* tslint:disable */
/* eslint-disable */
import type { CreateScheduledJobRequest } from '../models/CreateScheduledJobRequest';
import type { ScheduledJobCreateResponse } from '../models/ScheduledJobCreateResponse';
import type { ScheduledJobListResponse } from '../models/ScheduledJobListResponse';
import type { ScheduledJobRun } from '../models/ScheduledJobRun';
import type { ScheduledJobRunListResponse } from '../models/ScheduledJobRunListResponse';
import type { UpdateScheduledJobRequest } from '../models/UpdateScheduledJobRequest';
import type { CancelablePromise } from '../core/CancelablePromise';
import { OpenAPI } from '../core/OpenAPI';
import { request as __request } from '../core/request';
export class AutomationService {
/**
* @returns ScheduledJobListResponse Scheduled jobs for the project
* @throws ApiError
*/
public static listScheduledJobs({
projectId,
}: {
projectId: string,
}): CancelablePromise<ScheduledJobListResponse> {
return __request(OpenAPI, {
method: 'GET',
url: '/projects/{projectID}/jobs',
path: {
'projectID': projectId,
},
errors: {
401: `Error response`,
},
});
}
/**
* 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.
* @returns ScheduledJobCreateResponse Job created
* @throws ApiError
*/
public static createScheduledJob({
projectId,
requestBody,
}: {
projectId: string,
requestBody: CreateScheduledJobRequest,
}): CancelablePromise<ScheduledJobCreateResponse> {
return __request(OpenAPI, {
method: 'POST',
url: '/projects/{projectID}/jobs',
path: {
'projectID': projectId,
},
body: requestBody,
mediaType: 'application/json',
errors: {
400: `Error response`,
401: `Error response`,
409: `Error response`,
},
});
}
/**
* @returns ScheduledJobCreateResponse Job updated
* @throws ApiError
*/
public static updateScheduledJob({
projectId,
jobId,
requestBody,
}: {
projectId: string,
jobId: string,
requestBody: UpdateScheduledJobRequest,
}): CancelablePromise<ScheduledJobCreateResponse> {
return __request(OpenAPI, {
method: 'PATCH',
url: '/projects/{projectID}/jobs/{jobID}',
path: {
'projectID': projectId,
'jobID': jobId,
},
body: requestBody,
mediaType: 'application/json',
errors: {
404: `Error response`,
},
});
}
/**
* @returns void
* @throws ApiError
*/
public static deleteScheduledJob({
projectId,
jobId,
}: {
projectId: string,
jobId: string,
}): CancelablePromise<void> {
return __request(OpenAPI, {
method: 'DELETE',
url: '/projects/{projectID}/jobs/{jobID}',
path: {
'projectID': projectId,
'jobID': jobId,
},
errors: {
404: `Error response`,
},
});
}
/**
* Recent executions, newest first. History is capped at 200 runs per job.
* @returns ScheduledJobRunListResponse Job run history
* @throws ApiError
*/
public static listScheduledJobRuns({
projectId,
jobId,
limit = 50,
}: {
projectId: string,
jobId: string,
limit?: number,
}): CancelablePromise<ScheduledJobRunListResponse> {
return __request(OpenAPI, {
method: 'GET',
url: '/projects/{projectID}/jobs/{jobID}/runs',
path: {
'projectID': projectId,
'jobID': jobId,
},
query: {
'limit': limit,
},
errors: {
404: `Error response`,
},
});
}
/**
* Triggers an out-of-schedule manual run. Returns the run row in "running" state.
* @returns ScheduledJobRun Manual run queued
* @throws ApiError
*/
public static runScheduledJob({
projectId,
jobId,
}: {
projectId: string,
jobId: string,
}): CancelablePromise<ScheduledJobRun> {
return __request(OpenAPI, {
method: 'POST',
url: '/projects/{projectID}/jobs/{jobID}/run',
path: {
'projectID': projectId,
'jobID': jobId,
},
errors: {
404: `Error response`,
},
});
}
/**
* 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.
* @returns any text/event-stream of {type, occurred_at, data} messages
* @throws ApiError
*/
public static streamRealtimeEvents({
projectId,
apiKey,
token,
}: {
projectId: string,
apiKey?: string,
token?: string,
}): CancelablePromise<any> {
return __request(OpenAPI, {
method: 'GET',
url: '/projects/{projectID}/realtime/stream',
path: {
'projectID': projectId,
},
query: {
'api_key': apiKey,
'token': token,
},
errors: {
401: `Error response`,
},
});
}
}