fix: Chat search index, DMS group-share user_id, verify workflows and mail salt
- Implement CommSearchProvider: index_message(), reindex_all() with FTS + vector search - Add migration 0035: search_tsv + embedding columns on comm_messages - Fix DMS group-share: use current_user user_id instead of random uuid4() - Workflows already DB-configurable (seed from onboarding.py, loaded from DB) - Mail salt already dynamic with legacy fallback (migration 0026)
This commit is contained in:
@@ -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")
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user