MoviePilot/app/runtime/extensions/plugin/storage.py

"""
插件运行时持久化端口。
"""

from __future__ import annotations

import hashlib
import json
import threading
from collections.abc import Awaitable, Callable
from datetime import datetime
from typing import Any, NamedTuple, Optional

from pydantic import ValidationError

from app.runtime.extensions.plugin.database import PluginDatabase
from app.schemas.plugin import PluginInstance
from app.schemas.types import SystemConfigKey

ConfigReader = Callable[[Any], Any]
ConfigWriter = Callable[[Any, Any], Any]
AsyncConfigWriter = Callable[[Any, Any], Awaitable[Any]]
ConfigDeleter = Callable[[Any], bool]
# 插件配置端口按实例 ID 寻址，不经过任何字符串键：一个实例的业务参数存在它自己
# 那一行上，宿主的系统设置存储不必知道插件存在
InstanceConfigReader = Callable[[str], Any]
InstanceConfigWriter = Callable[[str, Any], Any]
AsyncInstanceConfigWriter = Callable[[str, Any], Awaitable[Any]]
InstanceConfigDeleter = Callable[[str], bool]
PluginDataDeleter = Callable[[str], Any]
# 日志等级覆盖端口的载荷：`(等级名, 失效时间)`，两者皆为 None 即未设置覆盖
LogLevelOverride = tuple[Optional[str], Optional[datetime]]
LogLevelReader = Callable[[str], LogLevelOverride]
LogLevelWriter = Callable[[str, Optional[str], Optional[datetime]], None]
PluginExists = Callable[[str], bool]


def _empty_read(_key: Any) -> Any:
    """
    组合根尚未装配时返回空配置。
    """
    return None


def _ignore_write(_key: Any, _value: Any) -> None:
    """
    组合根尚未装配时忽略同步配置写入。
    """


async def _ignore_async_write(_key: Any, _value: Any) -> None:
    """
    组合根尚未装配时忽略异步配置写入。
    """


def _ignore_delete(_key: Any) -> bool:
    """
    组合根尚未装配时报告配置未删除。
    """
    return False


def _empty_read_config(_instance_id: str) -> Any:
    """
    组合根尚未装配时报告实例没有业务参数。
    """
    return None


def _ignore_write_config(_instance_id: str, _config: Any) -> None:
    """
    组合根尚未装配时忽略同步业务参数写入。
    """


async def _ignore_async_write_config(_instance_id: str, _config: Any) -> None:
    """
    组合根尚未装配时忽略异步业务参数写入。
    """


def _ignore_delete_config(_instance_id: str) -> bool:
    """
    组合根尚未装配时报告业务参数未删除。
    """
    return False


def _ignore_plugin_data_delete(_plugin_id: str) -> None:
    """
    组合根尚未装配时忽略插件数据删除。
    """


def _no_log_level(_instance_id: str) -> LogLevelOverride:
    """
    组合根尚未装配时报告实例没有日志等级覆盖。
    """
    return (None, None)


def _ignore_log_level_write(
    _instance_id: str,
    _level: Optional[str],
    _expires_at: Optional[datetime],
) -> None:
    """
    组合根尚未装配时忽略日志等级落盘。
    """


class PluginStorage:
    """
    封装插件运行时所需的最小持久化能力。
    """

    def __init__(
            self,
            *,
            read: ConfigReader = _empty_read,
            write: ConfigWriter = _ignore_write,
            async_write: AsyncConfigWriter = _ignore_async_write,
            delete: ConfigDeleter = _ignore_delete,
            read_config: InstanceConfigReader = _empty_read_config,
            write_config: InstanceConfigWriter = _ignore_write_config,
            async_write_config: AsyncInstanceConfigWriter = _ignore_async_write_config,
            delete_config: InstanceConfigDeleter = _ignore_delete_config,
            delete_data: PluginDataDeleter = _ignore_plugin_data_delete,
            read_log_level: LogLevelReader = _no_log_level,
            write_log_level: LogLevelWriter = _ignore_log_level_write,
    ) -> None:
        """
        保存由启动组合根提供的读写函数。
        """
        self._read = read
        self._write = write
        self._async_write = async_write
        self._delete = delete
        self._read_config = read_config
        self._write_config = write_config
        self._async_write_config = async_write_config
        self._delete_config = delete_config
        self._delete_data = delete_data
        self._read_log_level = read_log_level
        self._write_log_level = write_log_level

    def read(self, key: Any) -> Any:
        """
        读取插件运行时配置。
        """
        return self._read(key)

    def write(self, key: Any, value: Any) -> Any:
        """
        同步保存插件运行时配置。
        """
        return self._write(key, value)

    async def async_write(self, key: Any, value: Any) -> Any:
        """
        异步保存插件运行时配置。
        """
        return await self._async_write(key, value)

    def delete(self, key: Any) -> bool:
        """
        删除插件运行时配置。
        """
        return self._delete(key)

    def read_config(self, instance_id: str) -> Any:
        """
        读取该实例配置行上的业务参数。
        """
        return self._read_config(instance_id)

    def write_config(self, instance_id: str, config: Any) -> Any:
        """
        同步把业务参数写进该实例的配置行。
        """
        return self._write_config(instance_id, config)

    async def async_write_config(self, instance_id: str, config: Any) -> Any:
        """
        异步把业务参数写进该实例的配置行。
        """
        return await self._async_write_config(instance_id, config)

    def delete_config(self, instance_id: str) -> bool:
        """
        清除该实例配置行上的业务参数。
        """
        return self._delete_config(instance_id)

    def delete_data(self, plugin_id: str) -> Any:
        """
        删除指定插件的业务数据。
        """
        return self._delete_data(plugin_id)

    def read_log_level(self, instance_id: str) -> LogLevelOverride:
        """
        读取实例配置里登记的日志等级覆盖与失效时间。
        """
        return self._read_log_level(instance_id)

    def write_log_level(
        self,
        instance_id: str,
        level: Optional[str],
        expires_at: Optional[datetime],
    ) -> None:
        """
        把日志等级覆盖写进该实例的配置行。
        """
        self._write_log_level(instance_id, level, expires_at)


class PluginConfigStore:
    """
    封装插件配置的存在性和强制删除规则。

    配置按实例 ID 直接落在该实例自己的配置行上，不再经过宿主系统设置存储的字符串键：
    插件配置与系统设置的生命周期、数量级和归属都不同，共用一张键值表会让系统设置被
    插件条目淹没，也会让「某插件的全部实例配置」只能靠键前缀去猜。
    """

    def __init__(
        self,
        *,
        storage: Callable[[], "PluginStorage"],
        database: Callable[[], PluginDatabase],
        plugin_exists: PluginExists,
    ) -> None:
        """
        保存持久化端口、自有数据库端口和运行态插件查询端口。
        """
        self._storage = storage
        self._database = database
        self._plugin_exists = plugin_exists

    def read(self, plugin_id: str) -> dict[str, object]:
        """
        读取配置并过滤历史空键。
        """
        if not self._plugin_exists(plugin_id):
            return {}
        config = self._storage().read_config(plugin_id)
        return {
            key: value
            for key, value in (config or {}).items()
            if key
        }

    def has_config(self, instance_id: str) -> bool:
        """
        判断该实例行上是否留有业务参数。

        不复用 :meth:`read`：那里要求插件类当前已登记，而这里问的恰恰是已停用、因而
        不在类注册表里的实例——走 :meth:`read` 会让每一个可恢复分身都报告「没有配置」。
        :param instance_id: 实例 ID
        :return: 该实例是否留有业务参数
        """
        return bool(self._storage().read_config(instance_id))

    def write(self, plugin_id: str, config: dict[str, object], force: bool = False) -> bool:
        """
        保存配置，默认拒绝不存在插件的配置写入。
        """
        if not force and not self._plugin_exists(plugin_id):
            return False
        self._storage().write_config(plugin_id, config)
        return True

    async def async_write(
        self,
        plugin_id: str,
        config: dict[str, object],
        force: bool = False,
    ) -> bool:
        """
        异步保存配置并保持同步写入的存在性规则。
        """
        if not force and not self._plugin_exists(plugin_id):
            return False
        await self._storage().async_write_config(plugin_id, config)
        return True

    def delete(self, plugin_id: str, force: bool = False) -> bool:
        """
        删除配置并保持停止插件后的强制删除能力。
        """
        if not force and not self._plugin_exists(plugin_id):
            return False
        return self._storage().delete_config(plugin_id)

    def read_log_level(self, instance_id: str) -> LogLevelOverride:
        """
        读取实例的日志等级覆盖；它与业务参数同属该实例的配置。
        """
        return self._storage().read_log_level(instance_id)

    def write_log_level(
        self,
        instance_id: str,
        level: Optional[str],
        expires_at: Optional[datetime],
    ) -> None:
        """
        写入实例的日志等级覆盖。
        """
        self._storage().write_log_level(instance_id, level, expires_at)

    def delete_data(self, plugin_id: str, force: bool = False) -> bool:
        """
        删除插件业务数据与自有数据库，并保持旧的布尔结果合同。
        """
        if not force and not self._plugin_exists(plugin_id):
            return False
        self._storage().delete_data(plugin_id)
        self._database().destroy(plugin_id)
        return True

    def delete_data_rows(self, plugin_id: str) -> None:
        """
        只删该实例在插件数据表里的行，不碰它的自有数据库。

        彻底清理要让用户逐项勾选删除范围，因而这两件事必须能分开做；
        :meth:`delete_data` 仍把它们捆在一起，服务「重置」那条既有语义。
        """
        self._storage().delete_data(plugin_id)

    def destroy_database(self, plugin_id: str) -> None:
        """
        销毁该实例的自有数据库并释放句柄。
        """
        self._database().destroy(plugin_id)


InstanceReader = Callable[[str], "PluginInstance | None"]
InstanceLister = Callable[[], "list[PluginInstance]"]
InstanceSourceLister = Callable[[str], "list[PluginInstance]"]
InstanceWriter = Callable[["PluginInstance"], None]
InstanceDeleter = Callable[[str], bool]
InstanceEnabler = Callable[[str, bool], bool]


def _empty_instance_get(_instance_id: str) -> PluginInstance | None:
    """
    组合根尚未装配时返回空实例描述。
    """
    return None


def _empty_instance_list() -> list[PluginInstance]:
    """
    组合根尚未装配时返回空实例列表。
    """
    return []


def _empty_instance_list_by_source(_source_plugin_id: str) -> list[PluginInstance]:
    """
    组合根尚未装配时返回空实例列表。
    """
    return []


def _ignore_instance_save(_instance: PluginInstance) -> None:
    """
    组合根尚未装配时忽略实例描述写入。
    """


def _ignore_instance_delete(_instance_id: str) -> bool:
    """
    组合根尚未装配时报告实例描述未删除。
    """
    return False


def _ignore_instance_enable(_instance_id: str, _is_enabled: bool) -> bool:
    """
    组合根尚未装配时报告启用位未写入。
    """
    return False


class PluginInstanceDirectory:
    """
    封装插件实例表的持久化能力。

    分身与源插件本体共用同一张表、同一套读写原语，两者靠 ``instance_id`` 是否等于
    ``source_plugin_id`` 区分；本类不做角色过滤，角色隔离由调用方
    （``PluginInstanceStore``）负责，因为只有调用方知道当前服务的是分身清单还是
    本体自身的那一行。

    ``list_all``/``get`` 一律返回全部登记行（含停用的），``list_enabled`` 才是运行期
    装载的取数口：是否实例化由 ``is_enabled`` 单独表达，读取口不替调用方把停用的行
    藏起来——藏起来会让卡片、卸载守卫和默认目标候选一并丢掉这些行。

    历史脏行如果无法投影为运行时 ``PluginInstance``，由启动组合根的读取适配器记录并
    跳过；本端口仍只向上层暴露可安全使用的实例描述，避免单个旧插件记录阻断整个启动。
    """

    def __init__(
            self,
            *,
            get: InstanceReader = _empty_instance_get,
            list_all: InstanceLister = _empty_instance_list,
            list_by_source: InstanceSourceLister = _empty_instance_list_by_source,
            save: InstanceWriter = _ignore_instance_save,
            delete: InstanceDeleter = _ignore_instance_delete,
            list_enabled: InstanceLister = _empty_instance_list,
            set_enabled: InstanceEnabler = _ignore_instance_enable,
            purge_invalid_empty_host: InstanceDeleter = _ignore_instance_delete,
    ) -> None:
        """
        保存由启动组合根提供的实例表读写函数。
        """
        self._get = get
        self._list_all = list_all
        self._list_by_source = list_by_source
        self._save = save
        self._delete = delete
        self._list_enabled = list_enabled
        self._set_enabled = set_enabled
        self._purge_invalid_empty_host = purge_invalid_empty_host

    def get(self, instance_id: str) -> PluginInstance | None:
        """
        按实例 ID 读取单条描述，不区分分身与本体。
        """
        return self._get(instance_id)

    def list_all(self) -> list[PluginInstance]:
        """
        列出表中全部描述，不区分分身与本体。
        """
        return self._list_all()

    def list_by_source(self, source_plugin_id: str) -> list[PluginInstance]:
        """
        按源插件 ID 列出其全部描述，不区分分身与本体。
        """
        return self._list_by_source(source_plugin_id)

    def save(self, instance: PluginInstance) -> None:
        """
        新增或更新一条描述，以 ``instance_id`` 为稳定键。
        """
        self._save(instance)

    def delete(self, instance_id: str) -> bool:
        """
        按实例 ID 删除一行，连同其配置，返回删除前是否存在。
        """
        return self._delete(instance_id)

    def list_enabled(self) -> list[PluginInstance]:
        """
        列出应当被实例化并启动的行，不含停用的，不区分分身与本体。
        """
        return self._list_enabled()

    def set_enabled(self, instance_id: str, is_enabled: bool) -> bool:
        """
        写入启用位，返回该行是否存在。
        """
        return self._set_enabled(instance_id, is_enabled)

    def purge_invalid_empty_host(self, instance_id: str) -> bool:
        """
        仅删除无法投影且不承载任何设置的历史本体行。
        """
        return self._purge_invalid_empty_host(instance_id)


class _LegacyInstanceEntry(NamedTuple):
    """
    旧 systemconfig 单键里的一条实例描述，连同其原始载荷的内容指纹。

    :param instance: 从历史配置解析的插件实例信息
    :param fingerprint: 原始载荷内容指纹，用于检测迁移后源数据是否变化
    """

    instance: PluginInstance
    fingerprint: str


def _payload_fingerprint(payload: dict[str, Any]) -> str:
    """
    为旧键里的一条原始载荷算内容指纹，用于判定旧版本此后有没有改动过它。

    指纹取自原始载荷而非模型 dump：模型字段将来增减会让全部指纹一起失配，把用户在
    独立表里的改动当成「旧载荷变化」整批盖回去。键排序后再序列化，旧版本重写整个键
    造成的字段顺序变化不会被误判成内容变化。

    :param payload: 旧键里的单条原始载荷
    :return: 十六进制摘要
    """
    canonical = json.dumps(payload, sort_keys=True, default=str, ensure_ascii=False)
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


class PluginInstanceStore:
    """
    管理共享源码的分身实例描述，并把源插件本体自身的那一行单独成一组读写口。

    两类记录同存一张表，靠 ``instance_id`` 是否等于 ``source_plugin_id`` 区分：
    ``all()``/``get()``/``save()``/``delete()``/``for_source()``/``enabled()`` 只服务分身，
    ``all_hosts()``/``get_host()``/``save_host()``/``enabled_hosts()`` 只服务本体，任何
    一侧都读不到、也改不到对方的记录——只有调用方知道自己要的是分身清单还是本体那一行。

    除 ``enabled()``/``enabled_hosts()`` 外的读取口一律返回全部登记行（含停用的）：
    停用的实例仍是一份在册配置，卡片要看得见、卸载守卫要拦得住、默认调用目标的候选
    清单也要列得出。是否应当被实例化单由 ``is_enabled`` 表达，运行期取数走前两者。
    """

    def __init__(
            self,
            *,
            storage: Callable[[], "PluginStorage"],
            directory: Callable[[], PluginInstanceDirectory],
    ) -> None:
        """
        保存独立表持久化端口，以及旧 systemconfig 单键端口供兜底导入使用。
        """
        self._storage = storage
        self._directory = directory
        self._bootstrap_checked = False
        # 引导只应发生一次，但完成标记要等合并真正成功才提交，两者之间存在窗口；
        # 用可重入锁把窗口关上，多线程同时首次访问时只有一个真的去合并
        self._bootstrap_lock = threading.RLock()

    def purge_invalid_empty_host(self, instance_id: str) -> bool:
        """
        让持久化端口直接核对历史空行，不经严格描述符投影或旧键导入。
        """
        return self._directory().purge_invalid_empty_host(instance_id)

    def _ensure_bootstrapped(self) -> None:
        """
        把旧 systemconfig 单键的实例描述变更合并进独立表，每进程成功一次。

        完成标记在合并与指纹落盘全部成功之后才提交：读取、列举、保存、写指纹任何
        一步遇到暂时性故障都会把异常抛给调用方，此时若标记已经落下，本进程后续调用
        就会跳过引导、静默返回一份没导完的分身清单，只能靠重启自愈。失败后重试仍然
        是一次性的——标记一旦提交，后续调用连旧键都不再读。

        :raise Exception: 导入失败时向上抛出，不吞掉持久化层错误
        """
        if self._bootstrap_checked:
            return
        with self._bootstrap_lock:
            # 持锁后复查：等锁期间前一个线程可能已经把引导做完了，不能重复导入
            if self._bootstrap_checked:
                return
            self._merge_legacy_instances()
            self._bootstrap_checked = True

    def _merge_legacy_instances(self) -> None:
        """
        执行一轮旧键与独立表的合并：新增、改写与删除三类变更都要同步过去。

        alembic 迁移已经搬过一轮，这里兜的是「库结构升级后又用旧版本写过分身」这类
        回滚往返：旧键刻意保留作回滚依据、从不清理，旧版本只认它，因此它在迁移之后
        仍可能长出新条目、被改写或被删条目。判据既不能是「表当前为空」——用户删光分身
        后表会重新变空，下次启动会把删掉的分身整批导回来；也不能是一条一次性的永久
        标记——标记落下之后旧版本新增的分身就永远看不见。

        因此按条目记住导入时的内容指纹：指纹对得上就跳过，表里那行用户怎么改都不会被
        旧副本盖回去；指纹对不上说明旧版本改写过该条目，合并进来；指纹没记过且表里也
        没有该行说明旧版本新增过它，导入。指纹没记过但表里已有该行（迁移刚搬过去的那
        一批），只认领指纹而不覆盖。指纹表始终与旧键当前内容对齐，重复启动不再产生写入。

        删除同样以指纹表为准：只清理「曾经认领过指纹、现在旧键里没有了」的行，新表里
        原生创建、从未在旧键出现过的分身因此不会被牵连。旧版本里的重命名表现为「旧 ID
        消失 + 新 ID 出现」，按同一套规则自然得到删旧建新。
        """
        storage = self._storage()
        recorded = storage.read(SystemConfigKey.PluginInstancesImported)
        imported: dict[str, str] = dict(recorded) if isinstance(recorded, dict) else {}
        legacy = self._legacy_instances()
        pending = {
            instance_id: entry
            for instance_id, entry in legacy.items()
            if imported.get(instance_id) != entry.fingerprint
        }
        dropped = [
            instance_id for instance_id in imported if instance_id not in legacy
        ]
        if pending or dropped:
            directory = self._directory()
            if pending:
                # 只在确有待定条目时扫一次现有行，让无变化的启动不额外查表
                present = {record.instance_id for record in directory.list_all()}
                for instance_id, entry in pending.items():
                    if instance_id in imported or instance_id not in present:
                        directory.save(entry.instance)
            for instance_id in dropped:
                directory.delete(instance_id)
        fingerprints = {
            instance_id: entry.fingerprint for instance_id, entry in legacy.items()
        }
        if fingerprints != imported:
            storage.write(SystemConfigKey.PluginInstancesImported, fingerprints)

    def _legacy_instances(self) -> dict[str, _LegacyInstanceEntry]:
        """
        解析旧 systemconfig 单键里的实例描述，兼容历史字典与列表两种载荷形态。
        """
        raw_instances = self._storage().read(SystemConfigKey.PluginInstances) or {}
        if isinstance(raw_instances, list):
            entries = {
                item.get("instance_id"): item
                for item in raw_instances
                if isinstance(item, dict) and item.get("instance_id")
            }
        elif isinstance(raw_instances, dict):
            entries = raw_instances
        else:
            return {}

        instances: dict[str, _LegacyInstanceEntry] = {}
        for instance_id, raw_instance in entries.items():
            try:
                payload = dict(raw_instance) if isinstance(raw_instance, dict) else {}
                payload.setdefault("instance_id", instance_id)
                instance = PluginInstance.model_validate(payload)
                instances[instance.instance_id] = _LegacyInstanceEntry(
                    instance=instance,
                    fingerprint=_payload_fingerprint(payload),
                )
            except (TypeError, ValidationError):
                continue
        return instances

    def all(self) -> dict[str, PluginInstance]:
        """
        读取全部登记的分身实例，不含源插件本体自身的那一行。
        """
        self._ensure_bootstrapped()
        return {
            record.instance_id: record
            for record in self._directory().list_all()
            if not record.is_host
        }

    def get(self, instance_id: str) -> PluginInstance | None:
        """
        读取指定分身实例描述；本体自身的那一行不会从这里返回。
        """
        self._ensure_bootstrapped()
        record = self._directory().get(instance_id)
        return record if record is not None and not record.is_host else None

    def save(self, instance: PluginInstance) -> None:
        """
        新增或更新分身实例描述，并以实例 ID 作为稳定持久化键。

        分身的实例 ID 必须区别于其源插件 ID：两者相等的那一行表示的是本体自身，
        按分身写入会把本体承载的业务参数顶掉。
        """
        self._ensure_bootstrapped()
        if instance.is_host:
            raise ValueError(
                f"分身实例 {instance.instance_id} 的 ID 不能等于其源插件 ID"
            )
        self._directory().save(instance)

    def delete(self, instance_id: str) -> bool:
        """
        删除指定分身实例描述连同其配置，返回删除前是否存在。
        """
        self._ensure_bootstrapped()
        record = self.get(instance_id)
        if record is None:
            return False
        return self._directory().delete(record.instance_id)

    def enabled(self) -> dict[str, PluginInstance]:
        """
        读取应当被实例化并启动的分身实例，不含停用的与本体记录。
        """
        self._ensure_bootstrapped()
        return {
            record.instance_id: record
            for record in self._directory().list_enabled()
            if not record.is_host
        }

    def enable(self, instance_id: str) -> bool:
        """
        启用指定分身实例，返回启用前它是否存在且处于停用状态。
        """
        self._ensure_bootstrapped()
        record = self.get(instance_id)
        if record is None or record.is_enabled:
            return False
        return self._directory().set_enabled(record.instance_id, True)

    def disable(self, instance_id: str) -> bool:
        """
        停用指定分身实例，返回停用前它是否存在且处于启用状态。

        停用不删行：业务参数、展示信息原样留在那一行，再次启用即恢复。要连配置一并
        抹掉走 :meth:`delete`。
        """
        self._ensure_bootstrapped()
        record = self.get(instance_id)
        if record is None or not record.is_enabled:
            return False
        return self._directory().set_enabled(record.instance_id, False)

    def all_hosts(self) -> dict[str, PluginInstance]:
        """
        一次性读取全部源插件本体记录，不含分身实例。

        供目录投影批量取数使用：按插件 ID 遍历卡片时只做内存字典查找，不再逐张卡片
        各查一次数据库。
        """
        self._ensure_bootstrapped()
        return {
            record.instance_id: record
            for record in self._directory().list_all()
            if record.is_host
        }

    def enabled_hosts(self) -> dict[str, PluginInstance]:
        """
        读取应当被装载的源插件本体记录，不含停用的与分身实例。

        本体与分身在这里终于用同一个判据：``is_enabled`` 决定一份配置是否应当被
        实例化并启动。安装清单退回去只回答「这个插件的包在不在磁盘上」，不再兼任
        运行开关——两者本是两件事，一处安装记录同时充当开关会让「装着但先不跑」
        无从表达。
        """
        self._ensure_bootstrapped()
        return {
            record.instance_id: record
            for record in self._directory().list_enabled()
            if record.is_host
        }

    def enable_host(self, plugin_id: str) -> bool:
        """
        把源插件本体登记为应当装载，没有本体记录时按默认视图建出。

        安装收尾必须调用：本体的装载判据已经归口到启用位，只往安装清单里加一条而
        不建出这一行，插件会装完却不加载。

        :param plugin_id: 插件 ID
        :return: 本次是否改变了状态；本体已处于启用状态时为 False
        """
        self._ensure_bootstrapped()
        host = self.get_host(plugin_id)
        if host is None:
            self.save_host(
                PluginInstance(
                    instance_id=plugin_id,
                    source_plugin_id=plugin_id,
                    is_enabled=True,
                )
            )
            return True
        if host.is_enabled:
            return False
        return self._directory().set_enabled(host.instance_id, True)

    def disable_host(self, plugin_id: str) -> bool:
        """
        停用源插件本体但保留其全部设置，返回停用前它是否存在且处于启用状态。

        业务参数保留——那是用户的数据，重装或重新启用后应当还在；默认目标置位与
        日志等级覆盖由启用位置假时一并清掉，它们只对在册实例有意义。
        """
        self._ensure_bootstrapped()
        host = self.get_host(plugin_id)
        if host is None or not host.is_enabled:
            return False
        return self._directory().set_enabled(host.instance_id, False)

    def get_host(self, plugin_id: str) -> PluginInstance | None:
        """
        读取源插件本体自身那一行；该插件从未登记过任何设置时为 None。
        """
        self._ensure_bootstrapped()
        record = self._directory().get(plugin_id)
        return record if record is not None and record.is_host else None

    def save_host(self, instance: PluginInstance) -> None:
        """
        新增或更新源插件本体自身那一行，本体的 ``source_plugin_id`` 恒等于自身 ID。

        归一而不是拒绝：本体行的一对身份列必然相等，调用方传进来的 ``source_plugin_id``
        对本体而言没有可选值，就地纠正比让调用方各自记住这条约束更不容易出错。
        """
        self._ensure_bootstrapped()
        self._directory().save(
            instance.model_copy(update={"source_plugin_id": instance.instance_id})
        )

    def for_source(self, source_plugin_id: str) -> list[PluginInstance]:
        """
        按持久化顺序返回引用同一源码插件的全部分身，不含本体自身那一行。
        """
        self._ensure_bootstrapped()
        return [
            record
            for record in self._directory().list_by_source(source_plugin_id)
            if not record.is_host
        ]


_plugin_storage = PluginStorage()
_plugin_instance_directory = PluginInstanceDirectory()


def configure_plugin_instance_directory(directory: PluginInstanceDirectory) -> None:
    """
    由启动组合根替换插件实例表持久化实现。
    """
    global _plugin_instance_directory
    _plugin_instance_directory = directory


def get_plugin_instance_directory() -> PluginInstanceDirectory:
    """
    返回当前插件实例表持久化端口。
    """
    return _plugin_instance_directory


def configure_plugin_storage(storage: PluginStorage) -> None:
    """
    由启动组合根替换插件运行时持久化实现。
    """
    global _plugin_storage
    _plugin_storage = storage


def get_plugin_storage() -> PluginStorage:
    """
    返回当前插件运行时持久化端口。
    """
    return _plugin_storage