CREATE TABLE participant_events (
  id bigserial PRIMARY KEY,
  user_id text NOT NULL,
  room_id text NOT NULL,
  event_type text NOT NULL,   -- 'join' or 'leave'
  event_ts timestamptz NOT NULL DEFAULT now(),
  livekit_participant_sid text
);

CREATE TABLE reconciled_minutes (
  id bigserial PRIMARY KEY,
  user_id text NOT NULL,
  period_start date NOT NULL,  -- day-start or week-start
  period_type text NOT NULL,   -- 'day' or 'week'
  minutes integer NOT NULL,
  source text NOT NULL,        -- 'app' | 'metrics_export'
  reconciled_at timestamptz NOT NULL DEFAULT now(),
  UNIQUE(user_id, period_start, period_type, source)
);

// POST /participant-event
// body: { user_id, room_id, event_type, livekit_participant_sid, event_ts? }
app.post('/participant-event', async (req, res) => {
  const { user_id, room_id, event_type, livekit_participant_sid, event_ts } = req.body;
  const ts = event_ts ? new Date(event_ts) : new Date();
  await db.query(
    `INSERT INTO participant_events (user_id, room_id, event_type, event_ts, livekit_participant_sid)
     VALUES ($1, $2, $3, $4, $5)`,
    [user_id, room_id, event_type, ts.toISOString(), livekit_participant_sid || null]
  );
  res.sendStatus(201);
});

async function processAppEventsForPeriod(periodStartDate) {
  // Example: periodStartDate = '2026-03-26' (UTC day start)
  const periodEnd = new Date(periodStartDate);
  periodEnd.setUTCDate(periodEnd.getUTCDate() + 1);

  // Get all users with events overlapping the period
  const { rows } = await db.query(
    `SELECT user_id, room_id, event_type, event_ts
     FROM participant_events
     WHERE event_ts >= $1::timestamptz - INTERVAL '1 day' -- safety margin
       AND event_ts < $2::timestamptz + INTERVAL '1 day'
     ORDER BY user_id, event_ts`,
    [periodStartDate, periodEnd.toISOString()]
  );

  // Pair joins/leaves per user -> build sessions
  const sessionsByUser = {};
  for (const r of rows) {
    const u = r.user_id;
    sessionsByUser[u] = sessionsByUser[u] || { stack: [], sessions: [] };
    const entry = sessionsByUser[u];

    if (r.event_type === 'join') {
      entry.stack.push(new Date(r.event_ts));
    } else if (r.event_type === 'leave') {
      // match last unmatched join
      const joinTs = entry.stack.pop() || null;
      if (joinTs) {
        entry.sessions.push({ start: joinTs, end: new Date(r.event_ts) });
      } else {
        // leave without join: ignore or create short session based on policy
      }
    }
  }

  // Close any open sessions at period end (policy dependent)
  for (const [user_id, data] of Object.entries(sessionsByUser)) {
    for (const openJoin of data.stack) {
      data.sessions.push({ start: openJoin, end: periodEnd }); // close at period end
    }

    // Calculate minutes overlapping the exact periodStartDate..periodEnd
    let totalSeconds = 0;
    const start = new Date(periodStartDate);
    const end = periodEnd;
    for (const s of data.sessions) {
      const overlapStart = s.start > start ? s.start : start;
      const overlapEnd = s.end < end ? s.end : end;
      if (overlapEnd > overlapStart) totalSeconds += (overlapEnd - overlapStart) / 1000;
    }
    const minutes = Math.ceil(totalSeconds / 60); // rounding choice

    // Upsert into reconciled_minutes
    await db.query(
      `INSERT INTO reconciled_minutes (user_id, period_start, period_type, minutes, source)
       VALUES ($1,$2,'day',$3,'app')
       ON CONFLICT (user_id, period_start, period_type, source)
       DO UPDATE SET minutes = EXCLUDED.minutes, reconciled_at = now()`,
      [user_id, periodStartDate, minutes]
    );
  }
}

async function ingestMetricsExport(startIso, endIso) {
  let nextPageToken = null;
  const perUserSeconds = new Map();

  do {
    const url = new URL('https://api.livekit.example/metrics/export'); // replace with actual endpoint
    url.searchParams.set('start', startIso);
    url.searchParams.set('end', endIso);
    if (nextPageToken) url.searchParams.set('pageToken', nextPageToken);

    const res = await fetch(url.toString(), {
      headers: { Authorization: `Bearer ${process.env.METRICS_API_KEY}` }
    });
    const payload = await res.json();
    // payload.items[]: each item should include participant id/user id and connected_seconds
    for (const item of payload.items || []) {
      const userId = item.user_id || item.participant_user_id || item.participant_sid;
      const seconds = item.connected_seconds || 0;
      perUserSeconds.set(userId, (perUserSeconds.get(userId) || 0) + seconds);
    }
    nextPageToken = payload.nextPageToken;
  } while (nextPageToken);

  // Upsert per user minutes
  for (const [userId, seconds] of perUserSeconds.entries()) {
    const minutes = Math.ceil(seconds / 60);
    await db.query(
      `INSERT INTO reconciled_minutes (user_id, period_start, period_type, minutes, source)
       VALUES ($1,$2,'day',$3,'metrics_export')
       ON CONFLICT (user_id, period_start, period_type, source)
       DO UPDATE SET minutes = EXCLUDED.minutes, reconciled_at = now()`,
      [userId, startIso.slice(0,10), minutes]
    );
  }
}

SELECT a.user_id,
  a.minutes as app_minutes,
  m.minutes as metrics_minutes,
  (a.minutes - m.minutes) AS delta
FROM reconciled_minutes a
LEFT JOIN reconciled_minutes m
  ON a.user_id = m.user_id
  AND a.period_start = m.period_start
  AND a.period_type = m.period_type
  AND m.source = 'metrics_export'
WHERE a.source = 'app' AND a.period_start = $1;