`poller.js`

385 lines
#!/usr/bin/env node
// discontinued-agent — poller daemon.
//
// Watches info@designerwallcoverings.com for vendor replies to DW's
// "Priority Request - Samples over 10 days old" sample-follow-up emails.
// When a vendor EXPLICITLY says a product is discontinued, it stamps the
// wallpaper record discontinued in FileMaker and closes the open sample order
// (so the >10-day follow-up system stops re-asking).
//
// Modes:
//   node poller.js            -> daemon: loops every POLL_INTERVAL_SEC seconds
//   node poller.js --once     -> run a single scan and exit
//
// SAFETY: unless DISCO_AUTOCOMMIT=1, this NEVER commits a FileMaker write. It
// stages the dry-run diff, logs it, and (for QUEUE cases) parks a memo.
//
// Cost: George (Gmail) + FileMaker Cloud are local/self-hosted = $0 per call.

import { readFileSync, writeFileSync, existsSync, appendFileSync, mkdirSync } from 'node:fs';
import { fileURLToPath } from 'node:url';
import { dirname, join } from 'node:path';

import * as George from './lib/george.js';
import * as FM from './lib/filemaker.js';
import {
  htmlToText,
  isDwOutbound,
  extractMfrNumbers,
  classifyReply,
  hashMsg,
} from './lib/parse.js';

const __dir = dirname(fileURLToPath(import.meta.url));
const DATA_DIR = join(__dir, 'data');
const PROCESSED_FILE = join(DATA_DIR, 'processed-threads.json');
const LEDGER_FILE = join(DATA_DIR, 'ledger.jsonl');

const AUTOCOMMIT = process.env.DISCO_AUTOCOMMIT === '1';
const QUEUE_DIR = process.env.QUEUE_DIR || join(process.env.HOME || '', '.claude/yolo-queue/pending-approval');
const GMAIL_SEARCH =
  process.env.GMAIL_SEARCH ||
  'subject:"Priority Request - Samples over 10 days old" newer_than:60d';
const POLL_INTERVAL_SEC = Number(process.env.POLL_INTERVAL_SEC || 10);
const MAX_RESULTS = Number(process.env.MAX_RESULTS || 100);
// Completion report: email a summary to info@ after any run that took action.
const REPORT_TO = process.env.DISCO_REPORT_TO || 'info@designerwallcoverings.com';
const REPORT_ENABLED = process.env.DISCO_REPORT !== '0';

function log(...a) {
  console.log(new Date().toISOString(), '[disco]', ...a);
}

// ---- state ----
function loadProcessed() {
  try {
    return JSON.parse(readFileSync(PROCESSED_FILE, 'utf8'));
  } catch {
    return {};
  }
}
function saveProcessed(state) {
  mkdirSync(DATA_DIR, { recursive: true });
  writeFileSync(PROCESSED_FILE, JSON.stringify(state, null, 2));
}
function appendLedger(entry) {
  mkdirSync(DATA_DIR, { recursive: true });
  appendFileSync(LEDGER_FILE, JSON.stringify({ ts: new Date().toISOString(), ...entry }) + '\n');
}

function queueMemo(name, body) {
  mkdirSync(QUEUE_DIR, { recursive: true });
  const stamp = new Date().toISOString().replace(/[:.]/g, '-');
  const path = join(QUEUE_DIR, `discontinued-agent-${name}-${stamp}.md`);
  writeFileSync(path, body);
  return path;
}

// ---- thread processing ----

// Build the {threadId -> [messages]} grouping from a flat list, ordered oldest->newest.
function groupByThread(messages) {
  const g = new Map();
  for (const m of messages) {
    if (!g.has(m.threadId)) g.set(m.threadId, []);
    g.get(m.threadId).push(m);
  }
  for (const [, arr] of g) {
    arr.sort((a, b) => Number(a.internalDate || 0) - Number(b.internalDate || 0));
  }
  return g;
}

// Given full messages for a thread, split into DW outbound + vendor replies.
function splitThread(fullMsgs) {
  const outbound = [];
  const vendor = [];
  for (const m of fullMsgs) {
    if (isDwOutbound(m)) outbound.push(m);
    else vendor.push(m);
  }
  return { outbound, vendor };
}

// Decide the disposition of one thread. Returns a plan object (no side effects).
function planThread(threadId, fullMsgs) {
  const { outbound, vendor } = splitThread(fullMsgs);
  const outboundText = outbound.map((m) => htmlToText(m.body || '')).join('\n\n');
  const mfrNumbers = extractMfrNumbers(outboundText);

  // Newest vendor reply drives the verdict.
  const latestVendor = vendor[vendor.length - 1] || null;
  const vendorText = latestVendor ? htmlToText(latestVendor.body || '') : '';
  const vendorFrom = latestVendor ? latestVendor.from : null;
  const cls = classifyReply(vendorText);

  return {
    threadId,
    vendorFrom,
    vendorMsgId: latestVendor ? latestVendor.id : null,
    vendorHash: hashMsg(vendorText),
    mfrNumbers,
    verdict: cls.verdict,
    reasons: cls.reasons,
    outboundCount: outbound.length,
    vendorCount: vendor.length,
    vendorText,
  };
}

// Execute the FileMaker writes for a COMMIT plan (dry-run unless AUTOCOMMIT).
async function applyCommit(plan) {
  const today = FM.fmToday();
  const F = FM.FIELDS;
  const perMfr = [];

  for (const mfr of plan.mfrNumbers) {
    let rows;
    try {
      rows = await FM.findByMfrPattern(mfr);
    } catch (e) {
      perMfr.push({ mfr, error: `find failed: ${e.message}` });
      continue;
    }
    const discoWrites = [];
    const closeWrites = [];

    for (const rec of rows) {
      const { recordId, fields } = FM.recRow(rec);
      const blankDisco = !String(fields[F.dateDiscontinued] || '').trim();
      const isOpenOrder =
        String(fields[F.account] || fields[F.company] || '').trim() &&
        String(fields[F.requestPrinted] || '').trim() &&
        !String(fields[F.dateSampleSent] || '').trim();

      // WRITE 1 — discontinue (only if currently blank)
      if (blankDisco) {
        const r = await FM.updateRecord(recordId, { [F.dateDiscontinued]: today }, { dryRun: !AUTOCOMMIT });
        discoWrites.push({ recordId, field: F.dateDiscontinued, ...r });
      }
      // WRITE 2 — close open order (blank Date WP Sample Sent on an open request)
      if (isOpenOrder) {
        const r = await FM.updateRecord(recordId, { [F.dateSampleSent]: today }, { dryRun: !AUTOCOMMIT });
        closeWrites.push({ recordId, field: F.dateSampleSent, ...r });
      }
    }
    perMfr.push({ mfr, rowsMatched: rows.length, discoWrites, closeWrites });
  }
  return { today, perMfr };
}

async function processThread(threadId, headers, state, actions = []) {
  // Fetch full bodies for each message in the thread.
  const full = [];
  for (const h of headers) {
    try {
      full.push(await George.getMessage(h.id));
    } catch (e) {
      log('getMessage failed', h.id, e.message);
    }
  }
  if (full.length === 0) return;

  const plan = planThread(threadId, full);

  // processed-state key: threadId + hash of the vendor message.
  const key = `${threadId}:${plan.vendorHash}`;
  if (state[key]) return; // already handled this exact vendor message

  if (plan.verdict === 'SKIP') {
    // Mark processed so we don't re-read it every cycle, but log nothing heavy.
    state[key] = { at: new Date().toISOString(), verdict: 'SKIP' };
    return;
  }

  if (plan.verdict === 'QUEUE') {
    const memoBody = [
      `# discontinued-agent — QUEUED for Steve's review`,
      ``,
      `- **When:** ${new Date().toISOString()}`,
      `- **Thread:** ${threadId}`,
      `- **Vendor:** ${plan.vendorFrom || 'unknown'}`,
      `- **Why queued (not auto-discontinued):** ${plan.reasons.join('; ')}`,
      `- **Candidate mfr#(s) in thread:** ${plan.mfrNumbers.join(', ') || '(none extracted)'}`,
      ``,
      `## Vendor reply (excerpt)`,
      '```',
      plan.vendorText.slice(0, 1500),
      '```',
      ``,
      `No FileMaker writes were made. Review and, if a discontinuation is warranted,`,
      `stamp \`Date Discontinued\` + close the open order manually or re-run with a`,
      `confirmed mfr# via the skill.`,
    ].join('\n');
    const memoPath = queueMemo(`queue-${threadId}`, memoBody);
    appendLedger({
      threadId,
      vendor: plan.vendorFrom,
      verdict: 'QUEUE',
      reasons: plan.reasons,
      mfrNumbers: plan.mfrNumbers,
      memoPath,
      fieldsWritten: [],
    });
    state[key] = { at: new Date().toISOString(), verdict: 'QUEUE', memoPath };
    actions.push({ verdict: 'QUEUE', threadId, vendor: plan.vendorFrom, mfrNumbers: plan.mfrNumbers, reasons: plan.reasons });
    log('QUEUE', threadId, plan.vendorFrom, '->', plan.reasons.join('; '));
    return;
  }

  // COMMIT path — but only if we have at least one clean mfr# to act on.
  if (plan.mfrNumbers.length === 0) {
    const memoPath = queueMemo(
      `nomfr-${threadId}`,
      `# discontinued-agent — discontinuation stated but no mfr# mapped\n\n` +
        `- Thread: ${threadId}\n- Vendor: ${plan.vendorFrom}\n` +
        `- The vendor reply says discontinued but no manufacturer number could be\n` +
        `  cleanly extracted from the DW outbound. Queued for Steve.\n\n` +
        '## Vendor reply\n```\n' + plan.vendorText.slice(0, 1500) + '\n```\n'
    );
    appendLedger({ threadId, vendor: plan.vendorFrom, verdict: 'QUEUE_NO_MFR', reasons: plan.reasons, mfrNumbers: [], memoPath, fieldsWritten: [] });
    state[key] = { at: new Date().toISOString(), verdict: 'QUEUE_NO_MFR', memoPath };
    actions.push({ verdict: 'QUEUE_NO_MFR', threadId, vendor: plan.vendorFrom, mfrNumbers: [], reasons: plan.reasons });
    log('QUEUE(no mfr#)', threadId, plan.vendorFrom);
    return;
  }

  const result = await applyCommit(plan);
  const fieldsWritten = [];
  const recordIds = [];
  for (const pm of result.perMfr) {
    for (const w of [...(pm.discoWrites || []), ...(pm.closeWrites || [])]) {
      recordIds.push(w.recordId);
      fieldsWritten.push({ recordId: w.recordId, field: w.field, committed: !!w.committed, changes: w.changes });
    }
  }

  appendLedger({
    threadId,
    vendor: plan.vendorFrom,
    verdict: AUTOCOMMIT ? 'COMMIT' : 'STAGED(dry-run)',
    reasons: plan.reasons,
    mfrNumbers: plan.mfrNumbers,
    today: result.today,
    recordIds,
    fieldsWritten,
    autocommit: AUTOCOMMIT,
  });

  // Only mark processed when we actually committed. In dry-run mode we leave it
  // un-processed so a later run with DISCO_AUTOCOMMIT=1 can complete the write.
  if (AUTOCOMMIT) {
    state[key] = { at: new Date().toISOString(), verdict: 'COMMIT', recordIds };
  }
  actions.push({
    verdict: AUTOCOMMIT ? 'COMMIT' : 'STAGED',
    threadId,
    vendor: plan.vendorFrom,
    mfrNumbers: plan.mfrNumbers,
    recordIds,
  });
  log(
    AUTOCOMMIT ? 'COMMIT' : 'STAGED(dry-run)',
    threadId,
    plan.vendorFrom,
    'mfr#', plan.mfrNumbers.join(','),
    'rows', recordIds.length
  );
}

// Build + send the run-completion report to info@ (only when actions occurred).
async function sendRunReport(actions) {
  if (!REPORT_ENABLED || actions.length === 0) return;
  const byVerdict = (v) => actions.filter((a) => a.verdict === v);
  const committed = [...byVerdict('COMMIT'), ...byVerdict('STAGED')];
  const queued = [...byVerdict('QUEUE'), ...byVerdict('QUEUE_NO_MFR')];
  const line = (a) =>
    `<li><b>${a.vendor || 'unknown vendor'}</b> — mfr# ${(a.mfrNumbers || []).join(', ') || '(none)'}` +
    (a.recordIds ? ` — ${a.recordIds.length} row(s)` : '') +
    (a.reasons ? ` — <i>${a.reasons.join('; ')}</i>` : '') +
    ` <span style="color:#888">[thread ${a.threadId}]</span></li>`;
  const mode = AUTOCOMMIT ? 'AUTO-COMMITTED to FileMaker' : 'STAGED (dry-run — DISCO_AUTOCOMMIT is OFF)';
  const body = [
    `<p><b>discontinued-agent run report</b> — ${new Date().toISOString()}</p>`,
    `<p>Discontinuations ${mode}: <b>${committed.length}</b> · Queued for review: <b>${queued.length}</b></p>`,
    committed.length ? `<p><b>Discontinued / closed:</b></p><ul>${committed.map(line).join('')}</ul>` : '',
    queued.length ? `<p><b>Queued for Steve (not auto-written):</b></p><ul>${queued.map(line).join('')}</ul>` : '',
    `<p style="color:#888">Ledger: data/ledger.jsonl · Queue memos: ~/.claude/yolo-queue/pending-approval/</p>`,
  ].join('\n');
  try {
    await George.sendMail({
      to: REPORT_TO,
      subject: `discontinued-agent — ${committed.length} discontinued, ${queued.length} queued`,
      body,
    });
    log(`report emailed to ${REPORT_TO} (${committed.length} disco / ${queued.length} queued)`);
  } catch (e) {
    log('WARN report email failed:', e.message);
  }
}

async function scanOnce() {
  const state = loadProcessed();
  let headers;
  try {
    headers = await George.listMessages(GMAIL_SEARCH, { maxResults: MAX_RESULTS });
  } catch (e) {
    log('George list failed:', e.message);
    return;
  }
  const threads = groupByThread(headers);
  log(`scanned ${headers.length} msgs across ${threads.size} threads`);

  const actions = [];
  for (const [threadId, msgs] of threads) {
    try {
      await processThread(threadId, msgs, state, actions);
    } catch (e) {
      log('thread error', threadId, e.message);
    }
  }
  saveProcessed(state);
  await sendRunReport(actions);
}

async function main() {
  const once = process.argv.includes('--once');
  log(`starting. autocommit=${AUTOCOMMIT ? 'ON' : 'OFF (dry-run/stage only)'} search="${GMAIL_SEARCH}"`);

  // Health preflight (non-fatal).
  try {
    const h = await George.health();
    log('George health:', h.status || 'ok');
  } catch (e) {
    log('WARN George health check failed:', e.message);
  }
  try {
    const p = await FM.ping();
    log('FileMaker ping:', p.ok ? `ok (${p.database} @ ${p.host})` : 'unexpected');
  } catch (e) {
    log('WARN FileMaker ping failed:', e.message);
  }

  if (once) {
    await scanOnce();
    log('single scan complete.');
    return;
  }

  // Daemon loop — every POLL_INTERVAL_SEC seconds, never overlapping.
  log(`daemon loop every ${POLL_INTERVAL_SEC}s`);
  for (;;) {
    try {
      await scanOnce();
    } catch (e) {
      log('scan error:', e.message);
    }
    await new Promise((r) => setTimeout(r, POLL_INTERVAL_SEC * 1000));
  }
}

main().catch((e) => {
  log('FATAL', e.stack || e.message);
  process.exit(1);
});