chat.postMessage is one per second, per channel

"""Measure how fast this app has been posting into one Slack channel.

Read only. chat.postMessage is on Slack's Special tier at roughly one message
per second per channel, and the honest way to test that would be to send two
hundred messages, which is the incident rather than the diagnostic. So this
reads one page of history instead, keeps the messages this app already wrote,
and does arithmetic on their ts values. A burst in the record is a burst that
happened and was allowed through on the burst allowance.
"""
import argparse
import logging
import math
import os
import sys

import requests

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

API = "https://slack.com/api/"

# The documented envelope for chat.postMessage and chat.update: about one
# message per second in one channel, bursts tolerated above it.
WINDOW = 1.0

# Fewer of this app's messages than this on the page and the sample is too thin
# to rule on. Saying so is more useful than a verdict drawn from three points.
MIN_SAMPLE = 5

# Block Kit's per-message ceiling. The budget that is not scarce here.
BLOCK_CEILING = 50


def parse_ts(value):
    """A Slack ts as a float, or None. Pure.

    ts is a Unix time with microseconds, delivered as a string, and it is also
    a message's identity, so it turns up in configuration and in stored thread
    anchors where it may have been mangled. Anything that is not a positive
    number is dropped rather than defaulted to now.
    """
    try:
        seconds = float(str(value).strip())
    except (TypeError, ValueError):
        return None
    if not math.isfinite(seconds) or seconds <= 0:
        return None
    return seconds


def app_messages(messages, identity):
    """The messages on this page that this app wrote. Pure.

    A positive filter, not a subtype blocklist. Everything else in the channel
    is somebody else's cadence, and counting it turns a busy channel into a
    finding about a sender that posted twice all week.

    Returns [(ts, matched_on), ...] in the order Slack returned them.
    """
    bot_id = str((identity or {}).get("bot_id") or "").strip()
    app_id = str((identity or {}).get("app_id") or "").strip()
    user_id = str((identity or {}).get("user_id") or "").strip()

    out = []
    for message in messages or []:
        matched = ""
        profile = message.get("bot_profile") or {}
        if bot_id and str(message.get("bot_id") or "") == bot_id:
            matched = "bot_id"
        elif app_id and str(message.get("app_id") or "") == app_id:
            matched = "app_id"
        elif app_id and str(profile.get("app_id") or "") == app_id:
            matched = "bot_profile"
        elif user_id and str(message.get("user") or "") == user_id:
            matched = "user_id"
        if not matched:
            continue
        stamp = parse_ts(message.get("ts"))
        if stamp is not None:
            out.append((stamp, matched))
    return out


def peak_per_second(stamps, window=WINDOW):
    """The most messages this sender put into any one window. Pure.

    An average hides this completely: five messages in one second and silence
    for an hour averages to nothing at all, and the limit compares against the
    peak. Sliding window, so a burst spanning a second boundary still counts.
    """
    ordered = sorted(s for s in stamps if s is not None)
    best, start = 0, 0
    for end in range(len(ordered)):
        while ordered[end] - ordered[start] >= window:
            start += 1
        best = max(best, end - start + 1)
    return best


def cadence_verdict(stamps, window=WINDOW, min_sample=MIN_SAMPLE):
    """Rule on one channel's send cadence. Pure. Returns (verdict, peak, detail)."""
    ordered = sorted(s for s in stamps if s is not None)
    if len(ordered) < min_sample:
        return ("no-sample", len(ordered),
                "only %d message(s) from this app on the page. That is too few to "
                "say anything about cadence, and a verdict drawn from it would be "
                "noise wearing a label." % len(ordered))

    peak = peak_per_second(ordered, window)
    span = ordered[-1] - ordered[0]
    if peak >= 3:
        return ("will-throttle", peak,
                "%d message(s) inside one second. This sender is already past the "
                "one per second envelope and is being carried by the burst "
                "allowance; a longer run in the same channel will be refused."
                % peak)
    if peak == 2:
        return ("at-the-edge", peak,
                "two messages inside one second, over a span of %.0fs. That is the "
                "top of the envelope. It survives today and will not survive a "
                "fan-out twice this size." % span)
    return ("paced", peak,
            "peak of %d message(s) in any one second window over %.0fs. Nothing "
            "here is racing the limit." % (peak, span))


def batch_plan(alerts, blocks_per_alert=2, ceiling=BLOCK_CEILING):
    """What N separate alerts cost, and what they cost batched. Pure.

    Returns (messages, seconds_now, seconds_batched). The block ceiling is a far
    larger budget than the one per second one, which is the whole argument: most
    fan-outs send N messages because N things happened, not because N messages
    were needed.
    """
    alerts = max(0, int(alerts))
    per_message = max(1, int(ceiling) // max(1, int(blocks_per_alert)))
    messages = -(-alerts // per_message) if alerts else 0
    return (messages, float(max(0, alerts - 1)), float(max(0, messages - 1)))


def identify(session):
    """The app's own identity. GET only."""
    body = session.get(API + "auth.test", timeout=30).json()
    if body.get("ok") is not True:
        raise SystemExit("auth.test answered 200 with ok: false, error=%s"
                         % body.get("error"))
    return body


def one_page(session, channel, limit):
    """One page of history. GET only, and one page on purpose: this is a
    question about the recent shape of the sending, not about the archive."""
    body = session.get(API + "conversations.history",
                       params={"channel": channel, "limit": limit},
                       timeout=30).json()
    if body.get("ok") is not True:
        log.error("%s conversations.history answered ok: false, error=%s",
                  channel, body.get("error"))
        return []
    return body.get("messages") or []


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("channels", nargs="+", help="channel IDs this app posts into")
    ap.add_argument("--token-env", default="SLACK_BOT_TOKEN",
                    help="environment variable holding the bot token")
    ap.add_argument("--app-id", default="",
                    help="the app ID, if you know it; catches messages sent by "
                         "other tokens of the same app")
    ap.add_argument("--limit", type=int, default=200,
                    help="messages to read per channel, one page only")
    ap.add_argument("--alerts", type=int, default=0,
                    help="size of a fan-out to cost against the batching plan")
    ap.add_argument("--blocks-per-alert", type=int, default=2,
                    help="blocks one alert occupies when batched")
    args = ap.parse_args()

    token = os.environ.get(args.token_env)
    if not token:
        log.error("set %s (channels:history is what the measurement needs)",
                  args.token_env)
        return 2

    s = requests.Session()
    s.headers.update({"Authorization": "Bearer " + token})

    identity = identify(s)
    if args.app_id:
        identity = dict(identity, app_id=args.app_id)
    log.info("identity  %s bot %s in %s", identity.get("user_id"),
             identity.get("bot_id"), identity.get("team"))
    if not args.app_id:
        log.info("no --app-id given, so messages sent by other tokens of this app "
                 "will not be matched and the sample may be smaller than the truth")

    hot = 0
    for channel in args.channels:
        page = one_page(s, channel, args.limit)
        mine = app_messages(page, identity)
        log.info("%-12s sample   %d message(s) from this app in the last %d",
                 channel, len(mine), len(page))

        verdict, peak, detail = cadence_verdict([ts for ts, _ in mine])
        if verdict in ("paced", "no-sample"):
            log.info("%-12s %-14s %s", channel, verdict, detail)
            continue

        hot += 1
        log.warning("%-12s %-14s %s", channel, verdict, detail)
        log.warning("  repair: one token bucket per channel ID at one per second, "
                    "and the sender queues rather than drops")
        log.warning("  repair: parallelise across channels if throughput matters. "
                    "Parallelising inside this one changes nothing, because the "
                    "limit follows the channel")
        log.warning("  note: chat.update shares this envelope, so editing one "
                    "message per token is refused for the same reason")

    if args.alerts:
        messages, before, after = batch_plan(args.alerts, args.blocks_per_alert)
        log.info("batching  %d alert(s) at %d block(s) each fit in %d message(s), "
                 "%.0fs instead of %.0fs", args.alerts, args.blocks_per_alert,
                 messages, after, before)

    log.info("%d channel(s) checked, %d sending faster than one per second",
             len(args.channels), hot)
    return 1 if hot else 0


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

/**
 * Measure how fast this app has been posting into one Slack channel.
 *
 * Read only. chat.postMessage is on Slack's Special tier at roughly one message
 * per second per channel, and the honest way to test that would be to send two
 * hundred messages, which is the incident rather than the diagnostic. So this
 * reads one page of history instead, keeps the messages this app already wrote,
 * and does arithmetic on their ts values.
 */

const API = 'https://slack.com/api/';

const WINDOW = 1.0;
const MIN_SAMPLE = 5;
const BLOCK_CEILING = 50;

/**
 * A Slack ts as a number, or null. Pure.
 * Anything that is not a positive finite number is dropped rather than
 * defaulted to now.
 */
export function parseTs(value) {
  const text = String(value ?? '').trim();
  if (!text) return null;
  const seconds = Number(text);
  if (!Number.isFinite(seconds) || seconds <= 0) return null;
  return seconds;
}

/**
 * The messages on this page that this app wrote. Pure.
 * A positive filter, not a subtype blocklist: everything else in the channel is
 * somebody else's cadence. Returns [[ts, matchedOn], ...].
 */
export function appMessages(messages, identity) {
  const botId = String(identity?.bot_id ?? '').trim();
  const appId = String(identity?.app_id ?? '').trim();
  const userId = String(identity?.user_id ?? '').trim();

  const out = [];
  for (const message of messages ?? []) {
    const profile = message.bot_profile ?? {};
    let matched = '';
    if (botId && String(message.bot_id ?? '') === botId) matched = 'bot_id';
    else if (appId && String(message.app_id ?? '') === appId) matched = 'app_id';
    else if (appId && String(profile.app_id ?? '') === appId) matched = 'bot_profile';
    else if (userId && String(message.user ?? '') === userId) matched = 'user_id';
    if (!matched) continue;
    const stamp = parseTs(message.ts);
    if (stamp !== null) out.push([stamp, matched]);
  }
  return out;
}

/**
 * The most messages this sender put into any one window. Pure.
 * An average hides this completely, and the limit compares against the peak.
 */
export function peakPerSecond(stamps, window = WINDOW) {
  const ordered = stamps.filter((s) => s !== null && s !== undefined)
    .slice().sort((a, b) => a - b);
  let best = 0;
  let start = 0;
  for (let end = 0; end < ordered.length; end += 1) {
    while (ordered[end] - ordered[start] >= window) start += 1;
    best = Math.max(best, end - start + 1);
  }
  return best;
}

/** Rule on one channel's send cadence. Pure. Returns [verdict, peak, detail]. */
export function cadenceVerdict(stamps, window = WINDOW, minSample = MIN_SAMPLE) {
  const ordered = stamps.filter((s) => s !== null && s !== undefined)
    .slice().sort((a, b) => a - b);
  if (ordered.length < minSample) {
    return ['no-sample', ordered.length,
      `only ${ordered.length} message(s) from this app on the page. That is too ` +
      'few to say anything about cadence, and a verdict drawn from it would be ' +
      'noise wearing a label.'];
  }

  const peak = peakPerSecond(ordered, window);
  const span = ordered[ordered.length - 1] - ordered[0];
  if (peak >= 3) {
    return ['will-throttle', peak,
      `${peak} message(s) inside one second. This sender is already past the one ` +
      'per second envelope and is being carried by the burst allowance; a longer ' +
      'run in the same channel will be refused.'];
  }
  if (peak === 2) {
    return ['at-the-edge', peak,
      `two messages inside one second, over a span of ${span.toFixed(0)}s. That ` +
      'is the top of the envelope. It survives today and will not survive a ' +
      'fan-out twice this size.'];
  }
  return ['paced', peak,
    `peak of ${peak} message(s) in any one second window over ${span.toFixed(0)}s. ` +
    'Nothing here is racing the limit.'];
}

/**
 * What N separate alerts cost, and what they cost batched. Pure.
 * Returns [messages, secondsNow, secondsBatched].
 */
export function batchPlan(alerts, blocksPerAlert = 2, ceiling = BLOCK_CEILING) {
  const count = Math.max(0, Math.trunc(Number(alerts) || 0));
  const perMessage = Math.max(1, Math.trunc(ceiling / Math.max(1, blocksPerAlert)));
  const messages = count ? Math.ceil(count / perMessage) : 0;
  return [messages, Math.max(0, count - 1), Math.max(0, messages - 1)];
}

async function call(token, method, params) {
  const qs = new URLSearchParams(params);
  const res = await fetch(`${API}${method}?${qs}`, {
    headers: { Authorization: `Bearer ${token}` },
  });
  return res.json();
}

/** One page of history. GET only, and one page on purpose. */
async function onePage(token, channel, limit) {
  const body = await call(token, 'conversations.history',
    { channel, limit: String(limit) });
  if (body.ok !== true) {
    console.error(`${channel} conversations.history answered ok: false, ` +
      `error=${body.error}`);
    return [];
  }
  return body.messages ?? [];
}

function arg(args, name, fallback = null) {
  const i = args.indexOf(name);
  return i === -1 ? fallback : args[i + 1];
}

function positionals(args) {
  const out = [];
  for (let i = 0; i < args.length; i += 1) {
    if (args[i].startsWith('--')) { i += 1; continue; }
    out.push(args[i]);
  }
  return out;
}

async function main() {
  const args = process.argv.slice(2);
  const channels = positionals(args);
  if (channels.length === 0) {
    console.error('usage: <channel id>... [--token-env SLACK_BOT_TOKEN] ' +
      '[--app-id A0...] [--limit 200] [--alerts 200] [--blocks-per-alert 2]');
    process.exitCode = 2;
    return;
  }
  const tokenEnv = arg(args, '--token-env', 'SLACK_BOT_TOKEN');
  const token = process.env[tokenEnv];
  if (!token) {
    console.error(`set ${tokenEnv} (channels:history is what the measurement needs)`);
    process.exitCode = 2;
    return;
  }

  const appId = arg(args, '--app-id', '');
  const limit = Number(arg(args, '--limit', 200));
  const alerts = Number(arg(args, '--alerts', 0));
  const blocksPerAlert = Number(arg(args, '--blocks-per-alert', 2));

  const auth = await call(token, 'auth.test', {});
  if (auth.ok !== true) {
    console.error(`auth.test answered 200 with ok: false, error=${auth.error}`);
    process.exitCode = 2;
    return;
  }
  const identity = appId ? { ...auth, app_id: appId } : auth;
  console.log(`identity  ${identity.user_id} bot ${identity.bot_id} in ${identity.team}`);
  if (!appId) {
    console.log('no --app-id given, so messages sent by other tokens of this app ' +
      'will not be matched and the sample may be smaller than the truth');
  }

  let hot = 0;
  for (const channel of channels) {
    const page = await onePage(token, channel, limit);
    const mine = appMessages(page, identity);
    console.log(`${channel.padEnd(12)} sample   ${mine.length} message(s) from ` +
      `this app in the last ${page.length}`);

    const [verdict, , detail] = cadenceVerdict(mine.map(([ts]) => ts));
    if (verdict === 'paced' || verdict === 'no-sample') {
      console.log(`${channel.padEnd(12)} ${verdict.padEnd(14)} ${detail}`);
      continue;
    }

    hot += 1;
    console.warn(`${channel.padEnd(12)} ${verdict.padEnd(14)} ${detail}`);
    console.warn('  repair: one token bucket per channel ID at one per second, and ' +
      'the sender queues rather than drops');
    console.warn('  repair: parallelise across channels if throughput matters. ' +
      'Parallelising inside this one changes nothing, because the limit follows ' +
      'the channel');
    console.warn('  note: chat.update shares this envelope, so editing one message ' +
      'per token is refused for the same reason');
  }

  if (alerts) {
    const [messages, before, after] = batchPlan(alerts, blocksPerAlert);
    console.log(`batching  ${alerts} alert(s) at ${blocksPerAlert} block(s) each ` +
      `fit in ${messages} message(s), ${after}s instead of ${before}s`);
  }

  console.log(`${channels.length} channel(s) checked, ${hot} sending faster than ` +
    'one per second');
  process.exitCode = hot ? 1 : 0;
}

// Only run when invoked directly, so importing this module in the tests does not
// execute main() and fail the file on a missing token.
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from slack_send_cadence import (app_messages, batch_plan, cadence_verdict,
                                  parse_ts, peak_per_second)

ME = {"user_id": "U0APPBOT11", "bot_id": "B0APPBOT99", "app_id": "A0APPBOT77"}

PAGE = [
    {"ts": "1770000000.000100", "bot_id": "B0APPBOT99", "text": "alert 1"},
    {"ts": "1770000000.400100", "bot_id": "B0APPBOT99", "text": "alert 2"},
    {"ts": "1770000000.800100", "bot_id": "B0APPBOT99", "text": "alert 3"},
    {"ts": "1770000004.000100", "user": "U0HUMAN123", "text": "what happened"},
    {"ts": "1770000004.100100", "bot_id": "B0OTHERBOT", "text": "unrelated bot"},
]


def test_a_ts_is_a_float_and_nonsense_is_dropped():
    assert parse_ts("1770000000.000100") == 1770000000.0001
    assert parse_ts(" 1770000000.5 ") == 1770000000.5
    assert parse_ts("") is None
    assert parse_ts(None) is None
    assert parse_ts("p1770000000.000100") is None
    assert parse_ts("-1") is None


def test_only_this_apps_messages_are_counted():
    mine = app_messages(PAGE, ME)
    assert [how for _, how in mine] == ["bot_id", "bot_id", "bot_id"]


def test_a_second_token_of_the_same_app_is_matched_through_bot_profile():
    page = PAGE + [{"ts": "1770000009.000100", "bot_id": "B0SECOND11",
                    "bot_profile": {"app_id": "A0APPBOT77"}}]
    assert app_messages(page, ME)[-1][1] == "bot_profile"


def test_without_an_app_id_the_second_token_is_invisible():
    page = PAGE + [{"ts": "1770000009.000100", "bot_id": "B0SECOND11",
                    "bot_profile": {"app_id": "A0APPBOT77"}}]
    identity = {"user_id": "U0APPBOT11", "bot_id": "B0APPBOT99"}
    assert len(app_messages(page, identity)) == 3


def test_a_burst_straddling_a_second_boundary_is_still_a_burst():
    # Grouping by whole second would file these as one and two.
    stamps = [1770000000.9, 1770000001.1, 1770000001.4]
    assert peak_per_second(stamps) == 3


def test_an_idle_sender_peaks_at_one():
    assert peak_per_second([1770000000.0, 1770000060.0, 1770000120.0]) == 1
    assert peak_per_second([]) == 0


def test_three_in_a_second_is_the_headline_verdict():
    stamps = [1770000000.0, 1770000000.3, 1770000000.6, 1770000000.9, 1770000010.0]
    verdict, peak, detail = cadence_verdict(stamps)
    assert verdict == "will-throttle"
    assert peak == 4
    assert "burst allowance" in detail


def test_two_in_a_second_is_the_edge_rather_than_a_failure():
    stamps = [1770000000.0, 1770000000.5, 1770000030.0, 1770000060.0, 1770000090.0]
    verdict, peak, _ = cadence_verdict(stamps)
    assert verdict == "at-the-edge"
    assert peak == 2


def test_a_paced_sender_is_cleared():
    stamps = [1770000000.0 + i * 5 for i in range(8)]
    assert cadence_verdict(stamps)[0] == "paced"


def test_a_thin_sample_produces_no_verdict_at_all():
    verdict, count, detail = cadence_verdict([1770000000.0, 1770000000.1])
    assert verdict == "no-sample"
    assert count == 2
    assert "too few" in detail


def test_batching_turns_two_hundred_alerts_into_eight_messages():
    messages, before, after = batch_plan(200, 2)
    assert messages == 8
    assert before == 199.0
    assert after == 7.0


def test_batching_degrades_sensibly_at_the_edges():
    assert batch_plan(0) == (0, 0.0, 0.0)
    assert batch_plan(1)[0] == 1
    # A fat alert that eats the whole block budget cannot be batched at all.
    assert batch_plan(9, 50) == (9, 8.0, 8.0)

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { appMessages, batchPlan, cadenceVerdict, parseTs, peakPerSecond }
  from './slack-send-cadence.mjs';

const ME = { user_id: 'U0APPBOT11', bot_id: 'B0APPBOT99', app_id: 'A0APPBOT77' };

const PAGE = [
  { ts: '1770000000.000100', bot_id: 'B0APPBOT99', text: 'alert 1' },
  { ts: '1770000000.400100', bot_id: 'B0APPBOT99', text: 'alert 2' },
  { ts: '1770000000.800100', bot_id: 'B0APPBOT99', text: 'alert 3' },
  { ts: '1770000004.000100', user: 'U0HUMAN123', text: 'what happened' },
  { ts: '1770000004.100100', bot_id: 'B0OTHERBOT', text: 'unrelated bot' },
];

test('a ts is a number and nonsense is dropped', () => {
  assert.equal(parseTs('1770000000.000100'), 1770000000.0001);
  assert.equal(parseTs(' 1770000000.5 '), 1770000000.5);
  assert.equal(parseTs(''), null);
  assert.equal(parseTs(null), null);
  assert.equal(parseTs('p1770000000.000100'), null);
  assert.equal(parseTs('-1'), null);
});

test('only this apps messages are counted', () => {
  assert.deepEqual(appMessages(PAGE, ME).map(([, how]) => how),
    ['bot_id', 'bot_id', 'bot_id']);
});

test('a second token of the same app is matched through bot_profile', () => {
  const page = [...PAGE, {
    ts: '1770000009.000100', bot_id: 'B0SECOND11',
    bot_profile: { app_id: 'A0APPBOT77' },
  }];
  assert.equal(appMessages(page, ME).at(-1)[1], 'bot_profile');
});

test('without an app id the second token is invisible', () => {
  const page = [...PAGE, {
    ts: '1770000009.000100', bot_id: 'B0SECOND11',
    bot_profile: { app_id: 'A0APPBOT77' },
  }];
  assert.equal(appMessages(page, { user_id: 'U0APPBOT11', bot_id: 'B0APPBOT99' }).length, 3);
});

test('a burst straddling a second boundary is still a burst', () => {
  assert.equal(peakPerSecond([1770000000.9, 1770000001.1, 1770000001.4]), 3);
});

test('an idle sender peaks at one', () => {
  assert.equal(peakPerSecond([1770000000, 1770000060, 1770000120]), 1);
  assert.equal(peakPerSecond([]), 0);
});

test('three in a second is the headline verdict', () => {
  const stamps = [1770000000, 1770000000.3, 1770000000.6, 1770000000.9, 1770000010];
  const [verdict, peak, detail] = cadenceVerdict(stamps);
  assert.equal(verdict, 'will-throttle');
  assert.equal(peak, 4);
  assert.match(detail, /burst allowance/);
});

test('two in a second is the edge rather than a failure', () => {
  const stamps = [1770000000, 1770000000.5, 1770000030, 1770000060, 1770000090];
  const [verdict, peak] = cadenceVerdict(stamps);
  assert.equal(verdict, 'at-the-edge');
  assert.equal(peak, 2);
});

test('a paced sender is cleared', () => {
  const stamps = Array.from({ length: 8 }, (_, i) => 1770000000 + i * 5);
  assert.equal(cadenceVerdict(stamps)[0], 'paced');
});

test('a thin sample produces no verdict at all', () => {
  const [verdict, count, detail] = cadenceVerdict([1770000000, 1770000000.1]);
  assert.equal(verdict, 'no-sample');
  assert.equal(count, 2);
  assert.match(detail, /too few/);
});

test('batching turns two hundred alerts into eight messages', () => {
  assert.deepEqual(batchPlan(200, 2), [8, 199, 7]);
});

test('batching degrades sensibly at the edges', () => {
  assert.deepEqual(batchPlan(0), [0, 0, 0]);
  assert.equal(batchPlan(1)[0], 1);
  assert.deepEqual(batchPlan(9, 50), [9, 8, 8]);
});