Dockerfile.playground

FROM python:3.13 AS base

ENV PATH=/opt/venv/bin:$PATH
ENV PYTHONUNBUFFERED=1
ENV PYTHONDONTWRITEBYTECODE=1

FROM base AS venv
WORKDIR /opt/venv
COPY *.txt .
RUN <<EOF
#!/usr/bin/env bash
set -euo pipefail

python -m venv .
. bin/activate
pip install --no-cache-dir --requirement requirements-dev.txt
EOF

FROM base
RUN mkdir -p /opt/engine /opt/game
WORKDIR /opt/venv
COPY --from=venv /opt/venv .
WORKDIR /opt/app
COPY . .
RUN <<EOF
cat <<CODE >> main.py
import asyncio
import io

import py7zr
from fastapi import APIRouter
from fastapi import Request
from fastapi import Response
from fastapi.responses import FileResponse
from rx.disposable import CompositeDisposable
from rx.operators import debounce
from rx.scheduler.eventloop import AsyncIOScheduler
from rx.subject import Subject
from starlette.types import Receive
from starlette.types import Scope
from starlette.types import Send
from watchdog.events import FileSystemEventHandler
from watchdog.events import PatternMatchingEventHandler
from watchdog.observers import Observer

def build() -> None:
    print("building...")


class FilteredEventHandler(PatternMatchingEventHandler):
    def __init__(self, notifier: Subject):
        super().__init__(patterns=["*.js", "*.wasm"], ignore_directories=True)
        self.notifier = notifier

    def on_modified(self, event):
        self.notifier.on_next(0)


class AllFilesEventHandler(FileSystemEventHandler):
    def __init__(self, notifier: Subject):
        self.notifier = notifier

    def on_modified(self, event):
        self.notifier.on_next(0)


def observe(path: str, handler: FileSystemEventHandler) -> Observer:
    observer = Observer()
    observer.schedule(handler, path, recursive=True)
    observer.start()
    return observer


@app.on_event("startup")
async def startup_event():
    loop = asyncio.get_event_loop()
    scheduler = AsyncIOScheduler(loop=loop)

    engine_notifier = Subject()
    engine_subscription = engine_notifier.pipe(debounce(3.0, scheduler=scheduler)).subscribe(lambda _: build())
    engine_handler = FilteredEventHandler(engine_notifier)
    engine_observer = observe("/opt/engine", engine_handler)

    game_notifier = Subject()
    game_subscription = game_notifier.pipe(debounce(3.0, scheduler=scheduler)).subscribe(lambda _: build())
    game_handler = AllFilesEventHandler(game_notifier)
    game_observer = observe("/opt/game", game_handler)

    app.state.observers = [engine_observer, game_observer]
    app.state.subscriptions = CompositeDisposable(engine_subscription, game_subscription)
    app.state.notifiers = [engine_notifier, game_notifier]


@app.on_event("shutdown")
async def shutdown_event():
    if hasattr(app.state, "observers"):
        for observer in app.state.observers:
            observer.stop()
            observer.join()
    if hasattr(app.state, "subscriptions"):
        app.state.subscriptions.dispose()
    if hasattr(app.state, "notifiers"):
        for notifier in app.state.notifiers:
            notifier.on_completed()

class SafeFileResponse(FileResponse):
    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        self.set_stat_headers = lambda *args, **kwargs: None
        await super().__call__(scope, receive, send)


playground = APIRouter(prefix="/playground")


@playground.get("/")
async def debug(request: Request):
    return templates.TemplateResponse("playground.html", context={"request": request})


@playground.get("/bundle.7z")
async def bundle():
    source = "/opt/game"
    stream = io.BytesIO()
    with py7zr.SevenZipFile(stream, mode="w") as archive:
        for root, dirs, files in os.walk(source):
            if ".git" in dirs:
                dirs.remove(".git")
            for file in files:
                path = os.path.join(root, file)
                arcname = os.path.relpath(path, source)
                archive.write(path, arcname)

    stream.seek(0)

    return Response(
        content=stream.read(),
        media_type="application/x-7z-compressed",
        headers={"Content-Disposition": "attachment; filename=bundle.7z"},
    )


@playground.get("/{filename}")
async def assets(filename: str):
    assets = {
        "carimbo.js": "application/javascript",
        "carimbo.wasm": "application/wasm",
    }

    path = f"/opt/engine/{filename}"
    media_type = assets.get(filename)
    if media_type:
        return SafeFileResponse(path, media_type=media_type)

    return Response(content="File not found", status_code=404)


app.include_router(playground)
CODE
EOF

ENTRYPOINT ["uvicorn"]
CMD ["main:app", "--host", "0.0.0.0", "--port", "3000", "--workers", "1"]