core.py

from __future__ import annotations

import base64
import hashlib
import hmac
import json
import mimetypes
import os
import re
import secrets
import shutil
import struct
import threading
import uuid
import zipfile
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Iterable

from cryptography.hazmat.primitives.ciphers.aead import AESGCM, ChaCha20Poly1305
from huggingface_hub import HfApi, hf_hub_download

MB = 1024 * 1024
GB = 1024 * MB
STATE_REMOTE_PATH = "state/state.json"
MAGIC = b"CVLT3"
CHUNK_SIZE = 4 * MB


def utcnow() -> datetime:
    return datetime.now(timezone.utc)


def iso(value: datetime | None = None) -> str:
    return (value or utcnow()).isoformat()


def parse_dt(value: str | None) -> datetime | None:
    if not value:
        return None
    try:
        parsed = datetime.fromisoformat(value)
        return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
    except (TypeError, ValueError):
        return None



@dataclass(frozen=True)
class Plan:
    key: str
    name: str
    quota_bytes: int
    reserved_bytes: int
    retention_days: int
    delete_after_age_even_if_downloaded: bool
    concurrent_downloads: int
    concurrent_uploads: int
    analytics_delay_hours: int
    analytics_history_days: int
    token_length: int
    folder_upload: bool
    encryption: str
    display_price: str
    old_price: str | None
    chat_access: bool
    priority: str


PLANS: dict[str, Plan] = {
    "free": Plan(
        key="free", name="Free", quota_bytes=250 * MB, reserved_bytes=250 * MB,
        retention_days=2, delete_after_age_even_if_downloaded=True,
        concurrent_downloads=1, concurrent_uploads=1,
        analytics_delay_hours=24, analytics_history_days=30,
        token_length=24, folder_upload=True, encryption="none",
        display_price="$0", old_price=None, chat_access=False, priority="Low",
    ),
    "plus": Plan(
        key="plus", name="Plus", quota_bytes=int(3.5 * GB), reserved_bytes=int(3.5 * GB),
        retention_days=7, delete_after_age_even_if_downloaded=False,
        concurrent_downloads=3, concurrent_uploads=3,
        analytics_delay_hours=12, analytics_history_days=365,
        token_length=8, folder_upload=True, encryption="aes256-gcm",
        display_price="$2.49", old_price="$5.00", chat_access=True, priority="Medium",
    ),
    "pro": Plan(
        key="pro", name="Pro", quota_bytes=int(6.5 * GB), reserved_bytes=int(6.5 * GB),
        retention_days=30, delete_after_age_even_if_downloaded=False,
        concurrent_downloads=6, concurrent_uploads=6,
        analytics_delay_hours=6, analytics_history_days=365,
        token_length=12, folder_upload=True, encryption="chacha20-poly1305",
        display_price="$4.99", old_price="$10.00", chat_access=True, priority="High",
    ),
    "admin": Plan(
        key="admin", name="Developer", quota_bytes=10**18, reserved_bytes=0,
        retention_days=36500, delete_after_age_even_if_downloaded=False,
        concurrent_downloads=10**9, concurrent_uploads=10**9,
        analytics_delay_hours=0, analytics_history_days=3650,
        token_length=16, folder_upload=True, encryption="chacha20-poly1305",
        display_price="Developer", old_price=None, chat_access=True, priority="Developer",
    ),
}


@dataclass(frozen=True)
class Settings:
    hf_token: str | None
    dataset_repo: str | None
    master_key: str
    admin_password: str
    public_base_url: str
    total_storage_capacity_bytes: int
    system_storage_reserve_bytes: int
    app_storage_limit_bytes: int
    purchase_safety_bytes: int
    temp_dir: Path
    local_root: Path
    local_mode: bool


def get_settings() -> Settings:
    temp_dir = Path(os.getenv("TEMP_DIR", "/tmp/cloudvault-temp"))
    local_root = Path(os.getenv("LOCAL_STORAGE_DIR", "/tmp/cloudvault-local"))
    temp_dir.mkdir(parents=True, exist_ok=True)
    local_root.mkdir(parents=True, exist_ok=True)
    token = os.getenv("HF_TOKEN") or None
    repo = os.getenv("HF_DATASET_REPO") or None
    forced_local = os.getenv("LOCAL_MODE", "").lower() in {"1", "true", "yes"}
    space_host = os.getenv("SPACE_HOST", "").strip()
    public_url = os.getenv("PUBLIC_BASE_URL", "").strip().rstrip("/")
    if not public_url and space_host:
        public_url = f"https://{space_host}"
    total_storage_capacity_bytes = int(float(os.getenv("TOTAL_STORAGE_GB", "95")) * GB)
    system_storage_reserve_bytes = int(float(os.getenv("SYSTEM_STORAGE_RESERVE_GB", "5")) * GB)
    system_storage_reserve_bytes = min(max(0, system_storage_reserve_bytes), total_storage_capacity_bytes)
    package_pool_bytes = max(0, total_storage_capacity_bytes - system_storage_reserve_bytes)
    configured_package_pool = int(float(os.getenv("APP_STORAGE_LIMIT_GB", "90")) * GB)
    return Settings(
        hf_token=token,
        dataset_repo=repo,
        master_key=os.getenv("APP_MASTER_KEY", "CHANGE-ME-IN-SPACE-SECRETS"),
        admin_password="eses1428",
        public_base_url=public_url,
        total_storage_capacity_bytes=total_storage_capacity_bytes,
        system_storage_reserve_bytes=system_storage_reserve_bytes,
        app_storage_limit_bytes=min(package_pool_bytes, max(0, configured_package_pool)),
        purchase_safety_bytes=int(float(os.getenv("PURCHASE_SAFETY_GB", "0")) * GB),
        temp_dir=temp_dir,
        local_root=local_root,
        local_mode=forced_local or not (token and repo),
    )


SETTINGS = get_settings()


def human_bytes(value: int) -> str:
    size = float(max(0, value))
    for unit in ("B", "KB", "MB", "GB", "TB"):
        if size < 1024 or unit == "TB":
            return f"{size:.0f} {unit}" if unit in {"B", "KB"} else f"{size:.2f} {unit}"
        size /= 1024
    return f"{size:.2f} TB"


def safe_filename(name: str) -> str:
    clean = Path(name or "file.bin").name.replace("\x00", "")
    clean = re.sub(r"[\r\n]", "", clean).strip()
    return clean[:180] or "file.bin"


def validate_registration(username: str, password: str) -> list[str]:
    username = (username or "").strip()
    password = password or ""
    errors: list[str] = []
    if len(username) < 3:
        errors.append("Username must contain at least 3 characters.")
    if len(username) > 32:
        errors.append("Username may contain at most 32 characters.")
    if not re.fullmatch(r"[A-Za-z0-9_.-]+", username):
        errors.append("Username may contain letters, numbers, dot, underscore and dash only.")
    if username.casefold() == "admin":
        errors.append("This username is reserved.")
    if len(password) < 6:
        errors.append("Password must contain at least 6 characters.")
    if not re.search(r"[A-Za-z]", password):
        errors.append("Password must contain at least one letter.")
    if not re.search(r"\d", password):
        errors.append("Password must contain at least one number.")
    return errors


def hash_password(password: str) -> str:
    salt = secrets.token_bytes(16)
    digest = hashlib.scrypt(password.encode("utf-8"), salt=salt, n=2**14, r=8, p=1, dklen=32)
    return "scrypt$" + base64.urlsafe_b64encode(salt).decode() + "$" + base64.urlsafe_b64encode(digest).decode()


def verify_password(password: str, stored: str | None) -> bool:
    try:
        algorithm, salt_text, digest_text = (stored or "").split("$", 2)
        if algorithm != "scrypt":
            return False
        salt = base64.urlsafe_b64decode(salt_text.encode())
        expected = base64.urlsafe_b64decode(digest_text.encode())
        actual = hashlib.scrypt(password.encode("utf-8"), salt=salt, n=2**14, r=8, p=1, dklen=32)
        return hmac.compare_digest(actual, expected)
    except (ValueError, TypeError):
        return False


def random_share_id(length: int) -> str:
    alphabet = "ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz23456789"
    return "".join(secrets.choice(alphabet) for _ in range(length))


class StorageBackend:
    def __init__(self, settings: Settings) -> None:
        self.settings = settings
        self.api = HfApi(token=settings.hf_token) if not settings.local_mode else None
        if self.api and settings.dataset_repo:
            self.api.create_repo(repo_id=settings.dataset_repo, repo_type="dataset", private=True, exist_ok=True)

    @property
    def persistent(self) -> bool:
        return not self.settings.local_mode

    def upload(self, local_path: Path, remote_path: str, message: str) -> None:
        if self.settings.local_mode:
            target = self.settings.local_root / remote_path
            target.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(local_path, target)
            return
        assert self.api and self.settings.dataset_repo
        self.api.upload_file(
            path_or_fileobj=str(local_path), path_in_repo=remote_path,
            repo_id=self.settings.dataset_repo, repo_type="dataset", commit_message=message,
        )

    def download(self, remote_path: str, local_path: Path) -> bool:
        local_path.parent.mkdir(parents=True, exist_ok=True)
        if self.settings.local_mode:
            source = self.settings.local_root / remote_path
            if not source.exists():
                return False
            shutil.copy2(source, local_path)
            return True
        assert self.settings.dataset_repo
        try:
            cached = hf_hub_download(
                repo_id=self.settings.dataset_repo, repo_type="dataset",
                filename=remote_path, token=self.settings.hf_token,
            )
            shutil.copy2(cached, local_path)
            return True
        except Exception:
            return False

    def delete_many(self, remote_paths: Iterable[str], message: str = "Delete CloudVaultHub files") -> None:
        paths = [p for p in dict.fromkeys(remote_paths) if p]
        if not paths:
            return
        if self.settings.local_mode:
            for remote in paths:
                (self.settings.local_root / remote).unlink(missing_ok=True)
            return
        assert self.api and self.settings.dataset_repo
        from huggingface_hub import CommitOperationDelete
        operations = [CommitOperationDelete(path_in_repo=path) for path in paths]
        try:
            self.api.create_commit(
                repo_id=self.settings.dataset_repo, repo_type="dataset",
                operations=operations, commit_message=message,
            )
        except Exception:
            pass


BACKEND = StorageBackend(SETTINGS)


class StateStore:
    def __init__(self, backend: StorageBackend, settings: Settings) -> None:
        self.backend = backend
        self.settings = settings
        self.lock = threading.RLock()
        self.state: dict[str, Any] = {
            "version": 3,
            "users": {},
            "files": {},
            "events": [],
            "chat": [],
            "chat_rooms": {},
            "support_tickets": [],
            "config": {},
        }
        self._load()

    def _load(self) -> None:
        changed = False
        loaded_existing = False
        with self.lock:
            temp = self.settings.temp_dir / f"state-load-{uuid.uuid4().hex}.json"
            if self.backend.download(STATE_REMOTE_PATH, temp):
                loaded_existing = True
                try:
                    loaded = json.loads(temp.read_text(encoding="utf-8"))
                    if isinstance(loaded, dict):
                        self.state.update(loaded)
                except (OSError, json.JSONDecodeError):
                    pass
                finally:
                    temp.unlink(missing_ok=True)
            changed |= self._normalize_unlocked()
            changed |= self._ensure_admin_unlocked()
            if changed or not loaded_existing:
                self._save_unlocked("Initialize or upgrade CloudVaultHub state")

    def _normalize_unlocked(self) -> bool:
        changed = False
        defaults = {
            "system_enabled": True,
            "chat_enabled": True,
            "plus_chat_enabled": True,
            "pro_chat_enabled": False,
            "support_enabled": True,
            "support_force_open": False,
            "support_auto_paused": False,
            "purchases_enabled": True,
            "registrations_enabled": True,
            "delete_paid_on_expiry": False,
            "shutdown_mode": False,
            "chat_mode": "single",
            "chat_room_capacity": 0,
        }
        if not isinstance(self.state.get("users"), dict):
            self.state["users"] = {}; changed = True
        if not isinstance(self.state.get("files"), dict):
            self.state["files"] = {}; changed = True
        if not isinstance(self.state.get("events"), list):
            self.state["events"] = []; changed = True
        if not isinstance(self.state.get("notifications"), list):
            self.state["notifications"] = []; changed = True
        if not isinstance(self.state.get("notification_reads"), dict):
            self.state["notification_reads"] = {}; changed = True
        if not isinstance(self.state.get("plan_features"), dict):
            self.state["plan_features"] = {}; changed = True
        if not isinstance(self.state.get("websites"), dict):
            self.state["websites"] = {}; changed = True
        if not isinstance(self.state.get("chat"), list):
            self.state["chat"] = []; changed = True
        if not isinstance(self.state.get("chat_rooms"), dict):
            self.state["chat_rooms"] = {}; changed = True
        if not isinstance(self.state.get("support_tickets"), list):
            self.state["support_tickets"] = []; changed = True
        for ticket in self.state["support_tickets"]:
            ticket.setdefault("id", uuid.uuid4().hex)
            ticket.setdefault("user_id", "")
            ticket.setdefault("username", "User")
            ticket.setdefault("plan", "plus")
            ticket.setdefault("subject", "")
            ticket.setdefault("message", "")
            ticket.setdefault("created_at", iso())
            ticket.setdefault("status", "open")
            ticket.setdefault("reply", "")
            ticket.setdefault("replied_at", None)
            ticket.setdefault("read_by_user", False)
        if not isinstance(self.state.get("config"), dict):
            self.state["config"] = {}; changed = True
        for key, value in defaults.items():
            if key not in self.state["config"]:
                self.state["config"][key] = value; changed = True
        # CloudVaultHub uses exactly one unlimited Global Chat room.
        if self.state["config"].get("chat_mode") != "single":
            self.state["config"]["chat_mode"] = "single"; changed = True
        if self.state["config"].get("chat_room_capacity") != 0:
            self.state["config"]["chat_room_capacity"] = 0; changed = True
        if "global-1" not in self.state["chat_rooms"]:
            self.state["chat_rooms"]["global-1"] = {
                "id": "global-1", "name": "Global 1", "paused": False, "created_at": iso()
            }
            changed = True
        if set(self.state["chat_rooms"]) != {"global-1"}:
            room1 = self.state["chat_rooms"].get("global-1", {
                "id": "global-1", "name": "Global Chat", "paused": False, "created_at": iso()
            })
            room1["name"] = "Global Chat"
            self.state["chat_rooms"] = {"global-1": room1}
            changed = True
        for message in self.state["chat"]:
            if message.get("room_id") != "global-1":
                message["room_id"] = "global-1"; changed = True
        for user in self.state["users"].values():
            if "role" not in user:
                user["role"] = "user"; changed = True
            normalized_role = str(user.get("role") or "user").strip().lower()
            if normalized_role not in {"user", "admin"}:
                normalized_role = "user"
            if user.get("role") != normalized_role:
                user["role"] = normalized_role; changed = True
            normalized_plan = str(user.get("plan") or "free").strip().lower()
            if normalized_role == "admin":
                normalized_plan = "admin"
            elif normalized_plan not in {"free", "plus", "pro"}:
                normalized_plan = "free"
            if user.get("plan") != normalized_plan:
                user["plan"] = normalized_plan; changed = True
            if "is_banned" not in user:
                user["is_banned"] = False; changed = True
            if "ban_until" not in user:
                user["ban_until"] = None; changed = True
            if "delete_on_expiry" not in user:
                user["delete_on_expiry"] = False; changed = True
            if user.get("chat_room_id") != "global-1":
                user["chat_room_id"] = "global-1"; changed = True
            billing_defaults = {
                "pending_purchase_plan": None,
                "pending_purchase_expires_at": None,
                "billing_subscription_id": None,
                "billing_customer_id": None,
                "billing_plan": None,
                "billing_status": None,
                "billing_email": None,
                "billing_auto_renew": False,
                "billing_renews_at": None,
                "billing_ends_at": None,
                "billing_anchor": None,
                "billing_card_brand": None,
                "billing_card_last_four": None,
                "billing_payment_processor": None,
                "billing_urls": {},
                "billing_updated_at": None,
            }
            for billing_key, billing_value in billing_defaults.items():
                if billing_key not in user:
                    user[billing_key] = billing_value
                    changed = True
        # Keep only the newest 400 messages in the single Global Chat room.
        normalized_chat = sorted(self.state["chat"], key=lambda m: m.get("created_at", ""))[-400:]
        for message in normalized_chat:
            message["room_id"] = "global-1"
        if len(normalized_chat) != len(self.state["chat"]):
            changed = True
        self.state["chat"] = normalized_chat
        self.state["version"] = 4
        return changed

    def _ensure_admin_unlocked(self) -> bool:
        """Guarantee exactly one canonical backend admin account."""
        users = self.state["users"]
        exact = next(
            (u for u in users.values() if str(u.get("username", "")).strip().casefold() == "admin"),
            None,
        )
        legacy_admins = [
            u for u in users.values()
            if str(u.get("role", "")).strip().lower() == "admin"
        ]
        admin = exact or (legacy_admins[0] if legacy_admins else None)
        now = utcnow()
        if admin is None:
            uid = "cloudvault-admin"
            admin = {
                "id": uid,
                "username": "admin",
                "password_hash": hash_password(self.settings.admin_password),
                "plan": "admin",
                "role": "admin",
                "plan_started_at": iso(now),
                "plan_expires_at": None,
                "created_at": iso(now),
                "is_banned": False,
                "ban_until": None,
                "delete_on_expiry": False,
                "last_chat_at": None,
                "chat_room_id": "global-1",
                "pending_purchase_plan": None,
                "pending_purchase_expires_at": None,
                "billing_subscription_id": None,
                "billing_customer_id": None,
                "billing_plan": None,
                "billing_status": None,
                "billing_email": None,
                "billing_auto_renew": False,
                "billing_renews_at": None,
                "billing_ends_at": None,
                "billing_anchor": None,
                "billing_card_brand": None,
                "billing_card_last_four": None,
                "billing_payment_processor": None,
                "billing_urls": {},
                "billing_updated_at": None,
            }
            users[uid] = admin
            return True

        changed = False
        # Migrate an old role=admin account to the canonical username instead of
        # creating a second hidden admin account.
        if str(admin.get("username", "")).strip().casefold() != "admin":
            admin["username"] = "admin"
            changed = True

        required = {
            "plan": "admin", "role": "admin", "is_banned": False,
            "ban_until": None, "plan_expires_at": None, "delete_on_expiry": False,
            "chat_room_id": admin.get("chat_room_id") or "global-1",
        }
        for key, value in required.items():
            if admin.get(key) != value:
                admin[key] = value
                changed = True
        if not verify_password(self.settings.admin_password, admin.get("password_hash")):
            admin["password_hash"] = hash_password(self.settings.admin_password)
            changed = True

        # No other account may retain backend admin privileges.
        for other in users.values():
            if other is admin:
                continue
            if str(other.get("role", "")).strip().lower() == "admin":
                other["role"] = "user"
                other["plan"] = "free"
                other["plan_expires_at"] = None
                other["delete_on_expiry"] = False
                changed = True
        return changed

    def _save_unlocked(self, message: str) -> None:
        temp = self.settings.temp_dir / f"state-save-{uuid.uuid4().hex}.json"
        temp.write_text(json.dumps(self.state, ensure_ascii=False, separators=(",", ":")), encoding="utf-8")
        self.backend.upload(temp, STATE_REMOTE_PATH, message)
        temp.unlink(missing_ok=True)

    def _username_exists_unlocked(self, username: str) -> bool:
        wanted = username.strip().casefold()
        return any(str(u.get("username", "")).casefold() == wanted for u in self.state["users"].values())

    def _is_admin_unlocked(self, user_id: str | None) -> bool:
        user = self.state["users"].get(user_id or "")
        return bool(user and user.get("role") == "admin")

    def _assert_admin_unlocked(self, user_id: str | None) -> None:
        if not self._is_admin_unlocked(user_id):
            raise PermissionError("Developer access is required.")

    def _effective_reserved_for_user_unlocked(
        self,
        user: dict[str, Any],
        now: datetime | None = None,
    ) -> int:
        if user.get("role") == "admin":
            return 0

        current_plan = PLANS.get(
            str(user.get("plan", "free")),
            PLANS["free"],
        )
        reserved = current_plan.reserved_bytes
        check_time = now or utcnow()

        pending_plan_key = str(
            user.get("pending_purchase_plan") or ""
        )
        pending_expires = parse_dt(
            user.get("pending_purchase_expires_at")
        )

        if (
            pending_plan_key in {"plus", "pro"}
            and pending_expires
            and pending_expires > check_time
        ):
            reserved = max(
                reserved,
                PLANS[pending_plan_key].reserved_bytes,
            )

        return reserved

    def _reserved_unlocked(
        self,
        exclude_user_id: str | None = None,
    ) -> int:
        now = utcnow()
        total = 0

        for uid, user in self.state["users"].items():
            if uid == exclude_user_id or user.get("role") == "admin":
                continue
            total += self._effective_reserved_for_user_unlocked(
                user,
                now,
            )

        return total


    def _used_unlocked(self, user_id: str | None = None) -> int:
        return sum(
            int(f.get("original_size", 0)) for f in self.state["files"].values()
            if user_id is None or f.get("owner_id") == user_id
        )

    def _admin_used_unlocked(self) -> int:
        admin_ids = {uid for uid, u in self.state["users"].items() if u.get("role") == "admin"}
        return sum(int(f.get("original_size", 0)) for f in self.state["files"].values() if f.get("owner_id") in admin_ids)

    def _package_used_unlocked(self) -> int:
        admin_ids = {uid for uid, u in self.state["users"].items() if u.get("role") == "admin"}
        return sum(
            int(f.get("original_size", 0))
            for f in self.state["files"].values()
            if f.get("owner_id") not in admin_ids
        )

    def _capacity_for_new_reservation_unlocked(
        self,
        bytes_needed: int,
        exclude_user_id: str | None = None,
        safety_bytes: int = 0,
    ) -> bool:
        projected = (
            self._reserved_unlocked(
                exclude_user_id=exclude_user_id
            )
            + max(0, int(bytes_needed))
            + max(0, int(safety_bytes))
        )
        return projected <= self.settings.app_storage_limit_bytes

    def _capacity_for_paid_plan_unlocked(
        self,
        user_id: str,
        plan_key: str,
    ) -> bool:
        return self._capacity_for_new_reservation_unlocked(
            PLANS[plan_key].reserved_bytes,
            exclude_user_id=user_id,
            safety_bytes=self.settings.purchase_safety_bytes,
        )

    def purchase_capacity(
        self,
        user_id: str,
        plan_key: str,
    ) -> dict[str, Any]:
        if plan_key not in {"plus", "pro"}:
            raise ValueError("Invalid paid package.")

        with self.lock:
            if user_id not in self.state["users"]:
                raise ValueError("Account not found.")

            projected_without_safety = (
                self._reserved_unlocked(
                    exclude_user_id=user_id
                )
                + PLANS[plan_key].reserved_bytes
            )
            remaining = max(
                0,
                self.settings.app_storage_limit_bytes
                - projected_without_safety,
            )
            return {
                "allowed": remaining >= self.settings.purchase_safety_bytes,
                "remaining_after_reservation": remaining,
                "required_safety": self.settings.purchase_safety_bytes,
                "target_reserved": PLANS[plan_key].reserved_bytes,
            }


    def config(self) -> dict[str, Any]:
        with self.lock:
            return dict(self.state["config"])

    def username_exists(self, username: str) -> bool:
        with self.lock:
            return self._username_exists_unlocked(username)

    def is_admin(self, user_id: str | None) -> bool:
        with self.lock:
            return self._is_admin_unlocked(user_id)

    def create_user(self, username: str, password_hash: str, plan_key: str) -> dict[str, Any]:
        if plan_key not in {"free", "plus", "pro"}:
            raise ValueError("Invalid package.")
        plan = PLANS[plan_key]
        with self.lock:
            cfg = self.state["config"]
            if not cfg.get("system_enabled", True):
                raise ValueError("The system is temporarily paused.")
            if not cfg.get("registrations_enabled", True):
                raise ValueError("New account registration is currently closed.")
            if plan_key in {"plus", "pro"} and not cfg.get("purchases_enabled", True):
                raise ValueError("Package activation is currently paused.")
            if self._username_exists_unlocked(username):
                raise ValueError("This username is already in use.")
            if not self._capacity_for_new_reservation_unlocked(plan.reserved_bytes):
                raise ValueError("The server does not have enough reserved storage to create this account.")
            user_id = uuid.uuid4().hex
            now = utcnow()
            user = {
                "id": user_id,
                "username": username.strip(),
                "password_hash": password_hash,
                "plan": plan_key,
                "role": "user",
                "plan_started_at": iso(now),
                "plan_expires_at": iso(now + timedelta(days=28)) if plan_key != "free" else None,
                "created_at": iso(now),
                "is_banned": False,
                "ban_until": None,
                "delete_on_expiry": bool(cfg.get("delete_paid_on_expiry", False) and plan_key != "free"),
                "last_chat_at": None,
                "chat_room_id": None,
                "pending_purchase_plan": None,
                "pending_purchase_expires_at": None,
                "billing_subscription_id": None,
                "billing_customer_id": None,
                "billing_plan": None,
                "billing_status": None,
                "billing_email": None,
                "billing_auto_renew": False,
                "billing_renews_at": None,
                "billing_ends_at": None,
                "billing_anchor": None,
                "billing_card_brand": None,
                "billing_card_last_four": None,
                "billing_payment_processor": None,
                "billing_urls": {},
                "billing_updated_at": None,
            }
            self.state["users"][user_id] = user
            self._save_unlocked("Create CloudVaultHub account")
            return dict(user)

    def authenticate(self, username: str, password: str) -> dict[str, Any] | None:
        with self.lock:
            wanted = (username or "").strip().casefold()
            user = next((u for u in self.state["users"].values() if str(u.get("username", "")).casefold() == wanted), None)
            if not user or not verify_password(password, user.get("password_hash")):
                return None
            now = utcnow()
            ban_until = parse_dt(user.get("ban_until"))
            if user.get("is_banned"):
                if ban_until and ban_until <= now:
                    user["is_banned"] = False
                    user["ban_until"] = None
                    self._save_unlocked("Automatically remove expired ban")
                else:
                    when = "permanently" if not ban_until else f"until {ban_until.strftime('%d %b %Y %H:%M UTC')}"
                    raise PermissionError(f"This account is banned {when}.")
            if user.get("role") != "admin" and not self.state["config"].get("system_enabled", True):
                raise PermissionError("CloudVaultHub is temporarily paused by the developer.")
            return dict(user)

    def get_user(self, user_id: str | None) -> dict[str, Any] | None:
        if not user_id:
            return None
        self.cleanup(user_id)
        with self.lock:
            user = self.state["users"].get(user_id)
            return dict(user) if user else None

    def list_users(self, admin_id: str) -> list[dict[str, Any]]:
        with self.lock:
            self._assert_admin_unlocked(admin_id)
            users = [dict(u) for u in self.state["users"].values() if u.get("role") != "admin"]
        return sorted(users, key=lambda u: u.get("created_at", ""), reverse=True)

    def change_plan(self, user_id: str, plan_key: str) -> dict[str, Any]:
        if plan_key not in {"free", "plus", "pro"}:
            raise ValueError("Invalid package.")
        plan = PLANS[plan_key]
        with self.lock:
            user = self.state["users"].get(user_id)
            if not user:
                raise ValueError("Account not found.")
            if user.get("role") == "admin":
                raise ValueError("The developer account does not use packages.")
            cfg = self.state["config"]
            if not cfg.get("system_enabled", True):
                raise ValueError("The system is temporarily paused.")
            if plan_key in {"plus", "pro"} and not cfg.get("purchases_enabled", True):
                raise ValueError("Package activation is currently paused.")
            if (
                plan_key in {"plus", "pro"}
                and not self._capacity_for_paid_plan_unlocked(
                    user_id,
                    plan_key,
                )
            ):
                raise ValueError(
                    "The server cannot reserve this package while keeping "
                    f"{human_bytes(self.settings.purchase_safety_bytes)} free."
                )
            now = utcnow()
            previous_plan = user.get("plan", "free")
            user["plan"] = plan_key
            if previous_plan == "free" and plan_key in {"plus", "pro"}:
                user["chat_room_id"] = None
            user["plan_started_at"] = iso(now)
            user["plan_expires_at"] = iso(now + timedelta(days=28)) if plan_key != "free" else None
            user["delete_on_expiry"] = bool(cfg.get("delete_paid_on_expiry", False) and plan_key != "free")
            removed = self._trim_user_unlocked(user_id, plan.quota_bytes)
            self._save_unlocked("Change CloudVaultHub package")
        self.backend.delete_many(removed, "Trim files after package change")
        return self.get_user(user_id) or {}

    def admin_change_user_plan(self, admin_id: str, user_id: str, plan_key: str, duration_days: int = 28) -> dict[str, Any]:
        """Developer-only package assignment that does not require a checkout."""
        if plan_key not in {"free", "plus", "pro"}:
            raise ValueError("Invalid package.")
        with self.lock:
            self._assert_admin_unlocked(admin_id)
            user = self.state["users"].get(user_id)
            if not user:
                raise ValueError("Account not found.")
            if user.get("role") == "admin":
                raise ValueError("The developer account cannot be assigned a customer package.")
            plan = PLANS[plan_key]
            if plan_key in {"plus", "pro"} and not self._capacity_for_paid_plan_unlocked(user_id, plan_key):
                raise ValueError("There is not enough reservable server storage for this package.")
            now = utcnow()
            user["plan"] = plan_key
            user["plan_started_at"] = iso(now)
            user["plan_expires_at"] = iso(now + timedelta(days=max(1, int(duration_days)))) if plan_key != "free" else None
            user["delete_on_expiry"] = False
            user["pending_purchase_plan"] = None
            user["pending_purchase_expires_at"] = None
            removed = self._trim_user_unlocked(user_id, plan.quota_bytes)
            self._save_unlocked("Developer changed CloudVaultHub user package")
        self.backend.delete_many(removed, "Trim files after developer package change")
        return self.get_user(user_id) or {}

    def begin_purchase(
        self,
        user_id: str,
        plan_key: str,
        hold_hours: int = 24,
    ) -> dict[str, Any]:
        if plan_key not in {"plus", "pro"}:
            raise ValueError("Invalid paid package.")

        with self.lock:
            user = self.state["users"].get(user_id)
            if not user:
                raise ValueError("Account not found.")
            if user.get("role") == "admin":
                raise ValueError(
                    "The developer account does not use paid packages."
                )

            cfg = self.state["config"]
            if not cfg.get("system_enabled", True):
                raise ValueError("The system is temporarily paused.")
            if not cfg.get("purchases_enabled", True):
                raise ValueError(
                    "Package purchases are currently paused."
                )

            subscription_id = str(
                user.get("billing_subscription_id") or ""
            )
            billing_status = str(
                user.get("billing_status") or ""
            )
            if (
                subscription_id
                and billing_status in {
                    "active",
                    "on_trial",
                    "past_due",
                    "paused",
                    "cancelled",
                }
            ):
                raise ValueError(
                    "This account already has a subscription. "
                    "Use Billing Settings or the Customer Portal."
                )

            if not self._capacity_for_paid_plan_unlocked(
                user_id,
                plan_key,
            ):
                raise ValueError(
                    "The server cannot reserve this package while keeping "
                    f"{human_bytes(self.settings.purchase_safety_bytes)} free."
                )

            expires_at = utcnow() + timedelta(
                hours=max(1, min(int(hold_hours), 72))
            )
            user["pending_purchase_plan"] = plan_key
            user["pending_purchase_expires_at"] = iso(expires_at)
            self._save_unlocked(
                f"Reserve CloudVaultHub {plan_key} checkout capacity"
            )
            return dict(user)

    def billing_info(
        self,
        user_id: str | None,
    ) -> dict[str, Any]:
        if not user_id:
            return {}

        self.cleanup(user_id)
        with self.lock:
            user = self.state["users"].get(user_id)
            if not user:
                return {}

            keys = [
                "id", "username", "plan", "plan_expires_at",
                "pending_purchase_plan", "pending_purchase_expires_at",
                "billing_subscription_id", "billing_customer_id",
                "billing_plan", "billing_status", "billing_email",
                "billing_auto_renew", "billing_renews_at",
                "billing_ends_at", "billing_anchor",
                "billing_card_brand", "billing_card_last_four",
                "billing_payment_processor", "billing_urls",
                "billing_updated_at",
            ]
            return {key: user.get(key) for key in keys}

    def find_user_by_billing_email(
        self,
        email: str | None,
    ) -> dict[str, Any] | None:
        wanted = str(email or "").strip().casefold()
        if not wanted:
            return None

        with self.lock:
            user = next(
                (
                    account
                    for account in self.state["users"].values()
                    if str(
                        account.get("billing_email") or ""
                    ).strip().casefold() == wanted
                ),
                None,
            )
            return dict(user) if user else None

    def apply_subscription_snapshot(
        self,
        user_id: str,
        plan_key: str,
        subscription_id: str,
        attributes: dict[str, Any],
        event_name: str = "subscription_updated",
    ) -> dict[str, Any]:
        if plan_key not in {"plus", "pro"}:
            raise ValueError("Unsupported subscription package.")

        removed: list[str] = []
        with self.lock:
            user = self.state["users"].get(user_id)
            if not user:
                raise ValueError("Account not found.")
            if user.get("role") == "admin":
                raise ValueError(
                    "The developer account cannot receive a subscription."
                )

            status = str(
                attributes.get("status") or ""
            ).strip().lower()
            cancelled = bool(
                attributes.get("cancelled")
                or status == "cancelled"
            )
            renews_at = attributes.get("renews_at")
            ends_at = attributes.get("ends_at")
            urls = attributes.get("urls")
            if not isinstance(urls, dict):
                urls = {}

            user["billing_subscription_id"] = (
                str(subscription_id or "")
                or user.get("billing_subscription_id")
            )
            user["billing_customer_id"] = attributes.get(
                "customer_id",
                user.get("billing_customer_id"),
            )
            user["billing_plan"] = plan_key
            user["billing_status"] = status or event_name
            user["billing_email"] = attributes.get(
                "user_email",
                user.get("billing_email"),
            )
            user["billing_auto_renew"] = (
                not cancelled
                and status not in {"expired", "unpaid"}
            )
            user["billing_renews_at"] = renews_at
            user["billing_ends_at"] = ends_at
            user["billing_anchor"] = attributes.get("billing_anchor")
            user["billing_card_brand"] = attributes.get("card_brand")
            user["billing_card_last_four"] = attributes.get("card_last_four")
            user["billing_payment_processor"] = attributes.get(
                "payment_processor"
            )
            if urls:
                user["billing_urls"] = urls
            user["billing_updated_at"] = iso()
            user["pending_purchase_plan"] = None
            user["pending_purchase_expires_at"] = None

            now = utcnow()
            if status == "expired":
                user["plan"] = "free"
                user["plan_started_at"] = iso(now)
                user["plan_expires_at"] = None
                user["billing_auto_renew"] = False
                removed.extend(
                    self._trim_user_unlocked(
                        user_id,
                        PLANS["free"].quota_bytes,
                    )
                )
            else:
                user["plan"] = plan_key
                user["plan_started_at"] = (
                    user.get("plan_started_at") or iso(now)
                )
                expiry_value = (
                    ends_at
                    if cancelled and ends_at
                    else renews_at
                )
                expiry = parse_dt(expiry_value)
                if expiry is None:
                    expiry = now + timedelta(days=28)
                user["plan_expires_at"] = iso(expiry)
                user["delete_on_expiry"] = False

            self._save_unlocked(
                f"Apply Lemon Squeezy event {event_name}"
            )

        self.backend.delete_many(
            removed,
            "Trim files after subscription expiry",
        )
        return self.get_user(user_id) or {}

    def used_bytes(self, user_id: str) -> int:
        with self.lock:
            return self._used_unlocked(user_id)

    def total_used_bytes(self) -> int:
        with self.lock:
            return self._used_unlocked()

    def total_reserved_bytes(self) -> int:
        with self.lock:
            return self._reserved_unlocked()

    def available_for_admin(self) -> int:
        with self.lock:
            return max(0, self.settings.app_storage_limit_bytes - self._used_unlocked())

    def list_files(self, user_id: str) -> list[dict[str, Any]]:
        self.cleanup(user_id)
        with self.lock:
            files = [dict(f) for f in self.state["files"].values() if f.get("owner_id") == user_id]
        return sorted(files, key=lambda item: item.get("created_at", ""), reverse=True)

    def _unique_token_unlocked(self, length: int) -> str:
        existing = {f.get("share_token") for f in self.state["files"].values()}
        for _ in range(300):
            token = random_share_id(length)
            if token not in existing:
                return token
        raise RuntimeError("Could not create a unique sharing ID.")

    def add_file(self, user_id: str, metadata: dict[str, Any]) -> dict[str, Any]:
        with self.lock:
            user = self.state["users"].get(user_id)
            if not user:
                raise ValueError("Account not found.")
            cfg = self.state["config"]
            if user.get("role") != "admin" and not cfg.get("system_enabled", True):
                raise ValueError("The system is temporarily paused.")
            size = int(metadata["original_size"])
            is_admin = str(user.get("role", "")).lower() == "admin"
            plan_key = str(user.get("plan", "free")).lower()
            plan = PLANS["admin"] if is_admin else PLANS.get(plan_key, PLANS["free"])
            if is_admin:
                remaining = max(0, self.settings.app_storage_limit_bytes - self._used_unlocked())
                if size > remaining:
                    raise ValueError(
                        f"Upload failed: developer storage is full. Available {human_bytes(remaining)}, file size {human_bytes(size)}."
                    )
            else:
                used_by_user = self._used_unlocked(user_id)
                remaining = max(0, plan.quota_bytes - used_by_user)
                if size > remaining:
                    raise ValueError(
                        f"Upload failed: your {plan.name} storage is full. Available {human_bytes(remaining)}, file size {human_bytes(size)}."
                    )
                server_remaining = max(0, self.settings.app_storage_limit_bytes - self._used_unlocked())
                if size > server_remaining:
                    raise ValueError(
                        f"Upload failed: CloudVaultHub server storage is full. Available {human_bytes(server_remaining)}, file size {human_bytes(size)}."
                    )
            metadata = dict(metadata)
            metadata["share_token"] = self._unique_token_unlocked(plan.token_length)
            metadata["share_password_hash"] = None
            self.state["files"][metadata["id"]] = metadata
            self._save_unlocked("Save CloudVaultHub file metadata")
            return dict(metadata)

    def get_file_by_token(self, token: str) -> dict[str, Any] | None:
        self.cleanup()
        with self.lock:
            for item in self.state["files"].values():
                if item.get("share_token") == token:
                    return dict(item)
        return None

    def delete_file(self, user_id: str, file_id: str) -> bool:
        remote = None
        with self.lock:
            item = self.state["files"].get(file_id)
            if not item or item.get("owner_id") != user_id:
                return False
            remote = item.get("remote_path")
            del self.state["files"][file_id]
            self.state["events"] = [e for e in self.state["events"] if e.get("file_id") != file_id]
            self._save_unlocked("Delete CloudVaultHub file")
        self.backend.delete_many([remote], "Delete CloudVaultHub file object")
        return True

    def _delete_user_unlocked(self, user_id: str) -> list[str]:
        user = self.state["users"].get(user_id)
        if not user or user.get("role") == "admin":
            return []
        remotes: list[str] = []
        for file_id, item in list(self.state["files"].items()):
            if item.get("owner_id") == user_id:
                remotes.append(item.get("remote_path", ""))
                del self.state["files"][file_id]
        self.state["events"] = [e for e in self.state["events"] if e.get("owner_id") != user_id]
        self.state["chat"] = [m for m in self.state["chat"] if m.get("sender_id") != user_id]
        self.state["support_tickets"] = [t for t in self.state["support_tickets"] if t.get("user_id") != user_id]
        del self.state["users"][user_id]
        return remotes

    def delete_account(self, user_id: str) -> bool:
        with self.lock:
            if user_id not in self.state["users"] or self._is_admin_unlocked(user_id):
                return False
            remotes = self._delete_user_unlocked(user_id)
            self._save_unlocked("Delete CloudVaultHub account")
        self.backend.delete_many(remotes, "Delete account objects")
        return True

    def record_download(self, file_id: str, country: str, device: str) -> None:
        with self.lock:
            item = self.state["files"].get(file_id)
            if not item:
                return
            now = iso()
            item["download_count"] = int(item.get("download_count", 0)) + 1
            item["last_download_at"] = now
            self.state["events"].append({
                "id": uuid.uuid4().hex,
                "file_id": file_id,
                "owner_id": item.get("owner_id"),
                "created_at": now,
                "country": (country or "Unknown")[:100],
                "device": (device or "Unknown")[:60],
            })
            cutoff = utcnow() - timedelta(days=400)
            self.state["events"] = [
                e for e in self.state["events"]
                if (parse_dt(e.get("created_at")) or utcnow()) >= cutoff
            ][-50000:]
            self._save_unlocked("Record CloudVaultHub download")

    def analytics(self, user_id: str, requested_days: int) -> dict[str, Any]:
        user = self.get_user(user_id)
        if not user:
            return {"events": [], "total": 0, "unique_files": 0, "countries": {}, "devices": {}, "days": 0, "delay": 0}
        is_admin = str(user.get("role", "")).lower() == "admin"
        plan_key = str(user.get("plan", "free")).lower()
        plan = PLANS["admin"] if is_admin else PLANS.get(plan_key, PLANS["free"])
        days = max(1, min(int(requested_days), plan.analytics_history_days))
        now = utcnow()
        start = now - timedelta(days=days)
        visible_until = now - timedelta(hours=plan.analytics_delay_hours)
        with self.lock:
            events = []
            for event in self.state["events"]:
                if event.get("owner_id") != user_id:
                    continue
                timestamp = parse_dt(event.get("created_at"))
                if timestamp and start <= timestamp <= visible_until:
                    events.append(dict(event))
        countries: dict[str, int] = {}
        devices: dict[str, int] = {}
        for event in events:
            country = event.get("country", "Unknown")
            device = event.get("device", "Unknown")
            countries[country] = countries.get(country, 0) + 1
            devices[device] = devices.get(device, 0) + 1
        return {
            "events": sorted(events, key=lambda e: e.get("created_at", ""), reverse=True),
            "total": len(events),
            "unique_files": len({e.get("file_id") for e in events}),
            "countries": countries,
            "devices": devices,
            "days": days,
            "delay": plan.analytics_delay_hours,
        }

    def _trim_user_unlocked(self, user_id: str, max_bytes: int) -> list[str]:
        owned = [f for f in self.state["files"].values() if f.get("owner_id") == user_id]
        owned.sort(key=lambda f: f.get("created_at", ""))
        used = sum(int(f.get("original_size", 0)) for f in owned)
        remotes: list[str] = []
        for item in owned:
            if used <= max_bytes:
                break
            used -= int(item.get("original_size", 0))
            remotes.append(item.get("remote_path", ""))
            self.state["files"].pop(item["id"], None)
            self.state["events"] = [e for e in self.state["events"] if e.get("file_id") != item["id"]]
        return remotes

    def _notify_user_unlocked(
        self,
        user_id: str,
        title: str,
        message: str,
        kind: str = "system",
    ) -> None:
        """Append a targeted notification while the store lock is already held."""
        if user_id not in self.state.get("users", {}):
            return
        notifications = self.state.setdefault("notifications", [])
        notifications.append({
            "id": uuid.uuid4().hex,
            "title": str(title or "CloudVaultHub update")[:120],
            "message": str(message or "")[:3000],
            "audience": [],
            "kind": str(kind or "system"),
            "user_id": user_id,
            "recipient_ids": [user_id],
            "created_at": iso(),
        })
        if len(notifications) > 2000:
            del notifications[:-2000]

    def cleanup(self, only_user_id: str | None = None) -> None:
        remotes: list[str] = []
        changed = False
        now = utcnow()
        with self.lock:
            user_ids = [only_user_id] if only_user_id else list(self.state["users"].keys())
            for uid in user_ids:
                user = self.state["users"].get(uid)
                if not user or user.get("role") == "admin":
                    continue
                ban_until = parse_dt(user.get("ban_until"))
                if user.get("is_banned") and ban_until and ban_until <= now:
                    user["is_banned"] = False
                    user["ban_until"] = None
                    changed = True
                pending_expires = parse_dt(
                    user.get("pending_purchase_expires_at")
                )
                if (
                    user.get("pending_purchase_plan")
                    and pending_expires
                    and pending_expires <= now
                ):
                    user["pending_purchase_plan"] = None
                    user["pending_purchase_expires_at"] = None
                    changed = True

                expires = parse_dt(user.get("plan_expires_at"))
                if user.get("plan") != "free" and expires and expires <= now:
                    should_delete = bool(user.get("delete_on_expiry") or self.state["config"].get("delete_paid_on_expiry", False))
                    if should_delete:
                        remotes.extend(self._delete_user_unlocked(uid))
                    else:
                        user["plan"] = "free"
                        user["plan_started_at"] = iso(now)
                        user["plan_expires_at"] = None
                        user["delete_on_expiry"] = False
                        user["billing_status"] = "expired"
                        user["billing_auto_renew"] = False
                        user["billing_ends_at"] = iso(now)
                        user["pending_purchase_plan"] = None
                        user["pending_purchase_expires_at"] = None
                        remotes.extend(self._trim_user_unlocked(uid, PLANS["free"].quota_bytes))
                        self._notify_user_unlocked(
                            uid,
                            "Your package expired",
                            "Your paid package ended and the account returned to Free. Files above the Free storage limit may have been removed.",
                            "account",
                        )
                    changed = True

            for file_id, item in list(self.state["files"].items()):
                if only_user_id and item.get("owner_id") != only_user_id:
                    continue
                user = self.state["users"].get(item.get("owner_id"))
                if not user:
                    remotes.append(item.get("remote_path", ""))
                    del self.state["files"][file_id]
                    changed = True
                    continue
                if user.get("role") == "admin":
                    continue
                uploaded_plan = PLANS.get(item.get("uploaded_under_plan", user.get("plan", "free")), PLANS["free"])
                created = parse_dt(item.get("created_at")) or now
                age = now - created
                delete = False
                if uploaded_plan.delete_after_age_even_if_downloaded and age >= timedelta(days=uploaded_plan.retention_days):
                    delete = True
                elif not uploaded_plan.delete_after_age_even_if_downloaded and int(item.get("download_count", 0)) == 0 and age >= timedelta(days=uploaded_plan.retention_days):
                    delete = True
                if delete:
                    display_name = str(item.get("display_name") or "A file")
                    retention_days = max(0, int(uploaded_plan.retention_days))
                    remotes.append(item.get("remote_path", ""))
                    del self.state["files"][file_id]
                    self.state["events"] = [e for e in self.state["events"] if e.get("file_id") != file_id]
                    self._notify_user_unlocked(
                        str(item.get("owner_id") or ""),
                        "File removed automatically",
                        f"{display_name} was deleted automatically after {retention_days} day(s) under the {uploaded_plan.name} retention rule.",
                        "file_deleted",
                    )
                    changed = True
            if changed:
                self._save_unlocked("Automatic CloudVaultHub cleanup")
        self.backend.delete_many(remotes, "Delete expired CloudVaultHub objects")

    # ---------------- Developer administration ----------------
    def admin_overview(self, admin_id: str) -> dict[str, Any]:
        self.cleanup()
        with self.lock:
            self._assert_admin_unlocked(admin_id)
            counts = {"free": 0, "plus": 0, "pro": 0}
            used_by_plan = {"free": 0, "plus": 0, "pro": 0, "admin": 0}
            for uid, user in self.state["users"].items():
                is_admin = str(user.get("role", "")).strip().lower() == "admin"
                plan_key = str(user.get("plan") or "free").strip().lower()
                if not is_admin and plan_key not in counts:
                    plan_key = "free"
                if not is_admin:
                    counts[plan_key] += 1
                bucket = "admin" if is_admin else plan_key
                used_by_plan[bucket] = used_by_plan.get(bucket, 0) + self._used_unlocked(uid)
            total_used = self._used_unlocked()
            package_used = self._package_used_unlocked()
            system_used = self._admin_used_unlocked()
            return {
                "counts": counts,
                "registered": sum(counts.values()),
                "used_by_plan": used_by_plan,
                "total_used": total_used,
                "remaining": max(0, self.settings.total_storage_capacity_bytes - total_used),
                "limit": self.settings.app_storage_limit_bytes,
                "total_capacity": self.settings.total_storage_capacity_bytes,
                "system_reserve": self.settings.system_storage_reserve_bytes,
                "system_used": system_used,
                "system_remaining": max(0, self.settings.system_storage_reserve_bytes - system_used),
                "package_capacity": self.settings.app_storage_limit_bytes,
                "package_used": package_used,
                "package_remaining": max(0, self.settings.app_storage_limit_bytes - package_used),
                "storage_pool_remaining": max(0, self.settings.app_storage_limit_bytes - total_used),
                "reserved": self._reserved_unlocked(),
                "config": dict(self.state["config"]),
                "chat_messages": len(self.state["chat"]),
                "support_tickets": len(self.state["support_tickets"]),
                "support_open": sum(1 for t in self.state["support_tickets"] if t.get("status") != "answered"),
            }

    def ban_user(self, admin_id: str, target_id: str, hours: float | None = None) -> None:
        with self.lock:
            self._assert_admin_unlocked(admin_id)
            target = self.state["users"].get(target_id)
            if not target or target.get("role") == "admin":
                raise ValueError("The selected account cannot be banned.")
            target["is_banned"] = True
            target["ban_until"] = iso(utcnow() + timedelta(hours=float(hours))) if hours and float(hours) > 0 else None
            self._save_unlocked("Ban CloudVaultHub account")

    def unban_user(self, admin_id: str, target_id: str) -> None:
        with self.lock:
            self._assert_admin_unlocked(admin_id)
            target = self.state["users"].get(target_id)
            if not target or target.get("role") == "admin":
                raise ValueError("Account not found.")
            target["is_banned"] = False
            target["ban_until"] = None
            self._save_unlocked("Unban CloudVaultHub account")

    def admin_delete_user(self, admin_id: str, target_id: str) -> None:
        with self.lock:
            self._assert_admin_unlocked(admin_id)
            remotes = self._delete_user_unlocked(target_id)
            if not remotes and target_id not in self.state["users"]:
                pass
            self._save_unlocked("Developer deleted account")
        self.backend.delete_many(remotes, "Developer delete account objects")

    def factory_reset(self, admin_id: str, confirmation: str) -> None:
        """Delete all user data and restore the service to first-install state."""
        if str(confirmation or "").strip() != "RESET CLOUDVAULTHUB":
            raise ValueError("Type RESET CLOUDVAULTHUB exactly.")
        with self.lock:
            self._assert_admin_unlocked(admin_id)
            # Delete every persisted object except the state file itself.
            remote_paths: list[str] = []
            if self.settings.local_mode:
                root = self.settings.local_root
                if root.exists():
                    for item in root.rglob("*"):
                        if item.is_file() and item.relative_to(root).as_posix() != STATE_REMOTE_PATH:
                            remote_paths.append(item.relative_to(root).as_posix())
            else:
                try:
                    assert self.backend.api and self.settings.dataset_repo
                    remote_paths = [
                        name for name in self.backend.api.list_repo_files(
                            repo_id=self.settings.dataset_repo, repo_type="dataset"
                        ) if name != STATE_REMOTE_PATH and name != ".gitattributes"
                    ]
                except Exception:
                    # Fall back to paths referenced by the current state.
                    for item in self.state.get("files", {}).values():
                        if isinstance(item, dict) and item.get("remote_path"):
                            remote_paths.append(str(item["remote_path"]))
                    for project in self.state.get("websites", {}).values():
                        for item in (project.get("entries", {}) if isinstance(project, dict) else {}).values():
                            if isinstance(item, dict) and item.get("remote_path"):
                                remote_paths.append(str(item["remote_path"]))
            self.backend.delete_many(remote_paths, "Factory reset CloudVaultHub storage")
            self.state = {
                "version": 4, "users": {}, "files": {}, "events": [], "chat": [],
                "chat_rooms": {}, "support_tickets": [], "config": {}, "websites": {},
                "notifications": [], "notification_reads": {}, "plan_features": {},
            }
            self._normalize_unlocked()
            self._ensure_admin_unlocked()
            self._save_unlocked("Factory reset CloudVaultHub")

    def delete_all_free(self, admin_id: str) -> int:
        with self.lock:
            self._assert_admin_unlocked(admin_id)
            targets = [uid for uid, u in self.state["users"].items() if u.get("role") != "admin" and u.get("plan") == "free"]
            remotes: list[str] = []
            for uid in targets:
                remotes.extend(self._delete_user_unlocked(uid))
            self._save_unlocked("Delete all Free accounts")
        self.backend.delete_many(remotes, "Delete all Free account objects")
        return len(targets)

    def delete_all_users(self, admin_id: str) -> int:
        with self.lock:
            self._assert_admin_unlocked(admin_id)
            targets = [uid for uid, u in self.state["users"].items() if u.get("role") != "admin"]
            remotes: list[str] = []
            for uid in targets:
                remotes.extend(self._delete_user_unlocked(uid))
            self._save_unlocked("Delete all CloudVaultHub user accounts")
        self.backend.delete_many(remotes, "Delete all user objects")
        return len(targets)

    def set_config(self, admin_id: str, **updates: Any) -> dict[str, Any]:
        allowed = {
            "system_enabled", "chat_enabled", "plus_chat_enabled", "pro_chat_enabled",
            "support_enabled", "support_force_open", "support_auto_paused", "purchases_enabled",
            "registrations_enabled", "delete_paid_on_expiry", "shutdown_mode",
        }
        with self.lock:
            self._assert_admin_unlocked(admin_id)
            for key, value in updates.items():
                if key in allowed:
                    self.state["config"][key] = bool(value)
            self._save_unlocked("Update CloudVaultHub developer controls")
            return dict(self.state["config"])

    def begin_shutdown(self, admin_id: str) -> int:
        with self.lock:
            self._assert_admin_unlocked(admin_id)
            self.state["config"].update({
                "system_enabled": False,
                "chat_enabled": False,
                "plus_chat_enabled": False,
                "pro_chat_enabled": False,
                "purchases_enabled": False,
                "registrations_enabled": False,
                "delete_paid_on_expiry": True,
                "shutdown_mode": True,
            })
            free_ids = [uid for uid, u in self.state["users"].items() if u.get("role") != "admin" and u.get("plan") == "free"]
            remotes: list[str] = []
            for uid in free_ids:
                remotes.extend(self._delete_user_unlocked(uid))
            for user in self.state["users"].values():
                if user.get("role") != "admin" and user.get("plan") in {"plus", "pro"}:
                    user["delete_on_expiry"] = True
            self._save_unlocked("Begin CloudVaultHub shutdown mode")
        self.backend.delete_many(remotes, "Shutdown mode delete Free objects")
        return len(free_ids)

    def cancel_shutdown(self, admin_id: str) -> None:
        with self.lock:
            self._assert_admin_unlocked(admin_id)
            self.state["config"].update({
                "system_enabled": True,
                "chat_enabled": True,
                "plus_chat_enabled": True,
                "pro_chat_enabled": True,
                "purchases_enabled": True,
                "registrations_enabled": True,
                "delete_paid_on_expiry": False,
                "shutdown_mode": False,
            })
            for user in self.state["users"].values():
                if user.get("role") != "admin":
                    user["delete_on_expiry"] = False
            self._save_unlocked("Cancel CloudVaultHub shutdown mode")

    # ---------------- Admin support tickets ----------------
    def _require_support_user_unlocked(self, user_id: str) -> dict[str, Any]:
        user = self.state["users"].get(user_id)
        if not user:
            raise PermissionError("Login required.")
        if user.get("role") == "admin":
            return user
        if user.get("plan") not in {"plus", "pro"}:
            raise PermissionError("Admin Support is available only for Plus and Pro accounts.")
        return user

    def support_runtime(self) -> dict[str, Any]:
        with self.lock:
            cfg = self.state["config"]
            enabled = bool(cfg.get("support_enabled", True))
            forced = bool(cfg.get("support_force_open", False))
            auto_paused = bool(cfg.get("support_auto_paused", False))
            return {"enabled": bool(enabled and (forced or not auto_paused)), "manual_enabled": enabled, "force_open": forced, "auto_paused": auto_paused}

    def set_support_memory_state(self, paused: bool) -> None:
        with self.lock:
            paused = bool(paused)
            if self.state["config"].get("support_auto_paused", False) != paused:
                self.state["config"]["support_auto_paused"] = paused
                self._save_unlocked("Update CloudVaultHub support memory protection")

    def submit_support_ticket(self, user_id: str, subject: str, message: str) -> dict[str, Any]:
        subject = (subject or "").strip(); message = (message or "").strip()
        if not subject: raise ValueError("Subject is required.")
        if len(subject) > 200: raise ValueError("Subject may contain at most 200 characters.")
        if not message: raise ValueError("Message is required.")
        if len(message) > 5000: raise ValueError("Message may contain at most 5000 characters.")
        with self.lock:
            user = self._require_support_user_unlocked(user_id)
            cfg = self.state["config"]
            if not cfg.get("support_enabled", True): raise PermissionError("Admin Support is currently paused.")
            if cfg.get("support_auto_paused", False) and not cfg.get("support_force_open", False): raise PermissionError("Admin Support is temporarily paused by server protection.")
            today = utcnow().date()
            for ticket in self.state["support_tickets"]:
                created = parse_dt(ticket.get("created_at"))
                if ticket.get("user_id") == user_id and created and created.date() == today:
                    raise ValueError("You can send only one Admin Support message per day.")
            ticket = {"id": uuid.uuid4().hex, "user_id": user_id, "username": user.get("username", "User"), "plan": user.get("plan", "plus"), "subject": subject, "message": message, "created_at": iso(), "status": "open", "reply": "", "replied_at": None, "read_by_user": False}
            self.state["support_tickets"].append(ticket)
            self._save_unlocked("Create CloudVaultHub support ticket")
            return dict(ticket)

    def user_support_tickets(self, user_id: str) -> list[dict[str, Any]]:
        with self.lock:
            self._require_support_user_unlocked(user_id)
            return sorted([dict(t) for t in self.state["support_tickets"] if t.get("user_id") == user_id], key=lambda t: t.get("created_at", ""), reverse=True)

    def admin_support_tickets(self, admin_id: str) -> list[dict[str, Any]]:
        with self.lock:
            self._assert_admin_unlocked(admin_id)
            priority = {"pro": 0, "plus": 1}
            tickets = [dict(t) for t in self.state["support_tickets"]]
            tickets.sort(key=lambda t: (priority.get(str(t.get("plan", "plus")), 2), -(parse_dt(t.get("created_at")).timestamp() if parse_dt(t.get("created_at")) else 0)))
            return tickets

    def get_support_ticket(self, actor_id: str, ticket_id: str) -> dict[str, Any] | None:
        with self.lock:
            actor = self.state["users"].get(actor_id)
            if not actor: return None
            for ticket in self.state["support_tickets"]:
                if ticket.get("id") == ticket_id and (actor.get("role") == "admin" or ticket.get("user_id") == actor_id): return dict(ticket)
            return None

    def reply_support_ticket(self, admin_id: str, ticket_id: str, reply: str) -> dict[str, Any]:
        reply = (reply or "").strip()
        if not reply: raise ValueError("Reply is required.")
        if len(reply) > 5000: raise ValueError("Reply may contain at most 5000 characters.")
        with self.lock:
            self._assert_admin_unlocked(admin_id)
            for ticket in self.state["support_tickets"]:
                if ticket.get("id") == ticket_id:
                    ticket.update({"reply": reply, "status": "answered", "replied_at": iso(), "read_by_user": False})
                    self._save_unlocked("Reply to CloudVaultHub support ticket")
                    return dict(ticket)
            raise ValueError("Support message not found.")

    def mark_support_read(self, user_id: str, ticket_id: str) -> dict[str, Any]:
        with self.lock:
            for ticket in self.state["support_tickets"]:
                if ticket.get("id") == ticket_id and ticket.get("user_id") == user_id:
                    ticket["read_by_user"] = True
                    self._save_unlocked("Mark CloudVaultHub support reply as read")
                    return dict(ticket)
            raise ValueError("Support message not found.")

    # ---------------- Legacy global chat (disabled) ----------------
    def _chat_eligible_unlocked(self, user: dict[str, Any] | None) -> bool:
        if not user:
            return False
        if user.get("role") == "admin":
            return True
        return PLANS.get(user.get("plan", "free"), PLANS["free"]).chat_access

    def _room_sort_key_unlocked(self, room_id: str) -> tuple[int, str]:
        try:
            return int(room_id.rsplit("-", 1)[-1]), room_id
        except (TypeError, ValueError):
            return 10**9, room_id

    def _ensure_room_unlocked(self, room_id: str) -> dict[str, Any]:
        room = self.state["chat_rooms"].get(room_id)
        if room:
            return room
        number, _ = self._room_sort_key_unlocked(room_id)
        if number >= 10**9:
            number = max([self._room_sort_key_unlocked(rid)[0] for rid in self.state["chat_rooms"]] or [0]) + 1
            room_id = f"global-{number}"
        room = {"id": room_id, "name": f"Global {number}", "paused": False, "created_at": iso()}
        self.state["chat_rooms"][room_id] = room
        return room

    def _room_member_count_unlocked(self, room_id: str, exclude_user_id: str | None = None) -> int:
        return sum(
            1 for uid, user in self.state["users"].items()
            if uid != exclude_user_id and self._chat_eligible_unlocked(user) and user.get("chat_room_id", "global-1") == room_id
        )

    def _new_room_unlocked(self) -> dict[str, Any]:
        numbers = [self._room_sort_key_unlocked(room_id)[0] for room_id in self.state["chat_rooms"]]
        number = max(numbers or [0]) + 1
        room_id = f"global-{number}"
        room = {"id": room_id, "name": f"Global {number}", "paused": False, "created_at": iso()}
        self.state["chat_rooms"][room_id] = room
        return room

    def _assign_room_unlocked(self, user_id: str, force: bool = False) -> str:
        user = self.state["users"].get(user_id)
        if not self._chat_eligible_unlocked(user):
            return ""
        cfg = self.state["config"]
        if cfg.get("chat_mode", "single") == "single":
            user["chat_room_id"] = "global-1"
            return "global-1"
        current = user.get("chat_room_id")
        if not force and current in self.state["chat_rooms"]:
            return str(current)
        capacity = int(cfg.get("chat_room_capacity", 200))
        available = [
            rid for rid, room in self.state["chat_rooms"].items()
            if not room.get("paused") and self._room_member_count_unlocked(rid, exclude_user_id=user_id) < capacity
        ]
        if not available:
            available = [self._new_room_unlocked()["id"]]
        chosen = secrets.choice(sorted(available, key=self._room_sort_key_unlocked))
        user["chat_room_id"] = chosen
        return chosen

    def chat_room_state(self, user_id: str) -> dict[str, Any]:
        with self.lock:
            user = self.state["users"].get(user_id)
            if not self._chat_eligible_unlocked(user):
                return {"mode": "single", "capacity": 0, "current": None, "can_choose": False, "rooms": []}
            user["chat_room_id"] = "global-1"
            room = self.state["chat_rooms"].setdefault(
                "global-1",
                {"id": "global-1", "name": "Global Chat", "paused": False, "created_at": iso()},
            )
            room["name"] = "Global Chat"
            return {
                "mode": "single",
                "capacity": 0,
                "current": "global-1",
                "can_choose": False,
                "rooms": [{
                    "id": "global-1",
                    "name": "Global Chat",
                    "paused": bool(room.get("paused", False)),
                    "members": self._room_member_count_unlocked("global-1"),
                    "messages": len(self.state["chat"]),
                }],
            }

    def change_chat_room(self, user_id: str, room_id: str) -> str:
        with self.lock:
            user = self.state["users"].get(user_id)
            if not self._chat_eligible_unlocked(user):
                raise PermissionError("Global Chat is available only to Plus and Pro accounts.")
            if self.state["config"].get("chat_mode", "single") != "multi":
                user["chat_room_id"] = "global-1"
                return "global-1"
            if user.get("role") != "admin" and user.get("plan") != "pro":
                raise PermissionError("Plus accounts are assigned to a room automatically.")
            room = self.state["chat_rooms"].get(room_id)
            if not room:
                raise ValueError("The selected chat room does not exist.")
            if room.get("paused") and user.get("role") != "admin":
                raise ValueError("The selected chat room is paused.")
            current = user.get("chat_room_id")
            capacity = int(self.state["config"].get("chat_room_capacity", 200))
            if current != room_id and user.get("role") != "admin" and self._room_member_count_unlocked(room_id) >= capacity:
                raise ValueError("The selected chat room is full.")
            user["chat_room_id"] = room_id
            self._save_unlocked("Change CloudVaultHub chat room")
            return room_id

    def set_chat_configuration(self, admin_id: str, mode: str, capacity: int) -> dict[str, Any]:
        with self.lock:
            self._assert_admin_unlocked(admin_id)
            self.state["config"]["chat_mode"] = "single"
            self.state["config"]["chat_room_capacity"] = 0
            room = self.state["chat_rooms"].setdefault(
                "global-1",
                {"id": "global-1", "name": "Global Chat", "paused": False, "created_at": iso()},
            )
            room["name"] = "Global Chat"
            self.state["chat_rooms"] = {"global-1": room}
            for user in self.state["users"].values():
                if self._chat_eligible_unlocked(user):
                    user["chat_room_id"] = "global-1"
            for message in self.state["chat"]:
                message["room_id"] = "global-1"
            self.state["chat"] = sorted(self.state["chat"], key=lambda m: m.get("created_at", ""))[-400:]
            self._save_unlocked("Use one unlimited Global Chat room")
            return dict(self.state["config"])

    def toggle_chat_room(self, admin_id: str, room_id: str) -> bool:
        with self.lock:
            self._assert_admin_unlocked(admin_id)
            room = self.state["chat_rooms"].get(room_id)
            if not room:
                raise ValueError("Chat room not found.")
            room["paused"] = not bool(room.get("paused"))
            self._save_unlocked("Toggle CloudVaultHub chat room")
            return bool(room["paused"])

    def clear_chat_room(self, admin_id: str, room_id: str) -> int:
        with self.lock:
            self._assert_admin_unlocked(admin_id)
            if room_id not in self.state["chat_rooms"]:
                raise ValueError("Chat room not found.")
            before = len(self.state["chat"])
            self.state["chat"] = [m for m in self.state["chat"] if m.get("room_id", "global-1") != room_id]
            removed = before - len(self.state["chat"])
            self._save_unlocked("Clear CloudVaultHub chat room")
            return removed

    def send_chat(self, user_id: str, text: str, room_id: str | None = None) -> dict[str, Any]:
        message = (text or "").strip()
        if not message:
            raise ValueError("Write a message first.")
        if len(message) > 1000:
            raise ValueError("Messages may contain at most 1000 characters.")
        with self.lock:
            user = self.state["users"].get(user_id)
            if not user:
                raise ValueError("Please log in again.")
            plan = PLANS.get(user.get("plan", "free"), PLANS["free"])
            if not self._chat_eligible_unlocked(user):
                raise PermissionError("Global Chat is available only to Plus and Pro accounts.")
            if user.get("is_banned"):
                raise PermissionError("This account is banned.")
            cfg = self.state["config"]
            if not cfg.get("system_enabled", True):
                raise PermissionError("The system is paused.")
            if not cfg.get("chat_enabled", True):
                raise PermissionError("Global Chat is paused by the developer.")
            if user.get("role") != "admin" and user.get("plan") == "plus" and not cfg.get("plus_chat_enabled", True):
                raise PermissionError("Global Chat is paused for Plus accounts.")
            if user.get("role") != "admin" and user.get("plan") == "pro" and not cfg.get("pro_chat_enabled", True):
                raise PermissionError("Global Chat is paused for Pro accounts.")
            user["chat_room_id"] = "global-1"
            assigned_room = "global-1"
            room = self.state["chat_rooms"].get(assigned_room)
            if room and room.get("paused") and user.get("role") != "admin":
                raise PermissionError("This chat room is paused by the developer.")
            now = utcnow()
            last = parse_dt(user.get("last_chat_at"))
            if last and (now - last).total_seconds() < 3:
                wait = max(1, 3 - int((now - last).total_seconds()))
                raise ValueError(f"Wait {wait} second(s) before sending another message.")
            record = {
                "id": uuid.uuid4().hex,
                "room_id": assigned_room,
                "sender_id": user_id,
                "sender_name": "admin" if user.get("role") == "admin" else user.get("username", "User"),
                "sender_plan": "admin" if user.get("role") == "admin" else user.get("plan", "free"),
                "text": message,
                "created_at": iso(now),
                "visible_at": iso(now + timedelta(seconds=2)),
            }
            user["last_chat_at"] = iso(now)
            self.state["chat"].append(record)
            self.state["chat"] = sorted(self.state["chat"], key=lambda m: m.get("created_at", ""))[-400:]
            self._save_unlocked("Send CloudVaultHub Global Chat message")
            return dict(record)

    def chat_messages(self, user_id: str, room_id: str | None = None) -> list[dict[str, Any]]:
        with self.lock:
            user = self.state["users"].get(user_id)
            if not self._chat_eligible_unlocked(user):
                return []
            assigned = self._assign_room_unlocked(user_id)
            now = utcnow()
            limit = 400 if user.get("role") == "admin" or user.get("plan") == "pro" else 200
            return [
                dict(m) for m in self.state["chat"]
                if m.get("room_id", "global-1") == assigned and (parse_dt(m.get("visible_at")) or now) <= now
            ][-limit:]

    def clear_chat(self, admin_id: str) -> None:
        with self.lock:
            self._assert_admin_unlocked(admin_id)
            self.state["chat"] = []
            self._save_unlocked("Clear CloudVaultHub Global Chat")


STORE = StateStore(BACKEND, SETTINGS)


def derive_file_key(user_id: str, file_id: str, algorithm: str) -> bytes:
    key_material = hashlib.sha256(SETTINGS.master_key.encode("utf-8")).digest()
    return hmac.new(key_material, f"{user_id}:{file_id}:{algorithm}".encode("utf-8"), hashlib.sha256).digest()


def encrypt_file(source: Path, destination: Path, user_id: str, file_id: str, algorithm: str) -> int:
    if algorithm == "none":
        shutil.copy2(source, destination)
        return destination.stat().st_size
    algorithm_id = 1 if algorithm == "aes256-gcm" else 2
    key = derive_file_key(user_id, file_id, algorithm)
    cipher = AESGCM(key) if algorithm_id == 1 else ChaCha20Poly1305(key)
    with source.open("rb") as fin, destination.open("wb") as fout:
        fout.write(MAGIC + bytes([algorithm_id]) + struct.pack(">I", CHUNK_SIZE))
        index = 0
        while True:
            chunk = fin.read(CHUNK_SIZE)
            if not chunk:
                break
            nonce = os.urandom(12)
            encrypted = cipher.encrypt(nonce, chunk, struct.pack(">Q", index))
            fout.write(nonce)
            fout.write(struct.pack(">I", len(encrypted)))
            fout.write(encrypted)
            index += 1
    return destination.stat().st_size


def decrypt_file(source: Path, destination: Path, user_id: str, file_id: str, algorithm: str) -> None:
    if algorithm == "none":
        shutil.copy2(source, destination)
        return
    with source.open("rb") as fin, destination.open("wb") as fout:
        magic = fin.read(len(MAGIC))
        if magic not in {MAGIC, b"CVLT2"}:
            raise ValueError("The encrypted file header is invalid.")
        algorithm_id_raw = fin.read(1)
        if not algorithm_id_raw:
            raise ValueError("The encrypted file is incomplete.")
        algorithm_id = algorithm_id_raw[0]
        chunk_raw = fin.read(4)
        if len(chunk_raw) != 4:
            raise ValueError("The encrypted file is incomplete.")
        _chunk_size = struct.unpack(">I", chunk_raw)[0]
        expected = 1 if algorithm == "aes256-gcm" else 2
        if algorithm_id != expected:
            raise ValueError("The encryption mode does not match the file metadata.")
        key = derive_file_key(user_id, file_id, algorithm)
        cipher = AESGCM(key) if algorithm_id == 1 else ChaCha20Poly1305(key)
        index = 0
        while True:
            nonce = fin.read(12)
            if not nonce:
                break
            if len(nonce) != 12:
                raise ValueError("The encrypted file is incomplete.")
            length_raw = fin.read(4)
            if len(length_raw) != 4:
                raise ValueError("The encrypted file is incomplete.")
            length = struct.unpack(">I", length_raw)[0]
            encrypted = fin.read(length)
            if len(encrypted) != length:
                raise ValueError("The encrypted file is incomplete.")
            fout.write(cipher.decrypt(nonce, encrypted, struct.pack(">Q", index)))
            index += 1


def zip_directory_files(paths: list[str], destination: Path) -> tuple[int, str]:
    valid = [Path(p) for p in paths if p and Path(p).is_file()]
    if not valid:
        raise ValueError("Choose a folder first.")
    total = 0
    with zipfile.ZipFile(destination, "w", compression=zipfile.ZIP_DEFLATED, allowZip64=True) as archive:
        used_names: set[str] = set()
        for index, path in enumerate(valid):
            name = safe_filename(path.name)
            if name in used_names:
                name = f"{index + 1}-{name}"
            used_names.add(name)
            archive.write(path, arcname=name)
            total += path.stat().st_size
    return total, "uploaded-folder.zip"


def estimate_upload_seconds(size_bytes: int, plan_key: str) -> int:
    # Approximate end-to-end speed. Real time depends on network and Hugging Face load.
    mb_per_second = {"free": 2.0, "plus": 5.0, "pro": 8.0, "admin": 12.0}.get(plan_key, 2.0)
    processing_overhead = {"free": 3, "plus": 5, "pro": 6, "admin": 3}.get(plan_key, 4)
    return max(2, int(size_bytes / MB / mb_per_second) + processing_overhead)


def format_eta(seconds: int) -> str:
    if seconds < 60:
        return f"about {seconds} seconds"
    minutes = max(1, round(seconds / 60))
    return f"about {minutes} minute{'s' if minutes != 1 else ''}"


def prepare_upload(source_path: str, folder_paths: list[str] | None, user_id: str, progress=None) -> dict[str, Any]:
    user = STORE.get_user(user_id)
    if not user:
        raise ValueError("Please log in again.")
    cfg = STORE.config()
    if user.get("role") != "admin" and not cfg.get("system_enabled", True):
        raise ValueError("The system is temporarily paused.")
    is_admin = str(user.get("role", "")).lower() == "admin"
    plan_key = str(user.get("plan", "free")).lower()
    plan = PLANS["admin"] if is_admin else PLANS.get(plan_key, PLANS["free"])
    file_id = uuid.uuid4().hex
    raw_temp: Path | None = None
    encrypted_temp: Path | None = None
    try:
        if folder_paths:
            if not plan.folder_upload:
                raise ValueError("Folder upload is not available on the Free package.")
            raw_temp = SETTINGS.temp_dir / f"folder-{file_id}.zip"
            original_size, display_name = zip_directory_files(folder_paths, raw_temp)
            content_type = "application/zip"
            is_folder = True
        else:
            if not source_path:
                raise ValueError("Choose a file or folder first.")
            source = Path(source_path)
            if not source.is_file():
                raise ValueError("The selected file could not be read.")
            raw_temp = SETTINGS.temp_dir / f"raw-{file_id}-{safe_filename(source.name)}"
            shutil.copy2(source, raw_temp)
            original_size = raw_temp.stat().st_size
            display_name = safe_filename(source.name)
            content_type = mimetypes.guess_type(display_name)[0] or "application/octet-stream"
            is_folder = False

        if original_size <= 0:
            raise ValueError("Empty files cannot be uploaded.")
        if is_admin:
            remaining = STORE.available_for_admin()
            if original_size > remaining:
                raise ValueError(
                    f"Upload failed: developer storage is full. Available {human_bytes(remaining)}, file size {human_bytes(original_size)}."
                )
        else:
            remaining = max(0, plan.quota_bytes - STORE.used_bytes(user_id))
            if original_size > remaining:
                raise ValueError(
                    f"Upload failed: your {plan.name} storage is full. Available {human_bytes(remaining)}, file size {human_bytes(original_size)}."
                )

        estimated_seconds = estimate_upload_seconds(original_size, plan.key)
        estimated_text = format_eta(estimated_seconds)
        if progress:
            progress(0.12, desc=f"Estimated completion: {estimated_text}")
            progress(0.25, desc=f"Preparing file · {estimated_text}")
        upload_path = raw_temp
        stored_size = original_size
        if plan.encryption != "none":
            encrypted_temp = SETTINGS.temp_dir / f"encrypted-{file_id}.bin"
            stored_size = encrypt_file(raw_temp, encrypted_temp, user_id, file_id, plan.encryption)
            upload_path = encrypted_temp
            if progress:
                progress(0.60, desc="Encrypting file")

        remote_path = f"objects/{user_id}/{file_id}.blob"
        if progress:
            progress(0.78, desc="Saving to persistent storage")
        BACKEND.upload(upload_path, remote_path, "Upload CloudVaultHub user file")
        metadata = {
            "id": file_id,
            "owner_id": user_id,
            "display_name": display_name,
            "remote_path": remote_path,
            "original_size": original_size,
            "stored_size": stored_size,
            "is_folder": is_folder,
            "content_type": content_type,
            "encryption": plan.encryption,
            "share_password_hash": None,
            "download_count": 0,
            "last_download_at": None,
            "created_at": iso(),
            "uploaded_under_plan": plan.key,
            "estimated_upload_seconds": estimated_seconds,
        }
        try:
            saved = STORE.add_file(user_id, metadata)
        except Exception:
            BACKEND.delete_many([remote_path], "Rollback incomplete CloudVaultHub upload")
            raise
        if progress:
            progress(1.0, desc="Upload completed")
        return saved
    finally:
        if raw_temp:
            raw_temp.unlink(missing_ok=True)
        if encrypted_temp:
            encrypted_temp.unlink(missing_ok=True)


def prepare_generated_folder_archive(
    archive_path: str | Path,
    display_name: str,
    user_id: str,
    progress=None,
) -> dict[str, Any]:
    user = STORE.get_user(user_id)
    if not user:
        raise ValueError("Please log in again.")
    cfg = STORE.config()
    if user.get("role") != "admin" and not cfg.get("system_enabled", True):
        raise ValueError("The system is temporarily paused.")
    if user.get("role") != "admin" and user.get("plan") != "pro":
        raise PermissionError("Bu özellik yalnızca Pro hesaplarda kullanılabilir.")

    source = Path(archive_path)
    if not source.is_file() or source.suffix.lower() != ".zip":
        raise ValueError("Hazırlanan web sitesi klasörü bulunamadı.")
    original_size = source.stat().st_size
    if original_size <= 0:
        raise ValueError("Boş klasör buluta eklenemez.")

    is_admin = str(user.get("role", "")).lower() == "admin"
    plan_key = str(user.get("plan", "free")).lower()
    plan = PLANS["admin"] if is_admin else PLANS.get(plan_key, PLANS["free"])
    if user.get("role") == "admin":
        if original_size > STORE.available_for_admin():
            raise ValueError("The remaining unreserved server storage is not enough for this upload.")
    elif STORE.used_bytes(user_id) + original_size > plan.quota_bytes:
        raise ValueError("There is not enough cloud storage space for this folder.")

    file_id = uuid.uuid4().hex
    encrypted_temp: Path | None = None
    try:
        upload_path = source
        stored_size = original_size
        if progress:
            progress(0.20, desc="Web sitesi klasörü hazırlanıyor")
        if plan.encryption != "none":
            encrypted_temp = SETTINGS.temp_dir / f"generated-folder-{file_id}.bin"
            stored_size = encrypt_file(source, encrypted_temp, user_id, file_id, plan.encryption)
            upload_path = encrypted_temp
            if progress:
                progress(0.60, desc="Klasör şifreleniyor")

        remote_path = f"objects/{user_id}/{file_id}.blob"
        BACKEND.upload(upload_path, remote_path, "Upload generated website folder")
        metadata = {
            "id": file_id,
            "owner_id": user_id,
            "display_name": safe_filename(display_name),
            "remote_path": remote_path,
            "original_size": original_size,
            "stored_size": stored_size,
            "is_folder": True,
            "content_type": "application/zip",
            "encryption": plan.encryption,
            "share_password_hash": None,
            "download_count": 0,
            "last_download_at": None,
            "created_at": iso(),
            "uploaded_under_plan": plan.key,
            "estimated_upload_seconds": estimate_upload_seconds(original_size, plan.key),
        }
        saved = STORE.add_file(user_id, metadata)
        if progress:
            progress(1.0, desc="Klasör bulut depolamaya eklendi")
        return saved
    finally:
        if encrypted_temp:
            encrypted_temp.unlink(missing_ok=True)


def materialize_download(item: dict[str, Any]) -> Path:
    file_id = item["id"]
    remote_temp = SETTINGS.temp_dir / f"download-remote-{file_id}-{uuid.uuid4().hex}.bin"
    output = SETTINGS.temp_dir / f"download-{uuid.uuid4().hex}-{safe_filename(item.get('display_name', 'file.bin'))}"
    if not BACKEND.download(item["remote_path"], remote_temp):
        raise FileNotFoundError("The stored object could not be downloaded.")
    try:
        decrypt_file(remote_temp, output, item["owner_id"], file_id, item.get("encryption", "none"))
    finally:
        remote_temp.unlink(missing_ok=True)
    return output


def guess_device(user_agent: str) -> str:
    value = (user_agent or "").lower()
    if "ipad" in value or "tablet" in value:
        return "Tablet"
    if "android" in value or "iphone" in value or "mobile" in value:
        return "Mobile"
    return "Desktop"