calebburke.dev/bin/dev

#!/usr/bin/env python3

import os
import re
import subprocess
import sys

# Support dashes in command names
COMMAND_TO_METHOD = {
    "ts-node": "ts_node",
    "check-types": "check_types",
    "bash-completions": "bash_completions",
    "plantuml-to-png": "plantuml_to_png",
}
METHOD_TO_COMMAND = {method: command for command, method in COMMAND_TO_METHOD.items()}

WAIT_FOR_PROCESS = "wait_for_process"


class DevHelper:
    def __init__(self):
        self.last_exit_status = 0

    # Core logic
    def call(self, *args, **kwargs):
        if not args:
            return self.compose(*args, **kwargs)

        command, rest = args[0], args[1:]
        method_name = COMMAND_TO_METHOD.get(command, command)
        method = getattr(self, method_name, None)
        if callable(method) and not method_name.startswith("_") and method_name != "call":
            return method(*rest, **kwargs)
        return self.compose(*args, **kwargs)

    def compose(self, *args, **kwargs):
        command = self._compose_command(*args, **kwargs)
        if not kwargs.get("silent"):
            print(f"Running: {command}")

        if kwargs.get("execution_mode") == WAIT_FOR_PROCESS:
            self._wait_for_process_with_logging(command)
        else:
            os.execv("/bin/sh", ["/bin/sh", "-c", command])

    # Primary command wrappers
    def build(self, *args, **kwargs):
        self.compose("build", *args, **kwargs)

    def compile(self, *args, **kwargs):
        self.run("api", "npm", "run", "build", execution_mode=WAIT_FOR_PROCESS)
        if self.last_exit_status != 0:
            sys.exit(self.last_exit_status)
        self.run("web", "npm", "run", "build")

    def up(self, *args, **kwargs):
        self.compose("up", "--remove-orphans", *args, **kwargs)

    def down(self, *args, **kwargs):
        self.compose("down", "--remove-orphans", *args, **kwargs)

    def logs(self, *args, **kwargs):
        self.compose("logs", "-f", *args, **kwargs)

    def run(self, *args, **kwargs):
        self.compose("run", "--rm", *args, **kwargs)

    def ps(self, *args, **kwargs):
        self.compose("ps", *args, **kwargs)

    # Custom helpers
    def api(self, *args, **kwargs):
        self.run("api", *args, **kwargs)

    def web(self, *args, **kwargs):
        self.run("web", *args, **kwargs)

    def check_types(self, *args, **kwargs):
        self.run("api", "npm", "run", "check-types", *args, **kwargs)

    def test(self, *args, **kwargs):
        service = args[0] if args else None
        if service == "api":
            self.test_api(*args[1:], **kwargs)
        elif service == "web":
            self.test_web(*args[1:], **kwargs)
        else:
            self.test_api(*args, **kwargs)

    def test_api(self, *args, **kwargs):
        args = self._reformat_project_relative_path_filter_for_vitest(list(args), "api/")
        self.run("test_api", "npm", "run", "test", *args, **kwargs)

    def test_web(self, *args, **kwargs):
        args = self._reformat_project_relative_path_filter_for_vitest(list(args), "web/")
        self.run("test_web", "npm", "run", "test", *args, **kwargs)

    def sqlcmd(self, *args, **kwargs):
        db_host = os.environ.get("DB_HOST", "localhost")
        db_user = os.environ.get("DB_USER", "sa")
        db_pass = os.environ.get("DB_PASS", "1m5ecure!")
        db_name = os.environ.get("DB_NAME", "YHSI")
        self.compose(
            "exec",
            "db",
            "/opt/mssql-tools/bin/sqlcmd",
            "-U",
            db_user,
            "-P",
            db_pass,
            "-H",
            db_host,
            "-d",
            db_name,
            "-I",  # enable quoted identifiers, e.g. "table"."column"
            *args,
            **kwargs,
        )

    def db(self, *args, **kwargs):
        self.compose("exec", "db", *args, **kwargs)

    def debug(self, *args, **kwargs):
        container_id = self._container_id("api")
        print("Waiting for breakpoint to trigger...")
        print("'ctrl-c' to exit.")
        command = f"docker attach --detach-keys ctrl-c {container_id}"
        print(f"Running: {command}")
        os.execv("/bin/sh", ["/bin/sh", "-c", command])

    def npm(self, *args, **kwargs):
        self.run("api", "npm", *args, **kwargs)

    def ts_node(self, *args, **kwargs):
        self.run("api", "npm", "run", "ts-node", *args, **kwargs)

    def knex(self, *args, **kwargs):
        if sys.platform.startswith("linux"):
            self.run("api", "npm", "run", "knex", *args, execution_mode=WAIT_FOR_PROCESS, **kwargs)

            file_or_directory = os.path.join(self._project_root(), "api/src/db/migrations")
            if not self._take_over_needed(file_or_directory):
                sys.exit(0)

            self.ownit(file_or_directory)
        else:
            self.run("api", "npm", "run", "knex", *args, **kwargs)

    def migrate(self, *args, **kwargs):
        action = args[0] if args else None
        self.knex(f"migrate:{action}", *args[1:], **kwargs)

    def seed(self, *args, **kwargs):
        action = args[0] if args else None
        self.knex(f"seed:{action}", *args[1:], **kwargs)

    def ownit(self, *args, **kwargs):
        file_or_directory = args[0] if args else None
        if file_or_directory is None:
            raise ValueError("Must provide a file or directory path.")

        if sys.platform.startswith("linux"):
            print(f"Take ownership of the file or directory? {file_or_directory}")
            command = f"sudo chown -R {self._user_id()}:{self._group_id()} {file_or_directory}"
            os.execv("/bin/sh", ["/bin/sh", "-c", command])
        else:
            raise NotImplementedError(f"Not implemented for platform {sys.platform}")

    def plantuml_to_png(self, *args, **kwargs):
        args = list(args)
        if not args:
            raise ValueError("Must provide a file path.")
        file_path = args.pop()

        png_path = re.sub(r"\.(wsd|pu|puml|plantuml|uml)$", ".png", file_path)

        command = f"curl {' '.join(args)} --data-binary @'{file_path}' http://localhost:9999/png > '{png_path}'"

        print(f"Running: {command}")
        os.execv("/bin/sh", ["/bin/sh", "-c", command])

    def bash_completions(self, *args, **kwargs):
        completions = sorted(
            METHOD_TO_COMMAND.get(name, name)
            for name in vars(DevHelper)
            if not name.startswith("_") and name != "call" and callable(getattr(DevHelper, name))
        )
        print(" ".join(completions))

    # Private helpers
    def _wait_for_process_with_logging(self, command):
        process = subprocess.Popen(
            ["/bin/sh", "-c", f"{command} 2>&1"],
            stdout=subprocess.PIPE,
            text=True,
        )
        for line in process.stdout:
            print(line, end="")
        process.wait()
        self.last_exit_status = process.returncode

    def _container_id(self, container_name, *args, **kwargs):
        command = self._compose_command("ps", "-q", container_name, *args, **kwargs)
        print(f"Running: {command}")
        result = subprocess.run(["/bin/sh", "-c", command], stdout=subprocess.PIPE, text=True)
        container_id = result.stdout.strip()
        print(f"Container id is: {container_id}")
        return container_id

    def _service_running(self, container_name):
        self.ps("-q", "--status=running", execution_mode=WAIT_FOR_PROCESS, silent=True)
        return self.last_exit_status == 0

    def _compose_command(self, *args, **kwargs):
        environment = kwargs.get("environment", "development")
        return f"cd {self._project_root()} && docker compose -f docker-compose.{environment}.yml {' '.join(args)}"

    def _project_root(self):
        return os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))

    def _take_over_needed(self, file_or_directory):
        result = subprocess.run(
            ["/bin/sh", "-c", f"find {file_or_directory} -not -user {self._user_id()} -print -quit | grep -q ."],
        )
        return result.returncode == 0

    def _user_id(self):
        if not sys.platform.startswith("linux"):
            raise NotImplementedError(f"Not implemented for platform {sys.platform}")
        return subprocess.run(["id", "-u"], stdout=subprocess.PIPE, text=True).stdout.strip()

    def _group_id(self):
        if not sys.platform.startswith("linux"):
            raise NotImplementedError(f"Not implemented for platform {sys.platform}")
        return subprocess.run(["id", "-g"], stdout=subprocess.PIPE, text=True).stdout.strip()

    def _reformat_project_relative_path_filter_for_vitest(self, args, prefix):
        if args and args[0].startswith(prefix):
            src_path_prefix = f"{prefix}src/"

            if args[0].startswith(src_path_prefix):
                # TODO: handle other file types
                args[0] = re.sub(r"\.ts$", ".test.ts", re.sub(f"^{re.escape(src_path_prefix)}", "tests/", args[0]))
            else:
                args[0] = re.sub(f"^{re.escape(prefix)}", "", args[0])

            print("Reformatted path filter from project relative to service relative for vitest.")

        return args


if __name__ == "__main__":
    DevHelper().call(*sys.argv[1:])