# ========= 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. =========

# The per-command metadata cases this file used to run now live in
# integ/unix/meta and integ/unix/meta_overlay, where the JSON battery runs
# them on 22 backends instead of the four here. What is left are the three
# scenarios the declarative harness cannot express: it can run commands and
# stat paths, but it cannot snapshot a workspace, reload it onto a fresh
# resource, or mutate a backend out of band. Retiring these needs snapshot
# and namespace support in the harness, not another case file.

import asyncio
import logging
import os
import shutil
import sys
import tempfile
import uuid
from pathlib import Path

import boto3
from moto.server import ThreadedMotoServer

# Importing mirage while this directory is on sys.path resolves the
# `redis` package to integ/redis.py (mirage swallows the resulting
# ImportError as "redis extra not installed", and the half-executed
# shadow module strips this directory from sys.path, breaking the
# caller's next sibling import). Import mirage with the directory off
# the path, mirroring integ/redis.py's own guard.
_INTEG_DIR = str(Path(__file__).resolve().parent)
_INTEG_ALIASES = {_INTEG_DIR, str(Path(__file__).parent), ""}
_ON_PATH = any(p in _INTEG_ALIASES for p in sys.path)
sys.path[:] = [p for p in sys.path if p not in _INTEG_ALIASES]

from mirage import MountMode, Workspace  # noqa: E402
from mirage.resource.ram import RAMResource  # noqa: E402
from mirage.resource.s3 import S3Config, S3Resource  # noqa: E402
from mirage.types import ConsistencyPolicy, FileStat, PathSpec  # noqa: E402

if _ON_PATH:
    sys.path.insert(0, _INTEG_DIR)

REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
BUCKET = "mirage-integ-meta"
CREDS = dict(aws_access_key_id="testing",
             aws_secret_access_key="testing",
             region_name="us-east-1")


def _meta_field(st: FileStat, field: str) -> str:
    if field == "mode":
        value = oct(st.mode)[2:] if st.mode is not None else "-"
    elif field == "uid":
        value = str(st.uid) if st.uid is not None else "-"
    elif field == "gid":
        value = str(st.gid) if st.gid is not None else "-"
    else:
        # First 19 chars ("2026-01-02T15:30:00") so the Z vs +00:00 suffix
        # never reaches the byte-diffed truth file.
        value = st.modified[:19] if st.modified else "-"
    return f"{field}={value}"


def meta_stat_line(st: FileStat, fields: tuple[str, ...]) -> str:
    return " ".join(_meta_field(st, field) for field in fields)


async def run_overlay_snapshot_roundtrip(ws: Workspace,
                                         fresh: S3Resource) -> None:
    # Overlay attrs live in namespace NODES, so they must survive a
    # snapshot even though the s3 resource is rebuilt fresh at load
    # (s3 snapshots redact creds and require a resources= override).
    await ws.execute("echo alpha > /data/f.txt")
    await ws.execute("chmod 601 /data/f.txt && chown 500:dev /data/f.txt"
                     " && touch -t 202601021530 /data/f.txt")
    snap = Path(tempfile.mkdtemp(prefix="mirage-meta-osnap-")) / "ws.tar"
    await ws.snapshot(str(snap))
    restored = await Workspace.load(str(snap), resources={"/data": fresh})
    st, _ = await restored.dispatch("stat",
                                    PathSpec.from_str_path("/data/f.txt"))
    print("=== overlay_snapshot_roundtrip ===")
    print(meta_stat_line(st, ("mode", "uid", "gid", "mtime")))
    await restored.execute("rm /data/f.txt")
    shutil.rmtree(snap.parent)


async def run_overlay_orphan_gc(config: S3Config) -> None:
    # A chmod on a slot-less backend (s3) creates an attribute overlay in
    # the namespace. When the object is deleted out-of-band (another agent,
    # the raw API), the overlay is orphaned. Under ALWAYS, a stat that the
    # backend reports gone must GC that orphaned node.
    ws = Workspace({"/data": S3Resource(config)},
                   mode=MountMode.WRITE,
                   consistency=ConsistencyPolicy.ALWAYS)
    await ws.execute("echo alpha > /data/g.txt && chmod 601 /data/g.txt")
    before = ws.namespace.meta_for("/data/g.txt") is not None
    mount = ws.namespace.mount_for("/data/g.txt")
    await mount.execute_op("unlink", "/data/g.txt")
    await ws.execute("stat /data/g.txt")
    after = ws.namespace.meta_for("/data/g.txt") is not None
    print("=== overlay_orphan_gc ===")
    print(f"before={before} after={after}")


async def run_snapshot_roundtrip() -> None:
    ws = Workspace({"/data": RAMResource()}, mode=MountMode.WRITE)
    await ws.execute("echo alpha > /data/f.txt")
    await ws.execute("chmod 601 /data/f.txt && chown 500:dev /data/f.txt"
                     " && touch -t 202601021530 /data/f.txt")
    snap = Path(tempfile.mkdtemp(prefix="mirage-meta-snap-")) / "ws.tar"
    await ws.snapshot(str(snap))
    restored = await Workspace.load(str(snap))
    result = await restored.execute("ls -l /data")
    print("=== snapshot_meta_roundtrip ===")
    print((await result.stdout_str()).rstrip())
    shutil.rmtree(snap.parent)


async def main() -> None:
    logging.getLogger("werkzeug").setLevel(logging.ERROR)
    server = ThreadedMotoServer(ip_address="127.0.0.1", port=0, verbose=False)
    server.start()
    host, port = server.get_host_and_port()
    endpoint = f"http://{host}:{port}"
    bucket = f"{BUCKET}-{uuid.uuid4().hex[:8]}"
    config = S3Config(bucket=bucket,
                      region="us-east-1",
                      endpoint_url=endpoint,
                      aws_access_key_id="testing",
                      aws_secret_access_key="testing",
                      path_style=True)
    try:
        boto3.client("s3", endpoint_url=endpoint,
                     **CREDS).create_bucket(Bucket=bucket)
        s3_ws = Workspace({"/data": S3Resource(config)}, mode=MountMode.WRITE)
        await run_overlay_snapshot_roundtrip(s3_ws, S3Resource(config))
        await run_overlay_orphan_gc(config)
    finally:
        server.stop()

    await run_snapshot_roundtrip()


if __name__ == "__main__":
    asyncio.run(main())
