# Index Virtual Terminal (IVT) Webserver
# Copyright (C) 2026 Morgana <morgana@icolotl.com>

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.

# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.

import asyncio
from random import choice, randint
from string import ascii_uppercase
from json import dumps, loads, JSONDecodeError
from argparse import ArgumentParser, ArgumentError, Namespace, _SubParsersAction
from typing import Literal
import sqlite3
from hashlib import scrypt
from os import urandom, listdir
from hmac import compare_digest
from colorsys import rgb_to_hsv

import requests
from websockets.asyncio.server import serve, ServerConnection


class IVTException(Exception): ...


HEX_CHARS = set("0123456789ABCDEF")
SCRYPT_PARAMS = {"n": 16384, "r": 8, "p": 1}


class IVTModel:
    """Represents a Index Virtual Terminal Worldstate. Only one instance of this class is created per server."""

    def __init__(self) -> None:
        self.users: dict[str, IVTSession] = {}
        self.subparsers: dict[str, ArgumentParser] = {}
        self.parser: ArgumentParser = self.makeParser()

        self.db = sqlite3.connect("data/akasha")

        cur = self.db.cursor()
        cur.execute("SELECT value FROM config WHERE key = 'light_url'")
        self.light_url = cur.fetchone()[0]

    async def newSession(self, websocket: ServerConnection) -> None:
        name = self.genNewAnonName()
        session = IVTSession(self, name, websocket)
        self.users[name] = session
        await session.mainloop()

    def genNewAnonName(self) -> str:
        name = "GUEST" + str(randint(0, 99)).zfill(2) + choice(ascii_uppercase)
        if name in self.users:
            return self.genNewAnonName()
        return name

    def changeUsername(self, oldname: str, newname: str) -> None:
        if oldname not in self.users:
            raise IVTException(f"unable to find user '{oldname}'")
        if newname in self.users:
            raise IVTException(f"user '{newname}' already exists")
        if oldname == newname:
            return
        self.users[newname] = self.users[oldname]
        self.users[newname].username = newname
        del self.users[oldname]

    def addSubparser(
        self, subparsers: _SubParsersAction[ArgumentParser], name: str
    ) -> ArgumentParser:
        subparser = subparsers.add_parser(
            name,
            suggest_on_error=False,
            exit_on_error=False,
            add_help=False,
            color=False,
        )
        self.subparsers[name] = subparser
        return subparser

    async def main(self) -> None:
        print("Starting INDEX Server...")
        try:
            async with serve(self.newSession, "localhost", 5417) as server:
                await server.serve_forever()
        except (InterruptedError, asyncio.exceptions.CancelledError):
            print("Exiting.")

    def makeParser(self) -> ArgumentParser:
        parser = ArgumentParser(
            "ivt",
            suggest_on_error=False,
            exit_on_error=False,
            add_help=False,
            color=False,
        )
        subparsers = parser.add_subparsers(required=True, dest="cmd")

        self.addSubparser(subparsers, "help")
        self.addSubparser(subparsers, "list")

        parser_usage = self.addSubparser(subparsers, "usage")
        parser_usage.add_argument("subparser", type=str, nargs="?")

        self.addSubparser(subparsers, "clear")

        parser_login = self.addSubparser(subparsers, "login")
        parser_login.add_argument("username", type=str)
        parser_login.add_argument("-r", "--register", action="store_true")

        parser_lighting = self.addSubparser(subparsers, "lighting")
        parser_lighting.add_argument("state", type=str, choices=["on", "off"])
        parser_lighting.add_argument("-l", "--light", type=int, default=5)
        parser_lighting.add_argument(
            "-c", "--color", type=str, nargs="?", const="#FF00AA", default="NIL"
        )

        parser_music = self.addSubparser(subparsers, "music")
        parser_music.add_argument("album", type=str, nargs="?")
        parser_music.add_argument("song", type=str, nargs="?")

        return parser

    def parse(self, line: str) -> ArgumentError | Namespace:
        try:
            return self.parser.parse_args(line.split())
        except ArgumentError as error:
            return error

    def checkPerms(self, username: str, command: str) -> bool:
        cur = self.db.cursor()
        res = cur.execute("SELECT public FROM commands WHERE command = ?", (command,))
        cmddata = res.fetchone()
        if not cmddata:
            return False
        if cmddata[0]:
            return True
        res = cur.execute(
            "SELECT 1 FROM perms WHERE username = ? AND command = ?",
            (username, command),
        )
        return res.fetchone() is not None

    def tool_disable_light(self, light: int):
        requests.put(
            f"{self.light_url}/lights/{light}/state",
            json={"on": False},
            timeout=5,
        )

    def tool_enable_light(self, light: int):
        requests.put(
            f"{self.light_url}/lights/{light}/state",
            json={"on": True, "ct": 366},
            timeout=5,
        )

    def tool_color_light(self, light: int, color: str):
        if color[0] == "#":
            color = color[1:]
        color = color.upper()
        if len(color) != 6 or not all(c in HEX_CHARS for c in color):
            color = "FF00AA"
        rgb = [int(color[i : i + 2], 16) / 255 for i in range(0, 6, 2)]
        hsv = rgb_to_hsv(*rgb)
        requests.put(
            f"{self.light_url}/lights/{light}/state",
            json={
                "on": True,
                "hue": int(hsv[0] * 65535),
                "sat": int(hsv[1] * 254),
                "bri": int(hsv[2] * 254),
            },
            timeout=5,
        )


class IVTSession:
    """Represents a Index Virtual Terminal Session. A new instance of this class is created for each connection."""

    def __init__(
        self, parent: IVTModel, username: str, websocket: ServerConnection
    ) -> None:
        self.parent = parent
        self.db = self.parent.db
        self.username = username
        self.socket = websocket
        self.pwdreturn: str | None = None
        self.pwddata: Namespace | None = None

    def changeUsername(self, newname: str) -> None:
        self.parent.changeUsername(self.username, newname)

    def checkPerms(self, command: str) -> bool:
        return self.parent.checkPerms(self.username, command)

    async def mainloop(self) -> None:
        await self.socket.send(dumps({"act": "activate", "username": self.username}))
        await self.send_text(f"Connection established; Logged in as {self.username}.")
        await self.send_display(
            [
                {"txt": "For more information, please see "},
                {"tag": "button", "txt": "help", "term": "help"},
                {"txt": ", "},
                {"tag": "button", "txt": "list", "term": "list"},
                {"txt": ", or "},
                {"tag": "button", "txt": "usage", "term": "usage"},
                {"txt": "."},
            ]
        )
        async for message in self.socket:
            if not isinstance(message, str):
                await self.send_error("incorrect data type")
                continue
            await self.process(message)

    async def process(self, line: str) -> None:
        if self.pwdreturn and self.pwddata:
            match self.pwdreturn:
                case "register":
                    await self.register_user(self.pwddata.username.upper(), line)
                case "login":
                    await self.login_user(self.pwddata.username.upper(), line)
            self.pwdreturn = None
            self.pwddata = None
            return
        result = self.parent.parse(line)
        if isinstance(result, ArgumentError):
            await self.send_error(result.message)
            return
        if not self.checkPerms(result.cmd):
            await self.send_error("permission denied")
            return
        match result.cmd:
            case "login":
                await self.login_prompt(result)
            case "usage":
                await self.display_usage(result)
            case "lighting":
                await self.set_lighting(result)
            case "music":
                await self.play_music(result)
            case "help":
                await self.display_help()
            case "list":
                await self.display_list()
            case "clear":
                await self.send_action("clear")
            case unknown:
                await self.send_error(f"command not found: {unknown}")

    async def login_prompt(self, result: Namespace):
        if result.register:
            await self.send_password_request(
                "register",
                result,
                f"REGISTER/{result.username.upper()}>",
                f"*** requesting password for {result.username.upper()} ***",
            )
        else:
            await self.send_password_request(
                "login",
                result,
                f"LOGIN/{result.username.upper()}>",
                f"*** requesting password for {result.username.upper()} ***",
            )

    async def register_user(self, username: str, password: str) -> None:
        cur = self.db.cursor()
        res = cur.execute("SELECT 1 FROM users WHERE username = ?", (username,))
        if res.fetchone():
            await self.send_error("user already exists")
            return
        salt = urandom(32)
        hash = scrypt(password.encode("utf-8"), salt=salt, **SCRYPT_PARAMS)
        cur.execute("INSERT INTO users VALUES (?, ?, ?)", (username, hash, salt))
        self.db.commit()
        await self.send_text(f"User {username} registered.")

    async def login_user(self, username: str, password: str) -> None:
        cur = self.db.cursor()
        res = cur.execute(
            "SELECT hash, salt FROM users WHERE username = ?", (username,)
        )
        userdata = res.fetchone()
        if not userdata:
            await self.send_error("invalid login credentials")
            return
        salt = userdata[1]
        expected_hash = userdata[0]
        actual_hash = scrypt(password.encode("utf-8"), salt=salt, **SCRYPT_PARAMS)
        if not compare_digest(expected_hash, actual_hash):
            await self.send_error("invalid login credentials")
            return
        self.username = username
        await self.socket.send(dumps({"act": "userchange", "username": username}))
        await self.send_text(f"Login successful; Switched to user {self.username}.")

    async def display_usage(self, result: Namespace) -> None:
        if result.subparser and result.subparser in self.parent.subparsers:
            await self.send_text(
                "\n" + self.parent.subparsers[result.subparser].format_help() + "\n"
            )
        else:
            await self.send_text("\n" + self.parent.parser.format_help() + "\n")

    async def display_list(self) -> None:
        await self.send_text("*** ivt command list ***")
        for cmd in self.parent.subparsers:
            await self.send_display(
                [
                    {"txt": " - "},
                    {"tag": "button", "txt": cmd, "term": cmd},
                    {"txt": " ("},
                    {"tag": "button", "txt": "usage", "term": f"usage {cmd}"},
                    {"txt": ")"},
                ]
            )

    async def display_help(self) -> None:
        await self.send_text(
            "The Index Virtual Terminal (IVT) is a web-based terminal emulator that interfaces with other Index applications, enabling their control from a centralized location."
        )
        await self.send_display(
            [
                {
                    "txt": "Many commands here do require logging in with an authorized Index account, which is possible with the "
                },
                {"tag": "button", "txt": "login <username>", "term": "usage login"},
                {
                    "txt": " command. For a list of all valid commands (plus shortcuts to their usage page), please see "
                },
                {"tag": "button", "txt": "list", "term": "list"},
                {
                    "txt": ". Lastly, do not forget that usage for any command can be seen with the "
                },
                {"tag": "button", "txt": "usage <command>", "term": "usage usage"},
                {"txt": " command."},
            ]
        )

    async def set_lighting(self, result: Namespace) -> None:
        if result.state == "on":
            if result.color and result.color != "NIL":
                self.parent.tool_color_light(result.light, result.color)
            else:
                self.parent.tool_enable_light(result.light)
        else:
            self.parent.tool_disable_light(result.light)

    async def play_music(self, result: Namespace) -> None:
        album = result.album
        song = result.song
        if not album:
            await self.send_text("*** ivt album list ***")
            for other in listdir("music"):
                await self.send_display(
                    [
                        {"txt": " - "},
                        {"tag": "button", "txt": other, "term": f"music {other}"},
                    ]
                )
            return
        if album not in listdir("music"):
            await self.send_error("album not found")
            return
        if not song:
            await self.send_text(f"*** songs on {album} ***")
            for other in listdir(f"music/{album}"):
                await self.send_display(
                    [
                        {"txt": " - "},
                        {
                            "tag": "button",
                            "txt": other.removesuffix(".mp3"),
                            "term": f"music {album} {other.removesuffix(".mp3")}",
                        },
                    ]
                )
            return
        if f"{song}.mp3" not in listdir(f"music/{album}"):
            await self.send_error("song not found")
            return
        await self.send_text(f"Playing {album}/{song}...")
        with open(f"music/{album}/{song}.mp3", "rb") as musicfile:
            await self.send_prelude("music")
            await self.socket.send(musicfile.read())

    async def send_password_request(
        self, pwdreturn: str, pwddata: Namespace, prompt: str, text: str
    ) -> None:
        self.pwdreturn = pwdreturn
        self.pwddata = pwddata
        await self.socket.send(
            dumps({"act": "password", "prompt": prompt, "text": text})
        )

    async def send_error(self, text: str) -> None:
        await self.socket.send(dumps({"act": "text", "text": f"error: {text}"}))

    async def send_text(self, text: str) -> None:
        await self.socket.send(dumps({"act": "text", "text": text}))

    async def send_action(self, action: str) -> None:
        await self.socket.send(dumps({"act": "action", "action": action}))

    async def send_prelude(self, action: str) -> None:
        await self.socket.send(dumps({"act": "prelude", "action": action}))

    async def send_display(self, msg) -> None:
        await self.socket.send(dumps({"act": "display", "msg": msg}))


if __name__ == "__main__":
    MAIN_MODEL = IVTModel()
    asyncio.run(MAIN_MODEL.main())