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

// Table<T>: the user-facing handle. Opening creates, validates and migrates;
// writes go through the typed encoders into the mapped file; reads are
// zero-copy Row<T> views; queries are lambdas or column expressions.
export module Crafter.DB:Table;
import std;
import :Error;
import :Reflect;
import :Format;
import :Formatter;
import :Schema;
import :File;
import :Storage;
import :Codec;
import :Row;
import :Query;
import :Migration;

export namespace Crafter::DB {
    template<class T>
    class Table {
        static_assert(CheckTable<T>());

    public:
        using Value = T;

        explicit Table(const std::filesystem::path& path, OpenOptions options = {}) : path(path), options(options) {
            Open();
        }

        Table(Table&&) noexcept = default;
        Table& operator=(Table&&) noexcept = default;
        Table(const Table&) = delete;
        Table& operator=(const Table&) = delete;

        // --- information ------------------------------------------------------

        const std::filesystem::path& Path() const noexcept { return path; }
        TableStats Stats() const noexcept { return file->Stats(); }
        // Live rows.
        std::uint64_t Size() const noexcept { return file->LiveCount(); }
        // Slots, tombstones included: the range of valid RowIds.
        std::uint64_t SlotCount() const noexcept { return file->SlotCount(); }
        bool Empty() const noexcept { return Size() == 0; }
        bool ReadOnly() const noexcept { return !file->Writable(); }
        TableFile& File() noexcept { return *file; }
        const TableFile& File() const noexcept { return *file; }

        // --- writing ----------------------------------------------------------

        RowId Append(const T& row) {
            const RowId id = AppendRows<T>(*file, std::span<const T>(&row, 1));
            AutoFlush();
            return id;
        }

        template<std::ranges::input_range R> requires std::convertible_to<std::ranges::range_reference_t<R>, const T&>
        RowId Append(R&& rows) {
            const RowId id = AppendRows<T>(*file, std::forward<R>(rows));
            AutoFlush();
            return id;
        }

        // One argument per member, in declaration order, encoded straight into
        // the columns (a std::string_view is enough for a std::string member).
        template<class... Args>
        RowId Emplace(const Args&... args) {
            static_assert(sizeof...(Args) == memberCount<T>, "Crafter.DB: Emplace takes exactly one argument per member of the table struct, in declaration order");
            file->RequireWritable();
            const RowId id = file->SlotCount();
            if constexpr (HasOrderedBy<T>) {
                std::optional<std::uint64_t> before;
                if (id != 0) before = id - 1;
                const OrderedKey<T> key = static_cast<OrderedKey<T>>(std::get<Layout<T>::orderedMember>(std::tie(args...)));
                CheckOrder<T>(*file, key, before, std::nullopt);
            }
            std::byte* out = file->BeginAppend(1);
            try {
                EmplaceRow<T>(out, file->Writer(), args...);
            } catch (...) {
                file->AbortAppend();
                throw;
            }
            file->CommitAppend(1);
            AutoFlush();
            return id;
        }

        // Replaces a live row in place. The new heap bytes are committed to
        // disk before the row starts pointing at them, so a crash leaves
        // either the old row or the new one, never a dangling handle. The row
        // bytes themselves are not atomic against a power loss mid-write.
        void Update(RowId id, const T& row) {
            file->RequireWritable();
            RequireLive(id);
            if constexpr (HasOrderedBy<T>) CheckNeighbours(id, row.*T::orderedBy);
            alignas(8) std::array<std::byte, Layout<T>::rowSize> buffer;
            HeapWriter& heap = file->Writer();
            const std::uint64_t heapBefore = heap.Cursor();
            try {
                EncodeRow<T>(row, buffer.data(), heap);
            } catch (...) {
                file->AbortAppend();
                throw;
            }
            if (heap.Cursor() != heapBefore) {
                file->CommitAppend(0);
                file->Flush(false);
            }
            std::byte* slot = file->MutableSlot(id);
            file->AddHeapGarbage(file->HeapBytesOf(slot));
            std::memcpy(slot, buffer.data(), Layout<T>::rowSize);
            AutoFlush();
        }

        // Sets one member of a live row, with the same durability order as
        // Update. Only that member's columns are written, so a concurrent
        // Erase of the row is not undone.
        template<auto Ptr, class V> requires std::is_same_v<MemberClass<Ptr>, T>
        void Set(RowId id, const V& value) {
            constexpr std::size_t index = MemberIndex<T, Ptr>();
            static_assert(index != noMember, "Crafter.DB: Set needs a pointer to a direct member of the table struct");
            constexpr std::size_t first = LeafStart<T, index>();
            constexpr std::size_t count = LeafCount<MemberType<index, T>>();
            file->RequireWritable();
            RequireLive(id);
            if constexpr (HasOrderedBy<T>) {
                if constexpr (SameMemberPointer<Ptr, T::orderedBy>()) CheckNeighbours(id, static_cast<OrderedKey<T>>(value));
            }
            std::byte* slot = file->MutableSlot(id);
            alignas(8) std::array<std::byte, Layout<T>::rowSize> buffer;
            std::memcpy(buffer.data(), slot, Layout<T>::rowSize);
            HeapWriter& heap = file->Writer();
            const std::uint64_t heapBefore = heap.Cursor();
            try {
                EncodeMember<T, index>(buffer.data(), heap, value);
            } catch (...) {
                file->AbortAppend();
                throw;
            }
            if (heap.Cursor() != heapBefore) {
                file->CommitAppend(0);
                file->Flush(false);
            }
            file->AddHeapGarbage(MemberHeapBytes<T, index>(slot));
            CopyLeaves<T, first, count>(slot, buffer.data());
            AutoFlush();
        }

        template<auto Ptr, class V> requires std::is_same_v<MemberClass<Ptr>, T>
        void Set(RowId id, Column<Ptr>, const V& value) {
            Set<Ptr>(id, value);
        }

        // Tombstones the row. Returns false when it was already erased.
        bool Erase(RowId id) {
            file->RequireWritable();
            RequireSlot(id);
            const bool erased = file->MarkDeleted(id);
            if (erased) AutoFlush();
            return erased;
        }

        template<class Pred>
        std::uint64_t EraseIf(Pred pred) {
            file->RequireWritable();
            std::uint64_t erased = 0;
            const std::uint64_t slots = file->SlotCount();
            for (std::uint64_t i = 0; i < slots; i++) {
                const Row<T> row(file->Slot(i), file.get());
                if (row.Erased() || !pred(row)) continue;
                if (file->MarkDeleted(i)) erased++;
            }
            if (erased != 0) AutoFlush();
            return erased;
        }

        void Flush() { file->Flush(false); }

        // Flushes and closes now, reporting failure as an exception. The
        // destructor does the same but cannot report, so call this when a
        // failed final flush must not go unnoticed. The table is unusable after.
        void Close() { file->Close(); }

        // Consistency scan of the whole file (see TableFile::Verify).
        VerifyResult Verify() const { return file->Verify(); }

        // Rewrites the file without tombstones and heap garbage. Invalidates
        // every RowId, Row and view taken before the call.
        void Compact() {
            file->RequireWritable();
            const TableStats stats = file->Stats();
            Rebuild([&](TableFile& target) { file->CompactInto(target); }, stats.liveCount * Layout<T>::rowSize, stats.heapUsed - std::min(stats.heapUsed, stats.heapGarbage));
        }

        // Removes every row. Invalidates RowIds, Rows and views.
        void Clear() {
            file->RequireWritable();
            Rebuild([](TableFile&) {}, 0, 0);
        }

        // --- reading ----------------------------------------------------------

        Row<T> operator[](RowId id) const noexcept { return { file->Slot(id), file.get() }; }

        Row<T> At(RowId id) const {
            RequireLive(id);
            return (*this)[id];
        }

        bool Contains(RowId id) const noexcept { return id < file->SlotCount() && !file->Deleted(id); }

        Row<T> Front() const {
            for (Row<T> row : Live()) return row;
            throw std::out_of_range("Table::Front on an empty table");
        }

        Row<T> Back() const {
            std::uint64_t index = file->SlotCount();
            while (index != 0) {
                index--;
                if (!file->Deleted(index)) return (*this)[index];
            }
            throw std::out_of_range("Table::Back on an empty table");
        }

        LiveRange<T> Live() const noexcept { return { file.get(), 0, file->SlotCount() }; }
        LiveIterator<T> begin() const noexcept { return Live().begin(); }
        std::default_sentinel_t end() const noexcept { return {}; }
        SlotView<T> Slots() const noexcept { return { file.get(), 0, file->SlotCount() }; }

        std::vector<T> Materialize() const {
            std::vector<T> out;
            out.reserve(Size());
            for (Row<T> row : Live()) out.push_back(row.Materialize());
            return out;
        }

        // Live rows of the slot range [first, last).
        Query<T, LiveRange<T>> Rows(RowId first, RowId last) const noexcept {
            const std::uint64_t end = std::min<std::uint64_t>(last, file->SlotCount());
            return Query<T, LiveRange<T>>(LiveRange<T>(file.get(), std::min(first, end), end));
        }

        // Lazy filtered view. For a column expression on an orderedBy table the
        // bounds on the ordered column are turned into a binary search first.
        template<class Pred>
        auto Where(Pred pred) const {
            std::uint64_t first = 0;
            std::uint64_t last = file->SlotCount();
            if constexpr (HasOrderedBy<T> && Expression<Pred>) {
                Bounds<OrderedKey<T>> bounds;
                CollectBounds<T>(pred, bounds);
                if (bounds.lower) first = bounds.lowerExclusive ? UpperBoundSlot<T>(*file, *bounds.lower, 0, last) : LowerBoundSlot<T>(*file, *bounds.lower, 0, last);
                if (bounds.upper) last = bounds.upperExclusive ? LowerBoundSlot<T>(*file, *bounds.upper, first, last) : UpperBoundSlot<T>(*file, *bounds.upper, first, last);
                if (last < first) last = first;
            }
            return Query<T, std::ranges::filter_view<LiveRange<T>, Pred>>(std::ranges::filter_view<LiveRange<T>, Pred>(LiveRange<T>(file.get(), first, last), std::move(pred)));
        }

        template<auto Ptr> requires std::is_same_v<MemberClass<Ptr>, T>
        auto Select() const {
            return std::views::transform(Live(), [](Row<T> row) { return row.template Get<Ptr>(); });
        }

        // --- ordered column ---------------------------------------------------

        template<auto Ptr>
        RowId LowerBound(const MemberTypeOf<Ptr>& key) const {
            RequireOrdered<Ptr>();
            return LowerBoundSlot<T>(*file, key, 0, file->SlotCount());
        }

        template<auto Ptr>
        RowId UpperBound(const MemberTypeOf<Ptr>& key) const {
            RequireOrdered<Ptr>();
            return UpperBoundSlot<T>(*file, key, 0, file->SlotCount());
        }

        // Live rows with from <= key <= to.
        template<auto Ptr>
        Query<T, LiveRange<T>> Range(const MemberTypeOf<Ptr>& from, const MemberTypeOf<Ptr>& to) const {
            RequireOrdered<Ptr>();
            const std::uint64_t slots = file->SlotCount();
            const std::uint64_t first = LowerBoundSlot<T>(*file, from, 0, slots);
            const std::uint64_t last = UpperBoundSlot<T>(*file, to, first, slots);
            return Rows(first, last);
        }

        // key >= from
        template<auto Ptr>
        Query<T, LiveRange<T>> From(const MemberTypeOf<Ptr>& from) const {
            RequireOrdered<Ptr>();
            return Rows(LowerBoundSlot<T>(*file, from, 0, file->SlotCount()), file->SlotCount());
        }

        // key > after
        template<auto Ptr>
        Query<T, LiveRange<T>> After(const MemberTypeOf<Ptr>& after) const {
            RequireOrdered<Ptr>();
            return Rows(UpperBoundSlot<T>(*file, after, 0, file->SlotCount()), file->SlotCount());
        }

        // key <= to
        template<auto Ptr>
        Query<T, LiveRange<T>> Until(const MemberTypeOf<Ptr>& to) const {
            RequireOrdered<Ptr>();
            return Rows(0, UpperBoundSlot<T>(*file, to, 0, file->SlotCount()));
        }

        // key < before
        template<auto Ptr>
        Query<T, LiveRange<T>> Before(const MemberTypeOf<Ptr>& before) const {
            RequireOrdered<Ptr>();
            return Rows(0, LowerBoundSlot<T>(*file, before, 0, file->SlotCount()));
        }

    private:
        template<auto Ptr>
        static consteval void RequireOrdered() {
            static_assert(HasOrderedBy<T>, "Crafter.DB: Range/From/After/Until/Before/LowerBound/UpperBound need `static constexpr auto orderedBy = &T::member;` on the table struct");
            if constexpr (HasOrderedBy<T>) {
                static_assert(SameMemberPointer<Ptr, T::orderedBy>(), "Crafter.DB: binary search is only available on the member named by `orderedBy`; use Where() for other columns");
            }
        }

        void CreateFile() {
            CreateParams params;
            params.schema = MakeSchemaInfo<T>();
            params.schemaVersion = Layout<T>::version;
            params.fingerprint = Layout<T>::fingerprint;
            params.orderedColumn = Layout<T>::orderedColumn;
            TableFile::Create(path, params, options);   // IOError(file_exists) when the file is already there
        }

        // One open() does the work: the file's own header says whether it is
        // current (done) or older (migrate, then reopen). A missing file is
        // created with a rename that never replaces, so two processes racing
        // to create the same table end up sharing one file.
        void Open() {
            if (options.mode == OpenMode::CreateNew) {
                CreateFile();
                file = std::make_unique<TableFile>(path, options);
                return;
            }
            std::unique_ptr<TableFile> opened;
            for (std::int32_t attempt = 0; !opened; attempt++) {
                try {
                    opened = std::make_unique<TableFile>(path, options);
                } catch (const IOError& error) {
                    if (error.code != std::errc::no_such_file_or_directory || options.mode == OpenMode::OpenExisting || options.readOnly || attempt != 0) throw;
                    try {
                        CreateFile();
                        file = std::make_unique<TableFile>(path, options);
                        return;
                    } catch (const IOError& created) {
                        if (created.code != std::errc::file_exists) throw;
                        // Lost the race: another process created it; open theirs.
                    }
                }
            }
            if (opened->Header().schemaVersion != Layout<T>::version) {
                Inspection info;
                info.header = opened->Header();
                info.schema = opened->Schema();
                opened.reset();   // the migration takes the lock itself
                MigrateTable<T>(path, info, options);
                opened = std::make_unique<TableFile>(path, options);
                if (opened->Header().schemaVersion != Layout<T>::version) {
                    throw SchemaError(std::format("{}: version {} on disk after migration, expected {}", path.string(), opened->Header().schemaVersion, Layout<T>::version));
                }
            }
            RequireFingerprint<T>(path, opened->Header(), opened->Schema());
            file = std::move(opened);
        }

        void AutoFlush() {
            if (options.durability == Durability::EveryAppend) file->Flush(false);
        }

        void RequireSlot(RowId id) const {
            if (id >= file->SlotCount()) throw std::out_of_range(std::format("RowId {} is outside the table ({} slots)", id, file->SlotCount()));
        }

        void RequireLive(RowId id) const {
            RequireSlot(id);
            if (file->Deleted(id)) throw std::out_of_range(std::format("RowId {} was erased", id));
        }

        template<class Key>
        void CheckNeighbours(RowId id, const Key& key) {
            std::optional<std::uint64_t> before;
            std::optional<std::uint64_t> after;
            if (id != 0) before = id - 1;
            if (id + 1 < file->SlotCount()) after = id + 1;
            CheckOrder<T>(*file, key, before, after);
        }

        template<class Fill>
        void Rebuild(Fill&& fill, std::uint64_t initialRowBytes, std::uint64_t initialHeapBytes) {
            CreateParams params;
            params.schema = file->Schema();
            params.schemaVersion = Layout<T>::version;
            params.fingerprint = Layout<T>::fingerprint;
            params.orderedColumn = Layout<T>::orderedColumn;
            params.initialRowBytes = initialRowBytes;
            params.initialHeapBytes = initialHeapBytes;
            params.migratedFromVersion = file->Header().migratedFromVersion;
            OpenOptions targetOptions = options;
            targetOptions.createDirectories = false;
            const std::filesystem::path temp = TableFile::TempPath(path);
            TableFile::Create(temp, params, targetOptions);
            std::unique_ptr<TableFile> target;
            try {
                target = std::make_unique<TableFile>(temp, targetOptions);
                fill(*target);
                target->Flush(false);
                TableFile::Replace(path, temp);
            } catch (...) {
                target.reset();
                Os::Remove(temp);
                throw;
            }
            target->SetPath(path);
            // The old inode is unlinked; a failing final flush of it is
            // irrelevant and must not leave this Table without a file.
            try {
                file->Close();
            } catch (...) {
            }
            file = std::move(target);
        }

        std::unique_ptr<TableFile> file;
        std::filesystem::path path;
        OpenOptions options;
    };
}