# Abacus Outbox Process

This Lambda function implements the **Transactional Outbox Pattern**. It reliably processes events stored in the `abacus_outbox` database table and publishes them to **Amazon EventBridge**.

## Overview

**Position in System:**
```
Something writes to abacus_outbox table
  ↓
[Outbox Process Lambda]        ← You are here
  ↓
EventBridge (publishes events)
  ↓
[Downstream Event Handlers]
```

**Dual Trigger Architecture:**

- **Kafka CDC (Primary):** Real-time event processing (<1s latency)
  - Triggered by database INSERT operations via Kafka/MSK
  - Event-driven, handles new events immediately

- **Scheduled Polling (Safety Net):** Reliable fallback and retry handler
  - Scheduled EventBridge Rule (e.g., every minute)
  - Catches missed CDC events, scheduled retries, manual status updates
  - Uses pessimistic locking (`FOR UPDATE SKIP LOCKED`)

**Why Both?** Defense-in-depth strategy: CDC provides speed, polling provides reliability.

**Processing Flow:**

1. **Kafka CDC Path:** Receives Debezium CDC events → Filters CREATE operations → Processes each event → Skips already COMPLETED (idempotency)
2. **MySQL Polling Path:** Fetches events with `FOR UPDATE SKIP LOCKED` → Validates status (PENDING or retriable FAILED) → Processes each event
3. **Common Processing:** Publishes to EventBridge → Updates status to COMPLETED → On failure: marks FAILED with exponential backoff retry schedule

Both paths use the same `OutboxProcessor` with optimistic locking for safe concurrent processing.

## Performance Characteristics

- **Typical Duration:** < 5 seconds
- **Memory Requirements:** Low
- **Expected Volume:** Low (TODO: Revise after adoption)

## Incoming Events

Lambda can be triggered by two event sources:

### 1. Scheduled Trigger / Direct Invocation
Empty events trigger MySQL polling with `FOR UPDATE SKIP LOCKED`. Processes events in FIFO order by `created_at`.

```json
{}
```

### 2. Kafka/MSK CDC Event (Event-Driven)
Debezium CDC events from Kafka/MSK triggered on `abacus_outbox` table INSERTs.

```json
{
  "eventSource": "aws:kafka",
  "eventSourceArn": "arn:aws:kafka:us-east-1:123456789012:cluster/msk-cluster",
  "records": {
    "cdc.royaltyAccounting.abacusOutbox-0": [
      {
        "topic": "cdc.royaltyAccounting.abacusOutbox",
        "partition": 0,
        "offset": 123,
        "timestamp": 1234567890000,
        "timestampType": "CREATE_TIME",
        "key": "base64-encoded-key",
        "value": "base64-encoded-debezium-event",
        "headers": []
      }
    ]
  }
}
```

Decoded CDC Payload (from `value` field above):
```json
{
  "before": null,
  "after": {
    "abacus_outbox_id": 123,
    "target_type": "file_upload",
    "target_id": 456,
    "event_type": "file_upload.completed",
    "correlation_id": "uuid-123",
    "details": "{\"upload_type\":\"adjustments\"}",
    "status": "pending",
    "retry_count": 0,
    "max_retries": 3
  },
  "op": "c"
}
```
**Note:** Only CREATE operations (`op: "c"`) are processed. UPDATE/DELETE/READ operations are filtered out.

## Outgoing Events

### 1. EventBridge Published Event
Successfully processed events are published to Amazon EventBridge:

```json
{
  "Source": "abacus.outbox",
  "DetailType": "file_upload.completed",
  "Detail": {
    "metadata": {
      "correlation_id": "550e8400-e29b-41d4-a716-446655440000",
      "outbox_event_id": 123,
      "target_id": 456,
      "target_type": "file_upload"
    },
    "data": {
      "upload_type": "adjustments"
    }
  }
}
```

**Key Fields:**
- `Source`: Always `abacus.outbox` (configurable via `EVENT_SOURCE` env var)
- `DetailType`: The event type from the outbox record (e.g., `file_upload.completed`)
- `metadata.outbox_event_id`: Use for deduplication in downstream consumers
- `metadata.correlation_id`: For distributed tracing across services
- `data`: Parsed business payload from `details` column

### 2. Lambda Response
Processing statistics returned by the Lambda function:

```json
{
  "total": 10,
  "processed": 8,
  "failed": 1,
  "skipped": 1
}
```

**Field Definitions:**
- `total`: Total events received/iterated
- `processed`: Successfully published to EventBridge and marked COMPLETED
- `failed`: Failed to publish or update (will be retried with exponential backoff)
- `skipped`: Events not processable (COMPLETED status, or FAILED with retries exhausted)

## Architecture & Design

*   **Dual-Path Iterable Pattern:** `DBOutboxIterable` (MySQL polling) and `KafkaOutboxIterable` (CDC parsing) both feed into unified `OutboxProcessor`
*   **Hexagonal Architecture:** Separation between core logic (`processor.py`), data access (`connectors/`), iterables, and schemas (`schemas/`)
*   **Transactional Integrity:** Per-event commits with optimistic locking
*   **Concurrency Safety:** `FOR UPDATE SKIP LOCKED` (MySQL) and idempotency checks + optimistic locking
*   **Reliability:** Exponential backoff retry scheduling
*   **Type Safety:** Pydantic models for strict typing

## Database Dependencies

**Tables:**
- `abacus_outbox` (read/write) - Source of pending events and status tracking

**Key Fields:**
- `status` - ENUM('pending', 'failed', 'completed')
- `retry_count` - Tracks retry attempts
- `max_retries` - Maximum retry limit (default: 5)
- `next_retry_at` - Timestamp for next retry attempt
- `correlation_id` - For distributed tracing
- `details` - JSON event payload

**Key Constraints:**
- `FOR UPDATE SKIP LOCKED` enables concurrent processing without race conditions
- Events are processed in `created_at` order (FIFO within each Lambda invocation)

## Edge Cases & Error Handling

| Scenario | Behavior | Retry Strategy |
| :--- | :--- | :--- |
| **Database Connection Failure** | Raises `TransientError`. Lambda fails. | Retried by EventBridge Scheduler / Trigger. |
| **EventBridge Publish Failure** | Event marked as `failed` in DB. Retry count increments. | **Exponential Backoff:** `next_retry_at` set to future time. |
| **Retry Exhausted** | Event remains `failed` with `retry_count >= max_retries`. No further retries. | **Manual Intervention Required:** Monitor for events stuck in failed state. |
| **Empty Batch** | Returns immediately with all counts as 0. No database queries after initial fetch. | N/A |
| **Concurrent Processing** | Multiple Lambda instances use `SKIP LOCKED` to process different events safely. | N/A |
| **Missing Correlation ID** | Generates a new UUID if `correlation_id` is null/missing. | N/A |
| **Invalid Details JSON** | Logs warning, attempts to parse as JSON. If parsing fails, publishes as-is or empty object. | If publish fails, handled as EventBridge publish failure. |
| **Optimistic Lock Failure (MySQL)** | EventBridge publish succeeds but another consumer already marked event COMPLETED. | **No Duplicate:** SKIP LOCKED prevents this scenario in MySQL polling path. |
| **Kafka CDC Redelivery** | Kafka redelivers same CDC event after successful processing. Event republished before optimistic lock check. | **Duplicate Publish Possible:** Downstream consumers MUST deduplicate using `outbox_event_id` field. |

## Deployment Configurations

### Production Configuration (Dual-Trigger)

**1. Kafka/MSK CDC Trigger:** Filter for CREATE operations only

| Property | Value |
| :--- | :--- |
| **StartingPosition** | LATEST |
| **Topics** | abacus.outbox |
| **BatchSize** | 10 |
| **Filter** | {"value": {"op": ["c"]}} |

**2. EventBridge Schedule Trigger:** Every minute, no input required

| Property | Value |
| :--- | :--- |
| **Schedule** | rate(1 minute) |
| **Input** | {} |

**Why `rate(1 minute)`?**
- Aligns with ~30s first retry (exponential backoff base)
- CDC fallback within ≤60s for missed events
- Fast processing of manual status updates (≤1 min)
- Negligible cost (~43,200 invocations/month, <100ms queries)

**Alternative Schedules:**

| Schedule | Trade-off | Recommendation |
| :--- | :--- | :--- |
| `rate(1 minute)` | Optimal retry alignment, fast CDC fallback | Recommended |
| `rate(2 minutes)` | 50% fewer invocations, slower fallback | If cost-sensitive |
| `rate(5 minutes)` | 80% fewer invocations, poor retry alignment | Not recommended |

### Development/Testing Configuration

Kafka CDC is optional. Polling provides full functionality with ~30s avg (~60s max) latency.

## Environment Variables

| Environment Variable | Default | Description |
| :--- | :--- | :--- |
| `BATCH_SIZE` | `10` | Maximum number of events to fetch from MySQL. **Only applies to scheduled/manual triggers.** Kafka CDC batch size is configured separately in terraform. |
| `EVENT_SOURCE` | `abacus.outbox` | The `Source` field for published EventBridge events. |
| `MYSQL_DB_HOST` | - | Database host. |
| `MYSQL_DB_PORT` | `3306` | Database port. |
| `MYSQL_DB_NAME` | `royalty_accounting` | Database name. |
| `MYSQL_DB_USER` | - | Database user. |

## Requirements

- Python 3.13
- [uv](https://docs.astral.sh/uv/) - Fast Python package manager
- Docker (for containerized testing)

## Local Development

### Setup
1.  Copy the environment file:
    ```bash
    cp .env.shadow .env
    ```
2.  Install dependencies:
    ```bash
    make env_dev
    ```

### Running Tests
*   **Unit Tests:** `make test` (runs pytest with coverage)
*   **Linting:** `make lint` (check) or `make lint_fix` (auto-fix)
*   **Formatting:** `make format` (check) or `make format_fix` (auto-fix)
*   **Docker Tests:** `make docker_test` (runs linting and tests inside a container)

### Running Locally
Run the Lambda function handler locally as a script:
```bash
make run
```

Or run as a Docker container:
```bash
make docker_local_up
```

## Deployment

Deploys via the [Jenkins pipeline](https://pipeline.theorchard.io/job/lambda-abacus-pipeline/). A webhook triggers the deploy for changes in this directory.