#lang racket/base

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Ordered PostgreSQL schema migrations for existing wiki databases.
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(require db
         racket/file
         racket/path
         racket/string
         "attachment-references.rkt"
         "concept-id.rkt"
         "config.rkt"
         "todo.rkt")

(provide current-schema-version
         database-schema-version
         migrate-database!)

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Supporting functions
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(define current-schema-version 21)

(define schema-1-statements
  (list
   #<<SQL
CREATE TABLE IF NOT EXISTS users (
  id BIGSERIAL PRIMARY KEY,
  username TEXT NOT NULL UNIQUE,
  display_name TEXT NOT NULL,
  password_hash TEXT NOT NULL,
  role TEXT NOT NULL CHECK(role IN ('reader', 'editor', 'admin')),
  enabled BOOLEAN NOT NULL DEFAULT TRUE,
  created_at BIGINT NOT NULL,
  updated_at BIGINT NOT NULL
)
SQL
   #<<SQL
CREATE TABLE IF NOT EXISTS sessions (
  token_hash TEXT PRIMARY KEY,
  user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
  csrf_token TEXT NOT NULL,
  created_at BIGINT NOT NULL,
  expires_at BIGINT NOT NULL
)
SQL
   "CREATE INDEX IF NOT EXISTS sessions_expires_idx ON sessions(expires_at)"
   #<<SQL
CREATE TABLE IF NOT EXISTS pages (
  id BIGSERIAL PRIMARY KEY,
  slug TEXT NOT NULL UNIQUE,
  title TEXT NOT NULL,
  markdown TEXT NOT NULL,
  tags TEXT NOT NULL DEFAULT '[]',
  current_version BIGINT NOT NULL DEFAULT 1,
  created_at BIGINT NOT NULL,
  updated_at BIGINT NOT NULL,
  created_by TEXT NOT NULL,
  updated_by TEXT NOT NULL,
  archived BOOLEAN NOT NULL DEFAULT FALSE,
  archived_at BIGINT,
  archived_by TEXT,
  search_document TSVECTOR NOT NULL
)
SQL
   "CREATE INDEX IF NOT EXISTS pages_search_idx ON pages USING GIN(search_document)"
   "CREATE INDEX IF NOT EXISTS pages_title_idx ON pages(lower(title))"
   #<<SQL
CREATE TABLE IF NOT EXISTS page_versions (
  id BIGSERIAL PRIMARY KEY,
  page_id BIGINT NOT NULL REFERENCES pages(id) ON DELETE CASCADE,
  version BIGINT NOT NULL,
  title TEXT NOT NULL,
  markdown TEXT NOT NULL,
  tags TEXT NOT NULL DEFAULT '[]',
  author TEXT NOT NULL,
  action TEXT NOT NULL,
  summary TEXT NOT NULL,
  created_at BIGINT NOT NULL,
  UNIQUE(page_id, version)
)
SQL
   "CREATE INDEX IF NOT EXISTS page_versions_page_idx ON page_versions(page_id, version DESC)"
   #<<SQL
CREATE TABLE IF NOT EXISTS attachments (
  id BIGSERIAL PRIMARY KEY,
  page_id BIGINT NOT NULL REFERENCES pages(id) ON DELETE CASCADE,
  original_name TEXT NOT NULL,
  stored_name TEXT NOT NULL,
  size BIGINT NOT NULL,
  uploaded_at BIGINT NOT NULL,
  uploaded_by TEXT NOT NULL,
  UNIQUE(page_id, stored_name)
)
SQL
   ))

(define (table-exists? db name)
  (if (query-maybe-value db "SELECT to_regclass($1) IS NOT NULL" (string-append "public." name))
      #t
      #f))

(define (ensure-schema-table! db)
  (query-exec db
              #<<SQL
CREATE TABLE IF NOT EXISTS wiki_schema (
  version INTEGER PRIMARY KEY,
  applied_at BIGINT NOT NULL
)
SQL
              ))

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
; goal   : Read the newest recorded wiki database schema version.
; pre    : db is an open PostgreSQL connection.
; post   : Schema tables have only been inspected.
; result : The highest recorded version, or 0 when wiki_schema does not exist.
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(define (database-schema-version db)
  (if (table-exists? db "wiki_schema")
      (query-value db "SELECT COALESCE(MAX(version), 0) FROM wiki_schema")
      0))

(define (record-schema-version! db version)
  (query-exec db
              "INSERT INTO wiki_schema(version, applied_at) VALUES ($1, $2) ON CONFLICT (version) DO NOTHING"
              version
              (current-seconds)))

(define (recognize-schema-1? db)
  (and (table-exists? db "users")
       (table-exists? db "sessions")
       (table-exists? db "pages")
       (table-exists? db "page_versions")
       (table-exists? db "attachments")))

(define (install-schema-1! db)
  (for ((statement (in-list schema-1-statements)))
    (query-exec db statement))
  (ensure-schema-table! db)
  (record-schema-version! db 1))

(define (recognize-or-install-schema-1! db)
  (cond
    ((table-exists? db "wiki_schema")
     (void))
    ((recognize-schema-1? db)
     (ensure-schema-table! db)
     (record-schema-version! db 1))
    (else
     (install-schema-1! db))))

(define (legacy-mime-type stored-name)
  (let ((lower (string-downcase stored-name)))
    (cond
      ((regexp-match? #px"[.]png$" lower) "image/png")
      ((regexp-match? #px"[.](jpg|jpeg)$" lower) "image/jpeg")
      ((regexp-match? #px"[.]gif$" lower) "image/gif")
      ((regexp-match? #px"[.]webp$" lower) "image/webp")
      ((regexp-match? #px"[.]pdf$" lower) "application/pdf")
      ((regexp-match? #px"[.]txt$" lower) "text/plain; charset=utf-8")
      (else "application/octet-stream"))))

(define (migrate-1->2! db config)
  (query-exec db "ALTER TABLE attachments ADD COLUMN IF NOT EXISTS mime_type TEXT")
  (query-exec db "ALTER TABLE attachments ADD COLUMN IF NOT EXISTS content BYTEA")
  (let ((rows
         (query-rows db
                     #<<SQL
SELECT a.id, p.slug, a.stored_name, a.content
FROM attachments a
JOIN pages p ON p.id = a.page_id
ORDER BY a.id
SQL
                     )))
    (for ((row (in-list rows)))
      (let ((attachment-id (vector-ref row 0))
            (slug (vector-ref row 1))
            (stored-name (vector-ref row 2))
            (content (vector-ref row 3)))
        (unless (bytes? content)
          (let ((path (build-path (uploads-directory config) slug stored-name)))
            (unless (file-exists? path)
              (error 'migrate-database!
                     "schema 1 -> 2 cannot migrate attachment ~a: missing file ~a"
                     stored-name
                     (path->string path)))
            (let ((bytes (file->bytes path)))
              (query-exec db
                          "UPDATE attachments SET content = $1, mime_type = $2, size = $3 WHERE id = $4"
                          bytes
                          (legacy-mime-type stored-name)
                          (bytes-length bytes)
                          attachment-id))))))
    (query-exec db "UPDATE attachments SET mime_type = 'application/octet-stream' WHERE mime_type IS NULL")
    (query-exec db "ALTER TABLE attachments ALTER COLUMN mime_type SET NOT NULL")
    (query-exec db "ALTER TABLE attachments ALTER COLUMN content SET NOT NULL")
    (record-schema-version! db 2)))


(define (replace-page-todos! db page-id markdown)
  (query-exec db "DELETE FROM todo_items WHERE page_id = $1" page-id)
  (for ((item (in-list (extract-todos markdown))))
    (query-exec db
                #<<SQL
INSERT INTO todo_items(page_id, item_number, line_number, text)
VALUES ($1, $2, $3, $4)
SQL
                page-id
                (hash-ref item 'number)
                (hash-ref item 'line)
                (hash-ref item 'text))))

(define (migrate-2->3! db)
  (query-exec db
              #<<SQL
CREATE TABLE IF NOT EXISTS todo_items (
  id BIGSERIAL PRIMARY KEY,
  page_id BIGINT NOT NULL REFERENCES pages(id) ON DELETE CASCADE,
  item_number INTEGER NOT NULL,
  line_number INTEGER NOT NULL,
  text TEXT NOT NULL,
  UNIQUE(page_id, item_number)
)
SQL
              )
  (query-exec db "CREATE INDEX IF NOT EXISTS todo_items_page_idx ON todo_items(page_id, item_number)")
  (for ((row (in-list (query-rows db "SELECT id, markdown FROM pages WHERE archived = FALSE"))))
    (replace-page-todos! db (vector-ref row 0) (vector-ref row 1)))
  (record-schema-version! db 3))


(define (migrate-3->4! db)
  (query-exec db
              #<<SQL
CREATE TABLE IF NOT EXISTS bookmarks (
  user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
  page_id BIGINT NOT NULL REFERENCES pages(id) ON DELETE CASCADE,
  section TEXT NOT NULL DEFAULT '',
  position INTEGER NOT NULL DEFAULT 0,
  created_at BIGINT NOT NULL,
  PRIMARY KEY(user_id, page_id)
)
SQL
              )
  (query-exec db
              "CREATE INDEX IF NOT EXISTS bookmarks_user_idx ON bookmarks(user_id, section, position, created_at)")
  (record-schema-version! db 4))

(define (migrate-4->5! db)
  (for ((row (in-list (query-rows db "SELECT id, markdown FROM pages WHERE archived = FALSE"))))
    (replace-page-todos! db (vector-ref row 0) (vector-ref row 1)))
  (record-schema-version! db 5))



(define (migrate-5->6! db)
  ;; Namespace is added here as well so a fresh migration chain can rebuild
  ;; attachment URLs before schema 7 creates the namespace indexes.
  (query-exec db "ALTER TABLE pages ADD COLUMN IF NOT EXISTS namespace TEXT NOT NULL DEFAULT ''")
  (query-exec db
              #<<SQL
CREATE TABLE IF NOT EXISTS attachment_references (
  id BIGSERIAL PRIMARY KEY,
  attachment_id BIGINT NOT NULL REFERENCES attachments(id) ON DELETE CASCADE,
  page_id BIGINT NOT NULL REFERENCES pages(id) ON DELETE CASCADE,
  page_version_id BIGINT REFERENCES page_versions(id) ON DELETE CASCADE,
  current_reference BOOLEAN NOT NULL,
  referenced_at BIGINT NOT NULL
)
SQL
              )
  (query-exec db
              "CREATE INDEX IF NOT EXISTS attachment_references_attachment_idx ON attachment_references(attachment_id, current_reference, referenced_at DESC)")
  (query-exec db
              "CREATE INDEX IF NOT EXISTS attachment_references_page_idx ON attachment_references(page_id, current_reference)")
  (rebuild-attachment-references! db)
  (record-schema-version! db 6))



(define (migrate-6->7! db)
  (query-exec db "ALTER TABLE pages ADD COLUMN IF NOT EXISTS namespace TEXT NOT NULL DEFAULT ''")
  (query-exec db "ALTER TABLE pages DROP CONSTRAINT IF EXISTS pages_slug_key")
  (query-exec db "CREATE UNIQUE INDEX IF NOT EXISTS pages_namespace_slug_idx ON pages(namespace, slug)")
  (query-exec db "CREATE INDEX IF NOT EXISTS pages_namespace_idx ON pages(lower(namespace), lower(title), title)")
  (record-schema-version! db 7))

(define (migrate-7->8! db)
  (query-exec db
              #<<SQL
CREATE TABLE IF NOT EXISTS page_aliases (
  id BIGSERIAL PRIMARY KEY,
  namespace TEXT NOT NULL DEFAULT '',
  slug TEXT NOT NULL,
  title TEXT NOT NULL,
  page_id BIGINT NOT NULL REFERENCES pages(id) ON DELETE CASCADE,
  created_at BIGINT NOT NULL,
  created_by TEXT NOT NULL,
  UNIQUE(namespace, slug)
)
SQL
              )
  (query-exec db
              "CREATE INDEX IF NOT EXISTS page_aliases_page_idx ON page_aliases(page_id, created_at DESC)")
  (record-schema-version! db 8))

(define (migrate-8->9! db)
  (query-exec db
              #<<SQL
CREATE TABLE IF NOT EXISTS concept_maps (
  id BIGSERIAL PRIMARY KEY,
  slug TEXT NOT NULL UNIQUE,
  title TEXT NOT NULL,
  document JSONB NOT NULL,
  current_version BIGINT NOT NULL DEFAULT 1,
  created_at BIGINT NOT NULL,
  updated_at BIGINT NOT NULL,
  created_by TEXT NOT NULL,
  updated_by TEXT NOT NULL,
  archived BOOLEAN NOT NULL DEFAULT FALSE,
  archived_at BIGINT,
  archived_by TEXT
)
SQL
              )
  (query-exec db
              "CREATE INDEX IF NOT EXISTS concept_maps_title_idx ON concept_maps(lower(title))")
  (record-schema-version! db 9))

(define (migrate-9->10! db)
  (query-exec db "ALTER TABLE users ADD COLUMN IF NOT EXISTS email TEXT")
  (query-exec db
              "CREATE UNIQUE INDEX IF NOT EXISTS users_email_unique_idx ON users(lower(email)) WHERE email IS NOT NULL")
  (query-exec db
              #<<SQL
CREATE TABLE IF NOT EXISTS password_reset_tokens (
  token_hash TEXT PRIMARY KEY,
  user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
  created_at BIGINT NOT NULL,
  expires_at BIGINT NOT NULL,
  used_at BIGINT
)
SQL
              )
  (query-exec db
              #<<SQL
CREATE TABLE IF NOT EXISTS wiki_settings (
  key TEXT PRIMARY KEY,
  value TEXT NOT NULL,
  updated_at BIGINT NOT NULL
)
SQL
              )
  (query-exec db
              "CREATE INDEX IF NOT EXISTS password_reset_tokens_user_idx ON password_reset_tokens(user_id, expires_at DESC)")
  (query-exec db
              "CREATE INDEX IF NOT EXISTS password_reset_tokens_expires_idx ON password_reset_tokens(expires_at)")
  (record-schema-version! db 10))

(define (migrate-10->11! db)
  (query-exec db
              #<<SQL
CREATE TABLE IF NOT EXISTS concept_map_versions (
  id BIGSERIAL PRIMARY KEY,
  concept_map_id BIGINT NOT NULL REFERENCES concept_maps(id) ON DELETE CASCADE,
  version BIGINT NOT NULL,
  title TEXT NOT NULL,
  document JSONB NOT NULL,
  author TEXT NOT NULL,
  action TEXT NOT NULL,
  summary TEXT NOT NULL,
  created_at BIGINT NOT NULL,
  UNIQUE(concept_map_id, version)
)
SQL
              )
  (query-exec db
              "CREATE INDEX IF NOT EXISTS concept_map_versions_map_idx ON concept_map_versions(concept_map_id, version DESC)")
  (query-exec db
              #<<SQL
INSERT INTO concept_map_versions
  (concept_map_id, version, title, document, author, action, summary, created_at)
SELECT id, current_version, title, document, updated_by, 'snapshot',
       'Current state when CMap history was enabled', updated_at
FROM concept_maps
ON CONFLICT (concept_map_id, version) DO NOTHING
SQL
              )
  (record-schema-version! db 11))

(define (migrate-11->12! db)
  ;; Older clients stored both automatic and explicit saves as `edit`. Preserve
  ;; the explicit saves, discard redundant autosave copies, and enforce the new
  ;; five-item manual history limit per CMap.
  (query-exec db
              #<<SQL
UPDATE concept_map_versions
SET action = 'manual'
WHERE action = 'edit'
  AND summary IN ('Manual save', 'Handmatig opgeslagen')
SQL
              )
  (query-exec db
              #<<SQL
DELETE FROM concept_map_versions
WHERE action = 'edit'
  AND summary IN ('Automatic save', 'Automatisch opgeslagen')
SQL
              )
  (query-exec db
              #<<SQL
WITH ranked AS (
  SELECT id,
         row_number() OVER (
           PARTITION BY concept_map_id
           ORDER BY version DESC
         ) AS position
  FROM concept_map_versions
  WHERE action = 'manual'
)
DELETE FROM concept_map_versions cmv
USING ranked
WHERE cmv.id = ranked.id AND ranked.position > 5
SQL
              )
  (record-schema-version! db 12))

(define (migrate-12->13! db)
  (query-exec db
              #<<SQL
CREATE TABLE IF NOT EXISTS people (
  id BIGSERIAL PRIMARY KEY,
  name TEXT NOT NULL,
  active BOOLEAN NOT NULL DEFAULT TRUE,
  created_at BIGINT NOT NULL,
  updated_at BIGINT NOT NULL
)
SQL
              )
  (query-exec db "CREATE UNIQUE INDEX IF NOT EXISTS people_name_unique_idx ON people(lower(name))")
  (query-exec
   db
   #<<SQL
WITH person_names AS (
  SELECT DISTINCT trim(tag ->> 'value') AS name
  FROM concept_maps cm
  CROSS JOIN LATERAL jsonb_array_elements(
    CASE WHEN jsonb_typeof(cm.document -> 'concepts') = 'array'
         THEN cm.document -> 'concepts' ELSE '[]'::jsonb END
  ) AS concept
  CROSS JOIN LATERAL jsonb_array_elements(
    CASE WHEN jsonb_typeof(concept -> 'tags') = 'array'
         THEN concept -> 'tags' ELSE '[]'::jsonb END
  ) AS tag
  WHERE tag ->> 'type' = 'person' AND trim(coalesce(tag ->> 'value', '')) <> ''
)
INSERT INTO people(name, active, created_at, updated_at)
SELECT name, TRUE, $1, $1 FROM person_names
ON CONFLICT (lower(name)) DO NOTHING
SQL
   (current-seconds))
  (record-schema-version! db 13))

(define (migrate-13->14! db)
  ;; Concept identity is shared between CMaps. Older generated numeric ids were
  ;; only unique inside one map, so namespace those before building the global
  ;; repository.
  (query-exec
   db
   #<<SQL
UPDATE concept_maps cm
SET document = cm.document || jsonb_build_object(
  'concepts', CASE
    WHEN jsonb_typeof(cm.document -> 'concepts') = 'array' THEN
      (SELECT coalesce(jsonb_agg(
         CASE WHEN concept ->> 'id' ~ '^concept-[0-9]+$'
              THEN jsonb_set(concept, '{id}', to_jsonb(concat('legacy:', cm.slug, ':', concept ->> 'id')))
              ELSE concept END
       ), '[]'::jsonb)
       FROM jsonb_array_elements(cm.document -> 'concepts') AS concept)
    ELSE coalesce(cm.document -> 'concepts', '[]'::jsonb)
  END,
  'items', CASE
    WHEN jsonb_typeof(cm.document -> 'items') = 'array' THEN
      (SELECT coalesce(jsonb_agg(
         CASE WHEN item ->> 'conceptId' ~ '^concept-[0-9]+$'
              THEN jsonb_set(item, '{conceptId}', to_jsonb(concat('legacy:', cm.slug, ':', item ->> 'conceptId')))
              ELSE item END
       ), '[]'::jsonb)
       FROM jsonb_array_elements(cm.document -> 'items') AS item)
    ELSE coalesce(cm.document -> 'items', '[]'::jsonb)
  END
)
WHERE jsonb_typeof(cm.document) = 'object'
SQL
   )
  (query-exec
   db
   #<<SQL
CREATE TABLE IF NOT EXISTS concept_definitions (
  id TEXT PRIMARY KEY,
  document JSONB NOT NULL,
  updated_at BIGINT NOT NULL,
  updated_by TEXT NOT NULL
)
SQL
   )
  ;; If a shared concept already has diverging copies, the copy from the most
  ;; recently edited active map becomes the initial canonical definition.
  (query-exec
   db
   #<<SQL
WITH candidates AS (
  SELECT concept ->> 'id' AS id,
         concept AS document,
         cm.updated_at,
         cm.updated_by,
         row_number() OVER (
           PARTITION BY concept ->> 'id'
           ORDER BY cm.updated_at DESC, cm.id DESC
         ) AS position
  FROM concept_maps cm
  CROSS JOIN LATERAL jsonb_array_elements(
    CASE WHEN jsonb_typeof(cm.document -> 'concepts') = 'array'
         THEN cm.document -> 'concepts' ELSE '[]'::jsonb END
  ) AS concept
  WHERE cm.archived = FALSE AND nullif(concept ->> 'id', '') IS NOT NULL
)
INSERT INTO concept_definitions(id, document, updated_at, updated_by)
SELECT id, document, updated_at, updated_by
FROM candidates
WHERE position = 1
ON CONFLICT (id) DO NOTHING
SQL
   )
  (record-schema-version! db 14))

(define (migrate-14->15! db)
  ;; Schema 14 originally filled the shared repository only from the concepts
  ;; array. Sub-CMap heads are placements too, but older documents often keep
  ;; their content solely on the item. Import those missing definitions.
  (query-exec
   db
   #<<SQL
WITH candidates AS (
  SELECT item ->> 'conceptId' AS id,
         jsonb_strip_nulls(jsonb_build_object(
           'id', item -> 'conceptId',
           'label', coalesce(repository.document, item) -> 'label',
           'synopsis', coalesce(repository.document, item) -> 'synopsis',
           'aspects', coalesce(repository.document, item) -> 'aspects',
           'tags', coalesce(repository.document, item) -> 'tags',
           'descriptionPageSlug', coalesce(repository.document, item) -> 'descriptionPageSlug',
           'pageSlug', coalesce(repository.document, item) -> 'pageSlug',
           'cmapSlug', coalesce(repository.document, item) -> 'cmapSlug',
           'parentCmapLink', coalesce(repository.document, item) -> 'parentCmapLink',
           'imageSource', coalesce(repository.document, item) -> 'imageSource'
         )) AS document,
         cm.updated_at,
         cm.updated_by,
         row_number() OVER (
           PARTITION BY item ->> 'conceptId'
           ORDER BY cm.updated_at DESC, cm.id DESC
         ) AS position
  FROM concept_maps cm
  CROSS JOIN LATERAL jsonb_array_elements(
    CASE WHEN jsonb_typeof(cm.document -> 'items') = 'array'
         THEN cm.document -> 'items' ELSE '[]'::jsonb END
  ) AS item
  LEFT JOIN LATERAL (
    SELECT concept AS document
    FROM jsonb_array_elements(
      CASE WHEN jsonb_typeof(cm.document -> 'concepts') = 'array'
           THEN cm.document -> 'concepts' ELSE '[]'::jsonb END
    ) AS concept
    WHERE concept ->> 'id' = item ->> 'conceptId'
    LIMIT 1
  ) repository ON TRUE
  WHERE cm.archived = FALSE
    AND nullif(item ->> 'conceptId', '') IS NOT NULL
    AND coalesce(item ->> 'kind', 'concept') <> 'phrase'
)
INSERT INTO concept_definitions(id, document, updated_at, updated_by)
SELECT id, document, updated_at, updated_by
FROM candidates
WHERE position = 1
ON CONFLICT (id) DO NOTHING
SQL
   )
  (record-schema-version! db 15))

(define (migrate-15->16! db)
  ;; A placement linked to a derived CMap and the head concept of that CMap
  ;; denote one concept. Retain aliases so old documents and open editors can
  ;; be canonicalized without making diagram structure part of the concept.
  (query-exec
   db
   #<<SQL
CREATE TABLE IF NOT EXISTS concept_aliases (
  alias_id TEXT PRIMARY KEY,
  canonical_id TEXT NOT NULL,
  created_at BIGINT NOT NULL,
  created_by TEXT NOT NULL,
  CHECK(alias_id <> canonical_id)
)
SQL
   )
  (query-exec db
              "CREATE INDEX IF NOT EXISTS concept_aliases_canonical_idx ON concept_aliases(canonical_id)")
  (query-exec
   db
   #<<SQL
WITH derived_roots AS (
  SELECT target.slug AS target_slug,
         root_item ->> 'conceptId' AS canonical_id
  FROM concept_maps target
  JOIN concept_maps source
    ON source.slug = target.document #>> '{derivedView,sourceCmapSlug}'
  CROSS JOIN LATERAL jsonb_array_elements(
    CASE WHEN jsonb_typeof(source.document -> 'items') = 'array'
         THEN source.document -> 'items' ELSE '[]'::jsonb END
  ) AS root_item
  WHERE target.archived = FALSE
    AND source.archived = FALSE
    AND root_item ->> 'id' = target.document #>> '{derivedView,rootItemId}'
    AND nullif(root_item ->> 'conceptId', '') IS NOT NULL
  UNION ALL
  SELECT target.slug AS target_slug,
         root_item ->> 'conceptId' AS canonical_id
  FROM concept_maps target
  CROSS JOIN LATERAL jsonb_array_elements(
    CASE WHEN jsonb_typeof(target.document -> 'items') = 'array'
         THEN target.document -> 'items' ELSE '[]'::jsonb END
  ) AS root_item
  LEFT JOIN LATERAL (
    SELECT concept AS document
    FROM jsonb_array_elements(
      CASE WHEN jsonb_typeof(target.document -> 'concepts') = 'array'
           THEN target.document -> 'concepts' ELSE '[]'::jsonb END
    ) AS concept
    WHERE concept ->> 'id' = root_item ->> 'conceptId'
    LIMIT 1
  ) repository ON TRUE
  WHERE target.archived = FALSE
    AND target.document -> 'derivedView' IS NULL
    AND nullif(root_item ->> 'conceptId', '') IS NOT NULL
    AND coalesce(root_item ->> 'kind', 'concept') <> 'phrase'
    AND nullif(root_item ->> 'parentSubmapId', '') IS NULL
    AND lower(trim(coalesce(repository.document ->> 'label', root_item ->> 'label', '')))
        = lower(trim(target.title))
), linked_placements AS (
  SELECT DISTINCT item ->> 'conceptId' AS alias_id,
         derived_roots.canonical_id
  FROM concept_maps cm
  CROSS JOIN LATERAL jsonb_array_elements(
    CASE WHEN jsonb_typeof(cm.document -> 'items') = 'array'
         THEN cm.document -> 'items' ELSE '[]'::jsonb END
  ) AS item
  JOIN derived_roots ON derived_roots.target_slug = item ->> 'cmapSlug'
  WHERE cm.archived = FALSE
    AND nullif(item ->> 'conceptId', '') IS NOT NULL
    AND item ->> 'conceptId' <> derived_roots.canonical_id
)
INSERT INTO concept_aliases(alias_id, canonical_id, created_at, created_by)
SELECT alias_id, canonical_id, $1, 'system:concept-identity-migration'
FROM linked_placements
ON CONFLICT (alias_id) DO UPDATE
SET canonical_id = EXCLUDED.canonical_id
SQL
   (current-seconds))
  ;; Preserve the newest available content when two ids are merged.
  (query-exec
   db
   #<<SQL
WITH resolved AS (
  SELECT coalesce(alias.canonical_id, definition.id) AS canonical_id,
         definition.document,
         definition.updated_at,
         definition.updated_by,
         row_number() OVER (
           PARTITION BY coalesce(alias.canonical_id, definition.id)
           ORDER BY definition.updated_at DESC, definition.id DESC
         ) AS position
  FROM concept_definitions definition
  LEFT JOIN concept_aliases alias ON alias.alias_id = definition.id
)
INSERT INTO concept_definitions(id, document, updated_at, updated_by)
SELECT canonical_id,
       jsonb_set(document - 'kind' - 'parentCmapLink', '{id}', to_jsonb(canonical_id)),
       updated_at,
       updated_by
FROM resolved
WHERE position = 1
ON CONFLICT (id) DO UPDATE
SET document = EXCLUDED.document,
    updated_at = EXCLUDED.updated_at,
    updated_by = EXCLUDED.updated_by
SQL
   )
  (record-schema-version! db 16))

(define (migrate-16->17! db)
  ;; Within one wiki a normalized concept name denotes one concept. Merge ids
  ;; that predate this invariant, while preserving a canonical id already
  ;; established through a linked CMap when one exists.
  (query-exec
   db
   #<<SQL
WITH named AS (
  SELECT definition.id,
         definition.document,
         definition.updated_at,
         definition.updated_by,
         lower(trim(definition.document ->> 'label')) AS name_key,
         coalesce(existing_alias.canonical_id, definition.id) AS resolved_id
  FROM concept_definitions definition
  LEFT JOIN concept_aliases existing_alias
    ON existing_alias.alias_id = definition.id
  WHERE nullif(trim(definition.document ->> 'label'), '') IS NOT NULL
), ranked AS (
  SELECT named.*,
         first_value(resolved_id) OVER (
           PARTITION BY name_key
           ORDER BY updated_at DESC, id DESC
         ) AS canonical_id,
         row_number() OVER (
           PARTITION BY name_key
           ORDER BY updated_at DESC, id DESC
         ) AS content_position
  FROM named
), new_aliases AS (
  SELECT DISTINCT id AS alias_id, canonical_id
  FROM ranked
  WHERE id <> canonical_id
), stored_aliases AS (
  INSERT INTO concept_aliases(alias_id, canonical_id, created_at, created_by)
  SELECT alias_id, canonical_id, $1, 'system:concept-name-migration'
  FROM new_aliases
  ON CONFLICT (alias_id) DO UPDATE
  SET canonical_id = EXCLUDED.canonical_id
  RETURNING alias_id
), winners AS (
  SELECT DISTINCT ON (canonical_id)
         canonical_id, document, updated_at, updated_by
  FROM ranked
  WHERE content_position = 1
  ORDER BY canonical_id, updated_at DESC, id DESC
)
INSERT INTO concept_definitions(id, document, updated_at, updated_by)
SELECT canonical_id,
       jsonb_set(document - 'kind' - 'parentCmapLink', '{id}', to_jsonb(canonical_id)),
       updated_at,
       updated_by
FROM winners
ON CONFLICT (id) DO UPDATE
SET document = EXCLUDED.document,
    updated_at = EXCLUDED.updated_at,
    updated_by = EXCLUDED.updated_by
SQL
   (current-seconds))
  (query-exec
   db
   #<<SQL
UPDATE concept_maps cm
SET document = cm.document || jsonb_build_object(
  'items', CASE
    WHEN jsonb_typeof(cm.document -> 'items') = 'array' THEN
      (SELECT coalesce(jsonb_agg(
         CASE WHEN alias.canonical_id IS NOT NULL
              THEN jsonb_set(item, '{conceptId}', to_jsonb(alias.canonical_id))
              ELSE item END
         ORDER BY ordinal
       ), '[]'::jsonb)
       FROM jsonb_array_elements(cm.document -> 'items') WITH ORDINALITY AS entry(item, ordinal)
       LEFT JOIN concept_aliases alias ON alias.alias_id = item ->> 'conceptId')
    ELSE coalesce(cm.document -> 'items', '[]'::jsonb)
  END,
  'concepts', CASE
    WHEN jsonb_typeof(cm.document -> 'concepts') = 'array' THEN
      (SELECT coalesce(jsonb_agg(rewritten ORDER BY ordinal), '[]'::jsonb)
       FROM (
         SELECT DISTINCT ON (coalesce(alias.canonical_id, concept ->> 'id'))
                CASE WHEN alias.canonical_id IS NOT NULL
                     THEN jsonb_set(concept, '{id}', to_jsonb(alias.canonical_id))
                     ELSE concept END AS rewritten,
                ordinal
         FROM jsonb_array_elements(cm.document -> 'concepts') WITH ORDINALITY
              AS entry(concept, ordinal)
         LEFT JOIN concept_aliases alias ON alias.alias_id = concept ->> 'id'
         ORDER BY coalesce(alias.canonical_id, concept ->> 'id'), ordinal DESC
       ) definitions)
    ELSE coalesce(cm.document -> 'concepts', '[]'::jsonb)
  END
)
WHERE EXISTS (
  SELECT 1
  FROM jsonb_array_elements(
    CASE WHEN jsonb_typeof(cm.document -> 'items') = 'array'
         THEN cm.document -> 'items' ELSE '[]'::jsonb END
  ) AS item
  JOIN concept_aliases alias ON alias.alias_id = item ->> 'conceptId'
)
OR EXISTS (
  SELECT 1
  FROM jsonb_array_elements(
    CASE WHEN jsonb_typeof(cm.document -> 'concepts') = 'array'
         THEN cm.document -> 'concepts' ELSE '[]'::jsonb END
  ) AS concept
  JOIN concept_aliases alias ON alias.alias_id = concept ->> 'id'
)
SQL
   )
  (query-exec
   db
   #<<SQL
DELETE FROM concept_definitions definition
USING concept_aliases alias
WHERE definition.id = alias.alias_id
SQL
   )
  (record-schema-version! db 17))

(define (migrate-17->18! db)
  ;; Normalize the current documents themselves. Compatibility aliases are not
  ;; part of the final model: every active placement is rewritten to the one
  ;; id selected for its normalized name, then duplicate definitions are gone.
  (query-exec
   db
   #<<SQL
CREATE TEMP TABLE concept_id_merge (
  old_id TEXT PRIMARY KEY,
  canonical_id TEXT NOT NULL
) ON COMMIT DROP
SQL
   )
  (query-exec
   db
   #<<SQL
WITH placements AS (
  SELECT item ->> 'conceptId' AS concept_id,
         lower(trim(coalesce(repository.document ->> 'label', item ->> 'label', ''))) AS name_key,
         cm.updated_at,
         cm.id AS map_id
  FROM concept_maps cm
  CROSS JOIN LATERAL jsonb_array_elements(
    CASE WHEN jsonb_typeof(cm.document -> 'items') = 'array'
         THEN cm.document -> 'items' ELSE '[]'::jsonb END
  ) AS item
  LEFT JOIN LATERAL (
    SELECT concept AS document
    FROM jsonb_array_elements(
      CASE WHEN jsonb_typeof(cm.document -> 'concepts') = 'array'
           THEN cm.document -> 'concepts' ELSE '[]'::jsonb END
    ) AS concept
    WHERE concept ->> 'id' = item ->> 'conceptId'
    LIMIT 1
  ) repository ON TRUE
  WHERE cm.archived = FALSE
    AND nullif(item ->> 'conceptId', '') IS NOT NULL
    AND coalesce(item ->> 'kind', 'concept') <> 'phrase'
), latest_name_per_id AS (
  SELECT DISTINCT ON (concept_id)
         concept_id, name_key, updated_at, map_id
  FROM placements
  WHERE name_key <> ''
  ORDER BY concept_id, updated_at DESC, map_id DESC
), ranked AS (
  SELECT concept_id,
         first_value(concept_id) OVER (
           PARTITION BY name_key
           ORDER BY updated_at DESC, map_id DESC, concept_id DESC
         ) AS canonical_id
  FROM latest_name_per_id
)
INSERT INTO concept_id_merge(old_id, canonical_id)
SELECT concept_id, canonical_id
FROM ranked
WHERE concept_id <> canonical_id
SQL
   )
  (query-exec
   db
   #<<SQL
UPDATE concept_maps cm
SET document = cm.document || jsonb_build_object(
  'items', CASE
    WHEN jsonb_typeof(cm.document -> 'items') = 'array' THEN
      (SELECT coalesce(jsonb_agg(
         CASE WHEN merge.canonical_id IS NOT NULL
              THEN jsonb_set(item, '{conceptId}', to_jsonb(merge.canonical_id))
              ELSE item END
         ORDER BY ordinal
       ), '[]'::jsonb)
       FROM jsonb_array_elements(cm.document -> 'items') WITH ORDINALITY AS entry(item, ordinal)
       LEFT JOIN concept_id_merge merge ON merge.old_id = item ->> 'conceptId')
    ELSE coalesce(cm.document -> 'items', '[]'::jsonb)
  END,
  'concepts', CASE
    WHEN jsonb_typeof(cm.document -> 'concepts') = 'array' THEN
      (SELECT coalesce(jsonb_agg(rewritten ORDER BY ordinal), '[]'::jsonb)
       FROM (
         SELECT DISTINCT ON (coalesce(merge.canonical_id, concept ->> 'id'))
                CASE WHEN merge.canonical_id IS NOT NULL
                     THEN jsonb_set(concept, '{id}', to_jsonb(merge.canonical_id))
                     ELSE concept END AS rewritten,
                ordinal
         FROM jsonb_array_elements(cm.document -> 'concepts') WITH ORDINALITY
              AS entry(concept, ordinal)
         LEFT JOIN concept_id_merge merge ON merge.old_id = concept ->> 'id'
         ORDER BY coalesce(merge.canonical_id, concept ->> 'id'), ordinal DESC
       ) definitions)
    ELSE coalesce(cm.document -> 'concepts', '[]'::jsonb)
  END
)
WHERE EXISTS (
  SELECT 1
  FROM jsonb_array_elements(
    CASE WHEN jsonb_typeof(cm.document -> 'items') = 'array'
         THEN cm.document -> 'items' ELSE '[]'::jsonb END
  ) AS item
  JOIN concept_id_merge merge ON merge.old_id = item ->> 'conceptId'
)
OR EXISTS (
  SELECT 1
  FROM jsonb_array_elements(
    CASE WHEN jsonb_typeof(cm.document -> 'concepts') = 'array'
         THEN cm.document -> 'concepts' ELSE '[]'::jsonb END
  ) AS concept
  JOIN concept_id_merge merge ON merge.old_id = concept ->> 'id'
)
SQL
   )
  (query-exec db "DELETE FROM concept_definitions")
  (query-exec
   db
   #<<SQL
WITH candidates AS (
  SELECT concept ->> 'id' AS id,
         concept - 'kind' - 'parentCmapLink' AS document,
         cm.updated_at,
         cm.updated_by,
         cm.id AS map_id
  FROM concept_maps cm
  CROSS JOIN LATERAL jsonb_array_elements(
    CASE WHEN jsonb_typeof(cm.document -> 'concepts') = 'array'
         THEN cm.document -> 'concepts' ELSE '[]'::jsonb END
  ) AS concept
  WHERE cm.archived = FALSE AND nullif(concept ->> 'id', '') IS NOT NULL
  UNION ALL
  SELECT item ->> 'conceptId' AS id,
         jsonb_strip_nulls(jsonb_build_object(
           'id', item -> 'conceptId',
           'label', item -> 'label',
           'synopsis', item -> 'synopsis',
           'aspects', item -> 'aspects',
           'tags', item -> 'tags',
           'descriptionPageSlug', item -> 'descriptionPageSlug',
           'pageSlug', item -> 'pageSlug',
           'cmapSlug', item -> 'cmapSlug',
           'imageSource', item -> 'imageSource'
         )) AS document,
         cm.updated_at,
         cm.updated_by,
         cm.id AS map_id
  FROM concept_maps cm
  CROSS JOIN LATERAL jsonb_array_elements(
    CASE WHEN jsonb_typeof(cm.document -> 'items') = 'array'
         THEN cm.document -> 'items' ELSE '[]'::jsonb END
  ) AS item
  WHERE cm.archived = FALSE
    AND nullif(item ->> 'conceptId', '') IS NOT NULL
    AND coalesce(item ->> 'kind', 'concept') <> 'phrase'
), ranked AS (
  SELECT candidates.*,
         row_number() OVER (
           PARTITION BY id
           ORDER BY updated_at DESC, map_id DESC
         ) AS position
  FROM candidates
)
INSERT INTO concept_definitions(id, document, updated_at, updated_by)
SELECT id, jsonb_set(document, '{id}', to_jsonb(id)), updated_at, updated_by
FROM ranked
WHERE position = 1
SQL
   )
  (query-exec db "DROP TABLE IF EXISTS concept_aliases")
  (record-schema-version! db 18))

(define (migrate-18->19! db)
  ;; Concept content must have exactly one source. Remove the old denormalized
  ;; copies from placements; linking phrases retain their own text because
  ;; they are relations rather than concepts.
  (query-exec
   db
   #<<SQL
UPDATE concept_maps cm
SET document = jsonb_set(
  cm.document,
  '{items}',
  (SELECT coalesce(jsonb_agg(
     CASE WHEN coalesce(item ->> 'kind', 'concept') = 'phrase'
          THEN item
          ELSE item
               - 'label'
               - 'synopsis'
               - 'aspects'
               - 'tags'
               - 'descriptionPageSlug'
               - 'pageSlug'
               - 'cmapSlug'
               - 'imageSource'
          END
     ORDER BY ordinal
   ), '[]'::jsonb)
   FROM jsonb_array_elements(
     CASE WHEN jsonb_typeof(cm.document -> 'items') = 'array'
          THEN cm.document -> 'items' ELSE '[]'::jsonb END
   ) WITH ORDINALITY AS entry(item, ordinal)),
  TRUE
)
WHERE jsonb_typeof(cm.document) = 'object'
SQL
   )
  (record-schema-version! db 19))

(define (migrate-19->20! db)
  ;; Current CMaps store concept references only. Full concept content lives in
  ;; concept_definitions and is hydrated into API responses on read. Historical
  ;; snapshots remain immutable.
  (query-exec
   db
   #<<SQL
UPDATE concept_maps cm
SET document = cm.document || jsonb_build_object(
  'items',
    (SELECT coalesce(jsonb_agg(
       CASE WHEN coalesce(item ->> 'kind', 'concept') = 'phrase'
            THEN item
            ELSE item
                 - 'label'
                 - 'synopsis'
                 - 'aspects'
                 - 'tags'
                 - 'descriptionPageSlug'
                 - 'pageSlug'
                 - 'cmapSlug'
                 - 'imageSource'
            END
       ORDER BY ordinal
     ), '[]'::jsonb)
     FROM jsonb_array_elements(
       CASE WHEN jsonb_typeof(cm.document -> 'items') = 'array'
            THEN cm.document -> 'items' ELSE '[]'::jsonb END
     ) WITH ORDINALITY AS entry(item, ordinal)),
  'concepts',
    (SELECT coalesce(jsonb_agg(jsonb_build_object('id', concept_id)
                               ORDER BY ordinal), '[]'::jsonb)
     FROM (
       SELECT DISTINCT ON (concept_id) concept_id, ordinal
       FROM (
         SELECT concept ->> 'id' AS concept_id, ordinal
         FROM jsonb_array_elements(
           CASE WHEN jsonb_typeof(cm.document -> 'concepts') = 'array'
                THEN cm.document -> 'concepts' ELSE '[]'::jsonb END
         ) WITH ORDINALITY AS concept_entry(concept, ordinal)
         UNION ALL
         SELECT item ->> 'conceptId' AS concept_id, 1000000000 + ordinal
         FROM jsonb_array_elements(
           CASE WHEN jsonb_typeof(cm.document -> 'items') = 'array'
                THEN cm.document -> 'items' ELSE '[]'::jsonb END
         ) WITH ORDINALITY AS item_entry(item, ordinal)
         WHERE coalesce(item ->> 'kind', 'concept') <> 'phrase'
       ) all_references
       WHERE nullif(concept_id, '') IS NOT NULL
       ORDER BY concept_id, ordinal
     ) concept_references)
)
WHERE jsonb_typeof(cm.document) = 'object'
SQL
   )
  (record-schema-version! db 20))

(define (migrate-20->21! db)
  ;; Current concept identity is a plain globally unique UUID. Preserve the
  ;; UUID portion of newer concept-<uuid> identifiers and allocate a UUID only
  ;; for genuinely old/local ids. CMap slugs and historical snapshots are not
  ;; identity aliases and deliberately remain unchanged.
  (query-exec
   db
   #<<SQL
CREATE TEMP TABLE concept_uuid_rekey (
  old_id TEXT PRIMARY KEY,
  new_id TEXT NOT NULL
) ON COMMIT DROP
SQL
   )
  (let* ((old-ids
          (query-list
           db
           #<<SQL
WITH all_current_ids AS (
  SELECT id AS old_id FROM concept_definitions
  UNION
  SELECT concept ->> 'id' AS old_id
  FROM concept_maps cm
  CROSS JOIN LATERAL jsonb_array_elements(
    CASE WHEN jsonb_typeof(cm.document -> 'concepts') = 'array'
         THEN cm.document -> 'concepts' ELSE '[]'::jsonb END
  ) AS concept
  UNION
  SELECT item ->> 'conceptId' AS old_id
  FROM concept_maps cm
  CROSS JOIN LATERAL jsonb_array_elements(
    CASE WHEN jsonb_typeof(cm.document -> 'items') = 'array'
         THEN cm.document -> 'items' ELSE '[]'::jsonb END
  ) AS item
  WHERE coalesce(item ->> 'kind', 'concept') <> 'phrase'
)
SELECT old_id
FROM all_current_ids
WHERE nullif(old_id, '') IS NOT NULL
ORDER BY old_id
SQL
           ))
         ;; Reserve every already canonical UUID before generating replacements,
         ;; so a random id can never collide with a UUID encountered later.
         (used-ids (make-hash)))
    (for ([old-id (in-list old-ids)])
      (let ((normalized (normalize-concept-id old-id)))
        (when normalized (hash-set! used-ids normalized #t))))
    (letrec ((fresh-unused-id
              (λ ()
                (let loop ()
                  (let ((candidate (new-concept-id)))
                    (if (hash-has-key? used-ids candidate)
                        (loop)
                        (begin
                          (hash-set! used-ids candidate #t)
                          candidate)))))))
      (for ([old-id (in-list old-ids)])
        (query-exec
         db
         "INSERT INTO concept_uuid_rekey(old_id, new_id) VALUES ($1, $2)"
         old-id
         (or (normalize-concept-id old-id) (fresh-unused-id)))))
  (query-exec
   db
   #<<SQL
UPDATE concept_maps cm
SET document = cm.document || jsonb_build_object(
  'items',
    (SELECT coalesce(jsonb_agg(
       CASE WHEN rekey.new_id IS NULL
            THEN item
            ELSE jsonb_set(item, '{conceptId}', to_jsonb(rekey.new_id))
            END
       ORDER BY ordinal
     ), '[]'::jsonb)
     FROM jsonb_array_elements(
       CASE WHEN jsonb_typeof(cm.document -> 'items') = 'array'
            THEN cm.document -> 'items' ELSE '[]'::jsonb END
     ) WITH ORDINALITY AS entry(item, ordinal)
     LEFT JOIN concept_uuid_rekey rekey ON rekey.old_id = item ->> 'conceptId'),
  'concepts',
    (SELECT coalesce(jsonb_agg(jsonb_build_object('id', new_id)
                               ORDER BY ordinal), '[]'::jsonb)
     FROM (
       SELECT DISTINCT ON (rekey.new_id) rekey.new_id, ordinal
       FROM jsonb_array_elements(
         CASE WHEN jsonb_typeof(cm.document -> 'concepts') = 'array'
              THEN cm.document -> 'concepts' ELSE '[]'::jsonb END
       ) WITH ORDINALITY AS entry(concept, ordinal)
       JOIN concept_uuid_rekey rekey ON rekey.old_id = concept ->> 'id'
       ORDER BY rekey.new_id, ordinal
     ) rewritten_concepts)
)
WHERE jsonb_typeof(cm.document) = 'object'
SQL
   )
  (query-exec
   db
   #<<SQL
CREATE TEMP TABLE rekeyed_concept_definitions ON COMMIT DROP AS
SELECT DISTINCT ON (rekey.new_id)
       rekey.new_id AS id,
       jsonb_set(definition.document, '{id}', to_jsonb(rekey.new_id)) AS document,
       definition.updated_at,
       definition.updated_by
FROM concept_definitions definition
JOIN concept_uuid_rekey rekey ON rekey.old_id = definition.id
ORDER BY rekey.new_id, definition.updated_at DESC, definition.id
SQL
   )
  (query-exec db "DELETE FROM concept_definitions")
  (query-exec
   db
   #<<SQL
INSERT INTO concept_definitions(id, document, updated_at, updated_by)
SELECT id, document, updated_at, updated_by
FROM rekeyed_concept_definitions
SQL
   )
  (query-exec
   db
   #<<SQL
ALTER TABLE concept_definitions
ADD CONSTRAINT concept_definitions_uuid_id_check
CHECK (id ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$')
SQL
   )
    (record-schema-version! db 21)))

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
; goal   : Bring a racket-wiki PostgreSQL database to the current schema.
; pre    : db is a writable PostgreSQL connection and config identifies the
;          data directory used by older racket-wiki versions.
; post   : Every required migration has been applied in order and recorded.
; result : The resulting schema version.
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(define (migrate-database! db config)
  (call-with-transaction
   db
   (λ ()
     (recognize-or-install-schema-1! db)
     (let loop ((version (database-schema-version db)))
       (cond
         ((< version 1)
          (error 'migrate-database! "unable to determine the existing wiki database schema"))
         ((= version 1) (migrate-1->2! db config) (loop (database-schema-version db)))
         ((= version 2) (migrate-2->3! db) (loop (database-schema-version db)))
         ((= version 3) (migrate-3->4! db) (loop (database-schema-version db)))
         ((= version 4) (migrate-4->5! db) (loop (database-schema-version db)))
         ((= version 5) (migrate-5->6! db) (loop (database-schema-version db)))
         ((= version 6) (migrate-6->7! db) (loop (database-schema-version db)))
         ((= version 7) (migrate-7->8! db) (loop (database-schema-version db)))
         ((= version 8) (migrate-8->9! db) (loop (database-schema-version db)))
         ((= version 9) (migrate-9->10! db) (loop (database-schema-version db)))
         ((= version 10) (migrate-10->11! db) (loop (database-schema-version db)))
         ((= version 11) (migrate-11->12! db) (loop (database-schema-version db)))
         ((= version 12) (migrate-12->13! db) (loop (database-schema-version db)))
         ((= version 13) (migrate-13->14! db) (loop (database-schema-version db)))
         ((= version 14) (migrate-14->15! db) (loop (database-schema-version db)))
         ((= version 15) (migrate-15->16! db) (loop (database-schema-version db)))
         ((= version 16) (migrate-16->17! db) (loop (database-schema-version db)))
         ((= version 17) (migrate-17->18! db) (loop (database-schema-version db)))
         ((= version 18) (migrate-18->19! db) (loop (database-schema-version db)))
         ((= version 19) (migrate-19->20! db) (loop (database-schema-version db)))
         ((= version 20) (migrate-20->21! db) (loop (database-schema-version db)))
         ((> version current-schema-version)
          (error 'migrate-database!
                 "database schema ~a is newer than this racket-wiki supports (~a)"
                 version
                 current-schema-version))
         (else version))))))