Ce que Marie consigne paraît chez les cinq cents autres

export interface Diffusion {

  readonly nom: string;

  publier(d: Depeche): Promise<void>;

  /**

   * Remettre aux seuls abonnés de CETTE instance.

   *

   * Pour ce que chaque instance apprend déjà par elle-même — les annonces du

   * journal, que Postgres notifie à toutes. Les republier par Redis les ferait

   * arriver autant de fois qu'il y a d'instances.

   */

  distribuer(d: Depeche): void;

  abonner(criseId: string, abonne: Abonne): () => void;

  marquerPresent(criseId: string, present: Present): Promise<void>;

  retirerPresent(criseId: string, compteId: string): Promise<void>;

export function diffusionMemoire(): Diffusion {

  const abonnes = new Map<string, Set<Abonne>>();

  const presences = new Map<string, Map<string, Present>>();

  const distribuer = (d: Depeche) => {

    for (const a of abonnes.get(d.criseId) ?? []) {

      try { a(d); } catch { /* un abonné fautif n'interrompt pas les autres */ }

    }

  };

  return {

    nom: "mémoire",

    async marquerPresent(criseId, present) {

      for (const [compteId, p] of groupe) if (p.depuis < limite) groupe.delete(compteId);

      return [...groupe.values()];

    },

    async publier(d) {

      for (const a of abonnes.get(d.criseId) ?? []) {

        try { a(d); } catch { /* un abonné fautif n'interrompt pas les autres */ }

      }

    },

    async publier(d) { distribuer(d); },

    distribuer,

    abonner(criseId, abonne) {

      if (!abonnes.has(criseId)) abonnes.set(criseId, new Set());

      abonnes.get(criseId)!.add(abonne);

  const lecteur = new Redis(url, options);

  const abonnes = new Map<string, Set<Abonne>>();

  const distribuer = (criseId: string, depeche: Depeche) => {

    for (const a of abonnes.get(criseId) ?? []) {

      try { a(depeche); } catch { /* idem */ }

    }

  };

  lecteur.on("message", (canal, brut) => {

    const criseId = canal.slice("kastell:crise:".length);

    let depeche: Depeche;

    try { depeche = JSON.parse(brut) as Depeche; } catch { return; }

    for (const a of abonnes.get(criseId) ?? []) {

      try { a(depeche); } catch { /* idem */ }

    }

    distribuer(criseId, depeche);

  });

  // Une panne du canal ne doit pas abattre l'application : la base reste la

  // source de vérité, le client rattrape par séquence.

      // Idem : la diffusion est un confort, la base est la source de vérité.

      await editeur.publish(CANAL(d.criseId), JSON.stringify(d)).catch(() => 0);

    },

    distribuer(d) { distribuer(d.criseId, d); },

    // Ensemble ordonné par horodatage : lister les présents revient à lire une

    // tranche, et les connexions perdues s'évaporent d'elles-mêmes.

-- ═══════════════════════════════════════════════════════════════════════════

-- Une entrée de main courante s'annonce à qui regarde (EF-401, EF-702).

--

-- ╔═════════════════════════════════════════════════════════════════════════╗

-- ║  Marie consigne. Yann, sur la main courante, ne voit rien venir.       ║

-- ╚═════════════════════════════════════════════════════════════════════════╝

--

-- Le flux temps réel ne portait que ce que le code publiait lui-même : les

-- messages, la mobilisation, les réunions. Aucune des trente écritures de la

-- main courante — consignation, décision, action, point de situation, tiers,

-- presse — ne publiait rien. L'écran des autres restait tel qu'il était au

-- chargement, jusqu'à ce qu'autre chose le fasse relire. Aucune garantie ne le

-- voyait : elles vérifiaient que le serveur publie un message, pas que l'écran

-- d'une autre personne bouge. Un parcours à deux navigateurs l'a trouvé.

--

-- Publier à chacune des trente écritures, c'était compter sur la trente et

-- unième pour ne pas l'oublier. L'annonce part donc du journal lui-même, et

-- par NOTIFY : Postgres ne la remet qu'à la validation de la transaction. Une

-- écriture annulée n'annonce rien, et un écran ne relit jamais avant que la

-- ligne soit visible.

--

-- La charge est minimale — crise, séquence, type, cellule — et c'est voulu :

-- le flux ne sert qu'à dire « relisez », la base fait foi. Le texte n'y

-- figure pas ; il n'a pas à voyager deux fois, ni à tenir dans les huit mille

-- octets d'une notification.

-- ═══════════════════════════════════════════════════════════════════════════

create or replace function journal_annoncer() returns trigger

language plpgsql as $$

begin

  perform pg_notify('kastell_journal', json_build_object(

    'crise_id', new.crise_id, 'seq', new.seq, 'type', new.type,

    'cellule_id', new.cellule_id)::text);

  return null;

end;

$$;

drop trigger if exists evenement_annonce on evenement;

create trigger evenement_annonce

  after insert on evenement

  for each row when (new.type like 'main\_courante.%')

  execute function journal_annoncer();

    charge: { salle_id: p.salle_id, participation_id: participationId, duree_secondes: duree },

    ctx,

  }));

  // L'entrée s'annonçait, la sortie non : le rail des autres montrait « en

  // appel, avec Marie » bien après que Marie eut raccroché, et l'on rejoignait

  // une salle vide en croyant y trouver quelqu'un.

  await publier({

    criseId, seq: null, genre: "reunion.quittee",

    charge: { salle_id: p.salle_id, compte_id: ctx.acteurId },

  });

  return duree;

}

import { promisify } from "node:util";

import { brotliCompress, constants, gzip } from "node:zlib";

import type { FastifyReply, FastifyRequest } from "fastify";

import type { Sql } from "../db/client.js";

/**

 * Les lectures que toute une crise demande au même instant (EX-14).

 *

 * Chaque entrée consignée fait relire le cockpit — et la main courante à qui

 * la regarde — sur chacun des écrans ouverts. Mesuré à cinq cents personnes :

 * cinq cents calculs identiques du même état, à la même milliseconde, et une

 * main courante de deux mille entrées sérialisée cent cinquante fois par

 * dépêche, soit 1,25 Mo chacune, non compressés. L'entrée devenait lisible

 * chez les autres au bout de vingt-huit secondes au 95ᵉ centile.

 *

 * Or ces lectures ne dépendent que du journal de la crise — et de la langue de

 * qui lit. Tant que le journal n'a pas avancé, la réponse est la même pour

 * tout le monde. On la calcule donc une fois par numéro de séquence, on la

 * compresse une fois par encodage, et les lecteurs simultanés attendent le

 * même calcul au lieu d'en lancer chacun un.

 *

 * Deux bornes à la fraîcheur, et elles suffisent :

 *

 *   — le numéro de séquence, relu à chaque requête (une clé primaire) : une

 *     écriture au journal rend la réponse caduque immédiatement ;

 *   — un âge maximal, pour ce qui change sans écrire au journal : une action

 *     qui passe en retard parce que l'heure tourne, un nom de compte corrigé.

 *

 * Ce qui n'est pas la vue commune — un filtre, un rembobinage — ne passe pas

 * par ici : ces lectures-là sont rares, et propres à qui les demande.

 */

export type Encodage = "br" | "gzip";

export interface Partage {

  json: string;

  compresse(e: Encodage): Promise<Buffer>;

}

interface Entree { seq: number; a: number; partage: Promise<Partage> }

const brotli = promisify(brotliCompress);

const gzipper = promisify(gzip);

const memoire = new Map<string, Entree>();

/** Assez pour quelques centaines de crises ouvertes à la fois ; au-delà, la plus ancienne sort. */

const TAILLE_MAX = 400;

export function compresser(corps: Buffer | string, e: Encodage): Promise<Buffer> {

  // Qualité modérée : compresser vite compte davantage que quelques octets de

  // moins, puisque c'est le serveur de crise qui paie à chaque lecture.

  return e === "br"

    ? brotli(corps, { params: { [constants.BROTLI_PARAM_QUALITY]: 4 } })

    : gzipper(corps, { level: 5 });

}

/** L'encodage que le client accepte, brotli d'abord. */

export function encodageAccepte(entete: string | string[] | undefined): Encodage | null {

  const v = (Array.isArray(entete) ? entete.join(",") : entete ?? "").toLowerCase();

  const accepte = (nom: string) => new RegExp(`(^|,)\\s*${nom}\\s*(;\\s*q=(?!0(\\.0*)?\\s*(,|$))[\\d.]+)?\\s*(,|$)`).test(v);

  return accepte("br") ? "br" : accepte("gzip") ? "gzip" : null;

}

/** Nombre de calculs réellement faits — pour la suite de vérification. */

export let calculsFaits = 0;

export async function lecturePartagee(

  sql: Sql, criseId: string, cle: string, ageMaxMs: number, calculer: () => Promise<unknown>,

): Promise<Partage> {

  const [c] = await sql<{ journal_seq: string | null }[]>`

    select journal_seq from crise where id = ${criseId}`;

  const seq = Number(c?.journal_seq ?? 0);

  const k = `${criseId}|${cle}`;

  const e = memoire.get(k);

  if (e && e.seq === seq && Date.now() - e.a < ageMaxMs) return e.partage;

  const partage = (async (): Promise<Partage> => {

    calculsFaits++;

    const json = JSON.stringify(await calculer());

    const compressions = new Map<Encodage, Promise<Buffer>>();

    return {

      json,

      compresse(enc) {

        if (!compressions.has(enc)) compressions.set(enc, compresser(json, enc));

        return compressions.get(enc)!;

      },

    };

  })();

  memoire.delete(k);

  memoire.set(k, { seq, a: Date.now(), partage });

  if (memoire.size > TAILLE_MAX) memoire.delete(memoire.keys().next().value!);

  // Un calcul qui échoue ne reste pas en mémoire : le suivant réessaie.

  partage.catch(() => { if (memoire.get(k)?.partage === partage) memoire.delete(k); });

  return partage;

}

/** Au-delà de ce poids, une réponse part compressée si le client l'accepte. */

export const SEUIL_COMPRESSION = 1024;

export async function envoyerPartage(req: FastifyRequest, reply: FastifyReply, p: Partage) {

  reply.header("content-type", "application/json; charset=utf-8").header("vary", "accept-encoding");

  const enc = encodageAccepte(req.headers["accept-encoding"]);

  if (enc && Buffer.byteLength(p.json) >= SEUIL_COMPRESSION) {

    reply.header("content-encoding", enc);

    return reply.send(await p.compresse(enc));

  }

  return reply.send(p.json);

}

export function enregistrerRoutesChat(app: FastifyInstance, sql: Sql): void {

  const acces = { preHandler: requerirAccesCrise(sql) };

  // ── Les annonces du journal (EF-401, EF-702) ──────────────────────────────

  //

  // Chaque entrée de main courante est notifiée par Postgres à la validation

  // (migration 0057). Chaque instance l'entend : elle la remet donc à ses seuls

  // abonnés, sans la republier — voir « distribuer ». Perdre l'écoute ne coûte

  // que la fraîcheur : le rattrapage relit la base à chaque reconnexion.

  const ecoute = sql.listen("kastell_journal", (brut) => {

    let a: { crise_id?: unknown; seq?: unknown; type?: unknown; cellule_id?: unknown };

    try { a = JSON.parse(brut) as typeof a; } catch { return; }

    if (typeof a.crise_id !== "string" || typeof a.seq !== "number" || typeof a.type !== "string") return;

    diffusion().distribuer({

      criseId: a.crise_id, seq: a.seq, genre: a.type,

      charge: { cellule_id: typeof a.cellule_id === "string" ? a.cellule_id : null },

    });

  }).catch((e: unknown) => {

    app.log.warn({ err: e }, "écoute du journal impossible : la main courante ne sera fraîche qu'au rattrapage");

    return null;

  });

  app.addHook("onClose", async () => { await (await ecoute)?.unlisten(); });

  const criseId = (req: FastifyRequest) =>

    z.object({ criseId: z.string().uuid() }).parse(req.params).criseId;

  const sousId = (req: FastifyRequest) =>

      await canal.marquerPresent(id, { compte_id: compte.id, nom: compte.nom, depuis: Date.now() });

      envoyer({ t: "presence", presents: await canal.presents(id) });

    };

    void annoncerPresence();

    // L'arrivée se dit aux autres, et pas seulement à qui arrive (EF-706).

    //

    // Seul l'arrivant recevait la liste : ceux qui étaient déjà là ne

    // l'apprenaient qu'à leur prochain pouls, vingt secondes plus tard — le

    // temps de conclure « personne d'autre n'est là » et d'appeler. Un parcours

    // à deux navigateurs l'a montré : Marie voyait Yann, Yann ne voyait pas

    // Marie. On diffuse l'arrivée seule, et non la liste entière : à cinq cents

    // présents, chaque arrivée enverrait la liste complète à chacun (EX-14).

    void annoncerPresence().then(() => canal.publier({

      criseId: id, seq: null, genre: "presence.arrivee",

      charge: { compte_id: compte.id, nom: compte.nom },

    }));

    const pouls = setInterval(() => void annoncerPresence(), PRESENCE_RAFRAICHISSEMENT_MS);

    socket.on("message", (brut: Buffer) => {

          // EX-13 : la reprise lit la base, jamais le canal. Le client donne la

          // dernière séquence qu'il a vue, on lui rend la suite.

          const depuis = Number(ordre.depuisSeq ?? 0);

          //

          // La reprise suit la même portée que le direct (EF-707a). Elle ne la

          // suivait pas : le flux filtrait les messages d'un salon privé par

          // destinataire, mais la reprise relisait le journal sans filtre — et

          // rendait leur texte, et le texte d'avant chaque correction, à toute

          // personne de la crise qui se reconnectait. Ce qui ne passe pas par

          // le direct ne passe pas davantage par la reprise.

          const evenements = await sql<{ seq: string; type: string; occurred_at: Date;

            acteur_id: string | null; charge: Record<string, unknown> }[]>`

            select seq, type, occurred_at, acteur_id, charge from evenement

             where crise_id = ${id} and seq > ${depuis} and type in ${sql(TYPES_FLUX)}

             order by seq limit 500`;

            select e.seq, e.type, e.occurred_at, e.acteur_id, e.charge from evenement e

             where e.crise_id = ${id} and e.seq > ${depuis} and e.type in ${sql(TYPES_FLUX)}

               and (e.charge->>'salon_id' is null

                    or exists (select 1 from salon s

                                where s.crise_id = e.crise_id

                                  and s.id = (e.charge->>'salon_id')::uuid

                                  and (not s.prive

                                       or exists (select 1 from salon_membre m

                                                   where m.crise_id = s.crise_id

                                                     and m.salon_id = s.id

                                                     and m.compte_id = ${compte.id}))))

             order by e.seq limit 500`;

          envoyer({

            t: "rattrapage",

            evenements: evenements.map((e) => ({ ...e, seq: Number(e.seq) })),