"""RQ background tasks for GitHub and LinkedIn sync.""" import logging from datetime import timedelta from typing import Any import django_rq from django.db import transaction from django.utils import timezone from .models import GitHubRepository, LinkedInPage, LinkedInSyncStatus from .services.github_client import GitHubRateLimitError from .services.github_sync import GitHubSyncService logger = logging.getLogger(__name__) # Retry configuration MAX_RETRIES = 3 RETRY_DELAYS = [60, 300, 900] # 1 min, 5 min, 15 min # Content types for fan-out CONTENT_TYPES = ["issues", "prs", "discussions"] def sync_github_repository( repository_id: str, historical_days: int = 90, priority: str = "default", retry_count: int = 0, ) -> dict[str, Any]: """Orchestrate sync for a single GitHub repository. This task fans out to separate content-type tasks (issues, PRs, discussions) which run in parallel across workers. Args: repository_id: The GitHubRepository primary key historical_days: Number of days of history to sync on initial sync priority: 'high' for manual sync, 'default' for scheduled retry_count: Current retry attempt (for automatic retries) Returns: Dictionary with queued task info """ try: repository = GitHubRepository.objects.select_related("installation", "source__workspace").get(pk=repository_id) except GitHubRepository.DoesNotExist: logger.error(f"Repository {repository_id} not found") return {"success": False, "error": "Repository not found"} # Skip if archived or paused if repository.is_archived: logger.info(f"Skipping archived repository: {repository.full_name}") return {"success": False, "error": "Repository is archived"} if repository.sync_status == "paused": logger.info(f"Skipping paused repository: {repository.full_name}") return {"success": False, "error": "Repository is paused"} # Skip if installation is suspended if repository.installation.is_suspended: logger.warning(f"Skipping repository with suspended installation: {repository.full_name}") return {"success": False, "error": "GitHub installation is suspended"} # Skip if already syncing if repository.sync_status == "syncing": logger.info(f"Skipping {repository.full_name} - already syncing") return {"success": False, "error": "Already syncing"} logger.info(f"Starting sync orchestration for repository: {repository.full_name}") # Initialize sync state _initialize_sync_state(repository) # Fan out to content-type tasks queue = django_rq.get_queue(priority) queued_tasks = [] for content_type in CONTENT_TYPES: queue.enqueue( sync_github_content_type, repository_id=repository_id, content_type=content_type, historical_days=historical_days, priority=priority, retry_count=retry_count, ) queued_tasks.append(content_type) logger.info(f"Queued {len(queued_tasks)} content-type tasks for {repository.full_name}") return { "success": True, "repository": repository.full_name, "queued_tasks": queued_tasks, } def sync_github_content_type( repository_id: str, content_type: str, historical_days: int = 90, priority: str = "default", retry_count: int = 0, ) -> dict[str, Any]: """Sync a specific content type (issues, PRs, or discussions) for a repository. Args: repository_id: The GitHubRepository primary key content_type: One of 'issues', 'prs', 'discussions' historical_days: Number of days of history to sync priority: Queue priority for retries retry_count: Current retry attempt Returns: Dictionary with sync results """ try: repository = GitHubRepository.objects.select_related("installation", "source__workspace").get(pk=repository_id) except GitHubRepository.DoesNotExist: logger.error(f"Repository {repository_id} not found") return {"success": False, "error": "Repository not found"} logger.info(f"Syncing {content_type} for {repository.full_name}") try: sync_service = GitHubSyncService(repository) since = sync_service._get_since_date(historical_days) # Sync the specific content type if content_type == "issues": count = sync_service.sync_issues(since) elif content_type == "prs": count = sync_service.sync_pull_requests(since) elif content_type == "discussions": count = sync_service.sync_discussions(since) else: return {"success": False, "error": f"Unknown content type: {content_type}"} # Mark this content type as complete (also finalizes if all done) all_complete = _mark_content_type_complete(repository_id, content_type, count) return { "success": True, "repository": repository.full_name, "content_type": content_type, "count": count, "all_complete": all_complete, } except GitHubRateLimitError: return _handle_rate_limit( repository_id, content_type, historical_days, priority, retry_count, repository.full_name ) except Exception as e: logger.exception(f"Sync {content_type} failed for {repository.full_name}: {e}") if retry_count < MAX_RETRIES and _is_transient_error(e): return _handle_transient_error(repository_id, content_type, historical_days, priority, retry_count, str(e)) # Mark as error but still track completion _mark_content_type_error(repository_id, content_type, str(e)) return {"success": False, "error": str(e)} def _initialize_sync_state(repository: GitHubRepository) -> None: """Initialize sync state for a new sync run.""" repository.sync_status = "syncing" repository.sync_cursors = { "sync_started_at": timezone.now().isoformat(), "issues": {"status": "pending", "count": 0}, "prs": {"status": "pending", "count": 0}, "discussions": {"status": "pending", "count": 0}, } repository.last_sync_error = "" repository.save(update_fields=["sync_status", "sync_cursors", "last_sync_error", "updated_at"]) def _mark_content_type_complete(repository_id: str, content_type: str, count: int) -> bool: """Mark a content type as complete and finalize if all content types are done. Uses select_for_update to prevent race conditions. Calls _finalize_sync inside the same transaction to avoid race conditions. """ with transaction.atomic(): repository = GitHubRepository.objects.select_for_update().get(pk=repository_id) cursors = repository.sync_cursors or {} cursors[content_type] = {"status": "complete", "count": count} repository.sync_cursors = cursors repository.save(update_fields=["sync_cursors", "updated_at"]) # Check if all content types are complete and finalize within same transaction all_done = _all_content_types_done(cursors) if all_done: _finalize_sync_internal(repository) return all_done def _mark_content_type_error(repository_id: str, content_type: str, error: str) -> bool: """Mark a content type as errored and finalize if all content types are done.""" with transaction.atomic(): repository = GitHubRepository.objects.select_for_update().get(pk=repository_id) cursors = repository.sync_cursors or {} cursors[content_type] = {"status": "error", "error": error[:200]} repository.sync_cursors = cursors repository.save(update_fields=["sync_cursors", "updated_at"]) # Check if all content types are done (including errors) and finalize all_done = _all_content_types_done(cursors) if all_done: _finalize_sync_internal(repository) return all_done def _all_content_types_done(cursors: dict) -> bool: """Check if all content types have finished (complete or error).""" for content_type in CONTENT_TYPES: status = cursors.get(content_type, {}).get("status", "pending") if status == "pending": return False return True def _finalize_sync_internal(repository: GitHubRepository) -> None: """Finalize the sync after all content types complete. This internal version takes an already-locked repository object. Must be called within an existing transaction with select_for_update. """ cursors = repository.sync_cursors or {} # Check if any content type had errors has_errors = any(cursors.get(ct, {}).get("status") == "error" for ct in CONTENT_TYPES) if has_errors: repository.sync_status = "error" errors = [ f"{ct}: {cursors.get(ct, {}).get('error', 'unknown')}" for ct in CONTENT_TYPES if cursors.get(ct, {}).get("status") == "error" ] repository.last_sync_error = "; ".join(errors)[:500] else: repository.sync_status = "active" repository.last_sync_at = timezone.now() repository.last_sync_error = "" # Calculate totals total_issues = cursors.get("issues", {}).get("count", 0) total_prs = cursors.get("prs", {}).get("count", 0) total_discussions = cursors.get("discussions", {}).get("count", 0) cursors["sync_completed_at"] = timezone.now().isoformat() cursors["totals"] = { "issues": total_issues, "prs": total_prs, "discussions": total_discussions, } repository.sync_cursors = cursors repository.save(update_fields=["sync_status", "sync_cursors", "last_sync_at", "last_sync_error", "updated_at"]) total = total_issues + total_prs + total_discussions logger.info(f"Sync finalized for repository {repository.full_name}: {total} items synced") def _finalize_sync(repository_id: str) -> None: """Finalize the sync after all content types complete. This is the public version that acquires its own lock. Prefer _finalize_sync_internal when already holding a lock. """ with transaction.atomic(): repository = GitHubRepository.objects.select_for_update().get(pk=repository_id) _finalize_sync_internal(repository) def _handle_rate_limit( repository_id: str, content_type: str, historical_days: int, priority: str, retry_count: int, full_name: str, ) -> dict[str, Any]: """Handle rate limiting with exponential backoff.""" if retry_count < MAX_RETRIES: delay = RETRY_DELAYS[min(retry_count, len(RETRY_DELAYS) - 1)] logger.warning( f"Rate limited for {full_name} ({content_type}), retrying in {delay}s " f"(attempt {retry_count + 1}/{MAX_RETRIES})" ) queue = django_rq.get_queue("default") queue.enqueue_in( timedelta(seconds=delay), sync_github_content_type, repository_id=repository_id, content_type=content_type, historical_days=historical_days, priority=priority, retry_count=retry_count + 1, ) return { "success": False, "error": f"Rate limited, scheduled retry in {delay}s", "retry_scheduled": True, } else: logger.error(f"Max retries exceeded for {full_name} ({content_type}) due to rate limiting") _mark_content_type_error(repository_id, content_type, "Rate limit exceeded, max retries reached") return { "success": False, "error": "Rate limit exceeded, max retries reached", } def _handle_transient_error( repository_id: str, content_type: str, historical_days: int, priority: str, retry_count: int, error: str, ) -> dict[str, Any]: """Handle transient errors with exponential backoff.""" delay = RETRY_DELAYS[min(retry_count, len(RETRY_DELAYS) - 1)] logger.warning(f"Transient error for {content_type}, retrying in {delay}s") queue = django_rq.get_queue("default") queue.enqueue_in( timedelta(seconds=delay), sync_github_content_type, repository_id=repository_id, content_type=content_type, historical_days=historical_days, priority=priority, retry_count=retry_count + 1, ) return { "success": False, "error": error, "retry_scheduled": True, } def sync_all_github_repositories() -> dict[str, Any]: """Sync all active GitHub repositories. This task should be scheduled to run every 15 minutes. Returns: Dictionary with overall sync results """ logger.info("Starting sync for all GitHub repositories") # Get all active repositories repositories = GitHubRepository.objects.filter( is_archived=False, installation__suspended_at__isnull=True, ).exclude(sync_status__in=["paused", "disabled"]) queued_count = 0 skipped_count = 0 queue = django_rq.get_queue("default") for repo in repositories: # Skip if already syncing if repo.sync_status == "syncing": logger.debug(f"Skipping {repo.full_name} - already syncing") skipped_count += 1 continue # Queue the orchestrator job queue.enqueue( sync_github_repository, repository_id=str(repo.pk), historical_days=90, priority="default", ) queued_count += 1 logger.info(f"Queued {queued_count} repositories for sync, skipped {skipped_count}") return { "success": True, "queued": queued_count, "skipped": skipped_count, } def _is_transient_error(error: Exception) -> bool: """Check if an error is likely transient and worth retrying.""" transient_messages = [ "connection", "timeout", "temporarily", "502", "503", "504", ] error_str = str(error).lower() return any(msg in error_str for msg in transient_messages) # ============================================================================= # LinkedIn Sync Tasks # ============================================================================= def sync_linkedin_page(page_id: int, full_sync: bool = False, retry_count: int = 0) -> dict[str, Any]: """Sync a single LinkedIn page with exponential backoff on failure. Args: page_id: The LinkedInPage primary key. full_sync: If True, fetch all historical notifications (60 days). If False (default), fetch only new notifications since last sync. retry_count: Current retry attempt for exponential backoff. Returns: Dictionary with sync results or error info. """ from .exceptions import LinkedInAPIError, LinkedInRateLimitError, LinkedInTokenExpiredError from .services.linkedin_sync import LinkedInSyncService try: page = LinkedInPage.objects.get(pk=page_id) except LinkedInPage.DoesNotExist: logger.error(f"LinkedIn page {page_id} not found") return {"error": f"LinkedIn page {page_id} not found"} # Skip if archived if page.is_archived: logger.info(f"Skipping archived LinkedIn page: {page.name}") return {"skipped": True, "reason": "Page is archived"} # Skip if paused (but allow if this is a full sync/backfill) if page.sync_status == LinkedInSyncStatus.PAUSED and not full_sync: logger.info(f"Skipping paused LinkedIn page: {page.name}") return {"skipped": True, "reason": "Page sync is paused"} # Skip if token expired if page.sync_status == LinkedInSyncStatus.EXPIRED: logger.warning(f"Skipping expired LinkedIn page: {page.name}") return {"skipped": True, "reason": "Token expired"} sync_type = "full backfill" if full_sync else "incremental" logger.info(f"Starting {sync_type} sync for LinkedIn page: {page.name}") try: service = LinkedInSyncService(page) stats = service.sync_page(full_sync=full_sync) stats["full_sync"] = full_sync return stats except LinkedInRateLimitError as e: # Handle rate limiting with exponential backoff retry_after = getattr(e, "retry_after", None) return _handle_linkedin_retry( page_id, full_sync, retry_count, page.name, error_msg="Rate limited", delay_override=retry_after, ) except LinkedInTokenExpiredError as e: # Token refresh is handled in the sync service # If we get here, the refresh failed logger.error(f"Token expired and refresh failed for {page.name}: {e}") return {"error": "Token expired and refresh failed. Please reconnect."} except LinkedInAPIError as e: if _is_transient_error(e): return _handle_linkedin_retry( page_id, full_sync, retry_count, page.name, error_msg=str(e), ) logger.error(f"LinkedIn API error syncing page {page.name}: {e}") return {"error": str(e)} except Exception as e: if _is_transient_error(e): return _handle_linkedin_retry( page_id, full_sync, retry_count, page.name, error_msg=str(e), ) logger.exception(f"Unexpected error syncing LinkedIn page {page.name}: {e}") return {"error": f"Unexpected error: {str(e)}"} def _handle_linkedin_retry( page_id: int, full_sync: bool, retry_count: int, page_name: str, error_msg: str, delay_override: int | None = None, ) -> dict[str, Any]: """Handle LinkedIn sync retry with exponential backoff. Args: page_id: The LinkedInPage primary key. full_sync: Whether to perform a full sync on retry. retry_count: Current retry attempt. page_name: Page name for logging. error_msg: Error message to log. delay_override: Optional delay in seconds (e.g., from rate limit header). Returns: Dictionary with retry info. """ if retry_count >= MAX_RETRIES: logger.error(f"Max retries exceeded for LinkedIn page {page_name}: {error_msg}") return { "success": False, "error": f"Max retries exceeded: {error_msg}", } delay = delay_override or RETRY_DELAYS[min(retry_count, len(RETRY_DELAYS) - 1)] logger.warning( f"Transient error for LinkedIn page {page_name}, retrying in {delay}s " f"(attempt {retry_count + 1}/{MAX_RETRIES}): {error_msg}" ) queue = django_rq.get_queue("default") queue.enqueue_in( timedelta(seconds=delay), sync_linkedin_page, page_id=page_id, full_sync=full_sync, retry_count=retry_count + 1, ) return { "success": False, "error": error_msg, "retry_scheduled": True, "retry_in_seconds": delay, } def sync_all_linkedin_pages() -> dict[str, Any]: """Sync all LinkedIn pages that are due for sync. This task should be scheduled to run periodically (e.g., every 15 minutes). Returns: Dictionary with overall sync results. """ logger.info("Starting sync for all LinkedIn pages") now = timezone.now() # Get pages that are active and due for sync pages = LinkedInPage.objects.filter( is_archived=False, sync_status__in=[LinkedInSyncStatus.ACTIVE, LinkedInSyncStatus.PENDING], ) pages_queued = 0 pages_skipped = 0 queue = django_rq.get_queue("default") for page in pages: # Check if page is due for sync based on interval if page.last_sync_at: next_sync = page.last_sync_at + timedelta(minutes=page.sync_interval_minutes) if now < next_sync: logger.debug(f"Skipping {page.name} - not due until {next_sync}") pages_skipped += 1 continue # Queue the sync task queue.enqueue(sync_linkedin_page, page_id=page.pk) pages_queued += 1 logger.info(f"Queued {pages_queued} LinkedIn pages for sync, skipped {pages_skipped}") return { "pages_queued": pages_queued, "pages_skipped": pages_skipped, }