Handling Retries

import crypto from 'crypto';

function idempotencyKey(eventData: any, timestamp: string): string {
  if (eventData.field === 'message' && eventData.payload.message_id) {
    // Outbound: state transition is unique
    return `msg:${eventData.payload.message_id}:${eventData.payload.message_status}`;
  }
  if (eventData.sub_type === 'message.received') {
    // Inbound: from + received_at + text hash
    const { from, received_at, text } = eventData.payload;
    const textHash = crypto.createHash('sha256').update(text ?? '').digest('hex').slice(0, 16);
    return `in:${from}:${received_at}:${textHash}`;
  }
  if (eventData.field === 'templates') {
    return `tpl:${eventData.payload.template_id}:${eventData.payload.status}`;
  }
  // Fallback: timestamp + canonical body hash
  const hash = crypto.createHash('sha256').update(JSON.stringify(eventData)).digest('hex');
  return `raw:${timestamp}:${hash}`;
}

async function handleWebhook(eventData: any, timestamp: string) {
  const key = idempotencyKey(eventData, timestamp);

  const existing = await db.webhookEvents.findUnique({ where: { idempotencyKey: key } });
  if (existing) {
    console.log(`Event ${key} already processed`);
    return;
  }

  await processEvent(eventData);

  await db.webhookEvents.create({
    data: { idempotencyKey: key, eventType: eventData.field, processedAt: new Date() }
  });
}

import hashlib, json

def idempotency_key(event_data: dict, timestamp: str) -> str:
    if event_data.get('field') == 'message' and event_data['payload'].get('message_id'):
        return f"msg:{event_data['payload']['message_id']}:{event_data['payload']['message_status']}"
    if event_data.get('sub_type') == 'message.received':
        p = event_data['payload']
        text_hash = hashlib.sha256((p.get('text') or '').encode()).hexdigest()[:16]
        return f"in:{p['from']}:{p['received_at']}:{text_hash}"
    if event_data.get('field') == 'templates':
        return f"tpl:{event_data['payload']['template_id']}:{event_data['payload']['status']}"
    body_hash = hashlib.sha256(json.dumps(event_data, sort_keys=True).encode()).hexdigest()
    return f"raw:{timestamp}:{body_hash}"

async def handle_webhook(event_data: dict, timestamp: str):
    key = idempotency_key(event_data, timestamp)

    existing = db.webhook_events.find_unique(where={"idempotency_key": key})
    if existing:
        print(f"Event {key} already processed")
        return

    await process_event(event_data)

    db.webhook_events.create({
        "idempotency_key": key,
        "event_type": event_data['field'],
        "processed_at": datetime.now()
    })

import (
    "crypto/sha256"
    "encoding/hex"
    "encoding/json"
    "fmt"
)

func idempotencyKey(event WebhookEvent, timestamp string) string {
    if event.Field == "message" && event.Payload.MessageID != "" {
        return fmt.Sprintf("msg:%s:%s", event.Payload.MessageID, event.Payload.MessageStatus)
    }
    if event.SubType != nil && *event.SubType == "message.received" {
        sum := sha256.Sum256([]byte(event.Payload.Text))
        return fmt.Sprintf("in:%s:%s:%s", event.Payload.From, event.Payload.ReceivedAt, hex.EncodeToString(sum[:])[:16])
    }
    if event.Field == "templates" {
        return fmt.Sprintf("tpl:%s:%s", event.Payload.TemplateID, event.Payload.Status)
    }
    body, _ := json.Marshal(event)
    sum := sha256.Sum256(body)
    return fmt.Sprintf("raw:%s:%s", timestamp, hex.EncodeToString(sum[:]))
}

func handleWebhook(event WebhookEvent, timestamp string) error {
    key := idempotencyKey(event, timestamp)

    var exists bool
    if err := db.QueryRow("SELECT EXISTS(SELECT 1 FROM webhook_events WHERE idempotency_key = $1)", key).Scan(&exists); err != nil {
        return err
    }
    if exists {
        log.Printf("Event %s already processed", key)
        return nil
    }

    if err := processEvent(event); err != nil {
        return err
    }

    _, err := db.Exec(
        "INSERT INTO webhook_events (idempotency_key, event_type, processed_at) VALUES ($1, $2, $3)",
        key, event.Field, time.Now(),
    )
    return err
}

await db.$transaction(async (tx) => {
  // Record event processing start
  await tx.webhookEvents.create({
    data: {
      eventId,
      eventType: field,
      status: 'processing'
    }
  });

  // Process business logic
  if (field === 'message') {
    await tx.messages.update({
      where: { id: eventData.payload.message_id },
      data: { status: eventData.payload.message_status }
    });
  }

  // Mark as completed
  await tx.webhookEvents.update({
    where: { eventId },
    data: { status: 'completed' }
  });
});

with db.transaction():
    # Record event processing start
    webhook_event = WebhookEvent(
        event_id=event_id,
        event_type=field,
        status="processing"
    )
    db.add(webhook_event)

    # Process business logic
    if field == 'message':
        message = db.messages.find_by_id(event_data['payload']['message_id'])
        message.status = event_data['payload']['message_status']

    # Mark as completed
    webhook_event.status = "completed"
    db.commit()