Integrations - Views/Webhooks Documentation

from django.urls import path
from .webhooks import SanityWebhookReceiver
from .views.manual_sync import ManualSyncView

urlpatterns = [
    # Single, unified webhook endpoint for all Sanity document types
    path('sanity-webhook/', SanityWebhookReceiver.as_view(), name='sanity-webhook'),

    # Manual sync endpoint for Sanity Studio document actions
    path('manual-sync/', ManualSyncView.as_view(), name='manual-sync'),
]

@method_decorator(csrf_exempt, name='dispatch')
class SanityWebhookReceiver(APIView):
    """
    Secure webhook receiver for Sanity CMS updates.

    Responsibilities:
    1. Verify HMAC-SHA256 signature
    2. Perform deduplication checks using Redis
    3. Queue Celery task for async processing
    4. Return fast 202 Accepted response
    """

    authentication_classes = []  # Webhooks use signature verification
    permission_classes = []      # Webhooks use signature verification

def get(self, request):
    """
    Simple GET endpoint for testing webhook URL accessibility.
    """
    logger.info("=== WEBHOOK GET REQUEST ===")
    logger.info(f"Request path: {request.path}")
    logger.info("Webhook endpoint is accessible")

    return Response(
        {
            'status': 'ok',
            'message': 'Sanity webhook endpoint is accessible',
            'endpoint': '/integrations/sanity-webhook/',
            'methods': ['GET', 'POST']
        },
        status=status.HTTP_200_OK
    )

curl https://api.example.com/integrations/sanity-webhook/

{
  "status": "ok",
  "message": "Sanity webhook endpoint is accessible",
  "endpoint": "/integrations/sanity-webhook/",
  "methods": ["GET", "POST"]
}

def post(self, request):
    """
    Handle incoming Sanity webhook with fast response.
    """
    logger.info("=== WEBHOOK ENDPOINT HIT ===")
    logger.info(f"Request method: {request.method}")
    logger.info(f"Request path: {request.path}")
    logger.info(f"Request headers: {dict(request.headers)}")
    logger.info(f"Request body length: {len(request.body)} bytes")
    logger.info("Received Sanity webhook")

    # 1. Parse webhook payload
    try:
        webhook_data = json.loads(request.body.decode('utf-8'))
    except (json.JSONDecodeError, UnicodeDecodeError) as e:
        logger.error(f"Failed to parse webhook JSON: {e}")
        return Response(
            {'error': 'Invalid JSON payload'},
            status=status.HTTP_400_BAD_REQUEST
        )

# Extract basic webhook info for logging and deduplication
document_type = webhook_data.get('_type', 'unknown')
sanity_id = webhook_data.get('_id', 'unknown')
revision = webhook_data.get('_rev', 'unknown')

logger.info(f"Verified webhook: {document_type} {sanity_id} (rev: {revision})")

# 2. Validate dataset environment match
sanity_dataset = request.headers.get('Sanity-Dataset')
if not self._is_dataset_valid(sanity_dataset):
    logger.warning(f"Dataset validation failed - received: {sanity_dataset}, expected: {getattr(settings, 'SANITY_DATASET', 'production')}")
    return Response(
        {'error': 'Dataset environment mismatch'},
        status=status.HTTP_400_BAD_REQUEST
    )

def _is_dataset_valid(self, received_dataset):
    """
    Validates that the webhook dataset matches the expected environment.
    Prevents production webhooks from processing on development servers and vice versa.
    """
    expected_dataset = getattr(settings, 'SANITY_DATASET', 'production')

    if not received_dataset:
        logger.warning("Missing 'Sanity-Dataset' header")
        return False

    # Check for exact match
    if received_dataset == expected_dataset:
        logger.debug(f"Dataset validation passed: {received_dataset}")
        return True

    # Allow development/staging flexibility if configured
    allow_cross_env = getattr(settings, 'SANITY_ALLOW_CROSS_ENV_WEBHOOKS', False)
    if allow_cross_env:
        logger.info(f"Cross-environment webhook allowed: {received_dataset} -> {expected_dataset}")
        return True

    return False

# 3. Verify HMAC-SHA256 signature
if not self._is_signature_valid(request):
    logger.warning("Webhook signature verification failed")
    return Response(
        {'error': 'Invalid webhook signature'},
        status=status.HTTP_401_UNAUTHORIZED
    )

def _is_signature_valid(self, request):
    """
    Verifies the Sanity webhook signature using URL-safe Base64 HMAC-SHA256.
    """
    webhook_secret = getattr(settings, 'SANITY_WEBHOOK_SECRET', None)
    if not webhook_secret:
        logger.error("SANITY_WEBHOOK_SECRET is not configured")
        return False

    # Get the signature header
    signature_header = request.headers.get('Sanity-Webhook-Signature')
    if not signature_header:
        logger.warning("Missing 'Sanity-Webhook-Signature' header")
        return False

    # Parse the header format: t=timestamp,v1=signature
    try:
        sig_map = {
            key_value.split('=')[0]: key_value.split('=')[1]
            for key_value in signature_header.split(',')
        }
        timestamp_str = sig_map.get('t')
        signature = sig_map.get('v1')

        if not timestamp_str or not signature:
            raise ValueError("Missing timestamp or signature in header")

        # Sanity timestamp is in milliseconds
        timestamp = int(timestamp_str)
    except (ValueError, IndexError) as e:
        logger.error(f"Invalid signature header format: {signature_header} - {e}")
        return False

    # Prevent replay attacks (5-minute tolerance)
    if abs(time.time() - (timestamp / 1000)) > 300:
        logger.warning("Webhook timestamp validation failed (possible replay attack)")
        return False

    # Construct the signed payload: timestamp_string + "." + raw_body
    signed_payload = f"{timestamp_str}.".encode('utf-8') + request.body

    # Calculate expected signature using URL-safe Base64
    digest = hmac.new(
        webhook_secret.encode('utf-8'),
        signed_payload,
        hashlib.sha256
    ).digest()

    expected_signature = base64.urlsafe_b64encode(digest).rstrip(b'=').decode('utf-8')

    # Securely compare the signatures
    return hmac.compare_digest(signature, expected_signature)

# DRAFT FILTERING: Safety net to skip draft documents
# PRIMARY: Sanity webhook should be configured with filter: !(_id in path("drafts.**"))
# SECONDARY: This code-level check prevents drafts if webhook config is missing/incorrect
if isinstance(sanity_id, str) and sanity_id.startswith('drafts.'):
    logger.info(f"Skipping draft document webhook: {sanity_id}")
    return Response(
        {
            'status': 'skipped',
            'message': 'Draft documents are not synced to Django',
            'document_type': document_type,
            'sanity_id': sanity_id,
            'reason': 'draft_document'
        },
        status=status.HTTP_200_OK
    )

# PING-PONG PREVENTION: Check if Django is currently mutating this document
# This must happen BEFORE queuing to prevent race condition
# Uses both Redis cache AND database for redundancy
from django.core.cache import cache
from .models import OutboundMutationLog

# Check 1: Redis cache (fast)
pre_mutation_key = f"django_mutating:{sanity_id}"
is_mutating_cache = cache.get(pre_mutation_key)

# Check 2: Database fallback (if Redis failed or cache miss)
is_mutating_db = OutboundMutationLog.is_django_mutation(sanity_id=sanity_id, transaction_id=revision)

if is_mutating_cache or is_mutating_db:
    source = 'cache' if is_mutating_cache else 'database'
    logger.info(f"PING-PONG PREVENTED ({source}): Webhook for {sanity_id} skipped - Django is currently updating this document")
    return Response(
        {
            'status': 'skipped',
            'message': 'Webhook originated from Django update - ping-pong prevented',
            'document_type': document_type,
            'sanity_id': sanity_id,
            'reason': 'django_mutation_in_progress',
            'detection_method': source
        },
        status=status.HTTP_200_OK
    )

# Skip processing for saleLineItem webhooks - managed as part of salesTransaction
if document_type == 'saleLineItem':
    logger.info(f"Skipping saleLineItem webhook {sanity_id} - managed as part of salesTransaction")
    return Response(
        {
            'status': 'skipped',
            'message': 'saleLineItem webhooks are managed as part of salesTransaction documents',
            'document_type': document_type,
            'sanity_id': sanity_id,
            'reason': 'line_items_managed_by_parent_transaction'
        },
        status=status.HTTP_200_OK
    )

# Skip processing for Sanity asset webhooks - not synced to Django
if document_type in ['sanity.imageAsset', 'sanity.fileAsset']:
    logger.info(f"Skipping Sanity asset webhook {sanity_id} - asset management is Sanity-only")
    return Response(
        {
            'status': 'skipped',
            'message': 'Sanity asset webhooks are not synced to Django',
            'document_type': document_type,
            'sanity_id': sanity_id,
            'reason': 'sanity_asset_not_synced'
        },
        status=status.HTTP_200_OK
    )

# 3. Deduplication check using Redis
from django.core.cache import cache

webhook_signature = hashlib.md5(
    f"{sanity_id}:{revision}:{document_type}".encode()
).hexdigest()

dedup_key = f"webhook_dedup:{webhook_signature}"

if cache.get(dedup_key):
    logger.info(f"Duplicate webhook detected - ignoring {document_type} {sanity_id}")
    return Response(
        {
            'status': 'duplicate',
            'message': 'Webhook already processed',
            'webhook_signature': webhook_signature,
        },
        status=status.HTTP_200_OK
    )

# 5. Mark webhook as processed (10 minute expiry)
cache.set(dedup_key, "1", 600)

# 6. Add sync metadata to webhook data
webhook_data['_sync_source'] = 'sanity_webhook'
webhook_data['_webhook_signature'] = webhook_signature

# TRANSACTION TRACKING: Extract transaction ID from webhook headers for ping-pong prevention
transaction_id = request.headers.get('sanity-transaction-id') or request.headers.get('Sanity-Transaction-Id')
if transaction_id:
    webhook_data['_transaction_id'] = transaction_id
    logger.debug(f"Webhook transaction ID: {transaction_id}")

# 7. Queue Celery task for async processing with Redis buffering
try:
    task_result = process_sanity_webhook_buffered.delay(webhook_data)
    logger.info(f"Queued buffered webhook processing task {task_result.id} for {document_type} {sanity_id}")

    # Always return 202 Accepted for successful queuing
    return Response(
        {
            'status': 'accepted',
            'message': 'Webhook received and queued for buffered processing',
            'task_id': task_result.id,
            'webhook_signature': webhook_signature,
            'document_type': document_type,
            'sanity_id': sanity_id,
            'processing_path': 'redis_buffered'
        },
        status=status.HTTP_202_ACCEPTED
    )

except Exception as e:
    logger.error(f"Failed to queue buffered webhook processing: {e}")
    return Response(
        {
            'status': 'error',
            'message': 'Failed to queue webhook for buffered processing',
            'error': str(e)
        },
        status=status.HTTP_500_INTERNAL_SERVER_ERROR
    )

┌─────────────────────────────────────────────────────────┐
│              Sanity CMS Webhook Trigger                 │
│         (Document created/updated/deleted)              │
└─────────────────────┬───────────────────────────────────┘
                      │
                      ▼
        ┌─────────────────────────────┐
        │   1. Parse JSON Payload     │
        └─────────────┬───────────────┘
                      │
                      ▼
        ┌─────────────────────────────┐
        │   2. Validate Dataset       │
        │   (production/staging)      │
        └─────────────┬───────────────┘
                      │
                      ▼
        ┌─────────────────────────────┐
        │   3. Verify HMAC Signature  │
        │   (HMAC-SHA256 + timestamp) │
        └─────────────┬───────────────┘
                      │
                      ▼
        ┌─────────────────────────────┐
        │   4. Filter Drafts          │
        │   (drafts.* → skip)         │
        └─────────────┬───────────────┘
                      │
                      ▼
        ┌─────────────────────────────┐
        │   5. Ping-Pong Prevention   │
        │   (Redis + DB check)        │
        └─────────────┬───────────────┘
                      │
                      ▼
        ┌─────────────────────────────┐
        │   6. Document Type Filter   │
        │   (saleLineItem, assets)    │
        └─────────────┬───────────────┘
                      │
                      ▼
        ┌─────────────────────────────┐
        │   7. Deduplication Check    │
        │   (MD5 hash + Redis)        │
        └─────────────┬───────────────┘
                      │
                      ▼
        ┌─────────────────────────────┐
        │   8. Add Metadata           │
        │   (_sync_source, etc.)      │
        └─────────────┬───────────────┘
                      │
                      ▼
        ┌─────────────────────────────┐
        │   9. Queue Celery Task      │
        │   (process_sanity_webhook)  │
        └─────────────┬───────────────┘
                      │
                      ▼
        ┌─────────────────────────────┐
        │   10. Return 202 Accepted   │
        │   (Fast response ~50-100ms) │
        └─────────────────────────────┘

{
  "status": "accepted",
  "message": "Webhook received and queued for buffered processing",
  "task_id": "a1b2c3d4-e5f6-7g8h-9i0j-k1l2m3n4o5p6",
  "webhook_signature": "5f4dcc3b5aa765d61d8327deb882cf99",
  "document_type": "product",
  "sanity_id": "product_123",
  "processing_path": "redis_buffered"
}

// Draft document
{
  "status": "skipped",
  "message": "Draft documents are not synced to Django",
  "document_type": "product",
  "sanity_id": "drafts.product_123",
  "reason": "draft_document"
}

// Ping-pong prevention
{
  "status": "skipped",
  "message": "Webhook originated from Django update - ping-pong prevented",
  "document_type": "product",
  "sanity_id": "product_123",
  "reason": "django_mutation_in_progress",
  "detection_method": "cache"
}

// Duplicate webhook
{
  "status": "duplicate",
  "message": "Webhook already processed",
  "webhook_signature": "5f4dcc3b5aa765d61d8327deb882cf99"
}

// Bad request (400)
{
  "error": "Invalid JSON payload"
}

// Dataset mismatch (400)
{
  "error": "Dataset environment mismatch"
}

// Invalid signature (401)
{
  "error": "Invalid webhook signature"
}

// Queue failure (500)
{
  "status": "error",
  "message": "Failed to queue webhook for buffered processing",
  "error": "Connection refused"
}

@method_decorator(csrf_exempt, name='dispatch')
class ManualSyncView(APIView):
    """
    Manual sync endpoint for Sanity Studio document actions.

    Accepts a Sanity document ID and type, fetches the current document,
    and syncs it to Django using the appropriate domain adapter.
    """

    authentication_classes = [PosGatewayAuthentication, WebSessionAuthentication]
    permission_classes = [IsAuthenticated]

def post(self, request):
    """
    Handle manual sync request from Sanity Studio.
    Fetches Django record and pushes to Sanity (Django → Sanity direction).

    Expected payload:
    {
        "sanity_id": "product_abc123",
        "document_type": "product"
    }
    """
    logger.info("=== MANUAL SYNC REQUEST (Django → Sanity) ===")
    logger.info(f"Request from: {request.META.get('REMOTE_ADDR')}")
    logger.info(f"Payload: {request.data}")

    # 1. Validate request payload
    sanity_id = request.data.get('sanity_id')
    document_type = request.data.get('document_type')

    if not sanity_id:
        return Response(
            {'success': False, 'error': 'Missing required field: sanity_id'},
            status=status.HTTP_400_BAD_REQUEST
        )

    if not document_type:
        return Response(
            {'success': False, 'error': 'Missing required field: document_type'},
            status=status.HTTP_400_BAD_REQUEST
        )

# 2. Find Django record by sanity_id
try:
    # Get the appropriate sync handler and model
    if document_type in ['product', 'productVariant', 'productType', 'brand', 'category', 'collection']:
        from ..sync.product_sync import ProductSyncHandler
        sync_handler = ProductSyncHandler()

        # Map to model class
        model_map = {
            'product': 'products.Product',
            'productVariant': 'products.ProductVariant',
            'productType': 'products.ProductType',
            'brand': 'products.Brand',
            'category': 'products.Category',
            'collection': 'products.Collection',
        }
        model_class_key = model_map.get(document_type)

    elif document_type == 'store':
        from ..sync.store_sync import StoreSyncHandler
        sync_handler = StoreSyncHandler()
        model_class_key = 'stores.Store'

    elif document_type in ['user', 'userEmployee', 'userCustomer', 'userManager']:
        from ..sync.user_sync import UserSyncHandler
        sync_handler = UserSyncHandler()
        model_class_key = 'users.User'

    elif document_type in ['salesTransaction', 'saleLineItem']:
        from ..sync.transaction_sync import TransactionSyncHandler
        sync_handler = TransactionSyncHandler()
        model_class_key = 'transactions.SaleTransaction' if document_type == 'salesTransaction' else 'transactions.SaleLineItem'

    # ... other document types ...

logger.info(f"Looking up Django {model_class_key} with sanity_id: {sanity_id}")

# Get the Django model
from django.apps import apps
app_label, model_name = model_class_key.split('.')
model_class = apps.get_model(app_label, model_name)

# Find Django instance by sanity_id
django_instance = model_class.objects.filter(sanity_id=sanity_id).first()

if not django_instance:
    logger.warning(f"No Django {model_class_key} found with sanity_id: {sanity_id}")
    return Response(
        {
            'success': False,
            'error': f'No Django record found with sanity_id: {sanity_id}',
            'sanity_id': sanity_id,
            'document_type': document_type,
            'model_type': model_class_key
        },
        status=status.HTTP_404_NOT_FOUND
    )

logger.info(f"Found Django instance: {django_instance.pk}")

# 3. Push Django data to Sanity (outbound sync)
try:
    logger.info(f"Pushing Django {model_class_key} {django_instance.pk} to Sanity")

    # Call sync_to_sanity (outbound direction)
    result_sanity_id = sync_handler.sync_to_sanity(django_instance, created=False)

    if result_sanity_id:
        logger.info(f"Successfully pushed Django data to Sanity: {result_sanity_id}")
        return Response(
            {
                'success': True,
                'sanity_id': result_sanity_id,
                'document_type': document_type,
                'model_type': model_class_key,
                'instance_id': str(django_instance.pk),
                'message': f'Django data successfully pushed to Sanity',
                'django_updated_at': str(getattr(django_instance, 'updated_at', None))
            },
            status=status.HTTP_200_OK
        )
    else:
        logger.error(f"sync_to_sanity returned None for {django_instance.pk}")
        return Response(
            {
                'success': False,
                'error': 'Sync completed but returned no sanity_id',
                'sanity_id': sanity_id,
                'document_type': document_type
            },
            status=status.HTTP_500_INTERNAL_SERVER_ERROR
        )

def get(self, request):
    """
    Simple GET endpoint for testing endpoint accessibility.
    """
    return Response(
        {
            'status': 'ok',
            'message': 'Manual sync endpoint is accessible (Django → Sanity direction)',
            'endpoint': '/integrations/manual-sync/',
            'method': 'POST',
            'direction': 'Django → Sanity (overwrites Sanity with Django data)',
            'required_fields': ['sanity_id', 'document_type'],
            'supported_types': [
                'product', 'store', 'taxJurisdiction', 'zone', 'user',
                'salesTransaction', 'saleLineItem', 'salesReturn', 'storeCredit',
                'payment', 'refund', 'paymentMethod', 'transactionPayment',
                'promoCode', 'promotionalCampaign', 'abTest'
            ]
        },
        status=status.HTTP_200_OK
    )

┌─────────────────────────────────────────────────────────┐
│         Sanity Studio Document Action Click             │
│     (User clicks "Sync from Django" button)             │
└─────────────────────┬───────────────────────────────────┘
                      │
                      ▼
        ┌─────────────────────────────┐
        │   1. Authenticate Request   │
        │   (PosGateway or WebAuth)   │
        └─────────────┬───────────────┘
                      │
                      ▼
        ┌─────────────────────────────┐
        │   2. Validate Payload       │
        │   (sanity_id, doc_type)     │
        └─────────────┬───────────────┘
                      │
                      ▼
        ┌─────────────────────────────┐
        │   3. Select Sync Handler    │
        │   (based on document_type)  │
        └─────────────┬───────────────┘
                      │
                      ▼
        ┌─────────────────────────────┐
        │   4. Lookup Django Record   │
        │   (by sanity_id)            │
        └─────────────┬───────────────┘
                      │
                      ▼
        ┌─────────────────────────────┐
        │   5. Call sync_to_sanity()  │
        │   (outbound sync)           │
        └─────────────┬───────────────┘
                      │
                      ▼
        ┌─────────────────────────────┐
        │   6. Return Success/Error   │
        │   (200 OK or error status)  │
        └─────────────────────────────┘

{
  "success": true,
  "sanity_id": "product_123",
  "document_type": "product",
  "model_type": "products.Product",
  "instance_id": "550e8400-e29b-41d4-a716-446655440000",
  "message": "Django data successfully pushed to Sanity",
  "django_updated_at": "2025-11-17T15:30:00Z"
}

// Missing required field (400)
{
  "success": false,
  "error": "Missing required field: sanity_id"
}

// Unsupported document type (400)
{
  "success": false,
  "error": "No sync handler for document type: unknownType"
}

// Django record not found (404)
{
  "success": false,
  "error": "No Django record found with sanity_id: product_999",
  "sanity_id": "product_999",
  "document_type": "product",
  "model_type": "products.Product"
}

// Sync failure (500)
{
  "success": false,
  "error": "Failed to push to Sanity: Connection timeout",
  "sanity_id": "product_123",
  "document_type": "product"
}

// sanity/schemas/actions/syncFromDjango.js
export default function syncFromDjango(props) {
  return {
    label: 'Sync from Django',
    onHandle: async () => {
      const {id, type} = props.draft || props.published;

      try {
        const response = await fetch('/integrations/manual-sync/', {
          method: 'POST',
          headers: {
            'Content-Type': 'application/json',
            'Authorization': `Bearer \${getAuthToken()}`
          },
          body: JSON.stringify({
            sanity_id: id,
            document_type: type
          })
        });

        const result = await response.json();

        if (result.success) {
          props.onComplete();
          alert('Successfully synced from Django!');
        } else {
          alert(`Sync failed: \${result.error}`);
        }
      } catch (error) {
        alert(`Sync error: \${error.message}`);
      }
    }
  };
}

// Sanity Studio webhook setup
{
  "url": "https://api.example.com/integrations/sanity-webhook/",
  "dataset": "production",
  "filter": "!(_id in path('drafts.**'))",  // PRIMARY draft filter
  "projection": "{...}",                    // Full document
  "includeDrafts": false,
  "httpMethod": "POST",
  "headers": {
    "Sanity-Dataset": "production"
  },
  "secret": "your-webhook-secret-here"
}

# Webhook logging
logger.info("=== WEBHOOK ENDPOINT HIT ===")
logger.info(f"Request method: {request.method}")
logger.info(f"Request headers: {dict(request.headers)}")
logger.info(f"Verified webhook: {document_type} {sanity_id} (rev: {revision})")
logger.info(f"Queued task {task_result.id} for {document_type} {sanity_id}")

# Manual sync logging
logger.info("=== MANUAL SYNC REQUEST (Django → Sanity) ===")
logger.info(f"Looking up Django {model_class_key} with sanity_id: {sanity_id}")
logger.info(f"Pushing Django {model_class_key} {django_instance.pk} to Sanity")
logger.info(f"Successfully pushed Django data to Sanity: {result_sanity_id}")

Webhook signature verification failed

PING-PONG PREVENTED (cache): Webhook for product_123 skipped

Duplicate webhook detected - ignoring product product_123

No Django record found with sanity_id: product_999

{
  "granularity": "hour",
  "data": [
    {
      "hour": "2026-02-24T10:00:00+00:00",
      "inbound_count": 42,
      "outbound_count": 15,
      "failed_count": 0,
      "total": 57
    }
  ]
}

{
  "total_received_24h": 156,
  "by_status": {
    "completed": 150,
    "failed": 4,
    "processing": 2
  },
  "by_document_type": [
    {"document_type": "product", "count": 80},
    {"document_type": "transaction", "count": 45}
  ],
  "avg_processing_ms": 230
}

# Test GET endpoint
curl https://api.example.com/integrations/sanity-webhook/

# Test POST with signature (requires valid Sanity webhook)
# (Signature must be generated by Sanity - cannot be manually created)

# Test GET endpoint
curl https://api.example.com/integrations/manual-sync/

# Test POST with authentication
curl -X POST https://api.example.com/integrations/manual-sync/ \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "sanity_id": "product_123",
    "document_type": "product"
  }'