orchestrata_ultimate.session.sync

Reference › orchestrata_ultimate › session › sync

SessionSync copies session data from the Redis hot store to the graph cold store. graph_available checks that the GraphSessionStore is present and connected; when it is not, flush_session returns False and sync_facts returns 0, so the sync degrades silently instead of failing. flush_session persists the session metadata and all its facts from the RedisSessionStore in one batch, and records the elapsed time. sync_facts(session_id, offset) persists only the facts from an offset onward in batches of SYNC_BATCH_SIZE. full_sync syncs every active session, link_sessions creates the succession link between two sessions, and sync_count and last_sync_ms report counters.

At a glance

Class diagram

Dependency diagram

API

Classes

SessionSync

Decorators: traces(spec='006', story='FR-007', reqs=['REQ-006-FR12-003', 'REQ-006-FR7-006']) · Line 21

Synchronizes session data from Redis to graph.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/session/sync.py

# mypy: ignore-errors
"""Redis to graph synchronization.

Flushes session data from Redis hot storage to graph cold storage.
"""

# Spec: 006 - Session Management

import logging
import time

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.session.constants import SYNC_BATCH_SIZE
from orchestrata_ultimate.session.graph_store import GraphSessionStore
from orchestrata_ultimate.session.redis_store import RedisSessionStore

logger = logging.getLogger(__name__)


@traces(spec="006", story="FR-007", reqs=["REQ-006-FR12-003", "REQ-006-FR7-006"])
class SessionSync:
    """Synchronizes session data from Redis to graph."""

    def __init__(
        self,
        redis_store: RedisSessionStore,
        graph_store: GraphSessionStore | None = None,
    ) -> None:
        self._redis = redis_store
        self._graph = graph_store
        self._sync_count = 0
        self._last_sync_ms = 0.0

    @property
    def graph_available(self) -> bool:
        """Check if graph backend is available."""
        return self._graph is not None and self._graph.connected

    @traces(spec="006", story="FR-012", reqs=["REQ-006-FR12-003"])
    def flush_session(self, session_id: str) -> bool:
        """Flush complete session data to graph."""
        graph = self._graph
        if graph is None or not graph.connected:
            return False

        start = time.monotonic()

        # Persist session metadata
        session_data = self._redis.get_session(session_id)
        if session_data:
            graph.persist_session(session_data)

        # Persist all facts
        facts = self._redis.get_facts(session_id)
        if facts:
            graph.persist_facts_batch(session_id, facts)

        elapsed_ms = (time.monotonic() - start) * 1000
        self._last_sync_ms = elapsed_ms
        self._sync_count += 1

        logger.info(
            "Flushed session %s to graph: %d facts in %.1fms",
            session_id,
            len(facts),
            elapsed_ms,
        )
        return True

    @traces(
        spec="006",
        story="FR-012",
        reqs=["REQ-006-FR12-003", "REQ-006-US-004-005"],
    )
    def sync_facts(self, session_id: str, offset: int = 0) -> int:
        """Sync facts from Redis to graph in batches."""
        graph = self._graph
        if graph is None or not graph.connected:
            return 0

        all_facts = self._redis.get_facts(session_id)
        to_sync = all_facts[offset:]

        synced = 0
        for i in range(0, len(to_sync), SYNC_BATCH_SIZE):
            batch = to_sync[i : i + SYNC_BATCH_SIZE]
            count = graph.persist_facts_batch(session_id, batch)
            synced += count

        self._sync_count += 1
        return synced

    def full_sync(self) -> int:
        """Sync all active sessions to graph."""
        if not self.graph_available:
            return 0

        active_ids = self._redis.get_active_sessions()
        total = 0
        for session_id in active_ids:
            if self.flush_session(session_id):
                total += 1
        return total

    def link_sessions(self, prev_id: str, next_id: str) -> bool:
        """Create session succession link in graph."""
        graph = self._graph
        if graph is None or not graph.connected:
            return False
        return graph.link_sessions(prev_id, next_id)

    @property
    @traces(spec="006", story="FR-012", reqs=["REQ-006-FR12-003"])
    def sync_count(self) -> int:
        """Total sync operations performed."""
        return self._sync_count

    @property
    @traces(spec="006", story="FR-012", reqs=["REQ-006-FR12-003"])
    def last_sync_ms(self) -> float:
        """Latency of last sync operation."""
        return self._last_sync_ms