Integrations Domain - Models

indexes = [
    Index(fields=['status', 'priority', 'received_at']),  # Priority queue queries
    Index(fields=['processing_path', 'status']),           # Path-specific monitoring
    Index(fields=['sanity_id', 'revision']),              # Document lookups
    Index(fields=['document_type', 'priority']),          # Type-based processing
]

UniqueConstraint(
    fields=['sanity_id', 'revision'],
    name='unique_sanity_webhook_event'
)

def add_processing_log_entry(self, stage: str, metadata: dict = None):

entry = {
    'timestamp': time.time(),
    'stage': stage,
    'status': self.status,
    'metadata': metadata or {}
}

if not self.processing_log:
    self.processing_log = []

self.processing_log.append(entry)
self.save(update_fields=['processing_log'])

webhook_event.add_processing_log_entry(
    stage='redis_buffer_write',
    metadata={'buffer_key': 'webhook:product:123', 'ttl': 300}
)

def transition_to_status(self, new_status: str, metadata: dict = None):

# Update timing fields based on status transitions
if new_status == 'processing' and not self.processing_started_at:
    self.processing_started_at = timezone.now()
elif new_status in ['completed', 'failed']:
    if not self.completed_at:
        self.completed_at = timezone.now()

self.add_processing_log_entry(f'status_transition', {
    'from_status': old_status,
    'to_status': new_status,
    **(metadata or {})
})

# Start processing
webhook_event.transition_to_status('processing', metadata={
    'worker_id': 'worker-1',
    'domain': 'products'
})

# Complete successfully
webhook_event.transition_to_status('completed', metadata={
    'duration_ms': 150,
    'records_created': 1
})

# Failed processing
webhook_event.transition_to_status('failed', metadata={
    'error_type': 'ValidationError',
    'retry_count': webhook_event.retry_count
})

def increment_retry(self, error_message: str = None):

if self.retry_count >= self.max_retries:
    self.transition_to_status('failed', {
        'retry_count': self.retry_count,
        'max_retries_exceeded': True,
        'final_error': error_message
    })
else:
    self.transition_to_status('retry', {
        'retry_count': self.retry_count,
        'error': error_message
    })

try:
    process_webhook(webhook_event)
except Exception as e:
    webhook_event.increment_retry(error_message=str(e))
    # If retry_count < max_retries: status='retry', will be reprocessed
    # If retry_count >= max_retries: status='failed', permanently failed

def get_processing_duration(self):
    if not self.processing_started_at:
        return None

    end_time = self.completed_at or timezone.now()
    return (end_time - self.processing_started_at).total_seconds()

# For completed webhook
duration = webhook_event.get_processing_duration()
print(f"Processing took {duration:.2f} seconds")

# For in-progress webhook
current_duration = webhook_event.get_processing_duration()
if current_duration > 30:
    print(f"WARNING: Webhook processing taking longer than expected: {current_duration}s")

def is_stale(self, max_age_minutes: int = 30):

if self.status in ['completed', 'failed']:
    return False  # Completed webhooks are never stale

age_minutes = (timezone.now() - self.received_at).total_seconds() / 60
return age_minutes > max_age_minutes

# Monitoring job to find stuck webhooks
stale_webhooks = WebhookEvent.objects.filter(
    status__in=['received', 'processing', 'retry']
)

for webhook in stale_webhooks:
    if webhook.is_stale(max_age_minutes=15):  # 15 min threshold
        print(f"ALERT: Stale webhook {webhook.id} - {webhook.document_type}")
        # Send alert, force retry, or manual intervention

indexes = [
    Index(fields=['sanity_id', 'created_at']),  # Lookup by document
    Index(fields=['transaction_id']),           # Lookup by transaction
    Index(fields=['created_at']),               # TTL cleanup queries
]

@classmethod
def is_django_mutation(cls, sanity_id: str = None, transaction_id: str = None) -> bool:

cutoff_time = timezone.now() - timedelta(minutes=5)

if sanity_id:
    return cls.objects.filter(
        sanity_id=sanity_id,
        created_at__gte=cutoff_time
    ).exists()

if transaction_id:
    return cls.objects.filter(
        transaction_id=transaction_id,
        created_at__gte=cutoff_time
    ).exists()

return False

# In webhook handler
def handle_product_webhook(webhook_data):
    sanity_id = webhook_data['_id']
    transaction_id = webhook_data.get('transactionId')

    # Check if we just sent this mutation
    if OutboundMutationLog.is_django_mutation(
        sanity_id=sanity_id,
        transaction_id=transaction_id
    ):
        logger.info(f"Ignoring webhook - we just sent this mutation")
        return  # Skip processing to prevent loop

    # Process webhook normally
    process_product_update(webhook_data)

@classmethod
def cleanup_old_records(cls):

cutoff_time = timezone.now() - timedelta(minutes=5)
deleted_count, _ = cls.objects.filter(created_at__lt=cutoff_time).delete()
if deleted_count > 0:
    logger.info(f"Cleaned up {deleted_count} old OutboundMutationLog records")
return deleted_count

# In Celery periodic task (celery.py or tasks.py)
from celery import shared_task
from nextango.apps.integrations.models import OutboundMutationLog

@shared_task
def cleanup_outbound_mutations():
    """Run every 5 minutes"""
    deleted = OutboundMutationLog.cleanup_old_records()
    return f"Cleaned up {deleted} mutation log records"

# In celerybeat schedule:
CELERYBEAT_SCHEDULE = {
    'cleanup-mutation-logs': {
        'task': 'cleanup_outbound_mutations',
        'schedule': crontab(minute='*/5'),  # Every 5 minutes
    },
}

indexes = [
    Index(fields=['status', 'created_at']),      # Status-based queries with time ordering
    Index(fields=['sanity_id', 'django_model']), # Document-specific lookups
    Index(fields=['direction', 'status']),       # Direction-filtered status queries
]

from nextango.apps.integrations.models import SyncAuditLog

# Create audit log for successful outbound sync
SyncAuditLog.objects.create(
    direction='outbound',
    status='completed',
    sanity_id='product-abc123',
    django_model='products.Product',
    message='Successfully synced Product "Red T-Shirt" to Sanity',
    payload={'product_id': 'abc123', 'action': 'update'},
    started_at=sync_start_time,
    completed_at=timezone.now()
)

# Query failed syncs for monitoring
failed_syncs = SyncAuditLog.objects.filter(
    status='failed',
    created_at__gte=timezone.now() - timedelta(hours=24)
).order_by('-created_at')

for sync in failed_syncs:
    print(f"Failed: {sync.django_model} - {sync.message}")

# Analyze sync patterns
from django.db.models import Count
sync_stats = SyncAuditLog.objects.values('direction', 'status').annotate(
    count=Count('id')
)
# Output: [{'direction': 'inbound', 'status': 'completed', 'count': 1523}, ...]

indexes = [
    Index(fields=['expires_at']),  # Cleanup expired locks
    Index(fields=['locked_at']),   # Lock age queries
]

def is_expired(self):
    from django.utils import timezone
    return timezone.now() > self.expires_at

@classmethod
def acquire_lock(cls, resource_id: str, timeout: int = 30):

expires_at = timezone.now() + timezone.timedelta(seconds=timeout)
lock, created = cls.objects.get_or_create(
    resource_id=resource_id,
    defaults={
        'expires_at': expires_at,
        'operation': 'sync',
        'locked_by': 'universal_sync'
    }
)

if 'users.UserRole' in resource_id:
    time.sleep(0.05)  # 50ms wait
    lock.refresh_from_db()
    if lock.is_expired():
        lock.expires_at = expires_at
        lock.locked_at = timezone.now()
        lock.save()
    else:
        # Skip silently, don't raise error
        logger.info(f"UserRole resource {resource_id} is locked, skipping")
        return

elif any(user_type in resource_id for user_type in
    ['users.Customer', 'users.Employee', 'users.Manager', 'users.Influencer']):
    time.sleep(0.1)  # 100ms wait
    lock.refresh_from_db()
    if lock.is_expired():
        lock.expires_at = expires_at
        lock.locked_at = timezone.now()
        lock.save()
    else:
        logger.info(f"User model resource {resource_id} is locked, skipping")
        return

else:
    time.sleep(0.1)
    raise RuntimeError(f"Resource {resource_id} is already locked")

finally:
    # Release lock
    if lock:
        try:
            lock.delete()
        except cls.DoesNotExist:
            pass  # Lock already deleted

from nextango.apps.integrations.models import SyncLock

# Basic usage
try:
    with SyncLock.acquire_lock('products.Product:abc123', timeout=60):
        # Perform sync operation
        sync_product_to_sanity(product)
        # Lock automatically released on exit
except RuntimeError as e:
    logger.warning(f"Could not acquire lock: {e}")
    # Handle lock contention

# UserRole sync (automatic graceful skip if locked)
with SyncLock.acquire_lock('users.UserRole:xyz789'):
    # If locked by another worker, this silently skips
    # No RuntimeError raised for UserRole
    sync_user_role_to_sanity(user_role)

# 1. Webhook arrives → Create WebhookEvent
webhook_event = WebhookEvent.objects.create(
    sanity_id=payload['_id'],
    document_type=payload['_type'],
    revision=payload['_rev'],
    payload=payload,
    status='received',
    priority='critical' if payload['_type'] == 'transaction' else 'normal'
)

# 2. Check ping-pong prevention
if OutboundMutationLog.is_django_mutation(
    sanity_id=webhook_event.sanity_id,
    transaction_id=payload.get('transactionId')
):
    webhook_event.transition_to_status('completed',
        metadata={'reason': 'ping-pong_prevented'})
    return

# 3. Acquire distributed lock
resource_id = f"{webhook_event.document_type}.{webhook_event.document_type.capitalize()}:{webhook_event.sanity_id}"

try:
    with SyncLock.acquire_lock(resource_id, timeout=30):
        # 4. Transition to processing
        webhook_event.transition_to_status('processing')

        # 5. Process webhook
        result = process_webhook_payload(webhook_event.payload)

        # 6. Create audit log
        SyncAuditLog.objects.create(
            direction='inbound',
            status='completed',
            sanity_id=webhook_event.sanity_id,
            django_model=f"{webhook_event.document_type}s.{webhook_event.document_type.capitalize()}",
            message=f"Successfully synced {webhook_event.document_type}",
            payload=webhook_event.payload
        )

        # 7. Mark webhook complete
        webhook_event.transition_to_status('completed')

except Exception as e:
    # 8. Handle failure with retry
    webhook_event.increment_retry(error_message=str(e))

    # 9. Create failure audit log
    SyncAuditLog.objects.create(
        direction='inbound',
        status='failed',
        sanity_id=webhook_event.sanity_id,
        django_model=f"{webhook_event.document_type}s.{webhook_event.document_type.capitalize()}",
        message=f"Failed to sync: {str(e)}",
        payload=webhook_event.payload
    )

# 1. Django model changes (via signal or explicit call)
def sync_product_to_sanity(product):
    # 2. Check if should skip (webhook processing context)
    from nextango.apps.integrations.sync.sync_context import should_skip_sync_to_sanity
    if should_skip_sync_to_sanity():
        return

    # 3. Acquire distributed lock
    resource_id = f"products.Product:{product.sanity_id}"

    try:
        with SyncLock.acquire_lock(resource_id, timeout=30):
            # 4. Perform Sanity API mutation
            sanity_client = SanityAPIClient()
            result = sanity_client.mutate_document(product.to_sanity_format())

            # 5. Record outbound mutation (ping-pong prevention)
            OutboundMutationLog.objects.create(
                sanity_id=product.sanity_id,
                transaction_id=result.get('transactionId'),
                mutation_type='update'
            )

            # 6. Create success audit log
            SyncAuditLog.objects.create(
                direction='outbound',
                status='completed',
                sanity_id=product.sanity_id,
                django_model='products.Product',
                message=f"Successfully synced Product '{product.name}' to Sanity",
                payload={'product_id': str(product.id)}
            )

    except RuntimeError as e:
        # Lock contention - skip this sync
        logger.info(f"Skipping sync for {product.sanity_id}: {e}")

    except Exception as e:
        # 7. Create failure audit log
        SyncAuditLog.objects.create(
            direction='outbound',
            status='failed',
            sanity_id=product.sanity_id,
            django_model='products.Product',
            message=f"Failed to sync Product: {str(e)}",
            payload={'product_id': str(product.id), 'error': str(e)}
        )
        raise

# Find webhooks stuck in processing
stale_webhooks = WebhookEvent.objects.filter(
    status__in=['received', 'processing', 'retry'],
    received_at__lt=timezone.now() - timedelta(minutes=30)
)

for webhook in stale_webhooks:
    duration = webhook.get_processing_duration() or 0
    print(f"Stale webhook: {webhook.document_type} {webhook.sanity_id} ({duration}s)")

# Get failed syncs grouped by model
from django.db.models import Count

failed_by_model = SyncAuditLog.objects.filter(
    status='failed',
    created_at__gte=timezone.now() - timedelta(hours=24)
).values('django_model').annotate(
    count=Count('id')
).order_by('-count')

for stat in failed_by_model:
    print(f"{stat['django_model']}: {stat['count']} failures")

# Find resources with high lock contention
active_locks = SyncLock.objects.filter(
    locked_at__gte=timezone.now() - timedelta(minutes=5)
).order_by('resource_id')

print(f"Active locks: {active_locks.count()}")
for lock in active_locks:
    age = (timezone.now() - lock.locked_at).total_seconds()
    print(f"  {lock.resource_id}: {age}s (expires in {(lock.expires_at - timezone.now()).total_seconds()}s)")

from celery import shared_task

@shared_task
def cleanup_integration_data():
    """Run daily to clean up old integration data"""

    # Clean up OutboundMutationLog (5 min TTL)
    mutations_deleted = OutboundMutationLog.cleanup_old_records()

    # Clean up old completed webhooks (>7 days)
    old_webhooks_deleted = WebhookEvent.objects.filter(
        status__in=['completed', 'failed'],
        completed_at__lt=timezone.now() - timedelta(days=7)
    ).delete()[0]

    # Clean up old audit logs (>30 days)
    old_audits_deleted = SyncAuditLog.objects.filter(
        created_at__lt=timezone.now() - timedelta(days=30)
    ).delete()[0]

    # Clean up expired locks
    expired_locks_deleted = SyncLock.objects.filter(
        expires_at__lt=timezone.now()
    ).delete()[0]

    return {
        'mutations_deleted': mutations_deleted,
        'webhooks_deleted': old_webhooks_deleted,
        'audits_deleted': old_audits_deleted,
        'locks_deleted': expired_locks_deleted
    }