stonks-oracle/services/shared/redis_keys.py

"""Redis key conventions and queue abstractions."""

import os

# --- Key prefixes ---
_STAGE = os.getenv("DEPLOY_STAGE", "")
PREFIX = f"stonks:{_STAGE}" if _STAGE else "stonks"

# Distributed locks
LOCK_PREFIX = f"{PREFIX}:lock"

# Rate limit counters
RATE_LIMIT_PREFIX = f"{PREFIX}:ratelimit"

# Job queues
QUEUE_PREFIX = f"{PREFIX}:queue"

# Dedupe markers
DEDUPE_PREFIX = f"{PREFIX}:dedupe"

# Cache
CACHE_PREFIX = f"{PREFIX}:cache"

# Retry backoff state
RETRY_PREFIX = f"{PREFIX}:retry"


def lock_key(resource: str) -> str:
    return f"{LOCK_PREFIX}:{resource}"


def rate_limit_key(source: str, window: str) -> str:
    return f"{RATE_LIMIT_PREFIX}:{source}:{window}"


def queue_key(queue_name: str) -> str:
    return f"{QUEUE_PREFIX}:{queue_name}"


def dedupe_key(content_hash: str) -> str:
    return f"{DEDUPE_PREFIX}:{content_hash}"


def cache_key(namespace: str, key: str) -> str:
    return f"{CACHE_PREFIX}:{namespace}:{key}"


def retry_key(job_id: str) -> str:
    return f"{RETRY_PREFIX}:{job_id}"


# Dead-letter queues
DLQ_PREFIX = f"{PREFIX}:dlq"


def dlq_key(queue_name: str) -> str:
    """Return the dead-letter queue key for a given source queue."""
    return f"{DLQ_PREFIX}:{queue_name}"


# --- Queue names ---
QUEUE_INGESTION = "ingestion"
QUEUE_PARSING = "parsing"
QUEUE_EXTRACTION = "extraction"
QUEUE_AGGREGATION = "aggregation"
QUEUE_RECOMMENDATION = "recommendation"
QUEUE_LAKE_PUBLISH = "lake_publish"
QUEUE_TRADE = "trade"
QUEUE_BROKER = "broker_orders"
QUEUE_MACRO_CLASSIFICATION = "macro_classification"
QUEUE_REPORT_GENERATION = "report_generation"
QUEUE_REPORT_GENERATION = "report_generation"

# --- Trading engine ---
QUEUE_TRADING_DECISIONS = "trading_decisions"
TRADING_DEDUPE_PREFIX = f"{PREFIX}:dedupe:trading"
TRADING_CB_PREFIX = f"{PREFIX}:trading:circuit_breaker"
TRADING_NOTIFICATION_RATE = f"{PREFIX}:trading:notification_rate"


def trading_dedupe_key(recommendation_id: str) -> str:
    """Return the deduplication key for a trading recommendation."""
    return f"{TRADING_DEDUPE_PREFIX}:{recommendation_id}"


def trading_cb_key(trigger_type: str) -> str:
    """Return the circuit breaker state key for a given trigger type."""
    return f"{TRADING_CB_PREFIX}:{trigger_type}"


def trading_notification_rate_key(channel: str) -> str:
    """Return the notification rate-limit key for a given channel."""
    return f"{TRADING_NOTIFICATION_RATE}:{channel}"


# --- Pipeline toggle ---
PIPELINE_ENABLED_KEY = f"{PREFIX}:pipeline:enabled"


async def is_pipeline_enabled(rds: "redis.asyncio.Redis") -> bool:  # type: ignore[name-defined]  # noqa: F821
    """Check whether the pipeline is enabled via the Redis toggle.

    Returns True (enabled) when the key is absent or set to anything
    other than ``"0"``.  Workers should call this at the top of each
    loop iteration and sleep when it returns False.
    """
    val = await rds.get(PIPELINE_ENABLED_KEY)
    return val != "0"