Kafka.hs

{- | Transport-neutral Kafka representation for outbox rows.

This module owns the conversion from 'Keiro.Outbox.Types.OutboxRow' to a
neutral 'KafkaProducerRecord' value. It deliberately does not import
@hw-kafka-client@ or @kafka-effectful@: keiro itself remains free of
librdkafka system-library requirements. The integration test package
(EP-22) bridges 'KafkaProducerRecord' to
@Kafka.Producer.Types.ProducerRecord@ from @hw-kafka-client@ inside its
own dependency scope.

A 'KafkaProducerRecord' carries everything the broker layer needs:
topic, optional partition key, the raw payload bytes from the EP-19
envelope, and the canonical header set. Building the record is pure;
publishing is the caller's responsibility.

The outbox worker opens one producer span around each claimed publish batch.
Adapters that need per-record broker visibility should add their own spans
around the actual Kafka produce calls.
-}
module Keiro.Outbox.Kafka (
    KafkaProducerRecord (..),
    outboxRowToKafkaRecord,
    integrationEventToKafkaRecord,
)
where

import Data.ByteString (ByteString)
import Data.Text.Encoding qualified as TE
import Keiro.Integration.Event (IntegrationEvent, integrationHeaders, integrationPayload)
import Keiro.Outbox.Types (OutboxRow (..))
import Keiro.Prelude

{- | A neutral Kafka producer record.

Fields:

* 'topic' — Kafka topic, taken from 'IntegrationEvent.destination'.
* 'key' — partition key bytes (UTF-8 encoded). 'Nothing' means
  Kafka round-robins the record across partitions and skips per-key
  ordering.
* 'payload' — exactly the bytes from
  'Keiro.Integration.Event.integrationPayload'.
* 'headers' — UTF-8 encoded view of
  'Keiro.Integration.Event.integrationHeaders'.

The byte encoding for keys and headers is UTF-8 by convention; Kafka
treats both as opaque bytes, so a future binary-key transport can drop
the encoding by populating 'key' directly.
-}
data KafkaProducerRecord = KafkaProducerRecord
    { topic :: !Text
    , key :: !(Maybe ByteString)
    , payload :: !ByteString
    , headers :: ![(ByteString, ByteString)]
    }
    deriving stock (Generic, Eq, Show)

-- | Build a 'KafkaProducerRecord' from a published outbox row.
outboxRowToKafkaRecord :: OutboxRow -> KafkaProducerRecord
outboxRowToKafkaRecord row = integrationEventToKafkaRecord (row ^. #event)

-- | Build a 'KafkaProducerRecord' directly from an 'IntegrationEvent'.
integrationEventToKafkaRecord :: IntegrationEvent -> KafkaProducerRecord
integrationEventToKafkaRecord event =
    KafkaProducerRecord
        { topic = event ^. #destination
        , key = fmap TE.encodeUtf8 (event ^. #key)
        , payload = integrationPayload event
        , headers = [(TE.encodeUtf8 n, TE.encodeUtf8 v) | (n, v) <- integrationHeaders event]
        }