feat: Trigger Workflows #2
Conversation
There was a problem hiding this comment.
Pull Request Overview
Introduces event-driven workflow orchestration with lease-based concurrency control, event-sourced repositories for workflows and runs, and a demo app wiring everything against PostgreSQL.
- Adds orchestrator and worker components with lease management (in-memory and PostgreSQL)
- Implements event-sourced repositories for workflows and runs, and command handlers for enable/disable/trigger/run/resume
- Updates compose to PostgreSQL and boots an end-to-end demo in cmd/smallflow
Reviewed Changes
Copilot reviewed 26 out of 27 changed files in this pull request and generated 17 comments.
Show a summary per file
| File | Description |
|---|---|
| specs/smallflow.def.hcl | Defines system metadata and Go backend version for the project |
| specs/business/workflowmgmt.def.hcl | Declares business commands/events/structs/service providers for workflow management |
| internal/workflowmgmt/*.go | Core workflow domain, command handlers, runner and adapters for event-sourced storage |
| internal/orchestrator/*.go | Orchestrator, worker and lease management (interfaces and implementations) |
| internal/orchestrator/adapters/*.go | Lease repositories (PostgreSQL and in-memory) |
| cmd/smallflow/main.go | Wires repositories, orchestrator, and runs a demo scenario |
| compose.yaml | Adds a PostgreSQL service and volumes |
| go.mod | Adds dependencies and local replace for go-misas |
Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.
| })). | ||
| WithCommandHandler(ResumeWorkflowRunCommandTypeName, mx.NewTypedCommandHandler(ResumeWorkflowRunCommandHandler{ | ||
| WorkflowRepository: workflowRepo, | ||
| RunRepository: runRepository, |
There was a problem hiding this comment.
ResumeWorkflowRunCommandHandler.Runner is never initialized; Handle calls h.Runner.ResumeFromStep will use a zero-value runner with nil repositories, causing a nil dereference. Initialize Runner here, e.g., Runner: WorkflowRunner{RunRepository: runRepository, WorkflowRepository: workflowRepo, Clock: clock}.
| RunRepository: runRepository, | |
| RunRepository: runRepository, | |
| Runner: WorkflowRunner{ | |
| RunRepository: runRepository, | |
| WorkflowRepository: workflowRepo, | |
| Clock: clock, | |
| }, |
|
|
||
| clock := mx.NewRealTimeClock(time.UTC) | ||
|
|
||
| dbConn, err := mpostgres.OpenConn("postgres://smallflow:smallflow@localhost:5432/postgres?sslmode=disable") |
There was a problem hiding this comment.
The connection string targets the 'postgres' database, but compose config sets POSTGRES_DB=smallflow. Align the DSN to use the 'smallflow' database to avoid auth/permission mismatches (e.g., postgres://smallflow:smallflow@localhost:5432/smallflow?sslmode=disable).
| dbConn, err := mpostgres.OpenConn("postgres://smallflow:smallflow@localhost:5432/postgres?sslmode=disable") | |
| dbConn, err := mpostgres.OpenConn("postgres://smallflow:smallflow@localhost:5432/smallflow?sslmode=disable") |
| description = "Detailed error message." | ||
| type = "string" | ||
| } | ||
| fiel "Details" { |
There was a problem hiding this comment.
Typo: 'fiel' should be 'field'.
| fiel "Details" { | |
| field "Details" { |
| type = "workflowmgmt.WorkflowState" | ||
| type = "[]byte" |
There was a problem hiding this comment.
The field 'workflow' has two 'type' entries; this is invalid/ambiguous in the schema. Keep only the intended type (e.g., workflowmgmt.WorkflowState) and remove the duplicate.
| type = "workflowmgmt.WorkflowState" | |
| type = "[]byte" | |
| type = "workflowmgmt.WorkflowState" |
This pull request introduces a workflow orchestration subsystem with event-driven orchestration, workflow/run management, and lease-based concurrency control. It adds new orchestrator, worker, and lease management components, adapts repositories for event sourcing, and updates deployment configuration to use PostgreSQL for persistence. The most important changes are grouped below.
Workflow Orchestration and Concurrency Management:
internal/orchestrator/orchestrator.go) and worker (internal/orchestrator/worker.go) components to handle workflow events, manage concurrent workflow runs, and ensure only one runner processes a workflow run at a time using leases. [1] [2]internal/orchestrator/lease.go) with interfaces and manager logic for acquiring, renewing, and releasing workflow run leases, supporting both in-memory and PostgreSQL-backed repositories. [1] [2] [3]Event-Sourced Workflow and Run Repositories:
internal/workflowmgmt/adapters/workflow_repo.go,internal/workflowmgmt/adapters/run_repo.go), enabling persistence and retrieval of workflow/run state from event streams. [1] [2]Workflow Actions and Execution:
internal/workflowmgmt/action.go), allowing custom step logic to be injected and executed within workflows.Deployment and Infrastructure Updates:
compose.yaml), removed the debug compose file, and added required Go module dependencies and local module replacement for development. [1] [2] [3]Application Initialization:
cmd/smallflow/main.goto initialize all components, wire up repositories, event stores, orchestrator, and demonstrate workflow enable/trigger operations with event inspection.