486 lines
19 KiB
Python
486 lines
19 KiB
Python
"""The link engine.
|
|
|
|
Deployment is a pure function from (enabled mods, load order) to a desired
|
|
`dest -> source` map, diffed against the last manifest and applied transactionally.
|
|
|
|
Safety rules (this writes into a ~100 GB game install):
|
|
1. Never delete anything that is not in our manifest *and* still matches what we recorded.
|
|
2. Never delete a PAK we did not create (falls out of rule 1).
|
|
3. Every destructive operation is preceded by a Plan the caller can show the user.
|
|
4. reframework/data/ is never touched.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
import uuid
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path, PurePosixPath
|
|
from typing import Callable
|
|
|
|
from . import paks
|
|
from .layout import PAK, Entry
|
|
from .paths import Paths
|
|
from .util import atomic_write_json, read_json
|
|
|
|
MANIFEST_VERSION = 1
|
|
ALLOWED_ROOTS = ("natives", "reframework", "pak_mods")
|
|
|
|
|
|
class DeployError(Exception):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DeployMod:
|
|
id: str
|
|
entries: tuple[Entry, ...]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Item:
|
|
dest: str
|
|
src: Path
|
|
mod: str
|
|
kind: str
|
|
|
|
|
|
@dataclass
|
|
class Conflict:
|
|
key: str # dest path, or a description for PAK asset overlaps
|
|
kind: str # loose | pak-pak | pak-loose
|
|
mods: list[str] # every mod involved, ascending priority
|
|
winner: str
|
|
losers: list[str]
|
|
count: int = 1 # number of overlapping assets (PAK overlaps are aggregated)
|
|
|
|
|
|
@dataclass
|
|
class Desired:
|
|
items: dict[str, Item] = field(default_factory=dict)
|
|
conflicts: list[Conflict] = field(default_factory=list)
|
|
slots: list[paks.Slot] = field(default_factory=list)
|
|
|
|
|
|
# -- destination validation ---------------------------------------------------
|
|
|
|
def validate_dest(dest: str, baseline_names: set[str] = frozenset()) -> str | None:
|
|
"""Return an error string if `dest` is not somewhere we may write, else None."""
|
|
if not dest or dest.startswith("/") or "\\" in dest or "\0" in dest:
|
|
return f"invalid destination {dest!r}"
|
|
parts = PurePosixPath(dest).parts
|
|
if any(p in ("..", ".") for p in parts):
|
|
return f"destination escapes the game directory: {dest}"
|
|
low = [p.lower() for p in parts]
|
|
if low[0] == "reframework" and len(low) > 1 and low[1] == "data":
|
|
return f"reframework/data is never managed: {dest}"
|
|
if len(parts) == 1:
|
|
if dest in baseline_names:
|
|
return f"refusing to shadow vanilla PAK {dest}"
|
|
if paks.parse_pak_name(dest) or dest.lower().endswith(".dll"):
|
|
return None
|
|
return f"root-level file not allowed: {dest}"
|
|
if low[0] not in ALLOWED_ROOTS:
|
|
return f"destination outside natives/ or reframework/: {dest}"
|
|
return None
|
|
|
|
|
|
# -- desired state --------------------------------------------------------------
|
|
|
|
def build_desired(mods: list[DeployMod], baseline: paks.Baseline,
|
|
hashes: Callable[[Path], frozenset] = paks.pak_hashes) -> Desired:
|
|
"""`mods` is in load order, lowest priority first; later mods win contested paths."""
|
|
des = Desired()
|
|
by_key: dict[str, list[Item]] = {}
|
|
for mod in mods:
|
|
for e in mod.entries:
|
|
if e.kind == PAK:
|
|
continue
|
|
by_key.setdefault(e.dest.lower(), []).append(Item(e.dest, e.src, mod.id, e.kind))
|
|
for contenders in by_key.values():
|
|
# a mod listing the same dest twice is not a conflict with itself
|
|
winner = contenders[-1]
|
|
des.items[winner.dest] = winner
|
|
ids = list(dict.fromkeys(c.mod for c in contenders))
|
|
if len(ids) > 1:
|
|
des.conflicts.append(Conflict(winner.dest, "loose", ids, winner.mod,
|
|
[i for i in ids if i != winner.mod]))
|
|
|
|
des.slots = paks.allocate([(m.id, list(m.entries)) for m in mods], baseline)
|
|
for s in des.slots:
|
|
des.items[s.dest] = Item(s.dest, s.src, s.mod_id, PAK)
|
|
|
|
des.conflicts.extend(_pak_conflicts(des, mods, hashes))
|
|
des.conflicts.sort(key=lambda c: (c.kind, c.key.lower()))
|
|
return des
|
|
|
|
|
|
def _pak_conflicts(des: Desired, mods: list[DeployMod], hashes) -> list[Conflict]:
|
|
"""Asset overlaps involving PAKs, matched by KPKA path hash (best effort)."""
|
|
if not des.slots:
|
|
return []
|
|
order = {m.id: i for i, m in enumerate(mods)}
|
|
owners: dict[tuple[int, int], list[tuple[str, str]]] = {} # hash -> [(mod, "pak"|"loose")]
|
|
for s in des.slots:
|
|
for h in hashes(s.src):
|
|
owners.setdefault(h, []).append((s.mod_id, "pak"))
|
|
if not owners:
|
|
return []
|
|
for it in des.items.values():
|
|
if it.kind != PAK and it.dest.lower().startswith("natives/"):
|
|
h = paks.path_hash(it.dest)
|
|
if h in owners:
|
|
owners[h].append((it.mod, "loose"))
|
|
groups: dict[tuple, dict] = {}
|
|
for h, lst in owners.items():
|
|
ids = tuple(sorted(dict.fromkeys(m for m, _ in lst), key=lambda m: order.get(m, 0)))
|
|
if len(ids) < 2:
|
|
continue
|
|
loose = {m for m, k in lst if k == "loose"}
|
|
g = groups.setdefault((("pak-loose" if loose else "pak-pak"), ids), {"count": 0, "loose": set()})
|
|
g["count"] += 1
|
|
g["loose"] |= loose
|
|
out = []
|
|
for (kind, ids), g in groups.items():
|
|
# the loose file loader takes precedence over PAK contents
|
|
winner = next((m for m in reversed(ids) if m in g["loose"]), ids[-1])
|
|
out.append(Conflict(f"{g['count']} shared asset(s)", kind, list(ids), winner,
|
|
[i for i in ids if i != winner], g["count"]))
|
|
return out
|
|
|
|
|
|
# -- plan ---------------------------------------------------------------------------
|
|
|
|
@dataclass
|
|
class Action:
|
|
op: str # add | remove | forget | skip
|
|
dest: str
|
|
item: Item | None = None
|
|
entry: dict | None = None
|
|
note: str = ""
|
|
shadow: bool = False # add: an unmanaged file occupies dest and will be backed up
|
|
|
|
def describe(self) -> str:
|
|
if self.op == "add":
|
|
extra = " (backs up existing file)" if self.shadow else ""
|
|
return f"+ {self.dest}{extra}"
|
|
if self.op == "remove":
|
|
return f"- {self.dest}" + (" (restores backup)" if self.entry and self.entry.get("backup") else "")
|
|
if self.op == "forget":
|
|
return f"? {self.dest} {self.note}"
|
|
return f"! {self.dest} {self.note}"
|
|
|
|
|
|
@dataclass
|
|
class Plan:
|
|
actions: list[Action] = field(default_factory=list)
|
|
warnings: list[str] = field(default_factory=list)
|
|
|
|
@property
|
|
def changes(self) -> list[Action]:
|
|
return [a for a in self.actions if a.op in ("add", "remove")]
|
|
|
|
@property
|
|
def is_empty(self) -> bool:
|
|
return not self.actions
|
|
|
|
def describe(self) -> list[str]:
|
|
return [a.describe() for a in self.actions] + [f"warning: {w}" for w in self.warnings]
|
|
|
|
|
|
@dataclass
|
|
class ApplyResult:
|
|
added: int = 0
|
|
removed: int = 0
|
|
warnings: list[str] = field(default_factory=list)
|
|
|
|
|
|
# -- engine ---------------------------------------------------------------------------
|
|
|
|
class _Txn:
|
|
"""Undo log so a mid-apply failure leaves the game directory as it was."""
|
|
|
|
def __init__(self):
|
|
self.undo: list[Callable[[], None]] = []
|
|
self.trash: list[Path] = []
|
|
|
|
def rollback(self) -> list[str]:
|
|
errs = []
|
|
for fn in reversed(self.undo):
|
|
try:
|
|
fn()
|
|
except Exception as e: # keep unwinding
|
|
errs.append(str(e))
|
|
return errs
|
|
|
|
|
|
class Deployer:
|
|
def __init__(self, game_dir: Path, paths: Paths, link_mode: str = "symlink"):
|
|
self.game = Path(game_dir)
|
|
self.paths = paths
|
|
self.link_mode = link_mode
|
|
if not self.game.is_dir():
|
|
raise DeployError(f"game directory not found: {self.game}")
|
|
self._real_game = Path(os.path.realpath(self.game))
|
|
|
|
# -- manifest ---------------------------------------------------------
|
|
def manifest(self) -> dict:
|
|
m = read_json(self.paths.manifest_file, None)
|
|
if not isinstance(m, dict) or "entries" not in m:
|
|
return {"version": MANIFEST_VERSION, "entries": {}, "dirs": []}
|
|
m.setdefault("dirs", [])
|
|
return m
|
|
|
|
def _save_manifest(self, m: dict) -> None:
|
|
atomic_write_json(self.paths.manifest_file, m)
|
|
|
|
def _inside(self, path: Path) -> bool:
|
|
real = Path(os.path.realpath(path))
|
|
return real == self._real_game or self._real_game in real.parents
|
|
|
|
def _p(self, dest: str) -> Path:
|
|
return self.game / dest
|
|
|
|
def check(self, dest: str, entry: dict) -> str:
|
|
"""'ok' if the on-disk state still matches what we recorded, 'missing', or 'modified'."""
|
|
p = self._p(dest)
|
|
if not os.path.lexists(p):
|
|
return "missing"
|
|
try:
|
|
mode = entry["mode"]
|
|
if mode == "symlink":
|
|
return "ok" if os.path.islink(p) and os.readlink(p) == entry["src"] else "modified"
|
|
if os.path.islink(p) or not os.path.isfile(p):
|
|
return "modified"
|
|
st = os.stat(p)
|
|
if mode == "hardlink":
|
|
return "ok" if (st.st_ino, st.st_dev) == (entry.get("ino"), entry.get("dev")) else "modified"
|
|
return "ok" if (st.st_size, st.st_mtime_ns) == (entry.get("size"), entry.get("mtime_ns")) else "modified"
|
|
except (OSError, KeyError):
|
|
return "modified"
|
|
|
|
def owned_paks(self) -> set[str]:
|
|
m = self.manifest()
|
|
return {d for d, e in m["entries"].items()
|
|
if e.get("kind") == PAK and self.check(d, e) == "ok"}
|
|
|
|
def desired_from_manifest(self, exclude_kind: str | None = None) -> Desired:
|
|
"""The currently-deployed state as a Desired, optionally minus one kind."""
|
|
des = Desired()
|
|
for d, e in self.manifest()["entries"].items():
|
|
if e.get("kind") != exclude_kind:
|
|
des.items[d] = Item(d, Path(e["src"]), e.get("mod", ""), e.get("kind", ""))
|
|
return des
|
|
|
|
# -- planning -----------------------------------------------------------
|
|
def plan(self, desired: Desired) -> Plan:
|
|
plan = Plan()
|
|
m = self.manifest()
|
|
entries: dict[str, dict] = m["entries"]
|
|
baseline = paks.load_baseline(self.paths.baseline_file)
|
|
vanilla = {n for c in (baseline.chains.values() if baseline else []) for n in c.get("files", {})}
|
|
|
|
for dest, entry in sorted(entries.items()):
|
|
state = self.check(dest, entry)
|
|
want = desired.items.get(dest)
|
|
if want is None:
|
|
if state == "ok":
|
|
plan.actions.append(Action("remove", dest, entry=entry))
|
|
elif state == "missing":
|
|
plan.actions.append(Action("forget", dest, entry=entry, note="already gone"))
|
|
else:
|
|
plan.actions.append(Action("forget", dest, entry=entry, note="changed since deployment; left in place"))
|
|
plan.warnings.append(f"{dest} was modified outside the manager; leaving it alone")
|
|
elif state == "modified":
|
|
plan.actions.append(Action("skip", dest, want, entry, "changed since deployment; not overwritten"))
|
|
plan.warnings.append(f"{dest} was modified outside the manager; skipping")
|
|
elif str(want.src) != entry["src"] or want.mod != entry.get("mod"):
|
|
plan.actions.append(Action("remove", dest, entry=entry, note="replaced"))
|
|
plan.actions.append(Action("add", dest, want))
|
|
elif state == "missing":
|
|
plan.actions.append(Action("add", dest, want, note="re-link"))
|
|
|
|
for dest, item in sorted(desired.items.items()):
|
|
if dest in entries:
|
|
continue
|
|
err = validate_dest(dest, vanilla)
|
|
if err:
|
|
plan.actions.append(Action("skip", dest, item, note=err))
|
|
plan.warnings.append(err)
|
|
continue
|
|
p = self._p(dest)
|
|
shadow = os.path.lexists(p)
|
|
if shadow and (os.path.isdir(p) and not os.path.islink(p)):
|
|
plan.actions.append(Action("skip", dest, item, note="a directory is in the way"))
|
|
plan.warnings.append(f"{dest}: a directory is in the way")
|
|
continue
|
|
if shadow and (self.paths.backup_dir / dest).exists():
|
|
plan.actions.append(Action("skip", dest, item, note="a stale backup already exists"))
|
|
plan.warnings.append(f"{dest}: stale backup exists at {self.paths.backup_dir / dest}")
|
|
continue
|
|
plan.actions.append(Action("add", dest, item, shadow=shadow))
|
|
return plan
|
|
|
|
# -- apply --------------------------------------------------------------
|
|
def apply(self, plan: Plan) -> ApplyResult:
|
|
m = self.manifest()
|
|
entries: dict[str, dict] = dict(m["entries"])
|
|
dirs: list[str] = list(m["dirs"])
|
|
res = ApplyResult(warnings=list(plan.warnings))
|
|
txn = _Txn()
|
|
try:
|
|
# removals first so a replaced path is free before its new link goes in
|
|
carry: dict[str, str] = {}
|
|
for a in plan.actions:
|
|
if a.op == "remove":
|
|
replaced = a.note == "replaced"
|
|
old = entries.pop(a.dest)
|
|
self._remove(a.dest, old, txn, restore=not replaced)
|
|
if replaced and old.get("backup"):
|
|
carry[a.dest] = old["backup"] # the shadowed original stays backed up
|
|
else:
|
|
res.removed += 1
|
|
elif a.op == "forget":
|
|
e = entries.pop(a.dest, None)
|
|
if a.note == "already gone" and e and e.get("backup"):
|
|
self._restore_backup(a.dest, txn)
|
|
for a in plan.actions:
|
|
if a.op == "add":
|
|
entries[a.dest] = self._add(a, txn, dirs, carry.get(a.dest, ""))
|
|
res.added += 1
|
|
m = {"version": MANIFEST_VERSION, "entries": entries, "dirs": dirs}
|
|
self._save_manifest(m)
|
|
except BaseException as e:
|
|
errs = txn.rollback()
|
|
if errs:
|
|
res.warnings.extend(f"rollback: {x}" for x in errs)
|
|
if isinstance(e, (OSError, DeployError)):
|
|
raise DeployError(f"deploy failed and was rolled back: {e}") from e
|
|
raise
|
|
for t in txn.trash:
|
|
shutil.rmtree(t, ignore_errors=True)
|
|
self._prune_dirs(m)
|
|
return res
|
|
|
|
def undeploy_all(self) -> ApplyResult:
|
|
return self.apply(self.plan(Desired()))
|
|
|
|
# -- primitives -----------------------------------------------------------
|
|
def _remove(self, dest: str, entry: dict, txn: _Txn, restore: bool = True) -> None:
|
|
p = self._p(dest)
|
|
mode = entry["mode"]
|
|
if mode == "symlink":
|
|
target = os.readlink(p)
|
|
os.unlink(p)
|
|
txn.undo.append(lambda: os.symlink(target, p))
|
|
elif mode == "hardlink":
|
|
src = entry["src"]
|
|
os.unlink(p)
|
|
txn.undo.append(lambda: os.link(src, p))
|
|
else:
|
|
stash = self.paths.trash_dir / uuid.uuid4().hex
|
|
stash.mkdir(parents=True)
|
|
txn.trash.append(stash)
|
|
shutil.move(str(p), str(stash / "f"))
|
|
txn.undo.append(lambda: shutil.move(str(stash / "f"), str(p)))
|
|
if restore and entry.get("backup"):
|
|
self._restore_backup(dest, txn)
|
|
|
|
def _restore_backup(self, dest: str, txn: _Txn) -> None:
|
|
bak = self.paths.backup_dir / dest
|
|
p = self._p(dest)
|
|
if not os.path.lexists(bak):
|
|
return
|
|
if os.path.lexists(p):
|
|
return # something else lives there now; keep the backup
|
|
os.makedirs(p.parent, exist_ok=True)
|
|
shutil.move(str(bak), str(p))
|
|
txn.undo.append(lambda: (os.makedirs(bak.parent, exist_ok=True), shutil.move(str(p), str(bak))))
|
|
self._prune_backup_dirs(bak.parent)
|
|
|
|
def _prune_backup_dirs(self, d: Path) -> None:
|
|
root = self.paths.backup_dir
|
|
while d != root and root in d.parents:
|
|
try:
|
|
d.rmdir()
|
|
except OSError:
|
|
return
|
|
d = d.parent
|
|
|
|
def _mkdirs(self, rel_parent: str, dirs: list[str], txn: _Txn) -> None:
|
|
cur = self.game
|
|
rel = ""
|
|
for part in PurePosixPath(rel_parent).parts:
|
|
cur = cur / part
|
|
rel = f"{rel}/{part}" if rel else part
|
|
if os.path.lexists(cur):
|
|
if not os.path.isdir(cur):
|
|
raise DeployError(f"{rel} exists and is not a directory")
|
|
continue
|
|
if not self._inside(cur.parent):
|
|
raise DeployError(f"{rel} would land outside the game directory")
|
|
os.mkdir(cur)
|
|
dirs.append(rel)
|
|
txn.undo.append(lambda c=cur: os.rmdir(c))
|
|
|
|
def _add(self, a: Action, txn: _Txn, dirs: list[str], carry_backup) -> dict:
|
|
assert a.item is not None
|
|
dest, src = a.dest, a.item.src
|
|
err = validate_dest(dest)
|
|
if err:
|
|
raise DeployError(err)
|
|
p = self._p(dest)
|
|
if not os.path.isfile(src):
|
|
raise DeployError(f"source file missing: {src}")
|
|
rel_parent = str(PurePosixPath(dest).parent)
|
|
if rel_parent != ".":
|
|
self._mkdirs(rel_parent, dirs, txn)
|
|
if not self._inside(p.parent):
|
|
raise DeployError(f"{dest} would land outside the game directory")
|
|
backup = carry_backup or ""
|
|
if os.path.lexists(p):
|
|
bak = self.paths.backup_dir / dest
|
|
if os.path.lexists(bak):
|
|
raise DeployError(f"stale backup exists: {bak}")
|
|
os.makedirs(bak.parent, exist_ok=True)
|
|
shutil.move(str(p), str(bak))
|
|
txn.undo.append(lambda: shutil.move(str(bak), str(p)))
|
|
backup = dest
|
|
mode = self._link(src, p, txn)
|
|
rec = {"mod": a.item.mod, "src": str(src), "mode": mode, "kind": a.item.kind, "backup": backup}
|
|
if mode != "symlink":
|
|
st = os.stat(p)
|
|
rec.update(size=st.st_size, mtime_ns=st.st_mtime_ns, ino=st.st_ino, dev=st.st_dev)
|
|
return rec
|
|
|
|
def _link(self, src: Path, dest: Path, txn: _Txn) -> str:
|
|
mode = self.link_mode
|
|
src_abs = os.path.abspath(src)
|
|
if mode == "symlink":
|
|
os.symlink(src_abs, dest)
|
|
elif mode == "hardlink":
|
|
try:
|
|
os.link(src_abs, dest)
|
|
except OSError:
|
|
shutil.copy2(src_abs, dest)
|
|
mode = "copy"
|
|
else:
|
|
shutil.copy2(src_abs, dest)
|
|
txn.undo.append(lambda: os.unlink(dest))
|
|
return mode
|
|
|
|
def _prune_dirs(self, m: dict) -> None:
|
|
"""Remove directories we created that are now empty (deepest first)."""
|
|
keep = []
|
|
for rel in sorted(m["dirs"], key=lambda r: -r.count("/")):
|
|
p = self._p(rel)
|
|
if not os.path.isdir(p):
|
|
continue
|
|
try:
|
|
p.rmdir()
|
|
except OSError:
|
|
keep.append(rel)
|
|
if len(keep) != len(m["dirs"]):
|
|
m["dirs"] = sorted(keep)
|
|
self._save_manifest(m)
|