diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -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: ` + `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. diff --git a/keiro-core.cabal b/keiro-core.cabal new file mode 100644 --- /dev/null +++ b/keiro-core.cabal @@ -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 diff --git a/src/Keiro/Codec.hs b/src/Keiro/Codec.hs new file mode 100644 --- /dev/null +++ b/src/Keiro/Codec.hs @@ -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) diff --git a/src/Keiro/EventStream.hs b/src/Keiro/EventStream.hs new file mode 100644 --- /dev/null +++ b/src/Keiro/EventStream.hs @@ -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) diff --git a/src/Keiro/EventStream/Validate.hs b/src/Keiro/EventStream/Validate.hs new file mode 100644 --- /dev/null +++ b/src/Keiro/EventStream/Validate.hs @@ -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: @ +'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) + } + ] diff --git a/src/Keiro/Integration/Event.hs b/src/Keiro/Integration/Event.hs new file mode 100644 --- /dev/null +++ b/src/Keiro/Integration/Event.hs @@ -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) diff --git a/src/Keiro/Prelude.hs b/src/Keiro/Prelude.hs new file mode 100644 --- /dev/null +++ b/src/Keiro/Prelude.hs @@ -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, + ) diff --git a/src/Keiro/Schema.hs b/src/Keiro/Schema.hs new file mode 100644 --- /dev/null +++ b/src/Keiro/Schema.hs @@ -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.@); 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" diff --git a/src/Keiro/Snapshot/Policy.hs b/src/Keiro/Snapshot/Policy.hs new file mode 100644 --- /dev/null +++ b/src/Keiro/Snapshot/Policy.hs @@ -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 diff --git a/src/Keiro/Stream.hs b/src/Keiro/Stream.hs new file mode 100644 --- /dev/null +++ b/src/Keiro/Stream.hs @@ -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:\@). + +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 +@\-\@. 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