id: dev_hello_snowflake
namespace: insights.playlists

description: |
  Safe sandbox demo — runs only against DEV_ENGINEERING.RROY.
  Creates a small test table, inserts rows, reads them back.
  Nothing touches FACTS QA or PROD.

labels:
  team: insights
  env: dev

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=DEV_ENGINEERING&schema=RROY"
      username: "{{ kv('SNOWFLAKE_USERNAME') }}"

tasks:

  # 1. Prove the connection works
  - id: ping
    type: io.kestra.plugin.jdbc.snowflake.Query
    fetchType: FETCH_ONE
    sql: |
      SELECT
        CURRENT_USER()      AS connected_as,
        CURRENT_WAREHOUSE() AS warehouse,
        CURRENT_DATABASE()  AS database,
        CURRENT_SCHEMA()    AS schema,
        CURRENT_TIMESTAMP() AS run_at

  - id: log_connection
    type: io.kestra.plugin.core.log.Log
    message: |
      Connected as: {{ outputs.ping.row.CONNECTED_AS }}
      Warehouse:    {{ outputs.ping.row.WAREHOUSE }}
      Database:     {{ outputs.ping.row.DATABASE }}
      Schema:       {{ outputs.ping.row.SCHEMA }}
      Time:         {{ outputs.ping.row.RUN_AT }}

  # 2. Create a small test table (safe — transient, no Fail Safe cost)
  - id: create_table
    type: io.kestra.plugin.jdbc.snowflake.Query
    fetchType: NONE
    sql: |
      CREATE TRANSIENT TABLE IF NOT EXISTS DEV_ENGINEERING.RROY.KESTRA_DEMO_PLAYLISTS (
        playlist_id     VARCHAR,
        playlist_name   VARCHAR,
        follower_count  NUMBER,
        loaded_by       VARCHAR DEFAULT CURRENT_USER(),
        loaded_at       TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP()
      )

  # 3. Insert a few sample rows (mimicking what the real pipeline processes)
  - id: insert_sample_rows
    type: io.kestra.plugin.jdbc.snowflake.Query
    fetchType: NONE
    sql: |
      INSERT INTO DEV_ENGINEERING.RROY.KESTRA_DEMO_PLAYLISTS (playlist_id, playlist_name, follower_count)
      VALUES
        ('37i9dQZF1DXcBWIGoYBM5M', 'New Music Friday',          7200000),
        ('37i9dQZF1DX0XUsuxWHRQd', 'RapCaviar',                14100000),
        ('37i9dQZF1DX4dyzvuaRJ0n', 'mint',                      6800000),
        ('37i9dQZF1DX4JAvHpjipBk', 'New Music Friday UK',       2100000),
        ('37i9dQZF1DX0Yxoavh5qJV', 'Hot Hits Australia',        1400000)

  # 4. Read them back — this is what outputs look like in Kestra
  - id: read_rows
    type: io.kestra.plugin.jdbc.snowflake.Query
    fetchType: FETCH
    sql: |
      SELECT
        playlist_id,
        playlist_name,
        follower_count,
        loaded_by,
        loaded_at
      FROM DEV_ENGINEERING.RROY.KESTRA_DEMO_PLAYLISTS
      ORDER BY follower_count DESC

  - id: log_results
    type: io.kestra.plugin.core.log.Log
    message: |
      Rows in KESTRA_DEMO_PLAYLISTS:
      {% for row in outputs.read_rows.rows %}
        {{ row.PLAYLIST_NAME }} — {{ row.FOLLOWER_COUNT }} followers (id: {{ row.PLAYLIST_ID }})
      {% endfor %}

  # 5. Cleanup — drop the table so the demo is repeatable
  - id: cleanup
    type: io.kestra.plugin.jdbc.snowflake.Query
    fetchType: NONE
    sql: |
      DROP TABLE IF EXISTS DEV_ENGINEERING.RROY.KESTRA_DEMO_PLAYLISTS

  - id: done
    type: io.kestra.plugin.core.log.Log
    message: "Demo complete. Table dropped. DEV_ENGINEERING.RROY is clean."