Building Internal Workflow Automation for Operational Teams with Go

Customer-facing automation gets most of the engineering attention in SaaS products. Internal workflows, the ones that run inside the operations team, are where actual hours get consumed every day. A mid-size company in Lebanon or Saudi Arabia may have staff spending two hours daily on repeatable manual tasks: reviewing new signups, notifying accounting of completed payments, updating order statuses, or generating daily summaries. A purpose-built internal workflow engine eliminates most of this without requiring a new code deployment for every business logic change.

What separates a workflow engine from a cron job

A cron job runs a fixed function on a schedule. A workflow engine executes a configurable sequence of steps in response to an event, with conditional branching, per-step error handling, full execution history, and the ability to modify workflows without deploying new code.

The critical difference is operational ownership. When the workflow definition is stored in the database rather than hardcoded, the operations team can modify business logic without opening a pull request. Engineers build the action library and the execution engine. Operations configures what runs and when.

This is the correct separation for internal automation in growing companies across MENA, where operations teams move quickly and engineering bandwidth is a constrained resource.

Schema for a tenant-scoped workflow system

CREATE TABLE workflow_definitions (
    id            UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id     UUID NOT NULL,
    name          TEXT NOT NULL,
    trigger_type  TEXT NOT NULL,
    definition    JSONB NOT NULL,
    is_active     BOOLEAN NOT NULL DEFAULT true,
    version       INT NOT NULL DEFAULT 1,
    created_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE workflow_runs (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    definition_id   UUID NOT NULL REFERENCES workflow_definitions(id),
    trigger_payload JSONB NOT NULL,
    status          TEXT NOT NULL DEFAULT 'running',
    current_step    INT NOT NULL DEFAULT 0,
    step_results    JSONB NOT NULL DEFAULT '[]',
    error_message   TEXT,
    started_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    completed_at    TIMESTAMPTZ
);

CREATE INDEX ON workflow_runs (definition_id, status, started_at DESC);
CREATE INDEX ON workflow_runs (status) WHERE status IN ('running', 'failed');

The step_results JSONB column records the outcome of each step including inputs, outputs, timing, and errors. When the operations team asks "what happened to order 12345," the workflow run record answers the question completely without requiring a developer.

The execution engine

type WorkflowEngine struct {
    actions map[string]ActionHandler
    db      WorkflowStore
}

type ActionHandler func(ctx context.Context, params map[string]any, payload map[string]any) error

type StepResult struct {
    StepName   string
    Status     string
    Output     map[string]any
    Error      string
    StartedAt  time.Time
    DurationMs int64
}

func (e *WorkflowEngine) Execute(ctx context.Context, runID uuid.UUID) error {
    run, def, err := e.db.LoadRunWithDefinition(ctx, runID)
    if err != nil {
        return err
    }

    steps := def.Steps
    for i := run.CurrentStep; i < len(steps); i++ {
        step := steps[i]
        start := time.Now()

        if step.Condition != nil && !evaluateCondition(step.Condition, run.TriggerPayload) {
            e.db.RecordStepResult(ctx, runID, StepResult{
                StepName: step.Name, Status: "skipped", StartedAt: start,
            })
            continue
        }

        handler, ok := e.actions[step.Action]
        if !ok {
            return fmt.Errorf("step %s: unknown action %q", step.Name, step.Action)
        }

        err := handler(ctx, step.Params, run.TriggerPayload)
        result := StepResult{
            StepName:   step.Name,
            StartedAt:  start,
            DurationMs: time.Since(start).Milliseconds(),
        }

        if err != nil {
            result.Status = "failed"
            result.Error = err.Error()
            if step.OnError == "stop" {
                e.db.RecordStepResult(ctx, runID, result)
                return e.db.MarkRunFailed(ctx, runID, i, err.Error())
            }
            // on_error: continue
        } else {
            result.Status = "completed"
        }

        e.db.RecordStepResult(ctx, runID, result)
        e.db.UpdateCurrentStep(ctx, runID, i+1)
    }

    return e.db.MarkRunComplete(ctx, runID)
}

Building an action library for MENA operations teams

The practical value of the engine depends on the action library. Actions that cover the majority of internal workflow needs:

// WhatsApp Business API notification
e.RegisterAction("whatsapp_notify", func(ctx context.Context, params, payload map[string]any) error {
    phone := resolveRef(params["phone"], payload)
    template := params["template"].(string)
    vars := resolveRefs(params["template_vars"], payload)
    return whatsappClient.SendTemplate(ctx, phone, template, vars)
})

// PostgreSQL record update
e.RegisterAction("update_record", func(ctx context.Context, params, payload map[string]any) error {
    table := params["table"].(string)
    id := resolveRef(params["id"], payload)
    fields := resolveRefs(params["fields"], payload)
    return db.UpdateRecord(ctx, table, id, fields)
})

// Outbound webhook
e.RegisterAction("webhook", func(ctx context.Context, params, payload map[string]any) error {
    url := params["url"].(string)
    body := mergeWithPayload(params["body"], payload)
    return httpPost(ctx, url, body)
})

// Email notification
e.RegisterAction("send_email", func(ctx context.Context, params, payload map[string]any) error {
    to := toStringSlice(params["to"])
    subject := renderTemplate(params["subject"].(string), payload)
    body := renderTemplate(params["template"].(string), payload)
    return emailClient.Send(ctx, to, subject, body)
})

// Slack or Teams notification
e.RegisterAction("team_notify", func(ctx context.Context, params, payload map[string]any) error {
    channel := params["channel"].(string)
    message := renderTemplate(params["message"].(string), payload)
    return slackClient.PostMessage(ctx, channel, message)
})

A sample workflow definition: new order processing

A workflow that runs automatically on every new order:

{
  "name": "New Order Processing",
  "trigger_type": "new_order",
  "steps": [
    {
      "name": "confirm_order",
      "action": "update_record",
      "params": {
        "table": "orders",
        "id": "{{trigger.order_id}}",
        "fields": { "status": "confirmed" }
      }
    },
    {
      "name": "notify_customer",
      "action": "whatsapp_notify",
      "params": {
        "phone": "{{trigger.customer_phone}}",
        "template": "order_confirmed",
        "template_vars": {
          "order_id": "{{trigger.order_id}}",
          "total": "{{trigger.total_amount}}"
        }
      },
      "on_error": "continue"
    },
    {
      "name": "alert_operations",
      "action": "send_email",
      "params": {
        "to": ["[email protected]"],
        "subject": "New Order: {{trigger.order_id}}",
        "template": "ops_new_order_summary"
      }
    }
  ]
}

The operations team can modify this definition through an admin interface without needing a developer.

Triggering workflows from application events

Workflows get triggered by application events without blocking the originating request:

func (s *OrderService) CreateOrder(ctx context.Context, req CreateOrderRequest) (*Order, error) {
    order, err := s.db.InsertOrder(ctx, req)
    if err != nil {
        return nil, err
    }

    // Non-blocking: workflow engine processes this asynchronously
    go s.workflowTrigger.Fire(context.Background(), order.TenantID, "new_order", map[string]any{
        "order_id":       order.ID,
        "customer_id":    order.CustomerID,
        "customer_phone": order.CustomerPhone,
        "total_amount":   order.TotalAmount,
    })

    return order, nil
}

The Fire call creates a workflow run record in the database and enqueues it for the worker pool. Order creation does not wait for workflow execution. If the workflow engine is temporarily unavailable, the run record is queued and processed on recovery.

Viewing workflow execution history

The step_results log on each run gives the operations team a complete execution trace:

SELECT
    wr.id,
    wd.name AS workflow_name,
    wr.status,
    wr.trigger_payload->>'order_id' AS order_id,
    wr.started_at,
    wr.completed_at,
    jsonb_array_length(wr.step_results) AS steps_completed
FROM workflow_runs wr
JOIN workflow_definitions wd ON wd.id = wr.definition_id
WHERE wr.trigger_payload->>'order_id' = '123e4567-e89b-12d3-a456-426614174000'
ORDER BY wr.started_at DESC;

This query is the answer to "what automation ran for this order" during a customer support investigation.

Key lessons from production

Store workflow definitions in the database, not in code. The operations team is the primary beneficiary. It lets them iterate on business logic without shipping code.

Capture full execution history on every run. The step_results log is the operations team's primary debugging tool.

on_error: continue for notification steps. A failed WhatsApp message should never stop an order from being confirmed. Mark the step failed in the log but keep the workflow running.

Decouple trigger from execution. Application code should not wait for workflow completion. The originating request should return as soon as the trigger is enqueued.