feat: implement update command

src/caelestia/parser.py

@@ -11,6 +11,7 @@ from caelestia.subcommands import (
     screenshot,
     shell,
     toggle,
+    update,
     wallpaper,
 )
 from caelestia.utils.dots.manifest import Manifest
@@ -161,6 +162,12 @@ def parse_args() -> tuple[argparse.ArgumentParser, argparse.Namespace]:
     install_parser.add_argument("--noconfirm", action="store_true", help="use defaults for all prompts")
     _set_install_epilog(install_parser)
 
+    # Create parser for update opts
+    update_parser = command_parser.add_parser("update", help="update the Caelestia dotfiles")
+    update_parser.set_defaults(cls=update.Command)
+    update_parser.add_argument("--aur-helper", choices=AUR_HELPERS, help="the AUR helper to use")
+    update_parser.add_argument("--noconfirm", action="store_true", help="use defaults for all prompts")
+
     return parser, parser.parse_args()
 
 

src/caelestia/subcommands/update.py

@@ -0,0 +1,202 @@
+import sys
+from argparse import Namespace
+from pathlib import Path
+
+from caelestia.utils.dots.deployer import Deployer
+from caelestia.utils.dots.diff import Changeset
+from caelestia.utils.dots.manifest import ComponentError, Manifest, ManifestError
+from caelestia.utils.dots.misc import build_local_packages, run_hooks
+from caelestia.utils.dots.packages import PackageInstaller
+from caelestia.utils.dots.source import DotsSource, SourceError
+from caelestia.utils.dots.state import DotsState
+from caelestia.utils.io import disable_input, fatal, info, log, prompt_selection, warn
+
+
+class Command:
+    args: Namespace
+
+    def __init__(self, args: Namespace) -> None:
+        self.args = args
+
+    def run(self) -> None:
+        if self.args.noconfirm:
+            disable_input()
+
+        state = DotsState.load()
+        if state.applied_rev is None:
+            fatal("dots not installed yet. Run `caelestia install` first.")
+
+        # Run system update
+        installer = PackageInstaller.get(self.args.aur_helper or state.aur_helper, self.args.noconfirm)
+        installer.system_update()
+
+        # Get manifest or exit if up to date
+        source, tip, manifest = self.fetch_manifest(state, state.applied_rev)
+
+        # Apply file changes
+        entries = manifest.enabled_entries()
+        try:
+            changeset = Changeset.compute(source, state.applied_rev, tip, entries, state.deployed_files)
+            source.checkout_tip()
+        except SourceError as e:
+            fatal(e)
+        new_files, placed = self.deploy_changeset(source, changeset)
+
+        # Install new/remove old packages
+        desired = manifest.enabled_packages()
+        state.packages = self.sync_packages(installer, state.packages, desired)
+
+        # Install new/remove old local PKGBUILD packages
+        desired_local = manifest.enabled_local_packages()
+        state.local_packages = self.sync_local_packages(installer, source, state.local_packages, desired_local)
+
+        # Run hooks
+        run_hooks(manifest, "post_update")
+
+        # Update state
+        deployed = dict(state.deployed_files)
+        for dest in (*changeset.deletes, *changeset.stale):
+            deployed.pop(str(dest), None)
+        deployed.update(placed)
+        state.deployed_files = deployed
+        state.applied_rev = tip
+        state.enabled_components = manifest.enabled_components
+        state.aur_helper = getattr(installer, "helper", state.aur_helper)
+        state.save()
+
+        self.summarize(changeset, new_files)
+
+    def fetch_manifest(self, state: DotsState, applied_rev: str) -> tuple[DotsSource, str, Manifest]:
+        print()
+        log("Fetching dots repo...")
+        source = DotsSource()
+        try:
+            source.ensure()
+            tip = source.tip_rev()
+            if tip == applied_rev:
+                info("Dots already up to date.")
+                sys.exit(0)
+
+            manifest = source.manifest_at(tip)
+            if source.has_rev(applied_rev):
+                known = set(source.manifest_at(applied_rev).components)
+            else:
+                # Treat all components as known if rev is invalid so we don't overwrite existing prefs
+                known = set(manifest.components)
+        except (SourceError, ManifestError) as e:
+            fatal(e)
+
+        # Enable components recorded at install time + any new components that are default on
+        enabled = [
+            name
+            for name, comp in manifest.components.items()
+            if name in state.enabled_components or (name not in known and comp.default)
+        ]
+
+        # Let the user opt into any new optional components
+        new_comps = [name for name, comp in manifest.components.items() if name not in known and not comp.default]
+        if new_comps:
+            info(f"New components: {', '.join(new_comps)}")
+            enabled += prompt_selection(new_comps, "Components to enable?")
+
+        disabled = [name for name in manifest.components if name not in enabled]
+        try:
+            manifest.resolve_components(enable=enabled, disable=disabled)
+        except ComponentError as e:
+            fatal(e)
+
+        info(f"Enabled components: {', '.join(enabled) or 'none'}")
+
+        return source, tip, manifest
+
+    def deploy_changeset(self, source: DotsSource, changeset: Changeset) -> tuple[list[Path], dict[str, str]]:
+        print()
+
+        if changeset.is_empty():
+            info("No configs to update.")
+            return [], {}
+
+        log("Updating configs...")
+        deployer = Deployer()
+
+        for repofile, dest in changeset.place:
+            src = source.working_path(repofile)
+            if not src.exists():
+                warn(f"missing in source, skipping: {repofile}")
+                continue
+            deployer.place_file(src, dest)
+            info(f"{repofile} -> {dest}")
+
+        new_files = []
+        for repofile, dest in changeset.conflicts:
+            src = source.working_path(repofile)
+            if not src.exists():
+                warn(f"missing in source, skipping: {repofile}")
+                continue
+            new_path = deployer.write_new(src, dest)
+            new_files.append(new_path)
+            warn(f"{dest} has local changes; upstream version written as {new_path.name}")
+
+        for dest in changeset.deletes:
+            deployer.remove(dest)
+            deployer.prune_empty_dirs(dest, Path.home())
+            info(f"Removed {dest}")
+
+        return new_files, deployer.deployed_files
+
+    def sync_packages(self, installer: PackageInstaller, current: list[str], desired: list[str]) -> list[str]:
+        to_install = [p for p in desired if p not in current]
+        to_remove = [p for p in current if p not in desired]
+        installed = list(current)
+
+        if to_install:
+            print()
+            info(f"Installing new packages: {', '.join(to_install)}")
+            installer.install(to_install)
+            installed.extend(p for p in to_install if p not in installed)
+
+        if to_remove:
+            print()
+            info(f"Packages no longer required: {', '.join(to_remove)}")
+            selected = prompt_selection(to_remove, "Packages to remove?")
+            if selected:
+                installer.remove(selected)
+                installed = [p for p in installed if p not in selected]
+
+        return installed
+
+    def sync_local_packages(
+        self, installer: PackageInstaller, source: DotsSource, current: dict[str, list[str]], desired: list[str]
+    ) -> dict[str, list[str]]:
+        to_build = [p for p in desired if p not in current]
+        to_remove = [p for p in current if p not in desired]
+        installed = dict(current)
+
+        if to_build:
+            print()
+            log(f"Building new local packages: {', '.join(to_build)}")
+            installed.update(build_local_packages(installer, source, to_build))
+
+        if to_remove:
+            print()
+            info(f"Local packages no longer required: {', '.join(to_remove)}")
+            selected = prompt_selection(to_remove, "Local packages to remove?")
+            if selected:
+                installer.remove([pkg for path in selected for pkg in current[path]])
+                for path in selected:
+                    installed.pop(path, None)
+
+        return installed
+
+    def summarize(self, changeset: Changeset, new_files: list[Path]) -> None:
+        print()
+        info(f"Updated {len(changeset.place)} file(s), removed {len(changeset.deletes)}, {len(new_files)} conflict(s).")
+        if new_files:
+            info("The following files were changed upstream but you had edited them locally.")
+            info("Your versions were kept; the upstream versions were written alongside as .new:")
+            for path in new_files:
+                info(f"  {path}")
+        if changeset.stale:
+            info("These files are no longer managed but differ from what was installed, so were kept:")
+            for path in changeset.stale:
+                info(f"  {path}")

src/caelestia/utils/dots/deployer.py

@@ -2,7 +2,10 @@ import shutil
 import tempfile
 from pathlib import Path
 
-from caelestia.utils.paths import dots_dir
+from caelestia.utils.paths import cache_dir, config_dir, data_dir, dots_dir, state_dir
+
+# Dirs to never prune even if empty
+_PROTECTED_DIRS = frozenset({Path.home(), config_dir, data_dir, state_dir, cache_dir})
 
 
 class Deployer:
@@ -65,3 +68,17 @@ class Deployer:
             path.unlink()
         elif path.is_dir():
             shutil.rmtree(path)
+
+    def prune_empty_dirs(self, start: Path, stop: Path) -> None:
+        """Removes dirs recursively from start to stop.
+
+        Will never prune protected dirs (home, config, cache, etc).
+        """
+
+        parent = start.parent
+        while parent != stop and stop in parent.parents and parent not in _PROTECTED_DIRS:
+            try:
+                parent.rmdir()
+            except OSError:
+                break
+            parent = parent.parent

src/caelestia/utils/dots/packages.py

@@ -73,6 +73,9 @@ class PackageInstaller(ABC):
     def build_install(self, directory: Path) -> list[str]:
         """Build and install the PKGBUILD in `directory`, returning the installed package names."""
 
+    @abstractmethod
+    def system_update(self) -> None: ...
+
 
 class NoopInstaller(PackageInstaller):
     """Used off Arch, where the dots' packages are not available via pacman/AUR."""
@@ -89,6 +92,9 @@ class NoopInstaller(PackageInstaller):
         info(f"Skipping local package build (not on Arch): {directory}")
         return []
 
+    def system_update(self) -> None:
+        info("Skipping system update (not on Arch)")
+
 
 class ArchInstaller(PackageInstaller):
     def __init__(self, helper: str, noconfirm: bool = False) -> None:
@@ -128,3 +134,6 @@ class ArchInstaller(PackageInstaller):
         subprocess.run(["makepkg", "-fsi", *self.flags], cwd=directory, env=env, check=True)
 
         return names
+
+    def system_update(self) -> None:
+        subprocess.run([self.helper, "-Syu", *self.flags], check=True)