diff --git a/app/plugins/builtins/kommunikation/conversations.py b/app/plugins/builtins/kommunikation/conversations.py new file mode 100644 index 0000000..bb6c72d --- /dev/null +++ b/app/plugins/builtins/kommunikation/conversations.py @@ -0,0 +1,343 @@ +"""Conversation CRUD and pin/mute actions for the kommunikation plugin.""" + +from __future__ import annotations + +import logging +import uuid +from typing import Any + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.event_bus import get_event_bus +from app.plugins.builtins.kommunikation.interactions import _get_unread_count +from app.plugins.builtins.kommunikation.messages import send_message +from app.plugins.builtins.kommunikation.models import ( + CommConversation, + CommConversationMute, + CommConversationPin, + CommParticipant, +) +from app.plugins.builtins.kommunikation.rbac import CommRBAC +from app.plugins.builtins.kommunikation.serializers import conversation_to_response + +logger = logging.getLogger(__name__) + + + +# ─── conversations ─── + + + +async def list_conversations( + db: AsyncSession, + tenant_id: uuid.UUID, + user_id: uuid.UUID, + include_archived: bool = False, +) -> list[dict[str, Any]]: + """List all conversations for a user.""" + # Get conversations where user is a participant + result = await db.execute( + select(CommConversation) + .join(CommParticipant, CommParticipant.conversation_id == CommConversation.id) + .where( + CommParticipant.participant_id == user_id, + CommParticipant.participant_type == "user", + CommParticipant.left_at.is_(None), + CommConversation.tenant_id == tenant_id, + CommConversation.deleted_at.is_(None), + ) + .order_by(CommConversation.last_msg_at.desc().nullslast()) + ) + conversations = result.scalars().all() + + # Get user's pinned conversations + pins_result = await db.execute( + select(CommConversationPin).where( + CommConversationPin.user_id == user_id, + CommConversationPin.tenant_id == tenant_id, + ) + ) + pinned_ids = {p.conversation_id for p in pins_result.scalars().all()} + + conv_list = [] + for conv in conversations: + if conv.is_archived and not include_archived: + continue + # Get participants + parts_result = await db.execute( + select(CommParticipant).where( + CommParticipant.conversation_id == conv.id, + CommParticipant.left_at.is_(None), + ) + ) + participants = list(parts_result.scalars().all()) + + # Get unread count + unread = await _get_unread_count(db, tenant_id, conv.id, user_id) + + conv_list.append( + conversation_to_response( + conv, participants, unread_count=unread, is_pinned_by_user=conv.id in pinned_ids + ) + ) + + # Sort: pinned first, then by last_msg_at + conv_list.sort(key=lambda c: (not c["is_pinned"], c["last_msg_at"] or ""), reverse=False) + # Actually: pinned first (True > False in reverse), then newest first + conv_list.sort(key=lambda c: c["last_msg_at"] or "0000", reverse=True) + conv_list.sort(key=lambda c: c["is_pinned"], reverse=True) + + return conv_list + + +async def get_conversation( + db: AsyncSession, + tenant_id: uuid.UUID, + conversation_id: uuid.UUID, + user_id: uuid.UUID, +) -> dict[str, Any] | None: + """Get a single conversation with participants.""" + result = await db.execute( + select(CommConversation).where( + CommConversation.id == conversation_id, + CommConversation.tenant_id == tenant_id, + CommConversation.deleted_at.is_(None), + ) + ) + conv = result.scalar_one_or_none() + if conv is None: + return None + + # Check user is participant + if not await CommRBAC.is_participant(db, conversation_id, user_id): + return None + + parts_result = await db.execute( + select(CommParticipant).where( + CommParticipant.conversation_id == conv.id, + CommParticipant.left_at.is_(None), + ) + ) + participants = list(parts_result.scalars().all()) + + # Check pinned + pin_result = await db.execute( + select(CommConversationPin).where( + CommConversationPin.conversation_id == conv.id, + CommConversationPin.user_id == user_id, + ) + ) + is_pinned = pin_result.scalar_one_or_none() is not None + + unread = await _get_unread_count(db, tenant_id, conv.id, user_id) + + return conversation_to_response(conv, participants, unread_count=unread, is_pinned_by_user=is_pinned) + + +async def create_conversation( + db: AsyncSession, + tenant_id: uuid.UUID, + user_id: uuid.UUID, + title: str | None = None, + participant_ids: list[str] | None = None, + is_direct: bool = False, + initial_message: str | None = None, + metadata: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Create a new conversation.""" + conv = CommConversation( + tenant_id=tenant_id, + title=title, + owner_id=user_id, + is_direct=is_direct, + created_by=user_id, + created_by_type="user", + metadata_=metadata or {}, + ) + from app.core.hooks import do_action + await do_action("comm.conversation.before_create", tenant_id=tenant_id, user_id=user_id) + db.add(conv) + await db.flush() + await do_action("comm.conversation.after_create", conversation_id=conv.id, tenant_id=tenant_id, user_id=user_id) + + # Add creator as admin + creator = CommParticipant( + tenant_id=tenant_id, + conversation_id=conv.id, + participant_id=user_id, + participant_type="user", + role="admin", + ) + db.add(creator) + + # Add other participants + for pid_str in (participant_ids or []): + try: + pid = uuid.UUID(pid_str) + if pid == user_id: + continue + p = CommParticipant( + tenant_id=tenant_id, + conversation_id=conv.id, + participant_id=pid, + participant_type="user", + role="member", + ) + db.add(p) + except ValueError: + logger.warning(f"Invalid participant UUID: {pid_str}") + + await db.flush() + + # Send initial message if provided + if initial_message: + await send_message( + db, tenant_id, conv.id, user_id, "user", + content=initial_message, content_format="text", + ) + + # Publish event + event_bus = get_event_bus() + await event_bus.publish("conversation.created", { + "conversation_id": str(conv.id), + "tenant_id": str(tenant_id), + "created_by": str(user_id), + }) + + # Get all participants for response + parts_result = await db.execute( + select(CommParticipant).where( + CommParticipant.conversation_id == conv.id, + CommParticipant.left_at.is_(None), + ) + ) + participants = list(parts_result.scalars().all()) + + return conversation_to_response(conv, participants) + + +async def update_conversation( + db: AsyncSession, + tenant_id: uuid.UUID, + conversation_id: uuid.UUID, + user_id: uuid.UUID, + title: str | None = None, + is_archived: bool | None = None, +) -> dict[str, Any] | None: + """Update a conversation.""" + result = await db.execute( + select(CommConversation).where( + CommConversation.id == conversation_id, + CommConversation.tenant_id == tenant_id, + CommConversation.deleted_at.is_(None), + ) + ) + conv = result.scalar_one_or_none() + if conv is None: + return None + + # Check locked + if conv.is_locked and title is not None: + # Only the locking plugin can change title on locked conversations + # Users cannot + pass + elif title is not None: + conv.title = title + conv.title_set_by = user_id + + if is_archived is not None: + conv.is_archived = is_archived + + await db.flush() + + return await get_conversation(db, tenant_id, conversation_id, user_id) + + +async def pin_conversation( + db: AsyncSession, + tenant_id: uuid.UUID, + conversation_id: uuid.UUID, + user_id: uuid.UUID, +) -> bool: + """Pin a conversation for a user.""" + existing = await db.execute( + select(CommConversationPin).where( + CommConversationPin.conversation_id == conversation_id, + CommConversationPin.user_id == user_id, + ) + ) + if existing.scalar_one_or_none() is None: + pin = CommConversationPin( + tenant_id=tenant_id, + conversation_id=conversation_id, + user_id=user_id, + ) + db.add(pin) + await db.flush() + return True + + +async def unpin_conversation( + db: AsyncSession, + conversation_id: uuid.UUID, + user_id: uuid.UUID, +) -> bool: + """Unpin a conversation for a user.""" + result = await db.execute( + select(CommConversationPin).where( + CommConversationPin.conversation_id == conversation_id, + CommConversationPin.user_id == user_id, + ) + ) + pin = result.scalar_one_or_none() + if pin: + await db.delete(pin) + await db.flush() + return True + + +async def mute_conversation( + db: AsyncSession, + tenant_id: uuid.UUID, + conversation_id: uuid.UUID, + user_id: uuid.UUID, +) -> bool: + """Mute a conversation for a user.""" + existing = await db.execute( + select(CommConversationMute).where( + CommConversationMute.conversation_id == conversation_id, + CommConversationMute.user_id == user_id, + ) + ) + if existing.scalar_one_or_none() is None: + mute = CommConversationMute( + tenant_id=tenant_id, + conversation_id=conversation_id, + user_id=user_id, + ) + db.add(mute) + await db.flush() + return True + + +async def unmute_conversation( + db: AsyncSession, + conversation_id: uuid.UUID, + user_id: uuid.UUID, +) -> bool: + """Unmute a conversation for a user.""" + result = await db.execute( + select(CommConversationMute).where( + CommConversationMute.conversation_id == conversation_id, + CommConversationMute.user_id == user_id, + ) + ) + mute = result.scalar_one_or_none() + if mute: + await db.delete(mute) + await db.flush() + return True + + +# ─── Participant Management ─── diff --git a/app/plugins/builtins/kommunikation/interactions.py b/app/plugins/builtins/kommunikation/interactions.py new file mode 100644 index 0000000..d26cd04 --- /dev/null +++ b/app/plugins/builtins/kommunikation/interactions.py @@ -0,0 +1,159 @@ +"""Reactions and read-state handling for the kommunikation plugin.""" + +from __future__ import annotations + +import logging +import uuid +from datetime import UTC, datetime +from typing import Any + +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.event_bus import get_event_bus +from app.plugins.builtins.kommunikation.models import ( + CommMessage, + CommMessageReaction, + CommMessageRead, +) + +logger = logging.getLogger(__name__) + + + +# ─── interactions ─── + + + +async def add_reaction( + db: AsyncSession, + tenant_id: uuid.UUID, + message_id: uuid.UUID, + user_id: uuid.UUID, + emoji: str, +) -> dict[str, Any] | None: + """Add an emoji reaction to a message.""" + existing = await db.execute( + select(CommMessageReaction).where( + CommMessageReaction.message_id == message_id, + CommMessageReaction.user_id == user_id, + CommMessageReaction.emoji == emoji, + ) + ) + if existing.scalar_one_or_none() is not None: + return None # Already reacted + + r = CommMessageReaction( + tenant_id=tenant_id, + message_id=message_id, + user_id=user_id, + emoji=emoji, + ) + db.add(r) + await db.flush() + + event_bus = get_event_bus() + await event_bus.publish("reaction.added", { + "message_id": str(message_id), + "emoji": emoji, + "user_id": str(user_id), + }) + + return { + "id": str(r.id), + "message_id": str(r.message_id), + "user_id": str(r.user_id), + "emoji": r.emoji, + } + + +async def remove_reaction( + db: AsyncSession, + message_id: uuid.UUID, + user_id: uuid.UUID, + emoji: str, +) -> bool: + """Remove an emoji reaction.""" + result = await db.execute( + select(CommMessageReaction).where( + CommMessageReaction.message_id == message_id, + CommMessageReaction.user_id == user_id, + CommMessageReaction.emoji == emoji, + ) + ) + r = result.scalar_one_or_none() + if r is None: + return False + await db.delete(r) + await db.flush() + return True + + +# ─── Read State ─── + + +async def mark_read( + db: AsyncSession, + tenant_id: uuid.UUID, + conversation_id: uuid.UUID, + user_id: uuid.UUID, + last_read_msg_id: str | None = None, +) -> bool: + """Mark conversation as read up to a message.""" + result = await db.execute( + select(CommMessageRead).where( + CommMessageRead.conversation_id == conversation_id, + CommMessageRead.user_id == user_id, + ) + ) + read = result.scalar_one_or_none() + + msg_id = uuid.UUID(last_read_msg_id) if last_read_msg_id else None + + if read is None: + read = CommMessageRead( + tenant_id=tenant_id, + conversation_id=conversation_id, + user_id=user_id, + last_read_msg_id=msg_id, + ) + db.add(read) + else: + read.last_read_msg_id = msg_id + read.last_read_at = datetime.now(UTC) + + await db.flush() + return True + + +async def _get_unread_count( + db: AsyncSession, + tenant_id: uuid.UUID, + conversation_id: uuid.UUID, + user_id: uuid.UUID, +) -> int: + """Get unread message count for a user in a conversation.""" + # Get last read message + read_result = await db.execute( + select(CommMessageRead).where( + CommMessageRead.conversation_id == conversation_id, + CommMessageRead.user_id == user_id, + ) + ) + read = read_result.scalar_one_or_none() + + query = select(func.count()).select_from(CommMessage).where( + CommMessage.conversation_id == conversation_id, + CommMessage.tenant_id == tenant_id, + CommMessage.deleted_at.is_(None), + CommMessage.sender_type != "system", # Don't count system messages? Or count all? + ) + + if read and read.last_read_at: + query = query.where(CommMessage.created_at > read.last_read_at) + + result = await db.execute(query) + return result.scalar() or 0 + + +# ─── Plugin Room Creation ─── diff --git a/app/plugins/builtins/kommunikation/messages.py b/app/plugins/builtins/kommunikation/messages.py new file mode 100644 index 0000000..04421fd --- /dev/null +++ b/app/plugins/builtins/kommunikation/messages.py @@ -0,0 +1,394 @@ +"""Message retrieval, sending and editing for the kommunikation plugin.""" + +from __future__ import annotations + +import logging +import uuid +from datetime import UTC, datetime +from typing import Any + +from sqlalchemy import func, select, update +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.event_bus import get_event_bus +from app.plugins.builtins.kommunikation.models import ( + CommConversation, + CommMessage, + CommMessageAttachment, + CommMessageBlock, + CommMessageEdit, + CommMessageReaction, + CommParticipant, +) +from app.plugins.builtins.kommunikation.participant_registry import get_participant_registry +from app.plugins.builtins.kommunikation.serializers import ( + conversation_to_response, + message_to_response, + parse_mentions, + participant_to_response, +) + +logger = logging.getLogger(__name__) + +MAX_TRIGGER_DEPTH = 3 + + +# ─── messages ─── + + + +async def get_messages( + db: AsyncSession, + tenant_id: uuid.UUID, + conversation_id: uuid.UUID, + page: int = 1, + page_size: int = 50, + before_id: uuid.UUID | None = None, +) -> dict[str, Any]: + """Get paginated messages for a conversation.""" + query = select(CommMessage).where( + CommMessage.conversation_id == conversation_id, + CommMessage.tenant_id == tenant_id, + CommMessage.deleted_at.is_(None), + ).order_by(CommMessage.created_at.desc()) + + if before_id: + before_msg = await db.execute( + select(CommMessage).where(CommMessage.id == before_id) + ) + before = before_msg.scalar_one_or_none() + if before: + query = query.where(CommMessage.created_at < before.created_at) + + query = query.offset((page - 1) * page_size).limit(page_size) + result = await db.execute(query) + messages = list(result.scalars().all()) + + # Get blocks, attachments, reactions for each message + msg_ids = [m.id for m in messages] + blocks_map: dict[uuid.UUID, list] = {} + attachments_map: dict[uuid.UUID, list] = {} + reactions_map: dict[uuid.UUID, list] = {} + + if msg_ids: + blocks_result = await db.execute( + select(CommMessageBlock).where( + CommMessageBlock.message_id.in_(msg_ids), + CommMessageBlock.deleted_at.is_(None), + ).order_by(CommMessageBlock.sort_order) + ) + for b in blocks_result.scalars().all(): + blocks_map.setdefault(b.message_id, []).append(b) + + atts_result = await db.execute( + select(CommMessageAttachment).where( + CommMessageAttachment.message_id.in_(msg_ids), + CommMessageAttachment.deleted_at.is_(None), + ) + ) + for a in atts_result.scalars().all(): + attachments_map.setdefault(a.message_id, []).append(a) + + reactions_result = await db.execute( + select(CommMessageReaction).where( + CommMessageReaction.message_id.in_(msg_ids), + ) + ) + for r in reactions_result.scalars().all(): + reactions_map.setdefault(r.message_id, []).append(r) + + items = [] + for msg in reversed(messages): # chronological order + items.append( + message_to_response( + msg, + blocks=blocks_map.get(msg.id, []), + attachments=attachments_map.get(msg.id, []), + reactions=reactions_map.get(msg.id, []), + ) + ) + + # Total count + count_result = await db.execute( + select(func.count()).select_from(CommMessage).where( + CommMessage.conversation_id == conversation_id, + CommMessage.tenant_id == tenant_id, + CommMessage.deleted_at.is_(None), + ) + ) + total = count_result.scalar() or 0 + + has_more = (page * page_size) < total + + return {"items": items, "total": total, "page": page, "has_more": has_more} + + +async def send_message( + db: AsyncSession, + tenant_id: uuid.UUID, + conversation_id: uuid.UUID, + sender_id: uuid.UUID | None, + sender_type: str, + content: str = "", + content_format: str = "text", + blocks: list[dict[str, Any]] | None = None, + reply_to_id: str | None = None, + attachments: list[dict[str, Any]] | None = None, + metadata: dict[str, Any] | None = None, + trigger_depth: int = 0, +) -> dict[str, Any]: + """Send a message to a conversation and trigger participant handlers.""" + # Create message + msg = CommMessage( + tenant_id=tenant_id, + conversation_id=conversation_id, + sender_id=sender_id, + sender_type=sender_type, + content=content, + content_format=content_format, + metadata_=metadata or {}, + ) + if reply_to_id: + try: + msg.reply_to_id = uuid.UUID(reply_to_id) + except ValueError: + pass + + from app.core.hooks import do_action + await do_action("comm.before_message", conversation_id=conversation_id, tenant_id=tenant_id, sender_id=sender_id) + db.add(msg) + await db.flush() + await do_action("comm.after_message", message_id=msg.id, conversation_id=conversation_id, tenant_id=tenant_id, sender_id=sender_id) + + # Create blocks + if blocks: + for i, block in enumerate(blocks): + b = CommMessageBlock( + tenant_id=tenant_id, + message_id=msg.id, + block_type=block["block_type"], + block_data=block["block_data"], + sort_order=i, + ) + db.add(b) + + # Create attachments + if attachments: + for att in attachments: + a = CommMessageAttachment( + tenant_id=tenant_id, + message_id=msg.id, + file_id=uuid.UUID(att["file_id"]) if att.get("file_id") else None, + file_source=att.get("file_source", "comm"), + file_name=att.get("file_name", ""), + file_type=att.get("file_type", "application/octet-stream"), + file_size=att.get("file_size"), + ) + db.add(a) + + await db.flush() + + # Update conversation last_msg + await db.execute( + update(CommConversation) + .where(CommConversation.id == conversation_id) + .values( + last_msg_at=datetime.now(UTC), + last_msg_preview=content[:200] if content else "", + last_msg_sender_type=sender_type, + ) + ) + + # Publish event + event_bus = get_event_bus() + await event_bus.publish("message.received", { + "conversation_id": str(conversation_id), + "message_id": str(msg.id), + "sender_type": sender_type, + "tenant_id": str(tenant_id), + "content": content, + "trigger_depth": trigger_depth, + }) + + # Trigger participant handlers (if not at max depth) + if trigger_depth < MAX_TRIGGER_DEPTH: + await _trigger_participants( + db, tenant_id, conversation_id, msg, trigger_depth + ) + + # Load blocks/attachments/reactions for response + blocks_result = await db.execute( + select(CommMessageBlock).where( + CommMessageBlock.message_id == msg.id, + CommMessageBlock.deleted_at.is_(None), + ).order_by(CommMessageBlock.sort_order) + ) + msg_blocks = list(blocks_result.scalars().all()) + + atts_result = await db.execute( + select(CommMessageAttachment).where( + CommMessageAttachment.message_id == msg.id, + CommMessageAttachment.deleted_at.is_(None), + ) + ) + msg_atts = list(atts_result.scalars().all()) + + return message_to_response(msg, blocks=msg_blocks, attachments=msg_atts) + + +async def _trigger_participants( + db: AsyncSession, + tenant_id: uuid.UUID, + conversation_id: uuid.UUID, + message: CommMessage, + trigger_depth: int, +) -> None: + """Trigger participant handlers for non-user participants.""" + # Get conversation participants + result = await db.execute( + select(CommParticipant).where( + CommParticipant.conversation_id == conversation_id, + CommParticipant.left_at.is_(None), + CommParticipant.participant_type != "user", + ) + ) + non_user_participants = list(result.scalars().all()) + + if not non_user_participants: + return + + # Get conversation info + conv_result = await db.execute( + select(CommConversation).where(CommConversation.id == conversation_id) + ) + conv = conv_result.scalar_one_or_none() + if conv is None: + return + + # Parse mentions + mentions = parse_mentions(message.content) + + # Build conversation dict + all_parts_result = await db.execute( + select(CommParticipant).where( + CommParticipant.conversation_id == conversation_id, + CommParticipant.left_at.is_(None), + ) + ) + all_parts = [participant_to_response(p) for p in all_parts_result.scalars().all()] + conv_dict = conversation_to_response(conv, []) + conv_dict["participants"] = all_parts + + msg_dict = message_to_response(message) + context = {"tenant_id": str(tenant_id), "trigger_depth": trigger_depth} + + registry = get_participant_registry() + + for p in non_user_participants: + handler = registry.get_handler(p.participant_type) + if handler is None: + continue + + try: + responses = await handler.on_message_received( + conversation_id=conversation_id, + message=msg_dict, + conversation=conv_dict, + mentions=mentions, + context=context, + ) + + if responses: + for resp in responses: + await send_message( + db, + tenant_id, + conversation_id, + sender_id=None, + sender_type=p.participant_type, + content=resp.get("content", ""), + content_format=resp.get("content_format", "text"), + blocks=resp.get("blocks"), + metadata={ + **(resp.get("metadata") or {}), + "triggered_by": str(message.id), + "trigger_depth": trigger_depth + 1, + }, + trigger_depth=trigger_depth + 1, + ) + except Exception: + logger.exception( + f"Participant handler error for type {p.participant_type}" + ) + + +async def edit_message( + db: AsyncSession, + tenant_id: uuid.UUID, + message_id: uuid.UUID, + user_id: uuid.UUID, + new_content: str, +) -> dict[str, Any] | None: + """Edit a message, storing the old version in history.""" + result = await db.execute( + select(CommMessage).where( + CommMessage.id == message_id, + CommMessage.tenant_id == tenant_id, + CommMessage.deleted_at.is_(None), + ) + ) + msg = result.scalar_one_or_none() + if msg is None: + return None + + # Get old blocks + blocks_result = await db.execute( + select(CommMessageBlock).where( + CommMessageBlock.message_id == message_id, + CommMessageBlock.deleted_at.is_(None), + ) + ) + old_blocks = [b.block_data for b in blocks_result.scalars().all()] + + # Save edit history + edit = CommMessageEdit( + tenant_id=tenant_id, + message_id=message_id, + old_content=msg.content, + old_blocks=old_blocks, + edited_by=user_id, + ) + db.add(edit) + + from app.core.hooks import do_action + await do_action("comm.before_edit", message_id=message_id, tenant_id=tenant_id, user_id=user_id) + + # Update message + msg.content = new_content + msg.edited_at = datetime.now(UTC) + await db.flush() + await do_action("comm.after_edit", message_id=message_id, tenant_id=tenant_id, user_id=user_id) + + return message_to_response(msg) + + +async def delete_message( + db: AsyncSession, + message_id: uuid.UUID, +) -> bool: + """Soft-delete a message.""" + result = await db.execute( + select(CommMessage).where(CommMessage.id == message_id) + ) + msg = result.scalar_one_or_none() + if msg is None: + return False + from app.core.hooks import do_action + await do_action("comm.before_delete", message_id=message_id) + msg.deleted_at = datetime.now(UTC) + await db.flush() + await do_action("comm.after_delete", message_id=message_id) + return True + + +# ─── Reactions ─── diff --git a/app/plugins/builtins/kommunikation/participants.py b/app/plugins/builtins/kommunikation/participants.py new file mode 100644 index 0000000..03c138a --- /dev/null +++ b/app/plugins/builtins/kommunikation/participants.py @@ -0,0 +1,127 @@ +"""Participant management for the kommunikation plugin.""" + +from __future__ import annotations + +import logging +import uuid +from datetime import UTC, datetime +from typing import Any + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.event_bus import get_event_bus +from app.plugins.builtins.kommunikation.models import ( + CommParticipant, +) +from app.plugins.builtins.kommunikation.serializers import participant_to_response + +logger = logging.getLogger(__name__) + + + +# ─── participants ─── + + + +async def add_participant( + db: AsyncSession, + tenant_id: uuid.UUID, + conversation_id: uuid.UUID, + participant_id: str, + participant_type: str = "user", + role: str = "member", + display_name: str | None = None, +) -> dict[str, Any] | None: + """Add a participant to a conversation.""" + try: + pid = uuid.UUID(participant_id) if participant_type == "user" else None + except ValueError: + return None + + existing = await db.execute( + select(CommParticipant).where( + CommParticipant.conversation_id == conversation_id, + CommParticipant.participant_id == pid if pid else CommParticipant.participant_type == participant_type, + CommParticipant.participant_type == participant_type, + CommParticipant.left_at.is_(None), + ) + ) + if existing.scalar_one_or_none() is not None: + return None # Already a participant + + p = CommParticipant( + tenant_id=tenant_id, + conversation_id=conversation_id, + participant_id=pid, + participant_type=participant_type, + role=role, + display_name=display_name, + ) + db.add(p) + await db.flush() + + # Publish event + event_bus = get_event_bus() + await event_bus.publish("participant.joined", { + "conversation_id": str(conversation_id), + "participant_id": participant_id, + "participant_type": participant_type, + "tenant_id": str(tenant_id), + }) + + return participant_to_response(p) + + +async def remove_participant( + db: AsyncSession, + conversation_id: uuid.UUID, + participant_id: uuid.UUID, +) -> bool: + """Remove a participant from a conversation (set left_at).""" + result = await db.execute( + select(CommParticipant).where( + CommParticipant.conversation_id == conversation_id, + CommParticipant.participant_id == participant_id, + CommParticipant.participant_type == "user", + CommParticipant.left_at.is_(None), + ) + ) + p = result.scalar_one_or_none() + if p is None: + return False + p.left_at = datetime.now(UTC) + await db.flush() + + event_bus = get_event_bus() + await event_bus.publish("participant.left", { + "conversation_id": str(conversation_id), + "participant_id": str(participant_id), + }) + return True + + +async def change_role( + db: AsyncSession, + conversation_id: uuid.UUID, + participant_id: uuid.UUID, + new_role: str, +) -> dict[str, Any] | None: + """Change a participant's role.""" + result = await db.execute( + select(CommParticipant).where( + CommParticipant.conversation_id == conversation_id, + CommParticipant.participant_id == participant_id, + CommParticipant.participant_type == "user", + CommParticipant.left_at.is_(None), + ) + ) + p = result.scalar_one_or_none() + if p is None: + return None + p.role = new_role + await db.flush() + return participant_to_response(p) + + +# ─── Messages ─── diff --git a/app/plugins/builtins/kommunikation/plugin_rooms.py b/app/plugins/builtins/kommunikation/plugin_rooms.py new file mode 100644 index 0000000..7e39a27 --- /dev/null +++ b/app/plugins/builtins/kommunikation/plugin_rooms.py @@ -0,0 +1,345 @@ +"""Plugin room creation and system channels for the kommunikation plugin.""" + +from __future__ import annotations + +import logging +import uuid +from datetime import UTC, datetime +from typing import Any + +from sqlalchemy import and_, select, update +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.event_bus import get_event_bus +from app.plugins.builtins.kommunikation.conversations import get_conversation +from app.plugins.builtins.kommunikation.models import ( + CommConversation, + CommConversationPin, + CommMessage, + CommMessageBlock, + CommParticipant, +) +from app.plugins.builtins.kommunikation.serializers import conversation_to_response + +logger = logging.getLogger(__name__) + + + +# ─── plugin_rooms ─── + + + +async def find_locked_room_id( + db: AsyncSession, + tenant_id: uuid.UUID, + plugin_name: str, + title: str, +) -> uuid.UUID | None: + """Find the conversation ID of a locked plugin room by tenant and title. + + Matches the same room semantics as ``create_plugin_room``: locked rooms + are owned by the plugin (``locked_by == plugin_name``) and soft-deleted + conversations are excluded. Returns ``None`` when no room exists. + """ + result = await db.execute( + select(CommConversation.id).where( + CommConversation.tenant_id == tenant_id, + CommConversation.title == title, + CommConversation.is_locked.is_(True), + CommConversation.locked_by == plugin_name, + CommConversation.deleted_at.is_(None), + ) + ) + return result.scalar_one_or_none() + + +async def create_plugin_room( + db: AsyncSession, + tenant_id: uuid.UUID, + user_id: uuid.UUID, + plugin_name: str, + title: str, + participant_type: str, + user_role: str = "member", +) -> dict[str, Any]: + """Create a locked, pinned room for a plugin (System, Live KI, Assistent). + + The room is locked (users can't change title/participants) and pinned for the user. + """ + # Check if room already exists for this user + plugin + result = await db.execute( + select(CommConversation).where( + CommConversation.tenant_id == tenant_id, + CommConversation.title == title, + CommConversation.is_locked.is_(True), + CommConversation.locked_by == plugin_name, + CommConversation.deleted_at.is_(None), + ).join(CommParticipant, CommParticipant.conversation_id == CommConversation.id).where( + CommParticipant.participant_id == user_id, + CommParticipant.participant_type == "user", + CommParticipant.left_at.is_(None), + ) + ) + existing = result.scalar_one_or_none() + if existing: + # Already exists — return it + return await get_conversation(db, tenant_id, existing.id, user_id) or {} + + # Create conversation + conv = CommConversation( + tenant_id=tenant_id, + title=title, + owner_id=user_id, + is_locked=True, + locked_by=plugin_name, + is_direct=False, + created_by=None, + created_by_type="plugin", + metadata_={"plugin": plugin_name}, + ) + db.add(conv) + await db.flush() + + # Add plugin as participant + plugin_p = CommParticipant( + tenant_id=tenant_id, + conversation_id=conv.id, + participant_id=None, + participant_type=participant_type, + role="admin", + display_name=title, + ) + db.add(plugin_p) + + # Add user as participant + user_p = CommParticipant( + tenant_id=tenant_id, + conversation_id=conv.id, + participant_id=user_id, + participant_type="user", + role=user_role, + ) + db.add(user_p) + + # Pin for user + pin = CommConversationPin( + tenant_id=tenant_id, + conversation_id=conv.id, + user_id=user_id, + ) + db.add(pin) + + await db.flush() + + # Publish event + event_bus = get_event_bus() + await event_bus.publish("conversation.created", { + "conversation_id": str(conv.id), + "tenant_id": str(tenant_id), + "created_by_type": "plugin", + "plugin_name": plugin_name, + }) + + parts = [plugin_p, user_p] + return conversation_to_response(conv, parts, is_pinned_by_user=True) + + + +# ─── System Channel ─── + + +async def get_or_create_system_channel( + db: AsyncSession, + tenant_id: uuid.UUID, +) -> CommConversation: + """Get or create the tenant-wide system channel. + + The system channel is a locked, is_system=True conversation that serves as + the central destination for system notifications, user alerts, and agent messages. + All users of the tenant are automatically added as participants. + """ + result = await db.execute( + select(CommConversation).where( + CommConversation.tenant_id == tenant_id, + CommConversation.is_system.is_(True), + CommConversation.deleted_at.is_(None), + ) + ) + conv = result.scalar_one_or_none() + if conv is not None: + return conv + + # Create the system channel + conv = CommConversation( + tenant_id=tenant_id, + title="System Channel", + is_pinned=False, + is_locked=True, + is_direct=False, + is_archived=False, + is_system=True, + created_by=None, + created_by_type="system", + metadata_={}, + ) + db.add(conv) + await db.flush() + + # Add all tenant users as participants + from app.models.user import User, UserTenant + + users_result = await db.execute( + select(User.id) + .join(UserTenant, UserTenant.user_id == User.id) + .where(UserTenant.tenant_id == tenant_id) + ) + user_ids = [row[0] for row in users_result.all()] + + for uid in user_ids: + p = CommParticipant( + tenant_id=tenant_id, + conversation_id=conv.id, + participant_id=uid, + participant_type="user", + role="member", + ) + db.add(p) + + await db.flush() + return conv + + +async def post_system_message( + db: AsyncSession, + tenant_id: uuid.UUID, + user_id: uuid.UUID, + message_type: str, + title: str, + body: str | None = None, + entity_type: str | None = None, + entity_id: uuid.UUID | None = None, + severity: str = "info", +) -> CommMessage | None: + """Post a typed system message to the tenant system channel. + + Creates a CommMessage in the system channel with: + - A text block containing title + body + - An action_card block with deep-link if entity_type/entity_id is set + - Block/message metadata: notification_type, severity, entity_ref + + Returns the created CommMessage, or None if the user has muted this type. + """ + # Check user preferences — reuse the notification preference system + from app.models.notification import NotificationPreference, NotificationType + + pref = await db.execute( + select(NotificationPreference).where( + and_( + NotificationPreference.user_id == user_id, + NotificationPreference.type_key == message_type, + NotificationPreference.tenant_id == tenant_id, + ) + ) + ) + pref_row = pref.scalar_one_or_none() + + if pref_row and not pref_row.is_enabled: + return None + + if not pref_row: + type_def = await db.execute( + select(NotificationType).where(NotificationType.type_key == message_type) + ) + type_row = type_def.scalar_one_or_none() + if type_row and not type_row.is_enabled_by_default: + return None + + # Get or create system channel + conv = await get_or_create_system_channel(db, tenant_id) + + # Build message content + content = title + if body: + content = f"{title}\n{body}" + + # Build metadata + msg_metadata: dict[str, Any] = { + "notification_type": message_type, + "severity": severity, + "target_user_id": str(user_id), + } + if entity_type and entity_id: + msg_metadata["entity_ref"] = { + "entity_type": entity_type, + "entity_id": str(entity_id), + } + + # Build blocks + blocks: list[dict[str, Any]] = [ + { + "block_type": "text", + "block_data": {"text": content, "title": title, "body": body or ""}, + } + ] + if entity_type and entity_id: + blocks.append( + { + "block_type": "action_card", + "block_data": { + "label": "Open", + "entity_type": entity_type, + "entity_id": str(entity_id), + }, + } + ) + + # Create message directly (not via send_message to avoid trigger_depth issues) + msg = CommMessage( + tenant_id=tenant_id, + conversation_id=conv.id, + sender_id=None, + sender_type="system", + content=content, + content_format="text", + metadata_=msg_metadata, + ) + db.add(msg) + await db.flush() + + # Create blocks + for i, block in enumerate(blocks): + b = CommMessageBlock( + tenant_id=tenant_id, + message_id=msg.id, + block_type=block["block_type"], + block_data=block["block_data"], + sort_order=i, + ) + db.add(b) + + await db.flush() + + # Update conversation last_msg + + await db.execute( + update(CommConversation) + .where(CommConversation.id == conv.id) + .values( + last_msg_at=datetime.now(UTC), + last_msg_preview=content[:200], + last_msg_sender_type="system", + ) + ) + + # Publish event + event_bus = get_event_bus() + await event_bus.publish("system.message.posted", { + "conversation_id": str(conv.id), + "message_id": str(msg.id), + "tenant_id": str(tenant_id), + "user_id": str(user_id), + "message_type": message_type, + "severity": severity, + }) + + return msg diff --git a/app/plugins/builtins/kommunikation/serializers.py b/app/plugins/builtins/kommunikation/serializers.py new file mode 100644 index 0000000..f653a7e --- /dev/null +++ b/app/plugins/builtins/kommunikation/serializers.py @@ -0,0 +1,129 @@ +"""Row/response serialization helpers for the kommunikation plugin.""" + +from __future__ import annotations + +import logging +import re +from typing import Any + +from app.plugins.builtins.kommunikation.models import ( + CommConversation, + CommMessage, + CommMessageAttachment, + CommMessageBlock, + CommMessageReaction, + CommParticipant, +) + +logger = logging.getLogger(__name__) + +MAX_TRIGGER_DEPTH = 3 + + +# ─── serializers ─── + +# ─── Mention Parsing ─── + +MENTION_PATTERN = re.compile(r"@(\w+)") + + +def parse_mentions(content: str) -> list[str]: + """Parse @mentions from message content. Returns list of mentioned types/names.""" + return MENTION_PATTERN.findall(content) + + +# ─── Conversation Helpers ─── + + +def conversation_to_response( + conv: CommConversation, + participants: list[CommParticipant], + unread_count: int = 0, + is_pinned_by_user: bool = False, +) -> dict[str, Any]: + """Convert a CommConversation to a response dict.""" + return { + "id": str(conv.id), + "title": conv.title, + "is_locked": conv.is_locked, + "locked_by": conv.locked_by, + "is_direct": conv.is_direct, + "is_archived": conv.is_archived, + "is_pinned": is_pinned_by_user, + "created_by": str(conv.created_by) if conv.created_by else None, + "created_by_type": conv.created_by_type, + "last_msg_at": conv.last_msg_at.isoformat() if conv.last_msg_at else None, + "last_msg_preview": conv.last_msg_preview, + "last_msg_sender_type": conv.last_msg_sender_type, + "participants": [participant_to_response(p) for p in participants], + "unread_count": unread_count, + "metadata": conv.metadata_ or {}, + } + + +def participant_to_response(p: CommParticipant) -> dict[str, Any]: + """Convert a CommParticipant to a response dict.""" + return { + "id": str(p.id), + "conversation_id": str(p.conversation_id), + "participant_id": str(p.participant_id) if p.participant_id else None, + "participant_type": p.participant_type, + "display_name": p.display_name, + "role": p.role, + "joined_at": p.joined_at.isoformat() if p.joined_at else None, + } + + +def message_to_response( + msg: CommMessage, + blocks: list[CommMessageBlock] | None = None, + attachments: list[CommMessageAttachment] | None = None, + reactions: list[CommMessageReaction] | None = None, +) -> dict[str, Any]: + """Convert a CommMessage to a response dict.""" + return { + "id": str(msg.id), + "conversation_id": str(msg.conversation_id), + "sender_id": str(msg.sender_id) if msg.sender_id else None, + "sender_type": msg.sender_type, + "content": msg.content, + "content_format": msg.content_format, + "metadata": msg.metadata_ or {}, + "reply_to_id": str(msg.reply_to_id) if msg.reply_to_id else None, + "is_pinned": msg.is_pinned, + "created_at": msg.created_at.isoformat() if msg.created_at else None, + "edited_at": msg.edited_at.isoformat() if msg.edited_at else None, + "blocks": [ + { + "id": str(b.id), + "block_type": b.block_type, + "block_data": b.block_data, + "sort_order": b.sort_order, + } + for b in (blocks or []) + ], + "attachments": [ + { + "id": str(a.id), + "file_id": str(a.file_id) if a.file_id else None, + "file_source": a.file_source, + "file_name": a.file_name, + "file_type": a.file_type, + "file_size": a.file_size, + "thumbnail_path": a.thumbnail_path, + } + for a in (attachments or []) + ], + "reactions": [ + { + "id": str(r.id), + "message_id": str(r.message_id), + "user_id": str(r.user_id), + "emoji": r.emoji, + } + for r in (reactions or []) + ], + } + + +# ─── Conversation CRUD ─── diff --git a/app/plugins/builtins/kommunikation/services.py b/app/plugins/builtins/kommunikation/services.py index ffe0b93..2a70c39 100644 --- a/app/plugins/builtins/kommunikation/services.py +++ b/app/plugins/builtins/kommunikation/services.py @@ -1,1364 +1,46 @@ -"""Business logic for the kommunikation plugin.""" +"""Business logic for the kommunikation plugin. -from __future__ import annotations +Re-export facade: implementations live in focused sub-modules +(serializers, conversations, participants, messages, interactions, +plugin_rooms). All public symbols stay importable from here. +""" -import logging -import re -import uuid -from datetime import UTC, datetime -from typing import Any - -from sqlalchemy import and_, func, select, update -from sqlalchemy.ext.asyncio import AsyncSession - -from app.core.event_bus import get_event_bus -from app.plugins.builtins.kommunikation.models import ( - CommConversation, - CommConversationMute, - CommConversationPin, - CommMessage, - CommMessageAttachment, - CommMessageBlock, - CommMessageEdit, - CommMessageReaction, - CommMessageRead, - CommParticipant, +from app.plugins.builtins.kommunikation.conversations import ( # noqa: F401 + create_conversation, + get_conversation, + list_conversations, + mute_conversation, + pin_conversation, + unmute_conversation, + unpin_conversation, + update_conversation, +) +from app.plugins.builtins.kommunikation.interactions import ( # noqa: F401 + add_reaction, + mark_read, + remove_reaction, +) +from app.plugins.builtins.kommunikation.messages import ( # noqa: F401 + MAX_TRIGGER_DEPTH, + delete_message, + edit_message, + get_messages, + send_message, +) +from app.plugins.builtins.kommunikation.participants import ( # noqa: F401 + add_participant, + change_role, + remove_participant, +) +from app.plugins.builtins.kommunikation.plugin_rooms import ( # noqa: F401 + create_plugin_room, + find_locked_room_id, + get_or_create_system_channel, + post_system_message, +) +from app.plugins.builtins.kommunikation.serializers import ( # noqa: F401 + conversation_to_response, + message_to_response, + parse_mentions, + participant_to_response, ) -from app.plugins.builtins.kommunikation.participant_registry import get_participant_registry -from app.plugins.builtins.kommunikation.rbac import CommRBAC - -logger = logging.getLogger(__name__) - -MAX_TRIGGER_DEPTH = 3 - - -# ─── Mention Parsing ─── - -MENTION_PATTERN = re.compile(r"@(\w+)") - - -def parse_mentions(content: str) -> list[str]: - """Parse @mentions from message content. Returns list of mentioned types/names.""" - return MENTION_PATTERN.findall(content) - - -# ─── Conversation Helpers ─── - - -def conversation_to_response( - conv: CommConversation, - participants: list[CommParticipant], - unread_count: int = 0, - is_pinned_by_user: bool = False, -) -> dict[str, Any]: - """Convert a CommConversation to a response dict.""" - return { - "id": str(conv.id), - "title": conv.title, - "is_locked": conv.is_locked, - "locked_by": conv.locked_by, - "is_direct": conv.is_direct, - "is_archived": conv.is_archived, - "is_pinned": is_pinned_by_user, - "created_by": str(conv.created_by) if conv.created_by else None, - "created_by_type": conv.created_by_type, - "last_msg_at": conv.last_msg_at.isoformat() if conv.last_msg_at else None, - "last_msg_preview": conv.last_msg_preview, - "last_msg_sender_type": conv.last_msg_sender_type, - "participants": [participant_to_response(p) for p in participants], - "unread_count": unread_count, - "metadata": conv.metadata_ or {}, - } - - -def participant_to_response(p: CommParticipant) -> dict[str, Any]: - """Convert a CommParticipant to a response dict.""" - return { - "id": str(p.id), - "conversation_id": str(p.conversation_id), - "participant_id": str(p.participant_id) if p.participant_id else None, - "participant_type": p.participant_type, - "display_name": p.display_name, - "role": p.role, - "joined_at": p.joined_at.isoformat() if p.joined_at else None, - } - - -def message_to_response( - msg: CommMessage, - blocks: list[CommMessageBlock] | None = None, - attachments: list[CommMessageAttachment] | None = None, - reactions: list[CommMessageReaction] | None = None, -) -> dict[str, Any]: - """Convert a CommMessage to a response dict.""" - return { - "id": str(msg.id), - "conversation_id": str(msg.conversation_id), - "sender_id": str(msg.sender_id) if msg.sender_id else None, - "sender_type": msg.sender_type, - "content": msg.content, - "content_format": msg.content_format, - "metadata": msg.metadata_ or {}, - "reply_to_id": str(msg.reply_to_id) if msg.reply_to_id else None, - "is_pinned": msg.is_pinned, - "created_at": msg.created_at.isoformat() if msg.created_at else None, - "edited_at": msg.edited_at.isoformat() if msg.edited_at else None, - "blocks": [ - { - "id": str(b.id), - "block_type": b.block_type, - "block_data": b.block_data, - "sort_order": b.sort_order, - } - for b in (blocks or []) - ], - "attachments": [ - { - "id": str(a.id), - "file_id": str(a.file_id) if a.file_id else None, - "file_source": a.file_source, - "file_name": a.file_name, - "file_type": a.file_type, - "file_size": a.file_size, - "thumbnail_path": a.thumbnail_path, - } - for a in (attachments or []) - ], - "reactions": [ - { - "id": str(r.id), - "message_id": str(r.message_id), - "user_id": str(r.user_id), - "emoji": r.emoji, - } - for r in (reactions or []) - ], - } - - -# ─── Conversation CRUD ─── - - -async def list_conversations( - db: AsyncSession, - tenant_id: uuid.UUID, - user_id: uuid.UUID, - include_archived: bool = False, -) -> list[dict[str, Any]]: - """List all conversations for a user.""" - # Get conversations where user is a participant - result = await db.execute( - select(CommConversation) - .join(CommParticipant, CommParticipant.conversation_id == CommConversation.id) - .where( - CommParticipant.participant_id == user_id, - CommParticipant.participant_type == "user", - CommParticipant.left_at.is_(None), - CommConversation.tenant_id == tenant_id, - CommConversation.deleted_at.is_(None), - ) - .order_by(CommConversation.last_msg_at.desc().nullslast()) - ) - conversations = result.scalars().all() - - # Get user's pinned conversations - pins_result = await db.execute( - select(CommConversationPin).where( - CommConversationPin.user_id == user_id, - CommConversationPin.tenant_id == tenant_id, - ) - ) - pinned_ids = {p.conversation_id for p in pins_result.scalars().all()} - - conv_list = [] - for conv in conversations: - if conv.is_archived and not include_archived: - continue - # Get participants - parts_result = await db.execute( - select(CommParticipant).where( - CommParticipant.conversation_id == conv.id, - CommParticipant.left_at.is_(None), - ) - ) - participants = list(parts_result.scalars().all()) - - # Get unread count - unread = await _get_unread_count(db, tenant_id, conv.id, user_id) - - conv_list.append( - conversation_to_response( - conv, participants, unread_count=unread, is_pinned_by_user=conv.id in pinned_ids - ) - ) - - # Sort: pinned first, then by last_msg_at - conv_list.sort(key=lambda c: (not c["is_pinned"], c["last_msg_at"] or ""), reverse=False) - # Actually: pinned first (True > False in reverse), then newest first - conv_list.sort(key=lambda c: c["last_msg_at"] or "0000", reverse=True) - conv_list.sort(key=lambda c: c["is_pinned"], reverse=True) - - return conv_list - - -async def get_conversation( - db: AsyncSession, - tenant_id: uuid.UUID, - conversation_id: uuid.UUID, - user_id: uuid.UUID, -) -> dict[str, Any] | None: - """Get a single conversation with participants.""" - result = await db.execute( - select(CommConversation).where( - CommConversation.id == conversation_id, - CommConversation.tenant_id == tenant_id, - CommConversation.deleted_at.is_(None), - ) - ) - conv = result.scalar_one_or_none() - if conv is None: - return None - - # Check user is participant - if not await CommRBAC.is_participant(db, conversation_id, user_id): - return None - - parts_result = await db.execute( - select(CommParticipant).where( - CommParticipant.conversation_id == conv.id, - CommParticipant.left_at.is_(None), - ) - ) - participants = list(parts_result.scalars().all()) - - # Check pinned - pin_result = await db.execute( - select(CommConversationPin).where( - CommConversationPin.conversation_id == conv.id, - CommConversationPin.user_id == user_id, - ) - ) - is_pinned = pin_result.scalar_one_or_none() is not None - - unread = await _get_unread_count(db, tenant_id, conv.id, user_id) - - return conversation_to_response(conv, participants, unread_count=unread, is_pinned_by_user=is_pinned) - - -async def create_conversation( - db: AsyncSession, - tenant_id: uuid.UUID, - user_id: uuid.UUID, - title: str | None = None, - participant_ids: list[str] | None = None, - is_direct: bool = False, - initial_message: str | None = None, - metadata: dict[str, Any] | None = None, -) -> dict[str, Any]: - """Create a new conversation.""" - conv = CommConversation( - tenant_id=tenant_id, - title=title, - owner_id=user_id, - is_direct=is_direct, - created_by=user_id, - created_by_type="user", - metadata_=metadata or {}, - ) - from app.core.hooks import do_action - await do_action("comm.conversation.before_create", tenant_id=tenant_id, user_id=user_id) - db.add(conv) - await db.flush() - await do_action("comm.conversation.after_create", conversation_id=conv.id, tenant_id=tenant_id, user_id=user_id) - - # Add creator as admin - creator = CommParticipant( - tenant_id=tenant_id, - conversation_id=conv.id, - participant_id=user_id, - participant_type="user", - role="admin", - ) - db.add(creator) - - # Add other participants - for pid_str in (participant_ids or []): - try: - pid = uuid.UUID(pid_str) - if pid == user_id: - continue - p = CommParticipant( - tenant_id=tenant_id, - conversation_id=conv.id, - participant_id=pid, - participant_type="user", - role="member", - ) - db.add(p) - except ValueError: - logger.warning(f"Invalid participant UUID: {pid_str}") - - await db.flush() - - # Send initial message if provided - if initial_message: - await send_message( - db, tenant_id, conv.id, user_id, "user", - content=initial_message, content_format="text", - ) - - # Publish event - event_bus = get_event_bus() - await event_bus.publish("conversation.created", { - "conversation_id": str(conv.id), - "tenant_id": str(tenant_id), - "created_by": str(user_id), - }) - - # Get all participants for response - parts_result = await db.execute( - select(CommParticipant).where( - CommParticipant.conversation_id == conv.id, - CommParticipant.left_at.is_(None), - ) - ) - participants = list(parts_result.scalars().all()) - - return conversation_to_response(conv, participants) - - -async def update_conversation( - db: AsyncSession, - tenant_id: uuid.UUID, - conversation_id: uuid.UUID, - user_id: uuid.UUID, - title: str | None = None, - is_archived: bool | None = None, -) -> dict[str, Any] | None: - """Update a conversation.""" - result = await db.execute( - select(CommConversation).where( - CommConversation.id == conversation_id, - CommConversation.tenant_id == tenant_id, - CommConversation.deleted_at.is_(None), - ) - ) - conv = result.scalar_one_or_none() - if conv is None: - return None - - # Check locked - if conv.is_locked and title is not None: - # Only the locking plugin can change title on locked conversations - # Users cannot - pass - elif title is not None: - conv.title = title - conv.title_set_by = user_id - - if is_archived is not None: - conv.is_archived = is_archived - - await db.flush() - - return await get_conversation(db, tenant_id, conversation_id, user_id) - - -async def pin_conversation( - db: AsyncSession, - tenant_id: uuid.UUID, - conversation_id: uuid.UUID, - user_id: uuid.UUID, -) -> bool: - """Pin a conversation for a user.""" - existing = await db.execute( - select(CommConversationPin).where( - CommConversationPin.conversation_id == conversation_id, - CommConversationPin.user_id == user_id, - ) - ) - if existing.scalar_one_or_none() is None: - pin = CommConversationPin( - tenant_id=tenant_id, - conversation_id=conversation_id, - user_id=user_id, - ) - db.add(pin) - await db.flush() - return True - - -async def unpin_conversation( - db: AsyncSession, - conversation_id: uuid.UUID, - user_id: uuid.UUID, -) -> bool: - """Unpin a conversation for a user.""" - result = await db.execute( - select(CommConversationPin).where( - CommConversationPin.conversation_id == conversation_id, - CommConversationPin.user_id == user_id, - ) - ) - pin = result.scalar_one_or_none() - if pin: - await db.delete(pin) - await db.flush() - return True - - -async def mute_conversation( - db: AsyncSession, - tenant_id: uuid.UUID, - conversation_id: uuid.UUID, - user_id: uuid.UUID, -) -> bool: - """Mute a conversation for a user.""" - existing = await db.execute( - select(CommConversationMute).where( - CommConversationMute.conversation_id == conversation_id, - CommConversationMute.user_id == user_id, - ) - ) - if existing.scalar_one_or_none() is None: - mute = CommConversationMute( - tenant_id=tenant_id, - conversation_id=conversation_id, - user_id=user_id, - ) - db.add(mute) - await db.flush() - return True - - -async def unmute_conversation( - db: AsyncSession, - conversation_id: uuid.UUID, - user_id: uuid.UUID, -) -> bool: - """Unmute a conversation for a user.""" - result = await db.execute( - select(CommConversationMute).where( - CommConversationMute.conversation_id == conversation_id, - CommConversationMute.user_id == user_id, - ) - ) - mute = result.scalar_one_or_none() - if mute: - await db.delete(mute) - await db.flush() - return True - - -# ─── Participant Management ─── - - -async def add_participant( - db: AsyncSession, - tenant_id: uuid.UUID, - conversation_id: uuid.UUID, - participant_id: str, - participant_type: str = "user", - role: str = "member", - display_name: str | None = None, -) -> dict[str, Any] | None: - """Add a participant to a conversation.""" - try: - pid = uuid.UUID(participant_id) if participant_type == "user" else None - except ValueError: - return None - - existing = await db.execute( - select(CommParticipant).where( - CommParticipant.conversation_id == conversation_id, - CommParticipant.participant_id == pid if pid else CommParticipant.participant_type == participant_type, - CommParticipant.participant_type == participant_type, - CommParticipant.left_at.is_(None), - ) - ) - if existing.scalar_one_or_none() is not None: - return None # Already a participant - - p = CommParticipant( - tenant_id=tenant_id, - conversation_id=conversation_id, - participant_id=pid, - participant_type=participant_type, - role=role, - display_name=display_name, - ) - db.add(p) - await db.flush() - - # Publish event - event_bus = get_event_bus() - await event_bus.publish("participant.joined", { - "conversation_id": str(conversation_id), - "participant_id": participant_id, - "participant_type": participant_type, - "tenant_id": str(tenant_id), - }) - - return participant_to_response(p) - - -async def remove_participant( - db: AsyncSession, - conversation_id: uuid.UUID, - participant_id: uuid.UUID, -) -> bool: - """Remove a participant from a conversation (set left_at).""" - result = await db.execute( - select(CommParticipant).where( - CommParticipant.conversation_id == conversation_id, - CommParticipant.participant_id == participant_id, - CommParticipant.participant_type == "user", - CommParticipant.left_at.is_(None), - ) - ) - p = result.scalar_one_or_none() - if p is None: - return False - p.left_at = datetime.now(UTC) - await db.flush() - - event_bus = get_event_bus() - await event_bus.publish("participant.left", { - "conversation_id": str(conversation_id), - "participant_id": str(participant_id), - }) - return True - - -async def change_role( - db: AsyncSession, - conversation_id: uuid.UUID, - participant_id: uuid.UUID, - new_role: str, -) -> dict[str, Any] | None: - """Change a participant's role.""" - result = await db.execute( - select(CommParticipant).where( - CommParticipant.conversation_id == conversation_id, - CommParticipant.participant_id == participant_id, - CommParticipant.participant_type == "user", - CommParticipant.left_at.is_(None), - ) - ) - p = result.scalar_one_or_none() - if p is None: - return None - p.role = new_role - await db.flush() - return participant_to_response(p) - - -# ─── Messages ─── - - -async def get_messages( - db: AsyncSession, - tenant_id: uuid.UUID, - conversation_id: uuid.UUID, - page: int = 1, - page_size: int = 50, - before_id: uuid.UUID | None = None, -) -> dict[str, Any]: - """Get paginated messages for a conversation.""" - query = select(CommMessage).where( - CommMessage.conversation_id == conversation_id, - CommMessage.tenant_id == tenant_id, - CommMessage.deleted_at.is_(None), - ).order_by(CommMessage.created_at.desc()) - - if before_id: - before_msg = await db.execute( - select(CommMessage).where(CommMessage.id == before_id) - ) - before = before_msg.scalar_one_or_none() - if before: - query = query.where(CommMessage.created_at < before.created_at) - - query = query.offset((page - 1) * page_size).limit(page_size) - result = await db.execute(query) - messages = list(result.scalars().all()) - - # Get blocks, attachments, reactions for each message - msg_ids = [m.id for m in messages] - blocks_map: dict[uuid.UUID, list] = {} - attachments_map: dict[uuid.UUID, list] = {} - reactions_map: dict[uuid.UUID, list] = {} - - if msg_ids: - blocks_result = await db.execute( - select(CommMessageBlock).where( - CommMessageBlock.message_id.in_(msg_ids), - CommMessageBlock.deleted_at.is_(None), - ).order_by(CommMessageBlock.sort_order) - ) - for b in blocks_result.scalars().all(): - blocks_map.setdefault(b.message_id, []).append(b) - - atts_result = await db.execute( - select(CommMessageAttachment).where( - CommMessageAttachment.message_id.in_(msg_ids), - CommMessageAttachment.deleted_at.is_(None), - ) - ) - for a in atts_result.scalars().all(): - attachments_map.setdefault(a.message_id, []).append(a) - - reactions_result = await db.execute( - select(CommMessageReaction).where( - CommMessageReaction.message_id.in_(msg_ids), - ) - ) - for r in reactions_result.scalars().all(): - reactions_map.setdefault(r.message_id, []).append(r) - - items = [] - for msg in reversed(messages): # chronological order - items.append( - message_to_response( - msg, - blocks=blocks_map.get(msg.id, []), - attachments=attachments_map.get(msg.id, []), - reactions=reactions_map.get(msg.id, []), - ) - ) - - # Total count - count_result = await db.execute( - select(func.count()).select_from(CommMessage).where( - CommMessage.conversation_id == conversation_id, - CommMessage.tenant_id == tenant_id, - CommMessage.deleted_at.is_(None), - ) - ) - total = count_result.scalar() or 0 - - has_more = (page * page_size) < total - - return {"items": items, "total": total, "page": page, "has_more": has_more} - - -async def send_message( - db: AsyncSession, - tenant_id: uuid.UUID, - conversation_id: uuid.UUID, - sender_id: uuid.UUID | None, - sender_type: str, - content: str = "", - content_format: str = "text", - blocks: list[dict[str, Any]] | None = None, - reply_to_id: str | None = None, - attachments: list[dict[str, Any]] | None = None, - metadata: dict[str, Any] | None = None, - trigger_depth: int = 0, -) -> dict[str, Any]: - """Send a message to a conversation and trigger participant handlers.""" - # Create message - msg = CommMessage( - tenant_id=tenant_id, - conversation_id=conversation_id, - sender_id=sender_id, - sender_type=sender_type, - content=content, - content_format=content_format, - metadata_=metadata or {}, - ) - if reply_to_id: - try: - msg.reply_to_id = uuid.UUID(reply_to_id) - except ValueError: - pass - - from app.core.hooks import do_action - await do_action("comm.before_message", conversation_id=conversation_id, tenant_id=tenant_id, sender_id=sender_id) - db.add(msg) - await db.flush() - await do_action("comm.after_message", message_id=msg.id, conversation_id=conversation_id, tenant_id=tenant_id, sender_id=sender_id) - - # Create blocks - if blocks: - for i, block in enumerate(blocks): - b = CommMessageBlock( - tenant_id=tenant_id, - message_id=msg.id, - block_type=block["block_type"], - block_data=block["block_data"], - sort_order=i, - ) - db.add(b) - - # Create attachments - if attachments: - for att in attachments: - a = CommMessageAttachment( - tenant_id=tenant_id, - message_id=msg.id, - file_id=uuid.UUID(att["file_id"]) if att.get("file_id") else None, - file_source=att.get("file_source", "comm"), - file_name=att.get("file_name", ""), - file_type=att.get("file_type", "application/octet-stream"), - file_size=att.get("file_size"), - ) - db.add(a) - - await db.flush() - - # Update conversation last_msg - await db.execute( - update(CommConversation) - .where(CommConversation.id == conversation_id) - .values( - last_msg_at=datetime.now(UTC), - last_msg_preview=content[:200] if content else "", - last_msg_sender_type=sender_type, - ) - ) - - # Publish event - event_bus = get_event_bus() - await event_bus.publish("message.received", { - "conversation_id": str(conversation_id), - "message_id": str(msg.id), - "sender_type": sender_type, - "tenant_id": str(tenant_id), - "content": content, - "trigger_depth": trigger_depth, - }) - - # Trigger participant handlers (if not at max depth) - if trigger_depth < MAX_TRIGGER_DEPTH: - await _trigger_participants( - db, tenant_id, conversation_id, msg, trigger_depth - ) - - # Load blocks/attachments/reactions for response - blocks_result = await db.execute( - select(CommMessageBlock).where( - CommMessageBlock.message_id == msg.id, - CommMessageBlock.deleted_at.is_(None), - ).order_by(CommMessageBlock.sort_order) - ) - msg_blocks = list(blocks_result.scalars().all()) - - atts_result = await db.execute( - select(CommMessageAttachment).where( - CommMessageAttachment.message_id == msg.id, - CommMessageAttachment.deleted_at.is_(None), - ) - ) - msg_atts = list(atts_result.scalars().all()) - - return message_to_response(msg, blocks=msg_blocks, attachments=msg_atts) - - -async def _trigger_participants( - db: AsyncSession, - tenant_id: uuid.UUID, - conversation_id: uuid.UUID, - message: CommMessage, - trigger_depth: int, -) -> None: - """Trigger participant handlers for non-user participants.""" - # Get conversation participants - result = await db.execute( - select(CommParticipant).where( - CommParticipant.conversation_id == conversation_id, - CommParticipant.left_at.is_(None), - CommParticipant.participant_type != "user", - ) - ) - non_user_participants = list(result.scalars().all()) - - if not non_user_participants: - return - - # Get conversation info - conv_result = await db.execute( - select(CommConversation).where(CommConversation.id == conversation_id) - ) - conv = conv_result.scalar_one_or_none() - if conv is None: - return - - # Parse mentions - mentions = parse_mentions(message.content) - - # Build conversation dict - all_parts_result = await db.execute( - select(CommParticipant).where( - CommParticipant.conversation_id == conversation_id, - CommParticipant.left_at.is_(None), - ) - ) - all_parts = [participant_to_response(p) for p in all_parts_result.scalars().all()] - conv_dict = conversation_to_response(conv, []) - conv_dict["participants"] = all_parts - - msg_dict = message_to_response(message) - context = {"tenant_id": str(tenant_id), "trigger_depth": trigger_depth} - - registry = get_participant_registry() - - for p in non_user_participants: - handler = registry.get_handler(p.participant_type) - if handler is None: - continue - - try: - responses = await handler.on_message_received( - conversation_id=conversation_id, - message=msg_dict, - conversation=conv_dict, - mentions=mentions, - context=context, - ) - - if responses: - for resp in responses: - await send_message( - db, - tenant_id, - conversation_id, - sender_id=None, - sender_type=p.participant_type, - content=resp.get("content", ""), - content_format=resp.get("content_format", "text"), - blocks=resp.get("blocks"), - metadata={ - **(resp.get("metadata") or {}), - "triggered_by": str(message.id), - "trigger_depth": trigger_depth + 1, - }, - trigger_depth=trigger_depth + 1, - ) - except Exception: - logger.exception( - f"Participant handler error for type {p.participant_type}" - ) - - -async def edit_message( - db: AsyncSession, - tenant_id: uuid.UUID, - message_id: uuid.UUID, - user_id: uuid.UUID, - new_content: str, -) -> dict[str, Any] | None: - """Edit a message, storing the old version in history.""" - result = await db.execute( - select(CommMessage).where( - CommMessage.id == message_id, - CommMessage.tenant_id == tenant_id, - CommMessage.deleted_at.is_(None), - ) - ) - msg = result.scalar_one_or_none() - if msg is None: - return None - - # Get old blocks - blocks_result = await db.execute( - select(CommMessageBlock).where( - CommMessageBlock.message_id == message_id, - CommMessageBlock.deleted_at.is_(None), - ) - ) - old_blocks = [b.block_data for b in blocks_result.scalars().all()] - - # Save edit history - edit = CommMessageEdit( - tenant_id=tenant_id, - message_id=message_id, - old_content=msg.content, - old_blocks=old_blocks, - edited_by=user_id, - ) - db.add(edit) - - from app.core.hooks import do_action - await do_action("comm.before_edit", message_id=message_id, tenant_id=tenant_id, user_id=user_id) - - # Update message - msg.content = new_content - msg.edited_at = datetime.now(UTC) - await db.flush() - await do_action("comm.after_edit", message_id=message_id, tenant_id=tenant_id, user_id=user_id) - - return message_to_response(msg) - - -async def delete_message( - db: AsyncSession, - message_id: uuid.UUID, -) -> bool: - """Soft-delete a message.""" - result = await db.execute( - select(CommMessage).where(CommMessage.id == message_id) - ) - msg = result.scalar_one_or_none() - if msg is None: - return False - from app.core.hooks import do_action - await do_action("comm.before_delete", message_id=message_id) - msg.deleted_at = datetime.now(UTC) - await db.flush() - await do_action("comm.after_delete", message_id=message_id) - return True - - -# ─── Reactions ─── - - -async def add_reaction( - db: AsyncSession, - tenant_id: uuid.UUID, - message_id: uuid.UUID, - user_id: uuid.UUID, - emoji: str, -) -> dict[str, Any] | None: - """Add an emoji reaction to a message.""" - existing = await db.execute( - select(CommMessageReaction).where( - CommMessageReaction.message_id == message_id, - CommMessageReaction.user_id == user_id, - CommMessageReaction.emoji == emoji, - ) - ) - if existing.scalar_one_or_none() is not None: - return None # Already reacted - - r = CommMessageReaction( - tenant_id=tenant_id, - message_id=message_id, - user_id=user_id, - emoji=emoji, - ) - db.add(r) - await db.flush() - - event_bus = get_event_bus() - await event_bus.publish("reaction.added", { - "message_id": str(message_id), - "emoji": emoji, - "user_id": str(user_id), - }) - - return { - "id": str(r.id), - "message_id": str(r.message_id), - "user_id": str(r.user_id), - "emoji": r.emoji, - } - - -async def remove_reaction( - db: AsyncSession, - message_id: uuid.UUID, - user_id: uuid.UUID, - emoji: str, -) -> bool: - """Remove an emoji reaction.""" - result = await db.execute( - select(CommMessageReaction).where( - CommMessageReaction.message_id == message_id, - CommMessageReaction.user_id == user_id, - CommMessageReaction.emoji == emoji, - ) - ) - r = result.scalar_one_or_none() - if r is None: - return False - await db.delete(r) - await db.flush() - return True - - -# ─── Read State ─── - - -async def mark_read( - db: AsyncSession, - tenant_id: uuid.UUID, - conversation_id: uuid.UUID, - user_id: uuid.UUID, - last_read_msg_id: str | None = None, -) -> bool: - """Mark conversation as read up to a message.""" - result = await db.execute( - select(CommMessageRead).where( - CommMessageRead.conversation_id == conversation_id, - CommMessageRead.user_id == user_id, - ) - ) - read = result.scalar_one_or_none() - - msg_id = uuid.UUID(last_read_msg_id) if last_read_msg_id else None - - if read is None: - read = CommMessageRead( - tenant_id=tenant_id, - conversation_id=conversation_id, - user_id=user_id, - last_read_msg_id=msg_id, - ) - db.add(read) - else: - read.last_read_msg_id = msg_id - read.last_read_at = datetime.now(UTC) - - await db.flush() - return True - - -async def _get_unread_count( - db: AsyncSession, - tenant_id: uuid.UUID, - conversation_id: uuid.UUID, - user_id: uuid.UUID, -) -> int: - """Get unread message count for a user in a conversation.""" - # Get last read message - read_result = await db.execute( - select(CommMessageRead).where( - CommMessageRead.conversation_id == conversation_id, - CommMessageRead.user_id == user_id, - ) - ) - read = read_result.scalar_one_or_none() - - query = select(func.count()).select_from(CommMessage).where( - CommMessage.conversation_id == conversation_id, - CommMessage.tenant_id == tenant_id, - CommMessage.deleted_at.is_(None), - CommMessage.sender_type != "system", # Don't count system messages? Or count all? - ) - - if read and read.last_read_at: - query = query.where(CommMessage.created_at > read.last_read_at) - - result = await db.execute(query) - return result.scalar() or 0 - - -# ─── Plugin Room Creation ─── - - -async def find_locked_room_id( - db: AsyncSession, - tenant_id: uuid.UUID, - plugin_name: str, - title: str, -) -> uuid.UUID | None: - """Find the conversation ID of a locked plugin room by tenant and title. - - Matches the same room semantics as ``create_plugin_room``: locked rooms - are owned by the plugin (``locked_by == plugin_name``) and soft-deleted - conversations are excluded. Returns ``None`` when no room exists. - """ - result = await db.execute( - select(CommConversation.id).where( - CommConversation.tenant_id == tenant_id, - CommConversation.title == title, - CommConversation.is_locked.is_(True), - CommConversation.locked_by == plugin_name, - CommConversation.deleted_at.is_(None), - ) - ) - return result.scalar_one_or_none() - - -async def create_plugin_room( - db: AsyncSession, - tenant_id: uuid.UUID, - user_id: uuid.UUID, - plugin_name: str, - title: str, - participant_type: str, - user_role: str = "member", -) -> dict[str, Any]: - """Create a locked, pinned room for a plugin (System, Live KI, Assistent). - - The room is locked (users can't change title/participants) and pinned for the user. - """ - # Check if room already exists for this user + plugin - result = await db.execute( - select(CommConversation).where( - CommConversation.tenant_id == tenant_id, - CommConversation.title == title, - CommConversation.is_locked.is_(True), - CommConversation.locked_by == plugin_name, - CommConversation.deleted_at.is_(None), - ).join(CommParticipant, CommParticipant.conversation_id == CommConversation.id).where( - CommParticipant.participant_id == user_id, - CommParticipant.participant_type == "user", - CommParticipant.left_at.is_(None), - ) - ) - existing = result.scalar_one_or_none() - if existing: - # Already exists — return it - return await get_conversation(db, tenant_id, existing.id, user_id) or {} - - # Create conversation - conv = CommConversation( - tenant_id=tenant_id, - title=title, - owner_id=user_id, - is_locked=True, - locked_by=plugin_name, - is_direct=False, - created_by=None, - created_by_type="plugin", - metadata_={"plugin": plugin_name}, - ) - db.add(conv) - await db.flush() - - # Add plugin as participant - plugin_p = CommParticipant( - tenant_id=tenant_id, - conversation_id=conv.id, - participant_id=None, - participant_type=participant_type, - role="admin", - display_name=title, - ) - db.add(plugin_p) - - # Add user as participant - user_p = CommParticipant( - tenant_id=tenant_id, - conversation_id=conv.id, - participant_id=user_id, - participant_type="user", - role=user_role, - ) - db.add(user_p) - - # Pin for user - pin = CommConversationPin( - tenant_id=tenant_id, - conversation_id=conv.id, - user_id=user_id, - ) - db.add(pin) - - await db.flush() - - # Publish event - event_bus = get_event_bus() - await event_bus.publish("conversation.created", { - "conversation_id": str(conv.id), - "tenant_id": str(tenant_id), - "created_by_type": "plugin", - "plugin_name": plugin_name, - }) - - parts = [plugin_p, user_p] - return conversation_to_response(conv, parts, is_pinned_by_user=True) - - - -# ─── System Channel ─── - - -async def get_or_create_system_channel( - db: AsyncSession, - tenant_id: uuid.UUID, -) -> CommConversation: - """Get or create the tenant-wide system channel. - - The system channel is a locked, is_system=True conversation that serves as - the central destination for system notifications, user alerts, and agent messages. - All users of the tenant are automatically added as participants. - """ - result = await db.execute( - select(CommConversation).where( - CommConversation.tenant_id == tenant_id, - CommConversation.is_system.is_(True), - CommConversation.deleted_at.is_(None), - ) - ) - conv = result.scalar_one_or_none() - if conv is not None: - return conv - - # Create the system channel - conv = CommConversation( - tenant_id=tenant_id, - title="System Channel", - is_pinned=False, - is_locked=True, - is_direct=False, - is_archived=False, - is_system=True, - created_by=None, - created_by_type="system", - metadata_={}, - ) - db.add(conv) - await db.flush() - - # Add all tenant users as participants - from app.models.user import User, UserTenant - - users_result = await db.execute( - select(User.id) - .join(UserTenant, UserTenant.user_id == User.id) - .where(UserTenant.tenant_id == tenant_id) - ) - user_ids = [row[0] for row in users_result.all()] - - for uid in user_ids: - p = CommParticipant( - tenant_id=tenant_id, - conversation_id=conv.id, - participant_id=uid, - participant_type="user", - role="member", - ) - db.add(p) - - await db.flush() - return conv - - -async def post_system_message( - db: AsyncSession, - tenant_id: uuid.UUID, - user_id: uuid.UUID, - message_type: str, - title: str, - body: str | None = None, - entity_type: str | None = None, - entity_id: uuid.UUID | None = None, - severity: str = "info", -) -> CommMessage | None: - """Post a typed system message to the tenant system channel. - - Creates a CommMessage in the system channel with: - - A text block containing title + body - - An action_card block with deep-link if entity_type/entity_id is set - - Block/message metadata: notification_type, severity, entity_ref - - Returns the created CommMessage, or None if the user has muted this type. - """ - # Check user preferences — reuse the notification preference system - from app.models.notification import NotificationPreference, NotificationType - - pref = await db.execute( - select(NotificationPreference).where( - and_( - NotificationPreference.user_id == user_id, - NotificationPreference.type_key == message_type, - NotificationPreference.tenant_id == tenant_id, - ) - ) - ) - pref_row = pref.scalar_one_or_none() - - if pref_row and not pref_row.is_enabled: - return None - - if not pref_row: - type_def = await db.execute( - select(NotificationType).where(NotificationType.type_key == message_type) - ) - type_row = type_def.scalar_one_or_none() - if type_row and not type_row.is_enabled_by_default: - return None - - # Get or create system channel - conv = await get_or_create_system_channel(db, tenant_id) - - # Build message content - content = title - if body: - content = f"{title}\n{body}" - - # Build metadata - msg_metadata: dict[str, Any] = { - "notification_type": message_type, - "severity": severity, - "target_user_id": str(user_id), - } - if entity_type and entity_id: - msg_metadata["entity_ref"] = { - "entity_type": entity_type, - "entity_id": str(entity_id), - } - - # Build blocks - blocks: list[dict[str, Any]] = [ - { - "block_type": "text", - "block_data": {"text": content, "title": title, "body": body or ""}, - } - ] - if entity_type and entity_id: - blocks.append( - { - "block_type": "action_card", - "block_data": { - "label": "Open", - "entity_type": entity_type, - "entity_id": str(entity_id), - }, - } - ) - - # Create message directly (not via send_message to avoid trigger_depth issues) - msg = CommMessage( - tenant_id=tenant_id, - conversation_id=conv.id, - sender_id=None, - sender_type="system", - content=content, - content_format="text", - metadata_=msg_metadata, - ) - db.add(msg) - await db.flush() - - # Create blocks - for i, block in enumerate(blocks): - b = CommMessageBlock( - tenant_id=tenant_id, - message_id=msg.id, - block_type=block["block_type"], - block_data=block["block_data"], - sort_order=i, - ) - db.add(b) - - await db.flush() - - # Update conversation last_msg - from datetime import datetime - - await db.execute( - update(CommConversation) - .where(CommConversation.id == conv.id) - .values( - last_msg_at=datetime.now(UTC), - last_msg_preview=content[:200], - last_msg_sender_type="system", - ) - ) - - # Publish event - event_bus = get_event_bus() - await event_bus.publish("system.message.posted", { - "conversation_id": str(conv.id), - "message_id": str(msg.id), - "tenant_id": str(tenant_id), - "user_id": str(user_id), - "message_type": message_type, - "severity": severity, - }) - - return msg