kiroku-store-migrations (empty) → 0.1.1.0

raw patch · 8 files changed

Files

CHANGELOG.md +24/−0
README.md +40/−0
app/Main.hs +11/−0
kiroku-store-migrations.cabal +77/−0
sql-migrations/2026-05-16-00-00-00-kiroku-bootstrap.sql +229/−0
sql-migrations/2026-05-26-00-00-00-add-subscription-dead-letters.sql +31/−0
src/Kiroku/Store/Migrations.hs +56/−0
test/Main.hs +229/−0

+ CHANGELOG.md view

@@ -0,0 +1,24 @@+# Changelog++## Unreleased++## 0.1.1.0 — 2026-05-31++### New Features++* Forward migration `2026-05-26-00-00-00-add-subscription-dead-letters.sql`:+  creates the `kiroku.dead_letters` table (per consumer-group member, with a+  foreign key to `kiroku.events`) and its recency index, supporting per-event+  dead-letter recording for subscriptions (MasterPlan 6 / plan 40).++## 0.1.0.0 — 2026-05-23++### New Features++* Initial release of the migration package.+* Embeds Kiroku's codd SQL migrations and exposes+  `Kiroku.Store.Migrations`.+* Provides the `kiroku-store-migrate` executable for applying the event store+  schema before application startup.+* Bootstraps the dedicated `kiroku` PostgreSQL schema and installs Kiroku+  tables, indexes, functions, and triggers there.

+ README.md view

@@ -0,0 +1,40 @@+# kiroku-store-migrations++`kiroku-store-migrations` owns schema evolution for `kiroku-store`.+It embeds Kiroku's timestamped SQL migrations and runs them through+`codd`, so a service can migrate its database before opening+`Kiroku.Store.withStore`.++Run the executable with codd's standard environment variables:++```bash+CODD_CONNECTION='host=/tmp port=5432 dbname=kiroku user=kiroku_admin' \+CODD_MIGRATION_DIRS=unused-for-embedded-migrations \+CODD_EXPECTED_SCHEMA_DIR=unused-for-unverified-embedded-migrations \+CODD_SCHEMAS=kiroku \+kiroku-store-migrate+```++The bootstrap migration creates a dedicated `kiroku` schema and installs+every Kiroku table, index, function, and trigger inside it, leaving+`public` free for application objects. `CODD_SCHEMAS=kiroku` tells codd to+track that schema. The runtime role therefore needs privileges on the+`kiroku` schema (for example `USAGE` plus the table privileges it uses),+not on `public`.++After migrations run, start the application normally:++```haskell+withStore (defaultConnectionSettings connString) app+```++`codd` is forward-only. Once a migration has run in production,+reverting the Haskell package does not undo the database change. Repair+state by restoring from backup or by shipping another forward migration.++This first implementation runs without codd expected-schema verification+because Kiroku does not yet ship a checked-in codd expected-schema+snapshot. `CODD_EXPECTED_SCHEMA_DIR` is still required by codd's settings+parser, but this executable does not read from it. Operators should treat+the migration table as the source of applied-version truth until strict+snapshots are added.

+ app/Main.hs view

@@ -0,0 +1,11 @@+module Main where++import Codd.Environment (getCoddSettings)+import Data.Time (secondsToDiffTime)+import Kiroku.Store.Migrations (runKirokuMigrationsNoCheck)++main :: IO ()+main = do+    settings <- getCoddSettings+    _ <- runKirokuMigrationsNoCheck settings (secondsToDiffTime 5)+    pure ()

+ kiroku-store-migrations.cabal view

@@ -0,0 +1,77 @@+cabal-version:      3.0+name:               kiroku-store-migrations+version:            0.1.1.0+synopsis:           Schema migrations for kiroku-store+description:+  Embedded codd migrations and a migration executable for installing and+  upgrading the PostgreSQL schema used by @kiroku-store@.++homepage:           https://github.com/shinzui/kiroku+bug-reports:        https://github.com/shinzui/kiroku/issues+author:             Nadeem Bitar+maintainer:         nadeem@gmail.com+license:            BSD-3-Clause+build-type:         Simple+category:           Database, Eventing+extra-doc-files:    CHANGELOG.md+extra-source-files:+  README.md+  sql-migrations/*.sql++source-repository head+  type:     git+  location: https://github.com/shinzui/kiroku.git++common common+  default-language:   GHC2024+  default-extensions:+    DeriveAnyClass+    DuplicateRecordFields+    OverloadedLabels+    OverloadedStrings++library+  import:          common+  exposed-modules: Kiroku.Store.Migrations+  hs-source-dirs:  src+  build-depends:+    , base        >=4.18   && <5+    , bytestring  >=0.11   && <0.13+    , codd        >=0.1.8  && <0.2+    , file-embed  >=0.0.15 && <0.0.17+    , streaming   >=0.2    && <0.3+    , text        >=2.0    && <2.2+    , time        >=1.12   && <1.15++executable kiroku-store-migrate+  import:         common+  main-is:        Main.hs+  hs-source-dirs: app+  ghc-options:    -threaded -rtsopts -with-rtsopts=-N+  build-depends:+    , base+    , codd+    , kiroku-store-migrations+    , time                     >=1.12 && <1.15++test-suite kiroku-store-migrations-test+  import:         common+  type:           exitcode-stdio-1.0+  main-is:        Main.hs+  hs-source-dirs: test+  ghc-options:    -threaded -rtsopts -with-rtsopts=-N+  build-depends:+    , aeson                    >=2.1  && <2.3+    , attoparsec+    , base                     >=4.18 && <5+    , codd+    , containers               >=0.6  && <0.8+    , ephemeral-pg             >=0.2  && <0.3+    , hasql                    >=1.10 && <1.11+    , hasql-pool               >=1.2  && <1.5+    , hspec                    >=2.10 && <2.12+    , kiroku-store             ^>=0.2+    , kiroku-store-migrations+    , text                     >=2.0  && <2.2+    , time                     >=1.12 && <1.15+    , vector                   >=0.13 && <0.14

+ sql-migrations/2026-05-16-00-00-00-kiroku-bootstrap.sql view

@@ -0,0 +1,229 @@+-- Kiroku Store bootstrap migration (codd)+-- Supports PostgreSQL 17+.+--+-- kiroku-store-migrations is the schema owner. codd applies this file+-- verbatim, records that the timestamped migration ran, and skips it on later+-- runs. Future schema changes add new timestamped SQL files in this directory+-- instead of changing kiroku-store.+--+-- All Kiroku-owned objects live in the dedicated `kiroku` schema, leaving+-- `public` free for application objects. Creating the schema and setting+-- search_path first means every unqualified object name in the rest of this+-- file resolves into the Kiroku schema.+CREATE SCHEMA IF NOT EXISTS kiroku;+SET search_path TO kiroku, pg_catalog;++-- PostgreSQL 18 provides pg_catalog.uuidv7(); PostgreSQL 17 needs this+-- Kiroku-schema fallback before events.event_id DEFAULT uuidv7() is parsed.+-- With search_path set above, the unqualified CREATE FUNCTION lands in the+-- Kiroku schema, and to_regprocedure('uuidv7()') resolves through search_path+-- (pg_catalog first for the built-in, then the Kiroku schema for the fallback).+DO $$+BEGIN+    IF to_regprocedure('pg_catalog.uuidv7()') IS NULL+       AND to_regprocedure('uuidv7()') IS NULL THEN+        EXECUTE $fn$+            CREATE FUNCTION uuidv7()+            RETURNS uuid+            AS $body$+            DECLARE+                unix_ts_ms bytea;+                uuid_bytes bytea;+            BEGIN+                unix_ts_ms = substring(int8send(floor(extract(epoch from clock_timestamp()) * 1000)::bigint) from 3);+                uuid_bytes = uuid_send(gen_random_uuid());+                uuid_bytes = overlay(uuid_bytes placing unix_ts_ms from 1 for 6);+                uuid_bytes = set_byte(uuid_bytes, 6, (b'0111' || get_byte(uuid_bytes, 6)::bit(4))::bit(8)::int);+                RETURN encode(uuid_bytes, 'hex')::uuid;+            END+            $body$+            LANGUAGE plpgsql+            VOLATILE+        $fn$;+    END IF;+END+$$;++-- Streams (including $all as stream_id = 0)+CREATE TABLE IF NOT EXISTS streams (+    stream_id    BIGSERIAL    PRIMARY KEY,+    stream_name  TEXT         NOT NULL,+    category     TEXT         GENERATED ALWAYS AS (split_part(stream_name, '-', 1)) STORED,+    stream_version BIGINT     NOT NULL DEFAULT 0,+    created_at   TIMESTAMPTZ  NOT NULL DEFAULT now(),+    deleted_at   TIMESTAMPTZ,+    CONSTRAINT ix_streams_stream_name UNIQUE (stream_name)+);++-- Seed the $all stream+INSERT INTO streams (stream_id, stream_name, stream_version)+VALUES (0, '$all', 0)+ON CONFLICT DO NOTHING;++-- Reset sequence past the reserved stream_id=0+SELECT setval('streams_stream_id_seq', GREATEST((SELECT MAX(stream_id) FROM streams), 1));++-- Events (flat table — stream membership tracked in stream_events)+CREATE TABLE IF NOT EXISTS events (+    event_id       UUID         PRIMARY KEY DEFAULT uuidv7(),+    event_type     TEXT         NOT NULL,+    causation_id   UUID,+    correlation_id UUID,+    data           JSONB        NOT NULL,+    metadata       JSONB,+    created_at     TIMESTAMPTZ  NOT NULL DEFAULT now()+);++-- Stream-event junction (each event gets 2+ rows: source stream + $all + any links)+CREATE TABLE IF NOT EXISTS stream_events (+    event_id                UUID   NOT NULL REFERENCES events(event_id),+    stream_id               BIGINT NOT NULL REFERENCES streams(stream_id),+    stream_version          BIGINT NOT NULL,+    original_stream_id      BIGINT NOT NULL,+    original_stream_version BIGINT NOT NULL,+    PRIMARY KEY (event_id, stream_id)+);++-- Indexes++-- Primary read path: fetch events from a stream in order+CREATE INDEX IF NOT EXISTS ix_stream_events_stream_version+    ON stream_events (stream_id, stream_version);++-- Event type filtering (for server-side subscription filtering)+CREATE INDEX IF NOT EXISTS ix_events_event_type+    ON events (event_type);++-- Correlation tracing+CREATE INDEX IF NOT EXISTS ix_events_correlation_id+    ON events (correlation_id) WHERE correlation_id IS NOT NULL;++-- Causation tracing+CREATE INDEX IF NOT EXISTS ix_events_causation_id+    ON events (causation_id) WHERE causation_id IS NOT NULL;++-- Category filtering (for readCategory — uses generated column, not LIKE)+CREATE INDEX IF NOT EXISTS ix_streams_category+    ON streams (category);++-- Category read path: find $all entries by originating stream, ordered by global position+-- Enables efficient category reads by allowing the planner to: look up category stream_ids →+-- index scan $all for each → merge ordered by stream_version+CREATE INDEX IF NOT EXISTS ix_stream_events_all_by_origin+    ON stream_events (original_stream_id, stream_version)+    WHERE stream_id = 0;++-- Subscriptions (checkpoint persistence for subscription positions).+-- consumer_group_member / consumer_group_size carry static consumer-group+-- topology (ExecPlan 28 / EP-1). Non-group subscriptions are member 0, size 1.+-- The unique key is composite (subscription_name, consumer_group_member) so each+-- group member persists its own checkpoint under one shared subscription name.+CREATE TABLE IF NOT EXISTS subscriptions (+    subscription_id       BIGSERIAL    PRIMARY KEY,+    subscription_name     TEXT         NOT NULL,+    stream_name           TEXT         NOT NULL DEFAULT '$all',+    last_seen             BIGINT       NOT NULL DEFAULT 0,+    consumer_group_member INT          NOT NULL DEFAULT 0,+    consumer_group_size   INT          NOT NULL DEFAULT 1,+    created_at            TIMESTAMPTZ  NOT NULL DEFAULT now(),+    updated_at            TIMESTAMPTZ  NOT NULL DEFAULT now()+);++-- Idempotent convergence for databases created before EP-1: add the columns if+-- missing, drop the old auto-named single-column unique constraint if present,+-- and install the composite unique index. All guarded so the bootstrap body is+-- still safe in disposable local databases; codd records the timestamped+-- migration and does not reapply it after a successful run.+ALTER TABLE subscriptions ADD COLUMN IF NOT EXISTS consumer_group_member INT NOT NULL DEFAULT 0;+ALTER TABLE subscriptions ADD COLUMN IF NOT EXISTS consumer_group_size   INT NOT NULL DEFAULT 1;+ALTER TABLE subscriptions DROP CONSTRAINT IF EXISTS subscriptions_subscription_name_key;+CREATE UNIQUE INDEX IF NOT EXISTS ix_subscriptions_name_member+    ON subscriptions (subscription_name, consumer_group_member);++-- Triggers++-- NOTIFY on stream changes (fires once per append, not per event)+CREATE OR REPLACE FUNCTION notify_events() RETURNS TRIGGER AS $$+BEGIN+    PERFORM pg_notify(+        TG_TABLE_SCHEMA || '.events',+        NEW.stream_name || ',' || NEW.stream_id || ',' || NEW.stream_version+    );+    RETURN NEW;+END;+$$ LANGUAGE plpgsql;++DROP TRIGGER IF EXISTS stream_events_notify ON streams;+CREATE TRIGGER stream_events_notify+    AFTER INSERT OR UPDATE ON streams+    FOR EACH ROW EXECUTE FUNCTION notify_events();++-- Immutability: prevent event mutation+CREATE OR REPLACE FUNCTION prevent_mutation() RETURNS TRIGGER AS $$+BEGIN+    RAISE EXCEPTION 'Immutable table: % cannot be updated', TG_TABLE_NAME;+END;+$$ LANGUAGE plpgsql;++DROP TRIGGER IF EXISTS no_update_events ON events;+CREATE TRIGGER no_update_events+    BEFORE UPDATE ON events+    FOR EACH ROW EXECUTE FUNCTION prevent_mutation();++DROP TRIGGER IF EXISTS no_update_stream_events ON stream_events;+CREATE TRIGGER no_update_stream_events+    BEFORE UPDATE ON stream_events+    FOR EACH ROW EXECUTE FUNCTION prevent_mutation();++-- Gated hard deletes (for maintenance/GDPR only)+CREATE OR REPLACE FUNCTION protect_deletion() RETURNS TRIGGER AS $$+BEGIN+    IF current_setting('kiroku.enable_hard_deletes', true) = 'on' THEN+        RETURN OLD;+    END IF;+    RAISE EXCEPTION 'Hard deletes require: SET LOCAL kiroku.enable_hard_deletes = ''on''';+END;+$$ LANGUAGE plpgsql;++DROP TRIGGER IF EXISTS no_delete_events ON events;+CREATE TRIGGER no_delete_events+    BEFORE DELETE ON events+    FOR EACH ROW EXECUTE FUNCTION protect_deletion();++DROP TRIGGER IF EXISTS no_delete_stream_events ON stream_events;+CREATE TRIGGER no_delete_stream_events+    BEFORE DELETE ON stream_events+    FOR EACH ROW EXECUTE FUNCTION protect_deletion();++DROP TRIGGER IF EXISTS no_delete_streams ON streams;+CREATE TRIGGER no_delete_streams+    BEFORE DELETE ON streams+    FOR EACH ROW EXECUTE FUNCTION protect_deletion();++-- TRUNCATE bypasses row-level triggers, so the BEFORE DELETE triggers above+-- do not protect against an operator running TRUNCATE on these tables. Add+-- statement-level BEFORE TRUNCATE triggers gated by the same GUC so the+-- protection is symmetric. See EP-1 F6.+CREATE OR REPLACE FUNCTION protect_truncation() RETURNS TRIGGER AS $$+BEGIN+    IF current_setting('kiroku.enable_hard_deletes', true) = 'on' THEN+        RETURN NULL;+    END IF;+    RAISE EXCEPTION 'TRUNCATE requires: SET LOCAL kiroku.enable_hard_deletes = ''on''';+END;+$$ LANGUAGE plpgsql;++DROP TRIGGER IF EXISTS no_truncate_events ON events;+CREATE TRIGGER no_truncate_events+    BEFORE TRUNCATE ON events+    FOR EACH STATEMENT EXECUTE FUNCTION protect_truncation();++DROP TRIGGER IF EXISTS no_truncate_stream_events ON stream_events;+CREATE TRIGGER no_truncate_stream_events+    BEFORE TRUNCATE ON stream_events+    FOR EACH STATEMENT EXECUTE FUNCTION protect_truncation();++DROP TRIGGER IF EXISTS no_truncate_streams ON streams;+CREATE TRIGGER no_truncate_streams+    BEFORE TRUNCATE ON streams+    FOR EACH STATEMENT EXECUTE FUNCTION protect_truncation();

+ sql-migrations/2026-05-26-00-00-00-add-subscription-dead-letters.sql view

@@ -0,0 +1,31 @@+-- Add the kiroku.dead_letters table (MasterPlan 6 / EP-2, docs/plans/40-...).+--+-- Forward, additive migration: codd applies this file once, records it, and+-- skips it on later runs. It does not edit the bootstrap migration and does not+-- mutate existing event data.+--+-- A dead-letter row records one event that a subscription handler asked to+-- "dead-letter" (return DeadLetter, or exhaust its bounded retry budget). The+-- event itself stays immutable in kiroku.events; this table references it by+-- event_id and global_position rather than copying the payload. The+-- consumer_group_member column (default 0, matching kiroku.subscriptions)+-- attributes the row to the member that produced it, so a consumer group's dead+-- letters are per-member. The worker writes a row and advances the member's+-- checkpoint in one atomic statement (see SQL.insertDeadLetterAndCheckpointStmt).++CREATE TABLE IF NOT EXISTS kiroku.dead_letters (+    dead_letter_id        BIGSERIAL    PRIMARY KEY,+    subscription_name     TEXT         NOT NULL,+    consumer_group_member INT          NOT NULL DEFAULT 0,+    global_position       BIGINT       NOT NULL,+    event_id              UUID         NOT NULL REFERENCES kiroku.events(event_id),+    reason                JSONB        NOT NULL,+    reason_summary        TEXT         NOT NULL,+    attempt_count         INT          NOT NULL,+    created_at            TIMESTAMPTZ  NOT NULL DEFAULT now(),+    UNIQUE (subscription_name, consumer_group_member, global_position, event_id)+);++-- Operator read path: list a subscription member's dead letters by recency.+CREATE INDEX IF NOT EXISTS ix_dead_letters_subscription_created_at+    ON kiroku.dead_letters (subscription_name, consumer_group_member, created_at);

+ src/Kiroku/Store/Migrations.hs view

@@ -0,0 +1,56 @@+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TemplateHaskell #-}++module Kiroku.Store.Migrations (+    kirokuMigrations,+    runKirokuMigrations,+    runKirokuMigrationsNoCheck,+) where++import Codd (ApplyResult (SchemasNotVerified), CoddSettings, VerifySchemas, applyMigrations, applyMigrationsNoCheck)+import Codd.Logging (runCoddLogger)+import Codd.Parsing (AddedSqlMigration, EnvVars, PureStream (..), parseAddedSqlMigration)+import Data.ByteString (ByteString)+import Data.FileEmbed (embedDir)+import Data.Text.Encoding qualified as TE+import Data.Time (DiffTime)+import Streaming.Prelude qualified as Streaming++-- | Kiroku's embedded SQL migrations, ordered by timestamped filename.+kirokuMigrations :: (MonadFail m, EnvVars m) => m [AddedSqlMigration m]+kirokuMigrations =+    traverse parseEmbeddedMigration embeddedMigrationFiles+  where+    parseEmbeddedMigration :: forall m. (MonadFail m, EnvVars m) => (FilePath, ByteString) -> m (AddedSqlMigration m)+    parseEmbeddedMigration (name, bytes) = do+        let stream :: PureStream m+            stream = PureStream $ Streaming.yield (TE.decodeUtf8 bytes)+        result <-+            parseAddedSqlMigration+                name+                stream+        case result of+            Left err -> fail ("Invalid Kiroku migration " <> name <> ": " <> err)+            Right migration -> pure migration++-- | Run Kiroku's embedded migrations through codd.+runKirokuMigrations :: CoddSettings -> DiffTime -> VerifySchemas -> IO ApplyResult+runKirokuMigrations settings connectTimeout verifySchemas =+    runCoddLogger $ do+        migrations <- kirokuMigrations+        applyMigrations settings (Just migrations) connectTimeout verifySchemas++{- | Run Kiroku's embedded migrations through codd without expected-schema+verification.++This is the right entry point until the caller owns a codd expected-schema+snapshot. Use 'runKirokuMigrations' when schema verification is configured.+-}+runKirokuMigrationsNoCheck :: CoddSettings -> DiffTime -> IO ApplyResult+runKirokuMigrationsNoCheck settings connectTimeout =+    runCoddLogger $ do+        migrations <- kirokuMigrations+        applyMigrationsNoCheck settings (Just migrations) connectTimeout (const (pure SchemasNotVerified))++embeddedMigrationFiles :: [(FilePath, ByteString)]+embeddedMigrationFiles = $(embedDir "sql-migrations")

+ test/Main.hs view

@@ -0,0 +1,229 @@+module Main where++import Codd (ApplyResult (SchemasNotVerified), CoddSettings (..))+import Codd.Parsing (connStringParser)+import Codd.Representations.Types (DbRep (..))+import Codd.Types (ConnectionString, SchemaAlgo (..), SchemaSelection (..), SqlSchema (..), TxnIsolationLvl (..), singleTryPolicy)+import Data.Aeson (Value (Null))+import Data.Aeson qualified as Aeson+import Data.Attoparsec.Text (endOfInput, parseOnly)+import Data.Map qualified as Map+import Data.Text (Text)+import Data.Text qualified as T+import Data.Time (secondsToDiffTime)+import Data.Vector qualified as Vector+import EphemeralPg qualified as Pg+import Hasql.Connection.Settings qualified as Conn+import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Pool qualified as Pool+import Hasql.Pool.Config qualified as Pool.Config+import Hasql.Session qualified as Session+import Hasql.Statement (Statement, preparable)+import Kiroku.Store+import Kiroku.Store.Migrations (runKirokuMigrationsNoCheck)+import Test.Hspec++main :: IO ()+main =+    hspec $+        describe "codd migration spike" $+            it "applies Kiroku migrations, opens the store without startup DDL, and is repeatable" $ do+                result <- Pg.withCached $ \db -> do+                    let connStr = Pg.connectionString db+                        coddSettings = testCoddSettings connStr++                    firstMigration <- runKirokuMigrationsNoCheck coddSettings (secondsToDiffTime 5)+                    firstMigration `shouldBeSchemasNotVerified` "first migration run"+                    assertBootstrapApplied connStr+                    assertSchemaPlacement connStr+                    assertDeadLettersTable connStr+                    assertDefaultUuidV7 connStr++                    withStore+                        (defaultConnectionSettings connStr)+                        $ \store -> do+                            let stream = StreamName "migration-consumer"+                                event = makeEvent "MigrationConsumerChecked" (Aeson.object [("ok", Aeson.Bool True)])+                            appendResult <- runStoreIO store $ appendToStream stream NoStream [event]+                            case appendResult of+                                Left err -> expectationFailure ("appendToStream failed: " <> show err)+                                Right _ -> pure ()++                            readResult <- runStoreIO store $ readStreamForward stream (StreamVersion 0) 10+                            case readResult of+                                Left err -> expectationFailure ("readStreamForward failed: " <> show err)+                                Right events -> Vector.length events `shouldBe` 1++                    secondMigration <- runKirokuMigrationsNoCheck coddSettings (secondsToDiffTime 5)+                    secondMigration `shouldBeSchemasNotVerified` "second migration run"+                    assertBootstrapApplied connStr+                    assertDefaultUuidV7 connStr+                case result of+                    Left err -> expectationFailure ("Failed to start ephemeral PostgreSQL: " <> show err)+                    Right () -> pure ()++testCoddSettings :: Text -> CoddSettings+testCoddSettings connStr =+    CoddSettings+        { migsConnString = parseConnString connStr+        , sqlMigrations = []+        , onDiskReps = Right (DbRep Null Map.empty Map.empty)+        , namespacesToCheck = IncludeSchemas [SqlSchema "kiroku"]+        , extraRolesToCheck = []+        , retryPolicy = singleTryPolicy+        , txnIsolationLvl = DbDefault+        , schemaAlgoOpts = SchemaAlgo False False False+        }++parseConnString :: Text -> ConnectionString+parseConnString connStr =+    case parseOnly (connStringParser <* endOfInput) connStr of+        Left err -> error ("Could not parse ephemeral PostgreSQL connection string for codd: " <> err)+        Right parsed -> parsed++makeEvent :: Text -> Value -> EventData+makeEvent typ payload =+    EventData+        { eventId = Nothing+        , eventType = EventType typ+        , payload = payload+        , metadata = Nothing+        , causationId = Nothing+        , correlationId = Nothing+        }++shouldBeSchemasNotVerified :: ApplyResult -> String -> Expectation+shouldBeSchemasNotVerified SchemasNotVerified _ = pure ()+shouldBeSchemasNotVerified _ label = expectationFailure (label <> " unexpectedly verified schemas")++assertBootstrapApplied :: Text -> IO ()+assertBootstrapApplied connStr = do+    pool <- Pool.acquire poolConfig+    result <- Pool.use pool (Session.statement () bootstrapStmt)+    Pool.release pool+    case result of+        Left err -> expectationFailure ("bootstrap verification query failed: " <> show err)+        Right True -> pure ()+        Right False -> expectationFailure "Kiroku bootstrap migration did not create the $all stream"+  where+    poolConfig =+        Pool.Config.settings+            [ Pool.Config.staticConnectionSettings (Conn.connectionString connStr)+            , Pool.Config.size 1+            ]++bootstrapStmt :: Statement () Bool+bootstrapStmt =+    preparable+        "SELECT EXISTS (SELECT 1 FROM kiroku.streams WHERE stream_id = 0 AND stream_name = '$all')"+        E.noParams+        (D.singleRow (D.column (D.nonNullable D.bool)))++{- | Assert that the migration installed every Kiroku table under the @kiroku@+schema and left @public@ free of Kiroku tables. The connection uses no special+@search_path@, so the schema-qualified @to_regclass@ checks prove placement+directly rather than relying on name resolution.+-}+assertSchemaPlacement :: Text -> IO ()+assertSchemaPlacement connStr = do+    pool <- Pool.acquire poolConfig+    result <- Pool.use pool (Session.statement () placementStmt)+    Pool.release pool+    case result of+        Left err -> expectationFailure ("schema placement query failed: " <> show err)+        Right (True, True) -> pure ()+        Right (kirokuPresent, publicAbsent) ->+            expectationFailure+                ( "expected all Kiroku tables in 'kiroku' and none in 'public'; got kirokuPresent="+                    <> show kirokuPresent+                    <> ", publicAbsent="+                    <> show publicAbsent+                )+  where+    poolConfig =+        Pool.Config.settings+            [ Pool.Config.staticConnectionSettings (Conn.connectionString connStr)+            , Pool.Config.size 1+            ]++placementStmt :: Statement () (Bool, Bool)+placementStmt =+    preparable+        "SELECT \+        \  (to_regclass('kiroku.streams') IS NOT NULL \+        \   AND to_regclass('kiroku.events') IS NOT NULL \+        \   AND to_regclass('kiroku.stream_events') IS NOT NULL \+        \   AND to_regclass('kiroku.subscriptions') IS NOT NULL), \+        \  (to_regclass('public.streams') IS NULL \+        \   AND to_regclass('public.events') IS NULL \+        \   AND to_regclass('public.stream_events') IS NULL \+        \   AND to_regclass('public.subscriptions') IS NULL)"+        E.noParams+        (D.singleRow ((,) <$> D.column (D.nonNullable D.bool) <*> D.column (D.nonNullable D.bool)))++{- | Assert that the forward migration installed the @kiroku.dead_letters@ table+(MasterPlan 6 / EP-40) — proving codd applied the timestamped migration after the+bootstrap, not only the bootstrap itself.+-}+assertDeadLettersTable :: Text -> IO ()+assertDeadLettersTable connStr = do+    pool <- Pool.acquire poolConfig+    result <- Pool.use pool (Session.statement () deadLettersStmt)+    Pool.release pool+    case result of+        Left err -> expectationFailure ("dead_letters verification query failed: " <> show err)+        Right True -> pure ()+        Right False -> expectationFailure "Kiroku migration did not create the kiroku.dead_letters table"+  where+    poolConfig =+        Pool.Config.settings+            [ Pool.Config.staticConnectionSettings (Conn.connectionString connStr)+            , Pool.Config.size 1+            ]++deadLettersStmt :: Statement () Bool+deadLettersStmt =+    preparable+        "SELECT to_regclass('kiroku.dead_letters') IS NOT NULL"+        E.noParams+        (D.singleRow (D.column (D.nonNullable D.bool)))++assertDefaultUuidV7 :: Text -> IO ()+assertDefaultUuidV7 connStr = do+    pool <- Pool.acquire poolConfig+    result <- Pool.use pool (Session.statement () defaultUuidStmt)+    versionResult <- Pool.use pool (Session.statement () serverVersionStmt)+    Pool.release pool+    case (result, versionResult) of+        (Right eventIdText, Right version)+            | T.length eventIdText > 14 && T.index eventIdText 14 == '7' -> pure ()+            | otherwise ->+                expectationFailure+                    ( "expected migration-created database default to generate UUIDv7 on PostgreSQL "+                        <> T.unpack version+                        <> ", got "+                        <> T.unpack eventIdText+                    )+        (Left err, _) -> expectationFailure ("default UUID insert failed: " <> show err)+        (_, Left err) -> expectationFailure ("server version query failed: " <> show err)+  where+    poolConfig =+        Pool.Config.settings+            [ Pool.Config.staticConnectionSettings (Conn.connectionString connStr)+            , Pool.Config.size 1+            ]++defaultUuidStmt :: Statement () Text+defaultUuidStmt =+    preparable+        "INSERT INTO kiroku.events (event_type, data) VALUES ('DefaultUuidGenerated', '{}'::jsonb) RETURNING event_id::text"+        E.noParams+        (D.singleRow (D.column (D.nonNullable D.text)))++serverVersionStmt :: Statement () Text+serverVersionStmt =+    preparable+        "SELECT current_setting('server_version_num')"+        E.noParams+        (D.singleRow (D.column (D.nonNullable D.text)))