# Accounting Run Calculation Pipeline

> Source: [Accounting Run (Notion)](https://www.notion.so/c9ecc13c18cb4362962ce54853992580)  
> Whimsical: [accounting_run_calculate](https://whimsical.com/accounting-run-calculate-9eUCj5dC1R9DGSCYvL94dg)  
> DAGs: [`ows-royalties-workflows`](https://github.com/theorchard/ows-royalties-workflows)

## Pipeline Overview

When a user clicks "Create" on a run controller, the following DAG steps execute:

```
1. Set run_status = "Running"
2. Snapshot contracts to S3
3. Flatten nested JSON → contract_denormalized_distro
4. Match transactions to contract terms → contract_transaction_staging
5. Calculate totals → accounting_run_results_distro_staging
6. Create run summary → ledger_accounting_run_balance
7. Export summary to S3
8. Set run_status = "Complete"
```

## Step 1: Set Status to "Running"

Updates `accounting_run.run_status` to inform users that the calculation has started.

## Step 2: Snapshot Contracts to S3

Takes a point-in-time snapshot of all `contract`, `contract_term`, and `contract_term_condition` data for the run controller. Exported as TSV files to S3.

**Why snapshot?** Fivetran replication from MySQL to Snowflake can take up to 5 minutes. The "accelerated" calculation can't wait. The snapshot also provides an audit trail of what each contract looked like at the time of any given run.

Snapshot files:
- `contract.tsv` — contract_id, account_id, term_start, term_end, payee_currency_code
- `contract-terms.tsv` — contract_term_id, term_type, attachments, conditions, term_rate, priority

S3 bucket structure: see [Whimsical](https://whimsical.com/sales-files-s3-bucket-structure-5B5NsjYGu1fBy15dqsEBRe)

## Step 3: Flatten Nested JSON (Distribution Only)

> [Code](https://github.com/theorchard/ows-royalties-workflows/blob/900ca4a973/dags/templates/accounting_run_calculate/snowflake_flatten_contract_json.py)

The `contract_term` and `contract_term_condition` tables contain JSON columns (attachments, conditions with stores/countries/transaction_types arrays). This step creates a cartesian product of every combination, inserting into `contract_denormalized_distro`.

Key fields in the denormalized table:
- `accounting_run_id`, `account_id`, `account_currency_code`
- `contract_id`, `contract_term_id`, `contract_term_condition_id`
- `term_type` (label, product, track), `term_rate`, `priority`
- `label_id`, `upc`, `isrc`, `country_id`, `store_id`, `transaction_type_id`

Only contracts with relevant `term_start`/`term_end` dates are included.

**Performance concern:** When users manually select nearly all stores/countries/types (e.g., 1855 of 1860) instead of "All", the cartesian product explodes. See [Calculation Timeout Investigation](../improvements/calculation-timeout.md).

## Step 4: Match Transactions to Contract Terms (Distribution Only)

> [Code](https://github.com/theorchard/ows-royalties-workflows/blob/900ca4a973/dags/templates/accounting_run_calculate/snowflake_match_contract_terms.py)

Compares `stmt_db_sales_distro` to `contract_denormalized_distro` to determine which contract terms apply to each transaction. Results go into `contract_transaction_staging`.

### Matching Criteria

Matching uses these fields, evaluated in order of term specificity:

1. **Track terms** — match on ISRC + UPC + label_id
2. **Product terms** — match on UPC + label_id
3. **Label terms** — match on label_id only

Within each level, additional filters apply:
- `country_id` (0 = all countries)
- `store_id` (0 = all stores)
- `transaction_type_id` (0 = all types)
- `priority` — for tiebreaking

**Conflict resolution:** If multiple contracts match the same transaction, the one with the highest `term_rate` wins (maximizing the client's net revenue).

**Unallocated sales:** Transactions that match no contract remain unallocated. They roll over to the next period.

### Double-Booking Risk

If two contracts in different run controllers share the same attachments (e.g., same UPC), both will match the same sales. This is intentional (enables "splits") but can accidentally double-book revenue. Keep sibling contracts in the same run controller.

## Step 5: Calculate Totals (Distribution Only)

> [Code](https://github.com/theorchard/ows-royalties-workflows/blob/900ca4a973/dags/templates/accounting_run_calculate/snowflake_calculate_run_results.py)

Applies term rates and currency conversions to produce `accounting_run_results_distro_staging`.

### Calculation Columns

**In sale currency:**
- `unit_price_sale_currency` = unit_price_usd / activity_rate
- `gross_revenue_sale_currency` = quantity * unit_price_sale_currency
- `withholding_tax_sale_currency` = withholding_tax_usd / activity_rate
- `gross_revenue_after_withholding_tax_sale_currency` = gross - WHT
- `net_revenue_sale_currency` = gross_after_WHT * (term_rate / 100)
- `distribution_fee_sale_currency` = (gross_after_WHT - net_revenue) * -1

**In payee currency** (same fields, multiplied by exchange_rate):
- `unit_price_payee_currency`, `gross_revenue_payee_currency`, etc.

## Step 6: Create Run Summary

Triggers the [`ledger_accounting_run_balance`](https://github.com/theorchard/lambda-abacus/blob/master/lambda/ledger_accounting_run_balance/README.md) lambda.

Sums results from the staging table by contract, rounded to two decimal places. Inserts into `ledger_accounting_run_balance` in MySQL with:
- `total_gross_revenue_amount`
- `total_net_revenue_amount`
- `distribution_fee`

This summary is what users see in the Abacus UI for review.

## Step 7: Export Summary to S3

Stages `ledger_accounting_run_balance` records as a TSV on S3. Updates `accounting_run.summary_export_url` so users can download it.

## Step 8: Set Status to "Complete"

Updates `accounting_run.run_status` to signal the calculation is done.