workflow

README

togo-framework/workflow

Part of the togo framework.

Install

togo install togo-framework/workflow

workflow — togo workflow engine

Run dynamic, multi-step workflows over the togo worker/queue. A workflow is an ordered list of steps; each step has a type that maps to a registered handler, so any plugin can be used as a workflow step (send a mail, charge a payment, call an AI model, sync contacts…). Steps run asynchronously through the kernel Queue, so long pipelines never block a request.

togo install togo-framework/workflow
togo install togo-framework/queue     # the runtime (memory/redis/db)

Register step types (from any plugin)

import "github.com/togo-framework/workflow"

func init() {
    workflow.RegisterStep("mail.send", func(ctx context.Context, sc *workflow.StepContext, with map[string]any) error {
        m, _ := mail.FromKernel(sc.Kernel)
        return m.Send(ctx, mail.Message{To: []string{with["to"].(string)}, Subject: with["subject"].(string)})
    })
}

Define + trigger a workflow

wf, _ := workflow.FromKernel(k)
wf.Define(workflow.Workflow{Name: "welcome", Steps: []workflow.Step{
    {Type: "mail.send", With: map[string]any{"to": "{{user.email}}", "subject": "Welcome!"}},
    {Type: "contacts.sync"},
}})

run, _ := wf.Trigger(ctx, "welcome", map[string]any{"user": user})
// run.Status: pending → running → done | failed; steps execute over the queue.

A step can set sc.Next to jump/branch (or -1 to end early) and read/write sc.Run.State shared across the run. With no queue installed, steps run inline.

MIT

---

Premium sponsors

ID8 Media  ·  One Studio

_{Support togo — become a sponsor.}

Documentation

Overview

Package workflow runs dynamic, multi-step workflows over the togo queue.

A workflow is an ordered list of steps; each step has a TYPE that maps to a registered handler (StepFunc). Any plugin can register step types via RegisterStep, so a workflow uses other plugins as providers (send a mail, charge a payment, call an AI model, …). Steps run asynchronously through the kernel Queue, so long pipelines never block a request; shared State is passed between steps in a run.

Index

• Constants
• func RegisterStep(typ string, fn StepFunc)
• type Run
• type Service
  • func FromKernel(k *togo.Kernel) (*Service, bool)
  • func (s *Service) Define(wf Workflow)
  • func (s *Service) Definitions() []string
  • func (s *Service) GetRun(id string) (*Run, bool)
  • func (s *Service) Trigger(ctx context.Context, name string, input map[string]any) (*Run, error)
• type Step
• type StepContext
• type StepFunc
• type Workflow

Constants

const (
	StatusPending = "pending"
	StatusRunning = "running"
	StatusDone    = "done"
	StatusFailed  = "failed"
)

Status values for a Run.

Functions

func RegisterStep

func RegisterStep(typ string, fn StepFunc)

RegisterStep registers a step type → handler (call from init() or a plugin's Boot). Other plugins register step types so workflows can use them.

Types

type Run

type Run struct {
	ID       string         `json:"id"`
	Workflow string         `json:"workflow"`
	Step     int            `json:"step"`
	Status   string         `json:"status"`
	State    map[string]any `json:"state"`
	Error    string         `json:"error,omitempty"`
	Started  time.Time      `json:"started"`
}

Run is a single execution of a workflow.

type Service

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

Service is the workflow runtime stored on the kernel (k.Get("workflow")).

func FromKernel

func FromKernel(k *togo.Kernel) (*Service, bool)

FromKernel fetches the workflow service from the kernel container.

func (*Service) Define

func (s *Service) Define(wf Workflow)

Define registers (or replaces) a workflow definition.

func (*Service) Definitions

func (s *Service) Definitions() []string

Definitions lists the registered workflow names.

func (*Service) GetRun

func (s *Service) GetRun(id string) (*Run, bool)

Run returns a run by id.

func (*Service) Trigger

func (s *Service) Trigger(ctx context.Context, name string, input map[string]any) (*Run, error)

Trigger starts a run of a workflow with the given initial State.

type Step

type Step struct {
	Type string         `json:"type"`           // a registered step type
	Name string         `json:"name,omitempty"` // optional label
	With map[string]any `json:"with,omitempty"` // step parameters
}

Step is one node in a workflow.

type StepContext

type StepContext struct {
	Kernel *togo.Kernel
	Run    *Run
	// Next overrides the next step index; -1 ends the run early. Zero value
	// (unset) advances to the following step.
	Next *int
}

StepContext is passed to a step handler.

type StepFunc

type StepFunc func(ctx context.Context, sc *StepContext, with map[string]any) error

StepFunc executes a step. Returning an error fails the run.

type Workflow

type Workflow struct {
	Name  string `json:"name"`
	Steps []Step `json:"steps"`
}

Workflow is an ordered pipeline of steps. Branching is expressed by step types that mutate State / set the next index via StepContext.