shopdb-flask/shopdb/plugins/cli.py

"""Flask CLI commands for plugin management."""

import json
from pathlib import Path

import click
from flask import current_app
from flask.cli import with_appcontext


# JSON-Schema primitive name -> Python type(s) for the no-dependency validator.
_SCHEMA_TYPES = {
    'string': str,
    'boolean': bool,
    'array': list,
    'object': dict,
    'integer': int,
    'number': (int, float),
}


def _load_manifest_schema() -> dict:
    """Load the packaged manifest schema (ships with the app, unlike docs/)."""
    schema_path = Path(__file__).with_name('manifest_schema.json')
    with open(schema_path) as f:
        return json.load(f)


def _check_against_schema(manifest: dict, schema: dict) -> list:
    """Lightweight schema check without a jsonschema dependency.

    Verifies required fields are present and that known typed fields hold the
    right JSON type; unknown fields are allowed (additionalProperties). Returns
    a list of human-readable error strings (empty = valid).
    """
    errors = []
    for field in schema.get('required', []):
        if field not in manifest:
            errors.append(f"missing required field '{field}'")
    props = schema.get('properties', {})
    for key, value in manifest.items():
        spec = props.get(key)
        if not spec:
            continue  # additionalProperties permitted
        expected = spec.get('type')
        pytype = _SCHEMA_TYPES.get(expected)
        # bool is a subclass of int; guard so a boolean does not pass 'integer'
        if pytype and (not isinstance(value, pytype)
                       or (expected in ('integer', 'number')
                           and isinstance(value, bool))):
            errors.append(f"field '{key}' should be {expected}")
        if spec.get('enum') and value not in spec['enum']:
            errors.append(
                f"field '{key}' must be one of {spec['enum']}, got '{value}'")
    return errors


@click.group('plugin')
def plugin_cli():
    """Plugin management commands."""
    pass


@plugin_cli.command('list')
@with_appcontext
def list_plugins():
    """List all available plugins."""
    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        return

    plugins = pm.discover_available()

    if not plugins:
        click.echo("No plugins found in plugins directory.")
        return

    # Format output
    click.echo("")
    click.echo(click.style("Available Plugins:", fg='cyan', bold=True))
    click.echo("-" * 60)

    for p in plugins:
        if p['enabled']:
            status = click.style("[Enabled]", fg='green')
        elif p['installed']:
            status = click.style("[Disabled]", fg='yellow')
        else:
            status = click.style("[Available]", fg='white')

        click.echo(f"  {p['name']:20} v{p['version']:10} {status}")
        if p['description']:
            click.echo(f"    {p['description'][:55]}...")
        if p['dependencies']:
            deps = ', '.join(p['dependencies'])
            click.echo(f"    Dependencies: {deps}")

    click.echo("")


@plugin_cli.command('install')
@click.argument('name')
@click.option('--skip-migrations', is_flag=True, help='Skip database migrations')
@with_appcontext
def install_plugin(name: str, skip_migrations: bool):
    """
    Install a plugin.

    Usage: flask plugin install printers
    """
    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        raise SystemExit(1)

    click.echo(f"Installing plugin: {name}")

    if pm.install_plugin(name, run_migrations=not skip_migrations):
        click.echo(click.style(f"Successfully installed {name}", fg='green'))
    else:
        click.echo(click.style(f"Failed to install {name}", fg='red'))
        raise SystemExit(1)


@plugin_cli.command('uninstall')
@click.argument('name')
@click.option('--remove-data', is_flag=True, help='Remove plugin database tables')
@click.confirmation_option(prompt='Are you sure you want to uninstall this plugin?')
@with_appcontext
def uninstall_plugin(name: str, remove_data: bool):
    """
    Uninstall a plugin.

    Usage: flask plugin uninstall printers
    """
    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        raise SystemExit(1)

    click.echo(f"Uninstalling plugin: {name}")

    if pm.uninstall_plugin(name, remove_data=remove_data):
        click.echo(click.style(f"Successfully uninstalled {name}", fg='green'))
    else:
        click.echo(click.style(f"Failed to uninstall {name}", fg='red'))
        raise SystemExit(1)


@plugin_cli.command('enable')
@click.argument('name')
@with_appcontext
def enable_plugin(name: str):
    """Enable a disabled plugin."""
    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        raise SystemExit(1)

    if pm.enable_plugin(name):
        click.echo(click.style(f"Enabled {name}", fg='green'))
        # Nudge the operator if the plugin's chain is ahead of the DB.
        try:
            if pm.migration_manager and \
                    pm.migration_manager.has_unapplied_migrations(name):
                click.echo(click.style(
                    f"  {name} has unapplied migrations - "
                    f"run 'flask plugin upgrade-all'", fg='yellow'))
        except Exception:
            pass
    else:
        click.echo(click.style(f"Failed to enable {name}", fg='red'))
        raise SystemExit(1)


@plugin_cli.command('disable')
@click.argument('name')
@with_appcontext
def disable_plugin(name: str):
    """Disable an enabled plugin."""
    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        raise SystemExit(1)

    if pm.disable_plugin(name):
        click.echo(click.style(f"Disabled {name}", fg='green'))
    else:
        click.echo(click.style(f"Failed to disable {name}", fg='red'))
        raise SystemExit(1)


@plugin_cli.command('info')
@click.argument('name')
@with_appcontext
def plugin_info(name: str):
    """Show detailed information about a plugin."""
    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        raise SystemExit(1)

    plugin_class = pm.loader.load_plugin_class(name)
    if not plugin_class:
        click.echo(click.style(f"Plugin {name} not found", fg='red'))
        raise SystemExit(1)

    try:
        temp = plugin_class()
        meta = temp.meta
    except Exception as e:
        click.echo(click.style(f"Error loading plugin: {e}", fg='red'))
        raise SystemExit(1)

    state = pm.registry.get(name)

    click.echo("")
    click.echo("=" * 50)
    click.echo(click.style(f"Plugin: {meta.name}", fg='cyan', bold=True))
    click.echo("=" * 50)
    click.echo(f"Version:      {meta.version}")
    click.echo(f"Description:  {meta.description}")
    click.echo(f"Author:       {meta.author or 'Unknown'}")
    click.echo(f"API Prefix:   {meta.api_prefix}")
    click.echo(f"Dependencies: {', '.join(meta.dependencies) or 'None'}")
    click.echo(f"Core Version: {meta.core_version}")
    click.echo("")

    if state:
        status = click.style('Enabled', fg='green') if state.enabled else click.style('Disabled', fg='yellow')
        click.echo(f"Status:       {status}")
        click.echo(f"Installed:    {state.installed_at}")
        click.echo(f"Migrations:   {len(state.migrations_applied)} applied")
    else:
        click.echo(f"Status:       {click.style('Not installed', fg='white')}")

    click.echo("")


@plugin_cli.command('new')
@click.argument('name')
@click.option('--description', default='', help='One-sentence plugin description')
@click.option('--overwrite', is_flag=True, help='Overwrite existing plugin directory')
@with_appcontext
def new_plugin(name: str, description: str, overwrite: bool):
    """Scaffold a new plugin from the bundled templates.

    Usage: flask plugin new cameras --description "Tracks shop-floor cameras"
    """
    from pathlib import Path
    from .scaffolder import scaffold_plugin, ScaffoldError

    plugins_dir = Path(current_app.root_path).parent / 'plugins'

    if not description:
        description = f'{name.capitalize()} plugin (TODO: replace this description)'

    try:
        target = scaffold_plugin(
            name=name,
            description=description,
            plugins_dir=plugins_dir,
            overwrite=overwrite,
        )
    except ScaffoldError as e:
        click.echo(click.style(f'Scaffold failed: {e}', fg='red'))
        raise SystemExit(1)

    click.echo(click.style(f'Created plugin at {target}', fg='green'))
    click.echo('')
    click.echo('Next steps:')
    click.echo(f'  1. Edit plugins/{name}/models/{name}.py with your domain fields')
    click.echo(f'  2. Edit plugins/{name}/api/routes.py with your endpoints')
    click.echo(f'  3. Add plugins/{name}/migrations/ with an Alembic chain that')
    click.echo(f'     creates your tables (per-plugin chain, NOT the core chain;')
    click.echo(f'     see ADR-008). Register the tables in PLUGIN_TABLE_OWNERS.')
    click.echo(f'  4. Run: flask plugin install {name}')
    click.echo(f'  5. Run: flask plugin upgrade-all')
    click.echo(f'  6. Run: pytest plugins/{name}/tests/')


def _dep_name(dep: str) -> str:
    """Bare plugin name from a dependency spec, dropping any PEP440 range."""
    for sep in ('>', '<', '=', '!', '~', ' '):
        dep = dep.split(sep)[0]
    return dep.strip()


def _validate_directory(pm, name: str) -> list:
    """Run the directory-mode checks, echoing each. Returns a failures list."""
    from shopdb import __contract_version__
    from ..exceptions import PluginError

    failures = []

    try:
        manifest = pm.loader.load_manifest(name)
    except PluginError as e:
        click.echo(click.style(f"  manifest: {e}", fg='red'))
        return [str(e)]
    click.echo(click.style(
        "  manifest loads + name matches directory", fg='green'))

    schema_errors = _check_against_schema(manifest, _load_manifest_schema())
    if schema_errors:
        for err in schema_errors:
            failures.append(f"schema: {err}")
            click.echo(click.style(f"  schema: {err}", fg='red'))
    else:
        click.echo(click.style("  schema OK", fg='green'))

    try:
        pm.loader.check_contract_version(name, __contract_version__)
        click.echo(click.style(
            f"  core_version admits contract {__contract_version__}", fg='green'))
    except PluginError as e:
        failures.append(str(e))
        click.echo(click.style(f"  core_version: {e}", fg='red'))

    available = set(pm.loader.discover_plugins())
    dep_failures = []
    for dep in manifest.get('dependencies', []):
        depname = _dep_name(dep)
        if depname not in available:
            dep_failures.append(depname)
            failures.append(f"dependency '{depname}' not found on disk")
            click.echo(click.style(
                f"  dependency '{depname}' not found on disk", fg='red'))
    if manifest.get('dependencies') and not dep_failures:
        click.echo(click.style("  dependencies present on disk", fg='green'))

    return failures


def _validate_artifact(pubkey_paths, artifact_path: str) -> list:
    """Verify a signed .shopdbplugin: signature + hashes + manifest schema +
    contract. Returns a failures list."""
    from shopdb import __contract_version__
    from packaging.specifiers import SpecifierSet
    from packaging.version import Version
    from .packaging import verify_artifact
    from .signing import load_trusted_keys

    keys = load_trusted_keys(pubkey_paths)
    if not keys:
        click.echo(click.style(
            "  no trusted keys (pass --pubkey or set PLUGIN_TRUSTED_KEYS)",
            fg='red'))

    manifest, errors = verify_artifact(artifact_path, keys)
    failures = list(errors)
    for err in errors:
        click.echo(click.style(f"  {err}", fg='red'))
    if not errors:
        click.echo(click.style(
            "  signature trusted + every file hash intact", fg='green'))

    if manifest is not None:
        schema_errors = _check_against_schema(
            manifest, _load_manifest_schema())
        for err in schema_errors:
            failures.append(f"schema: {err}")
            click.echo(click.style(f"  schema: {err}", fg='red'))
        if not schema_errors:
            click.echo(click.style("  schema OK", fg='green'))

        spec = manifest.get('core_version', '')
        if spec and Version(__contract_version__) not in SpecifierSet(spec):
            failures.append(
                f"core_version {spec} excludes contract {__contract_version__}")
            click.echo(click.style(
                f"  core_version {spec} excludes contract "
                f"{__contract_version__}", fg='red'))
        elif spec:
            click.echo(click.style(
                f"  core_version admits contract {__contract_version__}",
                fg='green'))

    return failures


@plugin_cli.command('validate')
@click.argument('target')
@click.option('--pubkey', 'pubkeys', multiple=True,
              type=click.Path(exists=True),
              help='Trusted public key PEM (artifact mode). Repeatable.')
@with_appcontext
def validate_plugin(target: str, pubkeys):
    """Validate a plugin - directory (by name) or a signed artifact (by path).

    Directory mode (`flask plugin validate printers`): manifest loads + name
    match, manifest schema, core_version admits this contract, dependencies
    exist on disk.

    Artifact mode (`flask plugin validate dist/printers-1.0.0.shopdbplugin`):
    signature against trusted keys (--pubkey, else PLUGIN_TRUSTED_KEYS), every
    file hash, no unexpected files, manifest schema + core_version. Exits
    non-zero on any failure.
    """
    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        raise SystemExit(1)

    is_artifact = target.endswith('.shopdbplugin') and Path(target).is_file()
    if is_artifact:
        keys = list(pubkeys) or current_app.config.get(
            'PLUGIN_TRUSTED_KEYS', [])
        failures = _validate_artifact(keys, target)
        label = Path(target).name
    else:
        failures = _validate_directory(pm, target)
        label = target

    click.echo("")
    if failures:
        click.echo(click.style(
            f"{label}: INVALID ({len(failures)} problem(s))", fg='red'))
        raise SystemExit(1)
    click.echo(click.style(f"{label}: valid", fg='green'))


@plugin_cli.command('keygen')
@click.option('--out', 'out_dir', default='.', type=click.Path(),
              help='Directory to write the key pair into')
@click.option('--name', 'key_name', default='curator',
              help='Base filename for the pair')
def keygen(out_dir: str, key_name: str):
    """Generate an ed25519 publisher key pair for signing plugin artifacts.

    Writes <name>.key (PRIVATE - keep offline with the curator) and <name>.pub
    (public - pin on each site via PLUGIN_TRUSTED_KEYS). Not app-bound.
    """
    from .signing import generate_keypair

    out = Path(out_dir)
    out.mkdir(parents=True, exist_ok=True)
    private_pem, public_pem = generate_keypair()

    private_path = out / f'{key_name}.key'
    public_path = out / f'{key_name}.pub'
    private_path.write_bytes(private_pem)
    public_path.write_bytes(public_pem)
    try:
        private_path.chmod(0o600)
    except OSError:
        pass

    click.echo(click.style(f"Private key: {private_path}", fg='green'))
    click.echo(click.style(f"Public key:  {public_path}", fg='green'))
    click.echo("")
    click.echo(click.style(
        "Keep the .key OFFLINE. Never place it on the plugin shelf. Distribute "
        "the .pub with each site's config and list it in PLUGIN_TRUSTED_KEYS.",
        fg='yellow'))


@plugin_cli.command('pack')
@click.argument('name')
@click.option('--key', 'key_path', required=True,
              type=click.Path(exists=True), help='Private signing key PEM')
@click.option('--publisher', default='',
              help='Publisher id recorded in the provenance')
@click.option('--out', 'out_dir', default=None, type=click.Path(),
              help='Output directory (default: the plugins/ directory)')
@with_appcontext
def pack_plugin(name: str, key_path: str, publisher: str, out_dir):
    """Validate a plugin directory, then emit a signed artifact.

    Produces <name>-<version>.shopdbplugin. Fails without packing if the
    directory does not validate.

    Usage: flask plugin pack printers --key curator.key --publisher west-jefferson
    """
    from .signing import load_private_key
    from .packaging import pack

    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        raise SystemExit(1)

    click.echo(f"Validating {name} ...")
    failures = _validate_directory(pm, name)
    if failures:
        click.echo("")
        click.echo(click.style(
            f"Refusing to pack {name}: {len(failures)} validation problem(s)",
            fg='red'))
        raise SystemExit(1)

    plugin_dir = Path(pm.loader.plugins_dir) / name
    private_key = load_private_key(Path(key_path).read_bytes())
    artifact = pack(plugin_dir, private_key,
                    publisher=publisher, out_dir=out_dir)
    click.echo("")
    click.echo(click.style(f"Packed + signed: {artifact}", fg='green'))


@plugin_cli.command('apply-profile')
@click.argument('profile', type=click.Path(exists=True, dir_okay=False))
@with_appcontext
def apply_profile(profile: str):
    """Install AND enable exactly the plugins named in a site profile.

    Declarative site setup: replaces the hand-ordered install/enable sequences
    in the deploy runbooks. Resolves the hard-dependency closure and applies it
    in dependency order; idempotent. Does NOT remove anything absent from the
    list.

    Profile JSON shape:
      { "site": "west-jefferson",
        "plugins": ["machines", "printers", "computers"],
        "locked": ["computers"] }

    Usage: flask plugin apply-profile site-profile.json
    """
    from ..exceptions import PluginError

    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        raise SystemExit(1)

    with open(profile) as f:
        data = json.load(f)
    names = data.get('plugins', [])
    if not isinstance(names, list) or not names:
        click.echo(click.style(
            "Profile has no 'plugins' list to apply", fg='red'))
        raise SystemExit(1)

    click.echo(f"Applying profile: {data.get('site', profile)}")
    try:
        result = pm.apply_profile(names, locked=data.get('locked'))
    except PluginError as e:
        click.echo(click.style(f"Profile failed: {e}", fg='red'))
        raise SystemExit(1)

    if result['installed']:
        click.echo(click.style(
            f"  installed: {', '.join(result['installed'])}", fg='green'))
    if result['enabled']:
        click.echo(click.style(
            f"  enabled:   {', '.join(result['enabled'])}", fg='green'))
    if result['already']:
        click.echo(click.style(
            f"  unchanged: {', '.join(result['already'])}", fg='white'))
    click.echo("")
    click.echo(click.style(
        "Run 'flask plugin upgrade-all' to apply plugin migrations, then "
        "restart so new blueprints/routes register.", fg='yellow'))


def _shelf_keys(pubkeys):
    """Trusted keys for shelf/artifact verification: explicit --pubkey first,
    else the site's PLUGIN_TRUSTED_KEYS."""
    from .signing import load_trusted_keys
    paths = list(pubkeys) or current_app.config.get('PLUGIN_TRUSTED_KEYS', [])
    return load_trusted_keys(paths), paths


@plugin_cli.command('stamp-bundled')
@click.option('--key', 'key_path', required=True, type=click.Path(exists=True),
              help='Private signing key PEM')
@click.option('--publisher', default='', help='Publisher id in the provenance')
@click.argument('name', required=False)
@with_appcontext
def stamp_bundled(key_path, publisher, name):
    """Write PROVENANCE.json/.sig into in-tree plugin dirs (image build step).

    Stamps one plugin (NAME) or every discovered plugin, so verify-at-load
    applies to bundled plugins identically to adopted ones. Run at image build
    with the site/build key, then set PLUGIN_REQUIRE_SIGNED.
    """
    from .signing import (load_private_key, build_provenance,
                          serialize_provenance, sign,
                          PROVENANCE_NAME, PROVENANCE_SIG)
    from .packaging import strip_bytecode

    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        raise SystemExit(1)

    private_key = load_private_key(Path(key_path).read_bytes())
    names = [name] if name else pm.loader.discover_plugins()
    plugins_dir = Path(pm.loader.plugins_dir)

    for plugin_name in names:
        plugin_dir = plugins_dir / plugin_name
        # Stamp a clean tree: no bytecode, so verify-at-load never trips on a
        # cache the provenance does not cover.
        strip_bytecode(plugin_dir)
        manifest = pm.loader.load_manifest(plugin_name)
        provenance = build_provenance(
            plugin_dir, manifest['name'], manifest['version'], publisher)
        provenance_bytes = serialize_provenance(provenance)
        (plugin_dir / PROVENANCE_NAME).write_bytes(provenance_bytes)
        (plugin_dir / PROVENANCE_SIG).write_bytes(
            sign(private_key, provenance_bytes))
        click.echo(click.style(f"  stamped {plugin_name}", fg='green'))


@plugin_cli.command('shelf-build')
@click.option('--dir', 'shelf_dir', required=True,
              type=click.Path(exists=True, file_okay=False),
              help='Shelf directory holding the artifacts')
@click.option('--key', 'key_path', required=True,
              type=click.Path(exists=True), help='Private signing key PEM')
@click.option('--serial', type=int, required=True,
              help='Index serial (must exceed the current published serial)')
@click.option('--publisher', default='')
def shelf_build(shelf_dir, key_path, serial, publisher):
    """Build + sign the shelf index from the artifacts in DIR.

    Carries forward the existing revoked list and refuses a serial that does not
    advance past the published one (a lower serial would let a site's
    anti-rollback reject the new index).
    """
    import json as _json
    from .signing import load_private_key
    from .shelf import build_index, INDEX_NAME

    shelf = Path(shelf_dir)
    revoked = []
    existing = shelf / INDEX_NAME
    if existing.exists():
        try:
            prev = _json.loads(existing.read_text())
            revoked = prev.get('revoked', [])
            if serial <= int(prev.get('serial', -1)):
                click.echo(click.style(
                    f"serial {serial} must exceed current "
                    f"{prev.get('serial')}", fg='red'))
                raise SystemExit(1)
        except (ValueError, TypeError):
            pass

    private_key = load_private_key(Path(key_path).read_bytes())
    index = build_index(shelf, private_key, serial,
                        publisher=publisher, revoked=revoked)
    total = sum(len(v) for v in index['plugins'].values())
    click.echo(click.style(
        f"Shelf index signed: serial {serial}, {total} artifact(s), "
        f"{len(revoked)} revoked.", fg='green'))


@plugin_cli.command('shelf-list')
@click.option('--pubkey', 'pubkeys', multiple=True, type=click.Path(exists=True))
@with_appcontext
def shelf_list(pubkeys):
    """Browse the configured shelf (verifies index signature + serial)."""
    from shopdb import __contract_version__
    from packaging.specifiers import SpecifierSet
    from packaging.version import Version
    from .shelf import load_index, read_last_serial, write_last_serial, is_revoked

    shelf_dir = current_app.config.get('PLUGIN_SHELF_DIR')
    if not shelf_dir:
        click.echo(click.style("PLUGIN_SHELF_DIR not configured", fg='red'))
        raise SystemExit(1)

    keys, _ = _shelf_keys(pubkeys)
    index, errors = load_index(shelf_dir, keys)
    for err in errors:
        click.echo(click.style(f"  {err}", fg='red'))
    if index is None:
        raise SystemExit(1)

    last = read_last_serial(current_app.instance_path)
    if index['serial'] < last:
        click.echo(click.style(
            f"REFUSING shelf: index serial {index['serial']} < last seen "
            f"{last} (rollback?)", fg='red'))
        raise SystemExit(1)
    if errors:
        raise SystemExit(1)
    write_last_serial(current_app.instance_path, index['serial'])

    click.echo("")
    click.echo(click.style(
        f"Shelf (serial {index['serial']}):", fg='cyan', bold=True))
    for name in sorted(index.get('plugins', {})):
        for entry in index['plugins'][name]:
            compat = 'ok'
            spec = entry.get('core_version', '')
            if spec and Version(__contract_version__) not in SpecifierSet(spec):
                compat = f'needs core {spec}'
            flags = []
            if is_revoked(index, name, entry['version']):
                flags.append('REVOKED')
            if compat != 'ok':
                flags.append(compat)
            tag = ('  [' + ', '.join(flags) + ']') if flags else ''
            click.echo(f"  {name:20} {entry['version']:10} "
                       f"({entry.get('tier', 'optional')}){tag}")


@plugin_cli.command('adopt')
@click.argument('spec')
@click.option('--pubkey', 'pubkeys', multiple=True, type=click.Path(exists=True))
@click.option('--force-downgrade', is_flag=True,
              help='Allow adopting a version lower than the installed one')
@with_appcontext
def adopt_plugin(spec, pubkeys, force_downgrade):
    """Pull a plugin from the shelf, verify it, and install + enable it.

    SPEC is a plugin name or name==version. Verifies the shelf index and the
    artifact signature + hashes, unpacks atomically only after verification,
    then installs + enables the plugin and its dependency closure.

    Usage: flask plugin adopt printers        (or printers==1.2.0)
    """
    from packaging.version import Version
    from ..exceptions import PluginError
    from .shelf import (load_index, read_last_serial, write_last_serial,
                        resolve_version, unpack_verified)

    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        raise SystemExit(1)

    name, _, version = spec.partition('==')
    name = name.strip()
    version = version.strip() or None

    shelf_dir = current_app.config.get('PLUGIN_SHELF_DIR')
    if not shelf_dir:
        click.echo(click.style("PLUGIN_SHELF_DIR not configured", fg='red'))
        raise SystemExit(1)

    keys, _ = _shelf_keys(pubkeys)
    index, errors = load_index(shelf_dir, keys)
    for err in errors:
        click.echo(click.style(f"  {err}", fg='red'))
    if index is None or errors:
        raise SystemExit(1)

    last = read_last_serial(current_app.instance_path)
    if index['serial'] < last:
        click.echo(click.style(
            f"REFUSING shelf: serial {index['serial']} < last seen {last} "
            f"(rollback?)", fg='red'))
        raise SystemExit(1)

    try:
        resolved_version, artifact_name, expected_sha = resolve_version(
            index, name, version)
    except (KeyError, ValueError) as e:
        click.echo(click.style(f"{e}", fg='red'))
        raise SystemExit(1)

    state = pm.registry.get(name)
    if state and not force_downgrade:
        try:
            if Version(resolved_version) < Version(state.version):
                click.echo(click.style(
                    f"Refusing downgrade {state.version} -> {resolved_version} "
                    f"(use --force-downgrade)", fg='red'))
                raise SystemExit(1)
        except Exception:
            pass

    artifact_path = Path(shelf_dir) / artifact_name
    plugins_dir = Path(pm.loader.plugins_dir)
    click.echo(f"Verifying + unpacking {name} {resolved_version} ...")
    try:
        unpack_verified(artifact_path, keys, plugins_dir, name,
                        expected_version=resolved_version,
                        expected_sha256=expected_sha)
    except ValueError as e:
        click.echo(click.style(f"  verification failed: {e}", fg='red'))
        raise SystemExit(1)

    # Clear any cached manifest so the freshly unpacked one is read.
    pm.loader._manifests.pop(name, None)
    try:
        result = pm.apply_profile([name])
    except PluginError as e:
        click.echo(click.style(f"  {e}", fg='red'))
        click.echo(click.style(
            "  (adopt the missing dependency plugins first)", fg='yellow'))
        raise SystemExit(1)

    write_last_serial(current_app.instance_path, index['serial'])
    click.echo("")
    click.echo(click.style(
        f"Adopted {name} {resolved_version}.", fg='green'))
    if result['installed']:
        click.echo(f"  installed: {', '.join(result['installed'])}")
    if result['enabled']:
        click.echo(f"  enabled:   {', '.join(result['enabled'])}")
    click.echo(click.style(
        "Run 'flask plugin upgrade-all' then restart so routes register.",
        fg='yellow'))


@plugin_cli.command('audit')
@click.option('--pubkey', 'pubkeys', multiple=True, type=click.Path(exists=True))
@with_appcontext
def audit_plugins(pubkeys):
    """Warn about installed plugins whose version is revoked on the shelf."""
    from .shelf import load_index, is_revoked

    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        raise SystemExit(1)

    shelf_dir = current_app.config.get('PLUGIN_SHELF_DIR')
    if not shelf_dir:
        click.echo(click.style("PLUGIN_SHELF_DIR not configured", fg='red'))
        raise SystemExit(1)

    keys, _ = _shelf_keys(pubkeys)
    index, errors = load_index(shelf_dir, keys)
    for err in errors:
        click.echo(click.style(f"  {err}", fg='red'))
    if index is None:
        raise SystemExit(1)

    flagged = []
    for name, state in pm.registry.get_all().items():
        if is_revoked(index, name, state.version):
            flagged.append(f"{name} {state.version}")

    click.echo("")
    if flagged:
        click.echo(click.style(
            "REVOKED plugins installed:", fg='red', bold=True))
        for item in flagged:
            click.echo(click.style(f"  {item}", fg='red'))
        raise SystemExit(1)
    click.echo(click.style("No installed plugin is revoked.", fg='green'))


@plugin_cli.command('migrate')
@click.argument('name')
@click.option('--revision', default='head', help='Target revision')
@with_appcontext
def migrate_plugin(name: str, revision: str):
    """Run migrations for a specific plugin."""
    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        raise SystemExit(1)

    if not pm.registry.is_installed(name):
        click.echo(click.style(f"Plugin {name} is not installed", fg='red'))
        raise SystemExit(1)

    click.echo(f"Running migrations for {name}...")

    if pm.migration_manager.run_plugin_migrations(name, revision):
        click.echo(click.style("Migrations completed", fg='green'))
    else:
        click.echo(click.style("Migration failed", fg='red'))
        raise SystemExit(1)


@plugin_cli.command('upgrade-all')
@with_appcontext
def upgrade_all_plugins():
    """Run pending migrations for every discovered plugin.

    Idempotent. Run this after `flask db upgrade` on every deploy and
    upgrade. It stamps each bundled plugin's anchor revision into
    alembic_version_<plugin> and applies any per-plugin migrations added
    after the ownership cutover (ADR-008). Safe to re-run at head.
    """
    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        raise SystemExit(1)
    results = pm.upgrade_all_plugins()
    if not results:
        click.echo("No plugins discovered.")
        return
    for name, status in sorted(results.items()):
        if status == 'ok':
            click.echo(click.style(f"  {name:20} ok", fg='green'))
        elif status == 'no-migrations':
            click.echo(click.style(f"  {name:20} no migrations", fg='yellow'))
        else:
            click.echo(click.style(f"  {name:20} {status}", fg='red'))


@plugin_cli.command('prune-schema')
@click.option('--yes', is_flag=True,
              help='Actually drop the tables (default is a dry-run preview)')
@click.option('--force', is_flag=True,
              help='Drop even tables that hold rows (DATA LOSS); default refuses')
@with_appcontext
def prune_schema(yes: bool, force: bool):
    """Drop tables owned by plugins this site did NOT install (ADR-014).

    Schema-lean per-site DBs: the shared core Alembic baseline creates every
    plugin's tables, so a lean site that omits a plugin still carries that
    plugin's (empty) tables. This drops the tables of every plugin in
    PLUGIN_TABLE_OWNERS that is not installed here, leaving core + chosen-plugin
    tables only. Run once at deploy AFTER `flask db upgrade` and
    `flask plugin upgrade-all`.

    Dry-run by default; pass --yes to execute. Drops by table name (no plugin
    code import) so it works on a lean image where the omitted plugin's
    directory is absent. Refuses to drop a non-empty table unless --force, so a
    misfire on a populated full site cannot silently delete data.
    """
    from sqlalchemy import create_engine, inspect, text

    from shopdb.extensions import db
    from shopdb.plugins.alembic_template import PLUGIN_TABLE_OWNERS

    pm = current_app.extensions.get('plugin_manager')
    if not pm:
        click.echo(click.style("Plugin manager not initialized", fg='red'))
        raise SystemExit(1)

    installed = {p['name'] for p in pm.discover_available() if p['installed']}
    engine = db.engine
    existing = set(inspect(engine).get_table_names())

    # A table is a prune candidate when its owning plugin is not installed here
    # AND the table actually exists in this database.
    victims = []
    for plugin, tables in PLUGIN_TABLE_OWNERS.items():
        if plugin in installed:
            continue
        for tablename in tables:
            if tablename in existing:
                victims.append((plugin, tablename))

    if not victims:
        click.echo("Nothing to prune: every plugin-owned table belongs to an "
                   "installed plugin.")
        return

    # On MySQL, use a SEPARATE engine in AUTOCOMMIT, not db.engine. Two reasons,
    # both of which burned real debugging time:
    #   1. db.engine's pool keeps connections idle-in-transaction (Flask-
    #      SQLAlchemy has no request-teardown in a CLI context); a DROP sharing
    #      that pool waits on their locks.
    #   2. Without engine-level AUTOCOMMIT, SQLAlchemy runs the COUNT probes in an
    #      open transaction (SET AUTOCOMMIT=0). Those reads hold shared metadata
    #      locks on every victim table, so the later DROP blocks on the metadata
    #      lock forever. Setting isolation_level on the CONNECTION (not the
    #      engine) silently did NOT take effect - it must be on the engine.
    # With engine-level AUTOCOMMIT every statement commits on its own, so no read
    # holds a lock into the DROP phase. lock_wait_timeout makes any residual
    # contention fail fast. SQLite has none of this (no metadata locks), and a
    # separate engine to an in-memory database would be a different, empty DB, so
    # there we just use db.engine. db.session is only used above, for pluginstate.
    dialect = engine.dialect.name
    db.session.remove()
    if dialect == 'mysql':
        prune_engine = create_engine(
            current_app.config['SQLALCHEMY_DATABASE_URI'],
            isolation_level='AUTOCOMMIT')
        own_engine = True
    else:
        prune_engine = engine
        own_engine = False

    try:
        with prune_engine.connect() as conn:
            if dialect == 'mysql':
                conn.execute(text('SET SESSION lock_wait_timeout=15'))

            # Row counts so a non-empty table is never dropped by accident.
            nonempty = []
            for plugin, tablename in victims:
                count = conn.execute(
                    text(f'SELECT COUNT(*) FROM `{tablename}`')).scalar()
                if count:
                    nonempty.append((plugin, tablename, count))

            click.echo(click.style(
                f"Plugins not installed here: "
                f"{', '.join(sorted({p for p, _ in victims}))}", fg='cyan'))
            click.echo(f"Tables to prune ({len(victims)}):")
            for plugin, tablename in victims:
                click.echo(f"  {tablename:32} ({plugin})")

            if nonempty and not force:
                click.echo("")
                click.echo(click.style(
                    "REFUSING: the following tables hold rows. Re-run with "
                    "--force to drop them anyway (this deletes data), or install "
                    "the owning plugin instead.", fg='red'))
                for plugin, tablename, count in nonempty:
                    click.echo(click.style(f"  {tablename:32} {count} rows",
                                           fg='red'))
                raise SystemExit(1)

            if not yes:
                click.echo("")
                click.echo("Dry-run. Re-run with --yes to drop the tables above.")
                return

            # Intra-plugin foreign keys mean drop order matters; disable the
            # checks for the batch rather than topologically sorting the tables
            # without the models. Restore the PRIOR foreign-key setting after,
            # not a hard ON: on SQLite the same connection is reused (StaticPool),
            # so forcing ON would leak into whatever ran next.
            fk_prev = None
            if dialect == 'mysql':
                conn.execute(text('SET FOREIGN_KEY_CHECKS=0'))
            elif dialect == 'sqlite':
                fk_prev = conn.execute(text('PRAGMA foreign_keys')).scalar()
                conn.execute(text('PRAGMA foreign_keys=OFF'))
            for _plugin, tablename in victims:
                conn.execute(text(f'DROP TABLE IF EXISTS `{tablename}`'))
            if dialect == 'mysql':
                conn.execute(text('SET FOREIGN_KEY_CHECKS=1'))
            elif dialect == 'sqlite':
                conn.execute(text(f'PRAGMA foreign_keys={int(fk_prev or 0)}'))
            conn.commit()
    finally:
        if own_engine:
            prune_engine.dispose()

    click.echo(click.style(f"Pruned {len(victims)} table(s). This database now "
                           f"carries core + installed-plugin tables only.",
                           fg='green'))