# CLAUDE.md

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

## Purpose

Consumes Neo4J CDC (Change Data Capture) events from 10 Kafka/MSK topics dealing with changes at the Node, as opposed to the Edge level,  (`cdc.musicGraph.*` and `cdc.musicGraphV5.*` variants of `orchardAsset`, `track`, `product`, `vendor`, `subAccount`), filters for significant changes, and triggers the fingerprinter AWS Step Function for relevant ones. Also handles primary track reassignment when a product deletion event is detected.

## Development Commands

Run from this directory:

```bash
# Run linting and tests
docker compose up --build lint-and-test

# Run with exit codes (CI-like)
docker compose up --exit-code-from lint-and-test --abort-on-container-exit --build lint-and-test

# Start Lambda locally (port 9000)
docker compose up --build -d function

# Invoke locally
curl -XPOST "http://localhost:9000/2015-03-31/functions/function/invocations" -d @tests/sample_events/neo4j_v5/track/isrc_added.json

# Skip linting
SKIP_LINT=true docker compose up --build lint-and-test

# Run a single test
TEST_ARGS="-k test_track_isrc_added" docker compose up --build lint-and-test

# Quick local invocation without Docker
python simulate_event.py
```

**Before running Docker:** ensure `~/.aws/credentials` has an up-to-date `default` profile (use `awsume` then copy the exported vars).

## Architecture

### Data Flow

```
MSK Kafka Event (base64-encoded CDC records)
  → decode_records()         # base64-decode, JSON-parse, attach topic as source key
  → filter_events()          # drop non-UPDATE ops; drop if only lastModifiedBy/lastModifiedAt changed
  → update_primary_track()   # for Product deletion events: reassign OSR primary track via GraphQL
  → generate_outputs()       # dispatch to per-entity event class → list of OutputEvent
  → throttle()               # wait if running SFN executions ≥ SFN_FINGERPRINTER_MAX_RUNNING (default 500)
  → execute_sfn()            # start fingerprinter SFN, idempotent via execution name
  → return summary {num_input_records, sfn_executions: {started, skipped}}
```

### Event Classes (`src/events/neo4j_v5/`)

Each entity type has its own class extending `BaseV5Event` that encodes which operations and property changes are fingerprint-worthy:

| Class | Triggers on |
|---|---|
| `Track` | ISRC added, name change, offerType change, creation (music+Orchard label only; ignores DELETE) |
| `Product` | Release status involving `in_content`, release date change, notForDistribution flag changes, deletions flag flipping to True (Orchard label only; UPDATE only) |
| `OrchardAsset` | CREATE only |
| `Vendor` | Status change to/from `signed` (UPDATE only) |
| `SubAccount` | `isDeleted` flag change (UPDATE only) |

`BaseV5Event` parses Neo4J CDC property format `{type: value}` (e.g. `{'S': 'foo'}`, `{'I64': 123}`, `{'B': True}`).

### OutputEvent

`exc_id()` produces the SFN execution name: `{data_type}-{data_id}-{action}-{unique_id}`. This makes executions idempotent — re-processing the same CDC event re-uses the same execution name and `ExecutionAlreadyExists` is caught and ignored.

`to_json()` serializes as `{id, label, operation, timestamp, payload}` for the fingerprinter SFN input.

### Primary Track Logic (`src/utils/events.py:update_primary_track`)

Only fires when a Product event has `is_deletion_event=True` (deletions flag changed False→True). Uses GraphQL to:
1. Fetch all tracks for the product's UPC
2. Find which track is being deleted (marked as primary)
3. Find OSRs sharing that track's ISRC
4. Reassign the OSR's primary track to the lowest-tuid active track

### Throttling (`src/utils/sfn_handler.py:throttle`)

Lists all RUNNING SFN executions; if `running + new_events > max`, sleeps 30 seconds and retries (up to 5 retries with exponential backoff + jitter on `ThrottlingException`).

## Key Configuration

`config.py` reads these env vars (set in `.env` for local dev):

| Variable | Required | Default |
|---|---|---|
| `SFN_FINGERPRINTER_ARN` | Yes | — |
| `SFN_FINGERPRINTER_MAX_RUNNING` | No | `500` |
| `ENVIRONMENT` | No | `dev` |
| `SENTRY_DSN` | No | — |

## Tests

Unit tests live in `tests/unit/`, organized to mirror `src/`. Sample CDC events in `tests/sample_events/neo4j_v5/<entity>/` are used in integration invocations. `tests/conftest.py` provides `neo4j_v5_cdc_node_event()` and `msk_event()` helpers for constructing mock MSK payloads.

`mypy.ini` is present — type-checking runs as part of lint.