# Kestra Playlist Pipeline — Local Dev

A local replica of the FACTS priority-playlist ingestion pipeline, orchestrated by [Kestra](https://kestra.io) instead of Snowflake native tasks and streams.

## What is this?

The production pipeline runs as Snowflake tasks that poll streams and MERGE data every minute. This project replaces those tasks with Kestra flows — an open-source workflow orchestrator — so we get:

| Pain point with Snowflake tasks | What Kestra gives instead |
|---|---|
| No execution history or per-run logs | Full timeline with per-task logs in the UI |
| Manual resume after task suspension | Automatic retry + UI pause/resume |
| No batch guard — 265M rows caused a 60-min MERGE timeout (Jun 26 2026) | `check_batch_size` task stops the run before MERGE if rows > 1M |
| SQL buried inside DDL files | SQL in plain `.sql` files, easy to test locally |
| Separate task graphs for QA vs PROD | One flow; a single KV value switches schema |

---

## How it works — the big picture

```
Your Mac
  │
  ├── Docker Desktop
  │     └── Kestra container  (localhost:8080)
  │           │
  │           ├── PostgreSQL  (stores execution history)
  │           │
  │           └── Kestra flows  (your YAML files)
  │                 │
  │                 └── Snowflake JDBC plugin
  │                       └──► Snowflake (DELPHI account, DEV_ENGINEERING.RROY)
  │
  └── ~/.ssh/snowflake/rsa_key_local.p8  (mounted read-only into container)
```

Kestra is a web app that runs your YAML flow definitions as jobs. Each "task" in a flow is a unit of work — here, most tasks are SQL queries sent to Snowflake via the JDBC driver. Kestra stores its own state (execution history, KV store) in PostgreSQL.

---

## Project structure

```
kestra-playlist-pipeline/
├── docker-compose.yml          # Runs Kestra + PostgreSQL
├── .env                        # Your local credentials (gitignored)
├── scripts/
│   └── load-flows.sh           # Pushes flow YAML files to Kestra via API
└── flows/
    ├── 00_dev_hello.yml        # Safe sandbox demo (DEV_ENGINEERING.RROY only)
    ├── 01_capture_upsert.yml   # Main pipeline: stream → capture → guard → MERGE
    ├── 02_metadata_update.yml  # Metadata: follower counts, artwork, track counts
    └── sql/
        ├── capture_events.sql  # INSERT from Chartmetric stream → stage table
        ├── upsert_placements.sql  # MERGE stage → placements table
        └── update_metadata.sql   # MERGE metadata stream → PRIORITY_PLAYLIST_METADATA
```

---

## Step-by-step setup

### Step 1 — Prerequisites

- Docker Desktop running on your Mac
- Your Snowflake RSA private key at `~/.ssh/snowflake/rsa_key_local.p8`
  - This is an unencrypted PKCS8 key already registered with your Snowflake user (`rroy`)
  - If you need to create one, see "Generating a new key pair" below
- Python 3 with `requests` (only needed for `scripts/load-flows.sh` or manual API calls)

### Step 2 — Configure `.env`

```bash
cp .env.example .env   # if .env doesn't exist yet
```

The `.env` is only used by `docker-compose.yml` when the container starts. The values here control what gets mounted and what environment variables Kestra sees inside the container.

Current working values for local dev:

```
SNOWFLAKE_ACCOUNT=delphi
SNOWFLAKE_USERNAME=RROY@SONYMUSIC-PDE.COM
SNOWFLAKE_PRIVATE_KEY_FILE=/Users/rajeev/.ssh/snowflake/rsa_key_local.p8
SNOWFLAKE_PRIVATE_KEY_PASSPHRASE=
SNOWFLAKE_WAREHOUSE=DEV_PERFORMANCE_WAREHOUSE
SNOWFLAKE_DATABASE=DEV_ENGINEERING
SNOWFLAKE_SCHEMA=RROY
SLACK_WEBHOOK_URL=
```

> **Important:** `SNOWFLAKE_USERNAME` must be your Snowflake **login name** (the email form), not the short username. This is used in the JWT subject when authenticating with key pair auth. Using `rroy` instead of `RROY@SONYMUSIC-PDE.COM` causes a "JWT token is invalid" error.

> **Important:** `SNOWFLAKE_PRIVATE_KEY_FILE` must be an absolute path. Docker does not expand `~`.

### Step 3 — Start Kestra

```bash
docker compose up -d
```

This starts two containers:
- `kestra` — the Kestra server (UI + API at http://localhost:8080)
- `postgres` — stores execution history and the KV store

Wait ~20 seconds for Kestra to fully start, then open http://localhost:8080.

### Step 4 — Load the flows

Kestra reads flow YAML files from the `/flows` directory inside the container (mounted from `./flows/`). However, Kestra needs the flows registered in its database, so push them via the API:

```bash
bash scripts/load-flows.sh
```

You should see:
```
Waiting for Kestra to be ready...
  loaded:  00_dev_hello.yml
  loaded:  01_capture_upsert.yml
  loaded:  02_metadata_update.yml
```

After loading, they appear in the UI under **Flows → insights.playlists**.

If you edit a flow YAML file, re-run `load-flows.sh` to push the update. (The script handles both create and update.)

### Step 5 — Configure the KV store

Kestra has a built-in KV store (key-value store per namespace). The flows read Snowflake credentials from it at runtime using `{{ kv('KEY_NAME') }}`. This is separate from the `.env` file — the `.env` is for Docker startup only.

Run this to populate the KV store:

```bash
BASE="http://localhost:8080/api/v1/namespaces/insights.playlists/kv"

# IMPORTANT: Use Content-Type: application/json with a JSON string value (not text/plain)
put_kv() { curl -s -X PUT "$BASE/$1" -H "Content-Type: application/json" -d "\"$2\""; }

put_kv SNOWFLAKE_HOST        "delphi.us-east-1.snowflakecomputing.com"
put_kv SNOWFLAKE_ACCOUNT     "delphi"
put_kv SNOWFLAKE_USERNAME    "RROY@SONYMUSIC-PDE.COM"
put_kv SNOWFLAKE_WAREHOUSE   "DEV_PERFORMANCE_WAREHOUSE"
put_kv SNOWFLAKE_DATABASE    "DEV_ENGINEERING"
put_kv SNOWFLAKE_SCHEMA      "RROY"
put_kv SNOWFLAKE_PRIVATE_KEY_PASSPHRASE ""
```

Verify:
```bash
curl -s "http://localhost:8080/api/v1/namespaces/insights.playlists/kv" | python3 -m json.tool
```

> **Why KV and not just env vars?** The JDBC URL is constructed at task runtime inside Kestra using Pebble templates (`{{ kv('...') }}`). The KV store lets you update credentials without restarting Docker. The env vars in `.env` are also available as `{{ envs.VAR_NAME }}` but are frozen at container start time.

### Step 6 — Run the demo flow

In the Kestra UI (http://localhost:8080):

1. Go to **Flows** in the left sidebar
2. Click `insights.playlists` namespace
3. Click `dev_hello_snowflake`
4. Click **Execute** (top right)
5. Watch the execution timeline — tasks light up green as they succeed

Or via API:
```bash
curl -s -X POST "http://localhost:8080/api/v1/executions/insights.playlists/dev_hello_snowflake"
```

The demo flow (`00_dev_hello.yml`):
1. **ping** — `SELECT CURRENT_USER(), CURRENT_WAREHOUSE(), ...` — proves connection works
2. **log_connection** — logs the result to the execution output
3. **create_table** — creates a transient table `DEV_ENGINEERING.RROY.KESTRA_DEMO_PLAYLISTS`
4. **insert_sample_rows** — inserts 5 fake playlists (New Music Friday, RapCaviar, etc.)
5. **read_rows** — reads them back
6. **log_results** — logs each row to the Kestra execution log
7. **cleanup** — `DROP TABLE` (so the demo is repeatable)
8. **done** — final log message

Everything runs against `DEV_ENGINEERING.RROY` only. Nothing touches FACTS QA or PROD.

---

## How credentials flow through the system

```
~/.ssh/snowflake/rsa_key_local.p8
    │
    │  (mounted read-only by docker-compose.yml)
    ▼
/secrets/rsa_key.p8  (inside the Kestra container)
    │
    │  (referenced in the JDBC URL)
    ▼
jdbc:snowflake://delphi.us-east-1.snowflakecomputing.com/
    ?account=delphi
    &authenticator=snowflake_jwt
    &private_key_file=/secrets/rsa_key.p8
    &warehouse=DEV_PERFORMANCE_WAREHOUSE

Username: RROY@SONYMUSIC-PDE.COM  (from KV store)
    │
    └──► JWT subject: DELPHI.RROY@SONYMUSIC-PDE.COM
         JWT issuer:  DELPHI.RROY@SONYMUSIC-PDE.COM.SHA256:<fingerprint>
         (Snowflake verifies against the stored RSA_PUBLIC_KEY_FP for user rroy)
```

The Snowflake JDBC plugin in Kestra constructs a JWT using the private key, signs it with RS256, and sends it to Snowflake instead of a password.

---

## The pipeline flows (01 and 02)

These replicate the production Snowflake task graph. They target `FACTS.QA` by default (set `SNOWFLAKE_SCHEMA=QA` in the KV store).

**`01_capture_upsert.yml`** runs every minute:
1. `check_stream` — `SYSTEM$STREAM_HAS_DATA(...)` — skip if no new events
2. `capture_events` — INSERT events from the Chartmetric stream into the stage table
3. `check_batch_size` — count rows captured; stop and alert Slack if > 1M (incident guard)
4. `upsert_placements` — MERGE stage → `PLAYLISTS_PRIORITY_PLACEMENTS_BY_PARTICIPANT_...`
5. `capture_processing_log` — INSERT a log row recording row counts and duration

**`02_metadata_update.yml`** runs every 5 minutes:
1. `check_metadata_stream` — skip if no new events
2. `update_metadata` — MERGE metadata stream → `PRIORITY_PLAYLIST_METADATA`

Both flows post to Slack on failure (`SLACK_WEBHOOK_URL` in KV store).

---

## Switching to QA or PROD

Change `SNOWFLAKE_SCHEMA` and `SNOWFLAKE_DATABASE` in the KV store:

```bash
BASE="http://localhost:8080/api/v1/namespaces/insights.playlists/kv"

# For QA:
curl -s -X PUT "$BASE/SNOWFLAKE_DATABASE" -H "Content-Type: application/json" -d '"FACTS"'
curl -s -X PUT "$BASE/SNOWFLAKE_SCHEMA"   -H "Content-Type: application/json" -d '"QA"'

# For PROD (be careful):
curl -s -X PUT "$BASE/SNOWFLAKE_DATABASE" -H "Content-Type: application/json" -d '"FACTS"'
curl -s -X PUT "$BASE/SNOWFLAKE_SCHEMA"   -H "Content-Type: application/json" -d '"PROD"'
```

The flows pick up the new value on the next execution — no restart needed.

---

## Tear down

```bash
docker compose down        # stops containers, keeps execution history
docker compose down -v     # stops containers AND wipes the postgres volume (clean slate)
```

---

## Generating a new RSA key pair (one-time setup)

If you don't have a key yet:

```bash
mkdir -p ~/.ssh/snowflake
openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out ~/.ssh/snowflake/rsa_key_local.p8 -nocrypt
openssl rsa -in ~/.ssh/snowflake/rsa_key_local.p8 -pubout -out ~/.ssh/snowflake/rsa_key_local.pub
chmod 400 ~/.ssh/snowflake/rsa_key_local.p8
```

Register the public key with your Snowflake user (run in a Snowflake worksheet):
```sql
-- Paste the contents of rsa_key_local.pub, removing the header/footer lines
ALTER USER rroy SET RSA_PUBLIC_KEY='MIIBIjANBgkq...';
```

Verify the fingerprint matches:
```bash
python3 -c "
import base64, hashlib
from cryptography.hazmat.primitives.serialization import load_pem_private_key, Encoding, PublicFormat
with open('/Users/rajeev/.ssh/snowflake/rsa_key_local.p8', 'rb') as f:
    pk = load_pem_private_key(f.read(), password=None)
der = pk.public_key().public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo)
print('SHA256:' + base64.b64encode(hashlib.sha256(der).digest()).decode())
"
```

This should match `RSA_PUBLIC_KEY_FP` shown by `DESC USER rroy` in Snowflake.