import { createProducer } from 'kafka.do'

export default {
  async fetch(request: Request, env: Env) {
    const producer = createProducer(env)

    // Send a single message
    const metadata = await producer.send({
      topic: 'orders',
      key: 'order-123',
      value: { orderId: '123', amount: 99.99 },
      headers: { source: 'web' }
    })

    console.log(`Message sent to partition ${metadata.partition} at offset ${metadata.offset}`)

    // Send multiple messages in a batch
    const results = await producer.sendBatch([
      { topic: 'orders', key: 'order-124', value: { orderId: '124', amount: 49.99 } },
      { topic: 'orders', key: 'order-125', value: { orderId: '125', amount: 149.99 } }
    ])

    await producer.close()

    return new Response('Messages sent')
  }
}

import { createConsumer } from 'kafka.do'

export default {
  async fetch(request: Request, env: Env) {
    const consumer = createConsumer(env, {
      groupId: 'order-processor',
      clientId: 'worker-1',
      autoCommit: true,
      fromBeginning: false
    })

    // Subscribe to topics
    await consumer.subscribe(['orders'])

    // Poll for messages
    const records = await consumer.poll(1000)

    for (const record of records) {
      console.log(`Received: ${record.key} = ${JSON.stringify(record.value)}`)
      console.log(`Topic: ${record.topic}, Partition: ${record.partition}, Offset: ${record.offset}`)
    }

    // Manual commit (if autoCommit is false)
    await consumer.commit()

    await consumer.close()

    return new Response(`Processed ${records.length} messages`)
  }
}

const consumer = createConsumer(env, { groupId: 'my-group' })
await consumer.subscribe(['orders'])

for await (const record of consumer) {
  console.log(`Processing: ${record.key}`)
  // Process each record as it arrives
}

import { createAdmin } from 'kafka.do'

export default {
  async fetch(request: Request, env: Env) {
    const admin = createAdmin(env)

    // Create a topic with 3 partitions
    await admin.createTopic({
      topic: 'orders',
      partitions: 3,
      config: {
        'retention.ms': '604800000' // 7 days
      }
    })

    // List all topics
    const topics = await admin.listTopics()
    console.log('Topics:', topics)

    // Describe a topic
    const metadata = await admin.describeTopic('orders')
    console.log('Partitions:', metadata.partitions.length)

    // Add more partitions
    await admin.createPartitions('orders', 6)

    // List consumer groups
    const groups = await admin.listGroups()

    // Describe a consumer group
    const groupInfo = await admin.describeGroup('order-processor')
    console.log('Group state:', groupInfo.state)
    console.log('Members:', groupInfo.members.length)

    // Get partition offsets
    const offsets = await admin.listOffsets('orders')
    for (const [partition, info] of offsets) {
      console.log(`Partition ${partition}: earliest=${info.earliest}, latest=${info.latest}`)
    }

    // Delete a topic
    await admin.deleteTopic('old-topic')

    // Delete a consumer group
    await admin.deleteGroup('old-group')

    await admin.close()

    return new Response('Admin operations complete')
  }
}

import { KafkaClient } from 'kafka.do/client'

// Create client pointing to your kafka.do deployment
const client = new KafkaClient({
  baseUrl: 'https://kafka.your-domain.workers.dev',
  clientId: 'my-app',
  timeout: 30000,
  headers: {
    'Authorization': 'Bearer your-token'
  }
})

// Check service health
const health = await client.health()
console.log('Status:', health.status)

// Producer operations
const producer = client.producer({ defaultTopic: 'events' })

await producer.send({
  key: 'user-123',
  value: { type: 'page_view', page: '/home' }
})

await producer.sendBatch([
  { key: 'user-123', value: { type: 'click', button: 'signup' } },
  { key: 'user-456', value: { type: 'page_view', page: '/about' } }
])

// Consumer operations
const consumer = client.consumer({
  groupId: 'analytics-processor',
  topics: ['events'],
  autoCommit: true
})

// Connect and join consumer group
const joinResult = await consumer.connect()
console.log('Member ID:', joinResult.memberId)

// Fetch messages from a partition
const { messages } = await consumer.fetch('events', 0, { offset: 0, limit: 100 })

for (const msg of messages) {
  console.log(`${msg.key}: ${JSON.stringify(msg.value)}`)
}

// Commit offsets
await consumer.commit()

// Get partition offsets
const offsets = await consumer.getOffsets('events', 0)
console.log(`Earliest: ${offsets.earliest}, Latest: ${offsets.latest}`)

// Disconnect from consumer group
await consumer.disconnect()

// Admin operations
const admin = client.admin()

await admin.createTopic({ topic: 'logs', partitions: 5 })

const topics = await admin.listTopics()
const topicInfo = await admin.describeTopic('logs')
const groups = await admin.listGroups()
const groupInfo = await admin.describeGroup('analytics-processor')

await admin.addPartitions('logs', 10)
await admin.deleteTopic('old-logs')
await admin.deleteGroup('old-group')

const consumer = createConsumer(env, config, {
  async onPartitionsAssigned(partitions) {
    console.log('Assigned:', partitions)
  },
  async onPartitionsRevoked(partitions) {
    console.log('Revoked:', partitions)
  }
})

import {
  KafkaPipeline,
  createKafkaPipeline,
  R2EventBridge,
  createR2EventBridge
} from 'kafka.do/integrations'

import { createKafkaPipeline, type CDCEvent } from 'kafka.do/integrations'

// Create a pipeline that routes events to topics based on database/collection
const pipeline = createKafkaPipeline({
  env,
  topicPattern: 'cdc.{db}.{coll}' // e.g., cdc.mydb.users
})

// Send a CDC event (typically from a MongoDB change stream)
await pipeline.send({
  eventId: 'evt-123',
  operationType: 'insert',
  ns: { db: 'mydb', coll: 'users' },
  documentKey: { _id: 'user-456' },
  fullDocument: { _id: 'user-456', name: 'Alice', email: 'alice@example.com' },
  timestamp: new Date().toISOString()
})

// Send multiple events in a batch
await pipeline.sendBatch(cdcEvents)

import { processCDCMessage, isInsertEvent } from 'kafka.do/integrations'

// Process CDC messages with typed handlers
await processCDCMessage(message, {
  database: 'mydb',
  collection: 'users',
  groupId: 'cdc-processor',
  onInsert: async (event) => console.log('New document:', event.fullDocument),
  onUpdate: async (event) => console.log('Updated:', event.updateDescription),
  onDelete: async (event) => console.log('Deleted:', event.documentKey)
})

import { createR2EventBridge, R2EventBridge } from 'kafka.do/integrations'

// Create an event bridge
const bridge = createR2EventBridge({
  env,
  topicPattern: 'r2.{bucket}', // e.g., r2.my-bucket
  bucketFilter: 'my-bucket',   // Optional: filter by bucket
  keyPrefixFilter: 'uploads/'  // Optional: filter by key prefix
})

// Process R2 events (typically in a Queue consumer)
export default {
  async queue(batch: MessageBatch, env: Env) {
    const bridge = createR2EventBridge({ env })

    for (const message of batch.messages) {
      const event = R2EventBridge.parseQueueMessage(message)
      if (event) {
        await bridge.processEvent(event)
      }
    }
  }
}

import { processR2Event, isR2ObjectCreated } from 'kafka.do/integrations'

// Process R2 events with typed handlers
await processR2Event(message, {
  bucketFilter: 'my-bucket',
  onObjectCreated: async (event) => {
    console.log('Created:', event.object.key, event.object.size)
  },
  onObjectDeleted: async (event) => {
    console.log('Deleted:', event.key)
  }
})

interface Env {
  TOPIC_PARTITION: DurableObjectNamespace
  CONSUMER_GROUP: DurableObjectNamespace
  CLUSTER_METADATA: DurableObjectNamespace
}