Platform Foundation & Workflow — Business Automation Architecture Research

Rule: Authorization Required
┌──────────────┬──────────────┬──────────────┬──────────┐
│ Payer Type   │ Service Line │ Visit Count  │ Result   │
├──────────────┼──────────────┼──────────────┼──────────┤
│ Medicaid     │ IOP          │ any          │ Required │
│ Medicaid     │ PHP          │ any          │ Required │
│ Commercial   │ IOP          │ > 10         │ Required │
│ Commercial   │ Residential  │ > 30         │ Required │
│ Self-Pay     │ any          │ any          │ None     │
└──────────────┴──────────────┴──────────────┴──────────┘

┌─────────────────────────────────────────────────────────────────────┐
│                        EVENT SOURCES                                │
├─────────────┬──────────────┬──────────────┬────────────────────────┤
│ Form Submit │ Domain Event │ Schedule     │ External Webhook       │
│ (FW-01/02)  │ (FW-16)      │ (pg_cron)    │ (PF-35)               │
└──────┬──────┴──────┬───────┴──────┬───────┴────────────┬──────────┘
       │             │              │                    │
       ▼             ▼              ▼                    ▼
┌─────────────────────────────────────────────────────────────────────┐
│                   fw_domain_events (Outbox Table)                   │
│  id | event_name | organization_id | payload | processed_at        │
└─────────────────────────────┬───────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────────┐
│            fw_process_domain_event() (DB Trigger)                   │
│  - Match fw_automation_rules by event_config                       │
│  - INSERT into fw_workflow_executions (status = 'queued')           │
│  - Match pf_event_subscriptions for external delivery              │
└─────────────────────────────┬───────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────────┐
│         fw_workflow_executions (Durable State Store)                │
│  id | rule_id | definition_id | status | current_step_id           │
│  trigger_payload | retry_count | next_retry_at | deadline_at       │
│  correlation_id | organization_id | created_at | updated_at        │
└─────────────────────────────┬───────────────────────────────────────┘
                              │
              ┌───────────────┤ (pg_cron every 10s → pg_net)
              ▼               │
┌─────────────────────────────┴───────────────────────────────────────┐
│              workflow-executor-worker (Edge Function)                │
│                                                                     │
│  1. SELECT executions WHERE status IN ('queued','retry_pending')    │
│     AND next_retry_at <= now() LIMIT 10 FOR UPDATE SKIP LOCKED     │
│                                                                     │
│  2. For each execution:                                             │
│     a. SET status = 'running', lock row                            │
│     b. Load workflow definition (nodes, edges)                      │
│     c. Resume from current_step_id (or start)                      │
│     d. Execute step:                                                │
│        - Action node → run action (email, webhook, update, etc.)   │
│        - Condition node → evaluate, choose branch                  │
│        - Approval node → create task, SET status = 'waiting'       │
│        - Delay node → SET next_retry_at = now() + delay            │
│        - Sub-workflow → create child execution                     │
│     e. Checkpoint: UPDATE current_step_id, write execution_step    │
│     f. On success → next step or status = 'completed'              │
│     g. On failure → apply retry policy or move to DLQ              │
│                                                                     │
│  3. Emit execution events for monitoring (FW-22)                    │
└─────────────────────────────────────────────────────────────────────┘

// workflow-executor-worker Edge Function
async function processQueuedExecutions(organizationId?: string) {
  // 1. Claim batch of queued executions (row-level locking prevents double-processing)
  const executions = await supabase.rpc('claim_queued_executions', {
    batch_size: 10,
    org_filter: organizationId
  });

  for (const execution of executions) {
    try {
      // 2. Load workflow definition
      const definition = await loadWorkflowDefinition(execution.definition_id);

      // 3. Resume from checkpoint
      const startStep = execution.current_step_id
        ? getNextStep(definition, execution.current_step_id)
        : getStartNode(definition);

      // 4. Execute steps until completion, wait, or failure
      let currentStep = startStep;
      while (currentStep) {
        const result = await executeStep(currentStep, execution);

        // 5. Checkpoint after each step
        await checkpointStep(execution.id, currentStep.id, result);

        if (result.status === 'waiting') break;      // Human-in-the-loop
        if (result.status === 'delayed') break;       // Timer wait
        if (result.status === 'failed') throw result.error;

        currentStep = getNextStep(definition, currentStep.id, result);
      }

      // 6. Mark complete if no more steps
      if (!currentStep) {
        await completeExecution(execution.id);
      }

    } catch (error) {
      // 7. Apply retry policy
      await handleExecutionFailure(execution, error);
    }
  }
}

-- Atomic claim of queued executions (prevents double-processing)
CREATE OR REPLACE FUNCTION claim_queued_executions(
  batch_size INTEGER DEFAULT 10,
  org_filter UUID DEFAULT NULL
)
RETURNS SETOF fw_workflow_executions AS $$
BEGIN
  RETURN QUERY
  UPDATE fw_workflow_executions
  SET status = 'running', updated_at = now()
  WHERE id IN (
    SELECT id FROM fw_workflow_executions
    WHERE status IN ('queued', 'retry_pending')
      AND (next_retry_at IS NULL OR next_retry_at <= now())
      AND (deadline_at IS NULL OR deadline_at > now())
      AND (org_filter IS NULL OR organization_id = org_filter)
    ORDER BY created_at ASC
    LIMIT batch_size
    FOR UPDATE SKIP LOCKED
  )
  RETURNING *;
END;
$$ LANGUAGE plpgsql;

function handleExecutionFailure(execution, error):
  rule = loadAutomationRule(execution.rule_id)
  retryPolicy = rule.retry_config ?? DEFAULT_RETRY_POLICY

  if execution.retry_count >= retryPolicy.max_retries:
    // Move to Dead Letter Queue
    insertDLQ(execution, error)
    updateExecution(execution.id, { status: 'failed' })
    notifyOwner(execution, 'workflow_failed')
    // Run compensation actions if defined
    if rule.compensation_actions:
      scheduleCompensation(execution, rule.compensation_actions)
  else:
    // Schedule retry with exponential backoff
    delay = retryPolicy.base_delay * (2 ^ execution.retry_count)
    delay = min(delay, retryPolicy.max_delay)
    updateExecution(execution.id, {
      status: 'retry_pending',
      retry_count: execution.retry_count + 1,
      next_retry_at: now() + delay,
      last_error: error.message
    })

┌─────────────────────────────────────────────────────────────────┐
│                    FW-17 Condition Builder                       │
│                                                                 │
│  ┌─────────────┐  ┌──────────────┐  ┌─────────────────────┐   │
│  │ Simple       │  │ Expression   │  │ Decision Table      │   │
│  │ Conditions   │  │ Builder      │  │ (NEW)               │   │
│  │ field=value  │  │ AND/OR/NOT   │  │ Multi-input/output  │   │
│  │ field>value  │  │ nested       │  │ DMN hit policies    │   │
│  └─────────────┘  └──────────────┘  └─────────────────────┘   │
│                                                                 │
│  All three evaluate to: { match: boolean, outputs: Record }     │
└─────────────────────────────────────────────────────────────────┘

                    ┌─────────────────────────────┐
                    │     Event Publishers         │
                    │  publishEvent() from any core│
                    └──────────────┬──────────────┘
                                   │
                                   ▼
┌──────────────────────────────────────────────────────────────────┐
│                    fw_domain_events (Event Store)                 │
│                                                                  │
│  Consumers:                                                      │
│  ┌──────────────────┐ ┌───────────────────┐ ┌────────────────┐  │
│  │ Automation Rules  │ │ External Webhooks │ │ Event Consumer │  │
│  │ (DB Trigger)      │ │ (PF-35 Subs)     │ │ (HTTP Handlers)│  │
│  │                   │ │                   │ │                │  │
│  │ fw_automation_    │ │ pf_event_         │ │ Teams notifs,  │  │
│  │ rules matching    │ │ subscriptions     │ │ CL transitions │  │
│  └────────┬─────────┘ └────────┬──────────┘ └───────┬────────┘  │
│           │                    │                     │           │
│           ▼                    ▼                     ▼           │
│  fw_workflow_executions   Outbound webhooks    Side effects      │
│  (queued → worker)        (with retry/DLQ)     (HTTP handlers)  │
└──────────────────────────────────────────────────────────────────┘
                                   │
                          On failure after retries
                                   │
                                   ▼
┌──────────────────────────────────────────────────────────────────┐
│                    fw_dead_letter_queue                           │
│                                                                  │
│  ┌──────────────────────────────────────────────────────────┐   │
│  │ Admin UI: Review → Retry / Discard / Investigate         │   │
│  │ Alerting: Notify ops when DLQ depth > threshold          │   │
│  │ Metrics:  DLQ depth, age, resolution rate                │   │
│  └──────────────────────────────────────────────────────────┘   │
└──────────────────────────────────────────────────────────────────┘

fw_workflow_events (enhanced):
  id, event_name, display_name, description, core,
  payload_schema (JSONB - JSON Schema),
  schema_version (INTEGER),
  deprecated_at (TIMESTAMPTZ),
  replacement_event (TEXT - event_name of successor),
  created_at, updated_at

Evolution rules:
  - Adding optional fields: same version (backward compatible)
  - Adding required fields: new version + migration period
  - Removing fields: deprecate first, remove in next major version
  - Renaming fields: add new field + deprecate old (never rename in-place)

CE (Referral) → PM (Scheduling) → CL (Assessment) → RH (Bed Assignment) → FA (Insurance Verify)
     │              │                   │                  │                    │
     ▼              ▼                   ▼                  ▼                    ▼
  Lead created   Intake appt      ASAM/clinical      Room assigned       Benefits verified
  Referral       scheduled        assessment          Lease agreement     Authorization
  acknowledged                    completed           signed              obtained

CL (Treatment Plan) → PM (Review Scheduling) → CL (Review Session) → FA (Authorization)
         │                    │                       │                      │
         ▼                    ▼                       ▼                      ▼
   Plan created with    Review scheduled         Review completed      Re-authorization
   30/60/90 day         for review date          with updates          submitted if
   review dates                                                         required

CL (Discharge Plan) → RH (Room Release) → FA (Final Billing) → CE (Aftercare)
        │                   │                    │                    │
        ▼                   ▼                    ▼                    ▼
  Discharge plan       Room marked          Final charges        Aftercare referrals
  approved by          for turnover         generated            sent
  clinical director    Move-out date set    Insurance billed     Follow-up scheduled

HR (Hire) → HR (Credentialing) → GR (Compliance Check) → PM (Provider Setup)
     │            │                      │                      │
     ▼            ▼                      ▼                      ▼
  Employee      Licenses verified    Background check       Provider profile
  created       Certifications       completed              created in
                tracked              Training assigned       scheduling system

PM (Encounter) → FA (Charge Capture) → FA (Claim Submit) → FA (Payment Post)
       │               │                      │                    │
       ▼               ▼                      ▼                    ▼
  Encounter       Charges generated     Claim submitted       Payment received
  completed       from fee schedule     to clearinghouse      ERA posted
                  (auto or manual)                            Reconciled

┌─────────────────────────────────────────────────────────────────────┐
│                        EVENT SOURCES                                │
├─────────────┬──────────────┬──────────────┬────────────────────────┤
│ Form Submit │ Domain Event │ Schedule     │ External Webhook       │
└──────┬──────┴──────┬───────┴──────┬───────┴────────────┬──────────┘
       │             │              │                    │
       ▼             ▼              ▼                    ▼
┌─────────────────────────────────────────────────────────────────────┐
│            fw_process_domain_event() (DB Trigger)                   │
│  - Match fw_automation_rules by event_config                       │
│  - pgmq.send('workflow_execution_queue', execution_payload)        │
│  - Match pf_event_subscriptions for external delivery              │
└─────────────────────────────┬───────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────────┐
│         pgmq: workflow_execution_queue (Durable Queue)             │
│  Messages: { execution_id, rule_id, trigger_payload, org_id }      │
│  Visibility timeout: 120 seconds                                   │
│  Archive: enabled (compliance audit trail)                         │
└─────────────────────────────┬───────────────────────────────────────┘
                              │
              ┌───────────────┤ (pg_cron every 10s → pg_net → Edge Function)
              ▼               │
┌─────────────────────────────┴───────────────────────────────────────┐
│              workflow-executor-worker (Edge Function)                │
│                                                                     │
│  1. pgmq.read('workflow_execution_queue', 10, 120) -- batch of 10  │
│                                                                     │
│  2. For each message:                                               │
│     a. Load execution from fw_workflow_executions                   │
│     b. Load workflow definition                                     │
│     c. Resume from current_step_id (or start)                      │
│     d. Execute step (action, condition, approval, delay, subflow)  │
│     e. Checkpoint: UPDATE current_step_id, write execution_step    │
│     f. On success → pgmq.delete() (or archive for audit)          │
│        If more steps → pgmq.send() new message for next step      │
│     g. On transient failure → let visibility timeout expire (auto  │
│        retry)                                                       │
│     h. On permanent failure → pgmq.send('workflow_dlq', msg)      │
│                                                                     │
│  3. Emit execution events for monitoring (FW-22)                    │
└─────────────────────────────────────────────────────────────────────┘

-- Create queues (run in migration)
SELECT pgmq.create('workflow_execution_queue');   -- Main execution queue
SELECT pgmq.create('workflow_dlq');               -- Dead letter queue
SELECT pgmq.create('event_forwarding_queue');     -- External webhook delivery
SELECT pgmq.create('notification_queue');         -- Async notification delivery

-- pg_cron job to invoke worker
SELECT cron.schedule(
  'process-workflow-queue',
  '10 seconds',
  $$SELECT net.http_post(
    url := current_setting('app.supabase_url') || '/functions/v1/workflow-executor-worker',
    headers := jsonb_build_object(
      'Authorization', 'Bearer ' || current_setting('app.service_role_key'),
      'Content-Type', 'application/json'
    ),
    body := '{"queue": "workflow_execution_queue", "batch_size": 10}'::jsonb
  )$$
);

{
  "name": "Authorization Requirements",
  "version": 2,
  "hitPolicy": "first",
  "rules": [
    {
      "conditions": {
        "all": [
          { "fact": "payer_category", "operator": "equal", "value": "medicaid" },
          { "fact": "service_line", "operator": "in", "value": ["residential", "php"] }
        ]
      },
      "event": {
        "type": "authorization_required",
        "params": { "requires_preauth": true, "review_interval_days": 7 }
      }
    },
    {
      "conditions": {
        "all": [
          { "fact": "payer_category", "operator": "equal", "value": "self_pay" }
        ]
      },
      "event": {
        "type": "no_authorization",
        "params": { "requires_preauth": false }
      }
    }
  ]
}

// In workflow-executor-worker or automation-executor
import { Engine } from 'json-rules-engine';

async function evaluateDecisionTable(
  tableId: string,
  facts: Record<string, unknown>,
  organizationId: string
): Promise<{ matched: boolean; outputs: Record<string, unknown>[] }> {
  const { data: table } = await supabase
    .from('fw_decision_tables')
    .select('rules, hit_policy')
    .eq('id', tableId)
    .eq('organization_id', organizationId)
    .single();

  const engine = new Engine(table.rules);
  const results = await engine.run(facts);

  // Apply hit policy
  if (table.hit_policy === 'first') {
    return { matched: results.events.length > 0, outputs: results.events.slice(0, 1) };
  }
  return { matched: results.events.length > 0, outputs: results.events };
}

CREATE TABLE fw_rule_evaluations (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  organization_id UUID NOT NULL,
  decision_table_id UUID REFERENCES fw_decision_tables,
  rule_version INTEGER NOT NULL,
  input_facts JSONB NOT NULL,
  matched_rules JSONB NOT NULL,
  output_results JSONB NOT NULL,
  evaluation_context TEXT, -- 'workflow_step', 'form_validation', 'api_request'
  source_execution_id UUID, -- link to fw_workflow_executions if triggered by workflow
  evaluated_by UUID, -- user who triggered the evaluation
  evaluated_at TIMESTAMPTZ DEFAULT now()
);

// Persist: Save XState snapshot to Postgres
const snapshot = actor.getPersistedSnapshot(); // Recursively serializes actor tree
await supabase.from('fw_workflow_instances').upsert({
  id: executionId,
  machine_snapshot: snapshot, // JSONB column
  updated_at: new Date().toISOString()
});

// Restore: Resume from persisted snapshot
const restoredSnapshot = (await supabase
  .from('fw_workflow_instances')
  .select('machine_snapshot')
  .eq('id', executionId)
  .single()).data.machine_snapshot;

const actor = createActor(workflowMachine, { snapshot: restoredSnapshot });
actor.start(); // Resumes exactly where it left off