# retoor <retoor@molodetz.nl>

import logging
import os
import shutil
from datetime import datetime, timezone
from pathlib import Path

from devplacepy.database import get_table, db, get_int_setting
from devplacepy.config import PROJECT_FILES_DIR
from devplacepy.utils import generate_uid
from devplacepy.attachments import _directory_for, _detect_mime

logger = logging.getLogger(__name__)

MAX_TEXT_CHARS = 400_000
MAX_FILES_PER_PROJECT = 5000
MAX_PATH_LENGTH = 1024
MAX_SEGMENT_LENGTH = 255
MAX_DEPTH = 32

TEXT_EXTENSIONS = {
    ".txt",
    ".md",
    ".markdown",
    ".rst",
    ".log",
    ".csv",
    ".tsv",
    ".py",
    ".pyi",
    ".ipynb",
    ".js",
    ".mjs",
    ".cjs",
    ".jsx",
    ".ts",
    ".tsx",
    ".json",
    ".jsonc",
    ".yaml",
    ".yml",
    ".toml",
    ".ini",
    ".cfg",
    ".conf",
    ".env",
    ".html",
    ".htm",
    ".xml",
    ".svg",
    ".css",
    ".scss",
    ".sass",
    ".less",
    ".c",
    ".h",
    ".cpp",
    ".cc",
    ".hpp",
    ".hh",
    ".cs",
    ".java",
    ".kt",
    ".kts",
    ".go",
    ".rs",
    ".rb",
    ".php",
    ".pl",
    ".pm",
    ".lua",
    ".r",
    ".dart",
    ".swift",
    ".scala",
    ".clj",
    ".ex",
    ".exs",
    ".erl",
    ".hs",
    ".ml",
    ".vue",
    ".svelte",
    ".sh",
    ".bash",
    ".zsh",
    ".fish",
    ".ps1",
    ".bat",
    ".cmd",
    ".sql",
    ".graphql",
    ".gql",
    ".proto",
    ".tf",
    ".hcl",
    ".dockerfile",
    ".gitignore",
    ".gitattributes",
    ".editorconfig",
    ".properties",
}
TEXT_FILENAMES = {
    "dockerfile",
    "makefile",
    "license",
    "readme",
    "changelog",
    "authors",
    "procfile",
    "rakefile",
    "gemfile",
    ".gitignore",
    ".gitattributes",
    ".env",
    ".editorconfig",
    ".dockerignore",
    ".npmrc",
    ".prettierrc",
    ".babelrc",
}


class ProjectFileError(ValueError):
    pass


def is_readonly(project_uid: str) -> bool:
    if "projects" not in db.tables:
        return False
    project = get_table("projects").find_one(uid=project_uid)
    return bool(project and project.get("read_only"))


def _guard_writable(project_uid: str) -> None:
    if is_readonly(project_uid):
        raise ProjectFileError("project is read-only; files cannot be modified")


def normalize_path(raw) -> str:
    if raw is None:
        raise ProjectFileError("path is required")
    text = str(raw).strip().replace("\\", "/")
    if "\x00" in text:
        raise ProjectFileError("path contains a null byte")
    parts = []
    for segment in text.split("/"):
        segment = segment.strip()
        if segment in ("", "."):
            continue
        if segment == "..":
            raise ProjectFileError("path may not contain '..'")
        if len(segment) > MAX_SEGMENT_LENGTH:
            raise ProjectFileError("path segment is too long")
        if any(ord(ch) < 32 for ch in segment):
            raise ProjectFileError("path contains control characters")
        parts.append(segment)
    if not parts:
        raise ProjectFileError("path is empty")
    if len(parts) > MAX_DEPTH:
        raise ProjectFileError("path is too deep")
    normalized = "/".join(parts)
    if len(normalized) > MAX_PATH_LENGTH:
        raise ProjectFileError("path is too long")
    return normalized


def _name_of(path: str) -> str:
    return path.rsplit("/", 1)[-1]


def _parent_of(path: str) -> str:
    return path.rsplit("/", 1)[0] if "/" in path else ""


def is_text_name(name: str) -> bool:
    lower = name.lower()
    if lower in TEXT_FILENAMES:
        return True
    return Path(name).suffix.lower() in TEXT_EXTENSIONS


def _table():
    return get_table("project_files")


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _file_url(directory: str, stored_name: str) -> str:
    return f"/static/uploads/project_files/{directory}/{stored_name}"


def get_node(project_uid: str, path: str):
    return _table().find_one(project_uid=project_uid, path=path, deleted_at=None)


def _count(project_uid: str) -> int:
    if "project_files" not in db.tables:
        return 0
    return _table().count(project_uid=project_uid, deleted_at=None)


def _guard_capacity(project_uid: str) -> None:
    if _count(project_uid) >= MAX_FILES_PER_PROJECT:
        raise ProjectFileError(
            f"project reached the {MAX_FILES_PER_PROJECT}-node limit"
        )


def _insert_dir(project_uid: str, user: dict, path: str) -> None:
    _table().insert(
        {
            "uid": generate_uid(),
            "project_uid": project_uid,
            "deleted_at": None,
            "deleted_by": None,
            "user_uid": user["uid"],
            "path": path,
            "name": _name_of(path),
            "parent_path": _parent_of(path),
            "type": "dir",
            "content": None,
            "is_binary": 0,
            "stored_name": None,
            "directory": None,
            "mime_type": None,
            "size": 0,
            "created_at": _now(),
            "updated_at": _now(),
        }
    )


def ensure_dirs(project_uid: str, user: dict, dir_path: str) -> None:
    if not dir_path:
        return
    current = ""
    for segment in dir_path.split("/"):
        current = f"{current}/{segment}" if current else segment
        existing = get_node(project_uid, current)
        if existing is None:
            _guard_capacity(project_uid)
            _insert_dir(project_uid, user, current)
        elif existing["type"] != "dir":
            raise ProjectFileError(f"'{current}' exists and is a file")


def make_dir(project_uid: str, user: dict, raw_path: str) -> dict:
    _guard_writable(project_uid)
    path = normalize_path(raw_path)
    existing = get_node(project_uid, path)
    if existing is not None:
        if existing["type"] != "dir":
            raise ProjectFileError(f"'{path}' exists and is a file")
        return existing
    ensure_dirs(project_uid, user, path)
    return get_node(project_uid, path)


def write_text_file(project_uid: str, user: dict, raw_path: str, content: str) -> dict:
    _guard_writable(project_uid)
    path = normalize_path(raw_path)
    content = content or ""
    if len(content) > MAX_TEXT_CHARS:
        raise ProjectFileError(
            f"file content exceeds the {MAX_TEXT_CHARS}-character limit"
        )
    ensure_dirs(project_uid, user, _parent_of(path))
    existing = get_node(project_uid, path)
    if existing is not None and existing["type"] == "dir":
        raise ProjectFileError(f"'{path}' is a directory")
    if existing is not None:
        if existing.get("is_binary"):
            _unlink_blob(existing)
        _table().update(
            {
                "uid": existing["uid"],
                "content": content,
                "is_binary": 0,
                "stored_name": None,
                "directory": None,
                "mime_type": "text/plain",
                "size": len(content.encode("utf-8")),
                "updated_at": _now(),
            },
            ["uid"],
        )
        return get_node(project_uid, path)
    _guard_capacity(project_uid)
    _table().insert(
        {
            "uid": generate_uid(),
            "project_uid": project_uid,
            "deleted_at": None,
            "deleted_by": None,
            "user_uid": user["uid"],
            "path": path,
            "name": _name_of(path),
            "parent_path": _parent_of(path),
            "type": "file",
            "content": content,
            "is_binary": 0,
            "stored_name": None,
            "directory": None,
            "mime_type": "text/plain",
            "size": len(content.encode("utf-8")),
            "created_at": _now(),
            "updated_at": _now(),
        }
    )
    return get_node(project_uid, path)


def _store_blob(data: bytes, filename: str) -> tuple[str, str, str]:
    uid = generate_uid()
    ext = Path(filename).suffix.lower()
    stored_name = f"{uid}{ext}"
    directory = _directory_for(uid)
    file_dir = PROJECT_FILES_DIR / directory
    file_dir.mkdir(parents=True, exist_ok=True)
    (file_dir / stored_name).write_bytes(data)
    return stored_name, directory, _detect_mime(data, filename)


def _unlink_blob(node: dict) -> None:
    stored_name = node.get("stored_name")
    directory = node.get("directory")
    if not (stored_name and directory):
        return
    try:
        (PROJECT_FILES_DIR / directory / stored_name).unlink(missing_ok=True)
    except OSError as exc:
        logger.warning(
            "Failed to delete project blob %s/%s: %s", directory, stored_name, exc
        )


def store_upload(
    project_uid: str, user: dict, dir_path: str, filename: str, data: bytes
) -> dict:
    _guard_writable(project_uid)
    max_bytes = get_int_setting("max_upload_size_mb", 10) * 1024 * 1024
    if len(data) > max_bytes:
        raise ProjectFileError(
            f"file exceeds the {get_int_setting('max_upload_size_mb', 10)}MB limit"
        )
    base = _name_of(normalize_path(filename))
    path = normalize_path(f"{dir_path}/{base}" if dir_path else base)

    if is_text_name(base):
        try:
            text = data.decode("utf-8")
            if len(text) <= MAX_TEXT_CHARS:
                return write_text_file(project_uid, user, path, text)
        except UnicodeDecodeError:
            pass

    ensure_dirs(project_uid, user, _parent_of(path))
    existing = get_node(project_uid, path)
    if existing is not None and existing["type"] == "dir":
        raise ProjectFileError(f"'{path}' is a directory")
    stored_name, directory, mime = _store_blob(data, base)
    if existing is not None:
        if existing.get("is_binary"):
            _unlink_blob(existing)
        _table().update(
            {
                "uid": existing["uid"],
                "content": None,
                "is_binary": 1,
                "stored_name": stored_name,
                "directory": directory,
                "mime_type": mime,
                "size": len(data),
                "updated_at": _now(),
            },
            ["uid"],
        )
        return get_node(project_uid, path)
    _guard_capacity(project_uid)
    _table().insert(
        {
            "uid": generate_uid(),
            "project_uid": project_uid,
            "deleted_at": None,
            "deleted_by": None,
            "user_uid": user["uid"],
            "path": path,
            "name": base,
            "parent_path": _parent_of(path),
            "type": "file",
            "content": None,
            "is_binary": 1,
            "stored_name": stored_name,
            "directory": directory,
            "mime_type": mime,
            "size": len(data),
            "created_at": _now(),
            "updated_at": _now(),
        }
    )
    return get_node(project_uid, path)


def _descendants(project_uid: str, path: str) -> list:
    prefix = f"{path}/"
    return [
        row
        for row in _table().find(project_uid=project_uid, deleted_at=None)
        if row["path"] == path or row["path"].startswith(prefix)
    ]


def move_node(project_uid: str, user: dict, raw_from: str, raw_to: str) -> dict:
    _guard_writable(project_uid)
    src_path = normalize_path(raw_from)
    dst_path = normalize_path(raw_to)
    if src_path == dst_path:
        raise ProjectFileError("source and target are the same")
    if dst_path == src_path or dst_path.startswith(f"{src_path}/"):
        raise ProjectFileError("cannot move a node into itself")
    src = get_node(project_uid, src_path)
    if src is None:
        raise ProjectFileError(f"'{src_path}' does not exist")
    if get_node(project_uid, dst_path) is not None:
        raise ProjectFileError(f"'{dst_path}' already exists")
    ensure_dirs(project_uid, user, _parent_of(dst_path))

    if src["type"] == "file":
        _table().update(
            {
                "uid": src["uid"],
                "path": dst_path,
                "name": _name_of(dst_path),
                "parent_path": _parent_of(dst_path),
                "updated_at": _now(),
            },
            ["uid"],
        )
        return get_node(project_uid, dst_path)

    for row in _descendants(project_uid, src_path):
        new_path = dst_path + row["path"][len(src_path) :]
        _table().update(
            {
                "uid": row["uid"],
                "path": new_path,
                "name": _name_of(new_path),
                "parent_path": _parent_of(new_path),
                "updated_at": _now(),
            },
            ["uid"],
        )
    return get_node(project_uid, dst_path)


def delete_node(project_uid: str, raw_path: str, deleted_by: str = "system") -> None:
    _guard_writable(project_uid)
    path = normalize_path(raw_path)
    node = get_node(project_uid, path)
    if node is None:
        raise ProjectFileError(f"'{path}' does not exist")
    stamp = _now()
    rows = _descendants(project_uid, path)
    for row in rows:
        _table().update(
            {"uid": row["uid"], "deleted_at": stamp, "deleted_by": deleted_by},
            ["uid"],
        )
    for row in rows:
        if row.get("is_binary"):
            _unlink_blob(row)


def soft_delete_all_project_files(project_uid: str, deleted_by: str) -> None:
    if "project_files" not in db.tables:
        return
    clear_sync_state(project_uid)
    stamp = _now()
    for row in _table().find(project_uid=project_uid, deleted_at=None):
        _table().update(
            {"uid": row["uid"], "deleted_at": stamp, "deleted_by": deleted_by},
            ["uid"],
        )


_SYNC_CLOCK_SKEW_SECONDS = 1.0
SYNC_STATE_TABLE = "project_file_sync_state"


def _epoch_of(iso_timestamp) -> float:
    if not iso_timestamp:
        return 0.0
    try:
        return datetime.fromisoformat(str(iso_timestamp)).timestamp()
    except (TypeError, ValueError):
        return 0.0


def _sync_state_table():
    return get_table(SYNC_STATE_TABLE)


def _load_sync_manifest(project_uid: str) -> dict:
    if SYNC_STATE_TABLE not in db.tables:
        return {}
    return {
        row["path"]: {"db_epoch": row["db_epoch"], "fs_epoch": row["fs_epoch"]}
        for row in _sync_state_table().find(project_uid=project_uid)
    }


def _save_sync_manifest(project_uid: str, old: dict, new: dict) -> None:
    table = _sync_state_table()
    for path in old:
        if path not in new:
            table.delete(project_uid=project_uid, path=path)
    for path, entry in new.items():
        if old.get(path) == entry:
            continue
        table.upsert(
            {"project_uid": project_uid, "path": path, **entry},
            ["project_uid", "path"],
        )


def clear_sync_state(project_uid: str) -> None:
    if SYNC_STATE_TABLE not in db.tables:
        return
    _sync_state_table().delete(project_uid=project_uid)


def _delete_db_row_for_sync(row: dict, deleted_by: str) -> None:
    _table().update(
        {"uid": row["uid"], "deleted_at": _now(), "deleted_by": deleted_by},
        ["uid"],
    )
    if row.get("is_binary"):
        _unlink_blob(row)


def _file_records(project_uid: str) -> dict:
    records: dict = {}
    for row in _table().find(project_uid=project_uid, deleted_at=None):
        if row["type"] != "file":
            continue
        records[row["path"]] = row
    return records


def _walk_workspace_files(src, skip):
    for root, dirs, files in src.walk():
        dirs[:] = [d for d in sorted(dirs) if d not in skip]
        for name in sorted(files):
            if name in skip or Path(name).suffix in IMPORT_SKIP_EXTENSIONS:
                continue
            full = Path(root) / name
            if full.is_symlink() or not full.is_file():
                continue
            try:
                if full.stat().st_size > IMPORT_MAX_FILE_BYTES:
                    continue
                relative = full.relative_to(src).as_posix()
                path = normalize_path(relative)
            except (OSError, ProjectFileError):
                continue
            yield path, full


def _workspace_records(workspace) -> dict:
    src = Path(workspace).resolve()
    records: dict = {}
    if not src.is_dir():
        return records
    for path, full in _walk_workspace_files(src, SYNC_SKIP_NAMES):
        records[path] = full
    return records


def sync_dir_bidirectional(project_uid: str, workspace, user: dict) -> dict:
    empty = {
        "exported": 0,
        "imported": 0,
        "deleted_in_project": 0,
        "deleted_in_workspace": 0,
    }
    if "project_files" not in db.tables:
        return empty
    readonly = is_readonly(project_uid)
    dest = Path(workspace).resolve()
    dest.mkdir(parents=True, exist_ok=True)
    db_files = _file_records(project_uid)
    fs_files = _workspace_records(dest)
    manifest = _load_sync_manifest(project_uid)
    new_manifest: dict = {}
    counts = dict(empty)

    for path in set(manifest) | set(db_files) | set(fs_files):
        row = db_files.get(path)
        fs_full = fs_files.get(path)
        entry = manifest.get(path)

        if row is not None and fs_full is not None:
            try:
                fs_epoch = fs_full.stat().st_mtime
            except OSError:
                continue
            db_epoch = _epoch_of(row.get("updated_at"))
            if (
                entry is not None
                and abs(db_epoch - entry["db_epoch"]) <= _SYNC_CLOCK_SKEW_SECONDS
                and abs(fs_epoch - entry["fs_epoch"]) <= _SYNC_CLOCK_SKEW_SECONDS
            ):
                new_manifest[path] = entry
                continue
            if db_epoch >= fs_epoch - _SYNC_CLOCK_SKEW_SECONDS or readonly:
                recorded = _record_export(row, dest)
                if recorded:
                    counts["exported"] += 1
                    new_manifest[path] = recorded
            else:
                recorded = _record_import(project_uid, user, path, fs_full, fs_epoch)
                if recorded:
                    counts["imported"] += 1
                    new_manifest[path] = recorded
            continue

        if row is not None and fs_full is None:
            db_epoch = _epoch_of(row.get("updated_at"))
            if (
                entry is None
                or readonly
                or db_epoch > entry["db_epoch"] + _SYNC_CLOCK_SKEW_SECONDS
            ):
                recorded = _record_export(row, dest)
                if recorded:
                    counts["exported"] += 1
                    new_manifest[path] = recorded
            else:
                _delete_db_row_for_sync(row, user["uid"])
                counts["deleted_in_project"] += 1
            continue

        if row is None and fs_full is not None:
            try:
                fs_epoch = fs_full.stat().st_mtime
            except OSError:
                continue
            if entry is None:
                if not readonly:
                    recorded = _record_import(project_uid, user, path, fs_full, fs_epoch)
                    if recorded:
                        counts["imported"] += 1
                        new_manifest[path] = recorded
                continue
            if not readonly and fs_epoch > entry["fs_epoch"] + _SYNC_CLOCK_SKEW_SECONDS:
                recorded = _record_import(project_uid, user, path, fs_full, fs_epoch)
                if recorded:
                    counts["imported"] += 1
                    new_manifest[path] = recorded
                continue
            try:
                fs_full.unlink()
                counts["deleted_in_workspace"] += 1
            except OSError:
                pass
            continue

    _save_sync_manifest(project_uid, manifest, new_manifest)
    return counts


def _record_export(row: dict, dest: Path):
    target = _export_node(row, dest)
    if target is None:
        return None
    db_epoch = _epoch_of(row.get("updated_at"))
    try:
        fs_epoch = target.stat().st_mtime
    except OSError:
        fs_epoch = db_epoch
    return {"db_epoch": db_epoch, "fs_epoch": fs_epoch}


def _record_import(project_uid: str, user: dict, path: str, fs_full: Path, fs_epoch: float):
    imported = _import_file(project_uid, user, path, fs_full)
    if imported is None:
        return None
    return {"db_epoch": _epoch_of(imported.get("updated_at")), "fs_epoch": fs_epoch}


def _export_node(row: dict, dest: Path):
    target = (dest / row["path"]).resolve()
    if target != dest and not target.is_relative_to(dest):
        return None
    target.parent.mkdir(parents=True, exist_ok=True)
    if target.is_symlink():
        target.unlink()
    if row.get("is_binary") and row.get("stored_name") and row.get("directory"):
        src = PROJECT_FILES_DIR / row["directory"] / row["stored_name"]
        try:
            shutil.copyfile(src, target)
        except (FileNotFoundError, OSError):
            logger.warning("Blob file missing during export: %s", src)
            return None
    else:
        target.write_text(row.get("content") or "", encoding="utf-8")
    return target


def _import_file(project_uid: str, user: dict, path: str, fs_full: Path):
    try:
        data = fs_full.read_bytes()
    except OSError:
        return None
    try:
        return store_upload(project_uid, user, _parent_of(path), _name_of(path), data)
    except ProjectFileError:
        return None


def node_to_dict(row: dict) -> dict:
    is_binary = bool(row.get("is_binary"))
    url = None
    if is_binary and row.get("stored_name") and row.get("directory"):
        url = _file_url(row["directory"], row["stored_name"])
    return {
        "uid": row["uid"],
        "path": row["path"],
        "name": row["name"],
        "type": row["type"],
        "is_binary": is_binary,
        "mime_type": row.get("mime_type"),
        "size": row.get("size", 0),
        "url": url,
        "updated_at": row.get("updated_at"),
    }


def list_files(project_uid: str) -> list:
    if "project_files" not in db.tables:
        return []
    rows = sorted(
        _table().find(project_uid=project_uid, deleted_at=None),
        key=lambda r: r["path"],
    )
    return [node_to_dict(row) for row in rows]


def count_files(project_uid: str) -> int:
    if "project_files" not in db.tables:
        return 0
    return _table().count(project_uid=project_uid, type="file", deleted_at=None)


def read_file(project_uid: str, raw_path: str) -> dict:
    path = normalize_path(raw_path)
    node = get_node(project_uid, path)
    if node is None:
        raise ProjectFileError(f"'{path}' does not exist")
    payload = node_to_dict(node)
    payload["content"] = (
        node.get("content")
        if node["type"] == "file" and not node.get("is_binary")
        else None
    )
    return payload


IMPORT_SKIP_NAMES = {
    ".git",
    "node_modules",
    "__pycache__",
    ".venv",
    "venv",
    ".mypy_cache",
    ".pytest_cache",
    ".DS_Store",
    ".idea",
    ".cache",
    "build",
    "dist",
    "target",
    "out",
    "bin",
    "obj",
    ".next",
    ".nuxt",
    ".gradle",
    ".tox",
    "cmake-build-debug",
    "cmake-build-release",
}
# Compiled/build-artifact extensions, regenerated wholesale on every build -
# never worth importing/syncing regardless of which directory they land in
# (unlike IMPORT_SKIP_NAMES, matched by suffix rather than exact name; see
# _walk_workspace_files). A missing exclusion here is what let an unlocked
# concurrent sync (see api._sync_dir_bidirectional_locked) leak millions of
# orphaned blobs from an actively-compiling container workspace.
IMPORT_SKIP_EXTENSIONS = {
    ".o",
    ".obj",
    ".pyc",
    ".pyo",
    ".class",
    ".so",
    ".dylib",
    ".dll",
    ".a",
    ".exe",
}
SYNC_SKIP_NAMES = IMPORT_SKIP_NAMES | {
    ".devplace_boot.py",
    ".devplace_boot.sh",
    ".devplace",
    ".dpc",
    "dpc.log",
}
IMPORT_MAX_FILE_BYTES = 25 * 1024 * 1024


def import_from_dir(project_uid: str, src_dir, user: dict, *, skip_names=None) -> int:
    _guard_writable(project_uid)
    src = Path(src_dir).resolve()
    if not src.is_dir():
        raise ProjectFileError("source directory does not exist")
    skip = set(skip_names) if skip_names is not None else set(IMPORT_SKIP_NAMES)
    imported = 0
    for path, full in _walk_workspace_files(src, skip):
        try:
            data = full.read_bytes()
        except OSError:
            continue
        try:
            store_upload(project_uid, user, _parent_of(path), _name_of(path), data)
            imported += 1
        except ProjectFileError:
            continue
    return imported


def export_to_dir(project_uid: str, subpath: str, dest_dir) -> int:
    dest = Path(dest_dir).resolve()
    dest.mkdir(parents=True, exist_ok=True)
    if subpath:
        base = normalize_path(subpath)
        rows = _descendants(project_uid, base)
        strip = _parent_of(base)
    else:
        rows = list(_table().find(project_uid=project_uid, deleted_at=None))
        strip = ""
    written = 0
    for row in sorted(rows, key=lambda r: r["path"]):
        relative = row["path"][len(strip) :].lstrip("/") if strip else row["path"]
        target = (dest / relative).resolve()
        if target != dest and not target.is_relative_to(dest):
            raise ProjectFileError("path escapes export directory")
        if row["type"] == "dir":
            target.mkdir(parents=True, exist_ok=True)
            continue
        target.parent.mkdir(parents=True, exist_ok=True)
        if target.is_symlink() or target.is_file():
            target.unlink()
        if row.get("is_binary") and row.get("stored_name") and row.get("directory"):
            src = PROJECT_FILES_DIR / row["directory"] / row["stored_name"]
            try:
                shutil.copyfile(src, target)
            except (FileNotFoundError, OSError):
                logger.warning("Blob file missing: %s", src)
                continue
        else:
            target.write_text(row.get("content") or "", encoding="utf-8")
        written += 1
    return written


def _load_text_node(project_uid: str, raw_path: str) -> tuple[dict, str]:
    path = normalize_path(raw_path)
    node = get_node(project_uid, path)
    if node is None:
        raise ProjectFileError(f"'{path}' does not exist")
    if node["type"] != "file":
        raise ProjectFileError(f"'{path}' is a directory, not a file")
    if node.get("is_binary"):
        raise ProjectFileError(f"'{path}' is a binary file and has no editable lines")
    return node, node.get("content") or ""


def _split_lines(content: str) -> tuple[list, bool]:
    if content == "":
        return [], False
    trailing = content.endswith("\n")
    body = content[:-1] if trailing else content
    return body.split("\n"), trailing


def _join_lines(lines: list, trailing: bool) -> str:
    if not lines:
        return ""
    return "\n".join(lines) + ("\n" if trailing else "")


def _split_content(content: str) -> tuple[list, bool]:
    if not content:
        return [], False
    return _split_lines(content)


def _tail_trailing(new_lines: list, tail: list, file_trailing: bool, content_trailing: bool) -> bool:
    if not new_lines:
        return False
    if tail:
        return file_trailing
    return content_trailing or file_trailing


def _save_text(project_uid: str, node: dict, new_content: str) -> dict:
    if len(new_content) > MAX_TEXT_CHARS:
        raise ProjectFileError(
            f"file content exceeds the {MAX_TEXT_CHARS}-character limit"
        )
    _table().update(
        {
            "uid": node["uid"],
            "content": new_content,
            "is_binary": 0,
            "stored_name": None,
            "directory": None,
            "mime_type": "text/plain",
            "size": len(new_content.encode("utf-8")),
            "updated_at": _now(),
        },
        ["uid"],
    )
    return get_node(project_uid, node["path"])


def read_lines(project_uid: str, raw_path: str, start: int = 1, end=None) -> dict:
    node, content = _load_text_node(project_uid, raw_path)
    lines, _ = _split_lines(content)
    total = len(lines)
    start = max(1, int(start))
    end = total if end is None else int(end)
    if end < 0:
        end = total
    end = min(end, total)
    selected = lines[start - 1 : end] if start <= total and end >= start else []
    return {
        "path": node["path"],
        "start": start,
        "end": end if selected else start - 1,
        "total_lines": total,
        "lines": selected,
        "content": "\n".join(selected),
    }


def replace_lines(
    project_uid: str, raw_path: str, start: int, end: int, content: str
) -> dict:
    _guard_writable(project_uid)
    node, original = _load_text_node(project_uid, raw_path)
    lines, file_trailing = _split_lines(original)
    total = len(lines)
    start = max(1, int(start))
    end = int(end)
    if start > total:
        raise ProjectFileError(
            f"start line {start} is past the end of the file ({total} lines)"
        )
    end = min(max(end, start - 1), total)
    replacement, content_trailing = _split_content(content)
    tail = lines[end:]
    new_lines = lines[: start - 1] + replacement + tail
    trailing = _tail_trailing(new_lines, tail, file_trailing, content_trailing)
    return _save_text(project_uid, node, _join_lines(new_lines, trailing))


def insert_lines(project_uid: str, raw_path: str, at: int, content: str) -> dict:
    _guard_writable(project_uid)
    node, original = _load_text_node(project_uid, raw_path)
    lines, file_trailing = _split_lines(original)
    total = len(lines)
    at = max(1, int(at))
    if at > total + 1:
        at = total + 1
    replacement, content_trailing = _split_content(content)
    tail = lines[at - 1 :]
    new_lines = lines[: at - 1] + replacement + tail
    trailing = _tail_trailing(new_lines, tail, file_trailing, content_trailing)
    return _save_text(project_uid, node, _join_lines(new_lines, trailing))


def delete_lines(project_uid: str, raw_path: str, start: int, end: int) -> dict:
    _guard_writable(project_uid)
    node, original = _load_text_node(project_uid, raw_path)
    lines, trailing = _split_lines(original)
    total = len(lines)
    start = max(1, int(start))
    end = min(int(end), total)
    if start > total or end < start:
        raise ProjectFileError(
            f"no lines in range {start}-{end} (file has {total} lines)"
        )
    new_lines = lines[: start - 1] + lines[end:]
    return _save_text(
        project_uid, node, _join_lines(new_lines, trailing if new_lines else False)
    )


def append_lines(project_uid: str, raw_path: str, content: str) -> dict:
    _guard_writable(project_uid)
    node, original = _load_text_node(project_uid, raw_path)
    lines, file_trailing = _split_lines(original)
    replacement, content_trailing = _split_content(content)
    new_lines = lines + replacement
    trailing = bool(new_lines) and (content_trailing or file_trailing)
    return _save_text(project_uid, node, _join_lines(new_lines, trailing))


def delete_all_project_files(project_uid: str) -> None:
    if "project_files" not in db.tables:
        return
    clear_sync_state(project_uid)
    for row in _table().find(project_uid=project_uid):
        if row.get("is_binary"):
            _unlink_blob(row)
    _table().delete(project_uid=project_uid)


def purge_soft_deleted_project_files(*, dry_run: bool = False) -> tuple[int, int]:
    if "project_files" not in db.tables:
        return 0, 0
    table = _table()
    rows = list(table.find(table.table.columns.deleted_at.isnot(None), is_binary=1))
    removed = 0
    freed = 0
    for row in rows:
        directory = row.get("directory")
        stored_name = row.get("stored_name")
        if directory and stored_name:
            path = PROJECT_FILES_DIR / directory / stored_name
            try:
                freed += path.stat().st_size
            except OSError:
                pass
            else:
                if not dry_run:
                    _unlink_blob(row)
        if not dry_run:
            table.delete(id=row["id"])
        removed += 1
    return removed, freed


def sweep_orphan_project_file_blobs(*, dry_run: bool = False) -> tuple[int, int]:
    if not PROJECT_FILES_DIR.exists():
        return 0, 0
    referenced = set()
    if "project_files" in db.tables:
        for row in _table().find(is_binary=1):
            directory = row.get("directory")
            stored_name = row.get("stored_name")
            if directory and stored_name:
                referenced.add((directory, stored_name))
    removed = 0
    freed = 0
    base = str(PROJECT_FILES_DIR)
    for root, _dirs, files in os.walk(base):
        directory = os.path.relpath(root, base)
        for name in files:
            if (directory, name) in referenced:
                continue
            file_path = os.path.join(root, name)
            try:
                size = os.path.getsize(file_path)
            except OSError:
                continue
            if not dry_run:
                try:
                    os.unlink(file_path)
                except OSError:
                    continue
            removed += 1
            freed += size
    return removed, freed
