build-cran-binaries/stats/collect.ts

/**
 * Pull the CDN access logs into `download_stats.events`.
 *
 *   bun stats/collect.ts               # fetch the last 70 hours, insert what is new
 *   bun stats/collect.ts --reclassify  # recompute derived columns after a rule change
 *
 * Bunny keeps logs for three days, so every run re-reads almost all of that
 * window. Events are keyed by Bunny's request id, which makes re-reading free
 * of double counts, and a run that fails is simply covered by the next one.
 *
 * Environment: BUNNYNET_API_KEY, PGPASS, STATS_CLIENT_SALT, and optionally
 * STATS_PULL_ZONE (default: cran-rpkgs) and STATS_OWN_NETWORKS (comma-separated
 * CIDRs of our own CI runners, kept out of the public numbers).
 */
import { classify, parseNetworks, toEvent, type Event, type LogEntry } from './classify';
import { connect, migrate } from './db';

const PULL_ZONE = Number(process.env.STATS_PULL_ZONE ?? 3857050);
const WINDOW_HOURS = 70;
const PAGE_SIZE = 10000;

const COLUMNS = [
  'request_id',
  'ts',
  'zone_id',
  'host',
  'status',
  'cache_status',
  'bytes',
  'path',
  'user_agent',
  'country',
  'asn',
  'asn_org',
  'client',
  'own',
  'ruleset',
  'kind',
  'outcome',
  'redirect_reason',
  'slot',
  'arch',
  'os',
  'r_minor',
  'r_version',
  'package',
  'version',
  'tool',
  'network',
] as const;

function requireEnv(name: string): string {
  const value = process.env[name];
  if (!value) throw new Error(`${name} is not set`);
  return value;
}

async function fetchPage(
  apiKey: string,
  from: Date,
  to: Date,
  offset: number,
): Promise<{ data: LogEntry[]; hasMore: boolean }> {
  const url = new URL(`https://logging.bunnycdn.com/v2/pullzones/${PULL_ZONE}/logs`);
  url.searchParams.set('from', from.toISOString());
  url.searchParams.set('to', to.toISOString());
  url.searchParams.set('order', 'asc');
  url.searchParams.set('limit', String(PAGE_SIZE));
  url.searchParams.set('offset', String(offset));

  for (let attempt = 1; ; attempt++) {
    const response = await fetch(url, { headers: { AccessKey: apiKey, Accept: 'application/json' } });
    if (response.ok) {
      const body = (await response.json()) as { data: LogEntry[]; pagination: { hasMore: boolean } };
      return { data: body.data, hasMore: body.pagination.hasMore };
    }
    if ((response.status === 429 || response.status >= 500) && attempt < 6) {
      await Bun.sleep(2000 * attempt);
      continue;
    }
    throw new Error(`Bunny logging API returned ${response.status}: ${await response.text()}`);
  }
}

async function collect(): Promise<void> {
  const apiKey = requireEnv('BUNNYNET_API_KEY');
  const salt = requireEnv('STATS_CLIENT_SALT');
  const ownNetworks = parseNetworks(process.env.STATS_OWN_NETWORKS ?? '');
  const hashClient = (key: string) => new Bun.CryptoHasher('sha256', salt).update(key).digest('hex').slice(0, 20);

  // Fixing `to` up front keeps offset pagination stable while new requests
  // keep arriving.
  const to = new Date();
  const from = new Date(to.getTime() - WINDOW_HOURS * 3600 * 1000);

  const events: Event[] = [];
  let fetched = 0;
  for (let offset = 0; ;) {
    const page = await fetchPage(apiKey, from, to, offset);
    fetched += page.data.length;
    for (const entry of page.data) {
      const event = toEvent(entry, hashClient, ownNetworks);
      if (event) events.push(event);
    }
    if (!page.hasMore || page.data.length === 0) break;
    offset += page.data.length;
  }

  const sql = connect();
  await migrate(sql);
  let inserted = 0;
  for (let i = 0; i < events.length; i += 1000) {
    const batch = events.slice(i, i + 1000);
    const rows = await sql`
      insert into download_stats.events ${sql(batch, ...COLUMNS)}
      on conflict (request_id) do nothing
      returning 1`;
    inserted += rows.length;
  }
  await sql`
    insert into download_stats.runs (zone_id, window_from, window_to, fetched, inserted)
    values (${PULL_ZONE}, ${from}, ${to}, ${fetched}, ${inserted})`;
  await sql.close();

  console.log(
    `fetched ${fetched} log entries from ${from.toISOString()} to ${to.toISOString()}, inserted ${inserted} new`,
  );
}

async function reclassify(): Promise<void> {
  const sql = connect();
  await migrate(sql);
  const rows: Event[] =
    await sql`select request_id, path, status, user_agent, asn, asn_org, own from download_stats.events`;
  let updated = 0;
  for (let i = 0; i < rows.length; i += 1000) {
    const batch = rows.slice(i, i + 1000).map((row) => ({ request_id: row.request_id, ...classify(row) }));
    await sql.begin(async (tx) => {
      for (const row of batch) {
        await tx`
          update download_stats.events set
            ruleset = ${row.ruleset}, kind = ${row.kind}, outcome = ${row.outcome},
            redirect_reason = ${row.redirect_reason}, slot = ${row.slot}, arch = ${row.arch}, os = ${row.os},
            r_minor = ${row.r_minor}, r_version = ${row.r_version}, package = ${row.package},
            version = ${row.version}, tool = ${row.tool}, network = ${row.network}
          where request_id = ${row.request_id}`;
      }
    });
    updated += batch.length;
  }
  await sql.close();
  console.log(`reclassified ${updated} events`);
}

if (import.meta.main) {
  await (process.argv.includes('--reclassify') ? reclassify() : collect());
}