Amnezia-Web-Panel-main/db/store.py

"""Panel data store backed by PostgreSQL 17.

Preserves the same dict shape that used to live in data.json so existing
FastAPI handlers keep working without a full rewrite.
"""

from __future__ import annotations

import json
import logging
import os
import shutil
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Optional
from uuid import UUID

from psycopg.types.json import Jsonb

from .connection import get_pool, init_schema

logger = logging.getLogger(__name__)

DEFAULT_SETTINGS = {
    'appearance': {
        'title': 'Amnezia',
        'logo': '❤️',
        'subtitle': 'Web Panel',
    },
    'sync': {
        'remnawave_url': '',
        'remnawave_api_key': '',
        'remnawave_sync': False,
        'remnawave_sync_users': False,
        'remnawave_create_conns': False,
        'remnawave_server_id': 0,
        'remnawave_protocol': 'awg',
        'xui_sync': False,
        'xui_url': '',
        'xui_username': '',
        'xui_password': '',
        'xui_api_token': '',
        'xui_sync_users': False,
        'xui_create_conns': False,
        'xui_server_id': 0,
        'xui_protocol': 'xray',
        'xui_inbound_id': 0,
        'xui_sub_url': '',
    },
    'guest': {
        'enabled': False,
        'token': '',
        'password_hash': None,
        'user_id': '',
        'allow_create': False,
        'create_protocol': 'xui',
        'create_server_id': 0,
        'create_inbound_id': 0,
    },
}


def _parse_ts(value: Any) -> Optional[datetime]:
    if value is None or value == '':
        return None
    if isinstance(value, datetime):
        return value
    text = str(value).strip()
    if not text:
        return None
    if text.endswith('Z'):
        text = text[:-1] + '+00:00'
    try:
        return datetime.fromisoformat(text)
    except ValueError:
        return None


def _ts_iso(value: Any) -> Optional[str]:
    if value is None:
        return None
    if isinstance(value, datetime):
        return value.isoformat()
    return str(value)


def _as_uuid(value: Any) -> UUID:
    return UUID(str(value))


def _merge_settings(raw: Optional[dict]) -> dict:
    settings = json.loads(json.dumps(DEFAULT_SETTINGS))
    if not isinstance(raw, dict):
        return settings
    for section, defaults in DEFAULT_SETTINGS.items():
        incoming = raw.get(section)
        if isinstance(incoming, dict) and isinstance(defaults, dict):
            merged = dict(defaults)
            merged.update(incoming)
            settings[section] = merged
        elif section in raw:
            settings[section] = raw[section]
    for key, value in raw.items():
        if key not in settings:
            settings[key] = value
    return settings


def _as_dict(value, default=None):
    """Coerce JSONB / legacy values to a plain dict."""
    if default is None:
        default = {}
    if value is None:
        return dict(default)
    if isinstance(value, dict):
        return dict(value)
    # Legacy: ssh.test_connection() used to return a plain string
    if isinstance(value, str):
        text = value.strip()
        if not text:
            return dict(default)
        if text.startswith('{') or text.startswith('['):
            try:
                parsed = json.loads(text)
                if isinstance(parsed, dict):
                    return parsed
            except Exception:
                pass
        return {'raw': text}
    try:
        return dict(value)
    except Exception:
        return {'raw': str(value)}


def _row_to_server(row) -> dict:
    return {
        'name': row['name'] or '',
        'host': row['host'] or '',
        'ssh_port': int(row['ssh_port'] or 22),
        'username': row['username'] or '',
        'password': row['password'],
        'private_key': row['private_key'],
        'server_info': _as_dict(row['server_info']),
        'protocols': _as_dict(row['protocols']),
    }


def _row_to_user(row) -> dict:
    return {
        'id': str(row['id']),
        'username': row['username'],
        'password_hash': row['password_hash'] or '',
        'role': row['role'] or 'user',
        'enabled': bool(row['enabled']),
        'created_at': _ts_iso(row['created_at']),
        'telegramId': row['telegram_id'],
        'email': row['email'],
        'description': row['description'],
        'traffic_limit': int(row['traffic_limit'] or 0),
        'traffic_used': int(row['traffic_used'] or 0),
        'traffic_total': int(row['traffic_total'] or 0),
        'traffic_reset_strategy': row['traffic_reset_strategy'] or 'never',
        'last_reset_at': _ts_iso(row['last_reset_at']),
        'expiration_date': _ts_iso(row['expiration_date']),
        'remnawave_uuid': row['remnawave_uuid'],
        'xui_email': row.get('xui_email'),
        'share_enabled': bool(row['share_enabled']),
        'share_token': row['share_token'],
        'share_password_hash': row['share_password_hash'],
    }


def _row_to_connection(row) -> dict:
    return {
        'id': str(row['id']),
        'user_id': str(row['user_id']),
        'server_id': int(row['server_id'] or 0),
        'protocol': row['protocol'] or '',
        'client_id': row['client_id'] or '',
        'name': row['name'] or '',
        'created_at': _ts_iso(row['created_at']),
        'last_bytes': int(row['last_bytes'] or 0),
    }


def _row_to_token(row) -> dict:
    return {
        'id': str(row['id']),
        'name': row['name'] or '',
        'token_hash': row['token_hash'] or '',
        'token_prefix': row['token_prefix'] or '',
        'user_id': str(row['user_id']),
        'created_at': _ts_iso(row['created_at']),
        'last_used_at': _ts_iso(row['last_used_at']),
    }


def _row_to_invite(row) -> dict:
    return {
        'id': str(row['id']),
        'name': row['name'] or '',
        'token': row['token'] or '',
        'enabled': bool(row['enabled']),
        'max_uses': int(row['max_uses'] or 0),
        'used_count': int(row['used_count'] or 0),
        'user_id': str(row['user_id']) if row['user_id'] else '',
        'protocol': row['protocol'] or 'xui',
        'server_id': int(row['server_id'] or 0),
        'xui_inbound_id': int(row['xui_inbound_id'] or 0),
        'password_hash': row['password_hash'],
        'expires_at': _ts_iso(row['expires_at']),
        'duration_days': int(row.get('duration_days') or 0),
        'note': row['note'] or '',
        'created_at': _ts_iso(row['created_at']),
    }


def load_data() -> dict:
    init_schema()
    pool = get_pool()
    with pool.connection() as conn:
        with conn.cursor() as cur:
            cur.execute(
                'SELECT position, name, host, ssh_port, username, password, '
                'private_key, server_info, protocols '
                'FROM servers ORDER BY position ASC'
            )
            servers = [_row_to_server(r) for r in cur.fetchall()]

            cur.execute(
                'SELECT id, username, password_hash, role, enabled, created_at, '
                'telegram_id, email, description, traffic_limit, traffic_used, '
                'traffic_total, traffic_reset_strategy, last_reset_at, '
                'expiration_date, remnawave_uuid, xui_email, share_enabled, share_token, '
                'share_password_hash FROM users ORDER BY created_at NULLS LAST, username'
            )
            users = [_row_to_user(r) for r in cur.fetchall()]

            cur.execute(
                'SELECT id, user_id, server_id, protocol, client_id, name, '
                'created_at, last_bytes FROM user_connections '
                'ORDER BY created_at NULLS LAST, id'
            )
            user_connections = [_row_to_connection(r) for r in cur.fetchall()]

            cur.execute(
                'SELECT id, name, token_hash, token_prefix, user_id, '
                'created_at, last_used_at FROM api_tokens '
                'ORDER BY created_at NULLS LAST, id'
            )
            api_tokens = [_row_to_token(r) for r in cur.fetchall()]

            cur.execute(
                'SELECT id, name, token, enabled, max_uses, used_count, user_id, '
                'protocol, server_id, xui_inbound_id, password_hash, expires_at, '
                'duration_days, note, created_at FROM invite_links '
                'ORDER BY created_at DESC NULLS LAST, name'
            )
            invite_links = [_row_to_invite(r) for r in cur.fetchall()]

            cur.execute('SELECT data FROM settings WHERE id = 1')
            settings_row = cur.fetchone()
            settings = _merge_settings(settings_row['data'] if settings_row else None)

    return {
        'servers': servers,
        'users': users,
        'user_connections': user_connections,
        'api_tokens': api_tokens,
        'invite_links': invite_links,
        'settings': settings,
    }


def save_data(data: dict) -> None:
    """Replace panel state in a single transaction (same semantics as rewriting data.json)."""
    init_schema()
    servers = data.get('servers') or []
    users = data.get('users') or []
    connections = data.get('user_connections') or []
    tokens = data.get('api_tokens') or []
    invite_links = data.get('invite_links') or []
    settings = _merge_settings(data.get('settings'))

    # Keep only connections whose user still exists
    user_ids = {str(u.get('id')) for u in users if u.get('id')}
    connections = [c for c in connections if str(c.get('user_id')) in user_ids]
    tokens = [t for t in tokens if str(t.get('user_id')) in user_ids]
    # Clear holder if user was deleted
    for link in invite_links:
        if link.get('user_id') and str(link['user_id']) not in user_ids:
            link['user_id'] = ''

    pool = get_pool()
    with pool.connection() as conn:
        with conn.cursor() as cur:
            # Order matters for FKs: children first on delete, parents first on insert
            cur.execute('DELETE FROM invite_links')
            cur.execute('DELETE FROM api_tokens')
            cur.execute('DELETE FROM user_connections')
            cur.execute('DELETE FROM users')
            cur.execute('DELETE FROM servers')

            for pos, server in enumerate(servers):
                cur.execute(
                    'INSERT INTO servers (position, name, host, ssh_port, username, '
                    'password, private_key, server_info, protocols) '
                    'VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)',
                    (
                        pos,
                        server.get('name') or '',
                        server.get('host') or '',
                        int(server.get('ssh_port') or 22),
                        server.get('username') or '',
                        server.get('password'),
                        server.get('private_key'),
                        Jsonb(_as_dict(server.get('server_info'))),
                        Jsonb(_as_dict(server.get('protocols'))),
                    ),
                )

            for user in users:
                cur.execute(
                    'INSERT INTO users ('
                    'id, username, password_hash, role, enabled, created_at, '
                    'telegram_id, email, description, traffic_limit, traffic_used, '
                    'traffic_total, traffic_reset_strategy, last_reset_at, '
                    'expiration_date, remnawave_uuid, xui_email, share_enabled, share_token, '
                    'share_password_hash'
                    ') VALUES ('
                    '%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s'
                    ')',
                    (
                        _as_uuid(user['id']),
                        user.get('username') or '',
                        user.get('password_hash') or '',
                        user.get('role') or 'user',
                        bool(user.get('enabled', True)),
                        _parse_ts(user.get('created_at')),
                        str(user['telegramId']) if user.get('telegramId') is not None else None,
                        user.get('email'),
                        user.get('description'),
                        int(user.get('traffic_limit') or 0),
                        int(user.get('traffic_used') or 0),
                        int(user.get('traffic_total') or 0),
                        user.get('traffic_reset_strategy') or 'never',
                        _parse_ts(user.get('last_reset_at')),
                        _parse_ts(user.get('expiration_date')),
                        user.get('remnawave_uuid'),
                        user.get('xui_email'),
                        bool(user.get('share_enabled', False)),
                        user.get('share_token'),
                        user.get('share_password_hash'),
                    ),
                )

            for conn_row in connections:
                cur.execute(
                    'INSERT INTO user_connections ('
                    'id, user_id, server_id, protocol, client_id, name, created_at, last_bytes'
                    ') VALUES (%s, %s, %s, %s, %s, %s, %s, %s)',
                    (
                        _as_uuid(conn_row['id']),
                        _as_uuid(conn_row['user_id']),
                        int(conn_row.get('server_id') or 0),
                        conn_row.get('protocol') or '',
                        conn_row.get('client_id') or '',
                        conn_row.get('name') or '',
                        _parse_ts(conn_row.get('created_at')),
                        int(conn_row.get('last_bytes') or 0),
                    ),
                )

            for token in tokens:
                cur.execute(
                    'INSERT INTO api_tokens ('
                    'id, name, token_hash, token_prefix, user_id, created_at, last_used_at'
                    ') VALUES (%s, %s, %s, %s, %s, %s, %s)',
                    (
                        _as_uuid(token['id']),
                        token.get('name') or '',
                        token.get('token_hash') or '',
                        token.get('token_prefix') or '',
                        _as_uuid(token['user_id']),
                        _parse_ts(token.get('created_at')),
                        _parse_ts(token.get('last_used_at')),
                    ),
                )

            for link in invite_links:
                uid = link.get('user_id') or None
                cur.execute(
                    'INSERT INTO invite_links ('
                    'id, name, token, enabled, max_uses, used_count, user_id, '
                    'protocol, server_id, xui_inbound_id, password_hash, expires_at, '
                    'duration_days, note, created_at'
                    ') VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)',
                    (
                        _as_uuid(link['id']),
                        link.get('name') or '',
                        link.get('token') or '',
                        bool(link.get('enabled', True)),
                        int(link.get('max_uses') or 0),
                        int(link.get('used_count') or 0),
                        _as_uuid(uid) if uid else None,
                        link.get('protocol') or 'xui',
                        int(link.get('server_id') or 0),
                        int(link.get('xui_inbound_id') or 0),
                        link.get('password_hash'),
                        _parse_ts(link.get('expires_at')),
                        int(link.get('duration_days') or 0),
                        link.get('note') or '',
                        _parse_ts(link.get('created_at')),
                    ),
                )

            cur.execute(
                'INSERT INTO settings (id, data) VALUES (1, %s) '
                'ON CONFLICT (id) DO UPDATE SET data = EXCLUDED.data',
                (Jsonb(settings),),
            )
        conn.commit()


def export_data_dict() -> dict:
    """Export current DB state as the legacy data.json document."""
    return load_data()


def is_database_empty() -> bool:
    init_schema()
    pool = get_pool()
    with pool.connection() as conn:
        with conn.cursor() as cur:
            cur.execute('SELECT COUNT(*) AS n FROM users')
            users = cur.fetchone()['n']
            cur.execute('SELECT COUNT(*) AS n FROM servers')
            servers = cur.fetchone()['n']
            return users == 0 and servers == 0


def import_from_json_file(path: str | os.PathLike, *, backup: bool = True) -> bool:
    """Import legacy data.json into Postgres. Returns True if import ran."""
    path = Path(path)
    if not path.exists():
        return False

    with path.open('r', encoding='utf-8') as f:
        data = json.load(f)

    if not isinstance(data, dict):
        raise ValueError(f'Invalid data.json: expected object, got {type(data).__name__}')

    data.setdefault('servers', [])
    data.setdefault('users', [])
    data.setdefault('user_connections', [])
    data.setdefault('api_tokens', [])
    data.setdefault('invite_links', [])
    data.setdefault('settings', {})

    save_data(data)

    if backup:
        stamp = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
        backup_path = path.with_name(f'data.json.migrated.{stamp}.bak')
        try:
            shutil.copy2(path, backup_path)
            logger.info('Backed up legacy data.json to %s', backup_path)
        except OSError as e:
            logger.warning('Could not backup data.json: %s', e)

    logger.info(
        'Imported data.json → PostgreSQL (%s servers, %s users, %s connections)',
        len(data.get('servers', [])),
        len(data.get('users', [])),
        len(data.get('user_connections', [])),
    )
    return True


def load_tunnel_state() -> dict:
    init_schema()
    pool = get_pool()
    with pool.connection() as conn:
        with conn.cursor() as cur:
            cur.execute('SELECT provider, data FROM tunnel_state')
            rows = cur.fetchall()
    return {row['provider']: dict(row['data'] or {}) for row in rows}


def save_tunnel_state(state: dict) -> None:
    init_schema()
    pool = get_pool()
    with pool.connection() as conn:
        with conn.cursor() as cur:
            cur.execute('DELETE FROM tunnel_state')
            for provider, payload in (state or {}).items():
                cur.execute(
                    'INSERT INTO tunnel_state (provider, data) VALUES (%s, %s)',
                    (str(provider), Jsonb(payload or {})),
                )
        conn.commit()


def update_tunnel_state(provider: str, **updates) -> None:
    state = load_tunnel_state()
    provider_state = state.get(provider, {})
    provider_state.update(updates)
    state[provider] = provider_state
    save_tunnel_state(state)


def clear_tunnel_state(provider: str) -> None:
    state = load_tunnel_state()
    if provider in state:
        state.pop(provider)
        save_tunnel_state(state)


def ensure_db_ready(legacy_data_file: Optional[str] = None) -> None:
    """Init schema and one-shot import from data.json when DB is empty."""
    init_schema()
    if legacy_data_file and is_database_empty() and os.path.exists(legacy_data_file):
        logger.info('Empty database — importing legacy %s', legacy_data_file)
        import_from_json_file(legacy_data_file)