> ## Documentation Index
> Fetch the complete documentation index at: https://docs-terra.withunify.org/llms.txt
> Use this file to discover all available pages before exploring further.

# Queue Architecture

> How Terra handles async operations with persistent queuing, retries, and dead-letter handling

# Queue Architecture

> When an applicant submits a form, we save it immediately. Everything else—Airtable sync, email, webhooks—happens asynchronously.

Form submissions are sacred. If Airtable is down or an email fails, the applicant shouldn't see an error. Terra uses a **persistent async queue** to decouple submission success from integration success.

***

## Why Async?

Consider what happens on submission:

1. Save submission to database
2. Sync to Airtable
3. Send confirmation email
4. Fire webhooks
5. Enrich with Plaid data

If any step fails synchronously, the applicant sees an error—even though their data was saved. This creates support tickets and anxiety.

With async processing:

```mermaid theme={null}
sequenceDiagram
    participant A as Applicant
    participant S as Server
    participant DB as Database
    participant Q as Queue
    participant I as Integrations

    A->>S: Submit form
    S->>DB: Save submission
    S->>Q: Enqueue jobs
    S->>A: Success! (~300ms)

    Note over Q,I: Async (background)
    Q->>I: Airtable sync
    Q->>I: Send email
    Q->>I: Fire webhooks
    I-->>Q: Retry on failure
```

The applicant sees success in \~300ms. Integrations process in the background.

***

## The async\_operations Table

The queue is a database table, not an external service like Redis or SQS:

```sql theme={null}
CREATE TABLE async_operations (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  operation_type TEXT NOT NULL,  -- 'webhook', 'airtable_sync', etc.
  payload JSONB NOT NULL,        -- Operation-specific data

  -- Linking
  submission_id UUID REFERENCES form_submissions(id),
  form_id UUID REFERENCES forms(id),
  workspace_id UUID REFERENCES folders(id),

  -- State
  status TEXT NOT NULL DEFAULT 'pending',
  attempts INT NOT NULL DEFAULT 0,
  max_attempts INT NOT NULL DEFAULT 3,

  -- Timing
  last_attempt_at TIMESTAMPTZ,
  next_retry_at TIMESTAMPTZ DEFAULT NOW(),
  completed_at TIMESTAMPTZ,

  -- Error tracking
  last_error TEXT,
  error_history JSONB DEFAULT '[]',
  result JSONB,

  -- Timestamps
  created_at TIMESTAMPTZ DEFAULT NOW(),
  updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- Index for queue processor
CREATE INDEX idx_pending_operations
ON async_operations(status, next_retry_at)
WHERE status IN ('pending', 'failed');
```

### Operation Types

| Type               | Purpose                       | Max Attempts |
| ------------------ | ----------------------------- | ------------ |
| `webhook`          | Fire webhook to external URL  | 5            |
| `airtable_sync`    | Sync submission to Airtable   | 3            |
| `email`            | Send email via Resend         | 3            |
| `sms`              | Send SMS via Twilio           | 3            |
| `plaid_enrichment` | Fetch Plaid verification data | 3            |
| `notification`     | Combined email + SMS          | 3            |

### Status Lifecycle

```mermaid theme={null}
stateDiagram-v2
    [*] --> pending: enqueue()
    pending --> processing: claim()
    processing --> completed: complete()
    processing --> failed: fail() + retries left
    failed --> pending: auto-retry
    failed --> dead: max retries exceeded
    dead --> pending: manual retry
```

***

## Enqueueing Operations

The `enqueueOperation` function adds jobs to the queue:

```typescript theme={null}
// src/lib/async-queue.ts

export async function enqueueOperation(
  operationType: OperationType,
  payload: Record<string, unknown>,
  options: EnqueueOptions = {}
): Promise<string | null> {
  const { data, error } = await supabaseAdmin.rpc("enqueue_async_operation", {
    p_operation_type: operationType,
    p_payload: payload,
    p_submission_id: options.submissionId || null,
    p_form_id: options.formId || null,
    p_workspace_id: options.workspaceId || null,
    p_max_attempts: options.maxAttempts || 3,
  });

  if (error) {
    console.error("[AsyncQueue] Failed to enqueue:", error);
    return null;
  }

  return data as string; // Returns operation ID
}
```

### Convenience Functions

```typescript theme={null}
// Webhook with 5 retries
await enqueueWebhook(webhookUrl, formId, submissionId, data);

// Airtable sync
await enqueueAirtableSync(formId, submissionId, data, schema, settings);

// Email + SMS notification
await enqueueNotification({
  formId,
  submissionId,
  eventType: "submission_receipt",
  email: { to: "user@example.com", subject: "...", templateParams: {...} },
  sms: { to: "+1234567890", body: "..." },
});
```

***

## Queue Processing

A cron-triggered API route processes the queue:

```typescript theme={null}
// /api/queue/process/route.ts

export async function GET() {
  // Get pending operations ready for processing
  const operations = await getPendingOperations(undefined, 10);

  for (const op of operations) {
    // Atomic claim (prevents double-processing)
    const claimed = await claimOperation(op.id);
    if (!claimed) continue;

    try {
      // Execute based on type
      switch (op.operation_type) {
        case "webhook":
          await processWebhook(op);
          break;
        case "airtable_sync":
          await processAirtableSync(op);
          break;
        case "email":
          await processEmail(op);
          break;
        // ...
      }

      // Mark complete
      await completeOperation(op.id, { processedAt: new Date().toISOString() });
    } catch (error) {
      // Mark failed (will retry if under max attempts)
      await failOperation(op.id, error.message);
    }
  }

  return Response.json({ processed: operations.length });
}
```

### Claiming Operations

The `claim_async_operation` function uses atomic updates to prevent double-processing:

```sql theme={null}
-- Stored procedure
CREATE OR REPLACE FUNCTION claim_async_operation(p_operation_id UUID)
RETURNS BOOLEAN AS $$
DECLARE
  v_claimed BOOLEAN;
BEGIN
  UPDATE async_operations
  SET status = 'processing',
      last_attempt_at = NOW(),
      attempts = attempts + 1,
      updated_at = NOW()
  WHERE id = p_operation_id
    AND status IN ('pending', 'failed')
    AND (next_retry_at IS NULL OR next_retry_at <= NOW())
  RETURNING TRUE INTO v_claimed;

  RETURN COALESCE(v_claimed, FALSE);
END;
$$ LANGUAGE plpgsql;
```

This ensures only one processor handles each operation, even with concurrent workers.

***

## Retry Strategy

Failed operations retry with exponential backoff:

```typescript theme={null}
// Calculated in fail_async_operation procedure
const backoffMinutes = Math.pow(2, attempts) * 5; // 5, 10, 20, 40 minutes...
const nextRetryAt = new Date(Date.now() + backoffMinutes * 60 * 1000);
```

| Attempt | Backoff | Total Wait  |
| ------- | ------- | ----------- |
| 1       | 5 min   | 5 min       |
| 2       | 10 min  | 15 min      |
| 3       | 20 min  | 35 min      |
| 4       | 40 min  | 75 min      |
| 5       | 80 min  | \~2.5 hours |

After `max_attempts`, the operation moves to `dead` status.

### Error History

Each failure is recorded:

```json theme={null}
{
  "error_history": [
    { "error": "Connection timeout", "timestamp": "2025-01-02T10:30:00Z", "attempt": 1 },
    { "error": "Rate limited", "timestamp": "2025-01-02T10:35:00Z", "attempt": 2 },
    { "error": "Rate limited", "timestamp": "2025-01-02T10:45:00Z", "attempt": 3 }
  ]
}
```

This helps debug persistent failures.

***

## Dead-Letter Handling

Operations that fail all retries need manual attention:

```typescript theme={null}
// Get dead operations
const deadOps = await getDeadLetterOperations(50);

// In admin UI, display for review
for (const op of deadOps) {
  console.log(`${op.operation_type}: ${op.last_error}`);
  console.log(`Attempts: ${op.attempts}, Created: ${op.created_at}`);
}

// Manual retry after fixing the issue
await retryOperation(operationId);

// Or cancel if it's no longer needed
await cancelOperation(operationId);
```

***

## Monitoring

### Queue Status View

A database view summarizes queue health:

```sql theme={null}
CREATE VIEW async_operations_status AS
SELECT
  operation_type,
  status,
  COUNT(*) as count,
  MIN(created_at) as oldest,
  MAX(created_at) as newest,
  AVG(attempts) as avg_attempts
FROM async_operations
GROUP BY operation_type, status;
```

```typescript theme={null}
// Get status summary
const status = await getQueueStatus();
// Returns: [
//   { operation_type: 'webhook', status: 'pending', count: 5, ... },
//   { operation_type: 'airtable_sync', status: 'completed', count: 142, ... },
// ]
```

### Notification Statistics

```typescript theme={null}
const stats = await getNotificationQueueStats();
// {
//   pending: 3,
//   processing: 1,
//   completed: 1250,
//   failed: 12,
//   todaysSent: 45
// }
```

***

## Operation Payloads

Each operation type has a specific payload structure:

### Webhook

```typescript theme={null}
interface WebhookPayload {
  webhookUrl: string;
  formId: string;
  submissionId: string;
  data: Record<string, unknown>;
  headers?: Record<string, string>;
}
```

### Airtable Sync

```typescript theme={null}
interface AirtableSyncPayload {
  formId: string;
  submissionId: string;
  data: Record<string, unknown>;
  schema: Record<string, unknown>;
  airtableSettings: {
    base_id: string;
    table_name: string;
    api_key: string;
  };
}
```

### Notification

```typescript theme={null}
interface NotificationPayload {
  formId: string;
  submissionId: string;
  eventType: "submission_receipt" | "status_change" | "reminder";
  email?: {
    to: string;
    subject: string;
    templateParams: { applicantName: string; programName: string; ... };
    from?: string;
    replyTo?: string;
  };
  sms?: {
    to: string;
    body: string;
  };
  metadata?: Record<string, unknown>;
}
```

***

## Why Database, Not Redis/SQS?

We use PostgreSQL instead of dedicated queue services because:

1. **Transactional consistency** — Enqueue in same transaction as submission
2. **No extra infrastructure** — One less service to manage
3. **Easy querying** — SQL for debugging and monitoring
4. **Persistence** — Survives restarts without configuration
5. **ACID guarantees** — Claims are atomic, no lost messages

For our volume (thousands of submissions/day), PostgreSQL handles it easily.

***

## Edge Cases

### Duplicate Prevention

Webhook idempotency keys prevent double-delivery:

```typescript theme={null}
const idempotencyKey = `${formId}-${submissionId}-${webhookId}`;
// External system can use this to deduplicate
```

### Long-Running Operations

Some operations (like large Airtable syncs) may timeout. The processor sets a maximum execution time and fails gracefully:

```typescript theme={null}
const controller = new AbortController();
const timeout = setTimeout(() => controller.abort(), 30000); // 30s timeout

try {
  await processWithTimeout(operation, controller.signal);
} finally {
  clearTimeout(timeout);
}
```

### Stuck Processing

If a processor crashes mid-operation, the operation stays in `processing` status. A cleanup job resets stale processing operations:

```sql theme={null}
-- Reset operations stuck in processing for > 5 minutes
UPDATE async_operations
SET status = 'pending',
    next_retry_at = NOW()
WHERE status = 'processing'
  AND last_attempt_at < NOW() - INTERVAL '5 minutes';
```

***

## Related

<CardGroup cols={2}>
  <Card title="Webhooks" icon="webhook" href="/core-systems/async/webhooks">
    Webhook delivery and HMAC signing
  </Card>

  <Card title="Background Jobs" icon="clock" href="/core-systems/async/background-jobs">
    Cron-triggered processing
  </Card>
</CardGroup>