feat(server): production-grade modular structure with algorithms

LICENSE

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2025 DevMind Contributors
+Copyright (c) 2025 Synapse Contributors
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

server/app/api/router.py

@@ -0,0 +1,7 @@
+from fastapi import APIRouter
+
+from app.api.routes import chat, health
+
+router = APIRouter()
+router.include_router(health.router)
+router.include_router(chat.router)

server/app/api/routes/chat.py

@@ -0,0 +1,36 @@
+from fastapi import APIRouter, Depends, HTTPException
+from fastapi.responses import StreamingResponse
+
+from app.core.prompts import SYSTEM_PROMPT
+from app.middleware.rate_limit import check_rate_limit
+from app.models.chat import ChatRequest
+from app.services.context import maybe_summarize, preprocess, trim_to_budget
+from app.services.groq_client import get_groq_client
+from app.services.streaming import stream_with_retry
+
+router = APIRouter()
+
+
+@router.post("/chat", dependencies=[Depends(check_rate_limit)])
+async def chat(body: ChatRequest) -> StreamingResponse:
+    """Stream an LLM response for the given conversation history."""
+    client = get_groq_client()
+
+    try:
+        messages = preprocess(body.messages)
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+
+    messages = maybe_summarize(client, messages)
+    messages = trim_to_budget(messages)
+
+    groq_messages = [{"role": "system", "content": SYSTEM_PROMPT}, *messages]
+
+    return StreamingResponse(
+        stream_with_retry(client, groq_messages),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "X-Accel-Buffering": "no",  # disable nginx buffering
+        },
+    )

server/app/api/routes/health.py

@@ -0,0 +1,8 @@
+from fastapi import APIRouter
+
+router = APIRouter()
+
+
+@router.get("/health")
+async def health() -> dict[str, str]:
+    return {"status": "ok"}

server/app/core/config.py

@@ -0,0 +1,32 @@
+from functools import lru_cache
+
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="ignore",
+    )
+
+    groq_api_key: str
+
+    # Model
+    groq_model: str = "llama-3.3-70b-versatile"
+
+    max_context_tokens: int = 6_000
+    summary_threshold: int = 20
+    max_message_chars: int = 8_000
+    rate_limit_requests: int = 20
+    rate_limit_window_s: int = 60
+    max_retries: int = 3
+    retry_base_delay_s: float = 0.5
+
+    # CORS — set as JSON array in env: CORS_ORIGINS='["https://example.com"]'
+    cors_origins: list[str] = ["http://localhost:5173"]
+
+
+@lru_cache
+def get_settings() -> Settings:
+    return Settings()

server/app/core/prompts.py

@@ -0,0 +1,11 @@
+SYSTEM_PROMPT = (
+    "You are Synapse, a helpful and knowledgeable AI assistant for developers. "
+    "You excel at explaining code, debugging, architecture decisions, and technical concepts. "
+    "Format code examples with appropriate markdown code blocks."
+)
+
+SUMMARIZE_PROMPT = (
+    "Summarize the following conversation history as concisely as possible. "
+    "Preserve all key facts, decisions, code snippets, and context that would be "
+    "needed to continue the conversation. Output only the summary, no preamble."
+)

server/app/main.py

@@ -0,0 +1,39 @@
+"""
+Synapse Backend — FastAPI application factory.
+"""
+
+import logging
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from app.api.router import router
+from app.core.config import get_settings
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
+
+
+def create_app() -> FastAPI:
+    cfg = get_settings()
+
+    app = FastAPI(
+        title="Synapse API",
+        version="2.0.0",
+        docs_url="/docs",
+        redoc_url="/redoc",
+    )
+
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=cfg.cors_origins,
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    app.include_router(router)
+
+    return app
+
+
+app = create_app()

server/app/middleware/rate_limit.py

@@ -0,0 +1,49 @@
+import time
+from collections import deque
+
+from fastapi import HTTPException, Request
+
+from app.core.config import get_settings
+
+_windows: dict[str, deque[float]] = {}
+
+
+def check_rate_limit(request: Request) -> None:
+    """
+    Sliding-window rate limiter (per IP).
+
+    Maintains a deque of request timestamps per IP. On each call:
+      1. Evict timestamps older than the window from the left — O(k) where k
+         is the number of expired entries (amortised O(1) per request).
+      2. Reject if the remaining count meets the limit.
+      3. Append the current timestamp.
+
+    A periodic sweep removes deques for IPs that have been idle longer than
+    one window, bounding memory to O(active IPs × window size).
+    """
+    cfg = get_settings()
+    now = time.monotonic()
+    ip = request.client.host if request.client else "unknown"
+    window = _windows.setdefault(ip, deque())
+
+    while window and now - window[0] > cfg.rate_limit_window_s:
+        window.popleft()
+
+    if len(window) >= cfg.rate_limit_requests:
+        raise HTTPException(
+            status_code=429,
+            detail=(
+                f"Rate limit exceeded: max {cfg.rate_limit_requests} requests "
+                f"per {cfg.rate_limit_window_s}s."
+            ),
+        )
+
+    window.append(now)
+
+    if len(_windows) > 10_000:
+        stale = [
+            k for k, v in _windows.items()
+            if not v or now - v[-1] > cfg.rate_limit_window_s
+        ]
+        for k in stale:
+            del _windows[k]

server/app/models/chat.py

@@ -0,0 +1,24 @@
+from pydantic import BaseModel, field_validator
+
+
+class Message(BaseModel):
+    role: str
+    content: str
+
+    @field_validator("role")
+    @classmethod
+    def validate_role(cls, v: str) -> str:
+        if v not in {"user", "assistant", "system"}:
+            raise ValueError("role must be 'user', 'assistant', or 'system'")
+        return v
+
+
+class ChatRequest(BaseModel):
+    messages: list[Message]
+
+    @field_validator("messages")
+    @classmethod
+    def validate_messages(cls, v: list[Message]) -> list[Message]:
+        if not v:
+            raise ValueError("messages list cannot be empty")
+        return v

server/app/services/context.py

@@ -0,0 +1,127 @@
+"""
+Context window management: token estimation, history trimming, and
+conversation summarization.
+"""
+
+import logging
+
+from groq import Groq
+
+from app.core.config import get_settings
+from app.core.prompts import SUMMARIZE_PROMPT
+from app.models.chat import Message
+
+logger = logging.getLogger(__name__)
+
+GroqMessage = dict[str, str]
+
+
+
+def estimate_tokens(text: str) -> int:
+    """
+    O(n) approximation: 1 token ≈ 4 UTF-8 characters for English prose.
+    Accurate enough for budget decisions without a full tokenizer dependency.
+    """
+    return max(1, len(text) // 4)
+
+
+def preprocess(messages: list[Message]) -> list[GroqMessage]:
+    """
+    Strip whitespace, enforce the per-message character cap, and drop empty
+    turns. Returns plain dicts ready for the Groq API.
+
+    Raises ValueError if every message is empty after sanitization.
+    """
+    cfg = get_settings()
+    result: list[GroqMessage] = []
+
+    for m in messages:
+        content = m.content.strip()
+        if not content:
+            continue
+        if len(content) > cfg.max_message_chars:
+            content = content[: cfg.max_message_chars] + "\n[Message truncated — input too long]"
+        result.append({"role": m.role, "content": content})
+
+    if not result:
+        raise ValueError("All messages were empty after sanitization")
+
+    return result
+
+
+def trim_to_budget(messages: list[GroqMessage]) -> list[GroqMessage]:
+    """
+    Greedy reverse scan: walk messages newest → oldest, accumulating token cost
+    until MAX_CONTEXT_TOKENS is exhausted. The result is the largest suffix of
+    the conversation that fits within the budget.
+
+    If messages were dropped, two framing turns are prepended so the model
+    knows the conversation was truncated and doesn't hallucinate earlier context.
+    """
+    cfg = get_settings()
+    budget = cfg.max_context_tokens
+    kept: list[GroqMessage] = []
+
+    for msg in reversed(messages):
+        cost = estimate_tokens(msg["content"])
+        if kept and budget - cost < 0:
+            break
+        budget -= cost
+        kept.append(msg)
+
+    kept.reverse()
+
+    if len(kept) < len(messages):
+        logger.info(
+            "Context trimmed: %d → %d messages (%d tokens remaining in budget)",
+            len(messages),
+            len(kept),
+            budget,
+        )
+        kept = [
+            {"role": "user", "content": "[Earlier messages omitted — context window full]"},
+            {"role": "assistant", "content": "Understood. I'll work from the available context."},
+            *kept,
+        ]
+
+    return kept
+
+
+def _summarize(client: Groq, messages: list[GroqMessage]) -> str:
+    """Non-streaming Groq call that returns a compact summary of older turns."""
+    transcript = "\n".join(f"{m['role'].upper()}: {m['content']}" for m in messages)
+    response = client.chat.completions.create(
+        model=get_settings().groq_model,
+        messages=[
+            {"role": "system", "content": SUMMARIZE_PROMPT},
+            {"role": "user", "content": transcript},
+        ],
+        stream=False,
+        temperature=0.3,
+        max_tokens=512,
+    )
+    return response.choices[0].message.content or ""
+
+
+def maybe_summarize(client: Groq, messages: list[GroqMessage]) -> list[GroqMessage]:
+    """
+    When history exceeds SUMMARY_THRESHOLD messages, split at the midpoint,
+    summarize the older half, and replace it with a single assistant turn.
+
+    This preserves the substance of earlier turns better than blind truncation
+    while keeping the payload sent to the model bounded.
+    """
+    cfg = get_settings()
+    if len(messages) <= cfg.summary_threshold:
+        return messages
+
+    split = len(messages) // 2
+    old, recent = messages[:split], messages[split:]
+
+    logger.info("Summarizing %d older messages (of %d total)", split, len(messages))
+    summary = _summarize(client, old)
+
+    return [
+        {"role": "assistant", "content": f"[Summary of earlier conversation]\n{summary}"},
+        *recent,
+    ]

server/app/services/groq_client.py

@@ -0,0 +1,13 @@
+from groq import Groq
+
+from app.core.config import get_settings
+
+_client: Groq | None = None
+
+
+def get_groq_client() -> Groq:
+    """Lazy singleton — instantiated once on first request."""
+    global _client
+    if _client is None:
+        _client = Groq(api_key=get_settings().groq_api_key)
+    return _client