Coverage for apps/communications/webhooks.py: 90%

1"""SendGrid Event Webhook: delivery and open tracking for outbound email (add-sendgrid-email). 

2 

3SendGrid POSTs batches of events (processed, delivered, open, click, bounce, dropped, …) to 

4``/api/webhooks/sendgrid``. Each batch is verified with SendGrid's Signed Event Webhook key 

5(``SENDGRID_WEBHOOK_PUBLIC_KEY``); each event is matched to our ``Communication`` by the 

6``comm_id`` custom arg (set when sending) or by SendGrid's message id, recorded as a 

7``CommunicationEvent`` (once — by ``sg_event_id``) and moves the message status forward. 

8""" 

9 

10from __future__ import annotations 

11 

12import json 

13import logging 

14from datetime import UTC, datetime 

15 

16from django.conf import settings 

17from django.db import transaction 

18from ninja import Router 

19 

20from apps.communications.models import Communication, CommunicationEvent 

21from apps.communications.status import EVENT_TO_STATUS, advance_status 

22from domain.result import AppError, DomainError 

23 

24logger = logging.getLogger(__name__) 

25 

26router = Router(tags=["webhooks"]) 

27 

28# SendGrid event name → our event type. Event types without a status change (deferred, spam 

29# report, unsubscribe) are still recorded on the timeline. 

30SENDGRID_EVENTS = { 

31 "processed": "sent", 

32 "delivered": "delivered", 

33 "open": "opened", 

34 "click": "clicked", 

35 "bounce": "bounced", 

36 "dropped": "failed", 

37 "deferred": "deferred", 

38 "spamreport": "spam_report", 

39 "unsubscribe": "unsubscribed", 

40 "group_unsubscribe": "unsubscribed", 

41} 

42 

43 

44def _verify(request) -> None: 

45 public_key = settings.SENDGRID_WEBHOOK_PUBLIC_KEY 

46 if not public_key: 

47 raise DomainError([AppError("NOT_FOUND", "SendGrid webhook is not configured.")]) 

48 from sendgrid.helpers.eventwebhook import EventWebhook, EventWebhookHeader 

49 

50 signature = request.headers.get(EventWebhookHeader.SIGNATURE, "") 

51 timestamp = request.headers.get(EventWebhookHeader.TIMESTAMP, "") 

52 try: 

53 webhook = EventWebhook() 

54 key = webhook.convert_public_key_to_ecdsa(public_key) 

55 valid = webhook.verify_signature(request.body.decode("utf-8"), signature, timestamp, key) 

56 except Exception: # malformed key / signature 

57 valid = False 

58 if not valid: 

59 raise DomainError([AppError("FORBIDDEN", "Invalid webhook signature.")]) 

60 

61 

62def _find_communication(event: dict) -> Communication | None: 

63 comm_id = str(event.get("comm_id") or "") 

64 if comm_id.isdigit(): 

65 comm = Communication.objects.filter(pk=int(comm_id), channel="email").first() 

66 if comm is not None: 

67 return comm 

68 sg_message_id = str(event.get("sg_message_id") or "") 

69 if sg_message_id: 

70 # sg_message_id = "<X-Message-Id>.filter…": match on the id we stored when sending. 

71 prefix = sg_message_id.split(".", 1)[0] 

72 return Communication.objects.filter(provider_message_id=prefix).first() 

73 return None 

74 

75 

76def record_sendgrid_event(event: dict) -> bool: 

77 """Apply one SendGrid event. Returns True when it was recorded (False: unknown event type, 

78 unknown message, or a duplicate).""" 

79 event_type = SENDGRID_EVENTS.get(str(event.get("event") or "")) 

80 if event_type is None: 

81 return False 

82 comm = _find_communication(event) 

83 if comm is None: 

84 return False 

85 sg_event_id = event.get("sg_event_id") or None 

86 if sg_event_id and CommunicationEvent.objects.filter(provider_event_id=sg_event_id).exists(): 

87 return False 

88 occurred_at = None 

89 if isinstance(event.get("timestamp"), int | float): 

90 occurred_at = datetime.fromtimestamp(event["timestamp"], tz=UTC) 

91 with transaction.atomic(): 

92 CommunicationEvent.objects.create( 

93 communication=comm, 

94 event_type=event_type, 

95 provider_event_id=sg_event_id, 

96 occurred_at=occurred_at, 

97 ) 

98 new_status = EVENT_TO_STATUS.get(event_type) 

99 if new_status: 99 ↛ 104line 99 didn't jump to line 104

100 advanced = advance_status(comm.status, new_status) 

101 if advanced != comm.status: 101 ↛ 104line 101 didn't jump to line 104

102 comm.status = advanced 

103 comm.save(update_fields=["status", "updated_at"]) 

104 return True 

105 

106 

107@router.post("/webhooks/sendgrid", auth=None) 

108def sendgrid_events(request): 

109 """Public endpoint for SendGrid's Event Webhook (signature-verified, no login).""" 

110 _verify(request) 

111 try: 

112 events = json.loads(request.body or b"[]") 

113 except ValueError as exc: 

114 raise DomainError([AppError("VALIDATION", "Body must be a JSON array of events.")]) from exc 

115 if not isinstance(events, list): 115 ↛ 116line 115 didn't jump to line 116 because the condition on line 115 was never true

116 raise DomainError([AppError("VALIDATION", "Body must be a JSON array of events.")]) 

117 recorded = 0 

118 for event in events: 

119 if isinstance(event, dict): 119 ↛ 118line 119 didn't jump to line 118 because the condition on line 119 was always true

120 try: 

121 recorded += record_sendgrid_event(event) 

122 except Exception: # one bad event must not make SendGrid retry the whole batch 

123 logger.exception("SendGrid event could not be recorded: %s", event.get("sg_event_id")) 

124 return {"received": len(events), "recorded": recorded}