// SPDX-License-Identifier: Apache-2.0
//
// SchFS copy-on-write B-tree — see btree.h for the contract.
//
// Leaf layout follows btrfs: the item array grows forward from just past the
// header while the values grow backward from the end of the block, so a leaf is
// full when the two meet. That lets one node hold a few large values or many
// small ones without fixing a record size in the format.
//
// Every mutation rewrites the path from the touched leaf up to the root. A node
// that already belongs to the transaction group being built is rewritten in
// place; anything older is copied first, so nothing reachable from a published
// root is ever modified and a crash at any instant leaves the previous tree
// intact.

#include "btree.h"

// memmove throughout rather than memcpy: a caller may legitimately hand
// leaf_put a value that points into the very block being written (the cursor
// hands out pointers into its own buffer), and the kernel declares memmove
// but not memcpy.
#ifdef SCHFS_HOSTED
#include <string.h>
#else
#include "types.h"
#include "defs.h"
#endif

/* ------------------------------------------------------------------ */
/* CRC32c (Castagnoli), table-free so this stays usable very early.    */
/* ------------------------------------------------------------------ */

uint32_t
schfs_crc32c(uint32_t crc, const void *buf, unsigned len)
{
    const uint8_t *p = (const uint8_t *)buf;
    crc = ~crc;
    while (len--) {
        crc ^= *p++;
        for (int i = 0; i < 8; i++)
            crc = (crc >> 1) ^ (0x82F63B78u & (uint32_t)(-(int32_t)(crc & 1)));
    }
    return ~crc;
}

/* ------------------------------------------------------------------ */
/* Keys                                                                */
/* ------------------------------------------------------------------ */

int
schfs_key_cmp(const struct schfs_key *a, const struct schfs_key *b)
{
    if (a->objectid < b->objectid) return -1;
    if (a->objectid > b->objectid) return  1;
    if (a->type     < b->type)     return -1;
    if (a->type     > b->type)     return  1;
    if (a->offset   < b->offset)   return -1;
    if (a->offset   > b->offset)   return  1;
    return 0;
}

/* ------------------------------------------------------------------ */
/* Node accessors                                                      */
/* ------------------------------------------------------------------ */

/* Short names for the accessors btree.h publishes. */
#define HDR(b)    SCHFS_HDR(b)
#define KPTRS(b)  SCHFS_KPTRS(b)
#define ITEMS(b)  SCHFS_ITEMS(b)

// Bytes still usable in a leaf: the gap between the end of the item array and
// the start of the value area.
static int
leaf_free_space(const void *blk)
{
    const struct schfs_node_hdr *h = (const struct schfs_node_hdr *)blk;
    const struct schfs_item *it = (const struct schfs_item *)
        ((const char *)blk + sizeof(*h));
    int items_end = (int)sizeof(*h) + h->nritems * (int)sizeof(struct schfs_item);
    int data_start = SCHFS_BLKSIZE;
    for (int i = 0; i < h->nritems; i++)
        if ((int)it[i].offset < data_start)
            data_start = (int)it[i].offset;
    return data_start - items_end;
}

// First key held anywhere in this subtree's node. The parent files a child
// under exactly this key; see the note on the separator invariant below.
static void
node_first_key(const void *blk, struct schfs_key *out)
{
    if (HDR(blk)->nritems == 0)
        memset(out, 0, sizeof(*out));
    else if (HDR(blk)->level)
        *out = KPTRS(blk)[0].key;
    else
        *out = ITEMS(blk)[0].key;
}

static void
node_seal(void *blk, uint64_t blocknr, uint64_t gen, uint32_t *crc_out)
{
    struct schfs_node_hdr *h = HDR(blk);
    h->magic      = SCHFS_NODE_MAGIC;
    h->blocknr    = blocknr;
    h->generation = gen;
    h->crc        = 0;
    uint32_t c    = schfs_crc32c(0, blk, SCHFS_BLKSIZE);
    h->crc        = c;
    if (crc_out)
        *crc_out = c;
}

// Read a block and validate it against the checksum recorded in its parent.
// want_crc of 0 means "no expectation" (used for the root, whose checksum lives
// in the uberblock and is passed in separately).
static int
node_read(struct schfs_bdev *dev, uint64_t blk, uint32_t want_crc, void *out)
{
    if (dev->bread(dev->ctx, blk, out) != 0)
        return -2;

    struct schfs_node_hdr *h = HDR(out);
    if (h->magic != SCHFS_NODE_MAGIC)
        return -2;
    // Self-pointer check: a block written to the wrong address still has a
    // valid checksum, so only this catches a misdirected write.
    if (h->blocknr != blk)
        return -2;
    // Two bounds the rest of the file then takes for granted.
    if (h->level > SCHFS_MAX_LEVEL)
        return -2;
    if (h->level ? h->nritems > SCHFS_MAX_KPTR : h->nritems > SCHFS_MAX_ITEMS)
        return -2;

    uint32_t stored = h->crc;
    h->crc = 0;
    uint32_t calc = schfs_crc32c(0, out, SCHFS_BLKSIZE);
    h->crc = stored;
    if (calc != stored)
        return -2;
    if (want_crc && want_crc != stored)
        return -2;
    return 0;
}

int
schfs_node_read(struct schfs_bdev *dev, uint64_t blk, uint32_t want_crc,
                void *out)
{
    return node_read(dev, blk, want_crc, out);
}

// Index of the first entry with key >= k, plus whether it matched exactly.
static int
node_search(const void *blk, const struct schfs_key *k, int *exact)
{
    const struct schfs_node_hdr *h = (const struct schfs_node_hdr *)blk;
    int lo = 0, hi = h->nritems - 1, found = -1;
    *exact = 0;

    while (lo <= hi) {
        int mid = (lo + hi) / 2;
        const struct schfs_key *mk = h->level
            ? &((const struct schfs_kptr *)((const char *)blk + sizeof(*h)))[mid].key
            : &((const struct schfs_item *)((const char *)blk + sizeof(*h)))[mid].key;
        int c = schfs_key_cmp(mk, k);
        if (c == 0) { *exact = 1; return mid; }
        if (c < 0)  { found = mid; lo = mid + 1; }
        else        { hi = mid - 1; }
    }
    // For an internal node we want the child that could contain k, i.e. the
    // last entry whose key is <= k (or the first child if k precedes all keys).
    return h->level ? (found < 0 ? 0 : found) : lo;
}

// Where a *new* separator key belongs in an internal node: the first index
// whose key is greater than k. Distinct from node_search, which answers "which
// child owns k" and would place a key smaller than every separator at index 1.
static int
kptr_insert_pos(const void *blk, const struct schfs_key *k)
{
    const struct schfs_node_hdr *h = (const struct schfs_node_hdr *)blk;
    const struct schfs_kptr *kp = (const struct schfs_kptr *)
        ((const char *)blk + sizeof(*h));
    int lo = 0, hi = h->nritems;
    while (lo < hi) {
        int mid = (lo + hi) / 2;
        if (schfs_key_cmp(&kp[mid].key, k) < 0)
            lo = mid + 1;
        else
            hi = mid;
    }
    return lo;
}

// Decide where a node's new contents go.
//
// A block stamped with the transaction group now being built was allocated
// after the last commit, so no published root can reach it: rewriting it in
// place is invisible to any reader of the committed tree and costs nothing.
// Anything older must be copied, which is the copy-on-write rule itself.
//
// Without this every single insert copies the entire root-to-leaf path, so a
// transaction touching a thousand records burns a thousand paths' worth of
// blocks before it can commit and release any of them.
static uint64_t
cow_block(struct schfs_btree *t, const void *blk, uint64_t old)
{
    if (HDR(blk)->generation == t->generation)
        return old;
    uint64_t nb = t->dev->balloc(t->dev->ctx);
    if (nb == 0)
        return 0;
    t->dev->bfree(t->dev->ctx, old);
    return nb;
}

// Seal a node into `blk_nr` and write it out.
static int
node_flush(struct schfs_btree *t, void *blk, uint64_t blk_nr, uint32_t *crc)
{
    node_seal(blk, blk_nr, t->generation, crc);
    return t->dev->bwrite(t->dev->ctx, blk_nr, blk) == 0 ? 0 : -2;
}

/* ------------------------------------------------------------------ */
/* Create                                                              */
/* ------------------------------------------------------------------ */

int
schfs_btree_create(struct schfs_btree *t, struct schfs_bdev *dev,
                   struct schfs_ctx *x)
{
    t->dev = dev;

    uint64_t b = dev->balloc(dev->ctx);
    if (b == 0)
        return -3;

    memset(x->a, 0, SCHFS_BLKSIZE);
    HDR(x->a)->level   = 0;
    HDR(x->a)->nritems = 0;
    if (node_flush(t, x->a, b, &t->root_crc) != 0)
        return -2;

    t->root     = b;
    t->root_gen = t->generation;
    return 0;
}

/* ------------------------------------------------------------------ */
/* Cursor                                                              */
/*                                                                     */
/* One block buffer, owned by the caller, plus a path of block numbers. */
/* Moving back up re-reads the parent instead of keeping it in memory,  */
/* which trades a buffer-cache hit for seven pages of stack.            */
/* ------------------------------------------------------------------ */

void
schfs_cursor_init(struct schfs_cursor *c, struct schfs_btree *t, void *blk)
{
    c->t     = t;
    c->blk   = blk;
    c->depth = 0;
    c->valid = 0;
}

// Load path[d]'s block into the cursor buffer, verifying it against the
// checksum recorded alongside it.
static int
cur_load(struct schfs_cursor *c, int d)
{
    return node_read(c->t->dev, c->path[d].blocknr, c->path[d].crc, c->blk);
}

// From the node currently in the buffer at depth `d`, walk down to a leaf,
// taking the first entry of each node (rightmost=0) or the last (rightmost=1).
// Leaves the leaf loaded and path[c->depth].index set. Returns 0, or -1 if the
// leaf reached is empty, or -2 on corruption.
static int
cur_descend(struct schfs_cursor *c, int d, int rightmost)
{
    for (;;) {
        struct schfs_node_hdr *h = HDR(c->blk);
        if (h->level == 0) {
            if (h->nritems == 0)
                return -1;
            c->depth = d;
            c->path[d].index = rightmost ? h->nritems - 1 : 0;
            return 0;
        }
        if (h->nritems == 0)
            return -2;
        int i = rightmost ? h->nritems - 1 : 0;
        c->path[d].index = i;
        struct schfs_kptr *kp = &KPTRS(c->blk)[i];
        if (d + 1 > SCHFS_MAX_LEVEL)
            return -2;
        c->path[d + 1].blocknr = kp->blocknr;
        c->path[d + 1].crc     = kp->crc;
        d++;
        if (cur_load(c, d) != 0)
            return -2;
    }
}

// Shared body of next/prev. `dir` is +1 or -1.
static int
cur_step(struct schfs_cursor *c, int dir)
{
    if (!c->valid)
        return -1;

    for (;;) {
        // Try to move within the leaf we are already holding.
        if (cur_load(c, c->depth) != 0) {
            c->valid = 0;
            return -2;
        }
        int n = HDR(c->blk)->nritems;
        int i = c->path[c->depth].index + dir;
        if (i >= 0 && i < n) {
            c->path[c->depth].index = i;
            return 0;
        }

        // Off the end of this leaf: climb until a parent has another child in
        // the direction of travel, then descend to that subtree's edge.
        int d = c->depth - 1;
        for (; d >= 0; d--) {
            if (cur_load(c, d) != 0) {
                c->valid = 0;
                return -2;
            }
            int pn = HDR(c->blk)->nritems;
            int pi = c->path[d].index + dir;
            if (pi >= 0 && pi < pn) {
                c->path[d].index = pi;
                struct schfs_kptr *kp = &KPTRS(c->blk)[pi];
                c->path[d + 1].blocknr = kp->blocknr;
                c->path[d + 1].crc     = kp->crc;
                if (cur_load(c, d + 1) != 0) {
                    c->valid = 0;
                    return -2;
                }
                int r = cur_descend(c, d + 1, dir < 0);
                if (r == -2) {
                    c->valid = 0;
                    return -2;
                }
                // r == -1 means that subtree bottomed out in an empty leaf. The
                // outer loop simply steps again from there. Delete frees leaves
                // that empty, so this is defensive rather than routine.
                if (r == 0)
                    return 0;
                break;
            }
        }
        if (d < 0) {
            c->valid = 0;
            return -1;              /* ran off the end of the tree */
        }
    }
}

int
schfs_cursor_next(struct schfs_cursor *c)
{
    return cur_step(c, +1);
}

int
schfs_cursor_prev(struct schfs_cursor *c)
{
    return cur_step(c, -1);
}

int
schfs_cursor_seek(struct schfs_cursor *c, const struct schfs_key *start)
{
    c->depth = 0;
    c->valid = 0;
    c->path[0].blocknr = c->t->root;
    c->path[0].crc     = c->t->root_crc;
    if (cur_load(c, 0) != 0)
        return -2;

    if (start == NULL) {
        int r = cur_descend(c, 0, 0);
        if (r == -2)
            return -2;
        if (r == -1)
            return -1;              /* empty tree */
        c->valid = 1;
        return 0;
    }

    int d = 0;
    for (;;) {
        struct schfs_node_hdr *h = HDR(c->blk);
        int exact;
        int i = node_search(c->blk, start, &exact);

        if (h->level == 0) {
            c->depth = d;
            if (i < h->nritems) {
                c->path[d].index = i;
                c->valid = 1;
                return 0;
            }
            // Every key in this leaf precedes `start`; the answer, if any, is
            // the next record in the tree. Park on the last item and step.
            if (h->nritems == 0) {
                // Empty leaf: fabricate a position so cur_step can climb out.
                c->path[d].index = 0;
                c->valid = 1;
                return schfs_cursor_next(c);
            }
            c->path[d].index = h->nritems - 1;
            c->valid = 1;
            return schfs_cursor_next(c);
        }

        if (h->nritems == 0)
            return -2;
        c->path[d].index = i;
        struct schfs_kptr *kp = &KPTRS(c->blk)[i];
        if (d + 1 > SCHFS_MAX_LEVEL)
            return -2;
        c->path[d + 1].blocknr = kp->blocknr;
        c->path[d + 1].crc     = kp->crc;
        d++;
        if (cur_load(c, d) != 0)
            return -2;
    }
}

const struct schfs_key *
schfs_cursor_key(struct schfs_cursor *c)
{
    if (!c->valid)
        return NULL;
    if (cur_load(c, c->depth) != 0) {
        c->valid = 0;
        return NULL;
    }
    if (c->path[c->depth].index >= HDR(c->blk)->nritems)
        return NULL;
    return &ITEMS(c->blk)[c->path[c->depth].index].key;
}

const void *
schfs_cursor_val(struct schfs_cursor *c, unsigned *len)
{
    if (!c->valid)
        return NULL;
    if (cur_load(c, c->depth) != 0) {
        c->valid = 0;
        return NULL;
    }
    if (c->path[c->depth].index >= HDR(c->blk)->nritems)
        return NULL;
    struct schfs_item *it = &ITEMS(c->blk)[c->path[c->depth].index];
    // A value must lie wholly inside the block and clear the item array. The
    // checksum already proves the block is the one that was written, but not
    // that the tool which wrote it was sane, and this pointer is about to be
    // handed to a caller that will read `size` bytes from it.
    int items_end = (int)sizeof(struct schfs_node_hdr)
                  + HDR(c->blk)->nritems * (int)sizeof(struct schfs_item);
    if ((int)it->offset < items_end ||
        (int)it->offset + (int)it->size > SCHFS_BLKSIZE)
        return NULL;
    if (len)
        *len = it->size;
    return (const char *)c->blk + it->offset;
}

int
schfs_btree_lookup_cur(struct schfs_cursor *c, const struct schfs_key *k,
                       void *buf, unsigned *len)
{
    int r = schfs_cursor_seek(c, k);
    if (r == -2)
        return -2;
    if (r != 0)
        return -1;

    const struct schfs_key *got = schfs_cursor_key(c);
    if (!got || schfs_key_cmp(got, k) != 0)
        return -1;

    unsigned vlen;
    const void *v = schfs_cursor_val(c, &vlen);
    if (!v)
        return -2;
    if (buf && len) {
        unsigned n = vlen < *len ? vlen : *len;
        memmove(buf, v, n);
    }
    if (len)
        *len = vlen;
    return 0;
}

int
schfs_btree_lookup(struct schfs_btree *t, struct schfs_ctx *x,
                   const struct schfs_key *k, void *buf, unsigned *len)
{
    struct schfs_cursor c;
    schfs_cursor_init(&c, t, x->a);
    return schfs_btree_lookup_cur(&c, k, buf, len);
}

int
schfs_btree_lookup_le(struct schfs_cursor *c, const struct schfs_key *k)
{
    int r = schfs_cursor_seek(c, k);

    if (r == -2)
        return -2;
    if (r == -1) {
        // Nothing at or after k, so the last record in the tree — if there is
        // one — is the predecessor. Descend rightmost from the root, which is
        // O(depth) rather than stepping the whole tree.
        c->depth = 0;
        c->valid = 0;
        c->path[0].blocknr = c->t->root;
        c->path[0].crc     = c->t->root_crc;
        if (cur_load(c, 0) != 0)
            return -2;
        if (cur_descend(c, 0, 1) != 0)
            return -1;
        c->valid = 1;
        return 0;
    }

    const struct schfs_key *got = schfs_cursor_key(c);
    if (got && schfs_key_cmp(got, k) == 0)
        return 0;                    /* exact hit */
    return schfs_cursor_prev(c);     /* -1 if k precedes everything */
}

/* ------------------------------------------------------------------ */
/* Leaf edit helpers                                                   */
/* ------------------------------------------------------------------ */

// Rebuild a leaf's value area compactly, using `tmp` as staging. Deleting and
// replacing leave holes; rather than track them, the leaf is repacked whenever
// it is modified, which is cheap next to the block write that follows.
static void
leaf_compact(void *blk, void *tmp)
{
    struct schfs_node_hdr *h = HDR(blk);
    struct schfs_item *it = ITEMS(blk);
    int n = h->nritems;
    int wr = SCHFS_BLKSIZE;

    memmove(tmp, blk, SCHFS_BLKSIZE);
    for (int i = 0; i < n; i++) {
        struct schfs_item *src = &((struct schfs_item *)
            ((char *)tmp + sizeof(*h)))[i];
        wr -= (int)src->size;
        memmove((char *)blk + wr, (char *)tmp + src->offset, src->size);
        it[i].key    = src->key;
        it[i].size   = src->size;
        it[i].offset = (uint32_t)wr;
    }
}

// Insert (or replace) one item in an already-copied leaf. Returns 0, or -1 if
// it does not fit and the leaf must be split. `tmp` is repacking scratch.
static int
leaf_put(void *blk, const struct schfs_key *k, const void *val, unsigned len,
         void *tmp)
{
    struct schfs_node_hdr *h = HDR(blk);
    struct schfs_item *it = ITEMS(blk);
    int exact;
    int i = node_search(blk, k, &exact);

    if (exact) {
        // Replace: drop the old item, then fall through to a clean insert so
        // the size may change freely.
        for (int j = i; j < h->nritems - 1; j++)
            it[j] = it[j + 1];
        h->nritems--;
        leaf_compact(blk, tmp);
        i = node_search(blk, k, &exact);
    }

    if (h->nritems >= SCHFS_MAX_ITEMS)
        return -1;
    int need = (int)sizeof(struct schfs_item) + (int)len;
    if (need > leaf_free_space(blk))
        return -1;

    for (int j = h->nritems; j > i; j--)
        it[j] = it[j - 1];

    int data_start = SCHFS_BLKSIZE;
    for (int j = 0; j < h->nritems; j++) {
        int idx = (j >= i) ? j + 1 : j;
        if ((int)it[idx].offset < data_start)
            data_start = (int)it[idx].offset;
    }
    int off = data_start - (int)len;

    it[i].key    = *k;
    it[i].offset = (uint32_t)off;
    it[i].size   = len;
    memmove((char *)blk + off, val, len);
    h->nritems++;
    return 0;
}

// Remove the item at index i.
static void
leaf_drop(void *blk, int i, void *tmp)
{
    struct schfs_node_hdr *h = HDR(blk);
    struct schfs_item *it = ITEMS(blk);
    for (int j = i; j < h->nritems - 1; j++)
        it[j] = it[j + 1];
    h->nritems--;
    leaf_compact(blk, tmp);
}

/* ------------------------------------------------------------------ */
/* Internal-node edit helpers                                          */
/* ------------------------------------------------------------------ */

// Insert a (key -> child) pair into an internal node that is already ours to
// write. Returns 0, or -1 if it is full.
static int
internal_put(void *blk, const struct schfs_key *k, uint64_t child,
             uint32_t crc, uint64_t gen)
{
    struct schfs_node_hdr *h = HDR(blk);
    struct schfs_kptr *kp = KPTRS(blk);

    if (h->nritems >= SCHFS_MAX_KPTR)
        return -1;

    int i = kptr_insert_pos(blk, k);
    for (int j = h->nritems; j > i; j--)
        kp[j] = kp[j - 1];

    kp[i].key        = *k;
    kp[i].blocknr    = child;
    kp[i].crc        = crc;
    kp[i].generation = gen;
    kp[i].pad        = 0;
    h->nritems++;
    return 0;
}

static void
internal_drop(void *blk, int i)
{
    struct schfs_node_hdr *h = HDR(blk);
    struct schfs_kptr *kp = KPTRS(blk);
    for (int j = i; j < h->nritems - 1; j++)
        kp[j] = kp[j + 1];
    h->nritems--;
}

/* ------------------------------------------------------------------ */
/* Split                                                               */
/* ------------------------------------------------------------------ */

// Move the upper half of `blk` into `nb`, which the caller then seals and
// writes at the returned block number. `tmp` is only touched for leaves.
// Returns 0 if no block could be allocated.
static uint64_t
node_split(struct schfs_btree *t, void *blk, void *nb, void *tmp,
           struct schfs_key *split_key)
{
    struct schfs_node_hdr *h = HDR(blk);
    int n = h->nritems;
    int half = n / 2;

    uint64_t newb = t->dev->balloc(t->dev->ctx);
    if (newb == 0)
        return 0;

    memset(nb, 0, SCHFS_BLKSIZE);
    HDR(nb)->level   = h->level;
    HDR(nb)->nritems = 0;

    if (h->level) {
        struct schfs_kptr *src = KPTRS(blk);
        struct schfs_kptr *dst = KPTRS(nb);
        for (int i = half; i < n; i++)
            dst[i - half] = src[i];
        HDR(nb)->nritems = (uint16_t)(n - half);
        h->nritems       = (uint16_t)half;
        *split_key = dst[0].key;
    } else {
        struct schfs_item *src = ITEMS(blk);
        struct schfs_item *dst = ITEMS(nb);
        int wr = SCHFS_BLKSIZE;
        for (int i = half; i < n; i++) {
            wr -= (int)src[i].size;
            memmove((char *)nb + wr, (char *)blk + src[i].offset, src[i].size);
            dst[i - half].key    = src[i].key;
            dst[i - half].size   = src[i].size;
            dst[i - half].offset = (uint32_t)wr;
        }
        HDR(nb)->nritems = (uint16_t)(n - half);
        h->nritems       = (uint16_t)half;
        leaf_compact(blk, tmp);
        *split_key = dst[0].key;
    }
    return newb;
}

/* ------------------------------------------------------------------ */
/* Insert                                                              */
/*                                                                     */
/* The separator invariant: kp[i].key is exactly the first key held in  */
/* child i. Every path rewrite therefore refreshes the parent's key as  */
/* well as its pointer, which is the whole reason a split key can       */
/* always be filed immediately after the child it split.                */
/*                                                                     */
/* The weaker "kp[i].key <= everything in child i" would also make      */
/* lookups work, and an earlier version relied on it. It breaks on      */
/* insert: descending inserts push a child's first key below its        */
/* parent's key for it, and splitting that child then yields a          */
/* separator smaller than kp[0].key, which has no valid position in a   */
/* sorted node. Keeping the keys exact removes the case entirely.       */
/* ------------------------------------------------------------------ */

struct path_ent {
    uint64_t blocknr;
    uint32_t crc;
    int      index;
};

int
schfs_btree_insert(struct schfs_btree *t, struct schfs_ctx *x,
                   const struct schfs_key *k, const void *val, unsigned len)
{
    struct path_ent path[SCHFS_MAX_LEVEL + 1];
    int depth = 0;

    if (len > SCHFS_MAX_INLINE)
        return -1;

    /* Descend, remembering the path so it can be rewritten on the way back up. */
    uint64_t cur = t->root;
    uint32_t crc = t->root_crc;
    for (;;) {
        if (depth > SCHFS_MAX_LEVEL)
            return -2;
        if (node_read(t->dev, cur, crc, x->a) != 0)
            return -2;

        path[depth].blocknr = cur;
        path[depth].crc     = crc;
        path[depth].index   = 0;

        if (HDR(x->a)->level == 0)
            break;
        if (HDR(x->a)->nritems == 0)
            return -2;

        int exact;
        int i = node_search(x->a, k, &exact);
        path[depth].index = i;
        cur = KPTRS(x->a)[i].blocknr;
        crc = KPTRS(x->a)[i].crc;
        depth++;
    }

    int root_level = HDR(x->a)->level + depth;

    struct schfs_key split_key;
    uint64_t sibling = 0;
    uint32_t sib_crc = 0;
    int split = 0;

    if (leaf_put(x->a, k, val, len, x->c) != 0) {
        sibling = node_split(t, x->a, x->b, x->c, &split_key);
        if (sibling == 0)
            return -3;
        split = 1;
        /* The item belongs in whichever half now covers its key. */
        if (schfs_key_cmp(k, &split_key) >= 0) {
            if (leaf_put(x->b, k, val, len, x->c) != 0)
                return -1;           /* value too big even for an empty leaf */
        } else {
            if (leaf_put(x->a, k, val, len, x->c) != 0)
                return -1;
        }
        /* node_split may have moved the sibling's first key. */
        node_first_key(x->b, &split_key);
        if (node_flush(t, x->b, sibling, &sib_crc) != 0)
            return -2;
    }

    uint64_t newb = cow_block(t, x->a, path[depth].blocknr);
    if (newb == 0)
        return -3;
    uint32_t child_crc;
    if (node_flush(t, x->a, newb, &child_crc) != 0)
        return -2;

    uint64_t child = newb;
    struct schfs_key child_first;
    node_first_key(x->a, &child_first);

    /* Walk back up, rewriting each parent to point at the new child. */
    for (int d = depth - 1; d >= 0; d--) {
        if (node_read(t->dev, path[d].blocknr, path[d].crc, x->a) != 0)
            return -2;

        struct schfs_kptr *kp = KPTRS(x->a);
        kp[path[d].index].key        = child_first;
        kp[path[d].index].blocknr    = child;
        kp[path[d].index].crc        = child_crc;
        kp[path[d].index].generation = t->generation;

        if (split) {
            if (internal_put(x->a, &split_key, sibling, sib_crc,
                             t->generation) != 0) {
                struct schfs_key up_key;
                uint64_t up = node_split(t, x->a, x->b, x->c, &up_key);
                if (up == 0)
                    return -3;
                if (schfs_key_cmp(&split_key, &up_key) >= 0)
                    internal_put(x->b, &split_key, sibling, sib_crc, t->generation);
                else
                    internal_put(x->a, &split_key, sibling, sib_crc, t->generation);
                node_first_key(x->b, &up_key);
                uint32_t up_crc;
                if (node_flush(t, x->b, up, &up_crc) != 0)
                    return -2;
                sibling   = up;
                sib_crc   = up_crc;
                split_key = up_key;
            } else {
                split = 0;
            }
        }

        uint64_t pnew = cow_block(t, x->a, path[d].blocknr);
        if (pnew == 0)
            return -3;
        if (node_flush(t, x->a, pnew, &child_crc) != 0)
            return -2;
        child = pnew;
        node_first_key(x->a, &child_first);
    }

    /* A split that reached the root means the tree gets one level taller. */
    if (split) {
        if (root_level + 1 > SCHFS_MAX_LEVEL)
            return -2;
        uint64_t rnew = t->dev->balloc(t->dev->ctx);
        if (rnew == 0)
            return -3;

        memset(x->a, 0, SCHFS_BLKSIZE);
        HDR(x->a)->level   = (uint16_t)(root_level + 1);
        HDR(x->a)->nritems = 2;
        struct schfs_kptr *kp = KPTRS(x->a);
        kp[0].key = child_first; kp[0].blocknr = child;   kp[0].crc = child_crc;
        kp[0].generation = t->generation;
        kp[1].key = split_key;   kp[1].blocknr = sibling; kp[1].crc = sib_crc;
        kp[1].generation = t->generation;

        if (node_flush(t, x->a, rnew, &child_crc) != 0)
            return -2;
        child = rnew;
    }

    t->root     = child;
    t->root_crc = child_crc;
    t->root_gen = t->generation;
    return 0;
}

/* ------------------------------------------------------------------ */
/* Delete                                                              */
/*                                                                     */
/* Emptied nodes are freed rather than left in place. Without that a    */
/* filesystem that is filled and emptied repeatedly accumulates empty   */
/* leaves that every scan still has to walk, and the tree never gets    */
/* shorter no matter how few records remain.                            */
/*                                                                     */
/* Nodes are still not *rebalanced* — an underfull node keeps its       */
/* records until it is refilled or emptied. Merging half-full siblings  */
/* is the hardest part of B-tree deletion and buys only density, so it  */
/* stays out until something measures a need for it.                    */
/* ------------------------------------------------------------------ */

int
schfs_btree_delete(struct schfs_btree *t, struct schfs_ctx *x,
                   const struct schfs_key *k)
{
    struct path_ent path[SCHFS_MAX_LEVEL + 1];
    int depth = 0;

    uint64_t cur = t->root;
    uint32_t crc = t->root_crc;
    for (;;) {
        if (depth > SCHFS_MAX_LEVEL)
            return -2;
        if (node_read(t->dev, cur, crc, x->a) != 0)
            return -2;
        path[depth].blocknr = cur;
        path[depth].crc     = crc;
        path[depth].index   = 0;
        if (HDR(x->a)->level == 0)
            break;
        if (HDR(x->a)->nritems == 0)
            return -2;
        int exact;
        int i = node_search(x->a, k, &exact);
        path[depth].index = i;
        cur = KPTRS(x->a)[i].blocknr;
        crc = KPTRS(x->a)[i].crc;
        depth++;
    }

    int exact;
    int i = node_search(x->a, k, &exact);
    if (!exact)
        return -1;

    leaf_drop(x->a, i, x->c);

    uint64_t child = 0;
    uint32_t child_crc = 0;
    struct schfs_key child_first;
    memset(&child_first, 0, sizeof(child_first));

    // An emptied leaf below the root is unlinked from its parent instead of
    // being rewritten; `drop` carries that decision up the path.
    int drop = (HDR(x->a)->nritems == 0 && depth > 0);

    if (drop) {
        t->dev->bfree(t->dev->ctx, path[depth].blocknr);
    } else {
        uint64_t newb = cow_block(t, x->a, path[depth].blocknr);
        if (newb == 0)
            return -3;
        if (node_flush(t, x->a, newb, &child_crc) != 0)
            return -2;
        child = newb;
        node_first_key(x->a, &child_first);
    }

    for (int d = depth - 1; d >= 0; d--) {
        if (node_read(t->dev, path[d].blocknr, path[d].crc, x->a) != 0)
            return -2;

        if (drop) {
            internal_drop(x->a, path[d].index);
            if (HDR(x->a)->nritems == 0 && d > 0) {
                t->dev->bfree(t->dev->ctx, path[d].blocknr);
                continue;            /* keep unlinking upward */
            }
            // A root that lost its last child becomes an empty leaf; leaving it
            // an internal node with no children makes every descent read a node
            // it cannot use and report corruption.
            if (HDR(x->a)->nritems == 0)
                HDR(x->a)->level = 0;
            drop = 0;
        } else {
            struct schfs_kptr *kp = KPTRS(x->a);
            kp[path[d].index].key        = child_first;
            kp[path[d].index].blocknr    = child;
            kp[path[d].index].crc        = child_crc;
            kp[path[d].index].generation = t->generation;
        }

        uint64_t pnew = cow_block(t, x->a, path[d].blocknr);
        if (pnew == 0)
            return -3;
        if (node_flush(t, x->a, pnew, &child_crc) != 0)
            return -2;
        child = pnew;
        node_first_key(x->a, &child_first);
    }

    t->root     = child;
    t->root_crc = child_crc;
    t->root_gen = t->generation;

    // Collapse a root that is down to one child. A tree grows a level per
    // split; without this it never gives one back, so a directory that briefly
    // held a million entries keeps paying five block reads per lookup forever.
    for (;;) {
        if (node_read(t->dev, t->root, t->root_crc, x->a) != 0)
            return -2;
        if (HDR(x->a)->level == 0 || HDR(x->a)->nritems != 1)
            break;
        uint64_t old = t->root;
        t->root     = KPTRS(x->a)[0].blocknr;
        t->root_crc = KPTRS(x->a)[0].crc;
        t->dev->bfree(t->dev->ctx, old);
    }
    t->root_gen = HDR(x->a)->generation;
    return 0;
}

/* ------------------------------------------------------------------ */
/* Range scan                                                          */
/* ------------------------------------------------------------------ */

int
schfs_btree_range(struct schfs_cursor *c, const struct schfs_key *start,
                  int (*fn)(const struct schfs_key *, const void *,
                            unsigned, void *),
                  void *arg)
{
    int r = schfs_cursor_seek(c, start);
    if (r == -2)
        return -2;
    if (r != 0)
        return 0;                    /* nothing at or after start */

    for (;;) {
        const struct schfs_key *k = schfs_cursor_key(c);
        if (!k)
            return 0;
        unsigned len;
        const void *v = schfs_cursor_val(c, &len);
        if (!v)
            return -2;

        // The key is copied out before the callback runs. Callbacks routinely
        // want to look something else up — "list a directory and stat each
        // entry" is the whole point of a range scan — and any other tree
        // operation may reuse this same block buffer. See the warning on
        // schfs_btree_range in btree.h: the *value* pointer has the same
        // hazard and must be copied by the callback if it needs it to survive.
        struct schfs_key kc = *k;

        int s = fn(&kc, v, len, arg);
        if (s)
            return s;

        r = schfs_cursor_next(c);
        if (r == -2)
            return -2;
        if (r != 0)
            return 0;
    }
}

/* ------------------------------------------------------------------ */
/* Host-only: whole-tree walk and structural check                     */
/* ------------------------------------------------------------------ */
#ifdef SCHFS_HOSTED

static int
walk(struct schfs_btree *t, uint64_t blk, uint32_t crc, int level_expect,
     int (*fn)(const struct schfs_key *, const void *, unsigned, void *),
     void *arg)
{
    char b[SCHFS_BLKSIZE];
    if (node_read(t->dev, blk, crc, b) != 0)
        return -2;
    if (level_expect >= 0 && HDR(b)->level != level_expect)
        return -2;

    if (HDR(b)->level == 0) {
        struct schfs_item *it = ITEMS(b);
        for (int i = 0; i < HDR(b)->nritems; i++) {
            int r = fn(&it[i].key, b + it[i].offset, it[i].size, arg);
            if (r)
                return r;
        }
        return 0;
    }

    struct schfs_kptr *kp = KPTRS(b);
    for (int i = 0; i < HDR(b)->nritems; i++) {
        int r = walk(t, kp[i].blocknr, kp[i].crc, HDR(b)->level - 1, fn, arg);
        if (r)
            return r;
    }
    return 0;
}

int
schfs_btree_iterate(struct schfs_btree *t,
                    int (*fn)(const struct schfs_key *, const void *,
                              unsigned, void *),
                    void *arg)
{
    return walk(t, t->root, t->root_crc, -1, fn, arg);
}

struct check_ctx {
    struct schfs_key last;
    int              have_last;
};

static int
check_order(const struct schfs_key *k, const void *v, unsigned len, void *arg)
{
    struct check_ctx *c = (struct check_ctx *)arg;
    (void)v; (void)len;
    if (c->have_last && schfs_key_cmp(&c->last, k) >= 0)
        return -1;                   /* out of order or duplicated */
    c->last = *k;
    c->have_last = 1;
    return 0;
}

// Verify the separator invariant the descent relies on: kp[i].key is exactly
// the first key in child i, every key in subtree i lies in
// [kp[i].key, kp[i+1].key), and no leaf value escapes its block.
static int
check_bounds(struct schfs_btree *t, uint64_t blk, uint32_t crc,
             int level_expect, const struct schfs_key *lo,
             const struct schfs_key *hi)
{
    char b[SCHFS_BLKSIZE];
    if (node_read(t->dev, blk, crc, b) != 0)
        return -2;
    if (level_expect >= 0 && HDR(b)->level != level_expect)
        return -2;

    int n = HDR(b)->nritems;

    if (HDR(b)->level == 0) {
        struct schfs_item *it = ITEMS(b);
        int items_end = (int)sizeof(struct schfs_node_hdr)
                      + n * (int)sizeof(struct schfs_item);
        for (int i = 0; i < n; i++) {
            if (i > 0 && schfs_key_cmp(&it[i - 1].key, &it[i].key) >= 0)
                return -3;           /* leaf keys out of order */
            if (lo && schfs_key_cmp(&it[i].key, lo) < 0)
                return -4;           /* record is below its subtree's range */
            if (hi && schfs_key_cmp(&it[i].key, hi) >= 0)
                return -5;           /* record is above its subtree's range */
            if ((int)it[i].offset < items_end ||
                (int)it[i].offset + (int)it[i].size > SCHFS_BLKSIZE)
                return -6;           /* value overlaps the item array or runs off the block */
        }
        return 0;
    }

    if (n == 0)
        return -7;                   /* internal node with no children */

    struct schfs_kptr *kp = KPTRS(b);
    for (int i = 0; i < n; i++) {
        if (i > 0 && schfs_key_cmp(&kp[i - 1].key, &kp[i].key) >= 0)
            return -3;               /* child pointers out of order */
        if (lo && i == 0 && schfs_key_cmp(&kp[0].key, lo) != 0)
            return -8;               /* first key does not match the parent's */
        int r = check_bounds(t, kp[i].blocknr, kp[i].crc, HDR(b)->level - 1,
                             &kp[i].key,
                             i + 1 < n ? &kp[i + 1].key : hi);
        if (r)
            return r;
    }
    return 0;
}

int
schfs_btree_check(struct schfs_btree *t)
{
    struct check_ctx c;
    c.have_last = 0;
    /* walk() verifies magic, self-pointer, checksum and level at every node;
       this adds the global key ordering invariant... */
    int r = schfs_btree_iterate(t, check_order, &c);
    if (r)
        return r;
    /* ...and this the separator invariant the descent depends on. */
    return check_bounds(t, t->root, t->root_crc, -1, NULL, NULL);
}

#endif /* SCHFS_HOSTED */