Tapbacks, Read Receipts & Typing Indicators

# message_id from the message.inbound event
curl -X POST https://api.ibubbl.com/v1/messages/8b3a4e72-1c2d-4e5f-9a0b-2c3d4e5f6a7b/reactions \
  -H "Authorization: Bearer $IBUBBL_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"type": "love"}'

const messageId = "8b3a4e72-1c2d-4e5f-9a0b-2c3d4e5f6a7b"; // from the message.inbound event

const res = await fetch(`https://api.ibubbl.com/v1/messages/${messageId}/reactions`, {
  method: "POST",
  headers: {
    Authorization: `Bearer ${process.env.IBUBBL_API_KEY}`,
    "Content-Type": "application/json",
  },
  body: JSON.stringify({ type: "love" }),
  signal: AbortSignal.timeout(10_000),
});

const data = await res.json();
if (!res.ok) throw new Error(`${res.status} ${data.detail?.error}`);
console.log(data.id, data.status); // keep data.id to remove it later

import os

import requests

API_KEY = os.environ["IBUBBL_API_KEY"]

message_id = "8b3a4e72-1c2d-4e5f-9a0b-2c3d4e5f6a7b"  # from the message.inbound event

res = requests.post(
    f"https://api.ibubbl.com/v1/messages/{message_id}/reactions",
    headers={"Authorization": f"Bearer {API_KEY}"},
    json={"type": "love"},
    timeout=10,
)
data = res.json()
if not res.ok:
    raise RuntimeError(f"{res.status_code} {data['detail']['error']}")
print(data["id"], data["status"])  # keep data["id"] to remove it later

# conversation_id from the message.inbound event
curl -X POST https://api.ibubbl.com/v1/conversations/3a4b5c6d-7e8f-9a0b-1c2d-3e4f5a6b7c8d/read \
  -H "Authorization: Bearer $IBUBBL_API_KEY"

const conversationId = "3a4b5c6d-7e8f-9a0b-1c2d-3e4f5a6b7c8d"; // from the message.inbound event

const res = await fetch(`https://api.ibubbl.com/v1/conversations/${conversationId}/read`, {
  method: "POST",
  headers: { Authorization: `Bearer ${process.env.IBUBBL_API_KEY}` },
  signal: AbortSignal.timeout(10_000),
});

const data = await res.json();
if (!res.ok) throw new Error(`${res.status} ${data.detail?.error}`);
console.log(data.read_receipt, data.read_receipt_skipped_reason);

import os

import requests

API_KEY = os.environ["IBUBBL_API_KEY"]

conversation_id = "3a4b5c6d-7e8f-9a0b-1c2d-3e4f5a6b7c8d"  # from the message.inbound event

res = requests.post(
    f"https://api.ibubbl.com/v1/conversations/{conversation_id}/read",
    headers={"Authorization": f"Bearer {API_KEY}"},
    timeout=10,
)
data = res.json()
if not res.ok:
    raise RuntimeError(f"{res.status_code} {data['detail']['error']}")
print(data["read_receipt"], data["read_receipt_skipped_reason"])

# The calls, in order. The ids come from the message.inbound event.
API=https://api.ibubbl.com
CONVERSATION_ID=3a4b5c6d-7e8f-9a0b-1c2d-3e4f5a6b7c8d
INBOUND_ID=5b6c7d8e-9f0a-1b2c-3d4e-5f6a7b8c9d0e

# 1. Read receipt, then the typing dots.
curl -X POST "$API/v1/conversations/$CONVERSATION_ID/read" -H "Authorization: Bearer $IBUBBL_API_KEY"
curl -X POST "$API/v1/conversations/$CONVERSATION_ID/typing" -H "Authorization: Bearer $IBUBBL_API_KEY"

# 2. Every 5 seconds while the reply is being written:
curl -X POST "$API/v1/conversations/$CONVERSATION_ID/typing" -H "Authorization: Bearer $IBUBBL_API_KEY"

# 3. Send the reply (it clears the dots). One key per inbound message, so a
#    repeated event can never send a second reply.
curl -X POST "$API/v1/messages" \
  -H "Authorization: Bearer $IBUBBL_API_KEY" \
  -H "Content-Type: application/json" \
  -d "{\"contact_id\": \"c8a1f4e2-3b9d-4f7c-a8e5-1d2c3b4a5e6f\", \"body\": \"Tuesday at 3 works. See you then!\", \"idempotency_key\": \"ai-reply-$INBOUND_ID\"}"

# 4. Only if you decided not to reply:
curl -X DELETE "$API/v1/conversations/$CONVERSATION_ID/typing" -H "Authorization: Bearer $IBUBBL_API_KEY"

const API = "https://api.ibubbl.com";
const HEADERS = {
  Authorization: `Bearer ${process.env.IBUBBL_API_KEY}`,
  "Content-Type": "application/json",
};

async function call(method: string, path: string, body?: unknown): Promise<any> {
  const res = await fetch(`${API}${path}`, {
    method,
    headers: HEADERS,
    body: body === undefined ? undefined : JSON.stringify(body),
    signal: AbortSignal.timeout(10_000),
  });
  const data = await res.json();
  if (!res.ok) throw new Error(`${res.status} ${data.detail?.error}`);
  return data;
}

// Run for a message.inbound event, after the signature check and
// de-duplication. generateReply is your model call.
export async function replyWithAI(
  event: { data: { message_id: string; conversation_id: string; body: string } },
  generateReply: (text: string) => Promise<string | null>,
): Promise<void> {
  const { message_id: inboundId, conversation_id: conversationId, body } = event.data;
  const typingPath = `/v1/conversations/${conversationId}/typing`;

  // Where to reply, and whether it's one-to-one (typing is 1:1 only).
  const inbound = await call("GET", `/v1/messages/${inboundId}`);
  if (inbound.conversation_kind === "group") return;

  // 1. Read receipt, then the dots. Neither may stop the reply.
  await call("POST", `/v1/conversations/${conversationId}/read`).catch(console.warn);
  const typing = await call("POST", typingPath).catch(() => null);

  // 2. Keep the dots going while the model works ("ignored" needs no refresh).
  const refresh =
    typing?.status === "started"
      ? setInterval(() => call("POST", typingPath).catch(() => {}), 5_000)
      : undefined;

  let sent = false;
  try {
    const reply = await generateReply(body);
    if (reply) {
      // 3. Send. The deterministic key makes a repeated event harmless.
      await call("POST", "/v1/messages", {
        contact_id: inbound.contact_id,
        body: reply,
        idempotency_key: `ai-reply-${inboundId}`,
      });
      sent = true;
    }
  } finally {
    clearInterval(refresh);
    // 4. No reply after all: stop the dots.
    if (!sent && typing?.status === "started") {
      await call("DELETE", typingPath).catch(() => {});
    }
  }
}

import os
import threading
from typing import Callable, Optional

import requests

API = "https://api.ibubbl.com"
API_KEY = os.environ["IBUBBL_API_KEY"]
HEADERS = {"Authorization": f"Bearer {API_KEY}"}


def call(method: str, path: str, body: Optional[dict] = None) -> dict:
    res = requests.request(method, f"{API}{path}", headers=HEADERS, json=body, timeout=10)
    data = res.json()
    if not res.ok:
        raise RuntimeError(f"{res.status_code} {data['detail']['error']}")
    return data


def reply_with_ai(event: dict, generate_reply: Callable[[str], Optional[str]]) -> None:
    """Run for a message.inbound event, after the signature check and
    de-duplication. generate_reply is your model call."""
    inbound_id = event["data"]["message_id"]
    conversation_id = event["data"]["conversation_id"]
    typing_path = f"/v1/conversations/{conversation_id}/typing"

    # Where to reply, and whether it's one-to-one (typing is 1:1 only).
    inbound = call("GET", f"/v1/messages/{inbound_id}")
    if inbound["conversation_kind"] == "group":
        return

    # 1. Read receipt, then the dots. Neither may stop the reply.
    try:
        call("POST", f"/v1/conversations/{conversation_id}/read")
    except Exception as exc:
        print("read receipt:", exc)
    try:
        typing = call("POST", typing_path)["status"] == "started"
    except Exception:
        typing = False

    # 2. Keep the dots going while the model works.
    done = threading.Event()

    def refresh() -> None:
        while not done.wait(5):
            try:
                call("POST", typing_path)
            except Exception:
                pass

    if typing:
        threading.Thread(target=refresh, daemon=True).start()

    sent = False
    try:
        reply = generate_reply(event["data"]["body"])
        if reply:
            # 3. Send. The deterministic key makes a repeated event harmless.
            call("POST", "/v1/messages", {
                "contact_id": inbound["contact_id"],
                "body": reply,
                "idempotency_key": f"ai-reply-{inbound_id}",
            })
            sent = True
    finally:
        done.set()
        # 4. No reply after all: stop the dots.
        if typing and not sent:
            try:
                call("DELETE", typing_path)
            except Exception:
                pass