Integrations - Celery Tasks

┌─────────────────────────────────────────────────────────────────┐
│                  HTTP Webhook Request                            │
│             (from Sanity CMS)                                    │
└───────────────────┬─────────────────────────────────────────────┘
                    │
                    ▼
┌─────────────────────────────────────────────────────────────────┐
│         webhooks.py - SanityWebhookReceiver                      │
│  • HMAC signature validation                                     │
│  • Immediate 200 OK response                                     │
│  • Call: process_sanity_webhook_buffered.delay()                 │
└───────────────────┬─────────────────────────────────────────────┘
                    │
                    ▼
┌─────────────────────────────────────────────────────────────────┐
│      Redis Ingest Buffer (RedisIngestBuffer)                     │
│  • Sub-second ingestion (~5-10ms)                                │
│  • Redis Streams for optimal performance                         │
│  • Automatic failover to Redis Lists                             │
│  • Returns ingestion_id immediately                              │
└───────────────────┬─────────────────────────────────────────────┘
                    │
                    ▼
┌─────────────────────────────────────────────────────────────────┐
│    event_persistence_worker (Celery Task)                        │
│  • Consume webhooks from Redis Stream                            │
│  • Create WebhookEvent database records                          │
│  • Deduplicate using MD5 hash                                    │
│  • Route to domain adapters                                      │
│  • Handle delete events via _handle_delete_event()               │
└───────────────────┬─────────────────────────────────────────────┘
                    │
        ┌───────────┴───────────┬──────────────────┬──────────────┐
        │                       │                  │              │
        ▼                       ▼                  ▼              ▼
┌─────────────┐   ┌──────────────────┐  ┌──────────────┐  ┌──────────────┐
│UserSync     │   │ProductSync       │  │StoreSync     │  │Transaction   │
│Adapter      │   │Adapter           │  │Adapter       │  │SyncAdapter   │
└─────────────┘   └──────────────────┘  └──────────────┘  └──────────────┘

process_sanity_webhook_buffered (HTTP Response)
    │
    ├─► RedisIngestBuffer.ingest_webhook()
    │   └─► Returns ingestion_id immediately
    │
    └─► event_persistence_worker (Celery Task Chain)
        │
        ├─► dependency_fetcher.ensure_dependencies_exist()
        │   ├─► cross_domain_resolver.resolve_reference()
        │   └─► Atomic locking prevents duplicate fetches
        │
        └─► Domain Adapter.process_event()
            └─► Domain Sync Handler.sync_from_sanity()

class RedisIngestBuffer:
    """
    High-performance Redis buffer for webhook ingestion with failover capabilities.
    Provides sub-second ingestion with enterprise reliability patterns.
    """

    WEBHOOK_STREAM_KEY = 'webhook_events_stream'
    WEBHOOK_BACKUP_LIST = 'webhook_events_backup'
    REDIS_FAILOVER_TIMEOUT = 5  # seconds

    def ingest_webhook(self, webhook_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        Ingest webhook with Redis-first approach and automatic failover.

        Flow:
        1. Generate unique ingestion_id
        2. Enrich webhook with metadata
        3. Serialize to JSON bytes
        4. Attempt Redis Stream ingestion (primary)
        5. Failover to Redis List (backup)
        6. Return ingestion result

        Returns:
            {
                'success': bool,
                'ingestion_id': str,
                'ingestion_timestamp': float,
                'storage_method': 'redis_stream' | 'redis_list_failover' | 'failed',
                'failover_used': bool
            }
        """

@shared_task(base=BaseAsyncTask, bind=True)
def process_sanity_webhook_buffered(self, webhook_data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Entry point for all Sanity webhooks.
    Buffers to Redis and returns immediately for sub-second HTTP response.

    Args:
        webhook_data: Complete webhook payload from Sanity

    Returns:
        {
            'success': bool,
            'processing_stage': 'buffered',
            'ingestion_id': str,
            'message': str
        }

    Performance:
        - HTTP response time: ~10-30ms
        - Includes HMAC validation, Redis ingestion
        - Actual sync happens asynchronously in event_persistence_worker
    """
    ingest_buffer = RedisIngestBuffer()
    ingestion_result = ingest_buffer.ingest_webhook(webhook_data)

    if not ingestion_result['success']:
        logger.error(f"Webhook ingestion failed: {ingestion_result.get('error_message')}")
        return {
            'success': False,
            'processing_stage': 'ingestion_failed',
            'error': ingestion_result.get('error_message')
        }

    # Webhook buffered successfully - HTTP 200 OK can be returned
    logger.info(f"Webhook buffered: {ingestion_result['ingestion_id']}")

    # Queue persistence worker to process from Redis
    event_persistence_worker.apply_async(
        args=[ingestion_result['ingestion_id']],
        queue='webhook_processing',
        priority=5  # High priority for webhook processing
    )

    return {
        'success': True,
        'processing_stage': 'buffered',
        'ingestion_id': ingestion_result['ingestion_id'],
        'storage_method': ingestion_result['storage_method'],
        'message': 'Webhook buffered and queued for processing'
    }

@shared_task(base=BaseAsyncTask, bind=True, max_retries=3)
def event_persistence_worker(self, ingestion_id: str) -> Dict[str, Any]:
    """
    Process webhook from Redis buffer to database persistence.

    Flow:
    1. Retrieve webhook from Redis Stream/List
    2. Calculate MD5 hash for deduplication
    3. Check if WebhookEvent already exists (idempotent)
    4. Create WebhookEvent database record
    5. Route to appropriate domain adapter
    6. Update WebhookEvent with processing result

    Args:
        ingestion_id: Unique ingestion identifier from RedisIngestBuffer

    Returns:
        {
            'success': bool,
            'webhook_event_id': int,
            'document_type': str,
            'processing_result': dict
        }

    Idempotency:
        - Uses MD5 hash of payload for deduplication
        - Safe to retry without creating duplicates
        - Existing events skip re-processing
    """
    ingest_buffer = RedisIngestBuffer()

    # Retrieve webhook from Redis
    webhook_data = ingest_buffer.retrieve_webhook(ingestion_id)
    if not webhook_data:
        logger.error(f"Could not retrieve webhook {ingestion_id} from Redis")
        return {
            'success': False,
            'error': 'Webhook not found in Redis buffer',
            'ingestion_id': ingestion_id
        }

    # Calculate deduplication hash
    payload_md5 = hashlib.md5(
        json.dumps(webhook_data, sort_keys=True).encode()
    ).hexdigest()

    # Check for existing event (idempotent)
    existing_event = WebhookEvent.objects.filter(
        sanity_id=webhook_data.get('_id'),
        payload_hash=payload_md5
    ).first()

    if existing_event:
        logger.info(f"Duplicate webhook detected: {existing_event.pk}")
        return {
            'success': True,
            'webhook_event_id': existing_event.pk,
            'duplicate': True,
            'message': 'Duplicate webhook skipped'
        }

    # Create WebhookEvent
    webhook_event = WebhookEvent.objects.create(
        document_type=webhook_data.get('_type'),
        sanity_id=webhook_data.get('_id'),
        operation=webhook_data.get('_operation', 'update'),
        payload=webhook_data,
        payload_hash=payload_md5,
        ingestion_id=ingestion_id,
        processing_stage='persisted'
    )

    logger.info(f"Created WebhookEvent {webhook_event.pk} for {webhook_event.document_type}")

    # Route to domain adapter
    from ..adapters import get_domain_adapter
    adapter_name = determine_adapter_for_document_type(webhook_event.document_type)
    adapter = get_domain_adapter(adapter_name)

    if not adapter:
        webhook_event.add_processing_log_entry('adapter_not_found', {
            'adapter_name': adapter_name,
            'document_type': webhook_event.document_type
        })
        webhook_event.processing_stage = 'failed'
        webhook_event.save()
        return {
            'success': False,
            'error': f'No adapter found for {adapter_name}',
            'webhook_event_id': webhook_event.pk
        }

    # Process through adapter
    processing_result = adapter.process_event(webhook_event.pk, priority='normal')

    # Update webhook event with result
    webhook_event.processing_stage = 'completed' if processing_result['success'] else 'failed'
    webhook_event.processing_result = processing_result
    webhook_event.save()

    return {
        'success': processing_result['success'],
        'webhook_event_id': webhook_event.pk,
        'document_type': webhook_event.document_type,
        'processing_result': processing_result
    }

def ensure_dependencies_exist(payload: Dict[str, Any]) -> Dict[str, Any]:
    """
    PRE-CHECK: Ensure all required dependencies exist before sync logic runs.

    This is the critical innovation that prevents sync failures:
    - Runs BEFORE adapters call sync handlers
    - Fetches missing references from Sanity
    - Uses atomic locking to prevent race conditions
    - Fails explicitly if dependencies can't be resolved

    Args:
        payload: Webhook payload containing references

    Returns:
        {
            'success': bool,
            'status': 'all_exist' | 'some_fetched' | 'fetch_failed',
            'fetched_count': int,
            'fetch_results': List[dict],
            'missing_dependencies': List[str] (if failed)
        }

    Raises:
        MissingReferenceError: If dependency cannot be resolved

    Example:
        # Product webhook references store that doesn't exist
        payload = {
            '_id': 'product-123',
            '_type': 'product',
            'store': {'_ref': 'store-456'}
        }

        # ensure_dependencies_exist() will:
        # 1. Detect store-456 reference
        # 2. Check if Store exists in Django
        # 3. If not, fetch from Sanity and create Store
        # 4. Return success with fetched_count=1
    """
    from .cross_domain_resolver import extract_references, resolve_reference

    # Extract all references from payload
    references = extract_references(payload)

    if not references:
        return {
            'success': True,
            'status': 'no_dependencies',
            'fetched_count': 0,
            'fetch_results': []
        }

    # Check which references are missing
    missing_references = []
    for ref in references:
        if not check_reference_exists(ref):
            missing_references.append(ref)

    if not missing_references:
        return {
            'success': True,
            'status': 'all_exist',
            'fetched_count': 0,
            'fetch_results': []
        }

    # Fetch missing references with atomic locking
    fetch_results = []
    for ref_id, ref_type in missing_references:
        lock_key = f"dependency_fetch:{ref_type}:{ref_id}"

        # Atomic lock acquisition
        lock_acquired = cache.add(lock_key, True, timeout=60)

        if not lock_acquired:
            # Another worker is fetching this dependency
            logger.info(f"Dependency {ref_id} already being fetched by another worker")
            # Wait briefly for other worker to complete
            time.sleep(0.5)
            continue

        try:
            # Fetch from Sanity and create Django instance
            result = resolve_reference(ref_id, ref_type)

            fetch_results.append({
                'reference_id': ref_id,
                'reference_type': ref_type,
                'success': result['success'],
                'created': result.get('created', False)
            })

            if not result['success']:
                raise MissingReferenceError(
                    f"Failed to resolve {ref_type} {ref_id}: {result.get('error')}"
                )

        finally:
            # Always release lock
            cache.delete(lock_key)

    return {
        'success': True,
        'status': 'some_fetched',
        'fetched_count': len(fetch_results),
        'fetch_results': fetch_results
    }


def check_reference_exists(reference: tuple) -> bool:
    """Check if referenced object exists in Django database."""
    ref_id, ref_type = reference

    # Map reference type to Django model
    model_mapping = {
        'user': 'users.User',
        'product': 'products.Product',
        'store': 'stores.Store',
        'category': 'products.Category',
        'brand': 'products.Brand',
        # ... more mappings
    }

    model_key = model_mapping.get(ref_type)
    if not model_key:
        logger.warning(f"No model mapping for reference type: {ref_type}")
        return False

    from django.apps import apps
    try:
        model_class = apps.get_model(model_key)
        return model_class.objects.filter(sanity_id=ref_id).exists()
    except Exception as e:
        logger.error(f"Error checking reference {ref_id}: {e}")
        return False


class MissingReferenceError(Exception):
    """Raised when a required dependency cannot be resolved."""
    pass

def extract_references(payload: Dict[str, Any], parent_path: str = '') -> List[tuple]:
    """
    Recursively extract all Sanity reference objects from payload.

    Sanity references have structure:
    {
        "_ref": "document-id",
        "_type": "reference"
    }

    Args:
        payload: Webhook payload or nested object
        parent_path: Path to current object (for debugging)

    Returns:
        List of (ref_id, ref_type) tuples

    Example:
        payload = {
            '_id': 'product-123',
            'store': {'_ref': 'store-456', '_type': 'reference'},
            'variants': [
                {'_ref': 'variant-789', '_type': 'reference'}
            ]
        }

        extract_references(payload) returns:
        [
            ('store-456', 'store'),
            ('variant-789', 'productVariant')
        ]
    """
    references = []

    if isinstance(payload, dict):
        # Check if this is a reference object
        if '_ref' in payload and '_type' in payload:
            ref_id = payload['_ref']
            # Infer type from ref_id prefix (e.g., "store-456" -> "store")
            ref_type = infer_type_from_ref_id(ref_id)
            references.append((ref_id, ref_type))

        # Recurse into nested objects
        for key, value in payload.items():
            nested_refs = extract_references(value, f"{parent_path}.{key}")
            references.extend(nested_refs)

    elif isinstance(payload, list):
        # Recurse into array elements
        for i, item in enumerate(payload):
            nested_refs = extract_references(item, f"{parent_path}[{i}]")
            references.extend(nested_refs)

    return references


def infer_type_from_ref_id(ref_id: str) -> str:
    """
    Infer document type from Sanity reference ID.

    Sanity IDs typically have format: "{type}-{uuid}"

    Examples:
        "store-a1b2c3" -> "store"
        "product-d4e5f6" -> "product"
        "user-g7h8i9" -> "user"
    """
    if '-' in ref_id:
        return ref_id.split('-')[0]
    return 'unknown'

def resolve_reference(ref_id: str, ref_type: str) -> Dict[str, Any]:
    """
    Resolve a Sanity reference by fetching from Sanity and creating Django instance.

    Flow:
    1. Fetch document from Sanity API
    2. Determine appropriate domain adapter
    3. Create WebhookEvent for the fetched document
    4. Process through adapter
    5. Return created Django instance

    Args:
        ref_id: Sanity document ID
        ref_type: Document type (user, product, store, etc.)

    Returns:
        {
            'success': bool,
            'ref_id': str,
            'ref_type': str,
            'django_instance_id': str,
            'created': bool,
            'error': str (if failed)
        }

    Example:
        # Product webhook references missing store
        resolve_reference('store-456', 'store')

        # This will:
        # 1. Fetch store-456 from Sanity
        # 2. Create WebhookEvent for store
        # 3. Process through StoreSyncAdapter
        # 4. Return created Store instance ID
    """
    from ..sanity_client import SanityAPIClient
    from ..adapters import get_domain_adapter
    from ..models import WebhookEvent

    logger.info(f"Resolving reference: {ref_type} {ref_id}")

    # Fetch from Sanity
    sanity_client = SanityAPIClient()
    try:
        document = sanity_client.get_document(ref_id)
    except Exception as e:
        logger.error(f"Failed to fetch {ref_id} from Sanity: {e}")
        return {
            'success': False,
            'ref_id': ref_id,
            'ref_type': ref_type,
            'error': f'Sanity fetch failed: {str(e)}'
        }

    if not document:
        return {
            'success': False,
            'ref_id': ref_id,
            'ref_type': ref_type,
            'error': f'Document {ref_id} not found in Sanity'
        }

    # Create WebhookEvent for dependency
    webhook_event = WebhookEvent.objects.create(
        document_type=document.get('_type', ref_type),
        sanity_id=ref_id,
        operation='dependency_fetch',
        payload=document,
        processing_stage='dependency_resolution'
    )

    # Determine adapter
    adapter_name = determine_adapter_for_document_type(webhook_event.document_type)
    adapter = get_domain_adapter(adapter_name)

    if not adapter:
        return {
            'success': False,
            'ref_id': ref_id,
            'ref_type': ref_type,
            'error': f'No adapter found for {adapter_name}'
        }

    # Process through adapter (with signal isolation to prevent outbound sync)
    from .signal_management import disable_sync_signals
    with disable_sync_signals():
        processing_result = adapter.process_event(webhook_event.pk, priority='dependency')

    if not processing_result['success']:
        return {
            'success': False,
            'ref_id': ref_id,
            'ref_type': ref_type,
            'error': f"Adapter processing failed: {processing_result.get('error')}"
        }

    return {
        'success': True,
        'ref_id': ref_id,
        'ref_type': ref_type,
        'django_instance_id': processing_result.get('instance_id'),
        'created': True
    }

from contextlib import contextmanager
from django.db.models.signals import post_save, post_delete, pre_save
from typing import List, Callable

_DISABLED_SIGNAL_RECEIVERS = []


@contextmanager
def disable_sync_signals():
    """
    Context manager to temporarily disable outbound sync signals.

    Critical for webhook processing to prevent infinite loops:
    - Webhook from Sanity creates/updates Django model
    - Without this: post_save signal would sync back to Sanity
    - With this: Signal is disabled during webhook processing

    Usage:
        with disable_sync_signals():
            user = User.objects.create(...)  # No outbound sync triggered

    How it works:
    1. Disconnects all sync-related signal handlers
    2. Executes code block
    3. Reconnects signal handlers
    4. Safe even if exception occurs (finally block)
    """
    # Collect signal handlers to disable
    sync_signal_handlers = _get_sync_signal_handlers()

    # Disconnect all sync signals
    for signal, receiver, sender in sync_signal_handlers:
        signal.disconnect(receiver, sender=sender, dispatch_uid=receiver.__name__)
        logger.debug(f"Disabled signal: {receiver.__name__} for {sender.__name__}")

    try:
        yield
    finally:
        # Reconnect all sync signals
        for signal, receiver, sender in sync_signal_handlers:
            signal.connect(receiver, sender=sender, dispatch_uid=receiver.__name__)
            logger.debug(f"Re-enabled signal: {receiver.__name__} for {sender.__name__}")


def _get_sync_signal_handlers() -> List[tuple]:
    """
    Identify all sync-related signal handlers across domains.

    Returns:
        List of (signal, receiver, sender) tuples
    """
    sync_handlers = []

    # Import domain signal modules
    from nextango.apps.users import signals as user_signals
    from nextango.apps.products import signals as product_signals
    from nextango.apps.stores import signals as store_signals
    from nextango.apps.promotions import signals as promo_signals
    from nextango.apps.transactions import signals as txn_signals

    # Collect handlers that contain "sync" or "sanity" in name
    for signal_module in [user_signals, product_signals, store_signals, promo_signals, txn_signals]:
        for attr_name in dir(signal_module):
            if 'sync' in attr_name.lower() or 'sanity' in attr_name.lower():
                handler = getattr(signal_module, attr_name)
                if callable(handler):
                    # Determine signal type and sender from handler
                    signal_info = _parse_signal_handler(handler)
                    if signal_info:
                        sync_handlers.append(signal_info)

    return sync_handlers


def _parse_signal_handler(handler: Callable) -> Optional[tuple]:
    """
    Parse signal handler to extract signal type and sender model.

    Uses introspection to determine which signal and model the handler is connected to.
    """
    # This uses Django's signal receiver registry
    # Implementation details omitted for brevity
    pass

# SyncContext provides thread-local flag that signals can check
from ..sync.sync_context import SyncContext

with SyncContext(source='sanity_webhook'):
    user = User.objects.create(...)
    # user._sync_source = 'sanity_webhook'

# In signal handler:
if instance._sync_source == 'sanity_webhook':
    return  # Skip outbound sync

from celery import shared_task
from .base import BaseAsyncTask


@shared_task(base=BaseAsyncTask, bind=True)
def queue_variant_sync_task(self, product_id: str, variant_data: Dict[str, Any]):
    """
    Queue variant sync as separate task to prevent blocking main product sync.

    Args:
        product_id: Parent product Sanity ID
        variant_data: ProductVariant payload

    Why separate task:
        - Products can have 100+ variants
        - Syncing all variants inline blocks HTTP response
        - Queuing allows parallel processing of variants
    """
    from ..sync.product_sync import ProductSyncHandler

    handler = ProductSyncHandler()
    result = handler.sync_from_sanity(variant_data, 'products.ProductVariant')

    return {
        'success': bool(result),
        'product_id': product_id,
        'variant_id': variant_data.get('_id')
    }


@shared_task(base=BaseAsyncTask, bind=True)
def queue_inventory_sync_task(self, product_id: str, inventory_data: Dict[str, Any]):
    """
    Queue inventory sync to handle stock level updates asynchronously.

    Args:
        product_id: Product Sanity ID
        inventory_data: Inventory payload with stock levels

    Why separate task:
        - Inventory updates are high-frequency (~1000/day)
        - Can batch multiple updates for same product
        - Prevents blocking product catalog updates
    """
    from ..sync.product_sync import ProductSyncHandler

    handler = ProductSyncHandler()
    result = handler.sync_inventory_from_sanity(inventory_data)

    return {
        'success': bool(result),
        'product_id': product_id
    }


@shared_task(base=BaseAsyncTask, bind=True)
def queue_delayed_product_sync_task(self, webhook_event_id: int, priority: str = 'normal'):
    """
    Delayed product sync for rate-limited scenarios.

    Called by ProductSyncAdapter when rate limit is exceeded.
    Delays processing by 60 seconds to spread load.

    Args:
        webhook_event_id: WebhookEvent PK
        priority: Processing priority

    Flow:
        - ProductSyncAdapter detects rate limit
        - Queues this task with countdown=60
        - Task executes after delay
        - Processes through adapter normally
    """
    from ..adapters import get_domain_adapter

    adapter = get_domain_adapter('product_sync_adapter')
    result = adapter.process_event(webhook_event_id, priority)

    return result

# In ProductSyncAdapter._process_business_logic()
if current_requests >= 50:
    # Queue for delayed processing
    from ...tasks.product_tasks import queue_delayed_product_sync_task
    queue_delayed_product_sync_task.apply_async(
        args=[webhook_event.pk, priority],
        countdown=60  # 1 minute delay
    )
    return {
        'success': True,
        'rate_limited': True,
        'queued_for_delayed_processing': True
    }

from celery import Task

class BaseAsyncTask(Task):
    """
    Base class for all webhook processing and sync-related Celery tasks.
    Provides common error handling, retry logic, and audit logging.
    """

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        """Called when task fails with systematic error logging."""
        logger.error(f"Task {task_id} ({self.name}) failed: {exc}", exc_info=einfo)

        # Log failure to sync audit if possible
        try:
            if 'sanity_id' in kwargs:
                SyncAuditLog.objects.create(
                    sanity_id=kwargs.get('sanity_id'),
                    django_model=kwargs.get('document_type', 'unknown'),
                    direction='inbound',
                    operation='webhook_processing',
                    status='failed',
                    message=f"Task failed: {str(exc)}"
                )
        except Exception as log_error:
            logger.warning(f"Failed to create audit log for failed task: {log_error}")

    def on_success(self, retval, task_id, args, kwargs):
        """Called when task succeeds with systematic success logging."""
        logger.info(f"Task {task_id} ({self.name}) completed successfully")

    def on_retry(self, exc, task_id, args, kwargs, einfo):
        """Called when task is retried with systematic retry logging."""
        logger.warning(f"Task {task_id} ({self.name}) retrying due to: {exc}")

class BaseWebhookProcessor:
    """
    Base class for domain-specific webhook processing.
    Provides shared infrastructure and common processing patterns.
    """

    def __init__(self):
        self.logger = logging.getLogger(self.__class__.__name__)

    def setup_sync_context(self, webhook_data: Dict[str, Any], document_type: str):
        """Setup sync context for webhook processing."""
        webhook_signature = hashlib.sha256(
            json.dumps(webhook_data, sort_keys=True).encode()
        ).hexdigest()[:16]

        sync_source = f'{document_type}_webhook_processing'
        return webhook_processing_context(
            webhook_signature=webhook_signature,
            sync_source=sync_source
        )

    def create_sync_audit_log(self, operation: str, django_model_key: str,
                            sanity_id: Optional[str] = None,
                            status: str = 'pending',
                            message: str = '') -> SyncAuditLog:
        """Create audit log entry."""
        return SyncAuditLog.objects.create(
            django_model=django_model_key,
            sanity_id=sanity_id,
            direction='inbound',
            status=status,
            message=message
        )

    def acquire_sync_lock(self, resource_id: str, operation: str) -> Optional[SyncLock]:
        """
        Acquire sync lock to prevent concurrent processing.
        Returns None if lock cannot be acquired.
        """
        try:
            lock, created = SyncLock.objects.get_or_create(
                resource_id=resource_id,
                defaults={'operation': operation}
            )

            if not created and not lock.is_expired():
                # Resource is locked
                raise RuntimeError(f"Resource {resource_id} is already locked")

            return lock

        except Exception as e:
            self.logger.error(f"Failed to acquire lock for {resource_id}: {e}")
            return None

# In webhooks.py
sanity_operation = request.headers.get('Sanity-Operation')
if sanity_operation == 'delete':
    webhook_data['_is_delete'] = True

# In cross_domain_resolver.py
def _handle_delete_event(webhook_data: Dict[str, Any], webhook_event) -> Dict[str, Any]:
    """
    Route delete events to appropriate domain adapter's handle_delete() method.

    Flow:
    1. Determine document type and adapter
    2. Find existing Django instance by sanity_id
    3. Call adapter.handle_delete(instance, webhook_event)
    4. Log result to audit trail
    """
    document_type = webhook_data.get('_type')
    sanity_id = webhook_data.get('_id')

    # Get domain adapter
    adapter = get_adapter_for_document_type(document_type)

    # Find existing instance
    instance = find_instance_by_sanity_id(sanity_id, document_type)

    if instance:
        return adapter.handle_delete(instance, webhook_event)
    else:
        return {'success': True, 'action': 'not_found_no_action'}

┌─────────────────────────────────────────────────────────────────┐
│              Sanity Delete Webhook                                │
│      (Sanity-Operation: delete header)                           │
└───────────────────┬─────────────────────────────────────────────┘
                    │
                    ▼
┌─────────────────────────────────────────────────────────────────┐
│         event_persistence_worker                                  │
│  • Check webhook_data['_is_delete']                              │
│  • Route to _handle_delete_event()                               │
└───────────────────┬─────────────────────────────────────────────┘
                    │
                    ▼
┌─────────────────────────────────────────────────────────────────┐
│         _handle_delete_event()                                    │
│  • Find instance by sanity_id                                    │
│  • Get domain adapter                                             │
│  • Call adapter.handle_delete(instance, webhook_event)           │
└───────────────────┬─────────────────────────────────────────────┘
                    │
        ┌───────────┴───────────────────────────────────┐
        │                                               │
        ▼                                               ▼
┌─────────────────┐                           ┌─────────────────┐
│  Product Domain │                           │  Other Domains  │
│  • Transaction- │                           │  • Standard     │
│    aware delete │                           │    soft-delete  │
│  • Cascade to   │                           │                 │
│    variants     │                           │                 │
└─────────────────┘                           └─────────────────┘

webhook_event.add_processing_log_entry('delete_processed', {
    'model': model_name,
    'pk': str(instance_pk),
    'sanity_id': sanity_id,
    'action': action  # 'soft_deleted_with_history', 'marked_deleted', etc.
})

HTTP Request → 200 OK Response: ~10-30ms
├─ HMAC validation: 2-5ms
├─ Redis ingestion: 5-10ms
└─ Task queueing: 3-5ms

Async Processing (event_persistence_worker): ~50-100ms
├─ Redis retrieval: 5-10ms
├─ Deduplication check: 5-10ms
├─ WebhookEvent creation: 30-50ms
└─ Adapter routing: 10-20ms

Dependency Resolution (if needed): ~100-500ms
├─ Dependency detection: 10-20ms
├─ Sanity API fetch: 100-300ms per dependency
└─ Dependency sync: 50-200ms per dependency

Business Logic Processing: Variable
├─ Adapter processing: 10-30ms
├─ Sync handler: 50-200ms
└─ Database operations: 30-100ms

Total (no dependencies): ~150-300ms
Total (with dependencies): ~500-2000ms

# settings.py
CELERY_TASK_ROUTES = {
    'nextango.apps.integrations.tasks.process_sanity_webhook_buffered': {
        'queue': 'webhook_ingestion',
        'routing_key': 'webhook.ingestion',
        'priority': 10  # Highest priority
    },
    'nextango.apps.integrations.tasks.event_persistence_worker': {
        'queue': 'webhook_processing',
        'routing_key': 'webhook.processing',
        'priority': 5  # High priority
    },
    'nextango.apps.integrations.tasks.queue_delayed_product_sync_task': {
        'queue': 'delayed_sync',
        'routing_key': 'sync.delayed',
        'priority': 1  # Low priority (rate-limited)
    },
}

@shared_task(
    base=BaseAsyncTask,
    bind=True,
    max_retries=3,
    default_retry_delay=60,  # 1 minute
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,
    retry_backoff_max=600,  # 10 minutes max
    retry_jitter=True
)
def event_persistence_worker(self, ingestion_id: str):
    # Task implementation
    pass

# celery_monitoring.py
from celery.signals import task_success, task_failure, task_retry

@task_success.connect
def task_success_handler(sender=None, result=None, **kwargs):
    """Track successful task completion."""
    task_name = sender.name
    # Increment success counter in monitoring system
    metrics.increment(f'celery.task.{task_name}.success')

@task_failure.connect
def task_failure_handler(sender=None, exception=None, **kwargs):
    """Track task failures."""
    task_name = sender.name
    # Increment failure counter
    metrics.increment(f'celery.task.{task_name}.failure')
    # Alert if failure rate exceeds threshold
    check_failure_rate_alert(task_name)

@task_retry.connect
def task_retry_handler(sender=None, reason=None, **kwargs):
    """Track task retries."""
    task_name = sender.name
    metrics.increment(f'celery.task.{task_name}.retry')

# views/manual_sync.py
def celery_health_check(request):
    """
    Health check endpoint for Celery infrastructure.
    Returns status of Redis, workers, and queue lengths.
    """
    from ..tasks.webhook_processing import RedisIngestBuffer

    buffer = RedisIngestBuffer()
    health_status = buffer.get_ingestion_health_status()

    # Check worker availability
    from celery import current_app
    inspect = current_app.control.inspect()
    active_workers = inspect.active()

    # Check queue lengths
    with current_app.connection() as conn:
        queue_lengths = {
            'webhook_ingestion': conn.default_channel.queue_declare('webhook_ingestion', passive=True).message_count,
            'webhook_processing': conn.default_channel.queue_declare('webhook_processing', passive=True).message_count,
        }

    return JsonResponse({
        'redis_health': health_status,
        'active_workers': len(active_workers) if active_workers else 0,
        'queue_lengths': queue_lengths,
        'status': 'healthy' if health_status['primary_redis'] and active_workers else 'degraded'
    })