"""
This program is free software: you can redistribute it and/or modify it under
the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License,
or (at your option) any later version.


This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. 
See the GNU General Public License for more details.


You should have received a copy of the GNU General Public License
 along with this program.  If not, see <https://www.gnu.org/licenses/>.

Copyright © 2019 Cloud Linux Software Inc.

This software is also available under ImunifyAV commercial license,
see <https://www.imunify360.com/legal/eula>

Tenant-aware path for multitenant cluster mode.

In standalone mode, TenantPath delegates to pathlib.Path.
In cluster mode (user set + k8s env), I/O operations route through
the file gateway so they reach the correct tenant's filesystem.

Pure path operations (name, parent, suffix, etc.) always work locally
since they only manipulate the path string.
"""

from __future__ import annotations

import asyncio
import atexit
import concurrent.futures
import glob as _glob
import logging
import os
from collections import namedtuple
from pathlib import Path, PurePath
from typing import Iterator, Union

from defence360agent.utils import is_cluster
from defence360agent.utils.tenant_path import (  # noqa: F401  re-export
    GLOBAL_TENANT,
    is_under_prefix,
    split_marked,
    split_prefixed,
    strip_prefix,
    strip_tenant,
    to_prefixed,
    under_prefix,
)

logger = logging.getLogger(__name__)

# Cluster-mode stat result. The gateway reports only size/mtime/mode, so
# st_ctime is aliased to st_mtime — a cluster caller reading st_ctime still
# gets a "changed since" signal. st_atime is unavailable and omitted.
_RemoteStat = namedtuple(
    "_RemoteStat", ("st_size", "st_mtime", "st_ctime", "st_mode")
)


_executor = concurrent.futures.ThreadPoolExecutor(1)
# Join the worker on interpreter exit (matches the project convention in
# safe_fileops.shutdown_process_pools / message_status_publisher).
atexit.register(_executor.shutdown)


def candidate_forms(path) -> "tuple[Path, ...]":
    """The forms an ignore-list entry may legitimately match.

    In cluster mode an entry is scoped by its own prefix: ``/app-1/...``
    covers one tenant, ``/*/...`` covers all of them. So a path is checked
    both as stored and with its tenant swapped for the wildcard. ``*`` is
    not a registered application, so the two can never collide.
    """
    if isinstance(path, TenantPath):
        path = path.to_prefixed()
    as_stored = Path(os.fspath(path))
    if not is_cluster():
        return (as_stored,)
    user, bare = split_prefixed(as_stored)
    if user is None or user == GLOBAL_TENANT:
        return (as_stored,)
    return (as_stored, Path(f"/{GLOBAL_TENANT}{bare}"))


def _run_async_sync(coro):
    """Run an async coroutine synchronously, even from inside a running loop."""
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(coro)
    # Already inside an event loop — offload to a worker thread
    return _executor.submit(asyncio.run, coro).result()


class TenantPath(os.PathLike):
    """Path that carries tenant (user) context.

    Identity is (user, path) — two TenantPaths with different users
    but the same filesystem path are distinct objects.
    """

    __slots__ = ("_path", "_user")

    def __init__(
        self, path: Union[str, Path, "TenantPath"], *, user: str | None = None
    ):
        if isinstance(path, TenantPath):
            self._path = path._path
            self._user = user if user is not None else path._user
        else:
            self._path = Path(path) if not isinstance(path, Path) else path
            self._user = user

    # -- identity --------------------------------------------------------

    @property
    def user(self) -> str | None:
        return self._user

    @property
    def path(self) -> Path:
        return self._path

    @classmethod
    def from_prefixed(
        cls, s: Union[str, os.PathLike], *, user: "str | None" = None
    ) -> "TenantPath":
        """Parse a canonical prefixed string back into a TenantPath.

        When ``user`` is given, that tenant wins and a matching redundant
        prefix is stripped; otherwise the first path component is taken as
        the tenant (syntactic — validate at input boundaries).
        """
        if user is not None:
            return cls(strip_prefix(s, user), user=user)
        parsed_user, bare = split_prefixed(s)
        return cls(bare, user=parsed_user)

    def to_prefixed(self) -> str:
        """The canonical serialized form: ``/<user><path>`` on k8s with a
        tenant, ``str(path)`` otherwise."""
        return to_prefixed(self._path, self._user)

    @property
    def display_path(self) -> str:
        """The bare in-container path — the only form humans ever see."""
        return str(self._path)

    @property
    def is_remote(self) -> bool:
        # An empty user is not a tenant — `bool("")` is False — so such a
        # path stays local instead of routing to the gateway with an empty
        # application_id. Some rows carry user="" (CharField default).
        return bool(self._user) and is_cluster()

    def __fspath__(self) -> str:
        return os.fspath(self._path)

    def __str__(self) -> str:
        return str(self._path)

    def __repr__(self) -> str:
        if self._user is not None:
            return f"TenantPath({str(self._path)!r}, user={self._user!r})"
        return f"TenantPath({str(self._path)!r})"

    def __eq__(self, other) -> bool:
        # Two TenantPaths are equal only if both user and path match.
        if isinstance(other, TenantPath):
            return (self._user, self._path) == (other._user, other._path)
        # Backward-compat: a TenantPath compares equal to a bare Path with
        # the same path (user-blind). Lots of code receives orig_file_path
        # (now a TenantPath) and compares it to a plain Path, e.g.
        # is_crontab() does ``path.parent == crontab_path()``. Dropping this
        # silently disabled those checks.
        if isinstance(other, PurePath):
            return self._path == other
        return NotImplemented

    def __hash__(self) -> int:
        # Hash on path only, NOT (user, path): otherwise the eq/hash invariant
        # breaks for TenantPath == Path (equal but different hash). Two
        # different-tenant paths may share a hash bucket; __eq__ still keeps
        # them distinct, so sets of TenantPaths stay correct.
        return hash(self._path)

    # -- pure path operations (no I/O) -----------------------------------

    @property
    def name(self) -> str:
        return self._path.name

    @property
    def parent(self) -> TenantPath:
        return TenantPath(self._path.parent, user=self._user)

    @property
    def parents(self):
        # Carry the tenant like .parent does — raw Path.parents would yield
        # user-less Paths that route remote I/O to the wrong filesystem.
        return tuple(
            TenantPath(p, user=self._user) for p in self._path.parents
        )

    # -- I/O operations --------------------------------------------------
    # Standalone: delegate to Path.
    # Cluster/k8s: route through file gateway (CleanupStorageK8s).

    def resolve(self) -> TenantPath:
        """Resolve symlinks (a filesystem syscall in standalone mode).
        In cluster mode, return as-is — there is no local fs to resolve."""
        if self.is_remote:
            return self
        return TenantPath(self._path.resolve(), user=self._user)

    def _assert_locally_resolvable(self) -> None:
        """Guard the local I/O branch. In cluster mode a path without a
        usable tenant cannot be resolved — it is not on the agent container
        and there is no app to ask — and stat'ing the wrong filesystem would
        misreport a tenant's file as missing (and get the hit deleted). This
        is an invalid object, so fail loud instead of returning a wrong value.
        """
        if is_cluster():
            raise ValueError(
                "TenantPath I/O in cluster mode requires a tenant; got "
                f"user={self._user!r}, path={str(self._path)!r}"
            )

    def exists(self) -> bool:
        if self.is_remote:
            return _run_async_sync(self._remote_exists())
        self._assert_locally_resolvable()
        return self._path.exists()

    async def exists_async(self) -> bool:
        """Non-blocking exists for use in async contexts."""
        if self.is_remote:
            return await self._remote_exists()
        self._assert_locally_resolvable()
        return self._path.exists()

    def stat(self):
        if self.is_remote:
            return _run_async_sync(self._remote_stat())
        self._assert_locally_resolvable()
        return self._path.stat()

    async def stat_async(self):
        """Async stat. Standalone returns a real os.stat_result; cluster mode
        returns a _RemoteStat (size/mtime/mode only). Unlike exists(), it does
        not swallow OSError — the caller must tell 'gone' from 'unverifiable'.
        """
        if self.is_remote:
            return await self._remote_stat()
        self._assert_locally_resolvable()
        return self._path.stat()

    def is_file(self) -> bool:
        if not self.is_remote:
            return self._path.is_file()
        return _run_async_sync(self._remote_stat_flag("is_file"))

    def read_bytes(self) -> bytes:
        if self.is_remote:
            return _run_async_sync(self._remote_read_bytes())
        self._assert_locally_resolvable()
        return self._path.read_bytes()

    # -- directory listing ------------------------------------------------

    @staticmethod
    def iglob(
        pattern: str, *, user: str | None = None
    ) -> Iterator[TenantPath]:
        """Expand a glob pattern. In cluster mode the agent can't glob the
        tenant filesystem locally, so expansion runs over the syncer
        (stat / list-dir), mirroring glob.iglob's non-recursive semantics."""
        # `user` (not `user is not None`) mirrors is_remote: an empty user is
        # not a tenant, so it globs locally instead of hitting the syncer with
        # an empty application_id.
        if user and is_cluster():
            for p in _run_async_sync(TenantPath._remote_glob(pattern, user)):
                yield TenantPath(p, user=user)
        else:
            for p in _glob.iglob(pattern):
                yield TenantPath(p, user=user)

    @staticmethod
    async def _remote_glob(pattern: str, user: str) -> list:
        import fnmatch
        import posixpath

        from imav.malwarelib.cleanup.storage import CleanupStorageK8s

        if not _glob.has_magic(pattern):
            # Concrete path: include it iff it exists, like glob.iglob (which
            # silently drops a non-existent literal path).
            try:
                await CleanupStorageK8s.stat_file(app_id=user, path=pattern)
            except FileNotFoundError:
                return []
            except OSError:
                # Gateway error: can't verify — keep the path rather than
                # silently drop one that may exist.
                pass
            return [pattern]

        dirname, basename = posixpath.split(pattern)
        if _glob.has_magic(dirname):
            parents = await TenantPath._remote_glob(dirname, user)
        else:
            parents = [dirname]

        matches: list = []
        for parent in parents:
            try:
                entries = await CleanupStorageK8s.list_dir(
                    app_id=user, path=parent
                )
            except (FileNotFoundError, OSError):
                # Missing / not a directory / unreadable — nothing to match
                # here, exactly like globbing a path the shell can't descend.
                continue
            for entry in entries:
                name = entry["name"]
                # Like the shell, a leading-dot name only matches a
                # leading-dot pattern.
                if name.startswith(".") and not basename.startswith("."):
                    continue
                if fnmatch.fnmatch(name, basename):
                    matches.append(posixpath.join(parent, name))
        return matches

    # -- remote I/O helpers (async, called via _run_async_sync) ----------

    async def _remote_exists(self) -> bool:
        # Lazy import to avoid circular: model → tenant_path → storage → model
        from imav.malwarelib.cleanup.storage import CleanupStorageK8s

        try:
            await CleanupStorageK8s.stat_file(
                app_id=self._user,
                path=str(self._path),
            )
            return True
        except FileNotFoundError:
            return False
        except OSError:
            # Gateway timeout, syncer down, etc. — assume file exists
            # so callers don't delete hits they can't verify.
            return True

    async def _remote_stat(self) -> _RemoteStat:
        from imav.malwarelib.cleanup.storage import CleanupStorageK8s

        # No try/except: FileNotFoundError ("gone") and OSError ("couldn't
        # check") must propagate so the caller can distinguish them.
        st = await CleanupStorageK8s.stat_file(
            app_id=self._user, path=str(self._path)
        )
        return _RemoteStat(
            st_size=st["size"],
            st_mtime=st["mtime"],
            st_ctime=st["mtime"],
            st_mode=st["mode"],
        )

    async def _remote_stat_flag(self, key: str) -> bool:
        from imav.malwarelib.cleanup.storage import CleanupStorageK8s

        try:
            st = await CleanupStorageK8s.stat_file(
                app_id=self._user, path=str(self._path)
            )
        except FileNotFoundError:
            return False
        except OSError:
            # Gateway error: don't claim a type, so the outdated-entry sweep
            # falls through to the directory branch instead of pruning.
            return False
        return bool(st[key])

    _READ_BYTES_CAP = 100 * 1024 * 1024  # 100 MiB

    async def _remote_read_bytes(self) -> bytes:
        import base64

        from imav.malwarelib.cleanup.storage import CleanupStorageK8s

        result = await CleanupStorageK8s.read_file(
            app_id=self._user,
            path=str(self._path),
            offset=0,
            limit=self._READ_BYTES_CAP,
        )
        # Fail loud rather than return a truncated file: a partial read would
        # corrupt any hashing/comparison the caller does on the bytes.
        if not result.get("eof", True):
            raise OSError(
                f"file exceeds {self._READ_BYTES_CAP} byte read cap "
                f"(path={self._path!r}, user={self._user!r})"
            )
        return base64.b64decode(result["data"])


# Cap concurrent existence checks so a large cleanup batch doesn't fan out
# hundreds of simultaneous file-gateway RPCs at the syncer in k8s.
EXISTS_GATHER_LIMIT = 50


async def safe_exists(path: "TenantPath") -> bool:
    """Existence check that never lets one bad hit abort a cleanup handler.

    ``exists_async`` raises for an invalid (tenantless) path in cluster mode;
    here that is logged and reported as existing (True) — never delete what
    we couldn't verify. Cancellation is re-raised, not swallowed, so agent
    deactivation still propagates.
    """
    try:
        return await path.exists_async()
    except asyncio.CancelledError:
        raise
    except Exception as exc:
        logger.warning(
            "existence check failed for %r (%s: %s); assuming it exists so "
            "cleanup does not drop an unverifiable hit",
            path,
            type(exc).__name__,
            exc,
        )
        return True


async def gather_exists(paths: "list[TenantPath]") -> list[bool]:
    """Check existence of many paths concurrently, bounded concurrency.

    Standalone paths resolve inline (a local stat, no await); remote paths
    issue gateway RPCs throttled by a semaphore. Per-path failures are
    isolated via :func:`safe_exists` so one bad hit can't abort the batch.
    """
    sem = asyncio.Semaphore(EXISTS_GATHER_LIMIT)

    async def _check(p: TenantPath) -> bool:
        async with sem:
            return await safe_exists(p)

    return await asyncio.gather(*(_check(p) for p in paths))