#!/usr/bin/env python3
"""VoltVault offline-first Docker upgrade transaction. Python 3.9+, stdlib only."""
import argparse
import copy
import csv
import hashlib
import json
import os
import platform
import re
import shutil
import signal
import subprocess
import sys
import tempfile
import time
import urllib.request
import uuid
import zipfile
from pathlib import Path

VERSION = '2.0.7'
PAYLOAD_SHA256 = '3c3475daa33feb6750b7bfdcc24fdccbf8c06d62a4dba695af4323ccc54b5a7e'
DOWNLOAD = 'https://voltvaultapp.com/downloads/voltvault-2.0.7-installer.zip'
SERVICES = ('voltvault-server', 'voltvault-web-client')
OPTIONAL_SERVICE = 'laya'
GUIDE = 'https://voltvaultapp.com/manual-setup.html#backup'
SENSITIVE_ENV = ('COMPOSE_FILE', 'COMPOSE_PROFILES', 'COMPOSE_ENV_FILES', 'COMPOSE_PROJECT_NAME')


class UpgradeError(Exception):
    pass


def require(condition, message):
    if not condition:
        raise UpgradeError(message)


def digest(path):
    h = hashlib.sha256()
    with open(path, 'rb') as file:
        for chunk in iter(lambda: file.read(1024 * 1024), b''):
            h.update(chunk)
    return h.hexdigest()


def private(path):
    """Protect recovery files, including resolved Compose secrets."""
    if os.name == 'nt':
        who = subprocess.check_output(['whoami', '/user', '/fo', 'csv', '/nh'], text=True)
        sid = re.search(r'S-1-5-[0-9-]+', who).group()
        result = subprocess.run(['icacls', str(path), '/inheritance:r', '/grant:r',
                                 f'*{sid}:(OI)(CI)F', '*S-1-5-18:(OI)(CI)F', '*S-1-5-32-544:(OI)(CI)F'],
                                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        require(result.returncode == 0, 'Could not protect recovery directory.')
    else:
        path.chmod(0o700 if path.is_dir() else 0o600)


def write_json(path, value):
    temp = path.with_suffix(path.suffix + '.new')
    with temp.open('w', encoding='utf-8') as file:
        json.dump(value, file, indent=2)
        file.flush()
        os.fsync(file.fileno())
    private(temp)
    os.replace(temp, path)


def safe(path):
    require(not path.is_symlink() and not (hasattr(path, 'is_junction') and path.is_junction()),
            f'Linked installation/recovery paths are unsupported: {path}')


def command(args, env=None, input_text=None, visible=False):
    # Captured Docker diagnostics/config can contain secrets; never echo them.
    try:
        result = subprocess.run([str(a) for a in args], env=env, input=input_text,
                                text=True, stdout=None if visible else subprocess.PIPE,
                                stderr=None if visible else subprocess.PIPE)
    except OSError as error:
        raise UpgradeError(f'Could not run {args[0]}. Check installation and permissions.') from error
    require(result.returncode == 0, f'{args[0]} {args[1] if len(args)>1 else ""} failed. Check Docker; configuration and backups are retained.')
    return result.stdout or ''


def docker(*args, **kwargs):
    return command(['docker', *args], **kwargs)


def containers():
    ids = docker('ps', '-a', '--format', '{{.ID}}').split()
    return json.loads(docker('inspect', *ids)) if ids else []


def labels(container):
    return container['Config'].get('Labels') or {}


def recovery_metadata(path):
    root = path / '.voltvault-updater'
    state_file = root / 'state.json'
    safe(root); safe(state_file)
    if not state_file.is_file():
        return None
    state = json.loads(state_file.read_text())
    require(state.get('format') == 1 and state.get('directory') == str(path), 'Invalid recovery metadata.')
    return state


def recovery_members(path, state, members):
    """Recover even if Compose was interrupted between container replacements."""
    transaction = Path(state['transaction'])
    require(transaction.parent == path / '.voltvault-updater' and transaction.name.startswith('transaction-'), 'Unsafe recovery path.')
    safe(transaction)
    previous = json.loads((transaction / 'previous.json').read_text())
    present = {labels(c)['com.docker.compose.service'] for c in members}
    for service in state.get('services', SERVICES):
        if service in present:
            continue
        spec = previous['services'][service]
        members.append({'Id': 'missing-' + service, 'Image': spec['image'],
                        'Config': {'Env': [f'{key}={str(value).replace("$$", "$")}' for key, value in spec.get('environment', {}).items()],
                                   'Labels': {'com.docker.compose.project': state['project'], 'com.docker.compose.service': service,
                                              'com.docker.compose.project.working_dir': str(path),
                                              'com.docker.compose.project.config_files': str(transaction / 'previous.json')}},
                        'Mounts': ([{'Type': 'volume', 'Name': name, 'Destination': target} for target, name in state['volumes'].items()]
                                   if service == SERVICES[0] else
                                   [{'Type': 'volume', 'Name': state['model_volume'], 'Destination': '/models'}]
                                   if service == OPTIONAL_SERVICE and state.get('model_volume') else [])})
    return members


def discover(requested):
    found = containers()
    groups = {}
    for container in found:
        meta = labels(container)
        if meta.get('com.docker.compose.service') in (*SERVICES, OPTIONAL_SERVICE):
            key = (meta.get('com.docker.compose.project'), meta.get('com.docker.compose.project.working_dir'))
            groups.setdefault(key, []).append(container)
    if requested:
        path = Path(requested).absolute()
        safe(path)
        path = path.resolve()
        matches = [(key, value) for key, value in groups.items()
                   if key[1] and Path(key[1]).resolve() == path]
        # Explicit directories may refer to the bundle root of a legacy deployment.
        if not matches:
            matches = [(key, value) for key, value in groups.items()
                       if key[1] and Path(key[1]).resolve() == path / 'voltvault-server']
    else:
        matches = list(groups.items())
    if not matches:
        fallback = Path(requested).resolve() if requested else (Path.home() / 'voltvault').resolve()
        state = recovery_metadata(fallback)
        if state and state['phase'] not in ('complete', 'aborted', 'rolled-back'):
            members = recovery_members(fallback, state, [])
            return fallback, state['project'], members, [Path(state['transaction']) / 'previous.json'], found
    require(len(matches) == 1, 'No unique VoltVault installation found. Use --dir PATH for the existing Compose directory. ' + GUIDE)
    (project, directory), members = matches[0]
    require(project and directory, 'Compose deployment ownership is missing. ' + GUIDE)
    path = Path(directory)
    safe(path)
    require(path.is_dir(), 'Docker labels point to an unavailable local directory. Run on the Docker host. ' + GUIDE)
    path = path.resolve()
    state = recovery_metadata(path)
    require(not state or state['project'] == project, 'Recovery project ownership differs.')
    enabled = any(line.strip() == 'LAYA_MATCH_ENABLED=true' for line in (path / '.env').read_text().splitlines()) if (path / '.env').is_file() else False
    expected = (*SERVICES, OPTIONAL_SERVICE) if enabled else SERVICES
    if not enabled:
        inactive = [c for c in members if labels(c).get('com.docker.compose.service') == OPTIONAL_SERVICE]
        require(not any(c.get('State', {}).get('Running', False) for c in inactive),
                'Laya is still running while disabled. Stop it with the installer --no-laya option before upgrading. ' + GUIDE)
        members = [c for c in members if labels(c).get('com.docker.compose.service') in SERVICES]
    if state and state['phase'] not in ('complete', 'aborted', 'rolled-back') and len(members) < len(state.get('services', expected)):
        members = recovery_members(path, state, members)
    require(len(members) == len(expected) and {labels(c).get('com.docker.compose.service') for c in members} == set(expected),
            'Only official full-stack deployments with an optional saved Laya worker are supported. ' + GUIDE)
    configs = {labels(c).get('com.docker.compose.project.config_files', '') for c in members}
    files = list(dict.fromkeys(Path(f) for config in configs for f in config.split(',') if f))
    if len(configs) != 1:
        require(state and all(path / '.voltvault-updater' in file.parents or str(file) in state['original_files'] for file in files),
                'Compose file ownership is ambiguous. ' + GUIDE)
        files = [Path(state['transaction']) / 'previous.json'] if state['phase'] not in ('complete', 'aborted', 'rolled-back') else [Path(f) for f in state['original_files']]
    for file in files:
        safe(file)
    require(files and all(f.is_file() for f in files), 'Original Compose files are unavailable. ' + GUIDE)
    return path, project, members, files, found


def clean_environment(env_file):
    env = os.environ.copy()
    # Compose interpolates inherited variables before .env. Saved settings win.
    keys = set(SENSITIVE_ENV)
    for line in env_file.read_text(encoding='utf-8-sig').splitlines():
        match = re.match(r'\s*(?:export\s+)?([A-Za-z_][A-Za-z_0-9]*)\s*=', line)
        if match:
            keys.add(match.group(1))
    for key in keys:
        env.pop(key, None)
    return env


NODE_HELPER = r'''
const fs=require('fs'), path=require('path'), crypto=require('crypto');
const Database=require('/app/node_modules/better-sqlite3');
let input=''; process.stdin.on('data',c=>input+=c); process.stdin.on('end',()=>{
try {
const request=JSON.parse(input), mode=request.mode;
function tree(root) {
  const out={};
  function walk(p,rel) {
    const st=fs.lstatSync(p);
    if(st.isSymbolicLink() || (!st.isFile() && !st.isDirectory())) throw Error('Unsupported file');
    const item={uid:st.uid,gid:st.gid,mode:st.mode&4095,dir:st.isDirectory()};
    if(st.isFile()) { const h=crypto.createHash('sha256'); const fd=fs.openSync(p,'r');
      const buf=Buffer.alloc(1024*1024); let n; while((n=fs.readSync(fd,buf,0,buf.length,null))>0) h.update(buf.subarray(0,n));
      fs.closeSync(fd); item.hash=h.digest('hex'); }
    out[rel]=item;
    if(st.isDirectory()) for(const name of fs.readdirSync(p).sort()) walk(path.join(p,name),rel?rel+'/'+name:name);
  } walk(root,''); return out;
}
if(mode==='clear') {
  for(const root of ['/data','/uploads']) for(const name of fs.readdirSync(root)) fs.rmSync(path.join(root,name),{recursive:true,force:true});
  console.log('{}'); return;
}
if(mode==='restore-permissions') {
  for(const [root,items] of Object.entries(request.manifest)) for(const [rel,item] of Object.entries(items)) {
    const p=path.join(root,rel); fs.chownSync(p,item.uid,item.gid); fs.chmodSync(p,item.mode);
  }
  console.log('{}'); return;
}
const db=new Database('/data/inventory.sqlite',{readonly:true,fileMustExist:true});
if(db.pragma('integrity_check',{simple:true})!=='ok') throw Error('Database integrity failed');
if(request.credential_key && db.prepare("SELECT 1 FROM sqlite_master WHERE type='table' AND name='printers'").get()) {
  const key=Buffer.from(request.credential_key,'hex');
  for(const row of db.prepare('SELECT access_code_encrypted FROM printers').all()) {
    const [version,iv,tag,data]=row.access_code_encrypted.split(':');
    if(version!=='v1') throw Error('Unsupported credential');
    const decipher=crypto.createDecipheriv('aes-256-gcm',key,Buffer.from(iv,'base64'));
    decipher.setAuthTag(Buffer.from(tag,'base64')); decipher.update(Buffer.from(data,'base64')); decipher.final();
  }
}
const quote=s=>'"'+s.replaceAll('"','""')+'"'; const tables={};
const specs=request.tables || Object.fromEntries(db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' AND name!='schema_migrations'").all().map(r=>[r.name,{columns:db.prepare('PRAGMA table_info('+quote(r.name)+')').all().map(c=>c.name)}]));
for(const [name,spec] of Object.entries(specs)) {
  const rows=db.prepare('SELECT '+spec.columns.map(quote).join(',')+' FROM '+quote(name)).raw().all();
  const serialized=rows.map(row=>JSON.stringify(row,(_,v)=>typeof v==='bigint'?String(v):Buffer.isBuffer(v)?v.toString('hex'):v)).sort();
  tables[name]={columns:spec.columns,count:rows.length,hash:crypto.createHash('sha256').update(JSON.stringify(serialized)).digest('hex')};
}
db.close();
console.log(JSON.stringify({tables,manifest:{'/data':tree('/data'),'/uploads':tree('/uploads')}}));
} catch(e) { console.error('Volume/database verification failed.'); process.exitCode=1; }
});
'''


def host_manifest(root):
    result = {}
    for entry in [root, *sorted(root.rglob('*'))]:
        safe(entry)
        require(entry.is_dir() or entry.is_file(), 'Unsupported backup file.')
        rel = entry.relative_to(root).as_posix()
        result['' if rel == '.' else rel] = {'dir': entry.is_dir()}
        if entry.is_file():
            result[rel]['hash'] = digest(entry)
    return result


def content_manifest(items):
    return {key: {k: v for k, v in item.items() if k in ('dir', 'hash')} for key, item in items.items()}


def release_architecture():
    platform_name = docker('info', '--format', '{{.OSType}}/{{.Architecture}}').strip()
    require('/' in platform_name, 'Could not determine Docker image architecture.')
    os_name, architecture = platform_name.split('/', 1)
    require(os_name == 'linux', 'Docker must use Linux containers.')
    architecture = {'aarch64': 'arm64', 'x86_64': 'amd64'}.get(architecture, architecture)
    require(architecture in ('arm64', 'amd64'), 'Prebuilt images support Linux ARM64 and AMD64.')
    return architecture


def release_images(manifest, architecture):
    with manifest.open(newline='') as file:
        reader = csv.DictReader(file, delimiter='\t')
        require(reader.fieldnames == ['arch', 'service', 'filename', 'sha256', 'tag', 'image_id'],
                'Invalid prebuilt image manifest columns.')
        rows = list(reader)
    require(len(rows) == 4 and {(row['arch'], row['service']) for row in rows} ==
            {(arch, service) for arch in ('arm64', 'amd64') for service in SERVICES},
            'Incomplete prebuilt image manifest.')
    selected = {}
    for row in rows:
        require(row['filename'] == f'{row["service"]}-{VERSION}-linux-{row["arch"]}.tar.gz' and
                re.fullmatch(r'[0-9a-f]{64}', row['sha256']) and
                re.fullmatch(r'sha256:[0-9a-f]{64}', row['image_id']) and
                row['tag'] == f'{row["service"]}:{VERSION}-{row["arch"]}',
                'Invalid prebuilt image manifest.')
        if row['arch'] == architecture:
            selected[row['service']] = row
    return selected


class Upgrader:
    def __init__(self, path, project, members, files, all_containers, bundle=None):
        self.path, self.project, self.members, self.files = path, project, members, files
        self.bundle = bundle
        self.root = path / '.voltvault-updater'
        self.state_file = self.root / 'state.json'
        self.lock = Path(str(path) + '.install-lock')
        self.env_file = path / '.env'
        safe(self.env_file)
        require(self.env_file.is_file(), 'Saved .env is missing; keys will not be regenerated. ' + GUIDE)
        self.env = clean_environment(self.env_file)
        saved = dict(line.split('=', 1) for line in self.env_file.read_text().splitlines() if '=' in line and not line.startswith('#'))
        self.tinyfish_api_key = saved.get('TINYFISH_API_KEY', '')
        require(not self.tinyfish_api_key or re.fullmatch(r'[A-Za-z0-9._-]+', self.tinyfish_api_key) is not None,
                'TINYFISH_API_KEY in the saved .env contains unsupported characters.')
        require(saved.get('LAYA_MATCH_ENABLED', 'false') in ('true', 'false'), 'Saved LAYA_MATCH_ENABLED must be true or false.')
        self.laya_enabled = saved.get('LAYA_MATCH_ENABLED', 'false') == 'true'
        require(not self.laya_enabled or saved.get('LAYA_MATCH_URL') == 'http://laya:8000',
                'The official Laya worker URL must be http://laya:8000.')
        self.services = (*SERVICES, OPTIONAL_SERVICE) if self.laya_enabled else SERVICES
        self.env_hash = digest(self.env_file)
        self.state = None
        self.locked = False
        safe(self.root)
        if self.state_file.exists():
            safe(self.state_file)
            self.state = json.loads(self.state_file.read_text())
            require(self.state.get('format') == 1 and self.state.get('project') == project and
                    self.state.get('directory') == str(path), 'Unrecognized recovery metadata. ' + GUIDE)
        managed = self.root / 'active.json'
        if any(self.root in file.parents for file in files):
            require(self.state is not None, 'Managed deployment metadata is missing. ' + GUIDE)
            self.files = [Path(f) for f in self.state['original_files']]
        require(len(self.files) == 1 and self.files[0].name in ('compose.yaml', 'compose.full-stack.yaml') and
                self.files[0].parent.resolve() == path,
                'Custom Compose layouts are unsupported. ' + GUIDE)
        if self.state and self.state['phase'] not in ('complete', 'aborted', 'rolled-back'):
            source_files = [self.transaction / 'previous.json']
        else:
            source_files = [managed] if managed.is_file() else self.files
        self.config = json.loads(self.compose(source_files, 'config', '--format', 'json'))
        self.validate(all_containers)

    def compose(self, files, *args, visible=False, profile=None):
        options = ['compose', '--project-directory', str(self.path), '--env-file', str(self.env_file), '-p', self.project]
        for file in files:
            options.extend(['-f', str(file)])
        if profile == 'laya' or (profile is None and self.laya_enabled):
            options.extend(['--profile', 'laya'])
        return docker(*options, *args, env=self.env, visible=visible)

    def validate(self, all_containers):
        services = self.config.get('services', {})
        require(set(services) == set(self.services), 'Custom Compose services are unsupported. ' + GUIDE)
        networks = self.config.get('networks', {})
        require(set(networks) <= {'default'} and not any(spec.get('external') or spec.get('driver', 'bridge') != 'bridge' for spec in networks.values()),
                'Shared/external Docker networks require the manual upgrade guide. ' + GUIDE)
        server, web = (services[s] for s in SERVICES)
        self.by_service = {labels(c)['com.docker.compose.service']: c for c in self.members}
        self.volumes = {}
        for service in self.services:
            spec = services[service]
            require(not any(spec.get(key) for key in ('privileged', 'devices', 'network_mode', 'secrets', 'configs', 'env_file',
                                                     'labels', 'entrypoint', 'command', 'extra_hosts')) and set(spec.get('networks', {})) <= {'default'},
                    'Custom Docker service configuration is unsupported. ' + GUIDE)
            require(spec.get('profiles', []) == (['laya'] if service == OPTIONAL_SERVICE else []),
                    'Unexpected Docker profile. ' + GUIDE)
            build = spec.get('build')
            if build:
                context = Path(build['context'].replace('$$', '$')).resolve()
                expected = (self.path / ('voltvault-server' if service == OPTIONAL_SERVICE else service)
                            if self.files[0].name == 'compose.yaml' else
                            self.path if service in (SERVICES[0], OPTIONAL_SERVICE) else self.path.parent / service)
                if service == OPTIONAL_SERVICE and self.state:
                    expected_managed = Path(self.state['transaction']) / 'source/voltvault/voltvault-server'
                else:
                    expected_managed = expected
                require(context in (expected.resolve(), expected_managed.resolve()) and
                        build.get('dockerfile', 'Dockerfile') == ('Dockerfile.laya' if service == OPTIONAL_SERVICE else 'Dockerfile') and
                        not any(build.get(k) for k in ('args', 'target', 'additional_contexts', 'dockerfile_inline')),
                        'Custom Docker build layout is unsupported. ' + GUIDE)
            else:
                require(self.state, 'Missing official build context. ' + GUIDE)
            current_env = dict(item.split('=', 1) for item in self.by_service[service]['Config']['Env'] if '=' in item)
            for key, value in spec.get('environment', {}).items():
                if self.state and self.state['phase'] in ('validating', 'validated', 'restoring', 'restore-failed') and service == SERVICES[0] and key in ('DEVICE_CREDENTIAL_KEY', 'BACKUP_SCHEDULE'):
                    require(current_env.get(key) in (str(value), '' if key == 'DEVICE_CREDENTIAL_KEY' else 'disabled'), 'Unrecognized validation configuration.')
                    continue
                # Compose's JSON export escapes literal dollars for safe round-tripping.
                require(str(value).replace('$$', '$') == current_env.get(key), f'Saved {key} differs from the existing container; reconcile configuration before upgrading. ' + GUIDE)
        settings = server.get('environment', {})
        require(settings.get('DATABASE_PATH') == '/data/inventory.sqlite' and settings.get('UPLOAD_DIR') == '/uploads',
                'Custom database/upload locations are unsupported. ' + GUIDE)
        require(settings.get('SESSION_SECRET') and settings.get('DEVICE_CREDENTIAL_KEY'),
                'Saved session/printer credential keys are missing; they will not be regenerated. ' + GUIDE)
        require(not server.get('ports') and not web.get('volumes'), 'Only private-API full-stack layouts are supported. ' + GUIDE)
        require(settings.get('LAYA_MATCH_ENABLED', 'false') == ('true' if self.laya_enabled else 'false') and
                settings.get('LAYA_MATCH_URL', 'http://laya:8000') == 'http://laya:8000',
                'Saved Laya settings differ from the running server. ' + GUIDE)
        if self.laya_enabled:
            laya = services[OPTIONAL_SERVICE]
            require(not laya.get('ports') and laya.get('environment', {}).get('HF_HOME') == '/models' and
                    len(laya.get('volumes', [])) == 1 and laya['volumes'][0]['target'] == '/models' and
                    laya['volumes'][0]['type'] == 'volume' and not laya['volumes'][0].get('read_only'),
                    'Laya must use one private local model-cache volume. ' + GUIDE)
            model_mount = next((m for m in self.by_service[OPTIONAL_SERVICE]['Mounts'] if m['Destination'] == '/models'), None)
            model_name = self.config.get('volumes', {}).get(laya['volumes'][0]['source'], {}).get('name')
            require(model_mount and model_mount['Type'] == 'volume' and model_mount['Name'] == model_name,
                    'Laya model-cache ownership differs. ' + GUIDE)
            require(json.loads(docker('volume', 'inspect', model_name))[0]['Name'] == model_name,
                    'The Laya model cache is missing. ' + GUIDE)
            self.model_volume = model_name
        else:
            self.model_volume = None
        mounts = server.get('volumes', [])
        require(len(mounts) == 2 and {m['target'] for m in mounts} == {'/data', '/uploads'} and
                all(m.get('type') == 'volume' and not m.get('read_only') for m in mounts),
                'Only dedicated named data/upload volumes are supported; bind mounts need the manual guide. ' + GUIDE)
        for mount in mounts:
            declaration = self.config.get('volumes', {}).get(mount['source'], {})
            require(not declaration.get('driver_opts') and declaration.get('driver', 'local') == 'local',
                    'Custom volume drivers are unsupported. ' + GUIDE)
            actual = next((m for m in self.by_service[SERVICES[0]]['Mounts'] if m['Destination'] == mount['target']), None)
            require(actual and actual['Type'] == 'volume' and actual['Name'] == declaration.get('name'),
                    'Compose and container volume ownership differ. ' + GUIDE)
            inspected = json.loads(docker('volume', 'inspect', actual['Name']))
            require(len(inspected) == 1 and inspected[0]['Name'] == actual['Name'], 'An existing data/upload volume is missing. Restore it before upgrading. ' + GUIDE)
            self.volumes[mount['target']] = actual['Name']
        require(len(set(self.volumes.values())) == 2, 'Data and uploads must be distinct volumes.')
        if self.laya_enabled:
            require(model_name not in self.volumes.values(), 'The model cache must not share inventory volumes. ' + GUIDE)
        for container in all_containers:
            if container['Id'] not in {c['Id'] for c in self.members}:
                if self.state and self.state['phase'] in ('restoring', 'restore-failed') and container.get('Name') == '/voltvault-restore-' + self.state['token'][:12] and container['Image'] == self.state['old_image']:
                    continue
                require(not any(m.get('Name') in self.volumes.values() for m in container.get('Mounts', [])),
                        'Another container owns these volumes. Stop and resolve that deployment first. ' + GUIDE)
        ports = web.get('ports', [])
        require(len(ports) == 1 and ports[0].get('target') == 80 and ports[0].get('published') and
                ports[0].get('protocol', 'tcp') == 'tcp', 'Custom browser port layouts are unsupported. ' + GUIDE)
        host = ports[0].get('host_ip', '0.0.0.0')
        self.url = f'http://{"127.0.0.1" if host in ("0.0.0.0", "::", "") else "["+host+"]" if ":" in host else host}:{ports[0]["published"]}'
        self.old_image = self.by_service[SERVICES[0]]['Image']

    def acquire(self):
        safe(self.lock)
        try:
            self.lock.mkdir()
        except FileExistsError:
            owner_file = self.lock / 'owner.json'
            require(owner_file.is_file() and not owner_file.is_symlink(), 'Another install/upgrade may be running. Do not remove its lock while active.')
            owner = json.loads(owner_file.read_text())
            require(owner.get('host') == platform.node(), 'Installation lock belongs to another host.')
            if os.name == 'nt':
                # os.kill(pid, 0) can terminate processes on Windows; query instead.
                listing = command(['tasklist', '/FI', f'PID eq {int(owner["pid"])}', '/FO', 'CSV', '/NH'])
                alive = bool(re.search(r'"' + str(int(owner['pid'])) + r'"', listing))
            else:
                try:
                    os.kill(owner['pid'], 0)
                    alive = True
                except ProcessLookupError:
                    alive = False
            if not alive:
                owner_file.unlink()
                self.lock.rmdir()
                self.lock.mkdir()
            else:
                raise UpgradeError('Another upgrade is running. Retry after it finishes.')
        self.locked = True
        private(self.lock)
        write_json(self.lock / 'owner.json', {'host': platform.node(), 'pid': os.getpid()})

    def release(self):
        if self.locked:
            (self.lock / 'owner.json').unlink(missing_ok=True)
            self.lock.rmdir()
            self.locked = False

    def save(self, phase):
        self.state['phase'] = phase
        write_json(self.state_file, self.state)

    def helper(self, request, image=None):
        return json.loads(docker('run', '--rm', '-i', '--network', 'none', '--user', '0', '--entrypoint', 'node',
                                 '--mount', f'type=volume,src={self.volumes["/data"]},dst=/data',
                                 '--mount', f'type=volume,src={self.volumes["/uploads"]},dst=/uploads',
                                 image or self.old_image, '-e', NODE_HELPER,
                                 input_text=json.dumps(request)))

    def snapshot(self, tables=None, image=None):
        return self.helper({'mode': 'snapshot', 'tables': tables,
                            'credential_key': self.config['services'][SERVICES[0]]['environment']['DEVICE_CREDENTIAL_KEY']}, image)

    def initialize(self):
        self.root.mkdir(exist_ok=True)
        private(self.root)
        token = uuid.uuid4().hex
        transaction = self.root / ('transaction-' + time.strftime('%Y%m%d-%H%M%S') + '-' + token[:8])
        transaction.mkdir()
        private(transaction)
        old = copy.deepcopy(self.config)
        old['name'] = self.project
        for service in self.services:
            old['services'][service].pop('build', None)
            old['services'][service]['image'] = self.by_service[service]['Image']
        write_json(transaction / 'previous.json', old)
        self.state = {'format': 1, 'directory': str(self.path), 'project': self.project,
                      'original_files': [str(f) for f in self.files], 'transaction': str(transaction),
                      'target': VERSION, 'payload_hash': PAYLOAD_SHA256, 'token': token,
                      'volumes': self.volumes, 'url': self.url, 'old_image': self.old_image,
                      'services': self.services, 'model_volume': self.model_volume,
                      'prior_active': (self.root / 'active.json').read_text() if (self.root / 'active.json').exists() else None}
        shutil.copy2(self.env_file, transaction / 'saved.env')
        private(transaction / 'saved.env')
        for index, file in enumerate(self.files):
            shutil.copy2(file, transaction / f'compose-{index}.saved')
            private(transaction / f'compose-{index}.saved')
        self.save('staging')

    @property
    def transaction(self):
        path = Path(self.state['transaction'])
        require(path.parent == self.root and path.name.startswith('transaction-'), 'Unsafe recovery metadata.')
        safe(path)
        return path

    def stage(self):
        payload = self.transaction / 'release.zip'
        if self.bundle is not None:
            print('[1/5] Copying and verifying the local release…', flush=True)
            require(digest(self.bundle) == PAYLOAD_SHA256,
                    'Local release verification failed; existing containers are untouched.')
            shutil.copyfile(self.bundle, payload)
        else:
            print('[1/5] Downloading and verifying the release…', flush=True)
            with urllib.request.urlopen(DOWNLOAD, timeout=300) as response, payload.open('wb') as output:
                require(response.geturl().startswith('https://'), 'Insecure release redirect rejected.')
                shutil.copyfileobj(response, output)
        require(digest(payload) == PAYLOAD_SHA256, 'Release verification failed; existing containers are untouched.')
        source = self.transaction / 'source'
        with zipfile.ZipFile(payload) as archive:
            for member in archive.infolist():
                parts = Path(member.filename).parts
                require(parts and parts[0] == 'voltvault' and '..' not in parts and not Path(member.filename).is_absolute()
                        and '\\' not in member.filename and ((member.external_attr >> 16) & 0o170000) != 0o120000,
                        'Unsafe release archive.')
            archive.extractall(source)
        image_manifest = source / 'voltvault/images.tsv'
        images = release_images(image_manifest, release_architecture()) if image_manifest.is_file() else None
        if images:
            image_dir = self.transaction / 'images'
            image_dir.mkdir()
            private(image_dir)
            for service in SERVICES:
                row = images[service]
                archive = image_dir / row['filename']
                try:
                    current = json.loads(docker('image', 'inspect', row['tag']))[0]
                except (UpgradeError, ValueError, IndexError):
                    current = None
                if current and current.get('Id') == row['image_id'] and current.get('Architecture') == row['arch']:
                    print(f'Reusing verified local {service} image.', flush=True)
                    continue
                url = 'https://voltvaultapp.com/downloads/' + row['filename']
                print(f'Downloading prebuilt {service} for linux/{row["arch"]}…', flush=True)
                with urllib.request.urlopen(url, timeout=300) as response, archive.open('wb') as output:
                    require(response.geturl().startswith('https://'), 'Insecure image redirect rejected.')
                    shutil.copyfileobj(response, output)
                require(digest(archive) == row['sha256'], 'Prebuilt image verification failed; existing containers are untouched.')
            print('[2/5] Loading verified images. Your existing installation is still running…', flush=True)
            for service in SERVICES:
                row = images[service]
                archive = image_dir / row['filename']
                if archive.is_file():
                    docker('load', '-i', str(archive), visible=True)
                image = json.loads(docker('image', 'inspect', row['tag']))[0]
                require(image.get('Id') == row['image_id'] and image.get('Os') == 'linux' and
                        image.get('Architecture') == row['arch'], 'Loaded image identity or architecture differs.')
                docker('tag', row['tag'], f'voltvault-upgrade-{service}:{VERSION}-{self.state["token"][:12]}')
        candidate = copy.deepcopy(self.config)
        candidate['name'] = self.project
        released = json.loads(self.compose([source / 'voltvault/compose.yaml'], 'config', '--format', 'json', profile='laya'))
        worker = released.get('services', {}).get(OPTIONAL_SERVICE)
        if worker:
            context = source / 'voltvault/voltvault-server'
            require((context / 'Dockerfile.laya').is_file(), 'The optional Laya worker is missing from the release.')
            worker['build'] = {'context': str(context).replace('$', '$$'), 'dockerfile': 'Dockerfile.laya'}
            candidate['services'][OPTIONAL_SERVICE] = worker
            candidate.setdefault('volumes', {})['laya-model-cache'] = (
                self.config.get('volumes', {}).get('laya-model-cache') or released['volumes']['laya-model-cache'])
        require(not self.laya_enabled or worker, 'The target release does not support the saved Laya worker.')
        candidate['services'][SERVICES[0]].setdefault('environment', {}).update({
            'LAYA_MATCH_ENABLED': 'true' if self.laya_enabled else 'false',
            'LAYA_MATCH_URL': 'http://laya:8000',
            'TINYFISH_API_KEY': self.tinyfish_api_key})
        for service in self.services:
            context = source / 'voltvault' / ('voltvault-server' if service == OPTIONAL_SERVICE else service)
            require((context / 'Dockerfile').is_file(), 'Incomplete release archive.')
            dockerfile = 'Dockerfile.laya' if service == OPTIONAL_SERVICE else 'Dockerfile'
            require((context / dockerfile).is_file(), 'Incomplete release archive.')
            if not images or service == OPTIONAL_SERVICE:
                candidate['services'][service]['build'] = {'context': str(context).replace('$', '$$'), 'dockerfile': dockerfile}
            else:
                candidate['services'][service].pop('build', None)
            candidate['services'][service]['image'] = f'voltvault-upgrade-{service}:{VERSION}-{self.state["token"][:12]}'
        write_json(self.transaction / 'build.json', candidate)
        if images:
            build_services = (OPTIONAL_SERVICE,) if self.laya_enabled else ()
        else:
            build_services = self.services
        if build_services:
            print('[2/5] Building enabled components. Your existing installation is still running…', flush=True)
            self.compose([self.transaction / 'build.json'], 'build', *build_services, visible=True)
        for service in SERVICES:
            candidate['services'][service].pop('build', None)
        write_json(self.transaction / 'candidate.json', candidate)
        closed = copy.deepcopy(candidate)
        closed['services'][SERVICES[1]].pop('ports', None)
        # Suppress background printer/backup writes only during closed-access checks.
        closed['services'][SERVICES[0]]['environment']['DEVICE_CREDENTIAL_KEY'] = ''
        closed['services'][SERVICES[0]]['environment']['BACKUP_SCHEDULE'] = 'disabled'
        closed['services'][SERVICES[0]]['environment']['UPGRADE_VALIDATION_MODE'] = 'true'
        write_json(self.transaction / 'closed.json', closed)
        self.save('staged')

    def stop(self):
        # Service selection avoids incidental helper containers; never remove volumes.
        self.compose([self.transaction / 'previous.json'], 'stop', *self.services)

    def backup(self):
        print('[3/5] Taking the app offline and creating a verified backup…', flush=True)
        self.save('stopping')
        self.stop()
        self.save('backing-up')
        baseline = self.snapshot()
        for target in ('data', 'uploads'):
            destination = self.transaction / ('backup-' + target)
            destination.mkdir()
            private(destination)
            docker('cp', f'{self.by_service[SERVICES[0]]["Id"]}:/{target}/.', str(destination))
            private(destination)
            require(host_manifest(destination) == content_manifest(baseline['manifest']['/' + target]),
                    'Backup copy verification failed. Upgrade has not begun.')
        write_json(self.transaction / 'baseline.json', baseline)
        self.save('backed-up')

    def validate_candidate(self):
        print('[4/5] Checking the new version while browser access is closed…', flush=True)
        self.save('validating')
        self.compose([self.transaction / 'closed.json'], 'up', '-d', '--no-build', '--wait', '--wait-timeout', '300')
        baseline = json.loads((self.transaction / 'baseline.json').read_text())
        # Stop the server before reading SQLite through a helper process.
        self.compose([self.transaction / 'closed.json'], 'stop', SERVICES[0])
        after = self.snapshot(baseline['tables'], self.config_image(SERVICES[0]))
        changed_tables = sorted(name for name in set(baseline['tables']) | set(after['tables'])
                                if baseline['tables'].get(name) != after['tables'].get(name))
        require(not changed_tables, 'Existing inventory/account/printer records changed during upgrade: ' +
                ', '.join(changed_tables) + '.')
        require(content_manifest(after['manifest']['/uploads']) == content_manifest(baseline['manifest']['/uploads']),
                'Upload verification failed.')
        self.compose([self.transaction / 'closed.json'], 'up', '-d', '--no-build', '--wait', '--wait-timeout', '300')
        # Caddy's internal proxy exercises both the browser and API without exposing ports.
        self.compose([self.transaction / 'closed.json'], 'exec', '-T', SERVICES[1],
                     'wget', '-q', '-O', '/dev/null', 'http://127.0.0.1/api/ready')
        self.save('validated')

    def config_image(self, service):
        return json.loads((self.transaction / 'candidate.json').read_text())['services'][service]['image']

    def reopen(self):
        print('[5/5] Restoring your browser address…', flush=True)
        candidate = json.loads((self.transaction / 'candidate.json').read_text())
        write_json(self.root / 'active.json', candidate)
        # Persist BEFORE publishing a port: recovery must never discard possible new writes.
        self.save('reopening')
        self.compose([self.root / 'active.json'], 'up', '-d', '--no-build', '--wait', '--wait-timeout', '300')
        opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
        with opener.open(self.url + '/api/ready', timeout=15) as response:
            require(response.status == 200, 'Browser readiness failed after access reopened.')
        self.save('complete')
        print(f'VoltVault {self.state["target"]} is ready. Open {self.url}\nVerified backup: {self.transaction}\nKeep an off-machine copy of this backup and your .env.', flush=True)

    def rollback(self):
        print('Upgrade failed; restoring the previous version while access is closed…', flush=True)
        phase = self.state['phase']
        require(phase not in ('reopening', 'rollback-reopening', 'complete', 'rolled-back'), 'Browser access may have reopened. Automatic data restore is disabled to protect new writes. ' + GUIDE)
        if phase in ('staging', 'staged'):
            self.save('aborted')
            return
        if phase in ('stopping', 'backing-up'):
            self.finish_rollback()
            return
        require(phase in ('backed-up', 'validating', 'validated', 'restoring', 'restore-failed'), 'Unrecognized recovery phase.')
        self.save('restoring')
        self.compose([self.transaction / 'previous.json'], 'stop', *self.services)
        baseline = json.loads((self.transaction / 'baseline.json').read_text())
        for target in ('data', 'uploads'):
            require(host_manifest(self.transaction / ('backup-' + target)) == content_manifest(baseline['manifest']['/' + target]),
                    'Recovery backup checksum mismatch; refusing to overwrite volumes.')
        self.helper({'mode': 'clear'})
        helper_name = 'voltvault-restore-' + self.state['token'][:12]
        # A stopped temporary container makes docker cp portable across Docker Desktop/Engine.
        existing = docker('ps', '-a', '--filter', 'name=^/' + helper_name + '$', '--format', '{{.ID}}').strip()
        if existing:
            docker('rm', existing)
        helper_id = docker('create', '--name', helper_name, '--network', 'none', '--user', '0', '--entrypoint', 'node',
                           '--mount', f'type=volume,src={self.volumes["/data"]},dst=/data',
                           '--mount', f'type=volume,src={self.volumes["/uploads"]},dst=/uploads',
                           self.old_image, '-e', 'process.exit(0)').strip()
        try:
            for target in ('data', 'uploads'):
                docker('cp', str(self.transaction / ('backup-' + target)) + '/.', f'{helper_id}:/{target}/')
        finally:
            docker('rm', helper_id)
        self.helper({'mode': 'restore-permissions', 'manifest': baseline['manifest']})
        restored = self.snapshot(baseline['tables'])
        require(restored['tables'] == baseline['tables'] and
                all(content_manifest(restored['manifest'][root]) == content_manifest(baseline['manifest'][root]) for root in ('/data', '/uploads')),
                'Restored volume verification failed.')
        if self.state['prior_active']:
            write_json(self.root / 'active.json', json.loads(self.state['prior_active']))
        else:
            (self.root / 'active.json').unlink(missing_ok=True)
        previous = json.loads((self.transaction / 'previous.json').read_text())
        previous['services'][SERVICES[1]].pop('ports', None)
        write_json(self.transaction / 'previous-closed.json', previous)
        self.compose([self.transaction / 'previous-closed.json'], 'up', '-d', '--no-build', '--wait', '--wait-timeout', '300')
        self.compose([self.transaction / 'previous-closed.json'], 'exec', '-T', SERVICES[1],
                     'wget', '-q', '-O', '/dev/null', 'http://127.0.0.1/api/ready')
        self.finish_rollback()

    def finish_rollback(self):
        self.save('rollback-reopening')
        self.compose([self.transaction / 'previous.json'], 'up', '-d', '--no-build', '--force-recreate', '--wait', '--wait-timeout', '300')
        self.save('rolled-back')
        print(f'Previous version restored. Open {self.url}\nBackup retained: {self.transaction}', flush=True)

    def execute(self, recover=False):
        current = 'unknown (older manual deployment)'
        marker = self.path / '.voltvault-install'
        if marker.is_file():
            current = marker.read_text().strip().split(':')[-1]
        if self.state and self.state['phase'] == 'complete':
            current = self.state['target']
            if current == VERSION and self.state['payload_hash'] == PAYLOAD_SHA256:
                require(not recover, 'No interrupted upgrade requires recovery.')
                print(f'Already up to date ({VERSION}). Open {self.url}')
                return
            require(tuple(map(int, current.split('.'))) < tuple(map(int, VERSION.split('.'))), 'Downgrades are unsupported. ' + GUIDE)
        elif current != 'unknown (older manual deployment)':
            require(tuple(map(int, current.split('.'))) <= tuple(map(int, VERSION.split('.'))), 'Downgrades are unsupported. ' + GUIDE)
        pending = self.state and self.state['phase'] not in ('complete', 'aborted', 'rolled-back')
        require(not pending or recover, 'An interrupted upgrade needs recovery. Rerun this command with --recover (PowerShell: -Recover).')
        require(not recover or pending, 'No interrupted upgrade requires recovery.')
        print(f'Installation: {self.path}\nCurrent release: {current}\nTarget release: {VERSION}\nBrowser: {self.url}\nThe app stays online while release images download; downtime depends on backup size and validation.', flush=True)
        if pending:
            print(f'Recovery phase: {self.state["phase"]}\nBackup: {self.transaction}', flush=True)
        finishing = pending and self.state['phase'] in ('reopening', 'rollback-reopening')
        if finishing:
            print('Access may have reopened. Recovery will retry the connection using the upgraded version and preserve current data.', flush=True)
        answer = input('Retry the browser connection without restoring data? [y/N] ' if finishing else 'Recover the previous version? [y/N] ' if recover else 'Back up and upgrade this installation? [y/N] ')
        if answer.strip().lower() not in ('y', 'yes'):
            print('Cancelled. Installation unchanged.')
            return
        self.acquire()
        try:
            require(digest(self.env_file) == self.env_hash and
                    (json.loads(self.state_file.read_text()) if self.state_file.exists() else None) == self.state,
                    'Installation changed while awaiting confirmation. Rerun to review the current deployment.')
            if recover:
                self.old_image = self.state['old_image']
                if finishing:
                    if self.state['phase'] == 'rollback-reopening':
                        self.finish_rollback()
                    else:
                        self.reopen()
                else:
                    try:
                        self.rollback()
                    except (Exception, KeyboardInterrupt):
                        if self.state['phase'] != 'rollback-reopening':
                            try:
                                self.compose([self.transaction / 'previous.json'], 'stop', *self.services)
                            except Exception:
                                pass
                            self.save('restore-failed')
                        print(f'Recovery needs attention. Backup: {self.transaction}\nRerun with --recover (PowerShell: -Recover).', file=sys.stderr)
                        raise
                return
            self.initialize()
            try:
                phase_started = time.monotonic()
                self.stage()
                print(f'Release preparation: {time.monotonic() - phase_started:.1f}s (app online)', flush=True)
                phase_started = time.monotonic()
                self.backup()
                print(f'Backup and access closure: {time.monotonic() - phase_started:.1f}s', flush=True)
                phase_started = time.monotonic()
                self.validate_candidate()
                print(f'Upgrade validation: {time.monotonic() - phase_started:.1f}s', flush=True)
                phase_started = time.monotonic()
                self.reopen()
                print(f'Browser reopen: {time.monotonic() - phase_started:.1f}s', flush=True)
            except (Exception, KeyboardInterrupt):
                if self.state['phase'] not in ('reopening', 'rollback-reopening', 'complete', 'rolled-back'):
                    try:
                        self.rollback()
                    except (Exception, KeyboardInterrupt):
                        if self.state['phase'] != 'rollback-reopening':
                            # Failed closed-access restoration must remain offline.
                            try:
                                self.compose([self.transaction / 'previous.json'], 'stop', *self.services)
                            except Exception:
                                pass
                            self.save('restore-failed')
                            print('Recovery needs attention; services should remain stopped.', file=sys.stderr)
                        else:
                            print('Previous data was retained/restored; browser access may be open. No further automatic database restore.', file=sys.stderr)
                        print(f'Backup: {self.transaction}\nRerun with --recover (PowerShell: -Recover).', file=sys.stderr)
                else:
                    print(f'Browser access may be open. Data has not been rolled back.\nBackup: {self.transaction}\nCheck Docker logs and {GUIDE}', file=sys.stderr)
                raise
        finally:
            self.release()


def main():
    require(sys.version_info >= (3, 9), 'Python 3.9 or newer is required.')
    os.umask(0o077)
    parser = argparse.ArgumentParser(description='Back up and upgrade an official full-stack VoltVault Docker installation.')
    parser.add_argument('--dir', help='Existing Compose directory (auto-detected by default)')
    parser.add_argument('--bundle', help='Verified local installer ZIP, for upgrading without website access')
    parser.add_argument('--recover', action='store_true', help='Restore an interrupted closed-access upgrade')
    args = parser.parse_args()
    bundle = None
    if args.bundle:
        bundle = Path(args.bundle).expanduser()
        require(bundle.is_file(), 'Local release file is missing; existing containers are untouched.')
        bundle = bundle.resolve()
        require(digest(bundle) == PAYLOAD_SHA256,
                'Local release verification failed; existing containers are untouched.')
    docker('info')
    require('--wait-timeout' in docker('compose', 'up', '--help'), 'Update Docker Compose; --wait-timeout is required.')
    # SIGTERM follows the same safe recovery path as Ctrl-C. SIGKILL uses saved phase metadata.
    signal.signal(signal.SIGTERM, lambda *_: (_ for _ in ()).throw(KeyboardInterrupt()))
    Upgrader(*discover(args.dir), bundle=bundle).execute(args.recover)


if __name__ == '__main__':
    try:
        main()
    except (UpgradeError, OSError, ValueError, EOFError, KeyboardInterrupt) as error:
        print(f'VoltVault: {error if isinstance(error, UpgradeError) else "Upgrade interrupted or failed. See the recovery instructions above."}', file=sys.stderr)
        sys.exit(1)