# ========= 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 functools
from collections.abc import Callable
from typing import Any

from mirage.accessor.base import Accessor
from mirage.cache.index import IndexCacheStore
from mirage.commands.builtin.generic.cp import walk
from mirage.commands.builtin.generic_bind.adapter import CommandIO, Operation
from mirage.commands.builtin.utils.output import format_optional_records
from mirage.commands.builtin.utils.slash_links import (is_slashed_link,
                                                       rm_link_refusal)
from mirage.commands.builtin.utils.verbose import removal_lines
from mirage.commands.config import CommandOpts
from mirage.commands.registry import command
from mirage.commands.spec import SPECS
from mirage.commands.spec.types import FlagView
from mirage.io.types import ByteSource, IOResult
from mirage.types import FileType, PathSpec


def make_rm(resource: str, io: CommandIO) -> Callable[..., Any]:
    """Build the no-real-directories rm override for one keyed store.

    Args:
        resource (str): resource name the command registers under.
        io (CommandIO): the backend's op table; must wire rm_r.
    """
    stat = io.stat
    readdir = io.readdir
    resolve_glob = io.resolve_glob
    unlink = io.require(Operation.UNLINK)
    rmdir = io.require(Operation.RMDIR)
    rm_r = io.rm_r
    if rm_r is None:
        raise NotImplementedError(
            "operation 'rm_r' is not supported on this backend")

    async def _rm(
        accessor: Accessor,
        path: PathSpec,
        recursive: bool = False,
        force: bool = False,
        remove_dir: bool = False,
        verbose: bool = False,
        *,
        index: IndexCacheStore,
    ) -> tuple[str | None, list[str]]:
        """Remove one operand, returning a GNU stderr line on failure.

        Args:
            accessor (Accessor): Backend handle.
            path (PathSpec): The operand to remove.
            recursive (bool): ``-r``; remove directories and their
                contents.
            force (bool): ``-f``; a missing operand is not an error.
            remove_dir (bool): ``-d``; remove empty directories.
            verbose (bool): ``-v``; collect one ``removed ...`` line per
                entry.
            index (IndexCacheStore): Cache index threaded into the core
                ops.

        Returns:
            tuple[str | None, list[str]]: A ``rm: cannot remove ...``
            line (or None when removed / skipped under ``-f``) and the
            verbose lines.
        """
        label = path.virtual
        try:
            s = await stat(accessor, path, index=index)
        except (FileNotFoundError, ValueError):
            if force:
                return None, []
            return (f"rm: cannot remove '{label}': "
                    "No such file or directory"), []
        if s.type == FileType.DIRECTORY:
            if recursive:
                lines = removal_lines(await walk(
                    functools.partial(readdir, accessor, index=index),
                    functools.partial(stat, accessor, index=index),
                    path)) if verbose else []
                await rm_r(accessor, path)
                return None, lines
            if remove_dir:
                children = await readdir(accessor, path, index)
                if children:
                    return (f"rm: cannot remove '{label}': "
                            "Directory not empty"), []
                await rmdir(accessor, path)
                return None, ([f"removed directory '{label}'"]
                              if verbose else [])
            return f"rm: cannot remove '{label}': Is a directory", []
        await unlink(accessor, path)
        return None, [f"removed '{label}'"] if verbose else []

    async def rm(
        accessor: Accessor,
        paths: list[PathSpec],
        texts: list[str],
        opts: CommandOpts,
    ) -> tuple[ByteSource | None, IOResult]:
        if not paths:
            raise ValueError("rm: missing operand")
        fl = FlagView(opts.flags, spec=SPECS["rm"])
        r = fl.as_bool("r") or fl.as_bool("R")
        f = fl.as_bool("f")
        v = fl.as_bool("v")
        d = fl.as_bool("d")
        paths = await resolve_glob(accessor, paths, opts.index)
        verbose_parts: list[str] = []
        errors: list[str] = []
        removed: dict[str, ByteSource] = {}
        links = opts.ns.links if opts.ns is not None else None
        for p in paths:
            # A link typed with a trailing slash is refused, never
            # followed: the shared helper keeps this identical to the
            # generic builder.
            if is_slashed_link(p, links):
                refusal = await rm_link_refusal(p, links, recursive=r, force=f)
                if refusal is not None:
                    errors.append(refusal)
                continue
            # GNU rm reports the operand and keeps removing the rest.
            error, entry_lines = await _rm(accessor,
                                           p,
                                           recursive=r,
                                           force=f,
                                           remove_dir=d,
                                           verbose=v,
                                           index=opts.index)
            if error is not None:
                errors.append(error)
                continue
            removed[p.mount_path] = b""
            verbose_parts.extend(entry_lines)
        output = format_optional_records(verbose_parts) if v else None
        stderr = ("\n".join(errors) + "\n").encode() if errors else None
        return output, IOResult(writes=removed,
                                stderr=stderr,
                                exit_code=1 if errors else 0)

    wrapped: Callable[..., Any] = command("rm",
                                          resource=resource,
                                          spec=SPECS["rm"],
                                          write=True)(rm)
    return wrapped
