# Runbook: Flowthrough Operations

Step-by-step operational guide for monitoring, managing, and troubleshooting the flowthrough adjustment processing system.

**Last Updated:** 2026-05-13

---

## Incident Triage — First 5 Minutes

If you're on-call and get paged, start here:

1. **Check Datadog** — any Lambda errors in the last 30 minutes?
   ```
   service:lambda-abacus-* status:error environment:prod
   ```
2. **Check DLQ depth:**
   ```bash
   aws sqs get-queue-attributes \
     --queue-url https://sqs.us-east-1.amazonaws.com/437795906767/prod-adjustment-file-ingest-workflow-dlq \
     --attribute-names ApproximateNumberOfMessages
   ```
3. **Check Step Function executions** — any Running or Failed in the last hour? [Prod Console](https://437795906767-d4xbig47.us-east-1.console.aws.amazon.com/states/home?region=us-east-1#/statemachines/view/arn:aws:states:us-east-1:437795906767:stateMachine:prod-adjustment-file-ingest-workflow)
4. **Check Airflow** — any stuck or failed DAG runs? [Prod Airflow](https://19a7ad7c-ad6d-47a0-8e89-389ecf29a83b-vpce.c7.us-east-1.airflow.amazonaws.com/home)
5. **Check DB connectivity:**
   ```sql
   SELECT 1;
   ```
6. **Check outbox** — any events stuck in pending?
   ```sql
   SELECT COUNT(*) FROM abacus_outbox
   WHERE status = 'pending'
     AND created_at < DATE_SUB(NOW(), INTERVAL 5 MINUTE);
   ```

Then use the [Severity Matrix](#severity-matrix) below to classify and the [Escalation Ladder](#escalation-ladder) to determine who to contact.

---

## Severity Matrix

| Symptom | Severity | Response Time | Action |
|---------|----------|---------------|--------|
| All uploads failing | **P0** | Immediate | Page team lead, investigate file upload SF + S3 |
| Apply Lambda creating incorrect ledger entries | **P0** | Immediate | Stop applies, escalate to finance + engineering |
| Payment allocation Lambda not firing after close_balance | **P1** | Within 1 hour | Impacts statement period close deadline |
| Step Function failure rate > 3 in 5 min | **P1** | Within 1 hour | Check InitializeBatch Lambda, EventBridge |
| DLQ depth >= 3 | **P2** | Within 4 hours | Replay DLQ messages, check EventBridge rule |
| DLQ depth > 0 but < 3 | **P3** | Next business day | Investigate and replay |
| Single file stuck in processing | **P3** | Next business day | Check abacus_state, re-trigger if needed |
| Validation Lambda slow but succeeding | **P3** | Next business day | Check file size, Snowflake performance |

---

## Escalation Ladder

| Level | Who | When | Contact |
|-------|-----|------|---------|
| L1 — On-call | Current on-call engineer | Any alert fires | PagerDuty |
| L2 — Team lead | Michael Rojas | L1 cannot resolve in 30 min, or financial impact | Slack DM or phone |
| L3 — Backend | Jonathan Raysor, Aarti Navalu, Serhii Laposhko | Lambda/API code issue | Slack: [#tmp-flowthrough-automation](https://sonymusic.enterprise.slack.com/archives/C09RJRL9924) |
| L4 — Data | Harmanpreet Singh | Snowflake view / data issue | Slack: [#tmp-flowthrough-automation](https://sonymusic.enterprise.slack.com/archives/C09RJRL9924) |
| L5 — DBA | DBA on-call | DB deadlocks, connection pool, schema issues | PagerDuty (DBA service) |
| L6 — Infra | Platform/infra on-call | Step Functions, EventBridge, IAM, Kafka CDC | PagerDuty (infra service) |
| L7 — Finance | Accounting stakeholders | Financial data corruption, incorrect payments | Contact via team lead |

---

## Table of Contents

1. [Where to Find Things](#1-where-to-find-things)
2. [Step Function Operations](#2-step-function-operations)
3. [Lambda Operations](#3-lambda-operations)
4. [Airflow DAG Operations](#4-airflow-dag-operations)
5. [S3 File Management](#5-s3-file-management)
6. [EventBridge Operations](#6-eventbridge-operations)
7. [Feature Flag Management](#7-feature-flag-management)
8. [Monitoring and Alerting](#8-monitoring-and-alerting)
9. [Common Operational Procedures](#9-common-operational-procedures)

---

## 1. Where to Find Things

### AWS Console Links

| Resource | QA | Prod |
|----------|-----|------|
| Step Functions | [QA Console](https://us-east-1.console.aws.amazon.com/states/home?region=us-east-1#/statemachines) - search `qa-adjustment-file-ingest-workflow` | [Prod Console](https://us-east-1.console.aws.amazon.com/states/home?region=us-east-1#/statemachines) - search `prod-adjustment-file-ingest-workflow` |
| Lambdas | [QA Console](https://us-east-1.console.aws.amazon.com/lambda/home?region=us-east-1#/functions) - search `adjustment` or `payment-allocation` | [Prod Console](https://us-east-1.console.aws.amazon.com/lambda/home?region=us-east-1#/functions) - search `adjustment` or `payment-allocation` |
| S3 Bucket | `qa-abacus-adjustments` | `prod-abacus-adjustments` |
| CloudWatch Logs | [QA Logs](https://us-east-1.console.aws.amazon.com/cloudwatch/home?region=us-east-1#logsV2:log-groups) - search `adjustment-file-ingest` | [Prod Logs](https://us-east-1.console.aws.amazon.com/cloudwatch/home?region=us-east-1#logsV2:log-groups) - search `adjustment-file-ingest` |
| EventBridge Rules | [QA Rules](https://us-east-1.console.aws.amazon.com/events/home?region=us-east-1#/rules) - search `adjustment-file-ingest` | [Prod Rules](https://us-east-1.console.aws.amazon.com/events/home?region=us-east-1#/rules) - search `adjustment-file-ingest` |
| DLQ (SQS) | `qa-adjustment-file-ingest-workflow-dlq` | `prod-adjustment-file-ingest-workflow-dlq` |

### Datadog Links

| Resource | Link |
|----------|------|
| Lambda Logs (Prod) | [Datadog Logs](https://sonymusic-pde.datadoghq.com/logs?query=service%3Alambda-abacus-adjustment-file-validation%20environment%3Aprod) |
| Lambda Logs (QA) | Same link, filter `environment:qa` |
| Step Function Monitors | Search Datadog Monitors for `adjustment-file-ingest` |

### Airflow Links

| DAG | QA | Prod |
|-----|-----|------|
| `adjustment_file_upload` | QA Airflow UI -> DAGs -> search `adjustment_file_upload` | Prod Airflow UI -> DAGs |
| `adjustment_file_import` | QA Airflow UI -> DAGs -> search `adjustment_file_import` | Prod Airflow UI -> DAGs |

### Code Repos

| Repo | What to look for |
|------|------------------|
| [lambda-abacus](https://github.com/theorchard/lambda-abacus) | Lambda source code in `lambda/adjustment_file_*` and `lambda/adjustments_apply` and `lambda/payment_allocation` |
| [ows-royalties](https://github.com/theorchard/ows-royalties) | Backend API in `royalties/blueprints/statement_period_adjustment_file.py`, models in `royalties/models/`, flowthrough in `abacus_contract/` and `abacus_flowthrough/` |
| [frontend-royalties](https://github.com/theorchard/frontend-royalties) | UI components in `src/components/adjustments/` |
| [terraform-infra](https://github.com/theorchard/terraform-infra) | Infrastructure in `accounting/qa/ows-royalties-workflows/` and `accounting/qa/lambda-abacus/` |
| [ows-royalties-workflows](https://github.com/theorchard/ows-royalties-workflows) | DAGs in `dags/adjustment_file_upload.py` and `dags/adjustment_file_import.py` |

### Database

| Environment | Host | Database |
|-------------|------|----------|
| QA | QA Aurora MySQL cluster | `royalty_accounting` |
| Prod | Prod Aurora MySQL cluster | `royalty_accounting` |

> **Read replica guidance:** For all diagnostic SELECT queries (everything in the debugging and data verification runbooks), use the **reader endpoint** (read replica). Only use the **writer endpoint** (primary) for UPDATE/DELETE operations, which require DBA approval in production.

---

## 2. Step Function Operations

### Viewing Executions

1. Go to AWS Step Functions Console
2. Find `{env}-adjustment-file-ingest-workflow`
3. Click "Executions" tab
4. Filter by status: Running, Succeeded, Failed, Timed Out, Aborted

### Understanding Execution Input

Each execution receives an event like:

```json
{
  "source": "abacus.outbox",
  "detail-type": "file_upload.completed",
  "detail": {
    "metadata": {
      "target_id": 12345,
      "target_type": "file_upload",
      "correlation_id": "abc-123"
    },
    "data": {
      "upload_type": "adjustments"
    }
  }
}
```

- `target_id` = `file_upload_id` in the database
- `correlation_id` = trace ID across all services

### Manually Starting an Execution

If you need to re-trigger processing for a file:

1. Go to Step Functions Console -> `{env}-adjustment-file-ingest-workflow`
2. Click "Start execution"
3. Paste the event JSON (get from DLQ or reconstruct from DB)
4. Click "Start execution"

**Warning:** Only do this if you've confirmed the file_upload record exists and has status `complete`.

### Handling Failed Executions

1. Click on the failed execution
2. Click on the failed state (red)
3. Check the "Error" and "Cause" fields
4. Check "Input" to see what data was passed
5. Check CloudWatch logs for the Lambda that failed

**Common failures:**
- `InitializeBatchError`: file_upload not found, wrong status, or statement_period issue
- `PrepareFileError` (QA only): The QA state machine includes a `PrepareFile` state that calls the deprecated `adjustment-file-prepare` Lambda — this will always fail. This is expected; the important thing is that the `InitializeBatch` step succeeds. See the [TRD Step Function section](../../technical-projects/flowthrough/TRD.md#63-step-functions) for details.

### Stopping a Running Execution

1. Go to the execution detail page
2. Click "Stop execution"
3. Optionally provide error and cause

---

## 3. Lambda Operations

### Lambda Reference

| Lambda | Function Name Pattern | Repo Path |
|--------|----------------------|-----------|
| Initialize | `{env}-lambda-abacus-adjustment-file-initialize` | `lambda/adjustment_file_initialize/` |
| Validation | `{env}-lambda-abacus-adjustment-file-validation` | `lambda/adjustment_file_validation/` |
| Import | `{env}-lambda-abacus-adjustment-file-import` | `lambda/adjustment_file_import/` |
| Apply | `{env}-lambda-abacus-adjustments-apply` | `lambda/adjustments_apply/` |
| Payment Allocation | `{env}-lambda-abacus-payment-allocation` | `lambda/payment_allocation/` |

### Viewing Lambda Logs

**Via Datadog:**

```
service:lambda-abacus-adjustment-file-validation environment:prod
```

Common log searches:
- Find a specific file: `"statement_period_adjustment_file_id" "12345"`
- Find errors: `service:lambda-abacus-adjustment-file-* status:error`

**Via CloudWatch:**

1. Go to Lambda Console -> select function
2. Click "Monitor" tab -> "View CloudWatch logs"
3. Or go directly to Log Group: `/aws/lambda/{env}-lambda-abacus-adjustment-file-*`

### Manually Invoking a Lambda

**Via AWS Console:**

1. Go to Lambda Console -> select function
2. Click "Test" tab
3. Create test event with the appropriate input:

**adjustment-file-initialize:**
```json
{
  "detail-type": "file_upload.completed",
  "detail": {
    "metadata": {
      "target_id": 12345,
      "target_type": "file_upload"
    },
    "data": {
      "upload_type": "adjustments"
    }
  }
}
```

**adjustment-file-validation:**
```json
{
  "target_id": 12345,
  "statement_period_id": 67
}
```
(`target_id` = `statement_period_adjustment_file_id`)

**adjustment-file-import:**
```json
{
  "abacus_event_id": 789,
  "target_id": 12345
}
```
(`target_id` = `statement_period_adjustment_file_id`)

**adjustments-apply:**

> **CAUTION — POINT OF NO RETURN:** Applying adjustments creates ledger entries (`ledger_adjustment_applied`, `ledger_account_contract`/`ledger_contract_flowthrough`, `ledger_deposit`) that are difficult to reverse. Reversal requires the multi-step rollback procedure in Section 9. Verify the adjustment data is correct BEFORE invoking. Use the [Data Verification Runbook](data-verification.md) Section 3 to confirm import data.

```json
{
  "abacus_event_id": 789,
  "target_id": 12345
}
```
(`target_id` = `statement_period_adjustment_file_id`)

**payment-allocation:**
```json
{
  "detail": {
    "metadata": {
      "target_id": 456
    }
  }
}
```
(`target_id` = `statement_period_payment_entity_id`)

**Via AWS CLI:**

```bash
aws lambda invoke \
  --function-name "{env}-lambda-abacus-adjustment-file-validation" \
  --payload '{"target_id": 12345, "statement_period_id": 67}' \
  --cli-binary-format raw-in-base64-out \
  output.json
```

### Running a Lambda Locally

See the Lambda's README in `lambda-abacus` for local setup. General steps:

1. `cd lambda/adjustment_file_validation`
2. Copy `.env.example` to `.env` and fill in DB/S3/Snowflake credentials
3. Edit `tests/sample_event.json` with your test data
4. Run `make run`

**Tip:** To avoid side effects, comment out the OWS service calls and S3 uploads before running locally. See [troubleshooting doc](../troubleshooting_adjustment_file_validation.md) for specifics.

### Lambda Environment Variables

All Lambdas share:

| Variable | Description |
|----------|-------------|
| `ENVIRONMENT` | dev, qa, uat, prod |
| `MYSQL_DB_HOST` | Aurora MySQL endpoint |
| `MYSQL_DB_USER` | Database user |
| `MYSQL_DB_PASS` | From Secrets Manager |
| `MYSQL_DB_NAME` | `royalty_accounting` |
| `MYSQL_DB_PORT` | `3306` |
| `SENTRY_DSN` | Sentry error tracking |
| `LOGGING_LEVEL` | INFO (default) |

Lambda-specific:

| Lambda | Variable | Description |
|--------|----------|-------------|
| Initialize | `ORCHARD_IDENTITY_ID` | Feature flag identity |
| Validation | `S3_BUCKET_NAME` | Adjustments S3 bucket |
| Validation | `SNOWFLAKE_CONFIG` | Snowflake connection JSON |
| Validation | `FIVETRAN_SCHEMA` | Fivetran replication schema |
| Import | `S3_BUCKET_NAME` | Adjustments S3 bucket |
| Import | `MYSQL_BATCH_SIZE` | Batch size for bulk inserts |
| Apply | `BATCH_SIZE_GET` | Batch size for reads (default 5000) |
| Apply | `BATCH_SIZE_POST` | Batch size for writes (default 100) |

---

## 4. Airflow DAG Operations

### Triggering a DAG Manually

1. Go to Airflow UI
2. Find `adjustment_file_upload` or `adjustment_file_import`
3. Click "Trigger DAG" (play button)
4. Provide configuration JSON if needed

### Viewing DAG Run Status

1. Go to Airflow UI -> DAGs -> `adjustment_file_upload`
2. Click on the run (by date/time)
3. View task status in Graph view or Tree view

### Common DAG Failures

| Task | Common Failure | Resolution |
|------|---------------|------------|
| `invoke_lambda_av_scan` | Lambda timeout, infected file | Check quarantine status in `file_upload` table |
| `validate_format` | Invalid .xlsx structure | Download file from S3, inspect manually |
| `invoke_adjustment_file_validation_lambda` | Lambda error | Check Lambda logs in Datadog |
| `adjustment_file_validation_check_result` | Sensor timeout (15min) | Lambda is still processing; check if file is very large |
| `check_adjustment_file_error_report` | Validation found errors | Expected behavior; check error report in S3 |
| `invoke_adjustment_file_import_lambda` | Import Lambda error | Check for deadlocks, DB connectivity |

### Clearing Failed Tasks

1. Go to the failed DAG run
2. Click on the failed task
3. Click "Clear" to retry
4. Optionally clear downstream tasks too

---

## 5. S3 File Management

### Accessing Files

**Via AWS CLI:**

```bash
# List files for a statement period
aws s3 ls s3://{env}-abacus-adjustments/adjustments/ --recursive

# Download an adjustment file
aws s3 cp s3://{env}-abacus-adjustments/{s3_key} ./local_file.xlsx

# Download error report
aws s3 cp s3://{env}-abacus-adjustments/{invalid_file_location} ./error_report.xlsx
```

**Via AWS Console:**

1. Go to S3 Console
2. Navigate to `{env}-abacus-adjustments`
3. Browse by path

### Getting S3 Paths from the Database

```sql
-- Get file locations for a specific adjustment file
SELECT
    statement_period_adjustment_file_id,
    file_name,
    valid_file_location,
    invalid_file_location,
    error_type,
    valid_row_count,
    invalid_row_count
FROM statement_period_adjustment_file
WHERE statement_period_adjustment_file_id = {id};
```

### Template File

The adjustment template is stored at:
```
s3://{env}-abacus-adjustments/abacus_adjustments_template/Adjustment_Template.xlsx
```

Download: Use the Abacus UI or `aws s3 cp`.

---

## 6. EventBridge Operations

### Viewing Events

**Via CloudWatch Logs Insights:**

EventBridge events are logged in the Step Function log group.

```
fields @timestamp, @message
| filter @message like /file_upload.completed/
| sort @timestamp desc
| limit 20
```

### Viewing DLQ Messages

Failed events land in the DLQ. To inspect:

```bash
# Receive messages (non-destructive peek)
aws sqs receive-message \
  --queue-url https://sqs.us-east-1.amazonaws.com/{account-id}/{env}-adjustment-file-ingest-workflow-dlq \
  --max-number-of-messages 10 \
  --visibility-timeout 0

# Get approximate message count
aws sqs get-queue-attributes \
  --queue-url https://sqs.us-east-1.amazonaws.com/{account-id}/{env}-adjustment-file-ingest-workflow-dlq \
  --attribute-names ApproximateNumberOfMessages
```

### Replaying DLQ Messages

1. Read the message from DLQ
2. Extract the event payload
3. Manually start a Step Function execution with that payload (see Section 2)
4. Delete the message from the DLQ after successful processing

---

## 7. Feature Flag Management

### Split.io Dashboard

Feature flags are managed in Split.io. Log into the Split.io dashboard to view/modify flags.

### Checking Flag Status

**In code:** Feature flags are checked via the Split.io SDK using the `ORCHARD_IDENTITY_ID`.

**Common flags and their effects:**

| Flag | OFF behavior | ON behavior |
|------|-------------|------------|
| `abacus_flowthrough_automation` | Initialize Lambda creates batch only | Also creates `statement_period_adjustment_file`, `abacus_state`, `abacus_event` |
| `ABACUS_APPLY_FLOWTHROUGH_PAYMENT` <!-- flag torn down 2026-06; see ACC-10470 --> | `apply_to_flowthrough_payment` field ignored | Field validated, imported, and used for `ledger_contract_flowthrough` entries |
| `abacus_auto_generate_adjustments_flowthrough` | No "Generate" button in UI | Users can auto-generate flowthrough batches |

### Enabling/Disabling Flags

1. Log into Split.io
2. Find the flag by name
3. Adjust targeting rules (by environment, identity, percentage)
4. Save changes (propagates within seconds)

**[DEPRECATED 2026-06 — flag torn down per ACC-10470]** ~~**Warning:** Changing `ABACUS_APPLY_FLOWTHROUGH_PAYMENT` mid-processing can cause inconsistent data. Only toggle between statement periods.~~

---

## 8. Monitoring and Alerting

### Datadog Monitors

| Monitor | What it means | Action |
|---------|---------------|--------|
| Step Function Failures >= 3 (5min) | Multiple files failing to process | Check Step Function executions for common error pattern |
| DLQ Depth >= 3 (5min) | Events not being processed | Check EventBridge rule, Step Function health |
| Step Function Duration >= 900s | Processing taking too long | Check for large files or Lambda issues |
| Lambda Errors (any) | A Lambda has thrown an exception | Check Datadog/CloudWatch logs for error details |
| Lambda Duration >= timeout | Lambda is hitting its time limit | Check file size, DB performance, Snowflake connectivity |

### Creating a Custom Dashboard

Key metrics to track:

```
# Step Function executions
aws.states.executions_started{statemachinename:*-adjustment-file-ingest-workflow}
aws.states.execution_succeeded{statemachinename:*-adjustment-file-ingest-workflow}
aws.states.execution_failed{statemachinename:*-adjustment-file-ingest-workflow}

# Lambda invocations
aws.lambda.invocations{functionname:*-adjustment-file-*}
aws.lambda.errors{functionname:*-adjustment-file-*}
aws.lambda.duration{functionname:*-adjustment-file-*}

# DLQ depth
aws.sqs.approximate_number_of_messages_visible{queuename:*-adjustment-file-ingest-workflow-dlq}
```

---

## 9. Common Operational Procedures

### Procedure: Re-process a Failed Adjustment File

1. **Identify the file:**
   ```sql
   SELECT * FROM statement_period_adjustment_file
   WHERE statement_period_adjustment_file_id = {id};
   ```

2. **Check the abacus_state:**
   ```sql
   SELECT * FROM abacus_state
   WHERE target_id = {id} AND target_type = 'statement_period_adjustment_file';
   ```

3. **Determine which step failed** (look for state = 'error')

4. **Fix the root cause** (file issue, Lambda bug, DB issue)

5. **Reset the state** (if needed):
   ```sql
   -- ALWAYS verify first: confirm exactly 1 row will be updated
   SELECT abacus_state_id, target_id, action, state
   FROM abacus_state
   WHERE target_id = {id}
     AND target_type = 'statement_period_adjustment_file'
     AND action = '{failed_action}';
   -- Verify row count = 1 before proceeding

   BEGIN;
   UPDATE abacus_state
   SET state = 'init'
   WHERE target_id = {id}
     AND target_type = 'statement_period_adjustment_file'
     AND action = '{failed_action}';

   -- Verify the update
   SELECT action, state FROM abacus_state
   WHERE target_id = {id}
     AND target_type = 'statement_period_adjustment_file';
   COMMIT; -- or ROLLBACK if verification fails
   ```

6. **Re-trigger** via Airflow DAG or Lambda manual invocation

### Procedure: Cancel an In-Progress Upload

1. **Via UI:** User clicks "Cancel Import" button -> soft deletes the file
2. **Via DB:**
   ```sql
   -- Pre-check: verify no applied ledger entries exist for this file
   SELECT COUNT(*) AS applied_count FROM ledger_adjustment_applied
   WHERE statement_period_adjustment_file_id = {id};
   -- If applied_count > 0, DO NOT delete. Applied adjustments are already in the ledger.
   -- Use the rollback procedure instead.

   BEGIN;
   UPDATE statement_period_adjustment_file
   SET deleted_at = NOW(), deleted_by = '{your_identity}'
   WHERE statement_period_adjustment_file_id = {id};
   COMMIT;
   ```
3. **Stop any running Step Function executions** for this file

### Procedure: Investigate Missing Payment Allocations

1. **Check if close_balance is complete:**
   ```sql
   SELECT * FROM abacus_state
   WHERE target_type = 'close_balance'
     AND target_id = {statement_period_payment_entity_id}
     AND state = 'complete';
   ```

2. **Check if the payment-allocation Lambda was triggered:**
   - Look in Datadog for `service:lambda-abacus-payment-allocation`
   - Check EventBridge for `close_balance.completed` events

3. **Check for unlinked adjustments:**
   ```sql
   SELECT COUNT(*) FROM ledger_adjustment_applied laa
   WHERE laa.statement_period_id = {id}
     AND laa.apply_to_flowthrough_payment = 1
     AND NOT EXISTS (
       SELECT 1 FROM payment_allocation_ledger_adjustment pala
       WHERE pala.ledger_adjustment_applied_id = laa.ledger_adjustment_applied_id
     );
   ```

4. **Manually trigger if needed** (see Lambda invocation in Section 3)

### Procedure: Rolling Back Applied Adjustments

**This is a destructive operation. Requires DBA + Finance approval before proceeding.**

> **Point of no return check:** If payment allocations have `payment_status != 'init'`, payments may have been processed downstream. **STOP and escalate to finance** — reversal may require a compensating adjustment rather than a delete.

#### Step 0: Scope the rollback

```sql
-- Identify which file(s) need rollback
SELECT
    spaf.statement_period_adjustment_file_id,
    spaf.file_name,
    COUNT(laa.ledger_adjustment_applied_id) AS applied_count,
    SUM(laa.adjustment_amount) AS total_amount
FROM statement_period_adjustment_file spaf
JOIN ledger_adjustment_applied laa
    ON laa.statement_period_adjustment_file_id = spaf.statement_period_adjustment_file_id
WHERE spaf.statement_period_adjustment_file_id = {id}
GROUP BY spaf.statement_period_adjustment_file_id, spaf.file_name;
```

#### Step 1: Check if payment allocations have been processed downstream

```sql
SELECT
    pa.payment_allocation_id,
    pa.payment_status,
    pa.ledger_status,
    pa.amount_to_payment,
    pa.currency_code
FROM payment_allocation pa
JOIN payment_allocation_ledger_adjustment pala
    ON pala.payment_allocation_id = pa.payment_allocation_id
JOIN ledger_adjustment_applied laa
    ON laa.ledger_adjustment_applied_id = pala.ledger_adjustment_applied_id
WHERE laa.statement_period_adjustment_file_id = {id};
-- If ANY row has payment_status != 'init' or ledger_status != 'init':
-- STOP. Escalate to finance. Do NOT proceed with deletion.
```

#### Step 2: Delete payment allocation links and orphaned allocations

> **Important:** Steps 2-3 should be executed in a single session. Do NOT commit Step 2 and walk away — if Step 3 fails after Step 2 commits, data will be in an inconsistent state (allocation links removed but ledger entries still present).

```sql
BEGIN;

-- 2a. Capture affected allocation IDs BEFORE deleting links
CREATE TEMPORARY TABLE _rollback_allocation_ids AS
SELECT DISTINCT pala.payment_allocation_id
FROM payment_allocation_ledger_adjustment pala
JOIN ledger_adjustment_applied laa
    ON laa.ledger_adjustment_applied_id = pala.ledger_adjustment_applied_id
WHERE laa.statement_period_adjustment_file_id = {id};

-- 2b. Delete links (FK child)
DELETE FROM payment_allocation_ledger_adjustment
WHERE ledger_adjustment_applied_id IN (
    SELECT ledger_adjustment_applied_id
    FROM ledger_adjustment_applied
    WHERE statement_period_adjustment_file_id = {id}
);

-- 2c. Delete ONLY the allocations from this file that now have zero remaining links
DELETE FROM payment_allocation
WHERE payment_allocation_id IN (SELECT payment_allocation_id FROM _rollback_allocation_ids)
  AND payment_allocation_id NOT IN (
      SELECT DISTINCT payment_allocation_id
      FROM payment_allocation_ledger_adjustment
  );

DROP TEMPORARY TABLE _rollback_allocation_ids;

-- Verify: should return 0 links for this file
SELECT COUNT(*) FROM payment_allocation_ledger_adjustment pala
JOIN ledger_adjustment_applied laa
    ON laa.ledger_adjustment_applied_id = pala.ledger_adjustment_applied_id
WHERE laa.statement_period_adjustment_file_id = {id};
```

**Do NOT commit yet — proceed immediately to Step 3.**

#### Step 3: Delete ledger entries (same transaction)

> **Note on FK columns:** The DELETE queries below assume `ledger_deposit`, `ledger_contract_flowthrough`, and `ledger_account_contract` have a `ledger_adjustment_applied_id` column. **Verify this against the actual schema before executing** — the ERD in the TRD may be incomplete. If these tables do not have this FK, you will need to identify the correct join path (e.g., matching on `account_id + contract_id + statement_period_id` with a source filter). Run the SELECT version of each query first to confirm it returns the expected rows.

```sql
-- 3a. Delete ledger deposit entries
-- PRE-CHECK: Run as SELECT first to verify row count
DELETE FROM ledger_deposit
WHERE ledger_adjustment_applied_id IN (
    SELECT ledger_adjustment_applied_id
    FROM ledger_adjustment_applied
    WHERE statement_period_adjustment_file_id = {id}
);

-- 3b. Delete flowthrough ledger entries
DELETE FROM ledger_contract_flowthrough
WHERE ledger_adjustment_applied_id IN (
    SELECT ledger_adjustment_applied_id
    FROM ledger_adjustment_applied
    WHERE statement_period_adjustment_file_id = {id}
);

-- 3c. Delete non-flowthrough ledger entries
DELETE FROM ledger_account_contract
WHERE ledger_adjustment_applied_id IN (
    SELECT ledger_adjustment_applied_id
    FROM ledger_adjustment_applied
    WHERE statement_period_adjustment_file_id = {id}
);

-- 3d. Delete applied adjustment details
DELETE FROM ledger_adjustment_detail_applied
WHERE ledger_adjustment_applied_id IN (
    SELECT ledger_adjustment_applied_id
    FROM ledger_adjustment_applied
    WHERE statement_period_adjustment_file_id = {id}
);

-- 3e. Delete applied adjustments
DELETE FROM ledger_adjustment_applied
WHERE statement_period_adjustment_file_id = {id};

-- Verify: should return 0
SELECT COUNT(*) FROM ledger_adjustment_applied
WHERE statement_period_adjustment_file_id = {id};

COMMIT; -- Commits both Step 2 and Step 3 together
-- or ROLLBACK if any verification fails
```

#### Step 4: Reset abacus_state

```sql
BEGIN;

-- Pre-check: verify exactly 1 row for apply_file
SELECT abacus_state_id, action, state
FROM abacus_state
WHERE target_id = {id}
  AND target_type = 'statement_period_adjustment_file'
  AND action = 'apply_file';

-- Reset apply_file state back to init
UPDATE abacus_state
SET state = 'init'
WHERE target_id = {id}
  AND target_type = 'statement_period_adjustment_file'
  AND action = 'apply_file';

-- Verify
SELECT action, state
FROM abacus_state
WHERE target_id = {id}
  AND target_type = 'statement_period_adjustment_file';
-- apply_file should now be 'init'; all others unchanged

COMMIT;
```

#### Step 5: Verify clean state

Run the end-to-end reconciliation query from [data-verification.md](data-verification.md) Section 6 to confirm the file has no remaining ledger entries. The file should still show as "Approved" (not "Applied") in the UI and can be re-applied after the underlying issue is fixed.

### Procedure: Handling Quarantined Files

If a file is quarantined (virus scan flagged it):

1. **Check file_upload status:**
   ```sql
   SELECT * FROM file_upload
   WHERE file_upload_id = {id} AND upload_status = 'quarantined';
   ```

2. The file has been moved to the quarantine bucket by the `abacus_file_upload_complete` Lambda

3. **Contact the uploader** to verify the file and re-upload a clean version

4. Quarantined files cannot be re-processed - a new upload is required