prover/fetcher.js

/**
 * prover/fetcher.js
 * ----------------------------------------------------------
 * Phase 3 component #1 of the off-chain prover.
 *
 * Responsibilities:
 *   1. Read config/assets.json -> [{ assetId, ipfsCid, expectedHash }]
 *   2. For each asset:
 *        a. Fetch the deed PDF from an IPFS gateway.
 *        b. Compute SHA-256 of the bytes.
 *        c. Compare against `expectedHash`.
 *        d. Emit a "ProofFresh" or "ProofMismatch" event for the
 *           submitter to relay on-chain.
 *
 * The fetcher is intentionally stateless. It is run on a cron
 * (every 5 minutes by default) and writes the latest result to
 * .local/state/prover-state.json so the dashboard can render it.
 *
 * Usage:
 *   node prover/fetcher.js               # one-shot
 *   node prover/fetcher.js --watch       # poll forever
 */

const fs = require('fs');
const path = require('path');
const crypto = require('crypto');
const notifier = require('./notifier');

const ROOT = path.resolve(__dirname, '..');
const ASSETS_PATH = path.join(ROOT, 'config', 'assets.json');
const STATE_DIR = path.join(ROOT, '.local', 'state');
const STATE_PATH = path.join(STATE_DIR, 'prover-state.json');

const { resolveCID, evaluateResolution } = require('./ipfsResolver');
const { scoreAsset } = require('./scorer');

const POLL_MS = Number(process.env.FETCHER_POLL_MS || 5 * 60 * 1000);

// Exponential backoff
function backoff(attempt, baseMs = 1000) {
  const delay = baseMs * Math.pow(2, attempt);
  return new Promise(resolve => setTimeout(resolve, delay));
}

// Structured logger
function logEvent(level, message, meta = {}) {
  const entry = { timestamp: new Date().toISOString(), level, message, ...meta };
  const output = JSON.stringify(entry);
  if (level === 'error') console.error(output);
  else console.log(output);
}

function loadAssets() {
  const raw = fs.readFileSync(ASSETS_PATH, 'utf8');
  const parsed = JSON.parse(raw);
  if (!Array.isArray(parsed)) throw new Error('assets.json must be an array');
  return parsed;
}

function sha256(buf) {
  return '0x' + crypto.createHash('sha256').update(buf).digest('hex');
}

// Phase 4: Gateway Quorum Resolution
async function resolveWithQuorum(cid, expectedHash) {
  logEvent('info', 'Starting gateway quorum resolution', { cid });

  const results = await resolveCID(cid);
  const outcome = evaluateResolution(results, expectedHash);

  const successes = results.filter(r => r.ok);
  const failures = results.filter(r => !r.ok);

  logEvent('info', 'Resolution complete', {
    cid,
    status: outcome.status,
    successes: successes.length,
    failures: failures.length,
    observedHash: outcome.observedHash
  });

  return {
    results,
    outcome,
    successes,
    failures
  };
}

async function checkAsset(asset, previousResult) {
  const start = Date.now();
  const result = {
    assetId: asset.assetId,
    ipfsCid: asset.ipfsCid,
    expectedHash: asset.expectedHash,
    actualHash: null,
    status: 'unknown',
    health: 'unknown',
    gateway: null,
    bytes: 0,
    error: null,
    durationMs: 0,
    checkedAt: new Date().toISOString(),
    consecutiveUnreachable: 0,
    // Phase 4 additions
    resolutionStatus: null,
    gatewayResults: [],
    evidence: null
  };

  try {
    const { results, outcome, successes, failures } = await resolveWithQuorum(asset.ipfsCid, asset.expectedHash);

    result.resolutionStatus = outcome.status;
    result.gatewayResults = results;
    const score = scoreAsset(results, asset.expectedHash);
    result.triggerScore = score.triggerScore;
    result.scenario = score.scenario;
    result.thresholdUsed = score.thresholdUsed;

    switch (outcome.status) {
      case 'CONSISTENT':
        result.actualHash = outcome.observedHash;
        result.status = 'fresh';
        result.health = failures.length === 0 ? 'healthy' : 'degraded';
        // Find first successful gateway for reporting
        const firstSuccess = successes[0];
        if (firstSuccess) {
          result.gateway = firstSuccess.gateway;
        }
        break;

      case 'HASH_MISMATCH':
        result.status = 'mismatch';
        result.health = 'compromised';
        result.evidence = outcome.evidence;
        logEvent('error', 'Hash mismatch detected', {
          assetId: asset.assetId,
          cid: asset.ipfsCid,
          evidence: outcome.evidence
        });
        break;

      case 'NETWORK_UNAVAILABLE':
        result.status = 'unreachable';
        result.error = 'all gateways failed';
        result.health = 'unreachable';
        logEvent('warn', 'Network unavailable', {
          assetId: asset.assetId,
          cid: asset.ipfsCid,
          failures: failures.length
        });
        break;
    }
  } catch (err) {
    result.status = 'error';
    result.error = err.message;
    result.health = 'error';
    logEvent('error', 'Resolution failed', {
      assetId: asset.assetId,
      error: err.message
    });
  }

  result.durationMs = Date.now() - start;

  // Compute consecutive unreachables (updated for Phase 4)
  const prevCount = previousResult?.consecutiveUnreachable || 0;
  if (result.status === 'unreachable' || result.resolutionStatus === 'NETWORK_UNAVAILABLE') {
    result.consecutiveUnreachable = prevCount + 1;
  } else {
    result.consecutiveUnreachable = 0;
  }

  // Apply threshold for network unavailable
  const threshold = Number(process.env.MAX_UNREACHABLE_RETRIES || 3);
  if ((result.status === 'unreachable' || result.resolutionStatus === 'NETWORK_UNAVAILABLE') &&
      result.consecutiveUnreachable >= threshold) {
    result.status = 'unreachable_threshold';
  }

  return result;
}

function loadPreviousState() {
  if (!fs.existsSync(STATE_PATH)) return {};
  try {
    return JSON.parse(fs.readFileSync(STATE_PATH, 'utf8'));
  } catch {
    return {};
  }
}

function persistState(results) {
  fs.mkdirSync(STATE_DIR, { recursive: true });
  const payload = {
    runAt: new Date().toISOString(),
    results,
    summary: {
      total: results.length,
      fresh: results.filter((r) => r.status === 'fresh').length,
      mismatch: results.filter((r) => r.status === 'mismatch').length,
      unreachable: results.filter((r) => r.status === 'unreachable').length,
      unreachable_threshold: results.filter((r) => r.status === 'unreachable_threshold').length,
      // Phase 4 additions
      network_unavailable: results.filter((r) => r.resolutionStatus === 'NETWORK_UNAVAILABLE').length,
      hash_mismatch: results.filter((r) => r.resolutionStatus === 'HASH_MISMATCH').length,
      consistent: results.filter((r) => r.resolutionStatus === 'CONSISTENT').length,
    },
  };
  fs.writeFileSync(STATE_PATH, JSON.stringify(payload, null, 2));
  return payload;
}

async function runOnce() {
  // Notify pipeline start
  try { notifier.fetchStart({ network: 'Polygon Amoy' }); } catch (_) {}

  const assets = loadAssets();
  const previous = loadPreviousState();
  const previousResults = previous.results || [];
  const prevMap = {};
  for (const r of previousResults) {
    prevMap[r.assetId] = r;
  }
  console.log(`[fetcher] checking ${assets.length} asset(s)`);
  const results = [];
  for (const asset of assets) {
    const r = await checkAsset(asset, prevMap[asset.assetId]);
    results.push(r);
    console.log(
      `[fetcher] ${r.assetId}  status=${r.status}  ` +
        (r.actualHash ? `hash=${r.actualHash.slice(0, 14)}…  ` : '') +
        (r.consecutiveUnreachable ? `consec=${r.consecutiveUnreachable}  ` : '') +
        `(${r.durationMs}ms)`
    );
  }
  const payload = persistState(results);
  console.log(
    `[fetcher] summary  fresh=${payload.summary.fresh}  ` +
      `mismatch=${payload.summary.mismatch}  unreachable=${payload.summary.unreachable}  threshold=${payload.summary.unreachable_threshold}`
  );
  console.log(
    `[fetcher] Phase 4:  consistent=${payload.summary.consistent}  ` +
      `hash_mismatch=${payload.summary.hash_mismatch}  network_unavailable=${payload.summary.network_unavailable}`
  );

  // Notify completion
  try {
    notifier.fetchComplete({
      network: 'Polygon Amoy',
      count: results.length,
      successful: results.filter(r => r.status === 'fresh' || r.status === 'mismatch').length,
      failed: results.filter(r => r.status === 'error' || r.status === 'unreachable' || r.status === 'unreachable_threshold').length,
    });
  } catch (_) {}

  return payload;
}

async function watch() {
  for (;;) {
    try {
      await runOnce();
    } catch (err) {
      console.error('[fetcher] run failed:', err.message || err);
    }
    await new Promise((r) => setTimeout(r, POLL_MS));
  }
}

if (require.main === module) {
  const watchMode = process.argv.includes('--watch');
  (watchMode ? watch() : runOnce()).catch((err) => {
    try { notifier.fetchError({ error: err.message, stack: err.stack }); } catch (_) {}
    console.error(err);
    process.exit(1);
  });
}

module.exports = { runOnce, checkAsset, sha256, loadAssets, loadPreviousState };