`lib/sessionize.js`

78 lines
'use strict';

// Sessionization: given an ordered stream of authenticated /826/ hits, group them
// into sessions per (client, remote_user, ip). A new session starts when the gap
// since the last hit exceeds SESSION_GAP. Duration = last_seen - started (dwell on
// the final page of a session is unknowable from logs alone — accepted approximation).

// Decide whether a parsed hit is in-scope for a client's time-on-server tracking.
function isTrackable(hit, clientSlug) {
  if (!hit || !hit.path) return false;
  // Scope to the /826/ area only.
  if (!(hit.path === '/826' || hit.path.startsWith('/826/'))) return false;
  // Only authenticated traffic counts as real client time (skips 401 pre-auth noise + bots).
  if (!hit.remote_user) return false;
  return true;
}

async function openSession(pool, clientSlug, h) {
  const { rows } = await pool.query(
    `INSERT INTO sessions (client_slug, remote_user, ip, user_agent, started_at, last_seen_at, hit_count, duration_seconds, is_open)
     VALUES ($1,$2,$3,$4,$5,$5,1,0,true)
     RETURNING id`,
    [clientSlug, h.remote_user, h.ip, h.ua, new Date(h.ts)]
  );
  return rows[0].id;
}

// Ingest one trackable hit. gapMs = inactivity threshold. Processes hits in time order.
async function ingestHit(pool, clientSlug, gapMs, h) {
  const { rows } = await pool.query(
    `SELECT id, started_at, last_seen_at FROM sessions
     WHERE client_slug=$1 AND is_open
       AND coalesce(remote_user,'')=coalesce($2,'') AND ip=$3
     ORDER BY last_seen_at DESC LIMIT 1`,
    [clientSlug, h.remote_user, h.ip]
  );

  let sessionId;
  if (rows.length) {
    const s = rows[0];
    const lastSeen = new Date(s.last_seen_at).getTime();
    const started = new Date(s.started_at).getTime();
    const withinGap = h.ts - lastSeen <= gapMs && h.ts >= lastSeen;
    if (withinGap) {
      const dur = Math.max(0, Math.round((h.ts - started) / 1000));
      await pool.query(
        `UPDATE sessions SET last_seen_at=$1, hit_count=hit_count+1, duration_seconds=$2 WHERE id=$3`,
        [new Date(h.ts), dur, s.id]
      );
      sessionId = s.id;
    } else {
      await pool.query(`UPDATE sessions SET is_open=false WHERE id=$1`, [s.id]);
      sessionId = await openSession(pool, clientSlug, h);
    }
  } else {
    sessionId = await openSession(pool, clientSlug, h);
  }

  await pool.query(
    `INSERT INTO page_hits (session_id, ts, method, path, status, bytes, referer, user_agent)
     VALUES ($1,$2,$3,$4,$5,$6,$7,$8)`,
    [sessionId, new Date(h.ts), h.method, h.path, h.status, h.bytes, h.referer, h.ua]
  );
  return sessionId;
}

// Mark sessions with no activity for > gapMs as closed (housekeeping for live mode).
async function closeStale(pool, clientSlug, gapMs) {
  await pool.query(
    `UPDATE sessions SET is_open=false
     WHERE client_slug=$1 AND is_open
       AND last_seen_at < now() - ($2::int * interval '1 millisecond')`,
    [clientSlug, gapMs]
  );
}

module.exports = { isTrackable, ingestHit, closeStale };