# 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 two MSK Kafka topics — `cdc.musicGraphV5.hasAsset` and `cdc.musicGraphV5.hasFingerprintRule` — filters them for relevance, then triggers the Fingerprinter Step Function (SFN) for each qualifying event.

## Development Commands

Run from this directory:

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

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

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

# Get HTML coverage report (served on port 8000 after tests complete)
COV_REPORT=html docker compose up --build lint-and-test

# Run lambda locally (port 9000)
docker compose up --build -d function

# Invoke with a sample event
./tests/sample_events/call.sh tests/sample_events/neo4j/has_assets/created_v5.json

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

Linting includes `flake8`, `yamllint`, and `mypy` (typed defs checked via `mypy.ini`).

## Architecture

### Processing Pipeline (`src/app.py` → handler)

1. **`decode_records()`** — Decodes base64-encoded MSK records, tags each with its source as `aws:kafka:<topic>`
2. **`filter_events()`** — Drops UPDATE events where only `lastModifiedBy`/`lastModifiedAt` changed (noise reduction)
3. **`generate_outputs()`** — Maps each `(source, raw_event)` tuple to a typed `InputEvent` subclass; calls `.output()` to get zero or more `OutputEvent` objects
4. **`throttle()`** — Lists running SFN executions; sleeps `SFN_THROTTLE_SLEEP_SECONDS` in a loop until `num_running + num_new < SFN_FINGERPRINTER_MAX_RUNNING`
5. **`execute_sfn()`** — Starts an SFN execution; silently skips `ExecutionAlreadyExists` (idempotency via `exc_id()`)

### Event Class Hierarchy

All event classes live in `src/events/`.

**`InputEvent`** (base, `src/events/__init__.py`) — has a single abstract method `.output() → OutputEvent | list[OutputEvent] | None`.

**`BaseV5Event`** (`neo4j_v5/base.py`) — parses raw Neo4j V5 CDC structure (txId, operation, before/after state, timestamp). Extended by node-level events:

| Class | Fires when |
|---|---|
| `Track` | Created with ISRC, or updated with ISRC assignment / name / offerType change; must be `type=music` and `Orchard` label |
| `Vendor` | Status transitions to or from `signed` |
| `SubAccount` | `isDeleted` changes |

**`BaseV5RelationshipEvent`** (`neo4j_v5/base.py`) — additionally parses `start`/`end` node metadata. Extended by relationship events:

| Class | Fires when |
|---|---|
| `HasAsset` | `HAS_ASSET` relationship between Orchard Track → OrchardAsset is **deleted** |
| `HasFingerprintRule` | Any `HAS_FINGERPRINT_RULE` change where start node is Track, SubAccount, or Vendor |

**`OutputEvent`** (`src/events/__init__.py`) — the payload sent to SFN. `exc_id()` returns `{data_type}-{data_id}-{action}-{unique_id}` (used as the SFN execution name for idempotency). `to_json()` formats `{id, label, operation, timestamp, payload}`.

### Key Configuration (`config.py`)

| Variable | Default | Purpose |
|---|---|---|
| `SFN_FINGERPRINTER_ARN` | required | ARN of the Fingerprinter SFN |
| `SFN_FINGERPRINTER_MAX_RUNNING` | 500 | Throttle ceiling for concurrent SFN executions |
| `SFN_THROTTLE_SLEEP_SECONDS` | 30 | Sleep duration per throttle loop iteration |
| `ENVIRONMENT` | dev | Controls secrets source (`LambdaSecretsManager` in qa/prod) |

## Testing

Test helpers in `tests/conftest.py` build realistic fixture data:
- `neo4j_v5_cdc_node_event(operation, labels, before, after)` — node CDC event
- `neo4j_v5_cdc_rel_event(operation, rel_type, start_labels, start_keys, end_labels, end_keys)` — relationship CDC event
- `msk_event(topic, value)` — wraps an event in MSK trigger format with base64 encoding

Sample JSON fixtures live in `tests/sample_events/neo4j/`. Use `tests/sample_events/call.sh <file>` to POST one to a running local function.

## Adding a New Event Type

1. Create `src/events/neo4j_v5/<entity>.py` extending `BaseV5Event` or `BaseV5RelationshipEvent`
2. Implement `.output()` returning `OutputEvent | list[OutputEvent] | None`
3. Export from `src/events/neo4j_v5/__init__.py`
4. Register the new Kafka topic → class mapping in `generate_outputs()` (`src/utils/events.py`)