Temps réel
Bus publish et subscribe du core sur LISTEN/NOTIFY de PostgreSQL, qui alimente les flux Server-Sent Events des routes /za/api.
core/realtime transmet un événement publié dans un processus à tous les abonnés de tous les processus de l'application, par LISTEN/NOTIFY de PostgreSQL. Il ne demande rien d'autre que la base que le projet a déjà, et il fonctionne avec plusieurs instances. Une route le relaie au navigateur avec apiEventStream().
Règle absolue : un événement est un signal, pas un enregistrement. « La commande 42 a changé », pas la commande. Un abonné lit l'état courant en commençant, puis le relit chaque fois qu'on lui signale un changement.
Import
import { publish, publishMany, subscribe, ensureRealtime, closeRealtime, RealtimeLostError } from '@fraza/core/realtime';Module serveur. Ne jamais l'importer depuis un fichier 'use client'. Le navigateur lit un flux avec openEventStream().
API
| Signature | Retour | Rôle |
|---|---|---|
publish(channel, data, { client }?) | Promise<void> | Publie un événement sur un canal. |
publishMany(channels, data, { client }?) | Promise<void> | Publie le même événement sur plusieurs canaux, en une instruction. |
subscribe(channel, { signal }?) | Promise<AsyncIterable> | S'abonne à un canal. |
ensureRealtime() | Promise<void> | Ouvre l'écoute du processus, ou lève un 503. |
closeRealtime() | Promise<void> | Termine les abonnements et ferme l'écoute. |
RealtimeLostError | classe | Erreur d'un abonnement dont les événements ne sont plus garantis. |
publish(channel, data, { client }?)
import { transaction } from '@fraza/core/database';
import { publish } from '@fraza/core/realtime';
await transaction(async (client) => {
await client.query('UPDATE orders SET status = $1 WHERE id = $2', ['shipped', id]);
await publish(`orders:${id}`, { status: 'shipped' }, { client });
});Sans client, l'événement passe par le pool et part tout de suite. Avec client, il part dans la transaction de l'appelant : PostgreSQL le livre au commit, et jamais si la transaction est annulée. Un abonné qui relit l'état à la réception voit donc toujours l'écriture qui l'a déclenché.
Le canal respecte /^[a-z][a-z0-9_.:-]{0,127}$/, par exemple notifications:42 ou orders.status:17. Préfixer par le nom du module évite qu'il en croise un autre. L'événement encodé en JSON ne dépasse pas 7 900 octets, sous le plafond de NOTIFY. Un canal mal formé ou un événement trop gros lève une TypeError.
publishMany(channels, data, { client }?)
await publishMany(recipients.map(id => `shop.orders.user:${id}`), null, { client });Les mêmes règles que publish(), pour une liste de canaux : une seule instruction SQL, quel que soit leur nombre, plutôt qu'un aller-retour par destinataire. Un canal en double n'est envoyé qu'une fois, et une liste vide ne fait rien. notify() s'en sert pour signaler chaque destinataire d'une notification.
subscribe(channel, { signal }?)
const events = await subscribe(`orders:${id}`, { signal });
send('state', await readOrderState(id)); // l'état courant, après l'abonnement
for await (const event of events) {
send('state', await readOrderState(id));
}L'abonnement est enregistré quand la promesse se résout, avant la première lecture. S'abonner d'abord, puis lire l'état : rien de ce qui est publié entre les deux n'est perdu.
L'itération rend le data de chaque événement. Elle se termine normalement quand signal est annulé ou que closeRealtime() s'exécute. Elle lève une RealtimeLostError quand la connexion d'écoute tombe, ou quand plus de 1 000 événements attendent sans être lus. Dans un producteur de flux, cette erreur ferme le flux : le navigateur se reconnecte et relit l'état.
drain(), sur l'abonnement, retire les événements en attente et rend leur nombre. Un abonné qui relit tout l'état à chaque signal s'en sert après une courte pause : une rafale de signaux donne une seule lecture.
const subscription = await subscribe(`orders:${id}`, { signal });
for await (const _ of subscription) {
await pause(250, signal);
subscription.drain();
send('state', await readOrderState(id));
}subscribe() appelle ensureRealtime() et lève donc le même 503 quand l'écoute est impossible.
ensureRealtime()
Ouvre l'écoute du processus si elle ne l'est pas, et rend la main tout de suite si elle l'est. Sans écoute possible, elle lève une ApiError 503 de code realtime_unavailable. Un handler l'attend avant de renvoyer son flux : le navigateur reçoit alors un vrai 503, que openEventStream() traite comme définitif, et la page revient à des appels périodiques.
Après un échec, les tentatives suivantes s'espacent, d'une seconde jusqu'à 30, et échouent immédiatement entre deux. Une base indisponible ne reçoit donc pas une tentative de connexion par requête.
closeRealtime()
Termine normalement chaque abonnement et ferme la connexion d'écoute. Le prochain abonnement en rouvre une. resetFrazaInitialization() et le harnais de test l'appellent. Un projet n'a pas à le faire.
Fonctionnement
Un seul canal PostgreSQL, fraza_realtime, porte tous les événements sous la forme {"c": canal, "d": data}. Le processus les répartit ensuite entre ses abonnés. Un seul LISTEN par processus suffit donc, quel que soit le nombre de canaux et d'abonnés, et aucun nom de canal n'est interpolé dans du SQL.
La connexion d'écoute est ouverte au premier abonnement, pas au démarrage : un projet qui ne diffuse rien n'ouvre rien. Elle est prise hors du pool, avec la même URL, la même politique TLS et le même délai de connexion que le pool, parce qu'un LISTEN garde sa connexion tant qu'il écoute. Le keepalive TCP est activé pour qu'une connexion longtemps inactive derrière un NAT ne meure pas sans que personne le sache.
Notes
Un pooler en mode transaction ne transmet pas LISTEN. PgBouncer dans ce mode, comme certains poolers d'hébergeurs, rend la connexion au pool après chaque transaction : l'écoute n'aboutit jamais. ensureRealtime() lève alors son 503, et les pages reviennent à leurs appels périodiques. Pour le temps réel, FRAZA_DATABASE_URL doit viser PostgreSQL directement, ou un pooler en mode session.
Un événement publié pendant une coupure est perdu. C'est la raison de la règle du signal : l'abonné qui revient relit l'état, il ne rejoue pas d'historique.
Le compteur d'attente protège la mémoire. Un abonné qui ne lit plus ses événements est coupé après 1 000 en attente, au lieu de les accumuler sans fin.