# TRD: Flowthrough Payment Automation & Paythrough Contract Deprecation

**Owner:** Michael Rojas
**Epics:** [ACC-9366](https://theorchard.atlassian.net/browse/ACC-9366), [ACC-9335](https://theorchard.atlassian.net/browse/ACC-9335)
**Last Updated:** 2026-05-13

---

## Table of Contents

1. [System Architecture](#1-system-architecture)
2. [Entity Relationship Diagrams](#2-entity-relationship-diagrams)
3. [Data Flow: Manual Adjustment Upload](#3-data-flow-manual-adjustment-upload)
4. [Data Flow: Auto-Generated Flowthrough Batch](#4-data-flow-auto-generated-flowthrough-batch)
5. [Data Flow: Payment Allocation](#5-data-flow-payment-allocation)
6. [Component Details](#6-component-details)
7. [Infrastructure](#7-infrastructure)
8. [Feature Flags](#8-feature-flags)
9. [API Reference](#9-api-reference)
10. [Security and Authorization](#10-security-and-authorization)

---

## 1. System Architecture

### High-Level Architecture

```mermaid
flowchart TD
    FE["Frontend\n(React/Vite)"]
    GQL["GraphQL Gateway\n(BFF)"]
    API["ows-royalties\n(Flask API)"]
    EVT["ows-abacus-event\n(Event Router)"]
    S3["S3: abacus-adjustments"]
    DB["MySQL: royalty_accounting"]
    OUTBOX["Outbox Processor\n(Lambda + Kafka CDC)"]
    EB["EventBridge"]
    SF_UPLOAD["Step Function:\nfile-upload-workflow"]
    SF_INGEST["Step Function:\nadjustment-file-ingest-workflow"]
    AIRFLOW["Airflow DAGs\n(MWAA)"]
    LAMBDAS["Lambda Functions\n(6 active)"]
    OWS["OWS Services\n(ledger, worksheet, state)"]

    FE -->|GraphQL| GQL -->|REST| API
    FE -->|Signed URL| S3
    API -->|Read/Write| DB
    API -->|Write event| DB
    DB -->|CDC / abacus_outbox| OUTBOX
    OUTBOX -->|Publish| EB
    EB -->|file_upload.completed| SF_UPLOAD
    SF_UPLOAD -->|Status updates| DB
    EB -->|upload_type: adjustments| SF_INGEST
    SF_INGEST -->|InitializeBatch| LAMBDAS
    LAMBDAS -->|Create abacus event| EVT
    EVT -->|MWAA CLI trigger| AIRFLOW
    AIRFLOW -->|Invoke| LAMBDAS
    LAMBDAS -->|Direct SQL| DB
    LAMBDAS -->|Service calls| OWS
    EB -->|close_balance.completed| LAMBDAS
```

### Component Overview

| Component | Technology | Repo | Purpose |
|-----------|-----------|------|---------|
| Frontend | React + Apollo GraphQL | `frontend-royalties` | Upload UI, batch management, status tracking |
| Backend API | Flask + SQLAlchemy | `ows-royalties` | REST endpoints, models, business logic |
| GraphQL Gateway | GraphQL (BFF) | `frontend-royalties` | Bridges frontend to backend REST APIs |
| Event Router | Flask | `ows-abacus-event` | Receives abacus events, maps to DAG IDs, triggers Airflow via MWAA CLI |
| Step Functions | AWS Step Functions | `terraform-infra` | File upload lifecycle + adjustment ingest initialization |
| Lambdas | Python (containerized) | `lambda-abacus` | File processing (init, validate, import, apply, allocate) |
| Airflow DAGs | Apache Airflow (MWAA) | `ows-royalties-workflows` | Orchestration for validation, import, and apply flows |
| Outbox Processor | Lambda | `lambda-abacus` | Reads `abacus_outbox` table (via Kafka CDC or schedule), publishes to EventBridge |
| Storage | S3 | `terraform-infra` | Adjustment file storage (source + error reports) |
| Database | Aurora MySQL | `ows-royalties` | Primary data store (`royalty_accounting`) |
| Event Bus | EventBridge | `terraform-infra` | Event-driven triggers (file upload complete, balance close) |
| Kafka CDC | MSK + CDC Connector | `terraform-infra` | Watches `abacus_outbox` table for new rows, triggers outbox processor |
| Feature Flags | Split.io | N/A | Feature gating for phased rollout |
| Monitoring | Datadog + CloudWatch | `terraform-infra` | Alerting, logs, metrics |

---

## 2. Entity Relationship Diagrams

### 2.1 Core Adjustment Tables

```mermaid
erDiagram
    statement_period {
        int statement_period_id PK
        string status "open|current|closed"
        date start_date
        date end_date
    }

    statement_period_adjustment_file {
        int statement_period_adjustment_file_id PK
        int statement_period_id FK
        int source_file_upload_id FK "nullable"
        string file_name "varchar(180)"
        enum batch_type "auto|manual|upload"
        string valid_file_location "S3 URL"
        string invalid_file_location "S3 URL"
        int valid_row_count
        int invalid_row_count
        decimal total_file_amount_multicurrency "25,12"
        decimal total_rounded_amount_multicurrency "20,2"
        string md5sum "varchar(35)"
        enum error_type "format_error|content_error|row_count_error|null"
        datetime deleted_at "soft delete"
        string deleted_by "soft delete"
        datetime created_at
        string created_by
    }

    statement_period_adjustment_batch_criteria {
        int statement_period_adjustment_batch_criteria_id PK
        int statement_period_adjustment_file_id FK
        json batch_criteria "payment_schedules, reference_payment_entities"
    }

    worksheet_adjustment {
        int worksheet_adjustment_id PK
        int statement_period_adjustment_file_id FK
        int abacus_event_id FK
        int account_id
        int contract_id
        int activity_statement_period_id FK
        int apply_to_statement_period_id FK
        int reference_adjustment_type_id FK
        decimal adjustment_amount
        string adjustment_currency_code
        string note
        string internal_note
        boolean apply_to_flowthrough_payment
        string created_by "lambda-abacus-adjustment-file-import"
    }

    worksheet_adjustment_detail {
        int worksheet_adjustment_detail_id PK
        int statement_period_adjustment_file_id FK
        int worksheet_adjustment_id FK
        int account_id
        int contract_id
        int activity_statement_period_id FK
        int apply_to_statement_period_id FK
        int reference_adjustment_type_id FK
        string currency_code
        decimal amount
        string upc
        string distribution_type
        string note
        string internal_note
        boolean apply_to_flowthrough_payment
    }

    statement_period ||--o{ statement_period_adjustment_file : "has many"
    statement_period_adjustment_file ||--o| statement_period_adjustment_batch_criteria : "has one"
    statement_period_adjustment_file ||--o{ worksheet_adjustment : "contains"
    worksheet_adjustment ||--o{ worksheet_adjustment_detail : "has details"
    statement_period_adjustment_file ||--o{ worksheet_adjustment_detail : "contains"
```

### 2.2 File Upload Tables

```mermaid
erDiagram
    file_upload_config {
        int file_upload_config_id PK
        string upload_type "adjustments|flowthrough"
        string s3_bucket
        string s3_key_template
        string event_name "nullable - triggers Lambda/DAG"
        int max_file_size_bytes
        int multipart_threshold_bytes
        datetime deleted_at "soft delete"
        string deleted_by
    }

    file_upload {
        int file_upload_id PK
        int file_upload_config_id FK
        string upload_type "adjustments|flowthrough"
        enum upload_status "init|scanning|complete|error|cancelled|quarantined"
        string s3_bucket
        string s3_key
        string original_filename
        string md5sum
        int file_size_bytes
        string asset_type "XLSX"
        datetime created_at
        string created_by
    }

    abacus_outbox {
        int abacus_outbox_id PK
        string event_type
        string event_source "abacus.outbox"
        json event_detail
        string status "pending|published|failed"
        datetime created_at
        datetime published_at
    }

    file_upload_config ||--o{ file_upload : "configures"
    file_upload ||--o| statement_period_adjustment_file : "source for"
    file_upload ||--o| abacus_outbox : "triggers event"
```

### 2.3 Flowthrough Contract Configuration

```mermaid
erDiagram
    contract {
        int contract_id PK
        string contract_name
        boolean is_paythrough_contract "DEPRECATED - to be removed"
        int account_id FK
    }

    contract_flowthrough {
        int contract_flowthrough_id PK
        int contract_id FK
        int reference_flowthrough_calculation_id FK
        decimal flowthrough_rate "5,2 - percentage"
        enum flowthrough_status "active|shutoff|paused"
        boolean has_automatic_shutoff "deprecated, default true"
        int recoupment_cap "nullable"
        text calculation_comment "MEDIUMTEXT"
        enum previous_flowthrough_status "active|shutoff|paused"
        string status_last_modified_by "varchar(180)"
        datetime status_last_modified
        datetime deleted_at "soft delete"
        string deleted_by
    }

    reference_flowthrough_calculation {
        int reference_flowthrough_calculation_id PK
        string flowthrough_calculation_name "varchar(180)"
        string flowthrough_calculation "varchar(180) - formula"
        string flowthrough_calculation_example "varchar(180)"
        string flowthrough_calculation_example_summary "varchar(180)"
    }

    contract ||--o| contract_flowthrough : "has flowthrough config"
    reference_flowthrough_calculation ||--o{ contract_flowthrough : "defines calculation"
```

### 2.4 DEPRECATED Tables (To Be Torn Down)

> **DEPRECATED:** The following tables were created during ACC-9366 (tickets ACC-9375, ACC-9376, ACC-9377, ACC-9378) but are **not used** by any Lambda, API, or frontend component. The `adjustment-file-initialize` Lambda creates a `worksheet_flowthrough_batch` record, but nothing downstream reads or updates it. All actual workflow state tracking goes through `abacus_state` + `abacus_event` (Section 2.6), and all adjustment data goes through `statement_period_adjustment_file` -> `worksheet_adjustment` (Section 2.1).
>
> **These tables should be torn down.** The `worksheet_flowthrough_batch` write in `adjustment-file-initialize` should be removed first, then the tables can be dropped.

```mermaid
erDiagram
    worksheet_flowthrough_batch {
        int worksheet_flowthrough_batch_id PK
        int statement_period_id FK
        int source_file_upload_id FK "nullable"
        enum batch_type "auto|manual|upload"
        enum batch_status "pending|validating|importing|pending_approval|approved|rejected|applied|error"
        json errors "array of error codes"
        int error_row_count
        int row_level_error_count
        int total_row_count
        decimal total_amount "30,2 - rounded sum"
        decimal total_amount_raw "40,12 - raw sum"
        datetime created_at
        string created_by
    }

    reference_worksheet_flowthrough_error {
        int reference_worksheet_flowthrough_error_id PK
        string error_code
        string error_message
        string error_description
    }

    worksheet_flowthrough {
        int worksheet_flowthrough_id PK
        int worksheet_flowthrough_batch_id FK
    }

    worksheet_flowthrough_file {
        int worksheet_flowthrough_file_id PK
        int worksheet_flowthrough_batch_id FK
    }

    worksheet_flowthrough_batch ||--o{ worksheet_flowthrough : "DEPRECATED"
    worksheet_flowthrough_batch }o--|| statement_period : "belongs to period"
    worksheet_flowthrough_batch }o--o| file_upload : "sourced from"
```

**DEPRECATED tables — to be torn down:**
- `worksheet_flowthrough_batch` — record created by `adjustment-file-initialize` Lambda but never read downstream; remove the write first, then drop
- `worksheet_flowthrough` — never populated; drop
- `worksheet_flowthrough_file` — never populated; drop
- `reference_worksheet_flowthrough_error` — populated with reference data but never queried; drop

### 2.5 Ledger and Payment Allocation

```mermaid
erDiagram
    ledger_adjustment_applied {
        int ledger_adjustment_applied_id PK
        int abacus_event_id FK
        int statement_period_adjustment_file_id FK
        int statement_period_id FK
        int account_id
        int contract_id
        int activity_statement_period_id
        int apply_to_statement_period_id
        int reference_adjustment_type_id
        decimal adjustment_amount
        string adjustment_currency_code
        decimal payee_amount "converted"
        string payee_currency_code
        boolean apply_to_flowthrough_payment
        string created_by
    }

    ledger_adjustment_detail_applied {
        int ledger_adjustment_detail_applied_id PK
        int ledger_adjustment_applied_id FK
    }

    ledger_account_contract {
        int ledger_account_contract_id PK
        int account_id
        int contract_id
        int statement_period_id
        decimal amount
        string currency_code
        string source "adjustment"
    }

    ledger_contract_flowthrough {
        int ledger_contract_flowthrough_id PK
        int account_id
        int contract_id
        int statement_period_id
        decimal amount
        string currency_code
        string source "flowthrough_adjustment"
    }

    ledger_deposit {
        int ledger_deposit_id PK
        decimal amount
        string currency_code
        string source "adjustment_rounding"
    }

    payment_allocation {
        int payment_allocation_id PK
        int contract_id
        string payee_type
        int payee_id
        int statement_period_id
        string payment_allocation_type
        decimal amount_to_payment
        string payment_status "init"
        decimal amount_to_ledger
        string ledger_status "init"
        string currency_code
        string description
        string created_by
        string last_modified_by
    }

    payment_allocation_ledger_adjustment {
        int payment_allocation_ledger_adjustment_id PK
        int payment_allocation_id FK
        int ledger_adjustment_applied_id FK
        string created_by
        string last_modified_by
    }

    statement_period_payment_entity {
        int statement_period_payment_entity_id PK
        int statement_period_id FK
        int reference_payment_entity_id
        boolean is_visible_to_customer
    }

    ledger_adjustment_applied ||--o{ ledger_adjustment_detail_applied : "has details"
    ledger_adjustment_applied ||--o{ payment_allocation_ledger_adjustment : "allocated to"
    payment_allocation ||--o{ payment_allocation_ledger_adjustment : "links adjustments"
    statement_period_payment_entity }o--|| statement_period : "belongs to"
```

### 2.6 State Tracking (Abacus Event System)

```mermaid
erDiagram
    abacus_state {
        int abacus_state_id PK
        int target_id "polymorphic FK"
        string target_type "e.g. statement_period_adjustment_file"
        string action "upload_file|validate_file|import_file|approve_file|apply_file"
        enum state "init|running|complete|error"
        datetime created_at
        datetime updated_at
    }

    abacus_event {
        int abacus_event_id PK
        int target_id "polymorphic FK"
        string target_type
        string event_type
        json event_data
        datetime created_at
    }

    abacus_state }o--|| statement_period_adjustment_file : "tracks workflow for"
    abacus_event }o--|| statement_period_adjustment_file : "logs events for"
```

### 2.7 Full System ERD (Simplified Relationships)

```mermaid
erDiagram
    %% Upload Flow
    file_upload_config ||--o{ file_upload : configures
    file_upload ||--o| statement_period_adjustment_file : "source for"

    %% Adjustment File Flow
    statement_period ||--o{ statement_period_adjustment_file : "has"
    statement_period_adjustment_file ||--o| statement_period_adjustment_batch_criteria : "criteria"
    statement_period_adjustment_file ||--o{ worksheet_adjustment : "contains"
    worksheet_adjustment ||--o{ worksheet_adjustment_detail : "details"

    %% Ledger Flow
    statement_period_adjustment_file ||--o{ ledger_adjustment_applied : "produces"
    ledger_adjustment_applied ||--o{ ledger_adjustment_detail_applied : "details"
    ledger_adjustment_applied ||--o| ledger_account_contract : "creates entry"
    ledger_adjustment_applied ||--o| ledger_contract_flowthrough : "creates FT entry"
    ledger_adjustment_applied ||--o| ledger_deposit : "rounding entry"

    %% Payment Allocation Flow
    ledger_adjustment_applied ||--o{ payment_allocation_ledger_adjustment : "links to"
    payment_allocation ||--o{ payment_allocation_ledger_adjustment : "groups into"
    statement_period_payment_entity ||--o{ payment_allocation : "triggers"

    %% Contract Config
    contract ||--o| contract_flowthrough : "FT config"
    reference_flowthrough_calculation ||--o{ contract_flowthrough : "method"

    %% State Tracking
    statement_period_adjustment_file ||--o{ abacus_state : "tracked by"
    statement_period_adjustment_file ||--o{ abacus_event : "events for"
```

---

## 3. Data Flow: Manual Adjustment Upload

### 3.1 End-to-End Sequence

```mermaid
sequenceDiagram
    actor User
    participant FE as Frontend
    participant API as ows-royalties
    participant S3 as S3 Bucket
    participant EB as EventBridge
    participant SF as Step Functions
    participant Lambda as Lambdas

    User->>FE: 1. Select .xlsx file
    alt New path (FF ENABLED)
        FE->>API: 2a. Initiate file upload (useAbacusInitiateFileUpload)
        API-->>FE: Signed S3 URL + file_upload record
        FE->>S3: 3. Upload file directly via signed URL
    else Legacy path (FF DISABLED)
        FE->>API: 2b. CreateStatementPeriodAdjustmentFile
        API-->>FE: adjustment file record
        FE->>API: 3. Get upload token
        FE->>S3: 4. Upload file to S3
        FE->>API: 5. Update DB with S3 location
    end
    API->>API: 6. file_upload status → complete
    API->>EB: 7. Emit file_upload.completed event
    EB->>SF: 8. Start adjustment-file-ingest-workflow
    SF->>Lambda: 9. InitializeBatch
    Lambda->>API: 10. Create statement_period_adjustment_file + abacus_state
    Lambda-->>SF: Batch details + S3 location
```

> **Feature flag gate:** The upload path is determined by `ABACUS_APPLY_FLOWTHROUGH_PAYMENT` <!-- flag torn down 2026-06; see ACC-10470 --> (note: the frontend code constant is misspelled as `ABACUS_APPLY_FLOWTROUGH_PAYMENT` — missing the second 'H' — but the Split.io flag value is correctly spelled `abacus_apply_flowthrough_payment`). When **enabled**, the frontend uses `useAbacusInitiateFileUpload` (new path) which creates the `file_upload` record server-side. When **disabled**, the legacy path uses `CreateStatementPeriodAdjustmentFile` + `useAbacusUploadToken`. The old `CreateStatementPeriodAdjustmentFile` mutation is still in the codebase but only executes when the flag is off.

### 3.2 Step-by-Step Flow

#### Phase 1: File Upload (Frontend -> S3)

**New path (when `ABACUS_APPLY_FLOWTHROUGH_PAYMENT` <!-- flag torn down 2026-06; see ACC-10470 --> flag is ON):**

1. **User selects `.xlsx` file** in the Adjustments page upload modal
2. **Frontend calls `useAbacusInitiateFileUpload('adjustments')`** which creates the `file_upload` record and returns a signed S3 upload URL
3. **Frontend uploads file directly to S3** via the signed URL
4. **ows-royalties detects upload completion**, marks `file_upload` as `complete`, and writes event to `abacus_outbox`

**Legacy path (when `ABACUS_APPLY_FLOWTHROUGH_PAYMENT` <!-- flag torn down 2026-06; see ACC-10470 --> flag is OFF):**

1. **User selects `.xlsx` file** in the Adjustments page upload modal
2. **Frontend creates DB record** via `CreateStatementPeriodAdjustmentFile` GraphQL mutation -> POST `/statement-period/{id}/adjustment-file`
3. **Frontend gets signed upload URL** via `useAbacusUploadToken`
4. **Frontend uploads file directly to S3** (`{env}-abacus-adjustments` bucket) with metadata: `asset_type`, `original_filename`, `object_type: "adjustment"`
5. **Frontend updates DB record** with S3 location via `UpdateStatementPeriodAdjustmentFile`
6. **ows-royalties marks `file_upload` as `complete`** and writes event to `abacus_outbox`

#### Phase 2: Event-Driven Processing (EventBridge -> Step Functions)

7. **Outbox processor publishes to EventBridge** with source `abacus.outbox`, detail-type `file_upload.completed`
8. **EventBridge rule matches** and starts Step Function execution (`{env}-adjustment-file-ingest-workflow`)

#### Phase 3: Lambda Pipeline

9. **InitializeBatch** (`adjustment-file-initialize`, 30s timeout):
   - Validates file_upload exists and status is `complete`
   - Validates upload_type is `adjustments`
   - Finds statement_period
   - Creates `worksheet_flowthrough_batch` record (note: this record is written but not read by any downstream step — **deprecated**, to be removed)
   - Creates `statement_period_adjustment_file` via ows-royalties (if FF enabled)
   - Creates `abacus_state` entries (if FF enabled)
   - Returns batch details + S3 location

10. **ReshapeAfterInit** (Pass state): Transforms event structure

11. ~~**PrepareFile**~~ (`adjustment-file-prepare`, **DEPRECATED** — deployed to QA but never integrated; to be torn down)

#### Phase 4: Validation (Airflow DAG: `adjustment_file_upload`)

The Airflow DAG runs these steps:

1. `notify_started` - Logs start
2. `invoke_lambda_av_scan` - Virus scan via Lambda
3. `validate_format` - Validates `.xlsx` format
4. `invoke_adjustment_file_validation_lambda` - Triggers validation Lambda
5. `adjustment_file_validation_check_result` - Polls validation status (30s interval, 15min timeout)
6. `check_adjustment_file_error_report` - Checks for errors
7. `notify_success` / `notify_failure` - Final notification

**Validation Lambda (`adjustment_file_validation`):**
- Downloads file from S3
- Validates Excel headers against expected mapping
- Validates each row:
  - Account ID, Contract ID exist
  - Amount is valid decimal
  - Currency code is valid uppercase ISO
  - Activity/Statement month/year are valid
  - Adjustment type is valid (lowercase)
  - `Apply to Flowthrough Payment` boolean (when FF enabled)
- Cross-references against Snowflake (`AdjustmentsValidationSnowflakeExecutor`)
- Adds "Validation Errors" column to Excel with per-row errors
- Uploads error report to S3
- Updates `statement_period_adjustment_file` with:
  - `valid_row_count`, `invalid_row_count`
  - `total_file_amount_multicurrency`, `total_rounded_amount_multicurrency`
  - `invalid_file_location` (S3 path)
  - `error_type` (content_error | format_error | null)

#### Phase 5: Import (Airflow DAG: `adjustment_file_import`)

1. `notify_started` - Logs start
2. `invoke_adjustment_file_import_lambda` - Triggers import Lambda
3. `notify_success` / `notify_failure` - Final notification

**Import Lambda (`adjustment_file_import`):**
- Downloads validated file from S3
- Reads Excel with pandas
- Cleans data (empty rows, special chars, whitespace)
- Inserts `worksheet_adjustment` records in batches (`MYSQL_BATCH_SIZE`)
- Groups detail rows by (account_id, contract_id, currency_code, activity/apply_to periods)
- Inserts `worksheet_adjustment_detail` records with deadlock retry (3 attempts)
- Sets `apply_to_flowthrough_payment` on each record (when FF enabled)

#### Phase 6: Approval + Application

- **User approves** batch via UI (restricted by `ABACUS_MANUAL_ADJUSTMENTS_APPROVED_USERS` flag)
- **User triggers apply** which fires `apply_pending_adjustments` event
- **Apply Lambda (`adjustments_apply`):**
  1. Gets current statement_period
  2. Gets exchange_rates for currency conversion
  3. Gets pending worksheet_adjustments
  4. Validates no payment entity has `close_balance = complete`
  5. For each adjustment:
     - Converts amount from `adjustment_currency_code` to `payee_currency_code`
     - Creates `ledger_adjustment_applied` record
  6. Bulk inserts `ledger_account_contract` (non-flowthrough) or `ledger_contract_flowthrough` (flowthrough)
  7. Creates `ledger_deposit` for rounding differences

---

## 4. Data Flow: Auto-Generated Flowthrough Batch

```mermaid
sequenceDiagram
    actor User
    participant FE as Frontend
    participant API as ows-royalties
    participant SF as Snowflake
    participant Lambda as Lambda
    participant S3 as S3 Bucket

    User->>FE: 1. Select payment entities + schedules
    FE->>API: 2. POST /generate (CreateAdjustmentFileAndBatchCriteria)
    API->>API: 3. Create statement_period_adjustment_file + batch_criteria
    API->>Lambda: 4. Trigger generate_flowthrough_adjustments via DAG
    Lambda->>SF: 5. Query VW_ABACUS_AUTOMATED_FLOWTHROUGH
    SF-->>Lambda: 6. Return matching rows
    Lambda->>Lambda: 7. Generate .xlsx from results
    Lambda->>S3: 8. Upload generated file
    loop Every 5 seconds
        FE->>API: 9. Poll auto-generation status
        API-->>FE: 10. Status: generating
    end
    API-->>FE: 11. Status: not_approved (or no_records / failed_to_generate)
    Note over FE,S3: File then enters standard validation → import → approve → apply pipeline
```

### Auto-Generation Steps

1. User selects batch type (flowthrough), reference payment entities, and payment schedules on `/adjustments/generate`
2. Frontend calls `CreateAdjustmentFileAndBatchCriteria` mutation -> POST `/statement-period/{id}/adjustments/generate`
3. Backend creates `statement_period_adjustment_file` (batch_type=auto) + `statement_period_adjustment_batch_criteria`
4. Backend triggers `generate_flowthrough_adjustments` Lambda via Airflow DAG
5. Lambda queries `VW_ABACUS_AUTOMATED_FLOWTHROUGH` Snowflake view with criteria
6. Lambda generates `.xlsx` file from query results
7. Lambda uploads generated file to S3 (`{env}-abacus-adjustments`)
8. File enters the standard validation -> import -> approve -> apply pipeline
9. Frontend polls every 5 seconds showing "Generating" status
10. On completion, batch appears as "Not Approved" (or "No Records" / "Failed to Generate")

### Snowflake View: `VW_ABACUS_AUTOMATED_FLOWTHROUGH`

Source data for auto-generation. Returns rows with:
- Account ID, Contract ID
- Amount, Currency Code
- Activity Month/Year, Statement Month/Year
- Adjustment Type
- UPC, Distribution Type
- Apply to Flowthrough Payment flag

Filtered by reference_payment_entity_id and payment_schedule.

---

## 5. Data Flow: Payment Allocation

```mermaid
sequenceDiagram
    participant EB as EventBridge
    participant Lambda as payment_allocation Lambda
    participant DB as MySQL (royalty_accounting)

    EB->>Lambda: close_balance.completed (statement_period_payment_entity_id)
    Lambda->>DB: 1. Lookup statement_period_payment_entity
    DB-->>Lambda: SPPE record
    Lambda->>DB: 2. Validate close_balance abacus_state = complete
    DB-->>Lambda: State confirmed

    Lambda->>DB: 3. Discover contracts with unlinked FT adjustments
    DB-->>Lambda: Contract IDs + adjustment counts

    loop For each contract batch
        Lambda->>DB: 4. Get adjustments for contract batch
        DB-->>Lambda: Adjustment records
        Lambda->>Lambda: 5. Group by (contract, payee, currency)
        Lambda->>DB: 6. INSERT payment_allocation (one per group)
        Lambda->>DB: 7. INSERT payment_allocation_ledger_adjustment (one per adjustment)
        Lambda->>DB: 8. COMMIT batch
    end

    Lambda-->>EB: Return allocations_created + ledger_adjustments_linked
```

### Payment Allocation Details

**Trigger:** EventBridge event `close_balance.completed` with `statement_period_payment_entity_id`

**Processing:**
1. Lookup `statement_period_payment_entity` record
2. Validate `close_balance` abacus_state is `complete` (raise `BalancesNotClosedError` if not)
3. Discover contracts with unlinked flowthrough adjustments:
   - Query `ledger_adjustment_applied` WHERE `apply_to_flowthrough_payment = 1`
   - Exclude already-linked (EXISTS in `payment_allocation_ledger_adjustment`)
   - Filter by `statement_period_id` and `reference_payment_entity_id`
4. Bin-pack contracts into batches (respecting `max_allowed_packet`)
5. For each batch:
   - Fetch adjustments for contract IDs
   - Group by `(contract_id, account_payee_id, adjustment_currency_code, payee_currency_code)`
   - Sum amounts per group
   - Create `payment_allocation` record (status: `init`)
   - Create `payment_allocation_ledger_adjustment` link for each adjustment
   - Commit
6. Return counts: `allocations_created`, `ledger_adjustments_linked`

**Idempotency:** Uses `NOT EXISTS` checks to skip already-linked adjustments. Safe to re-run.

---

## 6. Component Details

### 6.1 Lambda Functions

| Lambda | Timeout | Memory | Concurrency | Trigger | Purpose |
|--------|---------|--------|-------------|---------|---------|
| `outbox-process` | 120s | 1024 MB | — | Kafka CDC on `cdc.royaltyAccounting.abacusOutbox` + EventBridge schedule (backup) | Reads `abacus_outbox`, publishes events to EventBridge |
| `adjustment-file-initialize` | 300s (Lambda) / 30s (SF task timeout) | 1024 MB | 2 | Step Functions (`adjustment-file-ingest-workflow`) | Create batch record, validate file_upload, create abacus event |
| ~~`adjustment-file-prepare`~~ | 900s | 10240 MB | 4 | **DEPRECATED** | Deployed to QA but never integrated — tear down |
| `adjustment-file-validation` | 900s | 10240 MB | 16 | Airflow DAG (`adjustment_file_upload`) | Validate file content against rules + Snowflake |
| `adjustment-file-import` | 900s | 10240 MB | 16 | Airflow DAG (`adjustment_file_import`) | Import valid rows to worksheet tables |
| `adjustments-apply` | 900s | 1024 MB | 16 | Airflow DAG (`apply_pending_adjustments`) | Apply adjustments to ledger with currency conversion |
| `payment-allocation` | N/A | N/A | N/A | EventBridge (`close_balance.completed`) | Allocate flowthrough adjustments to payments |
| ~~`adjustment-file-complete`~~ | 900s | 1024 MB | 2 | **DEPRECATED** | Deployed to QA but never integrated — tear down |
| ~~`adjustment-file-process-batch`~~ | 900s | 10240 MB | 10 | **DEPRECATED** | Deployed to QA but never integrated — tear down |
| `generate-flowthrough-adjustments` | N/A | N/A | N/A | Airflow DAG (`auto_generate_adjustments`) | Generate adjustment files from Snowflake |

> **Note on timeouts:** `adjustment-file-initialize` has a 300-second Lambda timeout but a 30-second Step Functions task timeout. The SF task timeout is what matters operationally — if the Lambda hasn't returned in 30 seconds, the Step Function marks it as failed and retries (3x with exponential backoff).

### 6.2 Airflow DAGs

#### `adjustment_file_upload` DAG

```
notify_started
    └── invoke_lambda_av_scan
        └── validate_format
            └── invoke_adjustment_file_validation_lambda
                └── adjustment_file_validation_check_result (sensor, 30s poll, 15min timeout)
                    └── check_adjustment_file_error_report
                        └── notify_success
                            └── notify_failure (trigger_rule: one_failed)
```

> **Known issue:** AV scanning currently runs in **two places**: (1) the `abacus-file-upload-workflow` step function (during the `init` -> `scanning` -> `complete` transition), and (2) this DAG's `invoke_lambda_av_scan` task. This is redundant — the file is scanned twice. Deduplicating this is a known TODO.

**Schedule:** None (triggered on demand)
**Trigger:** `adjustment-file-initialize` Lambda creates an abacus event (`adjustment_file_upload`). The `ows-abacus-event` service receives this event, maps it to DAG ID `adjustment_file_upload`, and triggers the DAG via MWAA CLI (`dags trigger -c '{config}' adjustment_file_upload`).
**Task Files:** `ows-royalties-workflows/dags/tasks/adjustment_file_upload/`

#### `adjustment_file_import` DAG

```
notify_started
    └── invoke_adjustment_file_import_lambda
        ├── notify_success (trigger_rule: none_failed)
        └── notify_failure (trigger_rule: one_failed)
```

**Schedule:** None (triggered on demand)
**Trigger:** User clicks "IMPORT DATA" in the UI -> frontend calls `WorksheetAdjustmentAndDetailsImport` mutation -> backend creates abacus event (`adjustment_file_worksheet_import`) -> `ows-abacus-event` maps to DAG ID `adjustment_file_import` and triggers via MWAA CLI.
**Task Files:** `ows-royalties-workflows/dags/tasks/adjustment_file_import/`

#### `apply_pending_adjustments` DAG (not shown above)

This DAG is triggered when a user clicks "Apply" on an approved batch. It invokes the `adjustments-apply` Lambda.

**Schedule:** None (triggered on demand)
**Trigger:** User clicks "Apply" in the UI -> backend creates abacus event (`apply_pending_adjustments`) -> `ows-abacus-event` maps to DAG ID `apply_pending_adjustments` and triggers via MWAA CLI.
**Task Files:** `ows-royalties-workflows/dags/tasks/apply_pending_adjustments/`

#### How DAG triggering works

All Airflow DAGs in this system use `schedule_interval=None` (manual trigger only). They are triggered by the `ows-abacus-event` service, which:

1. Receives an event via `POST /abacus-event`
2. Looks up the event name in a mapping table to find the target DAG ID
3. Obtains an MWAA CLI token via `boto3` (`airflow:CreateCliToken`)
4. Executes `dags trigger -c '{config_json}' {dag_id}` via the MWAA CLI API

This means **every DAG trigger goes through `ows-abacus-event`** — there are no direct EventBridge-to-Airflow connections. The event router lives in the `ows-abacus-event` repo with IAM permissions defined in `terraform-infra/prod/ows-abacus-event/main.tf`.

### 6.3 Step Functions

#### `abacus-file-upload-workflow` (shared infrastructure)

This step function handles the generic file upload lifecycle — AV scanning, status transitions, and event emission. It is **not specific to flowthrough** but is the first step function in the adjustment pipeline.

**Trigger:** `file_upload` record created in ows-royalties
**What it does:** Runs AV scan, transitions `file_upload.upload_status` through `init` -> `scanning` -> `complete` (or `quarantined`/`error`), then writes `file_upload.completed` event to `abacus_outbox`
**Terraform:** `terraform-infra/prod/ows-royalties-workflows/file_upload_state_machine.tf`

> This is shared infrastructure used by all file upload types, not just adjustments. It lives in the prod account (`437795906767`) for both QA and Prod.

#### `adjustment-file-ingest-workflow`

#### QA State Machine

> **Note:** The QA definition includes `ReshapeAfterInit` and `PrepareFile` states, but `adjustment-file-prepare` is **deprecated** — it was deployed but never integrated into the active pipeline. The only active state is `InitializeBatch`. The `PrepareFile`/`ReshapeAfterInit` states should be removed from the QA state machine to match Prod.

```mermaid
stateDiagram-v2
    [*] --> InitializeBatch
    InitializeBatch --> ReshapeAfterInit: success
    InitializeBatch --> InitializeFailed: error
    ReshapeAfterInit --> PrepareFile
    PrepareFile --> [*]: success
    PrepareFile --> PrepareFailed: error

    note right of PrepareFile: DEPRECATED - to be torn down
    note right of InitializeBatch: Lambda: adjustment-file-initialize (30s timeout, 3x retry)
```

**Overall timeout:** 1200 seconds (20 minutes)

#### Prod State Machine

```mermaid
stateDiagram-v2
    [*] --> InitializeBatch
    InitializeBatch --> [*]: success
    InitializeBatch --> InitializeFailed: error

    note right of InitializeBatch: Lambda: adjustment-file-initialize (30s timeout, 3x retry)
```

**Key difference:** Prod only has `InitializeBatch`. The `adjustment-file-prepare`, `adjustment-file-complete`, and `adjustment-file-process-batch` Lambdas exist in QA Terraform but are **deprecated** and should be torn down.

### 6.4 EventBridge Rules

| Rule | Source | Detail Type | Filter | Target |
|------|--------|-------------|--------|--------|
| `{env}-adjustment-file-ingest-trigger` | `abacus.outbox` | `file_upload.completed` | `target_type: file_upload, upload_type: adjustments` | Step Function |
| Payment allocation trigger | `abacus.outbox` | `close_balance.completed` | N/A | `payment-allocation` Lambda |

**Dead Letter Queue:** `{env}-adjustment-file-ingest-workflow-dlq` (14-day retention)
**Retry:** 3 attempts, 24-hour max age

---

## 7. Infrastructure

### 7.1 Terraform File Locations

| Resource | QA Path | Prod Path |
|----------|---------|-----------|
| Step Function | `accounting/qa/ows-royalties-workflows/adjustment_file_ingest_state_machine.tf` | `prod/ows-royalties-workflows/adjustment_file_ingest_state_machine.tf` |
| IAM | `accounting/qa/ows-royalties-workflows/adjustment_file_ingest_iam.tf` | `prod/ows-royalties-workflows/adjustment_file_ingest_iam.tf` |
| EventBridge Trigger | `accounting/qa/ows-royalties-workflows/adjustment_file_ingest_trigger.tf` | `prod/ows-royalties-workflows/adjustment_file_ingest_trigger.tf` |
| Monitoring | `accounting/qa/ows-royalties-workflows/adjustment_file_ingest_monitoring.tf` | `prod/ows-royalties-workflows/adjustment_file_ingest_monitoring.tf` |
| Lambda: initialize | `accounting/qa/lambda-abacus/adjustment-file-initialize.tf` | `accounting/prod/lambda-abacus/adjustment-file-initialize.tf` |
| Lambda: validation | `accounting/qa/lambda-abacus/adjustment-file-validation.tf` | `accounting/prod/lambda-abacus/adjustment-file-validation.tf` |
| Lambda: import | `accounting/qa/lambda-abacus/adjustment-file-import.tf` | `accounting/prod/lambda-abacus/adjustment-file-import.tf` |
| Lambda: apply | `accounting/qa/lambda-abacus/adjustments-apply.tf` | `accounting/prod/lambda-abacus/adjustments-apply.tf` |
| ~~Lambda: prepare~~ (**DEPRECATED**) | `accounting/qa/lambda-abacus/adjustment-file-prepare.tf` | N/A — tear down |
| ~~Lambda: complete~~ (**DEPRECATED**) | `accounting/qa/lambda-abacus/adjustment-file-complete.tf` | N/A — tear down |
| ~~Lambda: process-batch~~ (**DEPRECATED**) | `accounting/qa/lambda-abacus/adjustment-file-process-batch.tf` | N/A — tear down |
| Lambda: payment-allocation | `accounting/qa/lambda-abacus/payment-allocation.tf` | N/A (TBD) |
| S3 Bucket | Cross-account from prod | `prod/ows-royalties-workflows/main.tf` |

### 7.2 S3 Buckets

| Bucket | Environment | Purpose | CORS |
|--------|-------------|---------|------|
| `qa-abacus-adjustments` | QA | Adjustment file storage | N/A |
| `prod-abacus-adjustments` | Prod | Adjustment file storage | PUT, POST from `abacus.theorchard.com` |

**File Structure in S3:**
```
{env}-abacus-adjustments/
├── adjustments/
│   ├── {statement_period_id}/
│   │   ├── {file_upload_id}/
│   │   │   ├── original.xlsx           # Uploaded file
│   │   │   └── error_report.xlsx       # Validation errors (if any)
│   │   └── ...
│   └── ...
└── templates/
    └── abacus_adjustments_template/
        └── Adjustment_Template.xlsx
```

### 7.3 Monitoring (Datadog)

| Monitor | Query | Warning | Critical |
|---------|-------|---------|----------|
| Step Function Failures | `sum:aws.states.execution_failed{statemachinename:*-adjustment-file-ingest-workflow}.as_count()` | 1 in 5min | 3 in 5min |
| DLQ Depth | `avg:aws.sqs.approximate_number_of_messages_visible{queuename:*-adjustment-file-ingest-workflow-dlq}` | 1 in 5min | 3 in 5min |
| Step Function Duration | `avg:aws.states.execution_time{statemachinename:*-adjustment-file-ingest-workflow}` | 600s | 900s |
| Lambda Errors (each) | Per-lambda error count | Varies | Any errors |
| Lambda Duration (each) | Per-lambda execution time | Varies | At timeout |

### 7.4 CloudWatch

- **Log Group:** `/aws/vendedlogs/states/{env}-adjustment-file-ingest-workflow`
- **Retention:** 365 days
- **X-Ray tracing:** Enabled
- **Log level:** ALL (includes execution data)

---

## 8. Feature Flags

| Flag | Service | Purpose | Impact |
|------|---------|---------|--------|
| `abacus_flowthrough_automation` | `adjustment-file-initialize` | Enables creation of `statement_period_adjustment_file` in OWS and `abacus_state`/`abacus_event` entries | When OFF: batch created but no OWS integration |
| `ABACUS_APPLY_FLOWTHROUGH_PAYMENT` <!-- flag torn down 2026-06; see ACC-10470 --> | `adjustment-file-validation`, `adjustment-file-import`, `adjustments-apply`, Frontend | Enables `apply_to_flowthrough_payment` field handling throughout the pipeline; gates new vs legacy upload path in frontend | When OFF: field ignored, no `ledger_contract_flowthrough` entries, legacy upload path active |

> **Known typo:** The frontend code constant is `ABACUS_APPLY_FLOWTROUGH_PAYMENT` (missing the second 'H' in FLOWTHROUGH). The Split.io flag value is correctly spelled `abacus_apply_flowthrough_payment` <!-- flag torn down 2026-06; see ACC-10470 -->. When searching the codebase, search for both spellings.
| `abacus_auto_generate_adjustments_flowthrough` | Frontend + `ows-royalties` | Enables the "Generate Adjustments" UI and auto-generation API | When OFF: manual upload only |
| `ABACUS_IMPORT_ADJUSTMENTS` | Frontend | Enables the adjustment import modal | When OFF: no upload UI |
| `ABACUS_MANUAL_ADJUSTMENTS_APPROVED_USERS` | Frontend | Restricts who can approve adjustment batches | When OFF: all users can approve |
| `ABACUS_PROGRESS_INDICATORS` <!-- flag torn down 2026-06; see ACC-10476 --> | Frontend | Shows estimated completion times during processing | When OFF: generic spinner |
| `ABACUS_ADJUSTMENTS_BATCH_PAGE_FILTERS` | Frontend | Enables advanced filtering on batch detail page | When OFF: no filters |

---

## 9. API Reference

### 9.1 ows-royalties REST Endpoints

#### Adjustment File Management

| Method | Path | Purpose |
|--------|------|---------|
| POST | `/statement-period/{id}/adjustment-file` | Create adjustment file record |
| GET | `/statement-period/{id}/adjustment-files` | List files for period |
| GET | `/statement-period-adjustment-file/{id}` | Get file details |
| PUT | `/statement-period-adjustment-file/{id}` | Update file (S3 location, counts) |
| DELETE | `/statement-period-adjustment-file/{id}` | Soft delete file |
| GET | `/statement-period-adjustment-files` | List all files with filters |
| GET | `/statement-period-adjustment-file/users/{action}` | Get users by action |
| GET | `/statement-period-adjustment-file/by-source-file-key/{key}` | Lookup by file upload key |
| POST | `/statement-period-adjustment-file/validate-adjustments` | Validate manual adjustments |
| GET | `/statement-period/{id}/adjustments/auto-generation/progress` | Auto-generation status |

#### File Downloads

| Method | Path | Purpose |
|--------|------|---------|
| GET | `/abacus-adjustments/download/template` | Get template presigned URL |
| GET | `/statement-period-adjustment-file/{id}/download/error` | Get error report presigned URL |
| GET | `/statement-period-adjustment-file/{id}/download/report` | Get valid file presigned URL |

#### Contract Flowthrough

| Method | Path | Purpose |
|--------|------|---------|
| POST | `/contract/{id}/contract-flowthrough/` | Create flowthrough config |
| GET | `/contract/{id}/contract-flowthrough/` | Get by contract |
| GET | `/contract-flowthrough/{id}/` | Get by ID |
| PUT | `/contract-flowthrough/{id}/` | Update config |
| DELETE | `/contract-flowthrough/{id}/` | Soft delete |

#### Reference Data

| Method | Path | Purpose |
|--------|------|---------|
| GET | `/reference-flowthrough-calculations/` | List calculation methods |
| GET | `/reference-flowthrough-calculation/{id}` | Get calculation details |

#### Worksheet Flowthrough Batch — **DEPRECATED**

> These endpoints correspond to the deprecated `worksheet_flowthrough_batch` table. They should be torn down along with the table.

| Method | Path | Purpose |
|--------|------|---------|
| ~~POST~~ | `/worksheet-flowthrough-batch/` | ~~Create batch~~ **DEPRECATED** |
| ~~GET~~ | `/worksheet-flowthrough-batch/{id}` | ~~Get batch~~ **DEPRECATED** |
| ~~PUT~~ | `/worksheet-flowthrough-batch/{id}` | ~~Update batch~~ **DEPRECATED** |
| ~~DELETE~~ | `/worksheet-flowthrough-batch/{id}` | ~~Delete batch~~ **DEPRECATED** |

### 9.2 GraphQL Queries (Frontend)

| Query | Description |
|-------|-------------|
| `GetStatementPeriodAdjustmentFilesList` | List all adjustment files with status |
| `GetStatementPeriodAdjustmentFileAndStatus` | File details + abacus_state |
| `GetInProgressAutoAdjustmentFile` | Track auto-generation progress |
| `GetAbacusWorksheetAdjustmentsAndDetails` | Adjustment records with filters |
| `GetStatementPeriodAdjustmentFilesSearch` | Search files |
| `GetStatementPeriodAdjustmentFileUsers` | Users who performed actions |

### 9.3 GraphQL Mutations (Frontend)

| Mutation | Description | Notes |
|----------|-------------|-------|
| `CreateStatementPeriodAdjustmentFile` | Create file record (fileName, statementPeriodId) | **Legacy only** — used when `ABACUS_APPLY_FLOWTHROUGH_PAYMENT` <!-- flag torn down 2026-06; see ACC-10470 --> FF is OFF |
| `UpdateStatementPeriodAdjustmentFile` | Update with S3 location | Legacy path only |
| `useAbacusInitiateFileUpload` | Initiate file upload with signed URL | **New path** — used when FF is ON |
| `WorksheetAdjustmentAndDetailsImport` | Trigger validation + import | |
| `CreateAdjustmentFileAndBatchCriteria` | Trigger auto-generation | |
| `SoftDeleteStatementPeriodAdjustmentFile` | Cancel/delete file | |

### 9.4 Frontend Routes

| Route | Component | Feature Flag |
|-------|-----------|-------------|
| `/adjustments` | `Adjustments` (list page) | None |
| `/adjustments/generate` | `GenerateAdjustmentsForm` | `ABACUS_AUTO_GENERATE_ADJUSTMENTS_FLOWTHROUGH` |
| `/adjustments/:batchId` | `AdjustmentsBatch` (detail) | None |

---

## 10. Security and Authorization

### Authentication

- All ows-royalties endpoints require `verify_rules_access_standalone(request)` authorization
- Returns 401 Unauthorized if access denied
- Uses OWS request/authorization framework with identity tokens

### S3 Access

- Frontend uploads via presigned URLs (1-hour expiration)
- Lambda functions access S3 via IAM roles with scoped policies
- Cross-account access for QA (S3 in prod account, Lambdas in QA account)

### Lambda IAM

- Each Lambda has a dedicated execution role
- Permissions scoped to: specific S3 bucket, specific OWS endpoints, Split.io API key
- Step Functions have a dedicated role with `lambda:InvokeFunction` for specific Lambdas only

### Feature Flag Access Control

- `ABACUS_MANUAL_ADJUSTMENTS_APPROVED_USERS` restricts approval to authorized users
- `ABACUS_IMPORT_ADJUSTMENTS` gates the entire upload feature
- Feature flag checks happen at both frontend (UI visibility) and backend (API validation)