diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,11 +2,56 @@ ## Unreleased +## 0.2.0.0 — 2026-07-11 + +### Breaking Changes + +* Migrated the package's runtime from Codd to `pg-migrate`. The public API now + exports the native `kirokuMigrations` component and `kirokuMigrationPlan` + instead of Codd settings, runner, ledger-status, and schema-check wrappers. +* Replaced timestamped runtime identities with manifest-ordered `0001` through + `0007` identities while preserving every SQL payload byte. +* Replaced the Codd CLI and `CODD_*` configuration surface with the standard + `pg-migrate-cli` command tree and `DATABASE_URL`. `verify` now compares the + declared plan with the `pgmigrate` ledger; it does not compare live schema + objects with an expected-schema snapshot. + +### New Features + +* Added a manifest-backed, compile-time-embedded migration component that + applications can compose with other libraries in explicit dependency order. +* Added checked-in Codd history mappings and `SamePayload` evidence for safe, + non-replaying import from current `codd` and legacy `codd_schema` ledgers. + Shared-ledger consumers can combine Kiroku's exported payloads and mappings + with their own components before importing. +* Added the standard `pg-migrate-cli` planning, inspection, execution, + verification, status, and numeric migration-authoring commands. +* Added fresh-apply, rerun, concurrent-apply, strict ledger verification, Codd + import, partial-row rejection, audit, and source-preservation coverage. The + full Kiroku store suite now consumes the same native plan through + `kiroku-test-support`. +* Appended `0008-schema-management-comment`, a non-destructive observable + native-runner canary. Fresh and imported-prefix tests prove it applies once, + verifies strictly, and reruns as `AlreadyApplied` without changing historical + payloads or Codd mappings. + +### Changed + +* Preserved the seven historical SQL payloads byte-for-byte while moving their + authoritative ordering to `migrations/manifest`; `migrations.lock` remains + the source evidence used during Codd history import. +* Removed Codd, `codd-extras`, `file-embed`, and `postgresql-simple` from the + normal library and executable dependency closure. +* Removed the orphaned Codd expected-schema snapshot, its writer executable, the + Cabal flag that gated it, and the accompanying Nix closure workaround. Codd + ledger history import remains supported independently through + `pg-migrate-import-codd`. + ## 0.1.1.0 — 2026-05-31 ### New Features -* Forward migration `2026-05-26-00-00-00-add-subscription-dead-letters.sql`: +* Forward migration `2026-05-29-15-26-04-add-subscription-dead-letters.sql`: creates the `kiroku.dead_letters` table (per consumer-group member, with a foreign key to `kiroku.events`) and its recency index, supporting per-event dead-letter recording for subscriptions (MasterPlan 6 / plan 40). diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -1,40 +1,118 @@ # kiroku-store-migrations -`kiroku-store-migrations` owns schema evolution for `kiroku-store`. -It embeds Kiroku's timestamped SQL migrations and runs them through -`codd`, so a service can migrate its database before opening -`Kiroku.Store.withStore`. +`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. -Run the executable with codd's standard environment variables: +## Public API +```haskell +import Kiroku.Store.Migrations + +kirokuMigrations :: Either DefinitionError MigrationComponent +kirokuMigrationPlan :: Either PlanError MigrationPlan +``` + +Applications with more than one component should consume `kirokuMigrations` +and build their own explicit dependency-ordered plan. The single-component +`kirokuMigrationPlan` is convenient for Kiroku-only deployments. + +Existing databases can import their Codd ledger through +`Kiroku.Store.Migrations.History.Codd`: + +```haskell +kirokuCoddHistoryMappings :: NonEmpty HistoryMapping +kirokuCoddSourcePayloads :: Map FilePath ByteString +kirokuCoddManifestText :: Text + +kirokuCoddSourceConfig + :: ConnectionProvider + -> Bool + -> Text + -> Confirmation + -> Either CoddDefinitionError CoddSourceConfig +``` + +The mapping selects the seven historical timestamped Codd names, verifies the +checked-in `migrations.lock` SHA-256 evidence against the exact embedded native +bytes, and maps them to `kiroku/0001-kiroku-bootstrap` through +`kiroku/0007-stream-truncate-before`. Import writes only the `pgmigrate` ledger; +it never executes already-applied SQL. Consumers with a shared Codd ledger can +combine the exported names, payload map, manifest text, and history mappings +with their own component evidence before constructing one atomic import. + +## CLI + +`kiroku-store-migrate` mounts the standard `pg-migrate-cli` command groups: + ```bash -CODD_CONNECTION='host=/tmp port=5432 dbname=kiroku user=kiroku_admin' \ -CODD_MIGRATION_DIRS=unused-for-embedded-migrations \ -CODD_EXPECTED_SCHEMA_DIR=unused-for-unverified-embedded-migrations \ -CODD_SCHEMAS=kiroku \ -kiroku-store-migrate +kiroku-store-migrate --help +kiroku-store-migrate plan +kiroku-store-migrate list +kiroku-store-migrate check kiroku-store-migrations/migrations/manifest +kiroku-store-migrate up --database-url "$DATABASE_URL" +kiroku-store-migrate verify --database-url "$DATABASE_URL" +kiroku-store-migrate status --database-url "$DATABASE_URL" ``` -The bootstrap migration creates a dedicated `kiroku` schema and installs -every Kiroku table, index, function, and trigger inside it, leaving -`public` free for application objects. `CODD_SCHEMAS=kiroku` tells codd to -track that schema. The runtime role therefore needs privileges on the -`kiroku` schema (for example `USAGE` plus the table privileges it uses), -not on `public`. +Database commands accept `--database-url`. When it is omitted the executable +uses `DATABASE_URL`. Local `plan`, `list`, `check`, and `new` commands need no +database environment variable. `verify` compares the declared plan strictly +with the `pgmigrate` ledger; it is not a live schema snapshot comparison. -After migrations run, start the application normally: +For Haskell callers: ```haskell -withStore (defaultConnectionSettings connString) app +import Database.PostgreSQL.Migrate +import Hasql.Connection.Settings qualified as Settings +import Kiroku.Store.Migrations + +main :: IO () +main = do + plan <- either (fail . show) pure kirokuMigrationPlan + result <- runMigrationPlan defaultRunOptions (Settings.connectionString databaseUrl) plan + either (fail . show) (const (pure ())) result ``` -`codd` is forward-only. Once a migration has run in production, -reverting the Haskell package does not undo the database change. Repair -state by restoring from backup or by shipping another forward migration. +## Authoring -This first implementation runs without codd expected-schema verification -because Kiroku does not yet ship a checked-in codd expected-schema -snapshot. `CODD_EXPECTED_SCHEMA_DIR` is still required by codd's settings -parser, but this executable does not read from it. Operators should treat -the migration table as the source of applied-version truth until strict -snapshots are added. +The authoritative source is `migrations/manifest`; each line names one SQL file +in execution order. Create the next numeric file with the standard CLI: + +```bash +kiroku-store-migrate new \ + --manifest kiroku-store-migrations/migrations/manifest \ + --description "add widget index" +``` + +The helper exclusively creates the inferred file and atomically appends its +name to the manifest. Never edit a released payload. Correct mistakes with a +new forward migration. The seven initial payloads intentionally retain their +exact historical bytes, including old comments, because Codd import uses +`SamePayload` evidence. + +Run the package suite after every migration change: + +```bash +cabal test kiroku-store-migrations:kiroku-store-migrations-test +``` + +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. + +## Recovery + +Migrations are forward-only. Before a persistent upgrade, take a backup. If an +applied migration is bad, either restore that backup or append a corrective +migration. Do not delete or rewrite an applied `pgmigrate.migrations` row except +through the reviewed `pg-migrate` repair workflow. + +The historical script under `ledger-fixups/` remains checked in only as source +evidence for databases that previously needed Codd timestamp repair. New native +migrations use component-local numeric identities and do not use timestamped +filenames. diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -1,11 +1,45 @@ -module Main where +module Main (main) where -import Codd.Environment (getCoddSettings) -import Data.Time (secondsToDiffTime) -import Kiroku.Store.Migrations (runKirokuMigrationsNoCheck) +import Data.Aeson qualified as Aeson +import Data.ByteString.Lazy.Char8 qualified as LazyByteString +import Data.Text qualified as Text +import Data.Text.IO qualified as Text.IO +import Database.PostgreSQL.Migrate (defaultRunOptions) +import Database.PostgreSQL.Migrate.CLI +import Hasql.Connection.Settings qualified as Settings +import Kiroku.Store.Migrations (kirokuMigrationPlan) +import Options.Applicative +import System.Environment (lookupEnv) +import System.Exit qualified as Exit main :: IO () main = do - settings <- getCoddSettings - _ <- runKirokuMigrationsNoCheck settings (secondsToDiffTime 5) - pure () + plan <- either (fail . show) pure kirokuMigrationPlan + command <- + 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 command + case commandOutputFormat command of + TextOutput -> Text.IO.putStrLn (renderMigrationCommandText outcome) + JsonOutput -> LazyByteString.putStrLn (Aeson.encode (renderMigrationCommandJson outcome)) + Exit.exitWith + (case exitClass outcome of ExitSuccess -> Exit.ExitSuccess; _ -> Exit.ExitFailure 1) + +commandOutputFormat :: MigrationCommand -> OutputFormat +commandOutputFormat command = + case command 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,10 +1,10 @@ cabal-version: 3.0 name: kiroku-store-migrations -version: 0.1.1.0 +version: 0.2.0.0 synopsis: Schema migrations for kiroku-store description: - Embedded codd migrations and a migration executable for installing and - upgrading the PostgreSQL schema used by @kiroku-store@. + Native pg-migrate component, Codd history mapping, and migration executable + for installing and upgrading the PostgreSQL schema used by @kiroku-store@. homepage: https://github.com/shinzui/kiroku bug-reports: https://github.com/shinzui/kiroku/issues @@ -15,8 +15,10 @@ category: Database, Eventing extra-doc-files: CHANGELOG.md extra-source-files: + migrations/*.sql + migrations/manifest + migrations.lock README.md - sql-migrations/*.sql source-repository head type: git @@ -32,16 +34,26 @@ library import: common - exposed-modules: Kiroku.Store.Migrations + exposed-modules: + Kiroku.Store.Migrations + Kiroku.Store.Migrations.History.Codd + Kiroku.Store.Migrations.New + + other-modules: + Kiroku.Store.Migrations.Internal.Definition + Kiroku.Store.Migrations.Internal.EmbedFile + hs-source-dirs: src build-depends: - , base >=4.18 && <5 - , bytestring >=0.11 && <0.13 - , codd >=0.1.8 && <0.2 - , file-embed >=0.0.15 && <0.0.17 - , streaming >=0.2 && <0.3 - , text >=2.0 && <2.2 - , time >=1.12 && <1.15 + , base >=4.18 && <5 + , bytestring >=0.11 && <0.13 + , containers >=0.6 && <0.8 + , filepath >=1.4 && <1.6 + , pg-migrate ^>=1.0.0.0 + , pg-migrate-embed ^>=1.0.0.0 + , pg-migrate-import-codd ^>=1.0.0.0 + , template-haskell >=2.20 && <2.24 + , text >=2.0 && <2.2 executable kiroku-store-migrate import: common @@ -49,10 +61,15 @@ hs-source-dirs: app ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends: - , base - , codd + , aeson >=2.1 && <2.3 + , base >=4.18 && <5 + , bytestring >=0.11 && <0.13 + , hasql >=1.10 && <1.11 , kiroku-store-migrations - , time >=1.12 && <1.15 + , optparse-applicative >=0.17 && <0.20 + , pg-migrate ^>=1.0.0.0 + , pg-migrate-cli ^>=1.0.0.0 + , text >=2.0 && <2.2 test-suite kiroku-store-migrations-test import: common @@ -61,17 +78,17 @@ hs-source-dirs: test ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends: - , aeson >=2.1 && <2.3 - , attoparsec - , base >=4.18 && <5 - , codd - , containers >=0.6 && <0.8 - , ephemeral-pg >=0.2 && <0.3 - , hasql >=1.10 && <1.11 - , hasql-pool >=1.2 && <1.5 - , hspec >=2.10 && <2.12 - , kiroku-store ^>=0.2 + , async + , base >=4.18 && <5 + , bytestring >=0.11 && <0.13 + , directory >=1.3 + , ephemeral-pg >=0.2 && <0.3 + , filepath >=1.4 && <1.6 + , hasql >=1.10 && <1.11 + , hspec >=2.10 && <2.12 , kiroku-store-migrations - , text >=2.0 && <2.2 - , time >=1.12 && <1.15 - , vector >=0.13 && <0.14 + , pg-migrate ^>=1.0.0.0 + , pg-migrate-import-codd ^>=1.0.0.0 + , pg-migrate-test-support ^>=1.0.0.0 + , temporary >=1.3 && <1.4 + , text >=2.0 && <2.2 diff --git a/migrations.lock b/migrations.lock new file mode 100644 --- /dev/null +++ b/migrations.lock @@ -0,0 +1,7 @@ +af7c777f8702d9b5bce6b3ea16540ab7542bc78725cb45e945c6d5b6025a4d66 2026-05-16-12-17-14-kiroku-bootstrap.sql +2771cdd2605a8808ea469c0bcba175c8ba4c5d2465f43eb05c56edb188fa8141 2026-05-29-15-26-04-add-subscription-dead-letters.sql +432796c59389d45478ba773b62e9f38d0a562807061eb49280d295ec83248351 2026-06-14-13-17-09-notify-trigger-append-guard.sql +f698db092ebf34beeddecc949bbd506afc0b9c63d3e9983ad5d14f203f2ee1cd 2026-06-14-13-25-40-dead-letters-event-id-index.sql +80c28c7743f1d9967eadf921c0509fc8477db16500466bf7a49c5f59aadb0fa2 2026-06-14-13-54-48-index-hygiene-and-streams-fillfactor.sql +a0e4338a2feadf8f01499a3809d3de78966966689b095ea46195c4d07743e4aa 2026-06-14-14-01-17-stream-name-length-check.sql +a4fe42ad8702b43c3d6f72edf263d6b301e783493d49f40a35dd14d2a7928fa7 2026-06-24-09-42-22-stream-truncate-before.sql diff --git a/migrations/0001-kiroku-bootstrap.sql b/migrations/0001-kiroku-bootstrap.sql new file mode 100644 --- /dev/null +++ b/migrations/0001-kiroku-bootstrap.sql @@ -0,0 +1,229 @@ +-- Kiroku Store bootstrap migration (codd) +-- Supports PostgreSQL 17+. +-- +-- kiroku-store-migrations is the schema owner. codd applies this file +-- verbatim, records that the timestamped migration ran, and skips it on later +-- runs. Future schema changes add new timestamped SQL files in this directory +-- instead of changing kiroku-store. +-- +-- All Kiroku-owned objects live in the dedicated `kiroku` schema, leaving +-- `public` free for application objects. Creating the schema and setting +-- search_path first means every unqualified object name in the rest of this +-- file resolves into the Kiroku schema. +CREATE SCHEMA IF NOT EXISTS kiroku; +SET search_path TO kiroku, pg_catalog; + +-- PostgreSQL 18 provides pg_catalog.uuidv7(); PostgreSQL 17 needs this +-- Kiroku-schema fallback before events.event_id DEFAULT uuidv7() is parsed. +-- With search_path set above, the unqualified CREATE FUNCTION lands in the +-- Kiroku schema, and to_regprocedure('uuidv7()') resolves through search_path +-- (pg_catalog first for the built-in, then the Kiroku schema for the fallback). +DO $$ +BEGIN + IF to_regprocedure('pg_catalog.uuidv7()') IS NULL + AND to_regprocedure('uuidv7()') IS NULL THEN + EXECUTE $fn$ + CREATE FUNCTION uuidv7() + RETURNS uuid + AS $body$ + DECLARE + unix_ts_ms bytea; + uuid_bytes bytea; + BEGIN + unix_ts_ms = substring(int8send(floor(extract(epoch from clock_timestamp()) * 1000)::bigint) from 3); + uuid_bytes = uuid_send(gen_random_uuid()); + uuid_bytes = overlay(uuid_bytes placing unix_ts_ms from 1 for 6); + uuid_bytes = set_byte(uuid_bytes, 6, (b'0111' || get_byte(uuid_bytes, 6)::bit(4))::bit(8)::int); + RETURN encode(uuid_bytes, 'hex')::uuid; + END + $body$ + LANGUAGE plpgsql + VOLATILE + $fn$; + END IF; +END +$$; + +-- Streams (including $all as stream_id = 0) +CREATE TABLE IF NOT EXISTS streams ( + stream_id BIGSERIAL PRIMARY KEY, + stream_name TEXT NOT NULL, + category TEXT GENERATED ALWAYS AS (split_part(stream_name, '-', 1)) STORED, + stream_version BIGINT NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + deleted_at TIMESTAMPTZ, + CONSTRAINT ix_streams_stream_name UNIQUE (stream_name) +); + +-- Seed the $all stream +INSERT INTO streams (stream_id, stream_name, stream_version) +VALUES (0, '$all', 0) +ON CONFLICT DO NOTHING; + +-- Reset sequence past the reserved stream_id=0 +SELECT setval('streams_stream_id_seq', GREATEST((SELECT MAX(stream_id) FROM streams), 1)); + +-- Events (flat table — stream membership tracked in stream_events) +CREATE TABLE IF NOT EXISTS events ( + event_id UUID PRIMARY KEY DEFAULT uuidv7(), + event_type TEXT NOT NULL, + causation_id UUID, + correlation_id UUID, + data JSONB NOT NULL, + metadata JSONB, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- Stream-event junction (each event gets 2+ rows: source stream + $all + any links) +CREATE TABLE IF NOT EXISTS stream_events ( + event_id UUID NOT NULL REFERENCES events(event_id), + stream_id BIGINT NOT NULL REFERENCES streams(stream_id), + stream_version BIGINT NOT NULL, + original_stream_id BIGINT NOT NULL, + original_stream_version BIGINT NOT NULL, + PRIMARY KEY (event_id, stream_id) +); + +-- Indexes + +-- Primary read path: fetch events from a stream in order +CREATE INDEX IF NOT EXISTS ix_stream_events_stream_version + ON stream_events (stream_id, stream_version); + +-- Event type filtering (for server-side subscription filtering) +CREATE INDEX IF NOT EXISTS ix_events_event_type + ON events (event_type); + +-- Correlation tracing +CREATE INDEX IF NOT EXISTS ix_events_correlation_id + ON events (correlation_id) WHERE correlation_id IS NOT NULL; + +-- Causation tracing +CREATE INDEX IF NOT EXISTS ix_events_causation_id + ON events (causation_id) WHERE causation_id IS NOT NULL; + +-- Category filtering (for readCategory — uses generated column, not LIKE) +CREATE INDEX IF NOT EXISTS ix_streams_category + ON streams (category); + +-- Category read path: find $all entries by originating stream, ordered by global position +-- Enables efficient category reads by allowing the planner to: look up category stream_ids → +-- index scan $all for each → merge ordered by stream_version +CREATE INDEX IF NOT EXISTS ix_stream_events_all_by_origin + ON stream_events (original_stream_id, stream_version) + WHERE stream_id = 0; + +-- Subscriptions (checkpoint persistence for subscription positions). +-- consumer_group_member / consumer_group_size carry static consumer-group +-- topology (ExecPlan 28 / EP-1). Non-group subscriptions are member 0, size 1. +-- The unique key is composite (subscription_name, consumer_group_member) so each +-- group member persists its own checkpoint under one shared subscription name. +CREATE TABLE IF NOT EXISTS subscriptions ( + subscription_id BIGSERIAL PRIMARY KEY, + subscription_name TEXT NOT NULL, + stream_name TEXT NOT NULL DEFAULT '$all', + last_seen BIGINT NOT NULL DEFAULT 0, + consumer_group_member INT NOT NULL DEFAULT 0, + consumer_group_size INT NOT NULL DEFAULT 1, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- Idempotent convergence for databases created before EP-1: add the columns if +-- missing, drop the old auto-named single-column unique constraint if present, +-- and install the composite unique index. All guarded so the bootstrap body is +-- still safe in disposable local databases; codd records the timestamped +-- migration and does not reapply it after a successful run. +ALTER TABLE subscriptions ADD COLUMN IF NOT EXISTS consumer_group_member INT NOT NULL DEFAULT 0; +ALTER TABLE subscriptions ADD COLUMN IF NOT EXISTS consumer_group_size INT NOT NULL DEFAULT 1; +ALTER TABLE subscriptions DROP CONSTRAINT IF EXISTS subscriptions_subscription_name_key; +CREATE UNIQUE INDEX IF NOT EXISTS ix_subscriptions_name_member + ON subscriptions (subscription_name, consumer_group_member); + +-- Triggers + +-- NOTIFY on stream changes (fires once per append, not per event) +CREATE OR REPLACE FUNCTION notify_events() RETURNS TRIGGER AS $$ +BEGIN + PERFORM pg_notify( + TG_TABLE_SCHEMA || '.events', + NEW.stream_name || ',' || NEW.stream_id || ',' || NEW.stream_version + ); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS stream_events_notify ON streams; +CREATE TRIGGER stream_events_notify + AFTER INSERT OR UPDATE ON streams + FOR EACH ROW EXECUTE FUNCTION notify_events(); + +-- Immutability: prevent event mutation +CREATE OR REPLACE FUNCTION prevent_mutation() RETURNS TRIGGER AS $$ +BEGIN + RAISE EXCEPTION 'Immutable table: % cannot be updated', TG_TABLE_NAME; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS no_update_events ON events; +CREATE TRIGGER no_update_events + BEFORE UPDATE ON events + FOR EACH ROW EXECUTE FUNCTION prevent_mutation(); + +DROP TRIGGER IF EXISTS no_update_stream_events ON stream_events; +CREATE TRIGGER no_update_stream_events + BEFORE UPDATE ON stream_events + FOR EACH ROW EXECUTE FUNCTION prevent_mutation(); + +-- Gated hard deletes (for maintenance/GDPR only) +CREATE OR REPLACE FUNCTION protect_deletion() RETURNS TRIGGER AS $$ +BEGIN + IF current_setting('kiroku.enable_hard_deletes', true) = 'on' THEN + RETURN OLD; + END IF; + RAISE EXCEPTION 'Hard deletes require: SET LOCAL kiroku.enable_hard_deletes = ''on'''; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS no_delete_events ON events; +CREATE TRIGGER no_delete_events + BEFORE DELETE ON events + FOR EACH ROW EXECUTE FUNCTION protect_deletion(); + +DROP TRIGGER IF EXISTS no_delete_stream_events ON stream_events; +CREATE TRIGGER no_delete_stream_events + BEFORE DELETE ON stream_events + FOR EACH ROW EXECUTE FUNCTION protect_deletion(); + +DROP TRIGGER IF EXISTS no_delete_streams ON streams; +CREATE TRIGGER no_delete_streams + BEFORE DELETE ON streams + FOR EACH ROW EXECUTE FUNCTION protect_deletion(); + +-- TRUNCATE bypasses row-level triggers, so the BEFORE DELETE triggers above +-- do not protect against an operator running TRUNCATE on these tables. Add +-- statement-level BEFORE TRUNCATE triggers gated by the same GUC so the +-- protection is symmetric. See EP-1 F6. +CREATE OR REPLACE FUNCTION protect_truncation() RETURNS TRIGGER AS $$ +BEGIN + IF current_setting('kiroku.enable_hard_deletes', true) = 'on' THEN + RETURN NULL; + END IF; + RAISE EXCEPTION 'TRUNCATE requires: SET LOCAL kiroku.enable_hard_deletes = ''on'''; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS no_truncate_events ON events; +CREATE TRIGGER no_truncate_events + BEFORE TRUNCATE ON events + FOR EACH STATEMENT EXECUTE FUNCTION protect_truncation(); + +DROP TRIGGER IF EXISTS no_truncate_stream_events ON stream_events; +CREATE TRIGGER no_truncate_stream_events + BEFORE TRUNCATE ON stream_events + FOR EACH STATEMENT EXECUTE FUNCTION protect_truncation(); + +DROP TRIGGER IF EXISTS no_truncate_streams ON streams; +CREATE TRIGGER no_truncate_streams + BEFORE TRUNCATE ON streams + FOR EACH STATEMENT EXECUTE FUNCTION protect_truncation(); diff --git a/migrations/0002-add-subscription-dead-letters.sql b/migrations/0002-add-subscription-dead-letters.sql new file mode 100644 --- /dev/null +++ b/migrations/0002-add-subscription-dead-letters.sql @@ -0,0 +1,31 @@ +-- Add the kiroku.dead_letters table (MasterPlan 6 / EP-2, docs/plans/40-...). +-- +-- Forward, additive migration: codd applies this file once, records it, and +-- skips it on later runs. It does not edit the bootstrap migration and does not +-- mutate existing event data. +-- +-- A dead-letter row records one event that a subscription handler asked to +-- "dead-letter" (return DeadLetter, or exhaust its bounded retry budget). The +-- event itself stays immutable in kiroku.events; this table references it by +-- event_id and global_position rather than copying the payload. The +-- consumer_group_member column (default 0, matching kiroku.subscriptions) +-- attributes the row to the member that produced it, so a consumer group's dead +-- letters are per-member. The worker writes a row and advances the member's +-- checkpoint in one atomic statement (see SQL.insertDeadLetterAndCheckpointStmt). + +CREATE TABLE IF NOT EXISTS kiroku.dead_letters ( + dead_letter_id BIGSERIAL PRIMARY KEY, + subscription_name TEXT NOT NULL, + consumer_group_member INT NOT NULL DEFAULT 0, + global_position BIGINT NOT NULL, + event_id UUID NOT NULL REFERENCES kiroku.events(event_id), + reason JSONB NOT NULL, + reason_summary TEXT NOT NULL, + attempt_count INT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (subscription_name, consumer_group_member, global_position, event_id) +); + +-- Operator read path: list a subscription member's dead letters by recency. +CREATE INDEX IF NOT EXISTS ix_dead_letters_subscription_created_at + ON kiroku.dead_letters (subscription_name, consumer_group_member, created_at); diff --git a/migrations/0003-notify-trigger-append-guard.sql b/migrations/0003-notify-trigger-append-guard.sql new file mode 100644 --- /dev/null +++ b/migrations/0003-notify-trigger-append-guard.sql @@ -0,0 +1,26 @@ +-- Guard append notifications at the trigger level. +-- +-- The trigger function and payload format stay unchanged: +-- stream_name,stream_id,stream_version +-- +-- INSERT covers newly-created streams. UPDATE covers later appends to an +-- existing stream. Both exclude the internal $all row (stream_id = 0), and the +-- UPDATE trigger only fires when an append advances stream_version, not for +-- lifecycle updates such as soft-delete or undelete. + +DROP TRIGGER IF EXISTS stream_events_notify ON kiroku.streams; + +DROP TRIGGER IF EXISTS stream_events_notify_insert ON kiroku.streams; +CREATE TRIGGER stream_events_notify_insert + AFTER INSERT ON kiroku.streams + FOR EACH ROW + WHEN (NEW.stream_id <> 0) + EXECUTE FUNCTION kiroku.notify_events(); + +DROP TRIGGER IF EXISTS stream_events_notify_update ON kiroku.streams; +CREATE TRIGGER stream_events_notify_update + AFTER UPDATE ON kiroku.streams + FOR EACH ROW + WHEN (NEW.stream_id <> 0 + AND NEW.stream_version IS DISTINCT FROM OLD.stream_version) + EXECUTE FUNCTION kiroku.notify_events(); diff --git a/migrations/0004-dead-letters-event-id-index.sql b/migrations/0004-dead-letters-event-id-index.sql new file mode 100644 --- /dev/null +++ b/migrations/0004-dead-letters-event-id-index.sql @@ -0,0 +1,10 @@ +-- Index dead_letters by event_id (MasterPlan 9 / EP-5, docs/plans/60-...). +-- +-- dead_letters.event_id has a FK to kiroku.events. The UNIQUE key leads with +-- subscription_name, so every referential-integrity check triggered by a +-- DELETE on kiroku.events (the hard-delete path) was a sequential scan of +-- dead_letters, and the hard-delete transaction's own dead-letter pre-delete +-- (Kiroku.Store.SQL.deleteDeadLettersForOrphanedEventsStmt) needs the same +-- access path. +CREATE INDEX IF NOT EXISTS ix_dead_letters_event_id + ON kiroku.dead_letters (event_id); diff --git a/migrations/0005-index-hygiene-and-streams-fillfactor.sql b/migrations/0005-index-hygiene-and-streams-fillfactor.sql new file mode 100644 --- /dev/null +++ b/migrations/0005-index-hygiene-and-streams-fillfactor.sql @@ -0,0 +1,33 @@ +-- Index hygiene and $all hot-row tuning (MasterPlan 9 / EP-5, docs/plans/60-...). + +-- 1. ix_events_event_type is referenced by no statement in kiroku-store; it is +-- pure write amplification on every append. Server-side event-type pushdown +-- (see the EventTypeFilter haddock in kiroku-store) should re-add a +-- fit-for-purpose index when it ships. +DROP INDEX IF EXISTS kiroku.ix_events_event_type; + +-- 2. Stream versions are unique per stream by construction (assigned under the +-- stream row lock; $all versions are the global position). Enforce it so a +-- version-assignment bug surfaces as a loud 23505 instead of silent +-- duplicates. Built as a new unique index, then the old non-unique index is +-- dropped (an index cannot be altered to unique in place). +CREATE UNIQUE INDEX IF NOT EXISTS ux_stream_events_stream_version + ON kiroku.stream_events (stream_id, stream_version); +DROP INDEX IF EXISTS kiroku.ix_stream_events_stream_version; + +-- 3. readDeadLetters orders by (global_position DESC, dead_letter_id DESC) -- +-- the store's canonical, deterministic "newest first". Re-key the read +-- index to match so the read is index-ordered instead of sorting each time. +CREATE INDEX IF NOT EXISTS ix_dead_letters_subscription_position + ON kiroku.dead_letters + (subscription_name, consumer_group_member, + global_position DESC, dead_letter_id DESC); +DROP INDEX IF EXISTS kiroku.ix_dead_letters_subscription_created_at; + +-- 4. The $all row (stream_id 0) is updated by every append in the database. +-- Its updated column (stream_version) is not indexed, so updates are +-- HOT-eligible when the page has free space; fillfactor 50 reserves that +-- space on newly written pages. Existing pages converge through normal +-- update/prune activity (VACUUM cannot run inside this migration's +-- transaction). Autovacuum tuning for this table is left to operators. +ALTER TABLE kiroku.streams SET (fillfactor = 50); diff --git a/migrations/0006-stream-name-length-check.sql b/migrations/0006-stream-name-length-check.sql new file mode 100644 --- /dev/null +++ b/migrations/0006-stream-name-length-check.sql @@ -0,0 +1,9 @@ +-- Defense-in-depth bound on stream-name length (MasterPlan 9 / EP-5, +-- docs/plans/60-...). The Haskell store validates this before any SQL +-- (StoreError StreamNameTooLong, maxStreamNameBytes = 512); the constraint +-- catches writers that bypass the library (raw SQL via runTransaction, +-- psql sessions). 512 bytes is far below pg_notify's 8,000-byte payload +-- limit, so the append-notification trigger can never abort on payload size. +ALTER TABLE kiroku.streams + ADD CONSTRAINT chk_streams_stream_name_length + CHECK (octet_length(stream_name) <= 512); diff --git a/migrations/0007-stream-truncate-before.sql b/migrations/0007-stream-truncate-before.sql new file mode 100644 --- /dev/null +++ b/migrations/0007-stream-truncate-before.sql @@ -0,0 +1,8 @@ +-- Logical truncate-before marker for close-the-book compaction +-- (ExecPlan docs/plans/65). Per-stream cursor: ordered stream reads return +-- only events whose stream_version >= truncate_before. Default 0 keeps all +-- events (per-stream versions are 1-based). Reversible; the global $all log +-- is never affected. UPDATE on streams is already permitted (soft-delete +-- uses it), so no trigger/GUC changes are needed. +ALTER TABLE kiroku.streams + ADD COLUMN IF NOT EXISTS truncate_before BIGINT NOT NULL DEFAULT 0; diff --git a/migrations/0008-schema-management-comment.sql b/migrations/0008-schema-management-comment.sql new file mode 100644 --- /dev/null +++ b/migrations/0008-schema-management-comment.sql @@ -0,0 +1,2 @@ +COMMENT ON SCHEMA kiroku IS + 'Managed by pg-migrate component kiroku through 0008-schema-management-comment'; diff --git a/migrations/manifest b/migrations/manifest new file mode 100644 --- /dev/null +++ b/migrations/manifest @@ -0,0 +1,8 @@ +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 diff --git a/sql-migrations/2026-05-16-00-00-00-kiroku-bootstrap.sql b/sql-migrations/2026-05-16-00-00-00-kiroku-bootstrap.sql deleted file mode 100644 --- a/sql-migrations/2026-05-16-00-00-00-kiroku-bootstrap.sql +++ /dev/null @@ -1,229 +0,0 @@ --- Kiroku Store bootstrap migration (codd) --- Supports PostgreSQL 17+. --- --- kiroku-store-migrations is the schema owner. codd applies this file --- verbatim, records that the timestamped migration ran, and skips it on later --- runs. Future schema changes add new timestamped SQL files in this directory --- instead of changing kiroku-store. --- --- All Kiroku-owned objects live in the dedicated `kiroku` schema, leaving --- `public` free for application objects. Creating the schema and setting --- search_path first means every unqualified object name in the rest of this --- file resolves into the Kiroku schema. -CREATE SCHEMA IF NOT EXISTS kiroku; -SET search_path TO kiroku, pg_catalog; - --- PostgreSQL 18 provides pg_catalog.uuidv7(); PostgreSQL 17 needs this --- Kiroku-schema fallback before events.event_id DEFAULT uuidv7() is parsed. --- With search_path set above, the unqualified CREATE FUNCTION lands in the --- Kiroku schema, and to_regprocedure('uuidv7()') resolves through search_path --- (pg_catalog first for the built-in, then the Kiroku schema for the fallback). -DO $$ -BEGIN - IF to_regprocedure('pg_catalog.uuidv7()') IS NULL - AND to_regprocedure('uuidv7()') IS NULL THEN - EXECUTE $fn$ - CREATE FUNCTION uuidv7() - RETURNS uuid - AS $body$ - DECLARE - unix_ts_ms bytea; - uuid_bytes bytea; - BEGIN - unix_ts_ms = substring(int8send(floor(extract(epoch from clock_timestamp()) * 1000)::bigint) from 3); - uuid_bytes = uuid_send(gen_random_uuid()); - uuid_bytes = overlay(uuid_bytes placing unix_ts_ms from 1 for 6); - uuid_bytes = set_byte(uuid_bytes, 6, (b'0111' || get_byte(uuid_bytes, 6)::bit(4))::bit(8)::int); - RETURN encode(uuid_bytes, 'hex')::uuid; - END - $body$ - LANGUAGE plpgsql - VOLATILE - $fn$; - END IF; -END -$$; - --- Streams (including $all as stream_id = 0) -CREATE TABLE IF NOT EXISTS streams ( - stream_id BIGSERIAL PRIMARY KEY, - stream_name TEXT NOT NULL, - category TEXT GENERATED ALWAYS AS (split_part(stream_name, '-', 1)) STORED, - stream_version BIGINT NOT NULL DEFAULT 0, - created_at TIMESTAMPTZ NOT NULL DEFAULT now(), - deleted_at TIMESTAMPTZ, - CONSTRAINT ix_streams_stream_name UNIQUE (stream_name) -); - --- Seed the $all stream -INSERT INTO streams (stream_id, stream_name, stream_version) -VALUES (0, '$all', 0) -ON CONFLICT DO NOTHING; - --- Reset sequence past the reserved stream_id=0 -SELECT setval('streams_stream_id_seq', GREATEST((SELECT MAX(stream_id) FROM streams), 1)); - --- Events (flat table — stream membership tracked in stream_events) -CREATE TABLE IF NOT EXISTS events ( - event_id UUID PRIMARY KEY DEFAULT uuidv7(), - event_type TEXT NOT NULL, - causation_id UUID, - correlation_id UUID, - data JSONB NOT NULL, - metadata JSONB, - created_at TIMESTAMPTZ NOT NULL DEFAULT now() -); - --- Stream-event junction (each event gets 2+ rows: source stream + $all + any links) -CREATE TABLE IF NOT EXISTS stream_events ( - event_id UUID NOT NULL REFERENCES events(event_id), - stream_id BIGINT NOT NULL REFERENCES streams(stream_id), - stream_version BIGINT NOT NULL, - original_stream_id BIGINT NOT NULL, - original_stream_version BIGINT NOT NULL, - PRIMARY KEY (event_id, stream_id) -); - --- Indexes - --- Primary read path: fetch events from a stream in order -CREATE INDEX IF NOT EXISTS ix_stream_events_stream_version - ON stream_events (stream_id, stream_version); - --- Event type filtering (for server-side subscription filtering) -CREATE INDEX IF NOT EXISTS ix_events_event_type - ON events (event_type); - --- Correlation tracing -CREATE INDEX IF NOT EXISTS ix_events_correlation_id - ON events (correlation_id) WHERE correlation_id IS NOT NULL; - --- Causation tracing -CREATE INDEX IF NOT EXISTS ix_events_causation_id - ON events (causation_id) WHERE causation_id IS NOT NULL; - --- Category filtering (for readCategory — uses generated column, not LIKE) -CREATE INDEX IF NOT EXISTS ix_streams_category - ON streams (category); - --- Category read path: find $all entries by originating stream, ordered by global position --- Enables efficient category reads by allowing the planner to: look up category stream_ids → --- index scan $all for each → merge ordered by stream_version -CREATE INDEX IF NOT EXISTS ix_stream_events_all_by_origin - ON stream_events (original_stream_id, stream_version) - WHERE stream_id = 0; - --- Subscriptions (checkpoint persistence for subscription positions). --- consumer_group_member / consumer_group_size carry static consumer-group --- topology (ExecPlan 28 / EP-1). Non-group subscriptions are member 0, size 1. --- The unique key is composite (subscription_name, consumer_group_member) so each --- group member persists its own checkpoint under one shared subscription name. -CREATE TABLE IF NOT EXISTS subscriptions ( - subscription_id BIGSERIAL PRIMARY KEY, - subscription_name TEXT NOT NULL, - stream_name TEXT NOT NULL DEFAULT '$all', - last_seen BIGINT NOT NULL DEFAULT 0, - consumer_group_member INT NOT NULL DEFAULT 0, - consumer_group_size INT NOT NULL DEFAULT 1, - created_at TIMESTAMPTZ NOT NULL DEFAULT now(), - updated_at TIMESTAMPTZ NOT NULL DEFAULT now() -); - --- Idempotent convergence for databases created before EP-1: add the columns if --- missing, drop the old auto-named single-column unique constraint if present, --- and install the composite unique index. All guarded so the bootstrap body is --- still safe in disposable local databases; codd records the timestamped --- migration and does not reapply it after a successful run. -ALTER TABLE subscriptions ADD COLUMN IF NOT EXISTS consumer_group_member INT NOT NULL DEFAULT 0; -ALTER TABLE subscriptions ADD COLUMN IF NOT EXISTS consumer_group_size INT NOT NULL DEFAULT 1; -ALTER TABLE subscriptions DROP CONSTRAINT IF EXISTS subscriptions_subscription_name_key; -CREATE UNIQUE INDEX IF NOT EXISTS ix_subscriptions_name_member - ON subscriptions (subscription_name, consumer_group_member); - --- Triggers - --- NOTIFY on stream changes (fires once per append, not per event) -CREATE OR REPLACE FUNCTION notify_events() RETURNS TRIGGER AS $$ -BEGIN - PERFORM pg_notify( - TG_TABLE_SCHEMA || '.events', - NEW.stream_name || ',' || NEW.stream_id || ',' || NEW.stream_version - ); - RETURN NEW; -END; -$$ LANGUAGE plpgsql; - -DROP TRIGGER IF EXISTS stream_events_notify ON streams; -CREATE TRIGGER stream_events_notify - AFTER INSERT OR UPDATE ON streams - FOR EACH ROW EXECUTE FUNCTION notify_events(); - --- Immutability: prevent event mutation -CREATE OR REPLACE FUNCTION prevent_mutation() RETURNS TRIGGER AS $$ -BEGIN - RAISE EXCEPTION 'Immutable table: % cannot be updated', TG_TABLE_NAME; -END; -$$ LANGUAGE plpgsql; - -DROP TRIGGER IF EXISTS no_update_events ON events; -CREATE TRIGGER no_update_events - BEFORE UPDATE ON events - FOR EACH ROW EXECUTE FUNCTION prevent_mutation(); - -DROP TRIGGER IF EXISTS no_update_stream_events ON stream_events; -CREATE TRIGGER no_update_stream_events - BEFORE UPDATE ON stream_events - FOR EACH ROW EXECUTE FUNCTION prevent_mutation(); - --- Gated hard deletes (for maintenance/GDPR only) -CREATE OR REPLACE FUNCTION protect_deletion() RETURNS TRIGGER AS $$ -BEGIN - IF current_setting('kiroku.enable_hard_deletes', true) = 'on' THEN - RETURN OLD; - END IF; - RAISE EXCEPTION 'Hard deletes require: SET LOCAL kiroku.enable_hard_deletes = ''on'''; -END; -$$ LANGUAGE plpgsql; - -DROP TRIGGER IF EXISTS no_delete_events ON events; -CREATE TRIGGER no_delete_events - BEFORE DELETE ON events - FOR EACH ROW EXECUTE FUNCTION protect_deletion(); - -DROP TRIGGER IF EXISTS no_delete_stream_events ON stream_events; -CREATE TRIGGER no_delete_stream_events - BEFORE DELETE ON stream_events - FOR EACH ROW EXECUTE FUNCTION protect_deletion(); - -DROP TRIGGER IF EXISTS no_delete_streams ON streams; -CREATE TRIGGER no_delete_streams - BEFORE DELETE ON streams - FOR EACH ROW EXECUTE FUNCTION protect_deletion(); - --- TRUNCATE bypasses row-level triggers, so the BEFORE DELETE triggers above --- do not protect against an operator running TRUNCATE on these tables. Add --- statement-level BEFORE TRUNCATE triggers gated by the same GUC so the --- protection is symmetric. See EP-1 F6. -CREATE OR REPLACE FUNCTION protect_truncation() RETURNS TRIGGER AS $$ -BEGIN - IF current_setting('kiroku.enable_hard_deletes', true) = 'on' THEN - RETURN NULL; - END IF; - RAISE EXCEPTION 'TRUNCATE requires: SET LOCAL kiroku.enable_hard_deletes = ''on'''; -END; -$$ LANGUAGE plpgsql; - -DROP TRIGGER IF EXISTS no_truncate_events ON events; -CREATE TRIGGER no_truncate_events - BEFORE TRUNCATE ON events - FOR EACH STATEMENT EXECUTE FUNCTION protect_truncation(); - -DROP TRIGGER IF EXISTS no_truncate_stream_events ON stream_events; -CREATE TRIGGER no_truncate_stream_events - BEFORE TRUNCATE ON stream_events - FOR EACH STATEMENT EXECUTE FUNCTION protect_truncation(); - -DROP TRIGGER IF EXISTS no_truncate_streams ON streams; -CREATE TRIGGER no_truncate_streams - BEFORE TRUNCATE ON streams - FOR EACH STATEMENT EXECUTE FUNCTION protect_truncation(); diff --git a/sql-migrations/2026-05-26-00-00-00-add-subscription-dead-letters.sql b/sql-migrations/2026-05-26-00-00-00-add-subscription-dead-letters.sql deleted file mode 100644 --- a/sql-migrations/2026-05-26-00-00-00-add-subscription-dead-letters.sql +++ /dev/null @@ -1,31 +0,0 @@ --- Add the kiroku.dead_letters table (MasterPlan 6 / EP-2, docs/plans/40-...). --- --- Forward, additive migration: codd applies this file once, records it, and --- skips it on later runs. It does not edit the bootstrap migration and does not --- mutate existing event data. --- --- A dead-letter row records one event that a subscription handler asked to --- "dead-letter" (return DeadLetter, or exhaust its bounded retry budget). The --- event itself stays immutable in kiroku.events; this table references it by --- event_id and global_position rather than copying the payload. The --- consumer_group_member column (default 0, matching kiroku.subscriptions) --- attributes the row to the member that produced it, so a consumer group's dead --- letters are per-member. The worker writes a row and advances the member's --- checkpoint in one atomic statement (see SQL.insertDeadLetterAndCheckpointStmt). - -CREATE TABLE IF NOT EXISTS kiroku.dead_letters ( - dead_letter_id BIGSERIAL PRIMARY KEY, - subscription_name TEXT NOT NULL, - consumer_group_member INT NOT NULL DEFAULT 0, - global_position BIGINT NOT NULL, - event_id UUID NOT NULL REFERENCES kiroku.events(event_id), - reason JSONB NOT NULL, - reason_summary TEXT NOT NULL, - attempt_count INT NOT NULL, - created_at TIMESTAMPTZ NOT NULL DEFAULT now(), - UNIQUE (subscription_name, consumer_group_member, global_position, event_id) -); - --- Operator read path: list a subscription member's dead letters by recency. -CREATE INDEX IF NOT EXISTS ix_dead_letters_subscription_created_at - ON kiroku.dead_letters (subscription_name, consumer_group_member, created_at); 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,56 +1,30 @@ -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE TemplateHaskell #-} - module Kiroku.Store.Migrations ( + DefinitionError, + MigrationComponent, + MigrationPlan, + PlanError, + kirokuMigrationPlan, kirokuMigrations, - runKirokuMigrations, - runKirokuMigrationsNoCheck, ) where -import Codd (ApplyResult (SchemasNotVerified), CoddSettings, VerifySchemas, applyMigrations, applyMigrationsNoCheck) -import Codd.Logging (runCoddLogger) -import Codd.Parsing (AddedSqlMigration, EnvVars, PureStream (..), parseAddedSqlMigration) -import Data.ByteString (ByteString) -import Data.FileEmbed (embedDir) -import Data.Text.Encoding qualified as TE -import Data.Time (DiffTime) -import Streaming.Prelude qualified as Streaming - --- | Kiroku's embedded SQL migrations, ordered by timestamped filename. -kirokuMigrations :: (MonadFail m, EnvVars m) => m [AddedSqlMigration m] -kirokuMigrations = - traverse parseEmbeddedMigration embeddedMigrationFiles - where - parseEmbeddedMigration :: forall m. (MonadFail m, EnvVars m) => (FilePath, ByteString) -> m (AddedSqlMigration m) - parseEmbeddedMigration (name, bytes) = do - let stream :: PureStream m - stream = PureStream $ Streaming.yield (TE.decodeUtf8 bytes) - result <- - parseAddedSqlMigration - name - stream - case result of - Left err -> fail ("Invalid Kiroku migration " <> name <> ": " <> err) - Right migration -> pure migration - --- | Run Kiroku's embedded migrations through codd. -runKirokuMigrations :: CoddSettings -> DiffTime -> VerifySchemas -> IO ApplyResult -runKirokuMigrations settings connectTimeout verifySchemas = - runCoddLogger $ do - migrations <- kirokuMigrations - applyMigrations settings (Just migrations) connectTimeout verifySchemas +import Data.List.NonEmpty (NonEmpty (..)) +import Database.PostgreSQL.Migrate ( + DefinitionError, + MigrationComponent, + MigrationPlan, + PlanError, + migrationPlan, + ) +import Kiroku.Store.Migrations.Internal.Definition (kirokuMigrations) -{- | Run Kiroku's embedded migrations through codd without expected-schema -verification. +{- | The complete single-component Kiroku migration plan. -This is the right entry point until the caller owns a codd expected-schema -snapshot. Use 'runKirokuMigrations' when schema verification is configured. +The embedded manifest is validated at compile time, so a component-definition +failure indicates a broken package invariant rather than an operator error. -} -runKirokuMigrationsNoCheck :: CoddSettings -> DiffTime -> IO ApplyResult -runKirokuMigrationsNoCheck settings connectTimeout = - runCoddLogger $ do - migrations <- kirokuMigrations - applyMigrationsNoCheck settings (Just migrations) connectTimeout (const (pure SchemasNotVerified)) - -embeddedMigrationFiles :: [(FilePath, ByteString)] -embeddedMigrationFiles = $(embedDir "sql-migrations") +kirokuMigrationPlan :: Either PlanError MigrationPlan +kirokuMigrationPlan = + 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 new file mode 100644 --- /dev/null +++ b/src/Kiroku/Store/Migrations/History/Codd.hs @@ -0,0 +1,99 @@ +{-# LANGUAGE TemplateHaskell #-} + +module Kiroku.Store.Migrations.History.Codd ( + kirokuCoddHistoryMappings, + kirokuCoddManifestText, + kirokuCoddSourceConfig, + kirokuCoddSourcePayloads, + kirokuLegacyMigrationNames, +) where + +import Data.Bifunctor (first) +import Data.ByteString (ByteString) +import Data.Foldable (toList) +import Data.List.NonEmpty (NonEmpty (..)) +import Data.Map.Strict qualified as Map +import Data.Text (Text) +import Database.PostgreSQL.Migrate ( + Confirmation, + ConnectionProvider, + EvidenceRequirement (Evidence), + HistoryMapping, + PayloadRelation (SamePayload), + historyMapping, + migrationId, + ) +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" + ] + +kirokuCoddHistoryMappings :: NonEmpty HistoryMapping +kirokuCoddHistoryMappings = + zipWithNonEmpty mapping kirokuLegacyMigrationNames nativeMigrationNames + where + mapping sourceFilename targetName = + historyMapping + (definitionInvariant (migrationId "kiroku" targetName)) + (Evidence sourceKey) + (SamePayload sourceKey) + where + sourceKey = definitionInvariant (first show (coddEvidenceKey sourceFilename)) + +kirokuCoddSourceConfig :: + ConnectionProvider -> + Bool -> + Text -> + Confirmation -> + Either CoddDefinitionError CoddSourceConfig +kirokuCoddSourceConfig sourceProvider strictSource 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" + ] + +kirokuCoddSourcePayloads :: Map.Map FilePath ByteString +kirokuCoddSourcePayloads = + 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 + +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 new file mode 100644 --- /dev/null +++ b/src/Kiroku/Store/Migrations/Internal/Definition.hs @@ -0,0 +1,23 @@ +{-# LANGUAGE TemplateHaskell #-} + +module Kiroku.Store.Migrations.Internal.Definition ( + embeddedMigrationEntries, + kirokuMigrations, +) where + +import Data.ByteString (ByteString) +import Data.List.NonEmpty (NonEmpty) +import Database.PostgreSQL.Migrate ( + DefinitionError, + MigrationComponent, + migrationComponentFromEmbeddedSql, + ) +import Database.PostgreSQL.Migrate.Embed (embedMigrationManifest) + +embeddedMigrationEntries :: NonEmpty (FilePath, ByteString) +embeddedMigrationEntries = + $(embedMigrationManifest "migrations/manifest") + +kirokuMigrations :: Either DefinitionError MigrationComponent +kirokuMigrations = + migrationComponentFromEmbeddedSql "kiroku" mempty embeddedMigrationEntries diff --git a/src/Kiroku/Store/Migrations/Internal/EmbedFile.hs b/src/Kiroku/Store/Migrations/Internal/EmbedFile.hs new file mode 100644 --- /dev/null +++ b/src/Kiroku/Store/Migrations/Internal/EmbedFile.hs @@ -0,0 +1,19 @@ +{-# LANGUAGE TemplateHaskell #-} + +module Kiroku.Store.Migrations.Internal.EmbedFile (embedTextFile) where + +import Data.ByteString qualified as ByteString +import Data.Text qualified as Text +import Data.Text.Encoding qualified as Text.Encoding +import Language.Haskell.TH (Exp (..), Lit (..), Q) +import Language.Haskell.TH.Syntax qualified as TH + +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)))) diff --git a/src/Kiroku/Store/Migrations/New.hs b/src/Kiroku/Store/Migrations/New.hs new file mode 100644 --- /dev/null +++ b/src/Kiroku/Store/Migrations/New.hs @@ -0,0 +1,40 @@ +module Kiroku.Store.Migrations.New ( + AuthoringError, + defaultMigrationsDir, + newMigrationFile, + migrationTemplate, +) 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, + newMigration, + newMigrationOptions, + ) +import System.FilePath (()) + +defaultMigrationsDir :: FilePath +defaultMigrationsDir = "migrations" + +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 + 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." + , "" + ] + ) diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -1,229 +1,396 @@ -module Main where +{-# LANGUAGE MultilineStrings #-} -import Codd (ApplyResult (SchemasNotVerified), CoddSettings (..)) -import Codd.Parsing (connStringParser) -import Codd.Representations.Types (DbRep (..)) -import Codd.Types (ConnectionString, SchemaAlgo (..), SchemaSelection (..), SqlSchema (..), TxnIsolationLvl (..), singleTryPolicy) -import Data.Aeson (Value (Null)) -import Data.Aeson qualified as Aeson -import Data.Attoparsec.Text (endOfInput, parseOnly) -import Data.Map qualified as Map +module Main (main) where + +import Control.Concurrent.Async (concurrently) +import Control.Exception (finally) +import Control.Monad (forM_) +import Data.ByteString (ByteString) +import Data.ByteString qualified as ByteString +import Data.Either (isLeft) +import Data.Foldable (toList) +import Data.Int (Int64) +import Data.List (sort) +import Data.List.NonEmpty (NonEmpty (..)) import Data.Text (Text) -import Data.Text qualified as T -import Data.Time (secondsToDiffTime) -import Data.Vector qualified as Vector +import Data.Text qualified as Text +import Data.Text.IO qualified as Text.IO +import Database.PostgreSQL.Migrate +import Database.PostgreSQL.Migrate.History.Codd +import Database.PostgreSQL.Migrate.Internal (migrationChecksumBytes) +import Database.PostgreSQL.Migrate.Test (withMigratedDatabase) import EphemeralPg qualified as Pg -import Hasql.Connection.Settings qualified as Conn -import Hasql.Decoders qualified as D -import Hasql.Encoders qualified as E -import Hasql.Pool qualified as Pool -import Hasql.Pool.Config qualified as Pool.Config +import Hasql.Connection qualified as Connection +import Hasql.Connection.Settings qualified as Settings +import Hasql.Decoders qualified as Decoders +import Hasql.Encoders qualified as Encoders import Hasql.Session qualified as Session -import Hasql.Statement (Statement, preparable) -import Kiroku.Store -import Kiroku.Store.Migrations (runKirokuMigrationsNoCheck) +import Hasql.Statement (Statement) +import Hasql.Statement qualified as Statement +import Kiroku.Store.Migrations +import Kiroku.Store.Migrations.History.Codd +import Kiroku.Store.Migrations.New +import Numeric qualified +import System.Directory (doesDirectoryExist, doesFileExist) +import System.FilePath (()) +import System.IO.Temp (withSystemTempDirectory) import Test.Hspec main :: IO () -main = - hspec $ - describe "codd migration spike" $ - it "applies Kiroku migrations, opens the store without startup DDL, and is repeatable" $ do - result <- Pg.withCached $ \db -> do - let connStr = Pg.connectionString db - coddSettings = testCoddSettings connStr +main = hspec $ do + describe "native Kiroku migration definition" $ do + it "tracks the eight native files in manifest order" $ do + directory <- findMigrationsDirectory + manifest <- Text.lines <$> Text.IO.readFile (directory "manifest") + manifest `shouldBe` Text.pack <$> nativeMigrationFiles - firstMigration <- runKirokuMigrationsNoCheck coddSettings (secondsToDiffTime 5) - firstMigration `shouldBeSchemasNotVerified` "first migration run" - assertBootstrapApplied connStr - assertSchemaPlacement connStr - assertDeadLettersTable connStr - assertDefaultUuidV7 connStr + 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) - withStore - (defaultConnectionSettings connStr) - $ \store -> do - let stream = StreamName "migration-consumer" - event = makeEvent "MigrationConsumerChecked" (Aeson.object [("ok", Aeson.Bool True)]) - appendResult <- runStoreIO store $ appendToStream stream NoStream [event] - case appendResult of - Left err -> expectationFailure ("appendToStream failed: " <> show err) - Right _ -> pure () + it "builds component kiroku and an eight-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` 8 - readResult <- runStoreIO store $ readStreamForward stream (StreamVersion 0) 10 - case readResult of - Left err -> expectationFailure ("readStreamForward failed: " <> show err) - Right events -> Vector.length events `shouldBe` 1 + 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"] - secondMigration <- runKirokuMigrationsNoCheck coddSettings (secondsToDiffTime 5) - secondMigration `shouldBeSchemasNotVerified` "second migration run" - assertBootstrapApplied connStr - assertDefaultUuidV7 connStr - case result of - Left err -> expectationFailure ("Failed to start ephemeral PostgreSQL: " <> show err) - Right () -> pure () + 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" -testCoddSettings :: Text -> CoddSettings -testCoddSettings connStr = - CoddSettings - { migsConnString = parseConnString connStr - , sqlMigrations = [] - , onDiskReps = Right (DbRep Null Map.empty Map.empty) - , namespacesToCheck = IncludeSchemas [SqlSchema "kiroku"] - , extraRolesToCheck = [] - , retryPolicy = singleTryPolicy - , txnIsolationLvl = DbDefault - , schemaAlgoOpts = SchemaAlgo False False False - } + describe "fresh native databases" $ do + it "applies all eight, 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 + verified <- verifyMigrationPlanWith defaultRunOptions provider plan >>= requireMigration + case verified of + VerificationReport verificationIssues applied _ _ -> do + verificationIssues `shouldBe` [] + length applied `shouldBe` 8 + either (expectationFailure . show) pure result -parseConnString :: Text -> ConnectionString -parseConnString connStr = - case parseOnly (connStringParser <* endOfInput) connStr of - Left err -> error ("Could not parse ephemeral PostgreSQL connection string for codd: " <> err) - Right parsed -> parsed + 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 8 AppliedNow, replicate 8 AlreadyApplied] -makeEvent :: Text -> Value -> EventData -makeEvent typ payload = - EventData - { eventId = Nothing - , eventType = EventType typ - , payload = payload - , metadata = Nothing - , causationId = Nothing - , correlationId = Nothing - } + describe "Codd history import" $ do + it "imports a current codd V5 ledger, verifies, and never replays SQL" $ + importFixture "codd" -shouldBeSchemasNotVerified :: ApplyResult -> String -> Expectation -shouldBeSchemasNotVerified SchemasNotVerified _ = pure () -shouldBeSchemasNotVerified _ label = expectationFailure (label <> " unexpectedly verified schemas") + it "imports the legacy codd_schema ledger shape" $ + importFixture "codd_schema" -assertBootstrapApplied :: Text -> IO () -assertBootstrapApplied connStr = do - pool <- Pool.acquire poolConfig - result <- Pool.use pool (Session.statement () bootstrapStmt) - Pool.release pool - case result of - Left err -> expectationFailure ("bootstrap verification query failed: " <> show err) - Right True -> pure () - Right False -> expectationFailure "Kiroku bootstrap migration did not create the $all stream" - where - poolConfig = - Pool.Config.settings - [ Pool.Config.staticConnectionSettings (Conn.connectionString connStr) - , Pool.Config.size 1 - ] + 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 -bootstrapStmt :: Statement () Bool -bootstrapStmt = - preparable - "SELECT EXISTS (SELECT 1 FROM kiroku.streams WHERE stream_id = 0 AND stream_name = '$all')" - E.noParams - (D.singleRow (D.column (D.nonNullable D.bool))) +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 + canaryId <- requireRight (migrationId "kiroku" "0008-schema-management-comment") + verifiedBeforeCanary <- verifyMigrationPlan defaultRunOptions settings plan >>= requireMigration + case verifiedBeforeCanary of + VerificationReport verificationIssues _ _ _ -> + verificationIssues + `shouldBe` [PendingMigration canaryId] + up <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration + reportOutcomes up `shouldBe` replicate 7 AlreadyApplied <> [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 + 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` (8, 7, True) -{- | Assert that the migration installed every Kiroku table under the @kiroku@ -schema and left @public@ free of Kiroku tables. The connection uses no special -@search_path@, so the schema-qualified @to_regclass@ checks prove placement -directly rather than relying on name resolution. --} -assertSchemaPlacement :: Text -> IO () -assertSchemaPlacement connStr = do - pool <- Pool.acquire poolConfig - result <- Pool.use pool (Session.statement () placementStmt) - Pool.release pool - case result of - Left err -> expectationFailure ("schema placement query failed: " <> show err) - Right (True, True) -> pure () - Right (kirokuPresent, publicAbsent) -> - expectationFailure - ( "expected all Kiroku tables in 'kiroku' and none in 'public'; got kirokuPresent=" - <> show kirokuPresent - <> ", publicAbsent=" - <> show publicAbsent - ) +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" + ] + +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 - poolConfig = - Pool.Config.settings - [ Pool.Config.staticConnectionSettings (Conn.connectionString connStr) - , Pool.Config.size 1 - ] + step value remaining = do + matches <- predicate value + values <- remaining + pure (if matches then value : values else values) -placementStmt :: Statement () (Bool, Bool) -placementStmt = - preparable - "SELECT \ - \ (to_regclass('kiroku.streams') IS NOT NULL \ - \ AND to_regclass('kiroku.events') IS NOT NULL \ - \ AND to_regclass('kiroku.stream_events') IS NOT NULL \ - \ AND to_regclass('kiroku.subscriptions') IS NOT NULL), \ - \ (to_regclass('public.streams') IS NULL \ - \ AND to_regclass('public.events') IS NULL \ - \ AND to_regclass('public.stream_events') IS NULL \ - \ AND to_regclass('public.subscriptions') IS NULL)" - E.noParams - (D.singleRow ((,) <$> D.column (D.nonNullable D.bool) <*> D.column (D.nonNullable D.bool))) +parseLockfile :: Text -> [(FilePath, Text)] +parseLockfile contents = + [ (Text.unpack filename, checksum) + | line <- Text.lines contents + , [checksum, filename] <- [Text.words line] + ] -{- | Assert that the forward migration installed the @kiroku.dead_letters@ table -(MasterPlan 6 / EP-40) — proving codd applied the timestamped migration after the -bootstrap, not only the bootstrap itself. --} -assertDeadLettersTable :: Text -> IO () -assertDeadLettersTable connStr = do - pool <- Pool.acquire poolConfig - result <- Pool.use pool (Session.statement () deadLettersStmt) - Pool.release pool - case result of - Left err -> expectationFailure ("dead_letters verification query failed: " <> show err) - Right True -> pure () - Right False -> expectationFailure "Kiroku migration did not create the kiroku.dead_letters table" +checksumText :: ByteString -> Text +checksumText = + Text.pack + . concatMap renderByte + . ByteString.unpack + . migrationChecksumBytes + . migrationFingerprint where - poolConfig = - Pool.Config.settings - [ Pool.Config.staticConnectionSettings (Conn.connectionString connStr) - , Pool.Config.size 1 - ] + renderByte byte = + case Numeric.showHex byte "" of + [digit] -> ['0', digit] + digits -> digits -deadLettersStmt :: Statement () Bool -deadLettersStmt = - preparable - "SELECT to_regclass('kiroku.dead_letters') IS NOT NULL" - E.noParams - (D.singleRow (D.column (D.nonNullable D.bool))) +dropSqlSuffix :: FilePath -> String +dropSqlSuffix = reverse . drop 4 . reverse -assertDefaultUuidV7 :: Text -> IO () -assertDefaultUuidV7 connStr = do - pool <- Pool.acquire poolConfig - result <- Pool.use pool (Session.statement () defaultUuidStmt) - versionResult <- Pool.use pool (Session.statement () serverVersionStmt) - Pool.release pool - case (result, versionResult) of - (Right eventIdText, Right version) - | T.length eventIdText > 14 && T.index eventIdText 14 == '7' -> pure () - | otherwise -> - expectationFailure - ( "expected migration-created database default to generate UUIDv7 on PostgreSQL " - <> T.unpack version - <> ", got " - <> T.unpack eventIdText - ) - (Left err, _) -> expectationFailure ("default UUID insert failed: " <> show err) - (_, Left err) -> expectationFailure ("server version query failed: " <> show err) +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 + +kirokuPgConfig :: Pg.Config +kirokuPgConfig = Pg.defaultConfig{Pg.user = "kiroku"} + +withKirokuPg :: (Pg.Database -> IO ()) -> IO () +withKirokuPg action = do + started <- Pg.startCached kirokuPgConfig 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_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)), + (obj_description(to_regnamespace('kiroku'), 'pg_namespace') = 'Managed by pg-migrate component kiroku through 0008-schema-management-comment') + ) AS checks(ok) + """ + Encoders.noParams + (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.bool))) + +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 + +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 - poolConfig = - Pool.Config.settings - [ Pool.Config.staticConnectionSettings (Conn.connectionString connStr) - , Pool.Config.size 1 - ] + 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" -defaultUuidStmt :: Statement () Text -defaultUuidStmt = - preparable - "INSERT INTO kiroku.events (event_type, data) VALUES ('DefaultUuidGenerated', '{}'::jsonb) RETURNING event_id::text" - E.noParams - (D.singleRow (D.column (D.nonNullable D.text))) +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))) -serverVersionStmt :: Statement () Text -serverVersionStmt = - preparable - "SELECT current_setting('server_version_num')" - E.noParams - (D.singleRow (D.column (D.nonNullable D.text))) +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