# resonance-engine-core

Backend service for the Resonance Engine fan data collection pipeline. Collects Spotify user data
(profile, top artists, top tracks, recently played, playlists, saved albums, saved tracks,
followed artists) and publishes results to Kafka.

---

## Architecture

The service follows a **fan-out / fan-collect** pattern:

```mermaid
flowchart TD
    EB(["⏰ Scheduler"])
    MANUAL(["🖱️ Manual / API"])
    S3(["🗂️ S3 file (not implemented)"])

    subgraph fanout["⚡ fan_fanout Lambda"]
        FF[/"create PipelineRun · batch & dispatch"/]
    end

    DSQL[("🗄️ Aurora DSQL\n─────────────\nfan_credentials\nfan_collection_state\npipeline_run\ndsp_client")]

    subgraph sqs["📬 SQS"]
        SI1[/"spotify_smf"/]
        SI2[/"spotify_songwhip"/]
        DLQ[/"DLQ · 14-day retention"/]
    end

    subgraph collect["⚡ fan_collect Lambda"]
        CI1[/"spotify_smf · 204 rps · 15 workers"/]
        CI2[/"spotify_songwhip · 11 rps · 5 workers"/]
    end

    EB & MANUAL & S3 --> fanout
    fanout -->|"reads"| DSQL
    fanout -->|"dispatch FanBatch"| sqs
    subgraph kafka["📨 Kafka"]
        KI1[/"fan-profile"/]
        KI2[/"fan-top-artists"/]
        KI3[/"fan-top-tracks"/]
        KI4[/"..."/]
    end

    sqs --> collect
    collect -->|"writes fan_collection_state\nrotates refresh_token"| DSQL
    SNOWFLAKE[("❄️ Snowflake")]

    collect -->|"publish"| kafka
    kafka -->|"sink"| SNOWFLAKE

    classDef trigger fill:#6366F1,color:#fff,stroke:#4338CA
    classDef storage fill:#10B981,color:#fff,stroke:#047857
    classDef queue fill:#8B5CF6,color:#fff,stroke:#6D28D9
    classDef info fill:#27272a,color:#a1a1aa,stroke:#3f3f46,stroke-dasharray:4 3

    classDef snowflake fill:#29B5E8,color:#fff,stroke:#1a8ab5

    class EB,MANUAL,S3 trigger
    class SNOWFLAKE snowflake
    class DSQL storage
    class SI1,SI2 info
    class FF,CI1,CI2,KI1,KI2,KI3,KI4 info
```

### ⚡ fan_fanout Lambda

Triggered by EventBridge Scheduler or manually via API. S3 file upload trigger is not yet implemented. Reads fan credentials and DSP client config from Aurora DSQL, creates a `pipeline_run` record, batches fans per DSP client, and dispatches `FanBatch` messages to SQS.

Handler: `app.runtime.main.fan_fanout`

### 📬 SQS

One queue per DSP client. Decouples fan_fanout from fan_collect and provides buffering and retry. Each queue feeds a dedicated fan_collect Lambda deployment. Failed batches are retried up to 3 times before moving to the DLQ.

| Queue env var | DSP client |
|---|---|
| `SQS_FAN_COLLECT_SPOTIFY_SMF_QUEUE_URL` | `spotify_smf` |
| `SQS_FAN_COLLECT_SPOTIFY_SONGWHIP_QUEUE_URL` | `spotify_songwhip` |

Each queue has a DLQ with 14-day retention. `ReportBatchItemFailures` is enabled on the fan_collect event source mapping — only failed fans within a batch are returned to the queue, not the entire batch.

### ⚡ fan_collect Lambda

One deployment per DSP client. Consumes `FanBatch` messages from SQS, calls the Spotify API for each fan, writes `fan_collection_state` to Aurora DSQL, and publishes collected fan data to Kafka topics. If Spotify returns a new refresh token during the token refresh call, `fan_credentials.refresh_token_encrypted` is rotated in the same transaction.

Handler: `app.runtime.main.fan_collect` (SQS trigger, `ReportBatchItemFailures` enabled)

| DSP client | `nominal_rps` | Concurrency |
|---|---|---|
| `spotify_smf` | 204 req/s | 15 |
| `spotify_songwhip` | 11 req/s | 5 |

### 🗄️ Aurora DSQL

**Why DSQL over standard Aurora / RDS**

At ~1B `fan_credentials`, fan_collect runs N Lambda instances concurrently all writing to the same tables. Standard single-writer Aurora would bottleneck on write throughput and require a connection pooler (PgBouncer / RDS Proxy) to survive Lambda bursts. DSQL is purpose-built for this pattern:

- **Distributed writer** — no single-writer bottleneck; writes scale horizontally
- **Serverless** — no capacity management, scales to zero between runs
- **IAM auth** — short-lived tokens per invocation, no password rotation
- **OCC model** — optimistic concurrency instead of row locks; conflicts resolve via retry rather than blocking. The codebase implements this in `_commit_batch_result()` with 5-attempt exponential backoff on `SerializationFailure`

**Schema**

| Table | Rows (est.) | Key columns |
|---|---|---|
| `fan_credentials` | ~1B | `id` (cursor), `dsp_user_id`, `dsp_id`, `client_id`, `refresh_token_encrypted`, `token_status` |
| `fan_collection_state` | ~1B | `dsp_user_id`, `dsp_id`, `last_collected_at`, `step_state` (JSONB — all 8 step timestamps), `consecutive_failures` |
| `pipeline_run` | low | `id`, `dsp_client_id`, `status`, counters (`fans_processed`, `errors`, `rate_limited`, …) |
| `dsp_client` | ~10 | `name`, `dsp_id`, `client_id`, `nominal_rps` |

**Bottlenecks and mitigations**

| Bottleneck | Root cause | Mitigation |
|---|---|---|
| `fan_credentials` cursor scan at 1B rows | fanout reads all qualifying fans page by page | `fan_credentials_fanout_idx ON (dsp_id, client_id, token_status, id)` — index-only scan, no heap fetch |
| `fan_collection_state` heap fetch on not_collected_since join | fanout LEFT JOINs by PK then reads `last_collected_at` — at 1B rows every join hits the heap | `fan_collection_state_staleness_idx` — covering index with `INCLUDE (last_collected_at)` eliminates heap fetch |
| Concurrent counter increments on `pipeline_run` | 15 Lambda instances all call `increment_batch_result()` on the same run row | OCC retry (5 attempts, exponential backoff) on `SerializationFailure` |
| Write contention on `fan_collection_state` upserts | 15 Lambdas × batch_size fans per invocation = high concurrent upsert rate | `fillfactor=70` leaves room for in-page updates; OCC handles conflicts without blocking |

### 📨 Kafka

Fan data sink. fan_collect publishes one message per fan per data type.

| Topic | Env var |
|---|---|
| `resonance-engine.fan-profile` | `KAFKA_TOPIC_FANS` |
| `resonance-engine.fan-top-artists` | `KAFKA_TOPIC_TOP_ARTISTS` |
| `resonance-engine.fan-top-tracks` | `KAFKA_TOPIC_TOP_TRACKS` |
| `resonance-engine.fan-recently-played` | `KAFKA_TOPIC_RECENTLY_PLAYED` |
| `resonance-engine.fan-playlists` | `KAFKA_TOPIC_PLAYLISTS` |
| `resonance-engine.fan-saved-albums` | `KAFKA_TOPIC_SAVED_ALBUMS` |
| `resonance-engine.fan-saved-tracks` | `KAFKA_TOPIC_SAVED_TRACKS` |
| `resonance-engine.fan-followed-artists` | `KAFKA_TOPIC_FOLLOWED_ARTISTS` |

### ❄️ Snowflake

Final data warehouse. Kafka topics are loaded into Snowflake via a sink connector.

---

## Encryption

All Spotify refresh tokens are stored encrypted in `fan_credentials.refresh_token_encrypted`.
Encryption and decryption happen entirely within the application — tokens are never sent to an
external service in plaintext.

Controlled by `ENCRYPTER_BACKEND`:

| Value | Mechanism | Config |
|-------|-----------|--------|
| `fernet` (**preferred**) | AES-128-CBC + HMAC-SHA256 via [cryptography.fernet](https://cryptography.io/en/latest/fernet/) | `FERNET_KEY` |
| `kms` | AWS KMS `Encrypt` / `Decrypt` API calls per token | `KMS_KEY_ARN` |

Fernet is preferred: encryption is in-process (no network round-trip per token), the key is a
single 32-byte secret managed outside AWS, and rotation is fully under application control.

**Ciphertext format:** `<key_id>:<fernet_token>` — the 8-character hex key ID is generated
independently of the key material and prepended to every ciphertext. `FernetEncrypter` uses the
prefix to select the correct decryption key without scanning all keys.

**Key format** (stored in Secrets Manager and `FERNET_KEYS` env var): `<key_id>:<fernet_key>`,
e.g. `a3f9c21b:AAAA...base64...`.

Generate a new Fernet key:

```python
from app.core.encrypter import FernetEncrypter
print(FernetEncrypter.generate_key())
# → a3f9c21b:AAAA...base64...
```

### Key rotation

Fernet is a symmetric scheme — rotating the key requires re-encrypting every stored token.
Rotation is zero-downtime: the Lambda can decrypt with both the old and new key during the
transition period, so collection continues uninterrupted.

```mermaid
flowchart TD
    S1["1 · invoke prepare_rotation Lambda\ngenerates new key, writes [new_key, old_key]\nto Secrets Manager"]
    S2["2 · redeploy Lambda\npicks up FERNET_KEYS=new_key,old_key from env\nall new encryptions now use new_key"]
    S3["3 · wait ~30 days\nactive fans are re-encrypted naturally\nduring fan_collect runs"]
    S4["4 · invoke rotate_keys Lambda (loop)\nre-encrypts remaining rows where prefix ≠ new_key_id\nwhen done=true + finalized=true: old_key removed from SM"]
    S5["5 · redeploy Lambda\npicks up FERNET_KEYS=new_key only"]

    S1 --> S2
    S2 -->|"collection continues\nuninterrupted"| S3
    S3 --> S4
    S4 -->|"paginate until\ndone=true"| S4
    S4 --> S5

    N1[/"decrypts with either key\nencrypts new tokens with new_key"/]
    N2[/"rotate_keys raises ValueError\nif SM not updated (step 1 skipped)\nor env out of sync (step 2 skipped)"/]

    S2 --- N1
    S4 --- N2

    classDef step fill:#1e293b,color:#e2e8f0,stroke:#334155
    classDef note fill:#27272a,color:#a1a1aa,stroke:#3f3f46,stroke-dasharray:4 3

    class S1,S2,S3,S4,S5 step
    class N1,N2 note
```

---

## Pipeline Run State Machine

```mermaid
stateDiagram-v2
    [*] --> queued : POST /pipeline/runs  or  fan_fanout Lambda

    queued --> running   : fan_fanout activates run
    queued --> cancelled : POST /runs/{id}/cancel
    queued --> cancelled : no matching credentials
    queued --> error     : stall (timeout / manual)

    running --> paused   : POST /runs/{id}/pause
    running --> done     : all batches complete
    running --> done     : timed_out / circuit_breaker
    running --> cancelled : POST /runs/{id}/cancel
    running --> error    : stall (timeout / manual)

    paused --> running   : POST /runs/{id}/resume
    paused --> cancelled : POST /runs/{id}/cancel
    paused --> error     : stall (timeout / manual)

    done      --> [*]
    cancelled --> [*]
    error     --> [*]
```

`finished_reason` qualifies terminal states:

| `status` | `finished_reason` | Trigger |
|----------|-------------------|---------|
| `done` | `done` | All batches processed |
| `done` | `timed_out` | Batch hit time limit |
| `done` | `circuit_breaker` | Too many consecutive rate limits |
| `cancelled` | `cancelled` | Manual cancel via API |
| `cancelled` | `no_credentials` | No fan credentials matched the run filters |
| `error` | `stalled` | Active run exceeded `PIPELINE_RUN_STALE_TIMEOUT_S` or manual stall |

`source` indicates how the run was created:

| `source` | Origin |
|----------|--------|
| `manual` | API (`POST /pipeline/run`) |
| `scheduled` | Triggered by EventBridge Scheduler |
| `triggered` | Programmatic / external trigger |

---

## Adding a new DSP client

1. Add the name to `DSPClientName` in `app/dsp/enums.py`.
2. Insert a row into `dsp_client` (migration) with `name`, `nominal_rps`, etc.
3. Register the backend in `app/dsp/gateway.py` (`get_dsp_gateway`).
4. Add per-client settings to `app/config.py` (batch size, concurrency, SQS queue URL) and the corresponding helper methods (`fan_fanout_batch_size_for`, `fan_collect_concurrency_for`, `sqs_fan_collect_queue_url_for`).
5. Create an SQS queue and wire its URL into the Lambda function's environment.

---

## Environment Variables

### App

| Variable | Default | Description |
|---|---|---|
| `ENVIRONMENT` | `dev` | Deployment environment label |
| `APP_DEBUG` | `false` | Enable debug mode |
| `CORS_ORIGINS` | `["http://localhost:5173"]` | Allowed CORS origins (JSON list) |
| `AWS_REGION_NAME` | `us-east-1` | AWS region |
| `LOGGING_DEBUG` | `false` | Enable rich console logging |

### Database

| Variable | Default | Description |
|---|---|---|
| `DB_HOST` | `localhost` | Postgres host |
| `DB_PORT` | `5432` | Postgres port |
| `DB_NAME` | `resonance` | Database name |
| `DB_USER` | `resonance` | Postgres user |
| `DB_PASSWORD` | — | Postgres password (not used with DSQL) |
| `DB_ECHO` | `false` | Log all SQL statements |
| `DB_POOL_SIZE` | `5` | SQLAlchemy connection pool size |
| `DB_POOL_MAX_OVERFLOW` | `10` | Max extra connections above pool size |
| `DB_POOL_RECYCLE` | `300` | Connection recycle interval (seconds) |

### Aurora DSQL

| Variable | Default | Description |
|---|---|---|
| `DSQL_ENDPOINT` | — | DSQL cluster hostname; omit to use plain Postgres |
| `DSQL_TOKEN_EXPIRES_IN` | `900` | IAM auth token TTL (seconds) |

### Encryption

| Variable | Default | Description |
|---|---|---|
| `ENCRYPTER_BACKEND` | `kms` | `fernet` or `kms` |
| `FERNET_KEYS` | — | Comma-separated list of `<key_id>:<fernet_key>` entries; first entry is the active key. Required when `ENCRYPTER_BACKEND=fernet` |
| `KMS_KEY_ARN` | — | Required when `ENCRYPTER_BACKEND=kms` |


### Fan fanout

| Variable | Default | Description |
|---|---|---|
| `FAN_FANOUT_DISPATCH_BACKEND` | `lambda` | `lambda` or `dummy` |
| `FAN_FANOUT_LAMBDA_FUNCTION_NAME` | — | Lambda function name (required when backend=`lambda`) |
| `FAN_FANOUT_BATCH_SIZE` | `100` | Default fans per SQS/Celery batch |
| `FAN_FANOUT_SPOTIFY_SMF_BATCH_SIZE` | — | Override batch size for spotify_smf |
| `FAN_FANOUT_SPOTIFY_SONGWHIP_BATCH_SIZE` | — | Override batch size for spotify_songwhip |
| `FAN_FANOUT_MAX_FANS` | `20000000` | Max fans queued per run |
| `FAN_FANOUT_TIME_LIMIT_S` | `900` | fanout Lambda execution time limit (seconds) |

### Fan collect

| Variable | Default | Description |
|---|---|---|
| `FAN_COLLECT_DISPATCH_BACKEND` | `sqs` | `sqs` or `dummy` |
| `FAN_COLLECT_TIME_LIMIT_S` | `600` | collect Lambda execution time limit (seconds) |
| `FAN_COLLECT_TIMEOUT_BUFFER_S` | `30` | Seconds before time limit to stop accepting new work |
| `FAN_COLLECT_RATE_LIMIT_CIRCUIT_BREAKER_THRESHOLD` | `3` | Consecutive rate limits before aborting a batch |
| `FAN_COLLECT_CONCURRENCY` | `15` | Default Lambda reserved concurrency |
| `FAN_COLLECT_SPOTIFY_SMF_CONCURRENCY` | `15` | Concurrency for spotify_smf |
| `FAN_COLLECT_SPOTIFY_SONGWHIP_CONCURRENCY` | `5` | Concurrency for spotify_songwhip |
| `FAN_COLLECT_PROFILE_INTERVAL_S` | `0` | Min seconds between profile collections per fan (`0` = every run) |
| `FAN_COLLECT_TOP_ARTISTS_INTERVAL_S` | `0` | Min seconds between top-artists collections |
| `FAN_COLLECT_TOP_TRACKS_INTERVAL_S` | `0` | Min seconds between top-tracks collections |
| `FAN_COLLECT_RECENTLY_PLAYED_INTERVAL_S` | `0` | Min seconds between recently-played collections |
| `FAN_COLLECT_PLAYLISTS_INTERVAL_S` | `0` | Min seconds between playlists collections |
| `FAN_COLLECT_SAVED_ALBUMS_INTERVAL_S` | `0` | Min seconds between saved-albums collections |
| `FAN_COLLECT_SAVED_TRACKS_INTERVAL_S` | `0` | Min seconds between saved-tracks collections |
| `FAN_COLLECT_FOLLOWED_ARTISTS_INTERVAL_S` | `0` | Min seconds between followed-artists collections |

### SQS

| Variable | Default | Description |
|---|---|---|
| `SQS_FAN_COLLECT_SPOTIFY_SMF_QUEUE_URL` | — | SQS queue URL for spotify_smf fan collect |
| `SQS_FAN_COLLECT_SPOTIFY_SONGWHIP_QUEUE_URL` | — | SQS queue URL for spotify_songwhip fan collect |

### Kafka

| Variable | Default | Description |
|---|---|---|
| `KAFKA_BOOTSTRAP_SERVERS` | — | Comma-separated broker list |
| `KAFKA_SECURITY_PROTOCOL` | `SSL` | Kafka security protocol |
| `KAFKA_TOPIC_FANS` | `resonance-engine.fan-profile` | Fan profile topic |
| `KAFKA_TOPIC_TOP_ARTISTS` | `resonance-engine.fan-top-artists` | Top artists topic |
| `KAFKA_TOPIC_TOP_TRACKS` | `resonance-engine.fan-top-tracks` | Top tracks topic |
| `KAFKA_TOPIC_RECENTLY_PLAYED` | `resonance-engine.fan-recently-played` | Recently played topic |
| `KAFKA_TOPIC_PLAYLISTS` | `resonance-engine.fan-playlists` | Playlists topic |
| `KAFKA_TOPIC_SAVED_ALBUMS` | `resonance-engine.fan-saved-albums` | Saved albums topic |
| `KAFKA_TOPIC_SAVED_TRACKS` | `resonance-engine.fan-saved-tracks` | Saved tracks topic |
| `KAFKA_TOPIC_FOLLOWED_ARTISTS` | `resonance-engine.fan-followed-artists` | Followed artists topic |

### Pipeline run

| Variable | Default | Description |
|---|---|---|
| `PIPELINE_RUN_STALE_TIMEOUT_S` | `1800` | Seconds before an active run is considered stalled |
| `PIPELINE_RUN_ESTIMATE_AVG_LATENCY_MS` | `135` | Avg fan collection latency used for ETA estimates |

### DSP

| Variable | Default | Description |
|---|---|---|
| `SPOTIFY_SMF_CLIENT_ID` | — | OAuth client ID for spotify_smf |
| `SPOTIFY_SMF_CLIENT_SECRET` | — | OAuth client secret for spotify_smf |
| `SPOTIFY_SONGWHIP_CLIENT_ID` | — | OAuth client ID for spotify_songwhip |
| `SPOTIFY_SONGWHIP_CLIENT_SECRET` | — | OAuth client secret for spotify_songwhip |
| `DSP_RATE_LIMIT_STATS_BACKEND` | `memory` | `redis` or `memory` — where rate-limit counters are stored |
| `REDIS_URL` | `redis://localhost:6379/0` | Redis connection URL (used when `DSP_RATE_LIMIT_STATS_BACKEND=redis`) |

### Data sink

| Variable | Default | Description |
|---|---|---|
| `DATA_SINK_BACKEND` | `kafka` | `kafka` or `dummy` |