Egutierrez
6b162deeb0
Añade GET /api/bench (SSE) y una seccion de simulador en index.html: un publisher inunda una room con miles de mensajes a N subscribers y una grafica en vivo anima el throughput. Las dos politicas de room se exponen como flags independientes (persist=JetStream, encrypt=E2E AEAD+Ed25519) mas tamano de payload, midiendo el coste de cada capa con la libreria cliente real. El benchmark usa peers efimeros propios, sin tocar los peers nombrados del sandbox manual. Verificado: las 4 combinaciones enc x persist con fan-out exacto. Bump app v0.2.0.
164 lines
7.2 KiB
Markdown
164 lines
7.2 KiB
Markdown
---
|
|
name: unibus
|
|
lang: go
|
|
domain: infra
|
|
version: 0.2.0
|
|
description: "Bus de mensajería unificado sobre NATS+JetStream con cifrado E2E por room (megolm/olm reducido): service de membresía/claves, librería cliente y peers demo."
|
|
tags: [service, messaging, nats, e2e]
|
|
uses_functions:
|
|
- generate_identity_go_cybersecurity
|
|
- seal_aead_go_cybersecurity
|
|
- open_aead_go_cybersecurity
|
|
- seal_key_box_go_cybersecurity
|
|
- open_key_box_go_cybersecurity
|
|
- sign_ed25519_go_cybersecurity
|
|
- verify_ed25519_go_cybersecurity
|
|
uses_types: []
|
|
framework: ""
|
|
entry_point: "cmd/membershipd"
|
|
dir_path: "projects/message_bus/apps/unibus"
|
|
repo_url: "https://gitea-dgg044oo04woo4ggcsws4gk0.organic-machine.com/dataforge/unibus"
|
|
service:
|
|
port: 8470
|
|
health_endpoint: /healthz
|
|
health_timeout_s: 3
|
|
systemd_unit: null
|
|
systemd_scope: null
|
|
restart_policy: none
|
|
runtime: manual
|
|
pc_targets:
|
|
- lucas-linux
|
|
is_local_only: true
|
|
e2e_checks:
|
|
- id: build
|
|
cmd: "CGO_ENABLED=0 go build ./..."
|
|
timeout_s: 180
|
|
- id: vet
|
|
cmd: "CGO_ENABLED=0 go vet ./..."
|
|
timeout_s: 120
|
|
- id: unit
|
|
cmd: "CGO_ENABLED=0 go test ./..."
|
|
timeout_s: 180
|
|
- id: smoke
|
|
cmd: "CGO_ENABLED=0 go build -o /tmp/unibus_membershipd ./cmd/membershipd && /tmp/unibus_membershipd --http-port 18470 --nats-port 14222 --db /tmp/unibus_smoke.db --store-dir /tmp/unibus_blobs --nats-store /tmp/unibus_js &"
|
|
health: "http://127.0.0.1:18470/healthz"
|
|
timeout_s: 30
|
|
---
|
|
|
|
## Qué es
|
|
|
|
`unibus` es un bus de mensajería unificado sobre NATS + JetStream. Una capa fina
|
|
encima de NATS aporta lo que NATS no tiene: membresía de rooms, invitaciones con
|
|
reparto de clave, y cifrado extremo a extremo (E2E) del payload por room, con
|
|
rotación de clave activa (forward secrecy) al expulsar a un miembro.
|
|
|
|
Todos los participantes —procesos/workers, interfaces de chat humanas, agentes
|
|
LLM— son peers de primera clase que hablan el mismo protocolo y se diferencian
|
|
solo por las rooms a las que se unen y por lo que hacen con lo que reciben.
|
|
|
|
### Piezas
|
|
|
|
| Componente | Ruta | Rol |
|
|
|---|---|---|
|
|
| `membershipd` | `cmd/membershipd` | Service (control plane): metadata de rooms, directorio de miembros, reparto de claves selladas, object store de media. Arranca NATS embebido si no se pasa `--nats-url`. |
|
|
| librería cliente | `pkg/client` | La API que consume cualquier peer. Crear/unirse a rooms, invitar, publicar/suscribir, request/reply, kick con rotación de clave, media cifrada. |
|
|
| `worker` | `cmd/worker` | Peer demo: publica un contador incremental cada segundo a una room cleartext. |
|
|
| `chat` | `cmd/chat` | Peer demo: suscriptor en vivo (modo simple) + demo de cifrado E2E y forward secrecy (`--demo-encrypted`). |
|
|
|
|
### Dos planos
|
|
|
|
- **Control plane**: HTTP autoritativo en `membershipd`. Quién está en cada room,
|
|
sus claves públicas, y la clave de room `K` sellada por epoch para cada miembro.
|
|
- **Data plane**: NATS. Los mensajes (frames) viajan por subject; los blobs de
|
|
media NO viajan por el bus, se cifran y se suben al object store, y por NATS
|
|
solo viaja una referencia (hash + nonce).
|
|
|
|
### Criptografía (importada del registry, NO reescrita)
|
|
|
|
El cifrado E2E se compone de las 7 primitivas del capability group
|
|
`e2e-messaging` (`docs/capabilities/e2e-messaging.md`), importadas del paquete
|
|
`fn-registry/functions/cybersecurity`. Esta app no reimplementa ninguna
|
|
primitiva criptográfica.
|
|
|
|
## Ejemplo
|
|
|
|
Demo end-to-end con `go run` (NATS embebido, nada que instalar):
|
|
|
|
```bash
|
|
cd projects/message_bus/apps/unibus
|
|
|
|
# 1. Service de membresía/claves (NATS embebido en :4250, HTTP en :8470)
|
|
go run ./cmd/membershipd
|
|
|
|
# 2. En otra terminal: peer publicador (proceso) — publica ticks cada 1s
|
|
go run ./cmd/worker
|
|
|
|
# 3. En otra terminal: peer suscriptor (chat humano) — imprime cada tick en vivo
|
|
go run ./cmd/chat
|
|
|
|
# 4. Demo de cifrado E2E + forward secrecy (contra el membershipd ya corriendo):
|
|
# A crea room cifrada, invita a B, A publica (B descifra), A expulsa a B,
|
|
# A publica de nuevo en el nuevo epoch (B ya NO puede descifrar).
|
|
go run ./cmd/chat --demo-encrypted
|
|
```
|
|
|
|
Para apuntar a un NATS externo en producción: `--nats-url nats://host:4222` en
|
|
`membershipd`, `worker` y `chat`.
|
|
|
|
## Cuando usarla
|
|
|
|
- Cuando necesites un tejido de mensajería donde procesos, humanos y agentes LLM
|
|
sean peers uniformes (mismo protocolo, distinta política por room).
|
|
- Cuando quieras rooms cifradas E2E con forward secrecy (paridad con Matrix) sin
|
|
montar un Synapse: `room.ModeMatrix`.
|
|
- Cuando quieras fan-out cleartext rápido para telemetría/coordinación de
|
|
procesos: `room.ModeNATS`.
|
|
- Como sustituto de la capa de transporte Matrix de `agents_and_robots` (fase
|
|
posterior; v1 valida el bus de forma autónoma).
|
|
|
|
## Gotchas
|
|
|
|
- **El service NO está endurecido (v1).** No hay TLS, ni rate-limit, ni auth en
|
|
las rutas GET de lectura. Confía en la red interna. Las rutas mutantes
|
|
(`/rooms`, `/invite`, `/rekey`) sí exigen firma Ed25519 del owner sobre los
|
|
bytes canónicos de la request. Endurecer es fase posterior.
|
|
- **Identidad = secreto crítico.** El archivo de identidad (`worker.id`,
|
|
`chat.id`) contiene las claves privadas (Ed25519 + X25519). Se escribe 0600.
|
|
Perderlo = mensajes ilegibles, sin recuperación. Trátalo como una clave SSH.
|
|
- **Las rooms reciben un ULID fresco al crearse.** No hay "crear o unirse por
|
|
nombre": cada `CreateRoom` produce un room nuevo. Los peers demo cleartext
|
|
comparten el *subject* (NATS enruta por subject), así que worker→chat funcionan
|
|
aunque cada uno tenga su propio room id mapeado al mismo subject.
|
|
- **La media no viaja por el bus.** `PublishMedia` cifra, sube al object store y
|
|
publica solo un `BlobRef`. El receptor, si ve `Frame.Blob != nil`, descarga y
|
|
descifra con `FetchMedia`. El frame de media NO lleva payload inline (su nonce
|
|
vive en `BlobRef.Nonce`); `Subscribe` no intenta descifrar payloads vacíos.
|
|
- **Forward secrecy depende del rekey.** `Kick` rota `K` a un epoch nuevo y la
|
|
re-sella solo para los miembros restantes. El expulsado pierde acceso a los
|
|
mensajes publicados después del kick, pero conserva los anteriores (las claves
|
|
de epochs pasados no se borran: cifraban datos que ya podía leer).
|
|
- **NATS embebido escribe JetStream en disco.** `--nats-store` apunta a
|
|
`local_files/jetstream`; borrarlo resetea el historial persistido.
|
|
- **Build sin CGO.** Usa el driver `modernc.org/sqlite` (pure-Go) y el paquete
|
|
`cybersecurity` del registry compila limpio con `CGO_ENABLED=0`. NO requiere
|
|
`fts5` ni `gcc`.
|
|
|
|
## Convención de subjects
|
|
|
|
```
|
|
proc.<svc>.<canal> telemetría/coordinación de procesos (proc.test.ticks)
|
|
rpc.<svc> request/reply (rpc.indexer)
|
|
room.<grupo> chat humano/grupo (room.general)
|
|
agent.<nombre>.{in,out} inbox/outbox de agente LLM (agent.scout.in)
|
|
```
|
|
|
|
## Capability growth log
|
|
|
|
- v0.2.0 (2026-06-03) — el playground gana un benchmark de rendimiento
|
|
(`GET /api/bench`, SSE): un publisher inunda una room con miles de mensajes a
|
|
N subscribers y una gráfica en vivo anima el throughput. Expone las dos
|
|
políticas como flags independientes (JetStream/`Persist` y encriptación
|
|
E2E/`Encrypt`) más tamaño de payload, de modo que se mide el coste de cada
|
|
capa (core NATS vs JetStream vs E2E vs E2E+JetStream) usando la librería
|
|
cliente real, sin reimplementar nada.
|