diff --git a/eventium-core.cabal b/eventium-core.cabal --- a/eventium-core.cabal +++ b/eventium-core.cabal @@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: eventium-core -version: 0.4.0 +version: 0.5.0 synopsis: Core module for eventium description: Eventium-core provides the core abstractions and utilities for building event sourcing systems in Haskell. It includes event store interfaces, command handlers, projections, event handlers, event publishers, @@ -48,6 +48,7 @@ Eventium.ProjectionCache.Cache Eventium.ProjectionCache.Types Eventium.ReadModel + Eventium.SchemaEvolution Eventium.Store.Class Eventium.Store.Queries Eventium.Store.Types @@ -91,6 +92,7 @@ Eventium.EventHandlerSpec Eventium.JsonSpec Eventium.ProjectionSpec + Eventium.SchemaEvolutionSpec Eventium.SumTypeSpec Eventium Eventium.Codec @@ -106,6 +108,7 @@ Eventium.ProjectionCache.Cache Eventium.ProjectionCache.Types Eventium.ReadModel + Eventium.SchemaEvolution Eventium.Store.Class Eventium.Store.Queries Eventium.Store.Types diff --git a/src/Eventium.hs b/src/Eventium.hs --- a/src/Eventium.hs +++ b/src/Eventium.hs @@ -13,6 +13,7 @@ import Eventium.Projection as X import Eventium.ProjectionCache.Cache as X import Eventium.ReadModel as X +import Eventium.SchemaEvolution as X import Eventium.Store.Class as X import Eventium.TypeEmbedding as X import Eventium.UUID as X diff --git a/src/Eventium/SchemaEvolution.hs b/src/Eventium/SchemaEvolution.hs new file mode 100644 --- /dev/null +++ b/src/Eventium/SchemaEvolution.hs @@ -0,0 +1,212 @@ +{-# LANGUAGE OverloadedStrings #-} + +-- | Event schema evolution via upcast-on-read against an immutable log. +-- +-- Stored events are never mutated. Instead, when an event type's shape changes +-- across a release, older stored events are normalized to the current shape +-- /on read/ by running a chain of pure, single-hop upcasters. This preserves the +-- append-only event log (the source of truth), makes @pg_dump@/restore +-- version-independent, and supports version-skipping (restoring an +-- N-releases-old backup) simply by running more hops. +-- +-- == Envelope +-- +-- Serialized payloads are wrapped in a versioned envelope: +-- +-- > { "schemaVersion": N, "payload": } +-- +-- Events written before this envelope existed lack the top-level +-- @schemaVersion@ key and are defined to be @schemaVersion = 1@ by convention; +-- their whole stored value is the payload. Existing rows are never rewritten. +-- +-- == Registry +-- +-- A 'SchemaRegistry' maps an event-type name to an ordered list of single-hop +-- upcasters @[v1->v2, v2->v3, ...]@. The current version of a type is +-- @1 + number of hops@, so the two can never drift. A type with no registered +-- upcasters is at version 1 and decodes with no transformation. +-- +-- == Codec +-- +-- 'upcastingValueCodec' is a drop-in 'Codec' between a domain event type and +-- its stored 'Value'. It is parameterized by an @eventTypeOf@ function that +-- reads the type name from a payload (the caller's tagging convention), which +-- keeps this module agnostic to how any particular app tags its events. Because +-- it is a plain 'Codec', it serves both synchronous (in-transaction) and +-- asynchronous (polling) consumers, and every event type and backend, with no +-- reader-level plumbing. +module Eventium.SchemaEvolution + ( -- * Versions and upcasters + EventTypeName, + SchemaVersion, + Upcaster, + + -- * Registry + SchemaRegistry, + emptyRegistry, + registerUpcasters, + currentVersion, + + -- * Upcaster combinators + atKey, + addFieldIfAbsent, + renameField, + removeField, + + -- * Envelope + wrapEnvelope, + unwrapEnvelope, + + -- * Codec + upcastingValueCodec, + ) +where + +import Data.Aeson + ( FromJSON, + Result (..), + ToJSON, + Value (..), + fromJSON, + object, + toJSON, + (.=), + ) +import Data.Aeson.Key (Key) +import qualified Data.Aeson.KeyMap as KeyMap +import Data.Map.Strict (Map) +import qualified Data.Map.Strict as Map +import Eventium.Codec (Codec (..)) +import Eventium.Store.Types (EventTypeName) + +-- | The schema version of an event type. Version numbering starts at 1. +type SchemaVersion = Int + +-- | A pure, single-hop transform from one schema version's JSON to the next. +type Upcaster = Value -> Value + +-- ----------------------------------------------------------------------------- +-- Upcaster combinators +-- ----------------------------------------------------------------------------- + +-- These build the common field-level transforms without hand-rolled 'KeyMap' +-- manipulation. Each is layout-agnostic and a no-op on non-object values, so +-- they compose freely (e.g. focus into a nested object with 'atKey', then edit +-- a field there). + +-- | Apply an upcaster to the value at @key@ within an object, leaving the rest +-- untouched. A no-op if @key@ is absent, or if the value is not an object. +-- Use this to focus a field-level transform into a nested object (e.g. an +-- Aeson tagged constructor's @contents@). +atKey :: Key -> Upcaster -> Upcaster +atKey key f (Object o) = case KeyMap.lookup key o of + Just v -> Object (KeyMap.insert key (f v) o) + Nothing -> Object o +atKey _ _ v = v + +-- | Insert @key = value@ only if @key@ is not already present. Idempotent and +-- never clobbers an existing value — the safe default for "a new field was +-- added with default @value@". +addFieldIfAbsent :: Key -> Value -> Upcaster +addFieldIfAbsent key value (Object o) = Object (KeyMap.insertWith (\_new old -> old) key value o) +addFieldIfAbsent _ _ v = v + +-- | Move the value at @from@ to @to@ (overwriting @to@ if present), dropping +-- @from@. A no-op if @from@ is absent. +renameField :: Key -> Key -> Upcaster +renameField from to (Object o) = case KeyMap.lookup from o of + Just v -> Object (KeyMap.insert to v (KeyMap.delete from o)) + Nothing -> Object o +renameField _ _ v = v + +-- | Remove @key@ from an object. A no-op if @key@ is absent. +removeField :: Key -> Upcaster +removeField key (Object o) = Object (KeyMap.delete key o) +removeField _ v = v + +-- | A registry of upcaster chains, keyed by event-type name. The list for each +-- type is ordered @[v1->v2, v2->v3, ...]@. +newtype SchemaRegistry = SchemaRegistry (Map EventTypeName [Upcaster]) + +-- | A registry with no upcasters: every event type is at version 1. +emptyRegistry :: SchemaRegistry +emptyRegistry = SchemaRegistry Map.empty + +-- | Register the ordered upcaster chain for an event type, replacing any +-- existing entry. The list must be ordered @[v1->v2, v2->v3, ...]@; its length +-- determines the type's current version. +registerUpcasters :: EventTypeName -> [Upcaster] -> SchemaRegistry -> SchemaRegistry +registerUpcasters eventType hops (SchemaRegistry m) = + SchemaRegistry (Map.insert eventType hops m) + +-- | The current schema version of an event type: @1 + number of hops@. An +-- unregistered type is at version 1. +currentVersion :: SchemaRegistry -> EventTypeName -> SchemaVersion +currentVersion reg eventType = 1 + length (hopsOf reg eventType) + +-- | All registered hops for an event type (@[]@ if unregistered). +hopsOf :: SchemaRegistry -> EventTypeName -> [Upcaster] +hopsOf (SchemaRegistry m) eventType = Map.findWithDefault [] eventType m + +-- | The hops needed to bring an event stored at @version@ up to current: drop +-- the @version - 1@ hops already baked into the stored shape. +hopsFrom :: SchemaRegistry -> EventTypeName -> SchemaVersion -> [Upcaster] +hopsFrom reg eventType version = drop (max 0 (version - 1)) (hopsOf reg eventType) + +envelopeVersionKey :: KeyMap.Key +envelopeVersionKey = "schemaVersion" + +envelopePayloadKey :: KeyMap.Key +envelopePayloadKey = "payload" + +-- | Wrap a payload in a versioned envelope. +wrapEnvelope :: SchemaVersion -> Value -> Value +wrapEnvelope version payload = + object [envelopeVersionKey .= version, envelopePayloadKey .= payload] + +-- | Peel an envelope into @(version, payload)@. A value that is not a +-- well-formed envelope — anything without both a numeric top-level +-- @schemaVersion@ and a @payload@ sibling — is legacy pre-envelope data and is +-- read as @(1, wholeValue)@. +unwrapEnvelope :: Value -> (SchemaVersion, Value) +unwrapEnvelope (Object o) + | Just version <- KeyMap.lookup envelopeVersionKey o >>= valueToInt, + Just payload <- KeyMap.lookup envelopePayloadKey o = + (version, payload) +unwrapEnvelope v = (1, v) + +valueToInt :: Value -> Maybe SchemaVersion +valueToInt v = case fromJSON v of + Success n -> Just n + Error _ -> Nothing + +-- | A drop-in 'Codec' between a domain event type and its stored 'Value', +-- applying schema evolution transparently. +-- +-- On 'encode', the current-version envelope is written. On 'decode', the +-- envelope is peeled (or defaulted to version 1 for legacy data), the event +-- type is resolved via @eventTypeOf@, the remaining upcaster hops are run, and +-- the result is decoded into the current type. Decoding yields 'Nothing' if the +-- upcasted value still does not match the current type. +upcastingValueCodec :: + (ToJSON a, FromJSON a) => + -- | Read the event-type name from a payload (the app's tagging convention). + (Value -> Maybe EventTypeName) -> + SchemaRegistry -> + Codec a Value +upcastingValueCodec eventTypeOf reg = + Codec + { encode = \a -> + let payload = toJSON a + version = maybe 1 (currentVersion reg) (eventTypeOf payload) + in wrapEnvelope version payload, + decode = \stored -> + let (version, payload) = unwrapEnvelope stored + upcasted = case eventTypeOf payload of + Just eventType -> foldl (\acc hop -> hop acc) payload (hopsFrom reg eventType version) + Nothing -> payload + in resultToMaybe (fromJSON upcasted) + } + where + resultToMaybe (Success a) = Just a + resultToMaybe (Error _) = Nothing diff --git a/src/Eventium/Store/Types.hs b/src/Eventium/Store/Types.hs --- a/src/Eventium/Store/Types.hs +++ b/src/Eventium/Store/Types.hs @@ -13,6 +13,7 @@ GlobalStreamEvent, -- * Event metadata + EventTypeName, EventMetadata (..), emptyMetadata, MetadataEnricher, @@ -45,9 +46,14 @@ import Web.HttpApiData import Web.PathPieces +-- | A name identifying an event type — e.g. the type name recorded in +-- 'EventMetadata', or the discriminator tag an app uses to key its schema +-- evolution registry (see "Eventium.SchemaEvolution"). +type EventTypeName = Text + -- | Metadata carried alongside every stored event. data EventMetadata = EventMetadata - { eventType :: !Text, + { eventType :: !EventTypeName, correlationId :: !(Maybe UUID), causationId :: !(Maybe UUID), createdAt :: !(Maybe UTCTime) diff --git a/tests/Eventium/SchemaEvolutionSpec.hs b/tests/Eventium/SchemaEvolutionSpec.hs new file mode 100644 --- /dev/null +++ b/tests/Eventium/SchemaEvolutionSpec.hs @@ -0,0 +1,173 @@ +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE OverloadedStrings #-} + +module Eventium.SchemaEvolutionSpec (spec) where + +import Data.Aeson + ( FromJSON, + ToJSON, + Value (..), + object, + toJSON, + (.=), + ) +import qualified Data.Aeson.KeyMap as KeyMap +import Data.Text (Text) +import Eventium.Codec (Codec (..)) +import Eventium.SchemaEvolution +import GHC.Generics (Generic) +import Test.Hspec + +-- A v2 shape that adds a boolean field ('enabled') absent from v1 events — +-- the canonical "required field added" migration. +data Gizmo = Gizmo + { kind :: Text, + size :: Int, + enabled :: Bool + } + deriving (Show, Eq, Generic) + +instance ToJSON Gizmo + +instance FromJSON Gizmo + +-- A type with a two-hop chain (v1 -> v2 -> v3), for version-skipping. +data Chain = Chain + { kind :: Text, + step :: Int + } + deriving (Show, Eq, Generic) + +instance ToJSON Chain + +instance FromJSON Chain + +-- A type with no registered upcasters, for the identity path. +data Plain = Plain + { kind :: Text, + note :: Text + } + deriving (Show, Eq, Generic) + +instance ToJSON Plain + +instance FromJSON Plain + +-- | Read the event-type name from a payload's @kind@ field. +kindOf :: Value -> Maybe EventTypeName +kindOf (Object o) = case KeyMap.lookup "kind" o of + Just (String t) -> Just t + _ -> Nothing +kindOf _ = Nothing + +-- v1 -> v2 for Gizmo: default the new field to False. Insert-if-absent so the +-- hop is idempotent and never clobbers an existing value. +injectEnabled :: Value -> Value +injectEnabled (Object o) = Object (KeyMap.insertWith (\_new old -> old) "enabled" (Bool False) o) +injectEnabled v = v + +setStep :: Int -> Value -> Value +setStep n (Object o) = Object (KeyMap.insert "step" (toJSON n) o) +setStep _ v = v + +registry :: SchemaRegistry +registry = + registerUpcasters "Gizmo" [injectEnabled] $ + registerUpcasters "Chain" [setStep 2, setStep 3] emptyRegistry + +codec :: Codec Gizmo Value +codec = upcastingValueCodec kindOf registry + +chainCodec :: Codec Chain Value +chainCodec = upcastingValueCodec kindOf registry + +plainCodec :: Codec Plain Value +plainCodec = upcastingValueCodec kindOf registry + +spec :: Spec +spec = do + describe "SchemaRegistry" $ do + it "current version is 1 + number of hops" $ do + currentVersion registry "Gizmo" `shouldBe` 2 + currentVersion registry "Chain" `shouldBe` 3 + + it "defaults unregistered types to version 1" $ + currentVersion registry "Unknown" `shouldBe` 1 + + describe "envelope" $ do + it "wraps a payload with its schema version" $ + wrapEnvelope 2 (object ["kind" .= ("Gizmo" :: Text)]) + `shouldBe` object ["schemaVersion" .= (2 :: Int), "payload" .= object ["kind" .= ("Gizmo" :: Text)]] + + it "unwraps an envelope to (version, payload)" $ + unwrapEnvelope (object ["schemaVersion" .= (2 :: Int), "payload" .= object ["kind" .= ("Gizmo" :: Text)]]) + `shouldBe` (2, object ["kind" .= ("Gizmo" :: Text)]) + + it "treats legacy bare JSON as version 1" $ + unwrapEnvelope (object ["kind" .= ("Gizmo" :: Text)]) + `shouldBe` (1, object ["kind" .= ("Gizmo" :: Text)]) + + it "is not fooled by a payload that merely has a schemaVersion field but no payload sibling" $ + unwrapEnvelope (object ["schemaVersion" .= (7 :: Int), "kind" .= ("Gizmo" :: Text)]) + `shouldBe` (1, object ["schemaVersion" .= (7 :: Int), "kind" .= ("Gizmo" :: Text)]) + + describe "upcastingValueCodec encode" $ do + it "wraps writes in the current-version envelope" $ + codec.encode (Gizmo "Gizmo" 100 True) + `shouldBe` object + [ "schemaVersion" .= (2 :: Int), + "payload" .= object ["kind" .= ("Gizmo" :: Text), "size" .= (100 :: Int), "enabled" .= True] + ] + + describe "upcastingValueCodec decode" $ do + it "decodes a legacy v1 event by running the chain (added field defaulted)" $ + codec.decode (object ["kind" .= ("Gizmo" :: Text), "size" .= (100 :: Int)]) + `shouldBe` Just (Gizmo "Gizmo" 100 False) + + it "round-trips a current-version event, preserving the field" $ + codec.decode (codec.encode (Gizmo "Gizmo" 250 True)) + `shouldBe` Just (Gizmo "Gizmo" 250 True) + + it "runs every hop for a legacy v1 event (version-skipping)" $ + chainCodec.decode (object ["kind" .= ("Chain" :: Text), "step" .= (1 :: Int)]) + `shouldBe` Just (Chain "Chain" 3) + + it "runs only the remaining hops for an event stored at an intermediate version" $ + chainCodec.decode (object ["schemaVersion" .= (2 :: Int), "payload" .= object ["kind" .= ("Chain" :: Text), "step" .= (99 :: Int)]]) + `shouldBe` Just (Chain "Chain" 3) + + it "is identity for an unregistered event type" $ + plainCodec.decode (object ["kind" .= ("Plain" :: Text), "note" .= ("hi" :: Text)]) + `shouldBe` Just (Plain "Plain" "hi") + + describe "upcaster combinators" $ do + it "addFieldIfAbsent inserts a default when the field is missing" $ + addFieldIfAbsent "enabled" (Bool False) (object ["kind" .= ("Gizmo" :: Text)]) + `shouldBe` object ["kind" .= ("Gizmo" :: Text), "enabled" .= False] + + it "addFieldIfAbsent leaves an existing field untouched" $ + addFieldIfAbsent "enabled" (Bool False) (object ["enabled" .= True]) + `shouldBe` object ["enabled" .= True] + + it "removeField drops a field" $ + removeField "legacy" (object ["kind" .= ("Gizmo" :: Text), "legacy" .= (1 :: Int)]) + `shouldBe` object ["kind" .= ("Gizmo" :: Text)] + + it "renameField moves a value to the new key" $ + renameField "amount" "size" (object ["amount" .= (10 :: Int)]) + `shouldBe` object ["size" .= (10 :: Int)] + + it "renameField is a no-op when the source key is absent" $ + renameField "amount" "size" (object ["size" .= (10 :: Int)]) + `shouldBe` object ["size" .= (10 :: Int)] + + it "atKey focuses an upcaster into a nested object" $ + atKey "contents" (addFieldIfAbsent "enabled" (Bool False)) (object ["tag" .= ("G" :: Text), "contents" .= object ["size" .= (1 :: Int)]]) + `shouldBe` object ["tag" .= ("G" :: Text), "contents" .= object ["size" .= (1 :: Int), "enabled" .= False]] + + it "atKey is a no-op when the focused key is absent" $ + atKey "contents" (addFieldIfAbsent "enabled" (Bool False)) (object ["tag" .= ("G" :: Text)]) + `shouldBe` object ["tag" .= ("G" :: Text)] + + it "combinators pass non-object values through unchanged" $ + addFieldIfAbsent "x" (Bool True) (String "scalar") `shouldBe` String "scalar"