ISO 20022 pacs.008: Parse, Idempotency & Gateway Latency

CREATE TABLE inbound_payments (
    id               UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    msg_id           VARCHAR(35) UNIQUE NOT NULL,   -- ISO 20022 MsgId — idempotency key
    uetr             UUID UNIQUE NOT NULL,           -- Unique End-to-end Transaction Ref
    end_to_end_id    VARCHAR(35) NOT NULL,
    amount           NUMERIC(18, 4) NOT NULL CHECK (amount > 0),
    currency         CHAR(3) NOT NULL,
    debtor_name      VARCHAR(140),
    debtor_account   VARCHAR(34),
    creditor_name    VARCHAR(140),
    creditor_account VARCHAR(34),
    raw_xml          TEXT,                           -- Store the entire raw XML for audit
    status           VARCHAR(20) NOT NULL DEFAULT 'RECEIVED',
    created_at       TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    processed_at     TIMESTAMP WITH TIME ZONE
);

-- UETR and msg_id are the natural idempotency keys of ISO 20022
CREATE INDEX idx_inbound_payments_uetr   ON inbound_payments(uetr);
CREATE INDEX idx_inbound_payments_status ON inbound_payments(status, created_at);

package main

import (
    "encoding/xml"
    "fmt"
    "io"
    "os"
)

// Struct strictly for CreditTransferInfo — we don't parse the entire document
type CreditTransferInfo struct {
    EndToEndId  string  `xml:"PmtId>EndToEndId"`
    UETR        string  `xml:"PmtId>UETR"`
    Amount      float64 `xml:"IntrBkSttlmAmt"`
    Currency    string  `xml:"IntrBkSttlmAmt>Ccy,attr"`
    DebtorName  string  `xml:"Dbtr>Nm"`
    CreditorAcc string  `xml:"CdtrAcct>Id>Othr>Id"`
}

// parseBulkPacs008 — Streaming parser, O(1) memory usage
func parseBulkPacs008(filePath string, handler func(CreditTransferInfo) error) error {
    file, err := os.Open(filePath)
    if err != nil {
        return fmt.Errorf("open file: %w", err)
    }
    defer file.Close()

    decoder := xml.NewDecoder(file)
    
    for {
        token, err := decoder.Token()
        if err == io.EOF {
            break
        }
        if err != nil {
            return fmt.Errorf("decode token: %w", err)
        }

        // Process only when encountering the start element CdtTrfTxInf
        if se, ok := token.(xml.StartElement); ok && se.Name.Local == "CdtTrfTxInf" {
            var tx CreditTransferInfo
            // DecodeElement parses only the current sub-tree, not the entire document
            if err := decoder.DecodeElement(&tx, &se); err != nil {
                return fmt.Errorf("decode element: %w", err)
            }
            
            // Process immediately, do not accumulate in memory
            if err := handler(tx); err != nil {
                return fmt.Errorf("handle transaction: %w", err)
            }
        }
    }
    
    return nil
}

// Usage:
func main() {
    err := parseBulkPacs008("bulk_payments.xml", func(tx CreditTransferInfo) error {
        // Insert directly into DB, without buffering in memory
        return insertInboundPayment(tx)
    })
    if err != nil {
        panic(err)
    }
}

# Kong Gateway config — ISO 20022 transformation plugin
plugins:
  - name: request-transformer
    config:
      # Transform internal JSON format to XML for SWIFT submission
      body: xml_transform
      
  - name: rate-limiting
    config:
      minute: 1000        # Rate limit per partner
      policy: redis       # Distributed rate limiting

  - name: request-size-limiting
    config:
      allowed_payload_size: 100  # 100KB max — prevent XML bomb attacks

type IdempotencyService struct {
    redis *redis.Client
    db    *sql.DB
}

// CheckAndProcess — Two-layer idempotency
func (s *IdempotencyService) CheckAndProcess(
    ctx context.Context,
    key string,
    processor func() (interface{}, error),
) (interface{}, bool, error) {
    
    // Layer 1: Pending lock (5 minutes) — prevents concurrent processing
    locked, err := s.redis.SetNX(ctx,
        "lock:"+key,
        "processing",
        5*time.Minute,
    ).Result()
    
    if err != nil {
        return nil, false, err
    }
    if !locked {
        // Already being processed — return 409 Conflict
        return nil, false, ErrAlreadyProcessing
    }
    defer s.redis.Del(ctx, "lock:"+key)
    
    // Layer 2: Result cache (24-48 hours) — returns cached response
    cached, err := s.redis.Get(ctx, "result:"+key).Result()
    if err == nil {
        // Cache hit — already processed, return cached result
        var result interface{}
        json.Unmarshal([]byte(cached), &result)
        return result, true, nil // true = was cached
    }
    
    // Process for the first time
    result, err := processor()
    if err != nil {
        return nil, false, err
    }
    
    // Cache the result for 48 hours
    resultJSON, _ := json.Marshal(result)
    s.redis.Set(ctx, "result:"+key, resultJSON, 48*time.Hour)
    
    return result, false, nil // false = freshly processed
}

// Usage in payment webhook handler:
func (h *WebhookHandler) HandleGatewayWebhook(w http.ResponseWriter, r *http.Request) {
    idempotencyKey := r.Header.Get("X-Message-ID") // Unique per payment
    
    result, wasCached, err := h.idempotency.CheckAndProcess(
        r.Context(),
        idempotencyKey,
        func() (interface{}, error) {
            return h.processPayment(r.Context(), r.Body)
        },
    )
    
    if err == ErrAlreadyProcessing {
        w.WriteHeader(http.StatusConflict) // 409
        return
    }
    
    if wasCached {
        w.Header().Set("X-Idempotent-Replayed", "true")
    }
    
    json.NewEncoder(w).Encode(result)
}

func TestIdempotencyPayloadMismatch(t *testing.T) {
    // Request 1: Amount = 1,000,000 VND
    resp1 := sendPaymentRequest("idempotency-key-001", 1_000_000)
    assert.Equal(t, 201, resp1.StatusCode)
    
    // Request 2: SAME key but DIFFERENT amount = 2,000,000 VND
    resp2 := sendPaymentRequest("idempotency-key-001", 2_000_000)
    
    // Must be rejected with 422 Unprocessable Entity
    assert.Equal(t, 422, resp2.StatusCode)
    assert.Contains(t, resp2.Body, "idempotency_key_mismatch")
}

func TestConcurrentDoubleSubmit(t *testing.T) {
    const idempotencyKey = "payment-unique-key-xyz"
    
    // Send 2 concurrent requests with the SAME idempotency key
    results := make(chan int, 2)
    go func() {
        resp := sendPayment(idempotencyKey, 500000)
        results <- resp.StatusCode
    }()
    go func() {
        resp := sendPayment(idempotencyKey, 500000)
        results <- resp.StatusCode
    }()
    
    status1 := <-results
    status2 := <-results
    
    // Exactly 1 request must be 201 Created, the other 409 Conflict or cached 200
    statusCodes := []int{status1, status2}
    createdCount := countOccurrences(statusCodes, 201)
    assert.Equal(t, 1, createdCount, "Only one request should be processed as new")
    
    // Must not be charged twice
    assert.Equal(t, expectedSingleCharge, getAccountDebit("account-A"))
}

# Generate bulk file with 100,000 transactions (~150MB XML)
python3 generate_bulk_pacs008.py --count 100000 > bulk_test.xml

# Run parser with 50MB memory limit
go test -run TestBulkXMLParsing -memprofile mem.prof
go tool pprof mem.prof

# Expectation: heap allocation does not exceed 20MB despite the 150MB file