Create danger_zone_service.py

app/services/danger_zone_service.py

@@ -0,0 +1,86 @@
+import tempfile
+import shutil
+import logging
+from typing import List, Dict, Any
+from git import Repo
+
+logger = logging.getLogger(__name__)
+
+class DangerZoneService:
+    @staticmethod
+    def simulate_destructive_action(
+        owner: str,
+        repo_name: str,
+        action: str,
+        target_ref: str,
+        current_local_shas: List[str] = None,
+    ) -> Dict[str, Any]:
+        """
+        Clones the repo into an ephemeral temp directory, performs the action,
+        and returns what would be lost. The temp dir is always cleaned up.
+        """
+        temp_dir = tempfile.mkdtemp(prefix="gitgud_danger_")
+        repo_url = f"https://github.com/{owner}/{repo_name}.git"
+
+        try:
+            logger.info(f"Cloning {repo_url} into isolation dir for simulation...")
+            repo = Repo.clone_from(repo_url, temp_dir, depth=100)  # shallow is enough
+
+            # Record reachable commits before the action
+            before_shas = {c.hexsha for c in repo.iter_commits()}
+
+            # Apply the simulated operation
+            if action in ("hard_reset", "force_push"):
+                repo.git.reset("--hard", target_ref)
+            else:
+                # Future-proof: you can add rebase/squash later
+                repo.git.reset("--hard", target_ref)
+
+            after_shas = {c.hexsha for c in repo.iter_commits()}
+            lost_shas = before_shas - after_shas
+
+            would_lose = []
+            for sha in lost_shas:
+                try:
+                    commit = repo.commit(sha)
+                    would_lose.append({
+                        "id": sha[:7],
+                        "author": commit.author.name if commit.author else "unknown",
+                        "message": commit.summary,
+                        "timestamp": commit.authored_date,
+                    })
+                except Exception:
+                    continue
+
+            # Optional: filter only commits that the client currently has locally
+            if current_local_shas:
+                local_set = set(current_local_shas)
+                would_lose = [c for c in would_lose if any(c["id"] in s or s.startswith(c["id"]) for s in local_set)]
+
+            summary = (
+                f"This {action.replace('_', ' ')} will eliminate "
+                f"{len(would_lose)} commits from the remote history."
+            )
+
+            return {
+                "wouldLose": would_lose,
+                "wouldRewrite": [],          # extend later for rebase/squash
+                "conflicts": [],
+                "summary": summary,
+            }
+
+        except Exception as e:
+            logger.error(f"Danger zone simulation failed: {e}", exc_info=True)
+            return {
+                "wouldLose": [],
+                "wouldRewrite": [],
+                "conflicts": [],
+                "summary": f"Simulation failed: {str(e)}",
+                "error": str(e),
+            }
+        finally:
+            # Always clean the isolation directory
+            try:
+                shutil.rmtree(temp_dir, ignore_errors=True)
+            except Exception:
+                pass