`tracker.js`

166 lines
'use strict';

// rentv-826-tracker — sessionizes the nginx access log for the /826/ client area
// into the per-client Postgres DB (rentv_826). Server-side only; no client-page changes.
//
//   node tracker.js                 -> one-time backfill (if needed) then live-tail LIVE_LOG
//   node tracker.js --backfill-only -> just backfill BACKFILL_LOG, then exit
//
// Env (see .env): DATABASE_URL, LIVE_LOG, BACKFILL_LOG, CLIENT_SLUG, SESSION_GAP_MIN

const fs = require('fs');
const path = require('path');
const readline = require('readline');
const { Pool } = require('pg');
const { parseLine } = require('./lib/parse');
const { isTrackable, ingestHit, closeStale } = require('./lib/sessionize');
const { writeDashboard } = require('./lib/dashboard');

// --- config -------------------------------------------------------------
function loadEnv() {
  const envPath = path.join(__dirname, '.env');
  if (fs.existsSync(envPath)) {
    for (const raw of fs.readFileSync(envPath, 'utf8').split('\n')) {
      const line = raw.trim();
      if (!line || line.startsWith('#')) continue;
      const i = line.indexOf('=');
      if (i === -1) continue;
      const k = line.slice(0, i).trim();
      if (!(k in process.env)) process.env[k] = line.slice(i + 1).trim();
    }
  }
}
loadEnv();

const CLIENT_SLUG = process.env.CLIENT_SLUG || 'boomer';
const GAP_MS = (parseInt(process.env.SESSION_GAP_MIN, 10) || 30) * 60 * 1000;
const LIVE_LOG = process.env.LIVE_LOG || '/var/log/nginx/rentv-826.access.log';
const BACKFILL_LOG = process.env.BACKFILL_LOG || '/var/log/nginx/rentv.access.log';
const POLL_MS = 3000;
const STATS_HTML = process.env.STATS_HTML || ''; // e.g. /var/www/rentv-826/stats.html; empty = disabled
const backfillOnly = process.argv.includes('--backfill-only');

async function refreshDashboard() {
  if (!STATS_HTML) return;
  try { await writeDashboard(pool, CLIENT_SLUG, STATS_HTML, fs); }
  catch (e) { log('dashboard write error:', e.message); }
}

const pool = new Pool({ connectionString: process.env.DATABASE_URL });

function log(...a) { console.log(new Date().toISOString(), ...a); }

// --- ingest-state helpers ----------------------------------------------
async function getState(file) {
  const { rows } = await pool.query(`SELECT inode, byte_offset FROM ingest_state WHERE log_file=$1`, [file]);
  return rows[0] || null;
}
async function setState(file, inode, offset) {
  await pool.query(
    `INSERT INTO ingest_state (log_file, inode, byte_offset, updated_at)
     VALUES ($1,$2,$3, now())
     ON CONFLICT (log_file) DO UPDATE SET inode=EXCLUDED.inode, byte_offset=EXCLUDED.byte_offset, updated_at=now()`,
    [file, inode, offset]
  );
}

// Read [start,end) of a file, process complete lines, return the offset up to the last newline.
async function processRange(file, start, end) {
  if (end <= start) return start;
  let consumed = start;
  const stream = fs.createReadStream(file, { start, end: end - 1, encoding: 'utf8' });
  let buf = '';
  for await (const chunk of stream) {
    buf += chunk;
    let nl;
    while ((nl = buf.indexOf('\n')) !== -1) {
      const line = buf.slice(0, nl);
      buf = buf.slice(nl + 1);
      consumed += Buffer.byteLength(line, 'utf8') + 1; // +1 for the '\n'
      const hit = parseLine(line);
      if (hit && isTrackable(hit, CLIENT_SLUG)) {
        try { await ingestHit(pool, CLIENT_SLUG, GAP_MS, hit); }
        catch (e) { log('ingest error:', e.message); }
      }
    }
  }
  // Any trailing partial line (no newline) is left for the next poll.
  return consumed;
}

// --- backfill -----------------------------------------------------------
async function backfill() {
  // Single-source mode: if the backfill log IS the live log, the offset-tracked live
  // tail reads it from byte 0 on first run — that initial pass IS the backfill. Running
  // a separate backfill here would double-count. So skip it.
  if (!BACKFILL_LOG || BACKFILL_LOG === LIVE_LOG) {
    log('single-source mode: first live-tail pass from offset 0 serves as backfill');
    return;
  }
  const marker = `BACKFILL:${BACKFILL_LOG}`;
  const done = await getState(marker);
  if (done) { log('backfill already done for', BACKFILL_LOG); return; }
  if (!fs.existsSync(BACKFILL_LOG)) { log('no backfill log at', BACKFILL_LOG); await setState(marker, 0, 0); return; }

  log('backfilling /826/ history from', BACKFILL_LOG, '...');
  const rl = readline.createInterface({ input: fs.createReadStream(BACKFILL_LOG, { encoding: 'utf8' }), crlfDelay: Infinity });
  let seen = 0, tracked = 0;
  for await (const line of rl) {
    seen++;
    const hit = parseLine(line);
    if (hit && isTrackable(hit, CLIENT_SLUG)) {
      tracked++;
      try { await ingestHit(pool, CLIENT_SLUG, GAP_MS, hit); }
      catch (e) { log('backfill ingest error:', e.message); }
    }
  }
  const sz = fs.statSync(BACKFILL_LOG).size;
  await setState(marker, 0, sz);
  log(`backfill complete: ${seen} lines scanned, ${tracked} /826/ authed hits ingested.`);
}

// --- live tail ----------------------------------------------------------
async function tailOnce() {
  let st;
  try { st = fs.statSync(LIVE_LOG); }
  catch { return; } // file not created yet (no /826/ traffic since nginx split)
  const state = await getState(LIVE_LOG);
  let start = 0;
  if (state) {
    if (Number(state.inode) === st.ino) start = Number(state.byte_offset);
    else start = 0; // rotation: inode changed, read from top of the new file
  }
  if (st.size < start) start = 0; // truncated
  if (st.size === start) return;
  const newOffset = await processRange(LIVE_LOG, start, st.size);
  await setState(LIVE_LOG, st.ino, newOffset);
}

async function liveLoop() {
  log('live-tailing', LIVE_LOG, `(client=${CLIENT_SLUG}, gap=${GAP_MS / 60000}min${STATS_HTML ? ', dashboard=' + STATS_HTML : ''})`);
  await refreshDashboard();
  let tick = 0;
  for (;;) {
    try {
      await tailOnce();
      if (++tick % 20 === 0) { await closeStale(pool, CLIENT_SLUG, GAP_MS); await refreshDashboard(); } // ~every 60s
    } catch (e) { log('tail error:', e.message); }
    await new Promise((r) => setTimeout(r, POLL_MS));
  }
}

// --- main ---------------------------------------------------------------
(async () => {
  try {
    await backfill();
    if (backfillOnly) { await pool.end(); return; }
    await liveLoop();
  } catch (e) {
    log('fatal:', e.message);
    process.exit(1);
  }
})();

process.on('SIGTERM', () => { log('SIGTERM, exiting'); pool.end().finally(() => process.exit(0)); });
process.on('SIGINT', () => { log('SIGINT, exiting'); pool.end().finally(() => process.exit(0)); });