Running in production
The parts that matter once the jobs are someone else's to operate: what gets written down, what survives a restart, and what operators can see.
Log policies
Persisted logs are the most expensive part of the system — one run is one row, but its logs are N. Logs are off by default; here is how to turn them on for the jobs that need them.
go
package main
import (
"context"
"fmt"
"time"
core "gitlab.finema.co/finema/idin-core/v2"
)
// --- Example 5: logs --------------------------------------------------------
//
// Persisted logs are the most expensive part of the system: a run record is one
// row, but its logs are N. A job running every 30s that logs 20 lines writes
// ~1.7M rows a month on its own. So storing them is **off by default**.
//
// Off does not mean blind. Under LogOff you still get:
// - the lines on stdout, tagged with job/run_id/attempt
// - live tailing while the run is in flight
// - the last ~50 lines attached to JobRun.Error when a run fails
//
// Turn persistence on per job, or per trigger, only where it earns its cost.
func registerLogPolicyJobs(reg *core.JobRegistry) {
// default (LogOff): noisy, frequent, boring when it works
_ = reg.Register(core.JobDef{
Name: "sync-users",
Schedule: core.Every(5 * time.Minute),
}, syncUsers)
// LogOnFailure: costs nothing on success (the buffer is dropped), and gives
// you the whole story when it breaks. Start here when you want more.
_ = reg.Register(core.JobDef{
Name: "sync-partners",
Schedule: core.Every(15 * time.Minute),
Logs: core.LogPolicyPtr(core.LogOnFailure),
}, syncUsers)
// LogAlways: important and infrequent, where the audit trail is the point.
// Keep the runs for a year but the logs for a month.
_ = reg.Register(core.JobDef{
Name: "monthly-settlement",
Schedule: core.Cron("0 4 1 * *"),
Logs: core.LogPolicyPtr(core.LogAlways),
RetainRuns: 365 * 24 * time.Hour,
RetainLogs: 30 * 24 * time.Hour,
}, syncUsers)
}
func syncUsers(c core.ICronjobContext) error {
c.Log().Info("sync started", "job", c.JobName())
c.Log().Info("sync finished", "updated", 12)
return nil
}
// triggerWithFullLogs is the operator's escape hatch: this one run keeps
// everything, while the scheduled ones stay silent.
func triggerWithFullLogs(ctx context.Context, runner *core.JobRunner, user string) (*core.JobRun, core.IError) {
return runner.Trigger(ctx, "sync-users", nil, core.TriggerOptions{
By: user,
CaptureLogs: core.LogPolicyPtr(core.LogAlways),
})
}
// watchRun tails a run as it happens. It reads from an in-memory hub, not the
// store, which is why it works whatever the log policy is — the usual "click
// run, watch it go" flow needs no storage at all.
func watchRun(runner *core.JobRunner, runID string) {
lines, unsubscribe := runner.TailLogs(runID)
defer unsubscribe()
timeout := time.After(5 * time.Minute)
for {
select {
case e, ok := <-lines:
if !ok {
return
}
fmt.Printf("[%s] %s %s\n", e.At.Format(time.TimeOnly), e.Level, e.Message)
case <-timeout:
return
}
}
}
// readStoredLogs pages through what was persisted, using seq as the cursor.
func readStoredLogs(ctx context.Context, runner *core.JobRunner, runID string) core.IError {
var after int64
for {
entries, err := runner.Logs(ctx, runID, after, 200)
if err != nil {
return err
}
if len(entries) == 0 {
return nil
}
for _, e := range entries {
fmt.Printf("%d %s %s\n", e.Seq, e.Level, e.Message)
}
after = entries[len(entries)-1].Seq
}
}Durable runs in SQL
The default backends keep everything in memory: zero configuration, but nothing survives a restart. Swap in the repository-backed store to keep run history.
go
package main
import (
"context"
"time"
core "gitlab.finema.co/finema/idin-core/v2"
"gitlab.finema.co/finema/idin-core/v2/jobstore"
"gitlab.finema.co/finema/idin-core/v2/repository"
)
// --- Example 7: durable runs ------------------------------------------------
//
// The default backends keep everything in memory: zero configuration, but
// queued runs disappear with the process. When you want history and restart
// safety, swap in the repository-backed store — GORM underneath, no Redis, no
// new infrastructure. The runs table doubles as the queue.
func durableRunner(app *core.App, reg *core.JobRegistry) (*core.JobRunner, core.IError) {
// create the tables (or take the DDL into your own migration tool)
if err := jobstore.Migrate(app.NewContext(context.Background()).DB()); err != nil {
return nil, err
}
store := jobstore.New(app)
queue := jobstore.NewQueue(app, jobstore.WithPollInterval(time.Second))
return core.NewJobRunner(app, reg,
core.WithJobStore(store),
core.WithJobQueue(queue),
core.WithWorkers(4),
), nil
}
// customPlacement puts job data on its own connection and its own table names —
// useful when the service already owns a schema convention.
func customPlacement(app *core.App, reg *core.JobRegistry) *core.JobRunner {
opts := []jobstore.Option{
jobstore.WithTables("ops_job_runs", "ops_job_run_logs"),
jobstore.WithConnection("ops"), // core.IContext.DBS("ops")
}
return core.NewJobRunner(app, reg,
core.WithJobStore(jobstore.New(app, opts...)),
core.WithJobQueue(jobstore.NewQueue(app, opts...)),
)
}
// Because core.JobRun is an ordinary core.IModel, job history is queryable with
// the same repository API as the rest of the service — no special client, and it
// joins with your own tables.
func failedRunsThisWeek(ctx core.IContext) ([]core.JobRun, core.IError) {
since := time.Now().AddDate(0, 0, -7)
return repository.New[core.JobRun](ctx).
Where("status = ?", core.RunFailed).
Where("created_at >= ?", since).
Order("created_at DESC").
FindAll()
}
// slowestJobs is the kind of report the store makes possible for free.
func slowestJobs(ctx core.IContext) ([]struct {
JobName string
AvgMS float64
}, core.IError) {
var out []struct {
JobName string
AvgMS float64
}
err := repository.New[core.JobRun](ctx).
Select("job_name, AVG(duration_ms) as avg_ms").
Where("status = ?", core.RunSucceeded).
Group("job_name").
Order("avg_ms DESC").
Limit(10).
Scan(&out)
return out, err
}
// purgeOldData keeps the tables from growing without bound. Register it on a
// schedule (see main.go) — retention for logs is deliberately shorter than for
// runs: statistics stay, chatter goes.
func purgeOldData(ctx context.Context, runner *core.JobRunner) (int64, core.IError) {
return runner.Purge(ctx)
}An admin API over the runner
The runner exposes everything an operator UI needs, so the HTTP layer is thin enough to keep in your own service — and yours to secure.
go
package main
import (
"net/http"
"strconv"
"github.com/labstack/echo/v4"
core "gitlab.finema.co/finema/idin-core/v2"
)
// --- Example 8: an admin API over the runner --------------------------------
//
// The runner exposes everything an operator UI needs, so the HTTP layer is thin
// enough to keep in your own service — and yours to secure.
//
// SECURITY: these endpoints run arbitrary business jobs. Mount them behind real
// authentication and authorisation, never on a public path.
func mountJobAdmin(e *core.Server, runner *core.JobRunner, adminOnly echo.MiddlewareFunc) {
if adminOnly == nil {
panic("job admin API mounted without authentication") // fail fast, loudly
}
g := e.Group("/_jobs", adminOnly)
// What jobs exist, and what each one takes. JobInfo already carries the
// parameter schema — name, kind (string/int/bool/date/enum/…), whether it is
// required, its enum values and description — so a UI can render the "run
// now" form without anyone hand-writing it:
//
// {
// "name": "sales-report",
// "schedule": "cron(30 1 * * *)",
// "params": [
// {"name": "date", "kind": "date", "required": true,
// "description": "Day to report on", "example": "2026-07-01"},
// {"name": "status", "kind": "enum", "enum": ["draft","sent","paid"], "default": "sent"},
// {"name": "limit", "kind": "int", "description": "Max rows"},
// {"name": "force", "kind": "bool"},
// {"name": "emails", "kind": "array", "items": {"kind": "string"}}
// ]
// }
g.GET("", func(c core.IHTTPContext) error {
return c.JSON(http.StatusOK, runner.Registry().Info())
})
// just the parameter schema of one job, for a form that loads on demand
g.GET("/:name/params", func(c core.IHTTPContext) error {
params := runner.Registry().Params(c.Param("name"))
if _, ok := runner.Registry().Def(c.Param("name")); !ok {
return core.Newf(http.StatusNotFound, "JOB_NOT_FOUND",
"job %s is not registered", c.Param("name"))
}
return c.JSON(http.StatusOK, params)
})
// run it now. ?wait=1 blocks for the outcome; otherwise you get the queued
// run back straight away and poll (or tail the logs).
g.POST("/:name/run", func(c core.IHTTPContext) error {
var params map[string]any
if err := c.BindOnly(¶ms); err != nil {
return err
}
opts := core.TriggerOptions{
By: userOf(c),
IdemKey: c.QueryParam("idem_key"),
}
if c.QueryParam("capture_logs") == "always" {
opts.CaptureLogs = core.LogPolicyPtr(core.LogAlways)
}
if c.QueryParam("wait") != "" {
run, err := runner.TriggerAndWait(c, c.Param("name"), params, opts)
if err != nil {
return err
}
return c.JSON(http.StatusOK, run)
}
run, err := runner.Trigger(c, c.Param("name"), params, opts)
if err != nil {
return err
}
return c.JSON(http.StatusAccepted, run)
})
g.POST("/:name/pause", func(c core.IHTTPContext) error {
if err := runner.Pause(c.Param("name")); err != nil {
return err
}
return c.NoContent(http.StatusNoContent)
})
g.POST("/:name/resume", func(c core.IHTTPContext) error {
if err := runner.Resume(c.Param("name")); err != nil {
return err
}
return c.NoContent(http.StatusNoContent)
})
// history, with the framework's usual pagination
g.GET("/runs", func(c core.IHTTPContext) error {
f := core.JobRunFilter{
JobName: c.QueryParam("job"),
Queue: c.QueryParam("queue"),
Trigger: core.Trigger(c.QueryParam("trigger")),
Page: c.GetPageOptions(),
}
if s := c.QueryParam("status"); s != "" {
f.Statuses = []core.RunStatus{core.RunStatus(s)}
}
page, err := runner.Runs(c, f)
if err != nil {
return err
}
return c.JSON(http.StatusOK, page)
})
g.GET("/runs/:id", func(c core.IHTTPContext) error {
run, err := runner.Run(c, c.Param("id"))
if err != nil {
return err
}
return c.JSON(http.StatusOK, run)
})
// persisted logs, paged by seq. Empty when the policy did not keep them —
// use the stream endpoint below while the run is in flight.
g.GET("/runs/:id/logs", func(c core.IHTTPContext) error {
after, _ := strconv.ParseInt(c.QueryParam("after_seq"), 10, 64)
entries, err := runner.Logs(c, c.Param("id"), after, 500)
if err != nil {
return err
}
return c.JSON(http.StatusOK, entries)
})
// live tail (server-sent events) — reads the in-memory hub, so it works
// whatever the log policy is
g.GET("/runs/:id/stream", func(c core.IHTTPContext) error {
lines, unsubscribe := runner.TailLogs(c.Param("id"))
defer unsubscribe()
res := c.Response()
res.Header().Set(echo.HeaderContentType, "text/event-stream")
res.Header().Set("Cache-Control", "no-cache")
res.WriteHeader(http.StatusOK)
for {
select {
case <-c.Request().Context().Done():
return nil
case e, ok := <-lines:
if !ok {
return nil
}
if _, err := res.Write([]byte("data: " + e.Level + " " + e.Message + "\n\n")); err != nil {
return nil
}
res.Flush()
}
}
})
g.POST("/runs/:id/cancel", func(c core.IHTTPContext) error {
reason := c.QueryParam("reason")
if err := runner.Cancel(c, c.Param("id"), userOf(c), reason); err != nil {
return err
}
return c.NoContent(http.StatusNoContent)
})
// replay a finished run as a new one (409 if it is still going, 403 if the
// job opted out of being replayed)
g.POST("/runs/:id/replay", func(c core.IHTTPContext) error {
run, err := runner.Replay(c, c.Param("id"), core.ReplayOptions{By: userOf(c)})
if err != nil {
return err
}
return c.JSON(http.StatusAccepted, run)
})
}
func userOf(c core.IHTTPContext) string {
if u := c.GetUser(); u != nil {
return u.ID
}
return "unknown"
}