orchestrata_ultimate.gateway.handlers.state

Reference › orchestrata_ultimate › gateway › handlers › state

StateHandler serves the client state-sync RPCs state.getAll, state.putAll, state.put and state.delete, used by the VizInt Command Center to keep UI state on the server. It borrows the Redis client from the ring_buffer and stores each connection's state as a Redis hash under a key built by _state_redis_key from the connection id. Values are saved as JSON with a timestamp. handle_get_all returns the entries, decoding JSON where possible; handle_put requires a key and raises RPCError with code -32602 when it is missing; handle_put_all and handle_delete follow the same pattern. When Redis is unavailable, reads return empty entries and writes return an error status. A Redis failure during a write raises RPCError. rpc wires it. See gateway.

At a glance

Class diagram

Dependency diagram

API

Classes

StateHandler

Decorators: traces(spec='016', story='State Sync', reqs=['REQ-016-274']) · Line 21

Redis-backed per-client state sync methods.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/gateway/handlers/state.py

# mypy: ignore-errors
"""State sync RPC handlers -- getAll, putAll, put, delete.

Implements SPEC-014 state synchronisation methods for UI.
"""

import json
import logging
import time
from collections.abc import Awaitable
from typing import Any, cast

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.gateway.exceptions import RPCError
from orchestrata_ultimate.gateway.ring_buffer import RedisRingBuffer

logger = logging.getLogger(__name__)


@traces(spec="016", story="State Sync", reqs=["REQ-016-274"])
class StateHandler:
    """Redis-backed per-client state sync methods."""

    def __init__(self, ring_buffer: RedisRingBuffer) -> None:
        """Wire handler to state subsystems.

        Args:
            ring_buffer: Redis ring buffer (used for client access).
        """
        self._ring_buffer = ring_buffer

    async def handle_get_all(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle state.getAll -- get all state from Redis.

        Args:
            connection_id: Caller connection ID.
            params: Ignored.

        Returns:
            All key-value entries for this client.
        """
        redis_client = self._ring_buffer.client
        if not redis_client:
            return {"entries": {}}
        key = _state_redis_key(connection_id)
        try:
            # redis-py's command mixins are shared between the sync and
            # async clients, so `hgetall()` is stub-typed `Awaitable[T] | T`;
            # `redis_client` here is always the async client, so it always
            # actually returns an awaitable.
            raw = await cast("Awaitable[dict[str, str]]", redis_client.hgetall(key))
            entries: dict[str, Any] = {}
            for k, v in raw.items():
                try:
                    entries[k] = json.loads(v)
                except json.JSONDecodeError, TypeError:
                    entries[k] = v
            return {"entries": entries}
        except Exception:
            logger.warning("state.getAll failed")
            return {"entries": {}}

    async def handle_put_all(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle state.putAll -- set multiple keys.

        Args:
            connection_id: Caller connection ID.
            params: entries dict of key -> {value, timestamp}.

        Returns:
            Status with count of keys set.
        """
        entries = params.get("entries", {})
        if not entries:
            return {"status": "ok", "count": 0}
        redis_client = self._ring_buffer.client
        if not redis_client:
            return {
                "status": "error",
                "error": "Redis unavailable",
            }
        key = _state_redis_key(connection_id)
        mapping: dict[str, str] = {}
        for k, v in entries.items():
            mapping[k] = json.dumps(v)
        try:
            # Stub deficiency (see handle_get_all's hgetall cast).
            await cast("Awaitable[int]", redis_client.hset(key, mapping=mapping))
            return {"status": "ok", "count": len(mapping)}
        except Exception as exc:
            raise RPCError(f"state.putAll failed: {exc}")

    async def handle_put(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle state.put -- set one key.

        Args:
            connection_id: Caller connection ID.
            params: key, value, optional timestamp.

        Returns:
            Status confirmation.
        """
        state_key = params.get("key", "")
        if not state_key:
            raise RPCError(
                "Missing 'key' parameter",
                rpc_code=-32602,
            )
        value = params.get("value")
        redis_client = self._ring_buffer.client
        if not redis_client:
            return {
                "status": "error",
                "error": "Redis unavailable",
            }
        key = _state_redis_key(connection_id)
        payload = json.dumps(
            {
                "value": value,
                "timestamp": params.get("timestamp", time.time()),
            }
        )
        try:
            # Stub deficiency (see handle_get_all's hgetall cast).
            await cast("Awaitable[int]", redis_client.hset(key, state_key, payload))
            return {"status": "ok", "key": state_key}
        except Exception as exc:
            raise RPCError(f"state.put failed: {exc}")

    async def handle_delete(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle state.delete -- delete one key.

        Args:
            connection_id: Caller connection ID.
            params: key to delete.

        Returns:
            Status confirmation.
        """
        state_key = params.get("key", "")
        if not state_key:
            raise RPCError(
                "Missing 'key' parameter",
                rpc_code=-32602,
            )
        redis_client = self._ring_buffer.client
        if not redis_client:
            return {
                "status": "error",
                "error": "Redis unavailable",
            }
        key = _state_redis_key(connection_id)
        try:
            # Stub deficiency (see handle_get_all's hgetall cast).
            await cast("Awaitable[int]", redis_client.hdel(key, state_key))
            return {"status": "ok", "key": state_key}
        except Exception as exc:
            raise RPCError(f"state.delete failed: {exc}")


def _state_redis_key(connection_id: str) -> str:
    """Build the Redis hash key for state sync.

    Args:
        connection_id: Client connection identifier.

    Returns:
        Redis key string.
    """
    return f"orchestrata:commandcenter:state:{connection_id}"