feat: add digest-gated internal main deployment

src/apps/core/management/commands/sanitize_cloned_environment.py

@@ -0,0 +1,97 @@
+"""Deactivate environment-bound state after restoring a database clone."""
+
+import json
+
+from apps.core.management.commands.base import BaseAppCommand
+from apps.core.models import BackgroundJob, JobStatus
+from apps.exchange.models import ExchangeConnection
+from apps.parsers.models import (
+    CheckoCollectionAttempt,
+    FinancialReport,
+    ParserLoadLog,
+)
+from django.db.models import Q
+from django.utils import timezone
+from django_celery_beat.models import PeriodicTask, PeriodicTasks
+
+CLONE_TERMINATION_MESSAGE = "Closed after database clone"
+
+
+class Command(BaseAppCommand):
+    """Make a restored database safe to start in another environment."""
+
+    help = "Disable schedules and close environment-bound state after a database clone"
+    use_transaction = True
+
+    def execute_command(self, *args, **options) -> str:
+        now = timezone.now()
+
+        periodic_tasks = PeriodicTask.objects.filter(enabled=True).update(
+            enabled=False,
+            date_changed=now,
+        )
+        if periodic_tasks:
+            PeriodicTasks.update_changed()
+
+        exchange_connections = ExchangeConnection.objects.filter(
+            Q(is_active=True)
+            | ~Q(password="")
+            | Q(last_checked_at__isnull=False)
+            | ~Q(last_error="")
+        ).update(
+            is_active=False,
+            password="",
+            last_checked_at=None,
+            last_error="",
+            updated_at=now,
+        )
+
+        background_jobs = BackgroundJob.objects.filter(
+            status__in=(JobStatus.PENDING, JobStatus.STARTED, JobStatus.RETRY)
+        ).update(
+            status=JobStatus.REVOKED,
+            completed_at=now,
+            updated_at=now,
+        )
+
+        parser_load_logs = ParserLoadLog.objects.filter(
+            status__in=(
+                ParserLoadLog.Status.PENDING,
+                ParserLoadLog.Status.IN_PROGRESS,
+            )
+        ).update(
+            status=ParserLoadLog.Status.FAILED,
+            error_message=CLONE_TERMINATION_MESSAGE,
+            updated_at=now,
+        )
+
+        checko_attempts = CheckoCollectionAttempt.objects.filter(
+            status=CheckoCollectionAttempt.Status.IN_PROGRESS
+        ).update(
+            status=CheckoCollectionAttempt.Status.FAILED,
+            error_message=CLONE_TERMINATION_MESSAGE,
+            updated_at=now,
+        )
+
+        financial_reports = FinancialReport.objects.filter(
+            status__in=(
+                FinancialReport.Status.PENDING,
+                FinancialReport.Status.PROCESSING,
+            )
+        ).update(
+            status=FinancialReport.Status.FAILED,
+            error_message=CLONE_TERMINATION_MESSAGE,
+            updated_at=now,
+        )
+
+        return json.dumps(
+            {
+                "background_jobs": background_jobs,
+                "checko_attempts": checko_attempts,
+                "exchange_connections": exchange_connections,
+                "financial_reports": financial_reports,
+                "parser_load_logs": parser_load_logs,
+                "periodic_tasks": periodic_tasks,
+            },
+            sort_keys=True,
+        )

src/apps/core/management/commands/validate_cloned_migrations.py

@@ -0,0 +1,40 @@
+"""Validate that a cloned database contains only migrations known to this image."""
+
+from django.core.management.base import BaseCommand, CommandError
+from django.db import DEFAULT_DB_ALIAS, connections
+from django.db.migrations.loader import MigrationLoader
+from django.db.migrations.recorder import MigrationRecorder
+
+
+class Command(BaseCommand):
+    """Reject a clone produced by code newer than the candidate image."""
+
+    help = "Fail if the database has applied migrations absent from this image"
+    requires_migrations_checks = False
+
+    def add_arguments(self, parser) -> None:
+        parser.add_argument(
+            "--database",
+            default=DEFAULT_DB_ALIAS,
+            choices=tuple(connections),
+            help="Database to validate",
+        )
+
+    def handle(self, *args, **options) -> str:
+        database = options["database"]
+        connection = connections[database]
+        disk_migrations = set(MigrationLoader(connection).disk_migrations)
+        applied_migrations = set(MigrationRecorder(connection).applied_migrations())
+        unknown_migrations = sorted(applied_migrations - disk_migrations)
+
+        if unknown_migrations:
+            formatted = ", ".join(
+                f"{app_label}.{migration_name}"
+                for app_label, migration_name in unknown_migrations
+            )
+            raise CommandError(
+                "Cloned database has applied migrations absent from this image: "
+                f"{formatted}"
+            )
+
+        return "Cloned database migrations are compatible with this image"