# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/across-what-write-to-temp-fsync-and-atomic-rename-gates-must-an-instance-persist-local-files-to-prevent-zero-byte-truncation-and-corrupted-state-files.md
# Remnant Warrant:    https://corpusfetch.com/c/across-what-write-to-temp-fsync-and-atomic-rename-gates-must-an-instance-persist-local-files-to-prevent-zero-byte-truncation-and-corrupted-state-files.warrant.json
# Markdown SHA-256:   8424e5815bd0135c49bff7cf31b76ce45661b693830e80e67eb421b0f4abf3c6
# Code AST SHA-256:   0ce90286166880d8c0f5a8b2f2e5c5d9810f4200fdc8a4629749c4dabddcb4bc
# Verified Classes:   AtomicFileWriter
# Defensive Guards:   1 fail-closed invariants (zero external deps)

import os
import tempfile
from pathlib import Path
from typing import Any


class AtomicFileWriter:
    '''
    Provides robust, crash-safe, and power-loss-durable local file persistence:
    1. Writes to a unique temporary file in the SAME directory (`EXDEV` prevention).
    2. Invokes `os.fsync()` on the file descriptor for data durability.
    3. Invokes `os.fsync()` on the parent directory descriptor for metadata durability.
    4. Swaps files atomically via `os.replace()` (zero truncation window).
    '''

    @staticmethod
    def write_file_atomically(
        target_path: str | Path,
        content: str | bytes,
        *,
        encoding: str = "utf-8",
        mode: int = 0o600,
    ) -> None:
        target = Path(target_path).resolve()
        parent_dir = target.parent
        parent_dir.mkdir(parents=True, exist_ok=True)

        data = content.encode(encoding) if isinstance(content, str) else content
        tmp_fd: int | None = None
        tmp_name: str | None = None

        try:
            # Gate 1: Create temp file in the EXACT same directory
            tmp_fd, tmp_name = tempfile.mkstemp(
                dir=str(parent_dir),
                prefix=f".tmp-{target.name}-",
            )

            # Write content to file descriptor
            with os.fdopen(tmp_fd, "wb") as f:
                tmp_fd = None  # ownership transferred to `with` block
                f.write(data)
                f.flush()
                # Gate 2: Explicit file fsync
                os.fsync(f.fileno())

            # Set requested POSIX permissions (e.g., 0o600 for secrets/config)
            os.chmod(tmp_name, mode)

            # Gate 4: Atomic POSIX replacement
            os.replace(tmp_name, target)

            # Gate 3: Parent directory fsync for metadata durability
            dir_fd = os.open(str(parent_dir), os.O_RDONLY)
            try:
                os.fsync(dir_fd)
            finally:
                os.close(dir_fd)

        except BaseException:
            # Cleanup temp file if any error occurred before atomic replace
            if tmp_fd is not None:
                try:
                    os.close(tmp_fd)
                except Exception:
                    pass
            if tmp_name is not None:
                try:
                    os.unlink(tmp_name)
                except Exception:
                    pass
            raise