.env.example

@@ -2,9 +2,13 @@
 LOG_LEVEL=INFO
 
 # Matrix Configuration
-HOMESERVER_URL = "https://matrix.org"
+MATRIX_HOMESERVER_URL = "https://matrix.org"
-USER_ID = "@botbot_user:matrix.org"
+MATRIX_USER_ID = "@botbot_user:matrix.org"
-PASSWORD = "botbot_password"
+MATRIX_PASSWORD = "botbot_password"
+MATRIX_LOGIN_TRIES = 5 # Number of login attempts before giving up, default is 5
+MATRIX_LOGIN_DELAY_INCREMENT = 5 # Delay increment,in seconds, between login attempts, default is 5
+AI_HANDLER_URL = "http://ai_service:8000"
 
 # OpenAI API Key
 OPENAI_API_KEY=your_openai_api_key_here
+AI_HANDLER_TOKEN=common_token_here

ai_service/Dockerfile

@@ -0,0 +1,12 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+COPY requirements.txt ./
+
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY main.py ./
+
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
+

ai_service/main.py

@@ -0,0 +1,45 @@
+import os
+from dotenv import load_dotenv
+from fastapi import FastAPI, Header, HTTPException
+from pydantic import BaseModel
+import openai
+import redis
+
+AI_TOKEN = os.environ["AI_HANDLER_TOKEN"]
+openai.api_key = os.environ["OPENAI_API_KEY"]
+r = redis.Redis.from_url(os.environ.get("REDIS_URL", "redis://redis:6379"))
+
+class MessagePayload(BaseModel):
+    roomId: str
+    userId: str
+    content: str
+    eventId: str
+    timestamp: int
+
+app = FastAPI()
+
+@app.post("/api/v1/message")
+async def message(
+    payload: MessagePayload,
+    authorization: str = Header(None)
+):
+    if authorization != f"Bearer {AI_TOKEN}":
+        raise HTTPException(status_code=401, detail="Unauthorized")
+
+    # Idempotency: ignore duplicates
+    if r.get(payload.eventId):
+        return {"reply": r.get(payload.eventId).decode()}
+
+    # Build prompt (very simple example)
+    prompt = f"User {payload.userId} said: {payload.content}\nBot:"
+    resp = openai.Completion.create(
+        model="text-davinci-003",
+        prompt=prompt,
+        max_tokens=150
+    )
+    reply = resp.choices[0].text.strip()
+
+    # Cache reply for idempotency
+    r.set(payload.eventId, reply, ex=3600)
+
+    return {"reply": reply}

ai_service/requirements.txt

@@ -0,0 +1,6 @@
+python-dotenv>=1.0.0
+openai
+fastapi>=0.95
+uvicorn>=0.22
+redis>=4.5
+pydantic>=1.10

docker-compose.yml

@@ -1,11 +1,29 @@
 services:
-  botbot:
+  matrix_service:
-    build: .
+    build: ./matrix_service
-    env_file:
+    environment:
-      - .env
+      - MATRIX_HOMESERVER_URL
+      - MATRIX_USER_ID
+      - MATRIX_PASSWORD
+      - AI_HANDLER_URL
+      - AI_HANDLER_TOKEN
     volumes:
-      - ./:/app           # Mount source for hot-reload
+      - ./matrix_service:/app
-      - matrix_data:/app/data  # Persist Matrix client store and tokens
+      - matrix_data:/app/data
+    depends_on:
+      - ai_service
+
+  ai_service:
+    build: ./ai_service
+    environment:
+      - OPENAI_API_KEY=${OPENAI_API_KEY}
+      - AI_HANDLER_TOKEN=${AI_HANDLER_TOKEN}
+      - REDIS_URL=redis://redis:6379
+    depends_on:
+      - redis
+
+  redis:
+    image: redis:7
     restart: unless-stopped
 
 volumes:

main.py

@@ -1,129 +0,0 @@
-import os
-import asyncio
-import logging
-from dotenv import load_dotenv
-from nio import AsyncClient, AsyncClientConfig, MatrixRoom, RoomMessageText, InviteMemberEvent
-from nio.responses import LoginResponse
-from openai import AsyncOpenAI
-
-# --- Load environment variables ---
-load_dotenv()
-HOMESERVER_URL = os.getenv("HOMESERVER_URL")
-USER_ID = os.getenv("USER_ID")
-PASSWORD = os.getenv("PASSWORD")
-LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper()
-OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
-
-if not OPENAI_API_KEY:
-    raise RuntimeError("OPENAI_API_KEY is not set in environment")
-
-# --- Initialize Async OpenAI client ---
-openai_client = AsyncOpenAI(api_key=OPENAI_API_KEY)
-
-# --- Logging Setup ---
-numeric_level = getattr(logging, LOG_LEVEL, logging.INFO)
-logging.basicConfig(
-    level=numeric_level,
-    format="%(asctime)s %(levelname)s %(name)s: %(message)s"
-)
-logger = logging.getLogger(__name__)
-
-async def trust_all_devices(client) -> None:
-    """
-    Programmatically verify all devices to allow sharing encryption keys.
-    """
-    for room_id in client.rooms:
-        try:
-            devices = await client.room_devices(room_id)
-            if isinstance(devices, dict):
-                for user, dev_ids in devices.items():
-                    if user == USER_ID:
-                        continue
-                    for dev_id in dev_ids:
-                        device = client.crypto.device_store.get_device(user, dev_id)
-                        if device and not client.crypto.device_store.is_device_verified(device):
-                            logger.info(f"Trusting {dev_id} for {user}")
-                            client.verify_device(device)
-        except Exception:
-            logger.exception(f"Error trusting devices in {room_id}")
-
-async def message_callback(room: MatrixRoom, event: RoomMessageText):
-    """Handle incoming text messages."""
-    if event.sender == USER_ID:
-        return
-    body = event.body.strip()
-    lower = body.lower()
-    logger.info("Received '%s' from %s in %s", body, event.sender, room.display_name)
-
-    send_kwargs = {
-        "room_id": room.room_id,
-        "message_type": "m.room.message",
-        "ignore_unverified_devices": True
-    }
-
-    # Simple ping
-    if lower == "!ping":
-        await client.room_send(**send_kwargs, content={"msgtype": "m.text", "body": "Pong!"})
-        return
-
-    # Ask OpenAI via chat completion
-    if lower.startswith("!ask "):
-        question = body[5:].strip()
-        if not question:
-            await client.room_send(**send_kwargs, content={"msgtype": "m.text", "body": "Provide a question after !ask."})
-            return
-        logger.info("Querying OpenAI: %s", question)
-        try:
-            response = await openai_client.chat.completions.create(
-                model="gpt-3.5-turbo",
-                messages=[
-                    {"role": "system", "content": "You are a helpful assistant."},
-                    {"role": "user",   "content": question}
-                ],
-                max_tokens=150,
-                timeout=300
-            )
-            answer = response.choices[0].message.content.strip()
-        except Exception:
-            logger.exception("OpenAI API error")
-            answer = "Sorry, I encountered an error contacting the AI service."
-        await client.room_send(**send_kwargs, content={"msgtype": "m.text", "body": answer})
-        return
-
-    # Greeting
-    if lower == "hello botbot":
-        await client.room_send(**send_kwargs, content={"msgtype": "m.text", "body": "Hello! How can I assist you today?"})
-
-async def main() -> None:
-    """Initialize and run the Matrix bot."""
-    global client
-    config = AsyncClientConfig(store_sync_tokens=True, encryption_enabled=True)
-    client = AsyncClient(HOMESERVER_URL, USER_ID, store_path="/app/data", config=config)
-
-    login_resp = await client.login(password=PASSWORD)
-    if isinstance(login_resp, LoginResponse):
-        logger.info("Logged in as %s", USER_ID)
-    else:
-        logger.error("Login failed: %s", login_resp)
-        return
-
-    await trust_all_devices(client)
-
-    # Auto-join and trust
-    async def on_invite(room, event):
-        if isinstance(event, InviteMemberEvent):
-            await client.join(room.room_id)
-            logger.info("Joined %s", room.room_id)
-            await trust_all_devices(client)
-    client.add_event_callback(on_invite, InviteMemberEvent)
-    client.add_event_callback(message_callback, RoomMessageText)
-
-    logger.info("Starting sync loop")
-    await client.sync_forever(timeout=30000)
-
-if __name__ == "__main__":
-    try:
-        asyncio.run(main())
-    except KeyboardInterrupt:
-        logger.info("Shutting down")
-        asyncio.run(client.close())

matrix_service/Dockerfile

@@ -4,10 +4,12 @@ FROM python:3.11-slim
 WORKDIR /app
 
 # Install system dependencies
-RUN apt-get update && \
+RUN apt-get update && apt-get install -y --no-install-recommends \
-    apt-get install -y --no-install-recommends \
+      libolm-dev \
-      libolm-dev build-essential python3-dev && \
+      build-essential \
-    rm -rf /var/lib/apt/lists/*
+      python3-dev \
+      && \
+    apt-get clean && rm -rf /var/lib/apt/lists/*
 
 # Install dependencies
 COPY requirements.txt .

matrix_service/main.py

@@ -0,0 +1,133 @@
+import os
+import logging
+from dotenv import load_dotenv
+
+import asyncio
+import httpx
+
+from nio import AsyncClient, AsyncClientConfig, MatrixRoom, RoomMessageText, InviteMemberEvent
+from nio.responses import LoginResponse
+
+
+# --- Load environment variables ---
+load_dotenv()
+MATRIX_HOMESERVER_URL = os.getenv("MATRIX_HOMESERVER_URL")
+MATRIX_USER_ID = os.getenv("MATRIX_USER_ID")
+MATRIX_PASSWORD = os.getenv("MATRIX_PASSWORD")
+MATRIX_LOGIN_TRIES = int(os.getenv("MATRIX_LOGIN_TRIES", 5))
+MATRIX_LOGIN_DELAY_INCREMENT = int(os.getenv("MATRIX_LOGIN_DELAY_INCREMENT", 5))
+LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper()
+AI_URL = os.getenv("AI_URL")
+AI_TOKEN = os.getenv("AI_TOKEN")
+
+
+
+# --- Logging Setup ---
+numeric_level = getattr(logging, LOG_LEVEL, logging.INFO)
+logging.basicConfig(
+    level=numeric_level,
+    format="%(asctime)s %(levelname)s %(name)s: %(message)s"
+)
+logger = logging.getLogger(__name__)
+
+
+async def main() -> None:
+
+    async def trust_all_devices(client) -> None:
+        """
+        Programmatically verify all devices to allow sharing encryption keys.
+        """
+        for room_id in client.rooms:
+            try:
+                devices = await client.room_devices(room_id)
+                if isinstance(devices, dict):
+                    for user, dev_ids in devices.items():
+                        if user == USER_ID:
+                            continue
+                        for dev_id in dev_ids:
+                            device = client.crypto.device_store.get_device(user, dev_id)
+                            if device and not client.crypto.device_store.is_device_verified(device):
+                                logger.info(f"Trusting {dev_id} for {user}")
+                                client.verify_device(device)
+            except Exception:
+                logger.exception(f"Error trusting devices in {room_id}")
+
+    async def on_invite(room, event):
+        """
+        Handle an invite event by joining the room and trusting all devices.
+        """
+        if isinstance(event, InviteMemberEvent):
+            await client.join(room.room_id)
+            logger.info("Joined %s", room.room_id)
+            await trust_all_devices(client)
+
+    async def on_message(room: MatrixRoom, event: RoomMessageText) -> None:
+        """
+        Handle incoming messages.
+        """
+
+        # Check if the message is from the bot itself
+        if event.sender == client.user_id:            
+            return
+        
+        logger.info("Received '%s' from %s in %s", event.body.strip(), event.sender, room.display_name)
+
+        
+        if isinstance(event, RoomMessageText):
+            logger.info(f"Received message in {room.room_id}: {event.body}")
+            payload = {
+                "roomId":    event["room_id"],
+                "userId":    event["sender"],
+                "content":   event["content"]["body"],
+                "eventId":   event["event_id"],
+                "timestamp": event["origin_server_ts"]
+            }
+            headers = {"Authorization": f"Bearer {AI_TOKEN}"}
+            async with httpx.AsyncClient() as http:
+                resp = await http.post(f"{AI_URL}/api/v1/message", json=payload, headers=headers)
+                resp.raise_for_status()
+                data = resp.json()
+            if data.get("reply"):
+                client.send_message(event["room_id"], data["reply"])
+
+    # --- Initialize the client ---
+    # Create the data directory if it doesn't exist
+    try:
+        os.makedirs("/app/data", exist_ok=True)
+    except Exception as e:
+        logger.error(f"Error creating data directory: {e}")
+        return
+    
+    # Initialize the client
+    config = AsyncClientConfig(store_sync_tokens=True, encryption_enabled=True)
+    client = AsyncClient(MATRIX_HOMESERVER_URL, MATRIX_USER_ID, store_path="/app/data", config=config)    
+  
+    for i in range(MATRIX_LOGIN_TRIES):
+        try:
+            login_response=await client.login(password=MATRIX_PASSWORD)
+            break
+        except Exception as e:
+            logger.error("Login failed: %s", login_response)
+            logger.error(f"Login attempt {i+1} failed: {e}")
+            if i == MATRIX_LOGIN_TRIES - 1:
+                return
+            await asyncio.sleep(MATRIX_LOGIN_DELAY_INCREMENT * (i + 1))
+    logger.info("Logged in successfully")   
+    
+    await trust_all_devices(client)
+
+    client.add_event_callback(on_invite, InviteMemberEvent)
+    client.add_event_callback(on_message, RoomMessageText)
+
+    logger.info("Starting sync loop")
+    await client.sync_forever(timeout=30000) # timeout should be moved to Variable
+
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        logger.info("Shutting down")
+        asyncio.run(client.close())
+