Build a Live Visitor Feed

const API = 'https://app.rybbit.io';
const SITE = '123';
const API_KEY = process.env.RYBBIT_API_KEY;
const POLL_MS = 10000; // every 10 seconds

async function get(path, params = {}) {
  const url = new URL(`${API}/api/sites/${SITE}${path}`);
  Object.entries(params).forEach(([k, v]) => url.searchParams.set(k, v));
  const res = await fetch(url, { headers: { Authorization: `Bearer ${API_KEY}` } });
  if (!res.ok) throw new Error(`${path} failed: ${res.status}`);
  return res.json();
}

async function liveCount() {
  const { count } = await get('/live-user-count', { minutes: 5 });
  return count;
}

function describe(e) {
  const where = e.city ? `${e.city}, ${e.country}` : e.country || 'Unknown';
  if (e.type === 'pageview')     return `👀 viewed ${e.pathname} — ${where}`;
  if (e.type === 'custom_event') return `⚡ ${e.event_name} (${e.pathname}) — ${where}`;
  if (e.type === 'outbound')     return `↗️  left via ${e.pathname} — ${where}`;
  return `• ${e.type} — ${where}`;
}

let cursor = null;     // where the next poll starts
const seen = new Set(); // event_cursor values already shown (polls overlap by a few seconds)
const SEEN_MAX = 5000;

// Timestamps are UTC, formatted "YYYY-MM-DD HH:MM:SS" with no zone marker
const toDate = (ts) => new Date(ts.replace(' ', 'T') + 'Z');

async function poll() {
  try {
    const polling = cursor !== null; // false for the initial load
    const params = polling ? { since_cursor: cursor } : { page_size: 20 };
    const [res, count] = await Promise.all([get('/events', params), liveCount()]);

    // Events come back newest-first; show oldest-first so the feed reads top-to-bottom
    for (const e of [...res.data].reverse()) {
      if (seen.has(e.event_cursor)) continue;
      seen.add(e.event_cursor);
      console.log(`[${toDate(e.timestamp).toLocaleTimeString()}] ${describe(e)}`);
    }
    // Keep memory bounded by forgetting the oldest cursors; the recent ones cover the overlap
    for (const old of seen) {
      if (seen.size <= SEEN_MAX) break;
      seen.delete(old);
    }

    if (polling) {
      cursor = res.cursor.nextCursor;
    } else if (res.cursor.newestCursor) {
      // Initial load: start 5 seconds before the newest event, the same overlap
      // later polls get (newestCursor is null on a site with no events yet)
      const [ms] = res.cursor.newestCursor.split('_');
      cursor = `${Number(ms) - 5000}_0000000000000000`;
    }

    process.stdout.write(`\n🟢 ${count} visitors online\n`);
    if (polling && res.cursor.hasMore) return poll(); // backlog: read the next 500 now
  } catch (err) {
    console.error('poll error:', err.message);
  }
}

console.log('Starting live feed… (Ctrl+C to stop)');
poll();
setInterval(poll, POLL_MS);

import os, time
from datetime import datetime, timezone
import requests

API = "https://app.rybbit.io"
SITE = "123"
API_KEY = os.environ["RYBBIT_API_KEY"]
POLL_SECONDS = 10

session = requests.Session()
session.headers["Authorization"] = f"Bearer {API_KEY}"


def get(path, **params):
    res = session.get(f"{API}/api/sites/{SITE}{path}", params=params)
    res.raise_for_status()
    return res.json()


def live_count():
    return get("/live-user-count", minutes=5)["count"]


def describe(e):
    where = f'{e["city"]}, {e["country"]}' if e.get("city") else e.get("country") or "Unknown"
    if e["type"] == "pageview":
        return f'👀 viewed {e["pathname"]} — {where}'
    if e["type"] == "custom_event":
        return f'⚡ {e["event_name"]} ({e["pathname"]}) — {where}'
    if e["type"] == "outbound":
        return f'↗️  left via {e["pathname"]} — {where}'
    return f'• {e["type"]} — {where}'


cursor = None  # where the next poll starts
seen = {}     # event_cursor values already shown, oldest first (polls overlap by a few seconds)
SEEN_MAX = 5000

print("Starting live feed… (Ctrl+C to stop)")
while True:
    has_more = False
    try:
        res = get("/events", **({"since_cursor": cursor} if cursor else {"page_size": 20}))

        # Events come back newest-first; show oldest-first so the feed reads top-to-bottom
        for e in reversed(res["data"]):
            if e["event_cursor"] in seen:
                continue
            seen[e["event_cursor"]] = True
            # Timestamps are UTC, formatted "YYYY-MM-DD HH:MM:SS" with no zone marker
            ts = datetime.strptime(e["timestamp"][:19], "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc)
            print(f"[{ts.astimezone().strftime('%H:%M:%S')}] {describe(e)}")
        # Keep memory bounded by forgetting the oldest cursors; the recent ones cover the overlap
        while len(seen) > SEEN_MAX:
            del seen[next(iter(seen))]

        if cursor:
            cursor = res["cursor"]["nextCursor"]
            has_more = res["cursor"]["hasMore"]
        elif res["cursor"]["newestCursor"]:
            # Initial load: start 5 seconds before the newest event, the same overlap
            # later polls get (newestCursor is None on a site with no events yet)
            ms = int(res["cursor"]["newestCursor"].split("_")[0])
            cursor = f"{ms - 5000}_0000000000000000"

        print(f"\n🟢 {live_count()} visitors online\n")
    except Exception as err:
        print(f"poll error: {err}")
    if not has_more:  # with a backlog, read the next 500 straight away
        time.sleep(POLL_SECONDS)

// Inside poll(), before the loop that prints new events and adds them to `seen`:
for (const e of res.data) {
  if (seen.has(e.event_cursor)) continue; // already handled in an earlier poll
  if (e.type === 'pageview' && e.pathname === '/enterprise') {
    await fetch(process.env.SLACK_WEBHOOK, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({
        text: `🏢 Enterprise page visit from ${e.city || e.country} (${e.referrer || 'direct'})`,
      }),
    });
  }
}