> ## Documentation Index
> Fetch the complete documentation index at: https://cona.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Temporal Workflow Architecture

> How Temporal orchestrates document creation from external sources and accounting impact processing

# Temporal Workflow Architecture

This document explains CONA's event-driven architecture for importing financial data and processing accounting impacts using Temporal workflows.

## Core Concept

CONA uses a **two-phase approach** to financial data processing:

1. **Ingestion Phase** - Import data from external sources (Shopify, PayPal) and create documents
2. **Accounting Phase** - Generate accounting impact (general ledger entries) for those documents

These phases are **completely decoupled** through a PostgreSQL queue, enabling independent scaling, resilient failure handling, and idempotent operations.

***

## Architecture Overview

```mermaid theme={null}
graph TB
    subgraph External["External Data Sources"]
        Shopify["Shopify API<br/>(Orders, Refunds, Payments)"]
        PayPal["PayPal CSV Upload<br/>(Transactions)"]
    end

    subgraph Temporal["Temporal Ingestion Workflows"]
        ShopifyWF["syncShopifyShopWorkflow<br/>━━━━━━━━━━━━━━━━━━<br/>1. Fetch orders/payments<br/>2. Process customers<br/>3. Process addresses<br/>4. Allocate numbers<br/>5. Create documents<br/>6. Create line items<br/>7. Set accounting_ready<br/>8. Enqueue to queue"]
        PayPalWF["syncPaypalPaymentsWorkflow<br/>━━━━━━━━━━━━━━━━━━<br/>1. Read CSV records<br/>2. Parse transactions<br/>3. Check duplicates<br/>4. Allocate numbers<br/>5. Create documents<br/>6. Create line items<br/>7. Set accounting_ready<br/>8. Enqueue to queue"]
    end

    subgraph Database["PostgreSQL Database"]
        Documents["📄 documents<br/>━━━━━━━━━<br/>• line_items<br/>• accounting_ready_at<br/>• accounting_ready_version: 1"]
        Queue["📋 accounting_work_queue<br/>━━━━━━━━━━━━━━━<br/>• document_id<br/>• version_hint: 1<br/>• state: pending"]
    end

    subgraph AccountingWF["Temporal Accounting Queue Workflow"]
        Loop["🔄 syncAccountingQueueWorkflow<br/>(infinite loop with continue-as-new)<br/>━━━━━━━━━━━━━━━━━━━━━━━━<br/>1. Poll for pending jobs (batch of 50)<br/>2. FOR UPDATE SKIP LOCKED (atomic lease)<br/>3. Process jobs concurrently (3 at a time)<br/>4. Create accounting impact (GL entries)<br/>5. Mark jobs completed<br/>6. Continue-as-new → back to step 1"]
    end

    GL["💰 general_ledger<br/>━━━━━━━━━<br/>• Debit entries<br/>• Credit entries<br/>• Reconciliation data"]

    Shopify -->|API calls| ShopifyWF
    PayPal -->|CSV upload| PayPalWF
    ShopifyWF -->|Create documents| Documents
    PayPalWF -->|Create documents| Documents
    Documents -->|Automatic trigger| Queue
    Queue -->|Poll & lease| Loop
    Loop -->|Create GL entries| GL

    style External fill:#e1f5ff
    style Temporal fill:#fff4e6
    style Database fill:#f3e5f5
    style AccountingWF fill:#e8f5e9
    style GL fill:#fce4ec
```

***

## Phase 1: Data Ingestion

### Purpose

Transform external financial data into standardized CONA documents with line items.

### How It Works

**Shopify Integration:**

* Fetches orders and payments via Shopify API
* Creates sales invoices, credit notes, and payment documents
* Processes in chunks for memory efficiency
* Handles customers, addresses, line items (products, shipping, discounts)

**PayPal Integration:**

* Reads transactions from CSV uploads
* Creates payment documents (main payment + optional fee documents)
* Tracks processing status per CSV record
* Handles duplicate detection via `paypal_payment_id`

### Key Operations

```mermaid theme={null}
flowchart LR
    Fetch[Fetch External Data] --> Process[Process Entities<br/>customers, addresses]
    Process --> Build[Build Documents<br/>with line items]
    Build --> Ready[Mark Ready<br/>accounting_ready_at=NOW<br/>version=1]
    Ready --> Enqueue[Enqueue to<br/>Accounting Queue]

    style Fetch fill:#e3f2fd
    style Process fill:#e3f2fd
    style Build fill:#f3e5f5
    style Ready fill:#fce4ec
    style Enqueue fill:#fff9c4
```

**Result:** Documents are created in the database with:

* Complete line item breakdown
* `accounting_ready_at` timestamp (signals ready for accounting)
* `accounting_ready_version = 1` (version tracking for changes)
* Corresponding job in `accounting_work_queue`

***

## Phase 2: Accounting Processing

### Purpose

Convert business documents into accounting entries (debits and credits) using posting matrix rules.

### How It Works

The `syncAccountingQueueWorkflow` runs as an **infinite loop**:

1. Polls the queue for pending jobs (batch of 50)
2. Atomically leases jobs using `FOR UPDATE SKIP LOCKED`
3. Processes 3 jobs concurrently
4. Creates general ledger entries based on posting rules
5. Marks jobs as completed
6. Uses `continueAsNew` to reset workflow history and loop forever

### Job State Machine

```mermaid theme={null}
stateDiagram-v2
    [*] --> pending: Job enqueued
    pending --> leased: Worker fetches job

    leased --> completed: Processing succeeded
    leased --> pending: Retry (transient error)
    leased --> failed: Max retries (6) exceeded
    leased --> cancelled: Document deleted
    leased --> pending: Version mismatch

    completed --> pending: Document updated<br/>(version incremented)

    failed --> pending: Manual reset

    completed --> [*]
    cancelled --> [*]
    failed --> [*]

    note right of pending
        Ready to process
        run_count: 0-6
    end note

    note right of leased
        Being processed
        Timeout: 15 min
    end note

    note right of failed
        Requires manual fix
        Check last_error
    end note
```

### Processing Flow

```mermaid theme={null}
sequenceDiagram
    participant Queue as Work Queue
    participant Worker as Accounting Worker
    participant DB as Database

    Note over Worker: Infinite loop

    Worker->>Queue: Fetch 50 pending jobs<br/>(FOR UPDATE SKIP LOCKED)
    Queue-->>Worker: Return leased jobs

    par Process 3 jobs concurrently
        Worker->>DB: Check document version

        alt Version matches
            Worker->>DB: Fetch document + line items
            Worker->>DB: Apply posting rules
            Worker->>DB: Create GL entries
            Worker->>Queue: Mark completed
        else Version mismatch
            Worker->>Queue: Reschedule with new version
        else Document deleted
            Worker->>Queue: Mark cancelled
        end
    end

    alt Jobs found
        Note over Worker: Continue immediately
    else No jobs
        Note over Worker: Sleep 10 seconds
    end

    Worker->>Worker: continueAsNew()<br/>(fresh history, loop forever)
```

***

## Version Tracking

### Why It Matters

Documents can be modified after creation (e.g., adding line items). Version tracking ensures accounting is always processed with the latest data.

### How It Works

```mermaid theme={null}
sequenceDiagram
    participant User
    participant Doc as Document
    participant Queue as Work Queue
    participant Worker

    Note over Doc,Queue: Initial Creation
    User->>Doc: Create document
    Doc->>Doc: version = 1
    Doc->>Queue: Enqueue job (version_hint: 1)

    Note over Worker: Processing
    Worker->>Queue: Fetch job (version_hint: 1)
    Worker->>Doc: Check version
    Doc-->>Worker: version = 1 ✓
    Worker->>Doc: Create GL entries
    Worker->>Queue: Mark completed

    Note over User,Doc: User modifies document
    User->>Doc: Add line item
    Doc->>Doc: version = 2 (increment!)
    Doc->>Queue: Re-enqueue job (version_hint: 2)

    Note over Worker: Reprocessing
    Worker->>Queue: Fetch job (version_hint: 2)
    Worker->>Doc: Check version
    Doc-->>Worker: version = 2 ✓
    Worker->>Doc: Delete old entries<br/>Create new entries
    Worker->>Queue: Mark completed

    rect rgb(255, 240, 240)
        Note over Worker,Queue: Version Mismatch Scenario
        Note over Worker: Worker has old version_hint
        Worker->>Doc: Check version
        Doc-->>Worker: version = 3 ✗ MISMATCH!
        Worker->>Queue: Reschedule with version_hint: 3
    end
```

**Key Points:**

* Documents start at version 1
* Adding line items to a document with accounting increments the version
* Workers check version before processing
* Mismatches cause automatic rescheduling with the correct version

***

## Key Design Patterns

### 1. Idempotency

Every operation can be safely retried without side effects:

* Duplicate documents detected by external IDs (`shopify_order_id`, `paypal_payment_id`)
* Jobs use unique constraint: one job per document per organization
* Accounting entries can be deleted and recreated

### 2. Atomic Operations

All state changes happen within database transactions:

* Document + line items created atomically
* Job state changes within transactions
* GL entries created atomically per document

### 3. Decoupled Phases

Ingestion and accounting are completely independent:

* Ingestion workflows can run without accounting being ready
* Accounting queue can catch up asynchronously
* Failures in one phase don't block the other

### 4. Resilient Failure Handling

**Transient errors:** Automatic retry with exponential backoff (5min, 15min, 45min, 2h, 6h)
**Permanent errors:** Jobs marked as `failed` after 6 attempts, require manual investigation
**Worker crashes:** Stale jobs (leased > 15min) automatically recovered

***

## Configuration

### Ingestion Workers

* **Shopify:** 10 concurrent activities, 20 concurrent workflows
* **PayPal:** 10 concurrent activities, 20 concurrent workflows
* Optimized for high-throughput batch processing

### Accounting Queue Worker

* **Concurrency:** 3 activities, 5 workflows
* **Batch size:** 50 jobs per poll
* **Poll interval:** 10 seconds when idle, immediate when active
* Optimized for consistent, controlled processing

### Performance Characteristics

* **Shopify sync:** \~250 orders/minute
* **PayPal CSV:** \~100 transactions/minute
* **Accounting queue:** \~22 jobs/minute average, \~50 jobs/minute peak

***

## Monitoring

### Quick Health Checks

```sql theme={null}
-- Pending jobs count
SELECT COUNT(*) FROM accounting_work_queue WHERE state = 'pending';

-- Failed jobs requiring attention
SELECT * FROM accounting_work_queue
WHERE state = 'failed'
ORDER BY updated_at DESC;

-- Stale jobs (potential worker crash)
SELECT * FROM accounting_work_queue
WHERE state = 'leased'
  AND last_attempted_at < NOW() - INTERVAL '15 minutes';
```

### Temporal UI

* View active workflow executions at `https://cloud.temporal.io`
* Track workflow history and activity retries
* Monitor continue-as-new chains
* Debug activity failures

***

## When Things Go Wrong

### Common Issues

**Queue growing faster than processing:**

1. Check worker is running
2. Increase batch size or concurrency
3. Scale horizontally (add more workers)

**Jobs stuck in `failed` state:**

1. Check `last_error` field for error message
2. Fix underlying issue (missing rules, bad data)
3. Reset job to `pending` state

**Documents not getting accounting impact:**

1. Check `accounting_ready_at` is set
2. Verify job exists in `accounting_work_queue`
3. Check posting rules exist for document type