feat: Function triggers — schedule, inbound hook, domain events

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>
This commit is contained in:
Tomas DvorakandDevin committed 2026-09-24 20:28:04 +02:00
1 parent c094afad65
commit a84049b8fc
40 files changed
+716 -157

No files matched your search

@@ -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
}
@@ -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
}
+33 -19
View File
@@ -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
}
+11 -8
View File
@@ -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 {
@@ -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)
+117 -26
View File
@@ -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 {
@@ -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)
}
}
}
+59 -15
View File
@@ -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(),
+75 -16
View File
@@ -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 {
@@ -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.