MapMil/centers/analytics/api/app/services/vpn.py

"""VPN settings persistence helpers."""

from __future__ import annotations

from contracts.queues import known_source_types
from contracts.vpn import VpnSettings
from sqlalchemy.orm import Session

from ..models import VpnSettingsRow
from ..schemas import VpnSettingsAdminRead, VpnSettingsInternalRead, VpnSettingsUpdate

SINGLETON_ID = 1


def _hint_secret(value: str | None) -> str | None:
    if not value:
        return None
    if len(value) <= 4:
        return "****"
    return f"…{value[-4:]}"


def ensure_vpn_settings(db: Session) -> VpnSettingsRow:
    row = db.query(VpnSettingsRow).filter(VpnSettingsRow.id == SINGLETON_ID).first()
    if row is not None:
        return row
    row = VpnSettingsRow(
        id=SINGLETON_ID,
        enabled=False,
        mode="subscription",
        subscription_interval_seconds=3600,
        proxied_source_types=[],
    )
    db.add(row)
    db.commit()
    db.refresh(row)
    return row


def row_to_contract(row: VpnSettingsRow) -> VpnSettings:
    return VpnSettings.model_validate(
        {
            "enabled": row.enabled,
            "mode": row.mode,
            "subscription_url": row.subscription_url,
            "subscription_interval_seconds": row.subscription_interval_seconds,
            "host": row.host,
            "port": row.port,
            "username": row.username,
            "password": row.password,
            "proxied_source_types": row.proxied_source_types or [],
        }
    )


def to_admin_read(row: VpnSettingsRow) -> VpnSettingsAdminRead:
    return VpnSettingsAdminRead(
        enabled=row.enabled,
        mode=row.mode,  # type: ignore[arg-type]
        subscription_url_set=bool(row.subscription_url),
        subscription_url_hint=_hint_secret(row.subscription_url),
        subscription_interval_seconds=row.subscription_interval_seconds,
        host=row.host,
        port=row.port,
        username=row.username,
        password_set=bool(row.password),
        proxied_source_types=list(row.proxied_source_types or []),
        available_source_types=known_source_types(),
    )


def to_internal_read(row: VpnSettingsRow) -> VpnSettingsInternalRead:
    return VpnSettingsInternalRead(
        enabled=row.enabled,
        mode=row.mode,  # type: ignore[arg-type]
        subscription_url=row.subscription_url,
        subscription_interval_seconds=row.subscription_interval_seconds,
        host=row.host,
        port=row.port,
        username=row.username,
        password=row.password,
        proxied_source_types=list(row.proxied_source_types or []),
    )


def apply_vpn_update(db: Session, payload: VpnSettingsUpdate) -> VpnSettingsRow:
    row = ensure_vpn_settings(db)
    data = {
        "enabled": row.enabled if payload.enabled is None else payload.enabled,
        "mode": row.mode if payload.mode is None else payload.mode,
        "subscription_url": row.subscription_url,
        "subscription_interval_seconds": (
            row.subscription_interval_seconds
            if payload.subscription_interval_seconds is None
            else payload.subscription_interval_seconds
        ),
        "host": row.host if payload.host is None else (payload.host.strip() or None),
        "port": row.port if payload.port is None else payload.port,
        "username": (
            row.username
            if payload.username is None
            else (payload.username.strip() or None)
        ),
        "password": row.password,
        "proxied_source_types": (
            list(row.proxied_source_types or [])
            if payload.proxied_source_types is None
            else payload.proxied_source_types
        ),
    }

    if payload.clear_subscription_url:
        data["subscription_url"] = None
    elif payload.subscription_url is not None:
        stripped = payload.subscription_url.strip()
        data["subscription_url"] = stripped or None

    if payload.clear_password:
        data["password"] = None
    elif payload.password is not None:
        stripped = payload.password.strip()
        data["password"] = stripped or None

    validated = VpnSettings.model_validate(data)

    row.enabled = validated.enabled
    row.mode = validated.mode
    row.subscription_url = validated.subscription_url
    row.subscription_interval_seconds = validated.subscription_interval_seconds
    row.host = validated.host
    row.port = validated.port
    row.username = validated.username
    row.password = validated.password
    row.proxied_source_types = validated.proxied_source_types
    db.commit()
    db.refresh(row)
    return row