# CLAUDE.md

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

## Commands

```bash
# Type check
uv run ty check

# Lint / format check
uv run ruff check
uv run ruff format --check

# Auto-fix lint + format
uv run ruff check --fix && uv run ruff format

# Run unit tests
uv run pytest tests/unit/

# Run a single test
uv run pytest tests/unit/path/to/test_file.py::TestClass::test_method

# Tests with coverage
uv run pytest --cov app tests/unit/ --cov-report xml

# Dev API server
uv run uvicorn app.api.main:app --reload --port=8000
```

## Architecture

**Resonance Engine** is an AWS Lambda pipeline that collects fan data from Digital Service Providers (DSPs) and publishes it to Kafka → Snowflake. Spotify is the primary DSP implementation; the architecture is designed to support additional DSPs via pluggable backends.

### End-to-end flow

```
EventBridge / POST /pipeline/run
        │
        ▼
fan_fanout Lambda          — reads fan_credentials from Aurora DSQL,
                             creates pipeline_run record, batches fans,
                             enqueues FanBatch messages to SQS (per DSP client)
        │
       SQS (per DSP)       — buffers batches; max 3 retries; 14-day DLQ
        │
        ▼
fan_collect Lambda         — consumes FanBatch, calls Spotify API per fan,
                             writes fan_collection_state to DSQL,
                             publishes results to Kafka
        │
       Kafka               — fans out to Snowflake (prod) / Postgres / Dummy (local)
```

### Key modules

| Module | Role |
|---|---|
| `app/pipeline/` | Core orchestration — `fan_fanout.py`, `fan_collect.py`, `services.py`, run state machine |
| `app/runtime/` | Lambda entry points (`main.py`) and handler logic with typed returns (`handlers.py`) |
| `app/dsp/` | Pluggable DSP backends (`spotify.py`, `faker.py`); rate-limit guard; per-client stats |
| `app/fandata/` | Fan data models, collection enums, swappable sinks (`kafka`, `postgres`, `dummy`) |
| `app/adapters/` | Thin clients for Postgres, Kafka, Redis, SQS, AWS Lambda, Spotify API, KMS |
| `app/api/` | FastAPI routers — `fandata`, `pipeline`, `dsp`, `config` |
| `app/core/` | Encryption (Fernet preferred; KMS fallback) |
| `app/cli/` | CLI commands for fandata and DB management |

### Critical design patterns

- **OCC on Aurora DSQL** — `SerializationFailure` exceptions are retried up to 5× with exponential backoff (see `app/adapters/db.py`).
- **DSP gateway** — all DSP access goes through the backend abstraction in `app/dsp/backends/base.py`; swap implementations via config.
- **Data sink factory** — `app/fandata/sinks/` returns Kafka, Postgres, or Dummy based on `settings.data_sink`; singletons via lazy proxy.
- **Typed handler returns** — `app/runtime/handlers.py` uses `TypedDict` + `Literal` discriminants for all Lambda return shapes.
- **Celery alternative** — the same handlers run as Celery tasks (Redis broker) for local dev; `task_acks_late=True`, `worker_prefetch_multiplier=1`.
- **Fernet key versioning** — every ciphertext is prefixed with an 8-char key ID (`<key_id>:<token>`). `FernetEncrypter` selects the decryption key from the prefix; no key-scanning needed.

### Fernet key rotation

Rotation is zero-downtime and relies on natural re-encryption by the pipeline over ~30 days:

```
1. invoke prepare_rotation   → generates new key, writes [new_key, old_key] to Secrets Manager
2. deploy Lambda             → Lambda env picks up FERNET_KEYS=[new_key, old_key]
                               from this point all new encryptions use new_key
3. wait ~30 days             → active fans re-encrypted naturally during fan_collect runs
4. invoke rotate_keys loop   → processes remaining inactive fans (filter: prefix != new_key)
                               when remaining=0: automatically removes old_key from SM
                               returns { done: true, finalized: true }
5. deploy Lambda             → Lambda env picks up FERNET_KEYS=[new_key]
```

`rotate_keys` enforces order: raises `ValueError` if SM hasn't been updated by `prepare_rotation` (step 1 skipped) or if Lambda env is out of sync with SM (step 2 skipped).

### Local dev

Docker Compose provides Postgres 17 (port 7432, DB `resonance_test`) and Redis 7. Test environment variables are injected via `pytest-env` from `pyproject.toml`.

## Testing

### Postgres in Docker

Tests use a dedicated `postgres:16-alpine` container defined in `tests/unit/docker-compose.yaml` (host port **7432**, DB `resonance_test`, user/password `postgres`). `pytest-docker` starts it automatically (`--docker-mode=auto`, the default) or you can start it yourself and pass `--docker-mode=external`.

The DB is created once per session and torn down afterwards. Each test that touches the DB is wrapped in a transaction that **rolls back automatically** — no manual cleanup needed.

### Marking tests that use the DB

```python
@pytest.mark.db
def test_something() -> None:
    ...
```

`@pytest.mark.db` ensures the DB schema is created, opens a session, and wraps the test in a transaction that rolls back on exit. Always add this marker to tests that read or write the DB.

### Creating model instances

Helpers live in `tests/unit/helpers.py`:

```python
from tests.unit.helpers import build_model, create_model

# Build in-memory only (no INSERT)
client = build_model(DSPClient, dsp_id=DSPId.spotify)

# INSERT into DB (rolled back after test)
client = create_model(DSPClient)
run    = create_model(PipelineRun, dsp_client_id=client.id, status=RunStatus.queued)
```

Factories are registered in `tests/unit/factories.py` and extend `SQLAlchemyFactory`:

```python
class DSPClientFactory(SQLAlchemyFactory[DSPClient]):
    dsp_id = DSPId.spotify

class PipelineRunFactory(SQLAlchemyFactory[PipelineRun]):
    filters = None
```

Add a new factory there whenever you add a new model that tests need to build.

### Overriding settings

Use the `override_settings` context manager from `tests/unit/helpers.py`:

```python
from tests.unit.helpers import override_settings

def test_stale_timeout() -> None:
    with override_settings(pipeline_run_stale_timeout_s=100):
        # settings.pipeline_run_stale_timeout_s == 100 inside this block
        ...
```

It patches `app.config.settings.__wrapped__` for the duration of the block and restores the original on exit. Pass any `settings` field as a keyword argument.

## Database patterns

### Transaction contexts

```python
from app.adapters.db import db

# Writes — commits on success, rolls back on exception
with db.transaction():
    run = services.create_queued_run(...)

# Decorator form
@db.transaction
def activate(run_id: uuid.UUID) -> None: ...

# Reads — AUTOCOMMIT isolation, always sees latest committed data
# Use for polling loops and reads that must bypass snapshot isolation
with db.autocommit():
    run = PipelineRun.query.where(PipelineRun.id == run_id).one_or_none()

# Decorator form
@db.autocommit
def _get_nominal_rps(client_name: str) -> int | None: ...
```

**Where to open a transaction:** at the representation layer (API route, CLI command, Lambda handler) **or** at the application layer (service function, use case) — but never both in the same call chain. `db.transaction()` raises `RuntimeError` on nesting, so pick one owner per flow and keep it consistent.

**Which context to use:** `db.transaction()` for any write; `db.autocommit()` when a read must see changes committed by other processes (polling loops, cursor pagination).

### ORM models

All models inherit from `app.adapters.db.Model` (a SQLAlchemy `MappedAsDataclass` + `DeclarativeBase`). Instances are plain dataclasses — fields are typed attributes, no magic accessors.

```python
from app.adapters.db import Model

class PipelineRun(Model, kw_only=True):
    __tablename__ = "pipeline_run"
    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, ...)
    status: Mapped[RunStatus]
    ...
```

Column type defaults from `Model.type_annotation_map`: `enum.Enum` → `VARCHAR(32)`, `datetime` → `TIMESTAMPTZ`.

### Querying: `Model.query`

`Model.query` is a class-level descriptor that returns a fresh `Query[T]` bound to the current session. Chain builder methods, terminate with an executor:

```python
# Builder methods (return Self, chainable)
PipelineRun.query.where(...).order_by(...).limit(10).offset(0)

# Executors
.all()          # Sequence[T]
.one()          # T — raises if 0 or >1
.one_or_none()  # T | None
.first()        # T | None
.count()        # int
.exists()       # bool
.get(pk)        # T | None
```

### Custom query classes with `.as_descriptor()`

When a model needs domain-specific query methods, subclass `Query` and assign it back via `.as_descriptor()`:

```python
from fansifter_common.adapters.db import Query

class PipelineRunQuery(Query["PipelineRun"]):
    def active(self) -> Self:
        return self.where(
            PipelineRun.status.in_([RunStatus.queued, RunStatus.running])
        )

class PipelineRun(Model, kw_only=True):
    ...
    query = PipelineRunQuery.as_descriptor()  # replaces default Query descriptor
```

Usage is identical to the default — the descriptor wires the subclass to the session automatically:

```python
stale = PipelineRun.query.active().where(PipelineRun.started_at < cutoff).all()
```

For raw SQL inside a custom query method, use `self.session` directly:

```python
def increment_batch_result(self, run_id: uuid.UUID, *, processed: int) -> None:
    self.session.execute(
        sa.update(PipelineRun).where(PipelineRun.id == run_id).values(...)
    )
```

Existing custom query classes: `PipelineRunQuery` (`app/pipeline/models.py`), `DSPClientQuery` (`app/dsp/models.py`), `FanCredentialsQuery` (`app/fandata/models.py`).