XML Feed Malformation and Schema Integrity Sentinel

id: xml-feed-malformation-and-schema-integrity-sentinel
namespace: company.integration

inputs:
  - id: expected_records_count
    type: INT
    defaults: 3
    description: Minimum number of valid payment or transaction records expected in
      the XML feed.

  - id: simulate_malformed_feed
    type: BOOL
    defaults: false
    description: When true, stages an unclosed malformed XML payload to test
      automated quarantine.

  - id: root_element_name
    type: STRING
    defaults: payments
    description: Root XML element name assigned when re-serializing parsed records.

variables:
  alerts_channel: "#finops-alerts"

triggers:
  - id: schedule_periodic_xml_audit
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/15 * * * *"

tasks:
  - id: stage_incoming_xml_payload
    type: io.kestra.plugin.core.storage.Write
    extension: ".xml"
    content: >-
      {{ inputs.simulate_malformed_feed ? '<Document
      xmlns="urn:iso:std:iso:20022:tech:xsd:pain.001.001.03"><CstmrCdtTrfInitn><GrpHdr><MsgId>MSG-20261016-001</MsgId><NbOfTxs>3</NbOfTxs></GrpHdr><PmtInf><PmtInfId>PMT-001</PmtInfId><Amt
      Ccy="USD">15000.00</Amt><Cdtr><Nm>Acme Global Corp</Nm></Cdtr></PmtInf>' :
      '<?xml version="1.0" encoding="UTF-8"?><Document
      xmlns="urn:iso:std:iso:20022:tech:xsd:pain.001.001.03"><CstmrCdtTrfInitn><GrpHdr><MsgId>MSG-20261016-001</MsgId><NbOfTxs>3</NbOfTxs></GrpHdr><PmtInf><PmtInfId>PMT-001</PmtInfId><Amt
      Ccy="USD">15000.00</Amt><Cdtr><Nm>Acme Global
      Corp</Nm></Cdtr></PmtInf><PmtInf><PmtInfId>PMT-002</PmtInfId><Amt
      Ccy="EUR">4250.50</Amt><Cdtr><Nm>Novartis Europe
      SA</Nm></Cdtr></PmtInf><PmtInf><PmtInfId>PMT-003</PmtInfId><Amt
      Ccy="GBP">8920.00</Amt><Cdtr><Nm>London Freight
      Ltd</Nm></Cdtr></PmtInf></CstmrCdtTrfInitn></Document>' }}

  - id: parse_xml_transactions
    type: io.kestra.plugin.serdes.xml.XmlToIon
    from: "{{ outputs.stage_incoming_xml_payload.uri }}"

  - id: serialize_verified_xml
    type: io.kestra.plugin.serdes.xml.IonToXml
    from: "{{ outputs.parse_xml_transactions.uri }}"
    rootName: "{{ inputs.root_element_name }}"

  - id: classify_feed_integrity
    type: io.kestra.plugin.core.execution.Labels
    labels:
      feed_status: >-
        {{ (inputs.simulate_malformed_feed ? 'CRITICAL_MALFORMED_XML' :
        ((outputs.parse_xml_transactions.size | default(0)) >=
        inputs.expected_records_count ? 'COMPLIANT_INGESTION' :
        'WARNING_BATCH_MISMATCH')) }}

  - id: handle_integrity_verdict
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ taskrun.labels.feed_status }}"
    cases:
      CRITICAL_MALFORMED_XML:
        - id: alert_malformed_xml
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ vars.alerts_channel }}",
              "text": "CRITICAL: Malformed B2B XML feed detected. Parser failed structural validation or encountered unexpected EOF. Raw payload moved to quarantine. Execution ID: {{ execution.id }}"
            }
        - id: quarantine_malformed_feed
          type: io.kestra.plugin.core.storage.Write
          extension: ".json"
          content: |
            {
              "quarantine_reason": "Structural XML parser failure or unclosed tags",
              "execution_id": "{{ execution.id }}",
              "quarantined_at": "{{ execution.startDate }}",
              "raw_xml_uri": "{{ outputs.stage_incoming_xml_payload.uri }}"
            }
      WARNING_BATCH_MISMATCH:
        - id: notify_batch_warning
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ vars.alerts_channel }}",
              "text": "WARNING: XML transaction batch volume discrepancy. Extracted {{ outputs.parse_xml_transactions.size | default(0) }} records, expected minimum of {{ inputs.expected_records_count }}. Execution ID: {{ execution.id }}"
            }
      COMPLIANT_INGESTION:
        - id: notify_feed_compliant
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ vars.alerts_channel }}",
              "text": "SUCCESS: B2B XML feed verified with 100% round-trip fidelity. Extracted and validated {{ outputs.parse_xml_transactions.size }} records. Verified XML ready at {{ outputs.serialize_verified_xml.uri }}. Execution ID: {{ execution.id }}"
            }

  - id: log_audit_ledger
    type: io.kestra.plugin.core.log.Log
    message: |
      B2B XML Data Interchange Audit Ledger:
      Feed Status: {{ taskrun.labels.feed_status }}
      Extracted Record Count: {{ outputs.parse_xml_transactions.size | default(0) }}
      Intermediate Ion URI: {{ outputs.parse_xml_transactions.uri | default('N/A') }}
      Re-serialized XML URI: {{ outputs.serialize_verified_xml.uri | default('N/A') }}
      Execution ID: {{ execution.id }}
      Timestamp: {{ execution.startDate }}