SaraView/Backend/Main.py

from __future__ import annotations

import json
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 FileResponse, HTMLResponse, PlainTextResponse, StreamingResponse

from Auth import (
    CreateSession,
    DeleteSession,
    GetCurrentUser,
    GetCurrentUserOrGuest,
    RequireAdmin,
    SessionCookieName,
)
from BookReader import ReadBookForFile
from Blame import LoadBlame, RenameUserInBlame, UpdateBlameOnSave
from Core import ResolvePath
from Embeds import BuildOgTags
from Editor import WriteTextFile
from Explorer import (
    ClassifyFile,
    CreateEntry,
    DeleteEntry,
    EnsureDirectoryNote,
    ListDirectory,
    ListRoots,
    MoveEntry,
)
from Groups import (
    CreateGroup,
    DeleteGroup,
    GuestGroupName,
    ListGroups,
    RenameUserInGroups,
    SetGroupMembers,
)
from Locks import (
    GetLockForPath,
    IsPathAccessible,
    LoadLocks,
    RemoveLock,
    RequirePathAccessible,
    SetLock,
)
from Portals import ResolvePortals
from Preview import GetMimeType, ReadTextFile, StreamBinaryFile, TextFileTypes
from Settings import (
    DefaultSettings,
    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,
)


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:
    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}


@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(GetCurrentUserOrGuest)):
    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)

        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}


@Application.get("/api/groups")
def GetGroupsEndpoint(CurrentUser: dict = Depends(GetCurrentUser)) -> list[dict]:
    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}


@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]:
    return LoadLocks()


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


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

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

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

    return [
        Entry
        for Entry in Entries
        if IsPathAccessible(
            Root,
            f"{Path}/{Entry['Name']}" if Path else Entry["Name"],
            Username,
            UserGroups,
        )
    ]


@Application.post("/api/note")
def PostNoteEndpoint(Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUser)) -> dict:
    RequirePathAccessible(Root, Path, CurrentUser)

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

    if not DirectoryPath.is_dir():
        raise HTTPException(status_code=400, detail="Not a directory.")

    NotePath = EnsureDirectoryNote(DirectoryPath)
    return {"Name": NotePath.name, "FileType": ClassifyFile(NotePath)}


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(GetCurrentUserOrGuest)
):
    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))

    return StreamBinaryFile(FilePath, request.headers.get("range"))


@Application.get("/api/file/resolved")
def GetResolvedFileEndpoint(
    Root: str, Path: str, CurrentUser: dict = Depends(GetCurrentUserOrGuest)
) -> dict:
    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 not in TextFileTypes:
        raise HTTPException(status_code=400, detail="Not a text file.")

    Content = ReadTextFile(FilePath)
    Resolved = ResolvePortals(
        Root, Path, Content, CurrentUser["Username"], CurrentUser.get("Groups", [])
    )
    return {"Content": Resolved}


@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(GetCurrentUserOrGuest)) -> 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(GetCurrentUserOrGuest)
):
    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(GetCurrentUserOrGuest)) -> 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 {}


@Application.get("/api/settings")
def GetSettingsEndpoint(CurrentUser: dict = Depends(GetCurrentUserOrGuest)) -> dict:
    if CurrentUser["Username"] is None:
        return dict(DefaultSettings)

    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(GetCurrentUserOrGuest)) -> 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}


@Application.get("/public/media/{RootName}/{FilePath:path}")
def GetPublicMediaEndpoint(RootName: str, FilePath: str, request: Request):
    if not IsPathAccessible(RootName, FilePath, "", [GuestGroupName]):
        raise HTTPException(status_code=404, detail="Not found.")

    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.")

    return StreamBinaryFile(ResolvedPath, request.headers.get("range"))


def GetForwardedScheme(request: Request) -> str:
    ForwardedProto = request.headers.get("x-forwarded-proto")
    if ForwardedProto:
        return ForwardedProto.split(",")[0].strip()

    CfVisitor = request.headers.get("cf-visitor")
    if CfVisitor:
        try:
            ParsedScheme = json.loads(CfVisitor).get("scheme")
            if ParsedScheme:
                return ParsedScheme
        except (json.JSONDecodeError, AttributeError):
            pass

    return request.url.scheme


FrontendDist = ApplicationDirectory.parent / "Frontend" / "dist"

if FrontendDist.is_dir():
    IndexHtmlTemplate = (FrontendDist / "index.html").read_text(encoding="utf-8")

    @Application.get("/{FullPath:path}")
    def ServeFrontendEndpoint(FullPath: str, request: Request):
        CandidatePath = (FrontendDist / FullPath).resolve()
        IsWithinDist = CandidatePath == FrontendDist or FrontendDist in CandidatePath.parents

        if FullPath and IsWithinDist and CandidatePath.is_file():
            return FileResponse(CandidatePath)

        OgTags = ""
        Segments = [Segment for Segment in FullPath.split("/") if Segment]

        if Segments:
            Scheme = GetForwardedScheme(request)
            Host = request.headers.get("x-forwarded-host", request.url.netloc)
            BaseUrl = f"{Scheme}://{Host}"
            OgTags = (
                BuildOgTags(Segments[0], "/".join(Segments[1:]), FullPath.endswith("/"), BaseUrl)
                or ""
            )

        return HTMLResponse(IndexHtmlTemplate.replace("<!--OgTags-->", OgTags))