SaraView/Backend/Main.py

from __future__ import annotations

import mimetypes
import sys
from pathlib import Path

ApplicationDirectory = Path(__file__).resolve().parent
if str(ApplicationDirectory) not in sys.path:
    sys.path.insert(0, str(ApplicationDirectory))

from fastapi import Depends, FastAPI, File, HTTPException, Request, Response, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import PlainTextResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles

from Auth import CreateSession, DeleteSession, GetCurrentUser, RequireAdmin, SessionCookieName
from BookReader import ReadBookForFile
from Blame import LoadBlame, RenameUserInBlame, UpdateBlameOnSave
from Core import ResolvePath
from Editor import WriteTextFile
from Explorer import ClassifyFile, CreateEntry, DeleteEntry, ListDirectory, ListRoots, MoveEntry
from Groups import (
    CreateGroup,
    DeleteGroup,
    ListGroups,
    RenameUserInGroups,
    SetGroupMembers,
)
from Locks import (
    GetLockForPath,
    IsPathAccessible,
    LoadLocks,
    RemoveLock,
    RequirePathAccessible,
    SetLock,
)
from Preview import GetFileSize, GetMimeType, ParseRangeHeader, ReadTextFile, TextFileTypes
from Settings import (
    DeleteTheme,
    ListThemeNames,
    LoadSettings,
    LoadTheme,
    PaletteRoles,
    RenameUserSettings,
    SaveSettings,
    SaveTheme,
)
from Users import (
    AnyUsersExist,
    ChangePassword,
    CreateUser,
    GetAvatarPath,
    ListUsers,
    RenameUser,
    RenameUserAvatar,
    SaveAvatar,
    UpdateUserColour,
    UserColourOptions,
    VerifyLogin,
)
from pydantic import BaseModel


Application = FastAPI(title="SaraView")

Application.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
    allow_credentials=True,
)


# --- Auth ---------------------------------------------------------------


class LoginRequest(BaseModel):
    Username: str
    Password: str


@Application.get("/api/auth/status")
def AuthStatusEndpoint() -> dict:
    return {"SetupComplete": AnyUsersExist()}


@Application.post("/api/auth/bootstrap")
def BootstrapEndpoint(Body: LoginRequest) -> dict:
    """Create the first (admin) account. Only usable while no users exist yet,
    so there's no default/baked-in password sitting in the image.
    """
    if AnyUsersExist():
        raise HTTPException(status_code=403, detail="Setup already completed.")

    try:
        CreateUser(Body.Username, Body.Password, True)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    return {"Created": True}


@Application.post("/api/auth/login")
def LoginEndpoint(Body: LoginRequest, response: Response) -> dict:
    User = VerifyLogin(Body.Username, Body.Password)

    if User is None:
        raise HTTPException(status_code=401, detail="Invalid username or password.")

    Token = CreateSession(Body.Username)
    response.set_cookie(
        key=SessionCookieName,
        value=Token,
        httponly=True,
        samesite="lax",
        max_age=60 * 60 * 24 * 30,
    )
    return User


@Application.post("/api/auth/logout")
def LogoutEndpoint(request: Request, response: Response) -> dict:
    Token = request.cookies.get(SessionCookieName)
    if Token:
        DeleteSession(Token)
    response.delete_cookie(SessionCookieName)
    return {"LoggedOut": True}


@Application.get("/api/auth/me")
def MeEndpoint(CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
    return CurrentUser


class CreateUserRequest(BaseModel):
    Username: str
    Password: str
    IsAdmin: bool = False
    Colour: str = "Purple"


@Application.get("/api/users")
def GetUsersEndpoint(CurrentUser: dict = Depends(GetCurrentUser)) -> list[dict]:
    return ListUsers()


@Application.post("/api/users")
def PostUsersEndpoint(Body: CreateUserRequest, CurrentUser: dict = Depends(RequireAdmin)) -> dict:
    try:
        CreateUser(Body.Username, Body.Password, Body.IsAdmin, Body.Colour)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    return {"Created": True}


# --- Profile (colour, avatar, username, password) --------------------------


@Application.get("/api/profile/colours")
def GetProfileColourOptionsEndpoint(CurrentUser: dict = Depends(GetCurrentUser)) -> list[str]:
    return UserColourOptions


class UpdateColourRequest(BaseModel):
    Colour: str


@Application.put("/api/profile/colour")
def PutProfileColourEndpoint(
    Body: UpdateColourRequest, CurrentUser: dict = Depends(GetCurrentUser)
) -> dict:
    try:
        UpdateUserColour(CurrentUser["Username"], Body.Colour)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    return {"Username": CurrentUser["Username"], "IsAdmin": CurrentUser["IsAdmin"], "Colour": Body.Colour}


@Application.post("/api/profile/avatar")
async def PostProfileAvatarEndpoint(
    Avatar: UploadFile = File(...), CurrentUser: dict = Depends(GetCurrentUser)
) -> dict:
    Extension = Path(Avatar.filename or "").suffix.lower() or ".png"
    FileBytes = await Avatar.read()

    try:
        SaveAvatar(CurrentUser["Username"], FileBytes, Extension)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    return {"Saved": True}


@Application.get("/api/profile/avatar")
def GetProfileAvatarEndpoint(Username: str, CurrentUser: dict = Depends(GetCurrentUser)):
    AvatarPath = GetAvatarPath(Username)

    if AvatarPath is None:
        raise HTTPException(status_code=404, detail="No avatar set.")

    MimeType = mimetypes.guess_type(str(AvatarPath))[0] or "application/octet-stream"
    return StreamingResponse(iter([AvatarPath.read_bytes()]), media_type=MimeType)


class ChangeUsernameRequest(BaseModel):
    NewUsername: str


@Application.put("/api/profile/username")
def PutProfileUsernameEndpoint(
    Body: ChangeUsernameRequest,
    request: Request,
    response: Response,
    CurrentUser: dict = Depends(GetCurrentUser),
) -> dict:
    OldUsername = CurrentUser["Username"]
    NewUsername = Body.NewUsername.strip()

    try:
        RenameUser(OldUsername, NewUsername)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    if NewUsername != OldUsername:
        RenameUserSettings(OldUsername, NewUsername)
        RenameUserAvatar(OldUsername, NewUsername)
        RenameUserInBlame(OldUsername, NewUsername)
        RenameUserInGroups(OldUsername, NewUsername)

        # Re-authenticate this same request under the new name so the
        # person doesn't get logged out by their own rename. Any OTHER
        # active session for the old name is left to fail naturally on its
        # next request — a re-login there is the simplest correct outcome.
        OldToken = request.cookies.get(SessionCookieName)
        if OldToken:
            DeleteSession(OldToken)

        NewToken = CreateSession(NewUsername)
        response.set_cookie(
            key=SessionCookieName,
            value=NewToken,
            httponly=True,
            samesite="lax",
            max_age=60 * 60 * 24 * 30,
        )

    return {"Username": NewUsername, "IsAdmin": CurrentUser["IsAdmin"], "Colour": CurrentUser["Colour"]}


class ChangePasswordRequest(BaseModel):
    CurrentPassword: str
    NewPassword: str


@Application.put("/api/profile/password")
def PutProfilePasswordEndpoint(
    Body: ChangePasswordRequest, CurrentUser: dict = Depends(GetCurrentUser)
) -> dict:
    try:
        ChangePassword(CurrentUser["Username"], Body.CurrentPassword, Body.NewPassword)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    return {"Changed": True}


# --- Groups -----------------------------------------------------------------


@Application.get("/api/groups")
def GetGroupsEndpoint(CurrentUser: dict = Depends(GetCurrentUser)) -> list[dict]:
    # Readable by anyone — picking an existing group as a lock target is
    # open to everyone, same as picking a user.
    return ListGroups()


class CreateGroupRequest(BaseModel):
    Name: str


@Application.post("/api/groups")
def PostGroupsEndpoint(Body: CreateGroupRequest, CurrentUser: dict = Depends(RequireAdmin)) -> dict:
    try:
        CreateGroup(Body.Name)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    return {"Created": True}


@Application.delete("/api/groups")
def DeleteGroupsEndpoint(Name: str, CurrentUser: dict = Depends(RequireAdmin)) -> dict:
    try:
        DeleteGroup(Name)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    return {"Deleted": True}


class SetGroupMembersRequest(BaseModel):
    Members: list[str]


@Application.put("/api/groups/members")
def PutGroupMembersEndpoint(
    Name: str, Body: SetGroupMembersRequest, CurrentUser: dict = Depends(RequireAdmin)
) -> dict:
    try:
        SetGroupMembers(Name, Body.Members)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    return {"Saved": True}


# --- Locked files/directories -----------------------------------------------
# Setting a lock is open to anyone who currently has access to the path -
# not admin-only. If you can see it, you're trusted to decide who else can.
# Admins do NOT bypass locks (see Locks.IsPathAccessible) - Admin is a
# management group, not a way to view everyone's private directories.


@Application.get("/api/lock")
def GetLockEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)
    return GetLockForPath(Root, Path) or {}


class SetLockRequest(BaseModel):
    Users: list[str] = []
    Groups: list[str] = []


@Application.put("/api/lock")
def PutLockEndpoint(
    Root: str, Path: str, Body: SetLockRequest, CurrentUser: dict = Depends(GetCurrentUser)
) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)

    try:
        SetLock(Root, Path, Body.Users, Body.Groups)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    return {"Locked": True}


@Application.delete("/api/lock")
def DeleteLockEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)
    RemoveLock(Root, Path)
    return {"Unlocked": True}


@Application.get("/api/locks")
def GetLocksEndpoint(CurrentUser: dict = Depends(RequireAdmin)) -> list[dict]:
    # The full list (across every root/path) stays admin-only — it's a
    # global overview, not a per-item action.
    return LoadLocks()


# --- Files ----------------------------------------------------------------


@Application.get("/api/roots")
def GetRootsEndpoint(CurrentUser: dict = Depends(GetCurrentUser)) -> list[dict]:
    return ListRoots()


@Application.get("/api/tree")
def GetTreeEndpoint(
    Root: str, Path: str = "", CurrentUser: dict = Depends(GetCurrentUser)
) -> list[dict]:
    RequirePathAccessible(Root, Path, CurrentUser)

    try:
        ShowHidden = LoadSettings(CurrentUser["Username"]).get("ShowHiddenFolders", False)
        Entries = ListDirectory(Root, Path, ShowHidden)
    except (ValueError, NotADirectoryError, FileNotFoundError) as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    Username = CurrentUser["Username"]
    UserGroups = CurrentUser.get("Groups", [])

    # A locked subdirectory (or anything inside one) shouldn't even appear
    # in its parent's listing for someone without access to it.
    return [
        Entry
        for Entry in Entries
        if IsPathAccessible(
            Root,
            f"{Path}/{Entry['Name']}" if Path else Entry["Name"],
            Username,
            UserGroups,
        )
    ]


class CreateEntryRequest(BaseModel):
    IsDirectory: bool


@Application.post("/api/tree")
def PostTreeEndpoint(
    Root: str, Path: str, Body: CreateEntryRequest, CurrentUser: dict = Depends(GetCurrentUser)
) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)

    try:
        CreateEntry(Root, Path, Body.IsDirectory)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))
    except FileExistsError as Error:
        raise HTTPException(status_code=409, detail=str(Error))
    except OSError as Error:
        raise HTTPException(status_code=500, detail=f"Could not create entry: {Error}")

    return {"Created": True}


class MoveEntryRequest(BaseModel):
    NewName: str | None = None
    DestinationPath: str | None = None


@Application.patch("/api/tree")
def PatchTreeEndpoint(
    Root: str, Path: str, Body: MoveEntryRequest, CurrentUser: dict = Depends(GetCurrentUser)
) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)

    if Body.DestinationPath:
        Destination = Body.DestinationPath
    elif Body.NewName:
        ParentPath = Path.rsplit("/", 1)[0] if "/" in Path else ""
        Destination = f"{ParentPath}/{Body.NewName}" if ParentPath else Body.NewName
    else:
        raise HTTPException(status_code=400, detail="Provide NewName or DestinationPath.")

    RequirePathAccessible(Root, Destination, CurrentUser)

    try:
        MoveEntry(Root, Path, Destination)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))
    except FileNotFoundError as Error:
        raise HTTPException(status_code=404, detail=str(Error))
    except FileExistsError as Error:
        raise HTTPException(status_code=409, detail=str(Error))
    except OSError as Error:
        raise HTTPException(status_code=500, detail=f"Could not move entry: {Error}")

    return {"Moved": True, "NewPath": Destination}


@Application.delete("/api/tree")
def DeleteTreeEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)

    try:
        DeleteEntry(Root, Path)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))
    except FileNotFoundError as Error:
        raise HTTPException(status_code=404, detail=str(Error))
    except OSError as Error:
        raise HTTPException(status_code=500, detail=f"Could not delete entry: {Error}")

    return {"Deleted": True}


@Application.get("/api/file")
def GetFileEndpoint(
    Root: str, Path: str, request: Request, CurrentUser: dict = Depends(GetCurrentUser)
):
    RequirePathAccessible(Root, Path, CurrentUser)

    try:
        FilePath = ResolvePath(Root, Path)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    if not FilePath.is_file():
        raise HTTPException(status_code=404, detail="File not found.")

    FileType = ClassifyFile(FilePath)

    if FileType in TextFileTypes:
        return PlainTextResponse(ReadTextFile(FilePath))

    FileSize = GetFileSize(FilePath)
    MimeType = GetMimeType(FilePath)
    RangeHeader = request.headers.get("range")

    if RangeHeader:
        Start, End = ParseRangeHeader(RangeHeader, FileSize)
        ChunkLength = End - Start + 1

        def IterateChunk():
            with FilePath.open("rb") as SourceFile:
                SourceFile.seek(Start)
                yield SourceFile.read(ChunkLength)

        Headers = {
            "Content-Range": f"bytes {Start}-{End}/{FileSize}",
            "Accept-Ranges": "bytes",
            "Content-Length": str(ChunkLength),
        }
        return StreamingResponse(
            IterateChunk(), status_code=206, media_type=MimeType, headers=Headers
        )

    def IterateFile():
        with FilePath.open("rb") as SourceFile:
            while Chunk := SourceFile.read(1024 * 1024):
                yield Chunk

    Headers = {"Accept-Ranges": "bytes", "Content-Length": str(FileSize)}
    return StreamingResponse(IterateFile(), media_type=MimeType, headers=Headers)


@Application.put("/api/file")
async def PutFileEndpoint(
    Root: str, Path: str, request: Request, CurrentUser: dict = Depends(GetCurrentUser)
):
    RequirePathAccessible(Root, Path, CurrentUser)

    try:
        FilePath = ResolvePath(Root, Path)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    OldContent = ""
    if FilePath.is_file():
        try:
            OldContent = FilePath.read_text(encoding="utf-8")
        except (UnicodeDecodeError, OSError):
            OldContent = ""

    Content = (await request.body()).decode("utf-8")

    try:
        WriteTextFile(FilePath, Content)
    except OSError as Error:
        raise HTTPException(status_code=500, detail=f"Could not write file: {Error}")

    UpdateBlameOnSave(Root, Path, OldContent, Content, CurrentUser["Username"])

    return {"Saved": True}


@Application.get("/api/blame")
def GetBlameEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)

    try:
        ResolvePath(Root, Path)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    RawBlame = LoadBlame(Root, Path)
    AllUsers = {User["Username"]: User for User in ListUsers()}
    ViewerUsername = CurrentUser["Username"]
    Enriched: dict[str, dict] = {}

    for LineNumber, Entry in RawBlame.items():
        AuthorUsername = Entry.get("Username")
        Timestamp = Entry.get("Timestamp")

        if AuthorUsername == ViewerUsername:
            IsVisible = True
        elif AuthorUsername in AllUsers:
            IsVisible = LoadSettings(AuthorUsername).get("VisibleInBlame", True)
        else:
            IsVisible = True

        if IsVisible:
            AuthorRecord = AllUsers.get(AuthorUsername, {})
            Enriched[LineNumber] = {
                "Username": AuthorUsername,
                "Timestamp": Timestamp,
                "Colour": AuthorRecord.get("Colour", "Purple"),
                "HasAvatar": AuthorUsername is not None and GetAvatarPath(AuthorUsername) is not None,
            }
        else:
            Enriched[LineNumber] = {"Username": None, "Hidden": True, "Timestamp": Timestamp}

    return Enriched


@Application.get("/api/raw/{RootName}/{FilePath:path}")
def GetRawFileEndpoint(
    RootName: str, FilePath: str, CurrentUser: dict = Depends(GetCurrentUser)
):
    """Serves a file at a URL that mirrors its real directory structure
    (unlike /api/file's query-param form), so relative links, stylesheets,
    and scripts INSIDE a previewed HTML document resolve correctly against
    sibling files in the same directory.
    """
    RequirePathAccessible(RootName, FilePath, CurrentUser)

    try:
        ResolvedPath = ResolvePath(RootName, FilePath)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    if not ResolvedPath.is_file():
        raise HTTPException(status_code=404, detail="File not found.")

    MimeType = mimetypes.guess_type(str(ResolvedPath))[0] or "application/octet-stream"

    def IterateFile():
        with ResolvedPath.open("rb") as SourceFile:
            while Chunk := SourceFile.read(1024 * 1024):
                yield Chunk

    return StreamingResponse(IterateFile(), media_type=MimeType)


@Application.get("/api/book")
def GetBookEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)

    try:
        FilePath = ResolvePath(Root, Path)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    Book = ReadBookForFile(FilePath)
    return Book or {}


# --- Settings / themes ------------------------------------------------------


@Application.get("/api/settings")
def GetSettingsEndpoint(CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
    return LoadSettings(CurrentUser["Username"])


@Application.put("/api/settings")
def PutSettingsEndpoint(NewSettings: dict, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
    return SaveSettings(CurrentUser["Username"], NewSettings)


@Application.get("/api/themes")
def GetThemesEndpoint(CurrentUser: dict = Depends(GetCurrentUser)) -> list[str]:
    return ListThemeNames()


@Application.get("/api/theme")
def GetThemeEndpoint(Name: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
    try:
        return LoadTheme(Name)
    except FileNotFoundError as Error:
        raise HTTPException(status_code=404, detail=str(Error))
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))


class SaveThemeRequest(BaseModel):
    Palette: dict[str, str]


@Application.put("/api/theme")
def PutThemeEndpoint(
    Name: str, Body: SaveThemeRequest, CurrentUser: dict = Depends(GetCurrentUser)
) -> dict:
    try:
        SaveTheme(Name, Body.Palette)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))

    return {"Saved": True, "Roles": PaletteRoles}


@Application.delete("/api/theme")
def DeleteThemeEndpoint(Name: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
    try:
        DeleteTheme(Name)
    except ValueError as Error:
        raise HTTPException(status_code=400, detail=str(Error))
    except FileNotFoundError as Error:
        raise HTTPException(status_code=404, detail=str(Error))

    return {"Deleted": True}


# In production, the frontend's built assets are copied alongside this file
# under Frontend/dist (see Dockerfile) and served directly by the backend.
# This stays unauthenticated — the SPA itself needs to load before it can
# even show a login screen.
FrontendDist = ApplicationDirectory.parent / "Frontend" / "dist"
if FrontendDist.is_dir():
    Application.mount("/", StaticFiles(directory=str(FrontendDist), html=True), name="Frontend")