"""Small, optional Redis cache for the homepage agent and skill catalogues.

MySQL remains the source of truth. Redis can use the dedicated
``api_read_cache`` connection, or reuse ``concurrency.redis`` when explicitly
enabled; failures and missing configuration fall through to the normal
MySQL-backed request path.
"""

from __future__ import annotations

import asyncio
import hashlib
import json
import logging
import secrets
import time
from typing import Any
from weakref import WeakKeyDictionary

from app.gateway.api_cache_catalog import API_CACHE_CATALOG, API_READ_CACHE_COLLECTIONS

logger = logging.getLogger(__name__)

_CACHE_SUFFIX = "api_read_cache:v1"
_DEFAULT_CACHE_TTL_SECONDS = 300
_MIN_CACHE_TTL_SECONDS = 1
_MAX_CACHE_TTL_SECONDS = 3600
_clients: dict[tuple[str, str, int, int], Any] = {}
_last_warning_at: dict[str, float] = {}
_VERSION_UNSET = object()
_api_read_cache_settings_cache: tuple[float, int | None, bool, int, dict[str, bool], bool, str] | None = None
_enabled_cache_ttl_seconds = 2.0
_save_locks: WeakKeyDictionary[asyncio.AbstractEventLoop, asyncio.Lock] = WeakKeyDictionary()
_enabled_locks: dict[tuple[int, int], asyncio.Lock] = {}
_fill_events: dict[tuple[int, str], asyncio.Event] = {}
_fill_local_payloads: dict[tuple[int, str], tuple[float, str, bytes]] = {}


def _redis_settings(config: Any, *, enabled: bool | None = None) -> tuple[str, str | None, str] | None:
    concurrency = getattr(config, "concurrency", None)
    concurrency_redis = getattr(concurrency, "redis", None)
    api_cache = getattr(config, "api_read_cache", None)
    api_enabled = bool(getattr(api_cache, "enabled", False)) if enabled is None else enabled
    concurrency_enabled = bool(getattr(concurrency_redis, "enabled", False))
    if not api_enabled:
        return None
    url = str(getattr(api_cache, "url", None) or (getattr(concurrency_redis, "url", None) if concurrency_enabled else None) or "").strip()
    if not url:
        return None
    namespace = str(getattr(api_cache, "namespace", "cmzs_deerflow:homepage") if api_enabled else f"{getattr(concurrency_redis, 'namespace', 'cmzs_deerflow:concurrency')}:homepage")
    password = getattr(api_cache, "password", None)
    if not getattr(api_cache, "url", None) and concurrency_enabled:
        password = password or getattr(concurrency_redis, "password", None)
    return url, password, namespace.strip().rstrip(":")


def _client(config: Any, *, enabled: bool | None = None, epoch: str | None = None) -> tuple[Any, str] | None:
    settings = _redis_settings(config, enabled=enabled)
    if settings is None:
        return None
    url, password, namespace = settings
    api_cache = getattr(config, "api_read_cache", None)
    redis_config = getattr(getattr(config, "concurrency", None), "redis", None)
    timeout_config = api_cache if (enabled if enabled is not None else getattr(api_cache, "enabled", False)) else redis_config
    connect_timeout_ms = int(getattr(timeout_config, "socket_connect_timeout_ms", 250) or 250)
    socket_timeout_ms = int(getattr(timeout_config, "socket_timeout_ms", 750) or 750)
    identity = (url, str(password or ""), connect_timeout_ms, socket_timeout_ms)
    client = _clients.get(identity)
    if client is None:
        try:
            from redis.asyncio import Redis
        except ImportError:
            _warn_once(namespace, "redis package is not installed")
            return None
        client = Redis.from_url(
            url,
            password=password,
            socket_connect_timeout=max(0.001, connect_timeout_ms / 1000),
            socket_timeout=max(0.001, socket_timeout_ms / 1000),
            decode_responses=False,
            # Homepage bursts commonly arrive in parallel (e.g. 70 users).
            # 70 users can load several homepage APIs at once; keep enough
            # headroom for the combined burst instead of queuing Redis reads.
            max_connections=256,
        )
        _clients[identity] = client
    return client, f"{namespace}:{_CACHE_SUFFIX}:epoch:{_cache_epoch(config) if epoch is None else epoch}"


def _cache_epoch(config: Any) -> str:
    current = _api_read_cache_settings_cache
    if current is not None and current[1] in (None, id(config)):
        return current[6]
    return "0"


def _scoped_version(epoch: str, generation: str) -> str:
    # Initial deployments keep the original generation value; later saves use
    # an opaque epoch marker so an in-flight old fill cannot enter a new cache.
    return generation if epoch == "0" else f"{epoch}.{generation}"


async def get_api_read_cache_settings(
    config: Any,
    *,
    force_refresh: bool = False,
) -> tuple[bool, int]:
    """Read the admin toggle and TTL from MySQL, refreshing at most every 2s.

    The YAML flag is retained as the compatibility default for deployments that
    have not yet created the settings row. Once an administrator saves the
    setting, MySQL is authoritative across restarts and worker containers.
    """

    global _api_read_cache_settings_cache
    now = time.monotonic()
    if not force_refresh and _api_read_cache_settings_cache is not None and now - _api_read_cache_settings_cache[0] < _enabled_cache_ttl_seconds and _api_read_cache_settings_cache[1] in (None, id(config)):
        return _api_read_cache_settings_cache[2], _api_read_cache_settings_cache[3]
    loop = asyncio.get_running_loop()
    lock_key = (id(loop), id(config))
    setting_lock = _enabled_locks.get(lock_key)
    if setting_lock is None:
        setting_lock = _enabled_locks[lock_key] = asyncio.Lock()
        if len(_enabled_locks) > 64:
            _enabled_locks.pop(next(iter(_enabled_locks)))
    async with setting_lock:
        now = time.monotonic()
        if not force_refresh and _api_read_cache_settings_cache is not None and now - _api_read_cache_settings_cache[0] < _enabled_cache_ttl_seconds and _api_read_cache_settings_cache[1] in (None, id(config)):
            return _api_read_cache_settings_cache[2], _api_read_cache_settings_cache[3]
        configured_default = bool(getattr(getattr(config, "api_read_cache", None), "enabled", False))
        enabled = configured_default
        ttl_seconds = _cache_ttl_seconds(config)
        collection_overrides: dict[str, bool] = {}
        browser_status_enabled = True
        cache_epoch = "0"
        try:
            from deerflow.persistence.api_read_cache_settings.model import ApiReadCacheSettingRow
            from deerflow.persistence.engine import get_session_factory

            session_factory = get_session_factory()
            if session_factory is not None:
                async with session_factory() as session:
                    row = await session.get(ApiReadCacheSettingRow, "default")
                if row is not None:
                    enabled = bool(row.enabled)
                    ttl_seconds = _cache_ttl_seconds(
                        config,
                        getattr(row, "ttl_seconds", ttl_seconds),
                    )
                    collection_overrides = dict(getattr(row, "collection_overrides", None) or {})
                    browser_status_enabled = bool(getattr(row, "browser_status_enabled", True))
                    cache_epoch = str(getattr(row, "cache_epoch", None) or "0")
        except Exception as exc:
            _warn_once("api_read_cache_settings", f"setting read failed ({type(exc).__name__})")
            if _api_read_cache_settings_cache is not None and _api_read_cache_settings_cache[1] in (None, id(config)):
                enabled = _api_read_cache_settings_cache[2]
                ttl_seconds = _api_read_cache_settings_cache[3]
                collection_overrides = _api_read_cache_settings_cache[4]
                browser_status_enabled = _api_read_cache_settings_cache[5]
                cache_epoch = _api_read_cache_settings_cache[6]
        if _api_read_cache_settings_cache is not None and _api_read_cache_settings_cache[0] > now and _api_read_cache_settings_cache[1] in (None, id(config)):
            # An administrator saved while this database read was in flight.
            # Do not replace the just-saved switch/epoch with its older row.
            return _api_read_cache_settings_cache[2], _api_read_cache_settings_cache[3]
        _api_read_cache_settings_cache = (now, id(config), enabled, ttl_seconds, collection_overrides, browser_status_enabled, cache_epoch)
        return enabled, ttl_seconds


async def is_api_read_cache_enabled(config: Any, *, force_refresh: bool = False) -> bool:
    return (await get_api_read_cache_settings(config, force_refresh=force_refresh))[0]


async def is_collection_cache_enabled(config: Any, collection: str) -> bool:
    enabled, _ = await get_api_read_cache_settings(config)
    return enabled and bool(_api_read_cache_settings_cache[4].get(collection, True))


async def is_browser_status_enabled(config: Any) -> bool:
    await get_api_read_cache_settings(config)
    return _api_read_cache_settings_cache[5]


async def get_api_cache_controls(config: Any) -> dict[str, Any]:
    _, ttl = await get_api_read_cache_settings(config)
    overrides = _api_read_cache_settings_cache[4]
    return {
        "browser_status_enabled": _api_read_cache_settings_cache[5],
        "collections": [{"id": key, "label": label, "paths": paths, "scope": scope, "enabled": overrides.get(key, True), "ttl_seconds": ttl} for key, label, paths, scope in API_CACHE_CATALOG],
    }


def set_api_read_cache_settings_local(
    enabled: bool,
    ttl_seconds: int,
    *,
    collection_overrides: dict[str, bool] | None = None,
    browser_status_enabled: bool | None = None,
    cache_epoch: str | None = None,
) -> None:
    """Make just-saved settings effective in this Gateway process immediately."""

    global _api_read_cache_settings_cache
    previous = _api_read_cache_settings_cache
    _api_read_cache_settings_cache = (
        time.monotonic(),
        None,
        bool(enabled),
        _cache_ttl_seconds(None, ttl_seconds),
        dict(collection_overrides if collection_overrides is not None else (previous[4] if previous else {})),
        browser_status_enabled if browser_status_enabled is not None else (previous[5] if previous else True),
        cache_epoch if cache_epoch is not None else (previous[6] if previous else "0"),
    )
    if previous is not None and previous[6] != _api_read_cache_settings_cache[6]:
        # Wake old-generation waiters immediately; their Redis locks naturally
        # expire and cannot block reads in the new persisted epoch.
        for event in _fill_events.values():
            event.set()
        _fill_events.clear()
        _fill_local_payloads.clear()


def set_api_read_cache_enabled_local(enabled: bool) -> None:
    """Backward-compatible local switch update that retains the cached TTL."""

    ttl_seconds = _api_read_cache_settings_cache[3] if _api_read_cache_settings_cache is not None else _DEFAULT_CACHE_TTL_SECONDS
    set_api_read_cache_settings_local(enabled, ttl_seconds)


async def save_api_read_cache_settings(
    enabled: bool | None,
    ttl_seconds: int | None = None,
    *,
    config: Any = None,
    collection_overrides: dict[str, bool] | None = None,
    browser_status_enabled: bool | None = None,
) -> tuple[bool, int]:
    """Serialize same-process patches; row locks protect concurrent DB workers."""

    loop = asyncio.get_running_loop()
    lock = _save_locks.setdefault(loop, asyncio.Lock())
    async with lock:
        return await _persist_api_read_cache_settings(
            enabled,
            ttl_seconds,
            config=config,
            collection_overrides=collection_overrides,
            browser_status_enabled=browser_status_enabled,
        )


async def _persist_api_read_cache_settings(
    enabled: bool | None,
    ttl_seconds: int | None,
    *,
    config: Any,
    collection_overrides: dict[str, bool] | None,
    browser_status_enabled: bool | None,
) -> tuple[bool, int]:
    """Persist the administrator toggle and optional TTL in the business DB."""

    from deerflow.persistence.api_read_cache_settings.model import ApiReadCacheSettingRow
    from deerflow.persistence.engine import get_session_factory

    session_factory = get_session_factory()
    if session_factory is None:
        raise RuntimeError("MySQL persistence is required to save the Redis cache setting")
    if collection_overrides is not None and set(collection_overrides) - set(API_READ_CACHE_COLLECTIONS):
        raise ValueError("Unknown API cache collection")
    from sqlalchemy.exc import IntegrityError

    for attempt in range(2):
        try:
            async with session_factory() as session:
                row = await session.get(ApiReadCacheSettingRow, "default", with_for_update=True)
                if row is None:
                    saved_enabled = bool(getattr(getattr(config, "api_read_cache", None), "enabled", False)) if enabled is None else enabled
                    saved_ttl = _cache_ttl_seconds(config, ttl_seconds)
                    row = ApiReadCacheSettingRow(
                        key="default",
                        enabled=saved_enabled,
                        ttl_seconds=saved_ttl,
                        collection_overrides={},
                        browser_status_enabled=True,
                    )
                    session.add(row)
                else:
                    if enabled is not None:
                        row.enabled = bool(enabled)
                    saved_ttl = _cache_ttl_seconds(
                        config,
                        ttl_seconds if ttl_seconds is not None else getattr(row, "ttl_seconds", None),
                    )
                    row.ttl_seconds = saved_ttl
                saved_enabled = bool(row.enabled)
                saved_overrides = {**(getattr(row, "collection_overrides", None) or {}), **(collection_overrides or {})}
                saved_browser_status = browser_status_enabled if browser_status_enabled is not None else bool(getattr(row, "browser_status_enabled", True))
                row.collection_overrides = saved_overrides
                row.browser_status_enabled = saved_browser_status
                saved_epoch = secrets.token_hex(16)
                row.cache_epoch = saved_epoch
                await session.commit()
            break
        except IntegrityError:
            # Another worker may have inserted the singleton settings row.
            # Reload it under the row lock and merge this patch exactly once.
            if attempt:
                raise
    set_api_read_cache_settings_local(saved_enabled, saved_ttl, collection_overrides=saved_overrides, browser_status_enabled=saved_browser_status, cache_epoch=saved_epoch)
    return saved_enabled, saved_ttl


async def save_api_read_cache_enabled(enabled: bool) -> None:
    """Backward-compatible switch update; preserves an existing TTL."""

    await save_api_read_cache_settings(enabled)


def api_read_cache_redis_configured(config: Any) -> bool:
    return _redis_settings(config, enabled=True) is not None


def _cache_ttl_seconds(config: Any, ttl_seconds: int | None = None) -> int:
    value = int(ttl_seconds if ttl_seconds is not None else getattr(getattr(config, "api_read_cache", None), "ttl_seconds", _DEFAULT_CACHE_TTL_SECONDS) or _DEFAULT_CACHE_TTL_SECONDS)
    return max(_MIN_CACHE_TTL_SECONDS, min(_MAX_CACHE_TTL_SECONDS, value))


def _redis_operation_timeout_seconds(config: Any) -> float:
    cache = getattr(config, "api_read_cache", None)
    connect_ms = int(getattr(cache, "socket_connect_timeout_ms", 250) or 250)
    socket_ms = int(getattr(cache, "socket_timeout_ms", 750) or 750)
    # Keep Redis optional and bounded, while allowing a private-network cache
    # to absorb short bursts before falling back to MySQL.
    return min(1.0, max(0.15, max(connect_ms, socket_ms) * 1.25 / 1000))


def _max_cache_value_bytes(config: Any) -> int:
    value = int(getattr(getattr(config, "api_read_cache", None), "max_value_bytes", 512 * 1024) or 512 * 1024)
    return max(1024, min(5 * 1024 * 1024, value))


def _warn_once(key: str, message: str) -> None:
    now = time.monotonic()
    if now - _last_warning_at.get(key, 0.0) >= 60:
        _last_warning_at[key] = now
        logger.warning("Homepage Redis cache unavailable; using MySQL: %s", message)


def _version_key(prefix: str, collection: str) -> str:
    return f"{prefix}:{collection}:version"


def _item_key(prefix: str, collection: str, version: str, logical_key: str) -> str:
    digest = hashlib.sha256(logical_key.encode("utf-8")).hexdigest()
    return f"{prefix}:{collection}:v{version}:{digest}"


async def get_cached_json(
    config: Any,
    collection: str,
    logical_key: str,
    *,
    version: str | None | object = _VERSION_UNSET,
) -> Any | None:
    raw = await get_cached_json_payload(config, collection, logical_key, version=version)
    if raw is None:
        return None
    try:
        return json.loads(raw)
    except Exception as exc:
        _warn_once(f"{collection}:decode", f"cached JSON decode failed ({type(exc).__name__})")
        return None


async def get_cached_json_payload(
    config: Any,
    collection: str,
    logical_key: str,
    *,
    version: str | None | object = _VERSION_UNSET,
) -> bytes | None:
    """Return the stored JSON bytes so HTTP routes can bypass revalidation."""

    if not await is_collection_cache_enabled(config, collection):
        return None
    client_info = _client(config, enabled=True)
    if client_info is None:
        return None
    client, prefix = client_info
    try:
        if version is _VERSION_UNSET:
            version = await get_cache_version(config, collection)
        if version is None:
            return None
        async with asyncio.timeout(_redis_operation_timeout_seconds(config)):
            raw = await client.get(_item_key(prefix, collection, str(version), logical_key))
        if raw is None:
            return None
        return raw if isinstance(raw, bytes) else str(raw).encode("utf-8")
    except Exception as exc:  # Redis is an optimization; MySQL remains authoritative.
        _warn_once(prefix, f"read failed ({type(exc).__name__})")
        return None


async def get_cache_version(config: Any, collection: str) -> str | None:
    """Read the collection generation used to guard against stale writes."""

    if not await is_collection_cache_enabled(config, collection):
        return None
    client_info = _client(config, enabled=True)
    if client_info is None:
        return None
    client, prefix = client_info
    epoch = _cache_epoch(config)
    try:
        async with asyncio.timeout(_redis_operation_timeout_seconds(config)):
            version = await client.get(_version_key(prefix, collection))
        generation = version.decode("ascii") if isinstance(version, bytes) else str(version or "0")
        return _scoped_version(epoch, generation)
    except Exception as exc:
        _warn_once(prefix, f"version read failed ({type(exc).__name__})")
        return None


async def wait_for_cached_json_or_claim_fill(
    config: Any,
    collection: str,
    logical_key: str,
    *,
    version: str | None,
    wait_ms: int = 1000,
) -> tuple[Any | None, str | None]:
    """Wait briefly for a concurrent miss to fill Redis, or claim the fill.

    The returned token must be passed to :func:`release_cache_fill` in a
    ``finally`` block. This prevents a burst of identical requests from all
    rebuilding the same large agent/skill response at once.
    """

    return await _wait_for_cached_value_or_claim_fill(
        config,
        collection,
        logical_key,
        version=version,
        wait_ms=wait_ms,
        raw_payload=False,
    )


async def wait_for_cached_json_payload_or_claim_fill(
    config: Any,
    collection: str,
    logical_key: str,
    *,
    version: str | None,
    wait_ms: int = 6000,
) -> tuple[bytes | None, str | None]:
    """Wait for an in-flight fill and return the cached HTTP JSON bytes."""

    value, token = await _wait_for_cached_value_or_claim_fill(
        config,
        collection,
        logical_key,
        version=version,
        wait_ms=wait_ms,
        raw_payload=True,
    )
    return value, token


async def _wait_for_cached_value_or_claim_fill(
    config: Any,
    collection: str,
    logical_key: str,
    *,
    version: str | None,
    wait_ms: int,
    raw_payload: bool,
) -> tuple[Any | bytes | None, str | None]:
    if version is None or not await is_collection_cache_enabled(config, collection):
        return None, None
    client_info = _client(config, enabled=True)
    if client_info is None:
        return None, None
    client, prefix = client_info
    digest = hashlib.sha256(logical_key.encode("utf-8")).hexdigest()
    lock_key = f"{prefix}:{collection}:fill:{digest}"
    event_key = (id(asyncio.get_running_loop()), lock_key)
    token = f"{_cache_epoch(config)}:{secrets.token_urlsafe(18)}"
    try:
        async with asyncio.timeout(_redis_operation_timeout_seconds(config)):
            if await client.set(lock_key, token, ex=15, nx=True):
                _fill_events[event_key] = asyncio.Event()
                return None, token
    except Exception as exc:
        _warn_once(prefix, f"fill lock unavailable ({type(exc).__name__})")
        return None, None

    wait_seconds = max(0, min(wait_ms, 8000)) / 1000
    local_event = _fill_events.get(event_key)
    if local_event is not None:
        try:
            async with asyncio.timeout(wait_seconds):
                await local_event.wait()
        except TimeoutError:
            pass
        if not await is_collection_cache_enabled(config, collection) or await get_cache_version(config, collection) != version:
            return None, None
        local_payload = _fill_local_payloads.get(event_key)
        if local_payload is not None:
            expires_at, payload_version, payload = local_payload
            if expires_at > time.monotonic() and payload_version == version:
                if raw_payload:
                    return payload, None
                try:
                    return json.loads(payload), None
                except Exception as exc:
                    _warn_once(f"{collection}:decode", f"local fill JSON decode failed ({type(exc).__name__})")
            else:
                _fill_local_payloads.pop(event_key, None)
        if raw_payload:
            cached = await get_cached_json_payload(config, collection, logical_key, version=version)
        else:
            cached = await get_cached_json(config, collection, logical_key, version=version)
        if cached is not None:
            return cached, None
    else:
        # Cross-process waiters cannot share the local event. Check Redis at a
        # low frequency instead of polling every 50-100ms under a cold burst.
        deadline = time.monotonic() + wait_seconds
        while time.monotonic() < deadline:
            await asyncio.sleep(min(1.0, max(0.0, deadline - time.monotonic())))
            if not await is_collection_cache_enabled(config, collection) or await get_cache_version(config, collection) != version:
                return None, None
            if raw_payload:
                cached = await get_cached_json_payload(config, collection, logical_key, version=version)
            else:
                cached = await get_cached_json(config, collection, logical_key, version=version)
            if cached is not None:
                return cached, None

    # The original filler may have failed. A single waiter gets one chance to
    # take over; remaining requests fall through to MySQL if the cache is down.
    if not await is_collection_cache_enabled(config, collection) or await get_cache_version(config, collection) != version:
        return None, None
    try:
        async with asyncio.timeout(_redis_operation_timeout_seconds(config)):
            if await client.set(lock_key, token, ex=15, nx=True):
                _fill_events[event_key] = asyncio.Event()
                return None, token
    except Exception as exc:
        _warn_once(prefix, f"fill lock retry failed ({type(exc).__name__})")
    return None, None


async def release_cache_fill(config: Any, collection: str, logical_key: str, token: str | None) -> None:
    """Release a miss lock only when it is still owned by this request."""

    if not token:
        return
    digest = hashlib.sha256(logical_key.encode("utf-8")).hexdigest()
    owner_epoch, separator, _ = token.partition(":")
    client_info = _client(config, enabled=True, epoch=owner_epoch if separator else None)
    if client_info is not None:
        client, prefix = client_info
        lock_key = f"{prefix}:{collection}:fill:{digest}"
        event = _fill_events.pop((id(asyncio.get_running_loop()), lock_key), None)
    else:
        lock_key = ""
        event = None
    if event is not None:
        event.set()
    if client_info is None:
        return
    try:
        async with asyncio.timeout(_redis_operation_timeout_seconds(config)):
            await client.eval(
                "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end",
                1,
                lock_key,
                token,
            )
    except Exception as exc:
        _warn_once(prefix, f"fill lock release failed ({type(exc).__name__})")


def _json_default(value: Any) -> Any:
    model_dump = getattr(value, "model_dump", None)
    if callable(model_dump):
        return model_dump(mode="json", by_alias=True)
    isoformat = getattr(value, "isoformat", None)
    if callable(isoformat):
        return isoformat()
    raise TypeError(f"Cannot cache {type(value).__name__}")


async def set_cached_json(
    config: Any,
    collection: str,
    logical_key: str,
    value: Any,
    *,
    expected_version: str | None | object = _VERSION_UNSET,
    ttl_seconds: int | None = None,
) -> None:
    enabled, configured_ttl = await get_api_read_cache_settings(config)
    if not enabled or not await is_collection_cache_enabled(config, collection):
        return
    client_info = _client(config, enabled=True)
    if client_info is None:
        return
    client, prefix = client_info
    try:
        if expected_version is _VERSION_UNSET:
            expected_version = await get_cache_version(config, collection)
        if expected_version is None:
            return
        encoded = json.dumps(value, ensure_ascii=False, separators=(",", ":"), default=_json_default).encode("utf-8")
        if len(encoded) > _max_cache_value_bytes(config):
            return
        async with asyncio.timeout(_redis_operation_timeout_seconds(config)):
            version = await client.get(_version_key(prefix, collection))
            version_text = version.decode("ascii") if isinstance(version, bytes) else str(version or "0")
            if _scoped_version(_cache_epoch(config), version_text) != expected_version:
                return
            await client.set(
                _item_key(prefix, collection, expected_version, logical_key),
                encoded,
                ex=_cache_ttl_seconds(
                    config,
                    ttl_seconds if ttl_seconds is not None else configured_ttl,
                ),
            )
            loop_id = id(asyncio.get_running_loop())
            fill_digest = hashlib.sha256(logical_key.encode("utf-8")).hexdigest()
            fill_event_key = (loop_id, f"{prefix}:{collection}:fill:{fill_digest}")
            if fill_event_key in _fill_events:
                now = time.monotonic()
                for key, (expires_at, _, _) in list(_fill_local_payloads.items()):
                    if expires_at <= now:
                        _fill_local_payloads.pop(key, None)
                _fill_local_payloads[fill_event_key] = (now + 2.0, str(expected_version), encoded)
                if len(_fill_local_payloads) > 128:
                    _fill_local_payloads.pop(next(iter(_fill_local_payloads)))
    except Exception as exc:
        _warn_once(prefix, f"write failed ({type(exc).__name__})")


async def invalidate_cached_collection(config: Any, collection: str, *, force: bool = False) -> None:
    """Bump a generation; old keys expire naturally, with no Redis key scan."""

    if not force and not await is_api_read_cache_enabled(config):
        return
    # A forced invalidation is used after an admin edits sensitive words while
    # Redis reads are disabled. The generation still advances so re-enabling
    # the cache cannot resurrect that older value within its one-hour TTL.
    client_info = _client(config, enabled=True)
    if client_info is None:
        return
    client, prefix = client_info
    try:
        async with asyncio.timeout(_redis_operation_timeout_seconds(config)):
            await client.incr(_version_key(prefix, collection))
    except Exception as exc:
        _warn_once(prefix, f"invalidation failed ({type(exc).__name__})")


async def close_api_read_cache() -> None:
    clients = list(_clients.values())
    _clients.clear()
    for client in clients:
        try:
            await client.aclose()
        except Exception:
            logger.debug("Failed to close homepage Redis cache client", exc_info=True)