# CLAUDE.md

This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.

## Purpose

This Lambda receives Kafka events representing potential sound recording deliveries to TikTok (Audio Fingerprinting), applies eligibility rules to filter them, and kicks off an AWS Step Function execution for each eligible delivery. Ineligible deliveries are logged back to Kafka for audit purposes.

## Development Commands

Run from this directory (`lambda/sr-delivery-tiktok-filter`):

```bash
# Run linting (flake8 + yamllint + mypy) and unit tests
docker compose up --build lint-and-test

# Run tests only (skip linting)
SKIP_LINT=1 docker compose up --build lint-and-test

# Run a specific test
TEST_ARGS="-k test_new_osr_full_delivery" docker compose up --build lint-and-test

# Run with HTML coverage report (served on port 8000)
COV_REPORT=html docker compose up --build lint-and-test

# Start the function locally (HTTP on port 9000)
docker compose up --build -d function

# Send a test event to the running function
./tests/call.sh tests/sample_event.json

# Simulate without Docker (requires local venv with dependencies)
python simulate_event.py
```

Linting runs `yamllint`, `mypy src/app.py tests/unit/`, and `flake8 src/ tests/ config.py`. mypy config is in `mypy.ini`.

## Architecture

**Trigger:** AWS MSK (Kafka) connector. Input events are base64-encoded JSON records in the MSK Lambda connector format (`eventSource: aws:kafka`). Decoding happens in `src/utils/input_events.py`.

**Eligibility decision flow** (`src/delivery_eligibility.py` → `get_eligible_events`):

1. Batch-check all input ISRCs against `ows-masters-registry` (one API call for all events).
2. Fetch `OrchardSoundRecording` version from S3 for each event (`src/common/connectors/s3_sound_recordings`).
3. Apply preliminary checks: requires a valid S3 version, eligible asset type (flac > wav preferred), and — for new OSRs — requires registry presence and at least one non-carveout monetize/block policy via `soundrecording_utils.metadata.consolidation_logic.consolidate_rights`.
4. Fetch delivery history from GraphQL for the filtered subset (`src/connectors/graphql.py`, `deliveryHistoryByOsrIds` query).
5. For each surviving event, determine delivery type:
   - `PROHIBITED_ATTRIBUTE` → takedown if previously live, skip otherwise.
   - New OSR in registry with policies → `FULL_DELIVERY`.
   - No registry or no policies → `TAKEDOWN_DELIVERY` (always sent, even if last delivery was already a takedown).
   - Never been fully delivered → `FULL_DELIVERY`.
   - Metadata unchanged (TikTok-relevant fields only) → skip.
   - Last delivery was a takedown → `FULL_DELIVERY`; otherwise `METADATA_UPDATE`.

**`upload_asset` flag:** Set to `True` on `FULL_DELIVERY`. On takedowns, it is `True` only if no prior `FULL_DELIVERY` exists in history and there are assets present.

**SFN throttling** (`src/utils/step_function.py`): Before each batch of 10 events, the handler polls `list_executions(RUNNING)` and sleeps 30 s if `running + new > SFN_DELIVERY_MAX_RUNNING` (default 500). Uses exponential backoff for AWS throttling exceptions.

**Ineligible event logging:** `src/utils/kafka_logger.py` produces a JSON message to `KAFKA_TOPIC` with `event_type: not_eligible` and the reason string from `src/constants.py`.

**Metadata comparison** (`src/utils/metadata_filter.py`): Before deciding to send a `METADATA_UPDATE`, the two versions are serialized to JSON after stripping TikTok-irrelevant fields (`audio_attributes`, `rights_attributes`, `offer_type`, `ownership_rights`, audit dates, non-tiktok rules, several product fields). Comparison is a simple string equality check after `sort_keys=True`.

## Key Configuration (`config.py`)

| Variable | Source | Notes |
|---|---|---|
| `SFN_DELIVERY_ARN` | env | Required — ARN of the TikTok delivery Step Function |
| `SFN_DELIVERY_MAX_RUNNING` | env | Max concurrent SFN executions (default 500) |
| `KAFKA_BROKERS` | env | Required for Kafka producer |
| `KAFKA_TOPIC` | env | Topic for ineligible delivery logs |
| `ADMIN_IDENTITY_ID` / `ADMIN_PROFILE_ID` | hardcoded | Used as GraphQL request headers |

## Testing Patterns

Unit tests live in `tests/unit/`. All external calls are mocked:
- `src.common.connectors.s3_sound_recordings.get_sound_recording_version` — returns `(json_str, None)` tuples; use `side_effect` list when multiple calls are expected (current + previous version).
- `src.connectors.ows_masters_registry.masterrights` — returns a list of `{isrc, territories}` dicts or `None`.
- `src.connectors.graphql.get_delivery_histories` — returns `{osr_id: {type, has_full_delivery, last_delivered_version_id}}`.
- `src.common.connectors.kafka_producer.produce_message` — verify `not_eligible` messages via `mock_kafka.call_args`.

Test fixtures in `tests/conftest.py` provide pre-built `OrchardSoundRecording`-shaped dicts and MSK-format event inputs (`new_osr_event_input`, `updated_osr_event_input`). Prohibited attribute IDs are patched per-test via `@patch('src.constants.PROHIBITED_AUDIO_ATTRIBUTES', [...])`.

The `has_full_delivery` flag in GraphQL history excludes `FULL_DELIVERY` records whose `stepFunctionName` ends in `-False` (indicating a dry-run/aborted SFN execution).