`modules/engine/scheduler.js`

192 lines
// Engine scheduler (agent 3). Fires APPROVED, due suggestions through the same
// live rails as /publish (channels.postLive). Safe-by-construction:
//   • ENGINE_SCHEDULER=off → never starts the interval (dev / test / staging).
//   • Every state change goes through store.transition() — the compare-and-set
//     that makes double-posting impossible (two ticks racing the same item: one
//     wins the approved→posting claim, the loser's transition returns null).
//   • Items are processed SEQUENTIALLY (never in parallel) to avoid platform
//     rate spikes.
//   • attempts is capped at 1: a failed post is NEVER auto-refired — Steve must
//     re-approve. This is deliberate. A post that threw/failed may still have
//     LANDED on the platform (the adapter succeeded but the response was lost),
//     so silently retrying risks a duplicate live post.
//   • CRASH SWEEP on start: any item stuck in 'posting' (the process died
//     mid-post) is moved to 'failed' with a verify-first note — NEVER
//     auto-refired, for the same land-before-crash reason.
const store = require('./store');

let _timer = null;
let _started = false;
let _ticking = false;
let _lastTick = null;

// ── logging ──────────────────────────────────────────────────────────────────
// Matches the repo's bracketed-prefix console style ([mcc]/[engine]).
const log = (...a) => console.log('[engine]', ...a);

// summarize(detail) — best-effort one-line error from an adapter result array
// when postLive returned no top-level `error` string (it doesn't on the happy
// path, but a per-adapter failure lives inside detail[].error).
function summarize(detail) {
  try {
    if (Array.isArray(detail)) {
      const errs = detail.map(d => d && d.error).filter(Boolean);
      if (errs.length) return errs.join('; ');
    }
    if (detail && typeof detail === 'object' && detail.error) return String(detail.error);
  } catch { /* fall through */ }
  return 'post failed';
}

// ── crash sweep ────────────────────────────────────────────────────────────────
// Any item left in 'posting' means the process died between the approved→posting
// claim and the posting→posted|failed resolve. The post MAY have landed, so we
// move it to 'failed' (never auto-refire) and flag it for manual verification.
function _sweep() {
  let swept = 0;
  for (const it of store.load()) {
    if (it.status === 'posting') {
      const done = store.transition(it.id, 'posting', 'failed', {
        lastError: 'interrupted mid-post — verify on platform before re-approving',
      });
      if (done) { swept++; log('crash-sweep: posting →', it.id, 'moved to failed (verify before re-approve)'); }
    }
  }
  if (swept) log(`crash-sweep swept ${swept} interrupted item(s)`);
  return swept;
}

// ── content mapping ────────────────────────────────────────────────────────────
// Build the channels.postLive `content` object from a queue item.
// Caption = the caption text, then (only if hashtags exist) a blank line + the
// hashtags joined by a single space. The product link is the conversion path:
// link-friendly channels carry it in the caption (Bluesky auto-facets it into a
// tappable link); YouTube carries it in the description. IG/TikTok captions don't
// render clickable links, so it's omitted there rather than shipped dead.
const LINK_CHANNELS = new Set(['bluesky', 'facebook', 'threads', 'linkedin']);
function buildContent(item) {
  const tags = Array.isArray(item.hashtags) ? item.hashtags.filter(Boolean) : [];
  const url = item.product && item.product.productUrl;
  // Bluesky hard-slices at 300 chars — budget the caption so the link never gets chopped.
  let cap = item.caption || '';
  if (item.channel === 'bluesky' && url) {
    const budget = 300 - (url.length + 2 + (tags.length ? tags.join(' ').length + 2 : 0));
    if (cap.length > budget) cap = cap.slice(0, Math.max(0, budget - 1)).replace(/\s+\S*$/, '') + '…';
  }
  const parts = [cap];
  if (url && LINK_CHANNELS.has(item.channel)) parts.push(url);
  if (tags.length) parts.push(tags.join(' '));
  const caption = parts.filter(Boolean).join('\n\n');
  let description = item.description || undefined;
  if (item.channel === 'youtube' && url) description = [description, url].filter(Boolean).join('\n\n');
  return {
    caption,
    mediaUrl: item.mediaUrl,
    videoUrl: item.videoUrl,
    privacy: item.privacy || undefined,
    title: item.channel === 'youtube' ? item.caption : undefined,
    description,
  };
}

// dueApproved() — items ready to fire this tick: approved, scheduled in the past,
// and under the attempts cap (never re-fire a failed post).
function dueApproved(items) {
  const now = Date.now();
  return items.filter(it =>
    it.status === 'approved' &&
    it.scheduledAt &&
    Number.isFinite(Date.parse(it.scheduledAt)) &&
    Date.parse(it.scheduledAt) <= now &&
    (it.attempts || 0) < 1
  );
}

// ── tick ────────────────────────────────────────────────────────────────────────
async function _tickOnce() {
  _lastTick = new Date().toISOString();
  const items = store.load();
  const due = dueApproved(items);
  if (!due.length) return;

  // require channels lazily + freshly so a test can monkey-patch postLive.
  const channels = require('../channels');

  for (const item of due) {
    // (1) CLAIM: compare-and-set approved→posting. If null, another tick or a
    //     manual action already took it — skip.
    const claimed = store.transition(item.id, 'approved', 'posting', {
      attempts: (item.attempts || 0) + 1,
    });
    if (!claimed) continue;

    // (2) POST — wrapped so one item's failure never kills the tick.
    try {
      const content = buildContent(claimed);
      const res = await channels.postLive(claimed.channel, content, {});
      const ok = !!(res && res.ok);
      store.transition(claimed.id, 'posting', ok ? 'posted' : 'failed', {
        detail: (res && res.detail) || null,
        outboxId: (res && res.outboxId) || null,
        lastError: ok ? null : ((res && res.error) || summarize(res && res.detail)),
        postedAt: ok ? new Date().toISOString() : null,
      });
      if (ok) log('posted', claimed.id, '→ outbox', (res && res.outboxId) || '(none)');
      else log('FAILED', claimed.id + ':', (res && res.error) || summarize(res && res.detail));
    } catch (err) {
      const msg = (err && err.message) || String(err);
      store.transition(claimed.id, 'posting', 'failed', { lastError: msg, postedAt: null });
      log('FAILED', claimed.id + ':', msg);
    }
  }
}

// tick() — the interval body: re-entrancy guard so a slow tick never overlaps.
async function tick() {
  if (_ticking) return;
  _ticking = true;
  try {
    await _tickOnce();
  } catch (e) {
    log('tick error:', (e && e.message) || String(e));
  } finally {
    _ticking = false;
  }
}

// ── lifecycle ────────────────────────────────────────────────────────────────
function start() {
  if (process.env.ENGINE_SCHEDULER === 'off') {
    log('scheduler disabled (ENGINE_SCHEDULER=off) — not starting');
    return;
  }
  if (_started) return;
  _started = true;

  // Crash sweep BEFORE the interval so interrupted items are resolved up front.
  _sweep();

  // Steady 60s cadence, plus one prompt tick 5s after boot. Both unref()'d so
  // they never keep the process alive on their own (mirrors social/index.js).
  _timer = setInterval(() => { tick().catch(() => {}); }, 60 * 1000);
  if (_timer.unref) _timer.unref();

  const kick = setTimeout(() => { tick().catch(() => {}); }, 5 * 1000);
  if (kick.unref) kick.unref();

  log('scheduler started (60s cadence)');
}

function getState() {
  return { started: _started, lastTick: _lastTick, ticking: _ticking };
}

module.exports = {
  start,
  getState,
  lastTick: () => _lastTick,
  // exported for tests only:
  _tickOnce,
  _sweep,
};