"""GraphQL types, queries, and mutations for GitHub integrations.""" from datetime import datetime from enum import Enum from typing import Annotated, cast from uuid import UUID import django_rq import strawberry from django.conf import settings from strawberry import ID from strawberry.types import Info from accounts.models import Workspace, WorkspaceMembership from members.models import Member from .models import GitHubInstallation, GitHubRepository, LinkedInPage, LinkedInSyncStatus # ============================================================================= # ENUMS # ============================================================================= @strawberry.enum class GitHubAccountType(Enum): """Type of GitHub account (user or organization).""" USER = "User" ORGANIZATION = "Organization" @strawberry.enum class GitHubRepositorySyncStatus(Enum): """Sync status for a GitHub repository.""" PENDING = "pending" SYNCING = "syncing" ACTIVE = "active" ERROR = "error" PAUSED = "paused" DISABLED = "disabled" @strawberry.enum class GitHubActivityType(Enum): """Type of GitHub activity.""" ISSUE = "issue" PULL_REQUEST = "pull_request" DISCUSSION = "discussion" @strawberry.enum class FeedItemTypeEnum(Enum): """Type of item in the unified activity feed.""" THREAD = "thread" SIGNAL = "signal" @strawberry.enum class ThreadTypeEnum(Enum): """Type of thread (conversation structure).""" CONVERSATION = "conversation" POST = "post" @strawberry.enum class SignalTypeEnum(Enum): """Type of signal (event structure).""" MENTION = "mention" STAR = "star" FORK = "fork" RELEASE = "release" FOLLOW = "follow" @strawberry.enum class LinkedInSyncStatusEnum(Enum): """Sync status for a LinkedIn page.""" PENDING = "PENDING" ACTIVE = "ACTIVE" SYNCING = "SYNCING" ERROR = "ERROR" PAUSED = "PAUSED" EXPIRED = "EXPIRED" # ============================================================================= # TYPES # ============================================================================= @strawberry.type class GitHubLabelType: """A GitHub label.""" name: str color: str @strawberry.type class GitHubActivityMetadataType: """GitHub-specific metadata for an activity item.""" type: GitHubActivityType number: int state: str html_url: str labels: list[GitHubLabelType] # PR-specific draft: bool | None = None merged: bool | None = None head_ref: str | None = None base_ref: str | None = None review_state: str | None = None # Discussion-specific category: str | None = None answered: bool | None = None @strawberry.type class GitHubInstallationType: """A GitHub App installation linked to a workspace.""" id: ID workspace_id: ID installation_id: str account_login: str account_type: GitHubAccountType account_avatar_url: str | None is_suspended: bool created_at: datetime updated_at: datetime @strawberry.field def repositories(self, info: Info) -> list["GitHubRepositoryType"]: """Get all repositories for this installation.""" # Get the installation from database try: installation = GitHubInstallation.objects.get(pk=self.id) repos = installation.repositories.filter(is_archived=False) return [GitHubRepositoryType.from_model(r) for r in repos] except GitHubInstallation.DoesNotExist: return [] @classmethod def from_model(cls, installation: GitHubInstallation) -> "GitHubInstallationType": return cls( id=ID(str(installation.pk)), workspace_id=ID(str(installation.workspace_id)), installation_id=str(installation.installation_id), account_login=installation.account_login, account_type=GitHubAccountType(installation.account_type), account_avatar_url=installation.account_avatar_url, is_suspended=installation.is_suspended, created_at=installation.created_at, updated_at=installation.updated_at, ) @strawberry.type class GitHubRepositoryType: """A GitHub repository being monitored.""" id: ID installation_id: ID source_id: ID repo_id: str owner: str name: str full_name: str private: bool sync_status: GitHubRepositorySyncStatus last_sync_at: datetime | None last_sync_error: str | None is_archived: bool created_at: datetime updated_at: datetime html_url: str @strawberry.field def item_count(self, info: Info) -> int: """Get the count of threads for this repository.""" try: repo = GitHubRepository.objects.get(pk=self.id) return repo.source.threads.count() except GitHubRepository.DoesNotExist: return 0 @classmethod def from_model(cls, repo: GitHubRepository) -> "GitHubRepositoryType": return cls( id=ID(str(repo.pk)), installation_id=ID(str(repo.installation_id)), source_id=ID(str(repo.source_id)), repo_id=str(repo.repo_id), owner=repo.owner, name=repo.name, full_name=repo.full_name, private=repo.private, sync_status=GitHubRepositorySyncStatus(repo.sync_status), last_sync_at=repo.last_sync_at, last_sync_error=repo.last_sync_error if repo.last_sync_error else None, is_archived=repo.is_archived, created_at=repo.created_at, updated_at=repo.updated_at, html_url=repo.html_url, ) @strawberry.type class AvailableGitHubRepositoryType: """A repository available for connection from GitHub App installation.""" repo_id: str owner: str name: str full_name: str private: bool description: str | None html_url: str is_connected: bool # ============================================================================= # LINKEDIN TYPES # ============================================================================= @strawberry.type class LinkedInPageType: """A connected LinkedIn organization page.""" id: ID workspace_id: ID source_id: ID | None organization_urn: str organization_id: str name: str vanity_name: str | None logo_url: str | None sync_status: LinkedInSyncStatusEnum sync_error: str | None last_sync_at: datetime | None next_sync_at: datetime | None sync_interval_minutes: int is_archived: bool is_token_expired: bool linkedin_url: str created_at: datetime updated_at: datetime @strawberry.field def item_count(self, info: Info) -> int: """Get the count of threads for this LinkedIn page.""" try: page = LinkedInPage.objects.get(pk=self.id) if page.source: return page.source.threads.count() return 0 except LinkedInPage.DoesNotExist: return 0 @classmethod def from_model(cls, page: LinkedInPage) -> "LinkedInPageType": return cls( id=ID(str(page.pk)), workspace_id=ID(str(page.workspace_id)), source_id=ID(str(page.source_id)) if page.source_id else None, organization_urn=page.organization_urn, organization_id=page.organization_id, name=page.name, vanity_name=page.vanity_name or None, logo_url=page.logo_url or None, sync_status=LinkedInSyncStatusEnum(page.sync_status), sync_error=page.sync_error or None, last_sync_at=page.last_sync_at, next_sync_at=page.next_sync_at, sync_interval_minutes=page.sync_interval_minutes, is_archived=page.is_archived, is_token_expired=page.is_token_expired, linkedin_url=page.linkedin_url, created_at=page.created_at, updated_at=page.updated_at, ) @strawberry.type class AvailableLinkedInOrganizationType: """A LinkedIn organization available for connection.""" organization_urn: str organization_id: str name: str vanity_name: str | None logo_url: str | None role: str is_connected: bool @strawberry.type class LinkedInConnectionErrorType: """Error details for LinkedIn connection failures.""" code: str message: str details: str | None = None @strawberry.type class PageInfoType: """Pagination info for connections.""" has_next_page: bool has_previous_page: bool start_cursor: str | None end_cursor: str | None @strawberry.type class ActivityFeedItemType: """A single item in the activity feed (based on Thread).""" id: ID source_id: ID source_name: str source_kind: str # e.g., "github_repo", "linkedin_page" external_id: str title: str content: str | None status: str activity_at: datetime created_at: datetime github: GitHubActivityMetadataType | None message_count: int # Feed-specific fields for unified view external_url: str | None activity_type: str | None actor_name: str | None actor_avatar_url: str | None @strawberry.field def author(self, info: Info) -> "ActivityAuthorType | None": """Get the author of this activity.""" # Would be populated from thread metadata return None @strawberry.type class ActivityAuthorType: """Author info for activity feed items.""" id: ID display_name: str avatar_url: str | None @classmethod def from_model(cls, member: Member) -> "ActivityAuthorType": return cls( id=ID(str(member.pk)), display_name=member.display_name, avatar_url=getattr(member, "avatar_url", None), ) # ============================================================================= # UNIFIED FEED TYPES (US4 - Type-Safe GraphQL API) # ============================================================================= @strawberry.type class GitHubSignalMetadataType: """GitHub-specific metadata for a signal event.""" # For stars: when the repo was starred starred_at: datetime | None = None # For forks: the full name of the fork fork_full_name: str | None = None # For releases: tag and release info tag_name: str | None = None release_name: str | None = None prerelease: bool | None = None @strawberry.type(name="FeedSignal") class FeedSignalType: """A signal event for the unified activity feed. Signals represent standalone events like mentions, stars, forks, releases. This type is used specifically in the unified feed context. """ id: ID workspace_id: ID source_id: ID signal_type: SignalTypeEnum occurred_at: datetime title: str external_url: str external_id: str | None body: str | None created_at: datetime updated_at: datetime # Resolved fields source_name: str source_kind: str # e.g., "github_repo", "linkedin_page" # Actor who triggered this signal (if known) actor_id: ID | None = None actor_name: str | None = None actor_avatar_url: str | None = None # GitHub-specific metadata github: GitHubSignalMetadataType | None = None @classmethod def from_model(cls, signal) -> "FeedSignalType": """Create a SignalType from a Signal model instance.""" # Extract GitHub metadata if present github_meta = None if signal.metadata and "github" in signal.metadata: gh = signal.metadata["github"] github_meta = GitHubSignalMetadataType( starred_at=gh.get("starred_at"), fork_full_name=gh.get("fork_full_name"), tag_name=gh.get("tag_name"), release_name=gh.get("release_name"), prerelease=gh.get("prerelease"), ) # Extract actor info actor_id = None actor_name = None actor_avatar_url = None if signal.actor: actor_id = ID(str(signal.actor.pk)) actor_name = signal.actor.display_name actor_avatar_url = getattr(signal.actor, "avatar_url", None) return cls( id=ID(str(signal.pk)), workspace_id=ID(str(signal.workspace_id)), source_id=ID(str(signal.source_id)), signal_type=SignalTypeEnum(signal.signal_type), occurred_at=signal.occurred_at, title=signal.title, external_url=signal.external_url, external_id=signal.external_id or None, body=signal.body or None, created_at=signal.created_at, updated_at=signal.updated_at, source_name=signal.source.name, source_kind=signal.source.kind, actor_id=actor_id, actor_name=actor_name, actor_avatar_url=actor_avatar_url, github=github_meta, ) @strawberry.type class SignalConnectionType: """Paginated signal results.""" items: list[FeedSignalType] page_info: PageInfoType total_count: int @strawberry.type(name="FeedThread") class FeedThreadType: """A conversation thread for the unified activity feed. Threads represent conversations with replies (issues, PRs, discussions). This type is used specifically in the unified feed context. """ id: ID workspace_id: ID source_id: ID external_id: str title: str status: str thread_type: ThreadTypeEnum activity_at: datetime | None created_at: datetime updated_at: datetime # Resolved fields source_name: str source_kind: str # e.g., "github_repo", "linkedin_page" message_count: int # GitHub-specific metadata github: GitHubActivityMetadataType | None = None @classmethod def from_model(cls, thread) -> "FeedThreadType": """Create a ThreadType from a Thread model instance.""" # Extract GitHub metadata if present github_meta = None if thread.metadata and "github" in thread.metadata: gh = thread.metadata["github"] github_meta = GitHubActivityMetadataType( type=GitHubActivityType(gh.get("type", "issue")), number=gh.get("number", 0), state=gh.get("state", "open"), html_url=gh.get("html_url", ""), labels=[ GitHubLabelType(name=lbl.get("name", ""), color=lbl.get("color", "")) for lbl in gh.get("labels", []) ], draft=gh.get("draft"), merged=gh.get("merged"), head_ref=gh.get("head_ref"), base_ref=gh.get("base_ref"), review_state=gh.get("review_state"), category=gh.get("category"), answered=gh.get("answered"), ) return cls( id=ID(str(thread.pk)), workspace_id=ID(str(thread.workspace_id)), source_id=ID(str(thread.source_id)), external_id=thread.external_id, title=thread.title, status=thread.status, thread_type=ThreadTypeEnum(thread.thread_type), activity_at=thread.activity_at, created_at=thread.created_at, updated_at=thread.updated_at, source_name=thread.source.name, source_kind=thread.source.kind, message_count=thread.messages.count(), github=github_meta, ) @strawberry.type class ThreadConnectionType: """Paginated thread results.""" items: list[FeedThreadType] page_info: PageInfoType total_count: int # Union type for feed content - either Thread or Signal FeedContentType = Annotated[FeedThreadType | FeedSignalType, strawberry.union("FeedContentType")] @strawberry.type class FeedItemType: """A single item in the unified activity feed.""" # Type for easy discrimination item_type: FeedItemTypeEnum # Timestamp for sorting (activity_at for threads, occurred_at for signals) activity_at: datetime # The actual content - either ThreadType or SignalType content: FeedContentType @strawberry.type class FeedConnectionType: """Connection type for paginated unified feed results.""" items: list[FeedItemType] page_info: PageInfoType total_count: int @strawberry.type class ActivityFeedConnectionType: """Paginated activity feed.""" items: list[ActivityFeedItemType] page_info: PageInfoType total_count: int # ============================================================================= # RESULT TYPES # ============================================================================= @strawberry.type class GitHubOAuthStartResult: """Result of starting GitHub OAuth flow.""" oauth_url: str state: str @strawberry.type class GitHubConnectRepoResult: """Result of connecting a repository.""" success: bool repository: GitHubRepositoryType | None = None error: str | None = None @strawberry.type class GitHubSyncResult: """Result of triggering a sync.""" success: bool items_synced: int | None = None error: str | None = None @strawberry.type class GitHubDisconnectResult: """Result of disconnecting.""" success: bool error: str | None = None # ============================================================================= # LINKEDIN RESULT TYPES # ============================================================================= @strawberry.type class LinkedInOAuthStartResult: """Result of starting LinkedIn OAuth flow.""" oauth_url: str state: str @strawberry.type class LinkedInConnectResult: """Result of connecting LinkedIn organizations.""" success: bool pages: list[LinkedInPageType] | None = None error: str | None = None @strawberry.type class LinkedInDisconnectResult: """Result of disconnecting a LinkedIn page.""" success: bool error: str | None = None @strawberry.type class LinkedInSyncResult: """Result of triggering a LinkedIn page sync.""" success: bool page: LinkedInPageType | None = None error: str | None = None @strawberry.type class LinkedInPauseResumeResult: """Result of pausing or resuming a LinkedIn page sync.""" success: bool page: LinkedInPageType | None = None error: str | None = None @strawberry.type class LinkedInSyncConfigResult: """Result of updating LinkedIn sync configuration.""" success: bool page: LinkedInPageType | None = None error: str | None = None # ============================================================================= # INPUT TYPES # ============================================================================= @strawberry.input class ActivityFeedFiltersInput: """Filters for the activity feed.""" source_ids: list[ID] | None = None activity_types: list[GitHubActivityType] | None = None author_ids: list[ID] | None = None status: str | None = None since_date: datetime | None = None until_date: datetime | None = None search_query: str | None = None @strawberry.input class ConnectGitHubRepoInput: """Input for connecting a GitHub repository.""" workspace_id: ID repo_id: str owner: str name: str full_name: str private: bool @strawberry.input class FeedFilterInput: """Filter options for the unified activity feed.""" # Filter by item type (thread, signal, or both) item_types: list[FeedItemTypeEnum] | None = None # Filter by thread types (when itemTypes includes THREAD) thread_types: list[ThreadTypeEnum] | None = None # Filter by signal types (when itemTypes includes SIGNAL) signal_types: list[SignalTypeEnum] | None = None # Filter by source IDs source_ids: list[ID] | None = None # Filter items after this date since: datetime | None = None # Filter items before this date until: datetime | None = None # ============================================================================= # HELPERS # ============================================================================= def check_workspace_access(user, workspace_id: ID, roles: list[str] | None = None) -> Workspace: """Check if user has access to workspace and return the workspace. Args: user: The authenticated user workspace_id: The workspace ID to check access for roles: Optional list of roles to require (e.g., ["owner", "admin"]) Returns: The Workspace object if access is granted Raises: Exception: If authentication or authorization fails """ if not user or not user.is_authenticated: raise Exception("Authentication required") from None try: workspace = Workspace.objects.get(pk=workspace_id) except Workspace.DoesNotExist: raise Exception("Workspace not found") from None membership_filter = {"user": user, "workspace": workspace} if roles: membership_filter["role__in"] = roles if not WorkspaceMembership.objects.filter(**membership_filter).exists(): raise Exception("Permission denied") from None return workspace def check_workspace_admin(user, workspace_id: ID) -> Workspace: """Check if user is a workspace admin (owner or admin role).""" return check_workspace_access(user, workspace_id, ["owner", "admin"]) # ============================================================================= # QUERIES # ============================================================================= @strawberry.type class IntegrationsQuery: """GraphQL queries for GitHub integrations.""" @strawberry.field def unified_activity_feed( self, info: Info, workspace_id: ID, first: int = 20, after: str | None = None, filters: FeedFilterInput | None = None, ) -> FeedConnectionType: """Get the unified activity feed combining threads and signals. Returns Thread and Signal items merged in chronological order (most recent first). Uses the FeedQueryService for efficient dual-cursor pagination. All workspace members can view this. """ from messages.models import Signal, Thread from messages.services.feed_query import FeedCursor, get_unified_feed check_workspace_access(info.context.user, workspace_id) # Parse cursor if provided cursor = None if after: try: cursor = FeedCursor.decode(after) except ValueError: pass # Invalid cursor, start from beginning # Convert filter input to service parameters item_types: list[str] | None = None thread_types: list[str] | None = None signal_types: list[str] | None = None source_ids: list[str | UUID] | None = None since: datetime | None = None until: datetime | None = None if filters: if filters.item_types: item_types = [t.value for t in filters.item_types] if filters.thread_types: thread_types = [t.value for t in filters.thread_types] if filters.signal_types: signal_types = [t.value for t in filters.signal_types] if filters.source_ids: source_ids = cast(list[str | UUID], [str(id) for id in filters.source_ids]) since = filters.since until = filters.until # Call the feed query service result = get_unified_feed( workspace_id=str(workspace_id), first=first, after=cursor, item_types=item_types, thread_types=thread_types, signal_types=signal_types, source_ids=source_ids, since=since, until=until, ) # Convert to GraphQL types feed_items: list[FeedItemType] = [] for item in result.items: content: FeedThreadType | FeedSignalType if isinstance(item, Thread): content = FeedThreadType.from_model(item) item_type = FeedItemTypeEnum.THREAD activity_at = item.activity_at or item.created_at else: # Signal content = FeedSignalType.from_model(item) item_type = FeedItemTypeEnum.SIGNAL activity_at = item.occurred_at feed_items.append( FeedItemType( item_type=item_type, activity_at=activity_at, content=content, ) ) # Build page info end_cursor = result.end_cursor.encode() if result.end_cursor else None page_info = PageInfoType( has_next_page=result.has_next_page, has_previous_page=after is not None, start_cursor=after, end_cursor=end_cursor, ) # Get total counts for both types (this is expensive but needed for total_count) # In production, you might want to cache this or make it optional thread_count = Thread.objects.filter(workspace_id=workspace_id).count() signal_count = Signal.objects.filter(workspace_id=workspace_id).count() return FeedConnectionType( items=feed_items, page_info=page_info, total_count=thread_count + signal_count, ) @strawberry.field def signals( self, info: Info, workspace_id: ID, first: int = 20, after: str | None = None, signal_types: list[SignalTypeEnum] | None = None, source_ids: list[ID] | None = None, ) -> SignalConnectionType: """Get signals for a workspace. Returns Signal items sorted by occurred_at descending. All workspace members can view this. """ import base64 from messages.models import Signal check_workspace_access(info.context.user, workspace_id) # Build queryset queryset = Signal.objects.filter(workspace_id=workspace_id).select_related("source", "actor") # Apply filters if signal_types: queryset = queryset.filter(signal_type__in=[t.value for t in signal_types]) if source_ids: queryset = queryset.filter(source_id__in=[str(id) for id in source_ids]) # Order by occurred_at descending queryset = queryset.order_by("-occurred_at") # Get total count total_count = queryset.count() # Apply cursor pagination if after: try: cursor_data = base64.b64decode(after).decode("utf-8") cursor_id = cursor_data.split(":")[1] queryset = queryset.filter(pk__lt=cursor_id) except (ValueError, IndexError): pass # Fetch one extra to check for next page items = list(queryset[: first + 1]) has_next_page = len(items) > first items = items[:first] # Convert to GraphQL types signal_items = [FeedSignalType.from_model(s) for s in items] # Build page info end_cursor = None if items: end_cursor = base64.b64encode(f"cursor:{items[-1].pk}".encode()).decode() page_info = PageInfoType( has_next_page=has_next_page, has_previous_page=after is not None, start_cursor=base64.b64encode(f"cursor:{items[0].pk}".encode()).decode() if items else None, end_cursor=end_cursor, ) return SignalConnectionType( items=signal_items, page_info=page_info, total_count=total_count, ) @strawberry.field def activity_feed( self, info: Info, workspace_id: ID, first: int = 20, after: str | None = None, filters: ActivityFeedFiltersInput | None = None, ) -> ActivityFeedConnectionType: """Get paginated activity feed for a workspace. Returns Threads sorted by activity_at descending. All workspace members can view this. """ import base64 from messages.models import Thread check_workspace_access(info.context.user, workspace_id) # Start with all threads for the workspace queryset = ( Thread.objects.filter(workspace_id=workspace_id).select_related("source").prefetch_related("messages") ) # Apply filters if provided if filters: if filters.source_ids: queryset = queryset.filter(source_id__in=[str(id) for id in filters.source_ids]) if filters.status: queryset = queryset.filter(status=filters.status) if filters.since_date: queryset = queryset.filter(activity_at__gte=filters.since_date) if filters.until_date: queryset = queryset.filter(activity_at__lte=filters.until_date) if filters.activity_types: type_filters = [t.value for t in filters.activity_types] queryset = queryset.filter(metadata__github__type__in=type_filters) if filters.search_query: queryset = queryset.filter(title__icontains=filters.search_query) # Order by activity_at descending (most recent first) queryset = queryset.order_by("-activity_at", "-created_at") # Get total count before pagination total_count = queryset.count() # Apply cursor-based pagination if after: try: cursor_data = base64.b64decode(after).decode("utf-8") cursor_id = int(cursor_data.split(":")[1]) queryset = queryset.filter(pk__lt=cursor_id) except (ValueError, IndexError): pass # Fetch one extra to check for next page items = list(queryset[: first + 1]) has_next_page = len(items) > first items = items[:first] # Build activity feed items feed_items = [] for thread in items: github_meta = None if thread.metadata and "github" in thread.metadata: gh = thread.metadata["github"] github_meta = GitHubActivityMetadataType( type=GitHubActivityType(gh.get("type", "issue")), number=gh.get("number", 0), state=gh.get("state", "open"), html_url=gh.get("html_url", ""), labels=[ GitHubLabelType(name=lbl.get("name", ""), color=lbl.get("color", "")) for lbl in gh.get("labels", []) ], draft=gh.get("draft"), merged=gh.get("merged"), head_ref=gh.get("head_ref"), base_ref=gh.get("base_ref"), review_state=gh.get("review_state"), category=gh.get("category"), answered=gh.get("answered"), ) # Get first message content as preview first_message = thread.messages.first() content = first_message.content if first_message else None # Extract author info from metadata metadata = thread.metadata or {} author_info = metadata.get("author", {}) or metadata.get("github", {}).get("author", {}) github_data = metadata.get("github", {}) # Get source kind with fallback source_kind_value = thread.source.kind if thread.source else "github_repo" feed_items.append( ActivityFeedItemType( id=ID(str(thread.pk)), source_id=ID(str(thread.source_id)), source_name=thread.source.name, source_kind=source_kind_value, external_id=thread.external_id, title=thread.title, content=content[:500] if content else None, status=thread.status, activity_at=thread.activity_at or thread.created_at, created_at=thread.created_at, github=github_meta, message_count=thread.messages.count(), # Feed-specific fields external_url=metadata.get("html_url") or github_data.get("html_url"), activity_type=metadata.get("type") or github_data.get("type"), actor_name=author_info.get("login"), actor_avatar_url=author_info.get("avatar_url"), ) ) # Build page info end_cursor = None if items: end_cursor = base64.b64encode(f"cursor:{items[-1].pk}".encode()).decode() page_info = PageInfoType( has_next_page=has_next_page, has_previous_page=after is not None, start_cursor=base64.b64encode(f"cursor:{items[0].pk}".encode()).decode() if items else None, end_cursor=end_cursor, ) return ActivityFeedConnectionType( items=feed_items, page_info=page_info, total_count=total_count, ) @strawberry.field def activity_item(self, info: Info, workspace_id: ID, item_id: ID) -> ActivityFeedItemType | None: """Get a single activity item by ID. All workspace members can view this. """ from messages.models import Thread check_workspace_access(info.context.user, workspace_id) try: thread = ( Thread.objects.select_related("source") .prefetch_related("messages") .get(pk=item_id, workspace_id=workspace_id) ) except Thread.DoesNotExist: return None github_meta = None if thread.metadata and "github" in thread.metadata: gh = thread.metadata["github"] github_meta = GitHubActivityMetadataType( type=GitHubActivityType(gh.get("type", "issue")), number=gh.get("number", 0), state=gh.get("state", "open"), html_url=gh.get("html_url", ""), labels=[ GitHubLabelType(name=lbl.get("name", ""), color=lbl.get("color", "")) for lbl in gh.get("labels", []) ], draft=gh.get("draft"), merged=gh.get("merged"), head_ref=gh.get("head_ref"), base_ref=gh.get("base_ref"), review_state=gh.get("review_state"), category=gh.get("category"), answered=gh.get("answered"), ) first_message = thread.messages.first() content = first_message.content if first_message else None # Extract author info from metadata metadata = thread.metadata or {} author_info = metadata.get("author", {}) or metadata.get("github", {}).get("author", {}) github_data = metadata.get("github", {}) # Get source kind with fallback source_kind_value = thread.source.kind if thread.source else "github_repo" return ActivityFeedItemType( id=ID(str(thread.pk)), source_id=ID(str(thread.source_id)), source_name=thread.source.name, source_kind=source_kind_value, external_id=thread.external_id, title=thread.title, content=content, status=thread.status, activity_at=thread.activity_at or thread.created_at, created_at=thread.created_at, github=github_meta, message_count=thread.messages.count(), # Feed-specific fields external_url=metadata.get("html_url") or github_data.get("html_url"), activity_type=metadata.get("type") or github_data.get("type"), actor_name=author_info.get("login"), actor_avatar_url=author_info.get("avatar_url"), ) @strawberry.field def github_installation(self, info: Info, workspace_id: ID) -> GitHubInstallationType | None: """Get the GitHub installation for a workspace. Returns None if no installation exists. Only workspace admins can query this. """ try: check_workspace_admin(info.context.user, workspace_id) except Exception: return None try: installation = GitHubInstallation.objects.get(workspace_id=workspace_id) return GitHubInstallationType.from_model(installation) except GitHubInstallation.DoesNotExist: return None @strawberry.field def github_repositories(self, info: Info, workspace_id: ID) -> list[GitHubRepositoryType]: """List connected GitHub repositories for a workspace. All workspace members can view this. """ user = info.context.user if not user or not user.is_authenticated: return [] has_access = WorkspaceMembership.objects.filter(user=user, workspace_id=workspace_id).exists() if not has_access: return [] try: installation = GitHubInstallation.objects.get(workspace_id=workspace_id) repos = installation.repositories.filter(is_archived=False) return [GitHubRepositoryType.from_model(r) for r in repos] except GitHubInstallation.DoesNotExist: return [] @strawberry.field def available_github_repos(self, info: Info, workspace_id: ID) -> list[AvailableGitHubRepositoryType]: """List available repositories from GitHub App installation. Only workspace admins can query this. This calls the GitHub API to list accessible repos. """ check_workspace_admin(info.context.user, workspace_id) try: installation = GitHubInstallation.objects.get(workspace_id=workspace_id) except GitHubInstallation.DoesNotExist: raise Exception("No GitHub installation found. Please connect GitHub first.") from None # Get connected repo IDs for comparison connected_repo_ids = set(installation.repositories.filter(is_archived=False).values_list("repo_id", flat=True)) # Call GitHub API to list repos from .services.github_client import GitHubClient client = GitHubClient(installation) try: repos = client.list_installation_repositories() except Exception as e: raise Exception(f"Failed to fetch repositories from GitHub: {e}") from None return [ AvailableGitHubRepositoryType( repo_id=str(repo["id"]), owner=repo["owner"]["login"], name=repo["name"], full_name=repo["full_name"], private=repo["private"], description=repo.get("description"), html_url=repo["html_url"], is_connected=repo["id"] in connected_repo_ids, ) for repo in repos ] # ========================================================================= # LinkedIn Queries # ========================================================================= @strawberry.field def linkedin_pages(self, info: Info, workspace_id: ID) -> list[LinkedInPageType]: """List connected LinkedIn pages for a workspace. All workspace members can view this. """ check_workspace_access(info.context.user, workspace_id) pages = LinkedInPage.objects.filter( workspace_id=workspace_id, is_archived=False, ) return [LinkedInPageType.from_model(p) for p in pages] @strawberry.field def available_linkedin_organizations(self, info: Info, workspace_id: ID) -> list[AvailableLinkedInOrganizationType]: """List organizations the user can connect from LinkedIn. Requires completing OAuth flow first (tokens cached). Only workspace admins can query this. """ from django.core.cache import cache from .services.linkedin_client import LinkedInClient check_workspace_admin(info.context.user, workspace_id) # Get cached OAuth tokens cache_key = f"linkedin_oauth_{workspace_id}" cached_tokens = cache.get(cache_key) if not cached_tokens: raise Exception("No LinkedIn authorization found. Please authorize LinkedIn first.") from None # Get organizations from LinkedIn API client = LinkedInClient(access_token=cached_tokens["access_token"]) orgs = client.get_organizations() # Get already connected org URNs connected_urns = set( LinkedInPage.objects.filter( workspace_id=workspace_id, is_archived=False, ).values_list("organization_urn", flat=True) ) return [ AvailableLinkedInOrganizationType( organization_urn=org.organization_urn, organization_id=org.organization_id, name=org.name, vanity_name=org.vanity_name, logo_url=org.logo_url, role=org.role, is_connected=org.organization_urn in connected_urns, ) for org in orgs ] # ============================================================================= # MUTATIONS # ============================================================================= @strawberry.type class IntegrationsMutation: """GraphQL mutations for GitHub integrations.""" @strawberry.mutation def github_start_oauth(self, info: Info, workspace_id: ID) -> GitHubOAuthStartResult: """Start GitHub OAuth flow. Returns a URL to redirect the user to for GitHub App installation. Only workspace admins can initiate this. """ import secrets from django.conf import settings from django.core.signing import TimestampSigner check_workspace_admin(info.context.user, workspace_id) # Generate state that includes workspace_id (signed for tamper protection) # Format: {random_token}:{signed_workspace_id} # This avoids session/cookie issues in cross-origin setups random_token = secrets.token_urlsafe(16) signer = TimestampSigner() signed_workspace_id = signer.sign(str(workspace_id)) state = f"{random_token}:{signed_workspace_id}" # Build the GitHub App installation URL app_slug = settings.GITHUB_APP_SLUG if not app_slug: raise Exception("GitHub App not configured") from None oauth_url = f"https://github.com/apps/{app_slug}/installations/new" f"?state={state}" return GitHubOAuthStartResult(oauth_url=oauth_url, state=state) @strawberry.mutation def github_disconnect_installation(self, info: Info, workspace_id: ID) -> GitHubDisconnectResult: """Disconnect the entire GitHub installation from a workspace. This archives all connected repositories but retains the synced data. Only workspace admins can do this. """ check_workspace_admin(info.context.user, workspace_id) try: installation = GitHubInstallation.objects.get(workspace_id=workspace_id) except GitHubInstallation.DoesNotExist: return GitHubDisconnectResult(success=False, error="No GitHub installation found") # Archive all repositories installation.repositories.update(is_archived=True, sync_status="disabled") # Delete the installation installation.delete() return GitHubDisconnectResult(success=True) @strawberry.mutation def github_connect_repo(self, info: Info, input: ConnectGitHubRepoInput) -> GitHubConnectRepoResult: """Connect a repository from the GitHub App installation. Creates a Source record and GitHubRepository record. Only workspace admins can do this. """ from sources.models import Source workspace = check_workspace_admin(info.context.user, input.workspace_id) try: installation = GitHubInstallation.objects.get(workspace=workspace) except GitHubInstallation.DoesNotExist: return GitHubConnectRepoResult(success=False, error="No GitHub installation found") # Check if repo is already connected if GitHubRepository.objects.filter(installation=installation, repo_id=int(input.repo_id)).exists(): return GitHubConnectRepoResult(success=False, error="Repository already connected") # Create Source record source = Source.objects.create( workspace=workspace, name=input.full_name, kind="github_repo", external_id=input.repo_id, metadata={"owner": input.owner, "name": input.name}, ) # Create GitHubRepository record repo = GitHubRepository.objects.create( installation=installation, source=source, repo_id=int(input.repo_id), owner=input.owner, name=input.name, full_name=input.full_name, private=input.private, sync_status="pending", ) # Queue initial sync import django_rq from .tasks import sync_github_repository queue = django_rq.get_queue("default") queue.enqueue( sync_github_repository, repository_id=str(repo.pk), historical_days=90, priority="default", ) return GitHubConnectRepoResult(success=True, repository=GitHubRepositoryType.from_model(repo)) @strawberry.mutation def github_disconnect_repo(self, info: Info, repository_id: ID) -> GitHubDisconnectResult: """Disconnect a repository. Archives the repository but retains the synced data. Only workspace admins can do this. """ try: repo = GitHubRepository.objects.select_related("installation__workspace").get(pk=repository_id) except GitHubRepository.DoesNotExist: return GitHubDisconnectResult(success=False, error="Repository not found") check_workspace_admin(info.context.user, ID(str(repo.installation.workspace_id))) # Archive the repository repo.is_archived = True repo.sync_status = "disabled" repo.save(update_fields=["is_archived", "sync_status", "updated_at"]) return GitHubDisconnectResult(success=True) @strawberry.mutation def github_sync_repo(self, info: Info, repository_id: ID) -> GitHubSyncResult: """Trigger an immediate sync for a repository. Only workspace admins can do this. """ try: repo = GitHubRepository.objects.select_related("installation__workspace").get(pk=repository_id) except GitHubRepository.DoesNotExist: return GitHubSyncResult(success=False, error="Repository not found") check_workspace_admin(info.context.user, ID(str(repo.installation.workspace_id))) if repo.is_archived: return GitHubSyncResult(success=False, error="Repository is archived") if repo.installation.is_suspended: return GitHubSyncResult(success=False, error="GitHub installation is suspended") # Queue immediate sync with high priority # Note: Don't set sync_status here - the worker's _initialize_sync_state handles it # Setting it here would cause a race condition where the worker sees "syncing" and skips import django_rq from .tasks import sync_github_repository queue = django_rq.get_queue("high") queue.enqueue( sync_github_repository, repository_id=str(repo.pk), historical_days=90, priority="high", ) return GitHubSyncResult(success=True, items_synced=0) @strawberry.mutation def github_set_repo_sync_paused(self, info: Info, repository_id: ID, paused: bool) -> GitHubRepositoryType: """Pause or resume sync for a repository. Only workspace admins can do this. """ try: repo = GitHubRepository.objects.select_related("installation__workspace").get(pk=repository_id) except GitHubRepository.DoesNotExist: raise Exception("Repository not found") from None check_workspace_admin(info.context.user, ID(str(repo.installation.workspace_id))) if paused: repo.sync_status = "paused" else: repo.sync_status = "active" if repo.last_sync_at else "pending" repo.save(update_fields=["sync_status", "updated_at"]) return GitHubRepositoryType.from_model(repo) # ========================================================================= # LinkedIn Mutations # ========================================================================= @strawberry.mutation def initiate_linkedin_oauth(self, info: Info, workspace_id: ID) -> LinkedInOAuthStartResult: """Start LinkedIn OAuth flow. Returns a URL to redirect the user to for LinkedIn authorization. Only workspace admins can initiate this. """ from .services.linkedin_client import LinkedInClient check_workspace_admin(info.context.user, workspace_id) client = LinkedInClient() state = client.generate_state(workspace_id=str(workspace_id)) oauth_url = client.get_authorization_url(state=state) return LinkedInOAuthStartResult(oauth_url=oauth_url, state=state) @strawberry.mutation def connect_linkedin_organizations( self, info: Info, workspace_id: ID, organization_urns: list[str], ) -> LinkedInConnectResult: """Connect selected LinkedIn organizations. Creates LinkedInPage and Source records for each organization. Only workspace admins can do this. """ from django.core.cache import cache from django.utils import timezone from sources.models import Source from .services.linkedin_client import LinkedInClient workspace = check_workspace_admin(info.context.user, workspace_id) # Get cached OAuth tokens cache_key = f"linkedin_oauth_{workspace_id}" cached_tokens = cache.get(cache_key) if not cached_tokens: return LinkedInConnectResult( success=False, error="No LinkedIn authorization found. Please authorize LinkedIn first.", ) # Fetch organizations from LinkedIn to validate client = LinkedInClient(access_token=cached_tokens["access_token"]) available_orgs = {org.organization_urn: org for org in client.get_organizations()} # Get already connected org URNs connected_urns = set( LinkedInPage.objects.filter( workspace=workspace, is_archived=False, ).values_list("organization_urn", flat=True) ) created_pages = [] for org_urn in organization_urns: # Skip if already connected if org_urn in connected_urns: continue # Validate user has access to this org org = available_orgs.get(org_urn) if not org: return LinkedInConnectResult( success=False, error=f"Organization {org_urn} not found or not authorized.", ) # Create Source record source = Source.objects.create( workspace=workspace, name=org.name, kind="linkedin_page", external_id=org.organization_urn, metadata={ "organization_id": org.organization_id, "vanity_name": org.vanity_name, }, ) # Create LinkedInPage record page = LinkedInPage.objects.create( workspace=workspace, source=source, organization_urn=org.organization_urn, organization_id=org.organization_id, name=org.name, vanity_name=org.vanity_name or "", logo_url=org.logo_url or "", authorized_user_urn="", # Will be set from person URN if available sync_status=LinkedInSyncStatus.PENDING, sync_interval_minutes=settings.LINKEDIN_SYNC_INTERVAL_MINUTES, ) # Set encrypted tokens page.access_token = cached_tokens["access_token"] page.refresh_token = cached_tokens.get("refresh_token", "") # Set token expiration expires_in = cached_tokens.get("expires_in", 5184000) page.token_expires_at = timezone.now() + timezone.timedelta(seconds=expires_in) page.save() created_pages.append(page) # Clear cached tokens after successful connection cache.delete(cache_key) # Trigger initial full sync (backfill) for each new page from .tasks import sync_linkedin_page queue = django_rq.get_queue("default") for page in created_pages: # Set status to SYNCING to indicate backfill is pending page.sync_status = LinkedInSyncStatus.SYNCING page.save(update_fields=["sync_status", "updated_at"]) # Enqueue full sync with backfill (full_sync=True for 60-day history) queue.enqueue(sync_linkedin_page, page_id=page.pk, full_sync=True) return LinkedInConnectResult( success=True, pages=[LinkedInPageType.from_model(p) for p in created_pages], ) @strawberry.mutation def disconnect_linkedin_page(self, info: Info, page_id: ID) -> LinkedInDisconnectResult: """Disconnect a LinkedIn page. Archives the page but retains the synced data. Only workspace admins can do this. """ try: page = LinkedInPage.objects.select_related("workspace").get(pk=page_id) except LinkedInPage.DoesNotExist: return LinkedInDisconnectResult(success=False, error="Page not found") check_workspace_admin(info.context.user, ID(str(page.workspace_id))) # Archive the page and pause sync page.is_archived = True page.sync_status = LinkedInSyncStatus.PAUSED page.save(update_fields=["is_archived", "sync_status", "updated_at"]) return LinkedInDisconnectResult(success=True) @strawberry.mutation def sync_linkedin_page(self, info: Info, page_id: ID) -> LinkedInSyncResult: """Manually trigger a sync for a LinkedIn page. Sets status to SYNCING and enqueues the sync task. Implements rate limiting to prevent abuse. Only workspace admins can do this. """ from django.core.cache import cache from .tasks import sync_linkedin_page as sync_task try: page = LinkedInPage.objects.select_related("workspace").get(pk=page_id) except LinkedInPage.DoesNotExist: return LinkedInSyncResult(success=False, error="Page not found") check_workspace_admin(info.context.user, ID(str(page.workspace_id))) # Rate limit: prevent sync more than once per minute per page rate_limit_key = f"linkedin_sync_rate_limit_{page_id}" if cache.get(rate_limit_key): return LinkedInSyncResult( success=False, error="Sync was triggered recently. Please wait a moment before syncing again.", ) # Skip if page is archived or has expired token if page.is_archived: return LinkedInSyncResult(success=False, error="Cannot sync archived page") if page.sync_status == LinkedInSyncStatus.EXPIRED: return LinkedInSyncResult(success=False, error="Token expired. Please reconnect the page.") # Set status to SYNCING page.sync_status = LinkedInSyncStatus.SYNCING page.sync_error = "" page.save(update_fields=["sync_status", "sync_error", "updated_at"]) # Enqueue sync task queue = django_rq.get_queue("default") queue.enqueue(sync_task, page_id=page.pk, full_sync=False) # Set rate limit (1 minute) cache.set(rate_limit_key, True, timeout=60) return LinkedInSyncResult( success=True, page=LinkedInPageType.from_model(page), ) @strawberry.mutation def update_linkedin_sync_config( self, info: Info, page_id: ID, sync_interval_minutes: int, ) -> LinkedInSyncConfigResult: """Update sync configuration for a LinkedIn page. Allows changing the sync interval. Only workspace admins can do this. """ try: page = LinkedInPage.objects.select_related("workspace").get(pk=page_id) except LinkedInPage.DoesNotExist: return LinkedInSyncConfigResult(success=False, error="Page not found") check_workspace_admin(info.context.user, ID(str(page.workspace_id))) # Validate minimum sync interval (15 minutes) min_interval = 15 if sync_interval_minutes < min_interval: return LinkedInSyncConfigResult( success=False, error=f"Minimum sync interval is {min_interval} minutes", ) # Update interval page.sync_interval_minutes = sync_interval_minutes page.save(update_fields=["sync_interval_minutes", "updated_at"]) return LinkedInSyncConfigResult( success=True, page=LinkedInPageType.from_model(page), ) @strawberry.mutation def pause_linkedin_page(self, info: Info, page_id: ID) -> LinkedInPauseResumeResult: """Pause syncing for a LinkedIn page. Sets status to PAUSED. The page will not sync until resumed. Only workspace admins can do this. """ try: page = LinkedInPage.objects.select_related("workspace").get(pk=page_id) except LinkedInPage.DoesNotExist: return LinkedInPauseResumeResult(success=False, error="Page not found") check_workspace_admin(info.context.user, ID(str(page.workspace_id))) # Skip if already paused if page.sync_status == LinkedInSyncStatus.PAUSED: return LinkedInPauseResumeResult( success=True, page=LinkedInPageType.from_model(page), ) # Pause syncing page.sync_status = LinkedInSyncStatus.PAUSED page.save(update_fields=["sync_status", "updated_at"]) return LinkedInPauseResumeResult( success=True, page=LinkedInPageType.from_model(page), ) @strawberry.mutation def resume_linkedin_page(self, info: Info, page_id: ID) -> LinkedInPauseResumeResult: """Resume syncing for a LinkedIn page. Sets status to ACTIVE and clears any previous error. Only workspace admins can do this. """ try: page = LinkedInPage.objects.select_related("workspace").get(pk=page_id) except LinkedInPage.DoesNotExist: return LinkedInPauseResumeResult(success=False, error="Page not found") check_workspace_admin(info.context.user, ID(str(page.workspace_id))) # Resume syncing and clear errors page.sync_status = LinkedInSyncStatus.ACTIVE page.sync_error = "" page.save(update_fields=["sync_status", "sync_error", "updated_at"]) return LinkedInPauseResumeResult( success=True, page=LinkedInPageType.from_model(page), )