Supabase Analytics Pipeline with DuckDB In-Memory Aggregations and Slack Metric Digest

id: supabase-analytics-duckdb-slack-digest
namespace: company.team
description: |
  Extract weekly analytics events from Supabase via PostgREST, compute aggregation metrics
  in-memory using DuckDB SQL, and push a weekly digest report to a Slack channel.

inputs:
  - id: supabase_url
    type: STRING
    displayName: Supabase Project URL
    description: The base REST URL of your Supabase project (e.g.
      https://xyzcompany.supabase.co).
    defaults: "https://xyzcompany.supabase.co"

  - id: table_name
    type: STRING
    displayName: Analytics Table Name
    description: The Supabase table containing user activity telemetry events.
    defaults: "user_events"

  - id: lookback_days
    type: INT
    displayName: Lookback Period (Days)
    description: Number of days of event history to extract from Supabase.
    defaults: 7

triggers:
  - id: weekly_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 9 * * 1"
    disabled: true

tasks:
  - id: extract_events
    type: io.kestra.plugin.supabase.Select
    description: Query raw user event telemetry from Supabase PostgREST API for the
      specified lookback window.
    url: "{{ inputs.supabase_url }}"
    apiKey: "{{ secret('SUPABASE_API_KEY') }}"
    table: "{{ inputs.table_name }}"
    select: "id,user_id,event_type,channel,created_at"
    filter: "created_at=gte.{{ execution.startDate | dateAdd(-inputs.lookback_days,
      'DAYS') }}"

  - id: analyze_metrics
    type: io.kestra.plugin.jdbc.duckdb.Query
    description: Perform fast SQL analytics on extracted Supabase records using
      DuckDB in-memory engine.
    url: "jdbc:duckdb:"
    inputFiles:
      events.json: "{{ outputs.extract_events.uri }}"
    sql: |
      SELECT
        event_type,
        count(DISTINCT user_id) AS active_users,
        count(*) AS total_events
      FROM read_json_auto('{{ workingDir }}/events.json')
      GROUP BY event_type
      ORDER BY total_events DESC;
    fetchType: FETCH

  - id: check_results
    type: io.kestra.plugin.core.flow.If
    description: Branch based on whether events were found in the lookback window.
    condition: "{{ outputs.analyze_metrics.size > 0 }}"
    then:
      - id: send_slack_digest
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Post the computed weekly analytics digest to Slack.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "📊 *Supabase Weekly Analytics Digest (Past {{ inputs.lookback_days }} Days)*\nTotal event categories: *{{ outputs.analyze_metrics.size }}*\nTop Event: *{{ outputs.analyze_metrics.rows[0].event_type }}* ({{ outputs.analyze_metrics.rows[0].total_events }} events from {{ outputs.analyze_metrics.rows[0].active_users }} active users)"
          }
    else:
      - id: log_no_events
        type: io.kestra.plugin.core.log.Log
        description: Log notice when no event records are present in the lookback window.
        message: "No user events were found in Supabase for the past {{
          inputs.lookback_days }} days."

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Send a failure alert to Slack if the analytics pipeline errors out.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": ":rotating_light: Flow {{ flow.namespace }}.{{ flow.id }} (Execution `{{ execution.id }}`) failed - Error: {{ error.message }}"
      }