Write-Audit-Publish (WAP) and Atomic Merge Gate for Apache Iceberg with Project Nessie

id: nessie-iceberg-branch-wap-and-atomic-merge-gate
namespace: company.dataops

description: |
  Automated Write-Audit-Publish (WAP) data lakehouse gate for Apache Iceberg tables cataloged in Project Nessie.
  Diffs an isolated ETL/staging branch against main, audits table modifications and commit drift,
  tracks branch staleness in Kestra KV, pauses for operator sign-off, executes an atomic zero-copy merge
  into main via Nessie REST v2 API, deletes the ephemeral staging branch, and verifies catalog consistency.

inputs:
  - id: nessie_url
    type: STRING
    displayName: Nessie Server API URL
    description: Base HTTP(S) URL of the Project Nessie REST v2 API endpoint (e.g.
      http://nessie.internal:19120/api/v2).
    defaults: "http://localhost:19120/api/v2"

  - id: target_branch
    type: STRING
    displayName: Target Production Branch
    description: Production branch in Nessie where verified Iceberg table metadata
      is merged.
    defaults: "main"

  - id: source_branch
    type: STRING
    displayName: Source Staging Branch
    description: Ephemeral staging or ETL branch holding uncommitted batch
      partitions or CDC updates.
    defaults: "etl_batch_staging"

  - id: max_commit_drift
    type: INT
    displayName: Max Allowed Commit Drift
    description: Maximum number of commits target production branch can be ahead of
      the staging branch before requiring a rebase.
    defaults: 10

  - id: max_modified_tables
    type: INT
    displayName: Max Modified Tables Threshold
    description: Safety limit on the number of Iceberg tables modified in a single
      merge batch to prevent accidental widespread overwrite.
    defaults: 25

  - id: delete_source_branch_on_merge
    type: BOOL
    displayName: Delete Staging Branch on Merge
    description: When true, automatically deletes the ephemeral source staging
      branch once the merge into target is verified.
    defaults: true

  - id: cursor_key
    type: STRING
    displayName: KV Cursor Key
    description: Kestra KV key used to persist per-branch WAP audit history and
      drift state across runs.
    defaults: "nessie_wap_gate_tracker"

  - id: dry_run
    type: BOOL
    displayName: Dry Run Mode
    description: When true, audits branch differences and logs proposed merge
      actions without mutating any Nessie catalog references.
    defaults: true

  - id: auto_merge
    type: BOOL
    displayName: Auto Merge on Approval
    description: When true and dry_run is false, pauses for operator approval and
      executes atomic merge and branch pruning.
    defaults: false

triggers:
  - id: periodic_branch_drift_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Audit active Nessie ETL branches for staleness and pending merge
      readiness every 2 hours.
    cron: "0 */2 * * *"
    disabled: true

  - id: on_demand_wap_merge_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated HTTP endpoint invoked by Spark, Flink, or dbt upon
      completing ETL writes to the staging branch.
    key: "{{ secret('NESSIE_GATE_WEBHOOK_KEY') }}"

tasks:
  - id: validate_inputs
    type: io.kestra.plugin.core.execution.Assert
    description: Assert operational branch names, URLs, and safety thresholds
      satisfy sanity constraints.
    conditions:
      - "{{ inputs.nessie_url | length > 0 }}"
      - "{{ inputs.target_branch | length > 0 }}"
      - "{{ inputs.source_branch | length > 0 }}"
      - "{{ inputs.target_branch != inputs.source_branch }}"
      - "{{ inputs.max_commit_drift >= 1 }}"
      - "{{ inputs.max_modified_tables >= 1 }}"
    errorMessage: "Invalid configuration: Nessie URL and branch names are required,
      source and target must differ, and thresholds must be positive."

  - id: get_target_branch_ref
    type: io.kestra.plugin.core.http.Request
    description: Query Nessie REST v2 API to resolve target production branch
      reference and current HEAD commit hash.
    uri: "{{ inputs.nessie_url }}/trees/{{ inputs.target_branch }}"
    method: GET
    headers:
      Accept: application/json
      Authorization: "Bearer {{ secret('NESSIE_AUTH_TOKEN') ?? '' }}"

  - id: get_source_branch_ref
    type: io.kestra.plugin.core.http.Request
    description: Query Nessie REST v2 API to resolve source staging branch reference
      and current HEAD commit hash.
    uri: "{{ inputs.nessie_url }}/trees/{{ inputs.source_branch }}"
    method: GET
    headers:
      Accept: application/json
      Authorization: "Bearer {{ secret('NESSIE_AUTH_TOKEN') ?? '' }}"

  - id: diff_and_audit_branch
    type: io.kestra.plugin.scripts.python.Script
    description: Compute catalog diff between target and source branches, detect
      modified Iceberg tables, evaluate commit drift, and check against KV
      state.
    containerImage: python:3.11-slim
    beforeCommands:
      - pip install requests
    env:
      NESSIE_URL: "{{ inputs.nessie_url }}"
      TARGET_BRANCH: "{{ inputs.target_branch }}"
      SOURCE_BRANCH: "{{ inputs.source_branch }}"
      TARGET_HASH: "{{ fromJson(outputs.get_target_branch_ref.body).reference.hash }}"
      SOURCE_HASH: "{{ fromJson(outputs.get_source_branch_ref.body).reference.hash }}"
      MAX_COMMIT_DRIFT: "{{ inputs.max_commit_drift }}"
      MAX_MODIFIED_TABLES: "{{ inputs.max_modified_tables }}"
      AUTH_TOKEN: "{{ secret('NESSIE_AUTH_TOKEN') ?? '' }}"
      EXECUTION_ID: "{{ execution.id }}"
      PREVIOUS_STATE_JSON: "{{ kv(inputs.cursor_key, errorOnMissing=false) ?? '{}' }}"
    script: |
      import json
      import os
      from datetime import datetime, timezone

      try:
          from kestra import Kestra
      except ImportError:
          Kestra = None

      import requests

      now_iso = datetime.now(timezone.utc).isoformat()
      nessie_url = os.environ.get("NESSIE_URL", "").rstrip("/")
      target_branch = os.environ.get("TARGET_BRANCH", "main")
      source_branch = os.environ.get("SOURCE_BRANCH", "etl_batch_staging")
      target_hash = os.environ.get("TARGET_HASH", "")
      source_hash = os.environ.get("SOURCE_HASH", "")
      max_drift = int(os.environ.get("MAX_COMMIT_DRIFT", "10"))
      max_modified_tables = int(os.environ.get("MAX_MODIFIED_TABLES", "25"))
      auth_token = os.environ.get("AUTH_TOKEN", "").strip()
      execution_id = os.environ.get("EXECUTION_ID", "manual")

      prev_state_raw = os.environ.get("PREVIOUS_STATE_JSON", "{}")
      try:
          prev_state = json.loads(prev_state_raw) if prev_state_raw.strip() else {}
      except Exception:
          prev_state = {}

      headers = {"Accept": "application/json"}
      if auth_token:
          headers["Authorization"] = f"Bearer {auth_token}"

      # 1. Fetch catalog diff between target and source
      diff_url = f"{nessie_url}/trees/{target_branch}@{target_hash}/diff/{source_branch}@{source_hash}"
      resp = requests.get(diff_url, headers=headers, timeout=20)
      if resp.status_code != 200:
          raise ValueError(f"Failed to fetch Nessie diff ({resp.status_code}): {resp.text}")

      diff_data = resp.json()
      differences = diff_data.get("differences", [])

      added_tables = []
      updated_tables = []
      deleted_tables = []

      for d in differences:
          key_elements = d.get("key", {}).get("elements", [])
          table_identifier = ".".join(key_elements)
          from_content = d.get("fromContent")
          to_content = d.get("toContent")

          if from_content is None and to_content is not None:
              added_tables.append(table_identifier)
          elif from_content is not None and to_content is None:
              deleted_tables.append(table_identifier)
          else:
              updated_tables.append(table_identifier)

      total_modified_tables = len(added_tables) + len(updated_tables) + len(deleted_tables)

      # 2. Query commit log to check drift / common ancestry
      target_log_url = f"{nessie_url}/trees/{target_branch}/history?maxRecords=50"
      target_log_resp = requests.get(target_log_url, headers=headers, timeout=15)
      target_commits = [c.get("commitMeta", {}).get("hash") for c in target_log_resp.json().get("logEntries", [])] if target_log_resp.status_code == 200 else []

      source_log_url = f"{nessie_url}/trees/{source_branch}/history?maxRecords=50"
      source_log_resp = requests.get(source_log_url, headers=headers, timeout=15)
      source_commits = [c.get("commitMeta", {}).get("hash") for c in source_log_resp.json().get("logEntries", [])] if source_log_resp.status_code == 200 else []

      # Compute divergence: commits on target branch not in source ancestry
      source_set = set(source_commits)
      drift_count = 0
      for c in target_commits:
          if c not in source_set:
              drift_count += 1
          else:
              break

      has_changes = total_modified_tables > 0 or source_hash != target_hash
      excessive_drift = drift_count > max_drift
      excessive_tables = total_modified_tables > max_modified_tables

      is_conflict_risk = excessive_drift or excessive_tables
      needs_merge = has_changes and not is_conflict_risk

      overall_status = "CRITICAL_DRIFT" if excessive_drift else ("TABLE_LIMIT_EXCEEDED" if excessive_tables else ("MERGE_READY" if needs_merge else "UP_TO_DATE"))

      branch_key = f"{target_branch}:{source_branch}"
      prev_entry = prev_state.get(branch_key, {})
      prev_consecutive = int(prev_entry.get("consecutive_pending_runs", 0))
      consecutive_runs = prev_consecutive + 1 if needs_merge else 0

      new_state_snapshot = dict(prev_state)
      new_state_snapshot[branch_key] = {
          "target_hash": target_hash,
          "source_hash": source_hash,
          "total_modified_tables": total_modified_tables,
          "drift_count": drift_count,
          "consecutive_pending_runs": consecutive_runs,
          "last_evaluated_at": now_iso,
      }

      summary_text = (
          f"Nessie WAP Gate: {overall_status} | Branch '{source_branch}' -> '{target_branch}'. "
          f"Modified tables: {total_modified_tables} (added={len(added_tables)}, updated={len(updated_tables)}, deleted={len(deleted_tables)}). "
          f"Target drift: {drift_count} commit(s). Needs merge: {needs_merge}."
      )
      print(summary_text)

      audit_manifest = {
          "overall_status": overall_status,
          "target_branch": target_branch,
          "source_branch": source_branch,
          "target_hash": target_hash,
          "source_hash": source_hash,
          "total_modified_tables": total_modified_tables,
          "added_tables": added_tables,
          "updated_tables": updated_tables,
          "deleted_tables": deleted_tables,
          "drift_count": drift_count,
          "excessive_drift": excessive_drift,
          "excessive_tables": excessive_tables,
          "needs_merge": needs_merge,
          "evaluated_at": now_iso,
          "execution_id": execution_id,
      }

      if Kestra:
          Kestra.outputs({
              "needs_merge": needs_merge,
              "overall_status": overall_status,
              "total_modified_tables": total_modified_tables,
              "drift_count": drift_count,
              "added_tables": added_tables,
              "updated_tables": updated_tables,
              "deleted_tables": deleted_tables,
              "summary": summary_text,
              "audit_manifest": audit_manifest,
              "new_state_snapshot": new_state_snapshot,
          })

  - id: update_branch_tracker_kv
    type: io.kestra.plugin.core.kv.Set
    description: Persist the latest Nessie WAP branch audit state and drift metrics
      into Kestra KV.
    key: "{{ inputs.cursor_key }}"
    value: "{{ outputs.diff_and_audit_branch.vars.new_state_snapshot | toJson }}"
    kvType: JSON

  - id: check_merge_eligible
    type: io.kestra.plugin.core.flow.If
    description: Evaluate whether the staging branch has unmerged Iceberg table
      changes without blocking drift or table limits.
    condition: "{{ outputs.diff_and_audit_branch.vars.needs_merge == true }}"
    then:
      - id: check_execution_mode
        type: io.kestra.plugin.core.flow.If
        description: Check whether active automated merge is armed or safe dry run
          auditing should proceed.
        condition: "{{ not inputs.dry_run and inputs.auto_merge }}"
        then:
          - id: pause_for_operator_approval
            type: io.kestra.plugin.core.flow.Pause
            description: Pause workflow execution for Data Steward / Platform Engineer
              sign-off before committing table changes into production.
            onResume:
              - id: approved
                type: BOOL
                defaults: false
                description: Set true to confirm atomic merge of the staging branch into the
                  target production branch.

          - id: branch_on_approval
            type: io.kestra.plugin.core.flow.If
            description: Execute atomic merge and branch cleanup only if human approval was
              explicitly confirmed.
            condition: "{{ outputs.pause_for_operator_approval.onResume.approved == true }}"
            then:
              - id: execute_nessie_merge
                type: io.kestra.plugin.core.http.Request
                description: Call Nessie REST v2 API to execute an atomic zero-copy merge of the
                  staging branch into target production.
                uri: "{{ inputs.nessie_url }}/trees/{{ inputs.target_branch }}/history/merge"
                method: POST
                headers:
                  Content-Type: application/json
                  Accept: application/json
                  Authorization: "Bearer {{ secret('NESSIE_AUTH_TOKEN') ?? '' }}"
                body: |
                  {
                    "fromRefName": "{{ inputs.source_branch }}",
                    "fromHash": "{{ fromJson(outputs.get_source_branch_ref.body).reference.hash }}",
                    "message": "Merged by Kestra WAP Gate [Execution: {{ execution.id }}]",
                    "defaultKeyMergeMode": "NORMAL",
                    "dryRun": false
                  }

              - id: verify_merge_commit
                type: io.kestra.plugin.core.http.Request
                description: Query target branch reference to confirm HEAD commit hash advanced
                  to the newly merged state.
                uri: "{{ inputs.nessie_url }}/trees/{{ inputs.target_branch }}"
                method: GET
                headers:
                  Accept: application/json
                  Authorization: "Bearer {{ secret('NESSIE_AUTH_TOKEN') ?? '' }}"

              - id: assert_target_advanced
                type: io.kestra.plugin.core.flow.If
                description: Confirm that the target branch HEAD hash advanced beyond the
                  pre-merge hash.
                condition: "{{ fromJson(outputs.verify_merge_commit.body).reference.hash ==
                  fromJson(outputs.get_target_branch_ref.body).reference.hash
                  }}"
                then:
                  - id: fail_unresolved_merge
                    type: io.kestra.plugin.core.execution.Fail
                    description: Abort with an error when the target branch commit hash did not
                      advance after the merge call.
                    errorMessage: "Nessie merge call completed but target branch {{
                      inputs.target_branch }} HEAD hash remained at {{
                      fromJson(outputs.get_target_branch_ref.body).reference.ha\
                      sh }}. Check Nessie server logs."
                else:
                  - id: log_merge_success
                    type: io.kestra.plugin.core.log.Log
                    description: Log successful atomic merge and new target branch commit hash.
                    message: "Nessie WAP merge succeeded! Target '{{ inputs.target_branch }}'
                      advanced to {{
                      fromJson(outputs.verify_merge_commit.body).reference.hash
                      }} with {{
                      outputs.diff_and_audit_branch.vars.total_modified_tables
                      }} modified Iceberg table(s)."

              - id: check_delete_staging_branch
                type: io.kestra.plugin.core.flow.If
                description: Check if automatic cleanup of the ephemeral staging branch is
                  requested.
                condition: "{{ inputs.delete_source_branch_on_merge == true }}"
                then:
                  - id: delete_staging_branch
                    type: io.kestra.plugin.core.http.Request
                    description: Delete the ephemeral staging branch via Nessie REST v2 API to keep
                      catalog branch namespace tidy.
                    uri: "{{ inputs.nessie_url }}/trees/{{ inputs.source_branch }}?expectedHash={{
                      fromJson(outputs.get_source_branch_ref.body).reference.ha\
                      sh }}"
                    method: DELETE
                    headers:
                      Accept: application/json
                      Authorization: "Bearer {{ secret('NESSIE_AUTH_TOKEN') ?? '' }}"

                  - id: log_staging_branch_pruned
                    type: io.kestra.plugin.core.log.Log
                    description: Confirm successful deletion of ephemeral staging branch.
                    message: "Ephemeral staging branch '{{ inputs.source_branch }}' deleted
                      successfully from Nessie catalog."

              - id: save_recovery_audit_kv
                type: io.kestra.plugin.core.kv.Set
                description: Persist the completed WAP merge record into Kestra KV for audit and
                  compliance.
                key: "nessie_wap_audit_{{ execution.id }}"
                value: "{{ outputs.diff_and_audit_branch.vars.audit_manifest | toJson }}"
                kvType: JSON

              - id: notify_resolution_slack
                type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
                description: Dispatch confirmation notification to Slack after successful
                  Iceberg catalog merge.
                url: "{{ secret('SLACK_WEBHOOK_URL') }}"
                payload: |
                  {
                    "attachments": [
                      {
                        "color": "#2EB67D",
                        "title": ":white_check_mark: Nessie Iceberg WAP Merge Completed",
                        "text": "*Catalog:* `{{ inputs.nessie_url }}`\n*Merged:* `{{ inputs.source_branch }}` -> `{{ inputs.target_branch }}`\n*New Target Hash:* `{{ fromJson(outputs.verify_merge_commit.body).reference.hash }}`\n*Modified Tables:* `{{ outputs.diff_and_audit_branch.vars.total_modified_tables }}`\n*Execution:* `{{ execution.id }}`"
                      }
                    ]
                  }
            else:
              - id: log_approval_declined
                type: io.kestra.plugin.core.log.Log
                description: Record that the operator declined approval for merging the staging
                  branch.
                message: "Operator declined approval for Nessie Iceberg merge. Staging branch
                  '{{ inputs.source_branch }}' left untouched."
        else:
          - id: log_dry_run_plan
            type: io.kestra.plugin.core.log.Log
            description: Log the proposed merge and affected tables without mutating catalog
              references.
            message: |
              Dry run active or auto_merge disabled (dry_run={{ inputs.dry_run }}, auto_merge={{ inputs.auto_merge }}).
              Staging branch '{{ inputs.source_branch }}' ready to merge into '{{ inputs.target_branch }}':
              - Modified Tables: {{ outputs.diff_and_audit_branch.vars.total_modified_tables }}
              - Added: {{ outputs.diff_and_audit_branch.vars.added_tables | toJson }}
              - Updated: {{ outputs.diff_and_audit_branch.vars.updated_tables | toJson }}
              - Deleted: {{ outputs.diff_and_audit_branch.vars.deleted_tables | toJson }}
              - Commit Drift: {{ outputs.diff_and_audit_branch.vars.drift_count }}
    else:
      - id: log_unmerged_or_conflict_state
        type: io.kestra.plugin.core.log.Log
        description: Log diagnostic information when branch is up to date or blocked by
          commit drift.
        message: "Nessie branch '{{ inputs.source_branch }}' status: {{
          outputs.diff_and_audit_branch.vars.overall_status }}. Drift: {{
          outputs.diff_and_audit_branch.vars.drift_count }} commit(s). Modified
          tables: {{ outputs.diff_and_audit_branch.vars.total_modified_tables
          }}."

errors:
  - id: alert_on_workflow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Dispatch emergency alert to Slack if the WAP gate encounters an
      unhandled execution error.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "attachments": [
          {
            "color": "danger",
            "title": ":rotating_light: Nessie WAP Gate Execution Failed",
            "text": "Failed execution `{{ execution.id }}` while evaluating or merging `{{ inputs.source_branch }}` into `{{ inputs.target_branch }}` on `{{ inputs.nessie_url }}`. Check Nessie server logs and permissions."
          }
        ]
      }

outputs:
  - id: target_branch
    type: STRING
    value: "{{ inputs.target_branch }}"

  - id: source_branch
    type: STRING
    value: "{{ inputs.source_branch }}"

  - id: modified_tables_count
    type: INT
    value: "{{ outputs.diff_and_audit_branch.vars.total_modified_tables ?? 0 }}"

  - id: status
    type: STRING
    value: "{{ (outputs.diff_and_audit_branch.vars.needs_merge ?? false) ?
      ((inputs.auto_merge and not inputs.dry_run and
      (outputs.pause_for_operator_approval.onResume.approved ?? false)) ?
      'MERGED' : 'PENDING_APPROVAL') :
      outputs.diff_and_audit_branch.vars.overall_status ?? 'UP_TO_DATE' }}"