Kafka.hs

{- | Reconstruct an 'IntegrationEvent' from Kafka payload bytes plus
headers.

This module is the receiving-side counterpart of
'Keiro.Outbox.Kafka.integrationEventToKafkaRecord'. It is pure: the
caller supplies the bytes and the @Text@-keyed header map produced by
its Kafka adapter, and gets back a decoded envelope or a typed error.

@keiro@ itself does not depend on @hw-kafka-client@ or
@shibuya-kafka-adapter@; the consumer adapter in EP-22 bridges the
broker library's header type to @[(Text, Text)]@ before calling
'integrationEventFromKafka'.
-}
module Keiro.Inbox.Kafka (
    KafkaInboundRecord (..),
    KafkaDecodeError (..),
    integrationEventFromKafka,
)
where

import Data.Aeson qualified as Aeson
import Data.ByteString (ByteString)
import Data.Maybe (mapMaybe)
import Data.Text qualified as Text
import Data.Text.Encoding qualified as TextEncoding
import Data.Text.Read qualified as TextRead
import Data.Time.Format.ISO8601 (iso8601ParseM)
import Data.UUID qualified as UUID
import Keiro.Inbox.Types (KafkaDeliveryRef (..))
import Keiro.Integration.Event (
    IntegrationContentType (..),
    IntegrationEvent (..),
    SchemaReference (..),
    TraceContext (..),
    headerAttributes,
    headerCausationId,
    headerContentType,
    headerCorrelationId,
    headerDestination,
    headerEventType,
    headerMessageId,
    headerOccurredAt,
    headerSchemaFingerprint,
    headerSchemaId,
    headerSchemaRegistry,
    headerSchemaSubject,
    headerSchemaVersion,
    headerSchemaVersionRef,
    headerSource,
    headerSourceEventId,
    headerSourceGlobalPosition,
    headerTraceParent,
    headerTraceState,
    parseContentType,
 )
import Keiro.Prelude
import Kiroku.Store.Types (EventId (..), GlobalPosition (..))

{- | A Kafka record as seen by the consumer-side adapter, decoupled from
the broker library's record type.
-}
data KafkaInboundRecord = KafkaInboundRecord
    { topic :: !Text
    , partition :: !Int64
    , offset :: !Int64
    , key :: !(Maybe Text)
    , payload :: !ByteString
    , headers :: ![(Text, Text)]
    , receivedAt :: !UTCTime
    }
    deriving stock (Generic, Eq, Show)

-- | Typed failures from 'integrationEventFromKafka'.
data KafkaDecodeError
    = MissingHeader !Text
    | InvalidIntHeader !Text !Text
    | InvalidUuidHeader !Text !Text
    | InvalidTimeHeader !Text !Text
    | InvalidJsonHeader !Text !Text
    deriving stock (Generic, Eq, Show)

{- | Reconstruct a full 'IntegrationEvent' plus the 'KafkaDeliveryRef'
recorded for diagnostics.

The reconstruction is faithful to the canonical header names defined
by 'Keiro.Integration.Event' (e.g. @keiro-message-id@, @keiro-source@,
@traceparent@). Missing required headers (@keiro-source@,
@keiro-destination@, @keiro-event-type@, @keiro-schema-version@,
@content-type@, @keiro-message-id@) produce 'MissingHeader'; malformed
numeric or UUID headers produce 'InvalidIntHeader' / 'InvalidUuidHeader'.
Optional headers are silently absent in the resulting envelope.
-}
integrationEventFromKafka ::
    KafkaInboundRecord ->
    Either KafkaDecodeError (IntegrationEvent, KafkaDeliveryRef)
integrationEventFromKafka record = do
    let hs = record ^. #headers
    source <- requireHeader hs headerSource
    destination <- requireHeader hs headerDestination
    eventType <- requireHeader hs headerEventType
    schemaVersionText <- requireHeader hs headerSchemaVersion
    schemaVersion <- parseInt headerSchemaVersion schemaVersionText
    contentTypeRaw <- requireHeader hs headerContentType
    messageId <- requireHeader hs headerMessageId
    schemaReference <- buildSchemaReference hs
    sourceEventId <- traverseLookup hs headerSourceEventId (fmap EventId . parseUuid headerSourceEventId)
    sourceGlobalPosition <-
        traverseLookup hs headerSourceGlobalPosition (fmap GlobalPosition . parseInt headerSourceGlobalPosition)
    causationId <- traverseLookup hs headerCausationId (fmap EventId . parseUuid headerCausationId)
    correlationId <- traverseLookup hs headerCorrelationId (fmap EventId . parseUuid headerCorrelationId)
    occurredAt <- fromMaybe (record ^. #receivedAt) <$> traverseLookup hs headerOccurredAt (parseTimeHeader headerOccurredAt)
    attributes <- traverseLookup hs headerAttributes (parseJsonHeader headerAttributes)
    let traceContext = case Prelude.lookup headerTraceParent hs of
            Nothing -> Nothing
            Just tp -> Just (TraceContext tp (Prelude.lookup headerTraceState hs))
        event =
            IntegrationEvent
                { messageId
                , source
                , destination
                , key = record ^. #key
                , eventType
                , schemaVersion
                , contentType = parseContentType contentTypeRaw
                , schemaReference
                , sourceEventId
                , sourceGlobalPosition
                , payloadBytes = record ^. #payload
                , occurredAt
                , causationId
                , correlationId
                , traceContext
                , attributes
                }
        kafka =
            KafkaDeliveryRef
                { topic = record ^. #topic
                , partition = record ^. #partition
                , offset = record ^. #offset
                }
    pure (event, kafka)

-- ---------------------------------------------------------------------------
-- Helpers
-- ---------------------------------------------------------------------------

requireHeader :: [(Text, Text)] -> Text -> Either KafkaDecodeError Text
requireHeader hs name = case Prelude.lookup name hs of
    Just v -> Right v
    Nothing -> Left (MissingHeader name)

traverseLookup ::
    [(Text, Text)] ->
    Text ->
    (Text -> Either KafkaDecodeError a) ->
    Either KafkaDecodeError (Maybe a)
traverseLookup hs name parser = case Prelude.lookup name hs of
    Nothing -> Right Nothing
    Just raw -> fmap Just (parser raw)

parseInt :: (Integral a) => Text -> Text -> Either KafkaDecodeError a
parseInt name raw = case TextRead.signed TextRead.decimal raw of
    Right (n, rest) | Text.null rest -> Right n
    _ -> Left (InvalidIntHeader name raw)

parseUuid :: Text -> Text -> Either KafkaDecodeError UUID.UUID
parseUuid name raw = case UUID.fromText raw of
    Just u -> Right u
    Nothing -> Left (InvalidUuidHeader name raw)

parseTimeHeader :: Text -> Text -> Either KafkaDecodeError UTCTime
parseTimeHeader name raw =
    maybe (Left (InvalidTimeHeader name raw)) Right (iso8601ParseM (Text.unpack raw))

parseJsonHeader :: Text -> Text -> Either KafkaDecodeError Value
parseJsonHeader name raw =
    case Aeson.eitherDecodeStrict (TextEncoding.encodeUtf8 raw) of
        Right value -> Right value
        Left _ -> Left (InvalidJsonHeader name raw)

buildSchemaReference ::
    [(Text, Text)] ->
    Either KafkaDecodeError (Maybe SchemaReference)
buildSchemaReference hs = do
    let registry = Prelude.lookup headerSchemaRegistry hs
        subject = Prelude.lookup headerSchemaSubject hs
        fingerprint = Prelude.lookup headerSchemaFingerprint hs
    versionRef <- traverseLookup hs headerSchemaVersionRef (parseInt headerSchemaVersionRef)
    schemaId <- traverseLookup hs headerSchemaId (parseInt headerSchemaId)
    let presentFields =
            mapMaybe id [registry, subject, fmap (Text.pack . show) versionRef, fmap (Text.pack . show) schemaId, fingerprint]
    if null presentFields
        then pure Nothing
        else
            pure
                ( Just
                    ( SchemaReference
                        { registry
                        , subject
                        , version = versionRef
                        , schemaId
                        , fingerprint
                        }
                    )
                )

{- | Silence unused-import warning when 'IntegrationContentType' is
imported only via the open re-export but referenced through
'parseContentType'.
-}
_unusedKeepContentType :: IntegrationContentType -> ()
_unusedKeepContentType _ = ()