Feat : Streamer watch list

packages/server/src/api.ts

@@ -6,11 +6,13 @@ import {
   DEFAULT_OBS_SETTINGS,
   isAgentAction,
   normalizeWatchSettings,
+  parseStripchatUsername,
 } from '@stream-control/shared';
 import { config } from './config.ts';
 import { generateToken, hashToken, issueSession, requireSession, safeEqual } from './auth.ts';
-import { agentsRepo, logsRepo } from './db.ts';
+import { agentsRepo, logsRepo, targetsRepo } from './db.ts';
 import { hub } from './hub.ts';
+import { watchlist } from './watchlist.ts';
 
 export const api: Router = Router();
 
@@ -188,6 +190,157 @@ api.post('/commands/bulk', async (req, res) => {
   res.json({ results });
 });
 
+// --- Veille : profils surveillés ---------------------------------------------
+
+api.get('/watchlist', (_req, res) => {
+  res.json({ targets: targetsRepo.list() });
+});
+
+/** Accepte une URL de profil complète ou un simple pseudo. */
+api.post('/watchlist', (req, res) => {
+  const input = typeof req.body?.url === 'string' ? req.body.url : '';
+  const username = parseStripchatUsername(input);
+  if (!username) {
+    res.status(400).json({
+      error: 'Lien ou pseudo Stripchat non reconnu (ex. https://fr.stripchat.com/pseudo)',
+    });
+    return;
+  }
+
+  const existing = targetsRepo.findByUsername('stripchat', username);
+  if (existing) {
+    res.status(409).json({ error: `« ${username} » est déjà surveillé`, target: existing });
+    return;
+  }
+
+  const target = targetsRepo.create({
+    id: randomUUID(),
+    username,
+    label: typeof req.body?.label === 'string' && req.body.label.trim() ? req.body.label.trim() : null,
+    agentId: typeof req.body?.agentId === 'string' ? req.body.agentId : null,
+    notify: req.body?.notify !== false,
+  });
+
+  hub.log(null, 'info', `Veille : « ${username} » ajouté`);
+  hub.publishTarget(target);
+  res.status(201).json({ target });
+
+  // Premier verdict sans attendre le prochain cycle.
+  void watchlist.checkOne(target.id);
+});
+
+api.patch('/watchlist/:id', (req, res) => {
+  const target = targetsRepo.get(req.params.id);
+  if (!target) {
+    res.status(404).json({ error: 'Profil surveillé introuvable' });
+    return;
+  }
+
+  targetsRepo.updateSettings(target.id, {
+    label: typeof req.body?.label === 'string' ? req.body.label.trim() || null : target.label,
+    agentId:
+      req.body?.agentId === null || typeof req.body?.agentId === 'string'
+        ? req.body.agentId
+        : target.agentId,
+    notify: typeof req.body?.notify === 'boolean' ? req.body.notify : target.notify,
+  });
+
+  const updated = targetsRepo.get(target.id);
+  if (!updated) {
+    res.status(500).json({ error: 'Mise à jour impossible' });
+    return;
+  }
+  hub.publishTarget(updated);
+  res.json({ target: updated });
+});
+
+api.delete('/watchlist/:id', (req, res) => {
+  const target = targetsRepo.get(req.params.id);
+  if (!target) {
+    res.status(404).json({ error: 'Profil surveillé introuvable' });
+    return;
+  }
+  targetsRepo.remove(target.id);
+  hub.publishTargetRemoval(target.id);
+  hub.log(null, 'info', `Veille : « ${target.username} » retiré`);
+  res.json({ ok: true });
+});
+
+api.post('/watchlist/:id/check', async (req, res) => {
+  const target = await watchlist.checkOne(req.params.id);
+  if (!target) {
+    res.status(404).json({ error: 'Profil surveillé introuvable' });
+    return;
+  }
+  res.json({ target });
+});
+
+/**
+ * Lance l'enregistrement du profil sur l'agent qui lui est assigné.
+ *
+ * Configure au passage la surveillance de l'agent sur ce pseudo : l'agent mettra
+ * l'enregistrement en pause pendant les shows privés sans réglage supplémentaire.
+ */
+api.post('/watchlist/:id/record', async (req, res) => {
+  const target = targetsRepo.get(req.params.id);
+  if (!target) {
+    res.status(404).json({ error: 'Profil surveillé introuvable' });
+    return;
+  }
+  if (!target.agentId) {
+    res.status(400).json({ error: 'Aucun agent assigné à ce profil' });
+    return;
+  }
+
+  const record = agentsRepo.get(target.agentId);
+  if (!record) {
+    res.status(404).json({ error: "L'agent assigné n'existe plus" });
+    return;
+  }
+
+  agentsRepo.updateSettings(record.id, {
+    name: record.name,
+    obs: record.obs,
+    autoConnectObs: record.autoConnectObs,
+    watch: normalizeWatchSettings({ ...record.watch, enabled: true, username: target.username }),
+    notes: record.notes,
+  });
+
+  const configured = agentsRepo.get(record.id);
+  if (configured) {
+    hub.pushConfig(configured);
+    hub.publishAgent(configured.id);
+  }
+
+  try {
+    await hub.sendCommand(record.id, 'record.start');
+    hub.log(
+      record.id,
+      'info',
+      `Enregistrement de « ${target.username} » démarré depuis la veille`,
+    );
+    res.json({ ok: true });
+  } catch (err) {
+    const message = err instanceof Error ? err.message : String(err);
+    hub.log(record.id, 'error', `Démarrage de « ${target.username} » en échec : ${message}`);
+    res.status(502).json({ ok: false, error: message });
+  }
+});
+
+api.post('/watchlist/:id/stop', async (req, res) => {
+  const target = targetsRepo.get(req.params.id);
+  if (!target?.agentId) {
+    res.status(400).json({ error: 'Aucun agent assigné à ce profil' });
+    return;
+  }
+  try {
+    await hub.sendCommand(target.agentId, 'record.stop');
+    res.json({ ok: true });
+  } catch (err) {
+    res.status(502).json({ ok: false, error: err instanceof Error ? err.message : String(err) });
+  }
+});
+
 // --- Divers -----------------------------------------------------------------
 
 api.get('/logs', (req, res) => {

packages/server/src/config.ts

@@ -45,6 +45,9 @@ export const config = {
   dbPath: path.resolve(repoRoot, process.env.DB_PATH ?? './data/stream-control.sqlite'),
 
   statusIntervalMs: int('STATUS_INTERVAL_MS', 2000),
+
+  /** Fréquence de sondage des profils surveillés (veille). */
+  watchlistIntervalMs: int('WATCHLIST_INTERVAL_MS', 30_000),
   agentTimeoutMs: int('AGENT_TIMEOUT_MS', 15_000),
   commandTimeoutMs: int('COMMAND_TIMEOUT_MS', 15_000),
 

packages/server/src/db.ts

@@ -6,13 +6,16 @@ import type {
   LogLevel,
   ObsSettings,
   Platform,
+  StreamState,
   WatchSettings,
+  WatchTarget,
 } from '@stream-control/shared';
 import {
   DEFAULT_OBS_SETTINGS,
   DEFAULT_WATCH_SETTINGS,
   normalizeWatchSettings,
   safeJsonParse,
+  stripchatProfileUrl,
 } from '@stream-control/shared';
 import { config } from './config.ts';
 
@@ -49,6 +52,26 @@ db.exec(`
   );
 
   CREATE INDEX IF NOT EXISTS idx_logs_ts ON logs (ts DESC);
+
+  -- Profils surveillés : le serveur sonde leur statut et signale les passages
+  -- en direct. Indépendant des agents : on peut veiller sans rien enregistrer.
+  CREATE TABLE IF NOT EXISTS watch_targets (
+    id              TEXT PRIMARY KEY,
+    provider        TEXT NOT NULL DEFAULT 'stripchat',
+    username        TEXT NOT NULL,
+    label           TEXT,
+    agent_id        TEXT,
+    notify          INTEGER NOT NULL DEFAULT 1,
+    state           TEXT NOT NULL DEFAULT 'unknown',
+    raw_status      TEXT,
+    state_since     INTEGER NOT NULL,
+    last_checked_at INTEGER,
+    last_error      TEXT,
+    created_at      INTEGER NOT NULL
+  );
+
+  CREATE UNIQUE INDEX IF NOT EXISTS idx_targets_identity
+    ON watch_targets (provider, username);
 `);
 
 /** Migrations additives : sûres à rejouer sur une base déjà peuplée. */
@@ -249,6 +272,136 @@ export const agentsRepo = {
   },
 };
 
+// --- Profils surveillés ------------------------------------------------------
+
+interface TargetRow {
+  id: string;
+  provider: string;
+  username: string;
+  label: string | null;
+  agent_id: string | null;
+  notify: number;
+  state: string;
+  raw_status: string | null;
+  state_since: number;
+  last_checked_at: number | null;
+  last_error: string | null;
+  created_at: number;
+}
+
+function toTarget(row: TargetRow): WatchTarget {
+  return {
+    id: row.id,
+    provider: (row.provider as WatchTarget['provider']) ?? 'stripchat',
+    username: row.username,
+    label: row.label,
+    url: stripchatProfileUrl(row.username),
+    agentId: row.agent_id,
+    notify: Number(row.notify) === 1,
+    state: (row.state as StreamState) ?? 'unknown',
+    rawStatus: row.raw_status,
+    stateSince: Number(row.state_since),
+    lastCheckedAt: row.last_checked_at === null ? null : Number(row.last_checked_at),
+    lastError: row.last_error,
+    createdAt: Number(row.created_at),
+  };
+}
+
+const targetStmts = {
+  list: db.prepare('SELECT * FROM watch_targets ORDER BY username COLLATE NOCASE'),
+  get: db.prepare('SELECT * FROM watch_targets WHERE id = ?'),
+  getByName: db.prepare('SELECT * FROM watch_targets WHERE provider = ? AND username = ?'),
+  insert: db.prepare(`
+    INSERT INTO watch_targets (id, provider, username, label, agent_id, notify,
+                               state, state_since, created_at)
+    VALUES (?, ?, ?, ?, ?, ?, 'unknown', ?, ?)
+  `),
+  updateSettings: db.prepare(
+    'UPDATE watch_targets SET label = ?, agent_id = ?, notify = ? WHERE id = ?',
+  ),
+  updateState: db.prepare(`
+    UPDATE watch_targets
+    SET state = ?, raw_status = ?, state_since = ?, last_checked_at = ?, last_error = ?
+    WHERE id = ?
+  `),
+  remove: db.prepare('DELETE FROM watch_targets WHERE id = ?'),
+};
+
+export const targetsRepo = {
+  list(): WatchTarget[] {
+    return (targetStmts.list.all() as unknown as TargetRow[]).map(toTarget);
+  },
+
+  get(id: string): WatchTarget | null {
+    const row = targetStmts.get.get(id) as unknown as TargetRow | undefined;
+    return row ? toTarget(row) : null;
+  },
+
+  findByUsername(provider: string, username: string): WatchTarget | null {
+    const row = targetStmts.getByName.get(provider, username) as unknown as TargetRow | undefined;
+    return row ? toTarget(row) : null;
+  },
+
+  create(input: {
+    id: string;
+    username: string;
+    provider?: string;
+    label?: string | null;
+    agentId?: string | null;
+    notify?: boolean;
+  }): WatchTarget {
+    const now = Date.now();
+    targetStmts.insert.run(
+      input.id,
+      input.provider ?? 'stripchat',
+      input.username,
+      input.label ?? null,
+      input.agentId ?? null,
+      input.notify === false ? 0 : 1,
+      now,
+      now,
+    );
+    const created = targetsRepo.get(input.id);
+    if (!created) throw new Error(`Échec de création du profil surveillé ${input.username}`);
+    return created;
+  },
+
+  updateSettings(
+    id: string,
+    settings: { label: string | null; agentId: string | null; notify: boolean },
+  ): void {
+    targetStmts.updateSettings.run(
+      settings.label,
+      settings.agentId,
+      settings.notify ? 1 : 0,
+      id,
+    );
+  },
+
+  updateState(
+    id: string,
+    state: {
+      state: StreamState;
+      rawStatus: string | null;
+      stateSince: number;
+      lastError: string | null;
+    },
+  ): void {
+    targetStmts.updateState.run(
+      state.state,
+      state.rawStatus,
+      state.stateSince,
+      Date.now(),
+      state.lastError,
+      id,
+    );
+  },
+
+  remove(id: string): void {
+    targetStmts.remove.run(id);
+  },
+};
+
 interface LogRow {
   id: number;
   agent_id: string | null;

packages/server/src/hub.ts

@@ -8,10 +8,11 @@ import type {
   LogLevel,
   ServerToAgent,
   ServerToDashboard,
+  WatchTarget,
 } from '@stream-control/shared';
 import { emptyStatus } from '@stream-control/shared';
 import { config } from './config.ts';
-import { agentsRepo, logsRepo, type AgentRecord } from './db.ts';
+import { agentsRepo, logsRepo, targetsRepo, type AgentRecord } from './db.ts';
 
 interface PendingCommand {
   resolve: (value: unknown) => void;
@@ -190,6 +191,7 @@ class Hub {
       type: 'snapshot',
       agents: this.views(),
       logs: logsRepo.recent(200),
+      targets: targetsRepo.list(),
     });
   }
 
@@ -211,6 +213,21 @@ class Hub {
     this.broadcast({ type: 'agent.removed', agentId });
   }
 
+  // --- Profils surveillés -------------------------------------------------
+
+  publishTarget(target: WatchTarget): void {
+    this.broadcast({ type: 'target', target });
+  }
+
+  publishTargetRemoval(targetId: string): void {
+    this.broadcast({ type: 'target.removed', targetId });
+  }
+
+  /** Passage en direct : le dashboard en fait une notification. */
+  publishTargetLive(target: WatchTarget): void {
+    this.broadcast({ type: 'target.live', target });
+  }
+
   /** Journalise un évènement : persistance + diffusion temps réel. */
   log(agentId: string | null, level: LogLevel, message: string, ts = Date.now()): LogEntry {
     const entry = logsRepo.append(agentId, level, message, ts);

packages/server/src/index.ts

@@ -8,6 +8,7 @@ import { verifySession } from './auth.ts';
 import { api } from './api.ts';
 import { hub } from './hub.ts';
 import { authenticateAgent, handleAgentConnection } from './agentGateway.ts';
+import { watchlist } from './watchlist.ts';
 
 const app = express();
 app.disable('x-powered-by');
@@ -91,6 +92,8 @@ const heartbeat = setInterval(() => {
 }, 5000);
 heartbeat.unref();
 
+watchlist.start();
+
 server.listen(config.port, config.host, () => {
   console.log(`stream-control · http://${config.host}:${config.port}`);
   console.log(`  agents     → ws://${config.host}:${config.port}/ws/agent`);

packages/server/src/watchlist.ts

@@ -0,0 +1,115 @@
+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 { hub } from './hub.ts';
+
+/**
+ * Sonde périodiquement les profils surveillés et signale les passages en direct.
+ *
+ * Côté serveur et non côté agent : un profil se surveille indépendamment de
+ * toute machine d'enregistrement, et une seule requête suffit quel que soit le
+ * nombre d'agents.
+ */
+class Watchlist {
+  private timer: NodeJS.Timeout | null = null;
+  private running = false;
+
+  start(): void {
+    if (this.timer) return;
+    this.timer = setInterval(() => void this.tick(), config.watchlistIntervalMs);
+    this.timer.unref?.();
+    void this.tick();
+  }
+
+  stop(): void {
+    if (this.timer) clearInterval(this.timer);
+    this.timer = null;
+  }
+
+  /** Sonde immédiate d'un seul profil, déclenchée depuis le dashboard. */
+  async checkOne(id: string): Promise<WatchTarget | null> {
+    const target = targetsRepo.get(id);
+    if (!target) return null;
+    await this.probe(target);
+    return targetsRepo.get(id);
+  }
+
+  private async tick(): Promise<void> {
+    if (this.running) return; // un cycle lent ne doit pas s'empiler sur le suivant
+    this.running = true;
+    try {
+      const targets = targetsRepo.list();
+      // Séquentiel et espacé : une poignée de profils ne justifie pas de
+      // marteler l'API en parallèle.
+      for (const target of targets) {
+        await this.probe(target);
+        await delay(250);
+      }
+    } finally {
+      this.running = false;
+    }
+  }
+
+  private async probe(target: WatchTarget): Promise<void> {
+    let next: StreamState;
+    let raw: string | null = null;
+    let error: string | null = null;
+
+    try {
+      const result = await fetchStripchatStatus(
+        target.username,
+        DEFAULT_WATCH_SETTINGS.privateStatuses,
+      );
+      next = result.state;
+      raw = result.raw;
+    } catch (err) {
+      // Une sonde en échec ne change pas l'état connu : on garde le dernier
+      // verdict fiable plutôt que d'annoncer un faux passage hors-ligne.
+      error = err instanceof Error ? err.message : String(err);
+      next = target.state;
+    }
+
+    const changed = next !== target.state;
+    targetsRepo.updateState(target.id, {
+      state: next,
+      rawStatus: raw ?? target.rawStatus,
+      stateSince: changed ? Date.now() : target.stateSince,
+      lastError: error,
+    });
+
+    const updated = targetsRepo.get(target.id);
+    if (!updated) return;
+
+    hub.publishTarget(updated);
+
+    if (changed) {
+      const name = updated.label ?? updated.username;
+      hub.log(null, 'info', `Veille : ${name} est passé « ${labelOf(next)} »`);
+
+      // Seul le passage effectif au flux public déclenche une notification.
+      if (next === 'public' && target.state !== 'public' && updated.notify) {
+        hub.publishTargetLive(updated);
+      }
+    }
+  }
+}
+
+function labelOf(state: StreamState): string {
+  switch (state) {
+    case 'public':
+      return 'en direct';
+    case 'private':
+      return 'en show privé';
+    case 'offline':
+      return 'hors-ligne';
+    default:
+      return 'inconnu';
+  }
+}
+
+function delay(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+export const watchlist = new Watchlist();