// ❌ JSON without schema — producer can change shape at any time
// Producer v1 publishes:
await producer.send({ topic: 'payments', messages: [{
  value: JSON.stringify({ paymentId: 'P1', amount: 9900 })
}]})

// Producer v2 renames the field — no compilation error, no runtime error
await producer.send({ topic: 'payments', messages: [{
  value: JSON.stringify({ paymentId: 'P1', amountCents: 9900 })  // 'amount' is gone
}]})

// Consumer reads from offset 0 — sees both shapes in the same topic
// It has no way to know which messages use v1 and which use v2 schema
const consumer = kafka.consumer({ groupId: 'analytics' })
await consumer.run({
  eachMessage: async ({ message }) => {
    const event = JSON.parse(message.value!.toString())
    const amount = event.amount   // undefined for all v2 messages — silent NaN corruption
    await analyticsDb.insert({ paymentId: event.paymentId, revenue: amount })
    // analytics table silently accumulates NaN values for all post-deploy payments
  }
})

// payments.avsc — Avro schema definition
const paymentSchema = {
  type:      'record',
  name:      'PaymentProcessed',
  namespace: 'com.company.payments',
  fields: [
    { name: 'paymentId',   type: 'string' },
    { name: 'amountCents', type: 'long' },
    { name: 'currency',    type: 'string' },
    { name: 'customerId',  type: 'string' },
    { name: 'processedAt', type: { type: 'long', logicalType: 'timestamp-millis' } },
    // ✅ New optional field with default — BACKWARD compatible
    { name: 'metadata',    type: ['null', 'string'], default: null },
  ]
}

// Producer: register + serialize
import { SchemaRegistry } from '@kafkajs/confluent-schema-registry'

const registry = new SchemaRegistry({ host: 'http://schema-registry:8081' })

const { id } = await registry.register({
  type:   SchemaType.AVRO,
  schema: JSON.stringify(paymentSchema),
}, { subject: 'payments-value' })

const encodedValue = await registry.encode(id, {
  paymentId:   'P-42',
  amountCents: 9900,
  currency:    'USD',
  customerId:  'C-99',
  processedAt: Date.now(),
  metadata:    null,
})

await producer.send({
  topic:    'payments',
  messages: [{ key: 'P-42', value: encodedValue }]
})

// Consumer: decode using schema registry (writer schema auto-resolved)
await consumer.run({
  eachMessage: async ({ message }) => {
    // Registry fetches writer schema by ID from the message header
    // Applies Avro resolution rules against the consumer's reader schema
    const event = await registry.decode(message.value!)
    // event is typed as PaymentProcessed — fully deserialized
    await analyticsDb.insert({
      paymentId: event.paymentId,
      revenue:   event.amountCents,
    })
  }
})

// ✅ BACKWARD compatible change — new field with default value
// Old schema:
{ name: 'paymentId', type: 'string' },
{ name: 'amountCents', type: 'long' },

// New schema (BACKWARD compatible):
{ name: 'paymentId',   type: 'string' },
{ name: 'amountCents', type: 'long' },
{ name: 'metadata',    type: ['null', 'string'], default: null },  // ← has default
// New consumer reading old message: metadata field gets its default (null) ✅
// Old consumer reading new message: unknown field 'metadata' is ignored ❌

// ❌ BACKWARD breaking change — removing a required field
// Old schema has: { name: 'customerId', type: 'string' }
// New schema removes it — new consumer reading old message cannot find a value for absent field
// (If the field had no default in the old schema, this is an error)

// ✅ FORWARD compatible change — new field with default in NEW schema
// Old consumer reading new message: sees unknown field 'metadata', ignores it ✅

// ❌ FORWARD breaking change — removing a field the old consumer expects
// Old consumer expects 'customerId' — new schema does not have it
// Avro resolution: field not in writer schema, not in reader schema = error

# Set compatibility on the subject in Schema Registry
curl -X PUT http://schema-registry:8081/config/payments-value \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"compatibility": "FULL_TRANSITIVE"}'

# .github/workflows/schema-check.yml
name: Schema Compatibility Check

on: [pull_request]

jobs:
  schema-check:
    runs-on: ubuntu-latest
    services:
      schema-registry:
        image: confluentinc/cp-schema-registry:7.6.0
        env:
          SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'localhost:9092'
          SCHEMA_REGISTRY_HOST_NAME: schema-registry
        ports: ['8081:8081']

    steps:
      - uses: actions/checkout@v4

      - name: Check schema compatibility
        run: |
          # Attempt to register changed schemas against the registry
          # Registry enforces BACKWARD_TRANSITIVE — incompatible changes fail here
          node scripts/check-schema-compatibility.mjs
        env:
          SCHEMA_REGISTRY_URL: http://localhost:8081

// scripts/check-schema-compatibility.mjs
import { SchemaRegistry, SchemaType } from '@kafkajs/confluent-schema-registry'
import { readdirSync, readFileSync } from 'fs'

const registry = new SchemaRegistry({ host: process.env.SCHEMA_REGISTRY_URL })

const schemas = readdirSync('schemas/').filter(f => f.endsWith('.avsc'))

for (const file of schemas) {
  const schema = readFileSync(`schemas/${file}`, 'utf-8')
  const subject = file.replace('.avsc', '-value')

  // testCompatibility throws if the schema is incompatible
  const result = await registry.testCompatibility(subject, {
    type: SchemaType.AVRO, schema
  })

  if (!result) {
    console.error(`❌ Schema ${subject} is incompatible with registered versions`)
    process.exit(1)
  }
  console.log(`✅ ${subject} is compatible`)
}