replypier/backend/app/services/agent_rooms.py

"""
Copyright 2024-2026 ReplyPier

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""

import asyncio
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Optional

from sqlalchemy.orm import Session

from app.core.auth import CHAT_VIEW_PERMISSIONS, has_any_permission
from app.core.logger import get_logger
from app.core.socketio import sio
from app.repositories.chat import ChatRepository
from app.repositories.session_to_agent import SessionToAgentRepository
from app.repositories.user import UserRepository
from app.services.chat_notifications import UNASSIGNED_CHAT_VIEWERS

# Per-organization room every inbox page joins (see handle_join_room), for
# updates about any of the org's conversations — including AI-owned ones,
# which have no handling agent to address.
ORG_INBOX_ROOM_PREFIX = "org_inbox_"
ORG_CHAT_UPDATES_PREFIX = "org_chat_updates_"
CHAT_WATCH_PREFIX = "chat_watch_"
logger = get_logger(__name__)
_INBOX_COALESCE_SECONDS = 0.25


@dataclass
class _InboxUpdate:
    dirty: bool = True
    task: Optional[asyncio.Task] = None


_pending_inbox_updates: dict[str, _InboxUpdate] = {}


# chat_reply `type`s the inbox appends to a conversation
USER_MESSAGE = 'user_message'
BOT_MESSAGE = 'bot'


def org_inbox_room(organization_id) -> str:
    return f"{ORG_INBOX_ROOM_PREFIX}{organization_id}"


def is_org_inbox_room(room: str) -> bool:
    return room.startswith(ORG_INBOX_ROOM_PREFIX)


def may_join_org_inbox(db: Session, socket_session: dict, room: str) -> bool:
    """Whether a dashboard socket may join an org inbox room: its own
    organization's, and only for people who can see chats nobody owns."""
    if str(socket_session.get('organization_id')) != room.removeprefix(ORG_INBOX_ROOM_PREFIX):
        return False
    user_id = socket_session.get('user_id')
    user = UserRepository(db).get_user(user_id) if user_id else None
    return (user is not None and user.is_active
            and has_any_permission(user, UNASSIGNED_CHAT_VIEWERS))


def may_join_chat_updates(db: Session, socket_session: dict, room: str) -> bool:
    """The organization-wide room carries only a content-free change signal."""
    organization_id = room.removeprefix(ORG_CHAT_UPDATES_PREFIX)
    if str(socket_session.get('organization_id')) != organization_id:
        return False
    user_id = socket_session.get('user_id')
    user = UserRepository(db).get_user(user_id) if user_id else None
    return bool(user and user.is_active and str(user.organization_id) == organization_id
                and has_any_permission(user, CHAT_VIEW_PERMISSIONS))


async def may_watch_chat(db: Session, socket_session: dict, room: str) -> bool:
    """Mirror the detail endpoint's organization and chat-scope checks."""
    session_id = room.removeprefix(CHAT_WATCH_PREFIX)
    user_id = socket_session.get('user_id')
    user = UserRepository(db).get_user(user_id) if user_id else None
    if not user or not user.is_active or not has_any_permission(user, CHAT_VIEW_PERMISSIONS):
        return False
    if str(user.organization_id) != str(socket_session.get('organization_id')):
        return False
    try:
        chat = SessionToAgentRepository(db).get_session(session_id)
    except (ValueError, TypeError):
        return False
    if not chat or str(chat.organization_id) != str(user.organization_id):
        return False
    if has_any_permission(user, ('view_all_chats',)):
        return True
    can_view_assigned = has_any_permission(user, ('view_assigned_chats',))
    return await ChatRepository(db).check_session_access(
        session_id=session_id,
        user_id=user.id if can_view_assigned else None,
        user_groups=[str(group.id) for group in user.groups] if can_view_assigned else [],
        include_unassigned=has_any_permission(user, ('view_unassigned_chats',)),
    )


async def notify_conversation_changed(organization_id, session_id) -> None:
    """Invalidate the selected thread and inbox after a committed change.

    Never send message text or identifiers to the org-wide room. The thread
    room also carries no content: every refresh goes through the normal HTTP
    authorization, including after a chat is reassigned.
    """
    if not organization_id or not session_id:
        return
    await _emit_change('chat_changed', f'{CHAT_WATCH_PREFIX}{session_id}')
    org_key = str(organization_id)
    state = _pending_inbox_updates.get(org_key)
    if state is not None:
        state.dirty = True
        return
    state = _InboxUpdate()
    _pending_inbox_updates[org_key] = state
    # One task per active organization, not one task per message. Each server
    # worker coalesces a burst before broadcasting it through the Redis manager.
    state.task = asyncio.create_task(_flush_inbox_changes(org_key, state))


async def _emit_change(event: str, room: str) -> None:
    try:
        await sio.emit(event, {}, room=room, namespace='/agent')
    except Exception as exc:
        # A broker outage must not turn an already-committed chat into an error.
        logger.warning('Failed to notify inbox of a conversation change: %s', type(exc).__name__)


async def _flush_inbox_changes(org_key: str, state: _InboxUpdate) -> None:
    try:
        while state.dirty:
            await asyncio.sleep(_INBOX_COALESCE_SECONDS)
            # Everything received before this point fits in this signal. An
            # arrival during publish sets dirty again and gets a trailing one.
            state.dirty = False
            await _emit_change('inbox_changed', f'{ORG_CHAT_UPDATES_PREFIX}{org_key}')
    finally:
        _pending_inbox_updates.pop(org_key, None)


async def emit_to_session_agents(event: str, payload: dict, session_id,
                                 user_id: Optional[object] = None) -> None:
    """Push a live update to the dashboard for one conversation.

    The inbox listens on the handling agent's personal room (user_<id>); the
    session room reaches anyone who joined that conversation explicitly.
    """
    if user_id is not None:
        await sio.emit(event, payload, room=f"user_{user_id}", namespace='/agent')
    await sio.emit(event, payload, room=str(session_id), namespace='/agent')


async def emit_to_org_inbox(event: str, payload: dict, organization_id,
                            user_id: Optional[object] = None) -> None:
    """Push a live update to every inbox open on the organization, and to the
    handling agent's personal room (where a chat they own is listening)."""
    await sio.emit(event, payload, room=org_inbox_room(organization_id), namespace='/agent')
    if user_id is not None:
        await sio.emit(event, payload, room=f"user_{user_id}", namespace='/agent')


async def relay_stored_message(session_record, stored, message_type: str) -> None:
    """Push a stored message into the dashboard conversation live, the way
    the widget's chat_reply events do — a customer message, or one sent from
    outside the inbox (e.g. by an external app) into a chat a human holds."""
    session_id = str(session_record.session_id)
    payload = {
        'message': stored.message,
        'message_id': stored.id,
        'type': message_type,
        'transfer_to_human': False,
        'session_id': session_id,
        # The inbox appends the message live off this field; without it the
        # handler builds an invalid Date and the message only shows on refetch.
        'created_at': (stored.created_at or datetime.now(timezone.utc)).isoformat(),
        # Carries e.g. the button-reply marker, so tags show live too
        'attributes': stored.attributes or {},
    }
    await notify_conversation_changed(session_record.organization_id, session_id)
    await emit_to_session_agents('chat_reply', payload, session_id, session_record.user_id)