"""GitHub sync service for syncing issues, PRs, and discussions. This service uses typed metadata contracts from integrations.contracts to ensure type-safe metadata storage. See Constitution B12 for metadata contract requirements. """ from __future__ import annotations import logging from dataclasses import dataclass, field from datetime import datetime, timedelta from typing import TYPE_CHECKING, Any, cast from django.db import transaction from django.utils import timezone as django_timezone from members.models import Member from messages.contracts import GitHubAuthorMeta, GitHubLabelMeta from messages.models import Message, Thread from ..contracts import GitHubDiscussionMeta, GitHubIssueMeta, GitHubPRMeta from ..models import GitHubRepository from .contributor_identity import ContributorIdentityService from .github_client import GitHubClient, GitHubRateLimitError if TYPE_CHECKING: pass # Future type-only imports can go here logger = logging.getLogger(__name__) @dataclass class SyncStats: """Statistics for a sync operation.""" issues_synced: int = 0 prs_synced: int = 0 discussions_synced: int = 0 comments_synced: int = 0 contributors_created: int = 0 errors: list[str] = field(default_factory=list) class GitHubSyncService: """Service for syncing GitHub data to local database.""" def __init__(self, repository: GitHubRepository): self.repository = repository self.installation = repository.installation self.source = repository.source self.workspace = self.source.workspace self.client = GitHubClient(self.installation) self.contributor_service = ContributorIdentityService(self.workspace) self.stats = SyncStats() def sync_all(self, historical_days: int = 90) -> dict[str, Any]: """Sync all content types from GitHub. Args: historical_days: Number of days of history to sync on initial sync Returns: Dictionary with sync statistics """ try: self._update_status("syncing") # Calculate since date for incremental sync since = self._get_since_date(historical_days) # Sync each content type self.sync_issues(since) self.sync_pull_requests(since) self.sync_discussions(since) # Update sync status self._update_status("active") self._update_cursors() total_synced = self.stats.issues_synced + self.stats.prs_synced + self.stats.discussions_synced logger.info(f"Sync complete for {self.repository.full_name}: " f"{total_synced} items synced") return self._stats_to_dict() except GitHubRateLimitError as e: error_msg = f"Rate limit exceeded: {e}" logger.warning(error_msg) self.stats.errors.append(error_msg) self._update_status("error", error_msg) raise except Exception as e: error_msg = f"Sync failed: {str(e)}" logger.exception(error_msg) self.stats.errors.append(error_msg) self._update_status("error", error_msg) raise def _stats_to_dict(self) -> dict[str, Any]: """Convert stats dataclass to dictionary.""" return { "issues_synced": self.stats.issues_synced, "prs_synced": self.stats.prs_synced, "discussions_synced": self.stats.discussions_synced, "comments_synced": self.stats.comments_synced, "contributors_created": self.stats.contributors_created, "errors": self.stats.errors, } def sync_issues(self, since: datetime | None = None) -> int: """Sync issues from GitHub. Args: since: Only sync issues updated after this date Returns: Number of issues synced """ logger.info(f"Syncing issues for {self.repository.full_name}") try: issues = self.client.list_issues( self.repository.owner, self.repository.name, since=since, state="all", ) for issue_data in issues: # Skip pull requests (they have a pull_request key) if "pull_request" in issue_data: continue self._upsert_thread_from_issue(issue_data) self.stats.issues_synced += 1 except Exception as e: error_msg = f"Failed to sync issues: {e}" logger.error(error_msg) self.stats.errors.append(error_msg) return self.stats.issues_synced def sync_pull_requests(self, since: datetime | None = None) -> int: """Sync pull requests from GitHub. Args: since: Only sync PRs updated after this date Returns: Number of PRs synced """ logger.info(f"Syncing pull requests for {self.repository.full_name}") try: prs = self.client.list_pull_requests( self.repository.owner, self.repository.name, since=since, state="all", ) for pr_data in prs: self._upsert_thread_from_pr(pr_data) self.stats.prs_synced += 1 except Exception as e: error_msg = f"Failed to sync pull requests: {e}" logger.error(error_msg) self.stats.errors.append(error_msg) return self.stats.prs_synced def sync_discussions(self, since: datetime | None = None) -> int: """Sync discussions from GitHub using GraphQL API. Args: since: Only sync discussions updated after this date Returns: Number of discussions synced """ logger.info(f"Syncing discussions for {self.repository.full_name}") try: discussions, _, _ = self.client.list_discussions( self.repository.owner, self.repository.name, ) for disc_data in discussions: # Filter by date if needed if since: updated_at = datetime.fromisoformat(disc_data["updatedAt"].replace("Z", "+00:00")) if updated_at < since: continue self._upsert_thread_from_discussion(disc_data) self.stats.discussions_synced += 1 except Exception as e: error_msg = f"Failed to sync discussions: {e}" logger.error(error_msg) self.stats.errors.append(error_msg) return self.stats.discussions_synced def sync_comments(self, thread: Thread, github_type: str) -> int: """Sync comments for a specific thread. Args: thread: The Thread to sync comments for github_type: Type of GitHub item ('issue', 'pull_request', 'discussion') Returns: Number of comments synced """ if not thread.metadata or "github" not in thread.metadata: return 0 github_meta = thread.metadata["github"] number = github_meta.get("number") if not number: return 0 count = 0 try: if github_type == "issue": comments = self.client.get_paginated( f"/repos/{self.repository.owner}/{self.repository.name}" f"/issues/{number}/comments" ) elif github_type == "pull_request": # Get both issue comments and review comments comments = self.client.get_paginated( f"/repos/{self.repository.owner}/{self.repository.name}" f"/issues/{number}/comments" ) review_comments = self.client.get_paginated( f"/repos/{self.repository.owner}/{self.repository.name}" f"/pulls/{number}/comments" ) comments.extend(review_comments) else: # Discussion comments are fetched via GraphQL - skip for now return 0 for comment in comments: self._upsert_message_from_comment(thread, comment) count += 1 # Update thread activity_at if count > 0: thread.activity_at = django_timezone.now() thread.save(update_fields=["activity_at", "updated_at"]) except Exception as e: logger.error(f"Failed to sync comments for thread {thread.pk}: {e}") self.stats.comments_synced += count return count def _upsert_thread_from_issue(self, issue: dict[str, Any]) -> Thread: """Create or update a Thread from GitHub issue data. Uses GitHubIssueMeta contract for type-safe metadata storage. See integrations/contracts.py for the metadata structure. """ external_id = f"issue:{issue['number']}" # Build typed metadata using GitHubIssueMeta contract author_meta: GitHubAuthorMeta = { "login": issue["user"]["login"], "id": issue["user"]["id"], "avatar_url": issue["user"]["avatar_url"], } labels: list[GitHubLabelMeta] = [ {"name": lbl["name"], "color": lbl["color"]} for lbl in issue.get("labels", []) ] github_meta: GitHubIssueMeta = { "type": "issue", "number": issue["number"], "state": issue["state"], "html_url": issue["html_url"], "labels": labels, "author": author_meta, } metadata: dict[str, Any] = {"github": github_meta} # Map GitHub state to internal status status = "closed" if issue["state"] == "closed" else "open" activity_at = datetime.fromisoformat(issue["updated_at"].replace("Z", "+00:00")) with transaction.atomic(): thread, created = Thread.objects.update_or_create( source=self.source, external_id=external_id, defaults={ "workspace": self.workspace, "title": issue["title"], "status": status, "metadata": metadata, "activity_at": activity_at, }, ) # Link author to member and update metadata with member ID if issue.get("user"): author = self.contributor_service.get_or_create_member_from_github_user(issue["user"]) # Type-safe: GitHubIssueMeta includes author_member_id as NotRequired github_data = cast(GitHubIssueMeta, thread.metadata.get("github", {})) github_data["author_member_id"] = str(author.pk) thread.metadata["github"] = github_data thread.save(update_fields=["metadata"]) # Create initial message from issue body if new if created and issue.get("body"): sent_at = datetime.fromisoformat(issue["created_at"].replace("Z", "+00:00")) msg_author: Member | None = None if issue.get("user"): msg_author = self.contributor_service.get_or_create_member_from_github_user(issue["user"]) Message.objects.create( thread=thread, external_id=f"issue:{issue['number']}:body", content=issue["body"], sent_at=sent_at, author=msg_author, metadata={"github": {"type": "issue_body"}}, ) return thread def _upsert_thread_from_pr(self, pr: dict[str, Any]) -> Thread: """Create or update a Thread from GitHub PR data. Uses GitHubPRMeta contract for type-safe metadata storage. See integrations/contracts.py for the metadata structure. """ external_id = f"pr:{pr['number']}" # Determine state if pr.get("merged"): state = "merged" elif pr["state"] == "closed": state = "closed" else: state = "open" # Build typed metadata using GitHubPRMeta contract author_meta: GitHubAuthorMeta = { "login": pr["user"]["login"], "id": pr["user"]["id"], "avatar_url": pr["user"]["avatar_url"], } labels: list[GitHubLabelMeta] = [{"name": lbl["name"], "color": lbl["color"]} for lbl in pr.get("labels", [])] github_meta: GitHubPRMeta = { "type": "pull_request", "number": pr["number"], "state": state, "html_url": pr["html_url"], "labels": labels, "author": author_meta, "draft": pr.get("draft", False), "merged": pr.get("merged", False), } # Add optional fields if present if pr.get("head", {}).get("ref"): github_meta["head_ref"] = pr["head"]["ref"] if pr.get("base", {}).get("ref"): github_meta["base_ref"] = pr["base"]["ref"] metadata: dict[str, Any] = {"github": github_meta} # Map to internal status status = "closed" if state in ("closed", "merged") else "open" activity_at = datetime.fromisoformat(pr["updated_at"].replace("Z", "+00:00")) with transaction.atomic(): thread, created = Thread.objects.update_or_create( source=self.source, external_id=external_id, defaults={ "workspace": self.workspace, "title": pr["title"], "status": status, "metadata": metadata, "activity_at": activity_at, }, ) # Link author to member and update metadata with member ID if pr.get("user"): author = self.contributor_service.get_or_create_member_from_github_user(pr["user"]) # Type-safe: GitHubPRMeta includes author_member_id as NotRequired github_data = cast(GitHubPRMeta, thread.metadata.get("github", {})) github_data["author_member_id"] = str(author.pk) thread.metadata["github"] = github_data thread.save(update_fields=["metadata"]) # Create initial message from PR body if new if created and pr.get("body"): sent_at = datetime.fromisoformat(pr["created_at"].replace("Z", "+00:00")) msg_author: Member | None = None if pr.get("user"): msg_author = self.contributor_service.get_or_create_member_from_github_user(pr["user"]) Message.objects.create( thread=thread, external_id=f"pr:{pr['number']}:body", content=pr["body"], sent_at=sent_at, author=msg_author, metadata={"github": {"type": "pr_body"}}, ) return thread def _upsert_thread_from_discussion(self, disc: dict[str, Any]) -> Thread: """Create or update a Thread from GitHub Discussion data. Uses GitHubDiscussionMeta contract for type-safe metadata storage. See integrations/contracts.py for the metadata structure. """ external_id = f"discussion:{disc['number']}" # Build typed metadata using GitHubDiscussionMeta contract # Note: Discussion author format differs from issues/PRs (uses databaseId, avatarUrl) author_meta: GitHubAuthorMeta = { "login": disc["author"]["login"], "id": disc["author"].get("databaseId", 0), } if disc["author"].get("avatarUrl"): author_meta["avatar_url"] = disc["author"]["avatarUrl"] labels: list[GitHubLabelMeta] = [ {"name": lbl["name"], "color": lbl.get("color", "")} for lbl in disc.get("labels", {}).get("nodes", []) ] github_meta: GitHubDiscussionMeta = { "type": "discussion", "number": disc["number"], "state": "answered" if disc.get("answer") else "open", "html_url": disc["url"], "labels": labels, "author": author_meta, "answered": disc.get("answer") is not None, } # Add optional category if present if disc.get("category", {}).get("name"): github_meta["category"] = disc["category"]["name"] metadata: dict[str, Any] = {"github": github_meta} status = "closed" if disc.get("answer") else "open" activity_at = datetime.fromisoformat(disc["updatedAt"].replace("Z", "+00:00")) with transaction.atomic(): thread, created = Thread.objects.update_or_create( source=self.source, external_id=external_id, defaults={ "workspace": self.workspace, "title": disc["title"], "status": status, "metadata": metadata, "activity_at": activity_at, }, ) # Link author to member (discussion author format is different) if disc.get("author") and disc["author"].get("databaseId"): author_data = { "id": disc["author"]["databaseId"], "login": disc["author"]["login"], "avatar_url": disc["author"].get("avatarUrl", ""), } author = self.contributor_service.get_or_create_member_from_github_user(author_data) # Type-safe: GitHubDiscussionMeta includes author_member_id as NotRequired github_data = cast(GitHubDiscussionMeta, thread.metadata.get("github", {})) github_data["author_member_id"] = str(author.pk) thread.metadata["github"] = github_data thread.save(update_fields=["metadata"]) # Create initial message from discussion body if new if created and disc.get("body"): sent_at = datetime.fromisoformat(disc["createdAt"].replace("Z", "+00:00")) msg_author: Member | None = None if disc.get("author") and disc["author"].get("databaseId"): author_data = { "id": disc["author"]["databaseId"], "login": disc["author"]["login"], "avatar_url": disc["author"].get("avatarUrl", ""), } msg_author = self.contributor_service.get_or_create_member_from_github_user(author_data) Message.objects.create( thread=thread, external_id=f"discussion:{disc['number']}:body", content=disc["body"], sent_at=sent_at, author=msg_author, metadata={"github": {"type": "discussion_body"}}, ) return thread def _upsert_message_from_comment(self, thread: Thread, comment: dict[str, Any]) -> Message: """Create or update a Message from a GitHub comment.""" external_id = f"comment:{comment['id']}" sent_at = datetime.fromisoformat(comment["created_at"].replace("Z", "+00:00")) metadata = { "github": { "type": "comment", "html_url": comment.get("html_url"), "author": { "login": comment["user"]["login"], "id": comment["user"]["id"], "avatar_url": comment["user"]["avatar_url"], }, } } # Get or create author member author: Member | None = None if comment.get("user"): author = self.contributor_service.get_or_create_member_from_github_user(comment["user"]) message, _ = Message.objects.update_or_create( thread=thread, external_id=external_id, defaults={ "content": comment["body"], "sent_at": sent_at, "author": author, "metadata": metadata, }, ) return message def _get_since_date(self, historical_days: int) -> datetime | None: """Get the 'since' date for incremental sync. When using last_sync_at, we subtract a 1-hour buffer to ensure we don't miss items that were updated just before the previous sync completed. This handles timing edge cases where an item is updated after the sync query runs but before last_sync_at is set. """ # Check if we have a last sync timestamp if self.repository.last_sync_at: # Subtract 1 hour buffer to catch any items we might have missed return self.repository.last_sync_at - timedelta(hours=1) # For initial sync, go back historical_days return django_timezone.now() - timedelta(days=historical_days) def _update_status(self, status: str, error: str | None = None) -> None: """Update repository sync status.""" self.repository.sync_status = status if error: self.repository.last_sync_error = error[:500] # Limit error length else: self.repository.last_sync_error = "" if status == "active": self.repository.last_sync_at = django_timezone.now() self.repository.save(update_fields=["sync_status", "last_sync_at", "last_sync_error", "updated_at"]) def _update_cursors(self) -> None: """Update sync cursors for incremental sync.""" self.repository.sync_cursors = { "last_sync_at": django_timezone.now().isoformat(), "issues_synced": self.stats.issues_synced, "prs_synced": self.stats.prs_synced, "discussions_synced": self.stats.discussions_synced, } self.repository.save(update_fields=["sync_cursors", "updated_at"])