# Adjustment File Prepare

Lambda function that validates adjustment files via DuckDB, extends rows with validation results, and loads data to MySQL staging for downstream processing.

## Overview

Triggered by EventBridge after batch initialization.

**Position in Workflow:**
```
EventBridge
  ↓ file_upload.completed
Adjustment File Initialize
  ↓ adjustment_batch.initialized
[Adjustment File Prepare]        ← You are here
  ↓ adjustment_batch.prepared
Adjustment File Process Batch
  ↓
Adjustment File Complete
```

**Workflow Steps:**
1. **Validate Event** - Parse and validate EventBridge event structure
2. **Update Status** - Set batch status to `VALIDATING`
3. **Download File** - Retrieve from S3 with checksum verification
4. **Load to DuckDB** - Auto-detect format (CSV/XLSX/Parquet/gzip) and load
5. **Validate Structure** - Check headers, normalize columns, enforce row limit (1M rows)
6. **Load References** - Import reference tables from Snowflake/MySQL into DuckDB
7. **Validate Business Rules** - Execute SQL validation for accounts, contracts, UPCs, periods, amounts, currencies
8. **Enrich Data** - Add batch_id, row_id, and validation_errors columns
9. **Export to S3** - Upload validated CSV to `staging/{batch_id}/prepared.csv`
10. **Stage to MySQL** - Load prepared data into `staging_adjustment_detail`
11. **Return Results** - Provide validation counts and financial totals

## Event Schema

**Input Event:**
```json
{
  "detail-type": "adjustment_batch.initialized",
  "detail": {
    "metadata": {
      "correlation_id": "550e8400-e29b-41d4-a716-446655440000",
      "target_id": 456,
      "target_type": "worksheet_flowthrough_batch"
    },
    "data": {
      "s3_bucket": "qa-abacus-adjustments",
      "s3_key": "uploads/2025/11/adjustment_file.xlsx"
    }
  }
}
```

**Output Response:**
```json
{
  "detail_type": "adjustment_batch.prepared",
  "detail": {
    "metadata": {
      "correlation_id": "550e8400-e29b-41d4-a716-446655440000",
      "target_id": 456,
      "target_type": "worksheet_flowthrough_batch"
    },
    "data": {
      "valid_row_count": 149850,
      "invalid_row_count": 150,
      "total_file_amount_multicurrency": 1234567.89,
      "total_rounded_amount_multicurrency": 1234568.00,
      "s3_bucket": "qa-abacus-adjustments",
      "s3_key": "staging/456/prepared.csv"
    }
  }
}
```

## Performance Characteristics

**File Limits:**

- **Maximum file size:** 1 GB
- **Maximum row count:** 1 million rows

> **TODO:** Explore limit increases. Linear forecasting predicts:
> - Row count: 8M rows
> - File size: ~1.5GB (XLSX) / ~4.36GB (CSV)
> - Processing time: 10-11.5 minutes
> - Acceptance Criteria: Must test with real-world and/or worst-case data

**Lambda Duration (XLSX files):**

> **TODO:** Update benchmarks with QA testing data using real-world / worst-case files.

| Rows | Size (MB) | Time |
|:--- | :--- | :--- |
| 50,000 | 12.77 | 53.40s |
| 100,000 | 22.25 | 54.28s |
| 250,000 | 50.62 | 1m10.74s |
| 500,000 | 97.90 | 1m30.74s |
| 1,000,000 | 192.47 | 2m11.42s |

| Stage | 50K rows (warm) | 1M rows (cold) | 1M rows (warm) |
|-------|-----------------|----------------|----------------|
| Downloading file * | 0.149s (0.3%) | 1.369s (0.9%) | 1.372s (1.1%) |
| Loading file | 1.412s (2.9%) | 22.867s (15.8%) | 22.016s (17.3%) |
| Loading references * | 8.767s (17.8%) | 22.474s (15.5%) | 6.371s (5.0%) |
| Validating file | 7.887s (16.1%) | 11.794s (8.1%) | 10.524s (8.3%) |
| Generating results | 0.087s (0.2%) | 0.406s (0.3%) | 0.428s (0.3%) |
| Uploading results * | 30.306s (61.7%) | 1m13.212s (50.4%) | 1m13.092s (57.3%) |
| Staging results * | 0.441s (0.9%) | 12.993s (9.0%) | 13.607s (10.7%) |
| Cleanup | 0.091s (0.2%) | 0.053s (0.0%) | 0.052s (0.0%) |
| **Total** | **49.14s** | **2m25.17s** | **2m7.46s** |

*Notes:*
1. *Stages marked with asterisk involve network I/O (S3 or Snowflake operations). Performance may vary based on database and network conditions.*

1. *"Warm" and "cold" refer to Snowflake warehouse state*

    *Warm warehouses are already running, while cold warehouses must resume from suspended state, significantly impacting reference data loading*

1. *Times are from local testing: personal machine, LocalStack, loading into local Docker S3 / DB.*

1. *Test files > 50K rows were created by duplicating the same 50K rows, which result in optimistic performance (duplicate reference data IDs, reducing lookup data and validation time). Real-world files with more unique data may show different performance.*

## Architecture

### Component Design

The application follows a modular structure separating concerns into distinct layers:

*   **Processor (`AdjustmentFilePrepareProcessor`)**
    *   The main orchestrator coordinating the pipeline stages: downloading, loading, validation, and staging.
    *   Manages the transaction lifecycle, batch status updates, and error handling (distinguishing between transient and permanent errors).

*   **Loaders**
    *   **File Loader (`AdjustmentFileLoader`)**: Handles ingestion of raw adjustment files (CSV, XLSX, Parquet) into DuckDB. It performs format detection, schema normalization, and enforces file-level constraints.
    *   **Reference Loader (`ReferenceLoader`)**: Orchestrates the loading of reference data from Snowflake into DuckDB. It uses a **Task Graph (DAG)** to execute independent tasks in parallel and manage task dependencies (e.g., creating optimized lookup tables only after raw data is loaded).

*   **Gateways (`SnowflakeGateway`)**
    *   Abstracts data retrieval from Snowflake with intelligent loading strategies to optimize performance:
        *   **Bulk Load**: Used for small reference tables or when filtering is inefficient.
        *   **Batch Load**: Used for large tables. It iterates through unique IDs from the input file in batches and fetches only relevant records from Snowflake.
        *   **Stage Load**: Used for large tables. It uploads unique IDs from the input file to a temporary Snowflake stage to efficiently filter and fetch only relevant records.

*   **Validator (`DuckDBValidator`)**
    *   Executes comprehensive business logic validation using high-performance SQL queries within DuckDB.
    *   Checks include: account existence, contract validity, UPC formats, currency codes, and financial constraints.
    *   Produces aggregated results (valid/invalid counts, financial totals)

*   **Repositories (`RoyaltyAccountingClient`)**
    *   Manages state in the MySQL database.
    *   Handles batch status transitions using **optimistic locking** to prevent race conditions.
    *   Stages the final validated data using `LOAD DATA` commands (Local or S3-based depending on environment).

### Key Architectural Patterns

*   **Hexagonal Architecture**: Separation of core business logic from external infrastructure (S3, Databases), allowing for easy testing and swapping of components.
*   **DuckDB Processing**: Uses an ephemeral, in-memory SQL engine for complex data joins and validation. This avoids the latency of round-trips to remote databases for individual row validation.
*   **Type Safety**: Extensive use of Pydantic models for strict validation of event inputs and response outputs.
*   **Idempotency**: Deterministic S3 naming and atomic database transactions enable safe retries.
*   **Environment-Adaptability**: Automatically switches staging strategies (`LOAD DATA LOCAL INFILE` for Dev vs `LOAD DATA FROM S3` for Prod) to optimize for local development speed and production security.

## Database Dependencies

**Source Databases:**
- **`art_relations`** - UPC mappings
- **`royalty_accounting`** - Reference and staging data

**Reference Tables (read-only):**
`account`, `account_contract`, `account_payment_term`, `account_upcs`, `ref_adjustment_type`, `close_balance_status`, `contract_term`, `currency_code`, `statement_period`

**Target Tables:**
- **`worksheet_flowthrough_batch`** (read/write) - Batch status tracking with optimistic locking
- **`staging_adjustment_detail`** (write) - Validated CSV data (valid + invalid rows)

## File Requirements

**Required Headers:**
- Account ID *
- Activity Month *
- Activity Year *
- Adjustment Type *
- Amount *
- Client Facing Comments *
- Contract ID *
- Currency *
- Statement Month *
- Statement Year *

**Optional Headers:**
- Apply to Flowthrough
- Distribution Type
- Internal Note
- UPC

## Error Handling

| Error Type | Retry | Batch Status |
| :--- | :--- | :--- |
| **S3/DB Connection Failure** | Retriable | Unchanged (Lambda retries) |
| **File Not Found** | Non-Retriable | `ERROR` (S3_OBJECT_NOT_FOUND) |
| **Invalid File Type** | Non-Retriable | `ERROR` (INVALID_FILE_TYPE) |
| **Missing Headers** | Non-Retriable | `ERROR` (MISSING_HEADERS) |
| **Empty File** | Non-Retriable | `ERROR` (EMPTY_FILE) |
| **Row Count Exceeded** | Non-Retriable | `ERROR` (ROW_COUNT_EXCEEDED) |
| **Parsing Error** | Non-Retriable | `ERROR` (FILE_PARSING_ERROR) |

Non-retriable errors update batch status to `ERROR` with specific error code before re-raising.

## Requirements

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

## Local Development

### MySQL Configuration

For local development, enable `LOAD DATA LOCAL INFILE` in MySQL:

```sql
-- View current setting
SHOW VARIABLES LIKE "local_infile";

-- Enable for local file loading
SET GLOBAL local_infile = 'ON';
```

### Development vs Production

| Environment | Staging Method | Benefit |
|------------|----------------|---------|
| **DEV** | `LOAD DATA LOCAL INFILE` | No S3 re-download needed |
| **Production** | `LOAD DATA FROM S3` | Direct Aurora MySQL loading |

### 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)
*   **Integration Tests:** `make docker_test_local_integration` (runs integration tests in Docker with full service dependencies)

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