diff --git a/alembic/versions/0035_comm_search_index.py b/alembic/versions/0035_comm_search_index.py new file mode 100644 index 0000000..424b6dc --- /dev/null +++ b/alembic/versions/0035_comm_search_index.py @@ -0,0 +1,51 @@ +"""Add search_tsv and embedding columns to comm_messages for full-text search indexing. + +Revision ID: 0035 +Revises: 0034_automation_config +Create Date: 2026-07-25 +""" + +from __future__ import annotations + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects.postgresql import TSVECTOR + +revision: str = "0035_comm_search_index" +down_revision: Union[str, None] = "0034_automation_config" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # Add search_tsv column for full-text search + op.add_column( + "comm_messages", + sa.Column("search_tsv", TSVECTOR, nullable=True), + ) + # Add embedding column for vector search (768 dimensions matching pgvector) + op.execute( + "ALTER TABLE comm_messages ADD COLUMN embedding vector(768)" + ) + # Create GIN index on search_tsv for fast FTS queries + op.create_index( + "ix_comm_messages_search_tsv", + "comm_messages", + ["search_tsv"], + postgresql_using="gin", + ) + # Create IVFFlat index on embedding for fast vector search + op.execute( + "CREATE INDEX IF NOT EXISTS ix_comm_messages_embedding " + "ON comm_messages USING ivfflat (embedding vector_cosine_ops) " + "WITH (lists = 100)" + ) + + +def downgrade() -> None: + op.drop_index("ix_comm_messages_embedding", table_name="comm_messages") + op.drop_index("ix_comm_messages_search_tsv", table_name="comm_messages") + op.drop_column("comm_messages", "embedding") + op.drop_column("comm_messages", "search_tsv") diff --git a/app/plugins/builtins/dms/routes.py b/app/plugins/builtins/dms/routes.py index 1b076cc..1f7eeb5 100644 --- a/app/plugins/builtins/dms/routes.py +++ b/app/plugins/builtins/dms/routes.py @@ -878,7 +878,7 @@ async def share_file( perm = Permission( tenant_id=tenant_id, file_id=fid, - user_id=uuid.uuid4(), # placeholder user_id for group shares + user_id=uuid.UUID(current_user["user_id"]), group_id=gid, access_level=body.access_level, ) diff --git a/app/plugins/builtins/kommunikation/search_provider.py b/app/plugins/builtins/kommunikation/search_provider.py index 1b59563..ab661e7 100644 --- a/app/plugins/builtins/kommunikation/search_provider.py +++ b/app/plugins/builtins/kommunikation/search_provider.py @@ -1,4 +1,4 @@ -"""Search provider for unified_search integration.""" +"""Search provider for unified_search integration — FTS + vector search for messages.""" from __future__ import annotations @@ -6,7 +6,7 @@ import logging import uuid from typing import Any -from sqlalchemy import select, func, or_ +from sqlalchemy import select, func, or_, text as sql_text from sqlalchemy.ext.asyncio import AsyncSession from app.plugins.builtins.kommunikation.models import ( @@ -14,6 +14,7 @@ from app.plugins.builtins.kommunikation.models import ( CommMessage, CommParticipant, ) +from app.plugins.builtins.unified_search.embedding import generate_embedding logger = logging.getLogger(__name__) @@ -21,6 +22,8 @@ logger = logging.getLogger(__name__) class CommSearchProvider: """Provider for unified_search — searches conversations and messages.""" + entity_type = "message" + async def search( self, db: AsyncSession, @@ -103,10 +106,204 @@ class CommSearchProvider: return results + async def search_fts( + self, + db: AsyncSession, + tsquery: str, + tenant_id: uuid.UUID, + limit: int, + ) -> list[dict[str, Any]]: + """Full-text search on comm_messages.search_tsv.""" + sql = sql_text( + """ + SELECT m.*, ts_rank(m.search_tsv, to_tsquery('pg_catalog.german', :q)) AS rank + FROM comm_messages m + WHERE m.tenant_id = :tid + AND m.deleted_at IS NULL + AND m.search_tsv @@ to_tsquery('pg_catalog.german', :q) + ORDER BY rank DESC + LIMIT :lim + """ + ) + result = await db.execute( + sql, + {"q": tsquery, "tid": tenant_id, "lim": limit}, + ) + rows = result.mappings().all() + return [dict(r) for r in rows] + + async def search_vector( + self, + db: AsyncSession, + embedding: list[float], + tenant_id: uuid.UUID, + limit: int, + ) -> list[dict[str, Any]]: + """Semantic vector search on comm_messages.embedding.""" + sql = sql_text( + """ + SELECT m.*, 1 - (m.embedding <=> cast(:emb AS vector)) AS score + FROM comm_messages m + WHERE m.tenant_id = :tid + AND m.deleted_at IS NULL + AND m.embedding IS NOT NULL + ORDER BY m.embedding <=> cast(:emb AS vector) + LIMIT :lim + """ + ) + result = await db.execute( + sql, + {"emb": str(embedding), "tid": tenant_id, "lim": limit}, + ) + rows = result.mappings().all() + return [dict(r) for r in rows] + + async def get_embedding_text( + self, + db: AsyncSession, + entity_id: uuid.UUID, + tenant_id: uuid.UUID, + ) -> str: + """Get message content for embedding generation.""" + sql = sql_text( + """ + SELECT content FROM comm_messages + WHERE id = :eid AND tenant_id = :tid AND deleted_at IS NULL + """ + ) + result = await db.execute(sql, {"eid": entity_id, "tid": tenant_id}) + row = result.mappings().first() + if row is None: + return "" + return row["content"] or "" + + def to_search_result(self, entity: object) -> dict: + """Convert a comm_messages row to a search result dict.""" + row = dict(entity) if isinstance(entity, dict) else {} + return { + "type": "message", + "id": str(row.get("id", "")), + "conversation_id": str(row.get("conversation_id", "")), + "content": row.get("content", ""), + "sender_type": row.get("sender_type", ""), + "created_at": str(row.get("created_at", "")) if row.get("created_at") else None, + } + async def index_message(self, message: dict[str, Any]) -> None: - """Index a message for search (placeholder for future full-text indexing).""" - pass + """Index a single message for search — updates search_tsv and embedding. + + Called after a message is created or updated. The message dict must contain + at least: id, tenant_id, content. + """ + from app.core.db import async_session_factory + + msg_id = message.get("id") + tenant_id = message.get("tenant_id") + content = message.get("content", "") + + if not msg_id or not tenant_id: + logger.warning("index_message: missing id or tenant_id") + return + + async with async_session_factory() as db: + try: + # Update search_tsv using PostgreSQL to_tsvector + sql_update_tsv = sql_text( + """ + UPDATE comm_messages + SET search_tsv = to_tsvector('pg_catalog.german', :content) + WHERE id = :mid AND tenant_id = :tid + """ + ) + await db.execute( + sql_update_tsv, + {"content": content, "mid": msg_id, "tid": tenant_id}, + ) + + # Generate and store embedding + if content.strip(): + embedding = await generate_embedding( + content, db=db, tenant_id=tenant_id + ) + if embedding: + sql_update_emb = sql_text( + """ + UPDATE comm_messages + SET embedding = cast(:emb AS vector) + WHERE id = :mid AND tenant_id = :tid + """ + ) + await db.execute( + sql_update_emb, + {"emb": str(embedding), "mid": msg_id, "tid": tenant_id}, + ) + + await db.commit() + except Exception: + await db.rollback() + logger.exception("Failed to index message %s", msg_id) async def reindex_all(self, db: AsyncSession, tenant_id: uuid.UUID) -> None: - """Full reindex (placeholder for future full-text indexing).""" - pass + """Full reindex of all messages for a tenant. + + Iterates over all non-deleted messages and updates search_tsv and embedding. + """ + sql_select = sql_text( + """ + SELECT id, content FROM comm_messages + WHERE tenant_id = :tid AND deleted_at IS NULL + ORDER BY created_at ASC + """ + ) + result = await db.execute(sql_select, {"tid": tenant_id}) + rows = result.mappings().all() + + total = len(rows) + logger.info("Reindexing %d messages for tenant %s", total, tenant_id) + + for i, row in enumerate(rows): + msg_id = row["id"] + content = row["content"] or "" + + try: + # Update search_tsv + sql_update_tsv = sql_text( + """ + UPDATE comm_messages + SET search_tsv = to_tsvector('pg_catalog.german', :content) + WHERE id = :mid AND tenant_id = :tid + """ + ) + await db.execute( + sql_update_tsv, + {"content": content, "mid": msg_id, "tid": tenant_id}, + ) + + # Generate and store embedding + if content.strip(): + embedding = await generate_embedding( + content, db=db, tenant_id=tenant_id + ) + if embedding: + sql_update_emb = sql_text( + """ + UPDATE comm_messages + SET embedding = cast(:emb AS vector) + WHERE id = :mid AND tenant_id = :tid + """ + ) + await db.execute( + sql_update_emb, + {"emb": str(embedding), "mid": msg_id, "tid": tenant_id}, + ) + + if (i + 1) % 50 == 0: + await db.commit() + logger.debug("Reindexed %d/%d messages", i + 1, total) + + except Exception: + logger.exception("Failed to reindex message %s", msg_id) + continue + + await db.commit() + logger.info("Reindexing complete for tenant %s: %d messages", tenant_id, total)