packages feed

keiro-core (empty) → 0.2.0.0

raw patch · 10 files changed

+1401/−0 lines, 10 filesdep +aesondep +aeson-casingdep +base

Dependencies added: aeson, aeson-casing, base, bytestring, deepseq, generic-lens, keiki, kiroku-store, lens, scientific, text, time, uuid

Files

+ CHANGELOG.md view
@@ -0,0 +1,68 @@+# Changelog++All notable changes to `keiro-core` are recorded here. The format follows+[Keep a Changelog](https://keepachangelog.com/), and the package follows the+[Haskell Package Versioning Policy](https://pvp.haskell.org/).++## [Unreleased]++_No unreleased changes._++## 0.2.0.0 — 2026-07-13++### Breaking Changes++- `keiro-core` now requires post-MP-16 Keiki 0.2 (`keiki >=0.2 && <0.3`) and+  Kiroku 0.3 (`kiroku-store >=0.3 && <0.4`). Stream validation runs Keiki's new+  head-recoverability, inversion-ambiguity, unguarded-input-read, and+  state-changing-silent-edge checks; any warning enabled by the selected+  `ValidationOptions` makes `mkEventStream` reject the stream at startup.+- `validateEventStreamWith` and `mkEventStreamWith` now force-enable Keiki's+  `checkHeadRecoverability` and `checkStateChangingEpsilon`. Caller-supplied+  options may only strengthen validation at Keiro's durable boundary; use the+  explicitly unsafe `mkEventStreamUnchecked` only for tests and emergency+  forensics, never for production streams.+- `mkEventStream` now rejects snapshot codecs that cannot encode their initial+  state and register file. Snapshot-enabled streams built from `emptyRegFile`+  must initialize every slot before validation; the labelled `uninit: <slot>`+  `ErrorCall` is reported as an `EventStreamWarning` instead of escaping at the+  first snapshot write.++### New Features++- New module `Keiro.Schema`, exporting `keiroSchema :: Text` (`"keiro"`). This+  is the single source of truth for the dedicated PostgreSQL schema that owns+  Keiro's framework tables: `keiro-migrations` creates them schema-qualified and+  the `keiro` runtime queries resolve against the same name.+- Added `mkEventStreamUnchecked`, which wraps an `EventStream` with no Keiki or+  Keiro checks at all. It exists for tests and emergency forensics; a stream+  admitted through it can silently lose state changes and fail hydration.++### Other Changes++- Keiki 0.2 and Kiroku 0.3 now resolve from Hackage; their obsolete Git package+  overrides and the local Cabal overlay are no longer needed.+- Added a `deepseq` dependency: `validateEventStream` forces the configured+  `stateCodec` over the initial state and registers, observing only `ErrorCall`+  so that any other exception stays programmer-visible.++## 0.1.0.0 — 2026-07-05++Initial Hackage release.++### Breaking Changes++- Finalized the pre-release stream naming, codec, and event-stream contracts,+  including safer stream category construction and validated event-stream+  boundaries.++### New Features++- Added shared `Keiro.Codec`, `Keiro.Stream`, `Keiro.EventStream`,+  `Keiro.EventStream.Validate`, `Keiro.Integration.Event`,+  `Keiro.Snapshot.Policy`, and `Keiro.Prelude` modules.+- Added replay-safety validation helpers built on keiki's transducer validator.++### Other Changes++- Added Haddock coverage and migration documentation for validated event streams.
+ keiro-core.cabal view
@@ -0,0 +1,65 @@+cabal-version:   3.0+name:            keiro-core+version:         0.2.0.0+synopsis:        Core contracts for Keiro packages+description:+  Stable stream, codec, event-stream, and integration-event contracts+  shared by Keiro packages.++license:         BSD-3-Clause+author:          Nadeem Bitar+maintainer:      nadeem@gmail.com+copyright:       2026 Nadeem Bitar+category:        Control+homepage:        https://github.com/shinzui/keiro#readme+bug-reports:     https://github.com/shinzui/keiro/issues+build-type:      Simple+tested-with:     GHC >=9.12 && <9.13+extra-doc-files: CHANGELOG.md++source-repository head+  type:     git+  location: https://github.com/shinzui/keiro.git++common warnings+  ghc-options:+    -Wall -Wcompat -Widentities -Wincomplete-record-updates+    -Wincomplete-uni-patterns -Wpartial-fields -Wredundant-constraints++common shared+  default-language:   GHC2024+  default-extensions:+    DeriveAnyClass+    DuplicateRecordFields+    MultilineStrings+    OverloadedLabels+    OverloadedStrings+    PackageImports++library+  import:          warnings, shared+  exposed-modules:+    Keiro.Codec+    Keiro.EventStream+    Keiro.EventStream.Validate+    Keiro.Integration.Event+    Keiro.Prelude+    Keiro.Schema+    Keiro.Snapshot.Policy+    Keiro.Stream++  hs-source-dirs:  src+  build-depends:+    , aeson         >=2.2+    , aeson-casing  >=0.2+    , base          >=4.21 && <5+    , bytestring    >=0.11+    , deepseq       >=1.5+    , generic-lens  >=2.2+    , keiki         >=0.2  && <0.3+    , kiroku-store  >=0.3  && <0.4+    , lens          >=5.2+    , scientific    >=0.3+    , text          >=2.1+    , time          >=1.12+    , uuid          >=1.3
+ src/Keiro/Codec.hs view
@@ -0,0 +1,312 @@+{- | Versioned encode/decode contract between a domain event type and its+stored JSON payload.++A 'Codec' is the single place a stream declares how its events are named,+serialized, and migrated. It pairs a current 'schemaVersion' with a chain+of 'Upcaster's so that payloads written under older versions are+transparently brought up to the current shape on read. Producers call+'encodeForAppend' (which stamps the schema version into event metadata);+consumers call 'decodeRecorded', which reads that stamp back, replays the+upcaster chain via 'migrateToCurrent', and then runs the current 'decode'.++Encoding can fail only with a misconfigured codec ('InvalidSchemaVersion')+or an event whose type tag is not in 'eventTypes' ('UnknownEventType').+Decoding additionally surfaces migration faults: a missing rung in the+upcaster chain ('GapInUpcasterChain'), an upcaster that rejected its input+('UpcasterError'), a stored future version ('VersionAhead'), an out-of-range+stored version ('UnknownVersion'), or a payload the current decoder rejects+('DecodeFailed').+-}+module Keiro.Codec (+    -- * Codec+    Codec (..),+    Upcaster,+    EventType (..),+    CodecError (..),+    CodecConfigError (..),+    mkCodec,++    -- * Encoding+    encodeForAppend,+    encodeForAppendWithMetadata,++    -- * Decoding+    decodeRecorded,+    decodeRaw,+    migrateToCurrent,++    -- * Metadata helpers+    extractSchemaVersion,+    metadataFor,+)+where++import Data.Aeson (Value (..))+import Data.Aeson.Key qualified as Key+import Data.Aeson.KeyMap qualified as KeyMap+import Data.List qualified as List+import Data.List.NonEmpty qualified as NonEmpty+import Data.Maybe qualified as Maybe+import Data.Scientific qualified as Scientific+import Keiro.Prelude+import Kiroku.Store.Types (+    EventData (..),+    EventType (..),+    RecordedEvent (..),+ )+import Prelude qualified++{- | One rung of an upcaster chain: the source schema version it upgrades+/from/, paired with a pure migration that rewrites a version-@n@ payload+into the version-@(n+1)@ shape. The stored event-type tag is supplied so+multi-event codecs can migrate by the authoritative wire tag rather than a+payload discriminator. A migration may reject malformed input with a 'Left'.+-}+type Upcaster = (Int, EventType -> Value -> Either Text Value)++{- | Everything a stream needs to serialize and deserialize its events.++* 'eventTypes' - the complete set of event-type tags this codec owns.+  Encoding and decoding reject any tag outside this set, so it doubles as+  the stream's event-type allow-list.+* 'eventType' - projects a domain value to its wire tag (must land in+  'eventTypes').+* 'schemaVersion' - the current payload version; must be @>= 1@. Stamped+  into event metadata on append and used as the migration target on read.+* 'encode' / 'decode' - the current-version JSON serialization. 'decode'+  receives the stored event-type tag and only sees payloads already migrated+  to 'schemaVersion'.+* 'upcasters' - migrations keyed by source version. To read a+  version-@n@ payload the codec applies the @n@, @n+1@, ... rungs in+  sequence until it reaches 'schemaVersion'; a missing rung is a+  'GapInUpcasterChain' or 'IncompleteUpcasterChain'.+-}+data Codec e = Codec+    { eventTypes :: !(NonEmpty EventType)+    , eventType :: !(e -> EventType)+    , schemaVersion :: !Int+    , encode :: !(e -> Value)+    , decode :: !(EventType -> Value -> Either Text e)+    , upcasters :: ![Upcaster]+    }+    deriving stock (Generic)++-- | Why an encode or decode could not be completed.+data CodecError+    = {- | The event-type tag is not one of the codec's 'eventTypes' (carries+      the offending tag and the allowed set).+      -}+      UnknownEventType !EventType ![EventType]+    | -- | The codec's 'schemaVersion' is not @>= 1@.+      InvalidSchemaVersion !Int+    | -- | A stored payload declared a version below @1@.+      UnknownVersion !Int+    | -- | A stored payload was written by a newer codec version.+      VersionAhead !Int !Int+    | {- | An upcaster rejected its input; carries the source version and the+      migration's error message.+      -}+      UpcasterError !Int !Text+    | -- | The current 'decode' rejected an already-migrated payload.+      DecodeFailed !Text+    | {- | The upcaster chain is missing a rung: migration reached version+      @n@ but the next available upcaster starts at a later version.+      -}+      GapInUpcasterChain !Int !Int+    | -- | The upcaster chain ended before reaching the codec's target version.+      IncompleteUpcasterChain !Int !Int+    | -- | A present schema-version stamp was malformed.+      MalformedSchemaVersionStamp !Value+    | -- | Caller-supplied metadata was not a JSON object.+      NonObjectCallerMetadata !Value+    deriving stock (Generic, Eq, Show)++-- | Why a raw 'Codec' record did not satisfy the construction invariants.+data CodecConfigError+    = CodecSchemaVersionInvalid !Int+    | CodecDuplicateEventTypes ![EventType]+    | CodecDuplicateUpcasterSources ![Int]+    | CodecUpcasterSourceOutOfRange !Int !Int+    | CodecUpcasterChainIncomplete ![Int] !Int+    deriving stock (Generic, Eq, Show)++schemaVersionKey :: Key.Key+schemaVersionKey = "schemaVersion"++{- | Validate a raw 'Codec' record before exposing it to runtime use.++The raw 'Codec' constructor remains exported as an escape hatch for tests and+low-level callers, but production definitions should prefer 'mkCodec' so+misconfigured schema versions and upcaster chains fail at construction time.+-}+mkCodec :: Codec e -> Either CodecConfigError (Codec e)+mkCodec codec+    | codec ^. #schemaVersion < 1 =+        Left (CodecSchemaVersionInvalid (codec ^. #schemaVersion))+    | Prelude.not (Prelude.null duplicateTypes) =+        Left (CodecDuplicateEventTypes duplicateTypes)+    | Prelude.not (Prelude.null duplicateSources) =+        Left (CodecDuplicateUpcasterSources duplicateSources)+    | Just source <- outOfRangeSource =+        Left (CodecUpcasterSourceOutOfRange source (codec ^. #schemaVersion))+    | Prelude.not (Prelude.null missingSources) =+        Left (CodecUpcasterChainIncomplete missingSources (codec ^. #schemaVersion))+    | otherwise =+        Right codec+  where+    sources = List.sort [source | (source, _) <- codec ^. #upcasters]+    expectedSources = [1 .. (codec ^. #schemaVersion) Prelude.- 1]+    duplicateTypes = duplicates (NonEmpty.toList (codec ^. #eventTypes))+    duplicateSources = duplicates sources+    outOfRangeSource =+        List.find+            (\source -> source < 1 Prelude.|| source >= codec ^. #schemaVersion)+            sources+    missingSources = expectedSources List.\\ sources++duplicates :: (Ord a) => [a] -> [a]+duplicates =+    Maybe.mapMaybe duplicateHead+        . List.group+        . List.sort+  where+    duplicateHead (x : _ : _) = Just x+    duplicateHead _ = Nothing++{- | Encode a domain event into 'EventData' ready for append, stamping the+codec's 'schemaVersion' into fresh metadata. Equivalent to+'encodeForAppendWithMetadata' with no caller metadata.+-}+encodeForAppend :: Codec e -> e -> Either CodecError EventData+encodeForAppend codec value = encodeForAppendWithMetadata codec Nothing value++{- | Encode a domain event into 'EventData', merging the codec's+'schemaVersion' into the supplied metadata object (if any).++Fails with 'InvalidSchemaVersion' when the codec's version is not @>= 1@,+'UnknownEventType' when 'eventType' produces a tag outside 'eventTypes', or+'NonObjectCallerMetadata' when the caller supplies non-object metadata. The+schema-version key always wins over any clashing key in the caller's metadata+so the stamp on disk is authoritative.+-}+encodeForAppendWithMetadata :: Codec e -> Maybe Value -> e -> Either CodecError EventData+encodeForAppendWithMetadata codec metadata value = do+    unless (codec ^. #schemaVersion > 0)+        $ Left (InvalidSchemaVersion (codec ^. #schemaVersion))+    let selectedType = codec ^. #eventType $ value+    unless (selectedType `List.elem` NonEmpty.toList (codec ^. #eventTypes))+        $ Left (UnknownEventType selectedType (NonEmpty.toList (codec ^. #eventTypes)))+    stampedMetadata <- metadataFor (codec ^. #schemaVersion) metadata+    pure+        EventData+            { eventId = Nothing+            , eventType = selectedType+            , payload = codec ^. #encode $ value+            , metadata = Just stampedMetadata+            , causationId = Nothing+            , correlationId = Nothing+            }++{- | Build the metadata object stored alongside an event, inserting the schema+version under the @schemaVersion@ key. A non-object @existing@ value is rejected+with 'NonObjectCallerMetadata'.+-}+metadataFor :: Int -> Maybe Value -> Either CodecError Value+metadataFor version existing =+    case existing of+        Just object@(Object _) ->+            Right (insertVersion object)+        Nothing ->+            Right (insertVersion (Object KeyMap.empty))+        Just value ->+            Left (NonObjectCallerMetadata value)+  where+    insertVersion (Object object) =+        Object+            $ object+            & KeyMap.insert schemaVersionKey (Number (Prelude.fromIntegral version))+    insertVersion value = value++{- | Decode a stored 'RecordedEvent' into a domain value. Reads the schema+version stamped in the event's metadata (defaulting to @1@ when absent),+migrates the payload up to the codec's current version, then runs 'decode'.+Rejects events whose type tag is not in 'eventTypes'.+-}+decodeRecorded :: Codec e -> RecordedEvent -> Either CodecError e+decodeRecorded codec recorded = do+    unless (isKnownEventType (recorded ^. #eventType) codec)+        $ Left (UnknownEventType (recorded ^. #eventType) (NonEmpty.toList (codec ^. #eventTypes)))+    version <- extractSchemaVersion recorded+    decodeRaw codec (recorded ^. #eventType) version (recorded ^. #payload)++{- | Decode a raw JSON payload whose source schema version and event type are+already known. Migrates from @version@ to the codec's current version and runs+'decode'. Useful when the version comes from somewhere other than+recorded-event metadata (e.g. a snapshot, job envelope, or replay tool).+-}+decodeRaw :: Codec e -> EventType -> Int -> Value -> Either CodecError e+decodeRaw codec selectedType version payload = do+    unless (isKnownEventType selectedType codec)+        $ Left (UnknownEventType selectedType (NonEmpty.toList (codec ^. #eventTypes)))+    migrated <- migrateToCurrent codec selectedType version payload+    case (codec ^. #decode) selectedType migrated of+        Right value -> Right value+        Left message -> Left (DecodeFailed message)++{- | Replay the upcaster chain to bring a payload from @sourceVersion@ up to+the codec's current 'schemaVersion'.++A payload exactly at the current version is returned unchanged. A payload from a+future version fails with 'VersionAhead'. Each step looks up the upcaster keyed+by the current version and applies it; the walk stops with+'GapInUpcasterChain' if a later rung exists but an earlier rung is missing,+'IncompleteUpcasterChain' if the chain ends early, 'UpcasterError' if a+migration rejects its input, or 'UnknownVersion' for a source version below @1@.+-}+migrateToCurrent :: Codec e -> EventType -> Int -> Value -> Either CodecError Value+migrateToCurrent codec selectedType sourceVersion payload+    | sourceVersion == codec ^. #schemaVersion = Right payload+    | sourceVersion > codec ^. #schemaVersion = Left (VersionAhead sourceVersion (codec ^. #schemaVersion))+    | sourceVersion < 1 = Left (UnknownVersion sourceVersion)+    | otherwise = go sourceVersion payload+  where+    go version current+        | version >= codec ^. #schemaVersion = Right current+        | otherwise =+            case Prelude.lookup version (codec ^. #upcasters) of+                Nothing ->+                    case nextChainStart version of+                        Just nextVersion -> Left (GapInUpcasterChain version nextVersion)+                        Nothing -> Left (IncompleteUpcasterChain version (codec ^. #schemaVersion))+                Just upcast ->+                    case upcast selectedType current of+                        Left message -> Left (UpcasterError version message)+                        Right next -> go (version Prelude.+ 1) next++    nextChainStart version =+        case [next | (next, _) <- codec ^. #upcasters, next > version] of+            next : _ -> Just next+            [] -> Nothing++{- | Read the schema version stamped into a recorded event's metadata. Defaults+to @1@ when metadata is absent or the object lacks the @schemaVersion@ key.+Present-but-malformed metadata fails with 'MalformedSchemaVersionStamp'.+-}+extractSchemaVersion :: RecordedEvent -> Either CodecError Int+extractSchemaVersion recorded =+    case recorded ^. #metadata of+        Nothing -> Right 1+        Just (Object object) ->+            case KeyMap.lookup schemaVersionKey object of+                Nothing -> Right 1+                Just stamp@(Number number) ->+                    maybe+                        (Left (MalformedSchemaVersionStamp stamp))+                        Right+                        (Scientific.toBoundedInteger number)+                Just stamp -> Left (MalformedSchemaVersionStamp stamp)+        Just metadata -> Left (MalformedSchemaVersionStamp metadata)++isKnownEventType :: EventType -> Codec e -> Bool+isKnownEventType selectedType codec =+    selectedType `List.elem` NonEmpty.toList (codec ^. #eventTypes)
+ src/Keiro/EventStream.hs view
@@ -0,0 +1,108 @@+{- | The complete description of one persistent event stream.++An 'EventStream' marries a pure keiki 'SymTransducer' (the decision logic+of a symbolic-register state machine) with everything keiro needs to run+it against a durable event store: where its initial state and registers+come from, how its emitted events are serialized ('Codec'), which physical+stream name to read and write, when to snapshot, and how to serialize that+snapshot. Command handling ("Keiro.Command") hydrates the machine from+stored events (optionally fast-forwarding from a snapshot), steps it with a+command, encodes the resulting events, and appends them. Public command+runners require a 'Keiro.EventStream.Validate.ValidatedEventStream', obtained+from 'Keiro.EventStream.Validate.mkEventStream' or+'Keiro.EventStream.Validate.mkEventStreamOrThrow', rather than a bare record+literal.++The type parameters thread through from the underlying transducer:++* @phi@ — the guard/predicate alphabet the transducer branches on.+* @rs@ — the register set ('RegFile' @rs@ holds the live values).+* @s@ — the control state.+* @ci@ — the command input the machine consumes.+* @co@ — the event output the machine emits (what 'eventCodec' serializes).+-}+module Keiro.EventStream (+    EventStream (..),+    Terminality (..),+    SnapshotPolicy (..),+    StateCodec (..),+)+where++import Keiki.Core (RegFile, SymTransducer)+import Keiro.Codec (Codec)+import Keiro.Prelude+import Keiro.Stream (Stream)+import Kiroku.Store.Types (StreamName, StreamVersion)++{- | A self-contained, persistable event stream definition.++* 'transducer' — the pure keiki state machine that turns a command into+  emitted events.+* 'initialState' \/ 'initialRegisters' — the machine's starting control+  state and register file, used when hydrating an empty stream.+* 'eventCodec' — serializes and migrates the emitted events (@co@) to and+  from stored payloads.+* 'resolveStreamName' — maps a typed 'Stream' handle to the physical+  'StreamName' read and appended to in the store.+* 'snapshotPolicy' — decides, per append, whether to persist a snapshot of+  the @(state, registers)@ pair.+* 'stateCodec' — how to serialize that snapshot. Set 'snapshotPolicy' and+  'stateCodec' coherently; 'Keiro.EventStream.Validate.mkEventStream' rejects+  a snapshotting policy without a state codec and returns the+  'Keiro.EventStream.Validate.ValidatedEventStream' that command runners accept.+-}+data EventStream phi rs s ci co = EventStream+    { transducer :: !(SymTransducer phi rs s ci co)+    , initialState :: !s+    , initialRegisters :: !(RegFile rs)+    , eventCodec :: !(Codec co)+    , resolveStreamName :: !(Stream (EventStream phi rs s ci co) -> StreamName)+    , snapshotPolicy :: !(SnapshotPolicy (s, RegFile rs))+    , stateCodec :: !(Maybe (StateCodec (s, RegFile rs)))+    }+    deriving stock (Generic)++-- | Whether the append that just happened reached a terminal stream state.+data Terminality = Terminal | NotTerminal+    deriving stock (Eq, Show, Generic)++{- | When to persist a snapshot of a stream's folded state.++* 'Never' — never snapshot; always rehydrate from the full event log.+* 'Every' @n@ — snapshot whenever the stream version is a multiple of @n@+  (a non-positive interval disables snapshotting).+* 'OnTerminal' — snapshot only when the machine has reached a final state.+* 'Custom' — an arbitrary predicate over terminality, folded @state@, and+  the current 'StreamVersion'.++See 'Keiro.Snapshot.Policy.shouldSnapshot' for the evaluation rules.+-}+data SnapshotPolicy state+    = Never+    | Every !Int+    | OnTerminal+    | Custom !(Terminality -> state -> StreamVersion -> Bool)+    deriving stock (Generic)++{- | How to serialize and deserialize a stream's snapshot state.++'stateCodecVersion' and 'shapeHash' together gate snapshot reuse: a stored+snapshot is only loaded when both match the current codec, so a change to+the snapshot encoding or to the shape of the folded state invalidates older+snapshots and forces a clean rehydration from events.++* 'stateCodecVersion' — bumped when the snapshot encoding changes+  incompatibly.+* 'shapeHash' — a digest of the folded-state shape; protects against+  silently loading a snapshot whose structure no longer matches.+* 'encode' \/ 'decode' — the JSON serialization of the @(state,+  registers)@ pair.+-}+data StateCodec state = StateCodec+    { stateCodecVersion :: !Int+    , shapeHash :: !Text+    , encode :: !(state -> Value)+    , decode :: !(Value -> Either Text state)+    }+    deriving stock (Generic)
+ src/Keiro/EventStream/Validate.hs view
@@ -0,0 +1,244 @@+{- | Replay-safety validation for keiro 'EventStream's.++An 'EventStream' pairs a pure keiki 'SymTransducer' with the durable plumbing+needed to replay it. keiki can prove, with no SMT solver, that a transducer is+/replay-safe/ — each emitted chain is recoverable from its first event, observed+events invert to one edge, input reads are guarded by the matching command+constructor, and output-free edges do not change durable state. The umbrella+also checks guard determinism and structural reachability. This module lifts+keiki's umbrella check ('Keiki.validateTransducer') to the 'EventStream'+boundary so a service can assert all of its streams are sound before hydration.++Every warning enabled by the selected 'ValidationOptions' makes construction+fail. This is intentionally stricter than a pure keiki use: events are keiro's+only durable state, so accepting an unreplayable shape would lose state or defer+the failure to production. Build custom options by updating+'defaultValidationOptions'. The replay-contract checks for head recoverability+and state-changing output-free edges are always forced on at this durable+boundary; caller-supplied options may only strengthen that contract.++* 'validateEventStream' \/ 'validateEventStreamWith' run the pure check and+  return labelled warnings (empty when the stream is sound).+* 'mkEventStream' is a fail-fast smart constructor: it returns @Left warnings@+  for an unsafe stream and @Right validatedStream@ for a sound one. The returned+  'ValidatedEventStream' is the value command runners accept; the bare+  'EventStream' record literal remains available only for construction,+  validation, and low-level internals.+-}+module Keiro.EventStream.Validate (+    EventStreamWarning (..),+    ValidatedEventStream,+    unvalidated,+    validateEventStream,+    validateEventStreamWith,+    mkEventStream,+    mkEventStreamWith,+    mkEventStreamOrThrow,++    -- * Unchecked escape hatch (tests and emergency forensics only)+    mkEventStreamUnchecked,+) where++import Control.DeepSeq (force)+import Control.Exception (ErrorCall, displayException, evaluate, try)+import Data.Text (Text)+import Data.Text qualified as Text+import GHC.Stack (HasCallStack)+import Keiki.Core (+    EdgeRef (..),+    HsPred,+    TransducerValidationWarning (..),+    ValidationOptions (..),+    defaultValidationOptions,+    validateTransducer,+ )+import Keiro.EventStream (EventStream (..), SnapshotPolicy (..), StateCodec (..))+import System.IO.Unsafe (unsafePerformIO)++{- | A validation warning about one event stream, tagged with the+caller-supplied label so a multi-aggregate service can tell which stream is at+fault.+-}+data EventStreamWarning = EventStreamWarning+    { eswStreamLabel :: !Text+    , eswReason :: !Text+    -- ^ rendered from the keiki warning+    }+    deriving stock (Eq, Show)++{- | An 'EventStream' that has passed keiki validation and keiro's stream-level+checks. Command runners require this wrapper instead of a bare 'EventStream'.+The constructor is intentionally not exported; use 'mkEventStream',+'mkEventStreamWith', or 'mkEventStreamOrThrow' to obtain a validated value.+'mkEventStreamUnchecked' exists only for tests and emergency forensics.+-}+newtype ValidatedEventStream phi rs s ci co+    = ValidatedEventStream (EventStream phi rs s ci co)++-- | Recover the underlying stream for internal runners and low-level helpers.+unvalidated :: ValidatedEventStream phi rs s ci co -> EventStream phi rs s ci co+unvalidated (ValidatedEventStream es) = es++{- | Run keiki's pure umbrella check over a stream's transducer with the default+options. This includes hidden-input, head recoverability, inversion ambiguity,+guarded input reads, state-changing epsilon edges, determinism, and dead-edge+checks. An empty list means the stream passed every enabled check. Pure; no+solver.+-}+validateEventStream ::+    (Bounded s, Enum s, Ord s, Show s) =>+    -- | caller-supplied stream label+    Text ->+    EventStream (HsPred rs ci) rs s ci co ->+    [EventStreamWarning]+validateEventStream = validateEventStreamWith defaultValidationOptions++{- | As 'validateEventStream', but with caller-chosen 'ValidationOptions'.+The head-recoverability and state-changing-epsilon checks are always forced on:+events are keiro's only durable state, so callers may narrow only checks with a+documented benign override.+-}+validateEventStreamWith ::+    (Bounded s, Enum s, Ord s, Show s) =>+    ValidationOptions ->+    -- | caller-supplied stream label+    Text ->+    EventStream (HsPred rs ci) rs s ci co ->+    [EventStreamWarning]+validateEventStreamWith opts label es =+    snapshotWarnings label es+        <> initialSnapshotEncodeWarnings label es+        <> [ EventStreamWarning{eswStreamLabel = label, eswReason = renderWarning w}+           | w <- validateTransducer (forceReplayContract opts) (transducer es)+           ]++-- | Force the replay-contract checks required at keiro's durable boundary.+forceReplayContract :: ValidationOptions -> ValidationOptions+forceReplayContract opts =+    opts+        { checkStateChangingEpsilon = True+        , checkHeadRecoverability = True+        }++{- | Build a validated event stream with the default validation options.+Returns the warnings (@Left@) for an unsafe stream, or a+'ValidatedEventStream' (@Right@) when it passes.+-}+mkEventStream ::+    (Bounded s, Enum s, Ord s, Show s) =>+    -- | caller-supplied stream label+    Text ->+    EventStream (HsPred rs ci) rs s ci co ->+    Either [EventStreamWarning] (ValidatedEventStream (HsPred rs ci) rs s ci co)+mkEventStream = mkEventStreamWith defaultValidationOptions++{- | Build a validated event stream with caller-chosen validation options.+The replay-contract checks for head recoverability and state-changing epsilon+edges cannot be disabled here: caller options may only strengthen the durable+boundary. Only narrow other checks for a documented benign warning.+-}+mkEventStreamWith ::+    (Bounded s, Enum s, Ord s, Show s) =>+    ValidationOptions ->+    -- | caller-supplied stream label+    Text ->+    EventStream (HsPred rs ci) rs s ci co ->+    Either [EventStreamWarning] (ValidatedEventStream (HsPred rs ci) rs s ci co)+mkEventStreamWith opts label es =+    case validateEventStreamWith opts label es of+        [] -> Right (ValidatedEventStream es)+        warns -> Left warns++{- | Partial constructor for generated code and test fixtures that have a+sibling validation proof. Hand-authored application wiring should prefer+'mkEventStream' and handle @Left@ explicitly.+-}+mkEventStreamOrThrow ::+    (HasCallStack, Bounded s, Enum s, Ord s, Show s) =>+    -- | caller-supplied stream label+    Text ->+    EventStream (HsPred rs ci) rs s ci co ->+    ValidatedEventStream (HsPred rs ci) rs s ci co+mkEventStreamOrThrow label es =+    case mkEventStream label es of+        Right validated -> validated+        Left warns ->+            error $+                "Keiro.EventStream.Validate.mkEventStreamOrThrow: "+                    <> Text.unpack label+                    <> " is not replay-safe: "+                    <> show warns++{- | Wrap an 'EventStream' /without validation/. This skips every keiki and+keiro check, including the replay-contract checks that 'mkEventStream'+force-enables. A stream admitted through this function can silently lose state+changes and fail hydration. Tests and emergency forensics only; never use it+for production streams. Prefer 'mkEventStream'.+-}+mkEventStreamUnchecked ::+    EventStream phi rs s ci co ->+    ValidatedEventStream phi rs s ci co+mkEventStreamUnchecked = ValidatedEventStream++{- | Render a keiki warning to a human-readable reason. All eight constructors+carry @tvwDetail@; the source vertex is @edgeSource . tvwEdge@ (or @tvwSource@+for pair warnings).+-}+renderWarning :: (Show s) => TransducerValidationWarning s -> Text+renderWarning w = case w of+    HiddenInput{tvwEdge = e, tvwDetail = d} ->+        "hidden-input @" <> showT (edgeSource e) <> ": " <> Text.pack d+    HeadUnrecoverable{tvwEdge = e, tvwDetail = d} ->+        "head-unrecoverable @" <> showT (edgeSource e) <> ": " <> Text.pack d+    InversionAmbiguity{tvwSource = s, tvwDetail = d} ->+        "inversion-ambiguity @" <> showT s <> ": " <> Text.pack d+    UnguardedInputRead{tvwEdge = e, tvwDetail = d} ->+        "unguarded-input-read @" <> showT (edgeSource e) <> ": " <> Text.pack d+    StateChangingEpsilon{tvwEdge = e, tvwDetail = d} ->+        "state-changing-epsilon @" <> showT (edgeSource e) <> ": " <> Text.pack d+    NondeterministicPair{tvwSource = s, tvwDetail = d} ->+        "nondeterministic @" <> showT s <> ": " <> Text.pack d+    PossiblyDeadEdge{tvwEdge = e, tvwDetail = d} ->+        "possibly-dead @" <> showT (edgeSource e) <> ": " <> Text.pack d+    OpaqueGuard{tvwEdge = e, tvwDetail = d} ->+        "opaque-guard @" <> showT (edgeSource e) <> ": " <> Text.pack d+  where+    showT = Text.pack . show++snapshotWarnings :: Text -> EventStream phi rs s ci co -> [EventStreamWarning]+snapshotWarnings label es =+    case (snapshotPolicy es, stateCodec es) of+        (Never, _) -> []+        (_, Just _) -> []+        (_, Nothing) ->+            [ EventStreamWarning+                { eswStreamLabel = label+                , eswReason = "snapshotPolicy is set but stateCodec is Nothing; snapshots would never be written"+                }+            ]++{- | Force the configured codec over the initial aggregate state while the+stream is being validated. This catches the labelled @uninit: <slot>@+'ErrorCall' thunks installed by 'Keiki.Generics.emptyRegFile' before a service+can accept commands for a snapshot-enabled stream.++The public validation API remains pure; this narrowly scoped exception spoon+observes only 'ErrorCall'. Any other exception remains a programmer-visible+failure instead of being converted into a warning.+-}+initialSnapshotEncodeWarnings :: Text -> EventStream phi rs s ci co -> [EventStreamWarning]+initialSnapshotEncodeWarnings label es =+    case stateCodec es of+        Nothing -> []+        Just codec -> unsafePerformIO $ do+            encoded <- try @ErrorCall (evaluate (force (encode codec (initialState es, initialRegisters es))))+            pure $ case encoded of+                Right _ -> []+                Left err ->+                    [ EventStreamWarning+                        { eswStreamLabel = label+                        , eswReason =+                            "stateCodec cannot encode the initial state/registers: "+                                <> Text.pack (displayException err)+                        }+                    ]
+ src/Keiro/Integration/Event.hs view
@@ -0,0 +1,310 @@+{- | 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)
+ src/Keiro/Prelude.hs view
@@ -0,0 +1,66 @@+{- | The curated prelude shared across the Keiro packages.++Re-exports a deliberately small slice of @base@, @aeson@, @text@, @time@,+and the full @lens@ surface so modules can open with a single+@import Keiro.Prelude@ and get a consistent, explicit set of names. The+implicit @base@ @Prelude@ is expected to be disabled+(@NoImplicitPrelude@); only the names listed here are in scope, which+keeps event-sourcing code uniform and avoids accidental partial functions.++The @generic-lens@ @Data.Generics.Labels@ orphan instances are imported+for their effect only, enabling the @^. #field@ overloaded-label optics+used throughout the codebase.+-}+module Keiro.Prelude (+    module X,+    module Control.Lens,+)+where++import "aeson" Data.Aeson as X (+    FromJSON,+    Options,+    SumEncoding (..),+    ToJSON,+    Value,+    defaultOptions,+    fromJSON,+    genericParseJSON,+    genericToEncoding,+    genericToJSON,+    parseJSON,+    toEncoding,+    toJSON,+ )+import "aeson-casing" Data.Aeson.Casing as X (aesonDrop, aesonPrefix, camelCase, pascalCase, snakeCase, trainCase)+import "base" Control.Applicative as X ((<|>))+import "base" Control.Monad as X (guard, unless, void, when)+import "base" Control.Monad.IO.Class as X (MonadIO, liftIO)+import "base" Data.Bool as X (Bool (..), otherwise)+import "base" Data.Either as X (Either (..), either)+import "base" Data.Eq as X (Eq (..))+import "base" Data.Foldable as X (foldl', for_, traverse_)+import "base" Data.Function as X (($), (&), (.))+import "base" Data.Functor as X (Functor (..), (<$>))+import "base" Data.Int as X (Int, Int64)+import "base" Data.List.NonEmpty as X (NonEmpty (..))+import "base" Data.Maybe as X (Maybe (..), fromMaybe, isJust, isNothing, maybe)+import "base" Data.Ord as X (Ord (..))+import "base" Data.Proxy as X (Proxy (..))+import "base" Data.Semigroup as X (Semigroup (..))+import "base" Data.String as X (String)+import "base" GHC.Generics as X (Generic)+import "generic-lens" Data.Generics.Labels ()+import "lens" Control.Lens+import "text" Data.Text as X (Text)+import "time" Data.Time as X (UTCTime, getCurrentTime)+import "base" Prelude as X (+    Applicative (..),+    Bounded (..),+    Enum (..),+    IO,+    Monad (..),+    Show (..),+    error,+    pure,+ )
+ src/Keiro/Schema.hs view
@@ -0,0 +1,17 @@+{- | The single source of truth for the name of the dedicated PostgreSQL schema+that owns all of Keiro's framework tables.++A PostgreSQL /schema/ is a namespace inside one database. Keiro's framework+tables (@keiro_snapshots@, @keiro_timers@, @keiro_outbox@, …) live in the schema+named by 'keiroSchema'. The migrations in @keiro-migrations@ create them+schema-qualified (@keiro.<table>@); runtime queries in the @keiro@ package+qualify against this same name. This is the literal string every part of the+system must agree on, so it is defined once here and imported elsewhere.+-}+module Keiro.Schema (keiroSchema) where++import Data.Text (Text)++-- | The schema that owns Keiro's framework tables: @"keiro"@.+keiroSchema :: Text+keiroSchema = "keiro"
+ src/Keiro/Snapshot/Policy.hs view
@@ -0,0 +1,60 @@+{- | Evaluation of a stream's 'SnapshotPolicy'.++This is the single decision procedure command handling consults after an+append to decide whether to persist a snapshot of the folded state. It+keeps the 'SnapshotPolicy' constructors purely declarative — the meaning+of each constructor lives here.+-}+module Keiro.Snapshot.Policy (+    shouldSnapshot,+    shouldSnapshotSpan,+)+where++import Keiro.EventStream (SnapshotPolicy (..), Terminality (..))+import Keiro.Prelude+import Kiroku.Store.Types (StreamVersion (..))+import Prelude qualified++{- | Decide whether to snapshot given a policy, a terminality flag, the+folded state, and the post-append stream version.++* 'Never' is always 'False'.+* 'Every' @n@ is 'True' when the version is a positive multiple of @n@+  (a non-positive interval never fires).+* 'OnTerminal' mirrors the @terminal@ flag — snapshot exactly when the+  machine has reached a final state.+* 'Custom' defers to the caller-supplied predicate over state and version.+-}+shouldSnapshot :: SnapshotPolicy state -> Terminality -> state -> StreamVersion -> Bool+shouldSnapshot Never _ _ _ = False+shouldSnapshot (Every interval) _ _ (StreamVersion version)+    | interval <= 0 = False+    | otherwise = version > 0 Prelude.&& version `Prelude.mod` Prelude.fromIntegral interval == 0+shouldSnapshot OnTerminal terminality _ _ = terminality == Terminal+shouldSnapshot (Custom decide) terminality state version = decide terminality state version++{- | Like 'shouldSnapshot', but evaluated over the half-open stream-version+span @(preVersion, postVersion]@ that one append covered.++For 'Every' @n@, this fires when any positive multiple of @n@ lies inside+the span, so a batch append that jumps over a boundary still snapshots at+the post-append version. The other policies ignore the span and behave like+'shouldSnapshot'.+-}+shouldSnapshotSpan ::+    SnapshotPolicy state ->+    Terminality ->+    state ->+    StreamVersion ->+    StreamVersion ->+    Bool+shouldSnapshotSpan Never _ _ _ _ = False+shouldSnapshotSpan (Every interval) _ _ (StreamVersion preVersion) (StreamVersion postVersion)+    | interval <= 0 = False+    | postVersion <= 0 = False+    | otherwise = postVersion `Prelude.div` n > preVersion `Prelude.div` n+  where+    n = Prelude.fromIntegral interval+shouldSnapshotSpan OnTerminal terminality _ _ _ = terminality == Terminal+shouldSnapshotSpan (Custom decide) terminality state _ postVersion = decide terminality state postVersion
+ src/Keiro/Stream.hs view
@@ -0,0 +1,151 @@+{- | A phantom-typed handle to a single event stream.++'Stream' wraps a 'StreamName' but carries a phantom type parameter @a@+identifying /which/ aggregate or event stream the name belongs to. This+lets the rest of the framework demand, say, a @Stream Order@ rather than a+bare 'StreamName', so a name for one aggregate cannot be passed where+another is expected. The wrapper is otherwise transparent — use 'stream'+to construct one and 'streamName' to recover the underlying name.+-}+module Keiro.Stream (+    Stream (..),+    stream,+    streamName,+    mapStreamName,++    -- * Safe, category-based construction+    StreamCategory,+    categoryText,+    CategoryError (..),+    category,+    categoryUnsafe,+    categoryName,+    StreamIdSegment (..),+    entityStream,+    entityStreamId,+)+where++import Data.Char qualified as Char+import Data.Text qualified as Text+import GHC.Stack (HasCallStack)+import Keiro.Prelude+import Kiroku.Store.Types (CategoryName (..), StreamName (..))+import Kiroku.Store.Types qualified as Store++-- | A 'StreamName' tagged with the phantom type @a@ of the stream it names.+newtype Stream a = Stream+    { name :: StreamName+    }+    deriving stock (Generic, Eq, Ord, Show)++-- | Build a 'Stream' handle from a raw stream-name 'Text'.+stream :: Text -> Stream a+stream name = Stream{name = StreamName name}++-- | Recover the underlying 'StreamName' from a 'Stream' handle.+streamName :: Stream a -> StreamName+streamName value = value ^. #name++{- | Transform the underlying 'StreamName' while preserving the phantom+type. Handy for namespacing or prefixing a stream name without losing the+compile-time tag.+-}+mapStreamName :: (StreamName -> StreamName) -> Stream a -> Stream a+mapStreamName f value = value & #name %~ f++{- | A validated stream /category/: the prefix that precedes the first @-@ in+every stream name belonging to this family. Kiroku defines a stream's category+as the substring before its first @-@ (see 'Kiroku.Store.Types.categoryName'),+so a category must itself contain no @-@. Carries the same phantom type @a@ as+the 'Stream' handles it produces. Write a compound category in camelCase (e.g.+@"hospitalSurge"@ for a saga over hospital surges); @:@ is reserved for the+workflow stream family (@wf:\<name\>@).++Named 'StreamCategory' (not @Category@) to avoid clashing with the+'Kiroku.Store.Subscription.Types.Category' subscription-target constructor,+which consumer code commonly imports alongside this module.+-}+newtype StreamCategory a = StreamCategory {categoryTextOf :: Text}+    deriving stock (Generic, Eq, Ord, Show)++-- | Why a 'Text' is not a valid 'StreamCategory'.+data CategoryError+    = -- | the empty string+      CategoryEmpty+    | -- | contains the reserved @-@ category/id boundary+      CategoryContainsSeparator !Text+    | -- | equals a store-reserved name (@$all@)+      CategoryReserved !Text+    | -- | contains whitespace or a control character+      CategoryContainsIllegalChar !Char !Text+    deriving stock (Eq, Show, Generic)++{- | Validate a 'Text' as a 'Category'. Rejects the empty string, any text+containing @-@ (Kiroku's category/id boundary, which would make 'categoryName'+ambiguous), whitespace/control characters, and the reserved name @$all@.+-}+category :: Text -> Either CategoryError (StreamCategory a)+category t+    | Text.null t = Left CategoryEmpty+    | t == "$all" = Left (CategoryReserved t)+    | Text.isInfixOf "-" t = Left (CategoryContainsSeparator t)+    | Just illegal <- Text.find (\c -> Char.isSpace c || Char.isControl c) t =+        Left (CategoryContainsIllegalChar illegal t)+    | otherwise = Right (StreamCategory t)++-- | Recover the validated category text.+categoryText :: StreamCategory a -> Text+categoryText (StreamCategory t) = t++{- | Partial constructor for static, known-good category literals at definition+sites. Calls 'error' on an invalid category; never pass user input. Intended for+top-level @fooCategory = categoryUnsafe "foo"@.+-}+categoryUnsafe :: (HasCallStack) => Text -> StreamCategory a+categoryUnsafe t =+    case category t of+        Right value -> value+        Left err -> error ("Keiro.Stream.categoryUnsafe: invalid category " <> show t <> ": " <> show err)++{- | The 'CategoryName' for category-scoped reads+('Kiroku.Store.Read.readCategory') and category subscription targets. For any+@cat@ and id segment, @categoryName cat@ equals+@Kiroku.Store.Types.categoryName (streamName (entityStream cat id))@ — the+category rule is single-sourced in kiroku.+-}+categoryName :: StreamCategory a -> CategoryName+categoryName (StreamCategory t) = CategoryName t++{- | Render a value as the /id segment/ of a stream name (the part after the+first @-@). The id may itself contain @-@ without corrupting the leading+category, but must be non-blank to keep the stream name distinct from a bare+category read.+-}+class StreamIdSegment i where+    renderIdSegment :: i -> Text++instance StreamIdSegment Text where+    renderIdSegment = id++instance StreamIdSegment String where+    renderIdSegment = Text.pack++{- | Build the per-entity 'Stream' handle for an aggregate instance, rendering+@\<category\>-\<id\>@. The phantom type is carried from the 'StreamCategory', so+the result is correctly tagged. The actual name mechanics are delegated to+kiroku's 'Store.streamNameInCategory', keeping the category rule single-sourced+in the store.+-}+entityStream :: (HasCallStack) => StreamCategory a -> Text -> Stream a+entityStream (StreamCategory c) idSeg =+    if Text.null (Text.strip idSeg)+        then error ("Keiro.Stream.entityStream: blank id segment for category " <> show c)+        else Stream{name = Store.streamNameInCategory (CategoryName c) idSeg}++{- | 'entityStream' for a typed id with a 'StreamIdSegment' instance. Domain id+types typically add @instance StreamIdSegment FooId where renderIdSegment =+Text.pack . show@.+-}+entityStreamId :: (HasCallStack, StreamIdSegment i) => StreamCategory a -> i -> Stream a+entityStreamId c = entityStream c . renderIdSegment