"""Turns a member's draft into one commit on the uploads repo's main branch (SPEC C-20…C-25, ARCHITECTURE §2.3). Pushes are serialized with a lock: two members pushing at once never fight over the shared working tree. """ import asyncio import logging from collections.abc import Awaitable, Callable from dataclasses import dataclass, field from datetime import datetime, timedelta from enum import StrEnum from pathlib import Path from sqlalchemy import func, select from app.db.models import ChangeOp, Push, User, utcnow from app.db.session import SessionMaker from app.repo.drafts import DraftView from app.repo.git import GitError, UploadsRepo log = logging.getLogger(__name__) BOT_NAME = "tgit-bot" BOT_EMAIL = "tgit-bot@users.noreply.tgit" PUSHES_PER_HOUR = 20 PUSH_ATTEMPTS = 3 class PushStatus(StrEnum): OK = "ok" CONFLICT = "conflict" # someone changed the same files in main since the draft began EMPTY = "empty" # the draft matches main already LIMIT = "limit" FAILED = "failed" @dataclass(frozen=True) class PushResult: status: PushStatus sha: str | None = None conflicts: list[str] = field(default_factory=list) class Publisher: def __init__( self, repo: UploadsRepo, sessions: SessionMaker, clock: Callable[[], datetime] = utcnow, before_push: Callable[[], Awaitable[None]] | None = None, # test hook: simulate a concurrent push ) -> None: self._repo = repo self._sessions = sessions self._clock = clock self._before_push = before_push self._lock = asyncio.Lock() async def push(self, member: User, draft: DraftView, message: str, *, overwrite: bool = False) -> PushResult: if await self._pushes_last_hour(member.tg_id) >= PUSHES_PER_HOUR: return PushResult(PushStatus.LIMIT) async with self._lock: try: return await self._push(member, draft, message, overwrite) except GitError: log.exception("push failed") await self._reset() return PushResult(PushStatus.FAILED) async def record( self, member: User, result: PushResult, message: str, draft: DraftView, chat_id: int, message_id: int ) -> None: assert result.sha is not None async with self._sessions.begin() as s: s.add( Push( user_id=member.tg_id, sha=result.sha, message=message, paths=[c.path for c in draft.changes], chat_id=chat_id, message_id=message_id, created_at=self._clock(), ) ) async def _push(self, member: User, draft: DraftView, message: str, overwrite: bool) -> PushResult: repo = self._repo await repo.refresh(force=True) if not overwrite: conflicts = await self._conflicts(draft) if conflicts: return PushResult(PushStatus.CONFLICT, conflicts=conflicts) await self._reset() self._apply(draft) await repo.git("add", "-A") try: await repo.git("diff", "--cached", "--quiet") return PushResult(PushStatus.EMPTY) # exit 0: nothing staged except GitError: pass # exit 1: there are changes handle = f"@{member.username}" if member.username else member.full_name full_message = f"{message.strip()}\n\nTelegram-User: {handle} ({member.tg_id})" await repo.git( "-c", f"user.name={BOT_NAME}", "-c", f"user.email={BOT_EMAIL}", "commit", "-q", "-m", full_message ) branch = repo.branch for attempt in range(1, PUSH_ATTEMPTS + 1): if self._before_push is not None: await self._before_push() try: await repo.git("push", "-q", "origin", f"HEAD:{branch}") break except GitError: if attempt == PUSH_ATTEMPTS: raise log.info("push rejected, rebasing (attempt %d)", attempt) await repo.git("fetch", "-q", "origin") try: await repo.git( "-c", f"user.name={BOT_NAME}", "-c", f"user.email={BOT_EMAIL}", "rebase", "-q", repo.ref ) except GitError: await repo.git("rebase", "--abort") return PushResult(PushStatus.CONFLICT, conflicts=await self._conflicts(draft)) sha = (await repo.git("rev-parse", "HEAD")).decode().strip() return PushResult(PushStatus.OK, sha=sha) async def _conflicts(self, draft: DraftView) -> list[str]: try: out = await self._repo.git("diff", "--name-only", draft.base_sha, self._repo.ref) except GitError: return [c.path for c in draft.changes] # base unknown (e.g. history rewritten): be safe changed = set(out.decode().splitlines()) return [c.path for c in draft.changes if c.path in changed] async def _reset(self) -> None: repo = self._repo await repo.git("checkout", "-q", "-B", repo.branch, repo.ref) await repo.git("reset", "-q", "--hard", repo.ref) await repo.git("clean", "-q", "-fdx") def _apply(self, draft: DraftView) -> None: root = self._repo.path.resolve() for change in draft.changes: target = (root / change.path).resolve() if root not in target.parents: # validated earlier; this is the last line of defence raise GitError(f"path escapes the repo: {change.path}") if change.op is ChangeOp.DELETE: target.unlink(missing_ok=True) _prune_empty_dirs(target.parent, root) else: target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(change.content or b"") async def _pushes_last_hour(self, user_id: int) -> int: async with self._sessions() as s: count = await s.scalar( select(func.count()) .select_from(Push) .where(Push.user_id == user_id, Push.created_at > self._clock() - timedelta(hours=1)) ) return count or 0 def _prune_empty_dirs(directory: Path, root: Path) -> None: while directory != root and directory.is_dir() and not any(directory.iterdir()): directory.rmdir() directory = directory.parent