Drive SQL Server from events, not SQL Server Agent's clock.

id: sql-server-cdc-debezium-notify
namespace: company.team
description: |
  React to SQL Server change data capture events in real time via Debezium, without
  standing up Kafka Connect. Every insert, update, or delete on a CDC-enabled table
  becomes one Kestra execution; high-value changes post a Microsoft Teams alert as they happen.

tasks:
  - id: log_change
    type: io.kestra.plugin.core.log.Log
    description: Record every captured change for audit purposes.
    message: "CDC event on {{ trigger.stream }}: {{ trigger.data }}"

  - id: high_value_change
    type: io.kestra.plugin.core.flow.If
    description: Only alert Teams when the changed order is above the review threshold.
    condition: "{{ trigger.data.total is defined and trigger.data.total > 5000 }}"
    then:
      - id: notify_teams
        type: io.kestra.plugin.microsoft365.teams.TeamsIncomingWebhook
        description: Post the high-value order change to the operations channel.
        url: "{{ secret('TEAMS_WEBHOOK_URL') }}"
        payload: |
          {
            "@type": "MessageCard",
            "@context": "http://schema.org/extensions",
            "summary": "High-value SQL Server order change",
            "sections": [
              {
                "activityTitle": "High-value order change captured via CDC",
                "activitySubtitle": "{{ trigger.stream }}",
                "text": "{{ trigger.data }}"
              }
            ]
          }

triggers:
  - id: capture_order_changes
    type: io.kestra.plugin.debezium.sqlserver.RealtimeTrigger
    description: Stream row-level changes from the CDC-enabled orders table in real
      time, one execution per row.
    hostname: "{{ secret('SQLSERVER_HOSTNAME') }}"
    port: "1433"
    username: "{{ secret('SQLSERVER_USERNAME') }}"
    password: "{{ secret('SQLSERVER_PASSWORD') }}"
    database: sales
    includedTables: sales.orders
    format: INLINE
    snapshotMode: INITIAL

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack if a captured change fails to process.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "SQL Server CDC processing failed (execution {{ execution.id }})."
      }

id: sql-server-scheduled-report-export
namespace: company.team

inputs:
  - id: report_date
    type: STRING
    required: false
    description: The day to report on (yyyy-MM-dd, UTC). Leave empty on a scheduled
      run to report on the day before the trigger fires; set it to backfill a
      specific past day from the UI.

variables:
  report_day: "{{ (inputs.report_date ?? ((trigger.date ?? execution.startDate) |
    dateAdd(-1, 'DAYS'))) | date('yyyy-MM-dd') }}"

pluginDefaults:
  - type: io.kestra.plugin.jdbc.sqlserver
    values:
      url: "{{ secret('SQLSERVER_URL') }}"
      username: "{{ secret('SQLSERVER_USERNAME') }}"
      password: "{{ secret('SQLSERVER_PASSWORD') }}"
      trustServerCertificate: true

tasks:
  - id: daily_sales_report
    type: io.kestra.plugin.jdbc.sqlserver.Query
    description: Aggregate the report day's orders by region, with the day bound as
      a parameter.
    fetchType: STORE
    parameters:
      day: "{{ vars.report_day }}"
    sql: |
      SELECT region, COUNT(*) AS order_count, SUM(total) AS total_sales
      FROM dbo.orders
      WHERE CAST(created_at AS DATE) = CAST(:day AS DATE)
      GROUP BY region
      ORDER BY total_sales DESC;

  - id: email_when_data
    type: io.kestra.plugin.core.flow.If
    description: Only build and send the report when the report day produced at
      least one row.
    condition: "{{ outputs.daily_sales_report.size > 0 }}"
    then:
      - id: report_to_csv
        type: io.kestra.plugin.serdes.csv.IonToCsv
        description: Convert the stored ION report into a CSV file suitable for an email
          attachment.
        from: "{{ outputs.daily_sales_report.uri }}"

      - id: email_report
        type: io.kestra.plugin.email.MailSend
        description: Email the daily sales report CSV to the business stakeholders.
        from: "{{ secret('REPORT_FROM_ADDRESS') }}"
        to: "{{ secret('REPORT_RECIPIENTS') }}"
        host: "{{ secret('SMTP_HOST') }}"
        port: 587
        username: "{{ secret('SMTP_USERNAME') }}"
        password: "{{ secret('SMTP_PASSWORD') }}"
        transportStrategy: SMTP_TLS
        subject: "SQL Server daily sales report ({{ vars.report_day }})"
        htmlTextContent: |
          Attached is the SQL Server sales report covering {{ vars.report_day }}.
          Rows in this report: {{ outputs.daily_sales_report.size }}
        attachments:
          - name: sales_report.csv
            uri: "{{ outputs.report_to_csv.uri }}"
            contentType: text/csv
    else:
      - id: log_no_data
        type: io.kestra.plugin.core.log.Log
        description: Record that there was nothing to report instead of emailing an
          empty attachment.
        message: "No SQL Server sales rows for {{ vars.report_day }}; skipping the
          report email."

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: Run the report every morning at 06:00 UTC for the prior day.
    cron: "0 6 * * *"
    timezone: Etc/UTC

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack if the morning report fails so a missed report is never silent.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "SQL Server daily report failed (execution {{ execution.id }})."
      }

id: sql-server-event-driven-queue-processor
namespace: company.team
description: |
  Start a flow the moment a SQL Server queue-style table has pending work, instead of
  polling on a fixed cron and hoping the interval lines up with real load. A trigger
  polls for rows in a NEW state, marks them PROCESSED in the same transaction so they
  never fire twice, and a ForEach processes each row independently.

tasks:
  - id: process_row
    type: io.kestra.plugin.core.flow.ForEach
    description: Process each pending row returned by the trigger independently.
    values: "{{ trigger.rows }}"
    tasks:
      - id: handle
        type: io.kestra.plugin.core.log.Log
        description: Replace with the real per-row action (an API call, a downstream
          write, a file export).
        message: "Processing order {{ json(taskrun.value).order_id }} for customer {{
          json(taskrun.value).customer_id }}"

  - id: notify
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Confirm the batch completed with how many rows were picked up this run.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "SQL Server queue processor picked up {{ trigger.size }} pending row(s) and processed them."
      }

triggers:
  - id: pending_rows
    type: io.kestra.plugin.jdbc.sqlserver.Trigger
    description: Poll every minute for rows still marked NEW, then flip them to
      PROCESSED in the same transaction so a slow consumer never
      double-processes a row.
    interval: PT1M
    url: "{{ secret('SQLSERVER_URL') }}"
    username: "{{ secret('SQLSERVER_USERNAME') }}"
    password: "{{ secret('SQLSERVER_PASSWORD') }}"
    trustServerCertificate: true
    sql: |
      SELECT order_id, customer_id, status
      FROM dbo.order_queue
      WHERE status = 'NEW';
    afterSQL: |
      UPDATE dbo.order_queue SET status = 'PROCESSED' WHERE status = 'NEW';
    fetchType: FETCH

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack if row processing fails partway through the batch.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "SQL Server queue processor failed partway through a batch (execution {{ execution.id }}). Check for rows stuck in PROCESSED with no downstream effect."
      }

id: sql-server-transactional-merge-upsert
namespace: company.team
description: |
  Stage new rows, MERGE them into a target table with dedup, and advance a watermark,
  all as one transaction. A failed MERGE rolls back everything, so the target never
  holds a half-applied batch and a rerun always resumes cleanly from the last commit.

tasks:
  - id: read_watermark
    type: io.kestra.plugin.core.kv.Get
    description: Read the last committed watermark; falls through to the epoch on first run.
    key: inventory_merge_watermark
    errorOnMissing: false

  - id: stage_new_rows
    type: io.kestra.plugin.jdbc.sqlserver.Query
    description: Pull only the rows changed since the last watermark into a staging
      result set.
    sql: |
      SELECT sku, warehouse_id, quantity, updated_at
      FROM dbo.inventory_updates
      WHERE updated_at > :watermark
      ORDER BY updated_at ASC;
    parameters:
      watermark: "{{ outputs.read_watermark.value ?? '1900-01-01T00:00:00' }}"
    fetchType: STORE

  - id: merge_and_advance_watermark
    type: io.kestra.plugin.jdbc.sqlserver.Queries
    description: >
      MERGE the staged rows into the target with dedup on the merge key, then
      advance the watermark, both in one transaction (transaction defaults to
      true), so a failed MERGE rolls back and the watermark is never advanced
      past what committed.
    sql: |
      MERGE dbo.inventory_current AS target
      USING dbo.inventory_staging AS source
      ON target.sku = source.sku AND target.warehouse_id = source.warehouse_id
      WHEN MATCHED THEN
        UPDATE SET quantity = source.quantity, updated_at = source.updated_at
      WHEN NOT MATCHED THEN
        INSERT (sku, warehouse_id, quantity, updated_at)
        VALUES (source.sku, source.warehouse_id, source.quantity, source.updated_at);
    afterSQL: |
      UPDATE dbo.merge_watermarks
      SET watermark = (SELECT MAX(updated_at) FROM dbo.inventory_staging)
      WHERE job_name = 'inventory_merge';

  - id: row_check
    type: io.kestra.plugin.core.flow.If
    description: Sanity-check that the merge actually touched rows when the stage
      extracted some.
    condition: "{{ outputs.stage_new_rows.size > 0 and
      (outputs.merge_and_advance_watermark.outputs | length) == 0 }}"
    then:
      - id: alert_no_op_merge
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Flag a suspicious run where rows were staged but the MERGE reported
          no output.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "SQL Server merge upsert staged {{ outputs.stage_new_rows.size }} rows but the MERGE step reported no output. Check dbo.inventory_staging for a schema mismatch."
          }

  - id: advance_watermark
    type: io.kestra.plugin.core.kv.Set
    description: Persist the new watermark in the KV Store only after the
      transactional MERGE has committed.
    key: inventory_merge_watermark
    value: "{{ outputs.stage_new_rows.rows | length > 0 and
      (outputs.stage_new_rows.rows | last).updated_at or
      (outputs.read_watermark.value ?? '1900-01-01T00:00:00') }}"

  - id: notify
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Confirm the merge completed with the row count and new watermark.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "SQL Server transactional merge staged {{ outputs.stage_new_rows.size }} rows. Watermark advanced to {{ outputs.advance_watermark.value }}."
      }

pluginDefaults:
  - type: io.kestra.plugin.jdbc.sqlserver.Query
    values:
      url: "{{ secret('SQLSERVER_URL') }}"
      username: "{{ secret('SQLSERVER_USERNAME') }}"
      password: "{{ secret('SQLSERVER_PASSWORD') }}"
      trustServerCertificate: true
  - type: io.kestra.plugin.jdbc.sqlserver.Queries
    values:
      url: "{{ secret('SQLSERVER_URL') }}"
      username: "{{ secret('SQLSERVER_USERNAME') }}"
      password: "{{ secret('SQLSERVER_PASSWORD') }}"
      trustServerCertificate: true

triggers:
  - id: every_15_minutes
    type: io.kestra.plugin.core.trigger.Schedule
    description: Run the incremental MERGE every 15 minutes. Adjust to match how
      fresh the target table needs to be.
    cron: "*/15 * * * *"
    disabled: true

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack if the stage or MERGE step fails; the watermark is not
      advanced so the next run retries the same window.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "SQL Server transactional merge failed and did not advance the watermark (execution {{ execution.id }})."
      }