diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,13 @@ # Changelog -## Unreleased +## 0.3.1.0 — 2026-08-13 + +### New Features + +* Added forward migration `0009`, which publishes the frozen, structurally read-only + `kiroku.subscription_checkpoints_v1` relation for exact durable subscription-member + checkpoints. Database readers can receive schema usage and view selection without access to + Kiroku's private checkpoint table; no role or grant is created automatically. ## 0.3.0.0 — 2026-07-14 diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -2,8 +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 seven immutable SQL payloads, so applications can compose it with other -libraries without copying Kiroku SQL. +and nine SQL payloads, so applications can compose it with other libraries +without copying Kiroku SQL. The first seven payloads are immutable historical +Codd bytes; `0008` and `0009` are native-only forward migrations. ## Public API @@ -101,9 +102,12 @@ It proves manifest order, legacy SHA-256 parity, fresh apply, strict verify, idempotent rerun, concurrent locking, current Codd V5 import, legacy `codd_schema` import, partial-row rejection, import audit records, and -source-ledger preservation. `cabal test kiroku-store:kiroku-store-test` consumes -the same native plan through `kiroku-test-support` and proves the complete store -behavior, including append and read scenarios. +source-ledger preservation. The relation contract suite also proves the frozen +`kiroku.subscription_checkpoints_v1` catalog, non-null value semantics, +owner-rights privilege isolation, structural read-only behavior, downstream +view survival, and indexed query plan. `cabal test kiroku-store:kiroku-store-test` +consumes the same native plan through `kiroku-test-support` and proves the +complete store behavior, including append and read scenarios. ## Recovery 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.3.0.0 +version: 0.3.1.0 synopsis: Schema migrations for kiroku-store description: Native pg-migrate component, Codd history mapping, and migration executable @@ -92,3 +92,4 @@ , pg-migrate-test-support ^>=1.1.0.0 , temporary >=1.3 && <1.4 , text >=2.0 && <2.2 + , time >=1.12 && <1.15 diff --git a/migrations/0009.sql b/migrations/0009.sql new file mode 100644 --- /dev/null +++ b/migrations/0009.sql @@ -0,0 +1,35 @@ +-- Publish the frozen, owner-rights, structurally read-only v1 checkpoint relation. +CREATE 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, + last_seen AS checkpoint_position, + updated_at AS checkpoint_updated_at + FROM kiroku.subscriptions +) +SELECT subscription_name, + consumer_group_member, + checkpoint_position, + checkpoint_updated_at +FROM checkpoint_rows; + +COMMENT ON VIEW kiroku.subscription_checkpoints_v1 IS + '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.'; + +COMMENT ON COLUMN kiroku.subscription_checkpoints_v1.subscription_name IS + 'Persisted subscription name; member zero does not distinguish a non-group subscription from member zero of a consumer group.'; + +COMMENT ON COLUMN kiroku.subscription_checkpoints_v1.consumer_group_member IS + 'Persisted consumer-group member key; member zero carries no topology classification.'; + +COMMENT ON COLUMN kiroku.subscription_checkpoints_v1.checkpoint_position IS + 'Exact persisted global position for this subscription member; an explicit reset may move it backward or forward.'; + +COMMENT ON COLUMN kiroku.subscription_checkpoints_v1.checkpoint_updated_at IS + 'Time of the latest checkpoint-row upsert; it does not imply position advancement or worker liveness.'; diff --git a/migrations/manifest b/migrations/manifest --- a/migrations/manifest +++ b/migrations/manifest @@ -6,3 +6,4 @@ 0006-stream-name-length-check.sql 0007-stream-truncate-before.sql 0008-schema-management-comment.sql +0009.sql diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -9,12 +9,14 @@ import Data.ByteString qualified as ByteString import Data.Either (isLeft) import Data.Foldable (toList) -import Data.Int (Int64) +import Data.Int (Int32, Int64) import Data.List (sort) import Data.List.NonEmpty (NonEmpty (..)) import Data.Text (Text) import Data.Text qualified as Text import Data.Text.IO qualified as Text.IO +import Data.Time (UTCTime (..), fromGregorian, secondsToDiffTime) +import Data.Unique (hashUnique, newUnique) import Database.PostgreSQL.Migrate import Database.PostgreSQL.Migrate.History.Codd import Database.PostgreSQL.Migrate.Internal (migrationChecksumBytes) @@ -24,6 +26,7 @@ import Hasql.Connection.Settings qualified as Settings import Hasql.Decoders qualified as Decoders import Hasql.Encoders qualified as Encoders +import Hasql.Errors qualified as Errors import Hasql.Session qualified as Session import Hasql.Statement (Statement) import Hasql.Statement qualified as Statement @@ -39,7 +42,7 @@ main :: IO () main = hspec $ do describe "native Kiroku migration definition" $ do - it "tracks the eight native files in manifest order" $ do + it "tracks the nine native files in manifest order" $ do directory <- findMigrationsDirectory manifest <- Text.lines <$> Text.IO.readFile (directory "manifest") manifest `shouldBe` Text.pack <$> nativeMigrationFiles @@ -52,7 +55,7 @@ bytes <- ByteString.readFile (directory nativeName) lookup legacyName lockEntries `shouldBe` Just (checksumText bytes) - it "builds component kiroku and an eight-migration plan" $ do + it "builds component kiroku and a nine-migration plan" $ do component <- requireRight kirokuMigrations component `seq` pure () plan <- requirePlan @@ -62,7 +65,7 @@ ] validateHistoryMappingTargets plan kirokuCoddHistoryMappings `shouldBe` Right () - length targetIds `shouldBe` 8 + length targetIds `shouldBe` length nativeMigrationFiles describe "native migration authoring" $ do it "creates the next numeric file and atomically appends the manifest" $ @@ -88,18 +91,18 @@ `shouldReturn` "0007-existing.sql\n" describe "fresh native databases" $ do - it "applies all eight, verifies strictly, and reports AlreadyApplied on rerun" $ do + it "applies all nine, 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 8 AlreadyApplied + 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` 8 + length applied `shouldBe` length nativeMigrationFiles either (expectationFailure . show) pure result it "serializes concurrent applies through the pg-migrate advisory lock" $ do @@ -111,8 +114,140 @@ (runMigrationPlan defaultRunOptions settings plan >>= requireMigration) (runMigrationPlan defaultRunOptions settings plan >>= requireMigration) sort [reportOutcomes first, reportOutcomes second] - `shouldBe` sort [replicate 8 AppliedNow, replicate 8 AlreadyApplied] + `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 "Codd history import" $ do it "imports a current codd V5 ledger, verifies, and never replays SQL" $ importFixture "codd" @@ -155,20 +290,23 @@ importCoddHistory defaultImportOptions config provider plan kirokuCoddHistoryMappings >>= requireRight importOutcomes first `shouldBe` replicate 7 Imported - canaryId <- requireRight (migrationId "kiroku" "0008-schema-management-comment") + pendingIds <- + traverse + (requireRight . migrationId "kiroku") + ["0008-schema-management-comment", "0009"] verifiedBeforeCanary <- verifyMigrationPlan defaultRunOptions settings plan >>= requireMigration case verifiedBeforeCanary of VerificationReport verificationIssues _ _ _ -> verificationIssues - `shouldBe` [PendingMigration canaryId] + `shouldBe` (PendingMigration <$> pendingIds) up <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration - reportOutcomes up `shouldBe` replicate 7 AlreadyApplied <> [AppliedNow] + reportOutcomes up `shouldBe` replicate 7 AlreadyApplied <> replicate 2 AppliedNow verifiedAfterCanary <- verifyMigrationPlan defaultRunOptions settings plan >>= requireMigration case verifiedAfterCanary of VerificationReport verificationIssues _ _ _ -> verificationIssues `shouldBe` [] rerun <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration - reportOutcomes rerun `shouldBe` replicate 8 AlreadyApplied + reportOutcomes rerun `shouldBe` replicate (length nativeMigrationFiles) AlreadyApplied second <- importCoddHistory defaultImportOptions config provider plan kirokuCoddHistoryMappings >>= requireRight @@ -178,7 +316,7 @@ sourceRows <- useSession connection (Session.statement () (sourceRowCountStatement sourceSchema)) sourceRows `shouldBe` 7 facts <- useSession connection (Session.statement () importFactsStatement) - facts `shouldBe` (8, 7, True) + facts `shouldBe` (fromIntegral (length nativeMigrationFiles), 7, True) nativeMigrationFiles :: [FilePath] nativeMigrationFiles = @@ -190,6 +328,7 @@ , "0006-stream-name-length-check.sql" , "0007-stream-truncate-before.sql" , "0008-schema-management-comment.sql" + , "0009.sql" ] findMigrationsDirectory :: IO FilePath @@ -319,6 +458,313 @@ "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 =