Receive messages

import express from "express";
import { Pool } from "pg";
import { verifySignature } from "./verify-signature"; // the verifier from the Signature verification guide
import { isFirstDelivery } from "./webhook-dedup"; // from the Retries and deduplication guide

const API = "https://api.ibubbl.com";
const HEADERS = {
  Authorization: `Bearer ${process.env.IBUBBL_API_KEY}`,
  "Content-Type": "application/json",
};
const AUTO_REPLY = "Thanks for your message! We'll get back to you shortly.";
const COOLDOWN_MS = 60 * 60 * 1000; // one auto-reply per conversation per hour

const db = new Pool({ connectionString: process.env.DATABASE_URL });
// Per instance, for the example. With several instances, keep it in the database too.
const lastAutoReplyAt = new Map<string, number>(); // conversation_id -> ms

const app = express();

// express.raw keeps the exact bytes iBubbl signed; express.json would not.
app.post("/ibubbl-webhooks", express.raw({ type: "application/json" }), async (req, res) => {
  // Node lower-cases header names: req.headers["X-iBubbl-Signature"] is undefined.
  const signature = req.headers["x-ibubbl-signature"];
  if (!verifySignature(req.body, signature, process.env.IBUBBL_WEBHOOK_SECRET!)) {
    return res.status(401).send("invalid signature");
  }

  const event = JSON.parse(req.body.toString("utf-8"));

  // At-least-once delivery: a unique constraint in your database lets each
  // event id through once, across every instance of your app.
  if (!(await isFirstDelivery(db, event))) return res.sendStatus(200);

  // Acknowledge now; do the work after the response.
  res.sendStatus(200);

  // Only a contact's message may trigger a send. Never react to
  // message.sent / delivered / read / failed: that loops.
  if (event.type !== "message.inbound") return;

  autoReply(event.data.message_id).catch((err) => console.error("auto-reply failed", err));
});

async function autoReply(inboundMessageId: string): Promise<void> {
  // Where to reply: contact_id for one-to-one, chat_guid for a group.
  const lookup = await fetch(`${API}/v1/messages/${inboundMessageId}`, {
    headers: HEADERS,
    signal: AbortSignal.timeout(10_000),
  });
  const inbound = await lookup.json();
  if (!lookup.ok) throw new Error(`lookup ${lookup.status} ${inbound.detail?.error}`);

  // Cooldown per conversation, claimed before sending.
  const last = lastAutoReplyAt.get(inbound.conversation_id) ?? 0;
  if (Date.now() - last < COOLDOWN_MS) return;
  lastAutoReplyAt.set(inbound.conversation_id, Date.now());

  const target =
    inbound.conversation_kind === "group"
      ? { chat_guid: inbound.chat_guid }
      : { contact_id: inbound.contact_id };

  // Deterministic key: a repeated event can never send a second reply.
  const payload = JSON.stringify({
    ...target,
    body: AUTO_REPLY,
    idempotency_key: `auto-reply-${inboundMessageId}`,
  });

  for (let attempt = 0; attempt < 3; attempt++) {
    let res: Response;
    try {
      res = await fetch(`${API}/v1/messages`, {
        method: "POST",
        headers: HEADERS,
        body: payload, // the same payload and key on every attempt
        signal: AbortSignal.timeout(10_000),
      });
    } catch {
      await new Promise((r) => setTimeout(r, 2 ** attempt * 1000)); // network error or timeout
      continue;
    }
    if (res.ok) return; // 202: accepted
    const error = (await res.json().catch(() => ({}))).detail?.error;
    if (res.status < 500) {
      // 4xx is final: contact_opted_out, unresponsive_contact, contact_blocked, ...
      console.warn(`auto-reply not sent: ${res.status} ${error}`);
      return;
    }
    await new Promise((r) => setTimeout(r, 2 ** attempt * 1000)); // 5xx: back off, retry
  }
}

app.listen(8000);

import json
import os
import time

import psycopg
import requests
from fastapi import BackgroundTasks, FastAPI, HTTPException, Request

from verify_signature import verify_signature  # the verifier from the Signature verification guide
from webhook_dedup import is_first_delivery  # from the Retries and deduplication guide

API = "https://api.ibubbl.com"
API_KEY = os.environ["IBUBBL_API_KEY"]
HEADERS = {"Authorization": f"Bearer {API_KEY}"}
AUTO_REPLY = "Thanks for your message! We'll get back to you shortly."
COOLDOWN_SECONDS = 60 * 60  # one auto-reply per conversation per hour

db = psycopg.connect(os.environ["DATABASE_URL"])
# Per instance, for the example. With several instances, keep it in the database too.
last_auto_reply_at: dict[str, float] = {}  # conversation_id -> unix time

app = FastAPI()


@app.post("/ibubbl-webhooks")
async def receive(request: Request, background: BackgroundTasks):
    # Read the raw bytes iBubbl signed, before any parsing.
    body = await request.body()
    # Starlette headers are case-insensitive; any spelling works.
    signature = request.headers.get("x-ibubbl-signature")
    if not verify_signature(body, signature, os.environ["IBUBBL_WEBHOOK_SECRET"]):
        raise HTTPException(status_code=401, detail="invalid signature")

    event = json.loads(body)

    # At-least-once delivery: a unique constraint in your database lets each
    # event id through once, across every instance of your app.
    if not is_first_delivery(db, event):
        return {"ok": True}

    # Only a contact's message may trigger a send. Never react to
    # message.sent / delivered / read / failed: that loops.
    if event["type"] == "message.inbound":
        # Runs after the response is sent, so iBubbl gets its 200 at once.
        background.add_task(auto_reply, event["data"]["message_id"])
    return {"ok": True}


def auto_reply(inbound_message_id: str) -> None:
    # Where to reply: contact_id for one-to-one, chat_guid for a group.
    lookup = requests.get(
        f"{API}/v1/messages/{inbound_message_id}", headers=HEADERS, timeout=10,
    )
    inbound = lookup.json()
    if not lookup.ok:
        raise RuntimeError(f"lookup {lookup.status_code} {inbound['detail']['error']}")

    # Cooldown per conversation, claimed before sending.
    conversation_id = inbound["conversation_id"]
    if time.time() - last_auto_reply_at.get(conversation_id, 0) < COOLDOWN_SECONDS:
        return
    last_auto_reply_at[conversation_id] = time.time()

    if inbound["conversation_kind"] == "group":
        target = {"chat_guid": inbound["chat_guid"]}
    else:
        target = {"contact_id": inbound["contact_id"]}

    # Deterministic key: a repeated event can never send a second reply.
    payload = {
        **target,
        "body": AUTO_REPLY,
        "idempotency_key": f"auto-reply-{inbound_message_id}",
    }

    for attempt in range(3):
        try:
            # The same payload and key on every attempt.
            res = requests.post(f"{API}/v1/messages", headers=HEADERS, json=payload, timeout=10)
        except requests.RequestException:
            time.sleep(2 ** attempt)  # network error or timeout
            continue
        if res.ok:
            return  # 202: accepted
        if res.status_code < 500:
            # 4xx is final: contact_opted_out, unresponsive_contact, contact_blocked, ...
            error = res.json().get("detail", {}).get("error") if res.content else None
            print(f"auto-reply not sent: {res.status_code} {error}")
            return
        time.sleep(2 ** attempt)  # 5xx: back off, retry