awg-dash/backend/awg_api/store/repository.py

"""Репозиторий сущностей состояния (issue #20).

Каждая мутация выполняется в одной транзакции, запись в ``audit_log``
(«кто/когда/что/old→new») коммитится там же. Чтения идут через
:meth:`Database.read` — под тем же ``RLock``, что и транзакции (#54), поэтому
читатель не видит незафиксированных строк и состояния, которое затем будет
откачено, и не может получить перемешанный результат операторов другого потока.
"""

from __future__ import annotations

import sqlite3
from dataclasses import replace
from typing import cast

from . import audit, validation
from .clock import utcnow
from .db import Database
from .errors import StorageError
from .models import (
    AuditEntry,
    ConfigMeta,
    Device,
    Entity,
    GeoIpSchedule,
    Mode,
    Rule,
    RuleException,
)


def _rule_from_row(row: sqlite3.Row) -> Rule:
    return Rule(
        id=int(row["id"]),
        name=str(row["name"]),
        kind=str(row["kind"]),
        match=str(row["match"]),
        target=str(row["target"]),
        priority=int(row["priority"]),
        enabled=bool(row["enabled"]),
        created_at=str(row["created_at"]),
        updated_at=str(row["updated_at"]),
    )


def _exception_from_row(row: sqlite3.Row) -> RuleException:
    return RuleException(
        id=int(row["id"]),
        match=str(row["match"]),
        reason=str(row["reason"]),
        enabled=bool(row["enabled"]),
        created_at=str(row["created_at"]),
        updated_at=str(row["updated_at"]),
    )


def _config_from_row(row: sqlite3.Row) -> ConfigMeta:
    return ConfigMeta(
        id=int(row["id"]),
        name=str(row["name"]),
        path=str(row["path"]),
        version=int(row["version"]),
        checksum=str(row["checksum"]),
        size_bytes=int(row["size_bytes"]),
        applied=bool(row["applied"]),
        created_at=str(row["created_at"]),
        updated_at=str(row["updated_at"]),
    )


def _device_from_row(row: sqlite3.Row) -> Device:
    return Device(
        id=int(row["id"]),
        mac=str(row["mac"]),
        name=str(row["name"]),
        address=str(row["address"]),
        policy=str(row["policy"]),
        last_seen=None if row["last_seen"] is None else str(row["last_seen"]),
        created_at=str(row["created_at"]),
        updated_at=str(row["updated_at"]),
    )


def _geoip_from_row(conn: sqlite3.Connection) -> GeoIpSchedule:
    row = conn.execute(
        "SELECT enabled, cron, provider, updated_at FROM geoip_schedule WHERE id = 1"
    ).fetchone()
    if row is None:
        return GeoIpSchedule()
    return GeoIpSchedule(
        enabled=bool(row["enabled"]),
        cron=str(row["cron"]),
        provider=str(row["provider"]),
        updated_at=str(row["updated_at"]),
    )


class Store:
    """Доступ к состоянию: чтение + аудируемые мутации."""

    def __init__(self, db: Database) -> None:
        self._db = db

    @property
    def database(self) -> Database:
        return self._db

    # -- чтение ---------------------------------------------------------
    # Чтения идут через Database.read() под тем же RLock, что транзакции
    # (#54): это исключает перемешивание операторов разных потоков и грязные
    # чтения незафиксированных строк.
    def get_mode(self) -> Mode:
        with self._db.read() as conn:
            row = conn.execute("SELECT mode FROM routing_mode WHERE id = 1").fetchone()
        if row is None:
            return Mode.OFF
        return Mode(str(row["mode"]))

    def get_geoip_schedule(self) -> GeoIpSchedule:
        with self._db.read() as conn:
            return _geoip_from_row(conn)

    def list_rules(self, *, enabled_only: bool = False) -> list[Rule]:
        sql = "SELECT * FROM routing_rules"
        if enabled_only:
            sql += " WHERE enabled = 1"
        sql += " ORDER BY priority, id"
        with self._db.read() as conn:
            rows = conn.execute(sql).fetchall()
        return [_rule_from_row(row) for row in rows]

    def get_rule(self, rule_id: int) -> Rule | None:
        with self._db.read() as conn:
            row = conn.execute("SELECT * FROM routing_rules WHERE id = ?", (rule_id,)).fetchone()
        return None if row is None else _rule_from_row(row)

    def get_rule_by_name(self, name: str) -> Rule | None:
        with self._db.read() as conn:
            row = conn.execute("SELECT * FROM routing_rules WHERE name = ?", (name,)).fetchone()
        return None if row is None else _rule_from_row(row)

    def list_exceptions(self) -> list[RuleException]:
        with self._db.read() as conn:
            rows = conn.execute("SELECT * FROM rule_exceptions ORDER BY id").fetchall()
        return [_exception_from_row(row) for row in rows]

    def get_exception(self, match: str) -> RuleException | None:
        with self._db.read() as conn:
            row = conn.execute("SELECT * FROM rule_exceptions WHERE match = ?", (match,)).fetchone()
        return None if row is None else _exception_from_row(row)

    def list_configs(self) -> list[ConfigMeta]:
        with self._db.read() as conn:
            rows = conn.execute("SELECT * FROM configs ORDER BY name").fetchall()
        return [_config_from_row(row) for row in rows]

    def get_config(self, name: str) -> ConfigMeta | None:
        with self._db.read() as conn:
            row = conn.execute("SELECT * FROM configs WHERE name = ?", (name,)).fetchone()
        return None if row is None else _config_from_row(row)

    def list_devices(self) -> list[Device]:
        with self._db.read() as conn:
            rows = conn.execute("SELECT * FROM devices ORDER BY name").fetchall()
        return [_device_from_row(row) for row in rows]

    def get_device(self, mac: str) -> Device | None:
        normalized = validation.normalize_mac(mac)
        with self._db.read() as conn:
            row = conn.execute("SELECT * FROM devices WHERE mac = ?", (normalized,)).fetchone()
        return None if row is None else _device_from_row(row)

    def list_audit(
        self, *, entity: str | None = None, entity_id: str | int | None = None, limit: int = 100
    ) -> list[AuditEntry]:
        with self._db.read() as conn:
            return audit.list_entries(conn, entity=entity, entity_id=entity_id, limit=limit)

    # -- мутации: правила ------------------------------------------------
    def upsert_rule(self, rule: Rule, *, actor: str) -> Rule:
        """Создать или обновить правило (по ``id`` либо по уникальному ``name``)."""
        validation.validate_name("name", rule.name)
        kind = validation.validate_rule_kind(rule.kind)
        match = validation.validate_match(kind, rule.match)
        target = validation.validate_target(rule.target)
        priority = validation.validate_priority(rule.priority)
        now = utcnow()
        with self._db.transaction() as conn:
            existing = self._find_rule_row(conn, rule.id, rule.name)
            core: tuple[str, str, str, str, int, int] = (
                rule.name,
                kind,
                match,
                target,
                priority,
                1 if rule.enabled else 0,
            )
            if existing is None:
                cursor = conn.execute(
                    "INSERT INTO routing_rules (name, kind, match, target, priority, enabled,"
                    " created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                    (*core, now, now),
                )
                new_id = int(cursor.lastrowid or 0)
                saved = replace(
                    rule,
                    id=new_id,
                    kind=kind,
                    match=match,
                    target=target,
                    priority=priority,
                    created_at=now,
                    updated_at=now,
                )
                audit.record(
                    conn,
                    actor=actor,
                    action="create",
                    entity=Entity.RULE.value,
                    entity_id=new_id,
                    old_value=None,
                    new_value=saved.as_dict(),
                )
            else:
                rule_id = int(existing["id"])
                conn.execute(
                    "UPDATE routing_rules SET name = ?, kind = ?, match = ?, target = ?,"
                    " priority = ?, enabled = ?, updated_at = ? WHERE id = ?",
                    (*core, now, rule_id),
                )
                saved = replace(
                    rule,
                    id=rule_id,
                    kind=kind,
                    match=match,
                    target=target,
                    priority=priority,
                    created_at=str(existing["created_at"]),
                    updated_at=now,
                )
                audit.record(
                    conn,
                    actor=actor,
                    action="update",
                    entity=Entity.RULE.value,
                    entity_id=rule_id,
                    old_value=_rule_from_row(existing).as_dict(),
                    new_value=saved.as_dict(),
                )
            return saved

    def delete_rule(self, rule_id: int, *, actor: str) -> bool:
        """Удалить правило; ``False`` — если правила не было."""
        with self._db.transaction() as conn:
            row = self._find_rule_row(conn, rule_id, None)
            if row is None:
                return False
            removed = _rule_from_row(row)
            conn.execute("DELETE FROM routing_rules WHERE id = ?", (int(row["id"]),))
            audit.record(
                conn,
                actor=actor,
                action="delete",
                entity=Entity.RULE.value,
                entity_id=int(row["id"]),
                old_value=removed.as_dict(),
                new_value=None,
            )
        return True

    @staticmethod
    def _find_rule_row(
        conn: sqlite3.Connection, rule_id: int | None, name: str | None
    ) -> sqlite3.Row | None:
        if rule_id is not None:
            row = conn.execute("SELECT * FROM routing_rules WHERE id = ?", (rule_id,)).fetchone()
            return cast("sqlite3.Row | None", row)
        if name is not None:
            row = conn.execute("SELECT * FROM routing_rules WHERE name = ?", (name,)).fetchone()
            return cast("sqlite3.Row | None", row)
        return None

    # -- мутации: исключения ---------------------------------------------
    def upsert_exception(self, exception: RuleException, *, actor: str) -> RuleException:
        match = validation.validate_exception_match(exception.match)
        now = utcnow()
        with self._db.transaction() as conn:
            existing = conn.execute(
                "SELECT * FROM rule_exceptions WHERE match = ?", (match,)
            ).fetchone()
            if existing is None:
                conn.execute(
                    "INSERT INTO rule_exceptions (match, reason, enabled, created_at, updated_at)"
                    " VALUES (?, ?, ?, ?, ?)",
                    (match, exception.reason, 1 if exception.enabled else 0, now, now),
                )
            else:
                conn.execute(
                    "UPDATE rule_exceptions SET reason = ?, enabled = ?, updated_at = ?"
                    " WHERE match = ?",
                    (exception.reason, 1 if exception.enabled else 0, now, match),
                )
            row = conn.execute("SELECT * FROM rule_exceptions WHERE match = ?", (match,)).fetchone()
            if row is None:  # pragma: no cover - невозможен после успешной записи
                raise StorageError("правило-исключение исчезло сразу после записи")
            saved = _exception_from_row(row)
            audit.record(
                conn,
                actor=actor,
                action="create" if existing is None else "update",
                entity=Entity.EXCEPTION.value,
                entity_id=saved.id,
                old_value=None if existing is None else _exception_from_row(existing).as_dict(),
                new_value=saved.as_dict(),
            )
            return saved

    def delete_exception(self, exception_id: int, *, actor: str) -> bool:
        with self._db.transaction() as conn:
            row = conn.execute(
                "SELECT * FROM rule_exceptions WHERE id = ?", (exception_id,)
            ).fetchone()
            if row is None:
                return False
            removed = _exception_from_row(row)
            conn.execute("DELETE FROM rule_exceptions WHERE id = ?", (exception_id,))
            audit.record(
                conn,
                actor=actor,
                action="delete",
                entity=Entity.EXCEPTION.value,
                entity_id=exception_id,
                old_value=removed.as_dict(),
                new_value=None,
            )
        return True

    # -- мутации: режим и GeoIP ------------------------------------------
    def set_mode(self, mode: str | Mode, *, actor: str) -> Mode:
        """Установить режим маршрутизации; в журнал пишется old→new."""
        validated = validation.validate_mode(mode)
        with self._db.transaction() as conn:
            previous = conn.execute("SELECT mode FROM routing_mode WHERE id = 1").fetchone()
            old = "off" if previous is None else str(previous["mode"])
            now = utcnow()
            conn.execute(
                "INSERT INTO routing_mode (id, mode, updated_at) VALUES (1, ?, ?)"
                " ON CONFLICT(id) DO UPDATE SET mode = excluded.mode,"
                " updated_at = excluded.updated_at",
                (validated.value, now),
            )
            audit.record(
                conn,
                actor=actor,
                action="set_mode",
                entity=Entity.MODE.value,
                entity_id=1,
                old_value={"mode": old},
                new_value={"mode": validated.value},
            )
        return validated

    def set_geoip_schedule(
        self,
        *,
        enabled: bool | None = None,
        cron: str | None = None,
        provider: str | None = None,
        actor: str,
    ) -> GeoIpSchedule:
        """Обновить расписание GeoIP (частичное обновление) и записать аудит."""
        now = utcnow()
        with self._db.transaction() as conn:
            # чтение old_value — внутри той же транзакции, что и запись (#55 D-2):
            # иначе при конкурентном обновлении журнал заявит устаревшее «старое».
            current = _geoip_from_row(conn)
            stored = GeoIpSchedule(
                enabled=current.enabled if enabled is None else enabled,
                cron=validation.validate_cron(current.cron if cron is None else cron),
                provider=validation.validate_provider(
                    current.provider if provider is None else provider
                ),
                updated_at=now,
            )
            conn.execute(
                "INSERT INTO geoip_schedule (id, enabled, cron, provider, updated_at)"
                " VALUES (1, ?, ?, ?, ?)"
                " ON CONFLICT(id) DO UPDATE SET enabled = excluded.enabled, cron = excluded.cron,"
                " provider = excluded.provider, updated_at = excluded.updated_at",
                (1 if stored.enabled else 0, stored.cron, stored.provider, now),
            )
            audit.record(
                conn,
                actor=actor,
                action="set_geoip_schedule",
                entity=Entity.GEOIP_SCHEDULE.value,
                entity_id=1,
                old_value=current.as_dict(),
                new_value=stored.as_dict(),
            )
        return stored

    # -- мутации: конфиги -------------------------------------------------
    def upsert_config(self, config: ConfigMeta, *, actor: str) -> ConfigMeta:
        name = validation.validate_name("name", config.name)
        path = validation.validate_config_path(config.path)
        now = utcnow()
        with self._db.transaction() as conn:
            existing = conn.execute("SELECT * FROM configs WHERE name = ?", (name,)).fetchone()
            if existing is None:
                cursor = conn.execute(
                    "INSERT INTO configs (name, version, path, checksum, size_bytes, applied,"
                    " created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                    (
                        name,
                        config.version,
                        path,
                        config.checksum,
                        config.size_bytes,
                        1 if config.applied else 0,
                        now,
                        now,
                    ),
                )
                config_id = int(cursor.lastrowid or 0)
                saved = replace(
                    config, id=config_id, name=name, path=path, created_at=now, updated_at=now
                )
                audit.record(
                    conn,
                    actor=actor,
                    action="create",
                    entity=Entity.CONFIG.value,
                    entity_id=config_id,
                    old_value=None,
                    new_value=saved.as_dict(),
                )
            else:
                config_id = int(existing["id"])
                conn.execute(
                    "UPDATE configs SET version = ?, path = ?, checksum = ?, size_bytes = ?,"
                    " applied = ?, updated_at = ? WHERE id = ?",
                    (
                        config.version,
                        path,
                        config.checksum,
                        config.size_bytes,
                        1 if config.applied else 0,
                        now,
                        config_id,
                    ),
                )
                saved = replace(
                    config,
                    id=config_id,
                    name=name,
                    path=path,
                    created_at=str(existing["created_at"]),
                    updated_at=now,
                )
                audit.record(
                    conn,
                    actor=actor,
                    action="update",
                    entity=Entity.CONFIG.value,
                    entity_id=config_id,
                    old_value=_config_from_row(existing).as_dict(),
                    new_value=saved.as_dict(),
                )
            return saved

    def mark_config_applied(self, name: str, *, applied: bool, actor: str) -> ConfigMeta:
        """Отметить конфиг применённым/снятым и записать аудит."""
        with self._db.transaction() as conn:
            row = conn.execute("SELECT * FROM configs WHERE name = ?", (name,)).fetchone()
            if row is None:
                raise StorageError(f"конфиг {name!r} не найден")
            now = utcnow()
            conn.execute(
                "UPDATE configs SET applied = ?, updated_at = ? WHERE id = ?",
                (1 if applied else 0, now, int(row["id"])),
            )
            updated = replace(_config_from_row(row), applied=applied, updated_at=now)
            audit.record(
                conn,
                actor=actor,
                action="set_applied",
                entity=Entity.CONFIG.value,
                entity_id=int(row["id"]),
                old_value=_config_from_row(row).as_dict(),
                new_value=updated.as_dict(),
            )
            return updated

    # -- мутации: устройства ----------------------------------------------
    def register_device(self, device: Device, *, actor: str) -> Device:
        mac = validation.normalize_mac(device.mac)
        name = validation.validate_name("name", device.name)
        policy = validation.validate_device_policy(device.policy)
        now = utcnow()
        with self._db.transaction() as conn:
            existing = conn.execute("SELECT * FROM devices WHERE mac = ?", (mac,)).fetchone()
            if existing is None:
                cursor = conn.execute(
                    "INSERT INTO devices (mac, name, address, policy, last_seen, created_at,"
                    " updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
                    (mac, name, device.address, policy, device.last_seen, now, now),
                )
                device_id = int(cursor.lastrowid or 0)
                saved = replace(
                    device,
                    id=device_id,
                    mac=mac,
                    name=name,
                    policy=policy,
                    created_at=now,
                    updated_at=now,
                )
                audit.record(
                    conn,
                    actor=actor,
                    action="create",
                    entity=Entity.DEVICE.value,
                    entity_id=device_id,
                    old_value=None,
                    new_value=saved.as_dict(),
                )
            else:
                device_id = int(existing["id"])
                conn.execute(
                    "UPDATE devices SET name = ?, address = ?, policy = ?, last_seen = ?,"
                    " updated_at = ? WHERE id = ?",
                    (name, device.address, policy, device.last_seen, now, device_id),
                )
                saved = replace(
                    device,
                    id=device_id,
                    mac=mac,
                    name=name,
                    policy=policy,
                    created_at=str(existing["created_at"]),
                    updated_at=now,
                )
                audit.record(
                    conn,
                    actor=actor,
                    action="update",
                    entity=Entity.DEVICE.value,
                    entity_id=device_id,
                    old_value=_device_from_row(existing).as_dict(),
                    new_value=saved.as_dict(),
                )
            return saved

    def delete_device(self, mac: str, *, actor: str) -> bool:
        normalized = validation.normalize_mac(mac)
        with self._db.transaction() as conn:
            row = conn.execute("SELECT * FROM devices WHERE mac = ?", (normalized,)).fetchone()
            if row is None:
                return False
            removed = _device_from_row(row)
            conn.execute("DELETE FROM devices WHERE mac = ?", (normalized,))
            audit.record(
                conn,
                actor=actor,
                action="delete",
                entity=Entity.DEVICE.value,
                entity_id=removed.id,
                old_value=removed.as_dict(),
                new_value=None,
            )
        return True

    # -- служебное --------------------------------------------------------
    def counts(self) -> dict[str, int]:
        """Количество строк в изменяемых таблицах (для отчётов и тестов)."""
        tables = {
            "rules": "routing_rules",
            "exceptions": "rule_exceptions",
            "configs": "configs",
            "devices": "devices",
            "audit": "audit_log",
            "snapshots": "state_snapshots",
        }
        result: dict[str, int] = {}
        with self._db.read() as conn:  # один снимок на все таблицы: счётчики согласованы
            for key, table in tables.items():
                row = conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()  # noqa: S608
                result[key] = int(row[0]) if row is not None else 0
        return result

    def raw(self) -> sqlite3.Connection:
        """Соединение для кода, которому нужен доступ ниже уровня репозитория.

        Лок не берётся: в рантайме используйте :meth:`Database.read` /
        :meth:`Database.transaction` (#54), иначе возможны гонки на общем
        соединении. Метод оставлен для однопоточного кода и миграций.
        """
        return self._db.connection