// ✅ Kinesis producer — PartitionKey determines shard assignment
import { KinesisClient, PutRecordCommand, PutRecordsCommand } from '@aws-sdk/client-kinesis'

const kinesis = new KinesisClient({ region: 'us-east-1' })

// Single record
await kinesis.send(new PutRecordCommand({
  StreamName:   'clickstream',
  PartitionKey: userId,           // same userId → same shard → ordered events per user
  Data:         Buffer.from(JSON.stringify({
    userId,
    eventType: 'page.view',
    path:      '/products/42',
    timestamp: Date.now(),
  })),
}))

// Batch: up to 500 records per PutRecords call (max 5 MB total)
const records = events.map(e => ({
  PartitionKey: e.userId,
  Data:         Buffer.from(JSON.stringify(e)),
}))

const result = await kinesis.send(new PutRecordsCommand({
  StreamName: 'clickstream',
  Records:    records,
}))

// PutRecords returns per-record success/failure — handle partial failures
const failed = result.Records?.filter(r => r.ErrorCode)
if (failed?.length) {
  // Retry only the failed records
  await retryFailedRecords(failed, records)
}

// Shard sizing formula
const writeThroughputMBps  = 50   // target: 50 MB/s ingest
const writePerShardMBps    = 1    // Kinesis limit: 1 MB/s per shard
const readConsumers        = 2    // dashboard Lambda + Flink job

// Standard GetRecords: all consumers share 2 MB/s read per shard
const readThroughputShared = 2    // MB/s per shard, shared
const requiredForReads     = Math.ceil((writeThroughputMBps * readConsumers) / readThroughputShared)

const requiredShards = Math.max(
  Math.ceil(writeThroughputMBps / writePerShardMBps),  // write constraint: 50
  requiredForReads,                                     // read constraint: 50
)

// With Enhanced Fan-Out: each consumer gets 2 MB/s dedicated — read constraint disappears
const requiredShardsEFO = Math.ceil(writeThroughputMBps / writePerShardMBps)  // 50

// ❌ Standard GetRecords — all consumers share 2 MB/s per shard
// At 50 shards × 2 MB/s = 100 MB/s total read bandwidth, split between all consumers
// 2 consumers → each gets 50 MB/s → may lag at high throughput

// ✅ Enhanced Fan-Out (EFO) — dedicated 2 MB/s per consumer per shard
// 2 EFO consumers × 50 shards × 2 MB/s = 200 MB/s total (no sharing)
import { RegisterStreamConsumerCommand, SubscribeToShardCommand } from '@aws-sdk/client-kinesis'

// Register this consumer as an EFO consumer (once per application)
const registration = await kinesis.send(new RegisterStreamConsumerCommand({
  StreamARN:    process.env.CLICKSTREAM_ARN!,
  ConsumerName: 'dashboard-lambda-efo',
}))

const consumerArn = registration.Consumer!.ConsumerARN!

// Subscribe to a specific shard — receives a push stream (HTTP/2 server push)
// Unlike GetRecords (polling), EFO uses long-lived connections
const response = await kinesis.send(new SubscribeToShardCommand({
  ConsumerARN: consumerArn,
  ShardId:     'shardId-000000000000',
  StartingPosition: { Type: 'LATEST' },
}))

// Process the event stream
for await (const event of response.EventStream!) {
  if (event.SubscribeToShardEvent) {
    for (const record of event.SubscribeToShardEvent.Records ?? []) {
      const data = JSON.parse(Buffer.from(record.Data!).toString())
      await processDashboardEvent(data)
    }
  }
}

// ✅ kafkajs client with MSK — no code changes vs self-managed Kafka
import { Kafka } from 'kafkajs'

const kafka = new Kafka({
  clientId: 'analytics-service',
  brokers:  process.env.MSK_BOOTSTRAP_SERVERS!.split(','),
  // MSK Provisioned: 'b-1.mycluster.xxxxx.c2.kafka.us-east-1.amazonaws.com:9092'
  // MSK Serverless:  'xxxxx.c2.kafka-serverless.us-east-1.amazonaws.com:9098' (IAM auth)
  ssl: true,
  sasl: {
    mechanism: 'aws',    // MSK IAM authentication
    authenticationProvider: {
      getToken: async () => {
        // MSK IAM: token generated from AWS SDK credentials (ECS task role)
        const { generateAuthToken } = await import('aws-msk-iam-sasl-signer-js')
        return generateAuthToken({ region: 'us-east-1' })
      }
    }
  }
})
// All kafkajs APIs work identically: producers, consumers, admin, Schema Registry

// MSK Serverless: throughput auto-scales — no shard count decision
// MSK Provisioned: choose broker instance type based on throughput
//
// Broker sizing guide (MSK Provisioned):
// kafka.m5.large:  ~150 MB/s per broker, 2 vCPU, 8 GB RAM  → small workloads
// kafka.m5.xlarge: ~350 MB/s per broker, 4 vCPU, 16 GB RAM → medium workloads
// kafka.m5.2xlarge: ~700 MB/s per broker                   → high throughput
//
// Minimum cluster: 3 brokers (AZ-distributed) for production HA
// Total cluster throughput ≈ broker_throughput × broker_count / replication_factor

// ✅ MSK IAM authentication with ECS task role — no credentials in code
// Requires: kafka-client-iam-auth-library on classpath (Java) 
// or aws-msk-iam-sasl-signer-js (Node.js)

import { generateAuthToken } from 'aws-msk-iam-sasl-signer-js'

const kafka = new Kafka({
  clientId: 'analytics-service',
  brokers:  [process.env.MSK_BOOTSTRAP!],
  ssl:      true,
  sasl: {
    mechanism: 'oauthbearer',
    oauthBearerProvider: async () => {
      const tokenProvider = await generateAuthToken({ region: 'us-east-1' })
      return {
        value: tokenProvider.token,
      }
    }
  }
})
// IAM policy required on ECS task role:
// kafka-cluster:Connect, kafka-cluster:AlterCluster, kafka-cluster:DescribeCluster
// kafka-cluster:CreateTopic, kafka-cluster:WriteData, kafka-cluster:ReadData