cronekorkn/left4me
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(l4d2-web): overlay builder registry with workshop builder

Browse source 
Adds l4d2web/services/overlay_builders.py with a BUILDERS dict mapping
Overlay.type to a builder class. ExternalBuilder is a no-op that just
ensures the overlay directory exists. WorkshopBuilder diff-applies
absolute symlinks under left4dead2/addons/ against the overlay's current
WorkshopItem associations: creates new ones, removes obsolete, leaves
unrelated files alone, and skips uncached items with a warning rather
than producing dangling symlinks.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
mwiegand 2026-05-07 16:40:30 +02:00
parent f0230e17d3
commit 700940d578
No known key found for this signature in database
2 changed files with 431 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

193
l4d2web/services/overlay_builders.py Normal file
View file

@ -0,0 +1,193 @@
"""Overlay builder registry.
Each `Overlay.type` maps to a builder. `build_overlay(overlay_id)` jobs (and
the synchronous `initialize_server` hook) dispatch through `BUILDERS`. Adding
a new overlay type means writing a new builder and registering it here — no
changes to the worker, the mount layer, or the blueprint editor.
"""
from __future__ import annotations
import os
from pathlib import Path
from typing import Callable, Protocol
from sqlalchemy import select
from l4d2host.paths import get_left4me_root
from l4d2web.db import session_scope
from l4d2web.models import Overlay, OverlayWorkshopItem, WorkshopItem
from l4d2web.services.workshop_paths import cache_path, workshop_cache_root
CancelCheck = Callable[[], bool]
LogSink = Callable[[str], None]
class OverlayBuilder(Protocol):
def build(
self,
overlay: Overlay,
*,
on_stdout: LogSink,
on_stderr: LogSink,
should_cancel: CancelCheck,
) -> None: ...
def _overlay_root(overlay: Overlay) -> Path:
return get_left4me_root() / "overlays" / overlay.path
class ExternalBuilder:
"""No-op builder for admin-managed overlays. Ensures the overlay directory
exists; everything inside it is the admin's responsibility (SFTP, etc.)."""
def build(
self,
overlay: Overlay,
*,
on_stdout: LogSink,
on_stderr: LogSink,
should_cancel: CancelCheck,
) -> None:
root = _overlay_root(overlay)
root.mkdir(parents=True, exist_ok=True)
on_stdout(f"external overlay {overlay.name!r} ready at {root}")
class WorkshopBuilder:
"""Diff-apply symlinks under `left4dead2/addons/` against the overlay's
current `WorkshopItem` associations. Cached items get an absolute symlink
into `workshop_cache/{steam_id}.vpk`. Items missing from cache are
skipped with a warning rather than turned into broken symlinks."""
def build(
self,
overlay: Overlay,
*,
on_stdout: LogSink,
on_stderr: LogSink,
should_cancel: CancelCheck,
) -> None:
addons_dir = _overlay_root(overlay) / "left4dead2" / "addons"
addons_dir.mkdir(parents=True, exist_ok=True)
with session_scope() as db:
items = db.scalars(
select(WorkshopItem)
.join(
OverlayWorkshopItem,
OverlayWorkshopItem.workshop_item_id == WorkshopItem.id,
)
.where(OverlayWorkshopItem.overlay_id == overlay.id)
).all()
# Detach items so we can use them outside the session.
items_data = [
(it.steam_id, it.last_downloaded_at) for it in items
]
cache_root = workshop_cache_root()
# desired: symlink-name -> absolute target path (only for cached items)
desired: dict[str, Path] = {}
skipped: list[str] = []
for steam_id, last_downloaded_at in items_data:
target = cache_path(steam_id)
if last_downloaded_at is None or not target.exists():
skipped.append(steam_id)
continue
desired[f"{steam_id}.vpk"] = target.resolve()
if should_cancel():
on_stderr("workshop build cancelled before applying symlinks")
return
# existing: symlink-name -> link target (only for symlinks pointing at our cache)
existing: dict[str, Path] = {}
for entry in os.scandir(addons_dir):
if not entry.is_symlink():
continue
try:
target = Path(os.readlink(entry.path))
except OSError:
continue
try:
resolved = target.resolve(strict=False)
except OSError:
continue
if not _is_under(resolved, cache_root):
continue
existing[entry.name] = resolved
created = 0
removed = 0
unchanged = 0
# Remove obsolete or stale symlinks first.
for name, current_target in existing.items():
if should_cancel():
on_stderr("workshop build cancelled mid-removal")
return
desired_target = desired.get(name)
if desired_target is None:
os.unlink(addons_dir / name)
removed += 1
elif current_target != desired_target:
os.unlink(addons_dir / name)
# will be recreated below
else:
unchanged += 1
# Recompute existing post-removal so the create loop knows what's left.
post_removal_existing = {
name for name in existing if name in desired and existing[name] == desired[name]
}
# Create new symlinks.
for name, target in desired.items():
if should_cancel():
on_stderr("workshop build cancelled mid-creation")
return
if name in post_removal_existing:
continue
link_path = addons_dir / name
# Defensive: if a non-symlink file collides with our name, leave it.
if link_path.exists() and not link_path.is_symlink():
on_stderr(
f"refusing to overwrite non-symlink at {link_path}; manual intervention required"
)
continue
if link_path.is_symlink():
# An obsolete symlink not in `existing` (target outside cache).
# We don't manage these — leave alone.
on_stderr(
f"refusing to overwrite foreign symlink at {link_path}"
)
continue
os.symlink(str(target), str(link_path))
created += 1
on_stdout(
f"workshop overlay {overlay.name!r}: created={created} "
f"removed={removed} unchanged={unchanged} "
f"skipped(uncached)={len(skipped)}"
)
for steam_id in skipped:
on_stderr(
f"workshop item {steam_id} skipped: not yet downloaded "
f"(refresh required before this overlay can mount it)"
)
def _is_under(path: Path, root: Path) -> bool:
try:
path.relative_to(root)
except ValueError:
return False
return True
BUILDERS: dict[str, OverlayBuilder] = {
"external": ExternalBuilder(),
"workshop": WorkshopBuilder(),
}

238
l4d2web/tests/test_overlay_builders.py Normal file
View file

@ -0,0 +1,238 @@
"""Tests for overlay builders (registry, ExternalBuilder, WorkshopBuilder)."""
from __future__ import annotations
import os
from datetime import UTC, datetime
from pathlib import Path
import pytest
from l4d2web.db import init_db, session_scope
from l4d2web.models import Overlay, OverlayWorkshopItem, User, WorkshopItem
from l4d2web.services import overlay_builders
@pytest.fixture
def env(tmp_path, monkeypatch):
monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'b.db'}")
monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path))
init_db()
yield tmp_path
def _create_user_and_overlay(name: str, type_: str) -> tuple[int, int]:
with session_scope() as s:
user = User(username="alice", password_digest="x")
s.add(user)
s.flush()
overlay = Overlay(name=name, path=str(7), type=type_, user_id=user.id)
s.add(overlay)
s.flush()
return user.id, overlay.id
def _add_workshop_item(steam_id: str, *, downloaded: bool, cache_root: Path, content: bytes = b"x") -> int:
if downloaded:
cache_root.mkdir(parents=True, exist_ok=True)
(cache_root / f"{steam_id}.vpk").write_bytes(content)
with session_scope() as s:
wi = WorkshopItem(
steam_id=steam_id,
title=f"item-{steam_id}",
filename=f"orig-{steam_id}.vpk",
file_url=f"https://example.com/{steam_id}.vpk",
file_size=len(content) if downloaded else 0,
time_updated=1700000000 if downloaded else 0,
last_downloaded_at=datetime.now(UTC) if downloaded else None,
)
s.add(wi)
s.flush()
return wi.id
def _associate(overlay_id: int, item_id: int) -> None:
with session_scope() as s:
s.add(OverlayWorkshopItem(overlay_id=overlay_id, workshop_item_id=item_id))
def _capture_logs():
out: list[str] = []
err: list[str] = []
return out, err, out.append, err.append
def test_registry_has_external_and_workshop() -> None:
assert "external" in overlay_builders.BUILDERS
assert "workshop" in overlay_builders.BUILDERS
def test_registry_unknown_type_raises_keyerror() -> None:
with pytest.raises(KeyError):
overlay_builders.BUILDERS["nope"]
def test_external_builder_is_idempotent_noop_with_log(env: Path) -> None:
_, overlay_id = _create_user_and_overlay("ext", "external")
out, err, on_stdout, on_stderr = _capture_logs()
with session_scope() as s:
overlay = s.query(Overlay).filter_by(id=overlay_id).one()
overlay_builders.BUILDERS["external"].build(
overlay, on_stdout=on_stdout, on_stderr=on_stderr, should_cancel=lambda: False
)
assert (env / "overlays" / "7").is_dir()
assert any("external overlay" in line for line in out), out
# Existing files in the overlay dir are not touched on subsequent build.
(env / "overlays" / "7" / "untouched.txt").write_text("data")
with session_scope() as s:
overlay = s.query(Overlay).filter_by(id=overlay_id).one()
overlay_builders.BUILDERS["external"].build(
overlay, on_stdout=on_stdout, on_stderr=on_stderr, should_cancel=lambda: False
)
assert (env / "overlays" / "7" / "untouched.txt").read_text() == "data"
def test_workshop_builder_creates_absolute_symlinks(env: Path) -> None:
_, overlay_id = _create_user_and_overlay("ws", "workshop")
cache_root = env / "workshop_cache"
item_a = _add_workshop_item("1001", downloaded=True, cache_root=cache_root, content=b"AAA")
item_b = _add_workshop_item("1002", downloaded=True, cache_root=cache_root, content=b"BBBB")
_associate(overlay_id, item_a)
_associate(overlay_id, item_b)
out, err, on_stdout, on_stderr = _capture_logs()
with session_scope() as s:
overlay = s.query(Overlay).filter_by(id=overlay_id).one()
overlay_builders.BUILDERS["workshop"].build(
overlay, on_stdout=on_stdout, on_stderr=on_stderr, should_cancel=lambda: False
)
addons = env / "overlays" / "7" / "left4dead2" / "addons"
link_a = addons / "1001.vpk"
link_b = addons / "1002.vpk"
assert link_a.is_symlink()
assert link_b.is_symlink()
# Targets must be ABSOLUTE so they resolve in the host's namespace.
assert os.path.isabs(os.readlink(link_a))
assert os.path.isabs(os.readlink(link_b))
# And they must resolve to the cache files.
assert link_a.resolve() == (cache_root / "1001.vpk").resolve()
assert link_b.resolve() == (cache_root / "1002.vpk").resolve()
def test_workshop_builder_skips_uncached_items_with_warning(env: Path) -> None:
_, overlay_id = _create_user_and_overlay("ws", "workshop")
cache_root = env / "workshop_cache"
cached = _add_workshop_item("1001", downloaded=True, cache_root=cache_root)
uncached = _add_workshop_item("9999", downloaded=False, cache_root=cache_root)
_associate(overlay_id, cached)
_associate(overlay_id, uncached)
out, err, on_stdout, on_stderr = _capture_logs()
with session_scope() as s:
overlay = s.query(Overlay).filter_by(id=overlay_id).one()
overlay_builders.BUILDERS["workshop"].build(
overlay, on_stdout=on_stdout, on_stderr=on_stderr, should_cancel=lambda: False
)
addons = env / "overlays" / "7" / "left4dead2" / "addons"
assert (addons / "1001.vpk").is_symlink()
assert not (addons / "9999.vpk").exists(), "must NOT create dangling symlink"
assert any("9999" in line and ("skip" in line.lower() or "uncached" in line.lower()) for line in err + out), err + out
def test_workshop_builder_rerun_is_idempotent(env: Path) -> None:
_, overlay_id = _create_user_and_overlay("ws", "workshop")
cache_root = env / "workshop_cache"
item = _add_workshop_item("1001", downloaded=True, cache_root=cache_root)
_associate(overlay_id, item)
addons = env / "overlays" / "7" / "left4dead2" / "addons"
# First run.
with session_scope() as s:
overlay = s.query(Overlay).filter_by(id=overlay_id).one()
overlay_builders.BUILDERS["workshop"].build(overlay, on_stdout=lambda _x: None, on_stderr=lambda _x: None, should_cancel=lambda: False)
first_inode = (addons / "1001.vpk").lstat().st_ino
# Second run — no-op.
out, err, on_stdout, on_stderr = _capture_logs()
with session_scope() as s:
overlay = s.query(Overlay).filter_by(id=overlay_id).one()
overlay_builders.BUILDERS["workshop"].build(overlay, on_stdout=on_stdout, on_stderr=on_stderr, should_cancel=lambda: False)
second_inode = (addons / "1001.vpk").lstat().st_ino
assert first_inode == second_inode, "symlink should be untouched on idempotent rebuild"
assert any("unchanged" in line.lower() for line in out), out
def test_workshop_builder_removes_obsolete_symlinks(env: Path) -> None:
_, overlay_id = _create_user_and_overlay("ws", "workshop")
cache_root = env / "workshop_cache"
item_a = _add_workshop_item("1001", downloaded=True, cache_root=cache_root)
item_b = _add_workshop_item("1002", downloaded=True, cache_root=cache_root)
_associate(overlay_id, item_a)
_associate(overlay_id, item_b)
addons = env / "overlays" / "7" / "left4dead2" / "addons"
with session_scope() as s:
overlay = s.query(Overlay).filter_by(id=overlay_id).one()
overlay_builders.BUILDERS["workshop"].build(overlay, on_stdout=lambda _x: None, on_stderr=lambda _x: None, should_cancel=lambda: False)
assert (addons / "1002.vpk").is_symlink()
# Remove the association for 1002.
with session_scope() as s:
s.query(OverlayWorkshopItem).filter_by(workshop_item_id=item_b).delete()
out, err, on_stdout, on_stderr = _capture_logs()
with session_scope() as s:
overlay = s.query(Overlay).filter_by(id=overlay_id).one()
overlay_builders.BUILDERS["workshop"].build(overlay, on_stdout=on_stdout, on_stderr=on_stderr, should_cancel=lambda: False)
assert (addons / "1001.vpk").is_symlink()
assert not (addons / "1002.vpk").exists()
# Cache file must remain — overlays are diff-applied, cache is shared.
assert (cache_root / "1002.vpk").exists()
def test_workshop_builder_leaves_unrelated_files_alone(env: Path) -> None:
_, overlay_id = _create_user_and_overlay("ws", "workshop")
cache_root = env / "workshop_cache"
item = _add_workshop_item("1001", downloaded=True, cache_root=cache_root)
_associate(overlay_id, item)
addons = env / "overlays" / "7" / "left4dead2" / "addons"
addons.mkdir(parents=True, exist_ok=True)
(addons / "manual_addon.vpk").write_bytes(b"hand-placed")
with session_scope() as s:
overlay = s.query(Overlay).filter_by(id=overlay_id).one()
overlay_builders.BUILDERS["workshop"].build(overlay, on_stdout=lambda _x: None, on_stderr=lambda _x: None, should_cancel=lambda: False)
# Manual file is preserved.
assert (addons / "manual_addon.vpk").read_bytes() == b"hand-placed"
# Workshop symlink is created alongside.
assert (addons / "1001.vpk").is_symlink()
def test_workshop_builder_honors_should_cancel(env: Path) -> None:
_, overlay_id = _create_user_and_overlay("ws", "workshop")
cache_root = env / "workshop_cache"
items = [_add_workshop_item(f"100{i}", downloaded=True, cache_root=cache_root) for i in range(3)]
for it in items:
_associate(overlay_id, it)
cancel_calls = {"n": 0}
def cancel():
cancel_calls["n"] += 1
return cancel_calls["n"] > 0 # cancel immediately
with session_scope() as s:
overlay = s.query(Overlay).filter_by(id=overlay_id).one()
# Should not crash; partial state is consistent (re-run heals).
overlay_builders.BUILDERS["workshop"].build(
overlay, on_stdout=lambda _x: None, on_stderr=lambda _x: None, should_cancel=cancel
)