Stop storing completed uploads; announce them through a live feed

Every auto-matched, duplicate or manually resolved upload left a completed
TempUpload row on the board until it was dismissed by hand, so the rows
accumulated without bound and the bulk dismiss (capped at 1000 ids) failed
once there were more. The original app never stored these: they are
notifications, not records.

- complete_temp_upload now appends {filename, J-ID, resolution, post} to a
  bounded recent_completions feed on UploadRun and deletes the staged row
- staging duplicates never create a board record either; the create response
  carries the J-ID and preview so the SPA can show the card immediately
- status_payload returns the feed (newest first, signed thumbnails) for the
  live board; finalize_round counts deleted matches in processed
- resolve/link-bulk return synthetic completion payloads
- migration 0012 adds the field and purges the existing completed backlog
  (and any stray staged files) on deploy

backend/apps/library/upload_pipeline.py

@@ -29,7 +29,7 @@ from django.db.models import F, Q
 from django.utils import timezone
 
 from . import e621, services
-from .models import TempUpload, UploadRun
+from .models import MediaItem, TempUpload, UploadRun
 
 logger = logging.getLogger(__name__)
 
@@ -46,6 +46,9 @@ MAX_ATTEMPTS = 3
 # Round-level e621 retries before the run is paused.
 ROUND_ATTEMPTS = 3
 ROUND_RETRY_SECONDS = 20
+# Completions kept in the live feed. The board only shows what happened while
+# the page was open, so a bounded rolling window is plenty.
+COMPLETION_FEED_LIMIT = 200
 
 WORK_STATUSES = (TempUpload.STATUS_PENDING, TempUpload.STATUS_VISUAL_MATCH)
 VIDEO_RE = r"\.(mp4|webm)$"
@@ -116,7 +119,7 @@ def waiting_counts(user):
     }
 
 
-def status_payload(user):
+def status_payload(user, request=None):
     """Cheap state for the shell/upload page to poll."""
     run = UploadRun.objects.filter(user=user).first()
     outstanding = count_outstanding(user)
@@ -141,10 +144,67 @@ def status_payload(user):
         "error": run.error if run is not None else "",
         "outstanding": outstanding,
         "waiting": waiting_counts(user),
+        "recent_completions": completion_payload(run, user, request=request),
         "updated_at": run.updated_at.isoformat() if run is not None else None,
     }
 
 
+def completion_payload(run, user, request=None):
+    """The live completion feed: filename -> J-ID for freshly indexed uploads.
+
+    Completed ``TempUpload`` rows are deleted, so this is the only place the
+    board learns about them. It is a notification feed, not durable state:
+    bounded, never dismissed, and ignored by fresh page loads.
+    """
+    entries = list(run.recent_completions or []) if run is not None else []
+    if not entries:
+        return []
+    ids = [entry.get("item_id") for entry in entries if entry.get("item_id")]
+    items = MediaItem.objects.in_bulk(ids)
+    out = []
+    for entry in reversed(entries):  # newest first
+        item = items.get(entry.get("item_id"))
+        if item is None:
+            continue
+        out.append(
+            {
+                "id": entry.get("id"),
+                "filename": entry.get("filename"),
+                "j_id": f"J-{item.id}",
+                "resolution": entry.get("resolution", ""),
+                "post_id": entry.get("post_id"),
+                "thumbnail_url": services.signed_media_url(
+                    item, user, "thumbnail", request=request
+                ),
+                "at": entry.get("at"),
+            }
+        )
+    return out
+
+
+def record_completion(temp, item, resolution=""):
+    """Append one completion to the owner's feed; never fails an import."""
+    entry = {
+        "id": str(temp.pk),
+        "filename": temp.original_filename,
+        "item_id": item.pk,
+        "resolution": resolution or temp.resolution or "",
+        "post_id": temp.e621_post_id,
+        "at": timezone.now().isoformat(),
+    }
+    try:
+        with transaction.atomic():
+            run, _ = UploadRun.objects.select_for_update().get_or_create(
+                user_id=temp.user_id
+            )
+            feed = list(run.recent_completions or [])
+            feed.append(entry)
+            run.recent_completions = feed[-COMPLETION_FEED_LIMIT:]
+            run.save(update_fields=["recent_completions", "updated_at"])
+    except Exception:  # noqa: BLE001 - a notification must not break an import
+        logger.exception("Could not record the completion of %s", temp.pk)
+
+
 def reap_stale_claims():
     """Queue rows left claimed by a recycled worker and pause dead runs."""
     cutoff = timezone.now() - STALE_CLAIM_AFTER
@@ -237,20 +297,17 @@ def run_pipeline(user_id):
                 break
             process_round(run, user, rows)
     except PipelinePaused as exc:
-        run.status = UploadRun.STATUS_PAUSED
-        run.phase = ""
-        run.error = str(exc)
-        run.save()
+        _save_run(run, status=UploadRun.STATUS_PAUSED, phase="", error=str(exc))
     except Exception as exc:  # noqa: BLE001 - surface crashes as a run error
         logger.exception("Upload pipeline for user %s failed", user_id)
-        run.status = UploadRun.STATUS_ERROR
-        run.phase = ""
-        run.error = f"The upload pipeline stopped: {exc}"
-        run.save()
+        _save_run(
+            run,
+            status=UploadRun.STATUS_ERROR,
+            phase="",
+            error=f"The upload pipeline stopped: {exc}",
+        )
     else:
-        run.status = UploadRun.STATUS_IDLE
-        run.phase = ""
-        run.save()
+        _save_run(run, status=UploadRun.STATUS_IDLE, phase="")
 
 
 def claim_round(user, size=CLAIM_SIZE):
@@ -460,7 +517,9 @@ def finalize_round(run, ids, matched):
         TempUpload.objects.filter(pk__in=release).update(claimed_at=None)
     _save_run(
         run,
-        processed=run.processed + len(finished),
+        # Completed rows are deleted as they are imported, so they cannot be
+        # seen in the refreshed rows; count the matches explicitly.
+        processed=run.processed + len(finished) + matched,
         failed=run.failed + len(failed - finished),
         matched=run.matched + matched,
     )