Core¶

async def get_state(request: Request) -> BackendState:
    return request.app.state.backend_state  # type: ignore

def get_distribution_path(dist_name: DistributionName, module_name: ModuleName) -> Path:
    """Get a distribution's source root or installed import-package root.

    Editable installs resolve to the project directory from `direct_url.json`.
    Installed distributions resolve from their metadata, optionally using an
    entry-point module to disambiguate distributions.

    See: https://packaging.python.org/en/latest/specifications/direct-url-data-structure/
    """
    # always try direct_url.json first (e.g. for the case of `uv sync --all-packages`)
    try:
        dist = Distribution.from_name(dist_name)
    except PackageNotFoundError as exc:
        raise FileNotFoundError(
            f"could not find distribution path for {dist_name}"
        ) from exc

    try:
        if direct_url_content := dist.read_text("direct_url.json"):
            direct_url_data = json.loads(direct_url_content)
            if (
                (url := direct_url_data.get("url"))
                # url may point to a git or zip archive, but since we only care
                # about local paths, we only handle the file:// scheme here
                and url.startswith("file://")
            ):
                parsed = urllib.parse.urlparse(url)
                if os.name == "nt":
                    path_str = urllib.request.url2pathname(parsed.path)
                    if parsed.netloc and parsed.netloc not in ("", "localhost"):
                        path_str = f"//{parsed.netloc}{path_str}"
                    project_path = Path(path_str)
                else:
                    project_path = Path(urllib.parse.unquote(parsed.path))
                if project_path.is_dir():
                    return project_path
    except (json.JSONDecodeError, FileNotFoundError):
        pass

    package_name = module_name.partition(".")[0]
    package_path = Path(str(dist.locate_file(package_name)))
    if package_path.is_dir():
        return package_path

    module_path = package_path.with_suffix(".py")
    if module_path.is_file():
        return module_path.parent

    raise FileNotFoundError(
        f"could not find module {module_name} in distribution {dist_name}"
    )

def resolve_frontend(plugin: Plugin) -> Path | None:
    if not plugin.frontend_path:
        return None
    return (
        get_distribution_path(plugin.dist_name, plugin.module_name)
        / plugin.frontend_path
    )

def load_enabled_plugins(
    config: Config,
) -> list[Plugin]:
    loaded_plugins = []
    enabled_plugin_names = set(config.core.plugins)

    for entry_point in scan_plugins():
        plugin_id: PluginId = entry_point.name
        if plugin_id not in enabled_plugin_names:
            continue
        if (plugin := load_plugin(entry_point)) is not None:
            loaded_plugins.append(plugin)

    return loaded_plugins

@asynccontextmanager
async def lifespan(
    app: FastAPI, backend_state: BackendState, loaded_plugins: Iterable[Plugin]
) -> AsyncGenerator[None, None]:
    async with AsyncExitStack() as stack:
        for plugin in loaded_plugins:
            if plugin.lifespan:
                logger.info(f"initializing lifespan for {plugin.id}")
                await stack.enter_async_context(plugin.lifespan(backend_state))

        app.state.backend_state = backend_state
        yield

def default_cache_dir() -> Path:
    if (xdg_cache := os.environ.get("XDG_CACHE_HOME")) is not None:
        cache_dir = Path(xdg_cache) / "tangram"
    else:
        cache_dir = Path(platformdirs.user_cache_dir(appname="tangram"))
    if not cache_dir.exists():
        cache_dir.mkdir(parents=True, exist_ok=True)

    return cache_dir

def make_cache_route_handler(
    entry: CacheEntry, state: BackendState
) -> Callable[..., Awaitable[FileResponse]]:
    """
    Factory function that creates a route handler for caching and serving files.
    Dynamically handles URL parameters found in both serve_route and origin.

    :param entry: Cache entry configuration
    :param state: Backend state with http_client for fetching remote resources
    :returns: Async function that handles the route with dynamic parameters
    """
    from inspect import Parameter, Signature

    # Extract parameter names from the serve_route (e.g., {fontstack}, {range})
    params = CACHE_PARAM_PATTERN.findall(entry.serve_route)

    async def cache_route_handler(**kwargs: str) -> FileResponse:
        if (local_path := entry.local_path) is None:
            local_path = default_cache_dir()
        else:
            local_path = local_path.expanduser()

        # Build the local file path by replacing parameters
        local_file = local_path
        for param in params:
            if param in kwargs:
                local_file = local_file / kwargs[param]

        logger.info(f"Serving cached file from {local_file}")

        if not local_file.exists():
            assert entry.origin is not None
            # Build the remote URL by replacing parameters
            remote_url = entry.origin
            for param, value in kwargs.items():
                remote_url = remote_url.replace(f"{{{param}}}", value)

            logger.info(f"Downloading from {remote_url} to {local_file}")
            c = await state.http_client.get(remote_url)
            c.raise_for_status()
            local_file.parent.mkdir(parents=True, exist_ok=True)
            local_file.write_bytes(c.content)

        return FileResponse(path=local_file, media_type=entry.media_type)

    # Create explicit parameters for the function signature
    sig_params = [
        Parameter(
            name=param,
            kind=Parameter.POSITIONAL_OR_KEYWORD,
            annotation=str,
        )
        for param in params
    ]
    cache_route_handler.__signature__ = Signature(  # type: ignore
        parameters=sig_params,
        return_annotation=FileResponse,
    )

    return cache_route_handler

def core_into_frontend_config(config: Config) -> FrontendConfig:
    if config.channel.public_url:
        channel_url = config.channel.public_url
    else:
        host = "localhost" if config.channel.host == "0.0.0.0" else config.channel.host
        channel_url = f"http://{host}:{config.channel.port}"
    frontend_channel = FrontendChannelConfig(url=channel_url)

    user_theme_names = {t.name for t in config.core.themes}
    merged_themes = [
        t for t in DEFAULT_THEMES if t.name not in user_theme_names
    ] + config.core.themes

    frontend_core = FrontendCoreConfig(theme=config.core.theme, themes=merged_themes)
    frontend_map = replace(
        config.map,
        styles=merge_styles(default_styles(), config.map.styles),
    )

    return FrontendConfig(
        core=frontend_core, map=frontend_map, channel=frontend_channel
    )

def create_app(
    backend_state: BackendState,
    loaded_plugins: Iterable[Plugin],
) -> FastAPI:
    app = FastAPI(
        lifespan=partial(
            lifespan, backend_state=backend_state, loaded_plugins=loaded_plugins
        ),
        default_response_class=ORJSONResponse,
    )

    frontend_config_instance = core_into_frontend_config(backend_state.config)
    frontend_adapter = TypeAdapter(FrontendConfig)
    manifest_core = to_frontend_manifest(frontend_adapter, frontend_config_instance)

    manifest_plugins = {}
    for plugin in loaded_plugins:
        backend_state.loaded_plugins[plugin.id] = plugin
        for router in plugin.routers:
            app.include_router(router)

        if (frontend_path_resolved := resolve_frontend(plugin)) is None:
            continue
        app.mount(
            f"/plugins/{plugin.id}",
            StaticFiles(directory=str(frontend_path_resolved)),
            name=plugin.id,
        )
        if not (plugin_json_path := frontend_path_resolved / "plugin.json").exists():
            continue
        with plugin_json_path.open("rb") as f:
            plugin_meta = json.load(f)

        manifest_plugins[plugin.id] = plugin_meta

        # NOTE: on the js side the `config` and `config_json_schema` keys
        # may be missing if the plugin does not configure them properly
        conf_dict = backend_state.config.plugins.get(plugin.id, {})
        if (backend_adapter := plugin.adapter()) is None:
            continue
        conf_instance = backend_adapter.validate_python(conf_dict)
        if not plugin.frontend_config_class:
            continue
        if plugin.into_frontend_config_function is None:
            logger.warning(
                f"expected `into_frontend_config_function` for"
                f" plugin {plugin.id} but it is None. "
                "Frontend config will be empty!"
            )
            continue
        frontend_instance = plugin.into_frontend_config_function(conf_instance)
        frontend_adapter_ = plugin.frontend_adapter()
        assert frontend_adapter_ is not None
        frontend_manifest = to_frontend_manifest(frontend_adapter_, frontend_instance)
        manifest_plugins[plugin.id] = {
            **plugin_meta,
            **frontend_manifest,
        }

    @app.post("/settings/validate/{plugin_id}")
    async def validate_settings(
        plugin_id: str,
        data: dict[str, Any],
        state: Annotated[BackendState, Depends(get_state)],
    ) -> dict[str, Any]:
        SUCCESS = {"success": True, "errors": {}}
        try:
            if plugin_id == "tangram_core":
                _ = parse_frontend_config(frontend_adapter, data)
                return SUCCESS
            plugin = state.loaded_plugins.get(plugin_id)
            assert plugin is not None, f"plugin {plugin_id} not found"

            if plugin.frontend_config_class:
                if (frontend_adapter_ := plugin.frontend_adapter()) is not None:
                    frontend_adapter_.validate_python(data)
                return SUCCESS

            # if no frontend config defined, validation always succeeds because
            # nothing to validate against
            return SUCCESS
        except ValidationError as e:
            errs = {
                ".".join(str(loc) for loc in err["loc"]): err["msg"]
                for err in e.errors()
            }
            return {"success": False, "errors": errs}

    @app.get("/manifest.json")
    async def get_manifest() -> ORJSONResponse:
        return ORJSONResponse(
            content={"core": manifest_core, "plugins": manifest_plugins}
        )

    # Cache mechanism - MUST be registered BEFORE the catch-all frontend mount
    for cache_entry in backend_state.config.cache.entries:
        logger.info(
            f"caching {cache_entry.origin} to {cache_entry.local_path} "
            f"and serving at {cache_entry.serve_route}"
        )
        route_handler = make_cache_route_handler(cache_entry, backend_state)

        logger.info(
            f"Registering route: GET {cache_entry.serve_route} with dynamic params"
        )
        app.add_api_route(
            cache_entry.serve_route,
            route_handler,
            methods=["GET"],
            name=f"cache-{cache_entry.serve_route.replace('/', '_')}",
        )

    if not (
        frontend_path := get_distribution_path("tangram_core", "tangram_core")
        / "dist-frontend"
    ).is_dir():
        raise ValueError(
            f"error: frontend {frontend_path} was not found, "
            "did you run `pnpm i && pnpm run build`?"
        )
    app.mount("/", StaticFiles(directory=str(frontend_path), html=True), name="core")

    return app

async def run_channel_service(config: Config) -> None:
    from . import _core

    _core.init_tracing_stderr(config.core.log_level)

    rust_config = _core.ChannelConfig(
        host=config.channel.host,
        port=config.channel.port,
        redis_url=config.core.redis_url,
        jwt_secret=config.channel.jwt_secret,
        jwt_expiration_secs=config.channel.jwt_expiration_secs,
        id_length=config.channel.id_length,
    )
    await _core.run(rust_config)

def get_log_config_dict(config: Config) -> dict[str, Any]:
    def format_time(dt: datetime) -> str:
        return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ ")

    return {
        "version": 1,
        "disable_existing_loggers": False,
        "handlers": {
            "default": {
                "class": "rich.logging.RichHandler",
                "log_time_format": format_time,
                "omit_repeated_times": False,
            },
        },
        "root": {"handlers": ["default"], "level": config.core.log_level.upper()},
    }