initial commit

.gitignore

@@ -0,0 +1,4 @@
+myenv/
+scripts/
+.env
+__pycache__/

Dockerfile

@@ -0,0 +1,16 @@
+FROM python:3.12.7
+WORKDIR /code
+COPY ./requirements.txt /code/requirements.txt
+RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
+
+RUN useradd user
+USER user
+
+ENV HOME=/home/user \
+    PATH=/home/user/.local/bin:$PATH
+
+WORKDIR $HOME/app
+
+COPY --chown=user . $HOME/app
+
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

app/api/v1/api_router.py

@@ -0,0 +1,11 @@
+from fastapi import APIRouter
+from app.api.v1.endpoints import auth, users, admin, chat, messages
+
+api_router = APIRouter()
+
+# Add the routers to the main API router
+api_router.include_router(auth.router, prefix="/auth", tags=["Authentication"])
+api_router.include_router(users.router, prefix="/users", tags=["Users"])
+api_router.include_router(admin.router, prefix="/admin", tags=["Admin Management"])
+api_router.include_router(messages.router, prefix="/messages", tags=["Message History"])
+api_router.include_router(chat.router, prefix="/chat", tags=["Real-Time Chat"])

app/api/v1/endpoints/admin.py

@@ -0,0 +1,414 @@
+import json
+from fastapi import APIRouter, Depends, HTTPException, status, Query
+from motor.motor_asyncio import AsyncIOMotorClient
+from sqlalchemy.orm import Session, joinedload
+from typing import List, Union, Optional
+import datetime
+
+from app.db.session import get_db, get_mongo_db
+from app.security.dependencies import get_current_user_from_cookie
+from app.security.hashing import Hasher
+from app.models import Admin, User, Group, GroupMember
+from app.schemas.user import UserOut, UserCreate, UserPasswordReset
+from app.schemas.group import GroupCreateWithMembers, GroupWithMembers, GroupOut
+from app.schemas.admin import ConversationSummary
+from app.schemas.message import MessageHistory
+from app.websocket.connection_manager import manager
+from app.cache.group_members import remove_group_from_cache, add_member_to_cache, remove_member_from_cache
+
+router = APIRouter()
+
+def get_admin_from_dependency(current_entity: Union[User, Admin] = Depends(get_current_user_from_cookie)) -> Admin:
+    """
+    A helper dependency to ensure the user is an Admin.
+    This replaces the old get_current_active_admin.
+    """
+    if not isinstance(current_entity, Admin):
+        raise HTTPException(status_code=403, detail="Operation not permitted. Requires admin privileges.")
+    if not current_entity.is_active:
+        raise HTTPException(status_code=400, detail="Inactive admin.")
+    return current_entity
+
+# --- User Management by Admin ---
+
+@router.post("/users", response_model=UserOut, status_code=status.HTTP_201_CREATED)
+def create_user_for_admin(
+    user_in: UserCreate,
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_admin_from_dependency)
+):
+    # ... (logic remains the same)
+    full_username = f"{user_in.username}{current_admin.admin_key}"
+    db_user = db.query(User).filter(User.username == full_username).first()
+    if db_user:
+        raise HTTPException(status_code=400, detail=f"Username '{user_in.username}' already exists in your tenant.")
+    hashed_password = Hasher.get_password_hash(user_in.password)
+    new_user = User(username=full_username, password_hash=hashed_password, admin_id=current_admin.id)
+    db.add(new_user)
+    db.commit()
+    db.refresh(new_user)
+    return new_user
+
+@router.get("/users/all", response_model=List[UserOut])
+def list_users_for_admin(
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_admin_from_dependency)
+):
+    return db.query(User).filter(User.admin_id == current_admin.id).all()
+
+@router.get("/users/active", response_model=List[UserOut])
+def list_active_users_for_admin(
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_admin_from_dependency)
+):
+    """Lists all active users in the admin's tenant."""
+    return db.query(User).filter(User.admin_id == current_admin.id, User.is_active == True).all()
+
+@router.get("/users/deactivated", response_model=List[UserOut])
+def list_deactivated_users_for_admin(
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_admin_from_dependency)
+):
+    """Lists all deactivated users in the admin's tenant."""
+    return db.query(User).filter(User.admin_id == current_admin.id, User.is_active == False).all()
+
+@router.get("/users/online", response_model=List[UserOut])
+def get_online_users(
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_admin_from_dependency)
+):
+    """Gets a list of all online users within the current admin's tenant."""
+    online_connection_ids = manager.get_all_connection_ids()
+    online_user_ids = [int(cid.split('-')[1]) for cid in online_connection_ids if cid.startswith('user-')]
+
+    # Fetch details only for online users that belong to this admin's tenant
+    online_users_in_tenant = db.query(User).filter(
+        User.id.in_(online_user_ids),
+        User.admin_id == current_admin.id
+    ).all()
+    return online_users_in_tenant
+
+@router.patch("/users/{user_id}/deactivate", status_code=status.HTTP_204_NO_CONTENT)
+async def deactivate_user(
+    user_id: int,
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_admin_from_dependency)
+):
+    # ... (logic remains the same)
+    user = db.query(User).filter(User.id == user_id, User.admin_id == current_admin.id).first()
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found in your tenant.")
+    user.is_active = False
+    db.commit()
+    user_connection_id = f"user-{user.id}"
+    if user_connection_id in manager.active_connections:
+        logout_command = json.dumps({"event": "force_logout", "reason": "Your account has been deactivated by the administrator."})
+        await manager.send_personal_message(logout_command, user_connection_id)
+        manager.disconnect(user_connection_id)
+    return
+
+@router.patch("/users/{user_id}/reactivate", status_code=status.HTTP_204_NO_CONTENT)
+def reactivate_user(
+    user_id: int,
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_admin_from_dependency)
+):
+    # ... (logic remains the same)
+    user = db.query(User).filter(User.id == user_id, User.admin_id == current_admin.id).first()
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found in your tenant.")
+    user.is_active = True
+    db.commit()
+    return
+
+@router.patch("/users/{user_id}/reset-password", status_code=status.HTTP_204_NO_CONTENT)
+async def reset_user_password(
+    user_id: int,
+    password_data: UserPasswordReset,
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_admin_from_dependency)
+):
+    """
+    Resets the password for a specific user within the admin's tenant.
+    """
+    # Find the user and verify they belong to the admin's tenant
+    user = db.query(User).filter(User.id == user_id, User.admin_id == current_admin.id).first()
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found in your tenant.")
+
+    # Hash the new password and update the user's record
+    hashed_password = Hasher.get_password_hash(password_data.new_password)
+    user.password_hash = hashed_password
+    
+    db.commit()
+    user_connection_id = f"user-{user.id}"
+    if user_connection_id in manager.active_connections:
+        logout_command = json.dumps({"event": "force_logout", "reason": "Please re-authenticate yourself as admin has reset your password."})
+        await manager.send_personal_message(logout_command, user_connection_id)
+        manager.disconnect(user_connection_id)
+    return
+
+# --- Group Management by Admin ---
+
+@router.post("/groups", response_model=GroupOut, status_code=status.HTTP_201_CREATED)
+def create_group_for_admin(
+    group_in: GroupCreateWithMembers,
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_admin_from_dependency)
+):
+    # ... (logic remains the same)
+    db_group = db.query(Group).filter(Group.name == group_in.name, Group.admin_id == current_admin.id).first()
+    if db_group:
+        raise HTTPException(status_code=400, detail=f"Group name '{group_in.name}' already exists in your tenant.")
+    new_group = Group(name=group_in.name, admin_id=current_admin.id)
+    db.add(new_group)
+    db.flush()
+    if group_in.members:
+        for user_id in group_in.members:
+            user = db.query(User).filter(User.id == user_id, User.admin_id == current_admin.id).first()
+            if user:
+                new_member = GroupMember(group_id=new_group.id, user_id=user.id)
+                db.add(new_member)
+                add_member_to_cache(new_group.id, f"user-{user.id}")
+    db.commit()
+    db.refresh(new_group)
+    return new_group
+
+@router.get("/groups/all", response_model=List[GroupWithMembers])
+def list_groups_with_members_for_admin(
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_admin_from_dependency)
+):
+    """Lists all groups in the admin's tenant, including their members."""
+    groups = (
+        db.query(Group)
+        .options(joinedload(Group.members).joinedload(GroupMember.user))
+        .filter(Group.admin_id == current_admin.id)
+        .all()
+    )
+    
+    result = []
+    for group in groups:
+        members_info = [
+            {"user_id": member.user.id, "username": member.user.username}
+            for member in group.members
+        ]
+        result.append({
+            "id": group.id,
+            "name": group.name,
+            "admin_id": group.admin_id,
+            "is_active": group.is_active,
+            "members": members_info
+        })
+    return result
+
+@router.get("/groups/active", response_model=List[GroupWithMembers])
+def list_active_groups_with_members(
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_admin_from_dependency)
+):
+    """Lists all active groups in the admin's tenant, including their members."""
+    groups = (
+        db.query(Group)
+        .options(joinedload(Group.members).joinedload(GroupMember.user))
+        .filter(Group.admin_id == current_admin.id, Group.is_active == True)
+        .all()
+    )
+    # ... (response formatting logic remains the same)
+    result = []
+    for group in groups:
+        members_info = [{"user_id": member.user.id, "username": member.user.username} for member in group.members]
+        result.append({"id": group.id, "name": group.name, "admin_id": group.admin_id, "is_active": group.is_active, "members": members_info})
+    return result
+
+@router.get("/groups/deactivated", response_model=List[GroupWithMembers])
+def list_deactivated_groups_with_members(
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_admin_from_dependency)
+):
+    """Lists all deactivated groups in the admin's tenant, including their members."""
+    groups = (
+        db.query(Group)
+        .options(joinedload(Group.members).joinedload(GroupMember.user))
+        .filter(Group.admin_id == current_admin.id, Group.is_active == False)
+        .all()
+    )
+    # ... (response formatting logic remains the same)
+    result = []
+    for group in groups:
+        members_info = [{"user_id": member.user.id, "username": member.user.username} for member in group.members]
+        result.append({"id": group.id, "name": group.name, "admin_id": group.admin_id, "is_active": group.is_active, "members": members_info})
+    return result
+
+@router.delete("/groups/{group_id}", status_code=status.HTTP_204_NO_CONTENT)
+def deactivate_group(
+    group_id: int,
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_admin_from_dependency)
+):
+    group = db.query(Group).filter(Group.id == group_id, Group.admin_id == current_admin.id).first()
+    if not group:
+        raise HTTPException(status_code=404, detail="Group not found in your tenant.")
+    group.is_active = False
+    db.commit()
+    remove_group_from_cache(group_id)
+    return
+
+@router.post("/groups/{group_id}/members/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
+def add_user_to_group(
+    group_id: int,
+    user_id: int,
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_current_user_from_cookie)
+):
+    """
+    Add a user from the admin's tenant to a group in the same tenant.
+    """
+    # Verify the group belongs to the admin
+    group = db.query(Group).filter(Group.id == group_id, Group.admin_id == current_admin.id).first()
+    if not group:
+        raise HTTPException(status_code=404, detail="Group not found in your tenant.")
+
+    # Verify the user belongs to the admin
+    user = db.query(User).filter(User.id == user_id, User.admin_id == current_admin.id).first()
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found in your tenant.")
+
+    # Check if the user is already a member
+    membership = db.query(GroupMember).filter(GroupMember.group_id == group_id, GroupMember.user_id == user_id).first()
+    if membership:
+        raise HTTPException(status_code=400, detail="User is already a member of this group.")
+
+    new_member = GroupMember(group_id=group_id, user_id=user_id)
+    db.add(new_member)
+    db.commit()
+    add_member_to_cache(group_id, f"user-{user_id}")
+    return
+
+@router.delete("/groups/{group_id}/members/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
+def remove_user_from_group(
+    group_id: int,
+    user_id: int,
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_current_user_from_cookie)
+):
+    """
+    Remove a user from a group.
+    """
+    # Verify the group belongs to the admin to allow this operation
+    group = db.query(Group).filter(Group.id == group_id, Group.admin_id == current_admin.id).first()
+    if not group:
+        raise HTTPException(status_code=404, detail="Group not found in your tenant.")
+        
+    membership = db.query(GroupMember).filter(GroupMember.group_id == group_id, GroupMember.user_id == user_id).first()
+    if not membership:
+        raise HTTPException(status_code=404, detail="User is not a member of this group.")
+
+    db.delete(membership)
+    db.commit()
+    remove_member_from_cache(group_id, f"user-{user_id}")
+    return
+
+@router.get("/conversations/users", response_model=List[ConversationSummary])
+async def list_user_to_user_conversations(
+    db: Session = Depends(get_db),
+    mongo_db: AsyncIOMotorClient = Depends(get_mongo_db),
+    current_admin: Admin = Depends(get_admin_from_dependency)
+):
+    """
+    Lists all private user-to-user conversations within the admin's tenant,
+    sorted by the most recent message.
+    """
+    # 1. Get all user IDs for the current admin's tenant
+    tenant_user_ids = [user.id for user in db.query(User.id).filter(User.admin_id == current_admin.id).all()]
+    if not tenant_user_ids:
+        return []
+
+    messages_collection = mongo_db["messages"]
+    
+    # 2. Run the aggregation pipeline in MongoDB
+    pipeline = [
+        # Match only private messages between users in this tenant
+        {"$match": {
+            "type": "private",
+            "sender.role": "user",
+            "receiver.role": "user",
+            "sender.id": {"$in": tenant_user_ids},
+            "receiver.id": {"$in": tenant_user_ids}
+        }},
+        # Group by a canonical key (sorted participants)
+        {"$group": {
+            "_id": {
+                "participants": {
+                    "$sortArray": {"input": ["$sender", "$receiver"], "sortBy": {"id": 1}}
+                }
+            },
+            "last_message_timestamp": {"$max": "$timestamp"},
+            "message_count": {"$sum": 1}
+        }},
+        # Sort conversations by the most recent activity
+        {"$sort": {"last_message_timestamp": -1}}
+    ]
+    
+    aggregation_result = await messages_collection.aggregate(pipeline).to_list(length=None)
+
+    # 3. Format the response
+    response = []
+    for item in aggregation_result:
+        participants = item["_id"]["participants"]
+        response.append({
+            "user_one": {"id": participants[0]["id"], "username": participants[0]["username"]},
+            "user_two": {"id": participants[1]["id"], "username": participants[1]["username"]},
+            "last_message_timestamp": item["last_message_timestamp"],
+            "message_count": item["message_count"]
+        })
+        
+    return response
+
+@router.get("/messages/users/{user1_id}/{user2_id}", response_model=MessageHistory)
+async def get_user_to_user_message_history(
+    user1_id: int,
+    user2_id: int,
+    before: Optional[str] = Query(None, description="ISO timestamp cursor for pagination"),
+    limit: int = Query(50, gt=0, le=100),
+    db: Session = Depends(get_db),
+    mongo_db: AsyncIOMotorClient = Depends(get_mongo_db),
+    current_admin: Admin = Depends(get_admin_from_dependency)
+):
+    """
+    Fetches the detailed message history for a specific user-to-user conversation.
+    """
+    # 1. Security Check: Verify both users belong to the admin's tenant
+    users = db.query(User).filter(User.id.in_([user1_id, user2_id]), User.admin_id == current_admin.id).all()
+    if len(users) != 2:
+        raise HTTPException(status_code=404, detail="One or both users not found in your tenant.")
+
+    # 2. Build the MongoDB query
+    messages_collection = mongo_db["messages"]
+    query = {
+        "$or": [
+            {"sender.id": user1_id, "receiver.id": user2_id},
+            {"sender.id": user2_id, "receiver.id": user1_id}
+        ],
+        "type": "private"
+    }
+
+    if before:
+        try:
+            cursor_time = datetime.datetime.fromisoformat(before.replace("Z", "+00:00"))
+            query["timestamp"] = {"$lt": cursor_time}
+        except ValueError:
+            raise HTTPException(status_code=400, detail="Invalid 'before' timestamp format.")
+            
+    # 3. Fetch messages
+    messages_cursor = messages_collection.find(query).sort("timestamp", -1).limit(limit)
+    messages = await messages_cursor.to_list(length=limit)
+
+    for msg in messages:
+        msg["_id"] = str(msg["_id"])
+
+    next_cursor = None
+    if len(messages) == limit:
+        next_cursor = messages[-1]['timestamp'].isoformat() + "Z"
+        
+    return {"messages": messages, "next_cursor": next_cursor}

app/api/v1/endpoints/auth.py

@@ -0,0 +1,111 @@
+from fastapi import APIRouter, Depends, HTTPException, status, Response, Request
+from sqlalchemy.orm import Session
+
+from app.db.session import get_db
+from app.schemas.user import UserLoginSchema
+from app.security.hashing import Hasher
+from app.security.jwt import create_access_token, create_refresh_token, verify_token
+from app.models import User, Admin, SuperAdmin
+
+router = APIRouter()
+
+def set_auth_cookies(response: Response, access_token: str, refresh_token: str):
+    """Utility function to set auth cookies on a response."""
+    response.set_cookie(
+        key="access_token", value=access_token, httponly=True, samesite="lax", secure=True
+    )
+    response.set_cookie(
+        key="refresh_token", value=refresh_token, httponly=True, samesite="lax", secure=True
+    )
+
+def delete_auth_cookies(response: Response):
+    """Utility function to delete auth cookies."""
+    response.delete_cookie(key="access_token")
+    response.delete_cookie(key="refresh_token")
+
+@router.post("/login", status_code=status.HTTP_204_NO_CONTENT)
+def login(response: Response, user_credentials: UserLoginSchema, db: Session = Depends(get_db)):
+    """
+    Handles login and sets HttpOnly cookies for access and refresh tokens.
+    Now includes a check to ensure the user/admin is active.
+    """
+    entity = None
+    role = None
+    tenant_id = None
+    
+    inactive_exception = HTTPException(
+        status_code=status.HTTP_403_FORBIDDEN,
+        detail="Your account has been deactivated. Please contact your administrator."
+    )
+
+    # Check across user, admin, and super_admin tables
+    user = db.query(User).filter(User.username == user_credentials.username).first()
+    if user:
+        if not user.is_active:
+            raise inactive_exception
+        if Hasher.verify_password(user_credentials.password, user.password_hash):
+            entity, role, tenant_id = user, "user", user.admin_id
+    
+    if not entity:
+        admin = db.query(Admin).filter(Admin.username == user_credentials.username).first()
+        if admin:
+            if not admin.is_active:
+                raise inactive_exception
+            if Hasher.verify_password(user_credentials.password, admin.password_hash):
+                entity, role, tenant_id = admin, "admin", admin.id
+
+    if not entity:
+        super_admin = db.query(SuperAdmin).filter(SuperAdmin.username == user_credentials.username).first()
+        if super_admin and Hasher.verify_password(user_credentials.password, super_admin.password_hash):
+            entity, role, tenant_id = super_admin, "super_admin", None
+
+    if not entity:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password")
+
+    token_data = {"sub": entity.username, "role": role, "tenant_id": tenant_id}
+    access_token = create_access_token(data=token_data)
+    refresh_token = create_refresh_token(data=token_data)
+
+    set_auth_cookies(response, access_token, refresh_token)
+    return
+
+@router.post("/refresh", status_code=status.HTTP_204_NO_CONTENT)
+def refresh_token(request: Request, response: Response, db: Session = Depends(get_db)):
+    """
+    Uses the refresh_token from cookies to issue a new access_token.
+    Now includes a check to ensure the user/admin is still active.
+    """
+    refresh_token = request.cookies.get("refresh_token")
+    if not refresh_token:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="No refresh token found")
+    
+    credentials_exception = HTTPException(status_code=401, detail="Could not validate refresh token")
+    token_data = verify_token(refresh_token, credentials_exception)
+    
+    # *** FIX IS HERE: Verify the user from the token is still active ***
+    entity_to_check = None
+    if token_data.role == "user":
+        entity_to_check = db.query(User).filter(User.username == token_data.username).first()
+    elif token_data.role == "admin":
+        entity_to_check = db.query(Admin).filter(Admin.username == token_data.username).first()
+
+    if entity_to_check and not entity_to_check.is_active:
+        # If the user has been deactivated, invalidate their session by deleting cookies
+        delete_auth_cookies(response)
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Your account has been deactivated. Session terminated."
+        )
+
+    # Re-create the payload for the new access token
+    new_token_data = {"sub": token_data.username, "role": token_data.role, "tenant_id": token_data.tenant_id}
+    new_access_token = create_access_token(data=new_token_data)
+    
+    response.set_cookie(key="access_token", value=new_access_token, httponly=True, samesite="lax", secure=True)
+    return
+
+@router.post("/logout", status_code=status.HTTP_204_NO_CONTENT)
+def logout(response: Response):
+    """Logs the user out by deleting the auth cookies."""
+    delete_auth_cookies(response)
+    return

app/api/v1/endpoints/chat.py

@@ -0,0 +1,250 @@
+import json
+from datetime import datetime
+import pytz
+from typing import Union
+from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Depends, Query, HTTPException, BackgroundTasks
+from sqlalchemy.orm import Session
+from motor.motor_asyncio import AsyncIOMotorClient
+
+from app.db.session import get_db, get_mongo_db
+from app.security.jwt import verify_token
+from app.models import User, Admin
+from app.websocket.connection_manager import manager
+from app.cache.group_members import get_group_members
+from app.cache.tenant_members import get_tenant_connection_ids
+
+router = APIRouter()
+
+async def broadcast_presence_update(tenant_id: int, user_id: int, role: str, status: str, db: Session):
+    """Broadcasts a user's online/offline status to all users in the same tenant using the cache."""
+    # Use the cache to get the list of who to notify
+    broadcast_list = get_tenant_connection_ids(tenant_id, db)
+    
+    payload = json.dumps({
+        "event": "presence_update",
+        "user": {"id": user_id, "role": role},
+        "status": status,
+        "timestamp": datetime.now(pytz.timezone('Asia/Kolkata')).isoformat()
+    })
+    
+    await manager.broadcast_to_users(payload, list(broadcast_list))
+
+# --- Background task for database updates on disconnect ---
+def update_last_seen(user_id: int, db: Session):
+    try:
+        user = db.query(User).filter(User.id == user_id).first()
+        if user:
+            user.last_seen = datetime.utcnow()
+            db.commit()
+    finally:
+        db.close()
+
+async def mark_messages_as_received(user_id: int, user_role: str, db: AsyncIOMotorClient):
+    """Background task to update 'sent' messages to 'received'."""
+    messages_collection = db["messages"]
+    sent_messages_cursor = messages_collection.find({
+        "receiver.id": user_id,
+        "receiver.role": user_role,
+        "status": "sent"
+    })
+    sent_messages = await sent_messages_cursor.to_list(length=None)
+    if not sent_messages: return
+
+    message_ids_to_update = [msg["_id"] for msg in sent_messages]
+    
+    await messages_collection.update_many(
+        {"_id": {"$in": message_ids_to_update}},
+        {"$set": {"status": "received"}}
+    )
+
+    for msg in sent_messages:
+        sender_connection_id = f"{msg['sender']['role']}-{msg['sender']['id']}"
+        status_update_payload = json.dumps({
+            "event": "status_update",
+            "message_id": str(msg["_id"]),
+            "status": "received"
+        })
+        await manager.send_personal_message(status_update_payload, sender_connection_id)
+
+
+@router.websocket("/ws")
+async def websocket_endpoint(
+    websocket: WebSocket,
+    background_tasks: BackgroundTasks,
+    db: Session = Depends(get_db),
+    mongo_db: AsyncIOMotorClient = Depends(get_mongo_db)
+):
+    try:
+        # 1. Extract the access_token from the WebSocket's cookies
+        token = websocket.cookies.get("access_token")
+        if not token:
+            await websocket.close(code=1008)
+            return
+
+        # 2. Verify the token
+        credentials_exception = HTTPException(status_code=403)
+        token_data = verify_token(token, credentials_exception)
+        
+        # 3. Fetch the user/admin from the database
+        entity: Union[User, Admin] = None
+        if token_data.role == 'user':
+            entity = db.query(User).filter(User.username == token_data.username).first()
+        elif token_data.role == 'admin':
+            entity = db.query(Admin).filter(Admin.username == token_data.username).first()
+
+        if not entity:
+            await websocket.close(code=1008)
+            return
+            
+    except Exception:
+        # If any part of auth fails, close the connection
+        await websocket.close(code=1008)
+        return
+
+    connection_id_str = f"{token_data.role}-{entity.id}"
+    await manager.connect(connection_id_str, websocket)
+
+    tenant_id = entity.id if isinstance(entity, Admin) else entity.admin_id
+    
+    online_connection_ids = manager.get_all_connection_ids()
+    all_tenant_members = get_tenant_connection_ids(tenant_id, db)
+
+    # 1. Identify which users in the tenant are offline
+    offline_user_ids = []
+    for member_cid in all_tenant_members:
+        if member_cid.startswith('user-') and member_cid not in online_connection_ids:
+            offline_user_ids.append(int(member_cid.split('-')[1]))
+    
+    # 2. Query the database for their last_seen timestamps in one go
+    offline_users_info = {
+        user.id: user.last_seen 
+        for user in db.query(User.id, User.last_seen).filter(User.id.in_(offline_user_ids)).all()
+    }
+
+    # 3. Build the initial state map with the correct data
+    initial_state = {}
+    for member_cid in all_tenant_members:
+        role, member_id_str = member_cid.split('-')
+        member_id = int(member_id_str)
+        
+        if member_id == entity.id and role == token_data.role:
+            continue
+
+        if member_cid in online_connection_ids:
+            status = "online"
+            last_seen = None
+        else:
+            status = "offline"
+            # Use the fetched timestamp if available
+            last_seen = offline_users_info.get(member_id)
+        
+        # Ensure timestamp is in ISO format if it exists
+        last_seen_iso = last_seen.isoformat() + "Z" if last_seen else None
+        initial_state[member_cid] = {"status": status, "lastSeen": last_seen_iso}
+
+    await manager.send_personal_message(json.dumps({
+        "event": "initial_presence_state",
+        "users": initial_state
+    }), connection_id_str)
+
+    # Announce the new user's arrival to everyone else
+    await broadcast_presence_update(tenant_id, entity.id, token_data.role, "online", db)
+    
+    background_tasks.add_task(mark_messages_as_received, entity.id, token_data.role, mongo_db)
+    
+    try:
+        while True:
+            data = await websocket.receive_text()
+            message_data = json.loads(data)
+            event_type = message_data.get("event", "new_message")
+            messages_collection = mongo_db["messages"]
+
+            if event_type == "messages_read":
+                partner = message_data.get("partner")
+                if not partner: continue
+                
+                messages_to_update = await messages_collection.find({
+                    "receiver.id": entity.id, "receiver.role": token_data.role,
+                    "sender.id": partner["id"], "sender.role": partner["role"],
+                    "status": {"$in": ["sent", "received"]}
+                }).to_list(length=None)
+
+                if not messages_to_update: continue
+                
+                msg_ids = [msg["_id"] for msg in messages_to_update]
+                await messages_collection.update_many(
+                    {"_id": {"$in": msg_ids}},
+                    {"$set": {"status": "read"}}
+                )
+                
+                partner_connection_id = f"{partner['role']}-{partner['id']}"
+                await manager.send_personal_message(json.dumps({
+                    "event": "status_update",
+                    "message_ids": [str(mid) for mid in msg_ids],
+                    "status": "read"
+                }), partner_connection_id)
+                continue
+
+            if event_type == "new_message":
+                raw_content = message_data.get("content")
+                if isinstance(raw_content, dict):
+                    content_obj = raw_content
+                else:
+                    continue
+                mongo_message = {
+                    "type": message_data.get("type"),
+                    "sender": {"id": entity.id, "role": token_data.role, "username": entity.username},
+                    "content": content_obj,
+                    "timestamp": datetime.now(pytz.utc),
+                    "status": "sent",
+                    "is_deleted": False
+                }
+
+                if mongo_message["type"] == "private":
+                    receiver_info = message_data.get("receiver")
+                    if not receiver_info: continue
+                    
+                    mongo_message["receiver"] = {
+                        "id": receiver_info['id'],
+                        "role": receiver_info['role'],
+                        "username": receiver_info['username']
+                    }
+                    
+                    result = await messages_collection.insert_one(mongo_message)
+                    mongo_message["_id"] = str(result.inserted_id)
+                    
+                    receiver_connection_id = f"{receiver_info['role']}-{receiver_info['id']}"
+                    await manager.broadcast_to_users(json.dumps(mongo_message, default=str), [receiver_connection_id])
+
+                elif mongo_message["type"] == "group":
+                    group_info = message_data.get("group")
+                    if not group_info: continue
+                    
+                    mongo_message["group"] = {
+                        "id": group_info["id"],
+                        "name": group_info["name"]
+                    }
+
+                    result = await messages_collection.insert_one(mongo_message)
+                    mongo_message["_id"] = str(result.inserted_id)
+
+                    member_ids = get_group_members(group_info["id"], db=db)
+                    connections = set(member_ids)
+                    connections.discard(connection_id_str)
+                    await manager.broadcast_to_users(json.dumps(mongo_message, default=str), list(connections))
+
+    except WebSocketDisconnect:
+        manager.disconnect(connection_id_str)
+        await broadcast_presence_update(tenant_id, entity.id, token_data.role, "offline", db)
+        if isinstance(entity, User):
+            # Create a new session for the background task
+            db_session_for_task = next(get_db())
+            background_tasks.add_task(update_last_seen, entity.id, db_session_for_task)
+    except Exception as e:
+        print(f"Error in WebSocket: {e}")
+        manager.disconnect(connection_id_str)
+        await broadcast_presence_update(tenant_id, entity.id, token_data.role, "offline", db)
+        if isinstance(entity, User):
+            db_session_for_task = next(get_db())
+            background_tasks.add_task(update_last_seen, entity.id, db_session_for_task)
+

app/api/v1/endpoints/messages.py

@@ -0,0 +1,87 @@
+from fastapi import APIRouter, Depends, HTTPException, Query, Path
+from sqlalchemy.orm import Session
+from motor.motor_asyncio import AsyncIOMotorClient
+from typing import Union, Optional
+import datetime
+
+from app.db.session import get_db, get_mongo_db
+from app.security.dependencies import get_current_user_from_cookie
+from app.models import User, Admin, Group, GroupMember
+from app.schemas.message import MessageHistory
+
+router = APIRouter()
+
+@router.get("/{conversation_type}/{partner_id}", response_model=MessageHistory)
+async def get_message_history(
+    conversation_type: str = Path(..., description="Type of conversation: 'private' or 'group'"),
+    partner_id: int = Path(..., description="ID of the user, admin, or group"),
+    partner_role: Optional[str] = Query(None, description="Role of the partner if private: 'user' or 'admin'"),
+    before: Optional[str] = Query(None, description="ISO timestamp cursor for pagination"),
+    limit: int = Query(50, gt=0, le=100),
+    current_entity: Union[User, Admin] = Depends(get_current_user_from_cookie),
+    db: Session = Depends(get_db),
+    mongo_db: AsyncIOMotorClient = Depends(get_mongo_db)
+):
+    """
+    Fetch the message history for a specific conversation with pagination.
+    """
+    messages_collection = mongo_db["messages"]
+    
+    entity_id = current_entity.id
+    entity_role = "admin" if isinstance(current_entity, Admin) else "user"
+    
+    query = {"type": conversation_type}
+
+    # --- Authorization and Query Building ---
+    if conversation_type == "private":
+        if not partner_role:
+            raise HTTPException(status_code=400, detail="Partner role is required for private chats.")
+        
+        # Build a query that finds messages between the two parties, regardless of who is sender/receiver
+        query["$or"] = [
+            {"sender.id": entity_id, "sender.role": entity_role, "receiver.id": partner_id, "receiver.role": partner_role},
+            {"sender.id": partner_id, "sender.role": partner_role, "receiver.id": entity_id, "receiver.role": entity_role}
+        ]
+    elif conversation_type == "group":
+        # Verify the current entity is a member of the group
+        group = db.query(Group).filter(Group.id == partner_id).first()
+        if not group: raise HTTPException(status_code=404, detail="Group not found.")
+        
+        is_member = False
+        if isinstance(current_entity, Admin) and current_entity.id == group.admin_id:
+            is_member = True
+        elif isinstance(current_entity, User):
+            membership = db.query(GroupMember).filter(GroupMember.group_id == partner_id, GroupMember.user_id == entity_id).first()
+            if membership: is_member = True
+        
+        if not is_member:
+            raise HTTPException(status_code=403, detail="You are not a member of this group.")
+        
+        query["group.id"] = partner_id
+    else:
+        raise HTTPException(status_code=400, detail="Invalid conversation type.")
+
+    # --- Pagination ---
+    if before:
+        try:
+            # Using timestamp for cursor-based pagination
+            cursor_time = datetime.datetime.fromisoformat(before.replace("Z", "+00:00"))
+            query["timestamp"] = {"$lt": cursor_time}
+        except ValueError:
+            raise HTTPException(status_code=400, detail="Invalid 'before' timestamp format.")
+
+    # --- Fetching Messages ---
+    messages_cursor = messages_collection.find(query).sort("timestamp", -1).limit(limit)
+    messages = await messages_cursor.to_list(length=limit)
+
+    # Convert MongoDB's _id to a string for Pydantic
+    for msg in messages:
+        msg["_id"] = str(msg["_id"])
+
+    # --- Determine the next cursor ---
+    next_cursor = None
+    if len(messages) == limit:
+        # The timestamp of the last message fetched is the cursor for the next page
+        next_cursor = messages[-1]['timestamp'].isoformat() + "Z"
+
+    return {"messages": messages, "next_cursor": next_cursor}

app/api/v1/endpoints/users.py

@@ -0,0 +1,159 @@
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy.orm import Session
+from motor.motor_asyncio import AsyncIOMotorClient
+from typing import Union
+
+from app.db.session import get_db, get_mongo_db
+from app.security.dependencies import get_current_user_from_cookie
+from app.models import User, Admin, Group, GroupMember
+from app.websocket.connection_manager import manager
+from typing import List
+from app.schemas.user import UserOut, AdminOut, SearchResult, ConversationList, ConversationPartner, MeOut, MeProfileOut
+
+router = APIRouter()
+
+@router.get("/me", response_model=MeProfileOut)
+def read_users_me(
+    # Use the new cookie-based dependency here
+    current_entity: Union[User, Admin] = Depends(get_current_user_from_cookie)
+):
+    """
+    Get the profile of the currently authenticated user or admin.
+    Authentication is now handled via HttpOnly cookies.
+    """
+    if isinstance(current_entity, User):
+        creator_name = current_entity.owner_admin.username
+        entity_type = "user"
+    elif isinstance(current_entity, Admin):
+        creator_name = "Super Admin"
+        entity_type = "admin"
+    else:
+        # This case handles if a Super Admin somehow hits this endpoint
+        creator_name = "System"
+        entity_type = "super_admin"
+
+    return MeProfileOut(
+        id=current_entity.id,
+        username=current_entity.username,
+        type=entity_type,
+        created_by=creator_name,
+        created_at=current_entity.created_at
+    )
+
+@router.get("/search", response_model=SearchResult)
+def search_for_entities(
+    query: str = Query(..., min_length=1, description="Search query for users, admins, or groups"),
+    # current_entity: Union[User, Admin] = Depends(get_current_active_user_or_admin),
+    current_entity: Union[User, Admin] = Depends(get_current_user_from_cookie),
+    db: Session = Depends(get_db)
+):
+    """
+    Search for users, the tenant admin, and groups within the same tenant.
+    - For Users: Shows other users in their tenant.
+    - For Admins: Shows all users and groups in their tenant.
+    """
+    if isinstance(current_entity, Admin):
+        tenant_id = current_entity.id
+        current_id = None
+    else: # It's a User
+        tenant_id = current_entity.admin_id
+        current_id = current_entity.id
+
+    # --- Search for users ---
+    user_query = db.query(User).filter(
+        User.username.ilike(f"%{query}%"),
+        User.admin_id == tenant_id
+    )
+    if current_id: # Exclude self from search if the searcher is a user
+        user_query = user_query.filter(User.id != current_id)
+    found_users = user_query.all()
+
+    # --- Search for the admin ---
+    found_admins = db.query(Admin).filter(
+        Admin.username.ilike(f"%{query}%"),
+        Admin.id == tenant_id
+    ).all()
+    
+    # --- Search for groups ---
+    group_query = db.query(Group).filter(
+        Group.name.ilike(f"%{query}%"),
+        Group.admin_id == tenant_id
+    )
+    # If the searcher is a standard user, only show groups they are a member of.
+    if isinstance(current_entity, User):
+        group_query = group_query.join(Group.members).filter(GroupMember.user_id == current_entity.id)
+    
+    found_groups = group_query.all()
+
+    return {"users": found_users, "admins": found_admins, "groups": found_groups}
+
+@router.get("/conversations", response_model=ConversationList)
+async def get_user_conversations(
+    # current_entity: Union[User, Admin] = Depends(get_current_active_user_or_admin),
+    current_entity: Union[User, Admin] = Depends(get_current_user_from_cookie),
+    db: Session = Depends(get_db),
+    mongo_db: AsyncIOMotorClient = Depends(get_mongo_db)
+):
+    messages_collection = mongo_db["messages"]
+    
+    entity_id = current_entity.id
+    entity_role = "admin" if isinstance(current_entity, Admin) else "user"
+    tenant_id = current_entity.id if isinstance(current_entity, Admin) else current_entity.admin_id
+
+    user_group_ids = []
+    if isinstance(current_entity, User):
+        memberships = db.query(GroupMember.group_id).filter_by(user_id=entity_id).all()
+        user_group_ids = [row.group_id for row in memberships]
+    else: # An admin is part of all groups in their tenant
+        groups = db.query(Group.id).filter_by(admin_id=tenant_id).all()
+        user_group_ids = [row.id for row in groups]
+
+    pipeline = [
+        {"$match": {
+            "$or": [
+                # Private messages sent to or from the user
+                {"sender.id": entity_id, "sender.role": entity_role},
+                {"receiver.id": entity_id, "receiver.role": entity_role},
+                # Messages from any group the user is a member of
+                {"group.id": {"$in": user_group_ids}}
+            ]
+        }},
+        {"$sort": {"timestamp": -1}},
+        {"$group": {
+            "_id": {
+                "$cond": {
+                    "if": {"$eq": ["$type", "private"]},
+                    "then": {
+                        "participants": {
+                           "$sortArray": { "input": [ "$sender", "$receiver" ], "sortBy": { "id": 1 } }
+                        }
+                    },
+                    "else": {"$concat": ["group-", {"$toString": "$group.id"}]}
+                }
+            },
+            "last_message_doc": {"$first": "$$ROOT"}
+        }},
+        {"$replaceRoot": {"newRoot": "$last_message_doc"}}
+    ]
+
+    latest_messages = await messages_collection.aggregate(pipeline).to_list(length=None)
+    
+    conversations = []
+    # The post-aggregation filtering is no longer needed, as the DB query is now correct.
+    for msg in sorted(latest_messages, key=lambda x: x['timestamp'], reverse=True):
+        last_message_text = msg.get("content", {}).get("text", "[attachment]")
+        
+        if msg['type'] == 'private':
+            partner = msg['receiver'] if msg['sender']['id'] == entity_id and msg['sender']['role'] == entity_role else msg['sender']
+            conversations.append(ConversationPartner(
+                id=partner['id'], name=partner['username'], type=partner['role'],
+                last_message=last_message_text, timestamp=msg['timestamp']
+            ))
+        elif msg['type'] == 'group':
+            group = msg['group']
+            conversations.append(ConversationPartner(
+                id=group['id'], name=group['name'], type='group',
+                last_message=last_message_text, timestamp=msg['timestamp']
+            ))
+
+    return {"conversations": conversations}

app/cache/group_members.py

@@ -0,0 +1,40 @@
+from typing import Dict, Set
+from threading import Lock
+from sqlalchemy.orm import Session
+from app.models import GroupMember, Group
+
+group_members_cache: Dict[int, Set[str]] = {}
+lock = Lock()
+
+def get_group_members(group_id: int, db: Session) -> Set[str]:
+    with lock:
+        if group_id in group_members_cache:
+            return group_members_cache[group_id]
+
+    group = db.query(Group).get(group_id)
+    if not group: return set()
+
+    members = db.query(GroupMember).filter_by(group_id=group.id).all()
+    connection_ids = {f"user-{m.user_id}" for m in members}
+    connection_ids.add(f"admin-{group.admin_id}")
+    
+    with lock:
+        group_members_cache[group_id] = connection_ids
+    return connection_ids
+
+def add_member_to_cache(group_id: int, connection_id: str):
+    with lock:
+        if group_id not in group_members_cache:
+            return
+        group_members_cache[group_id].add(connection_id)
+
+def remove_member_from_cache(group_id: int, connection_id: str):
+    with lock:
+        if group_id in group_members_cache:
+            group_members_cache[group_id].discard(connection_id)
+
+def remove_group_from_cache(group_id: int):
+    """Removes an entire group from the cache, e.g., when it's deactivated."""
+    with lock:
+        if group_id in group_members_cache:
+            del group_members_cache[group_id]

app/cache/tenant_members.py

@@ -0,0 +1,33 @@
+# app/cache/tenant_members.py
+from typing import Dict, List, Set
+from threading import Lock
+from sqlalchemy.orm import Session
+from app.models import User, Admin
+
+# Cache format: { tenant_id: {"user-1", "user-2", "admin-10"} }
+tenant_members_cache: Dict[int, Set[str]] = {}
+lock = Lock()
+
+def get_tenant_connection_ids(tenant_id: int, db: Session) -> Set[str]:
+    """
+    Returns cached connection IDs for a tenant.
+    If not cached, it queries the DB, populates the cache, and returns the data.
+    """
+    with lock:
+        if tenant_id in tenant_members_cache:
+            return tenant_members_cache[tenant_id]
+
+    # --- Not in cache, so load from PostgreSQL ---
+    tenant_users = db.query(User.id).filter(User.admin_id == tenant_id).all()
+    # The admin is also a member of their own tenant
+    tenant_admin = db.query(Admin.id).filter(Admin.id == tenant_id).first()
+
+    connection_ids = {f"user-{uid}" for uid, in tenant_users}
+    if tenant_admin:
+        connection_ids.add(f"admin-{tenant_admin.id}")
+    
+    # Populate the cache
+    with lock:
+        tenant_members_cache[tenant_id] = connection_ids
+    
+    return connection_ids

app/core/config.py

@@ -0,0 +1,38 @@
+# app/core/config.py
+import os
+from dotenv import load_dotenv
+from pathlib import Path
+
+# Load environment variables from .env file
+env_path = Path('.') / '.env'
+load_dotenv(dotenv_path=env_path)
+
+class Settings:
+    """
+    Application settings loaded from environment variables.
+    """
+    PROJECT_NAME: str = "Multi-Tenant Chat App"
+    PROJECT_VERSION: str = "1.0.0"
+
+    # PostgreSQL (Neon) settings
+    POSTGRES_USER: str = os.getenv("POSTGRES_USER", "postgres")
+    POSTGRES_PASSWORD = os.getenv("POSTGRES_PASSWORD", "password")
+    POSTGRES_SERVER: str = os.getenv("POSTGRES_SERVER", "localhost")
+    POSTGRES_PORT: str = os.getenv("POSTGRES_PORT", "5432")
+    POSTGRES_DB: str = os.getenv("POSTGRES_DB", "chat_app")
+    DATABASE_URL: str = os.getenv("DATABASE_URL")
+
+    # MongoDB (Atlas) settings
+    MONGO_DATABASE_URL: str = os.getenv("MONGO_DATABASE_URL")
+    MONGO_DB_NAME: str = os.getenv("MONGO_DB_NAME", "chat_app")
+
+
+    # JWT settings
+    SECRET_KEY: str = os.getenv("SECRET_KEY", "a_very_secret_key")
+    ALGORITHM: str = os.getenv("ALGORITHM", "HS256")
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", 30))
+
+    # API settings
+    API_V1_STR: str = "/api/v1"
+
+settings = Settings()

app/db/session.py

@@ -0,0 +1,52 @@
+# app/db/session.py
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from motor.motor_asyncio import AsyncIOMotorClient
+from app.core.config import settings
+
+# --- PostgreSQL (SQLAlchemy) Setup ---
+engine = create_engine(
+    settings.DATABASE_URL,
+    pool_pre_ping=True,
+    pool_recycle=1800
+)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+def get_db():
+    """
+    Dependency to get a DB session.
+    Ensures the session is always closed after the request.
+    """
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+# --- MongoDB (Motor) Setup ---
+class DataBase:
+    client: AsyncIOMotorClient = None
+
+db = DataBase()
+
+async def get_mongo_db():
+    """
+    Dependency to get the MongoDB database instance.
+    """
+    return db.client[settings.MONGO_DB_NAME]
+
+async def connect_to_mongo():
+    """
+    Connect to the MongoDB instance.
+    """
+    print("Connecting to MongoDB...")
+    db.client = AsyncIOMotorClient(settings.MONGO_DATABASE_URL)
+    print("Successfully connected to MongoDB.")
+
+async def close_mongo_connection():
+    """
+    Close the MongoDB connection.
+    """
+    print("Closing MongoDB connection...")
+    db.client.close()
+    print("MongoDB connection closed.")

app/models/__init__.py

@@ -0,0 +1,11 @@
+# app/models/__init__.py
+
+# This makes it easier to import models from other parts of the application.
+# For example: from app.models import User, Admin
+
+from .base import Base
+from .super_admin import SuperAdmin
+from .admin import Admin
+from .user import User
+from .group import Group
+from .group_member import GroupMember

app/models/admin.py

@@ -0,0 +1,16 @@
+from sqlalchemy import Boolean, Column, Integer, String, Text, DateTime
+from sqlalchemy.orm import relationship
+from .base import Base
+import datetime
+
+class Admin(Base):
+    __tablename__ = 'admins'
+    id = Column(Integer, primary_key=True, index=True)
+    username = Column(String(50), unique=True, nullable=False)
+    password_hash = Column(Text, nullable=False)
+    admin_key = Column(Text, unique=True, nullable=False)
+    is_active = Column(Boolean, default=True)
+    created_at = Column(DateTime, default=datetime.datetime.utcnow)
+    
+    users = relationship("User", back_populates="owner_admin", cascade="all, delete-orphan")
+    groups = relationship("Group", back_populates="owner_admin", cascade="all, delete-orphan")

app/models/base.py

@@ -0,0 +1,5 @@
+from sqlalchemy.orm import declarative_base
+
+# Create a Base class for all ORM models to inherit from.
+# This allows SQLAlchemy to discover all the models.
+Base = declarative_base()

app/models/group.py

@@ -0,0 +1,17 @@
+from sqlalchemy import Boolean, Column, Integer, Text, ForeignKey, DateTime, UniqueConstraint
+from sqlalchemy.orm import relationship
+from .base import Base
+import datetime
+
+class Group(Base):
+    __tablename__ = 'groups'
+    id = Column(Integer, primary_key=True, index=True)
+    name = Column(Text, nullable=False)
+    admin_id = Column(Integer, ForeignKey('admins.id', ondelete="CASCADE"), nullable=False)
+    is_active = Column(Boolean, default=True)
+    created_at = Column(DateTime, default=datetime.datetime.utcnow)
+
+    owner_admin = relationship("Admin", back_populates="groups")
+    members = relationship("GroupMember", back_populates="group", cascade="all, delete-orphan")
+    
+    __table_args__ = (UniqueConstraint('name', 'admin_id', name='_group_name_admin_uc'),)

app/models/group_member.py

@@ -0,0 +1,14 @@
+from sqlalchemy import Boolean, Column, Integer, ForeignKey, DateTime
+from sqlalchemy.orm import relationship
+from .base import Base
+import datetime
+
+class GroupMember(Base):
+    __tablename__ = 'group_members'
+    group_id = Column(Integer, ForeignKey('groups.id', ondelete="CASCADE"), primary_key=True)
+    user_id = Column(Integer, ForeignKey('users.id', ondelete="CASCADE"), primary_key=True)
+    is_admin = Column(Boolean, default=False)
+    joined_at = Column(DateTime, default=datetime.datetime.utcnow)
+
+    group = relationship("Group", back_populates="members")
+    user = relationship("User", back_populates="groups")

app/models/super_admin.py

@@ -0,0 +1,10 @@
+from sqlalchemy import Column, Integer, String, Text, DateTime
+from .base import Base
+import datetime
+
+class SuperAdmin(Base):
+    __tablename__ = 'super_admin'
+    id = Column(Integer, primary_key=True, index=True)
+    username = Column(String(50), unique=True, nullable=False)
+    password_hash = Column(Text, nullable=False)
+    created_at = Column(DateTime, default=datetime.datetime.utcnow)

app/models/user.py

@@ -0,0 +1,17 @@
+from sqlalchemy import Boolean, Column, Integer, Text, ForeignKey, DateTime
+from sqlalchemy.orm import relationship
+from .base import Base
+import datetime
+
+class User(Base):
+    __tablename__ = 'users'
+    id = Column(Integer, primary_key=True, index=True)
+    username = Column(Text, unique=True, nullable=False)
+    password_hash = Column(Text, nullable=False)
+    admin_id = Column(Integer, ForeignKey('admins.id', ondelete="CASCADE"), nullable=False)
+    is_active = Column(Boolean, default=True)
+    created_at = Column(DateTime, default=datetime.datetime.utcnow)
+    last_seen = Column(DateTime, default=datetime.datetime.utcnow)
+
+    owner_admin = relationship("Admin", back_populates="users")
+    groups = relationship("GroupMember", back_populates="user", cascade="all, delete-orphan")

app/schemas/admin.py

@@ -0,0 +1,28 @@
+from pydantic import BaseModel
+from typing import List
+import datetime
+
+class AdminUserCreate(BaseModel):
+    username: str
+    password: str
+
+class AdminUserUpdate(BaseModel):
+    is_active: bool
+
+class OnlineUser(BaseModel):
+    id: int
+    username: str
+    role: str
+
+class AllOnlineUsers(BaseModel):
+    users: List[OnlineUser]
+
+class UserPairInfo(BaseModel):
+    id: int
+    username: str
+
+class ConversationSummary(BaseModel):
+    user_one: UserPairInfo
+    user_two: UserPairInfo
+    last_message_timestamp: datetime.datetime
+    message_count: int

app/schemas/group.py

@@ -0,0 +1,31 @@
+from pydantic import BaseModel
+from typing import List, Optional
+
+class GroupMemberInfo(BaseModel):
+    user_id: int
+    username: str
+
+class GroupWithMembers(BaseModel):
+    id: int
+    name: str
+    admin_id: int
+    is_active: bool
+    members: List[GroupMemberInfo]
+
+    class Config:
+        orm_mode = True
+
+class GroupCreateWithMembers(BaseModel):
+    name: str
+    members: Optional[List[int]] = []
+
+class GroupCreate(BaseModel):
+    name: str
+
+class GroupOut(BaseModel):
+    id: int
+    name: str
+    admin_id: int
+
+    class Config:
+        orm_mode = True

app/schemas/message.py

@@ -0,0 +1,47 @@
+# app/schemas/message.py
+from pydantic import BaseModel, Field
+from typing import Optional
+import datetime
+
+# --- Nested objects for the main Message schema ---
+class MessageSender(BaseModel):
+    id: int
+    username: str
+    role: str
+
+class MessageReceiver(BaseModel):
+    id: int
+    username: str
+    role: str
+
+class MessageGroup(BaseModel):
+    id: int
+    name: str
+
+class MessageContent(BaseModel):
+    text: Optional[str] = None
+    image: Optional[str] = None
+    file: Optional[str] = None
+
+# --- Main Message Schema for API output ---
+class MessageOut(BaseModel):
+    id: str = Field(..., alias="_id") # Use MongoDB's _id
+    type: str
+    sender: MessageSender
+    receiver: Optional[MessageReceiver] = None
+    group: Optional[MessageGroup] = None
+    content: MessageContent
+    timestamp: datetime.datetime
+    status: str
+
+    class Config:
+        orm_mode = True
+        allow_population_by_field_name = True
+        json_encoders = {
+            datetime.datetime: lambda dt: dt.isoformat(),
+        }
+
+# --- Schema for the new message history endpoint response ---
+class MessageHistory(BaseModel):
+    messages: list[MessageOut]
+    next_cursor: Optional[str] = None

app/schemas/token.py

@@ -0,0 +1,15 @@
+from typing import Optional
+from pydantic import BaseModel
+
+class Token(BaseModel):
+    access_token: str
+    token_type: str
+
+class TokenData(BaseModel):
+    username: Optional[str] = None
+    role: Optional[str] = None
+    tenant_id: Optional[int] = None
+
+class TokenRequest(BaseModel):
+    username: str
+    password: str

app/schemas/user.py

@@ -0,0 +1,60 @@
+# app/schemas/user.py
+from pydantic import BaseModel, Field
+from typing import List, Union
+import datetime
+from .group import GroupOut
+
+class UserCreate(BaseModel):
+    username: str
+    password: str
+
+class MeProfileOut(BaseModel):
+    id: int
+    username: str
+    type: str
+    created_by: str
+    created_at: datetime.datetime
+
+class UserOut(BaseModel):
+    id: int
+    username: str
+    type: str = "User"
+    
+    class Config:
+        orm_mode = True
+
+class AdminOut(BaseModel):
+    id: int
+    username: str
+    type: str = "Admin"
+    admin_key: str
+
+    class Config:
+        orm_mode = True
+
+# A flexible response model for the /me endpoint
+MeOut = Union[UserOut, AdminOut]
+
+# A schema for the search result
+class SearchResult(BaseModel):
+    users: List[UserOut]
+    admins: List[AdminOut]
+    groups: List[GroupOut] 
+
+# Schemas for the conversation list
+class ConversationPartner(BaseModel):
+    id: int
+    name: str
+    type: str # "user", "group", or "admin"
+    last_message: str
+    timestamp: datetime.datetime
+
+class ConversationList(BaseModel):
+    conversations: List[ConversationPartner]
+
+class UserLoginSchema(BaseModel):
+    username: str
+    password: str
+
+class UserPasswordReset(BaseModel):
+    new_password: str

app/security/dependencies.py

@@ -0,0 +1,40 @@
+from fastapi import Depends, HTTPException, status, Request
+from sqlalchemy.orm import Session
+from typing import Union
+
+from app.db.session import get_db
+from app.security.jwt import verify_token
+from app.models import User, Admin, SuperAdmin
+
+def get_current_user_from_cookie(request: Request, db: Session = Depends(get_db)) -> Union[User, Admin, SuperAdmin]:
+    """
+    New primary dependency to get the current user from the access_token cookie.
+    """
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    
+    token = request.cookies.get("access_token")
+    if token is None:
+        raise credentials_exception
+        
+    token_data = verify_token(token, credentials_exception)
+    
+    # Based on the role in the token, fetch from the correct table
+    role = token_data.role
+    username = token_data.username
+    
+    user = None
+    if role == "user":
+        user = db.query(User).filter(User.username == username).first()
+    elif role == "admin":
+        user = db.query(Admin).filter(Admin.username == username).first()
+    elif role == "super_admin":
+        user = db.query(SuperAdmin).filter(SuperAdmin.username == username).first()
+
+    if user is None:
+        raise credentials_exception
+        
+    return user

app/security/hashing.py

@@ -0,0 +1,15 @@
+from passlib.context import CryptContext
+
+# Use bcrypt for password hashing
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+class Hasher:
+    @staticmethod
+    def verify_password(plain_password, hashed_password):
+        """Verifies a plain password against a hashed one."""
+        return pwd_context.verify(plain_password, hashed_password)
+
+    @staticmethod
+    def get_password_hash(password):
+        """Hashes a plain password."""
+        return pwd_context.hash(password)

app/security/jwt.py

@@ -0,0 +1,40 @@
+from datetime import datetime, timedelta
+from typing import Optional
+from jose import JWTError, jwt
+from app.core.config import settings
+from app.schemas.token import TokenData
+
+ACCESS_TOKEN_EXPIRE_MINUTES = 15  # 15 minutes
+REFRESH_TOKEN_EXPIRE_DAYS = 7     # 7 days
+
+def create_access_token(data: dict):
+    """Creates a short-lived access token."""
+    to_encode = data.copy()
+    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
+    return encoded_jwt
+
+def create_refresh_token(data: dict):
+    """Creates a long-lived refresh token."""
+    to_encode = data.copy()
+    expire = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
+    return encoded_jwt
+
+def verify_token(token: str, credentials_exception) -> TokenData:
+    """Verifies any JWT token and returns its payload."""
+    try:
+        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
+        username: str = payload.get("sub")
+        role: str = payload.get("role")
+        tenant_id: int = payload.get("tenant_id")
+
+        if username is None or role is None:
+            raise credentials_exception
+        
+        token_data = TokenData(username=username, role=role, tenant_id=tenant_id)
+        return token_data
+    except JWTError:
+        raise credentials_exception

app/websocket/connection_manager.py

@@ -0,0 +1,36 @@
+# app/websocket/connection_manager.py
+from fastapi import WebSocket
+from typing import Dict, List, Set
+
+class ConnectionManager:
+    def __init__(self):
+        # Maps user_id to their active WebSocket connection
+        self.active_connections: Dict[int, WebSocket] = {}
+
+    async def connect(self, user_id: int, websocket: WebSocket):
+        """Accept a new WebSocket connection."""
+        await websocket.accept()
+        self.active_connections[user_id] = websocket
+
+    def disconnect(self, user_id: int):
+        """Disconnect a WebSocket."""
+        if user_id in self.active_connections:
+            del self.active_connections[user_id]
+
+    async def send_personal_message(self, message: str, user_id: int):
+        """Send a message to a specific user."""
+        if user_id in self.active_connections:
+            websocket = self.active_connections[user_id]
+            await websocket.send_text(message)
+
+    async def broadcast_to_users(self, message: str, user_ids: List[int]):
+        """Send a message to a list of specific users."""
+        for user_id in user_ids:
+            if user_id in self.active_connections:
+                await self.send_personal_message(message, user_id)
+
+    def get_all_connection_ids(self) -> Set[str]:
+        """Returns a set of all active connection IDs."""
+        return set(self.active_connections.keys())
+
+manager = ConnectionManager()

main.py

@@ -0,0 +1,48 @@
+# main.py
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from app.core.config import settings
+from app.api.v1.api_router import api_router
+from app.db.session import close_mongo_connection, connect_to_mongo
+
+app = FastAPI(
+    title="Multi-Tenant Chat API",
+    openapi_url=f"{settings.API_V1_STR}/openapi.json"
+)
+
+origins = [
+    "http://localhost:3000"
+]
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=origins,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+@app.on_event("startup")
+async def startup_event():
+    """
+    Connect to MongoDB on startup.
+    """
+    await connect_to_mongo()
+
+@app.on_event("shutdown")
+async def shutdown_event():
+    """
+    Close MongoDB connection on shutdown.
+    """
+    await close_mongo_connection()
+
+
+# Include the API router
+app.include_router(api_router, prefix=settings.API_V1_STR)
+
+@app.get("/")
+def read_root():
+    """
+    Root endpoint for basic health check.
+    """
+    return {"message": "Welcome to the Multi-Tenant Chat API"}

requirements.txt

@@ -0,0 +1,11 @@
+fastapi
+uvicorn[standard]
+psycopg2-binary
+SQLAlchemy
+motor
+pydantic[email]
+python-dotenv
+passlib[bcrypt]
+python-jose[cryptography]
+python-multipart
+pytz