Fix async core calls, thread-local DB clients, WS case fix, schemas, mail

db.py

@@ -1,49 +1,42 @@
 """
 Database helper — thin wrapper around Supabase client.
+Each thread gets its own client instance to avoid HTTP/2 connection sharing issues.
 """
 import os
 import threading
-from supabase import create_client, Client
-from dotenv import load_dotenv
 from pathlib import Path
 
+from dotenv import load_dotenv
+from supabase import create_client, Client
+from supabase.client import ClientOptions
+
 load_dotenv(Path(__file__).parent.parent / ".env")
- 
-SUPABASE_URL = os.environ.get("SUPABASE_URL", "")
-SUPABASE_KEY = os.environ.get("SUPABASE_KEY", "")
-SUPABASE_SERVICE_KEY = os.environ.get("SUPABASE_SERVICE_KEY", "")
 
-_client = None
-_service_client = None
-_lock = threading.Lock()
+SUPABASE_URL = os.environ.get("SUPABASE_URL", "").strip()
+SUPABASE_KEY = (
+    os.environ.get("SUPABASE_KEY", "").strip()
+    or os.environ.get("SUPABASE_ANON_KEY", "").strip()
+)
+SUPABASE_SERVICE_KEY = os.environ.get("SUPABASE_SERVICE_KEY", "").strip()
+
+_local = threading.local()
+_opts = ClientOptions(postgrest_client_timeout=30.0)
 
 def get_client() -> Client:
-    """Returns a shared Supabase client."""
-    global _client
-    with _lock:
-        if _client is None:
-            opts = ClientOptions(postgrest_client_timeout=30.0)
-            _client = create_client(SUPABASE_URL, SUPABASE_KEY, options=opts)
-        return _client
+    """Returns a thread-local Supabase client (anon key)."""
+    if not getattr(_local, "client", None):
+        _local.client = create_client(SUPABASE_URL, SUPABASE_KEY, options=_opts)
+    return _local.client
 
 def new_client() -> Client:
     return get_client()
 
-from supabase.client import ClientOptions
-
 def get_service_client() -> Client:
-    """Returns a shared Supabase client with the service role key."""
-    global _service_client
-    with _lock:
-        if _service_client is None:
-            key = SUPABASE_SERVICE_KEY or SUPABASE_KEY
-            # Use ClientOptions to increase timeout and potentially stability
-            # postgrest_client_timeout defaults to 5.0, let's bump it
-            opts = ClientOptions(postgrest_client_timeout=30.0)
-            _service_client = create_client(SUPABASE_URL, key, options=opts)
-        return _service_client
+    """Returns a thread-local Supabase client (service key)."""
+    if not getattr(_local, "service_client", None):
+        key = SUPABASE_SERVICE_KEY or SUPABASE_KEY
+        _local.service_client = create_client(SUPABASE_URL, key, options=_opts)
+    return _local.service_client
 
 def reset_service_client():
-    global _service_client
-    with _lock:
-        _service_client = None
+    _local.service_client = None

engine.py

@@ -14,8 +14,8 @@ class Engine:
     # ========================
     # AUTH
     # ========================
-    def register_user(self, fn, ln, gender, uname, email, pw):
-        engine_user.register_user(fn, ln, gender, uname, email, pw)
+    def register_user(self, fn, ln, gender, uname, email, pw, theme=5):
+        engine_user.register_user(fn, ln, gender, uname, email, pw, theme)
 
     def login(self, username, password):
         return engine_user.login(username, password)
@@ -68,6 +68,18 @@ class Engine:
     def update_password(self, username, old_pass, new_pass):
         return engine_user.update_password(username, old_pass, new_pass)
 
+    def create_password_reset_token(self, username):
+        return engine_user.create_password_reset_token(username)
+
+    def validate_password_reset_token(self, token_plain):
+        return engine_user.validate_password_reset_token(token_plain)
+
+    def get_user_public_for_reset(self, username):
+        return engine_user.get_user_public_for_reset(username)
+
+    def confirm_password_reset(self, token_plain, new_password):
+        return engine_user.confirm_password_reset(token_plain, new_password)
+
     def delete_user(self, username):
         return engine_user.delete_user(username)
 

engine_user.py

@@ -1,9 +1,30 @@
 """User module"""
-import hashlib, secrets, time
+import hashlib
+import secrets
+import time
+from datetime import datetime, timedelta, timezone
+
 from db import get_service_client
 
 _dev_cache: set | None = None
 
+
+class PasswordResetUnavailableError(RuntimeError):
+    """Supabase is missing the password_reset_tokens table (migration not applied)."""
+
+
+RESET_TABLE_INSTRUCTIONS = (
+    "Password reset is not set up: the database table is missing. "
+    "In Supabase → SQL → run the script in docs/Design/password_reset_tokens.sql, then try again."
+)
+
+
+def _missing_password_reset_table(exc: BaseException) -> bool:
+    s = str(exc)
+    if "PGRST205" in s:
+        return True
+    return "password_reset_tokens" in s and "Could not find" in s
+
 def is_developer(username: str) -> bool:
     global _dev_cache
     if _dev_cache is None:
@@ -19,7 +40,15 @@ def _generate_salt(length=32):
 def _hash_password(password, salt):
     return hashlib.sha256((salt + password).encode()).hexdigest()
 
-def register_user(fn, ln, gender, uname, email, pw):
+def _clamp_theme(theme_val):
+    try:
+        t = int(theme_val)
+    except (TypeError, ValueError):
+        return 5
+    return max(0, min(6, t))
+
+
+def register_user(fn, ln, gender, uname, email, pw, theme=5):
     sb = get_service_client()
     existing = sb.table("users").select("username").eq("username", uname).execute()
     if existing.data:
@@ -30,10 +59,11 @@ def register_user(fn, ln, gender, uname, email, pw):
     else: avatar = "/static/assets/icons/other.png"
     salt = _generate_salt()
     h = _hash_password(pw, salt)
+    theme_i = _clamp_theme(theme)
     sb.table("users").insert({
         "username": uname, "firstName": fn, "lastName": ln, "gender": gender,
         "email": email, "password_hash": h, "password_salt": salt,
-        "bio": "", "followers": 0, "follows": 0, "avatar": avatar, "heroBanner": "", "theme": 4
+        "bio": "", "followers": 0, "follows": 0, "avatar": avatar, "heroBanner": "", "theme": theme_i
     }).execute()
     return True
 
@@ -81,7 +111,7 @@ def get_all_users_detailed():
 def get_theme(username):
     sb = get_service_client()
     rows = sb.table("users").select("theme").eq("username", username).execute()
-    if not rows.data: return "4"
+    if not rows.data: return "5"
     return str(rows.data[0]["theme"])
 
 def update_theme(username, theme):
@@ -139,6 +169,113 @@ def update_password(username, old_pass, new_pass):
     sb.table("users").update({"password_hash": new_hash, "password_salt": new_salt}).eq("username", username).execute()
     return True
 
+
+def _parse_expires_at(val):
+    if not val:
+        return None
+    s = str(val).replace("Z", "+00:00")
+    try:
+        return datetime.fromisoformat(s)
+    except Exception:
+        return None
+
+
+def create_password_reset_token(username: str):
+    """Create a one-time reset token; returns plaintext token for email, or None if user missing."""
+    if not user_exists(username):
+        return None
+    sb = get_service_client()
+    token_plain = secrets.token_urlsafe(32)
+    token_hash = hashlib.sha256(token_plain.encode()).hexdigest()
+    expires_at = datetime.now(timezone.utc) + timedelta(hours=1)
+    try:
+        sb.table("password_reset_tokens").insert(
+            {
+                "username": username,
+                "token_hash": token_hash,
+                "expires_at": expires_at.isoformat(),
+                "used_at": None,
+            }
+        ).execute()
+    except Exception as e:
+        if _missing_password_reset_table(e):
+            raise PasswordResetUnavailableError(RESET_TABLE_INSTRUCTIONS) from e
+        raise
+    return token_plain
+
+
+def validate_password_reset_token(token_plain: str):
+    """
+    Returns dict with username if token is valid and unused, else None.
+    """
+    if not token_plain or not isinstance(token_plain, str):
+        return None
+    sb = get_service_client()
+    token_hash = hashlib.sha256(token_plain.encode()).hexdigest()
+    try:
+        rows = (
+            sb.table("password_reset_tokens")
+            .select("username, expires_at, used_at")
+            .eq("token_hash", token_hash)
+            .execute()
+        )
+    except Exception as e:
+        if _missing_password_reset_table(e):
+            raise PasswordResetUnavailableError(RESET_TABLE_INSTRUCTIONS) from e
+        raise
+    if not rows.data:
+        return None
+    r = rows.data[0]
+    if r.get("used_at"):
+        return None
+    exp = _parse_expires_at(r.get("expires_at"))
+    if exp is None:
+        return None
+    if exp.tzinfo is None:
+        exp = exp.replace(tzinfo=timezone.utc)
+    if datetime.now(timezone.utc) > exp:
+        return None
+    return {"username": r["username"]}
+
+
+def get_user_public_for_reset(username: str):
+    """Safe fields for reset-password page."""
+    u = get_user_by_username(username)
+    if not u:
+        return None
+    return {
+        "username": u.get("username", ""),
+        "firstName": u.get("firstName", ""),
+        "lastName": u.get("lastName", ""),
+        "email": u.get("email", ""),
+        "avatar": get_avatar(username),
+    }
+
+
+def confirm_password_reset(token_plain: str, new_password: str):
+    if not new_password or len(new_password) < 1:
+        return False
+    info = validate_password_reset_token(token_plain)
+    if not info:
+        return False
+    username = info["username"]
+    sb = get_service_client()
+    new_salt = _generate_salt()
+    new_hash = _hash_password(new_password, new_salt)
+    sb.table("users").update({"password_hash": new_hash, "password_salt": new_salt}).eq(
+        "username", username
+    ).execute()
+    token_hash = hashlib.sha256(token_plain.encode()).hexdigest()
+    now = datetime.now(timezone.utc).isoformat()
+    try:
+        sb.table("password_reset_tokens").update({"used_at": now}).eq("token_hash", token_hash).execute()
+    except Exception as e:
+        if _missing_password_reset_table(e):
+            raise PasswordResetUnavailableError(RESET_TABLE_INSTRUCTIONS) from e
+        raise
+    return True
+
+
 def delete_user(username):
     sb = get_service_client()
     # Cascading deletes handle everything else!

mail.py

@@ -0,0 +1,73 @@
+"""Optional SMTP email for password reset and notifications."""
+
+import os
+import smtplib
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+
+
+def _smtp_configured():
+    return bool(os.environ.get("SMTP_HOST", "").strip())
+
+
+def send_password_reset_email(to_email: str, username: str, reset_url: str) -> bool:
+    """
+    Send password reset link. Returns True if sent via SMTP.
+    When SMTP is not configured, returns False (caller should log link instead).
+    """
+    host = os.environ.get("SMTP_HOST", "").strip()
+    if not host or not to_email:
+        return False
+
+    port = int(os.environ.get("SMTP_PORT", "587"))
+    user = os.environ.get("SMTP_USER", "").strip()
+    password_raw = os.environ.get("SMTP_PASSWORD", "").strip()
+    # Gmail shows app passwords as "abcd efgh ijkl mnop"; SMTP login uses the 16 chars without spaces.
+    password = "".join(password_raw.split()) if password_raw else ""
+    mail_from = os.environ.get("MAIL_FROM", user or "noreply@localhost").strip()
+
+    subject = "YapStation — reset your password"
+    body = (
+        f"Hi {username},\n\n"
+        f"We received a request to reset your YapStation password.\n\n"
+        f"Open this link (valid for 1 hour):\n{reset_url}\n\n"
+        f"If you did not request this, ignore this email.\n\n"
+        f"— YapStation"
+    )
+
+    msg = MIMEMultipart()
+    msg["Subject"] = subject
+    msg["From"] = mail_from
+    msg["To"] = to_email
+    msg.attach(MIMEText(body, "plain", "utf-8"))
+
+    try:
+        if port == 465:
+            with smtplib.SMTP_SSL(host, port, timeout=30) as smtp:
+                if user and password:
+                    smtp.login(user, password)
+                smtp.sendmail(mail_from, [to_email], msg.as_string())
+        else:
+            with smtplib.SMTP(host, port, timeout=30) as smtp:
+                smtp.ehlo()
+                smtp.starttls()
+                smtp.ehlo()
+                if user and password:
+                    smtp.login(user, password)
+                smtp.sendmail(mail_from, [to_email], msg.as_string())
+        return True
+    except smtplib.SMTPAuthenticationError as e:
+        print(
+            f"[mail] SMTP authentication failed for {user!r}: {e}\n"
+            "[mail] Gmail: create an App Password at "
+            "https://myaccount.google.com/apppasswords (enable 2-Step Verification first). "
+            "Put the 16-character password in SMTP_PASSWORD — not your normal Gmail password."
+        )
+        return False
+    except Exception as e:
+        print(f"[mail] SMTP send failed ({type(e).__name__}): {e}")
+        return False
+
+
+def smtp_configured() -> bool:
+    return _smtp_configured()

package-lock.json

@@ -0,0 +1,6 @@
+{
+  "name": "backend",
+  "lockfileVersion": 3,
+  "requires": true,
+  "packages": {}
+}

requirements.txt

@@ -4,3 +4,4 @@ python-dotenv
 pyjwt
 httpx
 supabase
+pydantic

schemas/API_INVENTORY.md

@@ -0,0 +1,46 @@
+# API response inventory (DTO sources)
+
+Data originates from Supabase via `engine_*` modules and is exposed as JSON through FastAPI (`server.py`). The browser calls `/api/*` (Next.js rewrite to FastAPI).
+
+## POST list item (feed, userPosts, station posts)
+
+Typical keys after engine normalization:
+
+| Field | Type (JSON) | Notes |
+|-------|----------------|-------|
+| id | string | |
+| username | string | Author or station pseudo-user |
+| content | string | |
+| media_path | string | URL or path |
+| timestamp | string | Unix or parsed string |
+| comments_count | string | Coerced in engine |
+| likes_count | string | |
+| station_name | string | Empty for personal posts |
+| avatar | string | User or station profile pic |
+| is_liked | boolean | When viewer set |
+
+## GET `/api/post/{post_id}`
+
+Success: `{ "status": "success", "post": { ... } }` — post includes `select("*")` columns as strings plus `is_liked`, `avatar`.
+
+Restricted: HTTP 403 `{ "status": "restricted", "username": ... }`.
+
+## POST `/api/post/feed` and GET `/api/post/feed`
+
+Success: `{ "status": "success", "feed": Post[], "posts": Post[] }` (same array twice for client compatibility).
+
+## POST `/api/post/userPosts`
+
+Success: `{ "status": "success", "posts": Post[] }` — optional `{ "blocked": true }` when viewer blocked.
+
+## POST `/api/station/get`
+
+Success: `{ "status": "success", "station": { station_name, admin, user_count, bio, hero_banner, profile_pic, users, members_info } | null }`.
+
+## GET `/api/user/{username}/full`
+
+Success: `{ "user": {...}, "avatar": str, "heroBanner": str }` — errors use `{ "error": str }`.
+
+## Optional pagination (feed)
+
+`FeedRequest` accepts optional `limit` (max 500) and `offset` for slicing the in-memory feed list after `get_feed` (does not change DB query yet).

schemas/__init__.py

@@ -0,0 +1,23 @@
+"""API DTOs and response mappers (Pydantic)."""
+
+from .dtos import (
+    FeedRequest,
+    UserPostsRequest,
+    map_post_dict,
+    map_posts_list,
+    build_feed_response,
+    build_user_posts_response,
+    build_post_detail_response,
+    build_station_get_response,
+)
+
+__all__ = [
+    "FeedRequest",
+    "UserPostsRequest",
+    "map_post_dict",
+    "map_posts_list",
+    "build_feed_response",
+    "build_user_posts_response",
+    "build_post_detail_response",
+    "build_station_get_response",
+]

schemas/dtos.py

@@ -0,0 +1,190 @@
+"""
+Pydantic DTOs for public API responses.
+Normalizes Supabase/engine dicts to a stable JSON contract for the frontend.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+
+class PostPublicDTO(BaseModel):
+    """Canonical post shape for lists and detail (unknown keys dropped)."""
+
+    model_config = ConfigDict(extra="ignore")
+
+    id: str
+    username: str = ""
+    content: str = ""
+    media_path: str = ""
+    timestamp: str = ""
+    comments_count: str = "0"
+    likes_count: str = "0"
+    station_name: str = ""
+    avatar: str = ""
+    is_liked: bool = False
+
+    @field_validator(
+        "id",
+        "username",
+        "content",
+        "media_path",
+        "timestamp",
+        "comments_count",
+        "likes_count",
+        "station_name",
+        "avatar",
+        mode="before",
+    )
+    @classmethod
+    def coerce_str_fields(cls, v: Any) -> str:
+        if v is None:
+            return ""
+        return str(v)
+
+    @field_validator("is_liked", mode="before")
+    @classmethod
+    def coerce_bool(cls, v: Any) -> bool:
+        if isinstance(v, bool):
+            return v
+        if v is None:
+            return False
+        s = str(v).lower()
+        return s in ("1", "true", "yes")
+
+
+class StationMemberDTO(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    username: str
+    avatar: Optional[str] = None
+
+    @field_validator("username", mode="before")
+    @classmethod
+    def username_str(cls, v: Any) -> str:
+        return "" if v is None else str(v)
+
+
+class StationPublicDTO(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    station_name: str = ""
+    admin: str = ""
+    user_count: str = ""
+    bio: str = ""
+    hero_banner: str = ""
+    profile_pic: str = ""
+    users: str = ""
+    members_info: list[StationMemberDTO] = Field(default_factory=list)
+
+    @field_validator(
+        "station_name",
+        "admin",
+        "user_count",
+        "bio",
+        "hero_banner",
+        "profile_pic",
+        "users",
+        mode="before",
+    )
+    @classmethod
+    def empty_str(cls, v: Any) -> str:
+        if v is None:
+            return ""
+        return str(v)
+
+
+class FeedRequest(BaseModel):
+    username: Optional[str] = None
+    viewer: Optional[str] = None
+    limit: Optional[int] = None
+    offset: Optional[int] = None
+
+    @field_validator("limit", mode="before")
+    @classmethod
+    def cap_limit(cls, v: Any) -> Optional[int]:
+        if v is None:
+            return None
+        try:
+            n = int(v)
+        except (TypeError, ValueError):
+            return None
+        return max(1, min(n, 500))
+
+    @field_validator("offset", mode="before")
+    @classmethod
+    def nonneg_offset(cls, v: Any) -> Optional[int]:
+        if v is None:
+            return None
+        try:
+            n = int(v)
+        except (TypeError, ValueError):
+            return None
+        return max(0, n)
+
+
+class UserPostsRequest(BaseModel):
+    username: str
+    viewer: Optional[str] = None
+
+
+def map_post_dict(raw: dict[str, Any]) -> PostPublicDTO:
+    """Map a single engine post dict to DTO (whitelisted fields)."""
+    return PostPublicDTO.model_validate(raw)
+
+
+def map_posts_list(items: list[Any]) -> list[dict[str, Any]]:
+    out: list[dict[str, Any]] = []
+    for item in items:
+        if not isinstance(item, dict):
+            continue
+        try:
+            out.append(map_post_dict(item).model_dump(mode="json"))
+        except Exception:
+            continue
+    return out
+
+
+def _slice_feed(posts: list[dict[str, Any]], limit: Optional[int], offset: Optional[int]) -> list[dict[str, Any]]:
+    if limit is None and offset is None:
+        return posts
+    start = offset or 0
+    if limit is None:
+        return posts[start:]
+    return posts[start : start + limit]
+
+
+def build_feed_response(
+    raw_posts: list[Any],
+    *,
+    limit: Optional[int] = None,
+    offset: Optional[int] = None,
+) -> dict[str, Any]:
+    mapped = map_posts_list(raw_posts)
+    mapped = _slice_feed(mapped, limit, offset)
+    return {"status": "success", "feed": mapped, "posts": mapped}
+
+
+def build_user_posts_response(
+    raw_posts: list[Any],
+    *,
+    blocked: bool = False,
+) -> dict[str, Any]:
+    mapped = map_posts_list(raw_posts)
+    body: dict[str, Any] = {"status": "success", "posts": mapped}
+    if blocked:
+        body["blocked"] = True
+    return body
+
+
+def build_post_detail_response(raw_post: dict[str, Any]) -> dict[str, Any]:
+    return {"status": "success", "post": map_post_dict(raw_post).model_dump(mode="json")}
+
+
+def build_station_get_response(raw_station: Optional[dict[str, Any]]) -> dict[str, Any]:
+    if not raw_station:
+        return {"status": "success", "station": None}
+    station = StationPublicDTO.model_validate(raw_station).model_dump(mode="json")
+    return {"status": "success", "station": station}

server.py

@@ -5,6 +5,7 @@ FastAPI server for real-time chat and social platform
 
 import asyncio
 import os
+import re
 import sys
 import time
 from pathlib import Path
@@ -20,7 +21,9 @@ load_dotenv(env_path)
 
 import engine
 import engineHelper as coreHelper
+import engine_user
 import jwt
+import mail
 from db import get_client, get_service_client
 
 core = engine.Engine()
@@ -30,6 +33,11 @@ app = FastAPI()
 
 @app.exception_handler(Exception)
 async def global_exception_handler(request: Request, exc: Exception):
+    if isinstance(exc, engine_user.PasswordResetUnavailableError):
+        return JSONResponse(
+            {"status": "error", "message": str(exc)},
+            status_code=503,
+        )
     return JSONResponse({"status": "fail", "message": str(exc)}, status_code=500)
 
 # Add CORS middleware  
@@ -37,23 +45,38 @@ FRONTEND_URL = os.environ.get("FRONTEND_URL", "https://yap-station.vercel.app")
 
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=[FRONTEND_URL, "http://localhost:3000"],
+    allow_origins=[
+        FRONTEND_URL,
+        "http://localhost:3000",
+        "http://127.0.0.1:3000",
+    ],
     allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
 )
 
 JWT_SECRET = os.environ.get("JWT_SECRET", "default_secret_key_if_not_found")
-from pydantic import BaseModel
+
+# Session cookie: browsers reject Secure cookies on plain HTTP. Local dev uses http://localhost:3000
+# (Next proxy). Set SESSION_COOKIE_SECURE=true in production (.env / host secrets) when using HTTPS.
+SESSION_COOKIE_SECURE = os.environ.get("SESSION_COOKIE_SECURE", "").strip().lower() in (
+    "1",
+    "true",
+    "yes",
+)
+SESSION_COOKIE_SAMESITE = "none" if SESSION_COOKIE_SECURE else "lax"
+
 from typing import Optional
 
-class FeedRequest(BaseModel):
-    username: Optional[str] = None
-    viewer: Optional[str] = None
+from schemas.dtos import (
+    FeedRequest,
+    UserPostsRequest,
+    build_feed_response,
+    build_post_detail_response,
+    build_station_get_response,
+    build_user_posts_response,
+)
 
-class UserPostsRequest(BaseModel):
-    username: str
-    viewer: Optional[str] = None
 JWT_ALGORITHM = "HS256"
 
 BASE = os.path.dirname(__file__)
@@ -66,6 +89,29 @@ else:
     os.makedirs("static_dummy", exist_ok=True)
     app.mount("/static", StaticFiles(directory="static_dummy"), name="static")
 
+
+def normalize_avatar_for_response(avatar: str | None) -> str:
+    """Make DB avatar values usable in <img src> (data URL, absolute http(s), or /static/…)."""
+    if not avatar:
+        return ""
+    a = str(avatar).strip()
+    if a.startswith("data:image/") or a.startswith("http://") or a.startswith("https://"):
+        return a
+    if a.startswith("/static"):
+        return a
+    compact = "".join(a.split())
+    if len(compact) >= 100 and re.fullmatch(r"[A-Za-z0-9+/=]+", compact):
+        if compact.startswith("iVBOR"):
+            mime = "image/png"
+        elif compact.startswith("/9j/"):
+            mime = "image/jpeg"
+        else:
+            mime = "image/png"
+        return f"data:{mime};base64,{compact}"
+    path = a.replace("../../frontend/", "").lstrip("/")
+    return f"/static/{path}" if path else ""
+
+
 def upload_media_to_supabase(data_url: str) -> str:
     """Upload a base64 data URL to Supabase Storage, return the public URL."""
     import base64, uuid, mimetypes
@@ -92,6 +138,35 @@ last_seen_ping = coreHelper.last_seen_ping
 broadcast_online = coreHelper.broadcast_online
 
 
+async def notify_station_members_ws(station_name: str, payload: dict) -> None:
+    """Deliver a WS payload to each connected station member (includes admin)."""
+    try:
+        members = await asyncio.to_thread(core.get_station_members, station_name, "")
+    except Exception:
+        members = []
+    seen: set[str] = set()
+    for u in members or []:
+        if not u:
+            continue
+        lk = str(u).lower()
+        if lk in seen:
+            continue
+        seen.add(lk)
+        await coreHelper.send_to_user(u, payload, connections)
+
+
+async def broadcast_station_post_ws(payload: dict) -> None:
+    """Station posts appear on Discover for many viewers — fan-out to all connections."""
+    await coreHelper.broadcast_to_all(payload)
+
+
+async def touch_station_catalog_ws(station_name: str) -> None:
+    """Light broadcast so browse pages (yap stations list, profile stations) can refresh."""
+    await coreHelper.broadcast_to_all(
+        {"type": "station_catalog_refresh", "stationName": station_name}
+    )
+
+
 def invalidate_conv_cache(username: str):
     if username in _conv_cache:
         del _conv_cache[username]
@@ -115,7 +190,7 @@ async def get_user_accounts(input: str):
     if not input or "@" not in input:
         return {"email": "", "usernames": [], "count": 0}
     
-    result = core.get_username(input)
+    result = await asyncio.to_thread(core.get_username, input)
     count = int(result.get("count", "0"))
     
     if count > 0:
@@ -134,7 +209,7 @@ async def check_email(req: Request):
     if not username or "@" not in username:
         return {"email": "", "usernames": [], "count": 0}
     
-    result = core.get_username(username)
+    result = await asyncio.to_thread(core.get_username, username)
     count = int(result.get("count", "0"))
     
     if count > 0:
@@ -182,7 +257,7 @@ async def admin_delete_user(data: dict):
     admin_user = data.get("admin_username")
     target_user = data.get("target_username")
     
-    if not core.is_developer(admin_user):
+    if not await asyncio.to_thread(core.is_developer, admin_user):
         return JSONResponse({"status": "error", "message": "Unauthorized"}, status_code=403)
     
     if not target_user:
@@ -193,7 +268,7 @@ async def admin_delete_user(data: dict):
         await coreHelper.evict_user_ws(target_user)
         
         # 2. Hard delete from DB
-        success = core.delete_user(target_user)
+        success = await asyncio.to_thread(core.delete_user, target_user)
         
         if success:
             return {"status": "success", "message": f"User {target_user} has been purged."}
@@ -211,7 +286,7 @@ def get_post_route(post_id: int, viewer: Optional[str] = None):
             return JSONResponse({"status": "fail", "message": "Post not found"}, status_code=404)
         if isinstance(post, dict) and post.get("restricted"):
             return JSONResponse({"status": "restricted", "username": post["username"]}, status_code=403)
-        return {"status": "success", "post": post}
+        return build_post_detail_response(post)
     except Exception as e:
         return JSONResponse({"status": "fail", "message": str(e)}, status_code=500)
 
@@ -233,14 +308,14 @@ async def admin_delete_station(data: dict):
     admin_user = data.get("admin_username")
     target_station = data.get("target_station")
     
-    if not core.is_developer(admin_user):
+    if not await asyncio.to_thread(core.is_developer, admin_user):
         return JSONResponse({"status": "error", "message": "Unauthorized"}, status_code=403)
     
     if not target_station:
         return {"status": "fail", "message": "Target station name required"}
 
     try:
-        success = core.force_delete_station(target_station)
+        success = await asyncio.to_thread(core.force_delete_station, target_station)
         if success:
             # Notify all users that the station is gone
             payload = {"type": "station_deleted", "stationName": target_station}
@@ -268,15 +343,15 @@ async def login(req: Request, response: Response):
     
     # If username doesn't contain @, it's already a username
     if username and "@" not in username:
-        if core.login(username, password):
+        if await asyncio.to_thread(core.login, username, password):
             token = create_jwt_token(username)
             response.set_cookie(
                 key="yap_session",
                 value=token,
                 httponly=True,
                 max_age=7 * 24 * 60 * 60,  # 7 days
-                samesite="none",
-                secure=True, # Set to True in prod with HTTPS
+                samesite=SESSION_COOKIE_SAMESITE,
+                secure=SESSION_COOKIE_SECURE,
                 path="/"
             )
             return {"status": "success", "username": username, "token": token}
@@ -284,7 +359,7 @@ async def login(req: Request, response: Response):
     
     # If username is an email, get the accounts
     if username and "@" in username:
-        result = core.get_username(username)
+        result = await asyncio.to_thread(core.get_username, username)
         if int(result.get("count", "0")) == 0:
             return {"status": "fail"}
         
@@ -294,15 +369,15 @@ async def login(req: Request, response: Response):
         # If password provided, try first account
         if password:
             username = usernames[0] if usernames else None
-            if username and core.login(username, password):
+            if username and await asyncio.to_thread(core.login, username, password):
                 token = create_jwt_token(username)
                 response.set_cookie(
                     key="yap_session",
                     value=token,
                     httponly=True,
                     max_age=7 * 24 * 60 * 60,  # 7 days
-                    samesite="none",
-                    secure=True, # Set to True in prod with HTTPS
+                    samesite=SESSION_COOKIE_SAMESITE,
+                    secure=SESSION_COOKIE_SECURE,
                     path="/"
                 )
                 return {"status": "success", "username": username, "token": token}
@@ -327,19 +402,104 @@ def terms_page():
 @app.post("/api/register")
 async def register(data: dict):
     try:
-        core.register_user(
+        theme = data.get("theme", 5)
+        await asyncio.to_thread(
+            core.register_user,
             data["firstName"],
             data["lastName"],
             data["gender"],
             data["username"],
             data["email"],
             data["password"],
+            theme,
         )
         return {"status": "success"}
     except Exception as e:
         return {"status": "error", "message": str(e)}
 
 
+@app.post("/api/auth/forgot-password")
+async def forgot_password(data: dict):
+    """Send reset email for a username (after client resolves email→username if needed)."""
+    username = (data.get("username") or "").strip()
+    generic_ok = {
+        "status": "success",
+        "message": "If an account exists for that username, we sent reset instructions to its email.",
+    }
+    if not username:
+        return JSONResponse(
+            {"status": "error", "message": "Username required"}, status_code=400
+        )
+    if not await asyncio.to_thread(core.user_exists, username):
+        return generic_ok
+    user_row = await asyncio.to_thread(core.get_user_by_username, username)
+    email_addr = (user_row.get("email") or "").strip()
+    if not email_addr:
+        return generic_ok
+    token_plain = await asyncio.to_thread(core.create_password_reset_token, username)
+    if not token_plain:
+        return JSONResponse(
+            {"status": "error", "message": "Could not create reset token"}, status_code=500
+        )
+    raw_base = os.environ.get(
+        "FRONTEND_URL",
+        os.environ.get("NEXT_PUBLIC_SITE_URL", "http://localhost:3000"),
+    ).strip()
+    if raw_base and not raw_base.startswith("http"):
+        raw_base = "https://" + raw_base
+    base = raw_base.rstrip("/")
+    reset_url = f"{base}/reset-password?token={token_plain}"
+    sent = mail.send_password_reset_email(email_addr, username, reset_url)
+    if not mail.smtp_configured():
+        print(
+            "[password-reset] SMTP not configured (set SMTP_HOST, SMTP_USER, SMTP_PASSWORD in .env). "
+            "Email was not sent; use this link or configure Gmail SMTP — see backend mail.py."
+        )
+        print(f"[password-reset] Reset URL for @{username}: {reset_url}")
+    elif not sent:
+        print("[password-reset] SMTP send failed (check credentials / Gmail app password). Link:")
+        print(f"[password-reset] Reset URL for @{username}: {reset_url}")
+    out = dict(generic_ok)
+    if os.environ.get("DEBUG_PASSWORD_RESET", "").strip().lower() in ("1", "true", "yes"):
+        out["dev_reset_url"] = reset_url
+    return out
+
+
+@app.get("/api/auth/reset-password/validate")
+def reset_password_validate(token: str = Query("")):
+    if not token:
+        return {"status": "error", "message": "Missing token"}
+    info = core.validate_password_reset_token(token)
+    if not info:
+        return {"status": "error", "message": "Invalid or expired link"}
+    username = info["username"]
+    preview = core.get_user_public_for_reset(username)
+    if not preview:
+        return {"status": "error", "message": "User not found"}
+    preview = dict(preview)
+    preview["avatar"] = normalize_avatar_for_response(preview.get("avatar"))
+    return {"status": "success", "user": preview}
+
+
+@app.post("/api/auth/reset-password/confirm")
+async def reset_password_confirm(data: dict):
+    token = (data.get("token") or "").strip()
+    pw = (data.get("newPassword") or data.get("password") or "").strip()
+    if not token:
+        return JSONResponse({"status": "error", "message": "Missing token"}, status_code=400)
+    if len(pw) < 6:
+        return JSONResponse(
+            {"status": "error", "message": "Password must be at least 6 characters"},
+            status_code=400,
+        )
+    ok = await asyncio.to_thread(core.confirm_password_reset, token, pw)
+    if not ok:
+        return JSONResponse(
+            {"status": "error", "message": "Invalid or expired token"}, status_code=400
+        )
+    return {"status": "success", "message": "Password updated. You can log in."}
+
+
 def create_page_response(filename: str):
     response = FileResponse(os.path.join(FRONTEND, filename))
     response.headers["Cache-Control"] = "no-store"
@@ -360,11 +520,21 @@ def get_current_user(response: Response, yap_session: str = Cookie(None)):
                 return {"status": "success", "username": username}
             else:
                 # User was deleted! Clear the cookie.
-                response.delete_cookie("yap_session", path="/", samesite="none", secure=True)
+                response.delete_cookie(
+                    "yap_session",
+                    path="/",
+                    samesite=SESSION_COOKIE_SAMESITE,
+                    secure=SESSION_COOKIE_SECURE,
+                )
                 return {"status": "fail", "message": "User no longer exists"}
         return {"status": "fail", "message": "Invalid token"}
     except jwt.ExpiredSignatureError:
-        response.delete_cookie("yap_session", path="/", samesite="none", secure=True)
+        response.delete_cookie(
+            "yap_session",
+            path="/",
+            samesite=SESSION_COOKIE_SAMESITE,
+            secure=SESSION_COOKIE_SECURE,
+        )
         return {"status": "fail", "message": "Session expired"}
     except Exception:
         return {"status": "fail", "message": "Auth check failed"}
@@ -377,8 +547,8 @@ async def logout(response: Response):
     response.delete_cookie(
         key="yap_session",
         httponly=True,
-        samesite="none",
-        secure=True
+        samesite=SESSION_COOKIE_SAMESITE,
+        secure=SESSION_COOKIE_SECURE,
     )
     return {"status": "success"}
 
@@ -532,7 +702,8 @@ async def update_user(data: dict):
     current_username = data.get("currentUsername", "")
     new_username = data.get("newUsername", "")
 
-    success = core.update_user_basic(
+    success = await asyncio.to_thread(
+        core.update_user_basic,
         current_username,
         new_username,
         data.get("email", ""),
@@ -565,7 +736,7 @@ async def delete_user(data: dict):
     username = data.get("username")
     if not username:
         return {"status": "fail", "message": "Username required"}
-    success = core.delete_user(username)
+    success = await asyncio.to_thread(core.delete_user, username)
     if success:
         return {"status": "success"}
     return {"status": "fail", "message": "Deletion failed"}
@@ -573,9 +744,7 @@ async def delete_user(data: dict):
 
 @app.post("/api/user/updatePassword")
 async def update_password(data: dict):
-    success = core.update_password(
-        data["username"], data["oldPassword"], data["newPassword"]
-    )
+    success = await asyncio.to_thread(core.update_password, data["username"], data["oldPassword"], data["newPassword"])
     return {"status": "success" if success else "fail"}
 
 
@@ -598,7 +767,7 @@ def get_user_full(username: str, viewer: str = None):
         avatar = r.get("avatar") or ""
         hero = r.get("heroBanner") or ""
         user = {k: str(v) for k, v in r.items() if k not in ("avatar", "heroBanner")}
-        return {"user": user, "avatar": avatar, "heroBanner": hero if hero != "null" else ""}
+        return {"user": user, "avatar": avatar, "heroBanner": hero if hero != "null" else "", "is_developer": core.is_developer(username)}
     except Exception as e:
         return {"error": str(e)}
 
@@ -609,9 +778,7 @@ def get_user_avatar(username: str, viewer: str = None):
             if core.is_blocked_either_way(viewer, username):
                 return {"error": "User blocked or unavailable"}
         avatar = core.get_avatar(username)
-        if avatar:
-            if not avatar.startswith("data:image/") and not avatar.startswith("/static"):
-                avatar = "/static/" + avatar.replace("../../frontend/", "")
+        avatar = normalize_avatar_for_response(avatar) if avatar else ""
         return {"username": username, "avatar": avatar}
     except Exception as e:
         return {"username": username, "avatar": "", "error": str(e)}
@@ -669,7 +836,7 @@ def get_user_theme(username: str):
 async def update_user_theme(data: dict):
     username = data["username"]
     theme = int(data["theme"])
-    core.update_theme(username, theme)
+    await asyncio.to_thread(core.update_theme, username, theme)
     payload = {"type": "theme_update", "username": username, "theme": theme}
     if username in connections:
         for ws in list(connections[username]):
@@ -680,16 +847,25 @@ async def update_user_theme(data: dict):
     return {"status": "ok"}
 
 
+def _row_username(row):
+    if isinstance(row, dict):
+        return str(row.get("username") or "")
+    return str(row)
+
+
 @app.get("/api/users")
 def get_all_users():
     users = core.get_all_users()
     result = []
     for u in users:
-        avatar = core.get_avatar(u)
+        uname = _row_username(u)
+        if not uname:
+            continue
+        avatar = core.get_avatar(uname)
         if avatar:
             if not avatar.startswith("/static"):
                 avatar = "/static/" + avatar.replace("../../frontend/", "")
-        result.append({"username": u, "avatar": avatar})
+        result.append({"username": uname, "avatar": avatar})
     return {"users": result}
 
 @app.post("/api/users")
@@ -701,14 +877,17 @@ def get_all_users_filtered_post(current_user: str = ""):
     users = core.get_all_users()
     result = []
     for u in users:
-        if current_user and u != current_user:
-            if core.is_blocked_either_way(current_user, u):
+        uname = _row_username(u)
+        if not uname:
+            continue
+        if current_user and uname != current_user:
+            if core.is_blocked_either_way(current_user, uname):
                 continue
-        avatar = core.get_avatar(u)
+        avatar = core.get_avatar(uname)
         if avatar:
             if not avatar.startswith("/static"):
                 avatar = "/static/" + avatar.replace("../../frontend/", "")
-        result.append({"username": u, "avatar": avatar})
+        result.append({"username": uname, "avatar": avatar})
     return {"users": result}
 
 
@@ -721,7 +900,7 @@ async def follow_user(data: dict):
     if follower == following:
         return {"status": "fail", "message": "Cannot follow yourself"}
     
-    result = core.follow_user(follower, following)
+    result = await asyncio.to_thread(core.follow_user, follower, following)
     if isinstance(result, dict) and result.get("status") == "success":
         follow_status = result.get("follow_status")
         
@@ -757,7 +936,7 @@ async def unfollow_user(data: dict):
     following = data.get("following")
     if not follower or not following:
         return {"status": "fail", "message": "Missing parameters"}
-    success = core.unfollow_user(follower, following)
+    success = await asyncio.to_thread(core.unfollow_user, follower, following)
     if success:
         await coreHelper.send_follow_notification(
             following, follower, "unfollow", connections
@@ -776,14 +955,14 @@ async def set_visibility_route(data: dict):
     visibility = data.get("visibility")
     if username is None or visibility is None:
         return {"status": "fail"}
-    core.set_visibility(username, visibility)
+    await asyncio.to_thread(core.set_visibility, username, visibility)
     return {"status": "success"}
 
 @app.post("/api/user/pendingFollows")
 async def get_pending_follows_route(data: dict):
     username = data.get("username")
     if not username: return {"status": "fail"}
-    pending = core.get_pending_follows(username)
+    pending = await asyncio.to_thread(core.get_pending_follows, username)
     # Process avatars
     for p in pending:
         avatar = p.get("avatar")
@@ -797,7 +976,7 @@ async def approve_follow_route(data: dict):
     follower = data.get("follower")
     following = data.get("following")
     if not follower or not following: return {"status": "fail"}
-    success = core.approve_follow(follower, following)
+    success = await asyncio.to_thread(core.approve_follow, follower, following)
     if success:
         # Notify the follower that they were accepted
         payload = {
@@ -824,7 +1003,7 @@ async def reject_follow_route(data: dict):
     follower = data.get("follower")
     following = data.get("following")
     if not follower or not following: return {"status": "fail"}
-    success = core.reject_follow(follower, following)
+    success = await asyncio.to_thread(core.reject_follow, follower, following)
     if success:
         # Notify the follower that they were rejected
         payload = {
@@ -844,7 +1023,7 @@ async def reject_follow_route(data: dict):
 async def get_follow_status_route(data: dict):
     follower = data.get("follower")
     following = data.get("following")
-    status = core.get_follow_status(follower, following)
+    status = await asyncio.to_thread(core.get_follow_status, follower, following)
     return {"status": "success", "follow_status": status}
 
 
@@ -854,7 +1033,7 @@ async def remove_follower(data: dict):
     follower = data.get("follower")
     if not username or not follower:
         return {"status": "fail", "message": "Missing parameters"}
-    success = core.unfollow_user(follower, username)
+    success = await asyncio.to_thread(core.unfollow_user, follower, username)
     if success:
         # Send notification only to the user who was removed (follower)
         await coreHelper.send_follow_notification(
@@ -936,7 +1115,7 @@ async def is_following(data: dict):
     following = data.get("following")
     if not follower or not following:
         return {"status": "fail"}
-    followers = core.get_following(follower)
+    followers = await asyncio.to_thread(core.get_following, follower)
     return {"status": "success", "isFollowing": following in followers}
 
 @app.post("/api/user/block")
@@ -952,7 +1131,7 @@ async def block_user(data: dict):
         return JSONResponse({"status": "fail", "message": "Cannot block yourself"}, status_code=400)
 
     
-    success = core.block_user(blocker, blocked)
+    success = await asyncio.to_thread(core.block_user, blocker, blocked)
     
     if success:
         # Broadcast block update
@@ -985,7 +1164,7 @@ async def unblock_user(data: dict):
         return JSONResponse({"status": "fail", "message": "Missing parameters"}, status_code=400)
 
     
-    success = core.unblock_user(blocker, blocked)
+    success = await asyncio.to_thread(core.unblock_user, blocker, blocked)
     
     if success:
         # Broadcast unblock update
@@ -1017,7 +1196,7 @@ async def check_is_blocked(data: dict):
     if not blocker or not blocked:
         return {"status": "fail", "message": "Missing parameters"}
     
-    is_blocked = core.is_blocked(blocker, blocked)
+    is_blocked = await asyncio.to_thread(core.is_blocked, blocker, blocked)
     return {"status": "success", "isBlocked": is_blocked}
 
 @app.post("/api/user/getBlockedUsers")
@@ -1027,11 +1206,11 @@ async def get_blocked_users(data: dict):
     if not username:
         return {"status": "fail", "message": "Username required"}
     
-    blocked_users = core.get_blocked_users(username)
+    blocked_users = await asyncio.to_thread(core.get_blocked_users, username)
     blocked_list = []
     
     for user in blocked_users:
-        avatar = core.get_avatar(user)
+        avatar = await asyncio.to_thread(core.get_avatar, user)
         if avatar:
             if not avatar.startswith("/static"):
                 avatar = "/static/" + avatar.replace("../../frontend/", "")
@@ -1063,7 +1242,7 @@ async def create_chat(data: dict):
         if not name:
             return {"error": "Group name required"}
         is_group = True
-    user_convs = core.get_user_conversations(creator)
+    user_convs = await asyncio.to_thread(core.get_user_conversations, creator)
     for c in user_convs:
         if isinstance(c, dict):
             conv_id = c.get("id")
@@ -1074,20 +1253,20 @@ async def create_chat(data: dict):
             conv_name = parts[1] if len(parts) > 1 else ""
         if not conv_id:
             continue
-        conv_users = core.get_conversation_users(int(conv_id))
+        conv_users = await asyncio.to_thread(core.get_conversation_users, int(conv_id))
         conv_users_sorted = sorted([u.strip() for u in conv_users if u])
         users_sorted = sorted([u.strip() for u in users if u])
         if conv_users_sorted == users_sorted and conv_name == name:
             return {"conversationId": conv_id}
-    conv_id = core.create_conversation(users, is_group, name, admin_username=creator)
+    conv_id = await asyncio.to_thread(core.create_conversation, users, is_group, name, admin_username=creator)
     payload = {"type": "conversation_created", "conversationId": conv_id}
     for u in users:
-        if u in connections:
-            for ws in list(connections[u]):
+        if u.lower() in connections:
+            for ws in list(connections[u.lower()]):
                 try:
                     await ws.send_json(payload)
                 except Exception:
-                    connections[u].discard(ws)
+                    connections[u.lower()].discard(ws)
     return {"conversationId": conv_id}
 
 
@@ -1101,23 +1280,23 @@ async def add_chat_member(data: dict):
     
     added_any = False
     for u_name in usernames:
-        if u_name and core.add_user_to_conversation(conv_id, u_name):
+        if u_name and await asyncio.to_thread(core.add_user_to_conversation, conv_id, u_name):
             added_any = True
             
     if added_any:
         # 4. Invalidate backend cache for all members
-        members = core.get_conversation_users(conv_id)
+        members = await asyncio.to_thread(core.get_conversation_users, conv_id)
         invalidate_conv_cache_for_all(members)
         
         # 5. Broadcast update to all members
         payload = {"type": "conversation_updated", "conversationId": conv_id}
         for u in members:
-            if u in connections:
-                for ws in list(connections[u]):
+            if u.lower() in connections:
+                for ws in list(connections[u.lower()]):
                     try:
                         await ws.send_json(payload)
                     except Exception:
-                        connections[u].discard(ws)
+                        connections[u.lower()].discard(ws)
         return {"status": "success"}
     return {"status": "fail"}
 
@@ -1135,12 +1314,12 @@ async def share_post_route(data: dict):
     
     for target in usernames:
         # 1. Find or Create direct chat
-        conv_id = core.find_conversation([sender, target], "")
+        conv_id = await asyncio.to_thread(core.find_conversation, [sender, target], "")
         if conv_id == -1:
-            conv_id = core.create_conversation([sender, target], False, "", admin_username=sender)
+            conv_id = await asyncio.to_thread(core.create_conversation, [sender, target], False, "", admin_username=sender)
             
         # 2. Save message as type 'post'
-        msg_id = core.send_message(sender, conv_id, str(post_id), "post")
+        msg_id = await asyncio.to_thread(core.send_message, sender, conv_id, str(post_id), "post")
         
         # 3. Broadcast to both participants
         invalidate_conv_cache(sender)
@@ -1171,28 +1350,28 @@ async def share_post_route(data: dict):
 async def remove_chat_member(data: dict):
     conv_id = int(data.get("conversationId"))
     username = data.get("username")
-    members_before = core.get_conversation_users(conv_id)
-    if core.remove_user_from_conversation(conv_id, username):
+    members_before = await asyncio.to_thread(core.get_conversation_users, conv_id)
+    if await asyncio.to_thread(core.remove_user_from_conversation, conv_id, username):
         # Invalidate backend cache for everyone who was in the group
         invalidate_conv_cache_for_all(members_before)
         
         # Broadcast update to remaining members and the removed member
         payload = {"type": "conversation_updated", "conversationId": conv_id}
         for u in set(members_before):
-            if u in connections:
-                for ws in list(connections[u]):
+            if u.lower() in connections:
+                for ws in list(connections[u.lower()]):
                     try:
                         await ws.send_json(payload)
                     except Exception:
-                        connections[u].discard(ws)
+                        connections[u.lower()].discard(ws)
         return {"status": "success"}
     return {"status": "fail"}
 
 @app.delete("/api/deleteConversation/{username}/{convId}")
 async def delete_conversation(username: str, convId: int):
     try:
-        members = core.get_conversation_users(convId)
-        convs = core.get_user_conversations(username)
+        members = await asyncio.to_thread(core.get_conversation_users, convId)
+        convs = await asyncio.to_thread(core.get_user_conversations, username)
         is_group = False
         for c in convs:
             if isinstance(c, dict):
@@ -1204,15 +1383,15 @@ async def delete_conversation(username: str, convId: int):
                 if parts[0] == str(convId):
                     is_group = parts[2] == "1"
                     break
-        core.delete_conversation(convId)
+        await asyncio.to_thread(core.delete_conversation, convId)
         payload = {"type": "conversation_deleted", "conversationId": convId}
         for u in members:
-            if u in connections:
-                for ws in list(connections[u]):
+            if u.lower() in connections:
+                for ws in list(connections[u.lower()]):
                     try:
                         await ws.send_json(payload)
                     except Exception:
-                        connections[u].discard(ws)
+                        connections[u.lower()].discard(ws)
         return {"status": "deleted", "type": "group" if is_group else "direct"}
     except Exception as e:
         return {"error": str(e)}
@@ -1225,8 +1404,8 @@ async def update_conv_avatar(data: dict):
         avatar = data["avatar"]
         if avatar and avatar.startswith("data:"):
             avatar = upload_media_to_supabase(avatar)
-        core.update_conversation_avatar(conv_id, avatar)
-        users = core.get_conversation_users(conv_id)
+        await asyncio.to_thread(core.update_conversation_avatar, conv_id, avatar)
+        users = await asyncio.to_thread(core.get_conversation_users, conv_id)
         payload = {
             "type": "conversation_avatar_update",
             "conversationId": conv_id,
@@ -1234,9 +1413,9 @@ async def update_conv_avatar(data: dict):
         }
         tasks = []
         for u in users:
-            if u in connections:
-                for ws in list(connections[u]):
-                    tasks.append(send_safe(ws, payload, connections[u]))
+            if u.lower() in connections:
+                for ws in list(connections[u.lower()]):
+                    tasks.append(send_safe(ws, payload, connections[u.lower()]))
         await asyncio.gather(*tasks, return_exceptions=True)
         return {"status": "ok"}
     except Exception as e:
@@ -1312,7 +1491,7 @@ async def delete_post(data: dict):
             {"status": "fail", "message": "Missing parameters"}, status_code=400
         )
 
-    success = core.delete_post(post_id, username)
+    success = await asyncio.to_thread(core.delete_post, post_id, username)
 
     if success:
         # Broadcast post deletion
@@ -1330,16 +1509,21 @@ async def delete_post(data: dict):
 
 
 @app.get("/api/post/feed")
-def get_feed_get(username: Optional[str] = None, viewer: Optional[str] = None):
+def get_feed_get(
+    username: Optional[str] = None,
+    viewer: Optional[str] = None,
+    limit: Optional[int] = None,
+    offset: Optional[int] = None,
+):
     try:
         current_user = viewer or username or ""
         feed = core.get_feed(current_user)
-        
+
         # Performance optimization: pre-fetch block list to avoid N+1 queries
         blocked_list = set()
         if current_user:
             blocked_list = set(core.get_blocked_users(current_user) + core.get_blockers(current_user))
-            
+
         filtered_feed = []
         for post in feed:
             post_user = post.get("username", "")
@@ -1347,7 +1531,10 @@ def get_feed_get(username: Optional[str] = None, viewer: Optional[str] = None):
                 if post_user in blocked_list:
                     continue
             filtered_feed.append(post)
-        return {"status": "success", "feed": filtered_feed, "posts": filtered_feed}
+
+        lim = max(1, min(int(limit), 500)) if limit is not None else None
+        off = max(0, int(offset)) if offset is not None else (0 if lim is not None else None)
+        return build_feed_response(filtered_feed, limit=lim, offset=off)
     except Exception as e:
         return JSONResponse({"status": "fail", "message": str(e)}, status_code=500)
 
@@ -1356,7 +1543,7 @@ def get_feed_post(data: FeedRequest):
     try:
         current_user = data.viewer or data.username or ""
         feed = core.get_feed(current_user)
-        return {"status": "success", "feed": feed, "posts": feed}
+        return build_feed_response(feed, limit=data.limit, offset=data.offset)
     except Exception as e:
         return JSONResponse({"status": "fail", "message": str(e)}, status_code=500)
 
@@ -1369,10 +1556,10 @@ def get_user_posts(data: UserPostsRequest):
         
         if viewer and viewer != username:
             if core.is_blocked_either_way(viewer, username):
-                return {"status": "success", "posts": [], "blocked": True}
+                return build_user_posts_response([], blocked=True)
 
         posts = core.get_user_posts(username, viewer)
-        return {"status": "success", "posts": posts}
+        return build_user_posts_response(posts)
     except Exception as e:
         
         return JSONResponse({"status": "fail", "message": str(e)}, status_code=500)
@@ -1387,10 +1574,10 @@ async def add_comment(data: dict):
         return JSONResponse(
             {"status": "fail", "message": "Missing parameters"}, status_code=400
         )
-    success = core.add_comment(post_id, username, content)
+    success = await asyncio.to_thread(core.add_comment, post_id, username, content)
     if success:
         # Get the newly added comment (get_comments returns newest first with desc=True)
-        comments = core.get_comments(post_id)
+        comments = await asyncio.to_thread(core.get_comments, post_id)
         new_comment = comments[0] if comments else None
         # Get updated comments count
         rows = get_client().table("posts").select("comments_count").eq("id", post_id).execute()
@@ -1439,6 +1626,7 @@ async def create_station(data: dict):
                 "type": "station_created",
                 "stationName": station_name,
                 "admin": admin_username,
+                "bio": bio,
             }
             # Broadcast to all
             asyncio.create_task(coreHelper.broadcast_to_all(payload))
@@ -1483,8 +1671,8 @@ async def add_user_to_station(data: dict):
             "stationName": station_name,
             "user": user_to_add,
         }
-        # Notify relevant users
-        asyncio.create_task(coreHelper.broadcast_to_all(payload))
+        asyncio.create_task(notify_station_members_ws(station_name, payload))
+        asyncio.create_task(touch_station_catalog_ws(station_name))
         return {"status": "success"}
     return {"status": "fail"}
 
@@ -1502,11 +1690,13 @@ async def remove_user_from_station(data: dict):
 
     success = await asyncio.to_thread(core.remove_user_from_station, station_name, admin_username, user_to_remove)
     if success:
-        await coreHelper.broadcast_to_all({
-            "type": "station_user_removed",
-            "stationName": station_name,
-            "user": user_to_remove,
-        })
+        asyncio.create_task(
+            notify_station_members_ws(
+                station_name,
+                {"type": "station_user_removed", "stationName": station_name, "user": user_to_remove},
+            )
+        )
+        asyncio.create_task(touch_station_catalog_ws(station_name))
         return {"status": "success"}
     return {"status": "fail"}
 
@@ -1521,7 +1711,6 @@ async def request_to_join_station(data: dict):
         )
     success = await asyncio.to_thread(core.request_to_join, station_name, username)
     if success:
-        # Get station info to find admin
         stations = await asyncio.to_thread(core.get_station, station_name)
         if stations:
             admin = stations[0].get("admin", "")
@@ -1530,14 +1719,8 @@ async def request_to_join_station(data: dict):
                 "stationName": station_name,
                 "user": username,
             }
-            # Only send to station admin if online
-            admin_key = admin.lower()
-            if admin_key in connections:
-                for ws in list(connections[admin_key]):
-                    try:
-                        await ws.send_json(payload)
-                    except Exception:
-                        connections[admin_key].discard(ws)
+            await coreHelper.send_to_user(admin, payload, connections)
+            await touch_station_catalog_ws(station_name)
     return {"status": "success" if success else "fail"}
 
 
@@ -1564,15 +1747,19 @@ async def approve_station_request(data: dict):
         )
     success = await asyncio.to_thread(core.approve_request, station_name, admin_username, user_to_approve)
     if success:
-        await coreHelper.broadcast_to_all({
-            "type": "station_user_approved",
-            "stationName": station_name,
-            "user": user_to_approve,
-        })
-        # Small delay before second broadcast or just one is enough
-        update_payload = {"type": "station_updated", "stationName": station_name}
-        asyncio.create_task(coreHelper.broadcast_to_all(update_payload))
-        
+        asyncio.create_task(
+            notify_station_members_ws(
+                station_name,
+                {"type": "station_user_approved", "stationName": station_name, "user": user_to_approve},
+            )
+        )
+        asyncio.create_task(
+            notify_station_members_ws(
+                station_name, {"type": "station_updated", "stationName": station_name}
+            )
+        )
+        asyncio.create_task(touch_station_catalog_ws(station_name))
+
     return {"status": "success" if success else "fail"}
 
 
@@ -1585,20 +1772,25 @@ async def reject_station_request(data: dict):
         return JSONResponse(
             {"status": "fail", "message": "Missing parameters"}, status_code=400
         )
-    success = core.reject_request(station_name, admin_username, user_to_reject)
+    success = await asyncio.to_thread(core.reject_request, station_name, admin_username, user_to_reject)
     if success:
-        payload = {
+        rej_payload = {
             "type": "station_request_rejected",
             "stationName": station_name,
             "user": user_to_reject,
         }
-        user_key = user_to_reject.lower()
-        if user_key in connections:
-            for ws in list(connections[user_key]):
-                try:
-                    await ws.send_json(payload)
-                except Exception:
-                    connections[user_key].discard(ws)
+        await coreHelper.send_to_user(user_to_reject, rej_payload, connections)
+        await coreHelper.send_to_user(
+            admin_username,
+            {
+                "type": "station_request_closed",
+                "stationName": station_name,
+                "user": user_to_reject,
+                "rejected": True,
+            },
+            connections,
+        )
+        asyncio.create_task(touch_station_catalog_ws(station_name))
     return {"status": "success" if success else "fail"}
 
 
@@ -1610,7 +1802,7 @@ async def get_pending_requests(data: dict):
         return JSONResponse(
             {"status": "fail", "message": "Missing parameters"}, status_code=400
         )
-    requests = core.get_pending_requests(station_name, admin_username)
+    requests = await asyncio.to_thread(core.get_pending_requests, station_name, admin_username)
     return {"status": "success", "requests": requests}
 
 
@@ -1624,19 +1816,14 @@ async def create_station_post(data: dict):
         return JSONResponse(
             {"status": "fail", "message": "Missing parameters"}, status_code=400
         )
-    new_post = core.create_station_post(station_name, username, content, media_path)
+    new_post = await asyncio.to_thread(core.create_station_post, station_name, username, content, media_path)
     if new_post:
         payload = {
             "type": "station_post_created",
             "stationName": station_name,
             "post": new_post,
         }
-        for user_conns in list(connections.values()):
-            for ws in list(user_conns):
-                try:
-                    await ws.send_json(payload)
-                except Exception:
-                    user_conns.discard(ws)
+        await broadcast_station_post_ws(payload)
         return {"status": "success", "post": new_post}
     return {"status": "fail"}
 
@@ -1653,19 +1840,14 @@ async def delete_station_post(data: dict):
         return JSONResponse(
             {"status": "fail", "message": "Missing parameters"}, status_code=400
         )
-    success = core.delete_station_post(post_id, station_name, username)
+    success = await asyncio.to_thread(core.delete_station_post, post_id, station_name, username)
     if success:
         payload = {
             "type": "station_post_deleted",
             "postId": post_id,
             "stationName": station_name,
         }
-        for user_conns in list(connections.values()):
-            for ws in list(user_conns):
-                try:
-                    await ws.send_json(payload)
-                except Exception:
-                    user_conns.discard(ws)
+        await broadcast_station_post_ws(payload)
         return {"status": "success"}
     return {"status": "fail"}
 
@@ -1679,8 +1861,8 @@ async def get_station_posts(data: dict):
             {"status": "fail", "message": "Station name required"}, status_code=400
         )
     try:
-        posts = core.get_station_posts(station_name, viewer)
-        return {"status": "success", "posts": posts}
+        posts = await asyncio.to_thread(core.get_station_posts, station_name, viewer)
+        return build_user_posts_response(posts)
     except Exception as e:
         return JSONResponse({"status": "fail", "message": str(e)}, status_code=500)
 
@@ -1693,8 +1875,9 @@ async def get_station(data: dict):
         return JSONResponse(
             {"status": "fail", "message": "Station name required"}, status_code=400
         )
-    stations = core.get_station(station_name, viewer)
-    return {"status": "success", "station": stations[0] if stations else None}
+    stations = await asyncio.to_thread(core.get_station, station_name, viewer)
+    raw = stations[0] if stations else None
+    return build_station_get_response(raw)
 
 
 
@@ -1706,7 +1889,7 @@ async def get_bulk_station_posts_route(data: dict):
         return {"status": "success", "posts": []}
     try:
         posts = await asyncio.to_thread(core.get_bulk_station_posts, station_names, viewer)
-        return {"status": "success", "posts": posts}
+        return build_user_posts_response(posts)
     except Exception as e:
         return JSONResponse({"status": "fail", "message": str(e)}, status_code=500)
 
@@ -1732,7 +1915,7 @@ async def get_station_members(data: dict):
         return JSONResponse(
             {"status": "fail", "message": "Station name required"}, status_code=400
         )
-    members = core.get_station_members(station_name, viewer)
+    members = await asyncio.to_thread(core.get_station_members, station_name, viewer)
     return {"status": "success", "members": members}
 
 
@@ -1748,7 +1931,8 @@ def update_station_bio(data: dict):
         success = core.update_station_bio(station_name, bio)
         if success:
             payload = {"type": "station_bio_updated", "stationName": station_name, "bio": bio}
-            asyncio.create_task(coreHelper.broadcast_to_all(payload))
+            asyncio.create_task(notify_station_members_ws(station_name, payload))
+            asyncio.create_task(touch_station_catalog_ws(station_name))
         return {"status": "success" if success else "fail"}
     except Exception as e:
         
@@ -1766,7 +1950,8 @@ def update_station_hero_banner(data: dict):
         success = core.update_station_hero_banner(station_name, hero_banner)
         if success:
             payload = {"type": "station_hero_banner_updated", "stationName": station_name, "heroBanner": hero_banner}
-            asyncio.create_task(coreHelper.broadcast_to_all(payload))
+            asyncio.create_task(notify_station_members_ws(station_name, payload))
+            asyncio.create_task(touch_station_catalog_ws(station_name))
         return {"status": "success" if success else "fail"}
     except Exception as e:
         
@@ -1784,7 +1969,8 @@ def update_station_profile_pic(data: dict):
         success = core.update_station_profile_pic(station_name, profile_pic)
         if success:
             payload = {"type": "station_profile_pic_updated", "stationName": station_name, "profilePic": profile_pic}
-            asyncio.create_task(coreHelper.broadcast_to_all(payload))
+            asyncio.create_task(notify_station_members_ws(station_name, payload))
+            asyncio.create_task(touch_station_catalog_ws(station_name))
         return {"status": "success" if success else "fail"}
     except Exception as e:
         
@@ -1799,10 +1985,25 @@ def update_station_name_fn(data: dict):
         if not old_name or not new_name:
             return JSONResponse({"status": "fail", "message": "Missing parameters"}, status_code=400)
             
+        members_snapshot = core.get_station_members(old_name, "")
         success = core.update_station_name(old_name, new_name)
         if success:
             payload = {"type": "station_name_updated", "oldName": old_name, "newName": new_name}
-            asyncio.create_task(coreHelper.broadcast_to_all(payload))
+
+            async def _fanout_rename():
+                seen: set[str] = set()
+                for u in members_snapshot or []:
+                    if not u:
+                        continue
+                    lk = str(u).lower()
+                    if lk in seen:
+                        continue
+                    seen.add(lk)
+                    await coreHelper.send_to_user(u, payload, connections)
+                await touch_station_catalog_ws(new_name)
+                await touch_station_catalog_ws(old_name)
+
+            asyncio.create_task(_fanout_rename())
         return {"status": "success" if success else "fail"}
     except Exception as e:
         
@@ -1816,7 +2017,7 @@ async def get_station_bio_fn(data: dict):
         return JSONResponse(
             {"status": "fail", "message": "Station name required"}, status_code=400
         )
-    bio = core.get_station_bio(station_name)
+    bio = await asyncio.to_thread(core.get_station_bio, station_name)
     return {"status": "success", "bio": bio}
 
 
@@ -1827,7 +2028,7 @@ async def get_station_hero_banner_fn(data: dict):
         return JSONResponse(
             {"status": "fail", "message": "Station name required"}, status_code=400
         )
-    hero_banner = core.get_station_hero_banner(station_name)
+    hero_banner = await asyncio.to_thread(core.get_station_hero_banner, station_name)
     return {"status": "success", "heroBanner": hero_banner}
 
 
@@ -1838,7 +2039,7 @@ async def get_station_profile_pic_fn(data: dict):
         return JSONResponse(
             {"status": "fail", "message": "Station name required"}, status_code=400
         )
-    profile_pic = core.get_station_profile_pic(station_name)
+    profile_pic = await asyncio.to_thread(core.get_station_profile_pic, station_name)
     return {"status": "success", "profilePic": profile_pic}
 
 
@@ -1849,7 +2050,7 @@ async def get_station_admin_fn(data: dict):
         return JSONResponse(
             {"status": "fail", "message": "Station name required"}, status_code=400
         )
-    admin = core.get_station_admin(station_name)
+    admin = await asyncio.to_thread(core.get_station_admin, station_name)
     return {"status": "success", "admin": admin}
 
 
@@ -1875,16 +2076,12 @@ async def like_post(data: dict):
     try:
         post_id = int(post_id)
         
-        def _do_like():
-            # If already liked, we still want to return success to keep UI in sync
-            success = core.like_post(post_id, username)
-            if success:
-                likes = core.modify_likes(post_id, +1)
-            else:
-                likes = core.get_likes_count(post_id)
-            return success, likes
-
-        success, likes = await asyncio.to_thread(_do_like)
+        # If already liked, we still want to return success to keep UI in sync
+        success = await asyncio.to_thread(core.like_post, post_id, username)
+        if success:
+            likes = await asyncio.to_thread(core.modify_likes, post_id, +1)
+        else:
+            likes = await asyncio.to_thread(core.get_likes_count, post_id)
         
         # Always broadcast the latest count
         asyncio.create_task(coreHelper.broadcast_like_update(post_id, likes))
@@ -1907,16 +2104,12 @@ async def unlike_post(data: dict):
     try:
         post_id = int(post_id)
         
-        def _do_unlike():
-            # Even if already unliked, return success to prevent UI from reverting
-            success = core.unlike_post(post_id, username)
-            if success:
-                likes = core.modify_likes(post_id, -1)
-            else:
-                likes = core.get_likes_count(post_id)
-            return success, likes
-
-        success, likes = await asyncio.to_thread(_do_unlike)
+        # Even if already unliked, return success to prevent UI from reverting
+        success = await asyncio.to_thread(core.unlike_post, post_id, username)
+        if success:
+            likes = await asyncio.to_thread(core.modify_likes, post_id, -1)
+        else:
+            likes = await asyncio.to_thread(core.get_likes_count, post_id)
 
         asyncio.create_task(coreHelper.broadcast_like_update(post_id, likes))
         return {"status": "success", "likes": likes}
@@ -1950,39 +2143,87 @@ async def ws_likes_endpoint(ws: WebSocket, username: str):
         coreHelper.unregister_ws(username, ws, connections)
 
 
-@app.get("/api/bitai/conversation/{username}")
-def get_bitai_conversation(username: str):
-    conv_id = core.get_or_create_bitai_conversation(username)
-    return {"conversationId": conv_id, "status": "success" if conv_id > 0 else "error"}
+# --- BitAI inline chat (@bitai in any thread + Groq) -------------------
 
 
-@app.post("/api/bitai/chat")
-async def bitai_chat(data: dict):
-    message = data.get("message", "")
-    user = data.get("username", "")
-    history = data.get("history", [])
+def _is_bitai_direct_dm(convo_id: int) -> bool:
+    """Exactly two participants: user + BitAI (private AI thread)."""
+    users = core.get_conversation_users(convo_id)
+    if len(users) != 2:
+        return False
+    return "BitAI" in users
 
-    if not message or not user:
-        return {"status": "error", "message": "Missing parameters"}
 
-    conv_id = core.get_or_create_bitai_conversation(user)
-    if conv_id <= 0:
-        return {"status": "error", "message": "Could not create conversation"}
+def _message_mentions_bitai(content: str) -> bool:
+    if not content or not isinstance(content, str):
+        return False
+    return re.search(r"@bitai\b", content, re.IGNORECASE) is not None
 
-    core.send_message(user, conv_id, message, "text", "")
 
-    try:
-        import httpx
+def _should_run_bitai_in_conversation(convo_id: int, sender: str, content: str, msg_type: str) -> bool:
+    if msg_type != "text" or sender == "BitAI":
+        return False
+    if not content or not str(content).strip():
+        return False
+    if _is_bitai_direct_dm(convo_id):
+        return True
+    return _message_mentions_bitai(content)
 
-        api_key = os.environ.get("GROQ_API_KEY", "")
 
-        if not api_key:
-            return {"status": "error", "response": "Me brain dead. You set GROQ_API_KEY in .env"}
+def _bitai_multi_audience(convo_id: int) -> bool:
+    """Reply visible to a group or two humans (not a solo BitAI DM)."""
+    users = set(core.get_conversation_users(convo_id))
+    if len(users) == 2 and "BitAI" in users:
+        return False
+    return len(users) >= 2
 
-        system_prompt = f"""You are BitAI. Your name is BitAI. You are a friendly AI assistant for YapStation - a cyberpunk social media platform.
-The user chatting with you is named {user}. Call them by their name!
+
+def _thread_to_llm_history(convo_id: int) -> list:
+    thread = core.get_messages(convo_id)
+    out = []
+    for m in thread:
+        if m.get("type") != "text":
+            continue
+        text = m.get("content", "") or ""
+        if m.get("sender") == "BitAI":
+            out.append({"role": "assistant", "content": text})
+        else:
+            out.append({"role": "user", "content": text})
+    return out
+
+
+def _prior_llm_history_for_latest(convo_id: int) -> list:
+    full = _thread_to_llm_history(convo_id)
+    if not full:
+        return []
+    return full[:-1][-8:]
+
+
+async def groq_bitai_completion(
+    acting_username: str,
+    latest_user_text: str,
+    prior_history: list,
+    convo_id: int,
+) -> tuple:
+    """
+    prior_history: {role, content} turns before the latest user message.
+    Returns ("success"|"error", response_string).
+    """
+    import httpx
+
+    api_key = os.environ.get("GROQ_API_KEY", "")
+    if not api_key:
+        return ("error", "Me brain dead. You set GROQ_API_KEY in .env")
+
+    group_note = ""
+    if _bitai_multi_audience(convo_id):
+        group_note = "\n\nYour message is shown to everyone in this chat. Keep it short. Someone used @bitai to ask you—answer them helpfully."
+
+    system_prompt = f"""You are BitAI. Your name is BitAI. You are a friendly AI assistant for YapStation - a cyberpunk social media platform.
+The user who asked is named {acting_username}. Call them by their name when it fits.
 YapStation features: chat with friends, yap stations (podcasts), posts, followers, themes.
 You help users with: chatting, advice, answering questions about the platform, fun conversation.
+{group_note}
 
 You speak CAVE MAN style. Short. Baby words. Mostly just pronouns and verbs. Friendly. Like speaking to a buddy.
 Examples: "Hey {{name}}! Me here. Whats up? Me help you?"
@@ -1990,14 +2231,15 @@ Or: "Hey {{name}}, me understand. You do this..."
 Or: "Cool cool! Me think you good. Go for it!"
 Or: "No stress! Me got you. Tell me more."
 
-Sometimes use the user's name ({user}) in your response when possible. Be casual, helpful, fun."""
+Sometimes use the user's name ({acting_username}) in your response when possible. Be casual, helpful, fun."""
 
-        messages = [{"role": "system", "content": system_prompt}]
-        for msg in history[-8:]:
-            role = "user" if msg.get("role") == "user" else "assistant"
-            messages.append({"role": role, "content": msg.get("content", "")})
-        messages.append({"role": "user", "content": message})
+    messages = [{"role": "system", "content": system_prompt}]
+    for msg in prior_history[-8:]:
+        role = "user" if msg.get("role") == "user" else "assistant"
+        messages.append({"role": role, "content": msg.get("content", "")})
+    messages.append({"role": "user", "content": latest_user_text})
 
+    try:
         async with httpx.AsyncClient(timeout=30.0) as client:
             response = await client.post(
                 "https://api.groq.com/openai/v1/chat/completions",
@@ -2017,14 +2259,111 @@ Sometimes use the user's name ({user}) in your response when possible. Be casual
             result = response.json()
             ai_response = result.get("choices", [{}])[0].get("message", {}).get("content", "")
             if ai_response:
-                core.send_message("BitAI", conv_id, ai_response, "text", "")
-                return {"status": "success", "response": ai_response}
-            return {"status": "error", "response": "Me no think. Try again."}
-        else:
-            return {"status": "error", "response": f"Me hurt. Error: {response.status_code}"}
+                return ("success", ai_response)
+            return ("error", "Me no think. Try again.")
+        return ("error", f"Me hurt. Error: {response.status_code}")
+    except Exception:
+        return ("error", "Me broken. Fix me.")
+
+
+async def broadcast_chat_participants(
+    convo_id: int,
+    sender: str,
+    msg_type: str,
+    content: str,
+    media: str,
+    reply_to,
+    connections: dict,
+):
+    """Invalidate conv caches and push message (+ notifications) to conversation members."""
+    users = await asyncio.to_thread(core.get_conversation_users, convo_id)
+    for u in users:
+        _conv_cache.pop(u, None)
+    users = await asyncio.to_thread(core.get_conversation_users, convo_id)
+    payload = {
+        "conversationId": convo_id,
+        "sender": sender,
+        "type": msg_type,
+        "content": content,
+        "media": media or "",
+        "time": int(time.time()),
+        "id": f"{convo_id}_{sender}_{int(time.time() * 1000)}",
+        "replyTo": reply_to,
+    }
+    for u in users:
+        if u.lower() not in connections:
+            continue
+        for conn in list(connections[u.lower()]):
+            try:
+                await conn.send_json(payload)
+                if u != sender:
+                    await conn.send_json(
+                        {
+                            "type": "notification",
+                            "sender": sender,
+                            "content": content,
+                            "conversationId": convo_id,
+                        }
+                    )
+            except Exception:
+                connections[u.lower()].discard(conn)
+
+
+async def _bitai_respond_in_conversation(convo_id: int, sender: str, content: str, connections: dict):
+    """Groq reply posted into this conversation and broadcast (like WhatsApp @bot)."""
+    prior = _prior_llm_history_for_latest(convo_id)
+    status, response_text = await groq_bitai_completion(sender, content, prior, convo_id)
+    if status != "success" or not response_text:
+        response_text = response_text or "Me no think. Try again."
+    try:
+        await asyncio.to_thread(core.send_message, "BitAI", convo_id, response_text, "text", "")
+        await broadcast_chat_participants(
+            convo_id, "BitAI", "text", response_text, "", None, connections
+        )
+    except Exception:
+        pass
+
+
+@app.get("/api/bitai/conversation/{username}")
+def get_bitai_conversation(username: str):
+    conv_id = core.get_or_create_bitai_conversation(username)
+    return {"conversationId": conv_id, "status": "success" if conv_id > 0 else "error"}
 
-    except Exception as e:
-        return {"status": "error", "response": "Me broken. Fix me."}
+
+@app.post("/api/bitai/chat")
+async def bitai_chat(data: dict):
+    message = data.get("message", "")
+    user = data.get("username", "")
+    history = data.get("history", [])
+
+    if not message or not user:
+        return {"status": "error", "message": "Missing parameters"}
+
+    conv_id = await asyncio.to_thread(core.get_or_create_bitai_conversation, user)
+    if conv_id <= 0:
+        return {"status": "error", "message": "Could not create conversation"}
+
+    await asyncio.to_thread(core.send_message, user, conv_id, message, "text", "")
+
+    prior = []
+    for msg in history[-8:]:
+        prior.append(
+            {
+                "role": "user" if msg.get("role") == "user" else "assistant",
+                "content": msg.get("content", ""),
+            }
+        )
+
+    status, response_text = await groq_bitai_completion(user, message, prior, conv_id)
+
+    if status == "success" and response_text:
+        await asyncio.to_thread(core.send_message, "BitAI", conv_id, response_text, "text", "")
+        await broadcast_chat_participants(
+            conv_id, "BitAI", "text", response_text, "", None, connections
+        )
+        return {"status": "success", "response": response_text}
+
+    return {"status": "error", "response": response_text or "Me broken. Fix me."}
 
 
 @app.websocket("/ws/{username}")
@@ -2139,20 +2478,21 @@ async def websocket_endpoint(ws: WebSocket, username: str):
                     convo_id = int(convo_id)
                 except Exception:
                     continue
-                users = core.get_conversation_users(convo_id)
+                users = await asyncio.to_thread(core.get_conversation_users, convo_id)
                 payload = {
                     "type": msg_type,
                     "sender": sender,
                     "conversationId": convo_id,
                 }
                 for u in users:
-                    if u == sender or u not in connections:
+                    ukey = u.lower()
+                    if u.lower() == sender.lower() or ukey not in connections:
                         continue
-                    for conn in list(connections[u]):
+                    for conn in list(connections[ukey]):
                         try:
                             await conn.send_json(payload)
                         except Exception:
-                            connections[u].discard(conn)
+                            connections[ukey].discard(conn)
                 continue
             convo_id = data.get("conversationId")
             sender = data.get("sender")
@@ -2169,10 +2509,10 @@ async def websocket_endpoint(ws: WebSocket, username: str):
                 continue
 
             # Blocking Check
-            users = core.get_conversation_users(convo_id)
+            users = await asyncio.to_thread(core.get_conversation_users, convo_id)
             is_blocked = False
             for u in users:
-                if u != sender and core.is_blocked_either_way(sender, u):
+                if u != sender and await asyncio.to_thread(core.is_blocked_either_way, sender, u):
                     is_blocked = True
                     break
             
@@ -2181,38 +2521,18 @@ async def websocket_endpoint(ws: WebSocket, username: str):
                 continue
 
             try:
-                core.send_message(sender, convo_id, content, msg_type, media, reply_to)
+                await asyncio.to_thread(core.send_message, sender, convo_id, content, msg_type, media, reply_to)
             except Exception:
                 continue
-            for u in users:
-                _conv_cache.pop(u, None)
-            users = core.get_conversation_users(convo_id)
-            payload = {
-                "conversationId": convo_id,
-                "sender": sender,
-                "type": msg_type,
-                "content": content,
-                "media": media,
-                "time": int(time.time()),
-                "id": f"{convo_id}_{sender}_{int(time.time() * 1000)}",
-                "replyTo": reply_to,
-            }
-            for u in users:
-                if u in connections:
-                    for conn in list(connections[u]):
-                        try:
-                            await conn.send_json(payload)
-                            if u != sender:
-                                await conn.send_json(
-                                    {
-                                        "type": "notification",
-                                        "sender": sender,
-                                        "content": content,
-                                        "conversationId": convo_id,
-                                    }
-                                )
-                        except Exception:
-                            connections[u].discard(conn)
+            await broadcast_chat_participants(
+                convo_id, sender, msg_type, content, media, reply_to, connections
+            )
+            if msg_type == "text" and _should_run_bitai_in_conversation(
+                convo_id, sender, content, msg_type
+            ):
+                asyncio.create_task(
+                    _bitai_respond_in_conversation(convo_id, sender, content, connections)
+                )
     except Exception:
         pass
     finally:

smoke_test.py

@@ -0,0 +1,64 @@
+"""
+Quick check that the FastAPI backend is up (default http://127.0.0.1:7860/).
+Uses stdlib only — no extra packages.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+import urllib.error
+import urllib.request
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Verify YapStation FastAPI is responding.")
+    parser.add_argument(
+        "--url",
+        default="",
+        help="Full base URL (overrides --host/--port). Example: http://127.0.0.1:7860/",
+    )
+    parser.add_argument("--host", default="127.0.0.1", help="Bind host of the API server")
+    parser.add_argument("--port", type=int, default=7860, help="Port (default 7860)")
+    args = parser.parse_args()
+
+    url = args.url.strip()
+    if not url:
+        url = f"http://{args.host}:{args.port}/"
+
+    req = urllib.request.Request(url, method="GET", headers={"Accept": "application/json"})
+    try:
+        with urllib.request.urlopen(req, timeout=8) as resp:
+            raw = resp.read().decode()
+            code = resp.status
+    except urllib.error.HTTPError as e:
+        print(f"FAIL: HTTP {e.code} from {url}")
+        return 1
+    except urllib.error.URLError as e:
+        print(f"FAIL: Could not reach {url}")
+        print(f"      ({e.reason})")
+        print("      Start the API with run-backend.bat or run-yapstation.bat.")
+        return 1
+    except Exception as e:
+        print(f"FAIL: {e}")
+        return 1
+
+    try:
+        data = json.loads(raw)
+    except json.JSONDecodeError:
+        print(f"FAIL: Non-JSON response (HTTP {code}) from {url}")
+        print(raw[:500])
+        return 1
+
+    if data.get("status") == "success":
+        msg = data.get("message", "")
+        print(f"OK: Backend is running ({code}) — {msg}")
+        print(f"    URL: {url}")
+        return 0
+
+    print(f"FAIL: Unexpected JSON from {url}: {data!r}")
+    return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

stack_test.py

@@ -0,0 +1,69 @@
+"""
+Verify FastAPI (:7860) and the Next.js -> API proxy (:3000/api/*) together.
+Stdlib only.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+import urllib.error
+import urllib.request
+
+
+def fetch_json(url: str, timeout: float = 8.0) -> tuple[int, dict]:
+    req = urllib.request.Request(url, headers={"Accept": "application/json"})
+    with urllib.request.urlopen(req, timeout=timeout) as resp:
+        raw = resp.read().decode()
+        code = resp.status
+    return code, json.loads(raw)
+
+
+def main() -> int:
+    p = argparse.ArgumentParser(description="Test backend + Next.js API proxy")
+    p.add_argument("--api-port", type=int, default=7860)
+    p.add_argument("--next-port", type=int, default=3000)
+    args = p.parse_args()
+
+    api_root = f"http://127.0.0.1:{args.api_port}/"
+    next_me = f"http://127.0.0.1:{args.next_port}/api/me"
+
+    ok = True
+
+    # 1) Direct FastAPI
+    try:
+        code, data = fetch_json(api_root)
+        if data.get("status") == "success":
+            print(f"OK: FastAPI direct      GET /     -> {code} {data.get('message', '')[:60]}")
+        else:
+            print(f"FAIL: FastAPI root JSON unexpected: {data!r}")
+            ok = False
+    except Exception as e:
+        print(f"FAIL: FastAPI not reachable at {api_root} ({e})")
+        ok = False
+
+    # 2) Through Next (same path the browser uses: /api/me)
+    try:
+        code, data = fetch_json(next_me)
+        if code == 200 and isinstance(data, dict) and "status" in data:
+            print(
+                f"OK: Next.js proxy       GET /api/me -> {code} status={data.get('status')!r}"
+            )
+        else:
+            print(f"FAIL: Next /api/me unexpected: HTTP {code} {data!r}")
+            ok = False
+    except Exception as e:
+        print(f"FAIL: Next.js not reachable or proxy error at {next_me}")
+        print(f"      ({e})")
+        print("      Start frontend: run-frontend.bat or run-yapstation.bat")
+        ok = False
+
+    if ok:
+        print()
+        print("Stack check passed: browser calls to /api/* on :3000 reach FastAPI on :7860.")
+        return 0
+    return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())