id: spotify_priority_playlist_capture_upsert
namespace: insights.playlists

description: |
  Captures CDC events from SPOTIFY_PRIORITY_PLAYLIST_EVENTS (Chartmetric data share stream)
  into the stage table, then merges into PLAYLISTS_PRIORITY_PLACEMENTS_BY_PARTICIPANT_ISRC_PLAYLIST_PUBLIC.

  Key improvements over Snowflake native tasks:
  - Batch size guard: alerts Slack and pauses if Chartmetric pushes a bulk refresh
    (the root cause of the Jun 26 2026 incident — 265M rows caused a 60-min MERGE timeout)
  - 30-min task-level timeout on the MERGE (not warehouse-wide)
  - Schema driven by env var: set SNOWFLAKE_SCHEMA=QA for local dev, PROD for production
  - Full execution history, logs, and replay via Kestra UI at http://localhost:8080

labels:
  team: insights
  pipeline: priority-playlists
  jira: IN-17505

# ── Plugin defaults ────────────────────────────────────────────────────────────
# All Snowflake tasks inherit these — no need to repeat the connection config.
pluginDefaults:
  - type: io.kestra.plugin.jdbc.snowflake.Query
    values:
      url: "jdbc:snowflake://{{ kv('SNOWFLAKE_HOST') }}/?account={{ kv('SNOWFLAKE_ACCOUNT') }}&authenticator=snowflake_jwt&private_key_file=/secrets/rsa_key.p8&warehouse={{ kv('SNOWFLAKE_WAREHOUSE') }}&db={{ kv('SNOWFLAKE_DATABASE') }}&schema={{ kv('SNOWFLAKE_SCHEMA') }}"
      username: "{{ kv('SNOWFLAKE_USERNAME') }}"

# ── Variables ──────────────────────────────────────────────────────────────────
variables:
  schema: "{{ kv('SNOWFLAKE_SCHEMA') }}"
  # Alert if a single CAPTURE run ingests more than this many rows.
  # 265M rows hit us on 2026-06-26 — 1M is a safe early-warning threshold.
  batch_alert_threshold: 1000000

# ── Trigger ───────────────────────────────────────────────────────────────────
triggers:
  - id: every_minute
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/1 * * * *"
    # To disable scheduled runs during local exploration, comment out the trigger
    # and use the "Execute" button in the Kestra UI instead.

# ── Tasks ─────────────────────────────────────────────────────────────────────
tasks:

  # 1. Check whether the stream has any pending CDC events.
  #    If empty, skip immediately — no warehouse compute consumed.
  - id: check_stream
    type: io.kestra.plugin.jdbc.snowflake.Query
    fetchType: FETCH_ONE
    sql: |
      SELECT SYSTEM$STREAM_HAS_DATA(
        'FACTS.{{ render(vars.schema) }}.SPOTIFY_PRIORITY_PLAYLIST_EVENTS'
      ) AS has_data

  - id: skip_if_empty
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.check_stream.row.HAS_DATA == false }}"
    then:
      - id: nothing_to_process
        type: io.kestra.plugin.core.log.Log
        message: "Stream is empty — skipping this run."

  # 2. Consume stream: INSERT CDC events into stage table.
  - id: capture_events
    type: io.kestra.plugin.jdbc.snowflake.Query
    fetchType: NONE
    sql: "{{ read('sql/capture_events.sql') | replace('{{schema}}', render(vars.schema)) }}"

  # 3. Batch size guard.
  #    If Chartmetric does another bulk refresh (like the Jun 26 incident),
  #    we alert immediately and stop — preventing the MERGE from timing out.
  - id: check_batch_size
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.capture_events.updatedCount > vars.batch_alert_threshold }}"
    then:
      - id: alert_large_batch
        type: io.kestra.plugin.notifications.slack.SlackIncomingWebhook
        url: "{{ envs.SLACK_WEBHOOK_URL }}"
        payload: |
          {
            "text": "⚠️ *Playlist Pipeline — Large Batch Detected*\n*Rows captured:* {{ outputs.capture_events.updatedCount | numberformat }}\n*Threshold:* {{ vars.batch_alert_threshold | numberformat }}\nThis volume will likely cause a MERGE timeout.\nPipeline paused. Check `FACTS.{{ render(vars.schema) }}.SPOTIFY_PRIORITY_PLAYLIST_EVENT_STAGE` and investigate before resuming.\n*Execution:* <{{ flow.baseUri }}/executions/{{ execution.id }}|View in Kestra>"
          }
      - id: fail_large_batch
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: |
          Captured {{ outputs.capture_events.updatedCount }} rows — exceeds threshold of {{ vars.batch_alert_threshold }}.
          Check for a Chartmetric bulk refresh. Pipeline stopped to prevent MERGE timeout.

  # 4. Merge stage into target placements table.
  #    timeout: PT30M means the task (and its Snowflake query) fails after 30 min,
  #    not after the warehouse-wide 60-min ceiling.
  - id: upsert_placements
    type: io.kestra.plugin.jdbc.snowflake.Query
    fetchType: NONE
    timeout: PT30M
    sql: "{{ read('sql/upsert_placements.sql') | replace('{{schema}}', render(vars.schema)) }}"

  # 5. Write a processing log entry (same logic as CAPTURE_SPOTIFY_PRIORITY_PLAYLIST_PROCESSING_LOG).
  - id: capture_processing_log
    type: io.kestra.plugin.jdbc.snowflake.Query
    fetchType: NONE
    sql: |
      INSERT INTO FACTS.{{ render(vars.schema) }}.SPOTIFY_PRIORITY_PLAYLIST_PROCESSING_LOG
      WITH deduplicated_stage AS (
          SELECT *
          FROM FACTS.{{ render(vars.schema) }}.SPOTIFY_PRIORITY_PLAYLIST_EVENT_STAGE
          QUALIFY ROW_NUMBER() OVER (
              PARTITION BY playlist_id, position
              ORDER BY created_at DESC NULLS LAST
          ) = 1
      )
      SELECT
          te.playlist_id,
          sp.name                    AS playlist_name,
          ARRAY_AGG(te.created_at)::ARRAY(TIMESTAMP_NTZ(9)) AS update_sent,
          sp.num_track_latest,
          COUNT(*)                   AS tracks_updated,
          ARRAY_AGG(
              OBJECT_CONSTRUCT(
                  'isrc', te.isrc,
                  'position', te.position + 1,
                  'track_name', te.track_name,
                  'created_at', te.created_at
              )
          ) WITHIN GROUP (ORDER BY te.position ASC) AS track_objects
      FROM deduplicated_stage te
      LEFT JOIN chartmetric.raw_data.spotify_playlist sp ON te.playlist_id = sp.playlist_id
      GROUP BY te.playlist_id, playlist_name, sp.num_track_latest

  # 6. Summary log — visible in the Kestra execution timeline.
  - id: log_summary
    type: io.kestra.plugin.core.log.Log
    message: |
      ✅ Run complete.
      Captured: {{ outputs.capture_events.updatedCount }} rows
      Upserted: {{ outputs.upsert_placements.updatedCount }} rows
      Schema:   FACTS.{{ render(vars.schema) }}

# ── Error handler ─────────────────────────────────────────────────────────────
# Runs when any task fails (except the intentional fail_large_batch above,
# which triggers its own alert first).
errors:
  - id: notify_failure
    type: io.kestra.plugin.notifications.slack.SlackIncomingWebhook
    url: "{{ envs.SLACK_WEBHOOK_URL }}"
    payload: |
      {
        "text": "❌ *Playlist Pipeline Failed*\n*Failed task:* `{{ task.id }}`\n*Error:* {{ error.message }}\n*Schema:* FACTS.{{ render(vars.schema) }}\n*Execution:* <{{ flow.baseUri }}/executions/{{ execution.id }}|View in Kestra>"
      }