Event.hs

-- | Public integration-event envelope.
--
-- A keiro integration event is a public message that crosses from one bounded
-- context to another over Kafka (or another transport). It is distinct from a
-- private 'Kiroku.Store.Types.RecordedEvent': domain events are internal facts
-- of one event stream, while integration events are stable public contracts
-- versioned independently from the private model.
--
-- This module owns the wire shape, identity rules, and pure encode/decode
-- helpers. Storage of the envelope ('Keiro.Outbox', 'Keiro.Inbox') and
-- Kafka-specific producer/consumer wrappers consume this contract; they do
-- not redefine it.
module Keiro.Integration.Event
  ( -- * Envelope
    IntegrationEvent (..),
    IntegrationContentType (..),
    SchemaReference (..),
    TraceContext (..),
    IntegrationEventError (..),

    -- * JSON convenience helpers
    encodeJsonIntegrationEvent,
    decodeJsonIntegrationEvent,

    -- * Wire mapping
    integrationPayload,
    integrationHeaders,
    headerMessageId,
    headerSource,
    headerDestination,
    headerEventType,
    headerSchemaVersion,
    headerContentType,
    headerSchemaRegistry,
    headerSchemaSubject,
    headerSchemaVersionRef,
    headerSchemaId,
    headerSchemaFingerprint,
    headerSourceEventId,
    headerSourceGlobalPosition,
    headerCausationId,
    headerCorrelationId,
    headerTraceParent,
    headerTraceState,
    headerOccurredAt,
    headerAttributes,
    contentTypeText,
    parseContentType,
  )
where

import Data.Aeson qualified as Aeson
import Data.Aeson.Types (parseEither)
import Data.ByteString (ByteString)
import Data.ByteString.Lazy qualified as Lazy
import Data.Text qualified as Text
import Data.Text.Encoding qualified as TextEncoding
import Data.Time.Format.ISO8601 (iso8601Show)
import Data.UUID qualified as UUID
import Keiro.Prelude
import Kiroku.Store.Types (EventId (..), GlobalPosition (..))

-- | The canonical integration-event envelope.
--
-- The envelope is byte-oriented so future schema-registry integration (Avro,
-- Protobuf, JSON Schema) does not require a table or API migration. JSON
-- remains the v1 encoding, but the contract itself does not commit to it.
--
-- Identity rules:
--
-- * 'messageId' is an application-level id (UUIDv7 or equivalent
--   time-ordered UUID) minted by the producer subscription when it writes
--   the outbox row. It is stable across publish retries because it lives in
--   the row; Kafka topic/partition/offset are delivery metadata only and
--   are /not/ the canonical dedupe key.
-- * 'sourceEventId' and 'sourceGlobalPosition' identify the private event
--   that produced this integration event. A single source event can fan out
--   to multiple integration events with distinct 'messageId's; consumers
--   can opt into source-position deduplication when they want to suppress
--   reissued public events sharing an upstream cause.
--
-- Routing:
--
-- * 'source' is the producing bounded context (e.g. @\"ordering\"@).
-- * 'destination' is the Kafka topic, conventionally including a contract
--   version (@\"billing.orders.v1\"@).
-- * 'key' partitions per-aggregate within the destination topic. Consumers
--   see events for the same 'key' in producer order under EP-20's per-key
--   head-of-line publisher policy.
data IntegrationEvent = IntegrationEvent
  { messageId :: !Text,
    source :: !Text,
    destination :: !Text,
    key :: !(Maybe Text),
    eventType :: !Text,
    schemaVersion :: !Int,
    contentType :: !IntegrationContentType,
    schemaReference :: !(Maybe SchemaReference),
    sourceEventId :: !(Maybe EventId),
    sourceGlobalPosition :: !(Maybe GlobalPosition),
    payloadBytes :: !ByteString,
    occurredAt :: !UTCTime,
    causationId :: !(Maybe EventId),
    correlationId :: !(Maybe EventId),
    traceContext :: !(Maybe TraceContext),
    attributes :: !(Maybe Value)
  }
  deriving stock (Eq, Show, Generic)

-- | Content type of the payload bytes.
--
-- 'ApplicationJson' is the v1 default. 'OtherContentType' is the open door
-- for Avro, Protobuf, or any future registry-backed binary format; a future
-- schema-registry adapter can populate 'schemaReference' alongside.
data IntegrationContentType
  = ApplicationJson
  | OtherContentType !Text
  deriving stock (Eq, Show, Generic)

-- | Optional registry-neutral schema reference.
--
-- Every field is optional because v1 does not require a registry. A future
-- adapter may populate any combination of registry name, subject, version,
-- numeric schema id, or fingerprint depending on the registry vendor. The
-- core envelope preserves all fields verbatim through publish and consume.
data SchemaReference = SchemaReference
  { registry :: !(Maybe Text),
    subject :: !(Maybe Text),
    version :: !(Maybe Int),
    schemaId :: !(Maybe Int64),
    fingerprint :: !(Maybe Text)
  }
  deriving stock (Eq, Show, Generic)

-- | W3C Trace Context propagation fields.
--
-- When a service captures @traceparent@ and (optionally) @tracestate@ from
-- the producing request and threads them through, both fields survive
-- publish and consume via Kafka headers so a consumer can continue the
-- same trace.
data TraceContext = TraceContext
  { traceparent :: !Text,
    tracestate :: !(Maybe Text)
  }
  deriving stock (Eq, Show, Generic)

-- | Typed decode errors. The encode side cannot fail; the decode side can
-- fail if payload bytes are malformed JSON, the JSON does not satisfy the
-- target type, or required envelope metadata is missing.
data IntegrationEventError
  = MalformedPayload !Text
  | DecodeFailed !Text
  | MissingField !Text
  | UnsupportedContentType !Text
  deriving stock (Eq, Show, Generic)

-- | The payload bytes as they should be put on the wire.
integrationPayload :: IntegrationEvent -> ByteString
integrationPayload = (^. #payloadBytes)

-- | The set of headers a transport should attach to a published message.
--
-- Every header value is a 'Text' for portability — Kafka header values are
-- arbitrary bytes, but text headers are the convention for human-readable
-- metadata. The publisher converts these to UTF-8 bytes; the consumer
-- decodes UTF-8 back to text.
--
-- Headers are emitted only when their source field is populated, so a
-- service that does not capture trace context, causation, or a schema
-- reference does not pay a header for it.
integrationHeaders :: IntegrationEvent -> [(Text, Text)]
integrationHeaders event =
  concat
    [ [ (headerMessageId, event ^. #messageId),
        (headerSource, event ^. #source),
        (headerDestination, event ^. #destination),
        (headerEventType, event ^. #eventType),
        (headerSchemaVersion, Text.pack (show (event ^. #schemaVersion))),
        (headerContentType, contentTypeText (event ^. #contentType))
      ],
      case event ^. #schemaReference of
        Nothing -> []
        Just ref ->
          concat
            [ maybeHeader headerSchemaRegistry (ref ^. #registry),
              maybeHeader headerSchemaSubject (ref ^. #subject),
              maybeHeader headerSchemaVersionRef (fmap (Text.pack . show) (ref ^. #version)),
              maybeHeader headerSchemaId (fmap (Text.pack . show) (ref ^. #schemaId)),
              maybeHeader headerSchemaFingerprint (ref ^. #fingerprint)
            ],
      maybeHeader headerSourceEventId (fmap (UUID.toText . eventIdToUuid) (event ^. #sourceEventId)),
      maybeHeader headerSourceGlobalPosition (fmap globalPositionText (event ^. #sourceGlobalPosition)),
      maybeHeader headerCausationId (fmap (UUID.toText . eventIdToUuid) (event ^. #causationId)),
      maybeHeader headerCorrelationId (fmap (UUID.toText . eventIdToUuid) (event ^. #correlationId)),
      case event ^. #traceContext of
        Nothing -> []
        Just tc ->
          (headerTraceParent, tc ^. #traceparent)
            : maybeHeader headerTraceState (tc ^. #tracestate),
      [(headerOccurredAt, Text.pack (iso8601Show (event ^. #occurredAt)))],
      maybeHeader headerAttributes (fmap (TextEncoding.decodeUtf8 . Lazy.toStrict . Aeson.encode) (event ^. #attributes))
    ]
  where
    maybeHeader name = maybe [] (\value -> [(name, value)])

-- | Canonical header names. Lower-case @kebab-case@ matches Kafka convention.
headerMessageId, headerSource, headerDestination, headerEventType, headerSchemaVersion, headerContentType :: Text
headerMessageId = "keiro-message-id"
headerSource = "keiro-source"
headerDestination = "keiro-destination"
headerEventType = "keiro-event-type"
headerSchemaVersion = "keiro-schema-version"
headerContentType = "content-type"

headerSchemaRegistry, headerSchemaSubject, headerSchemaVersionRef, headerSchemaId, headerSchemaFingerprint :: Text
headerSchemaRegistry = "keiro-schema-registry"
headerSchemaSubject = "keiro-schema-subject"
headerSchemaVersionRef = "keiro-schema-version-ref"
headerSchemaId = "keiro-schema-id"
headerSchemaFingerprint = "keiro-schema-fingerprint"

headerSourceEventId, headerSourceGlobalPosition, headerCausationId, headerCorrelationId :: Text
headerSourceEventId = "keiro-source-event-id"
headerSourceGlobalPosition = "keiro-source-global-position"
headerCausationId = "keiro-causation-id"
headerCorrelationId = "keiro-correlation-id"

headerTraceParent, headerTraceState :: Text
headerTraceParent = "traceparent"
headerTraceState = "tracestate"

headerOccurredAt, headerAttributes :: Text
headerOccurredAt = "keiro-occurred-at"
headerAttributes = "keiro-attributes"

-- | The canonical wire string for a 'IntegrationContentType'.
contentTypeText :: IntegrationContentType -> Text
contentTypeText = \case
  ApplicationJson -> "application/json"
  OtherContentType raw -> raw

-- | Parse a content-type header back into an 'IntegrationContentType'. The
-- JSON form round-trips to 'ApplicationJson'; everything else is preserved
-- verbatim as 'OtherContentType'.
parseContentType :: Text -> IntegrationContentType
parseContentType raw
  | normalized == "application/json" = ApplicationJson
  | otherwise = OtherContentType raw
  where
    normalized = Text.toLower (Text.strip (Text.takeWhile (/= ';') raw))

-- | Build an 'IntegrationEvent' from a JSON-serializable business payload.
--
-- @encodeJsonIntegrationEvent envelope value@ replaces 'contentType' with
-- 'ApplicationJson' and 'payloadBytes' with the UTF-8 encoding of
-- @value@'s JSON representation. Every other envelope field is taken from
-- @envelope@ verbatim, so the caller controls identity, routing, and
-- metadata.
encodeJsonIntegrationEvent ::
  (ToJSON a) =>
  -- | Envelope template carrying identity, routing, and metadata.
  IntegrationEvent ->
  -- | Business payload to encode as JSON.
  a ->
  IntegrationEvent
encodeJsonIntegrationEvent envelope value =
  envelope
    & #contentType
    .~ ApplicationJson
    & #payloadBytes
    .~ Lazy.toStrict (Aeson.encode value)

-- | Decode the JSON payload of an 'IntegrationEvent' into a business
-- type.
--
-- Returns 'MalformedPayload' if the bytes are not valid JSON,
-- 'DecodeFailed' if the JSON does not satisfy the target's 'FromJSON'
-- instance, and 'UnsupportedContentType' if the envelope's 'contentType'
-- is not 'ApplicationJson'.
decodeJsonIntegrationEvent ::
  (FromJSON a) =>
  IntegrationEvent ->
  Either IntegrationEventError a
decodeJsonIntegrationEvent event = do
  case event ^. #contentType of
    ApplicationJson -> pure ()
    OtherContentType raw -> Left (UnsupportedContentType raw)
  parsed <- case Aeson.eitherDecodeStrict (event ^. #payloadBytes) of
    Left err -> Left (MalformedPayload (Text.pack err))
    Right value -> Right (value :: Value)
  case parseEither parseJSON parsed of
    Left err -> Left (DecodeFailed (Text.pack err))
    Right value -> Right value

eventIdToUuid :: EventId -> UUID.UUID
eventIdToUuid (EventId uuid) = uuid

globalPositionText :: GlobalPosition -> Text
globalPositionText (GlobalPosition pos) = Text.pack (show pos)