`src/adapters/base.js`

243 lines
'use strict';
/**
 * Public-source adapter contract + common pipeline (spec §10).
 *
 * AdvertiserSourceAdapter is the base class every source implements:
 *   - policy                  the SourcePolicy this adapter runs under
 *   - discover(cursor)        async generator of DiscoveredSourceItem
 *   - fetch(item)             retrieve bytes (only when policy permits)
 *   - parse(input)            async generator of extractions
 *
 * runPipeline() provides the shared machinery the spec mandates:
 *   policy validation before every run, per-host throttle, exponential backoff
 *   with jitter, conditional-request hook, idempotent upsert hook, dead-letter
 *   recording into ingestion_jobs (status DEAD_LETTER), dry-run mode,
 *   incremental cursor, and job logging into ingestion_runs.
 */

const { assertSourceEnabledLegal } = require('../../lib/compliance/source-policy');

/**
 * Base adapter. Concrete adapters override discover/fetch/parse. The defaults
 * are deliberately inert (yield nothing / refuse) so a half-built adapter can
 * never accidentally fetch.
 */
class AdvertiserSourceAdapter {
  constructor(policy) {
    if (!policy || typeof policy !== 'object') {
      throw new Error('AdvertiserSourceAdapter: a SourcePolicy is required');
    }
    this.policy = policy;
  }

  // eslint-disable-next-line require-yield
  async *discover(_cursor) {
    // Default: discover nothing. Concrete adapters override.
    return;
  }

  async fetch(_item) {
    throw new Error(
      `${this.constructor.name}.fetch not implemented — adapter must define fetch or be manual_review_only`
    );
  }

  // eslint-disable-next-line require-yield
  async *parse(_input) {
    return;
  }
}

/** Exponential backoff with full jitter, capped. */
function backoffDelay(attempt, baseMs = 500, capMs = 30000) {
  const exp = Math.min(capMs, baseMs * 2 ** attempt);
  return Math.floor(Math.random() * exp); // full jitter
}

function sleep(ms) {
  return new Promise((r) => setTimeout(r, ms));
}

/** Simple per-host throttle used inside the pipeline (delegates to policy). */
async function throttle(policy, lastByHost, host) {
  const rpm = policy.maxRequestsPerMinute || Number(process.env.DEFAULT_REQUESTS_PER_MINUTE || 6);
  const minDelay = policy.minimumDelayMs != null ? policy.minimumDelayMs : Math.ceil(60000 / rpm);
  const last = lastByHost.get(host) || 0;
  const since = Date.now() - last;
  if (last && since < minDelay) await sleep(minDelay - since);
  lastByHost.set(host, Date.now());
}

function hostOf(url) {
  try {
    return new URL(url).hostname.toLowerCase();
  } catch (_e) {
    return 'unknown';
  }
}

/**
 * runPipeline(adapter, opts)
 *
 * opts:
 *   db        { query } — the db module (required for real logging; optional in
 *                         dry-run tests where you can pass a stub or omit)
 *   dryRun    boolean   — discover + parse but never fetch or upsert
 *   upsert    async(extraction, ctx) => void  — idempotent write callback
 *   cursor    string    — incremental cursor passed to discover()
 *   maxItems  number    — safety cap
 *   maxAttempts number  — per-item retry budget before dead-letter (default 3)
 *   conditionalRequest  async(item) => ({skip:boolean, headers?:object})
 *                        — conditional-request hook (etag/last-modified)
 *
 * Returns { runId, stats, cursor }.
 */
async function runPipeline(adapter, opts = {}) {
  if (!adapter || !(adapter instanceof AdvertiserSourceAdapter)) {
    throw new Error('runPipeline: adapter must extend AdvertiserSourceAdapter');
  }
  const policy = adapter.policy;

  // §10 — policy validation BEFORE every run. Throws on an illegal enabled source.
  assertSourceEnabledLegal(policy);

  const db = opts.db || null;
  const dryRun = opts.dryRun === true;
  const upsert = typeof opts.upsert === 'function' ? opts.upsert : null;
  const maxItems = opts.maxItems || 1000;
  const maxAttempts = opts.maxAttempts || 3;
  const conditionalRequest =
    typeof opts.conditionalRequest === 'function' ? opts.conditionalRequest : null;

  const stats = {
    discovered: 0,
    fetched: 0,
    skippedConditional: 0,
    parsed: 0,
    upserted: 0,
    deadLettered: 0,
    errors: 0,
  };
  const lastByHost = new Map();

  // Open a run row (ingestion_runs).
  let runId = null;
  if (db && db.query) {
    const r = await db.query(
      `INSERT INTO ingestion_runs (source_key, status, dry_run, stats)
       VALUES ($1, 'RUNNING', $2, $3::jsonb) RETURNING id`,
      [policy.sourceKey, dryRun, JSON.stringify(stats)]
    );
    runId = r.rows[0].id;
  }

  const deadLetter = async (item, err) => {
    stats.deadLettered += 1;
    if (db && db.query && runId) {
      await db.query(
        `INSERT INTO ingestion_jobs (run_id, job_type, payload, status, attempts, last_error)
         VALUES ($1, 'FETCH_PARSE', $2::jsonb, 'DEAD_LETTER', $3, $4)`,
        [runId, JSON.stringify(item || {}), maxAttempts, String((err && err.message) || err)]
      );
    }
  };

  let finalError = null;
  let cursor = opts.cursor;

  try {
    for await (const item of adapter.discover(opts.cursor)) {
      stats.discovered += 1;
      if (item && item.externalId) cursor = item.externalId; // incremental cursor
      if (stats.discovered > maxItems) break;

      if (dryRun) {
        // Dry-run: exercise discover only; no fetch, no writes.
        continue;
      }

      // Conditional-request hook (etag/last-modified) — skip unchanged.
      if (conditionalRequest) {
        try {
          const cr = await conditionalRequest(item);
          if (cr && cr.skip) {
            stats.skippedConditional += 1;
            continue;
          }
        } catch (_e) {
          /* non-fatal; proceed to fetch */
        }
      }

      // Fetch with retry + exponential backoff + jitter → dead-letter on give-up.
      let fetched = null;
      let attempt = 0;
      // eslint-disable-next-line no-constant-condition
      while (true) {
        try {
          await throttle(policy, lastByHost, hostOf(item.url));
          fetched = await adapter.fetch(item);
          stats.fetched += 1;
          break;
        } catch (err) {
          attempt += 1;
          stats.errors += 1;
          if (attempt >= maxAttempts) {
            await deadLetter(item, err);
            break;
          }
          await sleep(backoffDelay(attempt));
        }
      }
      if (!fetched) continue; // dead-lettered

      // Parse + idempotent upsert.
      try {
        for await (const extraction of adapter.parse({
          bytes: fetched.bytes,
          contentType: fetched.contentType,
          finalUrl: fetched.finalUrl,
        })) {
          stats.parsed += 1;
          if (upsert) {
            await upsert(extraction, { item, fetched, policy, runId });
            stats.upserted += 1;
          }
        }
      } catch (err) {
        stats.errors += 1;
        await deadLetter(item, err);
      }
    }
  } catch (err) {
    finalError = err;
  }

  // Close the run row.
  if (db && db.query && runId) {
    await db.query(
      `UPDATE ingestion_runs
         SET finished_at = now(),
             status = $2,
             stats = $3::jsonb,
             error = $4
       WHERE id = $1`,
      [
        runId,
        finalError ? 'FAILED' : 'COMPLETED',
        JSON.stringify(stats),
        finalError ? String(finalError.message || finalError) : null,
      ]
    );
  }

  if (finalError) throw finalError;
  return { runId, stats, cursor };
}

module.exports = {
  AdvertiserSourceAdapter,
  runPipeline,
  backoffDelay,
};