// SPDX-License-Identifier: LGPL-3.0-only
// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®

// The on-disk format: byte-exact header, commit record, extent table, schema
// blob and the 16-byte variable-length handle. Every struct here is a POD
// that is read and written with memcpy; the static_asserts pin each offset so
// a reordered member cannot silently change the file format. Little-endian
// only (asserted), which is every target Crafter builds for.
export module Crafter.DB:Format;
import std;

export namespace Crafter::DB {
    inline constexpr std::uint32_t pageSize = 4096;
    inline constexpr std::uint16_t formatVersion = 1;
    inline constexpr std::size_t maxExtents = 32;
    inline constexpr std::size_t regionSlots = 4;
    inline constexpr std::uint16_t regionCount = 2;   // 0 rows, 1 heap; 2-3 reserved
    inline constexpr std::uint32_t minExtentBytes = 16384;
    inline constexpr std::uint16_t noColumn = 0xFFFF;
    inline constexpr std::uint32_t inlineVarBytes = 12;  // strings up to this length live in the handle
    inline constexpr std::array<char, 8> fileMagic = { 'C', 'R', 'A', 'F', 'T', 'R', 'D', 'B' };
    inline constexpr std::array<char, 4> schemaMagic = { 'S', 'C', 'H', 'M' };
    inline constexpr std::uint32_t formatFlagLittleEndian = 1;
    inline constexpr std::uint64_t commitFlagCleanClose = 1;
    inline constexpr std::uint8_t rowFlagDeleted = 1;

    static_assert(std::endian::native == std::endian::little, "Crafter.DB stores little-endian and does not byte-swap");

    // Column kinds as stored in the descriptor table. Append only: values are
    // part of the file format.
    enum class Kind : std::uint8_t {
        Flags = 0,
        Bool = 1,
        Int8 = 2,
        Int16 = 3,
        Int32 = 4,
        Int64 = 5,
        UInt8 = 6,
        UInt16 = 7,
        UInt32 = 8,
        UInt64 = 9,
        Float32 = 10,
        Float64 = 11,
        TimePoint = 12,
        Duration = 13,
        String = 14,
        Blob = 15,
        FixedBytes = 16,
    };

    inline constexpr std::uint8_t columnFlagOptional = 1;  // payload of a std::optional; presenceOffset is valid
    inline constexpr std::uint8_t columnFlagEnum = 2;      // integer column backing an enum
    inline constexpr std::uint8_t columnFlagPresence = 4;  // the 1-byte presence column of a std::optional

    constexpr std::string_view KindName(Kind kind) noexcept {
        switch (kind) {
            case Kind::Flags: return "flags";
            case Kind::Bool: return "bool";
            case Kind::Int8: return "i8";
            case Kind::Int16: return "i16";
            case Kind::Int32: return "i32";
            case Kind::Int64: return "i64";
            case Kind::UInt8: return "u8";
            case Kind::UInt16: return "u16";
            case Kind::UInt32: return "u32";
            case Kind::UInt64: return "u64";
            case Kind::Float32: return "f32";
            case Kind::Float64: return "f64";
            case Kind::TimePoint: return "timepoint";
            case Kind::Duration: return "duration";
            case Kind::String: return "string";
            case Kind::Blob: return "blob";
            case Kind::FixedBytes: return "bytes";
        }
        return "?";
    }

    struct Extent {
        std::uint64_t fileOffset;
        std::uint64_t length;
    };
    static_assert(sizeof(Extent) == 16);

    // One of the two alternating commit slots. The writer fills the slot that
    // is not current, so a torn write can only damage the slot whose CRC then
    // fails, and the other slot still describes the previous commit.
    struct CommitRecord {
        std::uint64_t commitSeq;
        std::uint64_t rowCount;
        std::uint64_t liveCount;
        std::uint64_t heapUsed;
        std::uint64_t heapGarbage;
        std::array<std::uint32_t, regionSlots> extentCount;
        std::uint64_t fileLength;
        std::uint64_t modifiedUnixNanos;
        std::uint64_t flags;
        std::array<std::uint8_t, 428> reserved;
        std::uint32_t crc32c;
    };
    static_assert(sizeof(CommitRecord) == 512);
    static_assert(__builtin_offsetof(CommitRecord, extentCount) == 40);
    static_assert(__builtin_offsetof(CommitRecord, fileLength) == 56);
    static_assert(__builtin_offsetof(CommitRecord, flags) == 72);
    static_assert(__builtin_offsetof(CommitRecord, crc32c) == 508);

    struct FileHeader {
        std::array<char, 8> magic;
        std::uint16_t formatVersion;
        std::uint16_t headerPages;
        std::uint32_t formatFlags;
        std::uint32_t pageSize;
        std::uint32_t rowSize;
        std::uint16_t rowAlign;
        std::uint16_t columnCount;
        std::uint16_t orderedColumn;
        std::uint16_t regionCount;
        std::uint32_t schemaVersion;
        std::uint32_t reserved0;
        std::uint64_t schemaFingerprint;
        std::uint32_t schemaBlobLength;
        std::uint32_t minExtentBytes;
        std::uint64_t createdUnixNanos;
        std::array<std::uint8_t, 16> fileUuid;
        std::uint32_t migratedFromVersion;
        std::array<std::uint8_t, 44> reserved1;
        std::array<std::array<Extent, maxExtents>, regionSlots> extents;
        std::array<std::uint8_t, 896> reserved2;
        std::array<CommitRecord, 2> commits;
    };
    static_assert(sizeof(FileHeader) == pageSize);
    static_assert(__builtin_offsetof(FileHeader, schemaVersion) == 32);
    static_assert(__builtin_offsetof(FileHeader, schemaFingerprint) == 40);
    static_assert(__builtin_offsetof(FileHeader, createdUnixNanos) == 56);
    static_assert(__builtin_offsetof(FileHeader, fileUuid) == 64);
    static_assert(__builtin_offsetof(FileHeader, migratedFromVersion) == 80);
    static_assert(__builtin_offsetof(FileHeader, extents) == 128);
    static_assert(__builtin_offsetof(FileHeader, reserved2) == 2176);
    static_assert(__builtin_offsetof(FileHeader, commits) == 3072);

    struct SchemaBlobHeader {
        std::array<char, 4> magic;
        std::uint16_t schemaFormat;
        std::uint16_t columnCount;
        std::uint32_t rowSize;
        std::uint32_t namePoolLength;
    };
    static_assert(sizeof(SchemaBlobHeader) == 16);

    struct ColumnDescriptor {
        std::uint16_t offset;
        std::uint16_t size;
        Kind kind;
        std::uint8_t flags;
        std::uint16_t presenceOffset;
        std::uint16_t nameOffset;
        std::uint16_t nameLength;
        std::uint32_t paramNum;   // chrono period numerator; element alignment for blobs; N for fixed bytes
        std::uint32_t paramDen;   // chrono period denominator; element size for blobs
        std::uint32_t variant;    // element kind for blobs; clock code for time points; 0 otherwise
    };
    static_assert(sizeof(ColumnDescriptor) == 24);
    inline constexpr std::uint8_t maxKind = static_cast<std::uint8_t>(Kind::FixedBytes);

    // The 16-byte handle of a variable-length column (Umbra / "German string").
    // Bytes 4..16 are the first 12 bytes of the value, so a value of up to 12
    // bytes is entirely inline and contiguous at (handle + 4). Longer values
    // keep their 4-byte prefix inline and store an offset into the heap region
    // where the whole value, prefix included, lives contiguously.
    struct Var {
        std::uint32_t length;
        std::uint32_t prefix;
        std::uint64_t tailOrOffset;
    };
    static_assert(sizeof(Var) == 16 && alignof(Var) == 8);

    constexpr std::size_t KindSize(Kind kind) noexcept {
        switch (kind) {
            case Kind::Flags: case Kind::Bool: case Kind::Int8: case Kind::UInt8: return 1;
            case Kind::Int16: case Kind::UInt16: return 2;
            case Kind::Int32: case Kind::UInt32: case Kind::Float32: return 4;
            case Kind::Int64: case Kind::UInt64: case Kind::Float64: case Kind::TimePoint: case Kind::Duration: return 8;
            case Kind::String: case Kind::Blob: return sizeof(Var);
            case Kind::FixedBytes: return 0;  // size comes from the descriptor
        }
        return 0;
    }

    // FNV-1a 64. Chosen over std::hash because the result is part of the file
    // format and must not change with the standard library.
    inline constexpr std::uint64_t fnvOffset = 14695981039346656037ull;
    inline constexpr std::uint64_t fnvPrime = 1099511628211ull;

    constexpr std::uint64_t Fnv1a(std::uint64_t hash, std::string_view bytes) noexcept {
        for (const char c : bytes) {
            hash ^= static_cast<std::uint8_t>(c);
            hash *= fnvPrime;
        }
        return hash;
    }

    constexpr std::uint64_t Fnv1a(std::uint64_t hash, std::uint64_t value) noexcept {
        for (std::size_t i = 0; i < 8; i++) {
            hash ^= (value >> (8 * i)) & 0xFF;
            hash *= fnvPrime;
        }
        return hash;
    }

    // CRC-32C (Castagnoli), table driven. 508 bytes per commit is too little
    // for the hardware instruction to matter.
    namespace Detail {
        consteval std::array<std::uint32_t, 256> MakeCrcTable() {
            std::array<std::uint32_t, 256> table{};
            for (std::uint32_t i = 0; i < 256; i++) {
                std::uint32_t c = i;
                for (std::int32_t k = 0; k < 8; k++) c = (c & 1) ? (0x82F63B78u ^ (c >> 1)) : (c >> 1);
                table[i] = c;
            }
            return table;
        }
        inline constexpr std::array<std::uint32_t, 256> crcTable = MakeCrcTable();
    }

    constexpr std::uint32_t Crc32c(std::span<const std::byte> bytes) noexcept {
        std::uint32_t crc = 0xFFFFFFFFu;
        for (const std::byte b : bytes) crc = Detail::crcTable[(crc ^ static_cast<std::uint8_t>(b)) & 0xFF] ^ (crc >> 8);
        return crc ^ 0xFFFFFFFFu;
    }

    inline std::uint32_t CommitCrc(const CommitRecord& record) noexcept {
        return Crc32c(std::as_bytes(std::span(&record, 1)).first(__builtin_offsetof(CommitRecord, crc32c)));
    }

    inline bool CommitValid(const CommitRecord& record) noexcept {
        return record.commitSeq != 0 && CommitCrc(record) == record.crc32c;
    }

    // A parsed schema blob: what the file says its columns are. Column 0 is
    // always the flags byte at offset 0.
    struct SchemaInfo {
        std::vector<ColumnDescriptor> columns;
        std::string namePool;
        std::uint32_t rowSize = 0;
        std::uint16_t rowAlign = 1;

        std::string_view Name(std::size_t column) const noexcept {
            const ColumnDescriptor& d = columns[column];
            return std::string_view(namePool).substr(d.nameOffset, d.nameLength);
        }

        bool HasHeap() const noexcept {
            return std::ranges::any_of(columns, [](const ColumnDescriptor& d) { return d.kind == Kind::String || d.kind == Kind::Blob; });
        }

        std::vector<std::byte> Serialize() const {
            std::vector<std::byte> out(sizeof(SchemaBlobHeader) + columns.size() * sizeof(ColumnDescriptor) + namePool.size());
            const SchemaBlobHeader header{ schemaMagic, 1, static_cast<std::uint16_t>(columns.size()), rowSize, static_cast<std::uint32_t>(namePool.size()) };
            std::memcpy(out.data(), &header, sizeof(header));
            std::memcpy(out.data() + sizeof(header), columns.data(), columns.size() * sizeof(ColumnDescriptor));
            std::memcpy(out.data() + sizeof(header) + columns.size() * sizeof(ColumnDescriptor), namePool.data(), namePool.size());
            return out;
        }

        // Returns false when the bytes are not a well-formed blob. Every field
        // of every descriptor is range-checked here, because the engine and
        // DynamicRow index rows with them without further checks.
        static bool Parse(std::span<const std::byte> bytes, std::uint16_t rowAlign, SchemaInfo& out) {
            if (bytes.size() < sizeof(SchemaBlobHeader)) return false;
            SchemaBlobHeader header;
            std::memcpy(&header, bytes.data(), sizeof(header));
            if (header.magic != schemaMagic || header.schemaFormat != 1 || header.columnCount == 0 || header.rowSize == 0) return false;
            const std::size_t descriptorBytes = std::size_t(header.columnCount) * sizeof(ColumnDescriptor);
            if (bytes.size() < sizeof(header) + descriptorBytes + header.namePoolLength) return false;
            out.columns.resize(header.columnCount);
            std::memcpy(out.columns.data(), bytes.data() + sizeof(header), descriptorBytes);
            out.namePool.assign(reinterpret_cast<const char*>(bytes.data() + sizeof(header) + descriptorBytes), header.namePoolLength);
            out.rowSize = header.rowSize;
            out.rowAlign = rowAlign;
            for (std::size_t i = 0; i < out.columns.size(); i++) {
                const ColumnDescriptor& d = out.columns[i];
                if (static_cast<std::uint8_t>(d.kind) > maxKind) return false;
                if (std::size_t(d.nameOffset) + d.nameLength > out.namePool.size()) return false;
                if (d.size == 0 || std::size_t(d.offset) + d.size > out.rowSize) return false;
                if (d.kind != Kind::FixedBytes && KindSize(d.kind) != d.size) return false;
                if (d.presenceOffset != noColumn && d.presenceOffset >= out.rowSize) return false;
                if ((d.flags & columnFlagOptional) != 0 && d.presenceOffset == noColumn) return false;
                if (i == 0 && (d.kind != Kind::Flags || d.offset != 0)) return false;
                if (i != 0 && d.kind == Kind::Flags) return false;
            }
            return true;
        }

        // Canonical hash of the layout; the same function runs at compile time
        // on Layout<T> so the two sides agree byte for byte. The ordered
        // column is part of it: changing `orderedBy` changes what the file
        // promises about its rows.
        std::uint64_t Fingerprint(std::uint16_t orderedColumn) const noexcept {
            std::uint64_t hash = Fnv1a(fnvOffset, "crafter-db-v1");
            hash = Fnv1a(hash, rowSize);
            hash = Fnv1a(hash, rowAlign);
            hash = Fnv1a(hash, orderedColumn);
            for (std::size_t i = 0; i < columns.size(); i++) {
                const ColumnDescriptor& d = columns[i];
                hash = Fnv1a(hash, Name(i));
                hash = Fnv1a(hash, static_cast<std::uint64_t>(d.kind));
                hash = Fnv1a(hash, d.offset);
                hash = Fnv1a(hash, d.size);
                hash = Fnv1a(hash, d.flags);
                hash = Fnv1a(hash, d.paramNum);
                hash = Fnv1a(hash, d.paramDen);
                hash = Fnv1a(hash, d.variant);
            }
            return hash;
        }
    };
}