feat: Add FastAPI backend + React frontend

- FastAPI app.py with CORS and API routes
- Backend code (models, routes, services, system)
- React frontend (built static files)
- Requirements.txt for dependencies

Co-authored-by: openhands <openhands@all-hands.dev>

README.md

@@ -3,11 +3,13 @@ title: Chat App
 emoji: 💬
 colorFrom: blue
 colorTo: indigo
-sdk: static
-app_file: index.html
+sdk: gradio
+sdk_version: 6.11.0
+python_version: "3.11"
+app_file: app.py
 pinned: false
 license: mit
-short_description: 'Modern chat application with Three.js background'
+short_description: 'Modern chat application with FastAPI + React'
 ---
 
 # Chat-App 💬

app.py

@@ -0,0 +1,88 @@
+# Chat App - FastAPI + React (HuggingFace Space)
+
+from fastapi import FastAPI, Request
+from fastapi.responses import HTMLResponse, JSONResponse
+from fastapi.staticfiles import StaticFiles
+from fastapi.middleware.cors import CORSMiddleware
+import uvicorn
+import os
+
+app = FastAPI(title="Chat App", version="1.0.0")
+
+# Enable CORS
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Get the directory where this file is located
+BASE_DIR = os.path.dirname(os.path.abspath(__file__))
+
+# Mount frontend static files
+frontend_path = os.path.join(BASE_DIR, "frontend")
+if os.path.exists(frontend_path):
+    app.mount("/frontend", StaticFiles(directory=frontend_path), name="frontend")
+
+# API Routes
+@app.get("/")
+async def root(request: Request):
+    """Serve the frontend index.html"""
+    index_path = os.path.join(BASE_DIR, "index.html")
+    if os.path.exists(index_path):
+        with open(index_path, "r") as f:
+            content = f.read()
+        return HTMLResponse(content=content)
+    return {"message": "Chat App API", "status": "running"}
+
+@app.get("/api/health")
+async def health():
+    """Health check endpoint"""
+    return {"status": "healthy", "service": "chat-app"}
+
+@app.get("/api/v1")
+async def api_info():
+    """API info endpoint"""
+    return {
+        "name": "Chat App API",
+        "version": "1.0.0",
+        "endpoints": {
+            "auth": "/api/v1/auth",
+            "users": "/api/v1/users", 
+            "chat": "/api/v1/chat",
+            "groups": "/api/v1/groups"
+        }
+    }
+
+# Auth endpoints
+@app.post("/api/v1/auth/register")
+async def register(data: dict):
+    return {"message": "Register endpoint - requires database"}
+
+@app.post("/api/v1/auth/login")
+async def login(data: dict):
+    return {"message": "Login endpoint - requires database"}
+
+# Chat endpoints
+@app.get("/api/v1/chat/conversations")
+async def get_conversations():
+    return {"conversations": [], "message": "Requires database setup"}
+
+@app.post("/api/v1/chat/send")
+async def send_message(data: dict):
+    return {"message": "Send message - requires database"}
+
+# Group endpoints
+@app.get("/api/v1/groups")
+async def get_groups():
+    return {"groups": [], "message": "Requires database setup"}
+
+@app.post("/api/v1/groups/create")
+async def create_group(data: dict):
+    return {"message": "Create group - requires database"}
+
+if __name__ == "__main__":
+    port = int(os.environ.get("PORT", 7860))
+    uvicorn.run(app, host="0.0.0.0", port=port)

backend/app/config.py

@@ -0,0 +1,51 @@
+"""
+Configuration module - loads environment variables and secrets.
+DO NOT commit actual secrets to version control.
+"""
+from pydantic_settings import BaseSettings
+from typing import Optional
+import os
+
+
+class Settings(BaseSettings):
+    """Application settings loaded from environment variables."""
+
+    # App
+    APP_NAME: str = "ChatApp"
+    APP_VERSION: str = "1.0.0"
+    DEBUG: bool = False
+    SECRET_KEY: str = "change-me-in-production"  # TODO: Set via env
+
+    # Database
+    DATABASE_URL: str = "postgresql+asyncpg://user:pass@localhost:5432/chatapp"
+
+    # JWT
+    JWT_SECRET: str = "jwt-secret-change-me"
+    JWT_ALGORITHM: str = "HS256"
+    JWT_EXPIRATION_MINUTES: int = 60 * 24 * 7  # 7 days
+
+    # Redis (for async queue)
+    REDIS_URL: str = "redis://localhost:6379/0"
+
+    # GitHub (for storage backup)
+    GITHUB_TOKEN: Optional[str] = None
+    GITHUB_REPO: Optional[str] = None  # e.g., "username/chat-storage"
+
+    # HuggingFace (optional)
+    HF_TOKEN: Optional[str] = None
+    HF_SPACE_ID: Optional[str] = None
+
+    # Compression settings
+    COMPRESSION_ENABLED: bool = True
+    MAX_BATCH_SIZE_MB: int = 10
+    BATCH_INTERVAL_MINUTES: int = 60
+
+    # Storage
+    STORAGE_PATH: str = "./storage/data"
+
+    class Config:
+        env_file = ".env"
+        case_sensitive = True
+
+
+settings = Settings()

backend/app/dependencies.py

@@ -0,0 +1,49 @@
+"""
+Dependencies - shared FastAPI dependencies (auth, database, etc).
+"""
+from typing import Optional
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import create_async_engine, select
+from sqlalchemy.orm import sessionmaker, declarative_base
+
+from .config import settings
+
+# Security scheme
+security = HTTPBearer()
+
+# Database setup
+engine = create_async_engine(settings.DATABASE_URL, echo=settings.DEBUG)
+async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+
+Base = declarative_base()
+
+
+async def get_db() -> AsyncSession:
+    """Database session dependency."""
+    async with async_session() as session:
+        yield session
+
+
+async def get_current_user(
+    credentials: HTTPAuthorizationCredentials = Depends(security)
+) -> dict:
+    """
+    Validate JWT token and return current user.
+    In production, decode JWT and fetch user from DB.
+    """
+    # TODO: Implement actual JWT validation
+    # For now, return a placeholder user
+    token = credentials.credentials
+    # This would decode the token and verify it
+    return {"id": "user_placeholder", "username": "test_user"}
+
+
+async def get_optional_user(
+    credentials: Optional[HTTPAuthorizationCredentials] = Depends(security)
+) -> Optional[dict]:
+    """Optional auth - returns None if not authenticated."""
+    if not credentials:
+        return None
+    return await get_current_user(credentials)

backend/app/main.py

@@ -0,0 +1,83 @@
+"""
+Main FastAPI application entry point.
+"""
+from fastapi import FastAPI, WebSocket
+from fastapi.middleware.cors import CORSMiddleware
+from contextlib import asynccontextmanager
+
+from .config import settings
+from .routes import auth_router, user_router, chat_router, group_router, system_router
+from .sockets.ws import websocket_endpoint
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Application lifespan - startup and shutdown."""
+    # Startup
+    print(f"Starting {settings.APP_NAME} v{settings.APP_VERSION}")
+    
+    # TODO: Start worker and scheduler
+    # from .system import worker, batch_scheduler
+    # await worker.start()
+    # await batch_scheduler.start()
+    
+    yield
+    
+    # Shutdown
+    print("Shutting down...")
+    # await worker.stop()
+    # await batch_scheduler.stop()
+
+
+# Create FastAPI app
+app = FastAPI(
+    title=settings.APP_NAME,
+    version=settings.APP_VERSION,
+    description="Chat application with async message processing",
+    lifespan=lifespan
+)
+
+# CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # Configure appropriately for production
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+# Include routers
+app.include_router(auth_router, prefix="/api")
+app.include_router(user_router, prefix="/api")
+app.include_router(chat_router, prefix="/api")
+app.include_router(group_router, prefix="/api")
+app.include_router(system_router, prefix="/api")
+
+
+# WebSocket endpoint
+@app.websocket("/ws/{user_id}")
+async def websocket_route(websocket: WebSocket, user_id: str):
+    """WebSocket endpoint for real-time notifications."""
+    await websocket_endpoint(websocket, user_id)
+
+
+# Root endpoint
+@app.get("/")
+async def root():
+    """Root endpoint."""
+    return {
+        "name": settings.APP_NAME,
+        "version": settings.APP_VERSION,
+        "status": "running"
+    }
+
+
+@app.get("/api/health")
+async def health():
+    """Health check endpoint."""
+    return {"status": "healthy"}
+
+
+# Import all routes to ensure they're registered
+__all__ = ["app"]

backend/app/models/__init__.py

@@ -0,0 +1,10 @@
+"""
+Models package - all database models.
+"""
+from .user import User
+from .message import Message
+from .group import Group, GroupMember
+from .friend import Friend
+from .reaction import Reaction
+
+__all__ = ["User", "Message", "Group", "GroupMember", "Friend", "Reaction"]

backend/app/models/friend.py

@@ -0,0 +1,40 @@
+"""
+Friend model - represents friend relationships between users.
+"""
+from sqlalchemy import Column, String, DateTime, ForeignKey, Boolean, Enum as SQLEnum
+from sqlalchemy.sql import func
+from ..dependencies import Base
+import uuid
+import enum
+
+
+class FriendStatus(enum.Enum):
+    """Friend request status."""
+    PENDING = "pending"
+    ACCEPTED = "accepted"
+    BLOCKED = "blocked"
+
+
+class Friend(Base):
+    """Friend relationship model."""
+
+    __tablename__ = "friends"
+
+    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
+    
+    # User who initiated the friend request
+    user_id = Column(String, ForeignKey("users.id"), nullable=False, index=True)
+    
+    # User who received the request
+    friend_id = Column(String, ForeignKey("users.id"), nullable=False, index=True)
+    
+    # Status: pending, accepted, blocked
+    status = Column(String(20), default="pending")
+    
+    # Timestamps
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), onupdate=func.now())
+    accepted_at = Column(DateTime(timezone=True), nullable=True)
+
+    def __repr__(self):
+        return f"<Friend {self.user_id} <-> {self.friend_id}>"

backend/app/models/group.py

@@ -0,0 +1,58 @@
+"""
+Group model - represents chat groups.
+"""
+from sqlalchemy import Column, String, DateTime, Text, ForeignKey, Boolean, Integer
+from sqlalchemy.sql import func
+from ..dependencies import Base
+import uuid
+
+
+class Group(Base):
+    """Chat group model."""
+
+    __tablename__ = "groups"
+
+    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
+    name = Column(String(100), nullable=False)
+    description = Column(Text, nullable=True)
+    avatar = Column(Text, nullable=True)
+    
+    # Admin (owner)
+    owner_id = Column(String, ForeignKey("users.id"), nullable=False)
+    
+    # Settings
+    is_private = Column(Boolean, default=False)
+    max_members = Column(Integer, default=100)
+    
+    # Timestamps
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), onupdate=func.now())
+
+    def __repr__(self):
+        return f"<Group {self.name}>"
+
+
+class GroupMember(Base):
+    """Group member with roles."""
+
+    __tablename__ = "group_members"
+
+    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
+    group_id = Column(String, ForeignKey("groups.id"), nullable=False, index=True)
+    user_id = Column(String, ForeignKey("users.id"), nullable=False, index=True)
+    
+    # Role: 'admin', 'co_admin', 'member'
+    role = Column(String(20), default="member")
+    
+    # Custom nickname in group
+    nickname = Column(String(50), nullable=True)
+    
+    # Mute notifications?
+    is_muted = Column(Boolean, default=False)
+    is_banned = Column(Boolean, default=False)
+    
+    # Timestamps
+    joined_at = Column(DateTime(timezone=True), server_default=func.now())
+
+    def __repr__(self):
+        return f"<GroupMember {self.user_id} in {self.group_id}>"

backend/app/models/message.py

@@ -0,0 +1,49 @@
+"""
+Message model - represents chat messages.
+"""
+from sqlalchemy import Column, String, DateTime, Text, ForeignKey, Boolean, Integer
+from sqlalchemy.sql import func
+from sqlalchemy.orm import relationship
+from ..dependencies import Base
+import uuid
+
+
+class Message(Base):
+    """Chat message model."""
+
+    __tablename__ = "messages"
+
+    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
+    sender_id = Column(String, ForeignKey("users.id"), nullable=False, index=True)
+    
+    # Message type: 'text', 'image', 'file', 'system'
+    type = Column(String(20), default="text")
+    
+    # Content (text or file URL)
+    content = Column(Text, nullable=False)
+    
+    # For private chat (user-to-user)
+    receiver_id = Column(String, ForeignKey("users.id"), nullable=True, index=True)
+    
+    # For group chat
+    group_id = Column(String, ForeignKey("groups.id"), nullable=True, index=True)
+    
+    # Message status
+    is_deleted = Column(Boolean, default=False)
+    is_archived = Column(Boolean, default=False)
+    
+    # Batch processing info
+    batch_id = Column(String, nullable=True, index=True)
+    compressed = Column(Boolean, default=False)
+    
+    # Timestamps
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), onupdate=func.now())
+    deleted_at = Column(DateTime(timezone=True), nullable=True)
+
+    # Relationships
+    # sender = relationship("User", back_populates="messages")
+    # reactions = relationship("Reaction", back_populates="message")
+
+    def __repr__(self):
+        return f"<Message {self.id[:8]} from {self.sender_id[:8]}>"

backend/app/models/reaction.py

@@ -0,0 +1,26 @@
+"""
+Reaction model - emoji/like reactions on messages.
+"""
+from sqlalchemy import Column, String, DateTime, ForeignKey
+from sqlalchemy.sql import func
+from ..dependencies import Base
+import uuid
+
+
+class Reaction(Base):
+    """Message reaction (emoji) model."""
+
+    __tablename__ = "reactions"
+
+    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
+    message_id = Column(String, ForeignKey("messages.id"), nullable=False, index=True)
+    user_id = Column(String, ForeignKey("users.id"), nullable=False, index=True)
+    
+    # Emoji: '👍', '❤️', '😂', '😢', '😮', '😡', etc.
+    emoji = Column(String(10), nullable=False)
+    
+    # Timestamps
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+
+    def __repr__(self):
+        return f"<Reaction {self.emoji} on {self.message_id[:8]}>"

backend/app/models/user.py

@@ -0,0 +1,39 @@
+"""
+User model - represents a user in the chat system.
+"""
+from sqlalchemy import Column, String, DateTime, Boolean, Text
+from sqlalchemy.sql import func
+from ..dependencies import Base
+import uuid
+
+
+class User(Base):
+    """User account model."""
+
+    __tablename__ = "users"
+
+    id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
+    username = Column(String(50), unique=True, nullable=False, index=True)
+    email = Column(String(255), unique=True, nullable=False, index=True)
+    password_hash = Column(String(255), nullable=False)
+
+    # Profile
+    display_name = Column(String(100), nullable=True)
+    avatar = Column(Text, nullable=True)  # URL or base64
+    bio = Column(Text, nullable=True)
+    status = Column(String(20), default="offline")  # online, offline, away
+
+    # Settings
+    is_public = Column(Boolean, default=True)  # Public profile?
+    allow_friends = Column(Boolean, default=True)
+
+    # Timestamps
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), onupdate=func.now())
+    last_seen = Column(DateTime(timezone=True), nullable=True)
+
+    # Blocked users (relationship defined elsewhere)
+    # blocked_users = relationship("BlockedUser", back_populates="user")
+
+    def __repr__(self):
+        return f"<User {self.username}>"

backend/app/routes/__init__.py

@@ -0,0 +1,10 @@
+"""
+Routes package - API endpoints.
+"""
+from .auth import router as auth_router
+from .user import router as user_router
+from .chat import router as chat_router
+from .group import router as group_router
+from .system import router as system_router
+
+__all__ = ["auth_router", "user_router", "chat_router", "group_router", "system_router"]

backend/app/routes/auth.py

@@ -0,0 +1,185 @@
+"""
+Authentication routes - register, login, logout, token refresh.
+"""
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.security import HTTPBearer
+from pydantic import BaseModel, EmailStr
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select
+from typing import Optional
+
+from ..dependencies import get_db, get_current_user
+from ..models import User
+from ..utils.security import hash_password, verify_password, create_access_token
+from ..utils.avatar import generate_avatar
+
+router = APIRouter(prefix="/auth", tags=["Authentication"])
+security = HTTPBearer()
+
+
+# --- Schemas ---
+class RegisterRequest(BaseModel):
+    username: str
+    email: EmailStr
+    password: str
+
+
+class LoginRequest(BaseModel):
+    email: EmailStr
+    password: str
+
+
+class AuthResponse(BaseModel):
+    access_token: str
+    token_type: str = "bearer"
+    user: dict
+
+
+class UserResponse(BaseModel):
+    id: str
+    username: str
+    email: str
+    display_name: Optional[str] = None
+    avatar: Optional[str] = None
+    status: str = "offline"
+
+
+# --- Routes ---
+@router.post("/register", response_model=AuthResponse, status_code=status.HTTP_201_CREATED)
+async def register(
+    request: RegisterRequest,
+    db: AsyncSession = Depends(get_db)
+):
+    """Register a new user account."""
+    # Check if email already exists
+    result = await db.execute(select(User).where(User.email == request.email))
+    if result.scalar_one_or_none():
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Email already registered"
+        )
+    
+    # Check if username exists
+    result = await db.execute(select(User).where(User.username == request.username))
+    if result.scalar_one_or_none():
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Username already taken"
+        )
+    
+    # Create user
+    password_hash = hash_password(request.password)
+    avatar_url = await generate_avatar(request.username, request.username)
+    
+    user = User(
+        username=request.username,
+        email=request.email,
+        password_hash=password_hash,
+        display_name=request.username,
+        avatar=avatar_url,
+        status="offline"
+    )
+    
+    db.add(user)
+    await db.commit()
+    await db.refresh(user)
+    
+    # Generate token
+    access_token = create_access_token({"sub": user.id, "username": user.username})
+    
+    return AuthResponse(
+        access_token=access_token,
+        user={
+            "id": user.id,
+            "username": user.username,
+            "email": user.email,
+            "display_name": user.display_name,
+            "avatar": user.avatar,
+            "status": user.status
+        }
+    )
+
+
+@router.post("/login", response_model=AuthResponse)
+async def login(
+    request: LoginRequest,
+    db: AsyncSession = Depends(get_db)
+):
+    """Login with email and password."""
+    # Find user
+    result = await db.execute(select(User).where(User.email == request.email))
+    user = result.scalar_one_or_none()
+    
+    if not user or not verify_password(request.password, user.password_hash):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid email or password"
+        )
+    
+    # Update status
+    user.status = "online"
+    await db.commit()
+    
+    # Generate token
+    access_token = create_access_token({"sub": user.id, "username": user.username})
+    
+    return AuthResponse(
+        access_token=access_token,
+        user={
+            "id": user.id,
+            "username": user.username,
+            "email": user.email,
+            "display_name": user.display_name,
+            "avatar": user.avatar,
+            "status": user.status
+        }
+    )
+
+
+@router.post("/logout")
+async def logout(
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Logout and update user status."""
+    result = await db.execute(select(User).where(User.id == current_user["id"]))
+    user = result.scalar_one_or_none()
+    
+    if user:
+        user.status = "offline"
+        await db.commit()
+    
+    return {"message": "Logged out successfully"}
+
+
+@router.get("/me", response_model=UserResponse)
+async def get_me(
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Get current authenticated user profile."""
+    result = await db.execute(select(User).where(User.id == current_user["id"]))
+    user = result.scalar_one_or_none()
+    
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found")
+    
+    return UserResponse(
+        id=user.id,
+        username=user.username,
+        email=user.email,
+        display_name=user.display_name,
+        avatar=user.avatar,
+        status=user.status
+    )
+
+
+@router.post("/refresh")
+async def refresh_token(
+    current_user: dict = Depends(get_current_user)
+):
+    """Refresh access token."""
+    access_token = create_access_token(
+        {"sub": current_user["id"], "username": current_user.get("username", "user")}
+    )
+    return {"access_token": access_token, "token_type": "bearer"}

backend/app/routes/chat.py

@@ -0,0 +1,255 @@
+"""
+Chat routes - send messages, delete, reactions.
+"""
+from fastapi import APIRouter, Depends, HTTPException, status
+from pydantic import BaseModel
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, and_, or_
+from typing import Optional, List
+from datetime import datetime
+import uuid
+
+from ..dependencies import get_db, get_current_user
+from ..models import Message, Reaction, User
+from ..utils.formatter import clean_message
+
+router = APIRouter(prefix="/chat", tags=["Chat"])
+
+
+# --- Schemas ---
+class SendMessageRequest(BaseModel):
+    receiver_id: Optional[str] = None
+    group_id: Optional[str] = None
+    content: str
+    type: str = "text"  # text, image, file
+
+
+class MessageResponse(BaseModel):
+    id: str
+    sender_id: str
+    receiver_id: Optional[str] = None
+    group_id: Optional[str] = None
+    type: str
+    content: str
+    is_deleted: bool
+    created_at: datetime
+
+
+class ReactionRequest(BaseModel):
+    message_id: str
+    emoji: str
+
+
+class ReactionResponse(BaseModel):
+    id: str
+    message_id: str
+    user_id: str
+    emoji: str
+    created_at: datetime
+
+
+# --- Routes ---
+@router.post("/send", response_model=MessageResponse, status_code=status.HTTP_201_CREATED)
+async def send_message(
+    request: SendMessageRequest,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Send a message to a user or group."""
+    # Validate - must have either receiver_id or group_id
+    if not request.receiver_id and not request.group_id:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Must specify receiver_id or group_id"
+        )
+    
+    if request.receiver_id and request.group_id:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Cannot send to both receiver and group"
+        )
+    
+    # Clean message content
+    content = clean_message(request.content)
+    
+    # Create message
+    message = Message(
+        sender_id=current_user["id"],
+        receiver_id=request.receiver_id,
+        group_id=request.group_id,
+        content=content,
+        type=request.type,
+        compressed=False
+    )
+    
+    db.add(message)
+    await db.commit()
+    await db.refresh(message)
+    
+    # TODO: Push to async queue for batch processing
+    # await queue.push_message(message)
+    
+    return MessageResponse(
+        id=message.id,
+        sender_id=message.sender_id,
+        receiver_id=message.receiver_id,
+        group_id=message.group_id,
+        type=message.type,
+        content=message.content,
+        is_deleted=message.is_deleted,
+        created_at=message.created_at
+    )
+
+
+@router.get("/messages/{user_id}")
+async def get_private_messages(
+    user_id: str,
+    limit: int = 50,
+    offset: int = 0,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Get private messages between current user and another user."""
+    result = await db.execute(
+        select(Message).where(
+            and_(
+                or_(
+                    and_(Message.sender_id == current_user["id"], Message.receiver_id == user_id),
+                    and_(Message.sender_id == user_id, Message.receiver_id == current_user["id"])
+                ),
+                Message.group_id.is_(None),
+                Message.is_deleted == False
+            )
+        ).order_by(Message.created_at.desc()).limit(limit).offset(offset)
+    )
+    messages = result.scalars().all()
+    
+    return [
+        {
+            "id": m.id,
+            "sender_id": m.sender_id,
+            "receiver_id": m.receiver_id,
+            "content": m.content,
+            "type": m.type,
+            "created_at": m.created_at.isoformat()
+        }
+        for m in messages
+    ]
+
+
+@router.delete("/messages/{message_id}")
+async def delete_message(
+    message_id: str,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Delete a message (soft delete)."""
+    result = await db.execute(
+        select(Message).where(
+            and_(Message.id == message_id, Message.sender_id == current_user["id"])
+        )
+    )
+    message = result.scalar_one_or_none()
+    
+    if not message:
+        raise HTTPException(status_code=404, detail="Message not found or not authorized")
+    
+    message.is_deleted = True
+    message.deleted_at = datetime.utcnow()
+    await db.commit()
+    
+    return {"message": "Message deleted"}
+
+
+@router.post("/reactions", response_model=ReactionResponse)
+async def add_reaction(
+    request: ReactionRequest,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Add an emoji reaction to a message."""
+    # Verify message exists
+    result = await db.execute(select(Message).where(Message.id == request.message_id))
+    message = result.scalar_one_or_none()
+    
+    if not message:
+        raise HTTPException(status_code=404, detail="Message not found")
+    
+    # Check if user already reacted with this emoji
+    result = await db.execute(
+        select(Reaction).where(
+            and_(
+                Reaction.message_id == request.message_id,
+                Reaction.user_id == current_user["id"],
+                Reaction.emoji == request.emoji
+            )
+        )
+    )
+    existing = result.scalar_one_or_none()
+    
+    if existing:
+        raise HTTPException(status_code=400, detail="Reaction already exists")
+    
+    # Create reaction
+    reaction = Reaction(
+        message_id=request.message_id,
+        user_id=current_user["id"],
+        emoji=request.emoji
+    )
+    
+    db.add(reaction)
+    await db.commit()
+    await db.refresh(reaction)
+    
+    return ReactionResponse(
+        id=reaction.id,
+        message_id=reaction.message_id,
+        user_id=reaction.user_id,
+        emoji=reaction.emoji,
+        created_at=reaction.created_at
+    )
+
+
+@router.delete("/reactions/{reaction_id}")
+async def remove_reaction(
+    reaction_id: str,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Remove a reaction."""
+    result = await db.execute(
+        select(Reaction).where(
+            and_(Reaction.id == reaction_id, Reaction.user_id == current_user["id"])
+        )
+    )
+    reaction = result.scalar_one_or_none()
+    
+    if not reaction:
+        raise HTTPException(status_code=404, detail="Reaction not found")
+    
+    await db.delete(reaction)
+    await db.commit()
+    
+    return {"message": "Reaction removed"}
+
+
+@router.get("/messages/{message_id}/reactions")
+async def get_message_reactions(
+    message_id: str,
+    db: AsyncSession = Depends(get_db)
+):
+    """Get all reactions on a message."""
+    result = await db.execute(
+        select(Reaction).where(Reaction.message_id == message_id)
+    )
+    reactions = result.scalars().all()
+    
+    # Group by emoji
+    emoji_counts = {}
+    for r in reactions:
+        if r.emoji not in emoji_counts:
+            emoji_counts[r.emoji] = {"count": 0, "users": []}
+        emoji_counts[r.emoji]["count"] += 1
+        emoji_counts[r.emoji]["users"].append(r.user_id)
+    
+    return emoji_counts

backend/app/routes/group.py

@@ -0,0 +1,399 @@
+"""
+Group routes - create groups, manage members, admin controls.
+"""
+from fastapi import APIRouter, Depends, HTTPException, status
+from pydantic import BaseModel
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, and_, or_
+from typing import Optional, List
+from datetime import datetime
+import uuid
+
+from ..dependencies import get_db, get_current_user
+from ..models import Group, GroupMember, User, Message
+
+router = APIRouter(prefix="/groups", tags=["Group"])
+
+
+# --- Schemas ---
+class CreateGroupRequest(BaseModel):
+    name: str
+    description: Optional[str] = None
+    is_private: bool = False
+    max_members: int = 100
+
+
+class UpdateGroupRequest(BaseModel):
+    name: Optional[str] = None
+    description: Optional[str] = None
+    avatar: Optional[str] = None
+
+
+class GroupResponse(BaseModel):
+    id: str
+    name: str
+    description: Optional[str]
+    avatar: Optional[str]
+    owner_id: str
+    is_private: bool
+    max_members: int
+    created_at: datetime
+
+
+class AddMemberRequest(BaseModel):
+    user_id: str
+    role: str = "member"  # member, co_admin
+
+
+class UpdateMemberRequest(BaseModel):
+    nickname: Optional[str] = None
+    role: Optional[str] = None
+    is_muted: Optional[bool] = None
+
+
+class MemberResponse(BaseModel):
+    id: str
+    user_id: str
+    group_id: str
+    role: str
+    nickname: Optional[str]
+    is_muted: bool
+    joined_at: datetime
+
+
+# --- Routes ---
+@router.post("/", response_model=GroupResponse, status_code=status.HTTP_201_CREATED)
+async def create_group(
+    request: CreateGroupRequest,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Create a new chat group."""
+    group = Group(
+        name=request.name,
+        description=request.description,
+        owner_id=current_user["id"],
+        is_private=request.is_private,
+        max_members=request.max_members
+    )
+    
+    db.add(group)
+    await db.commit()
+    await db.refresh(group)
+    
+    # Add creator as admin
+    member = GroupMember(
+        group_id=group.id,
+        user_id=current_user["id"],
+        role="admin"
+    )
+    db.add(member)
+    await db.commit()
+    
+    return GroupResponse(
+        id=group.id,
+        name=group.name,
+        description=group.description,
+        avatar=group.avatar,
+        owner_id=group.owner_id,
+        is_private=group.is_private,
+        max_members=group.max_members,
+        created_at=group.created_at
+    )
+
+
+@router.get("/{group_id}", response_model=GroupResponse)
+async def get_group(
+    group_id: str,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Get group details."""
+    result = await db.execute(select(Group).where(Group.id == group_id))
+    group = result.scalar_one_or_none()
+    
+    if not group:
+        raise HTTPException(status_code=404, detail="Group not found")
+    
+    return GroupResponse(
+        id=group.id,
+        name=group.name,
+        description=group.description,
+        avatar=group.avatar,
+        owner_id=group.owner_id,
+        is_private=group.is_private,
+        max_members=group.max_members,
+        created_at=group.created_at
+    )
+
+
+@router.put("/{group_id}", response_model=GroupResponse)
+async def update_group(
+    group_id: str,
+    request: UpdateGroupRequest,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Update group (admin only)."""
+    # Check if user is admin
+    result = await db.execute(
+        select(GroupMember).where(
+            and_(
+                GroupMember.group_id == group_id,
+                GroupMember.user_id == current_user["id"],
+                GroupMember.role.in_(["admin", "co_admin"])
+            )
+        )
+    )
+    member = result.scalar_one_or_none()
+    
+    if not member:
+        raise HTTPException(status_code=403, detail="Not authorized")
+    
+    result = await db.execute(select(Group).where(Group.id == group_id))
+    group = result.scalar_one_or_none()
+    
+    if request.name:
+        group.name = request.name
+    if request.description is not None:
+        group.description = request.description
+    if request.avatar is not None:
+        group.avatar = request.avatar
+    
+    await db.commit()
+    await db.refresh(group)
+    
+    return GroupResponse(
+        id=group.id,
+        name=group.name,
+        description=group.description,
+        avatar=group.avatar,
+        owner_id=group.owner_id,
+        is_private=group.is_private,
+        max_members=group.max_members,
+        created_at=group.created_at
+    )
+
+
+@router.delete("/{group_id}")
+async def delete_group(
+    group_id: str,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Delete group (owner only)."""
+    result = await db.execute(select(Group).where(Group.id == group_id))
+    group = result.scalar_one_or_none()
+    
+    if not group:
+        raise HTTPException(status_code=404, detail="Group not found")
+    
+    if group.owner_id != current_user["id"]:
+        raise HTTPException(status_code=403, detail="Only owner can delete group")
+    
+    await db.delete(group)
+    await db.commit()
+    
+    return {"message": "Group deleted"}
+
+
+@router.get("/{group_id}/members")
+async def get_group_members(
+    group_id: str,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Get all group members."""
+    result = await db.execute(
+        select(GroupMember, User).join(User, GroupMember.user_id == User.id).where(
+            GroupMember.group_id == group_id,
+            GroupMember.is_banned == False
+        )
+    )
+    members = result.all()
+    
+    return [
+        {
+            "id": m.GroupMember.id,
+            "user_id": m.GroupMember.user_id,
+            "username": m.User.username,
+            "display_name": m.User.display_name,
+            "avatar": m.User.avatar,
+            "role": m.GroupMember.role,
+            "nickname": m.GroupMember.nickname,
+            "is_muted": m.GroupMember.is_muted,
+            "joined_at": m.GroupMember.joined_at.isoformat()
+        }
+        for m in members
+    ]
+
+
+@router.post("/{group_id}/members", status_code=status.HTTP_201_CREATED)
+async def add_member(
+    group_id: str,
+    request: AddMemberRequest,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Add a member to group (admin only)."""
+    # Check admin
+    result = await db.execute(
+        select(GroupMember).where(
+            and_(
+                GroupMember.group_id == group_id,
+                GroupMember.user_id == current_user["id"],
+                GroupMember.role.in_(["admin", "co_admin"])
+            )
+        )
+    )
+    if not result.scalar_one_or_none():
+        raise HTTPException(status_code=403, detail="Not authorized")
+    
+    # Check group exists
+    result = await db.execute(select(Group).where(Group.id == group_id))
+    group = result.scalar_one_or_none()
+    if not group:
+        raise HTTPException(status_code=404, detail="Group not found")
+    
+    # Check not already member
+    result = await db.execute(
+        select(GroupMember).where(
+            and_(GroupMember.group_id == group_id, GroupMember.user_id == request.user_id)
+        )
+    )
+    if result.scalar_one_or_none():
+        raise HTTPException(status_code=400, detail="Already a member")
+    
+    # Add member
+    member = GroupMember(
+        group_id=group_id,
+        user_id=request.user_id,
+        role=request.role
+    )
+    db.add(member)
+    await db.commit()
+    
+    return {"message": "Member added"}
+
+
+@router.delete("/{group_id}/members/{user_id}")
+async def remove_member(
+    group_id: str,
+    user_id: str,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Remove a member from group (admin only)."""
+    # Check admin or self-removal
+    result = await db.execute(
+        select(GroupMember).where(
+            and_(
+                GroupMember.group_id == group_id,
+                GroupMember.user_id == current_user["id"],
+                GroupMember.role.in_(["admin", "co_admin"])
+            )
+        )
+    )
+    is_admin = result.scalar_one_or_none() is not None
+    
+    if not is_admin and user_id != current_user["id"]:
+        raise HTTPException(status_code=403, detail="Not authorized")
+    
+    # Find and remove member
+    result = await db.execute(
+        select(GroupMember).where(
+            and_(GroupMember.group_id == group_id, GroupMember.user_id == user_id)
+        )
+    )
+    member = result.scalar_one_or_none()
+    
+    if not member:
+        raise HTTPException(status_code=404, detail="Member not found")
+    
+    await db.delete(member)
+    await db.commit()
+    
+    return {"message": "Member removed"}
+
+
+@router.put("/{group_id}/members/{user_id}", response_model=MemberResponse)
+async def update_member(
+    group_id: str,
+    user_id: str,
+    request: UpdateMemberRequest,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Update member (nickname, role, mute) - admin only."""
+    # Check admin
+    result = await db.execute(
+        select(GroupMember).where(
+            and_(
+                GroupMember.group_id == group_id,
+                GroupMember.user_id == current_user["id"],
+                GroupMember.role == "admin"
+            )
+        )
+    )
+    if not result.scalar_one_or_none():
+        raise HTTPException(status_code=403, detail="Not authorized")
+    
+    result = await db.execute(
+        select(GroupMember).where(
+            and_(GroupMember.group_id == group_id, GroupMember.user_id == user_id)
+        )
+    )
+    member = result.scalar_one_or_none()
+    
+    if not member:
+        raise HTTPException(status_code=404, detail="Member not found")
+    
+    if request.nickname is not None:
+        member.nickname = request.nickname
+    if request.role is not None:
+        member.role = request.role
+    if request.is_muted is not None:
+        member.is_muted = request.is_muted
+    
+    await db.commit()
+    
+    return MemberResponse(
+        id=member.id,
+        user_id=member.user_id,
+        group_id=member.group_id,
+        role=member.role,
+        nickname=member.nickname,
+        is_muted=member.is_muted,
+        joined_at=member.joined_at
+    )
+
+
+@router.get("/{group_id}/messages")
+async def get_group_messages(
+    group_id: str,
+    limit: int = 50,
+    offset: int = 0,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Get group messages."""
+    result = await db.execute(
+        select(Message).where(
+            and_(
+                Message.group_id == group_id,
+                Message.is_deleted == False
+            )
+        ).order_by(Message.created_at.desc()).limit(limit).offset(offset)
+    )
+    messages = result.scalars().all()
+    
+    return [
+        {
+            "id": m.id,
+            "sender_id": m.sender_id,
+            "content": m.content,
+            "type": m.type,
+            "created_at": m.created_at.isoformat()
+        }
+        for m in messages
+    ]

backend/app/routes/system.py

@@ -0,0 +1,119 @@
+"""
+System routes - batch triggers, queue control, monitoring.
+"""
+from fastapi import APIRouter, Depends, HTTPException, status
+from pydantic import BaseModel
+from typing import Optional, List, Dict
+from datetime import datetime, timedelta
+
+from ..dependencies import get_current_user
+
+router = APIRouter(prefix="/system", tags=["System"])
+
+
+# --- Schemas ---
+class BatchStatusResponse(BaseModel):
+    last_batch_time: Optional[datetime]
+    next_batch_time: Optional[datetime]
+    queue_size: int
+    processed_today: int
+
+
+class TriggerBatchRequest(BaseModel):
+    user_id: Optional[str] = None
+    group_id: Optional[str] = None
+
+
+class QueueStatsResponse(BaseModel):
+    size: int
+    oldest_message_age_seconds: int
+    avg_processing_time_ms: int
+
+
+# --- Routes ---
+@router.get("/health")
+async def health_check():
+    """Health check endpoint."""
+    return {
+        "status": "healthy",
+        "timestamp": datetime.utcnow().isoformat(),
+        "version": "1.0.0"
+    }
+
+
+@router.get("/batch/status", response_model=BatchStatusResponse)
+async def get_batch_status(current_user: dict = Depends(get_current_user)):
+    """Get batch processing status."""
+    # TODO: Pull actual stats from Redis/monitor
+    return BatchStatusResponse(
+        last_batch_time=datetime.utcnow() - timedelta(hours=1),
+        next_batch_time=datetime.utcnow() + timedelta(minutes=30),
+        queue_size=42,
+        processed_today=156
+    )
+
+
+@router.post("/batch/trigger", status_code=status.HTTP_202_ACCEPTED)
+async def trigger_batch(
+    request: TriggerBatchRequest,
+    current_user: dict = Depends(get_current_user)
+):
+    """
+    Manually trigger batch processing.
+    Can target specific user or group, or process all.
+    """
+    # TODO: Call controller to trigger batch
+    target = request.user_id or request.group_id or "all"
+    return {
+        "message": f"Batch processing triggered for {target}",
+        "status": "queued"
+    }
+
+
+@router.get("/queue/stats", response_model=QueueStatsResponse)
+async def get_queue_stats():
+    """Get queue statistics."""
+    # TODO: Pull from Redis
+    return QueueStatsResponse(
+        size=42,
+        oldest_message_age_seconds=300,
+        avg_processing_time_ms=15
+    )
+
+
+@router.get("/storage/stats")
+async def get_storage_stats():
+    """Get storage statistics."""
+    # TODO: Calculate from storage directory
+    return {
+        "total_files": 1250,
+        "total_size_mb": 45.2,
+        "compression_ratio": 68.5
+    }
+
+
+@router.post("/cleanup/old-messages")
+async def cleanup_old_messages(
+    days: int = 30,
+    current_user: dict = Depends(get_current_user)
+):
+    """Clean up messages older than specified days."""
+    # TODO: Implement cleanup
+    return {
+        "message": f"Cleanup scheduled for messages older than {days} days"
+    }
+
+
+@router.get("/logs")
+async def get_logs(
+    limit: int = 100,
+    level: Optional[str] = None
+):
+    """Get recent system logs."""
+    # TODO: Pull from log system
+    return {
+        "logs": [
+            {"timestamp": datetime.utcnow().isoformat(), "level": "INFO", "message": "System running"}
+        ],
+        "count": 1
+    }

backend/app/routes/user.py

@@ -0,0 +1,183 @@
+"""
+User routes - profile, avatar, block user.
+"""
+from fastapi import APIRouter, Depends, HTTPException, status
+from pydantic import BaseModel, EmailStr
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, or_
+from typing import Optional, List
+from datetime import datetime
+
+from ..dependencies import get_db, get_current_user
+from ..models import User, Friend
+from ..utils.avatar import generate_avatar
+
+router = APIRouter(prefix="/users", tags=["User"])
+
+
+# --- Schemas ---
+class UserProfileResponse(BaseModel):
+    id: str
+    username: str
+    email: str
+    display_name: Optional[str] = None
+    avatar: Optional[str] = None
+    bio: Optional[str] = None
+    status: str
+    is_public: bool = True
+    created_at: datetime
+
+
+class UpdateProfileRequest(BaseModel):
+    display_name: Optional[str] = None
+    bio: Optional[str] = None
+    is_public: Optional[bool] = None
+    allow_friends: Optional[bool] = None
+
+
+class BlockUserRequest(BaseModel):
+    user_id: str
+
+
+class FriendRequest(BaseModel):
+    user_id: str
+
+
+# --- Routes ---
+@router.get("/{user_id}", response_model=UserProfileResponse)
+async def get_user_profile(
+    user_id: str,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Get a user's public profile."""
+    result = await db.execute(select(User).where(User.id == user_id))
+    user = result.scalar_one_or_none()
+    
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found")
+    
+    # Check if user allows public profile
+    if not user.is_public and user.id != current_user["id"]:
+        raise HTTPException(status_code=403, detail="This profile is private")
+    
+    return UserProfileResponse(
+        id=user.id,
+        username=user.username,
+        email=user.email,
+        display_name=user.display_name,
+        avatar=user.avatar,
+        bio=user.bio,
+        status=user.status,
+        is_public=user.is_public,
+        created_at=user.created_at
+    )
+
+
+@router.put("/profile", response_model=UserProfileResponse)
+async def update_profile(
+    request: UpdateProfileRequest,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Update current user's profile."""
+    result = await db.execute(select(User).where(User.id == current_user["id"]))
+    user = result.scalar_one_or_none()
+    
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found")
+    
+    # Update fields
+    if request.display_name is not None:
+        user.display_name = request.display_name
+    if request.bio is not None:
+        user.bio = request.bio
+    if request.is_public is not None:
+        user.is_public = request.is_public
+    if request.allow_friends is not None:
+        user.allow_friends = request.allow_friends
+    
+    await db.commit()
+    await db.refresh(user)
+    
+    return UserProfileResponse(
+        id=user.id,
+        username=user.username,
+        email=user.email,
+        display_name=user.display_name,
+        avatar=user.avatar,
+        bio=user.bio,
+        status=user.status,
+        is_public=user.is_public,
+        created_at=user.created_at
+    )
+
+
+@router.put("/avatar")
+async def update_avatar(
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Update avatar (auto-generate or custom URL)."""
+    result = await db.execute(select(User).where(User.id == current_user["id"]))
+    user = result.scalar_one_or_none()
+    
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found")
+    
+    # Generate new avatar based on username
+    user.avatar = await generate_avatar(user.id, user.username)
+    await db.commit()
+    
+    return {"avatar": user.avatar}
+
+
+@router.post("/block")
+async def block_user(
+    request: BlockUserRequest,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Block a user (add to blocked list)."""
+    # In a real implementation, this would create a BlockedUser record
+    # For now, return success
+    return {"message": f"User {request.user_id} blocked"}
+
+
+@router.post("/unblock")
+async def unblock_user(
+    request: BlockUserRequest,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Unblock a user."""
+    return {"message": f"User {request.user_id} unblocked"}
+
+
+@router.get("/search/{query}")
+async def search_users(
+    query: str,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
+    """Search users by username or display name."""
+    result = await db.execute(
+        select(User).where(
+            or_(
+                User.username.ilike(f"%{query}%"),
+                User.display_name.ilike(f"%{query}%")
+            )
+        ).limit(20)
+    )
+    users = result.scalars().all()
+    
+    return [
+        {
+            "id": u.id,
+            "username": u.username,
+            "display_name": u.display_name,
+            "avatar": u.avatar,
+            "status": u.status
+        }
+        for u in users
+    ]

backend/app/services/__init__.py

@@ -0,0 +1,10 @@
+"""
+Services package - business logic layer.
+"""
+from .auth_service import AuthService
+from .chat_service import ChatService
+from .group_service import GroupService
+from .user_service import UserService
+from .file_service import FileService
+
+__all__ = ["AuthService", "ChatService", "GroupService", "UserService", "FileService"]

backend/app/services/auth_service.py

@@ -0,0 +1,70 @@
+"""
+Auth service - authentication business logic.
+"""
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select
+from typing import Optional
+
+from ..models import User
+from ..utils.security import hash_password, verify_password, create_access_token
+from ..utils.avatar import generate_avatar
+
+
+class AuthService:
+    """Authentication business logic."""
+
+    @staticmethod
+    async def register_user(db: AsyncSession, username: str, email: str, password: str) -> User:
+        """Register a new user."""
+        # Check existing
+        result = await db.execute(select(User).where(User.email == email))
+        if result.scalar_one_or_none():
+            raise ValueError("Email already registered")
+        
+        result = await db.execute(select(User).where(User.username == username))
+        if result.scalar_one_or_none():
+            raise ValueError("Username taken")
+        
+        # Create
+        user = User(
+            username=username,
+            email=email,
+            password_hash=hash_password(password),
+            display_name=username,
+            avatar=await generate_avatar(username, username),
+            status="offline"
+        )
+        db.add(user)
+        await db.commit()
+        await db.refresh(user)
+        return user
+
+    @staticmethod
+    async def login_user(db: AsyncSession, email: str, password: str) -> Optional[User]:
+        """Login user by email/password."""
+        result = await db.execute(select(User).where(User.email == email))
+        user = result.scalar_one_or_none()
+        
+        if not user or not verify_password(password, user.password_hash):
+            return None
+        
+        user.status = "online"
+        await db.commit()
+        return user
+
+    @staticmethod
+    def generate_token(user: User) -> str:
+        """Generate JWT token for user."""
+        return create_access_token({"sub": user.id, "username": user.username})
+
+    @staticmethod
+    async def logout_user(db: AsyncSession, user_id: str) -> bool:
+        """Logout user."""
+        result = await db.execute(select(User).where(User.id == user_id))
+        user = result.scalar_one_or_none()
+        
+        if user:
+            user.status = "offline"
+            await db.commit()
+            return True
+        return False

backend/app/services/chat_service.py

@@ -0,0 +1,115 @@
+"""
+Chat service - message handling and queue push.
+"""
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, and_, or_
+from typing import Optional, List
+from datetime import datetime
+
+from ..models import Message, User
+from ..utils.formatter import clean_message
+
+
+class ChatService:
+    """Chat business logic."""
+
+    @staticmethod
+    async def send_message(
+        db: AsyncSession,
+        sender_id: str,
+        content: str,
+        receiver_id: Optional[str] = None,
+        group_id: Optional[str] = None,
+        message_type: str = "text"
+    ) -> Message:
+        """Send a message (queued for batch processing)."""
+        # Clean content
+        content = clean_message(content)
+        
+        # Create message (initially uncompressed)
+        message = Message(
+            sender_id=sender_id,
+            receiver_id=receiver_id,
+            group_id=group_id,
+            content=content,
+            type=message_type,
+            compressed=False
+        )
+        
+        db.add(message)
+        await db.commit()
+        await db.refresh(message)
+        
+        # TODO: Push to async queue for batch processing
+        # from ..system.queue import message_queue
+        # await message_queue.push(message)
+        
+        return message
+
+    @staticmethod
+    async def get_private_messages(
+        db: AsyncSession,
+        user_id: str,
+        other_user_id: str,
+        limit: int = 50,
+        offset: int = 0
+    ) -> List[Message]:
+        """Get private messages between two users."""
+        result = await db.execute(
+            select(Message).where(
+                and_(
+                    or_(
+                        and_(Message.sender_id == user_id, Message.receiver_id == other_user_id),
+                        and_(Message.sender_id == other_user_id, Message.receiver_id == user_id)
+                    ),
+                    Message.group_id.is_(None),
+                    Message.is_deleted == False
+                )
+            ).order_by(Message.created_at.desc()).limit(limit).offset(offset)
+        )
+        return result.scalars().all()
+
+    @staticmethod
+    async def delete_message(db: AsyncSession, message_id: str, user_id: str) -> bool:
+        """Delete a message (soft delete)."""
+        result = await db.execute(
+            select(Message).where(
+                and_(Message.id == message_id, Message.sender_id == user_id)
+            )
+        )
+        message = result.scalar_one_or_none()
+        
+        if not message:
+            return False
+        
+        message.is_deleted = True
+        message.deleted_at = datetime.utcnow()
+        await db.commit()
+        return True
+
+    @staticmethod
+    async def get_conversations(db: AsyncSession, user_id: str) -> List[dict]:
+        """Get list of conversations for a user."""
+        # Get recent private conversations
+        result = await db.execute(
+            select(Message).where(
+                or_(
+                    Message.sender_id == user_id,
+                    Message.receiver_id == user_id
+                )
+            ).order_by(Message.created_at.desc()).limit(100)
+        )
+        messages = result.scalars().all()
+        
+        # Group by conversation partner
+        conversations = {}
+        for msg in messages:
+            partner_id = msg.receiver_id if msg.sender_id == user_id else msg.sender_id
+            if partner_id and partner_id not in conversations:
+                conversations[partner_id] = {
+                    "user_id": partner_id,
+                    "last_message": msg.content[:50],
+                    "last_time": msg.created_at
+                }
+        
+        return list(conversations.values())[:20]

backend/app/services/file_service.py

@@ -0,0 +1,137 @@
+"""
+File service - GitHub upload logic.
+"""
+from typing import Optional
+from pathlib import Path
+import asyncio
+from github import Github
+from ..config import settings
+
+
+class FileService:
+    """File storage and GitHub upload logic."""
+
+    def __init__(self):
+        self.github = None
+        if settings.GITHUB_TOKEN:
+            self.github = Github(settings.GITHUB_TOKEN)
+
+    async def upload_to_github(
+        self,
+        file_path: Path,
+        repo_path: str,
+        branch: str = "main"
+    ) -> Optional[str]:
+        """
+        Upload a file to GitHub repository.
+        
+        Args:
+            file_path: Local file path to upload
+            repo_path: GitHub repo (e.g., "username/repo")
+            branch: Target branch
+        
+        Returns:
+            GitHub file URL or None on failure
+        """
+        if not self.github:
+            print("GitHub token not configured")
+            return None
+
+        try:
+            repo = self.github.get_repo(repo_path)
+            
+            with open(file_path, "rb") as f:
+                content = f.read()
+            
+            # Get file path in repo
+            repo_path_str = str(file_path).replace("./", "")
+            
+            # Check if file exists
+            try:
+                existing_file = repo.get_contents(repo_path_str, ref=branch)
+                # Update existing file
+                repo.update_file(
+                    path=repo_path_str,
+                    message=f"Update {file_path.name}",
+                    content=content,
+                    sha=existing_file.sha,
+                    branch=branch
+                )
+            except Exception:
+                # Create new file
+                repo.create_file(
+                    path=repo_path_str,
+                    message=f"Upload {file_path.name}",
+                    content=content,
+                    branch=branch
+                )
+            
+            return f"https://github.com/{repo_path}/blob/{branch}/{repo_path_str}"
+        
+        except Exception as e:
+            print(f"GitHub upload error: {e}")
+            return None
+
+    async def download_from_github(
+        self,
+        repo_path: str,
+        file_path: str,
+        branch: str = "main"
+    ) -> Optional[bytes]:
+        """Download a file from GitHub."""
+        if not self.github:
+            return None
+
+        try:
+            repo = self.github.get_repo(repo_path)
+            contents = repo.get_contents(file_path, ref=branch)
+            return contents.decoded_content
+        except Exception as e:
+            print(f"GitHub download error: {e}")
+            return None
+
+    def get_local_storage_path(self, user_id: str, project_id: str, date_path: str) -> Path:
+        """Get local storage path for a user's project data."""
+        base = Path(settings.STORAGE_PATH) / "users" / user_id / "projects" / project_id / date_path
+        base.mkdir(parents=True, exist_ok=True)
+        return base
+
+    async def save_batch_file(
+        self,
+        user_id: str,
+        project_id: str,
+        filename: str,
+        content: bytes
+    ) -> Path:
+        """Save a batch file to local storage."""
+        from ..utils.time import get_date_path
+        date_path = get_date_path()
+        storage_dir = self.get_local_storage_path(user_id, project_id, date_path)
+        
+        file_path = storage_dir / filename
+        with open(file_path, "wb") as f:
+            f.write(content)
+        
+        return file_path
+
+    async def upload_batch(
+        self,
+        user_id: str,
+        project_id: str,
+        batch_dir: Path
+    ) -> bool:
+        """Upload all files in a batch directory to GitHub."""
+        if not settings.GITHUB_REPO:
+            return False
+        
+        success = True
+        for file_path in batch_dir.glob("*"):
+            if file_path.is_file():
+                result = await self.upload_to_github(
+                    file_path,
+                    settings.GITHUB_REPO
+                )
+                if not result:
+                    success = False
+        
+        return success

backend/app/services/group_service.py

@@ -0,0 +1,143 @@
+"""
+Group service - group management business logic.
+"""
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, and_
+from typing import Optional, List
+
+from ..models import Group, GroupMember, User
+
+
+class GroupService:
+    """Group business logic."""
+
+    @staticmethod
+    async def create_group(
+        db: AsyncSession,
+        name: str,
+        owner_id: str,
+        description: Optional[str] = None,
+        is_private: bool = False
+    ) -> Group:
+        """Create a new group."""
+        group = Group(
+            name=name,
+            owner_id=owner_id,
+            description=description,
+            is_private=is_private
+        )
+        db.add(group)
+        await db.commit()
+        await db.refresh(group)
+        
+        # Add owner as admin
+        member = GroupMember(
+            group_id=group.id,
+            user_id=owner_id,
+            role="admin"
+        )
+        db.add(member)
+        await db.commit()
+        
+        return group
+
+    @staticmethod
+    async def get_group(db: AsyncSession, group_id: str) -> Optional[Group]:
+        """Get group by ID."""
+        result = await db.execute(select(Group).where(Group.id == group_id))
+        return result.scalar_one_or_none()
+
+    @staticmethod
+    async def add_member(
+        db: AsyncSession,
+        group_id: str,
+        user_id: str,
+        role: str = "member"
+    ) -> GroupMember:
+        """Add a member to group."""
+        # Check not already member
+        result = await db.execute(
+            select(GroupMember).where(
+                and_(GroupMember.group_id == group_id, GroupMember.user_id == user_id)
+            )
+        )
+        if result.scalar_one_or_none():
+            raise ValueError("Already a member")
+        
+        member = GroupMember(
+            group_id=group_id,
+            user_id=user_id,
+            role=role
+        )
+        db.add(member)
+        await db.commit()
+        await db.refresh(member)
+        return member
+
+    @staticmethod
+    async def remove_member(db: AsyncSession, group_id: str, user_id: str) -> bool:
+        """Remove a member from group."""
+        result = await db.execute(
+            select(GroupMember).where(
+                and_(GroupMember.group_id == group_id, GroupMember.user_id == user_id)
+            )
+        )
+        member = result.scalar_one_or_none()
+        
+        if not member:
+            return False
+        
+        await db.delete(member)
+        await db.commit()
+        return True
+
+    @staticmethod
+    async def get_members(db: AsyncSession, group_id: str) -> List[GroupMember]:
+        """Get all group members."""
+        result = await db.execute(
+            select(GroupMember).where(GroupMember.group_id == group_id)
+        )
+        return result.scalars().all()
+
+    @staticmethod
+    async def is_admin(db: AsyncSession, group_id: str, user_id: str) -> bool:
+        """Check if user is admin or co_admin."""
+        result = await db.execute(
+            select(GroupMember).where(
+                and_(
+                    GroupMember.group_id == group_id,
+                    GroupMember.user_id == user_id,
+                    GroupMember.role.in_(["admin", "co_admin"])
+                )
+            )
+        )
+        return result.scalar_one_or_none() is not None
+
+    @staticmethod
+    async def is_owner(db: AsyncSession, group_id: str, user_id: str) -> bool:
+        """Check if user is owner."""
+        result = await db.execute(select(Group).where(Group.id == group_id))
+        group = result.scalar_one_or_none()
+        return group and group.owner_id == user_id
+
+    @staticmethod
+    async def update_member_role(
+        db: AsyncSession,
+        group_id: str,
+        user_id: str,
+        new_role: str
+    ) -> bool:
+        """Update member role (admin only)."""
+        result = await db.execute(
+            select(GroupMember).where(
+                and_(GroupMember.group_id == group_id, GroupMember.user_id == user_id)
+            )
+        )
+        member = result.scalar_one_or_none()
+        
+        if not member:
+            return False
+        
+        member.role = new_role
+        await db.commit()
+        return True

backend/app/services/user_service.py

@@ -0,0 +1,148 @@
+"""
+User service - user profile management.
+"""
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, and_, or_
+from typing import Optional, List
+
+from ..models import User, Friend
+from ..utils.avatar import generate_avatar
+
+
+class UserService:
+    """User business logic."""
+
+    @staticmethod
+    async def get_user(db: AsyncSession, user_id: str) -> Optional[User]:
+        """Get user by ID."""
+        result = await db.execute(select(User).where(User.id == user_id))
+        return result.scalar_one_or_none()
+
+    @staticmethod
+    async def get_by_email(db: AsyncSession, email: str) -> Optional[User]:
+        """Get user by email."""
+        result = await db.execute(select(User).where(User.email == email))
+        return result.scalar_one_or_none()
+
+    @staticmethod
+    async def get_by_username(db: AsyncSession, username: str) -> Optional[User]:
+        """Get user by username."""
+        result = await db.execute(select(User).where(User.username == username))
+        return result.scalar_one_or_none()
+
+    @staticmethod
+    async def update_profile(
+        db: AsyncSession,
+        user_id: str,
+        display_name: Optional[str] = None,
+        bio: Optional[str] = None,
+        is_public: Optional[bool] = None
+    ) -> Optional[User]:
+        """Update user profile."""
+        result = await db.execute(select(User).where(User.id == user_id))
+        user = result.scalar_one_or_none()
+        
+        if not user:
+            return None
+        
+        if display_name is not None:
+            user.display_name = display_name
+        if bio is not None:
+            user.bio = bio
+        if is_public is not None:
+            user.is_public = is_public
+        
+        await db.commit()
+        await db.refresh(user)
+        return user
+
+    @staticmethod
+    async def update_avatar(db: AsyncSession, user_id: str) -> Optional[str]:
+        """Generate and update avatar."""
+        result = await db.execute(select(User).where(User.id == user_id))
+        user = result.scalar_one_or_none()
+        
+        if not user:
+            return None
+        
+        user.avatar = await generate_avatar(user.id, user.username)
+        await db.commit()
+        return user.avatar
+
+    @staticmethod
+    async def search_users(db: AsyncSession, query: str, limit: int = 20) -> List[User]:
+        """Search users by username or display name."""
+        result = await db.execute(
+            select(User).where(
+                or_(
+                    User.username.ilike(f"%{query}%"),
+                    User.display_name.ilike(f"%{query}%")
+                )
+            ).limit(limit)
+        )
+        return result.scalars().all()
+
+    @staticmethod
+    async def add_friend(db: AsyncSession, user_id: str, friend_id: str) -> Friend:
+        """Send friend request."""
+        # Check not already friends
+        result = await db.execute(
+            select(Friend).where(
+                or_(
+                    and_(Friend.user_id == user_id, Friend.friend_id == friend_id),
+                    and_(Friend.user_id == friend_id, Friend.friend_id == user_id)
+                )
+            )
+        )
+        if result.scalar_one_or_none():
+            raise ValueError("Already friends or request pending")
+        
+        friend = Friend(
+            user_id=user_id,
+            friend_id=friend_id,
+            status="pending"
+        )
+        db.add(friend)
+        await db.commit()
+        await db.refresh(friend)
+        return friend
+
+    @staticmethod
+    async def accept_friend(db: AsyncSession, user_id: str, friend_id: str) -> bool:
+        """Accept friend request."""
+        result = await db.execute(
+            select(Friend).where(
+                and_(Friend.user_id == friend_id, Friend.friend_id == user_id, Friend.status == "pending")
+            )
+        )
+        friend = result.scalar_one_or_none()
+        
+        if not friend:
+            return False
+        
+        friend.status = "accepted"
+        await db.commit()
+        return True
+
+    @staticmethod
+    async def get_friends(db: AsyncSession, user_id: str) -> List[User]:
+        """Get user's friends."""
+        result = await db.execute(
+            select(Friend).where(
+                or_(
+                    and_(Friend.user_id == user_id, Friend.status == "accepted"),
+                    and_(Friend.friend_id == user_id, Friend.status == "accepted")
+                )
+            )
+        )
+        friends = result.scalars().all()
+        
+        friend_ids = []
+        for f in friends:
+            if f.user_id == user_id:
+                friend_ids.append(f.friend_id)
+            else:
+                friend_ids.append(f.user_id)
+        
+        result = await db.execute(select(User).where(User.id.in_(friend_ids)))
+        return result.scalars().all()

backend/app/sockets/ws.py

@@ -0,0 +1,89 @@
+"""
+WebSocket handler - real-time notifications (optional).
+"""
+from fastapi import WebSocket, WebSocketDisconnect
+from typing import List, Dict
+import json
+import asyncio
+
+
+class ConnectionManager:
+    """Manage WebSocket connections."""
+    
+    def __init__(self):
+        # user_id -> WebSocket
+        self.active_connections: Dict[str, WebSocket] = {}
+    
+    async def connect(self, user_id: str, websocket: WebSocket):
+        """Accept and store a WebSocket connection."""
+        await websocket.accept()
+        self.active_connections[user_id] = websocket
+    
+    def disconnect(self, user_id: str):
+        """Remove a connection."""
+        if user_id in self.active_connections:
+            del self.active_connections[user_id]
+    
+    async def send_personal(self, user_id: str, message: dict):
+        """Send message to specific user."""
+        if user_id in self.active_connections:
+            try:
+                await self.active_connections[user_id].send_text(json.dumps(message))
+            except Exception:
+                self.disconnect(user_id)
+    
+    async def broadcast(self, message: dict):
+        """Broadcast to all connected users."""
+        for ws in list(self.active_connections.values()):
+            try:
+                await ws.send_text(json.dumps(message))
+            except Exception:
+                pass
+
+
+manager = ConnectionManager()
+
+
+async def websocket_endpoint(websocket: WebSocket, user_id: str):
+    """
+    WebSocket endpoint for real-time notifications.
+    
+    Events:
+    - new_message: New message received
+    - message_deleted: Message was deleted
+    - friend_request: New friend request
+    - group_invite: Group invitation
+    """
+    await manager.connect(user_id, websocket)
+    try:
+        while True:
+            # Keep connection alive, listen for client messages
+            data = await websocket.receive_text()
+            # Could handle client acknowledgments here
+            pass
+    except WebSocketDisconnect:
+        manager.disconnect(user_id)
+
+
+async def notify_new_message(receiver_id: str, message: dict):
+    """Notify user of new message."""
+    await manager.send_personal(receiver_id, {
+        "type": "new_message",
+        "data": message
+    })
+
+
+async def notify_friend_request(user_id: str, request: dict):
+    """Notify user of friend request."""
+    await manager.send_personal(user_id, {
+        "type": "friend_request",
+        "data": request
+    })
+
+
+async def notify_group_invite(user_id: str, invite: dict):
+    """Notify user of group invite."""
+    await manager.send_personal(user_id, {
+        "type": "group_invite",
+        "data": invite
+    })

backend/app/system/__init__.py

@@ -0,0 +1,18 @@
+"""
+System package - async batch processing core.
+"""
+from .controller import SystemController, controller
+from .queue import MessageQueue, message_queue
+from .worker import Worker, worker
+from .batch import BatchProcessor, batch_processor
+from .scheduler import BatchScheduler, batch_scheduler
+from .monitor import SystemMonitor, system_monitor
+
+__all__ = [
+    "SystemController", "controller",
+    "MessageQueue", "message_queue",
+    "Worker", "worker",
+    "BatchProcessor", "batch_processor",
+    "BatchScheduler", "batch_scheduler",
+    "SystemMonitor", "system_monitor"
+]

backend/app/system/batch.py

@@ -0,0 +1,142 @@
+"""
+Batch builder - groups messages and compresses them.
+"""
+import json
+from typing import List, Optional
+from pathlib import Path
+from datetime import datetime
+import uuid
+
+from ..config import settings
+from ..utils.compression import compress_zstd
+from ..utils.time import get_date_path, utc_timestamp
+from ..services.file_service import FileService
+
+
+class BatchProcessor:
+    """
+    Batch processor - groups messages by user/project/time and compresses.
+    """
+    
+    def __init__(self, max_size_mb: int = 10):
+        self._max_size_bytes = max_size_mb * 1024 * 1024
+        self._current_batch: List[dict] = []
+        self._current_size = 0
+        self._batch_id = str(uuid.uuid4())[:8]
+        self._file_service = FileService()
+    
+    async def add_message(self, message: dict) -> bool:
+        """
+        Add a message to the current batch.
+        
+        Args:
+            message: Cleaned message dict
+            
+        Returns:
+            True if added, False if batch is full
+        """
+        msg_size = len(json.dumps(message).encode())
+        
+        # Check if adding would exceed limit
+        if self._current_size + msg_size > self._max_size_bytes:
+            # Compress current batch
+            await self.compress_and_save()
+            self._start_new_batch()
+        
+        self._current_batch.append(message)
+        self._current_size += msg_size
+        return True
+    
+    def is_ready(self) -> bool:
+        """Check if batch is ready for compression."""
+        # Ready if has messages and reaches size threshold (1MB)
+        return len(self._current_batch) > 0 and self._current_size > 1024 * 1024
+    
+    async def compress_and_save(self) -> Optional[Path]:
+        """
+        Compress current batch and save to storage.
+        
+        Returns:
+            Path to saved file or None on failure
+        """
+        if not self._current_batch:
+            return None
+        
+        try:
+            # Prepare batch data
+            batch_data = {
+                "batch_id": self._batch_id,
+                "created_at": datetime.utcnow().isoformat(),
+                "message_count": len(self._current_batch),
+                "messages": self._current_batch
+            }
+            
+            # Serialize and compress
+            json_data = json.dumps(batch_data, ensure_ascii=False)
+            compressed = compress_zstd(json_data.encode())
+            
+            # Generate filename
+            timestamp = utc_timestamp()
+            filename = f"{timestamp}_{self._batch_id}.zst"
+            
+            # Save to local storage
+            # In production, use user_id from messages
+            user_id = self._current_batch[0].get("sender_id", "default")
+            project_id = "chat_messages"
+            date_path = get_date_path()
+            
+            storage_path = self._file_service.get_local_storage_path(user_id, project_id, date_path)
+            file_path = storage_path / filename
+            
+            with open(file_path, "wb") as f:
+                f.write(compressed)
+            
+            # Create index file
+            await self._save_index(storage_path, filename, batch_data)
+            
+            # Optionally upload to GitHub
+            if settings.GITHUB_REPO:
+                await self._file_service.upload_to_github(file_path, settings.GITHUB_REPO)
+            
+            self._start_new_batch()
+            return file_path
+        
+        except Exception as e:
+            print(f"Batch save error: {e}")
+            return None
+    
+    async def _save_index(self, storage_path: Path, filename: str, batch_data: dict):
+        """Save or update index.json for the directory."""
+        index_path = storage_path / "index.json"
+        
+        index_data = []
+        if index_path.exists():
+            with open(index_path, "r") as f:
+                index_data = json.load(f)
+        
+        index_data.append({
+            "filename": filename,
+            "batch_id": self._batch_id,
+            "created_at": batch_data["created_at"],
+            "message_count": batch_data["message_count"],
+            "compressed_size": len(compress_zstd(json.dumps(batch_data).encode()))
+        })
+        
+        with open(index_path, "w") as f:
+            json.dump(index_data, f, indent=2)
+    
+    def _start_new_batch(self):
+        """Start a new batch."""
+        self._current_batch = []
+        self._current_size = 0
+        self._batch_id = str(uuid.uuid4())[:8]
+    
+    async def force_compress(self) -> Optional[Path]:
+        """Force compress current batch even if not full."""
+        if self._current_batch:
+            return await self.compress_and_save()
+        return None
+
+
+# Global batch processor
+batch_processor = BatchProcessor()

backend/app/system/controller.py

@@ -0,0 +1,78 @@
+"""
+System controller - main entry point for async batch processing.
+Receives messages and routes to appropriate handlers.
+"""
+from typing import Optional
+from ..models import Message
+from .queue import MessageQueue
+from .batch import BatchProcessor
+from .scheduler import BatchScheduler
+from .monitor import SystemMonitor
+
+
+class SystemController:
+    """
+    Main controller for async message processing.
+    Coordinates queue, batch, scheduler, and monitoring.
+    """
+    
+    def __init__(self):
+        self.queue = MessageQueue()
+        self.batch_processor = BatchProcessor()
+        self.scheduler = BatchScheduler()
+        self.monitor = SystemMonitor()
+    
+    async def receive_message(self, message: Message) -> bool:
+        """
+        Receive a message and add it to the processing queue.
+        
+        Args:
+            message: Message to process
+            
+        Returns:
+            True if queued successfully
+        """
+        try:
+            await self.queue.enqueue(message)
+            self.monitor.increment_queue_size()
+            return True
+        except Exception as e:
+            self.monitor.log_error(f"Failed to queue message: {e}")
+            return False
+    
+    async def trigger_batch(
+        self,
+        user_id: Optional[str] = None,
+        group_id: Optional[str] = None
+    ) -> dict:
+        """
+        Manually trigger batch processing.
+        
+        Args:
+            user_id: Optional specific user to process
+            group_id: Optional specific group to process
+            
+        Returns:
+            Status dict with processing results
+        """
+        return await self.batch_processor.process_batch(user_id, group_id)
+    
+    async def get_status(self) -> dict:
+        """Get system status."""
+        return {
+            "queue_size": self.queue.size(),
+            "monitor": self.monitor.get_stats(),
+            "scheduler_next": self.scheduler.get_next_run()
+        }
+    
+    async def start_scheduler(self):
+        """Start the batch scheduler."""
+        await self.scheduler.start()
+    
+    async def stop_scheduler(self):
+        """Stop the batch scheduler."""
+        await self.scheduler.stop()
+
+
+# Global controller instance
+controller = SystemController()

backend/app/system/monitor.py

@@ -0,0 +1,74 @@
+"""
+Monitor - logs, queue size, errors tracking.
+"""
+from typing import Dict, List
+from datetime import datetime
+from collections import deque
+
+
+class SystemMonitor:
+    """
+    System monitor - tracks metrics, logs, and errors.
+    """
+    
+    def __init__(self, max_log_entries: int = 1000):
+        self._queue_size = 0
+        self._processed_count = 0
+        self._batch_count = 0
+        self._error_count = 0
+        self._errors: deque = deque(maxlen=100)
+        self._max_log = max_log_entries
+    
+    def increment_queue_size(self, delta: int = 1):
+        """Increment queue size counter."""
+        self._queue_size += delta
+    
+    def decrement_queue_size(self, delta: int = 1):
+        """Decrement queue size counter."""
+        self._queue_size = max(0, self._queue_size - delta)
+    
+    def increment_processed(self, delta: int = 1):
+        """Increment processed messages counter."""
+        self._processed_count += delta
+    
+    def increment_batches(self, delta: int = 1):
+        """Increment batch count."""
+        self._batch_count += delta
+    
+    def increment_errors(self, delta: int = 1):
+        """Increment error counter."""
+        self._error_count += delta
+    
+    def log_error(self, message: str):
+        """Log an error message."""
+        self._errors.append({
+            "timestamp": datetime.utcnow().isoformat(),
+            "message": message
+        })
+        self.increment_errors()
+    
+    def get_stats(self) -> Dict:
+        """Get current statistics."""
+        return {
+            "queue_size": self._queue_size,
+            "processed_total": self._processed_count,
+            "batches_processed": self._batch_count,
+            "error_count": self._error_count,
+            "error_rate": self._error_count / max(1, self._processed_count)
+        }
+    
+    def get_recent_errors(self, limit: int = 10) -> List[Dict]:
+        """Get recent error entries."""
+        return list(self._errors)[-limit:]
+    
+    def reset(self):
+        """Reset all counters."""
+        self._queue_size = 0
+        self._processed_count = 0
+        self._batch_count = 0
+        self._error_count = 0
+        self._errors.clear()
+
+
+# Global monitor instance
+system_monitor = SystemMonitor()

backend/app/system/queue.py

@@ -0,0 +1,93 @@
+"""
+Message queue - buffer system for async message processing.
+"""
+from typing import Optional, List
+from collections import deque
+import asyncio
+import json
+from datetime import datetime
+
+
+class MessageQueue:
+    """
+    In-memory message queue with async support.
+    In production, use Redis for distributed processing.
+    """
+    
+    def __init__(self, max_size: int = 10000):
+        self._queue = deque()
+        self._max_size = max_size
+        self._lock = asyncio.Lock()
+    
+    async def enqueue(self, message) -> bool:
+        """
+        Add a message to the queue.
+        
+        Args:
+            message: Message object to queue
+            
+        Returns:
+            True if enqueued successfully
+        """
+        async with self._lock:
+            if len(self._queue) >= self._max_size:
+                return False
+            
+            # Serialize message for storage
+            msg_data = {
+                "id": message.id,
+                "sender_id": message.sender_id,
+                "receiver_id": message.receiver_id,
+                "group_id": message.group_id,
+                "content": message.content,
+                "type": message.type,
+                "created_at": message.created_at.isoformat() if message.created_at else None
+            }
+            self._queue.append(msg_data)
+            return True
+    
+    async def dequeue(self) -> Optional[dict]:
+        """
+        Remove and return the oldest message from queue.
+        
+        Returns:
+            Message data dict or None if empty
+        """
+        async with self._lock:
+            if not self._queue:
+                return None
+            return self._queue.popleft()
+    
+    async def peek(self) -> Optional[dict]:
+        """View the oldest message without removing it."""
+        async with self._lock:
+            if not self._queue:
+                return None
+            return self._queue[0]
+    
+    async def get_batch(self, size: int = 100) -> List[dict]:
+        """Get multiple messages at once for batch processing."""
+        async with self._lock:
+            batch = []
+            for _ in range(min(size, len(self._queue))):
+                if self._queue:
+                    batch.append(self._queue.popleft())
+            return batch
+    
+    async def clear(self):
+        """Clear all messages from queue."""
+        async with self._lock:
+            self._queue.clear()
+    
+    def size(self) -> int:
+        """Get current queue size (non-async)."""
+        return len(self._queue)
+    
+    async def is_empty(self) -> bool:
+        """Check if queue is empty."""
+        async with self._lock:
+            return len(self._queue) == 0
+
+
+# Global queue instance
+message_queue = MessageQueue()

backend/app/system/scheduler.py

@@ -0,0 +1,88 @@
+"""
+Scheduler - time-based trigger for batch processing (daily/hourly).
+"""
+import asyncio
+from datetime import datetime, timedelta
+from typing import Optional, Callable
+from ..config import settings
+
+
+class BatchScheduler:
+    """
+    Scheduler for automatic batch processing.
+    Triggers compression at configured intervals.
+    """
+    
+    def __init__(self):
+        self._running = False
+        self._task: Optional[asyncio.Task] = None
+        self._interval_minutes = settings.BATCH_INTERVAL_MINUTES
+        self._next_run: Optional[datetime] = None
+        self._last_run: Optional[datetime] = None
+    
+    async def start(self):
+        """Start the scheduler."""
+        self._running = True
+        self._calculate_next_run()
+        self._task = asyncio.create_task(self._run_loop())
+    
+    async def stop(self):
+        """Stop the scheduler."""
+        self._running = False
+        if self._task:
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+    
+    async def _run_loop(self):
+        """Main scheduler loop."""
+        while self._running:
+            try:
+                if self._should_run():
+                    await self._trigger_batch()
+            except Exception as e:
+                print(f"Scheduler error: {e}")
+            
+            # Check every minute
+            await asyncio.sleep(60)
+    
+    def _should_run(self) -> bool:
+        """Check if it's time to run."""
+        if not self._next_run:
+            return False
+        return datetime.utcnow() >= self._next_run
+    
+    def _calculate_next_run(self):
+        """Calculate next run time."""
+        self._next_run = datetime.utcnow() + timedelta(minutes=self._interval_minutes)
+    
+    async def _trigger_batch(self):
+        """Trigger batch processing."""
+        from .batch import batch_processor
+        from .monitor import system_monitor
+        
+        try:
+            await batch_processor.force_compress()
+            self._last_run = datetime.utcnow()
+            self._calculate_next_run()
+            system_monitor.increment_batches()
+        except Exception as e:
+            print(f"Scheduled batch error: {e}")
+    
+    def get_next_run(self) -> Optional[datetime]:
+        """Get next scheduled run time."""
+        return self._next_run
+    
+    def get_last_run(self) -> Optional[datetime]:
+        """Get last run time."""
+        return self._last_run
+    
+    async def run_now(self):
+        """Manually trigger a batch immediately."""
+        await self._trigger_batch()
+
+
+# Global scheduler
+batch_scheduler = BatchScheduler()

backend/app/system/worker.py

@@ -0,0 +1,100 @@
+"""
+Worker - process messages from queue (clean, group, compress).
+"""
+import asyncio
+from typing import Optional, List
+from datetime import datetime
+
+from .queue import message_queue
+from .batch import BatchProcessor
+from .monitor import SystemMonitor
+
+
+class Worker:
+    """
+    Background worker that processes messages from queue.
+    Handles cleaning, grouping, and preparing for compression.
+    """
+    
+    def __init__(self):
+        self.batch_processor = BatchProcessor()
+        self.monitor = SystemMonitor()
+        self._running = False
+        self._task: Optional[asyncio.Task] = None
+    
+    async def start(self, interval_seconds: int = 5):
+        """
+        Start the worker process.
+        
+        Args:
+            interval_seconds: How often to check queue
+        """
+        self._running = True
+        self._task = asyncio.create_task(self._run_loop(interval_seconds))
+    
+    async def stop(self):
+        """Stop the worker."""
+        self._running = False
+        if self._task:
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+    
+    async def _run_loop(self, interval: int):
+        """Main worker loop."""
+        while self._running:
+            try:
+                await self._process_queue()
+            except Exception as e:
+                self.monitor.log_error(f"Worker error: {e}")
+            
+            await asyncio.sleep(interval)
+    
+    async def _process_queue(self):
+        """Process pending messages in queue."""
+        # Get batch of messages
+        messages = await message_queue.get_batch(size=100)
+        
+        if not messages:
+            return
+        
+        # Process each message
+        for msg_data in messages:
+            try:
+                # Clean and normalize message
+                cleaned = await self._clean_message(msg_data)
+                
+                # Add to batch processor
+                await self.batch_processor.add_message(cleaned)
+                
+                self.monitor.increment_processed()
+            except Exception as e:
+                self.monitor.log_error(f"Message processing error: {e}")
+        
+        # Check if batch is ready to compress
+        await self._check_batch_ready()
+    
+    async def _clean_message(self, msg_data: dict) -> dict:
+        """Clean and normalize message data."""
+        # In production, use formatter.clean_message
+        content = msg_data.get("content", "")
+        
+        # Basic cleaning
+        cleaned_content = content.strip()
+        
+        return {
+            **msg_data,
+            "content": cleaned_content,
+            "processed_at": datetime.utcnow().isoformat()
+        }
+    
+    async def _check_batch_ready(self):
+        """Check if batch is ready for compression."""
+        if self.batch_processor.is_ready():
+            await self.batch_processor.compress_and_save()
+
+
+# Global worker instance
+worker = Worker()

backend/app/utils/__init__.py

@@ -0,0 +1,16 @@
+"""
+Utils package - helper utilities.
+"""
+from .security import hash_password, verify_password, create_access_token, decode_access_token
+from .avatar import generate_avatar, get_avatar_initials
+from .formatter import clean_message, normalize_content
+from .time import utc_now, utc_timestamp, format_date
+from .compression import compress_zstd, decompress_zstd
+
+__all__ = [
+    "hash_password", "verify_password", "create_access_token", "decode_access_token",
+    "generate_avatar", "get_avatar_initials",
+    "clean_message", "normalize_content",
+    "utc_now", "utc_timestamp", "format_date",
+    "compress_zstd", "decompress_zstd"
+]

backend/app/utils/avatar.py

@@ -0,0 +1,54 @@
+"""
+Avatar utilities - auto-generate avatars for users.
+"""
+import hashlib
+import base64
+from typing import Optional
+from io import BytesIO
+import json
+
+
+def generate_avatar_seed(user_id: str) -> str:
+    """Generate a deterministic seed from user ID for avatar."""
+    return hashlib.md5(user_id.encode()).hexdigest()[:16]
+
+
+def get_avatar_color(user_id: str) -> tuple:
+    """Generate a consistent color from user ID (RGB)."""
+    seed = generate_avatar_seed(user_id)
+    # Generate consistent color based on seed
+    r = int(seed[0:2], 16) % 200 + 30  # Avoid too dark
+    g = int(seed[2:4], 16) % 200 + 30
+    b = int(seed[4:6], 16) % 200 + 30
+    return (r, g, b)
+
+
+def get_avatar_initials(username: str) -> str:
+    """Get initials from username for avatar display."""
+    if not username:
+        return "?"
+    parts = username.split()
+    if len(parts) >= 2:
+        return (parts[0][0] + parts[1][0]).upper()
+    return username[:2].upper()
+
+
+def generate_identicon(user_id: str) -> str:
+    """
+    Generate a simple identicon (base64 encoded).
+    In production, use a proper library or service.
+    """
+    # Simplified placeholder - returns a colored background URL
+    color = get_avatar_color(user_id)
+    # In production: generate actual identicon or use a service like DiceBear
+    return f"https://api.dicebear.com/7.x/initials/svg?seed={user_id}"
+
+
+async def generate_avatar(user_id: str, username: str) -> str:
+    """
+    Generate an avatar URL for a user.
+    Can be customized to use different styles or services.
+    """
+    # Using DiceBear API for automatic avatar generation
+    # Options: 'initials', 'identicon', 'bottts', 'avataaars', etc.
+    return f"https://api.dicebear.com/7.x/initials/svg?seed={username}&backgroundColor={get_avatar_color(user_id)}"

backend/app/utils/compression.py

@@ -0,0 +1,85 @@
+"""
+Compression utilities - gzip and zstd compression for batch storage.
+"""
+import gzip
+import zstandard as zstd
+from typing import bytes
+from pathlib import Path
+import json
+
+
+def compress_gzip(data: bytes) -> bytes:
+    """Compress data using gzip."""
+    return gzip.compress(data, compresslevel=6)
+
+
+def decompress_gzip(data: bytes) -> bytes:
+    """Decompress gzip data."""
+    return gzip.decompress(data)
+
+
+def compress_zstd(data: bytes, level: int = 3) -> bytes:
+    """Compress data using zstd (better compression ratio)."""
+    cctx = zstd.ZstdCompressor(level=level)
+    return cctx.compress(data)
+
+
+def decompress_zstd(data: bytes) -> bytes:
+    """Decompress zstd data."""
+    dctx = zstd.ZstdDecompressor()
+    return dctx.decompress(data)
+
+
+async def compress_file(input_path: Path, output_path: Path, method: str = "zstd") -> bool:
+    """
+    Compress a file and save to output path.
+    
+    Args:
+        input_path: Source file path
+        output_path: Destination file path
+        method: 'zstd' or 'gzip'
+    """
+    try:
+        with open(input_path, "rb") as f:
+            data = f.read()
+        
+        if method == "zstd":
+            compressed = compress_zstd(data)
+        else:
+            compressed = compress_gzip(data)
+        
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        with open(output_path, "wb") as f:
+            f.write(compressed)
+        
+        return True
+    except Exception as e:
+        print(f"Compression error: {e}")
+        return False
+
+
+async def decompress_file(input_path: Path, output_path: Path, method: str = "zstd") -> bool:
+    """Decompress a file."""
+    try:
+        with open(input_path, "rb") as f:
+            data = f.read()
+        
+        if method == "zstd":
+            decompressed = decompress_zstd(data)
+        else:
+            decompressed = decompress_gzip(data)
+        
+        with open(output_path, "wb") as f:
+            f.write(decompressed)
+        
+        return True
+    except Exception as e:
+        print(f"Decompression error: {e}")
+        return False
+
+
+def get_compression_ratio(original_size: int, compressed_size: int) -> float:
+    """Calculate compression ratio."""
+    if original_size == 0:
+        return 0.0
+    return round((1 - compressed_size / original_size) * 100, 2)

backend/app/utils/formatter.py

@@ -0,0 +1,57 @@
+"""
+Formatter utilities - clean and normalize messages.
+"""
+import re
+from typing import Optional
+from html import escape
+
+
+def clean_message(text: str, max_length: int = 4000) -> str:
+    """
+    Clean and sanitize message text.
+    - Remove excessive whitespace
+    - Limit length
+    - Escape HTML
+    """
+    if not text:
+        return ""
+    
+    # Strip leading/trailing whitespace
+    text = text.strip()
+    
+    # Remove excessive internal whitespace (more than 2 spaces)
+    text = re.sub(r' {2,}', ' ', text)
+    
+    # Remove excessive newlines (more than 2 consecutive)
+    text = re.sub(r'\n{3,}', '\n\n', text)
+    
+    # Limit length
+    if len(text) > max_length:
+        text = text[:max_length]
+    
+    # Escape HTML to prevent XSS
+    text = escape(text)
+    
+    return text
+
+
+def normalize_content(content: str) -> str:
+    """Normalize message content for storage/search."""
+    # Lowercase for search indexing (optional)
+    # Remove special characters for search (optional)
+    return content.strip()
+
+
+def format_timestamp(timestamp) -> str:
+    """Format a timestamp for display."""
+    from datetime import datetime
+    if isinstance(timestamp, datetime):
+        return timestamp.strftime("%Y-%m-%d %H:%M")
+    return str(timestamp)
+
+
+def truncate_text(text: str, max_len: int = 50) -> str:
+    """Truncate text for previews."""
+    if len(text) <= max_len:
+        return text
+    return text[:max_len - 3] + "..."

backend/app/utils/security.py

@@ -0,0 +1,57 @@
+"""
+Security utilities - password hashing, JWT token handling.
+"""
+from passlib.context import CryptContext
+from jose import JWTError, jwt
+from datetime import datetime, timedelta
+from typing import Optional
+from ..config import settings
+
+# Password hashing
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+
+def hash_password(password: str) -> str:
+    """Hash a plain text password."""
+    return pwd_context.hash(password)
+
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    """Verify a password against its hash."""
+    return pwd_context.verify(plain_password, hashed_password)
+
+
+def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
+    """Create a JWT access token."""
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(minutes=settings.JWT_EXPIRATION_MINUTES)
+    
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, settings.JWT_SECRET, algorithm=settings.JWT_ALGORITHM)
+    return encoded_jwt
+
+
+def decode_access_token(token: str) -> Optional[dict]:
+    """Decode and verify a JWT token."""
+    try:
+        payload = jwt.decode(token, settings.JWT_SECRET, algorithms=[settings.JWT_ALGORITHM])
+        return payload
+    except JWTError:
+        return None
+
+
+def create_password_reset_token(email: str) -> str:
+    """Create a password reset token (time-limited)."""
+    delta = timedelta(hours=1)
+    return create_access_token({"sub": email, "type": "password_reset"}, delta)
+
+
+def verify_password_reset_token(token: str) -> Optional[str]:
+    """Verify password reset token and return email."""
+    payload = decode_access_token(token)
+    if payload and payload.get("type") == "password_reset":
+        return payload.get("sub")
+    return None

backend/app/utils/time.py

@@ -0,0 +1,79 @@
+"""
+Time utilities - timestamp helpers.
+"""
+from datetime import datetime, timezone, timedelta
+from typing import Optional
+import time
+
+
+def utc_now() -> datetime:
+    """Get current UTC time as timezone-aware datetime."""
+    return datetime.now(timezone.utc)
+
+
+def utc_timestamp() -> int:
+    """Get current UTC timestamp in seconds."""
+    return int(time.time())
+
+
+def format_date(dt: datetime, fmt: str = "%Y-%m-%d") -> str:
+    """Format datetime to string."""
+    if dt is None:
+        return ""
+    return dt.strftime(fmt)
+
+
+def format_datetime(dt: datetime) -> str:
+    """Format datetime for display."""
+    if dt is None:
+        return ""
+    return dt.strftime("%Y-%m-%d %H:%M:%S")
+
+
+def parse_date(date_str: str) -> Optional[datetime]:
+    """Parse date string to datetime."""
+    try:
+        return datetime.fromisoformat(date_str.replace("Z", "+00:00"))
+    except (ValueError, AttributeError):
+        return None
+
+
+def time_ago(dt: datetime) -> str:
+    """Get human-readable time ago string."""
+    if dt is None:
+        return "unknown"
+    
+    now = utc_now()
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    
+    diff = now - dt
+    
+    seconds = diff.total_seconds()
+    if seconds < 60:
+        return "just now"
+    elif seconds < 3600:
+        minutes = int(seconds / 60)
+        return f"{minutes}m ago"
+    elif seconds < 86400:
+        hours = int(seconds / 3600)
+        return f"{hours}h ago"
+    elif seconds < 604800:
+        days = int(seconds / 86400)
+        return f"{days}d ago"
+    else:
+        return format_date(dt)
+
+
+def get_date_path(dt: Optional[datetime] = None) -> str:
+    """Get date path for storage (YYYY/MM/DD)."""
+    if dt is None:
+        dt = utc_now()
+    return f"{dt.year}/{dt.month:02d}/{dt.day:02d}"
+
+
+def get_hour_bucket(dt: Optional[datetime] = None) -> int:
+    """Get hour bucket (0-23) for batching."""
+    if dt is None:
+        dt = utc_now()
+    return dt.hour

backend/requirements.txt

@@ -0,0 +1,41 @@
+# FastAPI and async
+fastapi==0.109.0
+uvicorn[standard]==0.27.0
+python-multipart==0.0.6
+pydantic==2.5.3
+pydantic-settings==2.1.0
+
+# Database
+sqlalchemy==2.0.25
+asyncpg==0.29.0
+alembic==1.13.1
+
+# Auth
+python-jose[cryptography]==3.3.0
+passlib[bcrypt]==1.7.4
+bcrypt==4.1.2
+
+# Async queue and workers
+aioredis==2.0.1
+celery==5.3.6
+httpx==0.26.0
+
+# Compression
+zstandard==0.22.0
+
+# Utils
+python-dotenv==1.0.0
+aiofiles==23.2.1
+python-dateutil==2.8.2
+
+# WebSocket
+websockets==12.0
+
+# GitHub API
+pygithub==2.2.0
+
+# Logging
+loguru==0.7.2
+
+# Optional: HuggingFace dataset upload
+# huggingface-hub==0.20.0

requirements.txt

@@ -0,0 +1,12 @@
+# HuggingFace Space Dependencies
+fastapi>=0.109.0
+uvicorn[standard]>=0.27.0
+python-multipart>=0.0.6
+python-jose[cryptography]>=3.3.0
+passlib[bcrypt]>=1.7.4
+sqlalchemy>=2.0.0
+asyncpg>=0.29.0
+redis>=5.0.0
+python-dotenv>=1.0.0
+pydantic>=2.5.0
+pydantic-settings>=2.1.0