import franz
import gleam/erlang/process
import gleam/io

pub fn main() {
  // 1. Connect to Kafka
  let name = process.new_name("my_kafka_client")
  let assert Ok(_) =
    franz.client()
    |> franz.endpoints([franz.Endpoint("localhost", 9092)])
    |> franz.option(franz.AutoStartProducers(True))
    |> franz.name(name)
    |> franz.start()

  let client = franz.named(name)

  // 2. Send a message
  let assert Ok(_) =
    franz.produce_sync(
      client: client,
      topic: "greetings",
      partition: franz.SinglePartition(0),
      key: <<"user:123">>,
      value: franz.Value(<<"Hello, Franz!">>, []),
    )

  io.println("Message sent successfully!")
}

import franz
import gleam/erlang/process

pub fn connect_to_kafka() {
  // Create a named client
  let name = process.new_name("my_client")
  let endpoints = [
    franz.Endpoint("broker1.example.com", 9092),
    franz.Endpoint("broker2.example.com", 9092),
  ]

  franz.client()
  |> franz.endpoints(endpoints)
  |> franz.option(franz.AutoStartProducers(True))
  |> franz.name(name)
  |> franz.start()
}

pub fn connect_with_auth() {
  // With SASL authentication
  let name = process.new_name("secure_client")

  franz.client()
  |> franz.endpoints([franz.Endpoint("secure.broker.com", 9093)])
  |> franz.sasl(franz.SaslCredentials(franz.ScramSha256, "username", "password"))
  |> franz.ssl(franz.SslEnabled)
  |> franz.name(name)
  |> franz.start()
}

import franz
import gleam/bit_array

pub fn send_user_event(client: franz.Client, user_id: String, event: String) {
  // Start a producer for the topic
  let assert Ok(_) =
    franz.producer(client, "user-events")
    |> franz.producer_option(franz.RequiredAcks(-1))
    |> franz.producer_option(franz.Compression(franz.Gzip))
    |> franz.producer_start()

  // Send the message synchronously
  franz.produce_sync(
    client: client,
    topic: "user-events",
    partition: franz.Partitioner(franz.Hash),  // Use hash partitioner based on key
    key: bit_array.from_string(user_id),
    value: franz.Value(
      bit_array.from_string(event),
      [#("event-type", "user-action")],  // Headers
    ),
  )
}

import franz
import gleam/io
import gleam/int

pub fn send_async_with_confirmation(client: franz.Client) {
  franz.produce_async(
    client: client,
    topic: "async-events",
    partition: franz.Partitioner(franz.Random),  // Random partition
    key: <<"">>,
    value: franz.Value(<<"async data">>, []),
    callback: fn(partition, offset) {
      let franz.Partition(p) = partition
      let franz.Offset(o) = offset
      io.println("Message delivered to partition " <> int.to_string(p))
      io.println("At offset " <> int.to_string(o))
    },
  )
}

import franz

pub fn fire_and_forget(client: franz.Client) {
  // No acknowledgment - fastest but may lose messages
  franz.produce(
    client: client,
    topic: "metrics",
    partition: franz.Partitioner(franz.Random),
    key: <<"">>,
    value: franz.Value(<<"metric data">>, []),
  )
}

import franz
import gleam/io
import gleam/erlang/process

pub fn start_consumer_group(client: franz.Client) {
  let name = process.new_name("analytics_consumer")

  franz.default_group_subscriber_config(
    name,
    client: client,
    group_id: "analytics-processors",
    topics: ["user-events", "system-events"],
    callback: fn(message, state) {
      case message {
        franz.KafkaMessage(offset, _key, value, _, _, _) -> {
          io.println("Processing message at offset " <> int.to_string(offset))
          // Process the message...

          // Commit the offset after processing
          franz.GroupCommit(state)
        }
        franz.KafkaMessageSet(topic, partition, _, messages) -> {
          io.println("Batch from " <> topic <> " partition " <> int.to_string(partition))
          // Process batch...
          franz.GroupCommit(state)
        }
      }
    },
    init_state: Nil,
  )
  |> franz.start_group_subscriber()
}

import franz
import gleam/io
import gleam/int
import gleam/erlang/process

pub fn subscribe_to_notifications(client: franz.Client) {
  let name = process.new_name("notification_subscriber")

  franz.default_topic_subscriber(
    name,
    client: client,
    topic: "notifications",
    callback: fn(partition, message, state) {
      let franz.KafkaMessage(offset, _key, value, _, _, _) = message
      io.println("Partition " <> int.to_string(partition) <> " offset " <> int.to_string(offset))
      // Process notification...
      franz.TopicAck(state)
    },
    initial_state: Nil,
  )
  |> franz.start_topic_subscriber()
}

import franz
import gleam/otp/supervision
import gleam/erlang/process

pub fn start_supervised() {
  let client_name = process.new_name("supervised_client")
  let consumer_name = process.new_name("supervised_consumer")

  let client_builder =
    franz.client()
    |> franz.endpoints([franz.Endpoint("localhost", 9092)])
    |> franz.name(client_name)

  let client = franz.named(client_name)

  let consumer_builder =
    franz.default_group_subscriber_config(
      consumer_name,
      client: client,
      group_id: "my-group",
      topics: ["events"],
      callback: fn(msg, state) { franz.GroupCommit(state) },
      init_state: Nil,
    )

  let children = [
    franz.supervised(client_builder),
    franz.group_subscriber_supervised(consumer_builder),
  ]

  supervision.start(children)
}

import franz
import gleam/erlang/process

let name = process.new_name("secure_client")

// PLAIN authentication (use with SSL!)
franz.default_client(name)
|> franz.endpoints([franz.Endpoint("kafka.example.com", 9093)])
|> franz.sasl(franz.SaslCredentials(franz.Plain, "username", "password"))

// SCRAM-SHA-256
franz.default_client(name)
|> franz.endpoints([franz.Endpoint("kafka.example.com", 9093)])
|> franz.sasl(franz.SaslCredentials(franz.ScramSha256, "username", "password"))

// SCRAM-SHA-512
franz.default_client(name)
|> franz.endpoints([franz.Endpoint("kafka.example.com", 9093)])
|> franz.sasl(franz.SaslCredentials(franz.ScramSha512, "username", "password"))

import franz
import gleam/erlang/process
import gleam/option.{Some}

let name = process.new_name("ssl_client")

// Simple SSL with system CA
franz.default_client(name)
|> franz.endpoints([franz.Endpoint("kafka.example.com", 9093)])
|> franz.ssl(franz.SslEnabled)

// Custom certificates (mTLS)
franz.default_client(name)
|> franz.endpoints([franz.Endpoint("kafka.example.com", 9093)])
|> franz.ssl(franz.SslWithOptions(
  cacertfile: Some("/path/to/ca.crt"),
  certfile: Some("/path/to/client.crt"),
  keyfile: Some("/path/to/client.key"),
  verify: franz.VerifyPeer,
))

import franz

pub fn handle_produce_result(result: Result(Nil, franz.ProduceError)) {
  case result {
    Ok(_) -> io.println("Success!")
    Error(franz.ProducerDown) -> restart_producer()
    Error(franz.ProducerMessageTooLarge) -> split_message()
    Error(franz.ProducerNotEnoughReplicas) -> retry_later()
    Error(error) -> log_error(error)
  }
}