diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,31 @@ # Changelog +## 0.7.0.0 — 2026-09-18 + +### Breaking Changes + +- Requires `keiro ^>=0.17.0.0` and `keiro-core ^>=0.17.0.0`, excluding the 0.16 compatibility + line. The exported read-model values now derive from truthful `ReadModelBlueprint` definitions; + their names, schemas, versions, shape hashes, SQL, and immediate behavior are unchanged. + +### Added + +- `Kioku.ProjectionCatalog`, which validates and binds two sources, three application projection + targets, two rebuild groups, two projection owners, and all nineteen query models. Catalog + registration and migration-time identity reconciliation derive from the same inventory. +- A public `ReadModelBlueprint` value beside every existing memory and session read model. + +### Changed + +- Queries use `runQueryWithFreshness Immediate` and no longer depend on Keiro's deprecated + consistency fields or `runQueryWith` compatibility layer. +- `withNoopAppEnv` validates both event streams and the projection catalog, then registers the + catalog before running user effects. Framework-owned timer writes remain explicit and atomic + with application projections because Keiro's exclusive target ownership cannot represent a + shared timer table. +- AI and distillation product records use strict fields, and the package participates in the + repository Haskell convention ratchet. + ## 0.6.0.0 — 2026-09-08 ### Breaking Changes diff --git a/kioku-core.cabal b/kioku-core.cabal --- a/kioku-core.cabal +++ b/kioku-core.cabal @@ -1,36 +1,42 @@ -cabal-version: 3.0 -name: kioku-core -version: 0.6.0.0 -synopsis: Reusable agent memory runtime +cabal-version: 3.0 +name: kioku-core +version: 0.7.0.0 +synopsis: Reusable agent memory runtime description: Core runtime for kioku. M1 establishes the application effect stack; later milestones add the memory and session aggregates. -license: BSD-3-Clause -license-file: LICENSE -author: Nadeem Bitar -maintainer: nadeem@gmail.com -copyright: 2026 Nadeem Bitar -category: Data -build-type: Simple -tested-with: GHC >=9.12 && <9.13 -homepage: https://github.com/shinzui/kioku -bug-reports: https://github.com/shinzui/kioku/issues +license: BSD-3-Clause +license-file: LICENSE +author: Nadeem Bitar +maintainer: nadeem@gmail.com +copyright: 2026 Nadeem Bitar +category: Data +build-type: Simple +tested-with: ghc >=9.12 && <9.13 +homepage: https://github.com/shinzui/kioku +bug-reports: https://github.com/shinzui/kioku/issues extra-doc-files: CHANGELOG.md source-repository head - type: git + type: git location: https://github.com/shinzui/kioku.git - subdir: kioku-core + subdir: kioku-core common warnings ghc-options: - -Wall -Wcompat -Widentities -Wincomplete-record-updates - -Wincomplete-uni-patterns -Wpartial-fields -Wredundant-constraints + -Wall + -Wcompat + -Widentities + -Wincomplete-record-updates + -Wincomplete-uni-patterns + -Wpartial-fields + -Wredundant-constraints + -Werror=deprecations -Werror=incomplete-patterns common shared - default-language: GHC2024 + default-language: GHC2024 default-extensions: BlockArguments DeriveAnyClass @@ -43,8 +49,8 @@ TemplateHaskell library - import: warnings, shared - hs-source-dirs: src + import: warnings, shared + hs-source-dirs: src exposed-modules: Kioku.AI.Config Kioku.AI.File @@ -71,6 +77,7 @@ Kioku.Memory.EventStream Kioku.Memory.ReadModel Kioku.Partition + Kioku.ProjectionCatalog Kioku.ReadModel Kioku.Recall Kioku.Recall.Capability @@ -84,8 +91,7 @@ -- Where Kioku's projections physically live is Kioku's business, so the -- qualified relation names stay internal rather than becoming a public API a -- consumer could pin its own SQL to. - other-modules: Kioku.Database.Schema - + other-modules: Kioku.Database.Schema -- Works around a GHC 9.12.4 coercion-optimiser bug that makes the *profiled* -- build of Kioku.Distill.L1 panic: -- @@ -105,48 +111,48 @@ -- mori://MMZK1526/mmzk-typeid/upstream-issues/mmzk-typeid-kindid-ghc-9-12-4-profiling-coercionkind-panic ghc-prof-options: -fno-opt-coercion build-depends: - , aeson >=2.2 && <2.3 - , baikai ^>=0.7.0.0 - , baikai-claude ^>=0.7.0.0 - , baikai-effectful ^>=0.4.0.1 - , baikai-openai ^>=0.7.0.0 - , base >=4.21 && <5 - , bytestring >=0.11 && <0.13 - , containers >=0.6 && <0.8 - , contravariant >=1.5 && <1.6 - , contravariant-extras >=0.3 && <0.4 - , crypton ^>=1.1.4 - , directory >=1.3 && <1.4 - , effectful >=2.5 && <2.7 - , effectful-core >=2.5 && <2.7 - , filepath >=1.4 && <1.6 - , generic-lens >=2.2 && <2.4 - , hasql >=1.6 && <1.11 - , hasql-pool >=1.2 && <1.5 - , hasql-transaction >=1.0 && <1.3 - , hs-opentelemetry-api >=1.0 && <1.1 - , keiki ^>=0.9.0.0 - , keiro ^>=0.16.0.0 - , keiro-core ^>=0.16.0.0 - , kioku-api ^>=0.6.0.0 - , kiroku-store ^>=0.8.0.0 - , lens >=5.2 && <5.4 - , mmzk-typeid >=0.7 && <0.8 - , shibuya-core ^>=0.9.0.0 - , shibuya-kiroku-adapter ^>=0.5.1.1 - , shikumi ^>=0.4.0.0 - , shikumi-trace ^>=0.3.0.0 - , text >=2.1 && <2.2 - , time >=1.12 && <1.15 - , unix >=2.8.8 && <2.9 - , unordered-containers >=0.2 && <0.3 - , uuid >=1.3 && <1.4 - , vector ^>=0.13.2.0 + aeson >=2.2 && <2.3, + baikai ^>=0.7.0.0, + baikai-claude ^>=0.7.0.0, + baikai-effectful ^>=0.4.0.1, + baikai-openai ^>=0.7.0.0, + base >=4.21 && <5, + bytestring >=0.11 && <0.13, + containers >=0.6 && <0.8, + contravariant >=1.5 && <1.6, + contravariant-extras >=0.3 && <0.4, + crypton ^>=1.1.4, + directory >=1.3 && <1.4, + effectful >=2.5 && <2.7, + effectful-core >=2.5 && <2.7, + filepath >=1.4 && <1.6, + generic-lens >=2.2 && <2.4, + hasql >=1.6 && <1.11, + hasql-pool >=1.2 && <1.5, + hasql-transaction >=1.0 && <1.3, + hs-opentelemetry-api >=1.0 && <1.1, + keiki ^>=0.9.0.0, + keiro ^>=0.17.0.0, + keiro-core ^>=0.17.0.0, + kioku-api ^>=0.7.0.0, + kiroku-store ^>=0.8.0.0, + lens >=5.2 && <5.4, + mmzk-typeid >=0.7 && <0.8, + shibuya-core ^>=0.9.0.0, + shibuya-kiroku-adapter ^>=0.5.1.1, + shikumi ^>=0.4.0.0, + shikumi-trace ^>=0.3.0.0, + text >=2.1 && <2.2, + time >=1.12 && <1.15, + unix >=2.8.8 && <2.9, + unordered-containers >=0.2 && <0.3, + uuid >=1.3 && <1.4, + vector ^>=0.13.2.0, test-suite kioku-test - import: warnings, shared - type: exitcode-stdio-1.0 - main-is: Main.hs + import: warnings, shared + type: exitcode-stdio-1.0 + main-is: Main.hs other-modules: Kioku.AIRuntimeSpec Kioku.AwaitingSpec @@ -156,6 +162,7 @@ Kioku.IdempotencySpec Kioku.MemorySpaceSpec Kioku.PortfolioAccessSpec + Kioku.ProjectionCatalogSpec Kioku.ReadModelReconcileSpec Kioku.RecallCompatSpec Kioku.RecallHarness @@ -172,39 +179,43 @@ Kioku.WorkspaceSpec hs-source-dirs: test - ghc-options: -threaded -rtsopts -with-rtsopts=-N + ghc-options: + -threaded + -rtsopts + -with-rtsopts=-N + build-depends: - , aeson >=2.2 - , async >=2.2 && <2.3 - , baikai ^>=0.7.0.0 - , base >=4.21 && <5 - , bytestring >=0.11 - , containers >=0.6 - , contravariant >=1.5 - , directory >=1.3 - , effectful >=2.5 - , effectful-core >=2.5 - , filepath >=1.4 - , generic-lens >=2.2 && <2.4 - , hasql >=1.6 - , hasql-transaction >=1.0 - , keiro ^>=0.16.0.0 - , keiro-core ^>=0.16.0.0 - , kioku-api ^>=0.6.0.0 - , kioku-core ^>=0.6.0.0 - , kioku-migrations:test-support ^>=0.6.0.0 - , kiroku-store ^>=0.8.0.0 - , lens >=5.2 - , shibuya-core ^>=0.9.0.0 - , shikumi ^>=0.4.0.0 - , shikumi-trace ^>=0.3.0.0 - , tasty >=1.5 - , tasty-expected-failure >=0.12 - , tasty-hunit >=0.10 - , temporary >=1.3 - , text >=2.1 - , time >=1.12 - , unix >=2.8.8 && <2.9 - , unordered-containers >=0.2 - , uuid >=1.3 - , vector + aeson >=2.2, + async >=2.2 && <2.3, + baikai ^>=0.7.0.0, + base >=4.21 && <5, + bytestring >=0.11, + containers >=0.6, + contravariant >=1.5, + directory >=1.3, + effectful >=2.5, + effectful-core >=2.5, + filepath >=1.4, + generic-lens >=2.2 && <2.4, + hasql >=1.6, + hasql-transaction >=1.0, + keiro ^>=0.17.0.0, + keiro-core ^>=0.17.0.0, + kioku-api ^>=0.7.0.0, + kioku-core ^>=0.7.0.0, + kioku-migrations:test-support ^>=0.7.0.0, + kiroku-store ^>=0.8.0.0, + lens >=5.2, + shibuya-core ^>=0.9.0.0, + shikumi ^>=0.4.0.0, + shikumi-trace ^>=0.3.0.0, + tasty >=1.5, + tasty-expected-failure >=0.12, + tasty-hunit >=0.10, + temporary >=1.3, + text >=2.1, + time >=1.12, + unix >=2.8.8 && <2.9, + unordered-containers >=0.2, + uuid >=1.3, + vector, diff --git a/src/Kioku/AI/Config.hs b/src/Kioku/AI/Config.hs --- a/src/Kioku/AI/Config.hs +++ b/src/Kioku/AI/Config.hs @@ -42,9 +42,9 @@ | InteractiveConfig InteractiveProvider InteractiveLaunchRequest data AIConfig = AIConfig - { distillationDefault :: DistillationConfig, - featureOverrides :: Map AIFeature DistillationConfig, - embeddingSettings :: Map AIFeature EmbeddingModel + { distillationDefault :: !DistillationConfig, + featureOverrides :: !(Map AIFeature DistillationConfig), + embeddingSettings :: !(Map AIFeature EmbeddingModel) } disabledAIConfig :: AIConfig diff --git a/src/Kioku/AI/Runtime.hs b/src/Kioku/AI/Runtime.hs --- a/src/Kioku/AI/Runtime.hs +++ b/src/Kioku/AI/Runtime.hs @@ -39,11 +39,11 @@ -- | Registries grant separate API and batch capabilities. The callback grants a -- fresh interactive session; presence of a terminal or executable grants none. data HostCapabilities = HostCapabilities - { apiRegistry :: Maybe Registry.ProviderRegistry, - batchRegistry :: Maybe Registry.ProviderRegistry, - allowInteractive :: Bool, - launchInteractive :: Maybe (AIFeature -> InteractiveLaunchRequest -> IO (Either AIExecutionError InteractiveLaunchResult)), - allowEmbeddingAPI :: Bool + { apiRegistry :: !(Maybe Registry.ProviderRegistry), + batchRegistry :: !(Maybe Registry.ProviderRegistry), + allowInteractive :: !Bool, + launchInteractive :: !(Maybe (AIFeature -> InteractiveLaunchRequest -> IO (Either AIExecutionError InteractiveLaunchResult))), + allowEmbeddingAPI :: !Bool } noHostCapabilities :: HostCapabilities diff --git a/src/Kioku/App.hs b/src/Kioku/App.hs --- a/src/Kioku/App.hs +++ b/src/Kioku/App.hs @@ -10,11 +10,19 @@ ) where +import Data.Text qualified as Text import Effectful (Eff, IOE, runEff) import Effectful.Error.Static (Error, runErrorNoCallStack) +import Keiro.Projection.Catalog (Validation (..)) import Keiro.Telemetry (KeiroMetrics) +import Kioku.Memory.EventStream (validateMemoryEventStream) import Kioku.Prelude +import Kioku.ProjectionCatalog + ( renderKiokuCatalogDiagnostics, + validateKiokuProjectionCatalog, + ) import Kioku.ReadModel (registerKiokuReadModels) +import Kioku.Session.EventStream (validateSessionEventStream) import Kiroku.Store.Connection (ConnectionSettings) import Kiroku.Store.Effect (Store, runStoreResource) import Kiroku.Store.Effect.Resource (KirokuStoreResource, withKirokuStore) @@ -42,12 +50,26 @@ withNoopAppEnv :: ConnectionSettings -> (AppEnv -> IO a) -> IO a withNoopAppEnv connectionSettings continue = do + validateRuntimeDefinitions tracer <- noopTracer let env = AppEnv {connectionSettings, tracer, metrics = Nothing} registration <- runAppIO env registerKiokuReadModels case registration of Left err -> fail ("Kioku read-model registration failed: " <> show err) - Right () -> continue env + Right (Left err) -> fail ("Kioku projection-catalog registration failed: " <> show err) + Right (Right _) -> continue env + +validateRuntimeDefinitions :: IO () +validateRuntimeDefinitions = do + case validateMemoryEventStream of + Left warnings -> fail ("Kioku memory event-stream validation failed: " <> show warnings) + Right _ -> pure () + case validateSessionEventStream of + Left warnings -> fail ("Kioku session event-stream validation failed: " <> show warnings) + Right _ -> pure () + case validateKiokuProjectionCatalog of + Failure diagnostics -> fail ("Kioku projection-catalog validation failed: " <> Text.unpack (renderKiokuCatalogDiagnostics diagnostics)) + Success _ -> pure () noopTracer :: IO Tracer noopTracer = do diff --git a/src/Kioku/Distill/Consolidate.hs b/src/Kioku/Distill/Consolidate.hs --- a/src/Kioku/Distill/Consolidate.hs +++ b/src/Kioku/Distill/Consolidate.hs @@ -30,28 +30,28 @@ deriving anyclass (ToSchema, FromModel) data ExistingMemory = ExistingMemory - { memoryId :: Field "existing memory identifier" Text, - memoryType :: Field "existing memory type or category" Text, - content :: Field "existing memory content" Text, - priority :: Field "existing memory priority where lower is more important" Int, - confidence :: Field "existing confidence label" Text + { memoryId :: !(Field "existing memory identifier" Text), + memoryType :: !(Field "existing memory type or category" Text), + content :: !(Field "existing memory content" Text), + priority :: !(Field "existing memory priority where lower is more important" Int), + confidence :: !(Field "existing confidence label" Text) } deriving stock (Generic, Eq, Show) deriving anyclass (ToSchema, FromModel, ToPrompt) data ConsolidateInput = ConsolidateInput - { scopeLabel :: Field "human-readable memory scope label" Text, - candidate :: ExtractedAtom, - existing :: [ExistingMemory] + { scopeLabel :: !(Field "human-readable memory scope label" Text), + candidate :: !ExtractedAtom, + existing :: ![ExistingMemory] } deriving stock (Generic, Eq, Show) deriving anyclass (ToSchema, FromModel, ToPrompt) data ConsolidationDecision = ConsolidationDecision - { action :: ConsolidationAction, - targetMemoryIds :: [Text], - resultContent :: Maybe Text, - rationale :: Field "one concise reason for the decision" Text + { action :: !ConsolidationAction, + targetMemoryIds :: ![Text], + resultContent :: !(Maybe Text), + rationale :: !(Field "one concise reason for the decision" Text) } deriving stock (Generic, Eq, Show) deriving anyclass (ToSchema, FromModel, ToPrompt) diff --git a/src/Kioku/Distill/Extract.hs b/src/Kioku/Distill/Extract.hs --- a/src/Kioku/Distill/Extract.hs +++ b/src/Kioku/Distill/Extract.hs @@ -20,18 +20,18 @@ import Shikumi.Signature (Signature, mkSignature) data ExtractInput = ExtractInput - { focus :: Field "the session focus, task, or topic" Text, - scopeLabel :: Field "human-readable memory scope label" Text, - conversation :: Field "recent turns, notes, or recorded memory evidence" Text + { focus :: !(Field "the session focus, task, or topic" Text), + scopeLabel :: !(Field "human-readable memory scope label" Text), + conversation :: !(Field "recent turns, notes, or recorded memory evidence" Text) } deriving stock (Generic, Eq, Show) deriving anyclass (ToSchema, FromModel, ToPrompt) data ExtractedAtom = ExtractedAtom - { atomType :: Field "one of: fact | pattern | preference | constraint | instruction" Text, - content :: Field "one concise durable memory sentence" Text, - priority :: Field "0=always inject; 100=default; larger=lower priority" Int, - confidence :: Field "one of: high | medium | low" Text + { atomType :: !(Field "one of: fact | pattern | preference | constraint | instruction" Text), + content :: !(Field "one concise durable memory sentence" Text), + priority :: !(Field "0=always inject; 100=default; larger=lower priority" Int), + confidence :: !(Field "one of: high | medium | low" Text) } deriving stock (Generic, Eq, Show) deriving anyclass (ToSchema, FromModel, ToPrompt) diff --git a/src/Kioku/Distill/Persona.hs b/src/Kioku/Distill/Persona.hs --- a/src/Kioku/Distill/Persona.hs +++ b/src/Kioku/Distill/Persona.hs @@ -18,8 +18,8 @@ import Shikumi.Signature (Signature, mkSignature) data PersonaInput = PersonaInput - { scopeLabel :: Field "human label of the scope" Text, - scenes :: Field "the scene blocks for this scope, newline-joined" Text + { scopeLabel :: !(Field "human label of the scope" Text), + scenes :: !(Field "the scene blocks for this scope, newline-joined" Text) } deriving stock (Generic, Eq, Show) deriving anyclass (ToSchema, FromModel, ToPrompt) diff --git a/src/Kioku/Distill/Runtime.hs b/src/Kioku/Distill/Runtime.hs --- a/src/Kioku/Distill/Runtime.hs +++ b/src/Kioku/Distill/Runtime.hs @@ -42,10 +42,10 @@ -- | Explicit controlled-runner seam for tests. Production construction accepts -- only a validated AIRuntime; its captured settings cannot be record-updated. data TestRunners = TestRunners - { runExtract :: ExtractInput -> IO (Either ShikumiError ExtractOutput), - runConsolidate :: ConsolidateInput -> IO (Either ShikumiError ConsolidationDecision), - runScene :: SceneInput -> IO (Either ShikumiError SceneOutput), - runPersona :: PersonaInput -> IO (Either ShikumiError PersonaOutput) + { runExtract :: !(ExtractInput -> IO (Either ShikumiError ExtractOutput)), + runConsolidate :: !(ConsolidateInput -> IO (Either ShikumiError ConsolidationDecision)), + runScene :: !(SceneInput -> IO (Either ShikumiError SceneOutput)), + runPersona :: !(PersonaInput -> IO (Either ShikumiError PersonaOutput)) } testDistillRuntime :: IO DistillRuntime diff --git a/src/Kioku/Distill/Scene.hs b/src/Kioku/Distill/Scene.hs --- a/src/Kioku/Distill/Scene.hs +++ b/src/Kioku/Distill/Scene.hs @@ -18,15 +18,15 @@ import Shikumi.Signature (Signature, mkSignature) data SceneInput = SceneInput - { scopeLabel :: Field "human label of the scope" Text, - atoms :: Field "the active memory atoms in this scope, newline-joined" Text + { scopeLabel :: !(Field "human label of the scope" Text), + atoms :: !(Field "the active memory atoms in this scope, newline-joined" Text) } deriving stock (Generic, Eq, Show) deriving anyclass (ToSchema, FromModel, ToPrompt) data SceneOutput = SceneOutput - { title :: Field "a short scene title, e.g. 'Testing & CI practices'" Text, - bodyMd :: Field "a markdown scene block summarizing the atoms as a narrative" Text + { title :: !(Field "a short scene title, e.g. 'Testing & CI practices'" Text), + bodyMd :: !(Field "a markdown scene block summarizing the atoms as a narrative" Text) } deriving stock (Generic, Eq, Show) deriving anyclass (ToSchema, FromModel, ToPrompt, Validatable) diff --git a/src/Kioku/Memory.hs b/src/Kioku/Memory.hs --- a/src/Kioku/Memory.hs +++ b/src/Kioku/Memory.hs @@ -62,7 +62,8 @@ import Effectful.Error.Static (Error) import Keiro.Command (CommandError (..), defaultRunCommandOptions) import Keiro.Projection (runCommandWithProjections) -import Keiro.ReadModel (ConsistencyMode (..), ReadModelError, runQueryWith) +import Keiro.Projection.Catalog (typedInlineProjections) +import Keiro.ReadModel (QueryFreshness (Immediate), ReadModelError, runQueryWithFreshness) import Kioku.Api.Access ( MemoryAccessContext, MemoryPermission (..), @@ -93,10 +94,10 @@ memoriesBySessionRowsReadModel, memoriesByTypeRowsReadModel, memoryByIdReadModel, - memoryInlineProjection, memorySupersessionChainReadModel, ) import Kioku.Prelude +import Kioku.ProjectionCatalog (kiokuProjectionCatalog, memoryProjectionSet) import Kiroku.Store.Effect (Store) import Kiroku.Store.Effect.Resource (KirokuStoreResource) import Kiroku.Store.Error (StoreError) @@ -518,9 +519,9 @@ MemoryId -> Eff es (Either ReadModelError (Maybe MemoryRow)) lookupMemory space mid = - runQueryWith + runQueryWithFreshness Nothing - Eventual + Immediate memoryByIdReadModel MemoryByIdQuery {memorySpaceId = space, memoryId = idText mid} @@ -538,9 +539,9 @@ Namespace -> Eff es (Either ReadModelError [MemoryRow]) getActiveRowsInNamespace space (Namespace ns) = - runQueryWith + runQueryWithFreshness Nothing - Eventual + Immediate memoriesByNamespaceRowsReadModel MemoriesByNamespaceQuery {memorySpaceId = space, namespace = ns} @@ -550,9 +551,9 @@ MemoryScope -> Eff es (Either ReadModelError [MemoryRow]) getActiveRowsByScope space scope = - runQueryWith + runQueryWithFreshness Nothing - Eventual + Immediate memoriesByScopeRowsReadModel MemoriesByScopeQuery { memorySpaceId = space, @@ -567,9 +568,9 @@ SessionId -> Eff es (Either ReadModelError [MemoryRow]) getRowsBySession space sid = - runQueryWith + runQueryWithFreshness Nothing - Eventual + Immediate memoriesBySessionRowsReadModel MemoriesBySessionQuery {memorySpaceId = space, sessionId = idText sid} @@ -580,9 +581,9 @@ MemoryType -> Eff es (Either ReadModelError [MemoryRow]) getActiveRowsByType space (Namespace ns) memoryType = - runQueryWith + runQueryWithFreshness Nothing - Eventual + Immediate memoriesByTypeRowsReadModel MemoriesByTypeQuery { memorySpaceId = space, @@ -596,9 +597,9 @@ MemoryId -> Eff es (Either ReadModelError [MemoryRow]) getSupersessionChain space mid = - runQueryWith + runQueryWithFreshness Nothing - Eventual + Immediate memorySupersessionChainReadModel MemorySupersessionChainQuery {memorySpaceId = space, memoryId = idText mid} @@ -608,13 +609,17 @@ MemoryCommand -> Eff es (Either MemoryWriteError MemoryId) runMemoryCommand mid cmd = do + -- Keiro's catalog cannot declare the shared framework-owned timer table as + -- a Kioku target. Derive Kioku's handler from the catalog, keep the timer + -- callback explicit, and run both in the append transaction; see + -- docs/adr/catalog-application-projections-not-framework-timers.md. result <- runCommandWithProjections defaultRunCommandOptions memoryEventStream (memoryStream mid) cmd - [memoryInlineProjection, l2SceneTimerScheduleProjection] + (typedInlineProjections kiokuProjectionCatalog memoryProjectionSet <> [l2SceneTimerScheduleProjection]) pure $ case result of Left err -> Left (MemoryCommandRejected err) diff --git a/src/Kioku/Memory/EventStream.hs b/src/Kioku/Memory/EventStream.hs --- a/src/Kioku/Memory/EventStream.hs +++ b/src/Kioku/Memory/EventStream.hs @@ -1,6 +1,7 @@ module Kioku.Memory.EventStream ( MemoryEventStream, memoryEventStream, + validateMemoryEventStream, memoryCodec, memoryStream, parseMemoryEvent, @@ -15,7 +16,7 @@ import Keiki.Generics (emptyRegFile) import Keiro.Codec (Codec (..), EventType (..)) import Keiro.EventStream (EventStream (..), SnapshotPolicy (..)) -import Keiro.EventStream.Validate (ValidatedEventStream, mkEventStreamOrThrow) +import Keiro.EventStream.Validate (EventStreamWarning, ValidatedEventStream, mkEventStream) import Keiro.Stream (Stream) import Keiro.Stream qualified as Stream import Kioku.Id (MemoryId, idText) @@ -30,7 +31,11 @@ memoryEventStream :: ValidatedEventStream (HsPred MemoryRegs MemoryCommand) MemoryRegs MemoryVertex MemoryCommand MemoryEvent memoryEventStream = - mkEventStreamOrThrow "kioku-memory" memoryEventStreamDefinition + either (error . ("invalid kioku memory event stream: " <>) . show) id validateMemoryEventStream + +-- | The explicit startup proof for the hand-written memory stream. +validateMemoryEventStream :: Either [EventStreamWarning] (ValidatedEventStream (HsPred MemoryRegs MemoryCommand) MemoryRegs MemoryVertex MemoryCommand MemoryEvent) +validateMemoryEventStream = mkEventStream "kioku-memory" memoryEventStreamDefinition memoryEventStreamDefinition :: MemoryEventStream memoryEventStreamDefinition = diff --git a/src/Kioku/Memory/ReadModel.hs b/src/Kioku/Memory/ReadModel.hs --- a/src/Kioku/Memory/ReadModel.hs +++ b/src/Kioku/Memory/ReadModel.hs @@ -14,15 +14,25 @@ MemoriesBySessionQuery (..), MemoriesByTypeQuery (..), MemorySupersessionChainQuery (..), + memoryByIdReadModelBlueprint, memoryByIdReadModel, + memoriesByNamespaceReadModelBlueprint, memoriesByNamespaceReadModel, + memoriesByNamespaceRowsReadModelBlueprint, memoriesByNamespaceRowsReadModel, + memoriesByScopeReadModelBlueprint, memoriesByScopeReadModel, + memoriesByScopeRowsReadModelBlueprint, memoriesByScopeRowsReadModel, + memoriesBySessionReadModelBlueprint, memoriesBySessionReadModel, + memoriesBySessionRowsReadModelBlueprint, memoriesBySessionRowsReadModel, + memoriesByTypeReadModelBlueprint, memoriesByTypeReadModel, + memoriesByTypeRowsReadModelBlueprint, memoriesByTypeRowsReadModel, + memorySupersessionChainReadModelBlueprint, memorySupersessionChainReadModel, ) where @@ -40,7 +50,12 @@ import Hasql.Statement (Statement, preparable) import Hasql.Transaction qualified as Tx import Keiro.Projection (InlineProjection (..)) -import Keiro.ReadModel (ConsistencyMode (..), ReadModel (..), StrongScope (..)) +import Keiro.ReadModel + ( QueryCursorAuthority (NoQueryCursor), + ReadModel, + ReadModelBlueprint (..), + immediateReadModel, + ) import Kioku.Api.Access (MemorySpaceId) import Kioku.Api.Scope (scopeFromColumns, scopeKindText, scopeNamespaceText, scopeRefText) import Kioku.Api.Types (MemoryRecord (..), confidenceToText, memoryTypeToText) @@ -189,142 +204,152 @@ } memoryByIdReadModel :: ReadModel MemoryByIdQuery (Maybe MemoryRow) -memoryByIdReadModel = - ReadModel +memoryByIdReadModel = immediateReadModel memoryByIdReadModelBlueprint + +memoryByIdReadModelBlueprint :: ReadModelBlueprint MemoryByIdQuery (Maybe MemoryRow) +memoryByIdReadModelBlueprint = + ReadModelBlueprint { name = "kioku-memory-by-id", schema = kiokuSchema, tableName = memoriesRelation, - subscriptionName = "kioku-memory-inline", version = memoryReadModelVersion, shapeHash = memoryReadModelShapeHash, - defaultConsistency = Eventual, - strongScope = EntireLog, + cursorAuthority = NoQueryCursor, query = \q -> Tx.statement q selectMemoryByIdStmt } memoriesByNamespaceReadModel :: ReadModel MemoriesByNamespaceQuery [MemoryRecord] -memoriesByNamespaceReadModel = - ReadModel +memoriesByNamespaceReadModel = immediateReadModel memoriesByNamespaceReadModelBlueprint + +memoriesByNamespaceReadModelBlueprint :: ReadModelBlueprint MemoriesByNamespaceQuery [MemoryRecord] +memoriesByNamespaceReadModelBlueprint = + ReadModelBlueprint { name = "kioku-memories-by-namespace", schema = kiokuSchema, tableName = memoriesRelation, - subscriptionName = "kioku-memory-inline", version = memoryReadModelVersion, shapeHash = memoryReadModelShapeHash, - defaultConsistency = Eventual, - strongScope = EntireLog, + cursorAuthority = NoQueryCursor, query = \q -> Tx.statement q selectActiveByNamespaceStmt } memoriesByNamespaceRowsReadModel :: ReadModel MemoriesByNamespaceQuery [MemoryRow] -memoriesByNamespaceRowsReadModel = - ReadModel +memoriesByNamespaceRowsReadModel = immediateReadModel memoriesByNamespaceRowsReadModelBlueprint + +memoriesByNamespaceRowsReadModelBlueprint :: ReadModelBlueprint MemoriesByNamespaceQuery [MemoryRow] +memoriesByNamespaceRowsReadModelBlueprint = + ReadModelBlueprint { name = "kioku-memory-rows-by-namespace", schema = kiokuSchema, tableName = memoriesRelation, - subscriptionName = "kioku-memory-inline", version = memoryReadModelVersion, shapeHash = memoryReadModelShapeHash, - defaultConsistency = Eventual, - strongScope = EntireLog, + cursorAuthority = NoQueryCursor, query = \q -> Tx.statement q selectActiveByNamespaceRowsStmt } memoriesByScopeReadModel :: ReadModel MemoriesByScopeQuery [MemoryRecord] -memoriesByScopeReadModel = - ReadModel +memoriesByScopeReadModel = immediateReadModel memoriesByScopeReadModelBlueprint + +memoriesByScopeReadModelBlueprint :: ReadModelBlueprint MemoriesByScopeQuery [MemoryRecord] +memoriesByScopeReadModelBlueprint = + ReadModelBlueprint { name = "kioku-memories-by-scope", schema = kiokuSchema, tableName = memoriesRelation, - subscriptionName = "kioku-memory-inline", version = memoryReadModelVersion, shapeHash = memoryReadModelShapeHash, - defaultConsistency = Eventual, - strongScope = EntireLog, + cursorAuthority = NoQueryCursor, query = \q -> Tx.statement q selectActiveByScopeStmt } memoriesByScopeRowsReadModel :: ReadModel MemoriesByScopeQuery [MemoryRow] -memoriesByScopeRowsReadModel = - ReadModel +memoriesByScopeRowsReadModel = immediateReadModel memoriesByScopeRowsReadModelBlueprint + +memoriesByScopeRowsReadModelBlueprint :: ReadModelBlueprint MemoriesByScopeQuery [MemoryRow] +memoriesByScopeRowsReadModelBlueprint = + ReadModelBlueprint { name = "kioku-memory-rows-by-scope", schema = kiokuSchema, tableName = memoriesRelation, - subscriptionName = "kioku-memory-inline", version = memoryReadModelVersion, shapeHash = memoryReadModelShapeHash, - defaultConsistency = Eventual, - strongScope = EntireLog, + cursorAuthority = NoQueryCursor, query = \q -> Tx.statement q selectActiveByScopeRowsStmt } memoriesBySessionReadModel :: ReadModel MemoriesBySessionQuery [MemoryRecord] -memoriesBySessionReadModel = - ReadModel +memoriesBySessionReadModel = immediateReadModel memoriesBySessionReadModelBlueprint + +memoriesBySessionReadModelBlueprint :: ReadModelBlueprint MemoriesBySessionQuery [MemoryRecord] +memoriesBySessionReadModelBlueprint = + ReadModelBlueprint { name = "kioku-memories-by-session", schema = kiokuSchema, tableName = memoriesRelation, - subscriptionName = "kioku-memory-inline", version = memoryReadModelVersion, shapeHash = memoryReadModelShapeHash, - defaultConsistency = Eventual, - strongScope = EntireLog, + cursorAuthority = NoQueryCursor, query = \q -> Tx.statement q selectBySessionStmt } memoriesBySessionRowsReadModel :: ReadModel MemoriesBySessionQuery [MemoryRow] -memoriesBySessionRowsReadModel = - ReadModel +memoriesBySessionRowsReadModel = immediateReadModel memoriesBySessionRowsReadModelBlueprint + +memoriesBySessionRowsReadModelBlueprint :: ReadModelBlueprint MemoriesBySessionQuery [MemoryRow] +memoriesBySessionRowsReadModelBlueprint = + ReadModelBlueprint { name = "kioku-memory-rows-by-session", schema = kiokuSchema, tableName = memoriesRelation, - subscriptionName = "kioku-memory-inline", version = memoryReadModelVersion, shapeHash = memoryReadModelShapeHash, - defaultConsistency = Eventual, - strongScope = EntireLog, + cursorAuthority = NoQueryCursor, query = \q -> Tx.statement q selectBySessionRowsStmt } memoriesByTypeReadModel :: ReadModel MemoriesByTypeQuery [MemoryRecord] -memoriesByTypeReadModel = - ReadModel +memoriesByTypeReadModel = immediateReadModel memoriesByTypeReadModelBlueprint + +memoriesByTypeReadModelBlueprint :: ReadModelBlueprint MemoriesByTypeQuery [MemoryRecord] +memoriesByTypeReadModelBlueprint = + ReadModelBlueprint { name = "kioku-memories-by-type", schema = kiokuSchema, tableName = memoriesRelation, - subscriptionName = "kioku-memory-inline", version = memoryReadModelVersion, shapeHash = memoryReadModelShapeHash, - defaultConsistency = Eventual, - strongScope = EntireLog, + cursorAuthority = NoQueryCursor, query = \q -> Tx.statement q selectByTypeStmt } memoriesByTypeRowsReadModel :: ReadModel MemoriesByTypeQuery [MemoryRow] -memoriesByTypeRowsReadModel = - ReadModel +memoriesByTypeRowsReadModel = immediateReadModel memoriesByTypeRowsReadModelBlueprint + +memoriesByTypeRowsReadModelBlueprint :: ReadModelBlueprint MemoriesByTypeQuery [MemoryRow] +memoriesByTypeRowsReadModelBlueprint = + ReadModelBlueprint { name = "kioku-memory-rows-by-type", schema = kiokuSchema, tableName = memoriesRelation, - subscriptionName = "kioku-memory-inline", version = memoryReadModelVersion, shapeHash = memoryReadModelShapeHash, - defaultConsistency = Eventual, - strongScope = EntireLog, + cursorAuthority = NoQueryCursor, query = \q -> Tx.statement q selectByTypeRowsStmt } memorySupersessionChainReadModel :: ReadModel MemorySupersessionChainQuery [MemoryRow] -memorySupersessionChainReadModel = - ReadModel +memorySupersessionChainReadModel = immediateReadModel memorySupersessionChainReadModelBlueprint + +memorySupersessionChainReadModelBlueprint :: ReadModelBlueprint MemorySupersessionChainQuery [MemoryRow] +memorySupersessionChainReadModelBlueprint = + ReadModelBlueprint { name = "kioku-memory-supersession-chain", schema = kiokuSchema, tableName = memoriesRelation, - subscriptionName = "kioku-memory-inline", version = memoryReadModelVersion, shapeHash = memoryReadModelShapeHash, - defaultConsistency = Eventual, - strongScope = EntireLog, + cursorAuthority = NoQueryCursor, query = \q -> Tx.statement q selectSupersessionChainStmt } diff --git a/src/Kioku/ProjectionCatalog.hs b/src/Kioku/ProjectionCatalog.hs new file mode 100644 --- /dev/null +++ b/src/Kioku/ProjectionCatalog.hs @@ -0,0 +1,247 @@ +-- | Kioku's complete application-owned projection inventory. +-- +-- The catalog deliberately stops at the @kioku@ schema boundary. Timer +-- scheduling writes Keiro's shared @keiro.keiro_timers@ table and therefore +-- remains an explicitly unmanaged transactional side effect at command call +-- sites; declaring that table here would falsely give one Kioku projection +-- exclusive ownership of framework state. +module Kioku.ProjectionCatalog + ( kiokuProjectionCatalog, + validateKiokuProjectionCatalog, + validateKiokuProjectionCatalogWith, + memoryProjectionSet, + sessionProjectionSet, + kiokuCatalogInventory, + kiokuCatalogFingerprint, + renderKiokuCatalogDiagnostics, + ) +where + +import Data.List.NonEmpty qualified as NonEmpty +import Data.Text qualified as Text +import Keiro.Projection (InlineProjection (..)) +import Keiro.Projection.Catalog + ( CatalogDiagnostic (..), + CatalogFingerprint, + CatalogInventory, + ClaimSite, + ProjectionCatalog (..), + ProjectionDefinition (..), + ProjectionHandler (..), + ProjectionReplayPolicy (..), + ProjectionSet (..), + QualifiedTable (..), + QueryModelBinding (..), + RebuildGroupDeclaration (..), + RebuildGroupId, + SomeProjectionSet (..), + SomeQueryModelBinding (..), + SourceDeclaration (..), + SourceId, + SourceScope (..), + TargetDeclaration (..), + TargetId, + TargetResetPolicy (..), + ValidatedProjectionCatalog, + Validation (..), + catalogFingerprint, + catalogInventory, + diagnosticCodeText, + mkClaimSite, + mkProjectionId, + mkQueryModelId, + mkRebuildGroupId, + mkSourceId, + mkTargetId, + replayAdapterFromCodec, + validateProjectionCatalog, + ) +import Keiro.ReadModel (ReadModel (..)) +import Kioku.Database.Schema (kiokuSchema, memoriesRelation, sessionsRelation, turnsRelation) +import Kioku.Memory.Domain (MemoryEvent) +import Kioku.Memory.EventStream (memoryCodec) +import Kioku.Memory.ReadModel + ( memoriesByNamespaceReadModel, + memoriesByNamespaceRowsReadModel, + memoriesByScopeReadModel, + memoriesByScopeRowsReadModel, + memoriesBySessionReadModel, + memoriesBySessionRowsReadModel, + memoriesByTypeReadModel, + memoriesByTypeRowsReadModel, + memoryByIdReadModel, + memoryInlineProjection, + memorySupersessionChainReadModel, + ) +import Kioku.Prelude +import Kioku.Session.Domain (SessionEvent) +import Kioku.Session.EventStream (sessionCodec) +import Kioku.Session.ReadModel + ( awaitingSessionsByCorrelationKeyReadModel, + sessionByIdReadModel, + sessionChainReadModel, + sessionDelegationChildrenReadModel, + sessionInlineProjection, + sessionsByFocusReadModel, + sessionsByNamespaceReadModel, + sessionsByScopeReadModel, + sessionsByStartedRangeReadModel, + turnsBySessionReadModel, + ) +import Kiroku.Store.Types (CategoryName (..)) + +memorySourceId, sessionSourceId :: SourceId +memorySourceId = must (mkSourceId "kioku-memory-events") +sessionSourceId = must (mkSourceId "kioku-session-events") + +memoriesTargetId, sessionsTargetId, turnsTargetId :: TargetId +memoriesTargetId = must (mkTargetId "kioku-memories") +sessionsTargetId = must (mkTargetId "kioku-sessions") +turnsTargetId = must (mkTargetId "kioku-turns") + +memoryGroupId, sessionGroupId :: RebuildGroupId +memoryGroupId = must (mkRebuildGroupId "kioku-memory") +sessionGroupId = must (mkRebuildGroupId "kioku-session") + +memoryProjectionSet :: ProjectionSet MemoryEvent +memoryProjectionSet = + ProjectionSet + { projectionSource = memorySourceId, + projectionDefinitions = + ProjectionDefinition + { projectionId = must (mkProjectionId "kioku-memory-inline"), + rebuildGroup = memoryGroupId, + ownedTargets = memoriesTargetId :| [], + replayPolicy = Replayable (replayAdapterFromCodec memoryCodec memoryInlineProjection.apply), + handlers = InlineHandler memoryInlineProjection (site "memory inline handler") :| [], + claimSite = site "memory projection owner" + } + :| [], + claimSite = site "memory projection set" + } + +sessionProjectionSet :: ProjectionSet SessionEvent +sessionProjectionSet = + ProjectionSet + { projectionSource = sessionSourceId, + projectionDefinitions = + ProjectionDefinition + { projectionId = must (mkProjectionId "kioku-session-inline"), + rebuildGroup = sessionGroupId, + ownedTargets = sessionsTargetId :| [turnsTargetId], + replayPolicy = Replayable (replayAdapterFromCodec sessionCodec sessionInlineProjection.apply), + handlers = InlineHandler sessionInlineProjection (site "session inline handler") :| [], + claimSite = site "session projection owner" + } + :| [], + claimSite = site "session projection set" + } + +kiokuProjectionCatalogDefinition :: ProjectionCatalog +kiokuProjectionCatalogDefinition = + ProjectionCatalog + { sources = + [ SourceDeclaration memorySourceId (CategorySource (CategoryName "kioku_memory")) "kioku-memory-codec-v1" (site "memory event source"), + SourceDeclaration sessionSourceId (CategorySource (CategoryName "kioku_session")) "kioku-session-codec-v1" (site "session event source") + ], + targets = + [ TargetDeclaration memoriesTargetId (QualifiedTable kiokuSchema memoriesRelation) ClearBeforeReplay [] (site "memories target"), + TargetDeclaration sessionsTargetId (QualifiedTable kiokuSchema sessionsRelation) ClearBeforeReplay [] (site "sessions target"), + TargetDeclaration turnsTargetId (QualifiedTable kiokuSchema turnsRelation) ClearBeforeReplay [sessionsTargetId] (site "turns target") + ], + rebuildGroups = + [ RebuildGroupDeclaration memoryGroupId [memoriesTargetId] [] (site "memory rebuild group"), + RebuildGroupDeclaration sessionGroupId [sessionsTargetId, turnsTargetId] [] (site "session rebuild group") + ], + projectionRevisions = [], + externalReadContracts = [], + subscriptions = [], + dedupKeys = [], + queryModels = memoryQueryBindings <> sessionQueryBindings, + projectionSets = + [ SomeProjectionSet memoryProjectionSet, + SomeProjectionSet sessionProjectionSet + ] + } + +memoryQueryBindings :: [SomeQueryModelBinding] +memoryQueryBindings = + [ binding memoryByIdReadModel memoryGroupId memoriesTargetId, + binding memoriesByNamespaceReadModel memoryGroupId memoriesTargetId, + binding memoriesByNamespaceRowsReadModel memoryGroupId memoriesTargetId, + binding memoriesByScopeReadModel memoryGroupId memoriesTargetId, + binding memoriesByScopeRowsReadModel memoryGroupId memoriesTargetId, + binding memoriesBySessionReadModel memoryGroupId memoriesTargetId, + binding memoriesBySessionRowsReadModel memoryGroupId memoriesTargetId, + binding memoriesByTypeReadModel memoryGroupId memoriesTargetId, + binding memoriesByTypeRowsReadModel memoryGroupId memoriesTargetId, + binding memorySupersessionChainReadModel memoryGroupId memoriesTargetId + ] + +sessionQueryBindings :: [SomeQueryModelBinding] +sessionQueryBindings = + [ binding sessionByIdReadModel sessionGroupId sessionsTargetId, + binding sessionsByNamespaceReadModel sessionGroupId sessionsTargetId, + binding sessionsByScopeReadModel sessionGroupId sessionsTargetId, + binding sessionsByFocusReadModel sessionGroupId sessionsTargetId, + binding sessionsByStartedRangeReadModel sessionGroupId sessionsTargetId, + binding sessionChainReadModel sessionGroupId sessionsTargetId, + binding sessionDelegationChildrenReadModel sessionGroupId sessionsTargetId, + binding awaitingSessionsByCorrelationKeyReadModel sessionGroupId sessionsTargetId, + binding turnsBySessionReadModel sessionGroupId turnsTargetId + ] + +binding :: ReadModel q r -> RebuildGroupId -> TargetId -> SomeQueryModelBinding +binding readModel rebuildGroup observedTarget = + SomeQueryModelBinding + QueryModelBinding + { queryModelId = must (mkQueryModelId readModel.name), + readModel, + rebuildGroup, + observedTargets = [observedTarget], + claimSite = site ("query model " <> readModel.name) + } + +validateKiokuProjectionCatalog :: Validation (NonEmpty CatalogDiagnostic) ValidatedProjectionCatalog +validateKiokuProjectionCatalog = validateKiokuProjectionCatalogWith id + +-- | Validate a deliberately transformed definition. Production code should +-- use 'validateKiokuProjectionCatalog'; this seam exists so tests can prove +-- malformed ownership and fingerprint declarations fail closed without +-- exporting Kioku's unvalidated catalog as a runtime value. +validateKiokuProjectionCatalogWith :: + (ProjectionCatalog -> ProjectionCatalog) -> + Validation (NonEmpty CatalogDiagnostic) ValidatedProjectionCatalog +validateKiokuProjectionCatalogWith modifyCatalog = + validateProjectionCatalog (modifyCatalog kiokuProjectionCatalogDefinition) + +kiokuProjectionCatalog :: ValidatedProjectionCatalog +kiokuProjectionCatalog = + case validateKiokuProjectionCatalog of + Success catalog -> catalog + Failure diagnostics -> + error (Text.unpack (renderKiokuCatalogDiagnostics diagnostics)) + +kiokuCatalogInventory :: CatalogInventory +kiokuCatalogInventory = catalogInventory kiokuProjectionCatalog + +kiokuCatalogFingerprint :: CatalogFingerprint +kiokuCatalogFingerprint = catalogFingerprint kiokuProjectionCatalog + +renderKiokuCatalogDiagnostics :: NonEmpty CatalogDiagnostic -> Text +renderKiokuCatalogDiagnostics diagnostics = + Text.intercalate + "; " + [ diagnosticCodeText diagnostic.diagnosticCode + <> " (" + <> diagnostic.diagnosticIdentity + <> "): " + <> diagnostic.diagnosticMessage + | diagnostic <- NonEmpty.toList diagnostics + ] + +site :: Text -> ClaimSite +site = must . mkClaimSite + +must :: (Show error) => Either error value -> value +must = either (error . show) id diff --git a/src/Kioku/ReadModel.hs b/src/Kioku/ReadModel.hs --- a/src/Kioku/ReadModel.hs +++ b/src/Kioku/ReadModel.hs @@ -12,9 +12,9 @@ -- 'Keiro.ReadModel.ReadModelStaleSchema'. -- -- 'reconcileReadModelRegistry' repairs those rows to the identity the compiled --- code expects, deriving every name, version, and shape hash from --- 'kiokuReadModelSchemas' — the same 'ReadModel' values the queries use, so the --- registry can never disagree with the code. The @kioku-migrate@ executable runs +-- code expects. 'kiokuReadModelSchemas' is derived from the same validated +-- projection catalog that registers and supplies the queries, so the migration +-- reconciler cannot drift into a second runtime inventory. The @kioku-migrate@ executable runs -- it immediately after applying migrations, which is why a read-model version -- bump needs no hand-written registry SQL. A host that applies migrations as a -- library (by running @Kioku.Migrations.kiokuMigrationPlan@ through pg-migrate) @@ -34,32 +34,11 @@ where import Effectful (Eff, (:>)) -import Keiro.ReadModel (ReadModel (..)) +import Keiro.Projection.Catalog (CatalogRegistration (..), catalogRegistrations) +import Keiro.ReadModel.Rebuild (CatalogRegistrationError, GroupRebuildMetadata, registerProjectionCatalog) import Keiro.ReadModel.Schema qualified as Schema -import Kioku.Memory.ReadModel - ( memoriesByNamespaceReadModel, - memoriesByNamespaceRowsReadModel, - memoriesByScopeReadModel, - memoriesByScopeRowsReadModel, - memoriesBySessionReadModel, - memoriesBySessionRowsReadModel, - memoriesByTypeReadModel, - memoriesByTypeRowsReadModel, - memoryByIdReadModel, - memorySupersessionChainReadModel, - ) import Kioku.Prelude -import Kioku.Session.ReadModel - ( awaitingSessionsByCorrelationKeyReadModel, - sessionByIdReadModel, - sessionChainReadModel, - sessionDelegationChildrenReadModel, - sessionsByFocusReadModel, - sessionsByNamespaceReadModel, - sessionsByScopeReadModel, - sessionsByStartedRangeReadModel, - turnsBySessionReadModel, - ) +import Kioku.ProjectionCatalog (kiokuProjectionCatalog) import Kiroku.Store.Effect (Store) -- | The registry identity of a read model: its logical name plus the schema @@ -71,47 +50,22 @@ } deriving stock (Eq, Show) -schemaOf :: ReadModel q r -> ReadModelSchema -schemaOf rm = ReadModelSchema rm.name rm.version rm.shapeHash +schemaOf :: CatalogRegistration -> ReadModelSchema +schemaOf registration = + ReadModelSchema registration.registryName registration.version registration.shapeHash -- | Every Kioku read model paired with the schema identity the current code -- expects. Ordered session models first, then memory models. kiokuReadModelSchemas :: [ReadModelSchema] -kiokuReadModelSchemas = - [ schemaOf sessionByIdReadModel, - schemaOf sessionsByNamespaceReadModel, - schemaOf sessionsByScopeReadModel, - schemaOf sessionsByFocusReadModel, - schemaOf sessionsByStartedRangeReadModel, - schemaOf sessionChainReadModel, - schemaOf sessionDelegationChildrenReadModel, - schemaOf awaitingSessionsByCorrelationKeyReadModel, - schemaOf turnsBySessionReadModel, - schemaOf memoryByIdReadModel, - schemaOf memoriesByNamespaceReadModel, - schemaOf memoriesByNamespaceRowsReadModel, - schemaOf memoriesByScopeReadModel, - schemaOf memoriesByScopeRowsReadModel, - schemaOf memoriesBySessionReadModel, - schemaOf memoriesBySessionRowsReadModel, - schemaOf memoriesByTypeReadModel, - schemaOf memoriesByTypeRowsReadModel, - schemaOf memorySupersessionChainReadModel - ] +kiokuReadModelSchemas = schemaOf <$> catalogRegistrations kiokuProjectionCatalog --- | Register every Kioku read model at application startup. +-- | Register Kioku's validated projection catalog at application startup. -- --- Keiro 0.3 deliberately stopped registering models on their first query. This --- operation is idempotent and leaves an existing row unchanged, allowing Keiro --- to continue failing closed when its version or shape hash is stale. -registerKiokuReadModels :: (Store :> es) => Eff es () -registerKiokuReadModels = - forM_ kiokuReadModelSchemas \schema -> - void $ - Schema.registerReadModel - schema.readModelName - schema.readModelVersion - schema.readModelShapeHash +-- This registers the query identities, rebuild groups, and persisted catalog +-- fingerprint in one idempotent handshake. A conflicting fingerprint or stale +-- query identity is returned as a typed registration error. +registerKiokuReadModels :: (Store :> es) => Eff es (Either CatalogRegistrationError [GroupRebuildMetadata]) +registerKiokuReadModels = registerProjectionCatalog kiokuProjectionCatalog -- | What reconciliation did to one read model's registry row. data ReconcileOutcome diff --git a/src/Kioku/Recall.hs b/src/Kioku/Recall.hs --- a/src/Kioku/Recall.hs +++ b/src/Kioku/Recall.hs @@ -97,7 +97,7 @@ import Hasql.Encoders qualified as E import Hasql.Statement (Statement, preparable) import Hasql.Transaction qualified as Tx -import Keiro.ReadModel (ConsistencyMode (..), ReadModelError, runQueryWith) +import Keiro.ReadModel (QueryFreshness (Immediate), ReadModelError, runQueryWithFreshness) import Kioku.AI.Config import Kioku.AI.Runtime (AIRuntime, runtimeEmbeddingModel) import Kioku.Api.Access (MemoryAccessContext, MemorySpaceId, memoryContextSpace) @@ -1152,9 +1152,9 @@ MemoryScope -> Eff es (Either ReadModelError [MemoryRecord]) getActiveByScope space scope = - runQueryWith + runQueryWithFreshness Nothing - Eventual + Immediate memoriesByScopeReadModel MemoriesByScopeQuery { memorySpaceId = space, @@ -1171,9 +1171,9 @@ Namespace -> Eff es (Either ReadModelError [MemoryRecord]) getActiveInNamespace space (Namespace ns) = - runQueryWith + runQueryWithFreshness Nothing - Eventual + Immediate memoriesByNamespaceReadModel MemoriesByNamespaceQuery {memorySpaceId = space, namespace = ns} @@ -1194,9 +1194,9 @@ Eff es (Either ReadModelError (Maybe MemoryRecord)) getById space mid = fmap (fmap (fmap memoryRowToRecord)) $ - runQueryWith + runQueryWithFreshness Nothing - Eventual + Immediate memoryByIdReadModel MemoryByIdQuery {memorySpaceId = space, memoryId = idText mid} @@ -1206,9 +1206,9 @@ SessionId -> Eff es (Either ReadModelError [MemoryRecord]) getBySession space sid = - runQueryWith + runQueryWithFreshness Nothing - Eventual + Immediate memoriesBySessionReadModel MemoriesBySessionQuery {memorySpaceId = space, sessionId = idText sid} @@ -1219,9 +1219,9 @@ MemoryType -> Eff es (Either ReadModelError [MemoryRecord]) getByType space (Namespace ns) mt = - runQueryWith + runQueryWithFreshness Nothing - Eventual + Immediate memoriesByTypeReadModel MemoriesByTypeQuery {memorySpaceId = space, namespace = ns, memoryType = memoryTypeToText mt} diff --git a/src/Kioku/Session.hs b/src/Kioku/Session.hs --- a/src/Kioku/Session.hs +++ b/src/Kioku/Session.hs @@ -56,7 +56,8 @@ import Effectful.Error.Static (Error) import Keiro.Command (CommandError (..), defaultRunCommandOptions) import Keiro.Projection (runCommandWithProjections) -import Keiro.ReadModel (ConsistencyMode (..), ReadModelError, runQueryWith) +import Keiro.Projection.Catalog (typedInlineProjections) +import Keiro.ReadModel (QueryFreshness (Immediate), ReadModelError, runQueryWithFreshness) import Kioku.Api.Access ( MemoryAccessContext, MemoryPermission (..), @@ -72,6 +73,7 @@ import Kioku.Distill.Timer (l1TimerScheduleProjection) import Kioku.Id (SessionId, idText) import Kioku.Prelude +import Kioku.ProjectionCatalog (kiokuProjectionCatalog, sessionProjectionSet) import Kioku.Session.Domain import Kioku.Session.EventStream (sessionEventStream, sessionStream) import Kioku.Session.ReadModel @@ -90,7 +92,6 @@ sessionByIdReadModel, sessionChainReadModel, sessionDelegationChildrenReadModel, - sessionInlineProjection, sessionsByFocusReadModel, sessionsByNamespaceReadModel, sessionsByScopeReadModel, @@ -675,9 +676,9 @@ SessionId -> Eff es (Either ReadModelError (Maybe SessionRow)) getById space sid = - runQueryWith + runQueryWithFreshness Nothing - Eventual + Immediate sessionByIdReadModel SessionByIdQuery {memorySpaceId = space, sessionId = idText sid} @@ -688,9 +689,9 @@ Int -> Eff es (Either ReadModelError [SessionRow]) getRecentInNamespace space ns limit = - runQueryWith + runQueryWithFreshness Nothing - Eventual + Immediate sessionsByNamespaceReadModel SessionsByNamespaceQuery {memorySpaceId = space, namespace = namespaceText ns, limit} @@ -700,9 +701,9 @@ MemoryScope -> Eff es (Either ReadModelError [SessionRow]) getByScope space scope = - runQueryWith + runQueryWithFreshness Nothing - Eventual + Immediate sessionsByScopeReadModel SessionsByScopeQuery { memorySpaceId = space, @@ -718,9 +719,9 @@ Text -> Eff es (Either ReadModelError [SessionRow]) getByFocus space ns focus = - runQueryWith + runQueryWithFreshness Nothing - Eventual + Immediate sessionsByFocusReadModel SessionsByFocusQuery {memorySpaceId = space, namespace = namespaceText ns, focus} @@ -732,9 +733,9 @@ UTCTime -> Eff es (Either ReadModelError [SessionRow]) getByStartedRange space ns startedAfter startedBefore = - runQueryWith + runQueryWithFreshness Nothing - Eventual + Immediate sessionsByStartedRangeReadModel SessionsByStartedRangeQuery { memorySpaceId = space, @@ -749,9 +750,9 @@ SessionId -> Eff es (Either ReadModelError [SessionRow]) getChain space sid = - runQueryWith + runQueryWithFreshness Nothing - Eventual + Immediate sessionChainReadModel SessionChainQuery {memorySpaceId = space, sessionId = idText sid} @@ -761,9 +762,9 @@ SessionId -> Eff es (Either ReadModelError [SessionRow]) getDelegationChildren space sid = - runQueryWith + runQueryWithFreshness Nothing - Eventual + Immediate sessionDelegationChildrenReadModel SessionDelegationChildrenQuery {memorySpaceId = space, parentSessionId = idText sid} @@ -774,9 +775,9 @@ Text -> Eff es (Either ReadModelError [SessionRow]) getAwaitingByCorrelationKey space ns correlationKey = - runQueryWith + runQueryWithFreshness Nothing - Eventual + Immediate awaitingSessionsByCorrelationKeyReadModel AwaitingSessionsByCorrelationKeyQuery { memorySpaceId = space, @@ -793,9 +794,9 @@ SessionId -> Eff es (Either ReadModelError [TurnRow]) getTurns space sid = - runQueryWith + runQueryWithFreshness Nothing - Eventual + Immediate turnsBySessionReadModel TurnsBySessionQuery {memorySpaceId = space, sessionId = idText sid} @@ -805,13 +806,17 @@ SessionCommand -> Eff es (Either SessionWriteError SessionId) runSessionCommand sid cmd = do + -- Keiro's catalog cannot declare the shared framework-owned timer table as + -- a Kioku target. Derive Kioku's handlers from the catalog, keep the timer + -- callback explicit, and run both in the append transaction; see + -- docs/adr/catalog-application-projections-not-framework-timers.md. result <- runCommandWithProjections defaultRunCommandOptions sessionEventStream (sessionStream sid) cmd - [sessionInlineProjection, l1TimerScheduleProjection] + (typedInlineProjections kiokuProjectionCatalog sessionProjectionSet <> [l1TimerScheduleProjection]) pure $ case result of Left err -> Left (SessionCommandRejected err) diff --git a/src/Kioku/Session/EventStream.hs b/src/Kioku/Session/EventStream.hs --- a/src/Kioku/Session/EventStream.hs +++ b/src/Kioku/Session/EventStream.hs @@ -1,6 +1,7 @@ module Kioku.Session.EventStream ( SessionEventStream, sessionEventStream, + validateSessionEventStream, sessionCodec, sessionStream, parseSessionEvent, @@ -15,7 +16,7 @@ import Keiki.Generics (emptyRegFile) import Keiro.Codec (Codec (..), EventType (..)) import Keiro.EventStream (EventStream (..), SnapshotPolicy (..)) -import Keiro.EventStream.Validate (ValidatedEventStream, mkEventStreamOrThrow) +import Keiro.EventStream.Validate (EventStreamWarning, ValidatedEventStream, mkEventStream) import Keiro.Stream (Stream) import Keiro.Stream qualified as Stream import Kioku.Id (SessionId, idText) @@ -30,7 +31,11 @@ sessionEventStream :: ValidatedEventStream (HsPred SessionRegs SessionCommand) SessionRegs SessionVertex SessionCommand SessionEvent sessionEventStream = - mkEventStreamOrThrow "kioku-session" sessionEventStreamDefinition + either (error . ("invalid kioku session event stream: " <>) . show) id validateSessionEventStream + +-- | The explicit startup proof for the hand-written session stream. +validateSessionEventStream :: Either [EventStreamWarning] (ValidatedEventStream (HsPred SessionRegs SessionCommand) SessionRegs SessionVertex SessionCommand SessionEvent) +validateSessionEventStream = mkEventStream "kioku-session" sessionEventStreamDefinition sessionEventStreamDefinition :: SessionEventStream sessionEventStreamDefinition = diff --git a/src/Kioku/Session/ReadModel.hs b/src/Kioku/Session/ReadModel.hs --- a/src/Kioku/Session/ReadModel.hs +++ b/src/Kioku/Session/ReadModel.hs @@ -16,14 +16,23 @@ SessionDelegationChildrenQuery (..), AwaitingSessionsByCorrelationKeyQuery (..), TurnsBySessionQuery (..), + sessionByIdReadModelBlueprint, sessionByIdReadModel, + sessionsByNamespaceReadModelBlueprint, sessionsByNamespaceReadModel, + sessionsByScopeReadModelBlueprint, sessionsByScopeReadModel, + sessionsByFocusReadModelBlueprint, sessionsByFocusReadModel, + sessionsByStartedRangeReadModelBlueprint, sessionsByStartedRangeReadModel, + sessionChainReadModelBlueprint, sessionChainReadModel, + sessionDelegationChildrenReadModelBlueprint, sessionDelegationChildrenReadModel, + awaitingSessionsByCorrelationKeyReadModelBlueprint, awaitingSessionsByCorrelationKeyReadModel, + turnsBySessionReadModelBlueprint, turnsBySessionReadModel, ) where @@ -36,7 +45,12 @@ import Hasql.Statement (Statement, preparable) import Hasql.Transaction qualified as Tx import Keiro.Projection (InlineProjection (..)) -import Keiro.ReadModel (ConsistencyMode (..), ReadModel (..), StrongScope (..)) +import Keiro.ReadModel + ( QueryCursorAuthority (NoQueryCursor), + ReadModel, + ReadModelBlueprint (..), + immediateReadModel, + ) import Kioku.Api.Access (MemorySpaceId) import Kioku.Api.Scope (scopeKindText, scopeNamespaceText, scopeRefText) import Kioku.Database.Schema @@ -294,128 +308,137 @@ } sessionByIdReadModel :: ReadModel SessionByIdQuery (Maybe SessionRow) -sessionByIdReadModel = - ReadModel +sessionByIdReadModel = immediateReadModel sessionByIdReadModelBlueprint + +sessionByIdReadModelBlueprint :: ReadModelBlueprint SessionByIdQuery (Maybe SessionRow) +sessionByIdReadModelBlueprint = + ReadModelBlueprint { name = "kioku-session-by-id", schema = kiokuSchema, tableName = sessionsRelation, - subscriptionName = "kioku-session-inline", version = sessionReadModelVersion, shapeHash = sessionReadModelShapeHash, - defaultConsistency = Eventual, - strongScope = EntireLog, + cursorAuthority = NoQueryCursor, query = \q -> Tx.statement q selectSessionByIdStmt } sessionsByNamespaceReadModel :: ReadModel SessionsByNamespaceQuery [SessionRow] -sessionsByNamespaceReadModel = - ReadModel +sessionsByNamespaceReadModel = immediateReadModel sessionsByNamespaceReadModelBlueprint + +sessionsByNamespaceReadModelBlueprint :: ReadModelBlueprint SessionsByNamespaceQuery [SessionRow] +sessionsByNamespaceReadModelBlueprint = + ReadModelBlueprint { name = "kioku-sessions-by-namespace", schema = kiokuSchema, tableName = sessionsRelation, - subscriptionName = "kioku-session-inline", version = sessionReadModelVersion, shapeHash = sessionReadModelShapeHash, - defaultConsistency = Eventual, - strongScope = EntireLog, + cursorAuthority = NoQueryCursor, query = \q -> Tx.statement q selectSessionsByNamespaceStmt } sessionsByScopeReadModel :: ReadModel SessionsByScopeQuery [SessionRow] -sessionsByScopeReadModel = - ReadModel +sessionsByScopeReadModel = immediateReadModel sessionsByScopeReadModelBlueprint + +sessionsByScopeReadModelBlueprint :: ReadModelBlueprint SessionsByScopeQuery [SessionRow] +sessionsByScopeReadModelBlueprint = + ReadModelBlueprint { name = "kioku-sessions-by-scope", schema = kiokuSchema, tableName = sessionsRelation, - subscriptionName = "kioku-session-inline", version = sessionReadModelVersion, shapeHash = sessionReadModelShapeHash, - defaultConsistency = Eventual, - strongScope = EntireLog, + cursorAuthority = NoQueryCursor, query = \q -> Tx.statement q selectSessionsByScopeStmt } sessionsByFocusReadModel :: ReadModel SessionsByFocusQuery [SessionRow] -sessionsByFocusReadModel = - ReadModel +sessionsByFocusReadModel = immediateReadModel sessionsByFocusReadModelBlueprint + +sessionsByFocusReadModelBlueprint :: ReadModelBlueprint SessionsByFocusQuery [SessionRow] +sessionsByFocusReadModelBlueprint = + ReadModelBlueprint { name = "kioku-sessions-by-focus", schema = kiokuSchema, tableName = sessionsRelation, - subscriptionName = "kioku-session-inline", version = sessionReadModelVersion, shapeHash = sessionReadModelShapeHash, - defaultConsistency = Eventual, - strongScope = EntireLog, + cursorAuthority = NoQueryCursor, query = \q -> Tx.statement q selectSessionsByFocusStmt } sessionsByStartedRangeReadModel :: ReadModel SessionsByStartedRangeQuery [SessionRow] -sessionsByStartedRangeReadModel = - ReadModel +sessionsByStartedRangeReadModel = immediateReadModel sessionsByStartedRangeReadModelBlueprint + +sessionsByStartedRangeReadModelBlueprint :: ReadModelBlueprint SessionsByStartedRangeQuery [SessionRow] +sessionsByStartedRangeReadModelBlueprint = + ReadModelBlueprint { name = "kioku-sessions-by-started-range", schema = kiokuSchema, tableName = sessionsRelation, - subscriptionName = "kioku-session-inline", version = sessionReadModelVersion, shapeHash = sessionReadModelShapeHash, - defaultConsistency = Eventual, - strongScope = EntireLog, + cursorAuthority = NoQueryCursor, query = \q -> Tx.statement q selectSessionsByStartedRangeStmt } sessionChainReadModel :: ReadModel SessionChainQuery [SessionRow] -sessionChainReadModel = - ReadModel +sessionChainReadModel = immediateReadModel sessionChainReadModelBlueprint + +sessionChainReadModelBlueprint :: ReadModelBlueprint SessionChainQuery [SessionRow] +sessionChainReadModelBlueprint = + ReadModelBlueprint { name = "kioku-session-chain", schema = kiokuSchema, tableName = sessionsRelation, - subscriptionName = "kioku-session-inline", version = sessionReadModelVersion, shapeHash = sessionReadModelShapeHash, - defaultConsistency = Eventual, - strongScope = EntireLog, + cursorAuthority = NoQueryCursor, query = \q -> Tx.statement q selectSessionChainStmt } sessionDelegationChildrenReadModel :: ReadModel SessionDelegationChildrenQuery [SessionRow] -sessionDelegationChildrenReadModel = - ReadModel +sessionDelegationChildrenReadModel = immediateReadModel sessionDelegationChildrenReadModelBlueprint + +sessionDelegationChildrenReadModelBlueprint :: ReadModelBlueprint SessionDelegationChildrenQuery [SessionRow] +sessionDelegationChildrenReadModelBlueprint = + ReadModelBlueprint { name = "kioku-session-delegation-children", schema = kiokuSchema, tableName = sessionsRelation, - subscriptionName = "kioku-session-inline", version = sessionReadModelVersion, shapeHash = sessionReadModelShapeHash, - defaultConsistency = Eventual, - strongScope = EntireLog, + cursorAuthority = NoQueryCursor, query = \q -> Tx.statement q selectDelegationChildrenStmt } awaitingSessionsByCorrelationKeyReadModel :: ReadModel AwaitingSessionsByCorrelationKeyQuery [SessionRow] -awaitingSessionsByCorrelationKeyReadModel = - ReadModel +awaitingSessionsByCorrelationKeyReadModel = immediateReadModel awaitingSessionsByCorrelationKeyReadModelBlueprint + +awaitingSessionsByCorrelationKeyReadModelBlueprint :: ReadModelBlueprint AwaitingSessionsByCorrelationKeyQuery [SessionRow] +awaitingSessionsByCorrelationKeyReadModelBlueprint = + ReadModelBlueprint { name = "kioku-sessions-awaiting-by-correlation-key", schema = kiokuSchema, tableName = sessionsRelation, - subscriptionName = "kioku-session-inline", version = sessionReadModelVersion, shapeHash = sessionReadModelShapeHash, - defaultConsistency = Eventual, - strongScope = EntireLog, + cursorAuthority = NoQueryCursor, query = \q -> Tx.statement q selectAwaitingByCorrelationKeyStmt } turnsBySessionReadModel :: ReadModel TurnsBySessionQuery [TurnRow] -turnsBySessionReadModel = - ReadModel +turnsBySessionReadModel = immediateReadModel turnsBySessionReadModelBlueprint + +turnsBySessionReadModelBlueprint :: ReadModelBlueprint TurnsBySessionQuery [TurnRow] +turnsBySessionReadModelBlueprint = + ReadModelBlueprint { name = "kioku-turns-by-session", schema = kiokuSchema, tableName = turnsRelation, - subscriptionName = "kioku-session-inline", version = turnReadModelVersion, shapeHash = turnReadModelShapeHash, - defaultConsistency = Eventual, - strongScope = EntireLog, + cursorAuthority = NoQueryCursor, query = \q -> Tx.statement q selectTurnsBySessionStmt } diff --git a/test/Kioku/ProjectionCatalogSpec.hs b/test/Kioku/ProjectionCatalogSpec.hs new file mode 100644 --- /dev/null +++ b/test/Kioku/ProjectionCatalogSpec.hs @@ -0,0 +1,353 @@ +module Kioku.ProjectionCatalogSpec (tests) where + +import Data.Foldable (for_) +import Data.List.NonEmpty (NonEmpty (..)) +import Data.List.NonEmpty qualified as NonEmpty +import Data.Set qualified as Set +import Data.Time (getCurrentTime) +import Data.Vector qualified as Vector +import Effectful (liftIO) +import Keiro.Command + ( SqlCommandOutcome (..), + SqlTransactionDecision (..), + defaultRunCommandOptions, + runCommandWithSqlEventsControlled, + ) +import Keiro.Projection (InlineProjection (..)) +import Keiro.Projection.Catalog + ( CatalogDiagnostic (..), + CatalogDiagnosticCode (..), + CatalogInventory (..), + InventoryQueryCursor, + InventoryQueryFreshness (..), + InventoryQueryModel (..), + ProjectionCatalog (..), + ProjectionDefinition (..), + ProjectionSet (..), + SomeProjectionSet (..), + SourceDeclaration (..), + Validation (..), + catalogFingerprintText, + catalogRegistrations, + mkClaimSite, + mkProjectionId, + mkTargetId, + typedInlineProjections, + ) +import Keiro.ReadModel + ( HeadScope (EntireVisibleLog), + QueryFreshness (WaitForHead), + ReadModelDefinitionError (ReadModelDefinitionMissingCursor), + headWaitingReadModel, + ) +import Keiro.ReadModel.Rebuild (registerProjectionCatalog) +import Keiro.Stream qualified as Stream +import Keiro.Timer (lookupTimer) +import Kioku.Api.Scope (MemoryScope (..), Namespace (..)) +import Kioku.Api.Types (Confidence (..), MemoryType (..)) +import Kioku.App (runAppIO, withNoopAppEnv) +import Kioku.Distill.L2 (l2SceneTimerId, l2SceneTimerScheduleProjection) +import Kioku.Id (genMemoryId, genSessionId, idText) +import Kioku.Memory qualified as Memory +import Kioku.Memory.Domain (MemoryCommand (..), RecordMemoryData (..)) +import Kioku.Memory.EventStream (memoryEventStream, memoryStream, validateMemoryEventStream) +import Kioku.Memory.ReadModel (memoryByIdReadModelBlueprint) +import Kioku.Migrations.TestSupport (withKiokuMigratedDatabase) +import Kioku.ProjectionCatalog + ( kiokuCatalogFingerprint, + kiokuCatalogInventory, + kiokuProjectionCatalog, + memoryProjectionSet, + sessionProjectionSet, + validateKiokuProjectionCatalog, + validateKiokuProjectionCatalogWith, + ) +import Kioku.Session qualified as Session +import Kioku.Session.Domain (StartSessionData (..)) +import Kioku.Session.EventStream (validateSessionEventStream) +import Kioku.SpaceFixtures (testActorPrincipal, testContext, testSpace) +import Kiroku.Store.Connection (defaultConnectionSettings) +import Kiroku.Store.Read (readStreamForward) +import Kiroku.Store.Types (StreamVersion (..)) +import Test.Tasty (TestTree, testGroup) +import Test.Tasty.HUnit (Assertion, assertBool, assertEqual, assertFailure, testCase) + +tests :: TestTree +tests = + testGroup + "ProjectionCatalog" + [ testCase "the Kioku catalog has one stable complete inventory" testInventory, + testCase "typed projection sets select only application-owned handlers" testTypedHandlers, + testCase "both hand-written event streams validate explicitly" testEventStreams, + testCase "duplicate target ownership is rejected with a stable diagnostic" testDuplicateOwnership, + testCase "an unknown owned target is rejected with a stable diagnostic" testUnknownOwnership, + testCase "a target without its supplier is rejected at validation" testMissingSupplier, + testCase "a cursorless query model cannot be made waiting" testWaitingWithoutCursor, + testCase "registered catalog serves immediate memory and session queries" testImmediateQueries, + testCase "a late projection failure rolls back events, rows, and timers" testProjectionRollback, + testCase "persisted catalog fingerprint drift refuses registration" testFingerprintDrift + ] + +testInventory :: Assertion +testInventory = do + case validateKiokuProjectionCatalog of + Failure diagnostics -> assertFailure ("valid Kioku catalog was rejected: " <> show diagnostics) + Success _ -> pure () + assertEqual "sources" 2 (length kiokuCatalogInventory.inventorySources) + assertEqual "targets" 3 (length kiokuCatalogInventory.inventoryTargets) + assertEqual "groups" 2 (length kiokuCatalogInventory.inventoryGroups) + assertEqual "projection owners" 2 (length kiokuCatalogInventory.inventoryProjections) + assertEqual "query models" 19 (length kiokuCatalogInventory.inventoryQueryModels) + assertEqual "registrations" 19 (length (catalogRegistrations kiokuProjectionCatalog)) + assertBool + "every query is immediate and cursorless" + ( all + (\query -> query.freshness == InventoryImmediate && noCursor query.cursor) + kiokuCatalogInventory.inventoryQueryModels + ) + assertEqual + "canonical fingerprint" + "catalog-v7:54f9ac55d73f40160d87b0ae49c2477a0de860c2e6e1e38df0591a2036d002e4" + (catalogFingerprintText kiokuCatalogFingerprint) + where + noCursor :: Maybe InventoryQueryCursor -> Bool + noCursor = maybe True (const False) + +testTypedHandlers :: Assertion +testTypedHandlers = do + assertEqual + "memory handler" + ["kioku-memory-inline"] + (map (.name) (typedInlineProjections kiokuProjectionCatalog memoryProjectionSet)) + assertEqual + "session handler" + ["kioku-session-inline"] + (map (.name) (typedInlineProjections kiokuProjectionCatalog sessionProjectionSet)) + +testEventStreams :: Assertion +testEventStreams = do + assertBool "memory stream" (isRight validateMemoryEventStream) + assertBool "session stream" (isRight validateSessionEventStream) + +testDuplicateOwnership :: Assertion +testDuplicateOwnership = + diagnosticsFor addDuplicateMemoryOwner + `shouldContainCode` TargetWithMultipleOwners + where + addDuplicateMemoryOwner catalog = + catalog + { projectionSets = catalog.projectionSets <> [SomeProjectionSet duplicateMemoryProjectionSet] + } + duplicateMemoryProjectionSet = + memoryProjectionSet + { projectionDefinitions = duplicateDefinition :| [] + } + duplicateDefinition = + case memoryProjectionSet.projectionDefinitions of + definition :| _ -> + definition + { projectionId = must (mkProjectionId "kioku-memory-inline-duplicate"), + claimSite = must (mkClaimSite "duplicate memory owner test") + } + +testMissingSupplier :: Assertion +testMissingSupplier = + diagnosticsFor (\catalog -> catalog {projectionSets = []}) + `shouldContainCode` TargetWithoutOwner + +testUnknownOwnership :: Assertion +testUnknownOwnership = + diagnosticsFor addUnknownTarget + `shouldContainCode` UnknownTargetReference + where + addUnknownTarget catalog = + catalog + { projectionSets = + [ SomeProjectionSet + memoryProjectionSet + { projectionDefinitions = + addTarget <$> memoryProjectionSet.projectionDefinitions + } + ] + <> drop 1 catalog.projectionSets + } + addTarget definition = + definition + { ownedTargets = + definition.ownedTargets + <> (must (mkTargetId "kioku-unknown") :| []) + } + +testWaitingWithoutCursor :: Assertion +testWaitingWithoutCursor = + case headWaitingReadModel EntireVisibleLog memoryByIdReadModelBlueprint of + Left err -> + assertEqual + "truthful construction refuses waiting without a cursor" + (ReadModelDefinitionMissingCursor "kioku-memory-by-id" (WaitForHead EntireVisibleLog)) + err + Right _ -> assertFailure "a cursorless Kioku read model became head-waiting" + +testImmediateQueries :: Assertion +testImmediateQueries = + withKiokuMigratedDatabase \connStr -> + withNoopAppEnv (defaultConnectionSettings connStr) \env -> do + result <- runAppIO env do + now <- liftIO getCurrentTime + sid <- liftIO genSessionId + sessionWrite <- + Session.startWithContext + testContext + StartSessionData + { sessionId = sid, + memorySpaceId = testSpace, + actorPrincipal = testActorPrincipal, + ownerPrincipal = Nothing, + agentId = "catalog-test", + focus = "prove immediate catalog queries", + scope = testScope, + subjectRef = Nothing, + previousSessionId = Nothing, + parentSessionId = Nothing, + delegationDepth = 0, + startedAt = now + } + liftIO case sessionWrite of + Left err -> assertFailure ("session command failed: " <> show err) + Right _ -> pure () + sessionRead <- Session.getById testSpace sid + liftIO case sessionRead of + Right (Just _) -> pure () + other -> assertFailure ("immediate session query failed: " <> show (() <$ other)) + + mid <- liftIO genMemoryId + memoryWrite <- + Memory.recordWithContext + testContext + RecordMemoryData + { memoryId = mid, + memorySpaceId = testSpace, + actorPrincipal = testActorPrincipal, + ownerPrincipal = Nothing, + agentId = "catalog-test", + sessionId = Just sid, + scope = testScope, + memoryType = MemoryFact, + content = "catalog registration is live", + priority = 50, + confidence = HighConfidence, + tags = Set.singleton "catalog", + supersedes = Nothing, + recordedAt = now + } + liftIO case memoryWrite of + Left err -> assertFailure ("memory command failed: " <> show err) + Right _ -> pure () + memoryRead <- Memory.getMemoryRowById testSpace mid + liftIO case memoryRead of + Right (Just _) -> pure () + other -> assertFailure ("immediate memory query failed: " <> show (() <$ other)) + case result of + Left err -> assertFailure ("store error while exercising catalog queries: " <> show err) + Right () -> pure () + where + testScope :: MemoryScope + testScope = ScopeGlobal (Namespace "kioku-catalog-test") + +testProjectionRollback :: Assertion +testProjectionRollback = + withKiokuMigratedDatabase \connStr -> + withNoopAppEnv (defaultConnectionSettings connStr) \env -> do + mid <- genMemoryId + now <- getCurrentTime + let record = + RecordMemoryData + { memoryId = mid, + memorySpaceId = testSpace, + actorPrincipal = testActorPrincipal, + ownerPrincipal = Nothing, + agentId = "catalog-test", + sessionId = Nothing, + scope = testScope, + memoryType = MemoryFact, + content = "must roll back", + priority = 50, + confidence = HighConfidence, + tags = Set.singleton "rollback", + supersedes = Nothing, + recordedAt = now + } + handlers = + typedInlineProjections kiokuProjectionCatalog memoryProjectionSet + <> [l2SceneTimerScheduleProjection] + failed <- + runAppIO env $ + runCommandWithSqlEventsControlled + defaultRunCommandOptions + memoryEventStream + (memoryStream mid) + (RecordMemory record) + ( \pairs _ -> do + for_ handlers \handler -> + for_ pairs \(event, recorded) -> handler.apply event recorded + pure (RollbackSqlTransaction ()) + ) + case failed of + Left err -> assertFailure ("store error while injecting rollback: " <> show err) + Right (Left err) -> assertFailure ("command failed before the injected rollback: " <> show err) + Right (Right (SqlCommandRolledBack ())) -> pure () + Right (Right other) -> assertFailure ("injected rollback committed: " <> show other) + + probe <- runAppIO env do + events <- readStreamForward (Stream.streamName (memoryStream mid)) (StreamVersion 0) 10 + row <- Memory.getMemoryRowById testSpace mid + timer <- lookupTimer (l2SceneTimerId testSpace testScope (idText mid)) + pure (Vector.length events, row, timer) + case probe of + Left err -> assertFailure ("store error while checking rollback: " <> show err) + Right (eventCount, row, timer) -> do + assertEqual "event append rolled back" 0 eventCount + assertEqual "application projection rolled back" (Right Nothing) row + assertEqual "framework timer rolled back" Nothing timer + where + testScope :: MemoryScope + testScope = ScopeGlobal (Namespace "kioku-catalog-rollback") + +testFingerprintDrift :: Assertion +testFingerprintDrift = + withKiokuMigratedDatabase \connStr -> + withNoopAppEnv (defaultConnectionSettings connStr) \env -> + case validateKiokuProjectionCatalogWith driftSources of + Failure diagnostics -> assertFailure ("drift fixture was structurally invalid: " <> show diagnostics) + Success drifted -> do + result <- runAppIO env (registerProjectionCatalog drifted) + case result of + Left err -> assertFailure ("store error while checking catalog drift: " <> show err) + Right (Left _) -> pure () + Right (Right _) -> assertFailure "registration accepted a changed persisted catalog fingerprint" + where + driftSources catalog = + catalog + { sources = + [ source {codecFingerprint = source.codecFingerprint <> "-drift"} + | source <- catalog.sources + ] + } + +diagnosticsFor :: (ProjectionCatalog -> ProjectionCatalog) -> [CatalogDiagnostic] +diagnosticsFor transform = + case validateKiokuProjectionCatalogWith transform of + Failure diagnostics -> NonEmpty.toList diagnostics + Success _ -> [] + +shouldContainCode :: [CatalogDiagnostic] -> CatalogDiagnosticCode -> Assertion +shouldContainCode diagnostics expected = + assertBool + ("expected diagnostic " <> show expected <> ", got " <> show diagnostics) + (any ((== expected) . (.diagnosticCode)) diagnostics) + +isRight :: Either left right -> Bool +isRight = either (const False) (const True) + +must :: (Show error) => Either error value -> value +must = either (error . show) id diff --git a/test/Kioku/ReadModelReconcileSpec.hs b/test/Kioku/ReadModelReconcileSpec.hs --- a/test/Kioku/ReadModelReconcileSpec.hs +++ b/test/Kioku/ReadModelReconcileSpec.hs @@ -26,10 +26,12 @@ import Effectful.Error.Static (Error) import Hasql.Transaction qualified as Tx import Keiro.ReadModel - ( ConsistencyMode (Eventual), - ReadModel (..), + ( QueryFreshness (Immediate), + ReadModel, + ReadModelBlueprint (..), ReadModelError (..), - runQueryWith, + immediateReadModel, + runQueryWithFreshness, ) import Kioku.Api.Scope (MemoryScope (..), Namespace (..)) import Kioku.App (AppEffects, runAppIO, withNoopAppEnv) @@ -138,9 +140,9 @@ withApp \sid -> do _ <- reconcileReadModelRegistry result <- - runQueryWith + runQueryWithFreshness Nothing - Eventual + Immediate preRelocationSessionByIdReadModel Session.SessionByIdQuery {memorySpaceId = testSpace, sessionId = idText sid} liftIO case result of @@ -156,7 +158,12 @@ -- the identity it carried before the projections moved. preRelocationSessionByIdReadModel :: ReadModel Session.SessionByIdQuery (Maybe Session.SessionRow) preRelocationSessionByIdReadModel = - Session.sessionByIdReadModel {version = 4, shapeHash = "kioku-session-v4"} + immediateReadModel + ( Session.sessionByIdReadModelBlueprint + { version = 4, + shapeHash = "kioku-session-v4" + } + ) -- | A second pass must write nothing. If it reported 'Reconciled' again, the reconciler -- would be rewriting @last_built_at@ on every @just migrate@ — and, worse, would be lying diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -8,6 +8,7 @@ import Kioku.IdempotencySpec qualified as IdempotencySpec import Kioku.MemorySpaceSpec qualified as MemorySpaceSpec import Kioku.PortfolioAccessSpec qualified as PortfolioAccessSpec +import Kioku.ProjectionCatalogSpec qualified as ProjectionCatalogSpec import Kioku.ReadModelReconcileSpec qualified as ReadModelReconcileSpec import Kioku.RecallCompatSpec qualified as RecallCompatSpec import Kioku.RecallSpec qualified as RecallSpec @@ -33,6 +34,7 @@ IdempotencySpec.tests, MemorySpaceSpec.tests, PortfolioAccessSpec.tests, + ProjectionCatalogSpec.tests, ReadModelReconcileSpec.tests, RecallSpec.tests, RecallCompatSpec.tests,