Chat de crise : ce qui entre au journal, et ce qui n'y entre pas

| **L1** | Noyau de persistance, main courante, rembobinage | 🟡 en cours — noyau, lecture à date, filtres et routes faits ; reste l'export PDF paginé de la main courante (EF-410) et la vue frise, qui suppose une interface |

| **L2** | Dossier de préparation (annuaire, tiers, fiches réflexes) | 🟢 fait — sauf le kit documentaire, qui suit le stockage objet (L5) |

| **L3** | Décisions, actions, points de situation, cockpit | 🟢 fait |

| L4+ | Chat, GED, visio, tableau blanc, mobilisation | ⚪ |

| **L4** | Chat de crise, flux temps réel | 🟢 fait — sauf les pièces jointes, qui suivent le stockage objet (L5) |

| L5+ | GED, visio, tableau blanc, mobilisation | ⚪ |

Ce qui fonctionne aujourd'hui : le journal inviolable et son chaînage

d'empreintes, le vérificateur indépendant, la reconstitution d'un état passé

invitations, multi-organisation), la console d'administration avec ses quotas,

le dossier de préparation versionné avec son export papier, la boucle

décisionnelle complète (décisions, arbitrages, actions, points de situation,

cockpit) et ses routes HTTP, et la pile Docker.

cockpit), le chat de crise avec son flux temps réel, et la pile Docker.

---

    indicateur.ts    ce qui manque, et le geste qui le comble

    campagne.ts      vérification semestrielle des coordonnées

    export.ts        tirage papier et tableur

  domaine/chat.ts     salons, messages, épingle, réactions, promotion

  domaine/directs.ts  messages directs — hors journal, à dessein

  adaptateurs/diffusion.ts  temps réel : mémoire en mono-hôte, Redis sinon

  administration/ console root — administrer sans lire

    console.ts       organisations, comptes, droit root

    etat.ts          état d'instance, intégrité globale, bris de glace

stocke ses paramètres, une migration reste possible sans invalider les

empreintes.

## Chat : ce qui entre au journal, et ce qui n'y entre pas

Les **messages de salon sont dans le journal**. Un échange de cellule est la

matière première de la crise : c'est là que les faits remontent et que les

arbitrages se préparent. Éditer laisse une trace et l'original reste lisible ;

supprimer retire de la conversation, jamais de la preuve — l'interface affiche

une pierre tombale plutôt qu'un blanc, parce qu'un message qui disparaît sans

trace fait douter de tout le reste.

Les **messages directs n'y sont pas** (EF-708, EB-10). Un outil qui

enregistrerait de manière inaltérable les conversations privées serait déserté,

et cesser de s'en servir est le pire résultat possible. Les gens ont besoin d'un

endroit où demander « tu comprends ce qui se passe ? » sans que ce soit versé au

dossier remis à l'assureur trois mois plus tard. La contrepartie est

l'honnêteté : l'exclusion est renvoyée par l'API à chaque lecture.

Le **flux temps réel ne sert qu'à la fraîcheur**. La reprise après coupure passe

par la base, indexée par séquence : un client qui revient redemande « tout ce

qui suit N ». Une dépêche perdue ne fait jamais de trou dans la conversation —

c'est ce qui permet de travailler en 3G depuis un parking.

## La boucle décisionnelle

Ce qui sépare Kastell d'une messagerie. L'opérationnelle publie un **point de

  },

  "dependencies": {

    "@fastify/cookie": "^11.1.2",

    "@fastify/websocket": "^11.3.0",

    "@kastell/shared": "workspace:*",

    "fastify": "^5.2.0",

    "ioredis": "^6.0.0",

    "postgres": "^3.4.5",

    "zod": "^3.24.1"

  },

  "devDependencies": {

    "@types/node": "^22.10.2",

    "@types/ws": "^8.18.1",

    "esbuild": "^0.24.2",

    "tsx": "^4.19.2"

    "tsx": "^4.19.2",

    "ws": "^8.21.3"

  }

}

 No newline at end of file

import Redis from "ioredis";

/**

 * Diffusion temps réel.

 *

 * Deux implémentations derrière une seule interface, comme pour les autres

 * adaptateurs : en mémoire quand l'instance est seule, par Redis dès qu'il y

 * en a plusieurs. Le mono-hôte est la cible de premier ordre et ne doit pas

 * payer une dépendance dont il n'a pas besoin.

 *

 * Ce canal ne porte que du transport. Le rattrapage après coupure ne passe

 * pas par lui mais par la base, indexée par numéro de séquence (EX-13) : un

 * client qui revient redemande « tout ce qui suit la séquence N ». Une

 * diffusion perdue ne fait donc jamais de trou dans la conversation.

 */

export interface Depeche {

  criseId: string;

  /** Numéro de séquence du journal, ou null pour un signal éphémère. */

  seq: number | null;

  genre: string;

  charge: Record<string, unknown>;

}

type Abonne = (d: Depeche) => void;

/** Présence (EF-706) : qui est réellement mobilisé, à 3 h du matin. */

export interface Present {

  compte_id: string;

  nom: string;

  depuis: number;

}

/** Au-delà, une présence non rafraîchie est tenue pour perdue. */

export const PRESENCE_TTL_MS = 45_000;

export const PRESENCE_RAFRAICHISSEMENT_MS = 20_000;

export interface Diffusion {

  readonly nom: string;

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

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

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

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

  presents(criseId: string): Promise<Present[]>;

  fermer(): Promise<void>;

}

/** Diffusion locale : suffisante tant qu'une seule instance sert les clients. */

export function diffusionMemoire(): Diffusion {

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

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

  return {

    nom: "mémoire",

    async marquerPresent(criseId, present) {

      if (!presences.has(criseId)) presences.set(criseId, new Map());

      presences.get(criseId)!.set(present.compte_id, { ...present, depuis: Date.now() });

    },

    async retirerPresent(criseId, compteId) {

      presences.get(criseId)?.delete(compteId);

    },

    async presents(criseId) {

      const limite = Date.now() - PRESENCE_TTL_MS;

      const groupe = presences.get(criseId);

      if (!groupe) return [];

      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 */ }

      }

    },

    abonner(criseId, abonne) {

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

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

      return () => {

        const groupe = abonnes.get(criseId);

        groupe?.delete(abonne);

        if (groupe?.size === 0) abonnes.delete(criseId);

      };

    },

    async fermer() { abonnes.clear(); presences.clear(); },

  };

}

const CANAL = (criseId: string) => `kastell:crise:${criseId}`;

export function diffusionRedis(url: string): Diffusion {

  const options = { lazyConnect: false, maxRetriesPerRequest: 3, enableOfflineQueue: true };

  const editeur = new Redis(url, options);

  const lecteur = new Redis(url, options);

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

  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 */ }

    }

  });

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

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

  editeur.on("error", () => {});

  lecteur.on("error", () => {});

  const CLE_PRESENCE = (criseId: string) => `kastell:presence:${criseId}`;

  return {

    nom: "redis",

    async publier(d) {

      // 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);

    },

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

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

    async marquerPresent(criseId, present) {

      await editeur.zadd(CLE_PRESENCE(criseId), Date.now(),

        `${present.compte_id}|${present.nom}`);

      await editeur.expire(CLE_PRESENCE(criseId), 3600);

    },

    async retirerPresent(criseId, compteId) {

      const membres = await editeur.zrange(CLE_PRESENCE(criseId), "0", "-1");

      const cible = membres.filter((m) => m.startsWith(`${compteId}|`));

      if (cible.length > 0) await editeur.zrem(CLE_PRESENCE(criseId), ...cible);

    },

    async presents(criseId) {

      const limite = Date.now() - PRESENCE_TTL_MS;

      await editeur.zremrangebyscore(CLE_PRESENCE(criseId), 0, limite);

      const membres = await editeur.zrangebyscore(

        CLE_PRESENCE(criseId), limite, "+inf", "WITHSCORES");

      const presents: Present[] = [];

      for (let i = 0; i < membres.length; i += 2) {

        const [compte_id, nom] = (membres[i] ?? "").split("|");

        if (compte_id) {

          presents.push({ compte_id, nom: nom ?? "", depuis: Number(membres[i + 1] ?? 0) });

        }

      }

      return presents;

    },

    abonner(criseId, abonne) {

      if (!abonnes.has(criseId)) {

        abonnes.set(criseId, new Set());

        // Le rejet doit être recueilli : une promesse abandonnée abat le

        // processus, et perdre le canal ne doit coûter que la fraîcheur — le

        // client rattrape par séquence.

        lecteur.subscribe(CANAL(criseId)).catch(() => {});

      }

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

      return () => {

        const groupe = abonnes.get(criseId);

        groupe?.delete(abonne);

        if (groupe?.size === 0) {

          abonnes.delete(criseId);

          lecteur.unsubscribe(CANAL(criseId)).catch(() => {});

        }

      };

    },

    async fermer() {

      abonnes.clear();

      await Promise.allSettled([editeur.quit(), lecteur.quit()]);

    },

  };

}

let courante: Diffusion = diffusionMemoire();

export function definirDiffusion(d: Diffusion): void {

  courante = d;

}

export function diffusion(): Diffusion {

  return courante;

}

export async function publier(d: Depeche): Promise<void> {

  await courante.publier(d);

}

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

-- 0008 — Chat de crise (§6.7)

--

-- Deux décisions structurent ce schéma.

--

-- 1. Les messages de salon SONT dans le journal. Un échange de cellule est la

--    matière première de la crise : c'est là que les faits remontent et que

--    les arbitrages se préparent. Éditer un message est un événement, et

--    l'original reste lisible (EF-702).

--

-- 2. Les messages directs N'Y SONT PAS (EB-10, EF-708). Un outil qui

--    enregistrerait les conversations privées de manière inaltérable serait

--    déserté — et cesser de s'en servir est le pire résultat possible. Cette

--    exclusion est affichée dans le produit, jamais tue.

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

-- ── Salons (EF-701) ────────────────────────────────────────────────────────

create table salon (

  crise_id   uuid        not null references crise(id) on delete restrict,

  id         uuid        not null,

  cellule_id uuid        references cellule(id) on delete set null,

  code       text        not null,

  libelle    text        not null,

  genre      text        not null check (genre in ('cellule', 'thematique')),

  seq_creation bigint    not null,

  cree_at    timestamptz not null,

  ferme_at   timestamptz,

  primary key (crise_id, id)

);

create unique index salon_code on salon (crise_id, code);

create index salon_cellule on salon (crise_id, cellule_id) where ferme_at is null;

-- ── Messages (EF-702, EF-705) ──────────────────────────────────────────────

create table message (

  crise_id      uuid        not null references crise(id) on delete restrict,

  id            uuid        not null,

  salon_id      uuid        not null,

  seq           bigint      not null,

  auteur_id     uuid        references compte(id) on delete set null,

  texte         text        not null,

  -- Fil de discussion : un message répond à un autre, sans imbrication au-delà.

  repond_a      uuid,

  mentions      uuid[]      not null default '{}',

  envoye_at     timestamptz not null,

  edite_at      timestamptz,

  edite_seq     bigint,

  supprime_at   timestamptz,

  supprime_seq  bigint,

  epingle_at    timestamptz,

  epingle_par   uuid        references compte(id) on delete set null,

  -- Renseigné quand le message a été versé en main courante (EF-704).

  promu_seq     bigint,

  recherche     tsvector generated always as (to_tsvector('french', texte)) stored,

  primary key (crise_id, id)

);

create index message_salon on message (crise_id, salon_id, seq);

create index message_fil on message (crise_id, repond_a) where repond_a is not null;

create index message_epingle on message (crise_id, salon_id) where epingle_at is not null;

create index message_recherche on message using gin (recherche);

create index message_mentions on message using gin (mentions);

-- ── Réactions ──────────────────────────────────────────────────────────────

-- Journalisées comme le reste : en cellule de crise, une réaction vaut souvent

-- accusé de lecture, et « qui a validé l'isolement du réseau » se relit.

create table message_reaction (

  crise_id   uuid        not null,

  message_id uuid        not null,

  compte_id  uuid        not null references compte(id) on delete cascade,

  emoji      text        not null,

  pose_at    timestamptz not null,

  primary key (crise_id, message_id, compte_id, emoji)

);

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

-- Messages directs — HORS JOURNAL, à dessein (EB-10, EF-708)

--

-- Aucun chaînage, aucune immuabilité, et l'export du dossier de crise les

-- ignore. Le produit l'affiche à l'utilisateur : on ne laisse pas croire à un

-- enregistrement plus large qu'il n'est.

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

create table conversation_directe (

  id       uuid        primary key default gen_random_uuid(),

  crise_id uuid        not null references crise(id) on delete cascade,

  cree_at  timestamptz not null default now()

);

create table conversation_participant (

  conversation_id uuid not null references conversation_directe(id) on delete cascade,

  compte_id       uuid not null references compte(id) on delete cascade,

  primary key (conversation_id, compte_id)

);

create table message_direct (

  id              uuid        primary key default gen_random_uuid(),

  conversation_id uuid        not null references conversation_directe(id) on delete cascade,

  auteur_id       uuid        not null references compte(id) on delete cascade,

  texte           text        not null,

  envoye_at       timestamptz not null default now(),

  supprime_at     timestamptz

);

create index message_direct_conversation on message_direct (conversation_id, envoye_at);