renatocaliari/cali-starstream

StarStream: Real-time Collaboration for StarHTML

StarStream provides a "convention over configuration" approach to real-time features in StarHTML. Follow these instructions to implement synchronized user experiences.

Core Integration Strategy

When tasked with adding real-time features, always follow this workflow:

1. Identify Features: Determine if the app needs simple broadcasting, persistence (PocketBase), or conflict-free editing (Loro).
2. Initialize Core: Add StarStreamPlugin(app) to the main application file.
3. Enable Modules: Toggle enable_presence, enable_typing, enable_cursor, or enable_history as needed.
4. Register SSE: Ensure a /stream route exists to provide the EventSource connection.
5. Emit Events: Use yield elements(...) in @sse routes to auto-broadcast UI updates.

Basic Initialization

from starhtml import *
from starstream import StarStreamPlugin

app, rt = star_app()
# Use defaults for auto-detection of topics via route parameters
stream = StarStreamPlugin(app, enable_presence=True, enable_typing=True)

@rt("/chat/{room_id}")
@sse
async def chat(room_id: str, msg: str):
    # Auto-broadcasts to topic "room:{room_id}"
    yield elements(Div(msg), "#messages", "append")

Plugin Selection Guide

• StarStreamCore: Simple broadcasting, ephemeral state.
• PocketBase: Persistent data with real-time sync.
• Loro: Multi-user document editing (CRDT).

Implementation Checklist

• [ ] StarStreamPlugin initialized with app.
• [ ] SSE endpoint /stream registered.
• [ ] Frontend uses stream.get_stream_element(topic).
• [ ] Route parameters match topic conventions (e.g., {room_id} -> room:{id}).
• [ ] (Optional) Presence/Typing heartbeats implemented in frontend.

Validation & Testing

Always validate the real-time implementation using the provided test suite or by running an example:

# Run core tests
pytest tests/test_core.py tests/test_presence.py

# Verify with a live example
python examples/full_features.py

Broadcasting

Fire-and-forget broadcast to all connected clients:

@rt("/todos/add", methods=["POST"])
@sse
def add_todo(text: str):
    todos.append(text)
    
    # Broadcast to all clients
    stream.broadcast(
        elements(render_todos(), "#todo-list"),
        target="todos"
    )
    
    # Response to current client
    yield elements(render_todos(), "#todo-list")

Targets

# Topic
stream.broadcast(msg, target="chat")

# Room
stream.broadcast(msg, target="room:123")

# User
stream.broadcast(msg, target="user:456")

# Default (global)
stream.broadcast(msg)

Observability

# Metrics
stats = stream.get_metrics("todos")

# Error handling
stream.set_error_hook(lambda topic, msg, err: logger.error(f"{topic}: {err}"))

Resources

• See REFERENCE.md for detailed plugin API and advanced usage.
• Full Demo: examples/full_features.py shows all features integrated.