data-contracts

Data Contracts

Ortak Doktrin

agents/shared/severity-rubric.md ve agents/shared/escalation-matrix.md default-load sayılır (agents/coordination.md §11). Bu skill'in çıktısı Critical / High / Medium / Low + kanıt formatında olmak zorunda — spekülatif Critical yasak. Sahiplik dışı bulgu ilgili agent'a delege; karar yetkisi eşiği aşılırsa kullanıcı onayı zorunlu.

Felsefe

• Contract = schema + semantics + SLA.
• Producer-owned.
• Versioning explicit — major BC-breaking; minor BC.
• Backward compatible default.
• DLQ for poison — schema invalid / business invariant fail.
• Schema registry merkez.
• Idempotent consumers — at-least-once + dedup.
• PII annotation — privacy bağı.

Ne Zaman Kullanılır

• Yeni event publish (producer çıkıyor)
• Yeni consumer onboard
• Breaking change kararı (major bump)
• Schema drift bulgu (CI fail veya production poison)
• Catalog API entity ekleme (Backstage)
• Privacy review PII field annotation
• DB warehouse contract (dbt)
• Kafka topic onboarding

Workflow

1) Contract design (producer)

# events/order-created.yaml (AsyncAPI 3.0)
asyncapi: 3.0.0
info:
  title: Order Created Event
  version: 1.0.0
  contact: { name: payments-team, email: payments@acme.com }
channels:
  events.order.created.v1: { address: events.order.created.v1 }
components:
  messages:
    OrderCreated:
      payload: { $ref: '#/components/schemas/OrderCreatedPayload' }
      headers:
        type: object
        properties:
          event_id: { type: string, format: uuid }
          schema_version: { type: string, const: "1.0.0" }
          trace_id: { type: string }
  schemas:
    OrderCreatedPayload:
      type: object
      required: [order_id, user_id, total_amount, currency, items]
      properties:
        order_id: { type: string, format: uuid }
        user_id: { type: string, x-pii: pseudo }
        total_amount: { type: integer, description: "minor units" }
        currency: { type: string, enum: [TRY, USD, EUR] }
        items: { type: array, items: { $ref: '...' } }
sla:
  latency_p99_ms: 500
  delivery_guarantee: at_least_once
  ordering: per_partition_by_user_id
  retention_days: 7
  availability: 99.9%
  consumer_lag_alarm_seconds: 60

2) Schema registry

# Confluent Schema Registry register
curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data @schema-payload.json \
  http://schema-registry/subjects/order.created-value/versions

Compatibility mode: BACKWARD (varsayılan).

3) Versioning karar

Bump	Trigger	Migration
Major 2.0	Required field remove / type değişti / semantics shift	Yeni topic v2; 6 ay sunset v1
Minor 1.1	Optional field added / enum extra (non-default)	BC; consumer ignore unknown
Patch 1.0.1	Doc / example / validation tighten (BC)	No action

Sessiz remove / anlam değişikliği yasak.

4) Producer impl

class OrderCreatedPublisher:
    SCHEMA_VERSION = "1.0.0"
    TOPIC = "events.order.created.v1"

    def publish(self, order: Order) -> None:
        payload = OrderCreatedPayload(
            order_id=str(order.id),
            user_id=pseudonymize(order.user_id),
            total_amount=order.total_minor_units,
            currency=order.currency,
            items=[LineItem.from_(i) for i in order.items],
        )
        validate(payload, schema=load_schema("order-created", self.SCHEMA_VERSION))
        kafka.send(self.TOPIC, value=payload, headers={
            "event_id": order.checkout_idempotency_key,
            "schema_version": self.SCHEMA_VERSION,
            "trace_id": get_trace_id(),
        })

5) Consumer impl (idempotent + DLQ)

class OrderCreatedHandler:
    def handle(self, msg: KafkaMessage) -> None:
        try:
            payload = OrderCreatedPayload.parse(msg.value)
        except SchemaValidationError as e:
            dead_letter(msg, reason="schema_invalid", error=str(e))
            return

        if seen(msg.headers["event_id"]):
            return

        if payload.total_amount < 0:
            dead_letter(msg, reason="business_invariant_fail")
            return

        with transactional():
            mark_seen(msg.headers["event_id"])
            send_confirmation_email(payload)

6) DLQ pattern

events.order.created.v1
  ↓ consumer
  ├── success → processed_events table
  ├── schema_invalid → dlq.events.order.created.schema (no retry)
  ├── business_invariant → dlq.events.order.created.business (no retry)
  └── transient → retry.events.order.created (exp backoff, max 3) → dlq

DLQ retention 30g; alert > 10 msg/dk per topic.

7) CI contract test

# .github/workflows/contract.yml
- run: npx @asyncapi/cli validate events/*.yaml
- run: |
    for s in events/*.yaml; do
      sr-cli compatibility-check --subject $(basename $s .yaml)-value --file $s
    done
- run: pact-broker can-i-deploy --pacticipant checkout-svc --version=$GIT_SHA
- run: |
    npx asyncapi generate types
    git diff --exit-code generated/    # drift check

8) Catalog API entity

Backstage catalog-info.yaml aynı repo'da + API entity dosyası:

apiVersion: backstage.io/v1alpha1
kind: API
metadata:
  name: order-created-event
  annotations:
    backstage.io/definition-at-location: url:https://github.com/.../order-created.yaml
spec:
  type: asyncapi
  lifecycle: production
  owner: group:payments-team
  definition: { $text: events/order-created.yaml }

9) PII annotation

properties:
  user_id: { type: string, x-pii: pseudo }
  raw_email: { type: string, x-pii: direct, x-retention-days: 1095 }

/privacy-review data-catalog otomatik update.

10) Database contract (dbt)

# dbt-models/orders.yml
version: 2
models:
  - name: orders
    config: { contract: { enforced: true }, on_schema_change: fail }
    columns:
      - name: order_id
        data_type: uuid
        constraints: [{ type: not_null }, { type: unique }]
      - name: total_amount_minor
        data_type: bigint
        constraints: [{ type: not_null }]

dbt run contract fail → pipeline red.

Checklist

• AsyncAPI/OpenAPI spec her event/endpoint
• Producer-owned, repo'da kanonik
• Schema Registry register (Confluent / Apicurio)
• Compatibility BACKWARD (default)
• Versioning explicit (semver)
• Producer schema validate (pre-publish)
• Consumer idempotency (event_id + processed_events table)
• DLQ topic + 30g retention + alert
• Retry topic + exp backoff + max 3
• PII annotation (x-pii: pseudo/direct)
• SLA (latency, delivery, lag)
• CI: AsyncAPI validate + schema compat + Pact + drift
• Backstage API entity
• dbt contract (warehouse table)
• Sunset plan major bump

Antipattern

• Schema yok / wiki.
• Producer-consumer copy-paste.
• Versioning yok.
• Sessiz field remove.
• Anlam değişikliği.
• PII annotation yok.
• DLQ yok.
• Idempotency yok.
• At-most-once kabul.
• Schema Registry yok.
• Compatibility NONE.
• CI contract test yok.
• SLA yok.
• Catalog API entity yok.
• Ownership belirsiz.

Örnek Agent Davranışı

User: /data-contract events.order.created
Agent (api-contract-guardian + system-design-architect +
       observability-engineer + privacy-engineer):

1. Discovery: `events.order.created` topic var, schema repo'da yok;
   3 producer + 7 consumer Slack üzerinden koordine.
2. Producer audit: checkout-svc payload üretiyor; field `user_email`
   plaintext (PII); idempotency key yok; schema_version header yok.
3. Consumer audit: 7 consumer DLQ kullanmıyor; 3'ü tekrarlı dedup yapmıyor;
   `notifications-svc` poison message 2 hafta retry storm.
4. CI: contract test yok; AsyncAPI spec yok; schema compat check yok.
5. Catalog: API entity yok; consumer/producer ilişki saklı.

Findings:
- Critical: 3 consumer dedup yok → duplicate process realized risk
- Critical: notifications-svc DLQ yok, retry storm 2h sustained
- High: PII user_email plaintext production (pseudo zorunlu)
- High: schema_version header yok → major bump migrate stratejisi yok
- Medium: CI contract test yok
- Medium: Backstage API entity yok

Action items: 12 issue + 4-week roadmap

Çıktı Formatı

# Data Contract Review: <event-or-api>

## Discovery
- Producer + Consumer inventory
- Schema location (repo / wiki / yok)

## Contract Spec (AsyncAPI / OpenAPI)
- Schema YAML
- SLA (latency + delivery + lag)
- PII annotation

## Versioning Strategy
- Current version
- Sunset plan (varsa major bump)

## Producer Impl

## Consumer Impl
- Idempotency (event_id + processed_events)
- DLQ pattern
- Retry pattern

## CI Contract Tests

## Catalog Integration
- Backstage API entity

## Findings (Critical/High/Medium/Low)

## Action Items
| P | Aksiyon | Sahip | Bitiş | Issue |