Merge pull request #181 from demilade18-git/fix/issues-124-125-126-127

BigBen-7 · web-flow · commit f0edeeed15df · 2026-03-28T15:44:51.000+01:00
fix: resolve issues #124, #125, #126, #127
diff --git a/src/chat.py b/src/chat.py
@@ -13,6 +13,7 @@
     log_info,
     log_warning,
 )
+from src.chat_store import chat_store
 
 
 class ChatMessage(BaseModel):
@@ -121,6 +122,8 @@ async def send_message(self, message: ChatMessage) -> bool:
                 self.message_history[message.conversation_id] = []
             self.message_history[message.conversation_id].append(message)
 
+        chat_store.save_message(message)
+
         if message.conversation_id in self.active_connections:
             disconnected: List[WebSocket] = []
             for websocket in self.active_connections[message.conversation_id]:
@@ -171,8 +174,27 @@ async def broadcast_event(
     def get_message_history(
         self, conversation_id: str, limit: int = 50
     ) -> List[ChatMessage]:
-        """Return the most recent messages for a conversation."""
+        """Return the most recent messages for a conversation.
+
+        Falls back to the DB when the in-memory cache is empty (e.g. after restart).
+        """
         messages = self.message_history.get(conversation_id, [])
+        if not messages:
+            db_rows = chat_store.get_messages(conversation_id, limit=limit)
+            messages = [
+                ChatMessage(
+                    id=r["id"],
+                    sender_id=r["sender_id"],
+                    sender_type=r["sender_type"],
+                    content=r["content"],
+                    timestamp=r["timestamp"],
+                    conversation_id=r["conversation_id"],
+                )
+                for r in db_rows
+            ]
+            if messages:
+                async_lock_safe: List[ChatMessage] = messages
+                self.message_history[conversation_id] = async_lock_safe
         return messages[-limit:] if len(messages) > limit else messages
 
     def get_user_conversations(self, user_id: str) -> List[str]:
@@ -200,6 +222,8 @@ async def escalate_conversation(
             self.conversation_assignments[conversation_id] = None
             self.conversation_escalated_at[conversation_id] = escalation.timestamp
 
+        chat_store.save_escalation(escalation)
+
         if conversation_id in self.active_connections:
             escalation_notification = {
                 "type": "escalation",
diff --git a/src/chat_store.py b/src/chat_store.py
@@ -0,0 +1,168 @@
+"""Persistent storage for chat messages and escalations using SQLAlchemy.
+
+NOTE: Active WebSocket connections are intentionally kept in-memory only
+(in ChatManager). Only messages and escalation events are persisted here.
+"""
+from __future__ import annotations
+
+import logging
+from datetime import datetime
+from typing import Any, Dict, List, Optional
+
+from sqlalchemy import Column, DateTime, Integer, String, Text, create_engine, text
+from sqlalchemy import MetaData, Table
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+
+import src.db as _db
+
+logger = logging.getLogger("veritix.chat_store")
+
+_MESSAGES_TABLE = "chat_messages"
+_ESCALATIONS_TABLE = "chat_escalations"
+
+
+def _get_engine():
+    return _db.get_engine()
+
+
+def _ensure_tables(engine) -> None:
+    metadata = MetaData()
+    Table(
+        _MESSAGES_TABLE,
+        metadata,
+        Column("id", String, primary_key=True),
+        Column("conversation_id", String, nullable=False),
+        Column("sender_id", String, nullable=False),
+        Column("sender_type", String, nullable=False),
+        Column("content", Text, nullable=False),
+        Column("timestamp", DateTime, nullable=False),
+        Column("metadata_json", Text),
+    )
+    Table(
+        _ESCALATIONS_TABLE,
+        metadata,
+        Column("id", String, primary_key=True),
+        Column("conversation_id", String, nullable=False),
+        Column("reason", String, nullable=False),
+        Column("timestamp", DateTime, nullable=False),
+        Column("metadata_json", Text),
+    )
+    with engine.begin() as conn:
+        metadata.create_all(conn)  # type: ignore[arg-type]
+
+
+class ChatStore:
+    """Persists chat messages and escalation events to Postgres."""
+
+    def __init__(self) -> None:
+        self._ready = False
+
+    def _init(self, engine) -> None:
+        if not self._ready:
+            try:
+                _ensure_tables(engine)
+                self._ready = True
+            except Exception as exc:
+                logger.error("ChatStore: failed to create tables: %s", exc)
+
+    # ------------------------------------------------------------------
+    # Messages
+    # ------------------------------------------------------------------
+
+    def save_message(self, message: Any) -> None:
+        """Persist a ChatMessage to the DB (best-effort)."""
+        engine = _get_engine()
+        if engine is None:
+            return
+        self._init(engine)
+        import json
+        try:
+            with engine.begin() as conn:
+                conn.execute(
+                    text(
+                        f"INSERT INTO {_MESSAGES_TABLE} "  # noqa: S608
+                        "(id, conversation_id, sender_id, sender_type, content, timestamp, metadata_json) "
+                        "VALUES (:id, :conv, :sender, :stype, :content, :ts, :meta) "
+                        "ON CONFLICT (id) DO NOTHING"
+                    ),
+                    {
+                        "id": message.id,
+                        "conv": message.conversation_id,
+                        "sender": message.sender_id,
+                        "stype": message.sender_type,
+                        "content": message.content,
+                        "ts": message.timestamp,
+                        "meta": json.dumps(message.metadata or {}),
+                    },
+                )
+        except Exception as exc:
+            logger.error("ChatStore: save_message failed: %s", exc)
+
+    def get_messages(self, conversation_id: str, limit: int = 50) -> List[Dict[str, Any]]:
+        """Retrieve the most recent messages for a conversation from DB."""
+        engine = _get_engine()
+        if engine is None:
+            return []
+        self._init(engine)
+        try:
+            with engine.connect() as conn:
+                rows = conn.execute(
+                    text(
+                        f"SELECT id, conversation_id, sender_id, sender_type, content, timestamp, metadata_json "  # noqa: S608
+                        f"FROM {_MESSAGES_TABLE} "
+                        "WHERE conversation_id = :conv "
+                        "ORDER BY timestamp DESC "
+                        "LIMIT :lim"
+                    ),
+                    {"conv": conversation_id, "lim": limit},
+                ).fetchall()
+            return [
+                {
+                    "id": r[0],
+                    "conversation_id": r[1],
+                    "sender_id": r[2],
+                    "sender_type": r[3],
+                    "content": r[4],
+                    "timestamp": r[5],
+                    "metadata": r[6],
+                }
+                for r in reversed(rows)
+            ]
+        except Exception as exc:
+            logger.error("ChatStore: get_messages failed: %s", exc)
+            return []
+
+    # ------------------------------------------------------------------
+    # Escalations
+    # ------------------------------------------------------------------
+
+    def save_escalation(self, escalation: Any) -> None:
+        """Persist an EscalationEvent to the DB (best-effort)."""
+        engine = _get_engine()
+        if engine is None:
+            return
+        self._init(engine)
+        import json
+        try:
+            with engine.begin() as conn:
+                conn.execute(
+                    text(
+                        f"INSERT INTO {_ESCALATIONS_TABLE} "  # noqa: S608
+                        "(id, conversation_id, reason, timestamp, metadata_json) "
+                        "VALUES (:id, :conv, :reason, :ts, :meta) "
+                        "ON CONFLICT (id) DO NOTHING"
+                    ),
+                    {
+                        "id": escalation.id,
+                        "conv": escalation.conversation_id,
+                        "reason": escalation.reason,
+                        "ts": escalation.timestamp,
+                        "meta": json.dumps(escalation.metadata or {}),
+                    },
+                )
+        except Exception as exc:
+            logger.error("ChatStore: save_escalation failed: %s", exc)
+
+
+# Singleton
+chat_store = ChatStore()
diff --git a/src/etl/__init__.py b/src/etl/__init__.py
@@ -584,4 +584,5 @@ def run_etl_once() -> None:
                 status=status,
                 rejected_count=rejected_count,
             )
+        ETL_JOBS_TOTAL.labels(status=status).inc()
         log_info("ETL job completed", {"status": status, "rejected_count": rejected_count})
diff --git a/src/event_store.py b/src/event_store.py
@@ -0,0 +1,64 @@
+"""Event store: loads events from Postgres event_sales_summary with a 60-second TTL cache."""
+import time
+from typing import Any, Dict, List, Optional
+
+from sqlalchemy import text
+
+import src.db as _db
+from src.logging_config import log_info
+
+_cache: Optional[List[Dict[str, Any]]] = None
+_cache_ts: float = 0.0
+_CACHE_TTL = 60.0  # seconds
+
+
+def get_events_from_db() -> List[Dict[str, Any]]:
+    """Return events from Postgres event_sales_summary, cached for 60 s.
+
+    Falls back to an empty list when the DB is unavailable.
+    """
+    global _cache, _cache_ts
+    now = time.monotonic()
+    if _cache is not None and (now - _cache_ts) < _CACHE_TTL:
+        return _cache
+
+    engine = _db.get_engine()
+    if engine is None:
+        return _cache or []
+
+    try:
+        with engine.connect() as conn:
+            rows = conn.execute(
+                text(
+                    "SELECT event_id, event_name, total_tickets, total_revenue, last_updated "
+                    "FROM event_sales_summary"
+                )
+            ).fetchall()
+        events: List[Dict[str, Any]] = [
+            {
+                "id": str(row[0]),
+                "name": str(row[1] or ""),
+                "description": "",
+                "event_type": "general",
+                "location": "",
+                "date": row[4].isoformat() if row[4] else "",
+                "price": float(row[3] or 0) / max(int(row[2] or 1), 1),
+                "capacity": int(row[2] or 0),
+            }
+            for row in rows
+        ]
+        _cache = events
+        _cache_ts = now
+        log_info("event_store: loaded events from DB", {"count": len(events)})
+        return events
+    except Exception as exc:
+        from src.logging_config import log_error
+        log_error("event_store: DB query failed", {"error": str(exc)})
+        return _cache or []
+
+
+def invalidate_cache() -> None:
+    """Force the next call to re-query the database."""
+    global _cache, _cache_ts
+    _cache = None
+    _cache_ts = 0.0
diff --git a/src/main.py b/src/main.py
@@ -42,7 +42,8 @@
     log_warning,
     setup_logging,
 )
-from src.mock_events import get_mock_events
+from src.event_store import get_events_from_db
+from src.mock_events import get_mock_events  # kept for test usage only
 from src.report_service import (
     _query_daily_sales,
     _query_invalid_scans,
@@ -531,7 +532,7 @@ def search_events(payload: SearchEventsRequest) -> Any:
     """Search for events using natural language keyword extraction."""
     try:
         keywords = extract_keywords(payload.query)
-        all_events = get_mock_events()
+        all_events = get_events_from_db() or get_mock_events()
         matching_events = filter_events_by_keywords(
             all_events,
             keywords,
@@ -570,8 +571,6 @@ def recommend_events(payload: RecommendRequest) -> RecommendResponse:
     user_id = payload.user_id
     # Prefer DB-sourced history; fall back to mock data when DB is unavailable.
     user_events_dict = get_user_events_from_db()
-    if not user_events_dict:
-        user_events_dict = mock_user_events
     similarity_matrix = build_item_similarity_matrix(user_events_dict)
     recommended = get_item_recommendations(
         user_id=user_id,
diff --git a/src/revenue_sharing_service.py b/src/revenue_sharing_service.py
@@ -4,6 +4,7 @@
 from decimal import ROUND_HALF_UP, Decimal
 from typing import Dict, List, Optional, Tuple
 
+from fastapi import HTTPException
 from src.logging_config import log_error, log_info
 from src.revenue_sharing_models import (
     EventRevenueInput,
@@ -52,10 +53,19 @@ def calculate_revenue_shares(self, input_data: EventRevenueInput) -> RevenueCalc
         
         # Get stakeholders for this event (in a real implementation, this would come from DB)
         stakeholders = self._get_default_stakeholders(input_data.event_id)
-        
+
         # Apply custom rules if provided
         rules = input_data.custom_rules or self._get_default_rules()
-        
+
+        # Guard: total rule percentages must not exceed 100%
+        total_pct = sum(rule.percentage for rule in rules)
+        if total_pct > 100.0:
+            breakdown = {rule.id: rule.percentage for rule in rules}
+            raise HTTPException(
+                status_code=400,
+                detail=f"Rule percentages sum to {total_pct:.2f}% which exceeds 100%. Breakdown: {breakdown}",
+            )
+
         # Calculate distributions
         distributions, remaining_balance = self._calculate_distributions(
             net_revenue, stakeholders, rules
diff --git a/tests/test_revenue_sharing.py b/tests/test_revenue_sharing.py

Original file line number	Diff line number	Diff line change
`@@ -584,4 +584,5 @@ def run_etl_once() -> None:`
`584`	`584`	`status=status,`
`585`	`585`	`rejected_count=rejected_count,`
`586`	`586`	`)`
	`587`	`+ ETL_JOBS_TOTAL.labels(status=status).inc()`
`587`	`588`	`log_info("ETL job completed", {"status": status, "rejected_count": rejected_count})`