# Architecture Overview

## Architecture Evolution (Phase 5)

The smoke test (2026-02-23) validated the batch pipeline end-to-end but exposed a parallelism bottleneck: the DDB export produces only 5 gzip files, capping concurrency at 5 Lambdas. Rather than patch the batch model, the architecture evolves to:

1. **Backfill** (one-time): Existing batch pipeline seeds `RE_FAN_TOKENS` in Snowflake with 159K fan tokens
2. **Re-collection** (scheduled): Snowflake exports token files to S3 → same SQS queue → Collector Worker. Snowflake `COPY INTO` controls file granularity (e.g. 1,000 fans/file), giving unlimited parallelism
3. **Fan onboarding** (near-real-time, future): DDB Streams → EventBridge Pipe → SQS FIFO → Collector Worker

The existing batch pipeline (Manifest Parser → SQS → Collector Worker) becomes a one-time backfill tool. Re-collection reuses the **same SQS queue and Collector Worker** — the only difference is the S3 file format (plain JSON from Snowflake vs DDB JSON from export). The Collector Worker auto-detects the format.

Note: Flows 0 and 2 are grouped together because they share the same SQS
queue and Collector Worker. Flow 1 (onboarding) is a separate path that
requires Songwhip account access and is implemented later (Phase 5c).

```
FLOW 0: INITIAL BACKFILL (one-time, existing batch pipeline)
═══════════════════════════════════════════════════════════

  S3 (.json.gz)  →  Manifest Parser  →  SQS  →  Collector Worker  →  Kafka
  [DDB export]                              │                           ↓
                                            │                    Snowflake Sink
                                            │                         ↓
                                            │              Landing → Streams → Tasks
                                            │                         ↓
                                            │                  RE_FAN_TOKENS
                                            │                  + Analytics Tables
                                            │
FLOW 2: RE-COLLECTION (scheduled)           │  ← same SQS queue
════════════════════════════════             │
                                            │
  Snowflake          S3                     │
  ┌──────────┐  COPY INTO   ┌──────────┐   │   ┌──────────────┐
  │RE_FAN_   │─────────────>│  JSON    │───>├──>│  Collector   │──> Kafka
  │TOKENS    │  @stage      │  files   │  S3│   │  Worker (λ)  │     ↓
  │(active,  │  (many small │  (1K fans│ notif  └──────────────┘  Snowflake
  │ stale)   │   files)     │  /file)  │        [auto-detects     (update
  └──────────┘              └──────────┘         file format]      tokens +
                                                                   data)

FLOW 1: FAN ONBOARDING (near-real-time, future — requires Songwhip access)
══════════════════════════════════════════════════════════════════════════

  Songwhip DDB                    RE Account
  ┌──────────┐  DDB Stream   ┌──────────────┐  Cross-acct   ┌──────────┐
  │ INSERT   │──────────────>│ EventBridge  │──────────────>│ SQS FIFO │
  │ presave  │               │ Pipe (filter)│  EventBridge  │ (dedup)  │
  └──────────┘               └──────────────┘               └────┬─────┘
                                                                 │
                              ┌──────────────┐              ┌────▼─────┐
                              │  Snowflake   │<── Kafka <───│Collector │
                              │  (tokens +   │              │Worker (λ)│
                              │   data)      │              └──────────┘
                              └──────────────┘
```

## Key Design Decisions

- **Token storage via Kafka pipeline**: `refresh_token` and `new_refresh_token` in the Kafka message; `RE_EXTRACT_TOKENS` task MERGEs into `RE_FAN_TOKENS`. Lambda never reads/writes Snowflake directly.
- **Re-collection reuses existing infrastructure**: Snowflake COPY INTO produces many small files → S3 notification → same SQS queue → same Collector Worker. No new Dispatcher Lambda needed. File format auto-detection in `_process_file()` handles both DDB JSON and plain Snowflake JSON.
- **Oldest-first re-collection priority**: Snowflake export orders by `LAST_COLLECTED_AT ASC` — stalest fans first. If interrupted, the most neglected fans were already prioritized.
- **Fan batch sizing via `MAX_FILE_SIZE`**: Snowflake COPY INTO controls file granularity. ~1,000 fans/file gives 159 files for 159K fans — each file is one Lambda invocation, achieving natural fan-out via SQS + Lambda concurrency.

## Data Flow (Backfill — Dev)

1. Manually export DynamoDB (Songwhip) from prod account and copy `.json.gz` files to dev S3 export bucket
2. S3 event triggers Manifest Parser Lambda on `manifest-summary.json` creation
3. Manifest Parser reads `manifest-files.json`, sends each `.json.gz` file path to SQS
4. Collector Worker consumes SQS messages, streams gzipped fan records from S3
5. For each fan: refresh via Spotify `/api/token` → call Spotify API endpoints
6. Produce results (including `refresh_token` + `new_refresh_token`) to MSK topic
7. Snowflake Sink Connector (Fargate) → landing table → Stream → Tasks
8. `RE_EXTRACT_TOKENS` task MERGEs tokens into `RE_FAN_TOKENS`; analytics tasks flatten endpoint data

## Data Flow (Re-collection — after backfill)

1. Snowflake `RE_RECOLLECTION_EXPORT` task runs COPY INTO @stage → produces JSON files in S3
2. S3 notification sends each file path to the existing SQS queue
3. Collector Worker auto-detects Snowflake JSON format (no DDB deserialization needed)
4. Same fan processing: refresh token → Spotify API → Kafka → Snowflake (updates `RE_FAN_TOKENS` + analytics)

## Production (Future — Dedicated Account)

Once the dedicated AWS account is provisioned and DynamoDB access is established, the full automated pipeline runs end-to-end. Backfill step 1 is automated via EventBridge + Step Functions. Fan onboarding (Flow 1) enables near-real-time processing of new presave signups via DDB Streams.