SQL Server Transactional MERGE Upsert with Watermark

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 }})."
      }