Real-time Updates

WebSocket for live order status. Streaming data to clients.

The Need

Sandy’s customers want live order tracking: “Your BLT is being prepared…”

WebSocket Adapter

from evoid.adapters.websocket import create_ws_app, on

ws_app = create_ws_app(name="sandy-ws")

async def handle_connect(ctx):
    return {"status": "connected"}

async def handle_message(ctx):
    data = ctx.intent.metadata.get("data", {})
    if data.get("type") == "track_order":
        order_id = data.get("order_id")
        # Send live updates
        return {"order_id": order_id, "status": "preparing"}
    return {"error": "unknown message type"}

async def handle_disconnect(ctx):
    return {"status": "disconnected"}

on(ws_app, "connect", handle_connect)
on(ws_app, "message", handle_message)
on(ws_app, "disconnect", handle_disconnect)

Server-Sent Events (SSE)

For simpler streaming:

from evoid import Intent, Level, add_intent

STREAM_ORDERS = Intent(
    name="stream_orders",
    level=Level.STANDARD,
)

async def handle_stream(ctx) -> list[dict]:
    """Return a list of events — adapter streams them."""
    return [
        {"event": "order_update", "data": {"id": 1, "status": "preparing"}},
        {"event": "order_update", "data": {"id": 1, "status": "ready"}},
    ]

add_intent(STREAM_ORDERS, handle_stream)

Custom SSE Adapter

import json
from starlette.applications import Starlette
from starlette.responses import StreamingResponse
from starlette.routing import Route
from evoid.core import Intent, Level, execute

async def handle_sse(request):
    async def generate():
        result = await execute(Intent(
            name="stream_orders",
            level=Level.STANDARD,
        ))
        if result.success:
            for event in result.value:
                yield f"data: {json.dumps(event)}\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

app = Starlette(routes=[
    Route("/events", handle_sse, methods=["GET"]),
])

What You Learned

Beyond Sandy: Game Integration

The same real-time patterns power game servers. The evoid-godot and evoid-transport plugins bring IOP to game development:

# Game client sends: EvoidApp.send_intent("player_move", {"x": 10, "y": 20})
# Server receives it as an Intent:

PLAYER_MOVE = Intent(
    name="game:my-game:player_move",
    level=Level.EPHEMERAL,  # Position updates are disposable
    metadata={"player_id": "abc", "x": 10, "y": 20},
)
# Pipeline: validate → handler (5s)
# Fast, no auth, no audit. Next frame corrects any errors.

# But a purchase? That's CRITICAL:
PURCHASE_ITEM = Intent(
    name="game:my-game:purchase_item",
    level=Level.CRITICAL,
    metadata={"player_id": "abc", "item": "sword", "price": 9.99},
)
# Pipeline: validate → authorize → audit → protect → handler (30s)
# Real money, full protection.

# The evoid-transport plugin picks the right channel:
# Channel 0 (RELIABLE) — card plays, purchases (CRITICAL)
# Channel 1 (UNRELIABLE) — position updates (EPHEMERAL)
# Channel 2 (CHAT) — chat messages (STANDARD)

# Same IOP concept: level determines infrastructure.
# Game state uses EPHEMERAL for speed.
# Game purchases use CRITICAL for safety.
```

Next: Plugin System

Let’s extend EVOID with plugins — Plugin System.