workflows

Documentation

Overview

Package workflows provides the WorkflowScheduler for cron-based session automation.

Index

• func RunRetentionSweep(ctx context.Context, entClient *ent.Client, ...)
• func StartRetentionEnforcer(ctx context.Context, entClient *ent.Client, ...)
• func ValidateCronExpression(expr string) error
• type Scheduler
  • func NewScheduler(repo session.WorkflowRepository, sessionSvc SessionServiceInterface, ...) *Scheduler
  • func (s *Scheduler) FireNow(ctx context.Context, wf *ent.Workflow, arg string) (string, error)
  • func (s *Scheduler) Reload(ctx context.Context, wf *ent.Workflow) error
  • func (s *Scheduler) Remove(workflowID string) error
  • func (s *Scheduler) Start(ctx context.Context)
  • func (s *Scheduler) Stop()
• type SessionServiceInterface

Functions

func RunRetentionSweep

func RunRetentionSweep(ctx context.Context, entClient *ent.Client, workflowRepo session.WorkflowRepository)

RunRetentionSweep performs a single retention sweep. Exported for use in tests.

func StartRetentionEnforcer

func StartRetentionEnforcer(
	ctx context.Context,
	entClient *ent.Client,
	workflowRepo session.WorkflowRepository,
	interval time.Duration,
)

StartRetentionEnforcer starts a background goroutine that periodically archives completed workflow sessions according to per-workflow retention settings:

• archive_after_hours > 0: archive completed sessions that stopped more than N hours ago (requires maybeAutoArchive to be suppressed for these workflows)
• keep_sessions > 0: keep only the N most recent completed sessions, archiving older ones

Guards:

• Never archives sessions with status Active (1), Creating (0), or Paused (2)
• archive_after_hours == 0 means disabled (skip time-based archival for that workflow)
• keep_sessions == 0 means disabled (keep all sessions)

The goroutine exits when ctx is cancelled.

func ValidateCronExpression

func ValidateCronExpression(expr string) error

ValidateCronExpression validates a 5-field cron expression. Exported so workflow_service.go can use it without importing the cron library directly.

Types

type Scheduler

type Scheduler struct {
	// contains filtered or unexported fields
}

Scheduler manages cron-based workflow execution.

func NewScheduler

func NewScheduler(repo session.WorkflowRepository, sessionSvc SessionServiceInterface, eventBus *events.EventBus) *Scheduler

NewScheduler creates a new WorkflowScheduler.

func (*Scheduler) FireNow

func (s *Scheduler) FireNow(ctx context.Context, wf *ent.Workflow, arg string) (string, error)

FireNow immediately fires a workflow outside of cron schedule. Returns the created session ID. Used by RunWorkflow RPC and internal cron trigger.

func (*Scheduler) Reload

func (s *Scheduler) Reload(ctx context.Context, wf *ent.Workflow) error

Reload registers or re-registers a workflow's cron job. Called after create/update. If cron_enabled is false, removes any existing entry.

func (*Scheduler) Remove

func (s *Scheduler) Remove(workflowID string) error

Remove removes a workflow's cron job by workflow ID string. Safe to call when no entry exists (no-op).

func (*Scheduler) Start

func (s *Scheduler) Start(ctx context.Context)

Start loads all enabled workflows and begins cron processing. Stops when ctx is cancelled.

func (*Scheduler) Stop

func (s *Scheduler) Stop()

Stop halts the cron engine. Called as a shutdown hook.

type SessionServiceInterface

type SessionServiceInterface interface {
	CreateSession(ctx context.Context, req *connect.Request[sessionv1.CreateSessionRequest]) (*connect.Response[sessionv1.CreateSessionResponse], error)
}

SessionServiceInterface is the minimal interface the scheduler needs from SessionService. Defined here to avoid a circular import: server/workflows does not import server/services.