packages feed

eventium-core 0.4.0 → 0.5.0

raw patch · 5 files changed

+397/−2 lines, 5 files

Files

eventium-core.cabal view
@@ -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
src/Eventium.hs view
@@ -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
+ src/Eventium/SchemaEvolution.hs view
@@ -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": <event-json> }+--+-- 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
src/Eventium/Store/Types.hs view
@@ -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)
+ tests/Eventium/SchemaEvolutionSpec.hs view
@@ -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"