diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -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 diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -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 diff --git a/kiroku-store-migrations.cabal b/kiroku-store-migrations.cabal --- a/kiroku-store-migrations.cabal +++ b/kiroku-store-migrations.cabal @@ -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 diff --git a/migrations/0016.sql b/migrations/0016.sql new file mode 100644 --- /dev/null +++ b/migrations/0016.sql @@ -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'; diff --git a/migrations/manifest b/migrations/manifest --- a/migrations/manifest +++ b/migrations/manifest @@ -13,3 +13,4 @@ 0013.sql 0014.sql 0015.sql +0016.sql diff --git a/src/Kiroku/Store/Migrations.hs b/src/Kiroku/Store/Migrations.hs --- a/src/Kiroku/Store/Migrations.hs +++ b/src/Kiroku/Store/Migrations.hs @@ -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 :| []) diff --git a/src/Kiroku/Store/Migrations/History/Codd.hs b/src/Kiroku/Store/Migrations/History/Codd.hs --- a/src/Kiroku/Store/Migrations/History/Codd.hs +++ b/src/Kiroku/Store/Migrations/History/Codd.hs @@ -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 diff --git a/src/Kiroku/Store/Migrations/Internal/Definition.hs b/src/Kiroku/Store/Migrations/Internal/Definition.hs --- a/src/Kiroku/Store/Migrations/Internal/Definition.hs +++ b/src/Kiroku/Store/Migrations/Internal/Definition.hs @@ -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 diff --git a/src/Kiroku/Store/Migrations/Internal/EmbedFile.hs b/src/Kiroku/Store/Migrations/Internal/EmbedFile.hs --- a/src/Kiroku/Store/Migrations/Internal/EmbedFile.hs +++ b/src/Kiroku/Store/Migrations/Internal/EmbedFile.hs @@ -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)))) diff --git a/src/Kiroku/Store/Migrations/New.hs b/src/Kiroku/Store/Migrations/New.hs --- a/src/Kiroku/Store/Migrations/New.hs +++ b/src/Kiroku/Store/Migrations/New.hs @@ -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.", + "" + ] + ) diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -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)))