Tasks and Job Runs

// package tasks - a leaf package that imports fluxaorm and nothing generated.
package tasks

import "github.com/latolukasz/fluxaorm/v2"

type SendWelcomeEmail struct {
    UserID uint64
    Locale string
}

// Optional: pick the queue. Without Queue() the task lands on fluxaorm.DefaultQueue ("default").
func (SendWelcomeEmail) Queue() fluxaorm.Queue { return "emails" }

type Queued interface {
    Queue() Queue
}

type Subjected interface {
    Subject() Subject
}

func TaskNameFor[T any]() TaskName // TaskNameFor[tasks.SendWelcomeEmail]() == "SendWelcomeEmail"

registry.RegisterTask(tasks.SendWelcomeEmail{}, fluxaorm.TaskOptions{})
registry.RegisterTask(tasks.TranscodeClip{}, fluxaorm.TaskOptions{MaxAttempts: 3, BaseBackoff: 30 * time.Second})
registry.RegisterEntity(fluxaorm.JobRunEntity{}) // required as soon as one task is registered

type TaskOptions struct {
    MaxAttempts int
    BaseBackoff time.Duration
    MaxBackoff  time.Duration
}

registry.RegisterTaskStream(fluxaorm.TaskStreamOptions{NatsPool: "events", MaxBytes: 1 << 30})

registry.RegisterConsumer(fluxaorm.ConsumerDef{Name: "emails", Queues: []fluxaorm.Queue{"emails"}})
registry.RegisterConsumer(fluxaorm.ConsumerDef{Name: "default-worker", Queues: []fluxaorm.Queue{fluxaorm.DefaultQueue}})

// DispatchSendWelcomeEmail publishes a SendWelcomeEmail task onto the "emails" queue
// and returns the ID of its job_runs row.
func DispatchSendWelcomeEmail(
    orm fluxaorm.Context,
    task *tasks.SendWelcomeEmail,
    opts ...fluxaorm.DispatchOption,
) (uint64, error)

func (b *EmailsBuilder) OnSendWelcomeEmail(
    handler func(ctx fluxaorm.Context, task *tasks.SendWelcomeEmail) error,
) *EmailsBuilder

func DispatchTask[T any](ctx Context, task *T, opts ...DispatchOption) (uint64, error)
func WithIdempotencyKey(key string) DispatchOption

var ErrDispatchInTransaction = errors.New(
    "task dispatch inside a transaction: the publish cannot roll back with it, so dispatch after commit")

u := entities.UserEntityProvider.New(ctx)
u.SetEmail("ann@example.com").SetName("Ann")
if err := ctx.Transaction(func(tx fluxaorm.Context) error { return tx.Save(u) }); err != nil {
    return err
}

// After the commit, never inside it.
runID, err := entities.DispatchSendWelcomeEmail(ctx,
    &tasks.SendWelcomeEmail{UserID: u.GetID(), Locale: "en"},
    fluxaorm.WithIdempotencyKey(fmt.Sprintf("welcome:%d", u.GetID())),
)

emails := fluxaorm.NewConsumer(engine, entities.ConsumerEmails).
    OnSendWelcomeEmail(func(ctx fluxaorm.Context, task *tasks.SendWelcomeEmail) error {
        return mailer.SendWelcome(ctx, task.UserID, task.Locale)
    }).
    Build() // panics if any task on "emails" has no handler

type JobRunEntity struct {
    ID         uint64     `orm:"table=job_runs"`
    Task       string     `orm:"required;length=100"`
    Queue      string     `orm:"required;length=50"`
    Status     string     `orm:"enum=pending,running,succeeded,failed,deduplicated;enumName=JobRunStatus;required"`
    Payload    string     `orm:"length=max;required"`
    Attempts   uint8      `orm:"required"`
    LastError  string     `orm:"length=max;required"`
    DurationMs uint32     `orm:"required"`
    StartedAt  *time.Time `orm:"time"`
    FinishedAt *time.Time `orm:"time"`
    CreatedAt  time.Time  `orm:"time"`
}
// Indexes: {Status, CreatedAt} and {Task, CreatedAt}

run, err := entities.JobRunsProvider.MustGetByID(ctx, runID)
run.GetTask()       // string
run.GetQueue()      // string
run.GetStatus()     // enums.JobRunStatus, compare with enums.JobRunStatusList.Succeeded etc.
run.GetPayload()    // string
run.GetAttempts()   // uint64
run.GetLastError()  // string
run.GetDurationMs() // uint64
run.GetStartedAt()  // *time.Time
run.GetFinishedAt() // *time.Time
run.GetCreatedAt()  // time.Time

failed, err := entities.JobRunsProvider.SearchMany(ctx, fluxaorm.NewQuery().
    Filter(entities.JobRunsProvider.Fields.Status.Is(enums.JobRunStatusList.Failed)).
    SortByDESC(entities.JobRunsProvider.Fields.CreatedAt).
    Pager(fluxaorm.NewPager(1, 50)))

func LoadJobRun(ctx Context, id uint64) (task, queue, status, payload string, err error)
func MarkJobRunStarted(ctx Context, id uint64, attempt uint8) error
func MarkJobRunFinished(ctx Context, id uint64, res JobRunResult) error
func RedispatchJobRun(ctx Context, runID uint64) (uint64, error)
func PurgeJobRuns(ctx Context, status string, olderThan time.Duration, limit int) (int, error)
func JobRunBacklog(ctx Context) (depth int, oldest time.Duration, err error)

type JobRunResult struct {
    Status   string        // a terminal status, or JobRunPending to hand the run back
    Err      string
    Duration time.Duration
}

// Cron, e.g. every minute.
depth, oldest, err := fluxaorm.JobRunBacklog(ctx)
if err == nil && depth > 0 && oldest > 15*time.Minute {
    alert("job runs stranded", depth, oldest)
}
pending, err := fluxaorm.ConsumerPending(ctx, entities.ConsumerEmails.Name()) // queue depth on JetStream

// Cron, e.g. hourly: keep the table bounded, one terminal status at a time.
for _, status := range []string{fluxaorm.JobRunSucceeded, fluxaorm.JobRunDeduplicated} {
    if _, err := fluxaorm.PurgeJobRuns(ctx, status, 7*24*time.Hour, 5000); err != nil {
        log.Printf("purge %s: %v", status, err)
    }
}
if _, err := fluxaorm.PurgeJobRuns(ctx, fluxaorm.JobRunFailed, 30*24*time.Hour, 5000); err != nil {
    log.Printf("purge failed: %v", err)
}

package main

import (
    "context"
    "fmt"
    "log"
    "os/signal"
    "syscall"
    "time"

    "github.com/latolukasz/fluxaorm/v2"

    "myapp/entities"
    "myapp/tasks"
)

type UserEntity struct {
    ID    uint64 `orm:"cdc"`
    Email string `orm:"required;length=255"`
    Name  string `orm:"required;length=100"`
}

func main() {
    registry := fluxaorm.NewRegistry()
    registry.RegisterMySQL("user:password@tcp(localhost:3306)/app", fluxaorm.DefaultPoolCode, &fluxaorm.MySQLOptions{})
    registry.RegisterRedis("localhost:6379", 0, fluxaorm.DefaultPoolCode, nil)
    registry.RegisterNats([]string{"nats://localhost:4222"}, fluxaorm.DefaultPoolCode, nil)
    registry.RegisterEntity(UserEntity{}, fluxaorm.JobRunEntity{})
    registry.RegisterTask(tasks.SendWelcomeEmail{}, fluxaorm.TaskOptions{MaxAttempts: 3})
    registry.RegisterConsumer(fluxaorm.ConsumerDef{Name: "user-indexer", Entities: []any{UserEntity{}}})
    registry.RegisterConsumer(fluxaorm.ConsumerDef{Name: "emails", Queues: []fluxaorm.Queue{"emails"}, AckWait: time.Minute})

    engine, err := registry.Validate()
    if err != nil {
        panic(err)
    }
    defer engine.Nats(fluxaorm.DefaultPoolCode).Close()

    appCtx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
    defer stop()
    ctx := engine.NewContext(appCtx)

    // Tables (job_runs included), then streams and durables.
    alters, err := fluxaorm.GetAlters(ctx)
    if err != nil {
        panic(err)
    }
    for _, alter := range alters {
        if err := alter.Exec(ctx); err != nil {
            panic(err)
        }
    }
    natsAlters, err := fluxaorm.GetNatsAlters(ctx)
    if err != nil {
        panic(err)
    }
    for _, alter := range natsAlters {
        if err := alter.Exec(ctx); err != nil {
            panic(err)
        }
    }

    emails := fluxaorm.NewConsumer(engine, entities.ConsumerEmails).
        OnSendWelcomeEmail(func(ctx fluxaorm.Context, task *tasks.SendWelcomeEmail) error {
            u, found, err := entities.UserEntityProvider.GetByID(ctx, task.UserID)
            if err != nil {
                return err // retried with backoff, up to MaxAttempts
            }
            if !found {
                return nil // nothing to do; ack
            }
            log.Printf("sending welcome email to %s in %s", u.GetEmail(), task.Locale)
            return nil
        }).
        Build()

    go func() {
        for appCtx.Err() == nil {
            if err := emails.Consume(appCtx, 10, time.Second); err != nil && appCtx.Err() == nil {
                log.Printf("emails consumer: %v", err)
                time.Sleep(time.Second)
            }
        }
    }()

    // Write in a transaction, dispatch after the commit.
    u := entities.UserEntityProvider.New(ctx)
    u.SetEmail("ann@example.com").SetName("Ann")
    if err := ctx.Transaction(func(tx fluxaorm.Context) error { return tx.Save(u) }); err != nil {
        panic(err)
    }
    runID, err := entities.DispatchSendWelcomeEmail(ctx,
        &tasks.SendWelcomeEmail{UserID: u.GetID(), Locale: "en"},
        fluxaorm.WithIdempotencyKey(fmt.Sprintf("welcome:%d", u.GetID())))
    if err != nil {
        log.Printf("dispatch failed, run %d stays pending: %v", runID, err)
    }

    // Maintenance cron.
    go func() {
        ticker := time.NewTicker(time.Minute)
        defer ticker.Stop()
        for {
            select {
            case <-appCtx.Done():
                return
            case <-ticker.C:
                cronCtx := engine.NewContext(appCtx)
                if depth, oldest, err := fluxaorm.JobRunBacklog(cronCtx); err == nil && depth > 0 {
                    log.Printf("job run backlog: %d pending, oldest %s", depth, oldest)
                }
                if _, err := fluxaorm.PurgeJobRuns(cronCtx, fluxaorm.JobRunSucceeded, 7*24*time.Hour, 5000); err != nil {
                    log.Printf("purge job runs: %v", err)
                }
            }
        }
    }()

    <-appCtx.Done()
}