Retain certificate PDF versions and delivery history during regeneration

docs/modules/program-certificates.md

@@ -121,5 +121,20 @@ Pillow и PDF используют одни и те же локальные TTF,
 - Старые GET download/archive и retry остаются совместимыми. Повторное скачивание
   текущего PDF не пересоздает файл. Ошибка генерации откатывает изменения транзакции.
 
-Миграция БД не требуется. Сначала API, затем frontend. Отправление проверяется
-отдельно от готовности PDF: `ready` означает доступность на платформе, не письмо.
+Отправление проверяется отдельно от готовности PDF: `ready` означает доступность
+на платформе, не письмо.
+
+## Замена PDF без разрыва ссылки
+
+`ProgramCertificate.rendered_batch` указывает на снимок фактически сохраненного
+PDF; `batch` — на текущее задание. Массовое обновление сохраняет прежний файл и
+его снимок, пока worker не создаст новую версию. PDF и `rendered_batch` меняются
+одним UPDATE. При ошибке старая ссылка продолжает возвращать прежний PDF, а
+журнал показывает `pdf_error` и расхождение с текущим шаблоном. Повтор использует
+нужный снимок, а успешно созданный PDF не генерируется заново при почтовой ошибке.
+
+Нужна миграция `0031_programcertificate_rendered_batch`: она добавляет nullable FK
+и запоминает существующий batch у строк с PDF. Байты, токены, статусы, даты и
+письма не меняются. Старые записи без provenance имеют совместимый fallback;
+перед изменением задания их исходный снимок запоминается. Сначала миграция/API
+и worker, затем frontend с распознаванием ошибки обновления.

partner_programs/certificate_models.py

@@ -53,6 +53,13 @@ class ProgramCertificate(models.Model):
     program = models.ForeignKey("PartnerProgram", on_delete=models.CASCADE)
     user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
     batch = models.ForeignKey(ProgramCertificateBatch, on_delete=models.RESTRICT)
+    rendered_batch = models.ForeignKey(
+        ProgramCertificateBatch,
+        null=True,
+        blank=True,
+        on_delete=models.RESTRICT,
+        related_name="rendered_certificates",
+    )
     name = models.CharField(max_length=500)
     email = models.EmailField()
     pdf = models.BinaryField(null=True)

partner_programs/certificate_views.py

@@ -2,7 +2,7 @@
 from datetime import timedelta
 
 from django.db import transaction
-from django.db.models import Count, Q
+from django.db.models import Count, F, Q
 from django.http import HttpResponse
 from django.shortcuts import get_object_or_404
 from django.utils import timezone
@@ -203,8 +203,11 @@ def journal_rows(program, params):
     query.is_valid(raise_exception=True)
     data = query.validated_data
     rows = ProgramCertificate.objects.filter(program=program).defer("pdf")
+    delivered = Q(sent_at__isnull=False) | Q(status="sent")
     if data["status"] == "unsent":
-        rows = rows.exclude(status="sent")
+        rows = rows.exclude(delivered)
+    elif data["status"] == "sent":
+        rows = rows.filter(delivered)
     elif data["status"] != "all":
         rows = rows.filter(status=data["status"])
     if data["search"]:
@@ -228,10 +231,14 @@ def get(self, request, pk):
         summary = ProgramCertificate.objects.filter(program=program).aggregate(
             total=Count("pk"),
             ready=Count("pk", filter=Q(pdf__isnull=False)),
-            sent=Count("pk", filter=Q(status="sent")),
+            sent=Count("pk", filter=Q(sent_at__isnull=False) | Q(status="sent")),
             failed=Count("pk", filter=Q(status="failed")),
-            unsent=Count("pk", filter=~Q(status="sent")),
-            outdated=Count("pk", filter=Q(batch_id__in=stale_batches)),
+            unsent=Count("pk", filter=Q(sent_at__isnull=True) & ~Q(status="sent")),
+            outdated=Count(
+                "pk",
+                filter=Q(rendered_batch_id__in=stale_batches)
+                | Q(rendered_batch__isnull=True, batch_id__in=stale_batches),
+            ),
         )
         offset = query["offset"]
         end = offset + query["limit"]
@@ -244,9 +251,13 @@ def get(self, request, pk):
                 "pdf_ready": row.pdf_ready,
                 "error": row.error,
                 "sent_at": row.sent_at,
-                "outdated": row.batch_id in stale_batches,
+                "outdated": service.outdated_certificate(row, draft),
+                "pdf_error": row.status == "failed"
+                and service.pdf_needs_render(row, row.pdf_ready),
             }
-            for row in rows.annotate(pdf_ready=~Q(pdf__isnull=True))[offset:end]
+            for row in rows.select_related("batch", "rendered_batch", "program").annotate(
+                pdf_ready=~Q(pdf__isnull=True)
+            )[offset:end]
         ]
         return Response({"count": rows.count(), "summary": summary, "results": result})
 
@@ -293,22 +304,26 @@ def post(self, request, pk):
         )
         if any(row.status in ("pending", "generating", "sending") for row in rows):
             raise ValidationError("Дождитесь завершения текущей обработки сертификатов.")
-        selected = [row for row in rows if action == "refresh" or row.status != "sent"]
+        selected = [
+            row
+            for row in rows
+            if action == "refresh"
+            or (action == "retry" and row.status == "failed")
+            or (row.status != "sent" and row.sent_at is None)
+        ]
         if not selected:
             return Response({"count": 0, "skipped": len(rows)}, status=202)
         draft = service.validate_draft(program, config.draft)
-        regenerate = [
-            row.pk
-            for row in selected
-            if action == "refresh" or service.outdated_certificate(row, draft)
-        ]
         for send_email in (False, True):
             group = [
                 row.pk
                 for row in selected
                 if (
-                    action == "send"
-                    or (action == "retry" and row.batch.snapshot["send_email"])
+                    row.sent_at is None
+                    and (
+                        action == "send"
+                        or (action == "retry" and row.batch.snapshot["send_email"])
+                    )
                 )
                 == send_email
             ]
@@ -321,15 +336,17 @@ def post(self, request, pk):
                 "send" if send_email else "refresh",
             )
             targets = ProgramCertificate.objects.filter(pk__in=group)
+            # Keep the current file's snapshot before moving work to a new batch.
+            targets.filter(pdf__isnull=False, rendered_batch__isnull=True).update(
+                rendered_batch_id=F("batch_id")
+            )
             targets.update(
                 batch=batch,
                 status="pending",
                 error="",
                 queued_at=None,
                 claimed_at=None,
-                sent_at=None,
             )
-            targets.filter(pk__in=regenerate).update(pdf=None)
         return Response(
             {"count": len(selected), "skipped": len(rows) - len(selected)}, status=202
         )
@@ -367,11 +384,13 @@ def current_certificate_pdf(program, config, row, public_base_url):
             timezone.localtime(row.created_at),
         )
         row.batch = service.manual_batch(program, draft, public_base_url, "refresh")
+        row.rendered_batch = row.batch
         row.pdf = content
-        row.status = "ready"
+        row.status = "sent" if row.sent_at or row.status == "sent" else "ready"
         row.error = ""
-        row.sent_at = None
-        row.save(update_fields=["batch", "pdf", "status", "error", "sent_at"])
+        row.save(
+            update_fields=["batch", "rendered_batch", "pdf", "status", "error", "sent_at"]
+        )
     return bytes(row.pdf)
 
 
@@ -383,7 +402,7 @@ def post(self, request, pk, certificate_id):
         revision_check(config, request.data)
         row = get_object_or_404(
             ProgramCertificate.objects.select_for_update(of=("self",)).select_related(
-                "batch", "program"
+                "batch", "rendered_batch", "program"
             ),
             program=program,
             pk=certificate_id,
@@ -413,7 +432,8 @@ def post(self, request, pk):
         rows = (
             selected_rows(program, request.data)
             .select_for_update(of=("self",))
-            .select_related("batch", "program")
+            .select_related("batch", "rendered_batch", "program")
+            .defer(None)
         )
         if rows.filter(status__in=["pending", "generating", "sending"]).exists():
             raise ValidationError(

partner_programs/migrations/0031_programcertificate_rendered_batch.py

@@ -0,0 +1,28 @@
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+def remember_rendered_batch(apps, schema_editor):
+    certificate = apps.get_model("partner_programs", "ProgramCertificate")
+    certificate.objects.using(schema_editor.connection.alias).filter(
+        pdf__isnull=False, rendered_batch__isnull=True
+    ).update(rendered_batch_id=models.F("batch_id"))
+
+
+class Migration(migrations.Migration):
+    dependencies = [("partner_programs", "0030_program_solution_foundation")]
+
+    operations = [
+        migrations.AddField(
+            model_name="programcertificate",
+            name="rendered_batch",
+            field=models.ForeignKey(
+                null=True,
+                blank=True,
+                on_delete=django.db.models.deletion.RESTRICT,
+                related_name="rendered_certificates",
+                to="partner_programs.programcertificatebatch",
+            ),
+        ),
+        migrations.RunPython(remember_rendered_batch, migrations.RunPython.noop),
+    ]

partner_programs/services/certificates.py

@@ -209,11 +209,20 @@ def template_values(draft, program_name):
 
 
 def outdated_certificate(row, draft):
-    return template_values(row.batch.snapshot, row.program.name) != template_values(
+    snapshot = (
+        row.rendered_batch.snapshot if row.rendered_batch_id else row.batch.snapshot
+    )
+    return template_values(snapshot, row.program.name) != template_values(
         draft, row.program.name
     )
 
 
+def pdf_needs_render(row, pdf_ready=None):
+    if pdf_ready is None:
+        pdf_ready = row.pdf is not None
+    return not pdf_ready or outdated_certificate(row, row.batch.snapshot)
+
+
 def manual_batch(program, draft, public_base_url, action):
     snapshot = {
         **draft,
@@ -622,8 +631,8 @@ def dispatch():
 def process(certificate_id):
     with transaction.atomic():
         row = (
-            ProgramCertificate.objects.select_for_update()
-            .select_related("batch", "program")
+            ProgramCertificate.objects.select_for_update(of=("self",))
+            .select_related("batch", "rendered_batch", "program")
             .get(pk=certificate_id)
         )
         if row.status != "pending" or not row.batch.active:
@@ -633,7 +642,7 @@ def process(certificate_id):
         row.attempts += 1
         row.save(update_fields=["status", "claimed_at", "attempts"])
     try:
-        if not row.pdf:
+        if pdf_needs_render(row):
             batch = row.batch
             row.pdf = render_pdf(
                 batch.snapshot,
@@ -643,7 +652,9 @@ def process(certificate_id):
                 str(row.pk).upper(),
                 timezone.localtime(row.created_at),
             )
-            ProgramCertificate.objects.filter(pk=row.pk).update(pdf=row.pdf)
+            ProgramCertificate.objects.filter(pk=row.pk).update(
+                pdf=row.pdf, rendered_batch_id=row.batch_id
+            )
         with transaction.atomic():
             # Lock the batch in the same order as pause/retry. A paused queue cannot send.
             batch = ProgramCertificateBatch.objects.select_for_update().get(
@@ -655,13 +666,14 @@ def process(certificate_id):
                 current.queued_at = None
                 current.save(update_fields=["status", "queued_at"])
                 return
-            if not batch.snapshot["send_email"]:
-                current.status = "ready"
+            should_send = batch.snapshot["send_email"] and current.sent_at is None
+            if not should_send:
+                current.status = "sent" if current.sent_at else "ready"
             else:
                 # Persist before the provider call; a crash must not trigger automatic resend.
                 current.status = "sending"
             current.save(update_fields=["status"])
-        if batch.snapshot["send_email"]:
+        if should_send:
             send_certificate(row)
             ProgramCertificate.objects.filter(pk=row.pk).update(
                 status="sent", sent_at=timezone.now(), error=""