""" Search Index Worker. Processes search index queue items with retry and dead letter support. """ import logging import traceback from datetime import timedelta from typing import Any, Callable, Dict, Optional from django.db import transaction from django.utils import timezone from .models import ( SearchIndexQueue, SearchIndexDeadLetter, QueueStatus, IndexOperation, ) logger = logging.getLogger(__name__) class SearchIndexWorker: """ Worker that processes search index queue items. Features: - Picks items from the queue ordered by priority - Indexes entities using registered indexers - Retries with exponential backoff on failure - Moves items to DLQ after max retries Example: worker = SearchIndexWorker() worker.register_indexer('task', task_indexer_func) worker.register_indexer('part', part_indexer_func) # Process queue processed = worker.process_batch(batch_size=50) """ # Registry of entity type → indexer function _indexers: Dict[str, Callable] = {} @classmethod def register_indexer( cls, entity_type: str, indexer: Callable[[str, Dict[str, Any], str], None], ) -> None: """ Register an indexer function for an entity type. The indexer function signature: def indexer(entity_id: str, payload: dict, operation: str) -> None Args: entity_type: Entity type string (e.g., 'task', 'part') indexer: Function that performs the indexing """ cls._indexers[entity_type] = indexer logger.info(f"Registered search indexer for '{entity_type}'") @classmethod def process_batch(cls, batch_size: int = 50) -> int: """ Process a batch of pending queue items. Args: batch_size: Maximum number of items to process Returns: Number of items processed """ now = timezone.now() # Get pending items, respecting retry timing items = SearchIndexQueue.objects.filter( status__in=[QueueStatus.PENDING, QueueStatus.FAILED], ).filter( # Either no retry scheduled or retry time has passed models_q_next_retry(now) ).order_by('-priority', 'created_at')[:batch_size] processed = 0 for item in items: cls._process_item(item) processed += 1 if processed: logger.info(f"Processed {processed} search index queue items") return processed @classmethod @transaction.atomic def _process_item(cls, item: SearchIndexQueue) -> None: """Process a single queue item.""" item.status = QueueStatus.PROCESSING item.save(update_fields=['status']) indexer = cls._indexers.get(item.entity_type) if not indexer: logger.warning( f"No indexer registered for entity type '{item.entity_type}'" ) item.error_message = f"No indexer for '{item.entity_type}'" item.status = QueueStatus.FAILED item.save(update_fields=['status', 'error_message']) return try: indexer( str(item.entity_id), item.payload, item.operation, ) item.status = QueueStatus.COMPLETED item.processed_at = timezone.now() item.save(update_fields=['status', 'processed_at']) logger.debug( f"Indexed {item.entity_type}:{item.entity_id} " f"({item.operation})" ) except Exception as e: item.retry_count += 1 item.error_message = str(e) if item.retry_count >= item.max_retries: # Move to Dead Letter Queue cls._move_to_dlq(item, e) else: # Schedule retry with exponential backoff backoff = min(2 ** item.retry_count * 30, 3600) # max 1 hour item.next_retry_at = timezone.now() + timedelta(seconds=backoff) item.status = QueueStatus.FAILED item.save(update_fields=[ 'status', 'retry_count', 'error_message', 'next_retry_at' ]) logger.warning( f"Search index failed for {item.entity_type}:{item.entity_id}, " f"retry {item.retry_count}/{item.max_retries} " f"scheduled at {item.next_retry_at}" ) @classmethod def _move_to_dlq(cls, item: SearchIndexQueue, error: Exception) -> None: """Move a failed item to the Dead Letter Queue.""" SearchIndexDeadLetter.objects.create( tenant=item.tenant, entity_type=item.entity_type, entity_id=item.entity_id, operation=item.operation, payload=item.payload, retry_count=item.retry_count, error_message=str(error), stack_trace=traceback.format_exc(), original_created_at=item.created_at, ) # Remove from main queue item.delete() logger.error( f"Moved to DLQ: {item.entity_type}:{item.entity_id} " f"after {item.retry_count} retries: {error}" ) @classmethod def requeue_from_dlq(cls, dlq_id) -> Optional[SearchIndexQueue]: """ Re-queue an item from the Dead Letter Queue. Args: dlq_id: UUID of the DLQ item Returns: New SearchIndexQueue item or None """ try: dlq_item = SearchIndexDeadLetter.objects.get(pk=dlq_id) except SearchIndexDeadLetter.DoesNotExist: return None new_item = SearchIndexQueue.objects.create( tenant=dlq_item.tenant, entity_type=dlq_item.entity_type, entity_id=dlq_item.entity_id, operation=dlq_item.operation, payload=dlq_item.payload, retry_count=0, status=QueueStatus.PENDING, ) dlq_item.resolved = True dlq_item.resolved_at = timezone.now() dlq_item.save(update_fields=['resolved', 'resolved_at']) logger.info( f"Re-queued from DLQ: {dlq_item.entity_type}:{dlq_item.entity_id}" ) return new_item @classmethod def get_queue_stats(cls) -> Dict[str, Any]: """Get queue statistics for monitoring.""" from django.db.models import Count, Q stats = SearchIndexQueue.objects.aggregate( pending=Count('id', filter=Q(status=QueueStatus.PENDING)), processing=Count('id', filter=Q(status=QueueStatus.PROCESSING)), completed=Count('id', filter=Q(status=QueueStatus.COMPLETED)), failed=Count('id', filter=Q(status=QueueStatus.FAILED)), total=Count('id'), ) dlq_stats = SearchIndexDeadLetter.objects.aggregate( unresolved=Count('id', filter=Q(resolved=False)), total=Count('id'), ) return { 'queue': stats, 'dead_letter_queue': dlq_stats, } def models_q_next_retry(now): """Helper to build Q filter for retry timing.""" from django.db.models import Q return Q(next_retry_at__isnull=True) | Q(next_retry_at__lte=now)