feat: add product archive retention cleanup

backend/service/cleanup_service.py

@@ -0,0 +1,143 @@
+"""Deterministic retention preview and physical cleanup for jobs and products."""
+
+from __future__ import annotations
+
+import shutil
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+from pydantic import BaseModel, Field
+
+from .metadata_store import MetadataStore
+from .product_archive_store import ProductArchiveStore
+from .schemas import ProductRecord
+from .storage import Storage
+
+
+TEMPORARY_RETENTION = timedelta(days=30)
+REMINDER_AGE = timedelta(days=23)
+
+
+class TemporaryJob(BaseModel):
+    job_id: str
+    created_at: datetime
+    size_bytes: int
+
+
+class CleanupReport(BaseModel):
+    temporary_jobs: list[TemporaryJob] = Field(default_factory=list)
+    pending_products: list[ProductRecord] = Field(default_factory=list)
+    reclaimable_bytes: int = 0
+    reminder_job_ids: list[str] = Field(default_factory=list)
+    deleted_job_ids: list[str] = Field(default_factory=list)
+    deleted_product_ids: list[str] = Field(default_factory=list)
+
+
+def _as_utc(value: datetime) -> datetime:
+    if value.tzinfo is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value.astimezone(timezone.utc)
+
+
+class CleanupService:
+    """Calculate candidates first and apply only those still eligible."""
+
+    def __init__(
+        self,
+        storage: Storage,
+        metadata_store: MetadataStore,
+        product_store: ProductArchiveStore,
+    ) -> None:
+        self.storage = storage
+        self.metadata_store = metadata_store
+        self.product_store = product_store
+
+    def archive_protected_job_ids(self) -> set[str]:
+        rows = self.product_store._fetchall(
+            """SELECT DISTINCT source_job_id
+               FROM product_wordcloud_archives
+               WHERE source_job_id != ''"""
+        )
+        return {str(row["source_job_id"]) for row in rows}
+
+    def _is_archive_protected(self, job_id: str) -> bool:
+        row = self.product_store._fetchone(
+            """SELECT 1 FROM product_wordcloud_archives
+               WHERE source_job_id = ? LIMIT 1""",
+            (job_id,),
+        )
+        return row is not None
+
+    def preview(self, now: datetime) -> CleanupReport:
+        now_utc = _as_utc(now)
+        protected = self.archive_protected_job_ids()
+        temporary_jobs: list[TemporaryJob] = []
+        reminder_job_ids: list[str] = []
+
+        for job in self.metadata_store.load_jobs():
+            if job.status != "success" or not job.artifacts.get("db") or job.job_id in protected:
+                continue
+            age = now_utc - _as_utc(job.created_at)
+            if age >= TEMPORARY_RETENTION:
+                temporary_jobs.append(
+                    TemporaryJob(
+                        job_id=job.job_id,
+                        created_at=job.created_at,
+                        size_bytes=self.storage.job_dir_size(job.job_id),
+                    )
+                )
+            elif age >= REMINDER_AGE:
+                reminder_job_ids.append(job.job_id)
+
+        temporary_jobs.sort(key=lambda item: (item.created_at, item.job_id))
+        reminder_job_ids.sort()
+        pending_products = self.product_store.due_product_cleanups(now_utc)
+        product_bytes = sum(
+            self._directory_size(self.product_store.root.parent / product.product_id)
+            for product in pending_products
+        )
+        return CleanupReport(
+            temporary_jobs=temporary_jobs,
+            pending_products=pending_products,
+            reclaimable_bytes=sum(item.size_bytes for item in temporary_jobs) + product_bytes,
+            reminder_job_ids=reminder_job_ids,
+        )
+
+    def apply(self, now: datetime) -> CleanupReport:
+        now_utc = _as_utc(now)
+        report = self.preview(now_utc)
+
+        for item in report.temporary_jobs:
+            # An archive can be committed after preview. Protect it at the last
+            # possible point before removing the original workspace.
+            if self._is_archive_protected(item.job_id):
+                continue
+            self.storage.remove_job_dir(item.job_id)
+            self.metadata_store.delete_job(item.job_id)
+            report.deleted_job_ids.append(item.job_id)
+
+        for candidate in report.pending_products:
+            # Re-read logical state so a concurrent restore wins over cleanup.
+            try:
+                product = self.product_store.get_product(candidate.product_id)
+            except ValueError:
+                continue
+            if (
+                product.status != "pending_cleanup"
+                or product.purge_after is None
+                or _as_utc(product.purge_after) > now_utc
+            ):
+                continue
+            product_dir = self.product_store.root.parent / product.product_id
+            if product_dir.exists():
+                shutil.rmtree(product_dir)
+            self.product_store.purge_product(product.product_id, now=now_utc)
+            report.deleted_product_ids.append(product.product_id)
+
+        return report
+
+    @staticmethod
+    def _directory_size(root: Path) -> int:
+        if not root.exists():
+            return 0
+        return sum(path.stat().st_size for path in root.rglob("*") if path.is_file())