# Enqueue Hive Backfill

Feeds bulk/historical AI-detection scans ([CDAM-4070](https://theorchard.atlassian.net/browse/CDAM-4070))
by enqueueing assets onto `{env}-hive-ai-detection-backfill-queue`, which triggers
[prod-lambda-assets-hive-ai-detection](https://github.com/theorchard/lambda-assets/tree/master/lambda/hive_ai_detection)
through an SQS event source mapping. One message per asset:

```json
{"ASSET_FINAL_ID": 20920529, "DURATION_MS": 183000}
```

By default the lambda is first-scan-wins: if an asset already has a scan, ows-assets
returns 409 and the new result is discarded (the Hive scan still runs and is billed
first, so don't re-enqueue scanned sets casually). To re-scan assets that already have
a scan (e.g. under a new model), enqueue with `OVERWRITE=true` — each message then
carries an `OVERWRITE` flag and the lambda posts with overwrite, so ows-assets replaces
the existing `hive_segment`/`hive_task` rows in place. A 409 while overwriting is
unexpected (the endpoint didn't honor the flag) and re-drives via the DLQ rather than
being silently skipped.

## Usage

1. Export the run list to `input/<name>.csv` with header `ID,DURATION` (query below).
2. Run:

```bash
source ~/.local/bin/awsume prod
ENVIRONMENT=prod CSV_FILE=<name>.csv docker compose run --rm --build cli
# to re-scan assets that already have a scan, replacing their rows in place:
ENVIRONMENT=prod CSV_FILE=<name>.csv OVERWRITE=true docker compose run --rm --build cli
```

On a partial `send_message_batch` failure the script raises with the failing offset;
resume with `START_OFFSET=<offset>`.

## Run-list query (Round 1: sale start 4/9/25 – 4/9/26)

Latest FLAC asset final per track and upload type, under Hive's 600s duration cap,
excluding assets already scanned by the current model. Update the model literal when
the production Hive model changes.

```sql
WITH latest_assets AS (
    SELECT
        af.id       AS id,
        af.duration AS duration
    FROM ORCHARD_APP_REPORTING_V2.ART_RELATIONS_PROD_ART_RELATIONS.RELEASES r
    INNER JOIN ORCHARD_APP_REPORTING_V2.ART_RELATIONS_PROD_ART_RELATIONS.TRACK t
        ON r.release_id = t.release_id AND t.track_type = 'music'
    INNER JOIN ORCHARD_APP_REPORTING_V2.ART_RELATIONS_PROD_ART_RELATIONS.PROJECT p
        ON r.project_id = p.project_id AND p.deletions = 'N'
    INNER JOIN ORCHARD_APP_REPORTING_V2.ART_RELATIONS_PROD_ART_RELATIONS.VENDOR v
        ON p.vendor_id = v.vendor_id AND v.status = 'signed'
    INNER JOIN ORCHARD_APP_REPORTING_V2.PROD_OWS_ASSETS_OWS_ASSETS.ASSET_UPLOAD au
        ON t.id = au.track_unique_id AND au.deleted = FALSE AND au.is_correction = FALSE
    INNER JOIN ORCHARD_APP_REPORTING_V2.PROD_OWS_ASSETS_OWS_ASSETS.ASSET_FINAL af
        ON au.id = af.asset_upload_id AND af.asset_type = 'FLAC'
    WHERE r.not_for_distribution = 'N'
    AND r.distribution_format_id = 1
    AND r.deletions = 'N'
    AND r.sale_start_date > '2025-04-09'::DATE
    AND r.sale_start_date <= '2026-04-09'::DATE
    AND r._FIVETRAN_DELETED != TRUE AND t._FIVETRAN_DELETED != TRUE
    AND p._FIVETRAN_DELETED != TRUE AND v._FIVETRAN_DELETED != TRUE
    AND au._FIVETRAN_DELETED != TRUE AND af._FIVETRAN_DELETED != TRUE
    QUALIFY ROW_NUMBER() OVER (
        PARTITION BY t.id, au.asset_upload_type_id
        ORDER BY au.id DESC, af.id DESC, r.sale_start_date DESC
    ) = 1
),
current_model_scanned AS (
    SELECT DISTINCT asset_final_id
    FROM ORCHARD_APP_REPORTING_V2.PROD_OWS_ASSETS_OWS_ASSETS.HIVE_TASK
    WHERE model = 'ai_music_classifier_HYMN_2026_04_24_v00'
    AND _FIVETRAN_DELETED != TRUE
),
previously_scanned AS (
    SELECT DISTINCT asset_final_id
    FROM ORCHARD_APP_REPORTING_V2.PROD_OWS_ASSETS_OWS_ASSETS.HIVE_SEGMENT
    WHERE _FIVETRAN_DELETED != TRUE
)
SELECT
    la.id       AS ID,
    la.duration AS DURATION
FROM latest_assets la
LEFT JOIN current_model_scanned cms
    ON cms.asset_final_id = la.id
LEFT JOIN previously_scanned ps
    ON ps.asset_final_id = la.id
WHERE cms.asset_final_id IS NULL
AND la.duration < 600 * 1000
-- previously scanned assets first: their old results are deleted before the
-- rescan, so draining them first closes the consumer-facing data gap soonest
ORDER BY (ps.asset_final_id IS NOT NULL) DESC, la.id;
```