five conversation webhooks is the cap, and the sixth is rejected

"""Report Twilio conversations at the five conversation-webhook ceiling (50361).

Read only. GET requests and nothing else: give this an API Key with read access
rather than the account auth token. The repair is printed, never performed,
because this script holds a credential to an account that can send messages and
spend money.
"""
import argparse
import logging
import os
import sys

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("twilio_conversation_webhook_limit_audit")

CONVERSATIONS = "https://conversations.twilio.com/v1"

LIMIT = 5


def destination(webhook):
    """A comparable key for where one conversation webhook delivers. Pure.

    Nothing dedupes on the URL when a webhook is created, so a retried create
    leaves two webhooks with two SIDs and one destination. Normalising case and
    the trailing slash is what makes those two compare equal; a Studio target
    has no URL at all and is keyed on its flow_sid instead.
    """
    cfg = webhook.get("configuration") or {}
    target = str(webhook.get("target") or "").strip().lower()
    if target == "studio":
        return "studio %s" % (str(cfg.get("flow_sid") or "").strip() or "(no flow)")
    url = str(cfg.get("url") or "").strip().lower().rstrip("/")
    method = str(cfg.get("method") or "").strip().upper()
    return "%s %s %s" % (target or "(no target)", method or "(no method)",
                         url or "(no url)")


def webhook_total(page):
    """The number of webhooks on a conversation. Pure.

    meta.total is the authority: counting the array is right only while five
    entries fit in one page, which stops being true the moment somebody passes a
    smaller PageSize. Counting is the fallback, not the method.
    """
    meta = page.get("meta") or {}
    raw = meta.get("total")
    try:
        if raw is not None:
            return int(raw)
    except (TypeError, ValueError):
        pass
    return len(page.get("webhooks") or [])


def verdict(total, webhooks):
    """Classify one conversation against the five-webhook cap. Pure.

    Duplicates matter more than the raw count: at the ceiling they are the free
    slot, and below it they mean the endpoint is being called twice for every
    event. Returns (state, detail).
    """
    seen = {}
    for w in webhooks or []:
        seen.setdefault(destination(w), []).append(str(w.get("sid") or "?"))
    dupes = {k: v for k, v in seen.items() if len(v) > 1}
    dupe_note = ""
    if dupes:
        first = sorted(dupes)[0]
        dupe_note = (" %d destination(s) are registered more than once, including "
                     "%s (%s)." % (len(dupes), first, ", ".join(dupes[first])))

    if total >= LIMIT and dupes:
        return ("at-limit-duplicates",
                "%d webhook(s): at the cap of %d, so the next create is rejected "
                "with 50361.%s Removing a duplicate frees a slot without losing "
                "an integration." % (total, LIMIT, dupe_note))

    if total >= LIMIT:
        return ("at-limit",
                "%d webhook(s): at the cap of %d. The next create is rejected "
                "with 50361, and the rejection lands on whichever integration "
                "deploys last." % (total, LIMIT))

    if dupes:
        return ("duplicates",
                "%d webhook(s), below the cap of %d, but%s Your endpoint is "
                "being called twice for every event." % (total, LIMIT, dupe_note))

    if total == LIMIT - 1:
        return ("near-limit",
                "%d webhook(s): one slot left before creates start failing with "
                "50361." % total)

    if total == 0:
        return ("none",
                "no conversation-scoped webhooks. Events reach the account or "
                "service-level configuration only.")

    return ("headroom", "%d webhook(s), %d slot(s) left" % (total, LIMIT - total))


def get(session, url, **params):
    r = session.get(url, params=params, timeout=30)
    if r.status_code in (401, 403):
        raise SystemExit("%d from Twilio: check TWILIO_ACCOUNT_SID and that the "
                         "API key belongs to that account with read access"
                         % r.status_code)
    r.raise_for_status()
    return r.json()


def paged(session, url, key, limit, **first):
    params = dict(first)
    params.setdefault("PageSize", 50)
    out = []
    while url and len(out) < limit:
        page = get(session, url, **params)
        out.extend(page.get(key, []))
        url = (page.get("meta") or {}).get("next_page_url")
        params = {}
    return out[:limit]


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--max-conversations", type=int, default=200,
                    help="stop after this many conversations; one GET each")
    args = ap.parse_args()

    account = os.environ.get("TWILIO_ACCOUNT_SID")
    key = os.environ.get("TWILIO_API_KEY")
    secret = os.environ.get("TWILIO_API_SECRET")
    if not (account and key and secret):
        log.error("set TWILIO_ACCOUNT_SID, TWILIO_API_KEY and TWILIO_API_SECRET "
                  "(an API Key with read access, not the auth token)")
        return 2

    session = requests.Session()
    session.auth = (key, secret)

    conversations = paged(session, CONVERSATIONS + "/Conversations",
                          "conversations", args.max_conversations)
    if not conversations:
        log.info("no conversations on this account")
        return 0

    bad = 0
    for conv in conversations:
        sid = conv.get("sid")
        page = get(session, "%s/Conversations/%s/Webhooks" % (CONVERSATIONS, sid),
                   PageSize=50)
        webhooks = page.get("webhooks") or []
        state, detail = verdict(webhook_total(page), webhooks)
        line = "%-19s %s  %s" % (state, sid, detail)

        if state in ("headroom", "none"):
            log.info(line)
            continue
        if state == "near-limit":
            log.info(line)
            continue

        bad += 1
        log.warning(line)
        for w in webhooks:
            log.warning("    %s  %s", w.get("sid"), destination(w))
        log.warning("  repair: remove the stale or duplicate webhook by SID at "
                    "%s/Conversations/%s/Webhooks/{WebhookSid}, or move the "
                    "integration onto the account-level webhook configuration so "
                    "it stops taking a slot on every conversation.",
                    CONVERSATIONS, sid)

    log.info("%d conversation(s), %d at the five webhook ceiling or carrying "
             "duplicates", len(conversations), bad)
    return 1 if bad else 0


if __name__ == "__main__":
    sys.exit(main())

/**
 * Report Twilio conversations at the five conversation-webhook ceiling (50361).
 *
 * Read only. GET requests and nothing else: give this an API Key with read
 * access rather than the account auth token. The repair is printed, never
 * performed.
 */
const CONVERSATIONS = 'https://conversations.twilio.com/v1';

const LIMIT = 5;

/**
 * A comparable key for where one conversation webhook delivers. Pure.
 *
 * Nothing dedupes on the URL when a webhook is created, so a retried create
 * leaves two webhooks with two SIDs and one destination. Normalising case and
 * the trailing slash is what makes those two compare equal; a Studio target has
 * no URL at all and is keyed on its flow_sid instead.
 */
export function destination(webhook) {
  const cfg = webhook.configuration ?? {};
  const target = String(webhook.target ?? '').trim().toLowerCase();
  if (target === 'studio') {
    return `studio ${String(cfg.flow_sid ?? '').trim() || '(no flow)'}`;
  }
  const url = String(cfg.url ?? '').trim().toLowerCase().replace(/\/+$/, '');
  const method = String(cfg.method ?? '').trim().toUpperCase();
  return `${target || '(no target)'} ${method || '(no method)'} ${url || '(no url)'}`;
}

/**
 * The number of webhooks on a conversation. Pure.
 *
 * meta.total is the authority: counting the array is right only while five
 * entries fit in one page. Counting is the fallback, not the method.
 */
export function webhookTotal(page) {
  const raw = (page.meta ?? {}).total;
  const n = Number(raw);
  if (raw !== null && raw !== undefined && raw !== '' && Number.isFinite(n)) {
    return n;
  }
  return (page.webhooks ?? []).length;
}

/**
 * Classify one conversation against the five-webhook cap. Pure.
 *
 * Duplicates matter more than the raw count: at the ceiling they are the free
 * slot, and below it they mean the endpoint is called twice for every event.
 * Returns [state, detail].
 */
export function verdict(total, webhooks) {
  const seen = new Map();
  for (const w of webhooks ?? []) {
    const key = destination(w);
    if (!seen.has(key)) seen.set(key, []);
    seen.get(key).push(String(w.sid ?? '?'));
  }
  const dupes = [...seen.entries()].filter(([, sids]) => sids.length > 1).sort();
  let dupeNote = '';
  if (dupes.length) {
    const [first, sids] = dupes[0];
    dupeNote = ` ${dupes.length} destination(s) are registered more than once, ` +
      `including ${first} (${sids.join(', ')}).`;
  }

  if (total >= LIMIT && dupes.length) {
    return ['at-limit-duplicates',
      `${total} webhook(s): at the cap of ${LIMIT}, so the next create is ` +
      `rejected with 50361.${dupeNote} Removing a duplicate frees a slot ` +
      'without losing an integration.'];
  }

  if (total >= LIMIT) {
    return ['at-limit',
      `${total} webhook(s): at the cap of ${LIMIT}. The next create is rejected ` +
      'with 50361, and the rejection lands on whichever integration deploys last.'];
  }

  if (dupes.length) {
    return ['duplicates',
      `${total} webhook(s), below the cap of ${LIMIT}, but${dupeNote} Your ` +
      'endpoint is being called twice for every event.'];
  }

  if (total === LIMIT - 1) {
    return ['near-limit',
      `${total} webhook(s): one slot left before creates start failing with 50361.`];
  }

  if (total === 0) {
    return ['none',
      'no conversation-scoped webhooks. Events reach the account or ' +
      'service-level configuration only.'];
  }

  return ['headroom', `${total} webhook(s), ${LIMIT - total} slot(s) left`];
}

function authHeader(key, secret) {
  return `Basic ${Buffer.from(`${key}:${secret}`).toString('base64')}`;
}

async function get(auth, url, params = {}) {
  const u = new URL(url);
  for (const [k, v] of Object.entries(params)) u.searchParams.set(k, v);
  const res = await fetch(u, { headers: { Authorization: auth } });
  if (res.status === 401 || res.status === 403) {
    throw new Error(`${res.status} from Twilio: check TWILIO_ACCOUNT_SID and ` +
                    'that the API key belongs to that account with read access');
  }
  if (!res.ok) throw new Error(`${res.status} from ${u.pathname}`);
  return res.json();
}

async function paged(auth, url, key, limit, first = {}) {
  let params = { PageSize: 50, ...first };
  const out = [];
  while (url && out.length < limit) {
    const page = await get(auth, url, params);
    out.push(...(page[key] ?? []));
    url = (page.meta ?? {}).next_page_url ?? null;
    params = {};
  }
  return out.slice(0, limit);
}

async function main() {
  const account = process.env.TWILIO_ACCOUNT_SID;
  const key = process.env.TWILIO_API_KEY;
  const secret = process.env.TWILIO_API_SECRET;
  if (!account || !key || !secret) {
    console.error('set TWILIO_ACCOUNT_SID, TWILIO_API_KEY and TWILIO_API_SECRET ' +
                  '(an API Key with read access, not the auth token)');
    process.exitCode = 2;
    return;
  }
  const auth = authHeader(key, secret);

  const conversations = await paged(auth, `${CONVERSATIONS}/Conversations`,
                                    'conversations', 200);
  if (!conversations.length) {
    console.log('no conversations on this account');
    return;
  }

  let bad = 0;
  for (const conv of conversations) {
    const page = await get(auth, `${CONVERSATIONS}/Conversations/${conv.sid}/Webhooks`,
                           { PageSize: 50 });
    const webhooks = page.webhooks ?? [];
    const [state, detail] = verdict(webhookTotal(page), webhooks);
    const line = `${state.padEnd(19)} ${conv.sid}  ${detail}`;

    if (state === 'headroom' || state === 'none' || state === 'near-limit') {
      console.log(line);
      continue;
    }

    bad += 1;
    console.warn(line);
    for (const w of webhooks) console.warn(`    ${w.sid}  ${destination(w)}`);
    console.warn('  repair: remove the stale or duplicate webhook by SID at ' +
                 `${CONVERSATIONS}/Conversations/${conv.sid}/Webhooks/{WebhookSid}, ` +
                 'or move the integration onto the account-level webhook ' +
                 'configuration so it stops taking a slot on every conversation.');
  }

  console.log(`${conversations.length} conversation(s), ${bad} at the five ` +
              'webhook ceiling or carrying duplicates');
  process.exitCode = bad ? 1 : 0;
}

// Only run when invoked directly. The test file imports this module, and without
// the guard main() would run there too, fail on the missing credentials and set a
// non-zero exit code that fails the whole test file even as every test passes.
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from twilio_conversation_webhook_limit_audit import (
    destination, verdict, webhook_total)

URL = "https://app.example.com/hook"


def hook(sid, url=URL, target="webhook", method="POST", flow=None):
    cfg = {"url": url, "method": method}
    if flow:
        cfg = {"flow_sid": flow}
    return {"sid": sid, "target": target, "configuration": cfg}


def distinct(n):
    return [hook("WH%d" % i, "%s/%d" % (URL, i)) for i in range(n)]


def test_five_distinct_webhooks_is_the_ceiling():
    state, detail = verdict(5, distinct(5))
    assert state == "at-limit"
    assert "50361" in detail


def test_a_duplicate_at_the_ceiling_is_a_free_slot():
    hooks = distinct(4) + [hook("WH9", "%s/0" % URL)]
    state, detail = verdict(5, hooks)
    assert state == "at-limit-duplicates"
    assert "frees a slot" in detail


def test_a_duplicate_below_the_ceiling_is_still_a_finding():
    state, detail = verdict(2, [hook("WH1"), hook("WH2")])
    assert state == "duplicates"
    assert "twice for every event" in detail


def test_four_distinct_webhooks_is_one_slot_from_failing():
    assert verdict(4, distinct(4))[0] == "near-limit"


def test_an_empty_conversation_is_not_a_finding():
    assert verdict(0, [])[0] == "none"
    assert verdict(2, distinct(2))[0] == "headroom"


def test_destination_ignores_case_and_a_trailing_slash():
    assert destination(hook("WH1", "https://App.Example.com/hook/")) == \
        destination(hook("WH2", URL))


def test_a_studio_target_is_keyed_on_the_flow():
    assert destination(hook("WH1", target="studio", flow="FW1")) == "studio FW1"
    assert destination(hook("WH1", target="studio", flow="FW1")) != \
        destination(hook("WH2", target="studio", flow="FW2"))


def test_meta_total_wins_over_the_length_of_the_page():
    # A smaller PageSize returns fewer entries than the conversation holds, and
    # counting the array would report headroom on a conversation at the cap.
    page = {"webhooks": distinct(2), "meta": {"total": 5}}
    assert webhook_total(page) == 5
    assert webhook_total({"webhooks": distinct(2), "meta": {}}) == 2

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { destination, verdict, webhookTotal }
  from './twilio-conversation-webhook-limit-audit.mjs';

const URL_ = 'https://app.example.com/hook';

const hook = (sid, { url = URL_, target = 'webhook', method = 'POST', flow = null } = {}) => ({
  sid, target, configuration: flow ? { flow_sid: flow } : { url, method },
});

const distinct = (n) =>
  Array.from({ length: n }, (_, i) => hook(`WH${i}`, { url: `${URL_}/${i}` }));

test('five distinct webhooks is the ceiling', () => {
  const [state, detail] = verdict(5, distinct(5));
  assert.equal(state, 'at-limit');
  assert.match(detail, /50361/);
});

test('a duplicate at the ceiling is a free slot', () => {
  const hooks = [...distinct(4), hook('WH9', { url: `${URL_}/0` })];
  const [state, detail] = verdict(5, hooks);
  assert.equal(state, 'at-limit-duplicates');
  assert.match(detail, /frees a slot/);
});

test('a duplicate below the ceiling is still a finding', () => {
  const [state, detail] = verdict(2, [hook('WH1'), hook('WH2')]);
  assert.equal(state, 'duplicates');
  assert.match(detail, /twice for every event/);
});

test('four distinct webhooks is one slot from failing', () => {
  assert.equal(verdict(4, distinct(4))[0], 'near-limit');
});

test('an empty conversation is not a finding', () => {
  assert.equal(verdict(0, [])[0], 'none');
  assert.equal(verdict(2, distinct(2))[0], 'headroom');
});

test('destination ignores case and a trailing slash', () => {
  assert.equal(destination(hook('WH1', { url: 'https://App.Example.com/hook/' })),
               destination(hook('WH2')));
});

test('a studio target is keyed on the flow', () => {
  assert.equal(destination(hook('WH1', { target: 'studio', flow: 'FW1' })),
               'studio FW1');
  assert.notEqual(destination(hook('WH1', { target: 'studio', flow: 'FW1' })),
                  destination(hook('WH2', { target: 'studio', flow: 'FW2' })));
});

test('meta.total wins over the length of the page', () => {
  assert.equal(webhookTotal({ webhooks: distinct(2), meta: { total: 5 } }), 5);
  assert.equal(webhookTotal({ webhooks: distinct(2), meta: {} }), 2);
});