Chat en tiempo real con Fitz: WebSockets tipados y AsyncA...

Sumá un dashboard de clicks en vivo al acortador de URLs de la Parte 2 usando @ws("/path"), WsConn<T> y el schema AsyncAPI 3.0 que Fitz genera automáticamente. Misma auth, mismos types, mismo binario.

La promesa

En la Parte 2 construimos un acortador de URLs: endpoints HTTP, ORM Postgres, auth JWT, binario nativo. Real y shippeable, pero cada interacción es request-response. Refrescá la página de stats para ver el click nuevo. Como 1998 otra vez.

Hoy lo hacemos vivo. Cuando alguien clickea un URL corto:

El handler HTTP sigue redirigiendo e incrementando el counter (igual que antes).
Un WebSocket suscrito a /dashboard recibe un mensaje tipado con el evento de click.
El dashboard se actualiza sin polling.

El diff completo contra el código de la Parte 2: ~40 líneas. Sin nuevos installs de librería. Sin paquete websockets, sin server socketio, sin redis para pub/sub. Solo @ws sobre una función.

El modelo de WebSocket tipado

type ClickEvent {
    code: Str,
    target_url: Str,
    timestamp: Str,
}

@authenticated
@ws("/dashboard")
async fn dashboard(conn: WsConn<ClickEvent>, user: User) {
    log.info("dashboard.connected", { user_email: user.email })
    loop {
        let msg = match conn.recv() {
            Ok(m) => m,
            Err(_) => break,   // cliente desconectó
        }
        // Por ahora solo hacemos echo. En un dashboard real ignoraríamos
        // los incoming y solo pusheríamos out — `broadcast` está abajo.
        conn.send(msg)
    }
}

Tres cosas en un decorador:

@ws("/dashboard") — registrar un endpoint WebSocket en ese path.
WsConn<ClickEvent> — la conexión tipada. Cada frame in o out se marshallea como ClickEvent.
@authenticated — la auth corre antes del upgrade WebSocket. Token malo → 401, sin socket abierto, sin recursos de red gastados.

@ws entiende el auth_provider de la Parte 2. La misma función de verificación de JWT gatea el dashboard.

Auto-marshalling, en ambas direcciones

La deserialización del body HTTP de la Parte 1 — JSON type-checked, defaults aplicados, fields faltantes detectados, extras rechazados — también funciona para frames WebSocket.

Cuando el dashboard envía un frame, Fitz serializa el ClickEvent a JSON y lo envía. Cuando un frame llega, Fitz deserializa el JSON en ClickEvent y valida. Si un frame está malformado, el recv() retorna Err.

No escribís ni una sola llamada a json.dumps/json.loads. El compilador lo hizo.

En contraste: el loop típico de WebSocket en Python:

# server típico de FastAPI / websockets
@app.websocket("/dashboard")
async def dashboard(websocket: WebSocket):
    await websocket.accept()
    while True:
        try:
            data = await websocket.receive_text()
            msg = ClickEvent.model_validate_json(data)  # pydantic
        except WebSocketDisconnect:
            break
        except ValidationError as e:
            await websocket.send_text(json.dumps({"error": str(e)}))
            continue
        await websocket.send_text(msg.model_dump_json())

En Fitz:

@ws("/dashboard")
async fn dashboard(conn: WsConn<ClickEvent>) {
    loop {
        let msg = match conn.recv() {
            Ok(m) => m,
            Err(_) => break,
        }
        conn.send(msg)
    }
}

La versión Python tiene la misma lógica pero tenés que deletrearla. La versión Fitz tiene la lógica en el type.

Broadcasteando clicks en vivo

Ahora cableamos el evento de click del redirect HTTP al dashboard. El truco: broadcast sobre un WsConn<T> envía a cada conexión sobre el mismo endpoint, no solo a la que lo llamó.

@get("/{code}")
async fn redirect(db: DbConn, code: Str) -> Result<HttpResponse> {
    let link: Link = match Link.where(fn(l) => l.code == code).first(db).await {
        Ok(l) => l,
        Err(_) => return Err("not found"),
    }
    // Igual que la Parte 2: spawnear un increment background.
    spawn(increment_clicks(db, link.id))
    // Nuevo: broadcastear el click event al dashboard.
    spawn(notify_dashboard(link.code, link.target_url))
    return Ok(redirect_to(link.target_url))
}

@background
async fn notify_dashboard(code: Str, target_url: Str) {
    let event = ClickEvent {
        code: code,
        target_url: target_url,
        timestamp: now_iso(),
    }
    // El runtime mantiene el broadcaster para `/dashboard` accesible
    // desde cualquier lugar vía el handle tipado.
    ws.broadcast("/dashboard", event)
}

El handler de redirect no cambió en shape — sigue retornando el response de redirect. Sumamos un spawn. La fn background notify_dashboard llama ws.broadcast que fanout a cada conexión actualmente suscrita a /dashboard.

El broadcast es fire-and-forget. Si no hay dashboards conectados, el call es no-op. Si hay 50 dashboards conectados, los 50 reciben el evento. El runtime maneja la lista de conexiones activas.

Heartbeat, horneado

Las conexiones WebSocket mueren silenciosamente en producción. Algún proxy decide que 60 segundos de idle es demasiado, droppea la conexión TCP, y tu cliente cree que sigue conectado. Cada librería WebSocket tiene que sumar heartbeats; en Fitz es una flag sobre el server:

@server(43929, ws_heartbeat_secs=30)
fn main() => 0

Cada 30 segundos, el runtime envía un frame Ping sobre cada conexión WebSocket. Si el cliente no responde con Pong, el runtime considera la conexión muerta y la cierra limpia. La mayoría de los proxies, incluyendo Nginx y Cloudflare, aceptan esto como "todavía vivo" y no la droppean.

Seteá ws_heartbeat_secs=0 para desactivar (default es 30).

Este es el tipo de feature que nunca te tomarías el trabajo de agregar en un proyecto chico, después pasarías un domingo debuggeando cuando se rompe en producción. Default on por la misma razón que tcp_keepalive está default on.

AsyncAPI generado automático

OpenAPI describe servicios HTTP. AsyncAPI es su hermano event-driven — mismo modelo de schema, pero para WebSockets, Kafka, MQTT, etc. Fitz genera AsyncAPI 3.0 automático, de la misma forma que genera OpenAPI para HTTP:

curl http://localhost:8080/asyncapi.json

{
  "asyncapi": "3.0.0",
  "info": { "title": "shortener", "version": "0.1.0" },
  "channels": {
    "/dashboard": {
      "messages": {
        "ClickEvent": {
          "payload": {
            "type": "object",
            "properties": {
              "code": { "type": "string" },
              "target_url": { "type": "string" },
              "timestamp": { "type": "string" }
            },
            "required": ["code", "target_url", "timestamp"]
          }
        }
      }
    }
  },
  "operations": {
    "/dashboard.receive": { ... },
    "/dashboard.send": { ... }
  },
  "components": {
    "securitySchemes": {
      "bearerAuth": { "type": "http", "scheme": "bearer", "bearerFormat": "JWT" }
    }
  }
}

Este es el API de eventos completo de tu servicio. No conozco otro lenguaje que auto-genere AsyncAPI desde código tipado. El ecosistema de AsyncAPI tiene generadores de clientes (TypeScript, Java, Python), lo que significa: dropeá el schema en studio.asyncapi.com o asyncapi generate, conseguí un cliente tipado para tu front-end.

Si marcás un server con @server(docs=false), ni OpenAPI ni AsyncAPI se exponen. Los defaults están on porque el costo es chico y el valor es grande.

Un cliente browser mínimo

Un cliente vanilla JS de 30 líneas para probar el dashboard:

<!doctype html>
<input id="token" placeholder="pegá JWT token">
<button id="connect">Conectar</button>
<ul id="events"></ul>
<script>
document.getElementById("connect").onclick = () => {
    const token = document.getElementById("token").value
    const ws = new WebSocket("ws://localhost:8080/dashboard", [], {
        headers: { Authorization: `Bearer ${token}` }
    })
    // Nota: los constructores de WebSocket del browser no aceptan headers
    // custom directamente. En producción, pondrías el token en un query
    // param (?token=...) y harías que el auth_provider lo lea de ahí.
    ws.onmessage = (e) => {
        const event = JSON.parse(e.data)
        const li = document.createElement("li")
        li.textContent = `${event.timestamp} • ${event.code} → ${event.target_url}`
        document.getElementById("events").prepend(li)
    }
    ws.onerror = (e) => console.error("ws error", e)
    ws.onclose = () => console.log("desconectado")
}
</script>

Poné esto en un archivo, abrí en el browser, pegá un JWT, clickeá el botón connect. Después en otra terminal:

# Conseguí un token (del POST /login de la Parte 2)
TOKEN=$(curl -s localhost:8080/login -X POST -H 'content-type: application/json' \
    -d '{"email":"ada@example.com","password":"secret-ada-123"}' | jq -r .token)

# Creá un URL corto
curl -X POST localhost:8080/shorten -H "Authorization: Bearer $TOKEN" \
    -H 'content-type: application/json' \
    -d '{"target_url":"https://github.com/Thegreekman76/fitz"}'

# Clickeá el URL corto — la lista `events` del browser se actualiza al instante
curl -I localhost:8080/abc123

El <ul> del dashboard se popula con el click event en el momento que el redirect ocurre. El browser tenía abierto un WebSocket; el click event se broadcasteó; el callback JS lo renderizó. En vivo.

Limitaciones y trade-offs

Honestamente:

Un tipo por endpoint. WsConn<ClickEvent> significa que cada frame in y out es ClickEvent. Si necesitás que ambas direcciones sean tipos distintos (In = ChatMessage, Out = ServerEvent), el workaround es hacer el tipo una suma (estilo union) — declarar un tipo Event más ancho con fields opcionales. La solución más limpia (WsConn<In, Out> generic de dos tipos) está en el roadmap.
Sin rooms / channels adentro de un endpoint. broadcast va a cada conexión sobre el endpoint. Si querés "broadcastear solo a users suscritos al proyecto 42", mantenés un Map<Int, Vec<WsConn>> vos o partís en múltiples endpoints (uno por proyecto — funciona bien para counts bajos).
Sin reconnect con replay de estado. Si el dashboard se desconecta, al reconectar ve solo eventos nuevos — no hay "dame los últimos 30 segundos que me perdí". Construir eso necesita un event log (una tabla Postgres poleada, o Redis Streams). Afuera de la capa WebSocket.
Frames binarios son soportados vía WsConn<Bytes>, pero no hablé de ellos acá. Útiles para uploads de archivos o audio streaming; el AsyncAPI emite format: binary.

Estos son los gaps honestos. El 90% de los casos de uso de WebSocket (updates en tiempo real, chat, dashboards en vivo, edición colaborativa multi-user para un solo documento) están cubiertos hoy.

Cómo esto compone con el resto de Fitz

Podés mezclar WebSockets con todo lo demás:

Auth: @authenticated/@admin/@requires("role") funcionan sobre @ws, evaluados antes del upgrade.
Middleware: middlewares corren antes del upgrade para cosas como rate limiting por IP.
ORM: el handler WebSocket puede tomar un DbConn y queryear la base de datos.
Async: recv/send/broadcast son awaiteables; combinalos con llamadas HTTP o queries Postgres adentro del loop.
Cron / spawn: jobs cron pueden ws.broadcast("/topic", event) para pushear updates periódicos.
OpenTelemetry: el chequeo de auth antes del upgrade emite un trace span; broadcasts subsecuentes también pueden trackearse.

Mismo lenguaje, mismos types, mismo binario. El WebSocket no es un mundo separado.

Qué necesitarías después para un producto real

El dashboard de arriba alcanza para demo "clicks de URL shortener en tiempo real". Para un producto en producción extenderías con:

Filtrado por usuario: solo pushear clicks para URLs que el user creó. Trivial — chequeá user.email == link.user_email antes de broadcastear, o partí en un endpoint por user con el email en el path.
Agregación: no enviar cada click; enviar una rate de clicks por código por segundo. Mantené estado en el handler, enviá frames agregados cada 1 segundo.
Integración con framework de frontend: tipos TypeScript desde el schema AsyncAPI. Corré asyncapi generate una vez.

Nada de esto cambia la estructura del código Fitz. Son todos "editá el call site del broadcast".

Probalo

# Linux / macOS / WSL
curl -sSf https://thegreekman76.github.io/fitz/install.sh | sh

# Windows (PowerShell)
irm https://thegreekman76.github.io/fitz/install.ps1 | iex

# Reabrí la terminal, después:
git clone https://github.com/Thegreekman76/fitz.git
cd fitz/boilerplates/api-websocket

# Leé el README, corré con docker compose o `fitz dev`
docker compose up

Para VSCode (recomendado — hover sobre WsConn<T>, autocomplete en conn.send/recv/broadcast): bajá el fitz-lang-<plataforma>.vsix desde la página de releases y code --install-extension fitz-lang-<plataforma>.vsix --force. El Language Server viene incluido.

El boilerplate api-websocket es un servidor de chat tipado. Más o menos el mismo shape que el dashboard de arriba — loop { recv; broadcast } con auth.

Para el shortener completo + dashboard en vivo, el boilerplate api-orm-full tiene la cosa entera ensamblada (rutas HTTP + ORM + dashboard WebSocket + cron job + auth JWT) en ~250 líneas total.

Repo: github.com/Thegreekman76/fitz
Boilerplate api-websocket: github.com/Thegreekman76/fitz/tree/main/boilerplates/api-websocket
Boilerplate api-orm-full (la cosa entera): github.com/Thegreekman76/fitz/tree/main/boilerplates/api-orm-full
Docs y curso: thegreekman76.github.io/fitz
Capítulo de la guía sobre WebSockets: thegreekman76.github.io/fitz/guide/#29-websockets
Roadmap: docs/roadmap.md
CHANGELOG: CHANGELOG.md
Issues: github.com/Thegreekman76/fitz/issues

Si construís algo con esto, escribime. Lo quiero ver.

Hasta la próxima.

推荐订阅源

DEV Community