path: root/src/srht_contrib/services/git.py

from __future__ import annotations

import copy
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
import logging
from typing import Any

from srht_contrib.config import Settings
from srht_contrib.schemas import NormalizedEvent
from srht_contrib.services.srht_client import SourceHutGraphQLClient
from srht_contrib.services.types import BackfillBatchResult
from srht_contrib.utils.dates import ensure_utc, parse_datetime
from srht_contrib.utils.identity import ActorIdentityResolver


logger = logging.getLogger(__name__)


REPOSITORY_LOG_QUERY = """
query RepositoryLog($username: String!, $repoName: String!, $cursor: Cursor) {
  user(username: $username) {
    repository(name: $repoName) {
      name
      owner {
        canonicalName
      }
      log(cursor: $cursor) {
        results {
          id
          shortId
          author {
            name
            email
            time
          }
          committer {
            name
            email
            time
          }
          message
        }
        cursor
      }
    }
  }
}
""".strip()


USER_REPOSITORIES_QUERY = """
query UserRepositories($username: String!, $cursor: Cursor) {
  user(username: $username) {
    repositories(cursor: $cursor) {
      results {
        name
        owner {
          canonicalName
        }
      }
      cursor
    }
  }
}
""".strip()


@dataclass(slots=True)
class GitPollResult:
    events: list[NormalizedEvent]
    cursor: str


class GitIngestionService:
    """Polls git.sr.ht repositories and normalizes commits for one actor."""

    service_name = "git"

    def __init__(self, client: SourceHutGraphQLClient, settings: Settings) -> None:
        self.client = client
        self.settings = settings
        self.identity_resolver = ActorIdentityResolver(settings.actor_aliases_json)

    def fetch_recent_events(
        self,
        actor: str,
        since: datetime | None = None,
        repositories: list[str] | None = None,
    ) -> GitPollResult:
        since_dt = ensure_utc(since or (datetime.now(tz=UTC) - timedelta(days=30)))
        discovered_repositories = repositories or self._repositories_for_actor(actor)
        if not discovered_repositories:
            logger.info("git poll skipped for actor=%s because no repositories were discovered", actor)
            return GitPollResult(events=[], cursor=datetime.now(tz=UTC).isoformat())

        events: list[NormalizedEvent] = []
        for repository in discovered_repositories:
            owner, repo_name = self._split_repository(actor, repository)
            repo_events = self._fetch_repository_commits(actor=actor, owner=owner, repo_name=repo_name, since=since_dt)
            events.extend(repo_events)

        logger.info("git poll complete for actor=%s normalized_events=%s", actor, len(events))
        return GitPollResult(events=events, cursor=datetime.now(tz=UTC).isoformat())

    def _repositories_for_actor(self, actor: str) -> list[str]:
        configured = {
            self._canonical_repository_name(actor, repository)
            for repository in self.settings.git_tracked_repositories
        }
        discovered = set(self._discover_owned_repositories(actor))
        repositories = sorted(configured | discovered)
        logger.info(
            "git repositories selected for actor=%s count=%s configured=%s discovered=%s",
            actor,
            len(repositories),
            len(configured),
            len(discovered),
        )
        return repositories

    def fetch_backfill_batch(self, actor: str, cursor_state: dict | None = None) -> BackfillBatchResult:
        state = {
            "discovery_cursor": None,
            "discovery_complete": False,
            "repository_queue": sorted(
                {
                    self._canonical_repository_name(actor, repository)
                    for repository in self.settings.git_tracked_repositories
                }
            ),
            "current_repository": None,
        }
        if cursor_state:
            state.update(copy.deepcopy(cursor_state))

        if not state["discovery_complete"]:
            data = self.client.execute(
                USER_REPOSITORIES_QUERY,
                {"username": actor.lstrip("~"), "cursor": state["discovery_cursor"]},
            )
            user = data.get("user") or {}
            repositories_page = user.get("repositories") or {}
            results = repositories_page.get("results") or []
            state["discovery_cursor"] = repositories_page.get("cursor")
            state["discovery_complete"] = not bool(state["discovery_cursor"])
            known = set(state["repository_queue"])
            current_repository = state.get("current_repository")
            if current_repository:
                known.add(current_repository["name"])
            for repository in results:
                if not isinstance(repository, dict):
                    continue
                name = repository.get("name")
                repository_owner = ((repository.get("owner") or {}).get("canonicalName") or actor).strip()
                if not name or not repository_owner:
                    continue
                canonical_name = f"{repository_owner}/{name}"
                if canonical_name not in known:
                    state["repository_queue"].append(canonical_name)
                    known.add(canonical_name)
            state["repository_queue"] = sorted(state["repository_queue"])
            logger.info(
                "git backfill discovery actor=%s page_count=%s queue=%s next_cursor=%s",
                actor,
                len(results),
                len(state["repository_queue"]),
                bool(state["discovery_cursor"]),
            )
            complete = state["discovery_complete"] and not state["repository_queue"] and not state["current_repository"]
            return BackfillBatchResult(events=[], cursor_state=state, complete=complete)

        if state["current_repository"] is None:
            if not state["repository_queue"]:
                return BackfillBatchResult(events=[], cursor_state=None, complete=True)
            state["current_repository"] = {"name": state["repository_queue"].pop(0), "cursor": None}

        repository_name = state["current_repository"]["name"]
        owner, repo_name = self._split_repository(actor, repository_name)
        data = self.client.execute(
            REPOSITORY_LOG_QUERY,
            {"username": owner, "repoName": repo_name, "cursor": state["current_repository"]["cursor"]},
        )
        user = data.get("user") or {}
        repository = user.get("repository") or {}
        log_page = repository.get("log") or {}
        commits = log_page.get("results") or []
        next_cursor = log_page.get("cursor")
        events = [
            normalized
            for commit in commits
            if isinstance(commit, dict)
            for normalized in [self._normalize_commit(actor=actor, repo_name=repo_name, commit=commit)]
            if normalized is not None
        ]
        logger.info(
            "git backfill actor=%s repository=%s commits=%s next_cursor=%s",
            actor,
            repository_name,
            len(commits),
            bool(next_cursor),
        )
        if next_cursor:
            state["current_repository"]["cursor"] = next_cursor
        else:
            state["current_repository"] = None

        complete = state["discovery_complete"] and not state["repository_queue"] and not state["current_repository"]
        return BackfillBatchResult(events=events, cursor_state=None if complete else state, complete=complete)

    def _discover_owned_repositories(self, actor: str) -> list[str]:
        owner = actor.lstrip("~")
        repositories: list[str] = []
        cursor: str | None = None

        for _ in range(50):
            data = self.client.execute(
                USER_REPOSITORIES_QUERY,
                {"username": owner, "cursor": cursor},
            )
            user = data.get("user") or {}
            repositories_page = user.get("repositories") or {}
            results = repositories_page.get("results") or []
            cursor = repositories_page.get("cursor")
            logger.info(
                "git repository discovery actor=%s page_count=%s next_cursor=%s",
                actor,
                len(results),
                bool(cursor),
            )

            for repository in results:
                if not isinstance(repository, dict):
                    continue
                name = repository.get("name")
                repository_owner = ((repository.get("owner") or {}).get("canonicalName") or actor).strip()
                if not name or not repository_owner:
                    continue
                repositories.append(f"{repository_owner}/{name}")

            if not cursor:
                break

        return repositories

    @staticmethod
    def _canonical_repository_name(default_actor: str, repository: str) -> str:
        owner, repo_name = GitIngestionService._split_repository(default_actor, repository)
        return f"~{owner}/{repo_name}"

    @staticmethod
    def _split_repository(default_actor: str, repository: str) -> tuple[str, str]:
        if "/" in repository:
            owner, repo_name = repository.split("/", 1)
            canonical_owner = owner if owner.startswith("~") else f"~{owner}"
            return canonical_owner.lstrip("~"), repo_name
        return default_actor.lstrip("~"), repository

    def _fetch_repository_commits(
        self,
        *,
        actor: str,
        owner: str,
        repo_name: str,
        since: datetime,
    ) -> list[NormalizedEvent]:
        events: list[NormalizedEvent] = []
        cursor: str | None = None

        for _ in range(50):
            data = self.client.execute(
                REPOSITORY_LOG_QUERY,
                {"username": owner, "repoName": repo_name, "cursor": cursor},
            )
            user = data.get("user") or {}
            repository = user.get("repository") or {}
            log_page = repository.get("log") or {}
            commits = log_page.get("results") or []
            cursor = log_page.get("cursor")
            logger.info(
                "git repository=%s/%s commit page count=%s next_cursor=%s",
                owner,
                repo_name,
                len(commits),
                bool(cursor),
            )

            stop_paging = False
            for commit in commits:
                if not isinstance(commit, dict):
                    continue
                commit_time = parse_datetime((commit.get("author") or {}).get("time"))
                if commit_time < since:
                    stop_paging = True
                    logger.info(
                        "git commit %s skipped because commit_time=%s is before since=%s",
                        commit.get("shortId") or commit.get("id"),
                        commit_time.isoformat(),
                        since.isoformat(),
                    )
                    continue

                normalized = self._normalize_commit(actor=actor, repo_name=repo_name, commit=commit)
                if normalized is not None:
                    logger.info(
                        "git commit accepted repo=%s shortId=%s author=%s email=%s",
                        repo_name,
                        commit.get("shortId"),
                        (commit.get("author") or {}).get("name"),
                        (commit.get("author") or {}).get("email"),
                    )
                    events.append(normalized)
                else:
                    logger.info(
                        "git commit skipped repo=%s shortId=%s author=%s email=%s",
                        repo_name,
                        commit.get("shortId"),
                        (commit.get("author") or {}).get("name"),
                        (commit.get("author") or {}).get("email"),
                    )

            if stop_paging or not cursor:
                break

        return events

    def _normalize_commit(
        self,
        *,
        actor: str,
        repo_name: str,
        commit: dict[str, Any],
    ) -> NormalizedEvent | None:
        author = commit.get("author") or {}
        candidate_aliases = [
            actor,
            author.get("email", ""),
            author.get("name", ""),
        ]
        matched_actor = None
        for candidate in candidate_aliases:
            canonical = self.identity_resolver.canonicalize(candidate)
            if canonical == actor:
                matched_actor = canonical
                break

        if matched_actor is None:
            return None

        commit_id = str(commit["id"])
        commit_time = parse_datetime(author["time"])
        return NormalizedEvent(
            service=self.service_name,
            event_type="commit",
            actor=matched_actor,
            repo_name=repo_name,
            resource_id=commit_id,
            external_uid=f"git:{repo_name}:{commit_id}",
            occurred_at=commit_time,
            weight=self.settings.event_weights["commit"],
            raw_payload_json=commit,
        )