Codec.hs

-- | How a job payload is turned into PGMQ JSON and back.
--
-- PGMQ stores message bodies as JSON ('Data.Aeson.Value'). A 'JobCodec' is the
-- adapter between a domain payload type @p@ and that JSON. Two are provided:
--
--   * 'aesonJobCodec' - raw @aeson@ encode/decode. A drop-in match for apps that
--     already use @ToJSON@/@FromJSON@ payloads, so migrating to @keiro-pgmq@ is
--     mechanical.
--   * 'keiroJobCodec' - the versioned upgrade. It bridges keiro's
--     'Keiro.Codec.Codec' by wrapping payloads in a @{ "v": <version>, "t":
--     <event-type>, "data": <payload> }@ envelope and replaying the codec's
--     upcaster chain on decode, giving job payloads the same schema-evolution
--     story event streams have.
--
-- When raising a 'keiroJobCodec' schema version, deploy upgraded workers before
-- upgraded producers. A worker that sees an envelope from a future schema version
-- returns 'JobPayloadFromFuture', and the job runner retries it after the queue's
-- default retry delay so a rolling deploy can complete. Those retries still
-- consume delivery attempts, so size @maxRetries * defaultRetryDelay@ to cover
-- the deploy window.
--
-- Do not switch a non-empty queue directly from 'aesonJobCodec' to
-- 'keiroJobCodec': the wire shape changes from the bare payload to the
-- @{"v","t","data"}@ envelope. Drain the queue first or use a transitional codec that
-- accepts both shapes; otherwise old in-flight messages are malformed and will be
-- dead-lettered.
module Keiro.PGMQ.Codec
  ( JobCodec (..),
    JobDecodeError (..),
    mkJobCodec,
    aesonJobCodec,
    keiroJobCodec,
  )
where

import "aeson" Data.Aeson (FromJSON, ToJSON, Value, object, parseJSON, toJSON, withObject, (.:), (.:?), (.=))
import "aeson" Data.Aeson.Types (parseEither)
import "base" Data.Bifunctor (first)
import "base" Data.List.NonEmpty qualified as NonEmpty
import "keiro-core" Keiro.Codec (Codec, EventType (..))
import "keiro-core" Keiro.Codec qualified as Codec
import "text" Data.Text (Text)
import "text" Data.Text qualified as Text

-- | Why a job payload could not be decoded.
data JobDecodeError
  = -- | The payload is malformed for this codec: poison, dead-letter it.
    JobPayloadMalformed !Text
  | -- | The payload was written by a newer schema version than this worker knows.
    --       Carries payload version, then this codec's version. This is transient
    --       during a rolling deploy: retry it, do not dead-letter it.
    JobPayloadFromFuture !Int !Int
  deriving stock (Eq, Show)

-- | How a job payload is turned into PGMQ JSON and back.
data JobCodec p = JobCodec
  { encodeJob :: p -> Value,
    decodeJob :: Value -> Either JobDecodeError p
  }

-- | Build a 'JobCodec' from an encoder and a legacy text-returning decoder.
mkJobCodec :: (p -> Value) -> (Value -> Either Text p) -> JobCodec p
mkJobCodec encode decode =
  JobCodec
    { encodeJob = encode,
      decodeJob = first JobPayloadMalformed . decode
    }

-- | The default codec: raw @aeson@. Use this when the payload type already has
-- @ToJSON@/@FromJSON@ instances and you do not need versioned schema evolution.
aesonJobCodec :: (ToJSON p, FromJSON p) => JobCodec p
aesonJobCodec = mkJobCodec toJSON (first Text.pack . parseEither parseJSON)

-- | Versioned bridge to keiro's 'Keiro.Codec.Codec'. On encode it produces
-- @{ "v": <schemaVersion>, "t": <event-type>, "data": <encode codec p> }@; on
-- decode it reads the version and event-type tag, runs the codec's upcaster chain
-- up to the current version via 'Keiro.Codec.migrateToCurrent', then runs the
-- codec's current decoder. Legacy envelopes without @"t"@ still decode for
-- single-event codecs. Codec migration faults are surfaced as
-- 'JobPayloadMalformed' (via 'show' of the 'Keiro.Codec.CodecError'), except for a
-- future envelope version, which surfaces as 'JobPayloadFromFuture'.
keiroJobCodec :: Codec p -> JobCodec p
keiroJobCodec codec =
  JobCodec
    { encodeJob = \p ->
        let EventType tag = Codec.eventType codec p
         in object
              [ "v" .= Codec.schemaVersion codec,
                "t" .= tag,
                "data" .= Codec.encode codec p
              ],
      decodeJob = \value -> do
        (version, tag, dataValue) <- parseEnvelope value
        selectedType <- resolveEventType codec tag
        let currentVersion = Codec.schemaVersion codec
        if version > currentVersion
          then Left (JobPayloadFromFuture version currentVersion)
          else pure ()
        migrated <-
          first (JobPayloadMalformed . Text.pack . show) $
            Codec.migrateToCurrent codec selectedType version dataValue
        first JobPayloadMalformed (Codec.decode codec selectedType migrated)
    }

-- | Parse the @{ "v", "t", "data" }@ envelope, surfacing aeson failures as malformed payloads.
parseEnvelope :: Value -> Either JobDecodeError (Int, Maybe EventType, Value)
parseEnvelope =
  first (JobPayloadMalformed . Text.pack) . parseEither parser
  where
    parser = withObject "Keiro.PGMQ.Codec envelope" $ \o -> do
      version <- o .: "v"
      tag <- fmap EventType <$> o .:? "t"
      dataValue <- o .: "data"
      pure (version, tag, dataValue)

resolveEventType :: Codec p -> Maybe EventType -> Either JobDecodeError EventType
resolveEventType _ (Just tag) = Right tag
resolveEventType codec Nothing =
  case NonEmpty.toList (Codec.eventTypes codec) of
    [tag] -> Right tag
    tags ->
      Left
        . JobPayloadMalformed
        $ "missing event-type tag in multi-event keiro job envelope; expected one of "
          <> Text.intercalate ", " (fmap eventTypeText tags)
  where
    eventTypeText (EventType tag) = tag