// SPDX-FileCopyrightText: © 2026 Vladimir Zorin <vladimir@deviant.guru>
// SPDX-License-Identifier: LicenseRef-OWL-1.0-or-later
// Licensed under OWL v1.0+. See LICENSE.

#include "mneme.h"

#include <errno.h>
#include <stdlib.h>
#include <string.h>
#include <sys/file.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <unistd.h>

/* Free all txn-owned arrays and zero the counts/caps so a stale
   handle that slips past the bindings fails cleanly instead of
   indexing freed memory. */
static void txn_free_arrays(mneme_txn_t *txn) {
    free(txn->dirty);
    for (int i = 0; i < txn->harvested_count; i++)
        free(txn->harvested[i].pages);
    free(txn->harvested);
    free(txn->reuse_pages);
    free(txn->consumed_reuse_pages);
    free(txn->fresh_free_pages);
    free(txn->pending_retire);
    txn->dirty                = NULL;
    txn->dirty_count          = 0;
    txn->dirty_cap            = 0;
    txn->harvested            = NULL;
    txn->harvested_count      = 0;
    txn->harvested_cap        = 0;
    txn->reuse_pages          = NULL;
    txn->reuse_count          = 0;
    txn->reuse_cap            = 0;
    txn->consumed_reuse_pages = NULL;
    txn->consumed_reuse_count = 0;
    txn->consumed_reuse_cap   = 0;
    txn->fresh_free_pages     = NULL;
    txn->fresh_free_count     = 0;
    txn->fresh_free_cap       = 0;
    txn->pending_retire       = NULL;
    txn->pending_retire_count = 0;
    txn->pending_retire_cap   = 0;
}

int mneme_txn_begin(mneme_db_t *db, int readonly, mneme_txn_t **out) {
    if (!db || db->closed)
        return -1;
    if (!readonly && (db->flags & MNEME_OPEN_READONLY))
        return -1;

    mneme_txn_t *txn = (mneme_txn_t *)calloc(1, sizeof(*txn));
    if (!txn)
        return -1;

    txn->db             = db;
    txn->readonly       = readonly;
    txn->reader_lock_fd = -1;

    if (readonly) {
        mneme_db_t snap = *db;
        snap.map        = NULL;
        snap.map_size   = 0;
        for (;;) {
            if (mneme_meta_refresh(db) < 0) {
                free(txn);
                return -1;
            }
            txn->reader_txn_id = db->meta.txn_id;
            if (mneme_reader_lock_acquire(txn) < 0) {
                free(txn);
                return -1;
            }
            if (mneme_meta_refresh(db) < 0) {
                mneme_reader_lock_release(txn);
                free(txn);
                return -1;
            }
            if (db->meta.txn_id != txn->reader_txn_id) {
                mneme_reader_lock_release(txn);
                continue;
            }
            snap.meta       = db->meta;
            snap.page_count = db->meta.page_count;
            if (mneme_mmap_refresh(&snap) < 0) {
                mneme_reader_lock_release(txn);
                free(txn);
                return -1;
            }
            txn->db = (mneme_db_t *)malloc(sizeof(*txn->db));
            if (!txn->db) {
                if (snap.map && snap.map_size > 0)
                    munmap(snap.map, snap.map_size);
                mneme_reader_lock_release(txn);
                free(txn);
                return -1;
            }
            *txn->db  = snap;
            txn->meta = txn->db->meta;
            break;
        }
    } else {
        if (db->write_txn_active) {
            /* flock is per-OFD: a nested LOCK_EX on the same fd would
               succeed and the inner commit's LOCK_UN would silently drop
               this handle's exclusivity. Refuse instead. */
            free(txn);
            return -1;
        }
        if (flock(db->fd, LOCK_EX) < 0) {
            free(txn);
            return -1;
        }
        struct stat st;
        if (fstat(db->fd, &st) == 0) {
            uint64_t file_pages = (uint64_t)st.st_size / MNEME_PAGE_SIZE;
            if (file_pages > db->page_count) {
                db->page_count = file_pages;
                if (mneme_mmap_refresh(db) < 0) {
                    flock(db->fd, LOCK_UN);
                    free(txn);
                    return -1;
                }
            }
        }
        if (mneme_meta_refresh(db) < 0) {
            flock(db->fd, LOCK_UN);
            free(txn);
            return -1;
        }
        txn->meta          = db->meta;
        txn->commit_txn_id = db->meta.txn_id + 1;
        txn->dirty_cap     = MNEME_DIRTY_INIT;
        txn->dirty         = (mneme_dirty_page_t *)calloc((size_t)txn->dirty_cap, sizeof(mneme_dirty_page_t));
        if (!txn->dirty) {
            flock(db->fd, LOCK_UN);
            free(txn);
            return -1;
        }
        if (mneme_retire_harvest_safe(txn, MNEME_RETIRE_HARVEST_BUDGET) < 0) {
            flock(db->fd, LOCK_UN);
            txn_free_arrays(txn);
            free(txn);
            return -1;
        }
        db->write_txn_active = 1;
    }

    txn->committed = 0;
    *out           = txn;
    return 0;
}

int mneme_txn_commit(mneme_txn_t *txn) {
    if (!txn || txn->committed)
        return -1;
    if (txn->readonly) {
        if (txn->db && txn->db->map && txn->db->map_size > 0)
            munmap(txn->db->map, txn->db->map_size);
        free(txn->db);
        txn->db = NULL;
        mneme_reader_lock_release(txn);
        txn->committed = 1;
        return 0;
    }

    int fd = txn->db->fd;
    if (mneme_retire_apply_harvested(txn) < 0)
        goto fail;
    if (mneme_retire_persist(txn) < 0)
        goto fail;

    off_t new_size = (off_t)txn->meta.page_count * MNEME_PAGE_SIZE;
    off_t cur_size = (off_t)txn->db->page_count * MNEME_PAGE_SIZE;
    if (new_size > cur_size && ftruncate(fd, new_size) < 0)
        goto fail;

    for (int i = 0; i < txn->dirty_count; i++) {
        mneme_dirty_page_t *dp = &txn->dirty[i];
        off_t off              = (off_t)dp->pgno * MNEME_PAGE_SIZE;
        if (pwrite(fd, dp->data, MNEME_PAGE_SIZE, off) != MNEME_PAGE_SIZE)
            goto fail;
    }

    if (!(txn->db->flags & MNEME_OPEN_NOSYNC) && fdatasync(fd) < 0)
        goto fail;

    txn->meta.txn_id   = txn->commit_txn_id;
    txn->meta.checksum = mneme_checksum(&txn->meta, offsetof(mneme_meta_t, checksum));
    uint32_t meta_pgno = (txn->meta.txn_id % 2 == 0) ? 0 : 1;

    uint8_t meta_buf[MNEME_PAGE_SIZE];
    memset(meta_buf, 0, MNEME_PAGE_SIZE);
    memcpy(meta_buf, &txn->meta, sizeof(txn->meta));
    if (pwrite(fd, meta_buf, MNEME_PAGE_SIZE, (off_t)meta_pgno * MNEME_PAGE_SIZE) != MNEME_PAGE_SIZE)
        goto fail;

    if (!(txn->db->flags & MNEME_OPEN_NOSYNC) && fdatasync(fd) < 0)
        goto fail;

    txn->db->meta       = txn->meta;
    txn->db->page_count = txn->meta.page_count;
    if (mneme_mmap_refresh(txn->db) < 0) {
        /* The commit is durable on disk but the handle can no longer map
           it. Tear the handle down fully here (closing the fd also drops
           the flock); mneme_db_close() early-returns on closed handles,
           so this is the only place these resources get released. */
        mneme_db_t *db = txn->db;
        if (db->map && db->map_size > 0) {
            munmap(db->map, db->map_size);
            db->map      = NULL;
            db->map_size = 0;
        }
        if (db->probe_fd >= 0) {
            close(db->probe_fd);
            db->probe_fd = -1;
        }
        close(db->fd);
        db->fd = -1;
        free(db->path);
        db->path             = NULL;
        db->write_txn_active = 0;
        db->closed           = 1;
        txn_free_arrays(txn);
        txn->committed = 1;
        return -2;
    }

    txn->db->write_txn_active = 0;
    flock(txn->db->fd, LOCK_UN);
    txn_free_arrays(txn);
    txn->committed = 1;
    return 0;

fail:
    txn->db->write_txn_active = 0;
    flock(txn->db->fd, LOCK_UN);
    txn_free_arrays(txn);
    txn->committed = 1;
    return -1;
}

void mneme_txn_abort(mneme_txn_t *txn) {
    if (!txn || txn->committed)
        return;
    if (txn->readonly) {
        if (txn->db && txn->db->map && txn->db->map_size > 0)
            munmap(txn->db->map, txn->db->map_size);
        free(txn->db);
        txn->db = NULL;
        mneme_reader_lock_release(txn);
    } else {
        txn->db->write_txn_active = 0;
        flock(txn->db->fd, LOCK_UN);
        txn_free_arrays(txn);
    }
    txn->committed = 1;
}