# CLAUDE.md

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

## What This Is

dbt-analytics — a dbt monorepo of local packages that transform ingested music industry data in Snowflake into analytics tables (streams, views, downloads, metrics, playlists, demographics, geographic rollups). Upstream data comes from `swf-feed-ingestion` which loads `FACT_ANALYTICS` and other source tables.

## Common Commands

```sh
# Dev setup
pyenv install 3.11.2
pip install pipenv
make pipenv_dev           # pipenv install --dev --ignore-pipfile + dbt deps
cp .env.shadow .env       # fill in Snowflake creds, set SNOWFLAKE_SCHEMA to <username>_DBT

# Integration tests (primary development workflow)
make test_integration_package package=streams    # single package (recommended)
scripts/test_integration_models.sh streams       # equivalent direct script
make test_integration                            # all packages (slow, runs sequentially)

# Unit / schema tests
make test           # dbt test --target test (all tests)
make test_unit      # dbt test -s 'test_type:unit' --target test
make test_schema    # dbt test --schema --target test

# Run models (Jenkins only — dev schemas lack source tables/permissions)
make run                  # incremental (seed + views)
make run_full_refresh     # full rebuild
make run_no_seed          # views only

# SQL formatting
pipenv run sqlfmt .

# Clean (fix unexpected test failures after branch switches)
make clean && make install_dbt_deps

# Docs
make generate_docs
make serve_docs

# List all models
make list_models
```

**Development note**: You don't run `dbt run` directly during development. Your dev schema (`DEV_ENGINEERING.<username>_DBT`) won't have source tables or permissions. Instead: write integration tests, and verify queries directly in Snowflake.

## Architecture

### Monorepo of Local Packages

The root `dbt_project.yml` imports 14 local packages + `metaplane/dbt_expectations` via `packages.yml`. Each package has its own `dbt_project.yml`, models, macros, and tests.

| Package | Purpose |
|---|---|
| `base` | View wrappers over raw source tables (materialized as views) |
| `utils` | Shared macros only — no models. Imported by all other packages. |
| `data_availability` | Watermark detection — determines `max_available_date` per store via IQR statistical analysis on `FACT_ANALYTICS` |
| `streams` | Core audio streaming tables at multiple aggregation levels (by track, product, participant, playlist, country, feed) |
| `metrics` | Downstream rollups of streams+downloads+views (all-time, 28-day, regional). Includes `EXPORT_` models for Opensearch. |
| `views` | YouTube video views (analogous to streams but for video) |
| `downloads` | Download counts (from `FACT_ANALYTICS`, filtered by `transactiontypeid`) |
| `playlists` | Playlist metadata, placement events, editorial/algorithmic classification, tracklist history. Includes hourly-built models. |
| `summary_streams` | Summary metrics by ISRC, playlist, source-of-streams |
| `summary_demographics` | Demographics rollups (age/gender) by artist/ISRC/label |
| `summary_geographics_v2` | Geographic rollups by artist/ISRC/label |
| `point_of_sale` | EU physical/digital point-of-sale data |
| `mappings` | Dimension relationship tables (track→video, product→participant, ISRC→release date, etc.) |
| `operations` | Orchestration: atomic table swap (`upgrade_main_pipeline`), RECENT→HISTORY monthly migration |

### Table Naming Convention

Formula: `<CATEGORY>_BY_<PRIMARY_KEYS>_<SUFFIX>_DBT`

- **Category prefix**: `STREAMS_`, `METRICS_`, `VIEWS_`, `DOWNLOADS_`, `SUMMARY_`, `TIKTOK_`, `MAPPINGS_`, `PLACEMENTS_`, `PLAYLISTS_`, `DATA_AVAILABILITY_`, `EXPORT_`, `MARKET_SIZE_`
- **Primary keys after BY**: `TRACK`, `PRODUCT`, `PARTICIPANT`, `ACCOUNT`, `VIDEO`, `CHANNEL`, `ISRC`, `PLAYLIST`, `COUNTRY`, `REGION`, `FEED`. DATE is omitted (implied by `_DAILY`).
- **Materialization suffix**: `_DAILY` (incremental, delete+insert), `_ROLLUP` (full-refresh), `_TOP` (limited content, full-refresh)
- **Split suffix**: `_RECENT` or `_HISTORY` for the RECENT/HISTORY pattern
- **All dbt-managed tables end in `_DBT`**

### DBT / Non-DBT Dual Table Pattern

Every production table exists as two objects:
- `<MODEL>_DBT` — dbt-managed, updated by dbt runs
- `<MODEL>` — production table consumed by apps

The `upgrade_view` macro atomically swaps `_DBT` into production using `ALTER TABLE ... SWAP WITH`, then clones `_DBT` back from the production table so both stay in sync, and resumes reclustering. **By default (`run_always=False`), the upgrade is silently skipped unless `FORCE_MODEL_UPGRADE=true` is set in the Jenkins job.** Models in the main pipeline are upgraded simultaneously via `operations/UPGRADE_MAIN_PIPELINE_DBT` to avoid inconsistencies. Models outside the main pipeline set `run_always=True` in their `upgrade_view` post-hook.

### RECENT / HISTORY Split

Large `_DAILY` tables are split:
- `<MODEL>_RECENT_DBT` — last ~90 days (since `split_date`), auto-clustered, receives daily incremental updates
- `<MODEL>_HISTORY_DBT` — older data, tagged `tag:history`, excluded from daily runs
- `V_<MODEL>` views UNION both for downstream consumers (tagged `tag:v_view`, excluded from tests)
- Monthly migration via `move_data_from_recent_to_history()` macro

### Incremental Strategy

`_DAILY` tables use `delete+insert` incremental strategy with a rolling refresh window:
- `DAYS_TO_RECALCULATE` env var (default 30) controls window size
- `START_DATE`/`END_DATE` env vars override `DAYS_TO_RECALCULATE` to target an arbitrary date range (used for backfills)
- `download_activity_date_filter()` macro generates the WHERE clause incorporating these vars
- `unique_key` must include both `download_activity_date` and `feed_id` (via CONCAT) to prevent data loss when feed filter is active

### Data Availability

The `data_availability` package determines `max_available_date` per store using IQR statistical analysis on `FACT_ANALYTICS`. All date-dependent macros in `utils/macros/dates/` query this. In `test` target, dates return hardcoded values (e.g., `2018-07-25`).

## Key Macros (`utils/macros/`)

### dates/
- `get_max_available_date()` — latest date with complete data across all stores
- `get_max_available_streaming_stores_date()` — streaming-store-specific
- `get_max_available_video_streaming_stores_date()` — YouTube-specific
- `get_max_available_tiktok_date()` — TikTok-specific
- `get_max_available_date_for_tiktok_gainers()` — TikTok gainers-specific
- `get_split_date()` — `DATE_TRUNC(month, DATEADD(day, -90, CURRENT_DATE))` boundary
- `get_current_date()` / `get_current_timestamp()` — helpers for test/prod-consistent date queries
- `download_activity_date_filter(max_available_date, date_column)` — rolling window WHERE clause (supports `START_DATE`/`END_DATE` override)
- `move_data_from_recent_to_history(table_name)` — monthly RECENT→HISTORY migration

### filters/
- `feed_filter(feed_id_notation='feed_id')` — renders `feed_id IN (...)` when `FEED_FILTER` env var set, else `TRUE`. **Must be the first WHERE clause in all `_DAILY` tables.**
- `distributor_filter(distributor_notation='distributor')` — renders `distributor IN (...)` when `DISTRIBUTOR_FILTER` env var set, else `TRUE`. Distributors: `theorchard`, `sme`, `awal`.

### operations/
- `upgrade_view(view_name, view_to_swap_in, run_always=False)` — atomic table swap + grant permissions + resume recluster
- `grant_permissions(view)` — grants SELECT to READ role, ALL to READWRITE role
- `cluster_table(table_name, cluster_scheme)` — sets clustering key + resumes recluster
- `unload_to_s3(table_name, s3_path)` — COPY INTO S3 as JSON for Opensearch exports

### source_of_streams/
- `create_udfs()` — creates SQL UDFs: `return_1_if_active()`, `return_1_if_passive()`, `return_1_if_collection()` for classifying stream source types. Run at `on-run-start`.

### playlists/
- `create_playlist_udfs()` — creates `CLASSIFY_PLAYLIST_OWNER()` (SQL UDF) and `placements_last_added_index()` (JavaScript UDF). Also run at `on-run-start`.

### Root
- `macros/get_custom_schema.sql` — delegates to `generate_schema_name_for_env` so all models land in the same schema

## INNER JOINs with Dimension Tables

Many models use INNER JOIN with dimension tables (e.g., `DIM_PLAYLIST`, `DIM_TRACK`, `DIM_PRODUCT`). This is **intentional design** to filter metrics to our internal catalog.

**Example from STREAMS_BY_TRACK_PLAYLIST_***:
```sql
INNER JOIN {{ ref('CLEAN_DIM_PLAYLIST_DBT') }} dp
    ON fa.playlistid = dp.playlistid
WHERE dp.playlisturl IS NOT NULL
```

**Design Intent**: Only report on entities in our dimension tables (our tracks, our releases, our artists).

**When data is dropped**: This indicates dimension table population issues, which could be:

1. **For internal entities** (tracks, releases, artists we distribute):
   - Missing from catalog system
   - ETL failure
   - Data quality issue
   - → **This is a bug that needs fixing**

2. **For external entities** (Spotify playlists, external artists):
   - Entity doesn't meet data provider thresholds (e.g., Spotify privacy thresholds)
   - Entity filtered by ETL logic
   - → **This may be expected behavior**

**Debugging missing data**:
1. Determine if entity is internal (our catalog) or external (partner-owned)
2. Check raw staging tables (`FACTS.PROD.STAGING_RAW_*`)
3. Verify dimension table has the entity
4. For external entities: Consider privacy thresholds or ETL filters
5. For internal entities: Investigate dimension population failure

## Integration Testing

**Always clear the `data/` folder before every integration test run** (e.g. `rm -rf data/`). Stale fixtures from prior runs will be picked up by `dbt seed` and silently corrupt results.

### Test Cycle (per package, via `scripts/test_integration_models.sh`)

1. Copy shared base fixtures from `base/tests/base_tables/` + package fixtures from `<package>/tests/integration/fixtures/full_refresh/` into `data/full_refresh/`
2. `dbt seed --target test --full-refresh`
3. `dbt run --target test --full-refresh --models base.* <package>.*` (excludes `MOVE_DATA_FROM_RECENT_TO_HISTORY` and `tag:v_view`)
4. `dbt test --target test` — SQL symmetric diff (actual MINUS expected UNION expected MINUS actual = 0 rows)
5. Repeat steps 1-4 with `data/incremental/` fixtures (no `--full-refresh`)

### Fixture Layout

```
base/tests/base_tables/              # shared source table fixtures (FACT_ANALYTICS.csv, DIM_*.csv, etc.)
  <SOURCE_TABLE>.csv                 # used by ALL packages — automatically merged in by test script

<package>/tests/integration/fixtures/
  full_refresh/
    <SOURCE_TABLE>.csv               # package-specific source overrides
    EXPECTED_<MODEL_NAME>.csv        # expected output
  incremental/
    <SOURCE_TABLE>.csv
    EXPECTED_<MODEL_NAME>.csv
```

If your model references an entirely new source table, add a fixture CSV in `base/tests/base_tables/` (for shared tables) or in the package's `full_refresh/` and `incremental/` directories.

### Updating Fixtures

When model changes cause test failures, query the actual result in Snowflake (`SELECT * FROM DEV_ENGINEERING.<schema>.<TABLE>_DBT`), download as CSV, and replace data in `EXPECTED_<MODEL_NAME>.csv`. Verify the output matches specifications before committing — do not blindly accept new output.

## Snowflake Connection

Two auth modes configured in `profiles/`:
- `sf_ssh_auth/` (default for dev) — SSH keypair, set `DBT_PROFILES_DIR=profiles/sf_ssh_auth`
- `sf_pswd_auth/` (Jenkins) — password-based

Targets: `dev` (hardcoded `DEV_ENGINEERING` database), `qa`, `prod`, `test`, `profile`. Dev setup requires creating schema `DEV_ENGINEERING.<username>_DBT`.

## Code Conventions

- All dbt column names must be **snake_case** (convert from UPPER_CASE source columns, e.g., `SELECT TRACKID AS track_id`)
- All models use `+transient: true` (Snowflake transient tables, no Fail-safe)
- SQL models use comments to group columns: `--primary key`, `--filter only`, `--metrics`
- `CREDIT_SAVINGS_MODE=enabled` in dev `.env` produces smaller datasets
- When renaming/deleting models, search all references across the org via [GitHub code search](https://github.com/search?q=org%3Atheorchard) and create a DB PR in `theorchard/database` to drop old Snowflake tables
- New models must be added to `TEMP_qa_dbt_models_refresh.sql` in the `sql-snowflake-utils` repo for QA cloning

## CI/CD

- **PR tests**: `dbt-analytics-pull-request` Jenkins job runs integration tests on every PR push. Kill stale runs manually if pushing multiple commits quickly.
- **Deploy**: Manual — kick off `dbt-analytics-pipeline` with `MODELS` param after merge, then use `dbt-scheduler-analytics-build-views` to build on QA (set `SNOWFLAKE_SCHEMA=QA`), verify, then build on PROD (set `SNOWFLAKE_SCHEMA=PROD`). The "Build views on QA" option has been removed from `dbt-analytics-pipeline`.
- **Daily scheduler**: `dbt-scheduler-analytics-pipeline` — builds streams+metrics, clones PROD→QA, exports to Elasticsearch.
- **Hourly playlists**: `dbt-scheduler-analytics-pipeline-playlists-hourly` — builds `tag:hourly_playlists` models.
- **Changing clustering keys**: requires CTAS approach to avoid credit spikes. Disable scheduler → DB PR → ownership grants → deploy → revert ownership → dbt PR → re-enable scheduler. See README for full procedure.

## Tags

- `tag:history` — `_HISTORY` tables, excluded from daily runs
- `tag:hourly_playlists` — models built every hour
- `tag:v_view` — union view wrappers (`V_` prefix), excluded from test runs
- `tag:tiktok_v2` — TikTok-specific models
- `tag:playlists_to_build_only_on_PROD_DBT_WAREHOUSE` — requires prod warehouse