Mock CTO Interview

# ======================================================================
# Q: CTO: "Design a notification service (email, SMS, push) for 1 million
# users."
#
# You, with a structure: 1) Requirements (volume, latency, user preferences,
# opt-out/GDPR). 2) An API that accepts notification requests (idempotency key).
# 3) Outbox and queue (RabbitMQ/SQS or Kafka), separate workers per channel. 4)
# Templates and localization. 5) Provider rate limits, retries with backoff,
# dead-letter queue. 6) Status stored per message, delivery webhooks. 7) Metrics
# (sent, failed, latency) and alerts. 8) Horizontal scaling with more workers.
# Priorities with separate queues, so that an OTP does not wait behind 1 million
# marketing messages.
# ======================================================================

CELERY_TASK_ROUTES = {
    'notifications.tasks.send_otp':       {'queue': 'critical'},
    'notifications.tasks.send_receipt':   {'queue': 'default'},
    'notifications.tasks.send_marketing': {'queue': 'bulk'},
}

@shared_task(rate_limit='50/s', autoretry_for=(ProviderError,),
             retry_backoff=True, max_retries=6)
def send_marketing(notification_id): ...

# celery -A proj worker -Q critical -c 8
# celery -A proj worker -Q default,bulk -c 4

# ======================================================================
# Q: CTO: "We run on ECS Fargate with RDS and SQS. How would you set up the
# Celery workers and what happens during a deploy?"
#
# You: "Separate ECS services for web, workers and beat, using the same image.
# Beat runs as a single task, otherwise scheduled jobs run twice. SQS is the
# broker, with a visibility timeout longer than the slowest task and a DLQ with
# a redrive policy. The result backend is not SQS, but Redis or the database, if
# needed. On deploy, ECS sends SIGTERM and Celery does a warm shutdown: it
# finishes the current task. Since we use acks_late and idempotent tasks,
# anything that gets cut off safely reruns. Workers autoscale based on queue
# length in CloudWatch. Migrations run as a one-off task before the new release
# and are always backwards compatible."
# ======================================================================

CELERY_BROKER_URL = 'sqs://'          # credentials from the ECS task role
CELERY_BROKER_TRANSPORT_OPTIONS = {
    'region': 'eu-central-1',
    'visibility_timeout': 3600,        # > the slowest task
    'predefined_queues': {
        'default': {'url': 'https://sqs.eu-central-1.amazonaws.com/<account>/default'},
    },
}
CELERY_TASK_ACKS_LATE = True
CELERY_WORKER_PREFETCH_MULTIPLIER = 1  # fair distribution across long tasks

# ======================================================================
# Q: CTO: "How do you publish the OrderPaid event so it is not lost if the
# broker goes down?"
#
# You: "Transactional outbox. The order change and the outbox row are written in
# the same transaction. A relay publishes the rows that have not been sent. With
# skip_locked, multiple relays can run without picking up the same events.
# Delivery is at-least-once, so consumers deduplicate by event_id."
# ======================================================================

with transaction.atomic():
    order.mark_paid()
    OutboxEvent.objects.create(event_type='order.paid', aggregate_id=order.id,
                               payload={'order_id': order.id})

@shared_task
def relay_outbox():
    with transaction.atomic():
        batch = (OutboxEvent.objects.select_for_update(skip_locked=True)
                 .filter(published_at=None).order_by('id')[:100])
        for ev in batch:
            broker.publish(ev.event_type, ev.payload, message_id=str(ev.id))
            ev.published_at = timezone.now()
            ev.save(update_fields=['published_at'])

# ======================================================================
# Q: CTO: "How do you process the payment provider's webhooks?"
#
# You: "The view does only three things: it verifies the HMAC signature, stores
# the event with a unique event_id and responds 200 immediately. The processing
# happens in a Celery task that is idempotent and changes the payment status
# through a state machine. If the provider sends the same event twice, the
# second one is ignored. A beat job resends any events that were left
# unprocessed."
# ======================================================================

@csrf_exempt
def psp_webhook(request):
    if not verify_signature(request, settings.PSP_WEBHOOK_SECRET):
        return HttpResponse(status=400)
    event = json.loads(request.body)
    _, created = WebhookEvent.objects.get_or_create(
        event_id=event['id'], defaults={'payload': event})
    if created:
        process_payment_event.delay(event['id'])
    return HttpResponse(status=200)

@shared_task(acks_late=True)
def process_payment_event(event_id):
    with transaction.atomic():
        ev = WebhookEvent.objects.select_for_update().get(event_id=event_id)
        if ev.processed_at:
            return                                    # already processed
        payment = Payment.objects.select_for_update().get(
            psp_reference=ev.payload['data']['reference'])
        payment.apply_event(ev.payload['type'])       # allowed transitions only
        ev.processed_at = timezone.now()
        ev.save(update_fields=['processed_at'])

# ======================================================================
# Q: CTO: "The mobile app retries after a timeout and the customer gets charged
# twice. How do you prevent it?"
#
# You: "With an Idempotency-Key. The client sends a unique key per payment
# attempt. I store it with a unique constraint on (user, key) together with a
# hash of the request body. If the same key arrives again, I return the stored
# response instead of charging again. If the first call is still running, I
# return 409. I also pass the same key to the PSP, so I am protected on its side
# too. I do not make the PSP call inside an open DB transaction."
# ======================================================================

def create_payment(request):
    key = request.headers.get('Idempotency-Key')
    if not key:
        return Response({'detail': 'Idempotency-Key required'}, status=400)
    body_hash = hashlib.sha256(request.body).hexdigest()
    record, created = IdempotencyRecord.objects.get_or_create(
        user=request.user, key=key, defaults={'request_hash': body_hash})
    if not created:
        if record.request_hash != body_hash:
            return Response(status=422)          # same key, different request
        if record.response is None:
            return Response(status=409)          # still in progress
        return Response(record.response, status=record.status_code)

    payment = payments.charge(request.user, request.data, idempotency_key=key)
    record.response = PaymentSerializer(payment).data
    record.status_code = 201
    record.save(update_fields=['response', 'status_code'])
    return Response(record.response, status=201)