> ## 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.

# Accounting job queue

# Accounting Job Queue System

The accounting job queue provides a durable, PostgreSQL-backed queue for processing accounting impacts. Jobs are persisted in the database, leased atomically using `FOR UPDATE SKIP LOCKED`, and processed by dedicated Temporal workers.

## Architecture

### Components

1. **Queue Table** (`accounting_work_queue`): PostgreSQL table storing job state
2. **Queue Functions** (`@cona/core`): Functions to enqueue and manage jobs
3. **Temporal Activities**: Fetch and batch process jobs
4. **Temporal Workflow**: Continuous polling loop that processes jobs
5. **Temporal Worker**: Dedicated worker pool for the queue

### Job States

* `pending`: Job is ready to be processed
* `leased`: Job has been claimed by a worker
* `completed`: Job successfully processed
* `failed`: Job failed after max retries
* `cancelled`: Job cancelled (e.g., document deleted)

## Usage

### Enqueuing Jobs

```typescript theme={null}
import { enqueueAccountingJob } from "@cona/core";

const result = await enqueueAccountingJob({
  organizationId: "org_123",
  documentId: "doc_456",
  versionHint: BigInt(1),
  readyAt: new Date(), // Optional, defaults to now
});
```

### Document Readiness

Before enqueuing, documents must be marked as ready:

```typescript theme={null}
await tx.documents.update({
  where: { id: documentId },
  data: {
    accounting_ready_at: new Date(),
    accounting_ready_version: { increment: 1 },
  },
});
```

## Retry Behavior

Failed jobs are automatically retried with exponential backoff:

* Attempt 1: 5 minutes
* Attempt 2: 15 minutes (5 × 3¹)
* Attempt 3: 45 minutes (5 × 3²)
* Attempt 4: 2 hours 15 minutes (5 × 3³)
* Attempt 5: 6 hours (5 × 3⁴, capped)
* Attempt 6+: 6 hours (capped)

After 6 attempts, jobs are marked as `failed` and require manual intervention.

## Performance

### Batch Processing

* **Batch size**: 50 jobs per poll
* **Concurrency**: 3 jobs processed in parallel
* **Throughput**: \~22 jobs/minute average, \~50 jobs/minute peak

### Adaptive Polling

The queue worker uses intelligent polling:

* **Jobs found**: Immediately poll again (zero sleep, minimal latency)
* **No jobs found**: Sleep 10 seconds before next poll (avoid DB load)

This ensures the queue drains quickly when active while being polite to the database when idle.

## Monitoring

### Key Metrics

* **Pending Jobs**: Count of jobs in `pending` state
* **Processing Time**: Time from `ready_at` to completion
* **Failure Rate**: Percentage of jobs reaching `failed` state
* **Stale Jobs**: Jobs in `leased` state > 15 minutes

### SQL Queries

```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 leased jobs
SELECT * FROM accounting_work_queue
WHERE state = 'leased'
  AND last_attempted_at < NOW() - INTERVAL '15 minutes';

-- Processing stats
SELECT state, COUNT(*) FROM accounting_work_queue GROUP BY state;
```

## Configuration

### Environment Variables

* `ACCOUNTING_QUEUE_BATCH_SIZE`: Number of jobs to fetch per poll (default: 50)
* `ACCOUNTING_QUEUE_POLL_INTERVAL_SECONDS`: Seconds between polls when idle (default: 10)
* `ACCOUNTING_QUEUE_RETRY_BASE_SECONDS`: Base retry delay in seconds (default: 300)

### Worker Configuration

Located in `apps/temporal-workers/src/worker.ts`:

```typescript theme={null}
const accountingQueueWorker = await Worker.create({
  taskQueue: ACCOUNTING_QUEUE_TASK_QUEUE,
  maxConcurrentActivityTaskExecutions: 3,
  maxConcurrentWorkflowTaskExecutions: 5,
  // ...
});
```

## Error Handling

### Transient Errors

Jobs with transient errors (network issues, temporary database unavailability) are automatically retried with exponential backoff.

### Document State Errors

* **Document deleted**: Job cancelled immediately
* **Document not ready** (`accounting_ready_at` is null): Job rescheduled for 30 seconds
* **Version mismatch**: Job rescheduled with updated version (5 second delay)

### Permanent Errors

Jobs that fail 6 times are marked as `failed`. These require:

1. Investigation of the `last_error` field
2. Manual correction of the underlying issue
3. Resetting the job to `pending` state:

```typescript theme={null}
import { markAccountingJobPending } from "@cona/core";

await markAccountingJobPending({
  organizationId: "org_123",
  documentId: "doc_456",
  versionHint: BigInt(currentVersion),
  delaySeconds: 0,
});
```

## Stale Job Recovery

Jobs stuck in `leased` state for > 15 minutes are automatically recovered and included in the next fetch. This handles worker crashes gracefully.

The fetch query includes:

```sql theme={null}
WHERE (
  (state = 'pending' AND ready_at <= NOW())
  OR (state = 'leased' AND last_attempted_at < NOW() - INTERVAL '15 minutes')
)
```

## Troubleshooting

### Jobs Not Processing

1. Check worker is running: `ps aux | grep temporal-worker`
2. Check Temporal UI for workflow status at `https://cloud.temporal.io`
3. Check database for pending jobs count

### High Failure Rate

1. Review `last_error` field in failed jobs
2. Check database connection pool availability
3. Check for schema mismatches or missing object types
4. Verify accounting rules are configured

### Growing Queue

If the queue is growing faster than it can be processed:

1. Increase `ACCOUNTING_QUEUE_BATCH_SIZE` (current: 50)
2. Scale worker instances horizontally
3. Check for slow `createFinancialEntries` calls
4. Verify database performance

### Starting the Queue Worker

Start the workflow via Temporal CLI or UI:

```bash theme={null}
temporal workflow start \
  --task-queue accounting-queue \
  --type syncAccountingQueueWorkflow \
  --workflow-id accounting-queue-worker-production \
  --input '{"batchSize": 50, "pollIntervalSeconds": 10}'
```

## Implementation Details

### Database Schema

```prisma theme={null}
model accounting_work_queue {
  id                String               @id @default(cuid())
  created_at        DateTime             @default(now())
  updated_at        DateTime             @updatedAt
  org_id            String
  document_id       String
  version_hint      BigInt
  ready_at          DateTime             @default(now())
  state             ACCOUNTING_JOB_STATE @default(pending)
  run_count         Int                  @default(0)
  last_attempted_at DateTime?
  last_error        String?

  organization organization @relation(fields: [org_id], references: [id])
  document     documents    @relation(fields: [document_id], references: [id])

  @@unique([org_id, document_id])
  @@index([state, ready_at])
  @@index([org_id])
}
```

### Key Files

* **Schema**: `packages/database/prisma/schema.prisma`
* **Core Queue**: `packages/core/src/accounting/queue/`
* **Activities**: `packages/temporal-workflows/src/activities/accounting/`
* **Workflows**: `packages/temporal-workflows/src/workflows/accounting/`
* **Worker**: `apps/temporal-workers/src/worker.ts`

### Migration from Old System

The old system used `accounting_impact_created_at` flag-based locking. The new queue system:

* ✅ Survives worker crashes (durable in PostgreSQL)
* ✅ Supports safe concurrent processing (`FOR UPDATE SKIP LOCKED`)
* ✅ Provides observability (query queue state)
* ✅ Handles retries with exponential backoff
* ✅ Decouples document creation from accounting processing
* ✅ Enables horizontal scaling

The `accounting_impact_created_at` flag is still set for backwards compatibility but is no longer used for locking.