kiroku-store-migrations 0.7.1.0 → 0.7.2.0

raw patch · 12 files changed

+1381/−1342 lines, 12 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +15/−0
• README.md +2/−2
• app/Main.hs +26/−26
• kiroku-store-migrations.cabal +1/−1
• migrations/0016.sql +13/−0
• migrations/manifest +1/−0
• src/Kiroku/Store/Migrations.hs +15/−15
• src/Kiroku/Store/Migrations/History/Codd.hs +47/−46
• src/Kiroku/Store/Migrations/Internal/Definition.hs +9/−8
• src/Kiroku/Store/Migrations/Internal/EmbedFile.hs +7/−7
• src/Kiroku/Store/Migrations/New.hs +20/−19
• test/Main.hs +1225/−1218

@@ -1,5 +1,20 @@ # Changelog +## 0.7.2.0 — 2026-10-26++### New Features++- Migration 0016 permits `category_namespace` durable subscription bindings using+  the existing `target_category` column. It changes subscription constraints only;+  no event/stream index, append column, or trigger is added. Existing all/category+  bindings remain valid.++### Other Changes++- Stop affected workers before applying migration 0016 and restart with+  kiroku-store 0.12.0.0 before using namespace targets. Older binaries must not+  manage namespace bindings.+ ## 0.7.1.0 — 2026-10-11  ### New Features

@@ -2,9 +2,9 @@  `kiroku-store-migrations` owns Kiroku's PostgreSQL schema as one native `pg-migrate` component named `kiroku`. The component embeds an ordered manifest-and fifteen SQL payloads, so applications can compose it with other libraries+and sixteen SQL payloads, so applications can compose it with other libraries without copying Kiroku SQL. The first seven payloads are immutable historical-Codd bytes; `0008` through `0015` are native-only forward migrations.+Codd bytes; `0008` through `0016` are native-only forward migrations.  ## Public API 

@@ -14,32 +14,32 @@  main :: IO () main = do-    plan <- either (fail . show) pure kirokuMigrationPlan-    cmd <--        execParser-            ( info-                (migrationCommandParser plan <**> helper)-                (fullDesc <> progDesc "Manage the Kiroku migration component")-            )-    defaultDatabaseUrl <- lookupEnv "DATABASE_URL"-    let defaultSettings =-            Settings.connectionString (Text.pack (maybe "" id defaultDatabaseUrl))-        environment = cliEnvironment defaultSettings plan defaultRunOptions-    outcome <- runMigrationCommand environment cmd-    case commandOutputFormat cmd of-        TextOutput -> Text.IO.putStrLn (renderMigrationCommandText outcome)-        JsonOutput -> LazyByteString.putStrLn (Aeson.encode (renderMigrationCommandJson outcome))-    Exit.exitWith-        (case exitClass outcome of ExitSucceeded -> Exit.ExitSuccess; _ -> Exit.ExitFailure 1)+  plan <- either (fail . show) pure kirokuMigrationPlan+  cmd <-+    execParser+      ( info+          (migrationCommandParser plan <**> helper)+          (fullDesc <> progDesc "Manage the Kiroku migration component")+      )+  defaultDatabaseUrl <- lookupEnv "DATABASE_URL"+  let defaultSettings =+        Settings.connectionString (Text.pack (maybe "" id defaultDatabaseUrl))+      environment = cliEnvironment defaultSettings plan defaultRunOptions+  outcome <- runMigrationCommand environment cmd+  case commandOutputFormat cmd of+    TextOutput -> Text.IO.putStrLn (renderMigrationCommandText outcome)+    JsonOutput -> LazyByteString.putStrLn (Aeson.encode (renderMigrationCommandJson outcome))+  Exit.exitWith+    (case exitClass outcome of ExitSucceeded -> Exit.ExitSuccess; _ -> Exit.ExitFailure 1)  commandOutputFormat :: MigrationCommand -> OutputFormat commandOutputFormat cmd =-    case cmd of-        Plan PlanOptions{output = OutputOptions format} -> format-        List ListOptions{output = OutputOptions format} -> format-        Check CheckOptions{output = OutputOptions format} -> format-        Status StatusOptions{output = OutputOptions format} -> format-        Verify VerifyOptions{output = OutputOptions format} -> format-        Up UpOptions{output = OutputOptions format} -> format-        Repair RepairOptions{output = OutputOptions format} -> format-        New NewOptions{output = OutputOptions format} -> format+  case cmd of+    Plan PlanOptions {output = OutputOptions format} -> format+    List ListOptions {output = OutputOptions format} -> format+    Check CheckOptions {output = OutputOptions format} -> format+    Status StatusOptions {output = OutputOptions format} -> format+    Verify VerifyOptions {output = OutputOptions format} -> format+    Up UpOptions {output = OutputOptions format} -> format+    Repair RepairOptions {output = OutputOptions format} -> format+    New NewOptions {output = OutputOptions format} -> format

@@ -1,6 +1,6 @@ cabal-version:      3.0 name:               kiroku-store-migrations-version:            0.7.1.0+version:            0.7.2.0 synopsis:           Schema migrations for kiroku-store description:   Native pg-migrate component, Codd history mapping, and migration executable

@@ -0,0 +1,13 @@+-- Allow category namespace subscription targets. Stop workers during cutover.+-- These are the generated constraint names from migration 0014 on PG 17/18.+-- Only subscription metadata changes; no event/stream index or append work.+ALTER TABLE kiroku.subscriptions+    DROP CONSTRAINT subscriptions_target_kind_check,+    DROP CONSTRAINT subscriptions_check,+    ADD CONSTRAINT subscriptions_target_kind_check+        CHECK (target_kind IN ('unbound', 'all', 'category', 'category_namespace')),+    ADD CONSTRAINT subscriptions_check+        CHECK ((target_kind IN ('category', 'category_namespace')) = (target_category IS NOT NULL));++COMMENT ON SCHEMA kiroku IS+    'Managed by pg-migrate component kiroku through 0016';

@@ -13,3 +13,4 @@ 0013.sql 0014.sql 0015.sql+0016.sql

@@ -1,30 +1,30 @@-module Kiroku.Store.Migrations (-    DefinitionError,+module Kiroku.Store.Migrations+  ( DefinitionError,     MigrationComponent,     MigrationPlan,     PlanError,     kirokuMigrationPlan,     kirokuMigrations,-) where+  )+where  import Data.List.NonEmpty (NonEmpty (..))-import Database.PostgreSQL.Migrate (-    DefinitionError,+import Database.PostgreSQL.Migrate+  ( DefinitionError,     MigrationComponent,     MigrationPlan,     PlanError,     migrationPlan,- )+  ) import Kiroku.Store.Migrations.Internal.Definition (kirokuMigrations) -{- | The complete single-component Kiroku migration plan.--The embedded manifest is validated at compile time, so a component-definition-failure indicates a broken package invariant rather than an operator error.--}+-- | The complete single-component Kiroku migration plan.+--+-- The embedded manifest is validated at compile time, so a component-definition+-- failure indicates a broken package invariant rather than an operator error. kirokuMigrationPlan :: Either PlanError MigrationPlan kirokuMigrationPlan =-    case kirokuMigrations of-        Left definitionError ->-            error ("invalid embedded Kiroku migration component: " <> show definitionError)-        Right component -> migrationPlan (component :| [])+  case kirokuMigrations of+    Left definitionError ->+      error ("invalid embedded Kiroku migration component: " <> show definitionError)+    Right component -> migrationPlan (component :| [])

@@ -1,12 +1,13 @@ {-# LANGUAGE TemplateHaskell #-} -module Kiroku.Store.Migrations.History.Codd (-    kirokuCoddHistoryMappings,+module Kiroku.Store.Migrations.History.Codd+  ( kirokuCoddHistoryMappings,     kirokuCoddManifestText,     kirokuCoddSourceConfig,     kirokuCoddSourcePayloads,     kirokuLegacyMigrationNames,-) where+  )+where  import Data.Bifunctor (first) import Data.ByteString (ByteString)@@ -14,86 +15,86 @@ import Data.List.NonEmpty (NonEmpty (..)) import Data.Map.Strict qualified as Map import Data.Text (Text)-import Database.PostgreSQL.Migrate (-    Confirmation,+import Database.PostgreSQL.Migrate+  ( Confirmation,     ConnectionProvider,     EvidenceRequirement (Evidence),     HistoryMapping,     PayloadRelation (SamePayload),     historyMapping,     migrationId,- )-import Database.PostgreSQL.Migrate.History.Codd (-    CoddDefinitionError,+  )+import Database.PostgreSQL.Migrate.History.Codd+  ( CoddDefinitionError,     CoddSourceConfig,     coddEvidenceKey,     coddSourceConfig,     parseCoddManifest,- )+  ) import Kiroku.Store.Migrations.Internal.Definition (embeddedMigrationEntries) import Kiroku.Store.Migrations.Internal.EmbedFile (embedTextFile)  kirokuLegacyMigrationNames :: NonEmpty FilePath kirokuLegacyMigrationNames =-    "2026-05-16-12-17-14-kiroku-bootstrap.sql"-        :| [ "2026-05-29-15-26-04-add-subscription-dead-letters.sql"-           , "2026-06-14-13-17-09-notify-trigger-append-guard.sql"-           , "2026-06-14-13-25-40-dead-letters-event-id-index.sql"-           , "2026-06-14-13-54-48-index-hygiene-and-streams-fillfactor.sql"-           , "2026-06-14-14-01-17-stream-name-length-check.sql"-           , "2026-06-24-09-42-22-stream-truncate-before.sql"-           ]+  "2026-05-16-12-17-14-kiroku-bootstrap.sql"+    :| [ "2026-05-29-15-26-04-add-subscription-dead-letters.sql",+         "2026-06-14-13-17-09-notify-trigger-append-guard.sql",+         "2026-06-14-13-25-40-dead-letters-event-id-index.sql",+         "2026-06-14-13-54-48-index-hygiene-and-streams-fillfactor.sql",+         "2026-06-14-14-01-17-stream-name-length-check.sql",+         "2026-06-24-09-42-22-stream-truncate-before.sql"+       ]  kirokuCoddHistoryMappings :: NonEmpty HistoryMapping kirokuCoddHistoryMappings =-    zipWithNonEmpty mapping kirokuLegacyMigrationNames nativeMigrationNames+  zipWithNonEmpty mapping kirokuLegacyMigrationNames nativeMigrationNames   where     mapping sourceFilename targetName =-        historyMapping-            (definitionInvariant (migrationId "kiroku" targetName))-            (Evidence sourceKey)-            (SamePayload sourceKey)+      historyMapping+        (definitionInvariant (migrationId "kiroku" targetName))+        (Evidence sourceKey)+        (SamePayload sourceKey)       where         sourceKey = definitionInvariant (first show (coddEvidenceKey sourceFilename))  kirokuCoddSourceConfig ::-    ConnectionProvider ->-    Bool ->-    Text ->-    Confirmation ->-    Either CoddDefinitionError CoddSourceConfig+  ConnectionProvider ->+  Bool ->+  Text ->+  Confirmation ->+  Either CoddDefinitionError CoddSourceConfig kirokuCoddSourceConfig sourceProvider strictSource reason confirmation =-    coddSourceConfig-        sourceProvider-        kirokuLegacyMigrationNames-        strictSource-        kirokuCoddSourcePayloads-        (Just (definitionInvariant (parseCoddManifest kirokuCoddManifestText)))-        reason-        confirmation+  coddSourceConfig+    sourceProvider+    kirokuLegacyMigrationNames+    strictSource+    kirokuCoddSourcePayloads+    (Just (definitionInvariant (parseCoddManifest kirokuCoddManifestText)))+    reason+    confirmation  nativeMigrationNames :: NonEmpty Text nativeMigrationNames =-    "0001-kiroku-bootstrap"-        :| [ "0002-add-subscription-dead-letters"-           , "0003-notify-trigger-append-guard"-           , "0004-dead-letters-event-id-index"-           , "0005-index-hygiene-and-streams-fillfactor"-           , "0006-stream-name-length-check"-           , "0007-stream-truncate-before"-           ]+  "0001-kiroku-bootstrap"+    :| [ "0002-add-subscription-dead-letters",+         "0003-notify-trigger-append-guard",+         "0004-dead-letters-event-id-index",+         "0005-index-hygiene-and-streams-fillfactor",+         "0006-stream-name-length-check",+         "0007-stream-truncate-before"+       ]  kirokuCoddSourcePayloads :: Map.Map FilePath ByteString kirokuCoddSourcePayloads =-    Map.fromList-        (zip (toList kirokuLegacyMigrationNames) (snd <$> toList embeddedMigrationEntries))+  Map.fromList+    (zip (toList kirokuLegacyMigrationNames) (snd <$> toList embeddedMigrationEntries))  kirokuCoddManifestText :: Text kirokuCoddManifestText = $(embedTextFile "migrations.lock")  zipWithNonEmpty :: (a -> b -> c) -> NonEmpty a -> NonEmpty b -> NonEmpty c zipWithNonEmpty combine (firstA :| restA) (firstB :| restB) =-    combine firstA firstB :| zipWith combine restA restB+  combine firstA firstB :| zipWith combine restA restB  definitionInvariant :: (Show error) => Either error value -> value definitionInvariant = either (error . ("invalid checked-in Kiroku migration definition: " <>) . show) id

@@ -7,24 +7,25 @@ -- reports "Up to date" and never invokes GHC. A clean build revalidates. {-# OPTIONS_GHC -fplugin=Database.PostgreSQL.Migrate.Embed.RecompilePlugin #-} -module Kiroku.Store.Migrations.Internal.Definition (-    embeddedMigrationEntries,+module Kiroku.Store.Migrations.Internal.Definition+  ( embeddedMigrationEntries,     kirokuMigrations,-) where+  )+where  import Data.ByteString (ByteString) import Data.List.NonEmpty (NonEmpty)-import Database.PostgreSQL.Migrate (-    DefinitionError,+import Database.PostgreSQL.Migrate+  ( DefinitionError,     MigrationComponent,     migrationComponentFromEmbeddedSql,- )+  ) import Database.PostgreSQL.Migrate.Embed (embedMigrationManifest)  embeddedMigrationEntries :: NonEmpty (FilePath, ByteString) embeddedMigrationEntries =-    $(embedMigrationManifest "migrations/manifest")+  $(embedMigrationManifest "migrations/manifest")  kirokuMigrations :: Either DefinitionError MigrationComponent kirokuMigrations =-    migrationComponentFromEmbeddedSql "kiroku" mempty embeddedMigrationEntries+  migrationComponentFromEmbeddedSql "kiroku" mempty embeddedMigrationEntries

@@ -10,10 +10,10 @@  embedTextFile :: FilePath -> Q Exp embedTextFile inputPath = do-    path <- TH.makeRelativeToProject inputPath-    TH.addDependentFile path-    bytes <- TH.runIO (ByteString.readFile path)-    case Text.Encoding.decodeUtf8' bytes of-        Left decodeError -> fail ("invalid UTF-8 in " <> path <> ": " <> show decodeError)-        Right contents ->-            pure (AppE (VarE 'Text.pack) (LitE (StringL (Text.unpack contents))))+  path <- TH.makeRelativeToProject inputPath+  TH.addDependentFile path+  bytes <- TH.runIO (ByteString.readFile path)+  case Text.Encoding.decodeUtf8' bytes of+    Left decodeError -> fail ("invalid UTF-8 in " <> path <> ": " <> show decodeError)+    Right contents ->+      pure (AppE (VarE 'Text.pack) (LitE (StringL (Text.unpack contents))))

@@ -1,18 +1,19 @@-module Kiroku.Store.Migrations.New (-    AuthoringError,+module Kiroku.Store.Migrations.New+  ( AuthoringError,     defaultMigrationsDir,     newMigrationFile,     migrationTemplate,-) where+  )+where  import Data.ByteString (ByteString) import Data.Text qualified as Text import Data.Text.Encoding qualified as Text.Encoding-import Database.PostgreSQL.Migrate.Embed (-    AuthoringError,+import Database.PostgreSQL.Migrate.Embed+  ( AuthoringError,     newMigration,     newMigrationOptions,- )+  ) import System.FilePath ((</>))  defaultMigrationsDir :: FilePath@@ -20,21 +21,21 @@  newMigrationFile :: FilePath -> String -> IO (Either AuthoringError FilePath) newMigrationFile migrationsDirectory description =-    case newMigrationOptions manifestPath Nothing (migrationTemplate description) of-        Left definitionError -> pure (Left definitionError)-        Right options -> newMigration options+  case newMigrationOptions manifestPath Nothing (migrationTemplate description) of+    Left definitionError -> pure (Left definitionError)+    Right options -> newMigration options   where     manifestPath = migrationsDirectory </> "manifest"  migrationTemplate :: String -> ByteString migrationTemplate description =-    Text.Encoding.encodeUtf8-        ( Text.unlines-            [ "-- " <> Text.pack description-            , ""-            , "-- Kiroku migrations are forward-only and applied exactly once by pg-migrate."-            , "-- Qualify every object with the kiroku schema and append corrections"-            , "-- as new manifest entries instead of editing released payloads."-            , ""-            ]-        )+  Text.Encoding.encodeUtf8+    ( Text.unlines+        [ "-- " <> Text.pack description,+          "",+          "-- Kiroku migrations are forward-only and applied exactly once by pg-migrate.",+          "-- Qualify every object with the kiroku schema and append corrections",+          "-- as new manifest entries instead of editing released payloads.",+          ""+        ]+    )

@@ -43,1221 +43,1228 @@  main :: IO () main = hspec $ do-    describe "native Kiroku migration definition" $ do-        it "tracks the fifteen native files in manifest order" $ do-            directory <- findMigrationsDirectory-            manifest <- Text.lines <$> Text.IO.readFile (directory </> "manifest")-            manifest `shouldBe` Text.pack <$> nativeMigrationFiles--        it "preserves every legacy payload byte recorded by migrations.lock" $ do-            directory <- findMigrationsDirectory-            lockPath <- findLockfile-            lockEntries <- parseLockfile <$> Text.IO.readFile lockPath-            forM_ (zip (toList kirokuLegacyMigrationNames) nativeMigrationFiles) $ \(legacyName, nativeName) -> do-                bytes <- ByteString.readFile (directory </> nativeName)-                lookup legacyName lockEntries `shouldBe` Just (checksumText bytes)--        it "builds component kiroku and a fifteen-migration plan" $ do-            component <- requireRight kirokuMigrations-            component `seq` pure ()-            plan <- requirePlan-            let targetIds =-                    [ requireRight (migrationId "kiroku" (Text.pack (dropSqlSuffix file)))-                    | file <- nativeMigrationFiles-                    ]-            validateHistoryMappingTargets plan kirokuCoddHistoryMappings-                `shouldBe` Right ()-            length targetIds `shouldBe` length nativeMigrationFiles--    describe "native migration authoring" $ do-        it "creates the next numeric file and atomically appends the manifest" $-            withSystemTempDirectory "kiroku-native-authoring" $ \directory -> do-                ByteString.writeFile (directory </> "0007-existing.sql") "SELECT 7;\n"-                ByteString.writeFile (directory </> "manifest") "0007-existing.sql\n"-                created <- newMigrationFile directory "add widget index"-                path <- requireRight created-                path `shouldBe` directory </> "0008.sql"-                body <- ByteString.readFile path-                body `shouldSatisfy` ByteString.isInfixOf "forward-only"-                Text.lines <$> Text.IO.readFile (directory </> "manifest")-                    `shouldReturn` ["0007-existing.sql", "0008.sql"]--        it "refuses to overwrite a pre-existing inferred migration" $-            withSystemTempDirectory "kiroku-native-exclusive" $ \directory -> do-                ByteString.writeFile (directory </> "manifest") "0007-existing.sql\n"-                ByteString.writeFile (directory </> "0007-existing.sql") "SELECT 7;\n"-                ByteString.writeFile (directory </> "0008.sql") "SELECT 8;\n"-                created <- newMigrationFile directory "must not overwrite"-                created `shouldSatisfy` isLeft-                Text.IO.readFile (directory </> "manifest")-                    `shouldReturn` "0007-existing.sql\n"--    describe "fresh native databases" $ do-        it "applies all fifteen, verifies strictly, and reports AlreadyApplied on rerun" $ do-            plan <- requirePlan-            result <- withMigratedDatabase plan $ \connection -> do-                assertSchema connection-                let provider = providerFor connection-                rerun <- runMigrationPlanWith defaultRunOptions provider plan >>= requireMigration-                reportOutcomes rerun `shouldBe` replicate (length nativeMigrationFiles) AlreadyApplied-                verified <- verifyMigrationPlanWith defaultRunOptions provider plan >>= requireMigration-                case verified of-                    VerificationReport verificationIssues applied _ _ -> do-                        verificationIssues `shouldBe` []-                        length applied `shouldBe` length nativeMigrationFiles-            either (expectationFailure . show) pure result--        it "serializes concurrent applies through the pg-migrate advisory lock" $ do-            plan <- requirePlan-            withKirokuPg $ \database -> do-                let settings = Pg.connectionSettings database-                (first, second) <--                    concurrently-                        (runMigrationPlan defaultRunOptions settings plan >>= requireMigration)-                        (runMigrationPlan defaultRunOptions settings plan >>= requireMigration)-                sort [reportOutcomes first, reportOutcomes second]-                    `shouldBe` sort-                        [ replicate (length nativeMigrationFiles) AppliedNow-                        , replicate (length nativeMigrationFiles) AlreadyApplied-                        ]--    describe "subscription checkpoint SQL relation" $ do-        it "publishes the frozen ordinary-view catalog contract" $ do-            plan <- requirePlan-            result <- withMigratedDatabase plan $ \connection -> do-                relation <- useSession connection (Session.statement () relationCatalogStatement)-                relation-                    `shouldBe` ( "v"-                               , "security_invoker=false"-                               , True-                               , "NO"-                               , "NO"-                               , "Stable read-only v1 relation of exact persisted subscription-member checkpoints; its columns and order are frozen, and its rows are unordered unless the caller supplies ORDER BY."-                               )-                columns <- useSession connection (Session.statement () relationColumnsStatement)-                columns-                    `shouldBe` [-                                   ( "subscription_name"-                                   , "text"-                                   , False-                                   , "Persisted subscription name; member zero does not distinguish a non-group subscription from member zero of a consumer group."-                                   )-                               ,-                                   ( "consumer_group_member"-                                   , "integer"-                                   , False-                                   , "Persisted consumer-group member key; member zero carries no topology classification."-                                   )-                               ,-                                   ( "checkpoint_position"-                                   , "bigint"-                                   , False-                                   , "Exact persisted global position for this subscription member; an explicit reset may move it backward or forward."-                                   )-                               ,-                                   ( "checkpoint_updated_at"-                                   , "timestamp with time zone"-                                   , False-                                   , "Time of the latest checkpoint-row upsert; it does not imply position advancement or worker liveness."-                                   )-                               ]-            either (expectationFailure . show) pure result--        it "returns zero rows for an empty checkpoint inventory" $ do-            plan <- requirePlan-            result <- withMigratedDatabase plan $ \connection -> do-                rows <- useSession connection (Session.statement () checkpointRowsStatement)-                rows `shouldBe` []-            either (expectationFailure . show) pure result--        it "returns exact non-null member rows and exposes reassignment only after commit" $ do-            plan <- requirePlan-            withKirokuPg $ \database -> do-                let settings = Pg.connectionSettings database-                _ <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration-                withConnection settings $ \writer ->-                    withConnection settings $ \reader -> do-                        useSession writer (Session.script checkpointFixtureSql)-                        rows <- useSession reader (Session.statement () checkpointRowsStatement)-                        rows-                            `shouldBe` [ ("grouped", 0, 20, checkpointFixtureTimestamp)-                                       , ("grouped", 1, 21, checkpointFixtureTimestamp)-                                       , ("grouped", 2, 22, checkpointFixtureTimestamp)-                                       , ("ordinary", 0, 11, checkpointFixtureTimestamp)-                                       ]--                        useSession writer (Session.script "BEGIN")-                        useSession writer (Session.script checkpointAdvanceSql)-                        checkpointPosition writer `shouldReturn` 5-                        checkpointPosition reader `shouldReturn` 11-                        useSession writer (Session.script "COMMIT")-                        checkpointPosition reader `shouldReturn` 5--                        useSession writer (Session.script "BEGIN")-                        useSession writer (Session.script checkpointRegressionSql)-                        checkpointPosition writer `shouldReturn` 2-                        checkpointPosition reader `shouldReturn` 5-                        useSession writer (Session.script "ROLLBACK")-                        checkpointPosition reader `shouldReturn` 5--        it "allows view-only readers while structurally rejecting owner updates" $ do-            plan <- requirePlan-            result <- withMigratedDatabase plan $ \connection ->-                withTemporaryReaderRole connection $ \role -> do-                    let identifier = quoteTestIdentifier role-                    useSession-                        connection-                        ( Session.script-                            ( "GRANT USAGE ON SCHEMA kiroku TO "-                                <> identifier-                                <> "; GRANT SELECT ON kiroku.subscription_checkpoints_v1 TO "-                                <> identifier-                                <> "; SET ROLE "-                                <> identifier-                            )-                        )-                    viewCount <- useSession connection (Session.statement () checkpointRowCountStatement)-                    viewCount `shouldBe` 0-                    privatePrivilege <- useSession connection (Session.statement () privateTablePrivilegeStatement)-                    privatePrivilege `shouldBe` False-                    directPrivateRead <- Connection.use connection (Session.statement () privateTableCountStatement)-                    directPrivateRead `shouldSatisfy` hasSqlState "42501"--                    useSession connection (Session.script "RESET ROLE")-                    ownerUpdate <- Connection.use connection (Session.statement () updatePublicViewStatement)-                    ownerUpdate `shouldSatisfy` hasSqlState "55000"-            either (expectationFailure . show) pure result--        it "preserves a downstream view while private checkpoint storage is replaced" $ do-            plan <- requirePlan-            withKirokuPg $ \database -> do-                let settings = Pg.connectionSettings database-                _ <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration-                withConnection settings $ \connection -> do-                    useSession connection (Session.script downstreamReplacementFixtureSql)-                    floors <- useSession connection (Session.statement () downstreamFloorsStatement)-                    floors `shouldBe` [("dependency", 42)]--        it "pushes a subscription filter to the existing checkpoint index" $ do-            plan <- requirePlan-            result <- withMigratedDatabase plan $ \connection -> do-                useSession connection (Session.script checkpointPlanFixtureSql)-                version <- useSession connection (Session.statement () serverVersionStatement)-                planLines <- useSession connection (Session.statement () checkpointExplainStatement)-                let planText = Text.unlines planLines-                Text.IO.putStrLn ("PostgreSQL " <> version <> " checkpoint relation plan:\n" <> planText)-                planText `shouldSatisfy` Text.isInfixOf "ix_subscriptions_name_member"-                Text.isInfixOf "CTE Scan" planText `shouldBe` False-            either (expectationFailure . show) pure result--    describe "history retention schema" $ do-        it "publishes the exact coordinator and lease table columns" $ do-            plan <- requirePlan-            result <- withMigratedDatabase plan $ \connection -> do-                columns <- useSession connection (Session.statement () historyRetentionColumnsStatement)-                columns-                    `shouldBe` [ ("history_retention_coordinator", "singleton", "boolean", True)-                               , ("history_retention_leases", "lease_id", "uuid", True)-                               , ("history_retention_leases", "owner", "text", True)-                               , ("history_retention_leases", "reason", "text", True)-                               , ("history_retention_leases", "protected_through", "bigint", True)-                               , ("history_retention_leases", "created_at", "timestamp with time zone", True)-                               , ("history_retention_leases", "renewed_at", "timestamp with time zone", True)-                               , ("history_retention_leases", "expires_at", "timestamp with time zone", True)-                               , ("history_retention_leases", "released_at", "timestamp with time zone", False)-                               ]-            either (expectationFailure . show) pure result--        it "installs the singleton, checks, partial index, function, and six triggers" $ do-            plan <- requirePlan-            result <- withMigratedDatabase plan $ \connection -> do-                facts <- useSession connection (Session.statement () historyRetentionFactsStatement)-                facts `shouldBe` (1, 9, True, True, 6)-            either (expectationFailure . show) pure result--    -- BUG-1. Migrations 0002 onward are parsed in sessions that never ran 0001,-    -- so a name that only 0001's `SET search_path` makes resolvable parses on a-    -- fresh install and fails on every ordinary upgrade. 0.3.2.x's 0010 named-    -- uuidv7() that way and could not be applied to a database bootstrapped-    -- through 0009 on PostgreSQL 17. These cases apply the tail of the plan in a-    -- session that carries none of 0001's state.-    describe "upgrades of already-bootstrapped databases" $ do-        it "applies the pending tail in a session that never ran the bootstrap" $ do-            plan <- requirePlan-            let bootstrapCount = 9 -- through 0009, so 0010 is in the pending tail-            throughBootstrap <- planThrough bootstrapCount-            withKirokuPg $ \database -> do-                let settings = Pg.connectionSettings database-                bootstrapped <--                    runMigrationPlan defaultRunOptions settings throughBootstrap-                        >>= requireMigration-                reportOutcomes bootstrapped-                    `shouldBe` replicate bootstrapCount AppliedNow--                -- A separate session, and one that cannot reach the Kiroku-                -- schema through search_path. The suite connects as role-                -- "kiroku", so the default "$user" entry resolves to the Kiroku-                -- schema and would hide exactly the name resolution this case-                -- exists to prove -- the same accident that let 0001's-                -- SET search_path carry a fresh install through 0.3.2.x's 0010.-                withConnection settings $ \upgradeSession -> do-                    useSession upgradeSession (Session.script "SET search_path TO pg_catalog")-                    upgraded <--                        runMigrationPlanWith defaultRunOptions (providerFor upgradeSession) plan-                            >>= requireMigration-                    reportOutcomes upgraded-                        `shouldBe` replicate bootstrapCount AlreadyApplied-                            <> replicate (length nativeMigrationFiles - bootstrapCount) AppliedNow--                verified <- verifyMigrationPlan defaultRunOptions settings plan >>= requireMigration-                case verified of-                    VerificationReport verificationIssues applied _ _ -> do-                        verificationIssues `shouldBe` []-                        length applied `shouldBe` length nativeMigrationFiles-                withConnection settings assertSchema--        it "adds unbound target metadata without rewriting legacy rows and drops stream_name" $ do-            plan <- requirePlan-            throughPrevious <- planThrough 13-            withKirokuPg $ \database -> do-                let settings = Pg.connectionSettings database-                _ <- runMigrationPlan defaultRunOptions settings throughPrevious >>= requireMigration-                withConnection settings $ \connection -> do-                    useSession connection (Session.script "INSERT INTO kiroku.subscriptions (subscription_name, last_seen) VALUES ('legacy-target', 7)")-                    before <- useSession connection (Session.statement () targetUpgradeFactsStatement)-                    _ <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration-                    after <- useSession connection (Session.statement () targetUpgradeFactsStatement)-                    -- Relation filenode and tuple ctid prove no heap rewrite.-                    after `shouldBe` before-                    facts <- useSession connection (Session.statement () targetBindingFactsStatement)-                    facts `shouldBe` ("unbound", Nothing, 7, False)--        it "derives group topology from legacy rows when 0013 upgrades a populated store" $ do-            plan <- requirePlan-            throughPrevious <- planThrough 12-            withKirokuPg $ \database -> do-                let settings = Pg.connectionSettings database-                _ <- runMigrationPlan defaultRunOptions settings throughPrevious >>= requireMigration-                withConnection settings $ \connection ->-                    useSession connection (Session.script "INSERT INTO kiroku.subscriptions (subscription_name, consumer_group_member, last_seen) VALUES ('group', 0, 5), ('group', 1, 20), ('incomplete', 0, 8), ('ordinary', 0, 10)")-                upgraded <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration-                reportOutcomes upgraded `shouldBe` replicate 12 AlreadyApplied <> replicate (length nativeMigrationFiles - 12) AppliedNow-                withConnection settings $ \connection -> do-                    let stmt =-                            Statement.preparable-                                "SELECT subscription_name, consumer_group_member, consumer_group_size, last_seen FROM kiroku.subscriptions ORDER BY subscription_name, consumer_group_member"-                                Encoders.noParams-                                (Decoders.rowList ((,,,) <$> Decoders.column (Decoders.nonNullable Decoders.text) <*> Decoders.column (Decoders.nonNullable Decoders.int4) <*> Decoders.column (Decoders.nonNullable Decoders.int4) <*> Decoders.column (Decoders.nonNullable Decoders.int8)))-                    facts <- useSession connection (Session.statement () stmt)-                    facts `shouldBe` [("group" :: Text, 0 :: Int32, 2 :: Int32, 5 :: Int64), ("group", 1, 2, 20), ("incomplete", 0, 1, 8), ("ordinary", 0, 1, 10)]-                    directory <- findMigrationsDirectory-                    sql <- Text.IO.readFile (directory </> "0013.sql")-                    useSession connection (Session.script sql)-                    useSession connection (Session.statement () stmt) `shouldReturn` facts--        -- BUG-2. 0012 copies each $all row's originating-stream category onto the-        -- junction row so category reads can range-scan an index. A store-        -- written before 0012 has $all rows without it; the backfill must fill-        -- every one, and afterwards the CHECK must refuse a new $all row that-        -- omits it, the immutability trigger must be back on, and the index must-        -- exist in exactly the shape the category reads rely on.-        it "backfills $all-row categories when 0012 upgrades a populated store" $ do-            plan <- requirePlan-            let throughCount = 11-            throughPrevious <- planThrough throughCount-            withKirokuPg $ \database -> do-                let settings = Pg.connectionSettings database-                _ <- runMigrationPlan defaultRunOptions settings throughPrevious >>= requireMigration-                withConnection settings $ \connection ->-                    useSession connection (Session.script preCategoryFixtureSql)-                upgraded <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration-                reportOutcomes upgraded-                    `shouldBe` replicate throughCount AlreadyApplied <> replicate (length nativeMigrationFiles - throughCount) AppliedNow-                withConnection settings $ \connection -> do-                    facts <- useSession connection (Session.statement () categoryBackfillFactsStatement)-                    facts-                        `shouldBe` ( True-                                   , 0-                                   , True-                                   , Just "CREATE INDEX ix_stream_events_all_by_category ON kiroku.stream_events USING btree (category, stream_version) INCLUDE (original_stream_id) WHERE (stream_id = 0)"-                                   , "O"-                                   )-                    missingCategory <- Connection.use connection (Session.statement () insertAllRowWithoutCategoryStatement)-                    missingCategory `shouldSatisfy` hasSqlState "23514"-                    assertSchema connection--        -- kiroku.uuidv7() is the component's version-independent generator, but-        -- it arrives by a different route on each major: 0001's fallback on-        -- PostgreSQL 17, 0010's alias for the builtin on PostgreSQL 18. This-        -- case pins the route to the running server, so an acceptance run that-        -- believes it exercised one major and actually ran the other fails here-        -- instead of passing quietly. Both routes need a run to be covered ---        -- see `just test-matrix`.-        it "publishes the generator by the route its PostgreSQL major requires" $ do-            plan <- requirePlan-            result <- withMigratedDatabase plan $ \connection -> do-                (major, builtinExists, generatorExists, leaseDefault) <--                    useSession connection (Session.statement () leaseDefaultStatement)-                Text.IO.putStrLn-                    ( "PostgreSQL "-                        <> Text.pack (show major)-                        <> " UUIDv7 generator: kiroku.uuidv7() via "-                        <> (if builtinExists then "0010's alias for the builtin" else "0001's fallback")-                    )-                generatorExists `shouldBe` True-                builtinExists `shouldBe` (major >= 18)-                leaseDefault `shouldBe` Just "kiroku.uuidv7()"-            either (expectationFailure . show) pure result--        it "acquires a lease from a session whose search_path excludes kiroku" $ do-            plan <- requirePlan-            result <- withMigratedDatabase plan $ \connection -> do-                useSession connection (Session.script "SET search_path TO pg_catalog")-                version <- useSession connection (Session.statement () insertLeaseStatement)-                version `shouldBe` 7-            either (expectationFailure . show) pure result--    describe "Codd history import" $ do-        it "imports a current codd V5 ledger, verifies, and never replays SQL" $-            importFixture "codd"--        it "imports the legacy codd_schema ledger shape" $-            importFixture "codd_schema"--        it "rejects a partial legacy row before creating the target ledger" $ do-            plan <- requirePlan-            withKirokuPg $ \database -> do-                let settings = Pg.connectionSettings database-                    provider = connectionProviderFromSettings settings-                withConnection settings $ \connection -> installCoddLedger connection "codd" True-                config <--                    requireRight-                        (kirokuCoddSourceConfig provider True "partial fixture must fail" Confirmed)-                imported <--                    importCoddHistory defaultImportOptions config provider plan kirokuCoddHistoryMappings-                imported `shouldSatisfy` \case-                    Left CoddPartialMigration{} -> True-                    _ -> False-                withConnection settings $ \connection -> do-                    targetExists <- useSession connection (Session.statement "pgmigrate" schemaExistsStatement)-                    targetExists `shouldBe` False--importFixture :: Text -> Expectation-importFixture sourceSchema = do-    plan <- requirePlan-    directory <- findMigrationsDirectory-    withKirokuPg $ \database -> do-        let settings = Pg.connectionSettings database-            provider = connectionProviderFromSettings settings-        withConnection settings $ \connection -> do-            applyNativeSqlFromDisk connection directory-            installCoddLedger connection sourceSchema False-        config <--            requireRight-                (kirokuCoddSourceConfig provider True "verified Kiroku Codd cutover" Confirmed)-        first <--            importCoddHistory defaultImportOptions config provider plan kirokuCoddHistoryMappings-                >>= requireRight-        importOutcomes first `shouldBe` replicate 7 Imported-        pendingIds <--            traverse-                (requireRight . migrationId "kiroku")-                ["0008-schema-management-comment", "0009", "0010", "0011", "0012", "0013", "0014", "0015"]-        verifiedBeforeCanary <- verifyMigrationPlan defaultRunOptions settings plan >>= requireMigration-        case verifiedBeforeCanary of-            VerificationReport verificationIssues _ _ _ ->-                verificationIssues-                    `shouldBe` (PendingMigration <$> pendingIds)-        up <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration-        reportOutcomes up `shouldBe` replicate 7 AlreadyApplied <> replicate (length nativeMigrationFiles - 7) AppliedNow-        verifiedAfterCanary <- verifyMigrationPlan defaultRunOptions settings plan >>= requireMigration-        case verifiedAfterCanary of-            VerificationReport verificationIssues _ _ _ ->-                verificationIssues `shouldBe` []-        rerun <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration-        reportOutcomes rerun `shouldBe` replicate (length nativeMigrationFiles) AlreadyApplied-        second <--            importCoddHistory defaultImportOptions config provider plan kirokuCoddHistoryMappings-                >>= requireRight-        importOutcomes second `shouldBe` replicate 7 AlreadyImported-        withConnection settings $ \connection -> do-            assertSchema connection-            sourceRows <- useSession connection (Session.statement () (sourceRowCountStatement sourceSchema))-            sourceRows `shouldBe` 7-            facts <- useSession connection (Session.statement () importFactsStatement)-            facts `shouldBe` (fromIntegral (length nativeMigrationFiles), 7, True)--nativeMigrationFiles :: [FilePath]-nativeMigrationFiles =-    [ "0001-kiroku-bootstrap.sql"-    , "0002-add-subscription-dead-letters.sql"-    , "0003-notify-trigger-append-guard.sql"-    , "0004-dead-letters-event-id-index.sql"-    , "0005-index-hygiene-and-streams-fillfactor.sql"-    , "0006-stream-name-length-check.sql"-    , "0007-stream-truncate-before.sql"-    , "0008-schema-management-comment.sql"-    , "0009.sql"-    , "0010.sql"-    , "0011.sql"-    , "0012.sql"-    , "0013.sql"-    , "0014.sql"-    , "0015.sql"-    ]--{- | The plan truncated to its first @count@ migrations, read from the checked-in-directory. It stands in for a database that a previous release bootstrapped: the-ledger it produces is byte-identical to the one the full plan would have written-for those entries, so the full plan later sees exactly the remainder as pending.--}-planThrough :: Int -> IO MigrationPlan-planThrough count = do-    directory <- findMigrationsDirectory-    entries <- traverse (readEntry directory) (take count nativeMigrationFiles)-    component <--        requireRight-            (migrationComponentFromEmbeddedSql "kiroku" mempty (NonEmpty.fromList entries))-    requireRight (migrationPlan (component :| []))-  where-    readEntry directory file = do-        bytes <- ByteString.readFile (directory </> file)-        pure (file, bytes)--findMigrationsDirectory :: IO FilePath-findMigrationsDirectory =-    findDirectory ["kiroku-store-migrations/migrations", "migrations"]--findLockfile :: IO FilePath-findLockfile =-    findFile ["kiroku-store-migrations/migrations.lock", "migrations.lock"]--findDirectory :: [FilePath] -> IO FilePath-findDirectory candidates = do-    existing <- filterM doesDirectoryExist candidates-    case existing of-        directory : _ -> pure directory-        [] -> expectationFailure ("could not find directory: " <> show candidates) >> pure "."--findFile :: [FilePath] -> IO FilePath-findFile candidates = do-    existing <- filterM doesFileExist candidates-    case existing of-        path : _ -> pure path-        [] -> expectationFailure ("could not find file: " <> show candidates) >> pure "."--filterM :: (value -> IO Bool) -> [value] -> IO [value]-filterM predicate = foldr step (pure [])-  where-    step value remaining = do-        matches <- predicate value-        values <- remaining-        pure (if matches then value : values else values)--parseLockfile :: Text -> [(FilePath, Text)]-parseLockfile contents =-    [ (Text.unpack filename, checksum)-    | line <- Text.lines contents-    , [checksum, filename] <- [Text.words line]-    ]--checksumText :: ByteString -> Text-checksumText =-    Text.pack-        . concatMap renderByte-        . ByteString.unpack-        . migrationChecksumBytes-        . migrationFingerprint-  where-    renderByte byte =-        case Numeric.showHex byte "" of-            [digit] -> ['0', digit]-            digits -> digits--dropSqlSuffix :: FilePath -> String-dropSqlSuffix = reverse . drop 4 . reverse--requirePlan :: IO MigrationPlan-requirePlan = requireRight kirokuMigrationPlan--requireRight :: (Show error) => Either error value -> IO value-requireRight = either (failure . show) pure--requireMigration :: (Show error) => Either error value -> IO value-requireMigration = requireRight--failure :: String -> IO value-failure message = expectationFailure message >> fail message--providerFor :: Connection.Connection -> ConnectionProvider-providerFor connection = connectionProvider (\action -> Right <$> action connection)--reportOutcomes :: MigrationReport -> [MigrationOutcome]-reportOutcomes MigrationReport{results} = outcome <$> toList results--importOutcomes :: HistoryImportReport -> [HistoryImportOutcome]-importOutcomes HistoryImportReport{importResults} = importOutcome <$> toList importResults--{- | Built on pg-migrate's stable per-user temporary root, which 'withMigratedDatabase'-also uses, so ephemeral-pg's startup sweep reclaims clusters left by killed runs.--}-kirokuPgConfig :: IO Pg.Config-kirokuPgConfig = do-    base <- defaultEphemeralConfig-    pure base{PgConfig.user = "kiroku"}--withKirokuPg :: (Pg.Database -> IO ()) -> IO ()-withKirokuPg action = do-    config <- kirokuPgConfig-    started <- Pg.startCached config Pg.defaultCacheConfig-    case started of-        Left startError -> expectationFailure (show startError)-        Right database -> action database `finally` Pg.stop database--withConnection :: Settings.Settings -> (Connection.Connection -> IO value) -> IO value-withConnection settings action = do-    acquired <- Connection.acquire settings-    connection <- requireRight acquired-    action connection `finally` Connection.release connection--useSession :: Connection.Connection -> Session.Session value -> IO value-useSession connection session =-    Connection.use connection session >>= requireRight--assertSchema :: Connection.Connection -> Expectation-assertSchema connection = do-    healthy <- useSession connection (Session.statement () schemaFactsStatement)-    healthy `shouldBe` True-    oversized <- Connection.use connection (Session.statement (Text.replicate 513 "x") oversizedStreamStatement)-    oversized `shouldSatisfy` isLeft--schemaFactsStatement :: Statement () Bool-schemaFactsStatement =-    Statement.preparable-        """-        SELECT bool_and(ok)-        FROM (VALUES-          (to_regnamespace('kiroku') IS NOT NULL),-          (to_regclass('kiroku.events') IS NOT NULL),-          (to_regclass('kiroku.streams') IS NOT NULL),-          (to_regclass('kiroku.dead_letters') IS NOT NULL),-          (EXISTS (SELECT 1 FROM pg_catalog.pg_trigger WHERE tgname = 'stream_events_notify_insert' AND NOT tgisinternal)),-          (EXISTS (SELECT 1 FROM pg_catalog.pg_indexes WHERE schemaname = 'kiroku' AND indexname = 'ix_dead_letters_event_id')),-          (EXISTS (SELECT 1 FROM pg_catalog.pg_indexes WHERE schemaname = 'kiroku' AND indexname = 'ix_stream_events_all_by_category')),-          (EXISTS (SELECT 1 FROM pg_catalog.pg_constraint WHERE conname = 'ck_stream_events_all_category')),-          (EXISTS (SELECT 1 FROM pg_catalog.pg_constraint WHERE conname = 'chk_streams_stream_name_length')),-          (EXISTS (SELECT 1 FROM pg_catalog.pg_attribute a JOIN pg_catalog.pg_class c ON c.oid = a.attrelid JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace WHERE n.nspname = 'kiroku' AND c.relname = 'streams' AND a.attname = 'truncate_before' AND NOT a.attisdropped)),-          (EXISTS (SELECT 1 FROM pg_catalog.pg_indexes WHERE schemaname = 'kiroku' AND indexname = 'ix_streams_browse_name' AND indexdef LIKE '%COLLATE "C"%' AND indexdef LIKE '%WHERE (stream_id <> 0)%')),-          (obj_description(to_regnamespace('kiroku'), 'pg_namespace') = 'Managed by pg-migrate component kiroku through 0015')-        ) AS checks(ok)-        """-        Encoders.noParams-        (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.bool)))--{- | One stream with one event, written with the five-column junction shape-every release before 0012 used: a home row and an @$all@ row, no category.--}-preCategoryFixtureSql :: Text-preCategoryFixtureSql =-    """-    INSERT INTO kiroku.streams (stream_name, stream_version) VALUES ('upgrade-1', 1);-    INSERT INTO kiroku.events (event_id, event_type, data)-    VALUES ('00000000-0000-7000-8000-000000000001', 'UpgradeFixture', '{}'::jsonb);-    INSERT INTO kiroku.stream_events-      (event_id, stream_id, stream_version, original_stream_id, original_stream_version)-    SELECT '00000000-0000-7000-8000-000000000001'::uuid, s.stream_id, 1, s.stream_id, 1-    FROM kiroku.streams AS s WHERE s.stream_name = 'upgrade-1'-    UNION ALL-    SELECT '00000000-0000-7000-8000-000000000001'::uuid, 0, 1, s.stream_id, 1-    FROM kiroku.streams AS s WHERE s.stream_name = 'upgrade-1';-    UPDATE kiroku.streams SET stream_version = 1 WHERE stream_id = 0;-    """--{- | After 0012: whether the fixture's @$all@ row carries its stream's category,-how many @$all@ rows lack one, whether the home row stayed NULL, the category-index definition, and the immutability trigger's enabled state.--}-categoryBackfillFactsStatement :: Statement () (Bool, Int64, Bool, Maybe Text, Text)-categoryBackfillFactsStatement =-    Statement.preparable-        """-        SELECT (SELECT se.category = s.category-                  FROM kiroku.stream_events AS se-                  JOIN kiroku.streams AS s ON s.stream_id = se.original_stream_id-                 WHERE se.stream_id = 0 AND s.stream_name = 'upgrade-1'),-               (SELECT count(*) FROM kiroku.stream_events-                 WHERE stream_id = 0 AND category IS NULL),-               (SELECT se.category IS NULL-                  FROM kiroku.stream_events AS se-                  JOIN kiroku.streams AS s ON s.stream_id = se.stream_id-                 WHERE s.stream_name = 'upgrade-1'),-               (SELECT indexdef FROM pg_catalog.pg_indexes-                 WHERE schemaname = 'kiroku' AND indexname = 'ix_stream_events_all_by_category'),-               (SELECT tgenabled::text FROM pg_catalog.pg_trigger-                 WHERE tgrelid = 'kiroku.stream_events'::regclass-                   AND tgname = 'no_update_stream_events')-        """-        Encoders.noParams-        ( Decoders.singleRow-            ( (,,,,)-                <$> Decoders.column (Decoders.nonNullable Decoders.bool)-                <*> Decoders.column (Decoders.nonNullable Decoders.int8)-                <*> Decoders.column (Decoders.nonNullable Decoders.bool)-                <*> Decoders.column (Decoders.nullable Decoders.text)-                <*> Decoders.column (Decoders.nonNullable Decoders.text)-            )-        )---- | An @$all@ row written the pre-0012 way, which the CHECK must now refuse.-insertAllRowWithoutCategoryStatement :: Statement () ()-insertAllRowWithoutCategoryStatement =-    Statement.preparable-        """-        WITH new_event AS (-          INSERT INTO kiroku.events (event_id, event_type, data)-          VALUES ('00000000-0000-7000-8000-000000000002', 'UpgradeFixture', '{}'::jsonb)-          RETURNING event_id-        )-        INSERT INTO kiroku.stream_events-          (event_id, stream_id, stream_version, original_stream_id, original_stream_version)-        SELECT new_event.event_id, 0, 2, s.stream_id, 2-        FROM new_event, kiroku.streams AS s-        WHERE s.stream_name = 'upgrade-1'-        """-        Encoders.noParams-        Decoders.noResult--{- | The server major, whether each UUIDv7 generator exists, and the stored-@lease_id@ default. Together these pin which route published-@kiroku.uuidv7()@ on the running server.--}-leaseDefaultStatement :: Statement () (Int32, Bool, Bool, Maybe Text)-leaseDefaultStatement =-    Statement.preparable-        """-        SELECT pg_catalog.current_setting('server_version_num')::int / 10000,-               to_regprocedure('pg_catalog.uuidv7()') IS NOT NULL,-               to_regprocedure('kiroku.uuidv7()') IS NOT NULL,-               (SELECT pg_catalog.pg_get_expr(stored.adbin, stored.adrelid)-                  FROM pg_catalog.pg_attrdef AS stored-                  JOIN pg_catalog.pg_attribute AS attribute-                    ON attribute.attrelid = stored.adrelid-                   AND attribute.attnum = stored.adnum-                 WHERE stored.adrelid = 'kiroku.history_retention_leases'::regclass-                   AND attribute.attname = 'lease_id')-        """-        Encoders.noParams-        ( Decoders.singleRow-            ( (,,,)-                <$> Decoders.column (Decoders.nonNullable Decoders.int4)-                <*> Decoders.column (Decoders.nonNullable Decoders.bool)-                <*> Decoders.column (Decoders.nonNullable Decoders.bool)-                <*> Decoders.column (Decoders.nullable Decoders.text)-            )-        )---- | Insert one lease and report the UUID version its default generated.-insertLeaseStatement :: Statement () Int32-insertLeaseStatement =-    Statement.preparable-        """-        INSERT INTO kiroku.history_retention_leases-            (owner, reason, protected_through, created_at, renewed_at, expires_at)-        VALUES-            ('kiroku-store-migrations-test', 'search_path independence', 0,-             pg_catalog.now(), pg_catalog.now(), pg_catalog.now() + interval '1 hour')-        RETURNING pg_catalog.get_byte(pg_catalog.uuid_send(lease_id), 6) >> 4-        """-        Encoders.noParams-        (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int4)))--historyRetentionColumnsStatement :: Statement () [(Text, Text, Text, Bool)]-historyRetentionColumnsStatement =-    Statement.preparable-        """-        SELECT relation.relname::text,-               attribute.attname::text,-               pg_catalog.format_type(attribute.atttypid, attribute.atttypmod),-               attribute.attnotnull-        FROM pg_catalog.pg_class AS relation-        JOIN pg_catalog.pg_namespace AS namespace-          ON namespace.oid = relation.relnamespace-        JOIN pg_catalog.pg_attribute AS attribute-          ON attribute.attrelid = relation.oid-        WHERE namespace.nspname = 'kiroku'-          AND relation.relname IN ('history_retention_coordinator', 'history_retention_leases')-          AND attribute.attnum > 0-          AND NOT attribute.attisdropped-        ORDER BY relation.relname, attribute.attnum-        """-        Encoders.noParams-        ( Decoders.rowList-            ( (,,,)-                <$> column Decoders.text-                <*> column Decoders.text-                <*> column Decoders.text-                <*> column Decoders.bool-            )-        )-  where-    column = Decoders.column . Decoders.nonNullable--historyRetentionFactsStatement :: Statement () (Int64, Int64, Bool, Bool, Int64)-historyRetentionFactsStatement =-    Statement.preparable-        """-        SELECT-          (SELECT count(*) FROM kiroku.history_retention_coordinator WHERE singleton),-          (SELECT count(*)-             FROM pg_catalog.pg_constraint AS con_record-             JOIN pg_catalog.pg_class AS relation ON relation.oid = con_record.conrelid-             JOIN pg_catalog.pg_namespace AS namespace ON namespace.oid = relation.relnamespace-            WHERE namespace.nspname = 'kiroku'-              AND relation.relname IN ('history_retention_coordinator', 'history_retention_leases')-              AND con_record.conname IN (-                'history_retention_coordinator_pkey',-                'chk_history_retention_coordinator_singleton',-                'history_retention_leases_pkey',-                'chk_history_retention_lease_owner_bytes',-                'chk_history_retention_lease_reason_bytes',-                'chk_history_retention_lease_frontier',-                'chk_history_retention_lease_renewal_time',-                'chk_history_retention_lease_expiry',-                'chk_history_retention_lease_release_time'-              )),-          EXISTS (-            SELECT 1-            FROM pg_catalog.pg_indexes-            WHERE schemaname = 'kiroku'-              AND indexname = 'ix_history_retention_leases_unreleased_expiry'-              AND indexdef LIKE '%WHERE (released_at IS NULL)'-          ),-          to_regprocedure('kiroku.protect_replay_history_from_destruction()') IS NOT NULL,-          (SELECT count(*)-             FROM pg_catalog.pg_trigger AS trigger_record-             JOIN pg_catalog.pg_class AS relation ON relation.oid = trigger_record.tgrelid-             JOIN pg_catalog.pg_namespace AS namespace ON namespace.oid = relation.relnamespace-            WHERE namespace.nspname = 'kiroku'-              AND relation.relname IN ('events', 'stream_events', 'streams')-              AND trigger_record.tgname IN ('protect_replay_history_delete', 'protect_replay_history_truncate')-              AND NOT trigger_record.tgisinternal)-        """-        Encoders.noParams-        ( Decoders.singleRow-            ( (,,,,)-                <$> column Decoders.int8-                <*> column Decoders.int8-                <*> column Decoders.bool-                <*> column Decoders.bool-                <*> column Decoders.int8-            )-        )-  where-    column = Decoders.column . Decoders.nonNullable--oversizedStreamStatement :: Statement Text ()-oversizedStreamStatement =-    Statement.preparable-        "INSERT INTO kiroku.streams (stream_name, stream_version) VALUES ($1, 0)"-        (Encoders.param (Encoders.nonNullable Encoders.text))-        Decoders.noResult--relationCatalogStatement :: Statement () (Text, Text, Bool, Text, Text, Text)-relationCatalogStatement =-    Statement.preparable-        """-        SELECT c.relkind::text,-               COALESCE(array_to_string(c.reloptions, ','), ''),-               c.relowner = source.relowner,-               view_contract.is_updatable::text,-               view_contract.is_insertable_into::text,-               pg_catalog.obj_description(c.oid, 'pg_class')-        FROM pg_catalog.pg_class AS c-        JOIN pg_catalog.pg_namespace AS n ON n.oid = c.relnamespace-        JOIN pg_catalog.pg_class AS source-          ON source.oid = to_regclass('kiroku.subscriptions')-        JOIN information_schema.views AS view_contract-          ON view_contract.table_schema = n.nspname-         AND view_contract.table_name = c.relname-        WHERE n.nspname = 'kiroku'-          AND c.relname = 'subscription_checkpoints_v1'-        """-        Encoders.noParams-        ( Decoders.singleRow-            ( (,,,,,)-                <$> column Decoders.text-                <*> column Decoders.text-                <*> column Decoders.bool-                <*> column Decoders.text-                <*> column Decoders.text-                <*> column Decoders.text-            )-        )-  where-    column = Decoders.column . Decoders.nonNullable--relationColumnsStatement :: Statement () [(Text, Text, Bool, Text)]-relationColumnsStatement =-    Statement.preparable-        """-        SELECT attribute.attname::text,-               pg_catalog.format_type(attribute.atttypid, attribute.atttypmod),-               attribute.attnotnull,-               pg_catalog.col_description(relation.oid, attribute.attnum)-        FROM pg_catalog.pg_class AS relation-        JOIN pg_catalog.pg_namespace AS namespace-          ON namespace.oid = relation.relnamespace-        JOIN pg_catalog.pg_attribute AS attribute-          ON attribute.attrelid = relation.oid-        WHERE namespace.nspname = 'kiroku'-          AND relation.relname = 'subscription_checkpoints_v1'-          AND attribute.attnum > 0-          AND NOT attribute.attisdropped-        ORDER BY attribute.attnum-        """-        Encoders.noParams-        ( Decoders.rowList-            ( (,,,)-                <$> column Decoders.text-                <*> column Decoders.text-                <*> column Decoders.bool-                <*> column Decoders.text-            )-        )-  where-    column = Decoders.column . Decoders.nonNullable--checkpointRowsStatement :: Statement () [(Text, Int32, Int64, UTCTime)]-checkpointRowsStatement =-    Statement.preparable-        """-        SELECT subscription_name,-               consumer_group_member,-               checkpoint_position,-               checkpoint_updated_at-        FROM kiroku.subscription_checkpoints_v1-        ORDER BY subscription_name, consumer_group_member-        """-        Encoders.noParams-        ( Decoders.rowList-            ( (,,,)-                <$> column Decoders.text-                <*> column Decoders.int4-                <*> column Decoders.int8-                <*> column Decoders.timestamptz-            )-        )-  where-    column = Decoders.column . Decoders.nonNullable--checkpointFixtureTimestamp :: UTCTime-checkpointFixtureTimestamp =-    UTCTime-        (fromGregorian 2026 8 13)-        (secondsToDiffTime (12 * 60 * 60 + 34 * 60 + 56))--checkpointFixtureSql :: Text-checkpointFixtureSql =-    """-    INSERT INTO kiroku.subscriptions-      (subscription_name, consumer_group_member, last_seen, updated_at)-    VALUES-      ('ordinary', 0, 11, '2026-08-13 12:34:56+00'),-      ('grouped', 0, 20, '2026-08-13 12:34:56+00'),-      ('grouped', 1, 21, '2026-08-13 12:34:56+00'),-      ('grouped', 2, 22, '2026-08-13 12:34:56+00')-    """--checkpointAdvanceSql :: Text-checkpointAdvanceSql =-    """-    UPDATE kiroku.subscriptions-    SET last_seen = 5-    WHERE subscription_name = 'ordinary'-      AND consumer_group_member = 0-    """--checkpointRegressionSql :: Text-checkpointRegressionSql =-    """-    UPDATE kiroku.subscriptions-    SET last_seen = 2-    WHERE subscription_name = 'ordinary'-      AND consumer_group_member = 0-    """--checkpointPosition :: Connection.Connection -> IO Int64-checkpointPosition connection =-    useSession connection (Session.statement "ordinary" checkpointPositionStatement)--checkpointPositionStatement :: Statement Text Int64-checkpointPositionStatement =-    Statement.preparable-        """-        SELECT checkpoint_position-        FROM kiroku.subscription_checkpoints_v1-        WHERE subscription_name = $1-          AND consumer_group_member = 0-        """-        (Encoders.param (Encoders.nonNullable Encoders.text))-        (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8)))--withTemporaryReaderRole :: Connection.Connection -> (Text -> IO value) -> IO value-withTemporaryReaderRole connection action = do-    unique <- newUnique-    let role =-            "kiroku_checkpoint_reader_"-                <> Text.replace "-" "n" (Text.pack (show (hashUnique unique)))-        identifier = quoteTestIdentifier role-    useSession connection (Session.script ("CREATE ROLE " <> identifier))-    action role-        `finally` useSession-            connection-            ( Session.script-                ( "RESET ROLE; DROP OWNED BY "-                    <> identifier-                    <> "; DROP ROLE "-                    <> identifier-                )-            )--quoteTestIdentifier :: Text -> Text-quoteTestIdentifier value =-    "\"" <> Text.replace "\"" "\"\"" value <> "\""--checkpointRowCountStatement :: Statement () Int64-checkpointRowCountStatement =-    Statement.preparable-        "SELECT count(*) FROM kiroku.subscription_checkpoints_v1"-        Encoders.noParams-        (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8)))--privateTablePrivilegeStatement :: Statement () Bool-privateTablePrivilegeStatement =-    Statement.preparable-        "SELECT has_table_privilege(current_user, 'kiroku.subscriptions', 'SELECT')"-        Encoders.noParams-        (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.bool)))--privateTableCountStatement :: Statement () Int64-privateTableCountStatement =-    Statement.preparable-        "SELECT count(*) FROM kiroku.subscriptions"-        Encoders.noParams-        (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8)))--updatePublicViewStatement :: Statement () ()-updatePublicViewStatement =-    Statement.unpreparable-        """-        UPDATE kiroku.subscription_checkpoints_v1-        SET checkpoint_position = checkpoint_position-        """-        Encoders.noParams-        Decoders.noResult--hasSqlState :: Text -> Either Errors.SessionError value -> Bool-hasSqlState expected = \case-    Left-        ( Errors.StatementSessionError-                _-                _-                _-                _-                _-                (Errors.ServerStatementError (Errors.ServerError actual _ _ _ _))-            ) -> actual == expected-    _ -> False--downstreamReplacementFixtureSql :: Text-downstreamReplacementFixtureSql =-    """-    INSERT INTO kiroku.subscriptions-      (subscription_name, consumer_group_member, last_seen,-       updated_at)-    VALUES ('dependency', 0, 42, '2026-08-13 12:34:56+00');--    CREATE VIEW public.subscription_checkpoint_floors AS-    SELECT subscription_name,-           min(checkpoint_position) AS checkpoint_floor-    FROM kiroku.subscription_checkpoints_v1-    GROUP BY subscription_name;--    CREATE TABLE kiroku.subscription_checkpoint_storage_v2 (-      subscription_name text NOT NULL,-      consumer_group_member integer NOT NULL,-      checkpoint_position bigint NOT NULL,-      checkpoint_updated_at timestamptz NOT NULL,-      PRIMARY KEY (subscription_name, consumer_group_member)-    );--    INSERT INTO kiroku.subscription_checkpoint_storage_v2-    SELECT subscription_name,-           consumer_group_member,-           last_seen,-           updated_at-    FROM kiroku.subscriptions;--    CREATE OR REPLACE VIEW kiroku.subscription_checkpoints_v1-        (subscription_name,-         consumer_group_member,-         checkpoint_position,-         checkpoint_updated_at)-    WITH (security_invoker = false)-    AS-    WITH checkpoint_rows AS NOT MATERIALIZED (-        SELECT subscription_name,-               consumer_group_member,-               checkpoint_position,-               checkpoint_updated_at-        FROM kiroku.subscription_checkpoint_storage_v2-    )-    SELECT subscription_name,-           consumer_group_member,-           checkpoint_position,-           checkpoint_updated_at-    FROM checkpoint_rows;--    DROP TABLE kiroku.subscriptions;-    """--downstreamFloorsStatement :: Statement () [(Text, Int64)]-downstreamFloorsStatement =-    Statement.preparable-        """-        SELECT subscription_name, checkpoint_floor-        FROM public.subscription_checkpoint_floors-        ORDER BY subscription_name-        """-        Encoders.noParams-        ( Decoders.rowList-            ( (,)-                <$> Decoders.column (Decoders.nonNullable Decoders.text)-                <*> Decoders.column (Decoders.nonNullable Decoders.int8)-            )-        )--checkpointPlanFixtureSql :: Text-checkpointPlanFixtureSql =-    """-    INSERT INTO kiroku.subscriptions-      (subscription_name, consumer_group_member, last_seen)-    SELECT 'subscription-' || lpad((value % 100)::text, 3, '0'),-           value::integer,-           value::bigint-    FROM generate_series(1, 10000) AS generated(value);--    ANALYZE kiroku.subscriptions;-    """--serverVersionStatement :: Statement () Text-serverVersionStatement =-    Statement.preparable-        "SELECT current_setting('server_version')"-        Encoders.noParams-        (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.text)))--checkpointExplainStatement :: Statement () [Text]-checkpointExplainStatement =-    Statement.preparable-        """-        EXPLAIN (COSTS OFF)-        SELECT min(checkpoint_position)-        FROM kiroku.subscription_checkpoints_v1-        WHERE subscription_name = 'subscription-042'-        """-        Encoders.noParams-        (Decoders.rowList (Decoders.column (Decoders.nonNullable Decoders.text)))--applyNativeSqlFromDisk :: Connection.Connection -> FilePath -> IO ()-applyNativeSqlFromDisk connection directory =-    forM_ (take 7 nativeMigrationFiles) $ \file -> do-        sql <- Text.IO.readFile (directory </> file)-        useSession connection (Session.script sql)--installCoddLedger :: Connection.Connection -> Text -> Bool -> IO ()-installCoddLedger connection sourceSchema partial =-    useSession connection (Session.script (coddFixtureSql sourceSchema partial))--coddFixtureSql :: Text -> Bool -> Text-coddFixtureSql sourceSchema partial =-    Text.unlines-        [ "CREATE SCHEMA " <> sourceSchema <> ";"-        , "CREATE TABLE " <> sourceSchema <> ".sql_migrations ("-        , "  id serial NOT NULL, migration_timestamp timestamptz NOT NULL,"-        , "  applied_at timestamptz, name text NOT NULL, application_duration interval,"-        , "  num_applied_statements int, no_txn_failed_at timestamptz, txnid bigint, connid int"-        , ");"-        , "INSERT INTO " <> sourceSchema <> ".sql_migrations"-        , "  (migration_timestamp, applied_at, name, application_duration, num_applied_statements, no_txn_failed_at, txnid, connid) VALUES"-        , Text.intercalate ",\n" (zipWith renderRow [1 :: Int ..] (toList kirokuLegacyMigrationNames)) <> ";"-        ]-  where-    renderRow index filename =-        "('2026-01-01 00:00:00+00'::timestamptz + interval '"-            <> Text.pack (show index)-            <> " seconds', "-            <> appliedAt index-            <> ", '"-            <> Text.pack filename-            <> "', interval '1 second', 1, "-            <> failureAt index-            <> ", 1, 1)"-    appliedAt index-        | partial && index == 4 = "NULL"-        | otherwise = "'2026-01-01 00:01:00+00'::timestamptz + interval '" <> Text.pack (show index) <> " seconds'"-    failureAt index-        | partial && index == 4 = "'2026-01-01 00:02:00+00'::timestamptz"-        | otherwise = "NULL"--schemaExistsStatement :: Statement Text Bool-schemaExistsStatement =-    Statement.preparable-        "SELECT to_regnamespace($1) IS NOT NULL"-        (Encoders.param (Encoders.nonNullable Encoders.text))-        (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.bool)))--sourceRowCountStatement :: Text -> Statement () Int64-sourceRowCountStatement sourceSchema =-    Statement.unpreparable-        ("SELECT count(*) FROM " <> sourceSchema <> ".sql_migrations")-        Encoders.noParams-        (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8)))--importFactsStatement :: Statement () (Int64, Int64, Bool)-importFactsStatement =-    Statement.preparable-        """-        SELECT-          (SELECT count(*) FROM pgmigrate.migrations),-          (SELECT count(*) FROM pgmigrate.history_imports),-          (SELECT bool_and(source_evidence #>> '{satisfying_evidence,0,details,adapter}' = 'codd') FROM pgmigrate.history_imports)-        """-        Encoders.noParams-        ( Decoders.singleRow-            ( (,,)-                <$> column Decoders.int8-                <*> column Decoders.int8-                <*> column Decoders.bool-            )-        )-  where-    column = Decoders.column . Decoders.nonNullable--targetUpgradeFactsStatement :: Statement () (Int64, Text)-targetUpgradeFactsStatement =-    Statement.preparable-        "SELECT pg_relation_filenode('kiroku.subscriptions')::bigint, ctid::text FROM kiroku.subscriptions WHERE subscription_name = 'legacy-target'"-        Encoders.noParams-        (Decoders.singleRow ((,) <$> Decoders.column (Decoders.nonNullable Decoders.int8) <*> Decoders.column (Decoders.nonNullable Decoders.text)))--targetBindingFactsStatement :: Statement () (Text, Maybe Text, Int64, Bool)-targetBindingFactsStatement =-    Statement.preparable-        """-        SELECT target_kind, target_category, last_seen,-          EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema = 'kiroku' AND table_name = 'subscriptions' AND column_name = 'stream_name')-        FROM kiroku.subscriptions WHERE subscription_name = 'legacy-target'-        """-        Encoders.noParams-        (Decoders.singleRow ((,,,) <$> Decoders.column (Decoders.nonNullable Decoders.text) <*> Decoders.column (Decoders.nullable Decoders.text) <*> Decoders.column (Decoders.nonNullable Decoders.int8) <*> Decoders.column (Decoders.nonNullable Decoders.bool)))+  describe "native Kiroku migration definition" $ do+    it "tracks the sixteen native files in manifest order" $ do+      directory <- findMigrationsDirectory+      manifest <- Text.lines <$> Text.IO.readFile (directory </> "manifest")+      manifest `shouldBe` Text.pack <$> nativeMigrationFiles++    it "preserves every legacy payload byte recorded by migrations.lock" $ do+      directory <- findMigrationsDirectory+      lockPath <- findLockfile+      lockEntries <- parseLockfile <$> Text.IO.readFile lockPath+      forM_ (zip (toList kirokuLegacyMigrationNames) nativeMigrationFiles) $ \(legacyName, nativeName) -> do+        bytes <- ByteString.readFile (directory </> nativeName)+        lookup legacyName lockEntries `shouldBe` Just (checksumText bytes)++    it "builds component kiroku and a sixteen-migration plan" $ do+      component <- requireRight kirokuMigrations+      component `seq` pure ()+      plan <- requirePlan+      let targetIds =+            [ requireRight (migrationId "kiroku" (Text.pack (dropSqlSuffix file)))+            | file <- nativeMigrationFiles+            ]+      validateHistoryMappingTargets plan kirokuCoddHistoryMappings+        `shouldBe` Right ()+      length targetIds `shouldBe` length nativeMigrationFiles++  describe "native migration authoring" $ do+    it "creates the next numeric file and atomically appends the manifest" $+      withSystemTempDirectory "kiroku-native-authoring" $ \directory -> do+        ByteString.writeFile (directory </> "0007-existing.sql") "SELECT 7;\n"+        ByteString.writeFile (directory </> "manifest") "0007-existing.sql\n"+        created <- newMigrationFile directory "add widget index"+        path <- requireRight created+        path `shouldBe` directory </> "0008.sql"+        body <- ByteString.readFile path+        body `shouldSatisfy` ByteString.isInfixOf "forward-only"+        Text.lines <$> Text.IO.readFile (directory </> "manifest")+          `shouldReturn` ["0007-existing.sql", "0008.sql"]++    it "refuses to overwrite a pre-existing inferred migration" $+      withSystemTempDirectory "kiroku-native-exclusive" $ \directory -> do+        ByteString.writeFile (directory </> "manifest") "0007-existing.sql\n"+        ByteString.writeFile (directory </> "0007-existing.sql") "SELECT 7;\n"+        ByteString.writeFile (directory </> "0008.sql") "SELECT 8;\n"+        created <- newMigrationFile directory "must not overwrite"+        created `shouldSatisfy` isLeft+        Text.IO.readFile (directory </> "manifest")+          `shouldReturn` "0007-existing.sql\n"++  describe "fresh native databases" $ do+    it "applies all sixteen, verifies strictly, and reports AlreadyApplied on rerun" $ do+      plan <- requirePlan+      result <- withMigratedDatabase plan $ \connection -> do+        assertSchema connection+        let provider = providerFor connection+        rerun <- runMigrationPlanWith defaultRunOptions provider plan >>= requireMigration+        reportOutcomes rerun `shouldBe` replicate (length nativeMigrationFiles) AlreadyApplied+        verified <- verifyMigrationPlanWith defaultRunOptions provider plan >>= requireMigration+        case verified of+          VerificationReport verificationIssues applied _ _ -> do+            verificationIssues `shouldBe` []+            length applied `shouldBe` length nativeMigrationFiles+      either (expectationFailure . show) pure result++    it "serializes concurrent applies through the pg-migrate advisory lock" $ do+      plan <- requirePlan+      withKirokuPg $ \database -> do+        let settings = Pg.connectionSettings database+        (first, second) <-+          concurrently+            (runMigrationPlan defaultRunOptions settings plan >>= requireMigration)+            (runMigrationPlan defaultRunOptions settings plan >>= requireMigration)+        sort [reportOutcomes first, reportOutcomes second]+          `shouldBe` sort+            [ replicate (length nativeMigrationFiles) AppliedNow,+              replicate (length nativeMigrationFiles) AlreadyApplied+            ]++  describe "subscription checkpoint SQL relation" $ do+    it "publishes the frozen ordinary-view catalog contract" $ do+      plan <- requirePlan+      result <- withMigratedDatabase plan $ \connection -> do+        relation <- useSession connection (Session.statement () relationCatalogStatement)+        relation+          `shouldBe` ( "v",+                       "security_invoker=false",+                       True,+                       "NO",+                       "NO",+                       "Stable read-only v1 relation of exact persisted subscription-member checkpoints; its columns and order are frozen, and its rows are unordered unless the caller supplies ORDER BY."+                     )+        columns <- useSession connection (Session.statement () relationColumnsStatement)+        columns+          `shouldBe` [ ( "subscription_name",+                         "text",+                         False,+                         "Persisted subscription name; member zero does not distinguish a non-group subscription from member zero of a consumer group."+                       ),+                       ( "consumer_group_member",+                         "integer",+                         False,+                         "Persisted consumer-group member key; member zero carries no topology classification."+                       ),+                       ( "checkpoint_position",+                         "bigint",+                         False,+                         "Exact persisted global position for this subscription member; an explicit reset may move it backward or forward."+                       ),+                       ( "checkpoint_updated_at",+                         "timestamp with time zone",+                         False,+                         "Time of the latest checkpoint-row upsert; it does not imply position advancement or worker liveness."+                       )+                     ]+      either (expectationFailure . show) pure result++    it "returns zero rows for an empty checkpoint inventory" $ do+      plan <- requirePlan+      result <- withMigratedDatabase plan $ \connection -> do+        rows <- useSession connection (Session.statement () checkpointRowsStatement)+        rows `shouldBe` []+      either (expectationFailure . show) pure result++    it "returns exact non-null member rows and exposes reassignment only after commit" $ do+      plan <- requirePlan+      withKirokuPg $ \database -> do+        let settings = Pg.connectionSettings database+        _ <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration+        withConnection settings $ \writer ->+          withConnection settings $ \reader -> do+            useSession writer (Session.script checkpointFixtureSql)+            rows <- useSession reader (Session.statement () checkpointRowsStatement)+            rows+              `shouldBe` [ ("grouped", 0, 20, checkpointFixtureTimestamp),+                           ("grouped", 1, 21, checkpointFixtureTimestamp),+                           ("grouped", 2, 22, checkpointFixtureTimestamp),+                           ("ordinary", 0, 11, checkpointFixtureTimestamp)+                         ]++            useSession writer (Session.script "BEGIN")+            useSession writer (Session.script checkpointAdvanceSql)+            checkpointPosition writer `shouldReturn` 5+            checkpointPosition reader `shouldReturn` 11+            useSession writer (Session.script "COMMIT")+            checkpointPosition reader `shouldReturn` 5++            useSession writer (Session.script "BEGIN")+            useSession writer (Session.script checkpointRegressionSql)+            checkpointPosition writer `shouldReturn` 2+            checkpointPosition reader `shouldReturn` 5+            useSession writer (Session.script "ROLLBACK")+            checkpointPosition reader `shouldReturn` 5++    it "allows view-only readers while structurally rejecting owner updates" $ do+      plan <- requirePlan+      result <- withMigratedDatabase plan $ \connection ->+        withTemporaryReaderRole connection $ \role -> do+          let identifier = quoteTestIdentifier role+          useSession+            connection+            ( Session.script+                ( "GRANT USAGE ON SCHEMA kiroku TO "+                    <> identifier+                    <> "; GRANT SELECT ON kiroku.subscription_checkpoints_v1 TO "+                    <> identifier+                    <> "; SET ROLE "+                    <> identifier+                )+            )+          viewCount <- useSession connection (Session.statement () checkpointRowCountStatement)+          viewCount `shouldBe` 0+          privatePrivilege <- useSession connection (Session.statement () privateTablePrivilegeStatement)+          privatePrivilege `shouldBe` False+          directPrivateRead <- Connection.use connection (Session.statement () privateTableCountStatement)+          directPrivateRead `shouldSatisfy` hasSqlState "42501"++          useSession connection (Session.script "RESET ROLE")+          ownerUpdate <- Connection.use connection (Session.statement () updatePublicViewStatement)+          ownerUpdate `shouldSatisfy` hasSqlState "55000"+      either (expectationFailure . show) pure result++    it "preserves a downstream view while private checkpoint storage is replaced" $ do+      plan <- requirePlan+      withKirokuPg $ \database -> do+        let settings = Pg.connectionSettings database+        _ <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration+        withConnection settings $ \connection -> do+          useSession connection (Session.script downstreamReplacementFixtureSql)+          floors <- useSession connection (Session.statement () downstreamFloorsStatement)+          floors `shouldBe` [("dependency", 42)]++    it "pushes a subscription filter to the existing checkpoint index" $ do+      plan <- requirePlan+      result <- withMigratedDatabase plan $ \connection -> do+        useSession connection (Session.script checkpointPlanFixtureSql)+        version <- useSession connection (Session.statement () serverVersionStatement)+        planLines <- useSession connection (Session.statement () checkpointExplainStatement)+        let planText = Text.unlines planLines+        Text.IO.putStrLn ("PostgreSQL " <> version <> " checkpoint relation plan:\n" <> planText)+        planText `shouldSatisfy` Text.isInfixOf "ix_subscriptions_name_member"+        Text.isInfixOf "CTE Scan" planText `shouldBe` False+      either (expectationFailure . show) pure result++  describe "history retention schema" $ do+    it "publishes the exact coordinator and lease table columns" $ do+      plan <- requirePlan+      result <- withMigratedDatabase plan $ \connection -> do+        columns <- useSession connection (Session.statement () historyRetentionColumnsStatement)+        columns+          `shouldBe` [ ("history_retention_coordinator", "singleton", "boolean", True),+                       ("history_retention_leases", "lease_id", "uuid", True),+                       ("history_retention_leases", "owner", "text", True),+                       ("history_retention_leases", "reason", "text", True),+                       ("history_retention_leases", "protected_through", "bigint", True),+                       ("history_retention_leases", "created_at", "timestamp with time zone", True),+                       ("history_retention_leases", "renewed_at", "timestamp with time zone", True),+                       ("history_retention_leases", "expires_at", "timestamp with time zone", True),+                       ("history_retention_leases", "released_at", "timestamp with time zone", False)+                     ]+      either (expectationFailure . show) pure result++    it "installs the singleton, checks, partial index, function, and six triggers" $ do+      plan <- requirePlan+      result <- withMigratedDatabase plan $ \connection -> do+        facts <- useSession connection (Session.statement () historyRetentionFactsStatement)+        facts `shouldBe` (1, 9, True, True, 6)+      either (expectationFailure . show) pure result++  -- BUG-1. Migrations 0002 onward are parsed in sessions that never ran 0001,+  -- so a name that only 0001's `SET search_path` makes resolvable parses on a+  -- fresh install and fails on every ordinary upgrade. 0.3.2.x's 0010 named+  -- uuidv7() that way and could not be applied to a database bootstrapped+  -- through 0009 on PostgreSQL 17. These cases apply the tail of the plan in a+  -- session that carries none of 0001's state.+  describe "upgrades of already-bootstrapped databases" $ do+    it "upgrades namespace constraints from 0015 and reruns without touching bound rows" $ do+      oldPlan <- planThrough 15+      plan <- requirePlan+      withKirokuPg $ \database -> do+        let settings = Pg.connectionSettings database+        _ <- runMigrationPlan defaultRunOptions settings oldPlan >>= requireMigration+        withConnection settings $ \connection ->+          useSession connection (Session.script "INSERT INTO kiroku.subscriptions(subscription_name, last_seen, target_kind, target_category) VALUES ('legacy-category', 42, 'category', 'pm')")+        upgraded <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration+        reportOutcomes upgraded `shouldBe` replicate 15 AlreadyApplied <> [AppliedNow]+        withConnection settings $ \connection ->+          useSession connection (Session.script "INSERT INTO kiroku.subscriptions(subscription_name, last_seen, target_kind, target_category) VALUES ('namespace', 0, 'category_namespace', 'pm'); DO $$ BEGIN IF NOT EXISTS(SELECT FROM kiroku.subscriptions WHERE subscription_name='legacy-category' AND last_seen=42 AND target_kind='category' AND target_category='pm') THEN RAISE EXCEPTION 'binding changed'; END IF; END $$")+        rerun <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration+        reportOutcomes rerun `shouldBe` replicate 16 AlreadyApplied++    it "applies the pending tail in a session that never ran the bootstrap" $ do+      plan <- requirePlan+      let bootstrapCount = 9 -- through 0009, so 0010 is in the pending tail+      throughBootstrap <- planThrough bootstrapCount+      withKirokuPg $ \database -> do+        let settings = Pg.connectionSettings database+        bootstrapped <-+          runMigrationPlan defaultRunOptions settings throughBootstrap+            >>= requireMigration+        reportOutcomes bootstrapped+          `shouldBe` replicate bootstrapCount AppliedNow++        -- A separate session, and one that cannot reach the Kiroku+        -- schema through search_path. The suite connects as role+        -- "kiroku", so the default "$user" entry resolves to the Kiroku+        -- schema and would hide exactly the name resolution this case+        -- exists to prove -- the same accident that let 0001's+        -- SET search_path carry a fresh install through 0.3.2.x's 0010.+        withConnection settings $ \upgradeSession -> do+          useSession upgradeSession (Session.script "SET search_path TO pg_catalog")+          upgraded <-+            runMigrationPlanWith defaultRunOptions (providerFor upgradeSession) plan+              >>= requireMigration+          reportOutcomes upgraded+            `shouldBe` replicate bootstrapCount AlreadyApplied+              <> replicate (length nativeMigrationFiles - bootstrapCount) AppliedNow++        verified <- verifyMigrationPlan defaultRunOptions settings plan >>= requireMigration+        case verified of+          VerificationReport verificationIssues applied _ _ -> do+            verificationIssues `shouldBe` []+            length applied `shouldBe` length nativeMigrationFiles+        withConnection settings assertSchema++    it "adds unbound target metadata without rewriting legacy rows and drops stream_name" $ do+      plan <- requirePlan+      throughPrevious <- planThrough 13+      withKirokuPg $ \database -> do+        let settings = Pg.connectionSettings database+        _ <- runMigrationPlan defaultRunOptions settings throughPrevious >>= requireMigration+        withConnection settings $ \connection -> do+          useSession connection (Session.script "INSERT INTO kiroku.subscriptions (subscription_name, last_seen) VALUES ('legacy-target', 7)")+          before <- useSession connection (Session.statement () targetUpgradeFactsStatement)+          _ <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration+          after <- useSession connection (Session.statement () targetUpgradeFactsStatement)+          -- Relation filenode and tuple ctid prove no heap rewrite.+          after `shouldBe` before+          facts <- useSession connection (Session.statement () targetBindingFactsStatement)+          facts `shouldBe` ("unbound", Nothing, 7, False)++    it "derives group topology from legacy rows when 0013 upgrades a populated store" $ do+      plan <- requirePlan+      throughPrevious <- planThrough 12+      withKirokuPg $ \database -> do+        let settings = Pg.connectionSettings database+        _ <- runMigrationPlan defaultRunOptions settings throughPrevious >>= requireMigration+        withConnection settings $ \connection ->+          useSession connection (Session.script "INSERT INTO kiroku.subscriptions (subscription_name, consumer_group_member, last_seen) VALUES ('group', 0, 5), ('group', 1, 20), ('incomplete', 0, 8), ('ordinary', 0, 10)")+        upgraded <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration+        reportOutcomes upgraded `shouldBe` replicate 12 AlreadyApplied <> replicate (length nativeMigrationFiles - 12) AppliedNow+        withConnection settings $ \connection -> do+          let stmt =+                Statement.preparable+                  "SELECT subscription_name, consumer_group_member, consumer_group_size, last_seen FROM kiroku.subscriptions ORDER BY subscription_name, consumer_group_member"+                  Encoders.noParams+                  (Decoders.rowList ((,,,) <$> Decoders.column (Decoders.nonNullable Decoders.text) <*> Decoders.column (Decoders.nonNullable Decoders.int4) <*> Decoders.column (Decoders.nonNullable Decoders.int4) <*> Decoders.column (Decoders.nonNullable Decoders.int8)))+          facts <- useSession connection (Session.statement () stmt)+          facts `shouldBe` [("group" :: Text, 0 :: Int32, 2 :: Int32, 5 :: Int64), ("group", 1, 2, 20), ("incomplete", 0, 1, 8), ("ordinary", 0, 1, 10)]+          directory <- findMigrationsDirectory+          sql <- Text.IO.readFile (directory </> "0013.sql")+          useSession connection (Session.script sql)+          useSession connection (Session.statement () stmt) `shouldReturn` facts++    -- BUG-2. 0012 copies each $all row's originating-stream category onto the+    -- junction row so category reads can range-scan an index. A store+    -- written before 0012 has $all rows without it; the backfill must fill+    -- every one, and afterwards the CHECK must refuse a new $all row that+    -- omits it, the immutability trigger must be back on, and the index must+    -- exist in exactly the shape the category reads rely on.+    it "backfills $all-row categories when 0012 upgrades a populated store" $ do+      plan <- requirePlan+      let throughCount = 11+      throughPrevious <- planThrough throughCount+      withKirokuPg $ \database -> do+        let settings = Pg.connectionSettings database+        _ <- runMigrationPlan defaultRunOptions settings throughPrevious >>= requireMigration+        withConnection settings $ \connection ->+          useSession connection (Session.script preCategoryFixtureSql)+        upgraded <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration+        reportOutcomes upgraded+          `shouldBe` replicate throughCount AlreadyApplied <> replicate (length nativeMigrationFiles - throughCount) AppliedNow+        withConnection settings $ \connection -> do+          facts <- useSession connection (Session.statement () categoryBackfillFactsStatement)+          facts+            `shouldBe` ( True,+                         0,+                         True,+                         Just "CREATE INDEX ix_stream_events_all_by_category ON kiroku.stream_events USING btree (category, stream_version) INCLUDE (original_stream_id) WHERE (stream_id = 0)",+                         "O"+                       )+          missingCategory <- Connection.use connection (Session.statement () insertAllRowWithoutCategoryStatement)+          missingCategory `shouldSatisfy` hasSqlState "23514"+          assertSchema connection++    -- kiroku.uuidv7() is the component's version-independent generator, but+    -- it arrives by a different route on each major: 0001's fallback on+    -- PostgreSQL 17, 0010's alias for the builtin on PostgreSQL 18. This+    -- case pins the route to the running server, so an acceptance run that+    -- believes it exercised one major and actually ran the other fails here+    -- instead of passing quietly. Both routes need a run to be covered --+    -- see `just test-matrix`.+    it "publishes the generator by the route its PostgreSQL major requires" $ do+      plan <- requirePlan+      result <- withMigratedDatabase plan $ \connection -> do+        (major, builtinExists, generatorExists, leaseDefault) <-+          useSession connection (Session.statement () leaseDefaultStatement)+        Text.IO.putStrLn+          ( "PostgreSQL "+              <> Text.pack (show major)+              <> " UUIDv7 generator: kiroku.uuidv7() via "+              <> (if builtinExists then "0010's alias for the builtin" else "0001's fallback")+          )+        generatorExists `shouldBe` True+        builtinExists `shouldBe` (major >= 18)+        leaseDefault `shouldBe` Just "kiroku.uuidv7()"+      either (expectationFailure . show) pure result++    it "acquires a lease from a session whose search_path excludes kiroku" $ do+      plan <- requirePlan+      result <- withMigratedDatabase plan $ \connection -> do+        useSession connection (Session.script "SET search_path TO pg_catalog")+        version <- useSession connection (Session.statement () insertLeaseStatement)+        version `shouldBe` 7+      either (expectationFailure . show) pure result++  describe "Codd history import" $ do+    it "imports a current codd V5 ledger, verifies, and never replays SQL" $+      importFixture "codd"++    it "imports the legacy codd_schema ledger shape" $+      importFixture "codd_schema"++    it "rejects a partial legacy row before creating the target ledger" $ do+      plan <- requirePlan+      withKirokuPg $ \database -> do+        let settings = Pg.connectionSettings database+            provider = connectionProviderFromSettings settings+        withConnection settings $ \connection -> installCoddLedger connection "codd" True+        config <-+          requireRight+            (kirokuCoddSourceConfig provider True "partial fixture must fail" Confirmed)+        imported <-+          importCoddHistory defaultImportOptions config provider plan kirokuCoddHistoryMappings+        imported `shouldSatisfy` \case+          Left CoddPartialMigration {} -> True+          _ -> False+        withConnection settings $ \connection -> do+          targetExists <- useSession connection (Session.statement "pgmigrate" schemaExistsStatement)+          targetExists `shouldBe` False++importFixture :: Text -> Expectation+importFixture sourceSchema = do+  plan <- requirePlan+  directory <- findMigrationsDirectory+  withKirokuPg $ \database -> do+    let settings = Pg.connectionSettings database+        provider = connectionProviderFromSettings settings+    withConnection settings $ \connection -> do+      applyNativeSqlFromDisk connection directory+      installCoddLedger connection sourceSchema False+    config <-+      requireRight+        (kirokuCoddSourceConfig provider True "verified Kiroku Codd cutover" Confirmed)+    first <-+      importCoddHistory defaultImportOptions config provider plan kirokuCoddHistoryMappings+        >>= requireRight+    importOutcomes first `shouldBe` replicate 7 Imported+    pendingIds <-+      traverse+        (requireRight . migrationId "kiroku")+        ["0008-schema-management-comment", "0009", "0010", "0011", "0012", "0013", "0014", "0015", "0016"]+    verifiedBeforeCanary <- verifyMigrationPlan defaultRunOptions settings plan >>= requireMigration+    case verifiedBeforeCanary of+      VerificationReport verificationIssues _ _ _ ->+        verificationIssues+          `shouldBe` (PendingMigration <$> pendingIds)+    up <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration+    reportOutcomes up `shouldBe` replicate 7 AlreadyApplied <> replicate (length nativeMigrationFiles - 7) AppliedNow+    verifiedAfterCanary <- verifyMigrationPlan defaultRunOptions settings plan >>= requireMigration+    case verifiedAfterCanary of+      VerificationReport verificationIssues _ _ _ ->+        verificationIssues `shouldBe` []+    rerun <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration+    reportOutcomes rerun `shouldBe` replicate (length nativeMigrationFiles) AlreadyApplied+    second <-+      importCoddHistory defaultImportOptions config provider plan kirokuCoddHistoryMappings+        >>= requireRight+    importOutcomes second `shouldBe` replicate 7 AlreadyImported+    withConnection settings $ \connection -> do+      assertSchema connection+      sourceRows <- useSession connection (Session.statement () (sourceRowCountStatement sourceSchema))+      sourceRows `shouldBe` 7+      facts <- useSession connection (Session.statement () importFactsStatement)+      facts `shouldBe` (fromIntegral (length nativeMigrationFiles), 7, True)++nativeMigrationFiles :: [FilePath]+nativeMigrationFiles =+  [ "0001-kiroku-bootstrap.sql",+    "0002-add-subscription-dead-letters.sql",+    "0003-notify-trigger-append-guard.sql",+    "0004-dead-letters-event-id-index.sql",+    "0005-index-hygiene-and-streams-fillfactor.sql",+    "0006-stream-name-length-check.sql",+    "0007-stream-truncate-before.sql",+    "0008-schema-management-comment.sql",+    "0009.sql",+    "0010.sql",+    "0011.sql",+    "0012.sql",+    "0013.sql",+    "0014.sql",+    "0015.sql",+    "0016.sql"+  ]++-- | The plan truncated to its first @count@ migrations, read from the checked-in+-- directory. It stands in for a database that a previous release bootstrapped: the+-- ledger it produces is byte-identical to the one the full plan would have written+-- for those entries, so the full plan later sees exactly the remainder as pending.+planThrough :: Int -> IO MigrationPlan+planThrough count = do+  directory <- findMigrationsDirectory+  entries <- traverse (readEntry directory) (take count nativeMigrationFiles)+  component <-+    requireRight+      (migrationComponentFromEmbeddedSql "kiroku" mempty (NonEmpty.fromList entries))+  requireRight (migrationPlan (component :| []))+  where+    readEntry directory file = do+      bytes <- ByteString.readFile (directory </> file)+      pure (file, bytes)++findMigrationsDirectory :: IO FilePath+findMigrationsDirectory =+  findDirectory ["kiroku-store-migrations/migrations", "migrations"]++findLockfile :: IO FilePath+findLockfile =+  findFile ["kiroku-store-migrations/migrations.lock", "migrations.lock"]++findDirectory :: [FilePath] -> IO FilePath+findDirectory candidates = do+  existing <- filterM doesDirectoryExist candidates+  case existing of+    directory : _ -> pure directory+    [] -> expectationFailure ("could not find directory: " <> show candidates) >> pure "."++findFile :: [FilePath] -> IO FilePath+findFile candidates = do+  existing <- filterM doesFileExist candidates+  case existing of+    path : _ -> pure path+    [] -> expectationFailure ("could not find file: " <> show candidates) >> pure "."++filterM :: (value -> IO Bool) -> [value] -> IO [value]+filterM predicate = foldr step (pure [])+  where+    step value remaining = do+      matches <- predicate value+      values <- remaining+      pure (if matches then value : values else values)++parseLockfile :: Text -> [(FilePath, Text)]+parseLockfile contents =+  [ (Text.unpack filename, checksum)+  | line <- Text.lines contents,+    [checksum, filename] <- [Text.words line]+  ]++checksumText :: ByteString -> Text+checksumText =+  Text.pack+    . concatMap renderByte+    . ByteString.unpack+    . migrationChecksumBytes+    . migrationFingerprint+  where+    renderByte byte =+      case Numeric.showHex byte "" of+        [digit] -> ['0', digit]+        digits -> digits++dropSqlSuffix :: FilePath -> String+dropSqlSuffix = reverse . drop 4 . reverse++requirePlan :: IO MigrationPlan+requirePlan = requireRight kirokuMigrationPlan++requireRight :: (Show error) => Either error value -> IO value+requireRight = either (failure . show) pure++requireMigration :: (Show error) => Either error value -> IO value+requireMigration = requireRight++failure :: String -> IO value+failure message = expectationFailure message >> fail message++providerFor :: Connection.Connection -> ConnectionProvider+providerFor connection = connectionProvider (\action -> Right <$> action connection)++reportOutcomes :: MigrationReport -> [MigrationOutcome]+reportOutcomes MigrationReport {results} = outcome <$> toList results++importOutcomes :: HistoryImportReport -> [HistoryImportOutcome]+importOutcomes HistoryImportReport {importResults} = importOutcome <$> toList importResults++-- | Built on pg-migrate's stable per-user temporary root, which 'withMigratedDatabase'+-- also uses, so ephemeral-pg's startup sweep reclaims clusters left by killed runs.+kirokuPgConfig :: IO Pg.Config+kirokuPgConfig = do+  base <- defaultEphemeralConfig+  pure base {PgConfig.user = "kiroku"}++withKirokuPg :: (Pg.Database -> IO ()) -> IO ()+withKirokuPg action = do+  config <- kirokuPgConfig+  started <- Pg.startCached config Pg.defaultCacheConfig+  case started of+    Left startError -> expectationFailure (show startError)+    Right database -> action database `finally` Pg.stop database++withConnection :: Settings.Settings -> (Connection.Connection -> IO value) -> IO value+withConnection settings action = do+  acquired <- Connection.acquire settings+  connection <- requireRight acquired+  action connection `finally` Connection.release connection++useSession :: Connection.Connection -> Session.Session value -> IO value+useSession connection session =+  Connection.use connection session >>= requireRight++assertSchema :: Connection.Connection -> Expectation+assertSchema connection = do+  healthy <- useSession connection (Session.statement () schemaFactsStatement)+  healthy `shouldBe` True+  oversized <- Connection.use connection (Session.statement (Text.replicate 513 "x") oversizedStreamStatement)+  oversized `shouldSatisfy` isLeft++schemaFactsStatement :: Statement () Bool+schemaFactsStatement =+  Statement.preparable+    """+    SELECT bool_and(ok)+    FROM (VALUES+      (to_regnamespace('kiroku') IS NOT NULL),+      (to_regclass('kiroku.events') IS NOT NULL),+      (to_regclass('kiroku.streams') IS NOT NULL),+      (to_regclass('kiroku.dead_letters') IS NOT NULL),+      (EXISTS (SELECT 1 FROM pg_catalog.pg_trigger WHERE tgname = 'stream_events_notify_insert' AND NOT tgisinternal)),+      (EXISTS (SELECT 1 FROM pg_catalog.pg_indexes WHERE schemaname = 'kiroku' AND indexname = 'ix_dead_letters_event_id')),+      (EXISTS (SELECT 1 FROM pg_catalog.pg_indexes WHERE schemaname = 'kiroku' AND indexname = 'ix_stream_events_all_by_category')),+      (EXISTS (SELECT 1 FROM pg_catalog.pg_constraint WHERE conname = 'ck_stream_events_all_category')),+      (EXISTS (SELECT 1 FROM pg_catalog.pg_constraint WHERE conname = 'chk_streams_stream_name_length')),+      (EXISTS (SELECT 1 FROM pg_catalog.pg_attribute a JOIN pg_catalog.pg_class c ON c.oid = a.attrelid JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace WHERE n.nspname = 'kiroku' AND c.relname = 'streams' AND a.attname = 'truncate_before' AND NOT a.attisdropped)),+      (EXISTS (SELECT 1 FROM pg_catalog.pg_indexes WHERE schemaname = 'kiroku' AND indexname = 'ix_streams_browse_name' AND indexdef LIKE '%COLLATE "C"%' AND indexdef LIKE '%WHERE (stream_id <> 0)%')),+      (obj_description(to_regnamespace('kiroku'), 'pg_namespace') = 'Managed by pg-migrate component kiroku through 0016')+    ) AS checks(ok)+    """+    Encoders.noParams+    (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.bool)))++-- | One stream with one event, written with the five-column junction shape+-- every release before 0012 used: a home row and an @$all@ row, no category.+preCategoryFixtureSql :: Text+preCategoryFixtureSql =+  """+  INSERT INTO kiroku.streams (stream_name, stream_version) VALUES ('upgrade-1', 1);+  INSERT INTO kiroku.events (event_id, event_type, data)+  VALUES ('00000000-0000-7000-8000-000000000001', 'UpgradeFixture', '{}'::jsonb);+  INSERT INTO kiroku.stream_events+    (event_id, stream_id, stream_version, original_stream_id, original_stream_version)+  SELECT '00000000-0000-7000-8000-000000000001'::uuid, s.stream_id, 1, s.stream_id, 1+  FROM kiroku.streams AS s WHERE s.stream_name = 'upgrade-1'+  UNION ALL+  SELECT '00000000-0000-7000-8000-000000000001'::uuid, 0, 1, s.stream_id, 1+  FROM kiroku.streams AS s WHERE s.stream_name = 'upgrade-1';+  UPDATE kiroku.streams SET stream_version = 1 WHERE stream_id = 0;+  """++-- | After 0012: whether the fixture's @$all@ row carries its stream's category,+-- how many @$all@ rows lack one, whether the home row stayed NULL, the category+-- index definition, and the immutability trigger's enabled state.+categoryBackfillFactsStatement :: Statement () (Bool, Int64, Bool, Maybe Text, Text)+categoryBackfillFactsStatement =+  Statement.preparable+    """+    SELECT (SELECT se.category = s.category+              FROM kiroku.stream_events AS se+              JOIN kiroku.streams AS s ON s.stream_id = se.original_stream_id+             WHERE se.stream_id = 0 AND s.stream_name = 'upgrade-1'),+           (SELECT count(*) FROM kiroku.stream_events+             WHERE stream_id = 0 AND category IS NULL),+           (SELECT se.category IS NULL+              FROM kiroku.stream_events AS se+              JOIN kiroku.streams AS s ON s.stream_id = se.stream_id+             WHERE s.stream_name = 'upgrade-1'),+           (SELECT indexdef FROM pg_catalog.pg_indexes+             WHERE schemaname = 'kiroku' AND indexname = 'ix_stream_events_all_by_category'),+           (SELECT tgenabled::text FROM pg_catalog.pg_trigger+             WHERE tgrelid = 'kiroku.stream_events'::regclass+               AND tgname = 'no_update_stream_events')+    """+    Encoders.noParams+    ( Decoders.singleRow+        ( (,,,,)+            <$> Decoders.column (Decoders.nonNullable Decoders.bool)+            <*> Decoders.column (Decoders.nonNullable Decoders.int8)+            <*> Decoders.column (Decoders.nonNullable Decoders.bool)+            <*> Decoders.column (Decoders.nullable Decoders.text)+            <*> Decoders.column (Decoders.nonNullable Decoders.text)+        )+    )++-- | An @$all@ row written the pre-0012 way, which the CHECK must now refuse.+insertAllRowWithoutCategoryStatement :: Statement () ()+insertAllRowWithoutCategoryStatement =+  Statement.preparable+    """+    WITH new_event AS (+      INSERT INTO kiroku.events (event_id, event_type, data)+      VALUES ('00000000-0000-7000-8000-000000000002', 'UpgradeFixture', '{}'::jsonb)+      RETURNING event_id+    )+    INSERT INTO kiroku.stream_events+      (event_id, stream_id, stream_version, original_stream_id, original_stream_version)+    SELECT new_event.event_id, 0, 2, s.stream_id, 2+    FROM new_event, kiroku.streams AS s+    WHERE s.stream_name = 'upgrade-1'+    """+    Encoders.noParams+    Decoders.noResult++-- | The server major, whether each UUIDv7 generator exists, and the stored+-- @lease_id@ default. Together these pin which route published+-- @kiroku.uuidv7()@ on the running server.+leaseDefaultStatement :: Statement () (Int32, Bool, Bool, Maybe Text)+leaseDefaultStatement =+  Statement.preparable+    """+    SELECT pg_catalog.current_setting('server_version_num')::int / 10000,+           to_regprocedure('pg_catalog.uuidv7()') IS NOT NULL,+           to_regprocedure('kiroku.uuidv7()') IS NOT NULL,+           (SELECT pg_catalog.pg_get_expr(stored.adbin, stored.adrelid)+              FROM pg_catalog.pg_attrdef AS stored+              JOIN pg_catalog.pg_attribute AS attribute+                ON attribute.attrelid = stored.adrelid+               AND attribute.attnum = stored.adnum+             WHERE stored.adrelid = 'kiroku.history_retention_leases'::regclass+               AND attribute.attname = 'lease_id')+    """+    Encoders.noParams+    ( Decoders.singleRow+        ( (,,,)+            <$> Decoders.column (Decoders.nonNullable Decoders.int4)+            <*> Decoders.column (Decoders.nonNullable Decoders.bool)+            <*> Decoders.column (Decoders.nonNullable Decoders.bool)+            <*> Decoders.column (Decoders.nullable Decoders.text)+        )+    )++-- | Insert one lease and report the UUID version its default generated.+insertLeaseStatement :: Statement () Int32+insertLeaseStatement =+  Statement.preparable+    """+    INSERT INTO kiroku.history_retention_leases+        (owner, reason, protected_through, created_at, renewed_at, expires_at)+    VALUES+        ('kiroku-store-migrations-test', 'search_path independence', 0,+         pg_catalog.now(), pg_catalog.now(), pg_catalog.now() + interval '1 hour')+    RETURNING pg_catalog.get_byte(pg_catalog.uuid_send(lease_id), 6) >> 4+    """+    Encoders.noParams+    (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int4)))++historyRetentionColumnsStatement :: Statement () [(Text, Text, Text, Bool)]+historyRetentionColumnsStatement =+  Statement.preparable+    """+    SELECT relation.relname::text,+           attribute.attname::text,+           pg_catalog.format_type(attribute.atttypid, attribute.atttypmod),+           attribute.attnotnull+    FROM pg_catalog.pg_class AS relation+    JOIN pg_catalog.pg_namespace AS namespace+      ON namespace.oid = relation.relnamespace+    JOIN pg_catalog.pg_attribute AS attribute+      ON attribute.attrelid = relation.oid+    WHERE namespace.nspname = 'kiroku'+      AND relation.relname IN ('history_retention_coordinator', 'history_retention_leases')+      AND attribute.attnum > 0+      AND NOT attribute.attisdropped+    ORDER BY relation.relname, attribute.attnum+    """+    Encoders.noParams+    ( Decoders.rowList+        ( (,,,)+            <$> column Decoders.text+            <*> column Decoders.text+            <*> column Decoders.text+            <*> column Decoders.bool+        )+    )+  where+    column = Decoders.column . Decoders.nonNullable++historyRetentionFactsStatement :: Statement () (Int64, Int64, Bool, Bool, Int64)+historyRetentionFactsStatement =+  Statement.preparable+    """+    SELECT+      (SELECT count(*) FROM kiroku.history_retention_coordinator WHERE singleton),+      (SELECT count(*)+         FROM pg_catalog.pg_constraint AS con_record+         JOIN pg_catalog.pg_class AS relation ON relation.oid = con_record.conrelid+         JOIN pg_catalog.pg_namespace AS namespace ON namespace.oid = relation.relnamespace+        WHERE namespace.nspname = 'kiroku'+          AND relation.relname IN ('history_retention_coordinator', 'history_retention_leases')+          AND con_record.conname IN (+            'history_retention_coordinator_pkey',+            'chk_history_retention_coordinator_singleton',+            'history_retention_leases_pkey',+            'chk_history_retention_lease_owner_bytes',+            'chk_history_retention_lease_reason_bytes',+            'chk_history_retention_lease_frontier',+            'chk_history_retention_lease_renewal_time',+            'chk_history_retention_lease_expiry',+            'chk_history_retention_lease_release_time'+          )),+      EXISTS (+        SELECT 1+        FROM pg_catalog.pg_indexes+        WHERE schemaname = 'kiroku'+          AND indexname = 'ix_history_retention_leases_unreleased_expiry'+          AND indexdef LIKE '%WHERE (released_at IS NULL)'+      ),+      to_regprocedure('kiroku.protect_replay_history_from_destruction()') IS NOT NULL,+      (SELECT count(*)+         FROM pg_catalog.pg_trigger AS trigger_record+         JOIN pg_catalog.pg_class AS relation ON relation.oid = trigger_record.tgrelid+         JOIN pg_catalog.pg_namespace AS namespace ON namespace.oid = relation.relnamespace+        WHERE namespace.nspname = 'kiroku'+          AND relation.relname IN ('events', 'stream_events', 'streams')+          AND trigger_record.tgname IN ('protect_replay_history_delete', 'protect_replay_history_truncate')+          AND NOT trigger_record.tgisinternal)+    """+    Encoders.noParams+    ( Decoders.singleRow+        ( (,,,,)+            <$> column Decoders.int8+            <*> column Decoders.int8+            <*> column Decoders.bool+            <*> column Decoders.bool+            <*> column Decoders.int8+        )+    )+  where+    column = Decoders.column . Decoders.nonNullable++oversizedStreamStatement :: Statement Text ()+oversizedStreamStatement =+  Statement.preparable+    "INSERT INTO kiroku.streams (stream_name, stream_version) VALUES ($1, 0)"+    (Encoders.param (Encoders.nonNullable Encoders.text))+    Decoders.noResult++relationCatalogStatement :: Statement () (Text, Text, Bool, Text, Text, Text)+relationCatalogStatement =+  Statement.preparable+    """+    SELECT c.relkind::text,+           COALESCE(array_to_string(c.reloptions, ','), ''),+           c.relowner = source.relowner,+           view_contract.is_updatable::text,+           view_contract.is_insertable_into::text,+           pg_catalog.obj_description(c.oid, 'pg_class')+    FROM pg_catalog.pg_class AS c+    JOIN pg_catalog.pg_namespace AS n ON n.oid = c.relnamespace+    JOIN pg_catalog.pg_class AS source+      ON source.oid = to_regclass('kiroku.subscriptions')+    JOIN information_schema.views AS view_contract+      ON view_contract.table_schema = n.nspname+     AND view_contract.table_name = c.relname+    WHERE n.nspname = 'kiroku'+      AND c.relname = 'subscription_checkpoints_v1'+    """+    Encoders.noParams+    ( Decoders.singleRow+        ( (,,,,,)+            <$> column Decoders.text+            <*> column Decoders.text+            <*> column Decoders.bool+            <*> column Decoders.text+            <*> column Decoders.text+            <*> column Decoders.text+        )+    )+  where+    column = Decoders.column . Decoders.nonNullable++relationColumnsStatement :: Statement () [(Text, Text, Bool, Text)]+relationColumnsStatement =+  Statement.preparable+    """+    SELECT attribute.attname::text,+           pg_catalog.format_type(attribute.atttypid, attribute.atttypmod),+           attribute.attnotnull,+           pg_catalog.col_description(relation.oid, attribute.attnum)+    FROM pg_catalog.pg_class AS relation+    JOIN pg_catalog.pg_namespace AS namespace+      ON namespace.oid = relation.relnamespace+    JOIN pg_catalog.pg_attribute AS attribute+      ON attribute.attrelid = relation.oid+    WHERE namespace.nspname = 'kiroku'+      AND relation.relname = 'subscription_checkpoints_v1'+      AND attribute.attnum > 0+      AND NOT attribute.attisdropped+    ORDER BY attribute.attnum+    """+    Encoders.noParams+    ( Decoders.rowList+        ( (,,,)+            <$> column Decoders.text+            <*> column Decoders.text+            <*> column Decoders.bool+            <*> column Decoders.text+        )+    )+  where+    column = Decoders.column . Decoders.nonNullable++checkpointRowsStatement :: Statement () [(Text, Int32, Int64, UTCTime)]+checkpointRowsStatement =+  Statement.preparable+    """+    SELECT subscription_name,+           consumer_group_member,+           checkpoint_position,+           checkpoint_updated_at+    FROM kiroku.subscription_checkpoints_v1+    ORDER BY subscription_name, consumer_group_member+    """+    Encoders.noParams+    ( Decoders.rowList+        ( (,,,)+            <$> column Decoders.text+            <*> column Decoders.int4+            <*> column Decoders.int8+            <*> column Decoders.timestamptz+        )+    )+  where+    column = Decoders.column . Decoders.nonNullable++checkpointFixtureTimestamp :: UTCTime+checkpointFixtureTimestamp =+  UTCTime+    (fromGregorian 2026 8 13)+    (secondsToDiffTime (12 * 60 * 60 + 34 * 60 + 56))++checkpointFixtureSql :: Text+checkpointFixtureSql =+  """+  INSERT INTO kiroku.subscriptions+    (subscription_name, consumer_group_member, last_seen, updated_at)+  VALUES+    ('ordinary', 0, 11, '2026-08-13 12:34:56+00'),+    ('grouped', 0, 20, '2026-08-13 12:34:56+00'),+    ('grouped', 1, 21, '2026-08-13 12:34:56+00'),+    ('grouped', 2, 22, '2026-08-13 12:34:56+00')+  """++checkpointAdvanceSql :: Text+checkpointAdvanceSql =+  """+  UPDATE kiroku.subscriptions+  SET last_seen = 5+  WHERE subscription_name = 'ordinary'+    AND consumer_group_member = 0+  """++checkpointRegressionSql :: Text+checkpointRegressionSql =+  """+  UPDATE kiroku.subscriptions+  SET last_seen = 2+  WHERE subscription_name = 'ordinary'+    AND consumer_group_member = 0+  """++checkpointPosition :: Connection.Connection -> IO Int64+checkpointPosition connection =+  useSession connection (Session.statement "ordinary" checkpointPositionStatement)++checkpointPositionStatement :: Statement Text Int64+checkpointPositionStatement =+  Statement.preparable+    """+    SELECT checkpoint_position+    FROM kiroku.subscription_checkpoints_v1+    WHERE subscription_name = $1+      AND consumer_group_member = 0+    """+    (Encoders.param (Encoders.nonNullable Encoders.text))+    (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8)))++withTemporaryReaderRole :: Connection.Connection -> (Text -> IO value) -> IO value+withTemporaryReaderRole connection action = do+  unique <- newUnique+  let role =+        "kiroku_checkpoint_reader_"+          <> Text.replace "-" "n" (Text.pack (show (hashUnique unique)))+      identifier = quoteTestIdentifier role+  useSession connection (Session.script ("CREATE ROLE " <> identifier))+  action role+    `finally` useSession+      connection+      ( Session.script+          ( "RESET ROLE; DROP OWNED BY "+              <> identifier+              <> "; DROP ROLE "+              <> identifier+          )+      )++quoteTestIdentifier :: Text -> Text+quoteTestIdentifier value =+  "\"" <> Text.replace "\"" "\"\"" value <> "\""++checkpointRowCountStatement :: Statement () Int64+checkpointRowCountStatement =+  Statement.preparable+    "SELECT count(*) FROM kiroku.subscription_checkpoints_v1"+    Encoders.noParams+    (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8)))++privateTablePrivilegeStatement :: Statement () Bool+privateTablePrivilegeStatement =+  Statement.preparable+    "SELECT has_table_privilege(current_user, 'kiroku.subscriptions', 'SELECT')"+    Encoders.noParams+    (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.bool)))++privateTableCountStatement :: Statement () Int64+privateTableCountStatement =+  Statement.preparable+    "SELECT count(*) FROM kiroku.subscriptions"+    Encoders.noParams+    (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8)))++updatePublicViewStatement :: Statement () ()+updatePublicViewStatement =+  Statement.unpreparable+    """+    UPDATE kiroku.subscription_checkpoints_v1+    SET checkpoint_position = checkpoint_position+    """+    Encoders.noParams+    Decoders.noResult++hasSqlState :: Text -> Either Errors.SessionError value -> Bool+hasSqlState expected = \case+  Left+    ( Errors.StatementSessionError+        _+        _+        _+        _+        _+        (Errors.ServerStatementError (Errors.ServerError actual _ _ _ _))+      ) -> actual == expected+  _ -> False++downstreamReplacementFixtureSql :: Text+downstreamReplacementFixtureSql =+  """+  INSERT INTO kiroku.subscriptions+    (subscription_name, consumer_group_member, last_seen,+     updated_at)+  VALUES ('dependency', 0, 42, '2026-08-13 12:34:56+00');++  CREATE VIEW public.subscription_checkpoint_floors AS+  SELECT subscription_name,+         min(checkpoint_position) AS checkpoint_floor+  FROM kiroku.subscription_checkpoints_v1+  GROUP BY subscription_name;++  CREATE TABLE kiroku.subscription_checkpoint_storage_v2 (+    subscription_name text NOT NULL,+    consumer_group_member integer NOT NULL,+    checkpoint_position bigint NOT NULL,+    checkpoint_updated_at timestamptz NOT NULL,+    PRIMARY KEY (subscription_name, consumer_group_member)+  );++  INSERT INTO kiroku.subscription_checkpoint_storage_v2+  SELECT subscription_name,+         consumer_group_member,+         last_seen,+         updated_at+  FROM kiroku.subscriptions;++  CREATE OR REPLACE VIEW kiroku.subscription_checkpoints_v1+      (subscription_name,+       consumer_group_member,+       checkpoint_position,+       checkpoint_updated_at)+  WITH (security_invoker = false)+  AS+  WITH checkpoint_rows AS NOT MATERIALIZED (+      SELECT subscription_name,+             consumer_group_member,+             checkpoint_position,+             checkpoint_updated_at+      FROM kiroku.subscription_checkpoint_storage_v2+  )+  SELECT subscription_name,+         consumer_group_member,+         checkpoint_position,+         checkpoint_updated_at+  FROM checkpoint_rows;++  DROP TABLE kiroku.subscriptions;+  """++downstreamFloorsStatement :: Statement () [(Text, Int64)]+downstreamFloorsStatement =+  Statement.preparable+    """+    SELECT subscription_name, checkpoint_floor+    FROM public.subscription_checkpoint_floors+    ORDER BY subscription_name+    """+    Encoders.noParams+    ( Decoders.rowList+        ( (,)+            <$> Decoders.column (Decoders.nonNullable Decoders.text)+            <*> Decoders.column (Decoders.nonNullable Decoders.int8)+        )+    )++checkpointPlanFixtureSql :: Text+checkpointPlanFixtureSql =+  """+  INSERT INTO kiroku.subscriptions+    (subscription_name, consumer_group_member, last_seen)+  SELECT 'subscription-' || lpad((value % 100)::text, 3, '0'),+         value::integer,+         value::bigint+  FROM generate_series(1, 10000) AS generated(value);++  ANALYZE kiroku.subscriptions;+  """++serverVersionStatement :: Statement () Text+serverVersionStatement =+  Statement.preparable+    "SELECT current_setting('server_version')"+    Encoders.noParams+    (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.text)))++checkpointExplainStatement :: Statement () [Text]+checkpointExplainStatement =+  Statement.preparable+    """+    EXPLAIN (COSTS OFF)+    SELECT min(checkpoint_position)+    FROM kiroku.subscription_checkpoints_v1+    WHERE subscription_name = 'subscription-042'+    """+    Encoders.noParams+    (Decoders.rowList (Decoders.column (Decoders.nonNullable Decoders.text)))++applyNativeSqlFromDisk :: Connection.Connection -> FilePath -> IO ()+applyNativeSqlFromDisk connection directory =+  forM_ (take 7 nativeMigrationFiles) $ \file -> do+    sql <- Text.IO.readFile (directory </> file)+    useSession connection (Session.script sql)++installCoddLedger :: Connection.Connection -> Text -> Bool -> IO ()+installCoddLedger connection sourceSchema partial =+  useSession connection (Session.script (coddFixtureSql sourceSchema partial))++coddFixtureSql :: Text -> Bool -> Text+coddFixtureSql sourceSchema partial =+  Text.unlines+    [ "CREATE SCHEMA " <> sourceSchema <> ";",+      "CREATE TABLE " <> sourceSchema <> ".sql_migrations (",+      "  id serial NOT NULL, migration_timestamp timestamptz NOT NULL,",+      "  applied_at timestamptz, name text NOT NULL, application_duration interval,",+      "  num_applied_statements int, no_txn_failed_at timestamptz, txnid bigint, connid int",+      ");",+      "INSERT INTO " <> sourceSchema <> ".sql_migrations",+      "  (migration_timestamp, applied_at, name, application_duration, num_applied_statements, no_txn_failed_at, txnid, connid) VALUES",+      Text.intercalate ",\n" (zipWith renderRow [1 :: Int ..] (toList kirokuLegacyMigrationNames)) <> ";"+    ]+  where+    renderRow index filename =+      "('2026-01-01 00:00:00+00'::timestamptz + interval '"+        <> Text.pack (show index)+        <> " seconds', "+        <> appliedAt index+        <> ", '"+        <> Text.pack filename+        <> "', interval '1 second', 1, "+        <> failureAt index+        <> ", 1, 1)"+    appliedAt index+      | partial && index == 4 = "NULL"+      | otherwise = "'2026-01-01 00:01:00+00'::timestamptz + interval '" <> Text.pack (show index) <> " seconds'"+    failureAt index+      | partial && index == 4 = "'2026-01-01 00:02:00+00'::timestamptz"+      | otherwise = "NULL"++schemaExistsStatement :: Statement Text Bool+schemaExistsStatement =+  Statement.preparable+    "SELECT to_regnamespace($1) IS NOT NULL"+    (Encoders.param (Encoders.nonNullable Encoders.text))+    (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.bool)))++sourceRowCountStatement :: Text -> Statement () Int64+sourceRowCountStatement sourceSchema =+  Statement.unpreparable+    ("SELECT count(*) FROM " <> sourceSchema <> ".sql_migrations")+    Encoders.noParams+    (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8)))++importFactsStatement :: Statement () (Int64, Int64, Bool)+importFactsStatement =+  Statement.preparable+    """+    SELECT+      (SELECT count(*) FROM pgmigrate.migrations),+      (SELECT count(*) FROM pgmigrate.history_imports),+      (SELECT bool_and(source_evidence #>> '{satisfying_evidence,0,details,adapter}' = 'codd') FROM pgmigrate.history_imports)+    """+    Encoders.noParams+    ( Decoders.singleRow+        ( (,,)+            <$> column Decoders.int8+            <*> column Decoders.int8+            <*> column Decoders.bool+        )+    )+  where+    column = Decoders.column . Decoders.nonNullable++targetUpgradeFactsStatement :: Statement () (Int64, Text)+targetUpgradeFactsStatement =+  Statement.preparable+    "SELECT pg_relation_filenode('kiroku.subscriptions')::bigint, ctid::text FROM kiroku.subscriptions WHERE subscription_name = 'legacy-target'"+    Encoders.noParams+    (Decoders.singleRow ((,) <$> Decoders.column (Decoders.nonNullable Decoders.int8) <*> Decoders.column (Decoders.nonNullable Decoders.text)))++targetBindingFactsStatement :: Statement () (Text, Maybe Text, Int64, Bool)+targetBindingFactsStatement =+  Statement.preparable+    """+    SELECT target_kind, target_category, last_seen,+      EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema = 'kiroku' AND table_name = 'subscriptions' AND column_name = 'stream_name')+    FROM kiroku.subscriptions WHERE subscription_name = 'legacy-target'+    """+    Encoders.noParams+    (Decoders.singleRow ((,,,) <$> Decoders.column (Decoders.nonNullable Decoders.text) <*> Decoders.column (Decoders.nullable Decoders.text) <*> Decoders.column (Decoders.nonNullable Decoders.int8) <*> Decoders.column (Decoders.nonNullable Decoders.bool)))