feat(migrations): add almost zero-downtime migration capability and linting

- CI linting of new migrations on pull requests
- Declarative dual-write trigger decorator (mirror_field)
- Reusable batched-backfill command (idempotent, resumable, throttled)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LfZvkigk8hdsKdEif3hzBi

Author: rtibbles

.github/workflows/pythontest.yml

@@ -62,6 +62,8 @@ jobs:
           - 6379:6379
     steps:
     - uses: actions/checkout@v6
+      with:
+        fetch-depth: 0
     - name: Set up minio
       run: |
         docker run -d -p 9000:9000 --name minio \
@@ -79,6 +81,17 @@ jobs:
       run: |
         # Use uv to install dependencies directly from requirements files
         uv pip sync requirements.txt requirements-dev.txt
+    - name: Lint new migrations for unsafe operations
+      if: github.event_name == 'pull_request'
+      env:
+        BASE_REF: ${{ github.base_ref }}
+        DJANGO_SETTINGS_MODULE: contentcuration.not_production_settings
+      run: |
+        set -euo pipefail
+        git fetch --no-tags origin "$BASE_REF"
+        base="$(git merge-base "origin/$BASE_REF" HEAD)"
+        test -n "$base"
+        python contentcuration/manage.py lintmigrations --git-commit-id "$base" --no-cache --warnings-as-errors
     - name: Test pytest
       run: |
         sh -c './contentcuration/manage.py makemigrations --check'

contentcuration/contentcuration/db/dual_write.py

@@ -0,0 +1,31 @@
+import hashlib
+
+import pgtrigger
+
+
+def mirror_field(source, target):
+    """Mirror Django field `source` into `target` via a BEFORE INSERT/UPDATE
+    trigger (expand/contract dual-write)."""
+
+    def decorator(model):
+        source_col = model._meta.get_field(source).column
+        target_col = model._meta.get_field(target).column
+        name = "mirror_{}_to_{}".format(source_col, target_col)
+        if len(name) > 43:  # stay safely under pgtrigger's trigger-name limit
+            digest = hashlib.sha1(
+                "{}_{}".format(source_col, target_col).encode()
+            ).hexdigest()[:8]
+            name = "mirror_{}".format(digest)
+        # Change-guard (IS DISTINCT FROM): keeps a read cutover from clobbering
+        # writes to the repointed column with the stale source value.
+        trigger = pgtrigger.Trigger(
+            name=name,
+            when=pgtrigger.Before,
+            operation=pgtrigger.Insert | pgtrigger.Update,
+            func="IF NEW.{s} IS DISTINCT FROM OLD.{s} THEN NEW.{t} = NEW.{s}; END IF; RETURN NEW;".format(
+                s=source_col, t=target_col
+            ),
+        )
+        return pgtrigger.register(trigger)(model)
+
+    return decorator

contentcuration/contentcuration/management/commands/backfill_column.py

@@ -0,0 +1,100 @@
+from django.apps import apps
+from django.core.exceptions import FieldDoesNotExist
+from django.core.management.base import BaseCommand
+from django.core.management.base import CommandError
+from django.db import transaction
+from django.db.models import F
+
+
+class Command(BaseCommand):
+    help = (
+        "Idempotent, resumable online backfill of one column into another, in batches."
+    )
+
+    def add_arguments(self, parser):
+        parser.add_argument("--model", required=True, help="app_label.ModelName")
+        parser.add_argument("--source-field", required=True)
+        parser.add_argument("--target-field", required=True)
+        parser.add_argument("--batch-size", type=int, default=10000)
+        parser.add_argument("--start-id", default=None, help="resume from this pk")
+        parser.add_argument(
+            "--progress-check",
+            action="store_true",
+            help="report unbackfilled rows, exit nonzero if any",
+        )
+
+    def _resolve_model_fields(self, model_label, source, target):
+        try:
+            model = apps.get_model(model_label)
+        except (LookupError, ValueError) as e:
+            raise CommandError("Bad --model {!r}: {}".format(model_label, e))
+        try:
+            model._meta.get_field(source)
+            model._meta.get_field(target)
+        except FieldDoesNotExist as e:
+            raise CommandError(str(e))
+        return model
+
+    def _batch_end_pk(self, queryset, pk_name, start_pk, batch_size):
+        """Last pk of the batch of `batch_size` rows starting at `start_pk`.
+
+        Returns None when fewer than `batch_size` rows remain at/after
+        `start_pk` — the final, short batch. Keyset paging by pk, so it works
+        for any pk type (int or UUID).
+        """
+        return (
+            queryset.filter(pk__gte=start_pk)
+            .order_by(pk_name)
+            .values_list("pk", flat=True)[batch_size - 1 : batch_size]
+            .first()
+        )
+
+    def handle(self, *args, **options):
+        if options["batch_size"] < 1:
+            raise CommandError("--batch-size must be >= 1")
+        source = options["source_field"]
+        target = options["target_field"]
+        model = self._resolve_model_fields(options["model"], source, target)
+
+        pk_name = model._meta.pk.name
+        batch_size = options["batch_size"]
+        only_unfilled = {target + "__isnull": True, source + "__isnull": False}
+        unfilled = model.objects.filter(**only_unfilled)
+        unfilled_pks = unfilled.order_by(pk_name).values_list("pk", flat=True)
+
+        if options["progress_check"]:
+            # exists(), not count() — the target table can have millions of rows.
+            if unfilled.exists():
+                raise CommandError("backfill incomplete: rows still pending")
+            self.stdout.write("Backfill complete: no rows pending.")
+            return
+
+        # Start at the first unfilled pk (>= --start-id if given); re-runs and
+        # resumes skip straight past an already-filled prefix.
+        batch_start = unfilled_pks
+        if options["start_id"] is not None:
+            batch_start = batch_start.filter(pk__gte=options["start_id"])
+        batch_start = batch_start.first()
+
+        total = 0
+        while batch_start is not None:
+            batch_end = self._batch_end_pk(
+                model.objects, pk_name, batch_start, batch_size
+            )
+            if batch_end is None:
+                window = {"pk__gte": batch_start}
+            else:
+                window = {"pk__gte": batch_start, "pk__lte": batch_end}
+            with transaction.atomic():
+                total += model.objects.filter(**window, **only_unfilled).update(
+                    **{target: F(source)}
+                )
+            self.stdout.write(
+                "backfilled through pk={} (updated {} so far)".format(
+                    batch_start if batch_end is None else batch_end, total
+                )
+            )
+            if batch_end is None:
+                break
+            batch_start = unfilled_pks.filter(pk__gt=batch_end).first()
+        self.stdout.write("Done. {} rows updated.".format(total))

contentcuration/contentcuration/not_production_settings.py

@@ -20,5 +20,14 @@
 
 AWS_AUTO_CREATE_BUCKET = True
 
+INSTALLED_APPS += ("django_migration_linter",)  # noqa F405
+
+MIGRATION_LINTER_OPTIONS = {
+    "exclude_apps": [
+        "kolibri_content"
+    ],  # SQLite content-export app; not on the safe-DDL Postgres backend
+    "sql_analyser": "postgresql",
+}
+
 # Use local instance for curriculum automation for development
 CURRICULUM_AUTOMATION_API_URL = "http://localhost:8000"

contentcuration/contentcuration/settings.py

@@ -92,6 +92,7 @@
     "django_celery_results",
     "kolibri_public",
     "automation",
+    "pgtrigger",
 )
 
 SESSION_ENGINE = "django.contrib.sessions.backends.cached_db"

contentcuration/contentcuration/tests/test_backfill_column.py

@@ -0,0 +1,191 @@
+import uuid
+from io import StringIO
+from unittest.mock import patch
+
+from django.core.management import call_command
+from django.core.management import CommandError
+from django.db import connection
+from django.db import models
+from django.db.models import F
+from django.test import SimpleTestCase
+from django.test import TransactionTestCase
+from django.test.utils import isolate_apps
+
+
+def _make_probe_class():
+    class Probe(models.Model):
+        source = models.IntegerField(null=True)
+        shadow = models.IntegerField(null=True)
+
+        class Meta:
+            app_label = "contentcuration"
+
+    return Probe
+
+
+def _make_uuid_probe_class():
+    class UUIDProbe(models.Model):
+        id = models.UUIDField(primary_key=True, default=uuid.uuid4)
+        source = models.IntegerField(null=True)
+        shadow = models.IntegerField(null=True)
+
+        class Meta:
+            app_label = "contentcuration"
+
+    return UUIDProbe
+
+
+@isolate_apps("contentcuration")
+class BackfillColumnTestCase(TransactionTestCase):
+    def _create_model(self, model):
+        with connection.schema_editor(atomic=False) as editor:
+            editor.create_model(model)
+        self.addCleanup(self._delete_model, model)
+
+    def _delete_model(self, model):
+        with connection.schema_editor(atomic=False) as editor:
+            editor.delete_model(model)
+
+    def _run_backfill(self, model, **kwargs):
+        out = StringIO()
+        with patch(
+            "contentcuration.management.commands.backfill_column.apps",
+            model._meta.apps,
+        ):
+            call_command(
+                "backfill_column",
+                stdout=out,
+                model="contentcuration.{}".format(model.__name__),
+                source_field="source",
+                target_field="shadow",
+                **kwargs,
+            )
+        return out.getvalue()
+
+    def _assert_all_synced(self, model):
+        self.assertEqual(
+            model.objects.count(), model.objects.filter(shadow=F("source")).count()
+        )
+
+    def _assert_synced_from(self, model, resume_pk):
+        below = model.objects.filter(pk__lt=resume_pk)
+        at_or_above = model.objects.filter(pk__gte=resume_pk)
+        self.assertEqual(below.count(), below.filter(shadow__isnull=True).count())
+        self.assertEqual(
+            at_or_above.count(), at_or_above.filter(shadow=F("source")).count()
+        )
+
+    def test_backfills_all_rows(self):
+        Probe = _make_probe_class()
+        self._create_model(Probe)
+        for i in range(1, 6):
+            Probe.objects.create(source=i * 10, shadow=None)
+
+        # batch_size=2 over 5 rows exercises the multi-batch loop.
+        self._run_backfill(Probe, batch_size=2)
+
+        self._assert_all_synced(Probe)
+
+    def test_idempotent(self):
+        Probe = _make_probe_class()
+        self._create_model(Probe)
+        for i in range(1, 4):
+            Probe.objects.create(source=i * 10, shadow=None)
+
+        self._run_backfill(Probe, batch_size=10)
+        output = self._run_backfill(Probe, batch_size=10)
+
+        self.assertIn("Done. 0 rows updated.", output)
+        self._assert_all_synced(Probe)
+
+    def test_resumable(self):
+        Probe = _make_probe_class()
+        self._create_model(Probe)
+        objs = sorted(
+            [Probe.objects.create(source=i * 10, shadow=None) for i in range(1, 6)],
+            key=lambda o: o.pk,
+        )
+        resume_pk = objs[2].pk
+
+        self._run_backfill(Probe, batch_size=10, start_id=resume_pk)
+
+        self._assert_synced_from(Probe, resume_pk)
+
+    def test_null_source_safe(self):
+        Probe = _make_probe_class()
+        self._create_model(Probe)
+        Probe.objects.create(source=None, shadow=None)
+        Probe.objects.create(source=42, shadow=None)
+
+        output = self._run_backfill(Probe, batch_size=10)
+
+        self.assertIn("Done.", output)
+        self.assertIsNone(Probe.objects.get(source__isnull=True).shadow)
+        self.assertEqual(Probe.objects.get(source=42).shadow, 42)
+
+    def test_backfills_uuid_pk_across_batches(self):
+        """Regression: paging must not assume an integer pk (File has a UUID pk)."""
+        UUIDProbe = _make_uuid_probe_class()
+        self._create_model(UUIDProbe)
+        for i in range(1, 6):
+            UUIDProbe.objects.create(source=i * 10, shadow=None)
+
+        # batch_size=2 forces the lower-bound advance where integer arithmetic on a
+        # UUID pk would blow up.
+        self._run_backfill(UUIDProbe, batch_size=2)
+
+        self._assert_all_synced(UUIDProbe)
+
+    def test_resumable_uuid_pk(self):
+        """--start-id must accept a UUID and resume from it."""
+        UUIDProbe = _make_uuid_probe_class()
+        self._create_model(UUIDProbe)
+        objs = sorted(
+            [UUIDProbe.objects.create(source=i * 10, shadow=None) for i in range(1, 6)],
+            key=lambda o: o.pk,
+        )
+        resume_pk = objs[2].pk
+
+        self._run_backfill(UUIDProbe, batch_size=10, start_id=str(resume_pk))
+
+        self._assert_synced_from(UUIDProbe, resume_pk)
+
+    def test_progress_check_passes_when_complete(self):
+        Probe = _make_probe_class()
+        self._create_model(Probe)
+        Probe.objects.create(source=1, shadow=1)
+        Probe.objects.create(source=None, shadow=None)  # null source doesn't count
+
+        output = self._run_backfill(Probe, progress_check=True)
+
+        self.assertIn("no rows pending", output)
+
+    def test_progress_check_fails_and_writes_nothing_when_incomplete(self):
+        Probe = _make_probe_class()
+        self._create_model(Probe)
+        Probe.objects.create(source=7, shadow=None)
+
+        with self.assertRaisesRegex(
+            CommandError, "backfill incomplete: rows still pending"
+        ):
+            self._run_backfill(Probe, progress_check=True)
+
+        # --progress-check must not write
+        self.assertIsNone(Probe.objects.get(source=7).shadow)
+
+
+class BackfillColumnArgValidationTestCase(SimpleTestCase):
+    def _call(self, **kwargs):
+        call_command(
+            "backfill_column",
+            model="contentcuration.Channel",
+            source_field="name",
+            target_field="name",
+            **kwargs,
+        )
+
+    def test_non_positive_batch_size_raises(self):
+        for bad in (0, -1):
+            with self.subTest(batch_size=bad):
+                with self.assertRaisesRegex(CommandError, "--batch-size must be >= 1"):
+                    self._call(batch_size=bad)