server/apps/codebase.py

import copy
import random
from typing import Any, Dict, List

_COMMIT_POOL = [
    {"hash": "a1b2c3d", "message": "fix: null pointer in auth middleware", "author": "alice", "files": ["auth/middleware.py"], "service": "auth"},
    {"hash": "e4f5g6h", "message": "feat: rate limiting for API gateway", "author": "bob", "files": ["gateway/ratelimit.py"], "service": "api-gateway"},
    {"hash": "i7j8k9l", "message": "refactor: database connection pool", "author": "carol", "files": ["db/pool.py"], "service": "auth"},
    {"hash": "m1n2o3p", "message": "fix: SQL injection in user search", "author": "dave", "files": ["search/query.py"], "service": "search"},
    {"hash": "q4r5s6t", "message": "perf: Redis session caching", "author": "eve", "files": ["payments/session.py"], "service": "payments"},
    {"hash": "u7v8w9x", "message": "feat: dark mode dashboard", "author": "frank", "files": ["frontend/theme.js"], "service": "notifications"},
    {"hash": "y1z2a3b", "message": "fix: XSS in comment renderer", "author": "alice", "files": ["frontend/comments.jsx"], "service": "notifications"},
    {"hash": "c4d5e6f", "message": "docs: API authentication guide", "author": "bob", "files": ["docs/auth.md"], "service": "auth"},
    {"hash": "g7h8i9j", "message": "feat: CSV data export endpoint", "author": "carol", "files": ["api/export.py"], "service": "api-gateway"},
    {"hash": "k1l2m3n", "message": "fix: websocket memory leak", "author": "dave", "files": ["payments/ws.py"], "service": "payments"},
    {"hash": "p2q3r4s", "message": "fix: checkout timeout on slow networks", "author": "eve", "files": ["checkout/flow.py"], "service": "checkout"},
    {"hash": "t5u6v7w", "message": "perf: inventory cache warm-up", "author": "frank", "files": ["inventory/cache.py"], "service": "inventory"},
]

_PR_POOL = [
    {"id": "PR-101", "title": "Fix auth rate limiting bug", "author": "alice", "status": "merged", "service": "auth"},
    {"id": "PR-102", "title": "Add payment retry logic", "author": "bob", "status": "open", "service": "payments"},
    {"id": "PR-103", "title": "Search index optimization", "author": "carol", "status": "merged", "service": "search"},
    {"id": "PR-104", "title": "Gateway health check endpoint", "author": "dave", "status": "open", "service": "api-gateway"},
    {"id": "PR-105", "title": "Checkout flow improvements", "author": "eve", "status": "merged", "service": "checkout"},
]


class CodebaseApp:
    def __init__(self, seed: int, services: List[str]):
        rng = random.Random(seed + 1337)
        relevant = [c for c in _COMMIT_POOL if c["service"] in services]
        other = [c for c in _COMMIT_POOL if c["service"] not in services]
        pool = relevant + rng.sample(other, min(3, len(other)))
        self.commits = rng.sample(pool, min(8, len(pool)))
        self.prs = [p for p in _PR_POOL if p["service"] in services]

    def list_commits(self, args: Dict[str, Any]) -> Dict[str, Any]:
        query = args.get("query", "").lower()
        service = args.get("service")
        limit = min(args.get("limit", 10), 20)
        offset = args.get("offset", 0)
        results = self.commits
        if query:
            results = [c for c in results if query in c["message"].lower() or query in c["hash"]]
        if service:
            results = [c for c in results if c.get("service") == service]
        return {"ok": True, "data": results[offset: offset + limit], "total": len(results)}

    def get_commit(self, args: Dict[str, Any]) -> Dict[str, Any]:
        h = args.get("hash", "")
        for c in self.commits:
            if c["hash"].startswith(h):
                return {"ok": True, "data": copy.deepcopy(c)}
        return {"ok": False, "error": f"Commit '{h}' not found"}

    def list_prs(self, args: Dict[str, Any]) -> Dict[str, Any]:
        status = args.get("status")
        results = self.prs if not status else [p for p in self.prs if p["status"] == status]
        return {"ok": True, "data": copy.deepcopy(results)}