revenue.js

/**
 * Revenue Tracking — lightweight version
 *
 * Scans Base Mainnet for USDC Transfer events to the payee wallet.
 * Short range per scan (last 100 blocks ≈ 200s) to keep up naturally with
 * the 60-second indexer interval. No per-tx lookups.
 *
 * Since x402 ERC-3009 payments emit a Transfer event where `from` = buyer,
 * we tag them by amount heuristic: small amounts (< 10 USDC) are very likely
 * x402 micropayments. Users can filter the dashboard.
 */

import { createPublicClient, http, parseAbiItem, formatUnits } from 'viem';
import { base } from 'viem/chains';
import { db } from './analytics.js';

const USDC = '0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913';
const PAYEE = process.env.WALLET_ADDRESS;
const RPC = process.env.BASE_RPC_URL || 'https://base.llamarpc.com';
const TRANSFER_EVENT = parseAbiItem('event Transfer(address indexed from, address indexed to, uint256 value)');

const client = createPublicClient({ chain: base, transport: http(RPC) });

// ============ DB SCHEMA ============

db.exec(`
  CREATE TABLE IF NOT EXISTS payments (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    tx_hash TEXT UNIQUE NOT NULL,
    block_number INTEGER NOT NULL,
    block_timestamp INTEGER NOT NULL,
    from_address TEXT NOT NULL,
    amount_wei TEXT NOT NULL,
    amount_usdc REAL NOT NULL,
    log_index INTEGER NOT NULL
  );
  CREATE INDEX IF NOT EXISTS idx_payments_block ON payments(block_number);
  CREATE INDEX IF NOT EXISTS idx_payments_ts ON payments(block_timestamp);
  CREATE INDEX IF NOT EXISTS idx_payments_amount ON payments(amount_usdc);

  CREATE TABLE IF NOT EXISTS indexer_state (
    key TEXT PRIMARY KEY,
    value TEXT NOT NULL
  );
`);

// Ensure enrichment columns exist
try { db.exec('ALTER TABLE payments ADD COLUMN tx_origin TEXT'); } catch {}
try { db.exec('ALTER TABLE payments ADD COLUMN is_x402 INTEGER DEFAULT 0'); } catch {}
try { db.exec('CREATE INDEX IF NOT EXISTS idx_payments_x402 ON payments(is_x402)'); } catch {}
try { db.exec('CREATE INDEX IF NOT EXISTS idx_payments_origin ON payments(tx_origin)'); } catch {}

const FACILITATOR = '0x227a3a70e889023c437c49fb15d492b06cfa4986';

const insertPayment = db.prepare(`
  INSERT OR IGNORE INTO payments
  (tx_hash, block_number, block_timestamp, from_address, amount_wei, amount_usdc, log_index)
  VALUES (?, ?, ?, ?, ?, ?, ?)
`);

const getUnenrichedSmallPayments = db.prepare(`
  SELECT id, tx_hash FROM payments
  WHERE tx_origin IS NULL AND amount_usdc < 10
  ORDER BY block_number DESC
  LIMIT ?
`);

const updatePaymentOrigin = db.prepare(`
  UPDATE payments SET tx_origin = ?, is_x402 = ? WHERE id = ?
`);

const getState = db.prepare('SELECT value FROM indexer_state WHERE key = ?');
const setState = db.prepare('INSERT OR REPLACE INTO indexer_state (key, value) VALUES (?, ?)');

// ============ INDEXER ============

async function getLastIndexedBlock() {
  const row = getState.get('last_indexed_block');
  return row ? parseInt(row.value) : null;
}

async function setLastIndexedBlock(n) {
  setState.run('last_indexed_block', String(n));
}

/**
 * Lightweight indexer: max 200 blocks per run, no per-tx lookups.
 * Uses approximate timestamps (block_number → current_timestamp - (blocks_behind * 2)).
 */
export async function indexPayments() {
  try {
    const latestBlock = Number(await client.getBlockNumber());
    let fromBlock = await getLastIndexedBlock();

    if (!fromBlock) {
      // First run: scan last 500 blocks (~16 min on Base)
      fromBlock = latestBlock - 500;
    } else {
      fromBlock = fromBlock + 1;
    }

    if (fromBlock > latestBlock) return { indexed: 0, upToBlock: latestBlock };

    // Cap per-scan range at 200 blocks to prevent slow runs
    const MAX_PER_RUN = 200;
    const toBlock = Math.min(fromBlock + MAX_PER_RUN - 1, latestBlock);

    const logs = await client.getLogs({
      address: USDC,
      event: TRANSFER_EVENT,
      args: { to: PAYEE },
      fromBlock: BigInt(fromBlock),
      toBlock: BigInt(toBlock),
    });

    // Approximate timestamp: assume 2-second blocks
    const nowMs = Date.now();
    const currentBlockMs = nowMs;
    const approxTs = (blockNum) => Math.floor((currentBlockMs - (latestBlock - blockNum) * 2000) / 1000);

    let indexed = 0;
    for (const log of logs) {
      const amountUsdc = parseFloat(formatUnits(log.args.value, 6));
      insertPayment.run(
        log.transactionHash,
        Number(log.blockNumber),
        approxTs(Number(log.blockNumber)),
        log.args.from.toLowerCase(),
        log.args.value.toString(),
        amountUsdc,
        log.logIndex,
      );
      indexed++;
    }

    await setLastIndexedBlock(toBlock);
    return { indexed, fromBlock, toBlock, latestBlock };
  } catch (e) {
    const msg = e.message || String(e);
    if (!msg.includes('rate limit')) {
      console.error('[revenue] indexer error:', msg.substring(0, 200));
    }
    return { error: msg.substring(0, 200) };
  }
}

// ============ DASHBOARD DATA ============

export function getRevenueStats() {
  const now = Math.floor(Date.now() / 1000);
  const h24 = now - 86400;
  const d7 = now - 7 * 86400;
  const d30 = now - 30 * 86400;

  const q = (sql, ...params) => db.prepare(sql).get(...params);
  const qAll = (sql, ...params) => db.prepare(sql).all(...params);

  // Only verified x402 payments (tx_origin == facilitator)
  const X402 = 'is_x402 = 1';
  const lastBlockRow = getState.get('last_indexed_block');
  const totalUnenriched = q('SELECT COUNT(*) c FROM payments WHERE tx_origin IS NULL AND amount_usdc < 10').c;
  const totalEnriched = q('SELECT COUNT(*) c FROM payments WHERE tx_origin IS NOT NULL').c;

  return {
    generatedAt: new Date().toISOString(),
    note: 'x402 filter: tx_origin == facilitator. Background enricher verifies ~30 tx/min.',
    totals: {
      allTime: {
        paymentCount: q(`SELECT COUNT(*) AS c FROM payments WHERE ${X402}`).c,
        totalUsdc: q(`SELECT COALESCE(SUM(amount_usdc), 0) AS s FROM payments WHERE ${X402}`).s,
        uniquePayers: q(`SELECT COUNT(DISTINCT from_address) AS c FROM payments WHERE ${X402}`).c,
      },
      last30Days: {
        paymentCount: q(`SELECT COUNT(*) AS c FROM payments WHERE ${X402} AND block_timestamp > ?`, d30).c,
        totalUsdc: q(`SELECT COALESCE(SUM(amount_usdc), 0) AS s FROM payments WHERE ${X402} AND block_timestamp > ?`, d30).s,
      },
      last7Days: {
        paymentCount: q(`SELECT COUNT(*) AS c FROM payments WHERE ${X402} AND block_timestamp > ?`, d7).c,
        totalUsdc: q(`SELECT COALESCE(SUM(amount_usdc), 0) AS s FROM payments WHERE ${X402} AND block_timestamp > ?`, d7).s,
      },
      last24Hours: {
        paymentCount: q(`SELECT COUNT(*) AS c FROM payments WHERE ${X402} AND block_timestamp > ?`, h24).c,
        totalUsdc: q(`SELECT COALESCE(SUM(amount_usdc), 0) AS s FROM payments WHERE ${X402} AND block_timestamp > ?`, h24).s,
      },
    },
    topPayers: qAll(`
      SELECT from_address, COUNT(*) AS payment_count, SUM(amount_usdc) AS total_usdc
      FROM payments WHERE ${X402} GROUP BY from_address ORDER BY total_usdc DESC LIMIT 10
    `),
    recentPayments: qAll(`
      SELECT tx_hash, block_number, block_timestamp, from_address, amount_usdc
      FROM payments WHERE ${X402} ORDER BY block_number DESC LIMIT 20
    `),
    dailyRevenue_last30d: qAll(`
      SELECT
        date(block_timestamp, 'unixepoch') AS day,
        COUNT(*) AS payments,
        SUM(amount_usdc) AS usdc
      FROM payments
      WHERE ${X402} AND block_timestamp > ?
      GROUP BY day
      ORDER BY day DESC
    `, d30),
    indexerState: {
      lastIndexedBlock: lastBlockRow ? Number(lastBlockRow.value) : null,
      totalScannedPayments: q('SELECT COUNT(*) AS c FROM payments').c,
      enrichmentProgress: {
        enriched: totalEnriched,
        pending: totalUnenriched,
      },
    },
  };
}

// ============ LAZY ENRICHER ============
// Fetches tx_origin for small payments (< 10 USDC) in the background,
// 5 per run with delays, so we eventually verify which are real x402.

export async function enrichPayments(batchSize = 5) {
  try {
    const rows = getUnenrichedSmallPayments.all(batchSize);
    let enriched = 0;
    let x402Found = 0;
    for (const row of rows) {
      try {
        const tx = await client.getTransaction({ hash: row.tx_hash });
        const origin = tx.from.toLowerCase();
        const isX402 = origin === FACILITATOR ? 1 : 0;
        updatePaymentOrigin.run(origin, isX402, row.id);
        enriched++;
        if (isX402) x402Found++;
        await new Promise(r => setTimeout(r, 200)); // rate-limit
      } catch (e) {
        // mark as enriched with unknown origin to avoid re-fetching
        updatePaymentOrigin.run('error', 0, row.id);
      }
    }
    if (x402Found > 0) console.log(`[revenue] enricher: ${x402Found} new x402 payments verified (batch: ${enriched})`);
    return { enriched, x402Found };
  } catch (e) {
    return { error: e.message };
  }
}

// ============ AUTO-INDEX + ENRICH ============

let indexerInterval = null;
let enricherInterval = null;

export function startIndexer(intervalMs = 60_000) {
  if (indexerInterval) return;
  // Initial run
  indexPayments().then(r => {
    if (r.indexed > 0) console.log(`[revenue] initial scan: ${r.indexed} payments, blocks ${r.fromBlock}-${r.toBlock}`);
  }).catch(() => {});
  indexerInterval = setInterval(() => {
    indexPayments().then(r => {
      if (r.indexed > 0) console.log(`[revenue] ${r.indexed} new payments (${r.fromBlock}-${r.toBlock})`);
    }).catch(() => {});
  }, intervalMs);

  // Enricher runs more frequently (every 10s, 5 tx per run = 30/min)
  enricherInterval = setInterval(() => {
    enrichPayments(5).catch(() => {});
  }, 10_000);
}

export function stopIndexer() {
  if (indexerInterval) { clearInterval(indexerInterval); indexerInterval = null; }
  if (enricherInterval) { clearInterval(enricherInterval); enricherInterval = null; }
}