From a84049b8fc14d9fbd59244dc5ffbd8c092343637 Mon Sep 17 00:00:00 2001 From: Tomas Dvorak Date: Thu, 24 Sep 2026 20:28:04 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20Function=20triggers=20=E2=80=94=20sched?= =?UTF-8?q?ule,=20inbound=20hook,=20domain=20events?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Migration 00016: function_id on scheduled_jobs/inbound_hooks, event_pattern on functions, 'event' in the run-trigger enum. - Jobs deliver to a function instead of HTTP when function_id is set; the run's status_code mirrors the process exit code. Clearing the field returns the job to URL delivery. - Inbound hooks gain mode "function" — received body becomes stdin payload, acked immediately, run recorded async. - event_pattern (document.*, *, exact type) fires the function detached on matching publishEvent — never blocks realtime/webhook fan-out. - One executeFunction path serves manual/schedule/hook/event; run rows record the trigger source. - Backend image now ships bun 1.3.14 so Functions work under compose. Dashboard: job target picker, hook function mode, event-pattern field. Verified live: job run → success (status_code 0), hook POST → run, document.created → event run with correct payload. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- CHANGELOG.md | 1 + DEPLOYMENT_GUIDE.md | 17 ++- README.md | 2 +- ROADMAP.md | 11 +- apps/backend/Dockerfile | 3 + .../db/migrations/00016_function_triggers.sql | 34 +++++ apps/backend/db/queries/functions.sql | 12 +- apps/backend/db/queries/inbound_hooks.sql | 4 +- apps/backend/db/queries/jobs.sql | 6 +- .../internal/database/db/functions.sql.go | 83 ++++++++-- .../internal/database/db/inbound_hooks.sql.go | 32 ++-- apps/backend/internal/database/db/jobs.sql.go | 52 ++++--- apps/backend/internal/database/db/models.go | 19 ++- apps/backend/internal/database/db/querier.go | 1 + apps/backend/internal/services/functions.go | 143 ++++++++++++++---- .../internal/services/functions_test.go | 14 ++ apps/backend/internal/services/inbound.go | 74 +++++++-- apps/backend/internal/services/jobs.go | 91 +++++++++-- .../internal/services/platform_service.go | 5 +- apps/backend/openapi/openapi.yaml | 37 ++++- apps/frontend/src/pages/AutomationPage.tsx | 78 ++++++++-- apps/frontend/src/pages/FunctionsPage.tsx | 50 +++++- apps/frontend/src/pages/IntegrationsPage.tsx | 29 +++- .../generated/models/CreateFunctionRequest.ts | 4 + .../models/CreateInboundHookRequest.ts | 5 + .../models/CreateScheduledJobRequest.ts | 9 +- .../src/generated/models/Function.ts | 4 + .../src/generated/models/FunctionRun.ts | 1 + .../src/generated/models/InboundHook.ts | 2 + .../src/generated/models/ScheduledJob.ts | 4 + .../generated/models/UpdateFunctionRequest.ts | 4 + .../models/UpdateScheduledJobRequest.ts | 4 + .../schemas/$CreateFunctionRequest.ts | 4 + .../schemas/$CreateInboundHookRequest.ts | 4 + .../schemas/$CreateScheduledJobRequest.ts | 7 +- .../src/generated/schemas/$Function.ts | 4 + .../src/generated/schemas/$InboundHook.ts | 3 + .../src/generated/schemas/$ScheduledJob.ts | 6 + .../schemas/$UpdateFunctionRequest.ts | 4 + .../schemas/$UpdateScheduledJobRequest.ts | 6 + 40 files changed, 716 insertions(+), 157 deletions(-) create mode 100644 apps/backend/db/migrations/00016_function_triggers.sql diff --git a/CHANGELOG.md b/CHANGELOG.md index cd7ecbf..5a0c679 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,7 @@ All notable changes to Primora. Format follows [Keep a Changelog](https://keepac - **Email templates + send log** (`core.email_log`, migration 00014) — the mailer renders registered templates (currently `invitation`) and every send — successful or failed — is recorded with template, recipient, subject, status, and error text. `GET /projects/:id/emails` lists recent sends; the Members page shows an "Email log" card under pending invitations. - **S3-compatible object storage backend** — `BACKEND_STORAGE_DRIVER=s3` swaps the local disk store for any S3-compatible API (AWS, MinIO, Garage, R2) via a stdlib-only SigV4 implementation — no AWS SDK dependency. `S3_ENDPOINT`/`S3_REGION`/`S3_BUCKET`/credentials configure it; `S3_PATH_STYLE` (default on) covers MinIO/Garage, off for AWS virtual-hosted. The configured bucket is created on startup if missing; objects stream with computed SHA-256, `storage_path` records `s3://bucket/key`. Verified live end-to-end (bucket → put → get → copy → list → delete). - **Functions** (`core.functions` + `core.function_runs`, migration 00015) — project-scoped JS/TS executed by a runtime binary on the backend host (`bun` or `deno`, per function). The JSON payload arrives on stdin and as `PRIMORA_PAYLOAD`; stdout/stderr are captured per run (capped at 64 KiB) with status, exit code and duration — `FUNCTIONS_RUNTIME`/`FUNCTIONS_TIMEOUT_SECONDS`/`FUNCTIONS_MAX_OUTPUT_BYTES` configure the runner. API: CRUD under `/projects/:id/functions`, `POST …/invoke` (502 + recorded run on failure), `GET …/runs`. Dashboard gains a Functions page: code editor, JSON payload invoke, live result, run history. Code executes with backend privileges — the documented self-hosted trust model. +- **Function triggers** (migration 00016) — three new ways to run a function, all through the same executor and run log. Scheduled jobs accept `function_id` instead of `url` (run status mirrors the exit code); inbound hooks gain `mode: "function"` — the received body becomes the function's stdin payload, acked immediately with the run recorded async; and `event_pattern` on a function (`document.*`, `*`, or an exact type) fires it detached whenever a matching domain event publishes — `{event, project_id, occurred_at, data}` as payload. Runs record trigger source: `manual`/`schedule`/`hook`/`event`. Dashboard: function event-trigger field, job target picker (URL or function), and inbound-hook function mode. ## [0.7.1] - 2026-09-23 diff --git a/DEPLOYMENT_GUIDE.md b/DEPLOYMENT_GUIDE.md index 6132c43..72fe1d5 100644 --- a/DEPLOYMENT_GUIDE.md +++ b/DEPLOYMENT_GUIDE.md @@ -46,6 +46,7 @@ docker compose -f docker-compose.yml -f docker-compose.dev.yml up -d | `S3_ENDPOINT` / `S3_REGION` / `S3_BUCKET` | S3 API endpoint (empty endpoint → AWS `s3..amazonaws.com`), region, bucket | with `s3` | | `S3_ACCESS_KEY_ID` / `S3_SECRET_ACCESS_KEY` | S3 credentials | with `s3` | | `S3_PREFIX` / `S3_PATH_STYLE` | optional key prefix; path-style URLs (default `true`, needed by MinIO/Garage — set `false` for AWS virtual-hosted) | no | +| `FUNCTIONS_RUNTIME` / `FUNCTIONS_TIMEOUT_SECONDS` / `FUNCTIONS_MAX_OUTPUT_BYTES` | Functions: runtime binary (`auto` resolves `bun`/`deno` from the backend's PATH — the backend image must ship one, or set this to its path), per-run timeout (default 30 s), captured-output cap (default 64 KiB) | no | | `AUTH_ADMIN_EMAILS` | comma-separated emails promoted to auth admin on boot | no | | `USER_RATE_LIMIT_PER_MINUTE` / `API_KEY_RATE_LIMIT_PER_MINUTE` | per-identity API rate limits (defaults 240 / 600) | no | | `SMTP_HOST` / `SMTP_PORT` / `SMTP_USER` / `SMTP_PASSWORD` / `SMTP_SECURE` / `MAIL_FROM` | transactional mail; without it verification/reset emails go nowhere | production | @@ -131,10 +132,18 @@ docker compose up -d # 3. rolling recreate; backend migrate ## Storage constraint -Object storage is the **local filesystem** (`backend_storage` volume). There -is no S3-compatible backend yet — the volume must be on durable, backed-up -disk, and a single host serves all objects. Plan capacity accordingly; -multi-node deployments need an external object store first. +Object storage defaults to the **local filesystem** (`backend_storage` +volume) — durable, backed-up disk on a single host. For multi-node or +external object stores set `BACKEND_STORAGE_DRIVER=s3` + the `S3_*` +variables above (AWS, MinIO, Garage, R2 all work). + +## Functions trust model + +Function code runs as a child process of the backend — same host, same +privileges, no sandbox. Treat every project member who can edit a function +as holding a shell on the backend host. If that is unacceptable for your +deployment, do not enable untrusted editors; isolated runtimes are on the +roadmap. ## Observability diff --git a/README.md b/README.md index 2a5de61..52a36e2 100644 --- a/README.md +++ b/README.md @@ -50,7 +50,7 @@ Supabase, with no hosted tier - your data stays on your hardware. - **Auth** - email/password plus optional GitHub, Google, Discord, and Microsoft OAuth. JWTs verified by the API against JWKS. - **Storage** - S3-style buckets on local disk or any S3-compatible backend (AWS, MinIO, Garage, R2), public or private. - **Collections** - schema-flexible JSON documents in Postgres JSONB, with filter/order queries. -- **Functions** - project-scoped JS/TS executed by bun/deno on the host, with per-run logs. +- **Functions** - project-scoped JS/TS executed by bun/deno on the host, with per-run logs. Trigger them four ways: manual invoke, cron schedules (`function_id` on a job), inbound hooks (`mode: "function"`), or domain events (`event_pattern` like `document.*`). - **Vault** - AES-256-GCM project secrets plus an Argon2id/XChaCha20 local vault in the CLI and desktop app. - **Automation** - scheduled jobs, inbound webhooks, telemetry alert rules, and `secret://` refs resolved at delivery. - **Realtime** - SSE event stream with a typed client SDK (`@primora/client`). diff --git a/ROADMAP.md b/ROADMAP.md index 5888b78..58f0dfd 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -357,11 +357,14 @@ Reusable templates + an operational send log. Ordered loosely by leverage. None committed; each gets scoped when picked. -- **Phase 9 — Functions** — ✅ first slice shipped: `core.functions` + +- **Phase 9 — Functions** — ✅ shipped: `core.functions` + `core.function_runs`, bun/deno exec runner (stdin payload, env injection, - capped output, timeout), invoke API + dashboard page. Remaining - candidates: schedule/event/HTTP triggers riding jobs + inbound hooks, - and isolated runtimes (sidecar/Firecracker) for untrusted code. + capped output, timeout), invoke API + dashboard page. Triggers shipped: + scheduled jobs (`function_id` on a job), inbound hooks (`mode: + "function"`), and domain events (`event_pattern`, e.g. `document.*`). + Runs record trigger source (`manual`/`schedule`/`hook`/`event`). + Remaining candidate: isolated runtimes (sidecar/Firecracker) for + untrusted code — functions still execute with backend host privileges. - **Phase 10 — Storage backends** — ✅ S3-compatible object storage shipped: `BACKEND_STORAGE_DRIVER=s3` + `S3_*` env vars, stdlib SigV4, verified against a real S3 server. Presigned-URL uploads for large files remain a diff --git a/apps/backend/Dockerfile b/apps/backend/Dockerfile index 1dbaca7..1657c66 100644 --- a/apps/backend/Dockerfile +++ b/apps/backend/Dockerfile @@ -34,6 +34,9 @@ RUN apt-get update \ && chmod +x /usr/local/bin/dbx-mcp \ && rm /tmp/dbx-mcp.tar.gz +# Functions runtime — bun static binary from the official image (multi-arch). +COPY --from=oven/bun:1.3.14 /usr/local/bin/bun /usr/local/bin/bun + COPY --from=builder /primora-backend /usr/local/bin/primora-backend COPY --from=builder /app/db ./db COPY --from=builder /app/openapi ./openapi diff --git a/apps/backend/db/migrations/00016_function_triggers.sql b/apps/backend/db/migrations/00016_function_triggers.sql new file mode 100644 index 0000000..55c4964 --- /dev/null +++ b/apps/backend/db/migrations/00016_function_triggers.sql @@ -0,0 +1,34 @@ +-- +goose Up +ALTER TABLE core.scheduled_jobs + ADD COLUMN function_id UUID REFERENCES core.functions(id) ON DELETE SET NULL; + +ALTER TABLE core.inbound_hooks + ADD COLUMN function_id UUID REFERENCES core.functions(id) ON DELETE CASCADE; +ALTER TABLE core.inbound_hooks + DROP CONSTRAINT inbound_hooks_mode_check; +ALTER TABLE core.inbound_hooks + ADD CONSTRAINT inbound_hooks_mode_check + CHECK (mode IN ('event','job','function')); + +ALTER TABLE core.functions + ADD COLUMN event_pattern TEXT NOT NULL DEFAULT ''; + +ALTER TABLE core.function_runs + DROP CONSTRAINT function_runs_trigger_check; +ALTER TABLE core.function_runs + ADD CONSTRAINT function_runs_trigger_check + CHECK (trigger IN ('manual','schedule','hook','event')); + +-- +goose Down +ALTER TABLE core.function_runs + DROP CONSTRAINT function_runs_trigger_check; +ALTER TABLE core.function_runs + ADD CONSTRAINT function_runs_trigger_check + CHECK (trigger IN ('manual','schedule','hook')); +ALTER TABLE core.functions DROP COLUMN IF EXISTS event_pattern; +ALTER TABLE core.inbound_hooks + DROP CONSTRAINT inbound_hooks_mode_check; +ALTER TABLE core.inbound_hooks + ADD CONSTRAINT inbound_hooks_mode_check CHECK (mode IN ('event','job')); +ALTER TABLE core.inbound_hooks DROP COLUMN IF EXISTS function_id; +ALTER TABLE core.scheduled_jobs DROP COLUMN IF EXISTS function_id; diff --git a/apps/backend/db/queries/functions.sql b/apps/backend/db/queries/functions.sql index 795d662..1fd16e7 100644 --- a/apps/backend/db/queries/functions.sql +++ b/apps/backend/db/queries/functions.sql @@ -1,6 +1,6 @@ -- name: CreateFunction :one -INSERT INTO core.functions (project_id, name, code, runtime) -VALUES ($1, $2, $3, $4) +INSERT INTO core.functions (project_id, name, code, runtime, event_pattern) +VALUES ($1, $2, $3, $4, $5) RETURNING *; -- name: GetFunction :one @@ -18,6 +18,7 @@ ORDER BY name; UPDATE core.functions SET code = COALESCE(sqlc.narg('code'), code), enabled = COALESCE(sqlc.narg('enabled'), enabled), + event_pattern = COALESCE(sqlc.narg('event_pattern'), event_pattern), updated_at = NOW() WHERE id = $1 AND project_id = $2 RETURNING *; @@ -35,3 +36,10 @@ SELECT * FROM core.function_runs WHERE function_id = $1 ORDER BY created_at DESC LIMIT $2; + +-- name: ListFunctionsForEvent :many +SELECT * FROM core.functions +WHERE project_id = $1 + AND enabled = TRUE + AND event_pattern <> '' + AND ($2 LIKE REPLACE(event_pattern, '*', '%') OR event_pattern = '*'); diff --git a/apps/backend/db/queries/inbound_hooks.sql b/apps/backend/db/queries/inbound_hooks.sql index 1a55588..fd86243 100644 --- a/apps/backend/db/queries/inbound_hooks.sql +++ b/apps/backend/db/queries/inbound_hooks.sql @@ -1,6 +1,6 @@ -- name: CreateInboundHook :one -INSERT INTO core.inbound_hooks (project_id, name, token, mode, job_id, secret, enabled) -VALUES ($1, $2, $3, $4, $5, $6, $7) +INSERT INTO core.inbound_hooks (project_id, name, token, mode, job_id, function_id, secret, enabled) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING *; -- name: ListInboundHooks :many diff --git a/apps/backend/db/queries/jobs.sql b/apps/backend/db/queries/jobs.sql index 98391c7..90115b1 100644 --- a/apps/backend/db/queries/jobs.sql +++ b/apps/backend/db/queries/jobs.sql @@ -17,8 +17,9 @@ INSERT INTO core.scheduled_jobs ( payload, enabled, next_run_at, - created_by_user_id -) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + created_by_user_id, + function_id +) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) RETURNING *; -- name: UpdateScheduledJob :one @@ -30,6 +31,7 @@ SET name = $3, next_run_at = $7, secret = CASE WHEN $8::boolean THEN $9 ELSE secret END, payload = CASE WHEN $10::boolean THEN $11 ELSE payload END, + function_id = CASE WHEN $12::boolean THEN $13 ELSE function_id END, updated_at = NOW() WHERE id = $1 AND project_id = $2 diff --git a/apps/backend/internal/database/db/functions.sql.go b/apps/backend/internal/database/db/functions.sql.go index 2d97169..027c43d 100644 --- a/apps/backend/internal/database/db/functions.sql.go +++ b/apps/backend/internal/database/db/functions.sql.go @@ -12,16 +12,17 @@ import ( ) const createFunction = `-- name: CreateFunction :one -INSERT INTO core.functions (project_id, name, code, runtime) -VALUES ($1, $2, $3, $4) -RETURNING id, project_id, name, code, runtime, enabled, created_at, updated_at +INSERT INTO core.functions (project_id, name, code, runtime, event_pattern) +VALUES ($1, $2, $3, $4, $5) +RETURNING id, project_id, name, code, runtime, enabled, created_at, updated_at, event_pattern ` type CreateFunctionParams struct { - ProjectID uuid.UUID `json:"project_id"` - Name string `json:"name"` - Code string `json:"code"` - Runtime string `json:"runtime"` + ProjectID uuid.UUID `json:"project_id"` + Name string `json:"name"` + Code string `json:"code"` + Runtime string `json:"runtime"` + EventPattern string `json:"event_pattern"` } func (q *Queries) CreateFunction(ctx context.Context, arg CreateFunctionParams) (CoreFunction, error) { @@ -30,6 +31,7 @@ func (q *Queries) CreateFunction(ctx context.Context, arg CreateFunctionParams) arg.Name, arg.Code, arg.Runtime, + arg.EventPattern, ) var i CoreFunction err := row.Scan( @@ -41,6 +43,7 @@ func (q *Queries) CreateFunction(ctx context.Context, arg CreateFunctionParams) &i.Enabled, &i.CreatedAt, &i.UpdatedAt, + &i.EventPattern, ) return i, err } @@ -60,7 +63,7 @@ func (q *Queries) DeleteFunction(ctx context.Context, arg DeleteFunctionParams) } const getFunction = `-- name: GetFunction :one -SELECT id, project_id, name, code, runtime, enabled, created_at, updated_at FROM core.functions WHERE id = $1 +SELECT id, project_id, name, code, runtime, enabled, created_at, updated_at, event_pattern FROM core.functions WHERE id = $1 ` func (q *Queries) GetFunction(ctx context.Context, id uuid.UUID) (CoreFunction, error) { @@ -75,12 +78,13 @@ func (q *Queries) GetFunction(ctx context.Context, id uuid.UUID) (CoreFunction, &i.Enabled, &i.CreatedAt, &i.UpdatedAt, + &i.EventPattern, ) return i, err } const getFunctionByName = `-- name: GetFunctionByName :one -SELECT id, project_id, name, code, runtime, enabled, created_at, updated_at FROM core.functions WHERE project_id = $1 AND name = $2 +SELECT id, project_id, name, code, runtime, enabled, created_at, updated_at, event_pattern FROM core.functions WHERE project_id = $1 AND name = $2 ` type GetFunctionByNameParams struct { @@ -100,6 +104,7 @@ func (q *Queries) GetFunctionByName(ctx context.Context, arg GetFunctionByNamePa &i.Enabled, &i.CreatedAt, &i.UpdatedAt, + &i.EventPattern, ) return i, err } @@ -188,7 +193,7 @@ func (q *Queries) ListFunctionRuns(ctx context.Context, arg ListFunctionRunsPara } const listFunctions = `-- name: ListFunctions :many -SELECT id, project_id, name, code, runtime, enabled, created_at, updated_at FROM core.functions +SELECT id, project_id, name, code, runtime, enabled, created_at, updated_at, event_pattern FROM core.functions WHERE project_id = $1 ORDER BY name ` @@ -211,6 +216,50 @@ func (q *Queries) ListFunctions(ctx context.Context, projectID uuid.UUID) ([]Cor &i.Enabled, &i.CreatedAt, &i.UpdatedAt, + &i.EventPattern, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listFunctionsForEvent = `-- name: ListFunctionsForEvent :many +SELECT id, project_id, name, code, runtime, enabled, created_at, updated_at, event_pattern FROM core.functions +WHERE project_id = $1 + AND enabled = TRUE + AND event_pattern <> '' + AND ($2 LIKE REPLACE(event_pattern, '*', '%') OR event_pattern = '*') +` + +type ListFunctionsForEventParams struct { + ProjectID uuid.UUID `json:"project_id"` + EventPattern string `json:"event_pattern"` +} + +func (q *Queries) ListFunctionsForEvent(ctx context.Context, arg ListFunctionsForEventParams) ([]CoreFunction, error) { + rows, err := q.db.Query(ctx, listFunctionsForEvent, arg.ProjectID, arg.EventPattern) + if err != nil { + return nil, err + } + defer rows.Close() + var items []CoreFunction + for rows.Next() { + var i CoreFunction + if err := rows.Scan( + &i.ID, + &i.ProjectID, + &i.Name, + &i.Code, + &i.Runtime, + &i.Enabled, + &i.CreatedAt, + &i.UpdatedAt, + &i.EventPattern, ); err != nil { return nil, err } @@ -226,16 +275,18 @@ const updateFunction = `-- name: UpdateFunction :one UPDATE core.functions SET code = COALESCE($3, code), enabled = COALESCE($4, enabled), + event_pattern = COALESCE($5, event_pattern), updated_at = NOW() WHERE id = $1 AND project_id = $2 -RETURNING id, project_id, name, code, runtime, enabled, created_at, updated_at +RETURNING id, project_id, name, code, runtime, enabled, created_at, updated_at, event_pattern ` type UpdateFunctionParams struct { - ID uuid.UUID `json:"id"` - ProjectID uuid.UUID `json:"project_id"` - Code *string `json:"code"` - Enabled *bool `json:"enabled"` + ID uuid.UUID `json:"id"` + ProjectID uuid.UUID `json:"project_id"` + Code *string `json:"code"` + Enabled *bool `json:"enabled"` + EventPattern *string `json:"event_pattern"` } func (q *Queries) UpdateFunction(ctx context.Context, arg UpdateFunctionParams) (CoreFunction, error) { @@ -244,6 +295,7 @@ func (q *Queries) UpdateFunction(ctx context.Context, arg UpdateFunctionParams) arg.ProjectID, arg.Code, arg.Enabled, + arg.EventPattern, ) var i CoreFunction err := row.Scan( @@ -255,6 +307,7 @@ func (q *Queries) UpdateFunction(ctx context.Context, arg UpdateFunctionParams) &i.Enabled, &i.CreatedAt, &i.UpdatedAt, + &i.EventPattern, ) return i, err } diff --git a/apps/backend/internal/database/db/inbound_hooks.sql.go b/apps/backend/internal/database/db/inbound_hooks.sql.go index ef736df..589bd16 100644 --- a/apps/backend/internal/database/db/inbound_hooks.sql.go +++ b/apps/backend/internal/database/db/inbound_hooks.sql.go @@ -13,19 +13,20 @@ import ( ) const createInboundHook = `-- name: CreateInboundHook :one -INSERT INTO core.inbound_hooks (project_id, name, token, mode, job_id, secret, enabled) -VALUES ($1, $2, $3, $4, $5, $6, $7) -RETURNING id, project_id, name, token, mode, job_id, secret, enabled, last_received_at, created_at, updated_at +INSERT INTO core.inbound_hooks (project_id, name, token, mode, job_id, function_id, secret, enabled) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8) +RETURNING id, project_id, name, token, mode, job_id, secret, enabled, last_received_at, created_at, updated_at, function_id ` type CreateInboundHookParams struct { - ProjectID uuid.UUID `json:"project_id"` - Name string `json:"name"` - Token string `json:"token"` - Mode string `json:"mode"` - JobID pgtype.UUID `json:"job_id"` - Secret []byte `json:"secret"` - Enabled bool `json:"enabled"` + ProjectID uuid.UUID `json:"project_id"` + Name string `json:"name"` + Token string `json:"token"` + Mode string `json:"mode"` + JobID pgtype.UUID `json:"job_id"` + FunctionID pgtype.UUID `json:"function_id"` + Secret []byte `json:"secret"` + Enabled bool `json:"enabled"` } func (q *Queries) CreateInboundHook(ctx context.Context, arg CreateInboundHookParams) (CoreInboundHook, error) { @@ -35,6 +36,7 @@ func (q *Queries) CreateInboundHook(ctx context.Context, arg CreateInboundHookPa arg.Token, arg.Mode, arg.JobID, + arg.FunctionID, arg.Secret, arg.Enabled, ) @@ -51,6 +53,7 @@ func (q *Queries) CreateInboundHook(ctx context.Context, arg CreateInboundHookPa &i.LastReceivedAt, &i.CreatedAt, &i.UpdatedAt, + &i.FunctionID, ) return i, err } @@ -71,7 +74,7 @@ func (q *Queries) DeleteInboundHook(ctx context.Context, arg DeleteInboundHookPa } const getInboundHook = `-- name: GetInboundHook :one -SELECT id, project_id, name, token, mode, job_id, secret, enabled, last_received_at, created_at, updated_at FROM core.inbound_hooks +SELECT id, project_id, name, token, mode, job_id, secret, enabled, last_received_at, created_at, updated_at, function_id FROM core.inbound_hooks WHERE id = $1 AND project_id = $2 ` @@ -95,12 +98,13 @@ func (q *Queries) GetInboundHook(ctx context.Context, arg GetInboundHookParams) &i.LastReceivedAt, &i.CreatedAt, &i.UpdatedAt, + &i.FunctionID, ) return i, err } const getInboundHookByToken = `-- name: GetInboundHookByToken :one -SELECT id, project_id, name, token, mode, job_id, secret, enabled, last_received_at, created_at, updated_at FROM core.inbound_hooks +SELECT id, project_id, name, token, mode, job_id, secret, enabled, last_received_at, created_at, updated_at, function_id FROM core.inbound_hooks WHERE token = $1 ` @@ -119,12 +123,13 @@ func (q *Queries) GetInboundHookByToken(ctx context.Context, token string) (Core &i.LastReceivedAt, &i.CreatedAt, &i.UpdatedAt, + &i.FunctionID, ) return i, err } const listInboundHooks = `-- name: ListInboundHooks :many -SELECT id, project_id, name, token, mode, job_id, secret, enabled, last_received_at, created_at, updated_at FROM core.inbound_hooks +SELECT id, project_id, name, token, mode, job_id, secret, enabled, last_received_at, created_at, updated_at, function_id FROM core.inbound_hooks WHERE project_id = $1 ORDER BY name ` @@ -150,6 +155,7 @@ func (q *Queries) ListInboundHooks(ctx context.Context, projectID uuid.UUID) ([] &i.LastReceivedAt, &i.CreatedAt, &i.UpdatedAt, + &i.FunctionID, ); err != nil { return nil, err } diff --git a/apps/backend/internal/database/db/jobs.sql.go b/apps/backend/internal/database/db/jobs.sql.go index 299c86c..a800923 100644 --- a/apps/backend/internal/database/db/jobs.sql.go +++ b/apps/backend/internal/database/db/jobs.sql.go @@ -22,9 +22,10 @@ INSERT INTO core.scheduled_jobs ( 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 + created_by_user_id, + function_id +) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) +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, function_id ` type CreateScheduledJobParams struct { @@ -37,6 +38,7 @@ type CreateScheduledJobParams struct { Enabled bool `json:"enabled"` NextRunAt pgtype.Timestamptz `json:"next_run_at"` CreatedByUserID pgtype.UUID `json:"created_by_user_id"` + FunctionID pgtype.UUID `json:"function_id"` } func (q *Queries) CreateScheduledJob(ctx context.Context, arg CreateScheduledJobParams) (CoreScheduledJob, error) { @@ -50,6 +52,7 @@ func (q *Queries) CreateScheduledJob(ctx context.Context, arg CreateScheduledJob arg.Enabled, arg.NextRunAt, arg.CreatedByUserID, + arg.FunctionID, ) var i CoreScheduledJob err := row.Scan( @@ -67,6 +70,7 @@ func (q *Queries) CreateScheduledJob(ctx context.Context, arg CreateScheduledJob &i.CreatedByUserID, &i.CreatedAt, &i.UpdatedAt, + &i.FunctionID, ) return i, err } @@ -133,7 +137,7 @@ func (q *Queries) FinishScheduledJobRun(ctx context.Context, arg FinishScheduled } 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 +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, function_id FROM core.scheduled_jobs WHERE id = $1 ` @@ -155,6 +159,7 @@ func (q *Queries) GetScheduledJobByID(ctx context.Context, id uuid.UUID) (CoreSc &i.CreatedByUserID, &i.CreatedAt, &i.UpdatedAt, + &i.FunctionID, ) return i, err } @@ -188,7 +193,7 @@ func (q *Queries) InsertScheduledJobRun(ctx context.Context, arg InsertScheduled } 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 +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, function_id FROM core.scheduled_jobs WHERE enabled AND next_run_at IS NOT NULL AND next_run_at <= $1 @@ -218,6 +223,7 @@ func (q *Queries) ListDueScheduledJobs(ctx context.Context, nextRunAt pgtype.Tim &i.CreatedByUserID, &i.CreatedAt, &i.UpdatedAt, + &i.FunctionID, ); err != nil { return nil, err } @@ -272,7 +278,7 @@ func (q *Queries) ListScheduledJobRuns(ctx context.Context, arg ListScheduledJob } 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 +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, function_id FROM core.scheduled_jobs WHERE project_id = $1 ORDER BY created_at ASC ` @@ -301,6 +307,7 @@ func (q *Queries) ListScheduledJobs(ctx context.Context, projectID uuid.UUID) ([ &i.CreatedByUserID, &i.CreatedAt, &i.UpdatedAt, + &i.FunctionID, ); err != nil { return nil, err } @@ -319,7 +326,7 @@ SET last_run_at = $2, 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 +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, function_id ` type MarkScheduledJobRanParams struct { @@ -352,6 +359,7 @@ func (q *Queries) MarkScheduledJobRan(ctx context.Context, arg MarkScheduledJobR &i.CreatedByUserID, &i.CreatedAt, &i.UpdatedAt, + &i.FunctionID, ) return i, err } @@ -386,24 +394,27 @@ SET name = $3, next_run_at = $7, secret = CASE WHEN $8::boolean THEN $9 ELSE secret END, payload = CASE WHEN $10::boolean THEN $11 ELSE payload END, + function_id = CASE WHEN $12::boolean THEN $13 ELSE function_id 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 +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, function_id ` 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"` + 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"` + Column12 bool `json:"column_12"` + FunctionID pgtype.UUID `json:"function_id"` } func (q *Queries) UpdateScheduledJob(ctx context.Context, arg UpdateScheduledJobParams) (CoreScheduledJob, error) { @@ -419,6 +430,8 @@ func (q *Queries) UpdateScheduledJob(ctx context.Context, arg UpdateScheduledJob arg.Secret, arg.Column10, arg.Payload, + arg.Column12, + arg.FunctionID, ) var i CoreScheduledJob err := row.Scan( @@ -436,6 +449,7 @@ func (q *Queries) UpdateScheduledJob(ctx context.Context, arg UpdateScheduledJob &i.CreatedByUserID, &i.CreatedAt, &i.UpdatedAt, + &i.FunctionID, ) return i, err } diff --git a/apps/backend/internal/database/db/models.go b/apps/backend/internal/database/db/models.go index dce04ea..d353de4 100644 --- a/apps/backend/internal/database/db/models.go +++ b/apps/backend/internal/database/db/models.go @@ -267,14 +267,15 @@ type CoreEvent struct { } type CoreFunction struct { - ID uuid.UUID `json:"id"` - ProjectID uuid.UUID `json:"project_id"` - Name string `json:"name"` - Code string `json:"code"` - Runtime string `json:"runtime"` - Enabled bool `json:"enabled"` - CreatedAt pgtype.Timestamptz `json:"created_at"` - UpdatedAt pgtype.Timestamptz `json:"updated_at"` + ID uuid.UUID `json:"id"` + ProjectID uuid.UUID `json:"project_id"` + Name string `json:"name"` + Code string `json:"code"` + Runtime string `json:"runtime"` + Enabled bool `json:"enabled"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` + EventPattern string `json:"event_pattern"` } type CoreFunctionRun struct { @@ -301,6 +302,7 @@ type CoreInboundHook struct { LastReceivedAt pgtype.Timestamptz `json:"last_received_at"` CreatedAt pgtype.Timestamptz `json:"created_at"` UpdatedAt pgtype.Timestamptz `json:"updated_at"` + FunctionID pgtype.UUID `json:"function_id"` } type CoreIntegration struct { @@ -393,6 +395,7 @@ type CoreScheduledJob struct { CreatedByUserID pgtype.UUID `json:"created_by_user_id"` CreatedAt pgtype.Timestamptz `json:"created_at"` UpdatedAt pgtype.Timestamptz `json:"updated_at"` + FunctionID pgtype.UUID `json:"function_id"` } type CoreScheduledJobRun struct { diff --git a/apps/backend/internal/database/db/querier.go b/apps/backend/internal/database/db/querier.go index e716868..bb4bd95 100644 --- a/apps/backend/internal/database/db/querier.go +++ b/apps/backend/internal/database/db/querier.go @@ -112,6 +112,7 @@ type Querier interface { ListEvents(ctx context.Context, arg ListEventsParams) ([]ListEventsRow, error) ListFunctionRuns(ctx context.Context, arg ListFunctionRunsParams) ([]CoreFunctionRun, error) ListFunctions(ctx context.Context, projectID uuid.UUID) ([]CoreFunction, error) + ListFunctionsForEvent(ctx context.Context, arg ListFunctionsForEventParams) ([]CoreFunction, error) ListInboundHooks(ctx context.Context, projectID uuid.UUID) ([]CoreInboundHook, error) ListIntegrations(ctx context.Context, projectID uuid.UUID) ([]CoreIntegration, error) ListInvitationsForOrganization(ctx context.Context, organizationID uuid.UUID) ([]ListInvitationsForOrganizationRow, error) diff --git a/apps/backend/internal/services/functions.go b/apps/backend/internal/services/functions.go index fa8f5e3..a43e729 100644 --- a/apps/backend/internal/services/functions.go +++ b/apps/backend/internal/services/functions.go @@ -148,23 +148,26 @@ func (l *limitedWriter) Write(p []byte) (int, error) { // --- service layer ------------------------------------------------------- type CreateFunctionInput struct { - Name string `json:"name" validate:"required"` - Code string `json:"code" validate:"required,max=262144"` - Runtime string `json:"runtime" validate:"omitempty,oneof=bun deno"` + Name string `json:"name" validate:"required"` + Code string `json:"code" validate:"required,max=262144"` + Runtime string `json:"runtime" validate:"omitempty,oneof=bun deno"` + EventPattern string `json:"event_pattern" validate:"omitempty,max=128"` } type UpdateFunctionInput struct { - Code *string `json:"code" validate:"omitempty,max=262144"` - Enabled *bool `json:"enabled"` + Code *string `json:"code" validate:"omitempty,max=262144"` + Enabled *bool `json:"enabled"` + EventPattern *string `json:"event_pattern" validate:"omitempty,max=128"` } type FunctionSummary struct { - ID string `json:"id"` - Name string `json:"name"` - Runtime string `json:"runtime"` - Enabled bool `json:"enabled"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` + ID string `json:"id"` + Name string `json:"name"` + Runtime string `json:"runtime"` + Enabled bool `json:"enabled"` + EventPattern string `json:"event_pattern,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` } type FunctionRunSummary struct { @@ -180,12 +183,13 @@ type FunctionRunSummary struct { func functionSummary(r db.CoreFunction) FunctionSummary { return FunctionSummary{ - ID: r.ID.String(), - Name: r.Name, - Runtime: r.Runtime, - Enabled: r.Enabled, - CreatedAt: r.CreatedAt.Time, - UpdatedAt: r.UpdatedAt.Time, + ID: r.ID.String(), + Name: r.Name, + Runtime: r.Runtime, + Enabled: r.Enabled, + EventPattern: r.EventPattern, + CreatedAt: r.CreatedAt.Time, + UpdatedAt: r.UpdatedAt.Time, } } @@ -215,10 +219,11 @@ func (s *PlatformService) CreateFunction(ctx context.Context, actor *models.Acto runtime = "bun" } row, err := s.repo.Queries().CreateFunction(ctx, db.CreateFunctionParams{ - ProjectID: projectID, - Name: name, - Code: input.Code, - Runtime: runtime, + ProjectID: projectID, + Name: name, + Code: input.Code, + Runtime: runtime, + EventPattern: strings.TrimSpace(input.EventPattern), }) if err != nil { return FunctionSummary{}, err @@ -262,11 +267,17 @@ func (s *PlatformService) UpdateFunction(ctx context.Context, actor *models.Acto if err := s.requireProjectRole(ctx, actor, row.ProjectID, "admin", "developer"); err != nil { return FunctionSummary{}, err } + var pattern *string + if input.EventPattern != nil { + p := strings.TrimSpace(*input.EventPattern) + pattern = &p + } updated, err := s.repo.Queries().UpdateFunction(ctx, db.UpdateFunctionParams{ - ID: functionID, - ProjectID: row.ProjectID, - Code: input.Code, - Enabled: input.Enabled, + ID: functionID, + ProjectID: row.ProjectID, + Code: input.Code, + Enabled: input.Enabled, + EventPattern: pattern, }) if err != nil { return FunctionSummary{}, err @@ -298,6 +309,13 @@ func (s *PlatformService) InvokeFunction(ctx context.Context, actor *models.Acto if err := s.requireProjectRole(ctx, actor, row.ProjectID, "admin", "developer"); err != nil { return FunctionRunSummary{}, err } + return s.executeFunction(ctx, row, payload, "manual") +} + +// executeFunction runs a function row and records the outcome — shared by +// manual invoke, schedule/hook delivery, and event triggers (no role check; +// callers hold their own auth context or are system paths). +func (s *PlatformService) executeFunction(ctx context.Context, row db.CoreFunction, payload json.RawMessage, trigger string) (FunctionRunSummary, error) { if !row.Enabled { return FunctionRunSummary{}, fmt.Errorf("function is disabled") } @@ -317,7 +335,7 @@ func (s *PlatformService) InvokeFunction(ctx context.Context, actor *models.Acto } run, err := s.repo.Queries().InsertFunctionRun(ctx, db.InsertFunctionRunParams{ FunctionID: row.ID, - Trigger: "manual", + Trigger: trigger, Status: result.Status, ExitCode: exit, Stdout: result.Stdout, @@ -330,6 +348,79 @@ func (s *PlatformService) InvokeFunction(ctx context.Context, actor *models.Acto return runSummary(run), nil } +// runJobFunction is the scheduler's delivery path for function-targeted +// jobs: load the function, run it with the resolved payload, record the +// run. The job run's status_code mirrors the process exit code. +func (s *PlatformService) runJobFunction(ctx context.Context, job db.CoreScheduledJob, run db.CoreScheduledJobRun, payload json.RawMessage) (*int32, error) { + fn, err := s.repo.Queries().GetFunction(ctx, uuid.UUID(job.FunctionID.Bytes)) + if err != nil { + return nil, fmt.Errorf("target function missing: %w", err) + } + summary, err := s.executeFunction(ctx, fn, payload, triggerFromJobRun(run.TriggeredBy)) + if err != nil { + return nil, err + } + code := int32(0) + if summary.ExitCode != nil { + code = *summary.ExitCode + } + if summary.Status != "success" { + return &code, fmt.Errorf("function %q %s: %s", fn.Name, summary.Status, truncate(summary.Stderr, 200)) + } + return &code, nil +} + +func truncate(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n] + "…" +} + +func triggerFromJobRun(triggeredBy string) string { + switch triggeredBy { + case "schedule", "hook", "manual": + return triggeredBy + default: + return "schedule" + } +} + +// triggerFunctionsForEvent fans a domain event out to every enabled +// function whose event_pattern matches ("document.*", "*", …). Runs are +// detached — event publishers never block on user code. +// jarvis: ceiling pattern is a single * suffix/prefix wildcard via SQL +// LIKE; upgrade to a real matcher if multi-segment globs are needed. +func (s *PlatformService) triggerFunctionsForEvent(ctx context.Context, projectID uuid.UUID, eventType string, data map[string]any) { + if s.functions == nil { + return + } + rows, err := s.repo.Queries().ListFunctionsForEvent(ctx, db.ListFunctionsForEventParams{ + ProjectID: projectID, + EventPattern: eventType, + }) + if err != nil || len(rows) == 0 { + return + } + payload, _ := json.Marshal(map[string]any{ + "event": eventType, + "project_id": projectID.String(), + "occurred_at": time.Now().UTC().Format(time.RFC3339), + "data": data, + }) + for _, fn := range rows { + fn := fn + go func() { + // Detached context: the publishing request may finish first. + runCtx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + if _, err := s.executeFunction(runCtx, fn, payload, "event"); err != nil { + s.logger.Warn("function event trigger failed", "function", fn.Name, "event", eventType, "error", err) + } + }() + } +} + func (s *PlatformService) ListFunctionRuns(ctx context.Context, actor *models.Actor, functionID uuid.UUID, limit int64) ([]FunctionRunSummary, error) { row, err := s.repo.Queries().GetFunction(ctx, functionID) if err != nil { diff --git a/apps/backend/internal/services/functions_test.go b/apps/backend/internal/services/functions_test.go index bd7bb03..256c453 100644 --- a/apps/backend/internal/services/functions_test.go +++ b/apps/backend/internal/services/functions_test.go @@ -70,3 +70,17 @@ func TestLimitedWriterCaps(t *testing.T) { t.Fatalf("got %q", buf.String()) } } + +func TestTriggerFromJobRun(t *testing.T) { + for in, want := range map[string]string{ + "schedule": "schedule", + "hook": "hook", + "manual": "manual", + "cli": "schedule", + "": "schedule", + } { + if got := triggerFromJobRun(in); got != want { + t.Fatalf("triggerFromJobRun(%q) = %q, want %q", in, got, want) + } + } +} diff --git a/apps/backend/internal/services/inbound.go b/apps/backend/internal/services/inbound.go index a7921a5..1866d5a 100644 --- a/apps/backend/internal/services/inbound.go +++ b/apps/backend/internal/services/inbound.go @@ -28,8 +28,9 @@ import ( const ( WebhookEventInboundReceived = "inbound.received" - InboundModeEvent = "event" - InboundModeJob = "job" + InboundModeEvent = "event" + InboundModeJob = "job" + InboundModeFunction = "function" ) type InboundHookSummary struct { @@ -40,6 +41,7 @@ type InboundHookSummary struct { URL string `json:"url"` Mode string `json:"mode"` JobID string `json:"job_id,omitempty"` + FunctionID string `json:"function_id,omitempty"` HasSecret bool `json:"has_secret"` Enabled bool `json:"enabled"` LastReceivedAt *time.Time `json:"last_received_at"` @@ -47,11 +49,12 @@ type InboundHookSummary struct { } type CreateInboundHookInput struct { - Name string `json:"name" validate:"required"` - Mode string `json:"mode"` - JobID string `json:"job_id"` - Secret string `json:"secret"` - Enabled *bool `json:"enabled"` + Name string `json:"name" validate:"required"` + Mode string `json:"mode"` + JobID string `json:"job_id"` + FunctionID string `json:"function_id"` + Secret string `json:"secret"` + Enabled *bool `json:"enabled"` } type InboundReceiveResult struct { @@ -75,6 +78,9 @@ func inboundHookSummary(row db.CoreInboundHook, publicURL string) InboundHookSum if row.JobID.Valid { out.JobID = uuid.UUID(row.JobID.Bytes).String() } + if row.FunctionID.Valid { + out.FunctionID = uuid.UUID(row.FunctionID.Bytes).String() + } if row.LastReceivedAt.Valid { t := row.LastReceivedAt.Time out.LastReceivedAt = &t @@ -108,8 +114,9 @@ func (s *PlatformService) CreateInboundHook(ctx context.Context, actor *models.A if mode == "" { mode = InboundModeEvent } - var jobID pgtype.UUID - if mode == InboundModeJob { + var jobID, functionID pgtype.UUID + switch mode { + case InboundModeJob: parsed, err := uuid.Parse(input.JobID) if err != nil { return InboundHookSummary{}, errors.New("job mode requires a valid job_id") @@ -119,8 +126,19 @@ func (s *PlatformService) CreateInboundHook(ctx context.Context, actor *models.A return InboundHookSummary{}, errors.New("job not found in this project") } jobID = pgtype.UUID{Bytes: parsed, Valid: true} - } else if mode != InboundModeEvent { - return InboundHookSummary{}, fmt.Errorf("invalid mode %q — event or job", input.Mode) + case InboundModeFunction: + parsed, err := uuid.Parse(input.FunctionID) + if err != nil { + return InboundHookSummary{}, errors.New("function mode requires a valid function_id") + } + fn, err := s.repo.Queries().GetFunction(ctx, parsed) + if err != nil || fn.ProjectID != projectID { + return InboundHookSummary{}, errors.New("function not found in this project") + } + functionID = pgtype.UUID{Bytes: parsed, Valid: true} + case InboundModeEvent: + default: + return InboundHookSummary{}, fmt.Errorf("invalid mode %q — event, job, or function", input.Mode) } secretBytes := []byte{} if input.Secret != "" { @@ -141,10 +159,11 @@ func (s *PlatformService) CreateInboundHook(ctx context.Context, actor *models.A ProjectID: projectID, Name: input.Name, Token: token, - Mode: mode, - JobID: jobID, - Secret: secretBytes, - Enabled: enabled, + Mode: mode, + JobID: jobID, + FunctionID: functionID, + Secret: secretBytes, + Enabled: enabled, }) if err != nil { if strings.Contains(err.Error(), "inbound_hooks_project_id_name_key") { @@ -230,6 +249,31 @@ func (s *PlatformService) ReceiveInboundHook(ctx context.Context, token string, return InboundReceiveResult{}, err } return InboundReceiveResult{Received: true, Event: WebhookEventJobRun, RunID: run.ID.String()}, nil + case InboundModeFunction: + if s.functions == nil { + return InboundReceiveResult{}, errors.New("functions runtime not configured") + } + if !hook.FunctionID.Valid { + return InboundReceiveResult{}, errors.New("hook has no function configured") + } + fn, err := s.repo.Queries().GetFunction(ctx, uuid.UUID(hook.FunctionID.Bytes)) + if err != nil { + return InboundReceiveResult{}, err + } + raw := body + if len(raw) == 0 { + raw = []byte(`{}`) + } + // Detached: the public endpoint acks immediately; the run row records + // the outcome. Hook callers must not wait on user code. + go func() { + runCtx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + if _, err := s.executeFunction(runCtx, fn, raw, "hook"); err != nil && s.logger != nil { + s.logger.Warn("hook function invoke failed", "hook", hook.Name, "function", fn.Name, "error", err) + } + }() + return InboundReceiveResult{Received: true, Event: "function.invoked"}, nil default: s.publishEvent(ctx, hook.ProjectID, WebhookEventInboundReceived, map[string]any{ "hook_id": hook.ID.String(), diff --git a/apps/backend/internal/services/jobs.go b/apps/backend/internal/services/jobs.go index 0a55ae8..82e4e6c 100644 --- a/apps/backend/internal/services/jobs.go +++ b/apps/backend/internal/services/jobs.go @@ -35,6 +35,7 @@ type ScheduledJobSummary struct { Name string `json:"name"` Schedule string `json:"schedule"` URL string `json:"url"` + FunctionID *string `json:"function_id,omitempty"` Payload json.RawMessage `json:"payload"` Enabled bool `json:"enabled"` HasSecret bool `json:"has_secret"` @@ -59,21 +60,23 @@ type ScheduledJobRunSummary struct { } 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"` + Name string `json:"name" validate:"required,min=2"` + Schedule string `json:"schedule" validate:"required"` + URL string `json:"url" validate:"required_without=FunctionID,omitempty,url"` + FunctionID string `json:"function_id" validate:"omitempty,uuid"` + 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"` + Name *string `json:"name" validate:"omitempty,min=2"` + Schedule *string `json:"schedule"` + URL *string `json:"url" validate:"omitempty,url"` + FunctionID *string `json:"function_id" validate:"omitempty,uuid"` + Secret *string `json:"secret"` + Payload map[string]any `json:"payload"` + Enabled *bool `json:"enabled"` } func toJobSummary(row db.CoreScheduledJob) ScheduledJobSummary { @@ -88,6 +91,10 @@ func toJobSummary(row db.CoreScheduledJob) ScheduledJobSummary { HasSecret: len(row.Secret) > 0, CreatedAt: row.CreatedAt.Time, } + if row.FunctionID.Valid { + id := uuid.UUID(row.FunctionID.Bytes).String() + out.FunctionID = &id + } if row.LastRunAt.Valid { t := row.LastRunAt.Time out.LastRunAt = &t @@ -176,9 +183,25 @@ func (s *PlatformService) CreateScheduledJob(ctx context.Context, actor *models. 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 + jobURL := "" + var functionID pgtype.UUID + if input.FunctionID != "" { + fid, err := uuid.Parse(input.FunctionID) + if err != nil { + return ScheduledJobSummary{}, fmt.Errorf("invalid function_id") + } + fn, err := s.repo.Queries().GetFunction(ctx, fid) + if err != nil || fn.ProjectID != projectID { + return ScheduledJobSummary{}, fmt.Errorf("function not found in this project") + } + functionID = pgtype.UUID{Bytes: fid, Valid: true} + jobURL = "function://" + fid.String() + } else { + var err error + jobURL, err = normalizeWebhookURL(input.URL) + if err != nil { + return ScheduledJobSummary{}, err + } } sched, err := parseSchedule(input.Schedule) if err != nil { @@ -222,6 +245,7 @@ func (s *PlatformService) CreateScheduledJob(ctx context.Context, actor *models. Enabled: enabled, NextRunAt: next, CreatedByUserID: createdBy, + FunctionID: functionID, }) if err != nil { if strings.Contains(err.Error(), "scheduled_jobs_project_id_name_key") { @@ -260,6 +284,28 @@ func (s *PlatformService) UpdateScheduledJob(ctx context.Context, actor *models. return ScheduledJobSummary{}, err } } + rotateFunction := input.FunctionID != nil + functionID := row.FunctionID + if rotateFunction { + raw := strings.TrimSpace(*input.FunctionID) + if raw == "" { + functionID = pgtype.UUID{} // explicit clear → back to URL delivery + if input.URL == nil && strings.HasPrefix(jobURL, "function://") { + return ScheduledJobSummary{}, fmt.Errorf("clearing function_id requires a url") + } + } else { + fid, ferr := uuid.Parse(raw) + if ferr != nil { + return ScheduledJobSummary{}, fmt.Errorf("invalid function_id") + } + fn, ferr := s.repo.Queries().GetFunction(ctx, fid) + if ferr != nil || fn.ProjectID != projectID { + return ScheduledJobSummary{}, fmt.Errorf("function not found in this project") + } + functionID = pgtype.UUID{Bytes: fid, Valid: true} + jobURL = "function://" + fid.String() + } + } enabled := row.Enabled if input.Enabled != nil { enabled = *input.Enabled @@ -302,6 +348,8 @@ func (s *PlatformService) UpdateScheduledJob(ctx context.Context, actor *models. Secret: secretBytes, Column10: rotatePayload, Payload: payload, + Column12: rotateFunction, + FunctionID: functionID, }) if err != nil { if strings.Contains(err.Error(), "scheduled_jobs_project_id_name_key") { @@ -410,6 +458,10 @@ type JobScheduler struct { leader *leaderState // 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) + // runFn delivers to a function instead of HTTP when job.FunctionID is set. + // The function receives the resolved job payload, not the run envelope. + // Owned by PlatformService — the scheduler stays transport-agnostic. + runFn func(ctx context.Context, job db.CoreScheduledJob, run db.CoreScheduledJobRun, payload json.RawMessage) (*int32, error) } type jobExec struct { @@ -559,7 +611,8 @@ func (j *JobScheduler) markRan(ctx context.Context, job db.CoreScheduledJob, sta } } -// deliver POSTs the job payload to its URL, signed like a webhook delivery. +// deliver POSTs the job payload to its URL — or invokes the target function +// when the job carries a function_id. func (j *JobScheduler) deliver(ctx context.Context, job db.CoreScheduledJob, run db.CoreScheduledJobRun) (*int32, error) { data, err := resolveSecretRefs(ctx, j.repo.ListProjectSecretValues, j.enc, job.ProjectID, job.Payload) if err != nil { @@ -575,6 +628,12 @@ func (j *JobScheduler) deliver(ctx context.Context, job db.CoreScheduledJob, run "occurred_at": time.Now().UTC().Format(time.RFC3339), "data": data, }) + if job.FunctionID.Valid { + if j.runFn == nil { + return nil, fmt.Errorf("functions runtime not configured") + } + return j.runFn(ctx, job, run, data) + } secret := "" if len(job.Secret) > 0 && j.enc != nil { if plain, err := j.enc.Decrypt(job.Secret); err == nil { diff --git a/apps/backend/internal/services/platform_service.go b/apps/backend/internal/services/platform_service.go index 61597fb..03a7101 100644 --- a/apps/backend/internal/services/platform_service.go +++ b/apps/backend/internal/services/platform_service.go @@ -40,6 +40,7 @@ type PlatformService struct { scheduler *JobScheduler alerts *alertEvaluator functions FunctionRunner + logger *slog.Logger } type BootstrapInput struct { @@ -226,11 +227,12 @@ type InvitationSummary struct { } func NewPlatformService(repo *repositories.CoreRepository, store storage.Store, mailer *Mailer, publicURL string, dbxClient *dbx.Client, enc *secrets.Encryptor, logger *slog.Logger, functions FunctionRunner) *PlatformService { - s := &PlatformService{repo: repo, store: store, mailer: mailer, publicURL: publicURL, dbx: dbxClient, hub: NewEventHub(), realtime: NewEventHub(), enc: enc, functions: functions} + s := &PlatformService{repo: repo, store: store, mailer: mailer, publicURL: publicURL, dbx: dbxClient, hub: NewEventHub(), realtime: NewEventHub(), enc: enc, functions: functions, logger: logger} if enc != nil { s.dispatcher = NewWebhookDispatcher(repo.Queries(), enc, logger) s.dispatcher.Start(context.Background()) s.scheduler = NewJobScheduler(repo.Queries(), enc, logger, s.publishEvent, repo.Pool()) + s.scheduler.runFn = s.runJobFunction s.scheduler.Start(context.Background()) s.alerts = newAlertEvaluator(repo.Queries(), s.publishEvent, logger, repo.Pool()) s.alerts.Start(context.Background()) @@ -327,6 +329,7 @@ func (s *PlatformService) publishEvent(ctx context.Context, projectID uuid.UUID, }) } s.dispatchEvent(ctx, projectID, eventType, data) + s.triggerFunctionsForEvent(ctx, projectID, eventType, data) } // SubscribeRealtime registers an SSE consumer for project domain events. diff --git a/apps/backend/openapi/openapi.yaml b/apps/backend/openapi/openapi.yaml index ca9d54b..a1c8d7d 100644 --- a/apps/backend/openapi/openapi.yaml +++ b/apps/backend/openapi/openapi.yaml @@ -4058,6 +4058,11 @@ components: description: Cron expression (5-field) or descriptor (@hourly, @every 30m). url: type: string + function_id: + type: string + format: uuid + nullable: true + description: When set, the schedule invokes this function instead of POSTing url. payload: type: object description: Static JSON payload merged into every delivery's data field. @@ -4093,6 +4098,11 @@ components: url: type: string format: uri + description: Delivery target. Required unless function_id is set. + function_id: + type: string + format: uuid + description: Invoke this function on schedule instead of POSTing url. secret: type: string description: HMAC key for X-Primora-Signature. Generated when omitted. @@ -4102,7 +4112,7 @@ components: enabled: type: boolean default: true - required: [name, schedule, url] + required: [name, schedule] UpdateScheduledJobRequest: type: object properties: @@ -4114,6 +4124,11 @@ components: url: type: string format: uri + function_id: + type: string + format: uuid + nullable: true + description: Set to retarget at a function; send "" to clear back to url delivery. secret: type: string description: Replace the signing secret. Omit to keep the current one. @@ -4309,9 +4324,11 @@ components: description: Full ingest URL — POST here from the external system. mode: type: string - enum: [event, job] + enum: [event, job, function] job_id: type: string + function_id: + type: string has_secret: type: boolean enabled: @@ -4339,11 +4356,14 @@ components: type: string mode: type: string - enum: [event, job] + enum: [event, job, function] default: event job_id: type: string description: Required when mode=job — scheduled job to trigger. + function_id: + type: string + description: Required when mode=function — invoked with the received body. secret: type: string description: Optional — stored encrypted; callers must then sign bodies with X-Primora-Signature. @@ -4402,6 +4422,9 @@ components: enum: [bun, deno] enabled: type: boolean + event_pattern: + type: string + description: 'Domain-event pattern that triggers this function — "document.*", "*", exact type. Empty = never.' created_at: type: string format: date-time @@ -4436,6 +4459,9 @@ components: type: string enum: [bun, deno] default: bun + event_pattern: + type: string + description: 'Optional — "document.*", "*", or an exact event type.' required: [name, code] UpdateFunctionRequest: type: object @@ -4444,6 +4470,9 @@ components: type: string enabled: type: boolean + event_pattern: + type: string + description: Replace the trigger pattern; "" disables event triggers. InvokeFunctionRequest: type: object properties: @@ -4456,7 +4485,7 @@ components: format: uuid trigger: type: string - enum: [manual, schedule, hook] + enum: [manual, schedule, hook, event] status: type: string enum: [success, error, timeout] diff --git a/apps/frontend/src/pages/AutomationPage.tsx b/apps/frontend/src/pages/AutomationPage.tsx index 781f7fc..94a9764 100644 --- a/apps/frontend/src/pages/AutomationPage.tsx +++ b/apps/frontend/src/pages/AutomationPage.tsx @@ -1,9 +1,11 @@ import { For, Show, createEffect, createSignal, onCleanup } from "solid-js"; import { AutomationService, + FunctionsService, OpenAPI, type ScheduledJob, type ScheduledJobRun, + type Function as PrimoraFunction, } from "@primora/api-client"; import { fetchApiToken } from "../lib/auth-client"; import { demoService } from "../lib/demo-mode"; @@ -63,7 +65,8 @@ export function AutomationPage(props: AutomationPageProps) { const [runs, setRuns] = createSignal>({}); const [expandedJob, setExpandedJob] = createSignal(); const [showCreate, setShowCreate] = createSignal(false); - const [jobForm, setJobForm] = createSignal({ name: "", schedule: "", url: "", payload: "", secret: "", enabled: true }); + const [jobForm, setJobForm] = createSignal({ name: "", schedule: "", url: "", payload: "", secret: "", enabled: true, target: "url" as "url" | "function", function_id: "" }); + const [functions, setFunctions] = createSignal([]); const [createdSecret, setCreatedSecret] = createSignal(""); const [loading, setLoading] = createSignal(false); const [busy, setBusy] = createSignal(); @@ -82,6 +85,10 @@ export function AutomationPage(props: AutomationPageProps) { try { const res = await svc(props.demoMode).listScheduledJobs({ projectId: props.projectID }); setJobs(res.items ?? []); + if (!props.demoMode) { + const fr = await FunctionsService.listFunctions({ projectId: props.projectID }); + setFunctions(fr.items ?? []); + } } catch (e) { setError(err(e)); } finally { @@ -199,14 +206,15 @@ export function AutomationPage(props: AutomationPageProps) { requestBody: { name: form.name.trim(), schedule: form.schedule.trim(), - url: form.url.trim(), + url: form.target === "url" ? form.url.trim() : undefined, + function_id: form.target === "function" ? form.function_id : undefined, secret: form.secret.trim() || undefined, payload, enabled: form.enabled, }, }); if (created.secret) setCreatedSecret(created.secret); - setJobForm({ name: "", schedule: "", url: "", payload: "", secret: "", enabled: true }); + setJobForm({ name: "", schedule: "", url: "", payload: "", secret: "", enabled: true, target: "url", function_id: "" }); setShowCreate(false); }, "Schedule created"); }; @@ -335,7 +343,9 @@ export function AutomationPage(props: AutomationPageProps) { {job.schedule} - {job.url} + {job.function_id + ? `ƒ ${functions().find((f) => f.id === job.function_id)?.name ?? "function"}` + : job.url} {job.enabled ? fmtTime(job.next_run_at) : "—"} @@ -545,16 +555,56 @@ export function AutomationPage(props: AutomationPageProps) {

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

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

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

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

+ + } + > + +

+ The function receives the resolved payload as JSON. Runs appear under the function's + history. +

+