A reply ts used as thread_ts, so the thread flattens

"""Find out whether the ts you thread under is a root or a reply.

Read only, and read only is enough because Slack answers the question directly:
conversations.replies given the ts of a reply returns the whole thread, rooted
at the real root, so the first message that comes back is where your reply would
actually have landed. Nothing is posted from here. Reproducing this by sending
is unusually pointless anyway, since the send succeeds either way and the only
evidence is which thread it turned up in.
"""
import argparse
import json
import logging
import os
import sys

import requests

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

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

# A reply that was also pushed into the channel. It is still a reply, and it is
# the one that most often gets captured as a parent, because it looks like a
# top level message in the channel where somebody found it.
BROADCAST = "thread_broadcast"


def ts_role(message):
    """Say whether one message is a thread root or a reply inside one. Pure.

    The distinguishing field is on the message itself and costs nothing to
    check: a root has no thread_ts, or a thread_ts equal to its own ts; a reply
    carries both, and they differ.
    """
    if not isinstance(message, dict) or not message.get("ts"):
        return ("unknown", "no message, so nothing can be said about the ts")
    ts = message.get("ts")
    thread_ts = message.get("thread_ts")
    if message.get("subtype") == BROADCAST:
        return ("broadcast", "a reply sent with reply_broadcast, so it appears in the "
                             "channel and is still a reply inside %s" % thread_ts)
    if not thread_ts or thread_ts == ts:
        if message.get("reply_count"):
            return ("root", "a thread root with %d repl(ies) already under it"
                    % message["reply_count"])
        return ("root", "a top level message and a valid thread root")
    return ("reply", "a reply inside thread %s; its own ts is %s and the two differ"
            % (thread_ts, ts))


def resolve_root(target):
    """Where a reply threaded under this ts would actually land. Pure.

    Slack does not refuse a reply's ts as a thread_ts. It reparents the reply
    to that reply's own root and returns ok, which is why this is the one
    threading fault with no error string attached to it.
    """
    role, detail = ts_role(target)
    if role == "unknown":
        return (None, False, "the target is not in this thread, which is a missing "
                             "parent rather than a flattened one")
    if role == "root":
        return (target.get("ts"), False, "the target is the root, so the reply lands "
                                         "where you meant it to")
    root = target.get("thread_ts")
    return (root, True, "the target is %s, so Slack silently reparents the reply to "
                        "%s and answers ok; the reply lands beside the message you "
                        "were replying to rather than under it" % (role, root))


def capture_rule(event):
    """The one line that captures a root from any message event. Pure.

    thread_ts or ts, in that order, which yields the root whether the event was
    a top level message or a reply. Storing response ts from a send is the
    mistake this replaces, because that ts is a root only when you posted the
    root.
    """
    if not isinstance(event, dict):
        return (None, "none")
    thread_ts = event.get("thread_ts")
    if thread_ts:
        return (thread_ts, "thread_ts")
    ts = event.get("ts")
    if ts:
        return (ts, "ts")
    return (None, "none")


def audit_stored(stored, index):
    """Check a map of stored parents against the messages Slack returned. Pure.

    stored: {label: ts}. index: {ts: message}, built from conversations.replies.
    A stored ts that resolves to a different root is a thread the app has been
    flattening, silently, for as long as that ts has been stored.
    """
    rows = []
    flattened = 0
    for label in sorted(stored or {}):
        ts = stored[label]
        target = (index or {}).get(ts)
        role, detail = ts_role(target)
        root, moved, why = resolve_root(target)
        if moved:
            flattened += 1
        rows.append({"label": label, "ts": ts, "role": role, "detail": detail,
                     "actual_root": root, "moved": moved, "why": why})
    return {
        "checked": len(rows),
        "flattened": flattened,
        "rows": rows,
        "verdict": "flattening" if flattened else "rooted",
    }


def _stored(args):
    out = {}
    if args.parents:
        body = json.loads(open(args.parents, encoding="utf-8").read())
        if isinstance(body, dict):
            for label, value in body.items():
                out[str(label)] = (value.get("thread_ts") if isinstance(value, dict)
                                   else str(value))
        else:
            for i, value in enumerate(body):
                out[str(i)] = str(value)
    for pair in args.parent:
        _, _, ts = pair.partition(":")
        out[pair] = ts
    return out


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--token-env", default="SLACK_BOT_TOKEN",
                    help="environment variable holding the bot token")
    ap.add_argument("--channel", required=True,
                    help="the channel the stored parents live in")
    ap.add_argument("--parent", action="append", default=[],
                    help="a stored parent as LABEL:THREAD_TS; repeatable")
    ap.add_argument("--parents", help="JSON file of stored parents")
    ap.add_argument("--limit", type=int, default=200,
                    help="replies read per thread")
    args = ap.parse_args()

    stored = _stored(args)
    if not stored:
        log.error("pass --parent LABEL:THREAD_TS or --parents FILE")
        return 2
    token = os.environ.get(args.token_env)
    if not token:
        log.error("set %s (channels:history is enough)", args.token_env)
        return 2

    s = requests.Session()
    s.headers.update({"Authorization": "Bearer " + token})
    who = s.get(API + "auth.test", timeout=30).json()
    if who.get("ok") is not True:
        log.error("auth.test  unavailable    %s", who.get("error"))
        return 2
    log.info("identity   %s in %s", who.get("user_id"), who.get("team"))

    index, broadcasts = {}, 0
    for label in sorted(stored):
        ts = stored[label]
        body = s.get(API + "conversations.replies", timeout=30,
                     params={"channel": args.channel, "ts": ts,
                             "limit": str(args.limit)}).json()
        if body.get("ok") is not True:
            log.warning("replies    unavailable    %s (%s): %s", label, ts,
                        body.get("error"))
            continue
        messages = body.get("messages") or []
        if messages:
            log.info("thread     %-16s root is %s with %d message(s) returned",
                     label, messages[0].get("ts"), len(messages))
        for m in messages:
            index[m.get("ts")] = m
            if m.get("subtype") == BROADCAST:
                broadcasts += 1

    a = audit_stored(stored, index)
    for row in a["rows"]:
        (log.warning if row["moved"] else log.info)(
            "stored     %-16s %-10s %s", row["label"], row["role"], row["why"])
    (log.warning if a["flattened"] else log.info)(
        "verdict    %-14s %d of %d stored parents are replies",
        a["verdict"], a["flattened"], a["checked"])
    if broadcasts:
        log.warning("broadcast  %d repl(ies) in these threads carry %s; each one was "
                    "pushed into the channel and is a candidate for being captured as "
                    "a parent by mistake", broadcasts, BROADCAST)

    if a["flattened"]:
        log.warning("  repair: store the root. From an event, that is "
                    "event.thread_ts or event.ts, which is the root in both cases")
        log.warning("  repair: from a send, keep the response ts only when you posted "
                    "the root; a reply's ts is not a parent")
        log.warning("  repair: set reply_broadcast deliberately and rarely. It puts the "
                    "reply in the channel, where the next person captures it as a root")
        return 1
    log.info("verdict    clear          every stored parent is a root")
    return 0


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

/**
 * Find out whether the ts you thread under is a root or a reply.
 *
 * Read only, and read only is enough because Slack answers the question
 * directly: conversations.replies given the ts of a reply returns the whole
 * thread, rooted at the real root, so the first message that comes back is
 * where your reply would actually have landed. Nothing is posted from here.
 */
import { readFile } from 'node:fs/promises';

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

// A reply that was also pushed into the channel. It is still a reply, and it
// is the one that most often gets captured as a parent, because it looks like
// a top level message in the channel where somebody found it.
const BROADCAST = 'thread_broadcast';

/**
 * Say whether one message is a thread root or a reply inside one. Pure.
 * A root has no thread_ts, or one equal to its own ts; a reply carries both,
 * and they differ.
 */
export function tsRole(message) {
  if (!message || typeof message !== 'object' || !message.ts) {
    return ['unknown', 'no message, so nothing can be said about the ts'];
  }
  const { ts } = message;
  const threadTs = message.thread_ts;
  if (message.subtype === BROADCAST) {
    return ['broadcast', 'a reply sent with reply_broadcast, so it appears in the '
      + `channel and is still a reply inside ${threadTs}`];
  }
  if (!threadTs || threadTs === ts) {
    if (message.reply_count) {
      return ['root', `a thread root with ${message.reply_count} repl(ies) already `
        + 'under it'];
    }
    return ['root', 'a top level message and a valid thread root'];
  }
  return ['reply', `a reply inside thread ${threadTs}; its own ts is ${ts} and the `
    + 'two differ'];
}

/**
 * Where a reply threaded under this ts would actually land. Pure.
 * Slack does not refuse a reply's ts as a thread_ts. It reparents the reply to
 * that reply's own root and returns ok, which is why this is the one threading
 * fault with no error string attached to it.
 */
export function resolveRoot(target) {
  const [role] = tsRole(target);
  if (role === 'unknown') {
    return [null, false, 'the target is not in this thread, which is a missing parent '
      + 'rather than a flattened one'];
  }
  if (role === 'root') {
    return [target.ts, false,
      'the target is the root, so the reply lands where you meant it to'];
  }
  const root = target.thread_ts;
  return [root, true, `the target is ${role}, so Slack silently reparents the reply `
    + `to ${root} and answers ok; the reply lands beside the message you were `
    + 'replying to rather than under it'];
}

/**
 * The one line that captures a root from any message event. Pure.
 * thread_ts or ts, in that order, which yields the root whether the event was
 * a top level message or a reply.
 */
export function captureRule(event) {
  if (!event || typeof event !== 'object') return [null, 'none'];
  if (event.thread_ts) return [event.thread_ts, 'thread_ts'];
  if (event.ts) return [event.ts, 'ts'];
  return [null, 'none'];
}

/**
 * Check a map of stored parents against the messages Slack returned. Pure.
 * stored: {label: ts}. index: {ts: message}. A stored ts that resolves to a
 * different root is a thread the app has been flattening, silently, for as
 * long as that ts has been stored.
 */
export function auditStored(stored, index) {
  const rows = [];
  let flattened = 0;
  for (const label of Object.keys(stored ?? {}).sort()) {
    const ts = stored[label];
    const target = (index ?? {})[ts];
    const [role, detail] = tsRole(target);
    const [root, moved, why] = resolveRoot(target);
    if (moved) flattened += 1;
    rows.push({ label, ts, role, detail, actualRoot: root, moved, why });
  }
  return {
    checked: rows.length,
    flattened,
    rows,
    verdict: flattened ? 'flattening' : 'rooted',
  };
}

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

function argAll(args, name) {
  const out = [];
  for (let i = 0; i < args.length; i += 1) {
    if (args[i] === name && args[i + 1] !== undefined) out.push(args[i + 1]);
  }
  return out;
}

async function storedParents(args) {
  const out = {};
  const file = arg(args, '--parents');
  if (file) {
    const body = JSON.parse(await readFile(file, 'utf8'));
    if (Array.isArray(body)) {
      body.forEach((value, i) => { out[String(i)] = String(value); });
    } else {
      for (const [label, value] of Object.entries(body)) {
        out[label] = (value && typeof value === 'object') ? value.thread_ts : String(value);
      }
    }
  }
  for (const pair of argAll(args, '--parent')) {
    const i = pair.indexOf(':');
    out[pair] = i === -1 ? '' : pair.slice(i + 1);
  }
  return out;
}

async function main() {
  const args = process.argv.slice(2);
  const channel = arg(args, '--channel');
  if (!channel) {
    console.error('pass --channel with the channel the stored parents live in');
    process.exitCode = 2;
    return;
  }
  const stored = await storedParents(args);
  if (!Object.keys(stored).length) {
    console.error('pass --parent LABEL:THREAD_TS or --parents FILE');
    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 enough)`);
    process.exitCode = 2;
    return;
  }
  const limit = arg(args, '--limit', '200');
  const headers = { Authorization: `Bearer ${token}` };

  const who = await (await fetch(`${API}auth.test`, { headers })).json();
  if (who.ok !== true) {
    console.error(`auth.test  unavailable    ${who.error}`);
    process.exitCode = 2;
    return;
  }
  console.log(`identity   ${who.user_id} in ${who.team}`);

  const index = {};
  let broadcasts = 0;
  for (const label of Object.keys(stored).sort()) {
    const ts = stored[label];
    const url = `${API}conversations.replies?channel=${encodeURIComponent(channel)}`
      + `&ts=${encodeURIComponent(ts)}&limit=${encodeURIComponent(limit)}`;
    const body = await (await fetch(url, { headers })).json();
    if (body.ok !== true) {
      console.warn(`replies    unavailable    ${label} (${ts}): ${body.error}`);
      continue;
    }
    const messages = body.messages ?? [];
    if (messages.length) {
      console.log(`thread     ${label.padEnd(16)} root is ${messages[0].ts} with `
        + `${messages.length} message(s) returned`);
    }
    for (const m of messages) {
      index[m.ts] = m;
      if (m.subtype === BROADCAST) broadcasts += 1;
    }
  }

  const a = auditStored(stored, index);
  for (const row of a.rows) {
    (row.moved ? console.warn : console.log)(
      `stored     ${row.label.padEnd(16)} ${row.role.padEnd(10)} ${row.why}`);
  }
  (a.flattened ? console.warn : console.log)(
    `verdict    ${a.verdict.padEnd(14)} ${a.flattened} of ${a.checked} stored parents `
    + 'are replies');
  if (broadcasts) {
    console.warn(`broadcast  ${broadcasts} repl(ies) in these threads carry ${BROADCAST}; `
      + 'each one was pushed into the channel and is a candidate for being captured as '
      + 'a parent by mistake');
  }

  if (a.flattened) {
    console.warn('  repair: store the root. From an event, that is event.thread_ts or '
      + 'event.ts, which is the root in both cases');
    console.warn('  repair: from a send, keep the response ts only when you posted the '
      + "root; a reply's ts is not a parent");
    console.warn('  repair: set reply_broadcast deliberately and rarely. It puts the '
      + 'reply in the channel, where the next person captures it as a root');
    process.exitCode = 1;
    return;
  }
  console.log('verdict    clear          every stored parent is a root');
}

// 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_thread_root import audit_stored, capture_rule, resolve_root, ts_role

ROOT = {"ts": "1700000000.0001", "text": "deploy started", "reply_count": 3}
REPLY = {"ts": "1700000100.0002", "thread_ts": "1700000000.0001", "text": "step 2"}
BCAST = {"ts": "1700000200.0003", "thread_ts": "1700000000.0001",
         "subtype": "thread_broadcast", "text": "still failing"}
INDEX = {m["ts"]: m for m in (ROOT, REPLY, BCAST)}


def test_a_root_is_a_message_whose_thread_ts_is_missing_or_its_own_ts():
    role, detail = ts_role(ROOT)
    assert role == "root"
    assert "3 repl(ies)" in detail
    assert ts_role({"ts": "1.0"})[0] == "root"
    assert ts_role({"ts": "1.0", "thread_ts": "1.0"})[0] == "root"


def test_a_reply_carries_both_fields_and_they_differ():
    role, detail = ts_role(REPLY)
    assert role == "reply"
    assert "1700000000.0001" in detail
    assert "the two differ" in detail


def test_a_broadcast_is_still_a_reply_however_it_looks_in_the_channel():
    role, detail = ts_role(BCAST)
    assert role == "broadcast"
    assert "still a reply" in detail


def test_a_missing_message_says_nothing_rather_than_guessing():
    assert ts_role(None)[0] == "unknown"
    assert ts_role({})[0] == "unknown"
    assert ts_role("1700000000.0001")[0] == "unknown"


def test_threading_under_a_root_lands_where_you_meant_it_to():
    root, moved, why = resolve_root(ROOT)
    assert root == "1700000000.0001"
    assert moved is False
    assert "where you meant it to" in why


def test_threading_under_a_reply_is_moved_and_nothing_is_returned_to_say_so():
    root, moved, why = resolve_root(REPLY)
    assert root == "1700000000.0001"
    assert moved is True
    assert "silently reparents" in why
    assert "answers ok" in why


def test_a_broadcast_reply_used_as_a_parent_is_moved_the_same_way():
    root, moved, why = resolve_root(BCAST)
    assert root == "1700000000.0001"
    assert moved is True
    assert "broadcast" in why


def test_a_target_that_is_not_in_the_thread_is_a_different_note():
    root, moved, why = resolve_root(None)
    assert root is None
    assert moved is False
    assert "missing parent" in why


def test_the_capture_rule_yields_the_root_for_a_reply_and_for_a_top_level_message():
    assert capture_rule(REPLY) == ("1700000000.0001", "thread_ts")
    assert capture_rule(ROOT) == ("1700000000.0001", "ts")
    assert capture_rule(BCAST)[0] == "1700000000.0001"


def test_the_capture_rule_holds_for_the_thing_it_was_written_to_prevent():
    stored, _source = capture_rule(REPLY)
    assert resolve_root(INDEX[stored])[1] is False


def test_the_capture_rule_refuses_to_invent_a_ts():
    assert capture_rule({}) == (None, "none")
    assert capture_rule(None) == (None, "none")


def test_the_audit_names_which_stored_parent_is_flattening_a_thread():
    a = audit_stored({"digest": ROOT["ts"], "incident": REPLY["ts"]}, INDEX)
    assert a["checked"] == 2
    assert a["flattened"] == 1
    assert a["verdict"] == "flattening"
    rows = {r["label"]: r for r in a["rows"]}
    assert rows["incident"]["moved"] is True
    assert rows["incident"]["actual_root"] == ROOT["ts"]
    assert rows["digest"]["moved"] is False


def test_the_audit_reports_rooted_when_every_stored_parent_is_a_root():
    a = audit_stored({"digest": ROOT["ts"]}, INDEX)
    assert a["flattened"] == 0
    assert a["verdict"] == "rooted"


def test_a_stored_ts_that_is_not_in_the_index_is_not_counted_as_flattened():
    a = audit_stored({"gone": "1699000000.0009"}, INDEX)
    assert a["flattened"] == 0
    assert a["rows"][0]["role"] == "unknown"
    assert a["rows"][0]["actual_root"] is None


def test_an_empty_audit_is_not_a_finding():
    a = audit_stored({}, INDEX)
    assert a["checked"] == 0
    assert a["verdict"] == "rooted"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import {
  auditStored, captureRule, resolveRoot, tsRole,
} from './slack-thread-root.mjs';

const ROOT = { ts: '1700000000.0001', text: 'deploy started', reply_count: 3 };
const REPLY = { ts: '1700000100.0002', thread_ts: '1700000000.0001', text: 'step 2' };
const BCAST = {
  ts: '1700000200.0003',
  thread_ts: '1700000000.0001',
  subtype: 'thread_broadcast',
  text: 'still failing',
};
const INDEX = Object.fromEntries([ROOT, REPLY, BCAST].map((m) => [m.ts, m]));

test('a root is a message whose thread_ts is missing or its own ts', () => {
  const [role, detail] = tsRole(ROOT);
  assert.equal(role, 'root');
  assert.match(detail, /3 repl\(ies\)/);
  assert.equal(tsRole({ ts: '1.0' })[0], 'root');
  assert.equal(tsRole({ ts: '1.0', thread_ts: '1.0' })[0], 'root');
});

test('a reply carries both fields and they differ', () => {
  const [role, detail] = tsRole(REPLY);
  assert.equal(role, 'reply');
  assert.match(detail, /1700000000\.0001/);
  assert.match(detail, /two differ/);
});

test('a broadcast is still a reply however it looks in the channel', () => {
  const [role, detail] = tsRole(BCAST);
  assert.equal(role, 'broadcast');
  assert.match(detail, /still a reply/);
});

test('a missing message says nothing rather than guessing', () => {
  assert.equal(tsRole(null)[0], 'unknown');
  assert.equal(tsRole({})[0], 'unknown');
  assert.equal(tsRole('1700000000.0001')[0], 'unknown');
});

test('threading under a root lands where you meant it to', () => {
  const [root, moved, why] = resolveRoot(ROOT);
  assert.equal(root, '1700000000.0001');
  assert.equal(moved, false);
  assert.match(why, /where you meant it to/);
});

test('threading under a reply is moved and nothing is returned to say so', () => {
  const [root, moved, why] = resolveRoot(REPLY);
  assert.equal(root, '1700000000.0001');
  assert.equal(moved, true);
  assert.match(why, /silently reparents/);
  assert.match(why, /answers ok/);
});

test('a broadcast reply used as a parent is moved the same way', () => {
  const [root, moved, why] = resolveRoot(BCAST);
  assert.equal(root, '1700000000.0001');
  assert.equal(moved, true);
  assert.match(why, /broadcast/);
});

test('a target that is not in the thread is a different note', () => {
  const [root, moved, why] = resolveRoot(null);
  assert.equal(root, null);
  assert.equal(moved, false);
  assert.match(why, /missing parent/);
});

test('the capture rule yields the root for a reply and for a top level message', () => {
  assert.deepEqual(captureRule(REPLY), ['1700000000.0001', 'thread_ts']);
  assert.deepEqual(captureRule(ROOT), ['1700000000.0001', 'ts']);
  assert.equal(captureRule(BCAST)[0], '1700000000.0001');
});

test('the capture rule holds for the thing it was written to prevent', () => {
  const [stored] = captureRule(REPLY);
  assert.equal(resolveRoot(INDEX[stored])[1], false);
});

test('the capture rule refuses to invent a ts', () => {
  assert.deepEqual(captureRule({}), [null, 'none']);
  assert.deepEqual(captureRule(null), [null, 'none']);
});

test('the audit names which stored parent is flattening a thread', () => {
  const a = auditStored({ digest: ROOT.ts, incident: REPLY.ts }, INDEX);
  assert.equal(a.checked, 2);
  assert.equal(a.flattened, 1);
  assert.equal(a.verdict, 'flattening');
  const rows = Object.fromEntries(a.rows.map((r) => [r.label, r]));
  assert.equal(rows.incident.moved, true);
  assert.equal(rows.incident.actualRoot, ROOT.ts);
  assert.equal(rows.digest.moved, false);
});

test('the audit reports rooted when every stored parent is a root', () => {
  const a = auditStored({ digest: ROOT.ts }, INDEX);
  assert.equal(a.flattened, 0);
  assert.equal(a.verdict, 'rooted');
});

test('a stored ts that is not in the index is not counted as flattened', () => {
  const a = auditStored({ gone: '1699000000.0009' }, INDEX);
  assert.equal(a.flattened, 0);
  assert.equal(a.rows[0].role, 'unknown');
  assert.equal(a.rows[0].actualRoot, null);
});

test('an empty audit is not a finding', () => {
  const a = auditStored({}, INDEX);
  assert.equal(a.checked, 0);
  assert.equal(a.verdict, 'rooted');
});