no status callback means delivery failures never reach you

"""Report Twilio Messaging Services with no delivery signal at all.

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_delivery_observability_audit")

MESSAGING = "https://messaging.twilio.com/v1"
EVENTS = "https://events.twilio.com/v1"

MESSAGE_EVENT = "com.twilio.messaging.message."


def message_streams(sinks, subscriptions):
    """Pair every subscription carrying a message event with the sink it feeds.

    Pure. `subscriptions` entries are the Subscription resource plus a "types"
    list, which is the SubscribedEvents subresource fetched alongside it. A sink
    that exists proves nothing on its own: it can be subscribed to voice events,
    or be subscribed correctly and sit in a status that is not active.

    Returns {"live": [sink sid, ...], "broken": [(sink sid, status), ...]}.
    """
    by_sid = {}
    for sink in sinks or []:
        by_sid[str(sink.get("sid") or "")] = sink

    live, broken = [], []
    for sub in subscriptions or []:
        types = [str(t.get("type") or "") for t in (sub.get("types") or [])]
        if not any(t.startswith(MESSAGE_EVENT) for t in types):
            continue
        sink_sid = str(sub.get("sink_sid") or "")
        sink = by_sid.get(sink_sid)
        status = str((sink or {}).get("status") or "missing").lower()
        if status == "active":
            live.append(sink_sid)
        else:
            broken.append((sink_sid or "?", status))
    return {"live": live, "broken": broken}


def verdict(service, streams=None):
    """Classify one Messaging Service's delivery observability. Pure.

    Returns (state, detail).
    """
    streams = streams or {"live": [], "broken": []}
    callback = str(service.get("status_callback") or "").strip()
    fallback = str(service.get("fallback_url") or "").strip()
    no_fallback = "" if fallback else " No fallback_url either."

    if callback:
        return ("callback", "status_callback posts terminal status and error_code "
                            "to %s.%s" % (callback, no_fallback))
    if streams["live"]:
        return ("streamed",
                "no status_callback, but Event Streams carries message events to "
                "active sink(s) %s.%s" % (", ".join(streams["live"]), no_fallback))
    if streams["broken"]:
        return ("sink-failed",
                "no status_callback, and the only message subscription feeds a "
                "sink that is not active: %s. Believed working, delivering "
                "nothing.%s"
                % (", ".join("%s (%s)" % pair for pair in streams["broken"]),
                   no_fallback))
    return ("blind",
            "no status_callback and no active subscription to "
            "com.twilio.messaging.message.*. Every delivery failure, opt-out and "
            "filtering code exists only in Twilio's logs.%s" % no_fallback)


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):
    params = {"PageSize": 100}
    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 load_subscriptions(session, limit):
    """Each subscription plus the event types it is actually subscribed to. The
    types live in a subresource, so the sink alone never answers the question."""
    subs = paged(session, "%s/Subscriptions" % EVENTS, "subscriptions", limit)
    for sub in subs:
        sub["types"] = paged(session, "%s/Subscriptions/%s/SubscribedEvents"
                             % (EVENTS, sub.get("sid")), "types", 200)
    return subs


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--max-services", type=int, default=200,
                    help="stop paging after this many Messaging Services")
    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)

    services = paged(session, "%s/Services" % MESSAGING, "services", args.max_services)
    if not services:
        log.info("no Messaging Services on this account")
        return 0

    sinks = paged(session, "%s/Sinks" % EVENTS, "sinks", 200)
    streams = message_streams(sinks, load_subscriptions(session, 200))

    bad = 0
    for svc in services:
        state, detail = verdict(svc, streams)
        line = "%-12s %s (%s)  %s" % (state, svc.get("sid"),
                                      svc.get("friendly_name", "?"), detail)
        if state in ("callback", "streamed"):
            log.info(line)
            continue
        bad += 1
        log.warning(line)
        log.warning("  repair: POST %s/Services/%s StatusCallback=https://.../twilio/"
                    "status FallbackUrl=https://.../twilio/fallback, then validate "
                    "X-Twilio-Signature, persist MessageStatus and ErrorCode, and "
                    "suppress the recipient on 21610.", MESSAGING, svc.get("sid"))

    log.info("%d service(s), %d with no delivery signal", len(services), bad)
    return 1 if bad else 0


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

/**
 * Report Twilio Messaging Services with no delivery signal at all.
 *
 * 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 MESSAGING = 'https://messaging.twilio.com/v1';
const EVENTS = 'https://events.twilio.com/v1';

const MESSAGE_EVENT = 'com.twilio.messaging.message.';

/**
 * Pair every subscription carrying a message event with the sink it feeds.
 * Pure. `subscriptions` entries are the Subscription resource plus a `types`
 * list, which is the SubscribedEvents subresource fetched alongside it. A sink
 * that exists proves nothing on its own: it can be subscribed to voice events,
 * or be subscribed correctly and sit in a status that is not active.
 * Returns { live: [sinkSid], broken: [[sinkSid, status]] }.
 */
export function messageStreams(sinks, subscriptions) {
  const bySid = new Map();
  for (const sink of sinks ?? []) bySid.set(String(sink.sid ?? ''), sink);

  const live = [];
  const broken = [];
  for (const sub of subscriptions ?? []) {
    const types = (sub.types ?? []).map((t) => String(t.type ?? ''));
    if (!types.some((t) => t.startsWith(MESSAGE_EVENT))) continue;
    const sinkSid = String(sub.sink_sid ?? '');
    const sink = bySid.get(sinkSid);
    const status = String(sink?.status ?? 'missing').toLowerCase();
    if (status === 'active') live.push(sinkSid);
    else broken.push([sinkSid || '?', status]);
  }
  return { live, broken };
}

/**
 * Classify one Messaging Service's delivery observability. Pure.
 * Returns [state, detail].
 */
export function verdict(service, streams = { live: [], broken: [] }) {
  const callback = String(service.status_callback ?? '').trim();
  const fallback = String(service.fallback_url ?? '').trim();
  const noFallback = fallback ? '' : ' No fallback_url either.';

  if (callback) {
    return ['callback',
      `status_callback posts terminal status and error_code to ${callback}.${noFallback}`];
  }
  if (streams.live.length) {
    return ['streamed',
      'no status_callback, but Event Streams carries message events to active ' +
      `sink(s) ${streams.live.join(', ')}.${noFallback}`];
  }
  if (streams.broken.length) {
    const named = streams.broken.map(([sid, status]) => `${sid} (${status})`).join(', ');
    return ['sink-failed',
      'no status_callback, and the only message subscription feeds a sink that is ' +
      `not active: ${named}. Believed working, delivering nothing.${noFallback}`];
  }
  return ['blind',
    'no status_callback and no active subscription to com.twilio.messaging.message.*. ' +
    `Every delivery failure, opt-out and filtering code exists only in Twilio's logs.${noFallback}`];
}

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();
}

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

async function loadSubscriptions(auth, limit = 200) {
  const subs = await paged(auth, `${EVENTS}/Subscriptions`, 'subscriptions', limit);
  for (const sub of subs) {
    sub.types = await paged(auth, `${EVENTS}/Subscriptions/${sub.sid}/SubscribedEvents`,
                            'types', 200);
  }
  return subs;
}

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 services = await paged(auth, `${MESSAGING}/Services`, 'services');
  if (services.length === 0) {
    console.log('no Messaging Services on this account');
    return;
  }

  const sinks = await paged(auth, `${EVENTS}/Sinks`, 'sinks');
  const streams = messageStreams(sinks, await loadSubscriptions(auth));

  let bad = 0;
  for (const svc of services) {
    const [state, detail] = verdict(svc, streams);
    const line = `${state.padEnd(12)} ${svc.sid} (${svc.friendly_name ?? '?'})  ${detail}`;
    if (state === 'callback' || state === 'streamed') { console.log(line); continue; }
    bad += 1;
    console.warn(line);
    console.warn(`  repair: POST ${MESSAGING}/Services/${svc.sid} StatusCallback=` +
                 'https://.../twilio/status FallbackUrl=https://.../twilio/fallback, ' +
                 'then validate X-Twilio-Signature, persist MessageStatus and ' +
                 'ErrorCode, and suppress the recipient on 21610.');
  }

  console.log(`${services.length} service(s), ${bad} with no delivery signal`);
  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_delivery_observability_audit import message_streams, verdict

SINK = "DG11111111111111111111111111111111"


def sink(status="active", sid=SINK):
    return {"sid": sid, "status": status, "sink_type": "webhook"}


def sub(types, sink_sid=SINK):
    return {"sid": "DF1", "sink_sid": sink_sid,
            "types": [{"type": t} for t in types]}


def test_a_messaging_subscription_on_an_active_sink_is_live():
    streams = message_streams(
        [sink()], [sub(["com.twilio.messaging.message.delivered",
                        "com.twilio.messaging.message.failed"])])
    assert streams == {"live": [SINK], "broken": []}


def test_voice_events_are_not_delivery_observability():
    streams = message_streams([sink()], [sub(["com.twilio.voice.insights.call-summary"])])
    assert streams == {"live": [], "broken": []}


def test_a_sink_that_is_not_active_is_broken_not_live():
    streams = message_streams(
        [sink(status="failed")], [sub(["com.twilio.messaging.message.delivered"])])
    assert streams["live"] == []
    assert streams["broken"] == [(SINK, "failed")]


def test_a_subscription_pointing_at_no_sink_at_all_is_broken():
    streams = message_streams([], [sub(["com.twilio.messaging.message.sent"])])
    assert streams["broken"] == [(SINK, "missing")]


def test_a_service_with_no_callback_and_no_stream_is_blind():
    state, detail = verdict({"sid": "MG1", "status_callback": None,
                             "fallback_url": None})
    assert state == "blind"
    assert "com.twilio.messaging.message." in detail
    assert "No fallback_url either." in detail


def test_the_status_callback_settles_it():
    state, detail = verdict({"status_callback": "https://app.example.com/twilio/status",
                             "fallback_url": "https://app.example.com/twilio/fallback"})
    assert state == "callback"
    assert "No fallback_url" not in detail


def test_event_streams_counts_when_the_sink_is_active():
    state, _ = verdict({"status_callback": ""}, {"live": [SINK], "broken": []})
    assert state == "streamed"


def test_a_failed_sink_is_worse_than_nothing_and_says_so():
    state, detail = verdict({"status_callback": ""},
                            {"live": [], "broken": [(SINK, "failed")]})
    assert state == "sink-failed"
    assert "Believed working" in detail

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { messageStreams, verdict } from './twilio-delivery-observability-audit.mjs';

const SINK = 'DG11111111111111111111111111111111';

const sink = (status = 'active', sid = SINK) => ({ sid, status, sink_type: 'webhook' });
const sub = (types, sinkSid = SINK) => ({
  sid: 'DF1', sink_sid: sinkSid, types: types.map((t) => ({ type: t })),
});

test('a messaging subscription on an active sink is live', () => {
  const streams = messageStreams([sink()], [sub([
    'com.twilio.messaging.message.delivered',
    'com.twilio.messaging.message.failed',
  ])]);
  assert.deepEqual(streams, { live: [SINK], broken: [] });
});

test('voice events are not delivery observability', () => {
  const streams = messageStreams([sink()], [sub(['com.twilio.voice.insights.call-summary'])]);
  assert.deepEqual(streams, { live: [], broken: [] });
});

test('a sink that is not active is broken, not live', () => {
  const streams = messageStreams([sink('failed')],
                                 [sub(['com.twilio.messaging.message.delivered'])]);
  assert.deepEqual(streams.live, []);
  assert.deepEqual(streams.broken, [[SINK, 'failed']]);
});

test('a subscription pointing at no sink at all is broken', () => {
  const streams = messageStreams([], [sub(['com.twilio.messaging.message.sent'])]);
  assert.deepEqual(streams.broken, [[SINK, 'missing']]);
});

test('a service with no callback and no stream is blind', () => {
  const [state, detail] = verdict({ sid: 'MG1', status_callback: null, fallback_url: null });
  assert.equal(state, 'blind');
  assert.match(detail, /com\.twilio\.messaging\.message\./);
  assert.match(detail, /No fallback_url either\./);
});

test('the status callback settles it', () => {
  const [state, detail] = verdict({
    status_callback: 'https://app.example.com/twilio/status',
    fallback_url: 'https://app.example.com/twilio/fallback',
  });
  assert.equal(state, 'callback');
  assert.ok(!/No fallback_url/.test(detail));
});

test('event streams counts when the sink is active', () => {
  const [state] = verdict({ status_callback: '' }, { live: [SINK], broken: [] });
  assert.equal(state, 'streamed');
});

test('a failed sink is worse than nothing and says so', () => {
  const [state, detail] = verdict({ status_callback: '' },
                                  { live: [], broken: [[SINK, 'failed']] });
  assert.equal(state, 'sink-failed');
  assert.match(detail, /Believed working/);
});