Jobs (cron, manual triggers, queue, status, logs)

A job is a named function you can put on a schedule and run by hand, with parameters. Every run — scheduled, manual, retried or replayed — becomes a JobRun record you can query, watch, cancel and re-run.

Migrating from the old scheduler? AddByCron / AddByDuration and the JobFunc signature are unchanged. What changed is that a tick now queues a run instead of calling your function directly, so scheduled jobs get status and logs for free.

The shape of it

  cron tick ──┐
  manual   ───┼─► JobRun{queued} ─► IJobQueue ─► worker ─► your handler
  retry ──────┘                                    │
                                                   └─► IJobStore (status, logs)

The scheduler never executes anything; it only enqueues. That is why manual and scheduled runs behave identically — there is one code path, not two.

Smallest useful setup

reg := core.NewJobRegistry()

_ = reg.Register(core.JobDef{
    Name:     "nightly-report",
    Schedule: core.Cron("0 2 * * *"),
}, func(c core.ICronjobContext) error {
    c.Log().Info("building report")
    return nil
})

runner := core.NewJobRunner(app, reg)   // in-memory queue + store, no config
runner.Start()

sc, _ := core.NewScheduler(app, runner)
_ = sc.Start()

Run it by hand, right now:

run, err := runner.Trigger(ctx, "nightly-report", nil, core.TriggerOptions{By: userID})

Parameters

RegisterJob decodes the run's JSON into your struct and validates it before the handler is called — the same contract as an HTTP payload, and the same validation builder, because a run context is a core.IContext:

type ReportParams struct {
    Date  *string `json:"date"`
    Email *string `json:"email"`
    Limit *int64  `json:"limit"`
}

func (p *ReportParams) Valid(ctx core.IContext) core.IError {
    v := valid.New(ctx)
    v.Str("date", p.Date).Date("2006-01-02")
    v.Str("email", p.Email).Email().Exists("users", "email")   // DB rules work too
    v.Int("limit", p.Limit).Between(1, 1000)
    return v.Error()
}

core.RegisterJob(reg, core.JobDef{Name: "report"},
    func(c core.ICronjobContext, p *ReportParams) error {
        date := utils.ToNonPointerOr(p.Date, yesterday())
        return build(c, date)
    })

runner.Trigger(ctx, "report", &ReportParams{Date: utils.ToPointer("2026-07-01")})

Everything in validation.md applies unchanged — field builders, Each/Nested, Unique/Exists with scopes, Check/CheckDB, custom messages. DB-backed rules query through the run's own connections, so they are cancelled with the run like any other query.

Violations fail the run before the handler is called, and land on the run itself in the same shape the HTTP layer returns:

{
  "status": "failed",
  "error": {
    "code": "INVALID_PARAMS",
    "status": 400,
    "fields": {
      "date":  { "code": "INVALID_DATE",  "message": "The date field must be a valid date" },
      "email": { "code": "NOT_EXISTS",    "message": "The email field does not exist" }
    }
  }
}

⚠️ A scheduled run passes no parameters, so the struct arrives with its zero values. If a field is Required(), every scheduled run will fail validation. For a job that is both scheduled and parameterised, validate the shape (Date(), Between(), …) and default the value in the handler, or keep the job manual-only.

IValidate (Valid() core.IError, no context) is supported too, for parameters with no DB rules. Both work whether the parameter type is a struct or a pointer to one.

Publishing the parameter schema

A job also advertises what it accepts, so an admin API can list the jobs and a UI can build the "run now" form on its own. Declare it on the JobDef, next to everything else about the job — the parameter struct stays a plain data type:

core.RegisterJob(reg, core.JobDef{
    Name: "sales-report",
    Params: core.Params(
        core.DateParam("date").Required().
            Desc("Day to report on").Example("2026-07-01"),
        core.EnumParam("status", "draft", "sent", "paid").Default("sent"),
        core.IntParam("limit").Desc("Maximum rows"),
        core.BoolParam("force"),
        core.ArrayParam("emails", core.StringParam("")),
        core.ObjectParam("address",
            core.StringParam("street").Required(),
            core.StringParam("zip"),
        ),
    ),
}, salesReport)

Constructors: StringParam · IntParam · NumberParam · BoolParam · DateParam · TimeParam · DateTimeParam · EnumParam(name, values…) · ArrayParam(name, item) · ObjectParam(name, fields…) · AnyParam · CustomParam(name, kind).

Chained on any of them:

.Required() .Desc(…) .Default(…) .Example(…) .Label(…)	what a form shows
.Enum(values…) .Options(Opt(v, label)…)	a fixed set, with display text when it differs from the value
.Min(n) .Max(n) .Between(a, b) .Pattern(re) .Multiline()	constraints the form can enforce before submitting
.Kind(…) .Meta(key, value) .With(fn)	anything this package does not model

Customising further

CustomParam takes any kind you like, and Meta carries whatever your UI needs — a widget, a group, an ordering hint. Both are passed through untouched, so a consumer that does not recognise them can fall back to a text input:

core.CustomParam("amount", "currency").
    Meta("currency", "THB").
    Meta("widget", "money-input").
    Between(1, 1_000_000)

With is the escape hatch for house conventions — wrap them once and reuse:

func tenantParam() *core.ParamSpec {
    return core.StringParam("tenant").Required().
        With(func(f *core.ParamField) { f.Pattern = "^t_[a-z0-9]+$" })
}

Letting the type describe itself

A parameter type may implement core.IParamsDescriber instead. It is the only way to build a schema from runtime state — allowed values loaded from the database, a feature flag — and it keeps the description next to the struct:

func (PayoutParams) ParamSchema() []core.ParamField {
    return core.Params(
        core.StringParam("tenant").Required(),
        core.EnumParam("currency", loadCurrencies()...),
    )
}

JobDef.Params wins when both are given. Whichever way the schema arrives, the name check below applies.

runner.Registry().Info()            // every job + its schedule, limits and params
runner.Registry().Params("report")  // just one job's parameters

{
  "name": "report",
  "schedule": "cron(0 2 * * *)",
  "queue": "heavy",
  "max_attempts": 3,
  "concurrency": "skip",
  "replayable": false,
  "paused": false,
  "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": "Maximum rows"},
    {"name": "force",  "kind": "bool"},
    {"name": "emails", "kind": "array", "items": {"kind": "string"}}
  ]
}

Kinds: string, int, number, bool, date, time, datetime, enum, array (with items), object (with fields), any.

A declared name that the parameter struct does not have is rejected at registration:

job sales-report declares parameters that its type does not have: statuss

That is the trade for keeping the schema out of the struct: it can drift, so the drift is caught at startup rather than becoming a form field the job silently ignores. (Free-form parameter types — a map[string]any — have nothing to check against, so their declared schema is taken as written.)

Declare nothing and the job still lists its parameters, read off the type:

core.RegisterJob(reg, core.JobDef{Name: "quick-report"}, salesReport)
// params: [{name: "date", kind: "string"}, {name: "limit", kind: "int"}, …]

That is enough for a rough form, but only a declared schema knows that a string is a date, which values an enum allows, or what a field means. Derived automatically: time.Time → datetime, slices → array, nested structs → object, embedded structs flattened, json:"-" and unexported fields left out.

Required() is documentation for the caller — it does not validate anything; Valid is still what enforces the rules. Keeping them separate is deliberate: a job that is also scheduled must tolerate empty parameters (see the warning above) even while the form marks them required for a human.

Status

run, _ := runner.Run(ctx, runID)          // one run
page, _ := runner.Runs(ctx, core.JobRunFilter{
    JobName:  "report",
    Statuses: []core.RunStatus{core.RunFailed},
    Page:     &core.PageOptions{Limit: 20},
})
done, _ := runner.Wait(ctx, runID)        // block until it finishes

queued ──► running ──► succeeded
   │          │
   │          ├──► failed      (attempts exhausted; otherwise back to queued)
   │          └──► canceled
   └──► skipped                (dropped by the concurrency policy)

A handler can report progress and a result:

c.Progress(50, "halfway")
c.SetResult(map[string]any{"rows": n})

Stopping

Three different things, deliberately kept apart:

runner.Cancel(ctx, runID, by, reason)  // one run
runner.Pause("report")                 // the whole job: no schedule, no triggers
runner.Stop(shutdownCtx)               // the process: drain, then requeue

A queued run is pulled out of the queue and never starts. A running one is asked to stop — and because Go cannot kill a goroutine, your handler has to cooperate:

for _, item := range items {
    if c.IsStopping() {
        return c.Err()
    }
    ...
}

Everything that goes through the context (DB, cache, HTTP, MQ) is cancelled for you; only your own loops need the check. A handler that ignores it is recorded as canceled after StopGrace and abandoned, with a warning — it will not pin a worker forever.

On shutdown, runs that do not finish in time go back on the queue. They are not failures; they never got the chance to fail.

How many run at once

core.NewJobRunner(app, reg,
    core.WithWorkers(4),               // this process
    core.WithQueueLimit("heavy", 2),   // this queue
)

core.JobDef{Name: "sync", MaxConcurrent: 1}   // this job (1 = singleton)

When a run hits a limit, JobDef.Concurrency decides:

policy	what happens	use it for
ConcurrencyEnqueue (default)	waits its turn, nothing is lost	ordinary work
ConcurrencySkip	finishes as skipped	anything on a short schedule
ConcurrencyReplace	cancels the older run and takes over	"latest wins" rebuilds

A frequent cron job with Enqueue is the classic trap: if a run takes longer than the interval, the queue grows forever. Use Skip there.

Retry, replay, idempotency

Four ideas people all call "run it again":

	who does it	result
Retry	the runner, after a failure, while MaxAttempts remain	same run, next attempt
Replay	a person, on a finished run	a new run, linked via ReplayOf/RootID
Cancel	a person, on an unfinished run	canceled
IdemKey	you, at trigger time	no duplicate run gets created

core.JobDef{Name: "recalc", MaxAttempts: 3,
    Backoff: core.ExponentialBackoff(time.Second, 2*time.Minute)}

runner.Replay(ctx, oldRunID, core.ReplayOptions{By: user})           // same params
runner.Replay(ctx, oldRunID, core.ReplayOptions{Params: newParams})  // fixed input

runner.Trigger(ctx, "settle", p, core.TriggerOptions{IdemKey: "settle:2026-07"})

Replay always creates a new record — history is never overwritten. Jobs that are dangerous to repeat opt out:

core.JobDef{Name: "payout", Replayable: core.BoolPtr(false)}   // replay → 403

Runs are delivered at least once (a worker can die and the run comes back): handlers must be idempotent.

Logs — off by default

Persisted logs are the most expensive part of the system. One run is one row; its logs are N. A job running every 30s that logs 20 lines writes ~1.7M rows a month by itself. So LogOff is the 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

core.NewJobRunner(app, reg,
    core.WithLogPolicy(core.LogOff),                       // default
    core.WithLogLimits(core.LogLimits{MaxLines: 1000, MinLevel: "info"}),
)

core.JobDef{Name: "sync",       Logs: core.LogPolicyPtr(core.LogOnFailure)}
core.JobDef{Name: "settlement", Logs: core.LogPolicyPtr(core.LogAlways)}

policy	cost	when
LogOff	none	the default
LogOnFailure	only failed runs	start here — successful runs are free
LogAlways	every line	important, infrequent jobs

One run at a time can ask for more — the operator's escape hatch:

runner.Trigger(ctx, "sync", nil, core.TriggerOptions{
    CaptureLogs: core.LogPolicyPtr(core.LogAlways),
})

Reading them:

entries, _ := runner.Logs(ctx, runID, afterSeq, 200)   // persisted, paged by seq
lines, stop := runner.TailLogs(runID)                  // live, no storage needed
defer stop()

Durable runs (SQL)

The default backends are in memory: no configuration, but queued runs disappear with the process. For history and restart safety, use the repository-backed store — GORM underneath, no extra infrastructure. The runs table doubles as the queue.

import "gitlab.finema.co/finema/idin-core/v2/jobstore"

_ = jobstore.Migrate(db)   // or take the DDL into your own migration tool

runner := core.NewJobRunner(app, reg,
    core.WithJobStore(jobstore.New(app)),
    core.WithJobQueue(jobstore.NewQueue(app)),
)

Placement is configurable, without any global state:

jobstore.New(app,
    jobstore.WithTables("ops_job_runs", "ops_job_run_logs"),
    jobstore.WithConnection("ops"),   // core.IContext.DBS("ops")
)

core.JobRun is an ordinary core.IModel, so job history is queryable with the repository you already use — and joins with your own tables:

repository.New[core.JobRun](ctx).
    Where("job_name = ? AND status = ?", "settlement", core.RunFailed).
    FindAll()

Both are interfaces (IJobStore, IJobQueue), so a Mongo or Redis backend is a drop-in replacement. A new backend is correct when it passes the shared conformance suite:

make test-integration        # or: go test --tags=integration ./...

It runs the same contract against every backend — in-memory, SQL through the repository, and (when test.env points at one) the database your service really uses. Adding a backend means adding one line to backends() in jobstore/repo_store_test.go.

A queue whose storage is the store implements core.IStoreBackedQueue (SharesStore() bool). The runner then skips Enqueue after writing a run: writing it is already what makes it available, and a second write can flip a run back to queued after another worker claimed it — running it twice.

Retention

core.JobDef{RetainRuns: 365 * 24 * time.Hour, RetainLogs: 30 * 24 * time.Hour}

Keep logs for less time than runs: the statistics are worth keeping, the chatter is not. Register the purge on a schedule:

sc.Add(core.JobDef{Name: "core.job.purge", Schedule: core.Cron("0 3 * * *")},
    func(c core.ICronjobContext) error {
        _, err := runner.Purge(c)
        return err
    })

Admin API

The runner exposes everything an operator UI needs, so the HTTP layer stays in your service, where your authentication lives. A complete example is in examples/jobs/08_http_admin.go: list jobs (with their parameter schema), run, pause/resume, list runs, read or stream logs, cancel, replay.

These endpoints run arbitrary business jobs. Mount them behind real authentication — never on a public path.

Scaling out later

Everything above assumes a single replica, where in-process concurrency limits are exactly correct. Running several replicas means replacing implementations, not rewriting jobs:

• ILimiter → a Redis-backed one, so MaxConcurrent holds across the cluster
• IJobQueue → Redis, if polling latency matters
• cancellation → a pub/sub signal instead of the store poll

Handlers and job definitions do not change.

Examples

Runnable, one file per topic, in examples/jobs/:

file	topic
01_basic.go	scheduled and manual-only jobs
02_params.go	typed, validated parameters
03_concurrency.go	limits and the three overlap policies
04_stop.go	cancellable jobs, pause, shutdown
05_logs.go	log policies, live tail, reading logs
06_replay.go	replay, retry, idempotency
07_store.go	durable runs, custom tables, reporting
08_http_admin.go	the admin API