# Phased Build Plan

## Phase 1: Infrastructure & Skeleton (Terraform)
- S3 bucket reference (DynamoDB export — existing `dev-mymac80` bucket)
- SQS Queue with Dead Letter Queue (DLQ)
- MSK topic creation (`resonance-engine.spotify-data` + `dlq.resonance-engine`)
- IAM Roles (least privilege)

## Phase 2: The Orchestrator
- **Dev**: Manual DynamoDB export from Songwhip prod account → copy to dev S3 bucket
- **Future (dedicated account)**: EventBridge rule → Step Function → `dynamodb:ExportTableToPointInTime`
- Manifest Parser Lambda: triggered by S3 `ObjectCreated` on `manifest-summary.json`
  - Reads `manifest-files.json`, sends each data file path to SQS

## Phase 3: The Collector Worker
- Consumes SQS messages (S3 file paths of gzipped fan records)
- Streams `.json.gz` from S3 (smart_open or equivalent)
- Token refresh: check `expires_at` → call Spotify `/api/token` → update Songwhip DDB
- Data extraction: call Spotify API endpoint(s)
- Sink: produce JSON to MSK topic (`resonance-engine.spotify-data`) using `confluent-kafka` with async writes + batching
- Throughput target: ~845 records/sec (73M fans / 24h window)

## Phase 4: Data Sink (Snowflake)
- Snowflake Sink Connector on Fargate (Kafka Connect with Snowpipe Streaming ingestion)
  - Follows org pattern from `terraform-infra/prod/kafka-infra/snowflake_sink*/`
  - Service naming: `kc-sfsink-resonance-engine`
  - Snowflake service user + key pair via Secrets Manager
  - Datadog monitoring via `terraform-datadog` kafka_connector module
- Raw landing table (VARIANT column for JSON)
- Snowflake Stream on landing table to capture new inserts
- Snowflake Task to transform raw JSON into final analytics table with flattened columns

## Phase 5: Architecture Evolution — Token Table & Re-Collection

The batch pipeline (Phases 1-4) becomes a one-time backfill tool. Phase 5 adds token persistence in Snowflake and a re-collection loop that reuses the existing SQS + Collector Worker infrastructure.

### Phase 5a: Token Table + Backfill Stream (DONE)
- `RE_FAN_TOKENS` table, `RE_SPOTIFY_DATA_STREAM_TOKENS` stream, `RE_EXTRACT_TOKENS` child task
- Collector Worker Kafka message includes `refresh_token` + `new_refresh_token`
- MERGE with recency guard (`COLLECTED_AT > LAST_COLLECTED_AT`) prevents stale overwrites
- Run 159K backfill to seed all tokens in Snowflake

### Phase 5b: Re-Collection Infrastructure
- Snowflake external stage (`RE_RECOLLECTION_STAGE`) pointing to S3 recollection prefix
- Snowflake scheduled task (`RE_RECOLLECTION_EXPORT`): COPY INTO @stage with `MAX_FILE_SIZE` to produce many small JSON files (~1,000 fans/file) ordered by `LAST_COLLECTED_AT ASC` (stalest first)
- S3 event notification → **existing SQS queue** (same queue used by Manifest Parser)
- Collector Worker format auto-detection: `_process_file()` checks first record shape — DDB JSON (has `Item` wrapper + type markers) vs plain Snowflake JSON (flat keys like `SPOTIFY_USER_ID`, `REFRESH_TOKEN`)
- No new Dispatcher Lambda, no new SQS queue — reuses the existing pipeline infrastructure
- **Files new:** `snowflake/recollection_stage.sql`, `snowflake/recollection_task.sql`
- **Files modified:** `collector-worker/handler.py` (format detection in `_process_file()`), `terraform/` (S3 notification for recollection prefix)

### Phase 5c: Fan Onboarding via DDB Streams (requires Songwhip account access)
- Enable DDB Streams on Songwhip table (if not already)
- EventBridge Pipe in Songwhip account: filter for `spotify-presave` INSERTs
- Cross-account EventBridge bus + rule + SQS FIFO queue (dedup by `spotify_user_id`) in RE account
- Collector Worker accepts onboarding messages: `{"source": "onboarding", "fan": {...}}`
- **Files new:** `terraform/eventbridge_onboarding.tf`, `terraform/sqs_onboarding.tf`
- **Files modified:** `collector-worker/handler.py` (onboarding mode), `terraform/collector_worker.tf` (new event source)

### Phase 5d: Retire Batch Pipeline
- Remove Manifest Parser Lambda + Terraform + tests (entire `lambdas/manifest-parser/` directory)
- Remove S3 file processing code from Collector Worker (`_deserialize_ddb_item()`, `_process_file()` DDB path)
- Remove/repurpose manifest-files SQS queue (may be reused for recollection S3 notifications)
- Update CLAUDE.md, README.md

## Week 1 Breadboard (Minimal Viable Pipe)
1. Manual DDB export of ~100 records from Songwhip prod account → copy to dev S3 export bucket
2. Manifest Parser Lambda parses it in the dev account
3. Single Worker refreshes one token and produces one JSON record to MSK topic