rose-ash/sx/app.py

from __future__ import annotations
import os
import path_setup  # noqa: F401

from bp import register_pages
from services import register_domain_services

SX_STANDALONE = os.getenv("SX_STANDALONE") == "true"


async def sx_docs_context() -> dict:
    """SX docs app context processor — fetches cross-service fragments."""
    from quart import request, g
    from shared.infrastructure.context import base_context
    from shared.infrastructure.cart_identity import current_cart_identity
    from shared.infrastructure.fragments import fetch_fragments

    ctx = await base_context()
    ctx["menu_items"] = []

    ident = current_cart_identity()
    user = getattr(g, "user", None)

    cart_params = {}
    if ident.get("user_id"):
        cart_params["user_id"] = ident["user_id"]
    if ident.get("session_id"):
        cart_params["session_id"] = ident["session_id"]

    auth_params = {"email": user.email} if user else None

    cart_mini, auth_menu, nav_tree = await fetch_fragments([
        ("cart", "cart-mini", cart_params or None),
        ("account", "auth-menu", auth_params),
        ("blog", "nav-tree", {"app_name": "sx", "path": request.path}),
    ], required=False)
    ctx["cart_mini"] = cart_mini
    ctx["auth_menu"] = auth_menu
    ctx["nav_tree"] = nav_tree

    return ctx


async def sx_standalone_context() -> dict:
    """Minimal context for standalone mode — no cross-service fragments."""
    from shared.infrastructure.context import base_context
    ctx = await base_context()
    ctx["menu_items"] = []
    ctx["cart_mini"] = ""
    ctx["auth_menu"] = ""
    ctx["nav_tree"] = ""
    return ctx


def create_app() -> "Quart":
    from shared.infrastructure.factory import create_base_app

    extra_kw = {}
    if SX_STANDALONE:
        extra_kw["no_oauth"] = True
        extra_kw["no_db"] = True

    app = create_base_app(
        "sx",
        context_fn=sx_standalone_context if SX_STANDALONE else sx_docs_context,
        domain_services_fn=register_domain_services,
        css_extras=[],  # No legacy CSS — SX uses CSSX + custom highlighting
        **extra_kw,
    )

    # Minimal shell — no Prism, no SweetAlert, no body.js
    # sx docs uses custom highlight.py, not Prism; body.js is for legacy apps
    # Load init SX from file — styles + nav behavior, no inline Python strings
    import os as _os
    _init_path = _os.path.join(_os.path.dirname(__file__), "sxc", "init-client.sx.txt")
    with open(_init_path) as _f:
        _init_sx = _f.read()
    app.config["SX_SHELL"] = {
        "head_scripts": [],  # no CDN scripts
        "body_scripts": [],  # no body.js
        "inline_head_js": "",  # no pre-boot JS (hover-capable, close-details unused)
        "inline_css": "",  # styles injected via CSSX from init.sx
        "init_sx": _init_sx,
    }

    app.url_map.strict_slashes = False

    from sxc.pages import setup_sx_pages
    setup_sx_pages()

    bp = register_pages(url_prefix="/")
    app.register_blueprint(bp)

    # Register SX-defined route handlers (defhandler with :path)
    from shared.sx.handlers import register_route_handlers
    n_routes = register_route_handlers(app, "sx")
    if n_routes:
        import logging
        logging.getLogger("sx.handlers").info(
            "Registered %d route handler(s) for sx", n_routes)

    from shared.sx.pages import auto_mount_pages
    auto_mount_pages(app, "sx")

    # --- GraphSX catch-all route: SX expression URLs ---
    from sxc.pages.sx_router import eval_sx_url, redirect_old_url

    @app.before_request
    async def sx_url_redirect():
        """Redirect old-style paths and bare root to /sx/ URLs (301)."""
        from quart import request, redirect as q_redirect
        path = request.path
        # Root → /sx/
        if path == "/":
            return q_redirect("/sx/", 301)
        # Skip non-page paths
        if path.startswith(("/static/", "/internal/", "/auth/")):
            return None
        # Skip SX expression URLs (already in new format under /sx/)
        if path.startswith("/sx/"):
            return None
        # Redirect bare /(...) to /sx/(...)
        if path.startswith("/(") or path.startswith("/~"):
            qs = request.query_string.decode()
            target = f"/sx{path}" + ("?" + qs if qs else "")
            return q_redirect(target, 301)
        new_url = redirect_old_url(path)
        if new_url:
            qs = request.query_string.decode()
            if qs:
                new_url += "?" + qs
            return q_redirect(new_url, 301)

    @app.before_request
    async def trailing_slash_redirect():
        from quart import request, redirect
        path = request.path
        # Skip SX expression URLs — they don't use trailing slashes
        if "(" in path or "/~" in path:
            return None
        if (path != "/"
                and path != "/sx/"
                and not path.endswith("/")
                and request.method == "GET"
                and not path.startswith(("/static/", "/internal/", "/auth/", "/sx/"))
                and "." not in path.rsplit("/", 1)[-1]):
            qs = request.query_string.decode()
            target = path + "/" + ("?" + qs if qs else "")
            return redirect(target, 301)

    @app.get("/sx/_engine")
    async def sx_engine_info():
        """Diagnostic: which SX engine is active."""
        import os, json
        info = {"engine": "python-ref", "ocaml": False}
        if os.environ.get("SX_USE_OCAML") == "1":
            try:
                from shared.sx.ocaml_bridge import get_bridge
                bridge = await get_bridge()
                engine = await bridge.ping()
                info = {"engine": engine, "ocaml": True, "pid": bridge._proc.pid}
            except Exception as e:
                info = {"engine": "ocaml-error", "ocaml": False, "error": str(e)}
        return json.dumps(info), 200, {"Content-Type": "application/json"}

    @app.get("/sx/")
    async def sx_home():
        """SX docs home page."""
        return await eval_sx_url("/")

    @app.get("/sx/<path:expr>")
    async def sx_eval_route(expr):
        """Catch-all: evaluate SX expression URLs under /sx/ prefix."""
        result = await eval_sx_url(f"/{expr}")
        if result is None:
            from quart import abort
            abort(404)
        return result

    @app.errorhandler(404)
    async def sx_not_found(e):
        from quart import request, make_response
        from shared.browser.app.utils.htmx import is_htmx_request
        from shared.sx.jinja_bridge import get_component_env, _get_request_context
        from shared.sx.types import Symbol, Keyword
        from shared.sx.helpers import full_page_sx, oob_page_sx, sx_response
        from shared.sx.pages import get_page_helpers
        from shared.sx.page import get_template_context
        import os

        path = request.path
        content_ast = [
            Symbol("~layouts/doc"), Keyword("path"), path,
            [Symbol("~not-found/content"), Keyword("path"), path],
        ]

        env = dict(get_component_env())
        env.update(get_page_helpers("sx"))
        ctx = _get_request_context()

        try:
            if os.environ.get("SX_USE_OCAML") == "1":
                from shared.sx.ocaml_bridge import get_bridge
                from shared.sx.parser import serialize
                bridge = await get_bridge()
                sx_text = serialize(content_ast)
                content_sx = await bridge.aser_slot(sx_text, ctx={"_helper_service": "sx"})
            else:
                from shared.sx.async_eval import async_eval_slot_to_sx
                content_sx = await async_eval_slot_to_sx(content_ast, env, ctx)
        except Exception:
            from shared.browser.app.errors import _sx_error_page
            html = _sx_error_page("404", "NOT FOUND",
                                  image="/static/errors/404.gif")
            return await make_response(html, 404)

        if is_htmx_request():
            return sx_response(
                await oob_page_sx(content=content_sx),
                status=404,
            )
        else:
            tctx = await get_template_context()
            html = await full_page_sx(tctx, header_rows="",
                                      content=content_sx)
            return await make_response(html, 404)

    return app


app = create_app()