feat: Functions — JS/TS execution with run log

core.functions + core.function_runs: project-scoped code executed by a
runtime binary on the backend host (bun or deno per function). Payload
arrives on stdin + PRIMORA_PAYLOAD env; stdout/stderr captured per run
(capped), with status/exit code/duration. Invoke returns 502 + the
recorded run on failure so errors stay observable.

Dashboard gains a Functions page: code editor, JSON payload invoke,
live result, run history. Runner config via FUNCTIONS_RUNTIME /
FUNCTIONS_TIMEOUT_SECONDS / FUNCTIONS_MAX_OUTPUT_BYTES.

Verified live: bun ran code, stdin payload delivered, error path returns
502 with captured stderr; timeout kills at deadline.

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 19:46:32 +02:00
1 parent 3aca5bd1e0
commit 32587c8063
37 files changed
+2157 -6

No files matched your search

@@ -0,0 +1,40 @@
-- +goose Up
CREATE TABLE core.functions (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
project_id UUID NOT NULL REFERENCES core.projects(id) ON DELETE CASCADE,
name TEXT NOT NULL,
code TEXT NOT NULL,
runtime TEXT NOT NULL DEFAULT 'bun' CHECK (runtime IN ('bun','deno')),
enabled BOOLEAN NOT NULL DEFAULT TRUE,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
UNIQUE (project_id, name)
);
CREATE TABLE core.function_runs (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
function_id UUID NOT NULL REFERENCES core.functions(id) ON DELETE CASCADE,
trigger TEXT NOT NULL CHECK (trigger IN ('manual','schedule','hook')),
status TEXT NOT NULL CHECK (status IN ('success','error','timeout')),
exit_code INT,
stdout TEXT NOT NULL DEFAULT '',
stderr TEXT NOT NULL DEFAULT '',
duration_ms INT NOT NULL DEFAULT 0,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX function_runs_function_idx ON core.function_runs(function_id, created_at DESC);
ALTER TABLE core.scheduled_job_runs
DROP CONSTRAINT scheduled_job_runs_triggered_by_check;
ALTER TABLE core.scheduled_job_runs
ADD CONSTRAINT scheduled_job_runs_triggered_by_check
CHECK (triggered_by IN ('schedule','manual','hook','function'));
-- +goose Down
ALTER TABLE core.scheduled_job_runs
DROP CONSTRAINT scheduled_job_runs_triggered_by_check;
ALTER TABLE core.scheduled_job_runs
ADD CONSTRAINT scheduled_job_runs_triggered_by_check
CHECK (triggered_by IN ('schedule','manual','hook'));
DROP TABLE IF EXISTS core.function_runs;
DROP TABLE IF EXISTS core.functions;
+37
View File
@@ -0,0 +1,37 @@
-- name: CreateFunction :one
INSERT INTO core.functions (project_id, name, code, runtime)
VALUES ($1, $2, $3, $4)
RETURNING *;
-- name: GetFunction :one
SELECT * FROM core.functions WHERE id = $1;
-- name: GetFunctionByName :one
SELECT * FROM core.functions WHERE project_id = $1 AND name = $2;
-- name: ListFunctions :many
SELECT * FROM core.functions
WHERE project_id = $1
ORDER BY name;
-- name: UpdateFunction :one
UPDATE core.functions
SET code = COALESCE(sqlc.narg('code'), code),
enabled = COALESCE(sqlc.narg('enabled'), enabled),
updated_at = NOW()
WHERE id = $1 AND project_id = $2
RETURNING *;
-- name: DeleteFunction :exec
DELETE FROM core.functions WHERE id = $1 AND project_id = $2;
-- name: InsertFunctionRun :one
INSERT INTO core.function_runs (function_id, trigger, status, exit_code, stdout, stderr, duration_ms)
VALUES ($1, $2, $3, $4, $5, $6, $7)
RETURNING *;
-- name: ListFunctionRuns :many
SELECT * FROM core.function_runs
WHERE function_id = $1
ORDER BY created_at DESC
LIMIT $2;
+1 -1
View File
@@ -112,7 +112,7 @@ func Bootstrap(ctx context.Context) (*App, error) {
if err != nil {
return nil, fmt.Errorf("encryption key: %w", err)
}
platform := services.NewPlatformService(repo, store, services.NewMailer(cfg), os.Getenv("VITE_APP_URL"), dbxClient, encryptor, logger)
platform := services.NewPlatformService(repo, store, services.NewMailer(cfg), os.Getenv("VITE_APP_URL"), dbxClient, encryptor, logger, services.NewExecRunner(cfg.FunctionsRuntime, cfg.FunctionsTimeout, cfg.FunctionsMaxBytes))
settings := services.NewSettingsService(repo, encryptor)
if cfg.Env == "production" {
+15
View File
@@ -7,6 +7,7 @@ import (
"path/filepath"
"strconv"
"strings"
"time"
)
type Config struct {
@@ -45,6 +46,12 @@ type Config struct {
S3SecretAccessKey string
S3Prefix string
S3PathStyle bool
// Functions runtime: binary name/path ("auto" resolves per-function
// runtime from PATH — "bun" or "deno"), timeout and output cap.
FunctionsRuntime string
FunctionsTimeout time.Duration
FunctionsMaxBytes int64
}
// devEncryptionKey is the built-in credential key for local development only.
@@ -80,6 +87,8 @@ func Load() (Config, error) {
S3AccessKeyID: os.Getenv("S3_ACCESS_KEY_ID"),
S3SecretAccessKey: os.Getenv("S3_SECRET_ACCESS_KEY"),
S3Prefix: os.Getenv("S3_PREFIX"),
FunctionsRuntime: getenv("FUNCTIONS_RUNTIME", "auto"),
FunctionsMaxBytes: 65536,
}
smtpPort, err := strconv.Atoi(getenv("SMTP_PORT", "1025"))
@@ -118,6 +127,12 @@ func Load() (Config, error) {
}
cfg.S3PathStyle = s3PathStyle
fnTimeout, err := strconv.Atoi(getenv("FUNCTIONS_TIMEOUT_SECONDS", "30"))
if err != nil {
return Config{}, fmt.Errorf("parse FUNCTIONS_TIMEOUT_SECONDS: %w", err)
}
cfg.FunctionsTimeout = time.Duration(fnTimeout) * time.Second
var missing []string
if cfg.DatabaseURL == "" {
missing = append(missing, "DATABASE_URL")
@@ -0,0 +1,260 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.29.0
// source: functions.sql
package db
import (
"context"
"github.com/google/uuid"
)
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
`
type CreateFunctionParams struct {
ProjectID uuid.UUID `json:"project_id"`
Name string `json:"name"`
Code string `json:"code"`
Runtime string `json:"runtime"`
}
func (q *Queries) CreateFunction(ctx context.Context, arg CreateFunctionParams) (CoreFunction, error) {
row := q.db.QueryRow(ctx, createFunction,
arg.ProjectID,
arg.Name,
arg.Code,
arg.Runtime,
)
var i CoreFunction
err := row.Scan(
&i.ID,
&i.ProjectID,
&i.Name,
&i.Code,
&i.Runtime,
&i.Enabled,
&i.CreatedAt,
&i.UpdatedAt,
)
return i, err
}
const deleteFunction = `-- name: DeleteFunction :exec
DELETE FROM core.functions WHERE id = $1 AND project_id = $2
`
type DeleteFunctionParams struct {
ID uuid.UUID `json:"id"`
ProjectID uuid.UUID `json:"project_id"`
}
func (q *Queries) DeleteFunction(ctx context.Context, arg DeleteFunctionParams) error {
_, err := q.db.Exec(ctx, deleteFunction, arg.ID, arg.ProjectID)
return err
}
const getFunction = `-- name: GetFunction :one
SELECT id, project_id, name, code, runtime, enabled, created_at, updated_at FROM core.functions WHERE id = $1
`
func (q *Queries) GetFunction(ctx context.Context, id uuid.UUID) (CoreFunction, error) {
row := q.db.QueryRow(ctx, getFunction, id)
var i CoreFunction
err := row.Scan(
&i.ID,
&i.ProjectID,
&i.Name,
&i.Code,
&i.Runtime,
&i.Enabled,
&i.CreatedAt,
&i.UpdatedAt,
)
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
`
type GetFunctionByNameParams struct {
ProjectID uuid.UUID `json:"project_id"`
Name string `json:"name"`
}
func (q *Queries) GetFunctionByName(ctx context.Context, arg GetFunctionByNameParams) (CoreFunction, error) {
row := q.db.QueryRow(ctx, getFunctionByName, arg.ProjectID, arg.Name)
var i CoreFunction
err := row.Scan(
&i.ID,
&i.ProjectID,
&i.Name,
&i.Code,
&i.Runtime,
&i.Enabled,
&i.CreatedAt,
&i.UpdatedAt,
)
return i, err
}
const insertFunctionRun = `-- name: InsertFunctionRun :one
INSERT INTO core.function_runs (function_id, trigger, status, exit_code, stdout, stderr, duration_ms)
VALUES ($1, $2, $3, $4, $5, $6, $7)
RETURNING id, function_id, trigger, status, exit_code, stdout, stderr, duration_ms, created_at
`
type InsertFunctionRunParams struct {
FunctionID uuid.UUID `json:"function_id"`
Trigger string `json:"trigger"`
Status string `json:"status"`
ExitCode *int32 `json:"exit_code"`
Stdout string `json:"stdout"`
Stderr string `json:"stderr"`
DurationMs int32 `json:"duration_ms"`
}
func (q *Queries) InsertFunctionRun(ctx context.Context, arg InsertFunctionRunParams) (CoreFunctionRun, error) {
row := q.db.QueryRow(ctx, insertFunctionRun,
arg.FunctionID,
arg.Trigger,
arg.Status,
arg.ExitCode,
arg.Stdout,
arg.Stderr,
arg.DurationMs,
)
var i CoreFunctionRun
err := row.Scan(
&i.ID,
&i.FunctionID,
&i.Trigger,
&i.Status,
&i.ExitCode,
&i.Stdout,
&i.Stderr,
&i.DurationMs,
&i.CreatedAt,
)
return i, err
}
const listFunctionRuns = `-- name: ListFunctionRuns :many
SELECT id, function_id, trigger, status, exit_code, stdout, stderr, duration_ms, created_at FROM core.function_runs
WHERE function_id = $1
ORDER BY created_at DESC
LIMIT $2
`
type ListFunctionRunsParams struct {
FunctionID uuid.UUID `json:"function_id"`
Limit int32 `json:"limit"`
}
func (q *Queries) ListFunctionRuns(ctx context.Context, arg ListFunctionRunsParams) ([]CoreFunctionRun, error) {
rows, err := q.db.Query(ctx, listFunctionRuns, arg.FunctionID, arg.Limit)
if err != nil {
return nil, err
}
defer rows.Close()
var items []CoreFunctionRun
for rows.Next() {
var i CoreFunctionRun
if err := rows.Scan(
&i.ID,
&i.FunctionID,
&i.Trigger,
&i.Status,
&i.ExitCode,
&i.Stdout,
&i.Stderr,
&i.DurationMs,
&i.CreatedAt,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const listFunctions = `-- name: ListFunctions :many
SELECT id, project_id, name, code, runtime, enabled, created_at, updated_at FROM core.functions
WHERE project_id = $1
ORDER BY name
`
func (q *Queries) ListFunctions(ctx context.Context, projectID uuid.UUID) ([]CoreFunction, error) {
rows, err := q.db.Query(ctx, listFunctions, projectID)
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,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const updateFunction = `-- name: UpdateFunction :one
UPDATE core.functions
SET code = COALESCE($3, code),
enabled = COALESCE($4, enabled),
updated_at = NOW()
WHERE id = $1 AND project_id = $2
RETURNING id, project_id, name, code, runtime, enabled, created_at, updated_at
`
type UpdateFunctionParams struct {
ID uuid.UUID `json:"id"`
ProjectID uuid.UUID `json:"project_id"`
Code *string `json:"code"`
Enabled *bool `json:"enabled"`
}
func (q *Queries) UpdateFunction(ctx context.Context, arg UpdateFunctionParams) (CoreFunction, error) {
row := q.db.QueryRow(ctx, updateFunction,
arg.ID,
arg.ProjectID,
arg.Code,
arg.Enabled,
)
var i CoreFunction
err := row.Scan(
&i.ID,
&i.ProjectID,
&i.Name,
&i.Code,
&i.Runtime,
&i.Enabled,
&i.CreatedAt,
&i.UpdatedAt,
)
return i, err
}
@@ -266,6 +266,29 @@ type CoreEvent struct {
Ts pgtype.Timestamptz `json:"ts"`
}
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"`
}
type CoreFunctionRun struct {
ID uuid.UUID `json:"id"`
FunctionID uuid.UUID `json:"function_id"`
Trigger string `json:"trigger"`
Status string `json:"status"`
ExitCode *int32 `json:"exit_code"`
Stdout string `json:"stdout"`
Stderr string `json:"stderr"`
DurationMs int32 `json:"duration_ms"`
CreatedAt pgtype.Timestamptz `json:"created_at"`
}
type CoreInboundHook struct {
ID uuid.UUID `json:"id"`
ProjectID uuid.UUID `json:"project_id"`
@@ -30,6 +30,7 @@ type Querier interface {
CreateCollection(ctx context.Context, arg CreateCollectionParams) (CoreCollection, error)
CreateDBConnection(ctx context.Context, arg CreateDBConnectionParams) (CoreDbConnection, error)
CreateDocument(ctx context.Context, arg CreateDocumentParams) (CoreDocument, error)
CreateFunction(ctx context.Context, arg CreateFunctionParams) (CoreFunction, error)
CreateInboundHook(ctx context.Context, arg CreateInboundHookParams) (CoreInboundHook, error)
CreateIntegration(ctx context.Context, arg CreateIntegrationParams) (CoreIntegration, error)
CreateInvitation(ctx context.Context, arg CreateInvitationParams) (CoreProjectInvitation, error)
@@ -44,6 +45,7 @@ type Querier interface {
DeleteComponent(ctx context.Context, arg DeleteComponentParams) error
DeleteDBConnection(ctx context.Context, arg DeleteDBConnectionParams) (CoreDbConnection, error)
DeleteDocument(ctx context.Context, arg DeleteDocumentParams) error
DeleteFunction(ctx context.Context, arg DeleteFunctionParams) error
DeleteInboundHook(ctx context.Context, arg DeleteInboundHookParams) error
DeleteIntegration(ctx context.Context, arg DeleteIntegrationParams) (CoreIntegration, error)
DeleteOrganizationByID(ctx context.Context, id uuid.UUID) (CoreOrganization, error)
@@ -67,6 +69,8 @@ type Querier interface {
GetDBConnectionByID(ctx context.Context, id uuid.UUID) (CoreDbConnection, error)
GetDBConnectionByName(ctx context.Context, arg GetDBConnectionByNameParams) (CoreDbConnection, error)
GetDocumentByID(ctx context.Context, arg GetDocumentByIDParams) (CoreDocument, error)
GetFunction(ctx context.Context, id uuid.UUID) (CoreFunction, error)
GetFunctionByName(ctx context.Context, arg GetFunctionByNameParams) (CoreFunction, error)
GetInboundHook(ctx context.Context, arg GetInboundHookParams) (CoreInboundHook, error)
GetInboundHookByToken(ctx context.Context, token string) (CoreInboundHook, error)
GetIntegrationByID(ctx context.Context, id uuid.UUID) (CoreIntegration, error)
@@ -86,6 +90,7 @@ type Querier interface {
HasFingerprintSeen(ctx context.Context, arg HasFingerprintSeenParams) (bool, error)
InsertEmailLog(ctx context.Context, arg InsertEmailLogParams) (CoreEmailLog, error)
InsertEvent(ctx context.Context, arg InsertEventParams) (CoreEvent, error)
InsertFunctionRun(ctx context.Context, arg InsertFunctionRunParams) (CoreFunctionRun, error)
InsertScheduledJobRun(ctx context.Context, arg InsertScheduledJobRunParams) (CoreScheduledJobRun, error)
InsertWebhookDelivery(ctx context.Context, arg InsertWebhookDeliveryParams) (CoreWebhookDelivery, error)
// Per-component most recent heartbeat — heartbeat_silence detection.
@@ -105,6 +110,8 @@ type Querier interface {
ListEnabledAlertRules(ctx context.Context) ([]CoreAlertRule, error)
ListErrorGroups(ctx context.Context, arg ListErrorGroupsParams) ([]ListErrorGroupsRow, error)
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)
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)
@@ -150,6 +157,7 @@ type Querier interface {
UpdateBucketByID(ctx context.Context, arg UpdateBucketByIDParams) (CoreBucket, error)
UpdateCollection(ctx context.Context, arg UpdateCollectionParams) (CoreCollection, error)
UpdateDocument(ctx context.Context, arg UpdateDocumentParams) (CoreDocument, error)
UpdateFunction(ctx context.Context, arg UpdateFunctionParams) (CoreFunction, error)
UpdateIntegrationHealth(ctx context.Context, arg UpdateIntegrationHealthParams) error
UpdateOrganizationByID(ctx context.Context, arg UpdateOrganizationByIDParams) (CoreOrganization, error)
UpdateOrganizationMemberRole(ctx context.Context, arg UpdateOrganizationMemberRoleParams) (CoreOrganizationMember, error)
+146
View File
@@ -0,0 +1,146 @@
package handlers
import (
"encoding/json"
"net/http"
"strconv"
"github.com/gin-gonic/gin"
apperrors "github.com/tdvorak/primora/apps/backend/internal/response"
"github.com/tdvorak/primora/apps/backend/internal/middleware"
"github.com/tdvorak/primora/apps/backend/internal/services"
)
func (h *HTTPHandler) listFunctions(c *gin.Context) {
actor, projectID, ok := h.actorAndProject(c)
if !ok {
return
}
items, err := h.Platform.ListFunctions(c.Request.Context(), actor, projectID)
if err != nil {
h.handleError(c, err)
return
}
c.JSON(http.StatusOK, gin.H{"items": items})
}
func (h *HTTPHandler) createFunction(c *gin.Context) {
actor, projectID, ok := h.actorAndProject(c)
if !ok {
return
}
var body services.CreateFunctionInput
if !h.bindAndValidate(c, &body) {
return
}
fn, err := h.Platform.CreateFunction(c.Request.Context(), actor, projectID, body, middleware.RequestIDFromContext(c))
if err != nil {
h.handleError(c, err)
return
}
c.JSON(http.StatusCreated, fn)
}
func (h *HTTPHandler) getFunction(c *gin.Context) {
actor, _, ok := h.actorAndProject(c)
if !ok {
return
}
functionID, ok := parseUUIDParam(c, "functionID")
if !ok {
return
}
fn, code, err := h.Platform.GetFunctionCode(c.Request.Context(), actor, functionID)
if err != nil {
h.handleError(c, err)
return
}
c.JSON(http.StatusOK, gin.H{"function": fn, "code": code})
}
func (h *HTTPHandler) updateFunction(c *gin.Context) {
actor, _, ok := h.actorAndProject(c)
if !ok {
return
}
functionID, ok := parseUUIDParam(c, "functionID")
if !ok {
return
}
var body services.UpdateFunctionInput
if !h.bindAndValidate(c, &body) {
return
}
fn, err := h.Platform.UpdateFunction(c.Request.Context(), actor, functionID, body, middleware.RequestIDFromContext(c))
if err != nil {
h.handleError(c, err)
return
}
c.JSON(http.StatusOK, fn)
}
func (h *HTTPHandler) deleteFunction(c *gin.Context) {
actor, _, ok := h.actorAndProject(c)
if !ok {
return
}
functionID, ok := parseUUIDParam(c, "functionID")
if !ok {
return
}
if err := h.Platform.DeleteFunction(c.Request.Context(), actor, functionID, middleware.RequestIDFromContext(c)); err != nil {
h.handleError(c, err)
return
}
c.Status(http.StatusNoContent)
}
func (h *HTTPHandler) invokeFunction(c *gin.Context) {
actor, _, ok := h.actorAndProject(c)
if !ok {
return
}
functionID, ok := parseUUIDParam(c, "functionID")
if !ok {
return
}
var body struct {
Payload json.RawMessage `json:"payload"`
}
// An empty body is a valid invoke — defaults to {}.
if c.Request.Body != nil && c.Request.ContentLength > 0 {
if err := c.ShouldBindJSON(&body); err != nil {
apperrors.Abort(c, http.StatusBadRequest, "invalid_json", err.Error())
return
}
}
run, err := h.Platform.InvokeFunction(c.Request.Context(), actor, functionID, body.Payload, middleware.RequestIDFromContext(c))
if err != nil {
h.handleError(c, err)
return
}
status := http.StatusOK
if run.Status != "success" {
status = http.StatusBadGateway
}
c.JSON(status, run)
}
func (h *HTTPHandler) listFunctionRuns(c *gin.Context) {
actor, _, ok := h.actorAndProject(c)
if !ok {
return
}
functionID, ok := parseUUIDParam(c, "functionID")
if !ok {
return
}
limit, _ := strconv.ParseInt(c.DefaultQuery("limit", "50"), 10, 64)
runs, err := h.Platform.ListFunctionRuns(c.Request.Context(), actor, functionID, limit)
if err != nil {
h.handleError(c, err)
return
}
c.JSON(http.StatusOK, gin.H{"items": runs})
}
+7
View File
@@ -84,6 +84,13 @@ func (h *HTTPHandler) Register(router *gin.Engine) {
api.POST("/projects/:projectID/inbound-hooks", h.createInboundHook)
api.DELETE("/projects/:projectID/inbound-hooks/:hookID", h.deleteInboundHook)
api.GET("/projects/:projectID/emails", h.listEmailLog)
api.GET("/projects/:projectID/functions", h.listFunctions)
api.POST("/projects/:projectID/functions", h.createFunction)
api.GET("/projects/:projectID/functions/:functionID", h.getFunction)
api.PUT("/projects/:projectID/functions/:functionID", h.updateFunction)
api.DELETE("/projects/:projectID/functions/:functionID", h.deleteFunction)
api.POST("/projects/:projectID/functions/:functionID/invoke", h.invokeFunction)
api.GET("/projects/:projectID/functions/:functionID/runs", h.listFunctionRuns)
api.GET("/projects/:projectID/events", h.listEvents)
api.GET("/projects/:projectID/issues", h.listIssues)
api.GET("/projects/:projectID/components", h.listComponents)
+356
View File
@@ -0,0 +1,356 @@
package services
import (
"bytes"
"context"
"encoding/json"
"fmt"
"os"
"os/exec"
"path/filepath"
"regexp"
"strings"
"time"
"github.com/google/uuid"
"github.com/tdvorak/primora/apps/backend/internal/database/db"
"github.com/tdvorak/primora/apps/backend/internal/models"
)
var functionNamePattern = regexp.MustCompile(`^[a-zA-Z][a-zA-Z0-9_-]{0,63}$`)
// FunctionRunner executes a function's code with a JSON payload on stdin
// and returns the outcome. Implemented by execRunner (a runtime binary on
// the backend host); tests substitute a fake.
type FunctionRunner interface {
Run(ctx context.Context, runtime, code string, payload json.RawMessage) FunctionResult
}
// FunctionResult is one execution's captured outcome.
type FunctionResult struct {
Status string // success | error | timeout
ExitCode *int32
Stdout string
Stderr string
DurationMs int64
}
// execRunner shells out to a JS runtime binary. User code runs with the
// backend's privileges — this is the documented self-hosted model (same as
// n8n custom-code nodes); operators harden at the container/VM boundary.
// jarvis: ceiling unsandboxed exec; upgrade to Firecracker/gVisor or a
// sidecar runtime pool when multi-tenant isolation is needed.
type execRunner struct {
bin string
timeout time.Duration
maxOutput int64
resolveBin func(runtime string) string
}
func NewExecRunner(bin string, timeout time.Duration, maxOutput int64) *execRunner {
return &execRunner{
bin: bin,
timeout: timeout,
maxOutput: maxOutput,
resolveBin: func(runtime string) string {
if bin != "" && bin != "auto" {
return bin
}
return runtime // "bun" or "deno" from PATH
},
}
}
func (r *execRunner) Run(ctx context.Context, runtime, code string, payload json.RawMessage) FunctionResult {
bin := r.resolveBin(runtime)
if bin == "" {
return FunctionResult{Status: "error", Stderr: "no functions runtime configured"}
}
dir, err := os.MkdirTemp("", "primora-fn-*")
if err != nil {
return FunctionResult{Status: "error", Stderr: err.Error()}
}
defer os.RemoveAll(dir)
file := filepath.Join(dir, "fn.ts")
if err := os.WriteFile(file, []byte(code), 0o600); err != nil {
return FunctionResult{Status: "error", Stderr: err.Error()}
}
ctx, cancel := context.WithTimeout(ctx, r.timeout)
defer cancel()
var cmd *exec.Cmd
switch runtime {
case "deno":
// --no-prompt: never block on interactive permission prompts.
cmd = exec.CommandContext(ctx, bin, "run", "--no-prompt", file)
default: // bun
cmd = exec.CommandContext(ctx, bin, "run", file)
}
cmd.Stdin = bytes.NewReader(payload)
cmd.Env = append(os.Environ(),
"PRIMORA_PAYLOAD="+string(payload),
"PRIMORA_FUNCTION_RUNTIME="+runtime,
)
var stdout, stderr bytes.Buffer
cmd.Stdout = &limitedWriter{w: &stdout, max: r.maxOutput}
cmd.Stderr = &limitedWriter{w: &stderr, max: r.maxOutput}
start := time.Now()
runErr := cmd.Run()
duration := time.Since(start).Milliseconds()
res := FunctionResult{
Status: "success",
Stdout: stdout.String(),
Stderr: stderr.String(),
DurationMs: duration,
}
if ctx.Err() == context.DeadlineExceeded {
res.Status = "timeout"
res.Stderr = strings.TrimSpace(res.Stderr + "\nexecution timed out")
return res
}
if runErr != nil {
res.Status = "error"
if exitErr, ok := runErr.(*exec.ExitError); ok {
code := int32(exitErr.ExitCode())
res.ExitCode = &code
} else {
res.Stderr = strings.TrimSpace(runErr.Error() + "\n" + res.Stderr)
}
}
return res
}
// limitedWriter caps captured output; excess bytes are discarded, not
// stored — a chatty function can't grow function_runs rows unboundedly.
type limitedWriter struct {
w *bytes.Buffer
max int64
}
func (l *limitedWriter) Write(p []byte) (int, error) {
remaining := l.max - int64(l.w.Len())
if remaining > 0 {
if int64(len(p)) > remaining {
l.w.Write(p[:remaining])
} else {
l.w.Write(p)
}
}
return len(p), nil
}
// --- 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"`
}
type UpdateFunctionInput struct {
Code *string `json:"code" validate:"omitempty,max=262144"`
Enabled *bool `json:"enabled"`
}
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"`
}
type FunctionRunSummary struct {
ID string `json:"id"`
Trigger string `json:"trigger"`
Status string `json:"status"`
ExitCode *int32 `json:"exit_code,omitempty"`
Stdout string `json:"stdout"`
Stderr string `json:"stderr"`
DurationMs int64 `json:"duration_ms"`
CreatedAt time.Time `json:"created_at"`
}
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,
}
}
func runSummary(r db.CoreFunctionRun) FunctionRunSummary {
return FunctionRunSummary{
ID: r.ID.String(),
Trigger: r.Trigger,
Status: r.Status,
ExitCode: r.ExitCode,
Stdout: r.Stdout,
Stderr: r.Stderr,
DurationMs: int64(r.DurationMs),
CreatedAt: r.CreatedAt.Time,
}
}
func (s *PlatformService) CreateFunction(ctx context.Context, actor *models.Actor, projectID uuid.UUID, input CreateFunctionInput, requestID string) (FunctionSummary, error) {
if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer"); err != nil {
return FunctionSummary{}, err
}
name := strings.TrimSpace(input.Name)
if !functionNamePattern.MatchString(name) {
return FunctionSummary{}, fmt.Errorf("invalid function name")
}
runtime := input.Runtime
if runtime == "" {
runtime = "bun"
}
row, err := s.repo.Queries().CreateFunction(ctx, db.CreateFunctionParams{
ProjectID: projectID,
Name: name,
Code: input.Code,
Runtime: runtime,
})
if err != nil {
return FunctionSummary{}, err
}
return functionSummary(row), nil
}
func (s *PlatformService) ListFunctions(ctx context.Context, actor *models.Actor, projectID uuid.UUID) ([]FunctionSummary, error) {
if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer", "viewer"); err != nil {
return nil, err
}
rows, err := s.repo.Queries().ListFunctions(ctx, projectID)
if err != nil {
return nil, err
}
out := make([]FunctionSummary, len(rows))
for i, r := range rows {
out[i] = functionSummary(r)
}
return out, nil
}
// GetFunctionCode returns the function's source — code is not a secret,
// project members already see it in the dashboard editor.
func (s *PlatformService) GetFunctionCode(ctx context.Context, actor *models.Actor, functionID uuid.UUID) (FunctionSummary, string, error) {
row, err := s.repo.Queries().GetFunction(ctx, functionID)
if err != nil {
return FunctionSummary{}, "", err
}
if err := s.requireProjectRole(ctx, actor, row.ProjectID, "admin", "developer", "viewer"); err != nil {
return FunctionSummary{}, "", err
}
return functionSummary(row), row.Code, nil
}
func (s *PlatformService) UpdateFunction(ctx context.Context, actor *models.Actor, functionID uuid.UUID, input UpdateFunctionInput, requestID string) (FunctionSummary, error) {
row, err := s.repo.Queries().GetFunction(ctx, functionID)
if err != nil {
return FunctionSummary{}, err
}
if err := s.requireProjectRole(ctx, actor, row.ProjectID, "admin", "developer"); err != nil {
return FunctionSummary{}, err
}
updated, err := s.repo.Queries().UpdateFunction(ctx, db.UpdateFunctionParams{
ID: functionID,
ProjectID: row.ProjectID,
Code: input.Code,
Enabled: input.Enabled,
})
if err != nil {
return FunctionSummary{}, err
}
return functionSummary(updated), nil
}
func (s *PlatformService) DeleteFunction(ctx context.Context, actor *models.Actor, functionID uuid.UUID, requestID string) error {
row, err := s.repo.Queries().GetFunction(ctx, functionID)
if err != nil {
return err
}
if err := s.requireProjectRole(ctx, actor, row.ProjectID, "admin", "developer"); err != nil {
return err
}
return s.repo.Queries().DeleteFunction(ctx, db.DeleteFunctionParams{
ID: functionID,
ProjectID: row.ProjectID,
})
}
// InvokeFunction executes a function with a JSON payload, records the run,
// and returns its summary.
func (s *PlatformService) InvokeFunction(ctx context.Context, actor *models.Actor, functionID uuid.UUID, payload json.RawMessage, requestID string) (FunctionRunSummary, error) {
row, err := s.repo.Queries().GetFunction(ctx, functionID)
if err != nil {
return FunctionRunSummary{}, err
}
if err := s.requireProjectRole(ctx, actor, row.ProjectID, "admin", "developer"); err != nil {
return FunctionRunSummary{}, err
}
if !row.Enabled {
return FunctionRunSummary{}, fmt.Errorf("function is disabled")
}
if len(payload) == 0 {
payload = json.RawMessage(`{}`)
}
if !json.Valid(payload) {
return FunctionRunSummary{}, fmt.Errorf("payload must be valid JSON")
}
if s.functions == nil {
return FunctionRunSummary{}, fmt.Errorf("functions runtime not configured")
}
result := s.functions.Run(ctx, row.Runtime, row.Code, payload)
var exit *int32
if result.ExitCode != nil {
exit = result.ExitCode
}
run, err := s.repo.Queries().InsertFunctionRun(ctx, db.InsertFunctionRunParams{
FunctionID: row.ID,
Trigger: "manual",
Status: result.Status,
ExitCode: exit,
Stdout: result.Stdout,
Stderr: result.Stderr,
DurationMs: int32(result.DurationMs),
})
if err != nil {
return FunctionRunSummary{}, err
}
return runSummary(run), nil
}
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 {
return nil, err
}
if err := s.requireProjectRole(ctx, actor, row.ProjectID, "admin", "developer", "viewer"); err != nil {
return nil, err
}
if limit <= 0 || limit > 200 {
limit = 50
}
rows, err := s.repo.Queries().ListFunctionRuns(ctx, db.ListFunctionRunsParams{
FunctionID: row.ID,
Limit: int32(limit),
})
if err != nil {
return nil, err
}
out := make([]FunctionRunSummary, len(rows))
for i, r := range rows {
out[i] = runSummary(r)
}
return out, nil
}
@@ -0,0 +1,72 @@
package services
import (
"bytes"
"context"
"strings"
"testing"
"time"
)
// These tests run the real bun binary — skipped when it is not installed,
// same convention as exec-dependent tests elsewhere.
func bunAvailable(t *testing.T) {
t.Helper()
r := NewExecRunner("auto", 5*time.Second, 4096)
res := r.Run(context.Background(), "bun", `console.log("probe")`, nil)
if strings.Contains(res.Stderr, "no functions runtime") || (res.Status == "error" && res.ExitCode == nil && strings.Contains(res.Stderr, "executable file not found")) {
t.Skip("bun not installed")
}
}
func TestExecRunnerSuccess(t *testing.T) {
bunAvailable(t)
r := NewExecRunner("auto", 10*time.Second, 4096)
res := r.Run(context.Background(), "bun",
`const p = JSON.parse(await new Response(Bun.stdin.stream()).text()); console.log("got:" + p.x);`,
[]byte(`{"x":42}`))
if res.Status != "success" {
t.Fatalf("status=%s stderr=%s", res.Status, res.Stderr)
}
if !strings.Contains(res.Stdout, "got:42") {
t.Fatalf("stdout=%q", res.Stdout)
}
if res.DurationMs <= 0 {
t.Fatalf("duration=%d", res.DurationMs)
}
}
func TestExecRunnerErrorCapturesStderr(t *testing.T) {
bunAvailable(t)
r := NewExecRunner("auto", 10*time.Second, 4096)
res := r.Run(context.Background(), "bun", `throw new Error("test-boom")`, nil)
if res.Status != "error" {
t.Fatalf("status=%s", res.Status)
}
if res.ExitCode == nil || *res.ExitCode == 0 {
t.Fatalf("exit=%v", res.ExitCode)
}
if !strings.Contains(res.Stderr, "test-boom") {
t.Fatalf("stderr=%q", res.Stderr)
}
}
func TestExecRunnerTimeout(t *testing.T) {
bunAvailable(t)
r := NewExecRunner("auto", 500*time.Millisecond, 4096)
res := r.Run(context.Background(), "bun", `await new Promise(r => setTimeout(r, 10000))`, nil)
if res.Status != "timeout" {
t.Fatalf("status=%s", res.Status)
}
}
func TestLimitedWriterCaps(t *testing.T) {
var buf bytes.Buffer
lw := &limitedWriter{w: &buf, max: 5}
if n, err := lw.Write([]byte("hello world")); err != nil || n != 11 {
t.Fatalf("write n=%d err=%v", n, err)
}
if buf.String() != "hello" {
t.Fatalf("got %q", buf.String())
}
}
@@ -39,6 +39,7 @@ type PlatformService struct {
dispatcher *WebhookDispatcher
scheduler *JobScheduler
alerts *alertEvaluator
functions FunctionRunner
}
type BootstrapInput struct {
@@ -224,8 +225,8 @@ type InvitationSummary struct {
Status string `json:"status"`
}
func NewPlatformService(repo *repositories.CoreRepository, store storage.Store, mailer *Mailer, publicURL string, dbxClient *dbx.Client, enc *secrets.Encryptor, logger *slog.Logger) *PlatformService {
s := &PlatformService{repo: repo, store: store, mailer: mailer, publicURL: publicURL, dbx: dbxClient, hub: NewEventHub(), realtime: NewEventHub(), enc: enc}
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}
if enc != nil {
s.dispatcher = NewWebhookDispatcher(repo.Queries(), enc, logger)
s.dispatcher.Start(context.Background())
+268
View File
@@ -2167,6 +2167,182 @@ paths:
$ref: "#/components/schemas/EmailLogListResponse"
"401":
$ref: "#/components/responses/ErrorResponse"
/projects/{projectID}/functions:
get:
tags: [Functions]
operationId: listFunctions
summary: List functions in a project.
security:
- bearerAuth: []
- apiKeyAuth: []
parameters:
- $ref: "#/components/parameters/ProjectID"
responses:
"200":
description: Functions
content:
application/json:
schema:
$ref: "#/components/schemas/FunctionListResponse"
"401":
$ref: "#/components/responses/ErrorResponse"
post:
tags: [Functions]
operationId: createFunction
summary: Create a function (name, JS/TS code, runtime bun|deno).
security:
- bearerAuth: []
- apiKeyAuth: []
parameters:
- $ref: "#/components/parameters/ProjectID"
requestBody:
required: true
content:
application/json:
schema:
$ref: "#/components/schemas/CreateFunctionRequest"
responses:
"201":
description: Created function
content:
application/json:
schema:
$ref: "#/components/schemas/Function"
"400":
$ref: "#/components/responses/ErrorResponse"
/projects/{projectID}/functions/{functionID}:
get:
tags: [Functions]
operationId: getFunction
summary: Function metadata plus its source code.
security:
- bearerAuth: []
- apiKeyAuth: []
parameters:
- $ref: "#/components/parameters/ProjectID"
- name: functionID
in: path
required: true
schema: { type: string, format: uuid }
responses:
"200":
description: Function with code
content:
application/json:
schema:
$ref: "#/components/schemas/FunctionDetailResponse"
"404":
$ref: "#/components/responses/ErrorResponse"
put:
tags: [Functions]
operationId: updateFunction
summary: Update function code and/or enabled state.
security:
- bearerAuth: []
- apiKeyAuth: []
parameters:
- $ref: "#/components/parameters/ProjectID"
- name: functionID
in: path
required: true
schema: { type: string, format: uuid }
requestBody:
required: true
content:
application/json:
schema:
$ref: "#/components/schemas/UpdateFunctionRequest"
responses:
"200":
description: Updated function
content:
application/json:
schema:
$ref: "#/components/schemas/Function"
"404":
$ref: "#/components/responses/ErrorResponse"
delete:
tags: [Functions]
operationId: deleteFunction
summary: Delete a function and its run history.
security:
- bearerAuth: []
- apiKeyAuth: []
parameters:
- $ref: "#/components/parameters/ProjectID"
- name: functionID
in: path
required: true
schema: { type: string, format: uuid }
responses:
"204":
description: Deleted
"404":
$ref: "#/components/responses/ErrorResponse"
/projects/{projectID}/functions/{functionID}/invoke:
post:
tags: [Functions]
operationId: invokeFunction
summary: Execute the function with a JSON payload on stdin; returns the recorded run. 502 when the run fails or times out.
security:
- bearerAuth: []
- apiKeyAuth: []
parameters:
- $ref: "#/components/parameters/ProjectID"
- name: functionID
in: path
required: true
schema: { type: string, format: uuid }
requestBody:
required: false
content:
application/json:
schema:
$ref: "#/components/schemas/InvokeFunctionRequest"
responses:
"200":
description: Recorded run (status success)
content:
application/json:
schema:
$ref: "#/components/schemas/FunctionRun"
"502":
description: Recorded run (status error|timeout)
content:
application/json:
schema:
$ref: "#/components/schemas/FunctionRun"
"404":
$ref: "#/components/responses/ErrorResponse"
/projects/{projectID}/functions/{functionID}/runs:
get:
tags: [Functions]
operationId: listFunctionRuns
summary: Recent runs for a function.
security:
- bearerAuth: []
- apiKeyAuth: []
parameters:
- $ref: "#/components/parameters/ProjectID"
- name: functionID
in: path
required: true
schema: { type: string, format: uuid }
- name: limit
in: query
schema:
type: integer
default: 50
maximum: 200
responses:
"200":
description: Function runs
content:
application/json:
schema:
$ref: "#/components/schemas/FunctionRunListResponse"
"404":
$ref: "#/components/responses/ErrorResponse"
components:
securitySchemes:
bearerAuth:
@@ -4213,3 +4389,95 @@ components:
items:
$ref: "#/components/schemas/EmailLogEntry"
required: [items]
Function:
type: object
properties:
id:
type: string
format: uuid
name:
type: string
runtime:
type: string
enum: [bun, deno]
enabled:
type: boolean
created_at:
type: string
format: date-time
updated_at:
type: string
format: date-time
required: [id, name, runtime, enabled, created_at, updated_at]
FunctionListResponse:
type: object
properties:
items:
type: array
items:
$ref: "#/components/schemas/Function"
required: [items]
FunctionDetailResponse:
type: object
properties:
function:
$ref: "#/components/schemas/Function"
code:
type: string
required: [function, code]
CreateFunctionRequest:
type: object
properties:
name:
type: string
code:
type: string
runtime:
type: string
enum: [bun, deno]
default: bun
required: [name, code]
UpdateFunctionRequest:
type: object
properties:
code:
type: string
enabled:
type: boolean
InvokeFunctionRequest:
type: object
properties:
payload: {}
FunctionRun:
type: object
properties:
id:
type: string
format: uuid
trigger:
type: string
enum: [manual, schedule, hook]
status:
type: string
enum: [success, error, timeout]
exit_code:
type: integer
nullable: true
stdout:
type: string
stderr:
type: string
duration_ms:
type: integer
created_at:
type: string
format: date-time
required: [id, trigger, status, stdout, stderr, duration_ms, created_at]
FunctionRunListResponse:
type: object
properties:
items:
type: array
items:
$ref: "#/components/schemas/FunctionRun"
required: [items]