ForcePilot/backend/package/yuxi/channels/adapters/slack/dm_cache.py

from __future__ import annotations

import asyncio
from collections import OrderedDict

from slack_sdk.errors import SlackApiError
from slack_sdk.web.async_client import AsyncWebClient

from yuxi.utils.logging_config import logger

MAX_DM_CACHE_SIZE = 1024


class DmChannelCache:
    def __init__(self, max_size: int = MAX_DM_CACHE_SIZE):
        self._cache: OrderedDict[str, str] = OrderedDict()
        self._max_size = max_size
        self._lock = asyncio.Lock()

    async def get(self, user_id: str) -> str | None:
        async with self._lock:
            channel_id = self._cache.get(user_id)
            if channel_id is not None:
                self._cache.move_to_end(user_id)
            return channel_id

    async def put(self, user_id: str, channel_id: str) -> None:
        async with self._lock:
            if user_id in self._cache:
                self._cache.move_to_end(user_id)
            else:
                if len(self._cache) >= self._max_size:
                    self._cache.popitem(last=False)
            self._cache[user_id] = channel_id

    async def clear(self) -> None:
        async with self._lock:
            self._cache.clear()

    @property
    def size(self) -> int:
        return len(self._cache)

    async def resolve(self, client: AsyncWebClient, user_id: str) -> str | None:
        cached = await self.get(user_id)
        if cached:
            return cached

        try:
            result = await client.conversations_open(users=[user_id])
            if not result.get("ok"):
                logger.warning(f"conversations.open failed for {user_id}: {result.get('error')}")
                return None
            channel_id = result.get("channel", {}).get("id", "")
            if channel_id:
                await self.put(user_id, channel_id)
            return channel_id
        except SlackApiError as e:
            logger.warning(f"Failed to resolve DM channel for {user_id}: {e}")
            return None