id: spotify_priority_playlist_metadata_update
namespace: insights.playlists

description: |
  Consumes the PRIORITY_PLAYLIST_METADATA_UPDATE_STREAM and merges follower counts,
  artwork URLs, track counts, and curator info into PRIORITY_PLAYLIST_METADATA.

  Runs every 5 minutes (less frequent than the placements pipeline — metadata
  changes are low-volume and don't need sub-minute latency).

  Equivalent to Snowflake task: UPDATE_SPOTIFY_PRIORITY_PLAYLIST_METADATA

labels:
  team: insights
  pipeline: priority-playlists

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:
  schema: "{{ kv('SNOWFLAKE_SCHEMA') }}"

triggers:
  - id: every_5_minutes
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/5 * * * *"

tasks:

  - id: check_metadata_stream
    type: io.kestra.plugin.jdbc.snowflake.Query
    fetchType: FETCH_ONE
    sql: |
      SELECT SYSTEM$STREAM_HAS_DATA(
        'FACTS.{{ render(vars.schema) }}.PRIORITY_PLAYLIST_METADATA_UPDATE_STREAM'
      ) AS has_data

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

  - id: update_metadata
    type: io.kestra.plugin.jdbc.snowflake.Query
    fetchType: NONE
    timeout: PT10M
    sql: "{{ read('sql/update_metadata.sql') | replace('{{schema}}', render(vars.schema)) }}"

  - id: log_summary
    type: io.kestra.plugin.core.log.Log
    message: |
      ✅ Metadata update complete.
      Updated: {{ outputs.update_metadata.updatedCount }} rows
      Schema:  FACTS.{{ render(vars.schema) }}

errors:
  - id: notify_failure
    type: io.kestra.plugin.notifications.slack.SlackIncomingWebhook
    url: "{{ envs.SLACK_WEBHOOK_URL }}"
    payload: |
      {
        "text": "❌ *Playlist Metadata Update 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>"
      }