verae-zapier-simulator/src/monitors.js

/**
 * Scan a trace for assumption breaks, delays, failures, recoveries,
 * and suggest changes before the Zapier app goes live.
 * @module monitors
 */

const ASSUMPTIONS = [
  {
    id: 'zapier-no-nats',
    title: 'Zapier never publishes or subscribes to NATS',
    check(events) {
      const bad = events.filter(
        (e) =>
          e.module === 'zapier-platform-app' &&
          (String(e.address).startsWith('verae.') || String(e.hop).includes('nats')),
      );
      return bad.length
        ? { ok: false, detail: `Zapier hop used NATS address ${bad[0].address}` }
        : { ok: true, detail: 'Zapier hops are HTTPS only' };
    },
  },
  {
    id: 'chain-no-bytes',
    title: 'Main chain never receives file bytes or private metadata',
    check(events) {
      const chain = events.filter((e) => e.module === 'verae-chain-client' && e.payload);
      for (const e of chain) {
        const p = e.payload;
        if (p.contentBase64 || p.files || p.privateMetadata) {
          return { ok: false, detail: `chain payload included ${Object.keys(p).join(',')}` };
        }
      }
      return { ok: true, detail: 'chain payloads are hash/root only' };
    },
  },
  {
    id: 'bloom-silence',
    title: 'Bloom miss produces no archive.reply packet',
    check(events) {
      const leaks = events.filter(
        (e) => e.module.startsWith('archive-') || e.module.startsWith('tree-node'),
      ).filter((e) => e.status === 'silent' && e.address.includes('archive.reply'));
      return leaks.length
        ? { ok: false, detail: 'silent hop still addressed a reply subject' }
        : { ok: true, detail: 'misses logged as silent with no reply address' };
    },
  },
  {
    id: 'archive-query-broadcast',
    title: 'archive.query is broadcast (not a shared queue group)',
    check(events) {
      const queries = events.filter((e) => e.address === 'verae.archive.query');
      const replies = events.filter((e) => String(e.address).startsWith('verae.archive.reply.'));
      if (!queries.length) return { ok: true, detail: 'no archive query this run' };
      return { ok: true, detail: `${queries.length} query hop(s), ${replies.length} reply hop(s)` };
    },
  },
];

const DELAY_MS = 200;

/**
 * @param {object[]} events
 * @param {object} [ctx]
 */
export function analyze(events, ctx = {}) {
  const findings = [];
  const suggestions = [];

  for (const a of ASSUMPTIONS) {
    const r = a.check(events);
    findings.push({
      kind: 'assumption',
      id: a.id,
      title: a.title,
      ok: r.ok,
      detail: r.detail,
    });
  }

  const errors = events.filter((e) => e.status === 'error');
  const retries = events.filter((e) => e.status === 'retry');
  const delays = events.filter((e) => e.status === 'delay' || (e.durationMs || 0) >= DELAY_MS);

  for (const e of errors) {
    findings.push({
      kind: 'error',
      id: `err-${e.seq}`,
      title: `${e.module} ${e.hop} failed`,
      ok: false,
      detail: e.note || e.address,
    });
  }
  for (const e of delays) {
    findings.push({
      kind: 'delay',
      id: `delay-${e.seq}`,
      title: `${e.module} took ${e.durationMs || e.tMs}ms`,
      ok: false,
      detail: e.note || 'exceeds 200ms hop budget used for Zapier wait steps',
    });
  }
  if (retries.length) {
    findings.push({
      kind: 'recovery',
      id: 'retries',
      title: `${retries.length} recovery retry hop(s)`,
      ok: true,
      detail: retries.map((e) => `${e.module}:${e.hop}`).join(', '),
    });
  }

  const last = events[events.length - 1];
  if (last?.status === 'error' && retries.length === 0) {
    suggestions.push({
      id: 'add-retry',
      severity: 'high',
      text: 'Add a single retry on jobs.watch / jobs.events drop before returning pending to Zapier. Zapier will otherwise mark the step failed with no jobId recovery path.',
    });
  }

  const lookup = events.find((e) => e.hop === 'hash_lookup' || e.module === 'hash-lookup');
  const treeQuery = events.some((e) => e.payload?.kinds?.includes?.('tree') || e.hop === 'tree_query');
  const chainMiss = events.some(
    (e) => e.module === 'verae-chain-client' && e.hop === 'lookup' && e.status === 'silent',
  );
  if (chainMiss && !treeQuery) {
    suggestions.push({
      id: 'tree-lookup-action',
      severity: 'high',
      text: 'Central chain miss for a SHA-256 that may live only in a bulk Merkle summary. Add Zapier search “Find Hash (tree nodes)” that sets includeTree=true so middleware broadcasts verae.archive.query kinds=["tree"].',
    });
  }

  const puts = events.filter((e) => e.address === 'verae.archive.put');
  const queries = events.filter((e) => e.address === 'verae.archive.query');
  if (puts.length && !queries.length && ctx.includeAttached) {
    suggestions.push({
      id: 'query-after-put',
      severity: 'high',
      text: 'Archive puts ran but wait path never published verae.archive.query. Wire aggregator into createTimestampAndWait when includeAttached is true.',
    });
  }
  if (puts.length && !ctx.includeAttached && ctx.action !== 'batch_timestamp') {
    suggestions.push({
      id: 'include-attached-default',
      severity: 'medium',
      text: 'Metadata/files were stored off-chain but the Zapier step did not request includeAttached. Document the flag and default it on for “Create Timestamp and Wait”.',
    });
  }

  const billing = events.find((e) => e.payload?.status === 402 || e.note?.includes?.('402'));
  if (billing) {
    suggestions.push({
      id: '402-upgrade-url',
      severity: 'medium',
      text: 'Map HTTP 402 to Zapier QuotaExceeded and always include details.upgradeUrl from zappier-edge so the user can reload credit without leaving the Zap editor.',
    });
  }

  const waitPending = events.some((e) => e.payload?.status === 'pending' && e.hop === 'wait');
  if (waitPending) {
    suggestions.push({
      id: 'wait-timeout',
      severity: 'medium',
      text: 'Wait returned pending. Pair every Create Timestamp (async) Zap with the Timestamp Completed REST Hook, or raise WAIT_TIMEOUT_MS only after measuring chain poll interval.',
    });
  }

  const natsFromZapier = events.filter(
    (e) => e.module === 'zapier-platform-app' && String(e.address).startsWith('verae.'),
  );
  if (natsFromZapier.length) {
    suggestions.push({
      id: 'never-bind-nats',
      severity: 'high',
      text: 'Zapier cloud must not reach NATS. Keep NS1 nats-server on 127.0.0.1:4222; only middleware workers subscribe.',
    });
  }

  if (ctx.action === 'batch_timestamp') {
    suggestions.push({
      id: 'batch-docs',
      severity: 'low',
      text: 'User docs must say batch items are not individually itemized on the main chain. Lookup of a member hash requires the tree-node search, not Find Timestamp by SHA256 alone.',
    });
  }

  const silentAll =
    queries.length > 0 &&
    !events.some((e) => String(e.address).startsWith('verae.archive.reply.') && e.status === 'ok');
  if (silentAll && (ctx.includeAttached || ctx.includeTree)) {
    findings.push({
      kind: 'error',
      id: 'all-archives-silent',
      title: 'All archives/tree nodes silent during query',
      ok: false,
      detail: 'Aggregator completed with zero replies. Distinguish “hash unknown” from “archives unreachable”.',
    });
    suggestions.push({
      id: 'archive-quorum',
      severity: 'high',
      text: 'Treat “zero replies + known puts” as archive outage, not a miss. Require at least one reply OR an explicit bloom-miss count from a live node heartbeat.',
    });
  }

  const ok = findings.filter((f) => f.ok === false).length === 0;
  return {
    ok,
    findings,
    suggestions,
    stats: {
      hops: events.length,
      errors: errors.length,
      retries: retries.length,
      delays: delays.length,
      natsHops: events.filter((e) => String(e.address).startsWith('verae.')).length,
    },
  };
}