compiler.py

"""Compile a .bbproj block tree into a runnable nerimity.js bot, server-side.

This is the whole security model. The client uploads *block structure* — opcodes, literal
field values and nesting — never JavaScript. Every line of emitted code comes from
catalog.json, which ships with the host and is not user-writable. An unknown opcode is
rejected outright, and every literal is escaped through json.dumps or validated as a
number, so a field value cannot break out of its expression and become code.

Ported from the editor's original GDScript generator; catalog.json is generated from the
same catalogue the editor renders from, so the two cannot drift.
"""
from __future__ import annotations

import json
import re
from pathlib import Path
from typing import Any

CATALOG_PATH = Path(__file__).parent / "catalog.json"
INDENT = "  "

# event name -> (handler signature, __ctx body)
EVENT_CONTEXTS: dict[str, tuple[str, str]] = {
    "Ready": ("()", "{}"),
    "MessageCreate": ("(message)",
        "{ message, channel: message?.channel, user: message?.user, "
        "member: message?.member, server: message?.channel?.server }"),
    "MessageUpdate": ("(message)",
        "{ message, channel: message?.channel, user: message?.user, "
        "member: message?.member, server: message?.channel?.server }"),
    "MessageDelete": ("(payload)",
        "{ channel: client.channels?.cache?.get(payload?.channelId) }"),
    "MessageButtonClick": ("(button)",
        "{ button, message: button?.message, user: button?.user, channel: button?.channel }"),
    "MessageReactionAdded": ("(reaction, user)",
        "{ reaction, user, message: reaction?.message, channel: reaction?.channel }"),
    "MessageReactionRemoved": ("(reaction, user)",
        "{ reaction, user, message: reaction?.message, channel: reaction?.channel }"),
    "ServerMemberJoined": ("(member)", "{ member, user: member?.user, server: member?.server }"),
    "ServerMemberLeft": ("(member)", "{ member, user: member?.user, server: member?.server }"),
}

MAX_BLOCKS = 2000       # a project this large is a bug or an attack, not a bot
MAX_DEPTH = 60
MAX_COMMAND_OPTIONS = 20    # keeps a command hat's preamble bounded

# Command options are written the way people write lists: "count sides" or "count, sides".
OPTION_SEPARATOR = re.compile(r"[\s,]+")


class CompileError(Exception):
    """Raised for anything malformed; the message goes back to the editor verbatim."""


class Compiler:
    def __init__(self, catalog: dict | None = None) -> None:
        if catalog is None:
            catalog = json.loads(CATALOG_PATH.read_text())
        self.blocks: dict[str, dict] = catalog["blocks"]
        self.variables: dict[str, str] = {}
        self.lists: dict[str, str] = {}
        self.warnings: list[str] = []
        self._count = 0

    # -- helpers ------------------------------------------------------------

    def spec(self, opcode: str) -> dict:
        spec = self.blocks.get(opcode)
        if spec is None:
            raise CompileError(f"unknown block '{opcode}'")
        return spec

    @staticmethod
    def _ident(prefix: str, name: str) -> str:
        safe = re.sub(r"[^a-z0-9_]", "_", str(name).lower())
        if not safe or safe[0].isdigit():
            safe = "x" + safe
        return prefix + safe

    def _arg_def(self, spec: dict, name: str) -> dict | None:
        for a in spec["args"]:
            if a["name"] == name:
                return a
        return None

    @staticmethod
    def _command_options(node: dict) -> list[str]:
        """The option names declared on a command hat, in order and de-duplicated.

        Only the literal field is read. A reporter plugged into OPTIONS is ignored — option
        names have to be known while compiling, because each one becomes a variable.
        """
        raw = str((node.get("fields") or {}).get("OPTIONS", "") or "").strip()
        if not raw:
            return []
        out: list[str] = []
        for name in OPTION_SEPARATOR.split(raw):
            if name and name not in out:
                out.append(name)
        return out[:MAX_COMMAND_OPTIONS]

    # -- literals -----------------------------------------------------------

    def _literal(self, spec: dict, arg: dict, fields: dict) -> str:
        raw = fields.get(arg["name"], arg["default"])
        raw = "" if raw is None else str(raw)
        kind = arg["type"]

        if kind == "variable":
            self.variables.setdefault(raw, self._ident("v_", raw))
            return self.variables[raw]
        if kind == "list":
            self.lists.setdefault(raw, self._ident("l_", raw))
            return self.lists[raw]
        if kind == "boolean":
            return "false"                      # empty hexagonal socket
        if kind == "dropdown":
            # Only values the catalogue offers; never echo an arbitrary string back into
            # the template, even though json.dumps would already neutralise it.
            options = arg.get("options") or []
            if options and raw not in options:
                raw = str(arg["default"] or options[0])
            return json.dumps(raw)
        if kind == "number":
            try:
                float(raw)
                return raw
            except ValueError:
                return json.dumps(raw)
        return json.dumps(raw)

    def _compile_arg(self, spec: dict, arg: dict, node: dict, depth: int) -> str:
        child = (node.get("inputs") or {}).get(arg["name"])
        if child:
            return self._expression(child, depth + 1)
        return self._literal(spec, arg, node.get("fields") or {})

    def _substitute(self, spec: dict, template: str, node: dict, depth: int) -> str:
        out = template
        for arg in spec["args"]:
            if arg["type"] == "substack":
                continue
            out = out.replace("{%s}" % arg["name"], self._compile_arg(spec, arg, node, depth))
        return out

    # -- expressions --------------------------------------------------------

    def _expression(self, node: dict, depth: int) -> str:
        self._guard_limits(depth)
        spec = self.spec(node.get("opcode", ""))
        js = self._substitute(spec, spec["js"], node, depth)
        if js.startswith("await "):
            return "(%s)" % js
        return js

    # -- statements ---------------------------------------------------------

    def _guard_limits(self, depth: int) -> None:
        self._count += 1
        if self._count > MAX_BLOCKS:
            raise CompileError(f"project has more than {MAX_BLOCKS} blocks")
        if depth > MAX_DEPTH:
            raise CompileError(f"blocks nested deeper than {MAX_DEPTH}")

    def _stack(self, node: dict | None, depth: int) -> str:
        lines: list[str] = []
        current = node
        seen = 0
        while current:
            piece = self._statement(current, depth)
            if piece:
                lines.append(piece)
            current = current.get("next")
            seen += 1
            if seen > MAX_BLOCKS:
                raise CompileError("block chain is impossibly long")
        return "\n".join(lines)

    def _statement(self, node: dict, depth: int) -> str:
        self._guard_limits(depth)
        spec = self.spec(node.get("opcode", ""))
        if not spec["js"].strip():
            return ""

        pad = INDENT * depth
        template = self._substitute(spec, spec["js"], node, depth)

        substack_index = 0
        for arg in spec["args"]:
            if arg["type"] != "substack":
                continue
            token = "{SUBSTACK}" if substack_index == 0 else "{SUBSTACK%d}" % (substack_index + 1)
            inner = self._stack((node.get("substacks") or {}).get(arg["name"]), depth + 1)
            template = template.replace(token, inner)
            substack_index += 1

        if node.get("opcode") == "control_stop":
            if str((node.get("fields") or {}).get("WHAT", "this script")) == "all":
                template = "process.exit(0);"

        block_id = self._safe_id(node.get("id", ""))
        out = ['%s__bb.step("%s");' % (pad, block_id)]
        for line in template.split("\n"):
            if not line.strip():
                continue
            out.append(line if line.startswith(INDENT * (depth + 1)) else pad + line)
        return "\n".join(out)

    @staticmethod
    def _safe_id(block_id: Any) -> str:
        # Block ids are echoed into a string literal for error attribution; keep them to a
        # character set that cannot terminate it.
        return re.sub(r"[^A-Za-z0-9_.-]", "", str(block_id))[:64]

    # -- top level ----------------------------------------------------------

    def _collect_data(self, node: dict | None) -> None:
        if not node:
            return
        spec = self.spec(node.get("opcode", ""))
        fields = node.get("fields") or {}
        for arg in spec["args"]:
            value = str(fields.get(arg["name"], arg["default"]) or "")
            if arg["type"] == "variable":
                self.variables.setdefault(value, self._ident("v_", value))
            elif arg["type"] == "list":
                self.lists.setdefault(value, self._ident("l_", value))
        # A command hat's options are variables too, declared by naming them rather than by
        # sitting in a variable slot.
        if node.get("opcode") == "event_command":
            for option in self._command_options(node):
                self.variables.setdefault(option, self._ident("v_", option))
        for child in (node.get("inputs") or {}).values():
            self._collect_data(child)
        for child in (node.get("substacks") or {}).values():
            self._collect_data(child)
        self._collect_data(node.get("next"))

    def _hat(self, node: dict) -> str:
        spec = self.spec(node["opcode"])
        body = self._stack(node.get("next"), 1)

        if spec["event"] == "__broadcast":
            name_arg = self._arg_def(spec, "NAME")
            name = self._literal(spec, name_arg, node.get("fields") or {}) if name_arg else "''"
            lines = ["__bb.onBroadcast(%s, __bb.handler('broadcast', async () => {" % name,
                     INDENT + "const __ctx = {};"]
            if body:
                lines.append(body)
            lines.append("}));")
            return "\n".join(lines)

        signature, ctx = EVENT_CONTEXTS.get(spec["event"], ("()", "{}"))
        lines = ["client.on(Events.%s, __bb.handler('%s', async %s => {"
                 % (spec["event"], node["opcode"], signature),
                 INDENT + "const __ctx = %s;" % ctx]
        if spec["event"] == "Ready":
            lines.append(INDENT + "__bb.ready(client);")
        else:
            lines.append(INDENT + "if (__ctx.user && client.user && "
                                  "__ctx.user.id === client.user.id) return;")
        if spec["guard"]:
            lines.append(INDENT + "if (!(%s)) return;"
                         % self._substitute(spec, spec["guard"], node, 0))
        # Options are read after the guard, so they only overwrite their variables when this
        # command is the one that actually ran.
        if node["opcode"] == "event_command":
            options = self._command_options(node)
            if options:
                lines.append(INDENT + "const __args = __ctx.message?.command?.args ?? [];")
                for index, option in enumerate(options):
                    ident = self.variables.setdefault(option, self._ident("v_", option))
                    lines.append(INDENT + "%s = __bb.commandOption(__args, %d, %d);"
                                 % (ident, index, len(options)))
        if body:
            lines.append(body)
        lines.append("}));")
        return "\n".join(lines)

    def _command_registration(self, scripts: list[dict]) -> str:
        entries: list[str] = []
        seen: set[str] = set()
        for node in scripts:
            if node.get("opcode") != "event_command":
                continue
            if (node.get("inputs") or {}).get("OPTIONS"):
                self.warnings.append(
                    "A command hat uses a block for its options. Option names have to be "
                    "plain text, because each one becomes a variable — that command will "
                    "run with no options.")
            if (node.get("inputs") or {}).get("NAME"):
                self.warnings.append(
                    "A command hat uses a block for its name, so it can't be registered "
                    "automatically — type it as plain text to make it appear in Nerimity's "
                    "command list.")
                continue
            fields = node.get("fields") or {}
            name = str(fields.get("NAME", "")).strip().lstrip("/")
            if not name or name in seen:
                continue
            seen.add(name)
            help_text = str(fields.get("HELP", "")).strip() or name
            # Nerimity shows `args` beside the command in its picker, so the option names
            # double as the usage hint people read before typing.
            usage = " ".join(self._command_options(node))
            entries.append("    { name: %s, description: %s, args: %s }"
                           % (json.dumps(name), json.dumps(help_text), json.dumps(usage)))
        if not entries:
            return ""
        return "\n".join([
            "client.on(Events.Ready, __bb.handler('register_commands', async () => {",
            "  await __bb.registerCommands(client, process.env.NERIMITY_TOKEN, [",
            ",\n".join(entries),
            "  ]);",
            "}));",
        ])

    def compile(self, project: dict, name: str = "bot") -> str:
        scripts = project.get("scripts")
        if not isinstance(scripts, list):
            raise CompileError("project has no scripts array")

        self.variables.clear()
        self.lists.clear()
        self.warnings.clear()
        self._count = 0

        for node in scripts:
            if not isinstance(node, dict):
                raise CompileError("a script entry is not an object")
            self._collect_data(node)

        handlers: list[str] = []
        orphans = 0
        for node in scripts:
            if self.spec(node.get("opcode", ""))["shape"] != "hat":
                orphans += 1
                continue
            handlers.append(self._hat(node))

        if orphans:
            self.warnings.append(
                "%d loose block stack%s no event block on top, so %s never run."
                % (orphans, " has" if orphans == 1 else "s have",
                   "it will" if orphans == 1 else "they will"))
        if not handlers:
            self.warnings.append(
                "This project has no event blocks, so the bot will connect and do nothing.")

        out = [
            "'use strict';",
            "// Generated by the bbuilder host from block structure — do not edit.",
            "// project: %s" % re.sub(r"[\r\n]", " ", str(name))[:80],
            "",
            "const { Client, Events } = require('@nerimity/nerimity.js');",
            "const __bb = require('./bb-runtime.js');",
            "",
            "const client = new Client();",
            "",
        ]
        if self.variables or self.lists:
            out.append("// --- variables ---")
            # Two different names can sanitise to the same identifier ("my var" and "my-var",
            # or an option and a variable spelled with different capitals). They then share
            # one variable, which is the sane reading — but it must only be declared once.
            out += ["let %s = 0;" % v for v in dict.fromkeys(self.variables.values())]
            out += ["let %s = [];" % v for v in dict.fromkeys(self.lists.values())]
            out.append("")

        registration = self._command_registration(scripts)
        if registration:
            out.append("// --- slash command registration ---")
            out.append(registration)
            out.append("")

        for handler in handlers:
            out.append(handler)
            out.append("")

        out.append("client.login(process.env.NERIMITY_TOKEN);")
        out.append("")
        return "\n".join(out)


def compile_project(project: dict, name: str = "bot") -> tuple[str, list[str]]:
    c = Compiler()
    return c.compile(project, name), list(c.warnings)