Feat : Streamers page revamp

packages/server/src/api.ts

@@ -13,7 +13,7 @@ import {
 } from '@stream-control/shared';
 import { config } from './config.ts';
 import { generateToken, hashToken, issueSession, requireSession, safeEqual } from './auth.ts';
-import { agentsRepo, logsRepo, targetsRepo } from './db.ts';
+import { agentsRepo, logsRepo, sessionsRepo, spansRepo, targetsRepo } from './db.ts';
 import { hub } from './hub.ts';
 import { getPushoverSettings, savePushoverSettings, sendPushover } from './pushover.ts';
 import { startTargetRecording } from './recorder.ts';
@@ -286,6 +286,7 @@ api.patch('/watchlist/:id', (req, res) => {
     agentId,
     notify: typeof req.body?.notify === 'boolean' ? req.body.notify : target.notify,
     autoRecord,
+    favorite: typeof req.body?.favorite === 'boolean' ? req.body.favorite : target.favorite,
   });
 
   const updated = targetsRepo.get(target.id);
@@ -372,6 +373,24 @@ api.post('/watchlist/:id/stop', async (req, res) => {
   }
 });
 
+/**
+ * Frise de diffusion : les streams observés et ce qui en a été capturé.
+ *
+ * Une seule route pour la frise globale et celle d'un profil — c'est le même
+ * assemblage, seul le filtre change. Le client recoupe les deux listes par
+ * `targetId`, il a déjà les profils par le flux temps réel.
+ */
+api.get('/timeline', (req, res) => {
+  const days = Math.min(Math.max(Number.parseInt(String(req.query.days ?? '7'), 10) || 7, 1), 120);
+  const targetId = typeof req.query.targetId === 'string' ? req.query.targetId : null;
+  const from = Date.now() - days * 86_400_000;
+
+  res.json({
+    sessions: sessionsRepo.since(from, targetId),
+    spans: spansRepo.since(from, targetId),
+  });
+});
+
 // --- Divers -----------------------------------------------------------------
 
 /** Historique d'une VM : les mêmes entrées que le journal, filtrées et bornées. */

packages/server/src/db.ts

@@ -8,6 +8,8 @@ import type {
   ObsSettings,
   Platform,
   RecordingSettings,
+  RecordingSpan,
+  StreamSession,
   StreamState,
   BrowserSettings,
   WatchSettings,
@@ -81,6 +83,35 @@ db.exec(`
   CREATE UNIQUE INDEX IF NOT EXISTS idx_targets_identity
     ON watch_targets (provider, username);
 
+  -- Diffusions observées, une ligne par passage en direct. Le CASCADE est
+  -- voulu : ne plus suivre un profil, c'est aussi oublier son historique.
+  CREATE TABLE IF NOT EXISTS stream_sessions (
+    id         INTEGER PRIMARY KEY AUTOINCREMENT,
+    target_id  TEXT NOT NULL REFERENCES watch_targets (id) ON DELETE CASCADE,
+    started_at INTEGER NOT NULL,
+    ended_at   INTEGER
+  );
+
+  CREATE INDEX IF NOT EXISTS idx_sessions_target ON stream_sessions (target_id, started_at DESC);
+  CREATE INDEX IF NOT EXISTS idx_sessions_started ON stream_sessions (started_at DESC);
+
+  -- Intervalles réellement capturés par une VM. La colonne last_seen_at est
+  -- réécrite à chaque statut reçu pendant la capture : c'est elle qui permet de
+  -- refermer honnêtement un intervalle dont on n'a jamais vu la fin (agent
+  -- disparu, serveur redémarré) sans lui attribuer la durée de l'interruption.
+  CREATE TABLE IF NOT EXISTS recording_spans (
+    id           INTEGER PRIMARY KEY AUTOINCREMENT,
+    target_id    TEXT REFERENCES watch_targets (id) ON DELETE CASCADE,
+    agent_id     TEXT,
+    started_at   INTEGER NOT NULL,
+    ended_at     INTEGER,
+    last_seen_at INTEGER NOT NULL
+  );
+
+  CREATE INDEX IF NOT EXISTS idx_spans_target ON recording_spans (target_id, started_at DESC);
+  CREATE INDEX IF NOT EXISTS idx_spans_started ON recording_spans (started_at DESC);
+  CREATE INDEX IF NOT EXISTS idx_spans_open ON recording_spans (agent_id) WHERE ended_at IS NULL;
+
   -- Réglages globaux, hors agents et profils : une ligne par clé, en JSON.
   -- Pensé pour grandir (un jour d'autres canaux que Pushover) sans nouvelle
   -- table à chaque fois.
@@ -119,6 +150,8 @@ addColumnIfMissing('watch_targets', 'last_live_ended_at', 'INTEGER');
 // Enregistrement automatique dès que le profil est en direct depuis assez
 // longtemps. Desactivé par défaut : rien ne doit s'enclencher sans un geste.
 addColumnIfMissing('watch_targets', 'auto_record', 'INTEGER NOT NULL DEFAULT 0');
+// Épinglage en tête de liste, sans autre effet.
+addColumnIfMissing('watch_targets', 'favorite', 'INTEGER NOT NULL DEFAULT 0');
 
 /**
  * `idle` — le « revient bientôt » de Stripchat — relevait de la clôture et non de
@@ -426,6 +459,7 @@ interface TargetRow {
   agent_id: string | null;
   notify: number;
   auto_record: number;
+  favorite: number;
   state: string;
   raw_status: string | null;
   state_since: number;
@@ -449,6 +483,7 @@ function toTarget(row: TargetRow): WatchTarget {
     url: stripchatProfileUrl(row.username),
     agentId: row.agent_id,
     notify: Number(row.notify) === 1,
+    favorite: Number(row.favorite) === 1,
     autoRecord: Number(row.auto_record) === 1,
     state: (row.state as StreamState) ?? 'unknown',
     rawStatus: row.raw_status,
@@ -473,7 +508,7 @@ const targetStmts = {
     VALUES (?, ?, ?, ?, ?, ?, 'unknown', ?, ?)
   `),
   updateSettings: db.prepare(
-    'UPDATE watch_targets SET label = ?, agent_id = ?, notify = ?, auto_record = ? WHERE id = ?',
+    'UPDATE watch_targets SET label = ?, agent_id = ?, notify = ?, auto_record = ?, favorite = ? WHERE id = ?',
   ),
   updateState: db.prepare(`
     UPDATE watch_targets
@@ -531,6 +566,7 @@ export const targetsRepo = {
       agentId: string | null;
       notify: boolean;
       autoRecord: boolean;
+      favorite: boolean;
     },
   ): void {
     targetStmts.updateSettings.run(
@@ -538,6 +574,7 @@ export const targetsRepo = {
       settings.agentId,
       settings.notify ? 1 : 0,
       settings.autoRecord ? 1 : 0,
+      settings.favorite ? 1 : 0,
       id,
     );
   },
@@ -574,6 +611,146 @@ export const targetsRepo = {
   },
 };
 
+// --- Historique de diffusion -------------------------------------------------
+
+const sessionStmts = {
+  openFor: db.prepare(
+    'SELECT * FROM stream_sessions WHERE target_id = ? AND ended_at IS NULL ORDER BY started_at DESC',
+  ),
+  insert: db.prepare('INSERT INTO stream_sessions (target_id, started_at) VALUES (?, ?)'),
+  // MAX() borne la fin au démarrage : l'heure de fin vient de la plateforme et
+  // rien ne garantit qu'elle soit postérieure à celle du début qu'elle nous
+  // avait donnée. Une durée négative rendrait la frise illisible.
+  close: db.prepare(`
+    UPDATE stream_sessions SET ended_at = MAX(started_at, ?)
+    WHERE target_id = ? AND ended_at IS NULL
+  `),
+  since: db.prepare(
+    'SELECT * FROM stream_sessions WHERE started_at >= ? ORDER BY started_at',
+  ),
+  sinceForTarget: db.prepare(
+    'SELECT * FROM stream_sessions WHERE target_id = ? AND started_at >= ? ORDER BY started_at',
+  ),
+};
+
+interface SessionRow {
+  id: number;
+  target_id: string;
+  started_at: number;
+  ended_at: number | null;
+}
+
+const toSession = (row: SessionRow): StreamSession => ({
+  id: Number(row.id),
+  targetId: row.target_id,
+  startedAt: Number(row.started_at),
+  endedAt: num(row.ended_at),
+});
+
+export const sessionsRepo = {
+  /**
+   * Ouvre une diffusion — sans effet s'il en reste déjà une ouverte.
+   *
+   * L'idempotence n'est pas un luxe : au redémarrage du serveur, un profil déjà
+   * en direct ne produit aucune transition, et la diffusion en cours doit être
+   * reprise telle quelle plutôt que coupée en deux.
+   */
+  open(targetId: string, startedAt: number): void {
+    const existing = sessionStmts.openFor.get(targetId);
+    if (existing) return;
+    sessionStmts.insert.run(targetId, startedAt);
+  },
+
+  close(targetId: string, endedAt: number): void {
+    sessionStmts.close.run(endedAt, targetId);
+  },
+
+  since(from: number, targetId?: string | null): StreamSession[] {
+    const rows = (
+      targetId
+        ? sessionStmts.sinceForTarget.all(targetId, from)
+        : sessionStmts.since.all(from)
+    ) as unknown as SessionRow[];
+    return rows.map(toSession);
+  },
+};
+
+interface SpanRow {
+  id: number;
+  target_id: string | null;
+  agent_id: string | null;
+  started_at: number;
+  ended_at: number | null;
+  last_seen_at: number;
+}
+
+const toSpan = (row: SpanRow): RecordingSpan => ({
+  id: Number(row.id),
+  targetId: row.target_id,
+  agentId: row.agent_id,
+  startedAt: Number(row.started_at),
+  endedAt: num(row.ended_at),
+});
+
+const spanStmts = {
+  openFor: db.prepare(
+    'SELECT * FROM recording_spans WHERE agent_id = ? AND ended_at IS NULL ORDER BY started_at DESC',
+  ),
+  insert: db.prepare(`
+    INSERT INTO recording_spans (target_id, agent_id, started_at, last_seen_at)
+    VALUES (?, ?, ?, ?)
+  `),
+  touch: db.prepare('UPDATE recording_spans SET last_seen_at = ? WHERE id = ?'),
+  // Le profil n'est pas toujours connu à l'ouverture : une capture lancée
+  // depuis l'onglet Agents précède parfois le calage de la surveillance.
+  attach: db.prepare('UPDATE recording_spans SET target_id = ? WHERE id = ? AND target_id IS NULL'),
+  close: db.prepare('UPDATE recording_spans SET ended_at = MAX(started_at, ?) WHERE id = ?'),
+  sweep: db.prepare(`
+    UPDATE recording_spans SET ended_at = last_seen_at
+    WHERE ended_at IS NULL AND last_seen_at < ?
+  `),
+  since: db.prepare('SELECT * FROM recording_spans WHERE started_at >= ? ORDER BY started_at'),
+  sinceForTarget: db.prepare(
+    'SELECT * FROM recording_spans WHERE target_id = ? AND started_at >= ? ORDER BY started_at',
+  ),
+};
+
+export const spansRepo = {
+  openForAgent(agentId: string): RecordingSpan | null {
+    const row = spanStmts.openFor.get(agentId) as unknown as SpanRow | undefined;
+    return row ? toSpan(row) : null;
+  },
+
+  open(agentId: string, targetId: string | null, at: number): void {
+    spanStmts.insert.run(targetId, agentId, at, at);
+  },
+
+  touch(id: number, at: number, targetId: string | null): void {
+    spanStmts.touch.run(at, id);
+    if (targetId) spanStmts.attach.run(targetId, id);
+  },
+
+  close(id: number, endedAt: number): void {
+    spanStmts.close.run(endedAt, id);
+  },
+
+  /**
+   * Referme les captures dont on n'a jamais vu la fin, à leur dernière preuve
+   * de vie plutôt qu'à maintenant : une VM disparue une semaine ne doit pas
+   * laisser croire à une semaine d'enregistrement.
+   */
+  sweepStale(deadline: number): void {
+    spanStmts.sweep.run(deadline);
+  },
+
+  since(from: number, targetId?: string | null): RecordingSpan[] {
+    const rows = (
+      targetId ? spanStmts.sinceForTarget.all(targetId, from) : spanStmts.since.all(from)
+    ) as unknown as SpanRow[];
+    return rows.map(toSpan);
+  },
+};
+
 interface LogRow {
   id: number;
   agent_id: string | null;

packages/server/src/hub.ts

@@ -13,7 +13,17 @@ import type {
 } from '@stream-control/shared';
 import { emptyStatus } from '@stream-control/shared';
 import { config } from './config.ts';
-import { agentsRepo, logsRepo, targetsRepo, type AgentRecord } from './db.ts';
+import { agentsRepo, logsRepo, spansRepo, targetsRepo, type AgentRecord } from './db.ts';
+
+/**
+ * Au-delà de ce silence, une capture en cours est considérée comme terminée.
+ *
+ * Généreux face aux deux secondes du cycle de statut : un agent qui redémarre
+ * pendant un enregistrement doit retrouver son intervalle plutôt que d'en
+ * ouvrir un second. Le délai ne coûte rien en justesse — l'intervalle est
+ * refermé à sa dernière preuve de vie, pas à l'heure du balayage.
+ */
+const SPAN_STALE_MS = 60_000;
 
 interface PendingCommand {
   resolve: (value: unknown) => void;
@@ -87,9 +97,44 @@ class Hub {
   updateStatus(agentId: string, status: AgentStatus): void {
     this.statuses.set(agentId, status);
     this.markSeen(agentId);
+    this.trackRecording(agentId, status);
     this.publishAgent(agentId);
   }
 
+  /**
+   * Tient à jour l'intervalle de capture en cours de cette VM.
+   *
+   * Une pause OBS (show privé) n'interrompt pas l'intervalle : elle n'écrit
+   * rien mais ne clôt pas le fichier, et le flux public s'est de toute façon
+   * arrêté pendant ce temps — la frise le montre déjà comme une coupure de
+   * diffusion. Découper ici laisserait croire à deux captures distinctes.
+   */
+  private trackRecording(agentId: string, status: AgentStatus): void {
+    const now = Date.now();
+    const open = spansRepo.openForAgent(agentId);
+
+    if (!status.recording) {
+      if (open) spansRepo.close(open.id, now);
+      return;
+    }
+
+    if (open) {
+      // Le profil n'est cherché que tant qu'il manque : une capture déjà
+      // rattachée l'est pour de bon, inutile de reposer la question toutes les
+      // deux secondes.
+      spansRepo.touch(open.id, now, open.targetId ? null : this.recordedTargetId(agentId));
+    } else {
+      spansRepo.open(agentId, this.recordedTargetId(agentId), now);
+    }
+  }
+
+  /** Le profil que cette VM capture, d'après le pseudo sur lequel sa veille est calée. */
+  private recordedTargetId(agentId: string): string | null {
+    const record = agentsRepo.get(agentId);
+    if (!record?.watch.username) return null;
+    return targetsRepo.findByUsername(record.watch.provider, record.watch.username)?.id ?? null;
+  }
+
   resolveCommand(agentId: string, requestId: string, ok: boolean, data: unknown, error?: string): void {
     const pending = this.connections.get(agentId)?.pending.get(requestId);
     if (!pending) return;
@@ -271,6 +316,12 @@ class Hub {
 
   /** Coupe les agents silencieux : le heartbeat n'arrive plus. */
   reapStale(): void {
+    // Même ménage côté captures : une VM disparue en plein enregistrement
+    // laisse un intervalle ouvert. On le referme à sa dernière preuve de vie,
+    // pas à maintenant — sans quoi une VM éteinte une semaine passerait pour
+    // avoir enregistré une semaine.
+    spansRepo.sweepStale(Date.now() - SPAN_STALE_MS);
+
     const deadline = Date.now() - config.agentTimeoutMs;
     for (const [agentId, connection] of this.connections) {
       if (connection.lastSeenAt < deadline) {

packages/server/src/watchlist.ts

@@ -1,7 +1,7 @@
 import type { StreamState, WatchTarget } from '@stream-control/shared';
 import { DEFAULT_WATCH_SETTINGS, fetchStripchatStatus } from '@stream-control/shared';
 import { config } from './config.ts';
-import { targetsRepo } from './db.ts';
+import { sessionsRepo, targetsRepo } from './db.ts';
 import { hub } from './hub.ts';
 import { notifyTargetLive } from './pushover.ts';
 import { startTargetRecording } from './recorder.ts';
@@ -89,6 +89,14 @@ class Watchlist {
 
     const changed = next !== target.state;
 
+    // Diffusion en cours : sans condition sur `changed`, car l'ouverture est
+    // idempotente. C'est ce qui rattrape un profil déjà en direct au démarrage
+    // du serveur — aucune transition ne se produira pour lui, et sa diffusion
+    // resterait sinon absente de la frise jusqu'à la suivante.
+    if (next === 'public') {
+      sessionsRepo.open(target.id, statusChangedAt ?? Date.now());
+    }
+
     // Fin de diffusion : on fige le stream qui vient de se terminer. Son début
     // est le statusChangedAt d'avant la transition, que la plateforme vient de
     // remplacer par celui du nouveau statut.
@@ -97,6 +105,7 @@ class Watchlist {
     if (changed && target.state === 'public') {
       lastLiveStartedAt = target.statusChangedAt ?? target.stateSince;
       lastLiveEndedAt = Date.now();
+      sessionsRepo.close(target.id, lastLiveEndedAt);
     }
 
     targetsRepo.updateState(target.id, {