Delete VPS document stage after verified local storage sync

app/modules/documents/services.py

@@ -53,6 +53,11 @@ UDIN_REQUIRED_DOCUMENT_TYPES = {"AUDIT_REPORT", "SIGNED_OUTPUT", "ACKNOWLEDGEMEN
 
 MAX_UPLOAD_BYTES = int(os.getenv("DOCUMENT_MAX_UPLOAD_MB", "50")) * 1024 * 1024
 
+# A server copy is retained only as a temporary transfer buffer. It is deleted
+# after the local storage agent confirms the exact expected SHA-256 hash.
+# Emergency rollback: DOCUMENT_DELETE_VPS_AFTER_LOCAL_SYNC=false
+DELETE_VPS_AFTER_LOCAL_SYNC = (os.getenv("DOCUMENT_DELETE_VPS_AFTER_LOCAL_SYNC", "true").strip().lower() not in {"0", "false", "no", "off"})
+
 
 def _resolve_document_storage_root(env_name: str, default_leaf: str) -> Path:
     """Return a short absolute storage root.
@@ -815,6 +820,139 @@ def acknowledge_storage_job(db: Session, *, node: BranchStorageNode, job: Docume
     return False
 
 
+def _prune_empty_stage_directories(file_path: Path, root: Path) -> None:
+    """Remove empty parent folders without crossing the configured root."""
+    root = root.resolve()
+    parent = file_path.parent
+    while parent != root:
+        try:
+            parent.rmdir()
+        except OSError:
+            break
+        parent = parent.parent
+
+
+def _verified_stage_path(version) -> Path | None:
+    """Resolve the staged path and reject traversal outside storage root."""
+    if not version or not getattr(version, "local_relative_path", None):
+        return None
+    root = DEFAULT_STORAGE_ROOT.resolve()
+    candidate = (DEFAULT_STORAGE_ROOT / Path(version.local_relative_path)).resolve()
+    try:
+        candidate.relative_to(root)
+    except ValueError:
+        return None
+    return candidate
+
+
+def _cleanup_verified_vps_stage(*, job, version) -> dict[str, object]:
+    """Delete one VPS stage only after verified branch-local storage."""
+    result: dict[str, object] = {"enabled": DELETE_VPS_AFTER_LOCAL_SYNC, "deleted": False}
+    if not DELETE_VPS_AFTER_LOCAL_SYNC:
+        result["reason"] = "cleanup_disabled"
+        return result
+    if not job or job.status != "completed":
+        result["reason"] = "job_not_completed"
+        return result
+    if getattr(job, "vps_cleanup_status", "pending") == "deleted":
+        result.update({"deleted": True, "already_completed": True, "reason": "already_deleted"})
+        return result
+
+    expected = (job.expected_hash_sha256 or "").strip().lower()
+    acknowledged = (job.acknowledged_hash_sha256 or "").strip().lower()
+    if not expected or acknowledged != expected:
+        result["reason"] = "local_hash_not_verified"
+        return result
+    if not (job.local_final_path or "").strip():
+        result["reason"] = "local_final_path_missing"
+        return result
+
+    job.vps_cleanup_attempts = int(job.vps_cleanup_attempts or 0) + 1
+    job.vps_cleanup_status = "deleting"
+    job.vps_cleanup_error = None
+    stage_path = _verified_stage_path(version)
+    if stage_path is None:
+        job.vps_cleanup_status = "failed"
+        job.vps_cleanup_error = "Unsafe or missing staged path."
+        result["reason"] = "unsafe_stage_path"
+        return result
+
+    try:
+        deleted_bytes = int(stage_path.stat().st_size) if stage_path.exists() and stage_path.is_file() else 0
+        if stage_path.exists():
+            if not stage_path.is_file():
+                raise OSError("Staged path is not a regular file.")
+            stage_path.unlink()
+            _prune_empty_stage_directories(stage_path, DEFAULT_STORAGE_ROOT)
+        job.vps_cleanup_status = "deleted"
+        job.vps_cleanup_error = None
+        job.vps_deleted_at_utc = _now_utc()
+        job.vps_deleted_bytes = deleted_bytes
+        result.update({
+            "deleted": True,
+            "already_missing": deleted_bytes == 0,
+            "deleted_bytes": deleted_bytes,
+            "reason": "deleted_after_verified_local_sync",
+        })
+        return result
+    except OSError as exc:
+        job.vps_cleanup_status = "retry" if int(job.vps_cleanup_attempts or 0) < 20 else "failed"
+        job.vps_cleanup_error = f"{type(exc).__name__}: {str(exc)[:900]}"
+        result.update({"reason": "delete_failed", "error": job.vps_cleanup_error})
+        return result
+
+
+def cleanup_completed_storage_job_vps_stage(db: Session, job: DocumentStorageJob) -> dict[str, object]:
+    version = db.get(EngagementDocumentVersion, job.version_id) if job else None
+    return _cleanup_verified_vps_stage(job=job, version=version)
+
+
+def cleanup_completed_permanent_storage_job_vps_stage(db: Session, job: PermanentDocumentStorageJob) -> dict[str, object]:
+    version = db.get(PermanentClientDocumentVersion, job.version_id) if job else None
+    return _cleanup_verified_vps_stage(job=job, version=version)
+
+
+def retry_verified_vps_cleanup_for_node(db: Session, node: BranchStorageNode, limit: int = 50) -> dict[str, int]:
+    """Retry interrupted/failed deletion during heartbeat or tunnel sync."""
+    summary = {"checked": 0, "deleted": 0, "failed": 0}
+    if not DELETE_VPS_AFTER_LOCAL_SYNC or not node:
+        return summary
+    retry_states = ["pending", "retry", "failed", "deleting"]
+    engagement_jobs = db.execute(
+        select(DocumentStorageJob)
+        .where(
+            DocumentStorageJob.storage_node_id == node.id,
+            DocumentStorageJob.status == "completed",
+            DocumentStorageJob.vps_cleanup_status.in_(retry_states),
+        )
+        .order_by(DocumentStorageJob.completed_at_utc.asc(), DocumentStorageJob.id.asc())
+        .limit(limit)
+    ).scalars().all()
+    remaining = max(0, limit - len(engagement_jobs))
+    permanent_jobs = []
+    if remaining:
+        permanent_jobs = db.execute(
+            select(PermanentDocumentStorageJob)
+            .where(
+                PermanentDocumentStorageJob.storage_node_id == node.id,
+                PermanentDocumentStorageJob.status == "completed",
+                PermanentDocumentStorageJob.vps_cleanup_status.in_(retry_states),
+            )
+            .order_by(PermanentDocumentStorageJob.completed_at_utc.asc(), PermanentDocumentStorageJob.id.asc())
+            .limit(remaining)
+        ).scalars().all()
+
+    for job in engagement_jobs:
+        summary["checked"] += 1
+        result = cleanup_completed_storage_job_vps_stage(db, job)
+        summary["deleted" if result.get("deleted") else "failed"] += 1
+    for job in permanent_jobs:
+        summary["checked"] += 1
+        result = cleanup_completed_permanent_storage_job_vps_stage(db, job)
+        summary["deleted" if result.get("deleted") else "failed"] += 1
+    return summary
+
+
 def list_storage_jobs(db: Session, tenant_id: int | None = None, branch_id: int | None = None, status: str | None = None, limit: int = 200):
     stmt = select(DocumentStorageJob).order_by(DocumentStorageJob.created_at_utc.desc())
     if tenant_id: