feat: implement telegram alert engine and war room dashboard

.env.example

@@ -0,0 +1,4 @@
+SCRAPER_SCHEDULE_ENABLED=false
+SIMILARITY_DEVICE=cpu
+SIMILARITY_MIN_RESULTS=10
+SCRAPER_MAX_RESULTS=25

README.md

@@ -5,3 +5,62 @@ A Bangla-first AI commerce OS for marketplace sellers. Monitors competitor price
 The frontend currently stores JWTs in `localStorage`, which is vulnerable to token theft if an XSS occurs. For production deployments, prefer httpOnly cookies (or another mitigated storage strategy) and enforce a strong CSP + input sanitization to reduce XSS risk.
 
 Database bootstrap helpers run only when `DB_BOOTSTRAP_ON_STARTUP` is enabled (defaults to `true` in development and `false` otherwise). Use migrations for predictable schema changes in multi-instance environments.
+
+## Telegram product intelligence snapshots
+
+**Frontend flow**
+1. Open **Telegram Alerts**.
+2. In **Product Intelligence Snapshot**, select a product and click **Send Snapshot**.
+3. A success toast appears and the realtime activity feed updates.
+
+**API endpoint**
+`POST /api/v1/telegram/send-product-snapshot`
+
+**cURL example**
+```bash
+curl -X POST http://localhost:8000/api/v1/telegram/send-product-snapshot \
+  -H "Authorization: Bearer TOKEN" \
+  -H "Content-Type: application/json" \
+  -d '{"product_id":"UUID"}'
+```
+
+**Postman example**
+1. Method: `POST`
+2. URL: `http://localhost:8000/api/v1/telegram/send-product-snapshot`
+3. Headers: `Authorization: Bearer TOKEN`, `Content-Type: application/json`
+4. Body (raw JSON): `{"product_id":"UUID"}`
+
+**Expected Telegram output**
+```
+📊 BazarSync AI Product Intelligence
+
+🛒 Product:
+Dove Body Lotion 200ml
+
+💰 Current Price:
+৳320
+
+📦 Current Stock:
+45 units
+
+🚚 Supplier Lead Time:
+5 days
+
+📉 Competitor Median Price:
+৳305
+
+⚠️ Price Gap:
+-4.7%
+
+🤖 Latest AI Recommendation:
+PRICE LOWER
+
+🧠 AI Reason:
+Competitor prices are currently lower while 2 sellers are out-of-stock. Lowering price slightly may capture additional demand.
+
+📈 Confidence:
+HIGH
+
+⏰ Updated:
+2026-05-28 10:45 PM
+```

backend/alembic/env.py

@@ -3,20 +3,28 @@
 
 from sqlalchemy import engine_from_config, pool
 from alembic import context
+from dotenv import load_dotenv
 
 # Import your SQLAlchemy Base
 from app.db.base import Base
 
 # Import ALL models so Alembic can detect them
-from app.models.seller import Seller
-from app.models.product import Product
-from app.models.scrape_run import ScrapeRun
+from app.models.alert import Alert
+from app.models.alert_delivery_log import AlertDeliveryLog
 from app.models.competitor import Competitor
 from app.models.price_history import PriceHistory
+from app.models.product import Product
+from app.models.recommendation import Recommendation
+from app.models.scrape_run import ScrapeRun
+from app.models.seller import Seller
+from app.models.telegram_connection import SellerTelegramConnection
+from app.models.telegram_snapshot_log import TelegramProductSnapshotLog
 
 # Alembic Config object
 config = context.config
 
+load_dotenv()
+
 # Load DATABASE_URL from environment variable
 database_url = os.getenv("DATABASE_URL")
 
@@ -83,11 +91,4 @@ def run_migrations_online() -> None:
 if context.is_offline_mode():
     run_migrations_offline()
 else:
-    run_migrations_online()
-
-import os
-
-config.set_main_option(
-    "sqlalchemy.url",
-    os.getenv("DATABASE_URL")
-)
+    run_migrations_online()

backend/alembic/versions/1c8c3e4d9a5b_add_product_and_competitor_metadata.py

@@ -14,7 +14,7 @@
 
 # revision identifiers, used by Alembic.
 revision: str = "1c8c3e4d9a5b"
-down_revision: Union[str, Sequence[str], None] = "9ba821df6da5"
+down_revision: Union[str, Sequence[str], None] = "d4f5a8c9b2e1"
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 

backend/alembic/versions/4f3b6e8a92b1_add_telegram_connections_and_delivery_logs.py

@@ -0,0 +1,173 @@
+"""add telegram connections and delivery logs
+
+Revision ID: 4f3b6e8a92b1
+Revises: 1c8c3e4d9a5b
+Create Date: 2026-05-28 00:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = "4f3b6e8a92b1"
+down_revision: Union[str, Sequence[str], None] = "1c8c3e4d9a5b"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    inspector = sa.inspect(op.get_bind())
+    if "seller_telegram_connections" not in inspector.get_table_names():
+        op.create_table(
+            "seller_telegram_connections",
+            sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
+            sa.Column("seller_id", postgresql.UUID(as_uuid=True), nullable=False, unique=True),
+            sa.Column("telegram_chat_id", sa.String(length=120), nullable=True),
+            sa.Column("telegram_username", sa.String(length=120), nullable=True),
+            sa.Column("telegram_first_name", sa.String(length=120), nullable=True),
+            sa.Column("telegram_last_name", sa.String(length=120), nullable=True),
+            sa.Column("connection_token", sa.String(length=128), nullable=True),
+            sa.Column("token_expires_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column(
+                "is_connected",
+                sa.Boolean(),
+                nullable=False,
+                server_default=sa.text("false"),
+            ),
+            sa.Column("connected_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("last_alert_sent_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column(
+                "created_at",
+                sa.DateTime(timezone=True),
+                server_default=sa.text("now()"),
+                nullable=False,
+            ),
+            sa.Column(
+                "updated_at",
+                sa.DateTime(timezone=True),
+                server_default=sa.text("now()"),
+                nullable=False,
+            ),
+            sa.ForeignKeyConstraint(["seller_id"], ["sellers.id"], ondelete="CASCADE"),
+        )
+        existing_indexes: set[str] = set()
+    else:
+        existing_indexes = {
+            index["name"]
+            for index in inspector.get_indexes("seller_telegram_connections")
+        }
+
+    if "ix_seller_telegram_connections_seller_id" not in existing_indexes:
+        op.create_index(
+            "ix_seller_telegram_connections_seller_id",
+            "seller_telegram_connections",
+            ["seller_id"],
+            unique=True,
+        )
+    if "ix_seller_telegram_connections_token" not in existing_indexes:
+        op.create_index(
+            "ix_seller_telegram_connections_token",
+            "seller_telegram_connections",
+            ["connection_token"],
+            unique=False,
+        )
+    if "ix_seller_telegram_connections_chat_id" not in existing_indexes:
+        op.create_index(
+            "ix_seller_telegram_connections_chat_id",
+            "seller_telegram_connections",
+            ["telegram_chat_id"],
+            unique=False,
+        )
+    if "ix_seller_telegram_connections_is_connected" not in existing_indexes:
+        op.create_index(
+            "ix_seller_telegram_connections_is_connected",
+            "seller_telegram_connections",
+            ["is_connected"],
+            unique=False,
+        )
+    if "alerts" in inspector.get_table_names():
+        op.create_table(
+            "alert_delivery_logs",
+            sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
+            sa.Column("seller_id", postgresql.UUID(as_uuid=True), nullable=False),
+            sa.Column("alert_id", postgresql.UUID(as_uuid=True), nullable=False),
+            sa.Column("telegram_chat_id", sa.String(length=120), nullable=True),
+            sa.Column(
+                "delivery_status",
+                sa.Enum(
+                    "delivered",
+                    "failed",
+                    "pending",
+                    "duplicate",
+                    name="alert_delivery_status",
+                ),
+                nullable=False,
+                server_default="pending",
+            ),
+            sa.Column("error_message", sa.Text(), nullable=True),
+            sa.Column("sent_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column(
+                "created_at",
+                sa.DateTime(timezone=True),
+                server_default=sa.text("now()"),
+                nullable=False,
+            ),
+            sa.ForeignKeyConstraint(["seller_id"], ["sellers.id"], ondelete="CASCADE"),
+            sa.ForeignKeyConstraint(["alert_id"], ["alerts.id"], ondelete="CASCADE"),
+        )
+        op.create_index(
+            "ix_alert_delivery_logs_seller_id",
+            "alert_delivery_logs",
+            ["seller_id"],
+            unique=False,
+        )
+        op.create_index(
+            "ix_alert_delivery_logs_alert_id",
+            "alert_delivery_logs",
+            ["alert_id"],
+            unique=False,
+        )
+        op.create_index(
+            "ix_alert_delivery_logs_status",
+            "alert_delivery_logs",
+            ["delivery_status"],
+            unique=False,
+        )
+        op.create_index(
+            "ix_alert_delivery_logs_sent_at",
+            "alert_delivery_logs",
+            ["sent_at"],
+            unique=False,
+        )
+
+
+def downgrade() -> None:
+    inspector = sa.inspect(op.get_bind())
+    if "alert_delivery_logs" in inspector.get_table_names():
+        op.drop_index("ix_alert_delivery_logs_sent_at", table_name="alert_delivery_logs")
+        op.drop_index("ix_alert_delivery_logs_status", table_name="alert_delivery_logs")
+        op.drop_index("ix_alert_delivery_logs_alert_id", table_name="alert_delivery_logs")
+        op.drop_index("ix_alert_delivery_logs_seller_id", table_name="alert_delivery_logs")
+        op.drop_table("alert_delivery_logs")
+
+    if "seller_telegram_connections" in inspector.get_table_names():
+        op.drop_index(
+            "ix_seller_telegram_connections_is_connected",
+            table_name="seller_telegram_connections",
+        )
+        op.drop_index(
+            "ix_seller_telegram_connections_chat_id",
+            table_name="seller_telegram_connections",
+        )
+        op.drop_index(
+            "ix_seller_telegram_connections_token",
+            table_name="seller_telegram_connections",
+        )
+        op.drop_index(
+            "ix_seller_telegram_connections_seller_id",
+            table_name="seller_telegram_connections",
+        )
+        op.drop_table("seller_telegram_connections")

backend/alembic/versions/7d2c3f6a9b0e_add_snapshot_logs.py

@@ -0,0 +1,62 @@
+"""add telegram snapshot logs
+
+Revision ID: 7d2c3f6a9b0e
+Revises: 4f3b6e8a92b1
+Create Date: 2026-05-28 23:59:00.000000
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = "7d2c3f6a9b0e"
+down_revision: Union[str, Sequence[str], None] = "4f3b6e8a92b1"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    inspector = sa.inspect(op.get_bind())
+    if "snapshot_logs" not in inspector.get_table_names():
+        op.create_table(
+            "snapshot_logs",
+            sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
+            sa.Column("seller_id", postgresql.UUID(as_uuid=True), nullable=False),
+            sa.Column("product_id", postgresql.UUID(as_uuid=True), nullable=False),
+            sa.Column("telegram_chat_id", sa.String(length=120), nullable=False),
+            sa.Column(
+                "sent_at",
+                sa.DateTime(timezone=True),
+                server_default=sa.text("now()"),
+                nullable=False,
+            ),
+            sa.Column("status", sa.String(length=50), nullable=False),
+            sa.Column("error_message", sa.Text(), nullable=True),
+            sa.ForeignKeyConstraint(["seller_id"], ["sellers.id"], ondelete="CASCADE"),
+            sa.ForeignKeyConstraint(["product_id"], ["products.id"], ondelete="CASCADE"),
+        )
+        op.create_index(
+            "ix_snapshot_logs_seller_id", "snapshot_logs", ["seller_id"], unique=False
+        )
+        op.create_index(
+            "ix_snapshot_logs_product_id", "snapshot_logs", ["product_id"], unique=False
+        )
+        op.create_index(
+            "ix_snapshot_logs_status", "snapshot_logs", ["status"], unique=False
+        )
+        op.create_index(
+            "ix_snapshot_logs_sent_at", "snapshot_logs", ["sent_at"], unique=False
+        )
+
+
+def downgrade() -> None:
+    inspector = sa.inspect(op.get_bind())
+    if "snapshot_logs" in inspector.get_table_names():
+        op.drop_index("ix_snapshot_logs_sent_at", table_name="snapshot_logs")
+        op.drop_index("ix_snapshot_logs_status", table_name="snapshot_logs")
+        op.drop_index("ix_snapshot_logs_product_id", table_name="snapshot_logs")
+        op.drop_index("ix_snapshot_logs_seller_id", table_name="snapshot_logs")
+        op.drop_table("snapshot_logs")

backend/alembic/versions/91e0e754a8e6_merge_heads.py

@@ -0,0 +1,28 @@
+"""merge heads
+
+Revision ID: 91e0e754a8e6
+Revises: 9b2d6a4d2a19, 7d2c3f6a9b0e
+Create Date: 2026-05-29 00:52:16.179202
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '91e0e754a8e6'
+down_revision: Union[str, Sequence[str], None] = ('9b2d6a4d2a19', '7d2c3f6a9b0e')
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    pass
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    pass