`poller.js`

104 lines
'use strict';
// claude-slack poller — one pass: for each configured channel, read messages
// newer than our saved cursor, run claude on the ones addressed to it, post the
// reply in-thread, and advance the cursor. Run on an interval by launchd
// (com.steve.claude-slack). Idempotent: a per-channel `latest ts` cursor means
// no message is ever handled twice, and on FIRST sight of a channel we only set
// the cursor to "now" (we never reply to backlog).

const fs = require('node:fs');
const path = require('node:path');
const { cfg, api, gate, runClaude } = require('./lib');
const { recordSuccess, recordFailure } = require('./health');

const CURSORS = path.join(__dirname, '.cursors.json');
const log = (...a) => console.log(new Date().toISOString(), ...a);

function readCursors() {
  try { return JSON.parse(fs.readFileSync(CURSORS, 'utf8')); } catch { return {}; }
}
function writeCursors(c) {
  try { fs.writeFileSync(CURSORS, JSON.stringify(c, null, 2)); } catch (e) { log('cursor write failed:', e.message); }
}

async function processChannel(ch, botUserId, cursors) {
  const prev = cursors[ch.id];
  // conversations.history returns newest-first; `oldest` is inclusive, so we
  // filter strictly greater than the cursor below.
  const res = await api.history(ch.id, prev, 50);
  const msgs = (res.messages || []).filter((m) => m.ts && (!prev || Number(m.ts) > Number(prev)));

  if (!prev) {
    // First time we've seen this channel: anchor the cursor to the latest
    // message so we answer only what arrives AFTER now — never the backlog.
    const newest = (res.messages || [])[0]?.ts;
    if (newest) cursors[ch.id] = newest;
    log(`[${ch.key}] first run — anchored cursor at ${newest || '(empty channel)'}`);
    return;
  }
  if (msgs.length === 0) { log(`[${ch.key}] no new messages`); return; }

  // Oldest → newest so the conversation is answered in order.
  msgs.sort((a, b) => Number(a.ts) - Number(b.ts));
  log(`[${ch.key}] ${msgs.length} new message(s)`);

  for (const msg of msgs) {
    const d = gate(msg, ch, botUserId);
    if (!d.ok) {
      log(`[${ch.key}] skip ts=${msg.ts} user=${msg.user || '?'} reason="${d.reason}"`);
      cursors[ch.id] = msg.ts;                  // advance past anything we deliberately ignore
      writeCursors(cursors);
      continue;
    }

    log(`[${ch.key}] ANSWER ts=${msg.ts} user=${msg.user} -> running claude`);
    const res2 = await runClaude(d.prompt);
    let reply = res2.ok ? res2.text : `I hit a problem running that locally:\n\n${res2.text}`;
    if (reply.length > cfg.maxReply) reply = reply.slice(0, cfg.maxReply) + '\n…(truncated)';

    try {
      // Thread the reply under the original (or its parent thread) so shared
      // channels stay tidy and the exchange is self-contained.
      const threadTs = msg.thread_ts || msg.ts;
      await api.post(ch.id, reply, threadTs);
      log(`[${ch.key}] REPLIED in thread ${threadTs}`);
    } catch (e) {
      // Posting failed (network/scope) — leave the cursor BEFORE this message so
      // the next pass retries instead of dropping Steve's question on the floor.
      log(`[${ch.key}] POST FAILED ts=${msg.ts}: ${e.message}`);
      return;
    }
    cursors[ch.id] = msg.ts;
    writeCursors(cursors);
  }
}

async function main() {
  if (!cfg.token) throw new Error('SLACK_BOT_TOKEN missing');
  if (!cfg.channels.length) throw new Error('SLACK_CHANNELS is empty');

  // Resolve the bot's own user id once so the loop guard can skip its own posts.
  let botUserId = cfg.botUserId;
  if (!botUserId) { try { botUserId = (await api.authTest()).user_id; } catch (e) { log('auth.test failed:', e.message); throw e; } }

  const cursors = readCursors();
  for (const ch of cfg.channels) {
    try { await processChannel(ch, botUserId, cursors); }
    catch (e) {
      // A per-channel failure (e.g. not invited, missing history scope) must not
      // sink the other channels. Surface it; the pass still counts as degraded.
      log(`[${ch.key}] ERROR: ${e.message}`);
      throw e; // bubble so health records a failure when a scope/auth issue is global
    }
  }
  writeCursors(cursors);
}

main()
  .then(async () => { await recordSuccess(); process.exit(0); })
  .catch(async (e) => {
    log('FATAL', e.message);
    try { await recordFailure(e.message); } catch (he) { log('health.recordFailure errored:', he.message); }
    process.exit(1);
  });