# ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========

import dataclasses
import posixpath

from mirage.commands.spec import SPECS, parse_command
from mirage.runtime.types import DispatchFn
from mirage.types import FileStat, FileType, PathSpec
from mirage.utils.path import CycleError
from mirage.workspace.executor.builtins.links.probe import stat_or_none
from mirage.workspace.executor.builtins.shared import (Result, fail, ok,
                                                       split_flags)
from mirage.workspace.mount.namespace import Namespace


def link_flags(args: list[str | PathSpec], known: str) -> set[str]:
    flags, _ = split_flags(args, known)
    return flags


def follow_parent(namespace: Namespace, virtual: str) -> str:
    """Resolve every component of a path but the last one.

    POSIX resolves a path one component at a time, and only the last one
    is exempt for an lstat-style command: ``stat dlink/f2`` reports
    ``f2`` because ``dlink`` was resolved on the way to it, while
    ``stat dlink`` reports the link. A no-follow command therefore still
    needs its operand's directory prefix resolved.

    Args:
        namespace (Namespace): addressing authority holding the links.
        virtual (str): absolute virtual path.

    Raises:
        CycleError: when a prefix loops past the hop limit (ELOOP).
    """
    parent, _, name = virtual.rstrip("/").rpartition("/")
    if not name:
        return virtual
    resolved = namespace.follow(parent or "/")
    return resolved.rstrip("/") + "/" + name


def follow_paths(
    namespace: Namespace,
    items: list[str | PathSpec],
    follow_last: bool = True,
    slash_follows: bool = True,
) -> list[str | PathSpec]:
    """Rewrite path operands through the symlink table (open(2) semantics).

    The directory prefix always resolves; ``follow_last`` decides the
    final component, which is the whole difference between open(2) and
    lstat(2). A trailing slash overrides it per operand, because POSIX
    reads ``dlink/`` as ``dlink/.`` and there is no ``.`` to reach
    without resolving the link first (GNU: ``stat dlink`` is a symbolic
    link, ``stat dlink/`` is a directory).

    Non-path items and paths that resolve to themselves pass through
    untouched. A rewritten spec keeps the user-typed form in ``raw_path``
    so error messages still name the operand as typed; the mount re-stamps
    ``resource_path`` at dispatch.

    Args:
        namespace (Namespace): addressing authority holding the link table.
        items (list[str | PathSpec]): classified command parts.
        follow_last (bool): whether the command resolves the final
            component of its own accord (open(2) rather than lstat(2)).
        slash_follows (bool): whether a trailing slash may override
            ``follow_last``; False only for ``tar``, which strips the
            slash before it stats.

    Raises:
        CycleError: when a path loops past the hop limit (ELOOP).
    """
    out: list[str | PathSpec] = []
    for item in items:
        if not isinstance(item, PathSpec):
            out.append(item)
            continue
        last = follow_last or (slash_follows and item.raw_path.endswith("/"))
        try:
            virtual = (namespace.follow(item.virtual)
                       if last else follow_parent(namespace, item.virtual))
        except CycleError:
            raise CycleError(item.raw_path) from None
        if virtual == item.virtual:
            out.append(item)
            continue
        out.append(
            dataclasses.replace(item,
                                virtual=virtual,
                                directory=virtual[:virtual.rfind("/") + 1]
                                or "/",
                                resource_path=""))
    return out


def accepts_line(name: str, args: tuple[str, ...], items: list[str | PathSpec],
                 cwd: str) -> bool:
    """Whether the command layer will act on this line as written.

    A link entry lives in the namespace, so ``strip_link_operands``
    removes it before the command runs, and the command layer can
    neither see that nor undo it. GNU validates the whole line first and
    removes nothing when it refuses: ``rm --bogus dlink`` reports the
    option and ``unlink dlink other`` reports the extra operand, both
    with every link still in place. So the strip runs only for a line
    that layer accepts, and a refused one falls through to it unchanged
    to be reported there. Option errors are the parser's, which reports
    rather than raises them; unlink's one-operand grammar is its
    builder's, and reporting it needs the operands to arrive intact.

    Args:
        name (str): command name.
        args (tuple[str, ...]): the line's words after the name.
        items (list[str | PathSpec]): classified command parts.
        cwd (str): session working directory, which the parser resolves
            path operands against.
    """
    spec = SPECS.get(name)
    if spec is None:
        return True
    parsed = parse_command(spec, list(args), cwd)
    if parsed.invalid_options or parsed.ambiguous_options:
        return False
    if name == "unlink":
        return sum(1 for i in items if isinstance(i, PathSpec)) <= 1
    return True


async def strip_link_operands(
    namespace: Namespace,
    items: list[str | PathSpec],
) -> tuple[list[str | PathSpec], int]:
    """Unlink and drop ``rm``/``unlink`` operands that are symlinks.

    GNU ``rm`` removes the link itself and never follows it; a dangling
    link removes fine. Remaining operands stay for backend dispatch.

    An operand typed with a trailing slash is deliberately kept: the
    slash asked for a directory, and GNU refuses rather than removing
    the link (``rm dlink/`` is "Is a directory", ``unlink dlink/`` is
    "Not a directory"). Removing it here would delete exactly what the
    slash was protecting, so the command reports it instead.

    Args:
        namespace (Namespace): addressing authority holding the link table.
        items (list[str | PathSpec]): classified command parts.

    Returns:
        tuple[list[str | PathSpec], int]: surviving parts and the number
        of link entries removed.
    """
    removed = 0
    kept: list[str | PathSpec] = []
    for item in items:
        if (isinstance(item, PathSpec) and not item.raw_path.endswith("/")
                and namespace.is_link(item.virtual)):
            await namespace.unlink(item.virtual)
            removed += 1
            continue
        kept.append(item)
    return kept, removed


async def _slashed_link_refusal(
    namespace: Namespace,
    dispatch: DispatchFn,
    src: PathSpec,
    dst: PathSpec,
    dst_stat: FileStat | None,
) -> Result:
    """GNU's refusal for an ``mv`` source that is a link typed with a slash.

    rename(2) never follows, so the slash is not resolved away: POSIX
    reads ``dlink/`` as ``dlink/.``, which asks for a directory the call
    will not resolve, and GNU refuses with everything left in place --
    where a bare ``dlink`` renames the link entry. Which of the four
    wordings applies follows mv's own order, source stat before
    destination type before the rename itself, and all four are pinned
    against GNU coreutils 9.7.

    Args:
        namespace (Namespace): addressing authority holding the link table.
        dispatch (DispatchFn): op dispatcher used to stat the target.
        src (PathSpec): the slashed link source.
        dst (PathSpec): destination as typed.
        dst_stat (FileStat | None): the destination's stat, None when it
            does not exist.
    """
    followed = namespace.follow(src.virtual)
    target = await stat_or_none(dispatch, PathSpec.from_str_path(followed))
    if target is None:
        return fail(
            "mv", f"mv: cannot stat '{src.raw_path}': "
            "No such file or directory\n")
    if target.type != FileType.DIRECTORY:
        return fail("mv",
                    f"mv: cannot stat '{src.raw_path}': Not a directory\n")
    if dst_stat is not None and dst_stat.type != FileType.DIRECTORY:
        return fail(
            "mv", f"mv: cannot overwrite non-directory '{dst.raw_path}' "
            f"with directory '{src.raw_path}'\n")
    landing = dst.raw_path
    if dst_stat is not None:
        landing = landing.rstrip("/") + "/" + posixpath.basename(src.virtual)
    return fail(
        "mv", f"mv: cannot move '{src.raw_path}' to '{landing}': "
        "Not a directory\n")


async def prepare_mv(
    namespace: Namespace,
    dispatch: DispatchFn,
    items: list[str | PathSpec],
) -> tuple[list[str | PathSpec], str | None, tuple[str, str] | None, Result
           | None]:
    """Adjust a two-operand ``mv`` for node-meta operands.

    A link source renames the link entry itself. A destination that is
    (a link to) a directory receives the move inside it (rename(2)
    preceded by mv's dst stat); any other destination is replaced, so its
    node entry, link or overlay attrs alike, drops once the backend move
    succeeds. A plain source that carries overlay attributes has its meta
    travel with the file once the backend move succeeds.

    Args:
        namespace (Namespace): addressing authority holding the node table.
        dispatch (DispatchFn): op dispatcher used to stat the destination.
        items (list[str | PathSpec]): classified command parts.

    Returns:
        tuple: (possibly rewritten parts, node entry to drop after a
        successful backend move (the replaced destination), (src, dst)
        meta rename to apply after a successful backend move, early
        result when the mv completed as a pure namespace rename).
    """
    paths = [p for p in items if isinstance(p, PathSpec)]
    if len(paths) != 2:
        return items, None, None, None
    src, dst = paths

    # Where the move lands: inside a directory destination (followed, so
    # node-meta keys line up with the followed paths stat merges on), else
    # the destination itself, replaced like rename(2).
    followed = namespace.follow(dst.virtual)
    stat = await stat_or_none(dispatch, PathSpec.from_str_path(followed))
    into_dir = stat is not None and stat.type == FileType.DIRECTORY
    if into_dir:
        target_dst = (followed.rstrip("/") + "/" +
                      posixpath.basename(src.virtual))
    else:
        target_dst = dst.virtual

    if namespace.is_link(src.virtual):
        if src.raw_path.endswith("/"):
            return items, None, None, await _slashed_link_refusal(
                namespace, dispatch, src, dst, stat)
        # The move is a node-table rename, which the door answers: a
        # link has no backend entry for the generic mv to move. Reaching
        # the table directly from here would skip the admission gates
        # every other mv passes, so the dispatch is the point.
        try:
            await dispatch("rename",
                           src,
                           dst=PathSpec.from_str_path(target_dst))
        except PermissionError:
            return items, None, None, fail(
                "mv", f"mv: cannot move '{src.raw_path}' to "
                f"'{dst.raw_path}': Permission denied\n")
        return items, None, None, ok("mv")

    post_rename: tuple[str, str] | None = None
    if namespace.meta_for(src.virtual) is not None:
        post_rename = (src.virtual, target_dst)

    rewritten = items
    if into_dir and namespace.is_link(dst.virtual):
        rewritten = follow_paths(namespace, items)
    return rewritten, target_dst, post_rename, None
