refactor(kommunikation): split god-object services.py into 6 focused sub-modules with re-export facade — behavior identical (comm suite 132P/1F/6E pre-existing, failures byte-identical to pre-split baseline)
Check Cross-Plugin Imports / check (push) Has been cancelled
Check Cross-Plugin Imports / check (push) Has been cancelled
This commit is contained in:
Agent Zero
@@ -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 ───
|
||||
Reference in New Issue
Block a user