// SPDX-License-Identifier: Apache-2.0
//
// Attributes, and the index over them.
//
// An attribute is a named value on an object. The interesting part is not
// storing them -- that is another key type in a tree that already exists -- but
// the index: an indexed attribute gets a second set of records keyed by *value*
// rather than by object, so "every object whose `tag` is `holiday`" is a range
// scan of a few keys instead of a walk of the whole filesystem.
//
// That is the whole of what "a filesystem you can query" means here, and the
// reason the design was one B-tree with many key types from the start. There is
// no query engine, no parser and no planner; there is an index, which is the
// part that actually makes the question fast. See docs/architecture/schfs.md §6.
//
// ---------------------------------------------------------------------------
// Packed lists, again
//
// Attributes are keyed (ino, ATTR, hash-of-name) and index entries
// (name-hash, INDEX, hash-of-value). Both hash something that can collide, so
// both store a *packed list* of records rather than a single one -- the same
// shape directory entries use, for the same reason. A 64-bit hash makes a
// collision vanishingly unlikely and "vanishingly unlikely" is not "impossible".

#include "schfs_int.h"

#ifdef SCHFS_HOSTED
#include <string.h>
#else
#include "types.h"
#include "defs.h"
#endif

/* ------------------------------------------------------------------ */
/* Reading                                                             */
/* ------------------------------------------------------------------ */

// Find one attribute in the packed list under its name's hash.
//
// Returns 1 and fills the caller's buffers if present, 0 if absent, -2 on
// corruption. `*valuelen` is set to the value's true length even when the
// buffer is too small, so a caller can size a second call -- which is what
// getxattr(2) promises.
int
schfs_attr_get(struct schfs_vol *v, uint64_t ino, const char *name,
               unsigned namelen, void *out, unsigned cap, unsigned *valuelen,
               uint16_t *type)
{
    if (namelen == 0 || namelen > SCHFS_ATTR_NAME_MAX)
        return 0;

    struct schfs_key k;
    schfs_mkkey(&k, ino, SCHFS_KEY_ATTR, schfs_namehash(name, namelen));

    unsigned len = SCHFS_MAX_INLINE;
    int r = schfs_btree_lookup(&v->tree, &v->ctx, &k, v->dirbuf, &len);
    if (r == -1)
        return 0;
    if (r != 0)
        return r;
    if (len > SCHFS_MAX_INLINE)
        return -2;

    unsigned off = 0;
    while (off + sizeof(struct schfs_attr_rec) <= len) {
        const struct schfs_attr_rec *a =
            (const struct schfs_attr_rec *)(v->dirbuf + off);
        unsigned step = SCHFS_ATTR_LEN(a->namelen, a->valuelen);
        if (a->namelen == 0 || a->namelen > SCHFS_ATTR_NAME_MAX ||
            a->valuelen > SCHFS_ATTR_VALUE_MAX || off + step > len)
            return -2;

        const char *nm = (const char *)a + sizeof(*a);
        if (a->namelen == namelen && memcmp(nm, name, namelen) == 0) {
            if (valuelen) *valuelen = a->valuelen;
            if (type)     *type     = a->type;
            if (out && cap >= a->valuelen)
                memmove(out, nm + a->namelen, a->valuelen);
            else if (out)
                return -3;           /* buffer too small; length is set */
            return 1;
        }
        off += step;
    }
    return 0;
}

// Hand every attribute on an object to a callback, in the order they are
// stored. Used by listxattr and by the reindexing walk.
int
schfs_attr_iterate(struct schfs_vol *v, struct schfs_cursor *c, uint64_t ino,
                   int (*fn)(const char *name, unsigned namelen,
                             const void *val, unsigned vallen,
                             uint16_t type, void *arg),
                   void *arg)
{
    struct schfs_key start;
    schfs_mkkey(&start, ino, SCHFS_KEY_ATTR, 0);

    // Re-point at the volume's current tree. The root moves on every commit
    // under copy-on-write, so a cursor is bound per operation rather than once.
    schfs_cursor_init(c, &v->tree, c->blk);
    if (schfs_cursor_seek(c, &start) != 0)
        return 0;

    for (;;) {
        const struct schfs_key *k = schfs_cursor_key(c);
        if (!k || k->objectid != ino || k->type != SCHFS_KEY_ATTR)
            return 0;

        unsigned len;
        const char *val = (const char *)schfs_cursor_val(c, &len);
        if (!val || len > SCHFS_MAX_INLINE)
            return -2;

        // Copied out before the callback runs: the callback may look something
        // up, and that reuses the very buffer this points into. The same trap
        // that once reparented directories under their own children.
        memmove(v->dirbuf, val, len);

        unsigned off = 0;
        while (off + sizeof(struct schfs_attr_rec) <= len) {
            const struct schfs_attr_rec *a =
                (const struct schfs_attr_rec *)(v->dirbuf + off);
            unsigned step = SCHFS_ATTR_LEN(a->namelen, a->valuelen);
            if (a->namelen == 0 || a->namelen > SCHFS_ATTR_NAME_MAX ||
                a->valuelen > SCHFS_ATTR_VALUE_MAX || off + step > len)
                return -2;
            const char *nm = (const char *)a + sizeof(*a);
            int stop = fn(nm, a->namelen, nm + a->namelen, a->valuelen,
                          a->type, arg);
            if (stop)
                return 0;
            off += step;
        }

        if (schfs_cursor_next(c) != 0)
            return 0;
    }
}

/* ------------------------------------------------------------------ */
/* The index                                                           */
/*                                                                     */
/* Which attributes are indexed is one record, and it is consulted on   */
/* every attribute write -- so it is read into the volume rather than   */
/* fetched each time. See schfs_index_reload.                          */
/* ------------------------------------------------------------------ */

// Is `name` an indexed attribute?
//
// Answered from the cached list rather than the tree. setxattr asks this every
// time, and the answer is almost always no; a tree descent to say so would make
// every attribute write pay for a feature it is not using.
static int
index_wanted(struct schfs_vol *v, const char *name, unsigned namelen)
{
    unsigned off = 0;
    while (off + 1 <= v->idxnames_len) {
        unsigned l = (unsigned char)v->idxnames[off];
        if (l == 0 || off + 1 + l > v->idxnames_len)
            return 0;
        if (l == namelen && memcmp(v->idxnames + off + 1, name, namelen) == 0)
            return 1;
        off += 1 + l;
    }
    return 0;
}

// Read the indexed-attribute list off the disk into the volume.
//
// Called at mount and after any change to the list. A volume with no indexes --
// which is every volume until somebody asks for one -- ends up with an empty
// list and index_wanted answers no without touching anything.
int
schfs_index_reload(struct schfs_vol *v)
{
    struct schfs_key k;
    schfs_mkkey(&k, SCHFS_INDEX_OBJECTID, SCHFS_KEY_INDEX, SCHFS_INDEX_LIST_OFF);

    unsigned len = sizeof(v->idxnames);
    int r = schfs_btree_lookup(&v->tree, &v->ctx, &k, v->idxnames, &len);
    if (r == -1) {
        v->idxnames_len = 0;
        return 0;
    }
    if (r != 0)
        return r;
    v->idxnames_len = len;
    return 0;
}

// Add one object to the index for (attribute name, value).
static int
index_add(struct schfs_vol *v, const char *name, unsigned namelen,
          const void *val, unsigned vallen, uint64_t ino)
{
    struct schfs_key k;
    schfs_mkkey(&k, schfs_namehash(name, namelen), SCHFS_KEY_INDEX,
                schfs_namehash((const char *)val, vallen));

    unsigned len = SCHFS_MAX_INLINE;
    int r = schfs_btree_lookup(&v->tree, &v->ctx, &k, v->dirbuf, &len);
    if (r == -1)
        len = 0;
    else if (r != 0)
        return r;

    // Already listed? An object that sets the same value twice must not appear
    // twice, or a query would return it twice.
    unsigned off = 0;
    while (off + sizeof(struct schfs_index_rec) <= len) {
        const struct schfs_index_rec *ir =
            (const struct schfs_index_rec *)(v->dirbuf + off);
        unsigned step = SCHFS_INDEX_LEN(ir->valuelen);
        if (off + step > len)
            return -2;
        if (ir->objectid == ino)
            return 0;
        off += step;
    }

    unsigned add = SCHFS_INDEX_LEN(vallen);
    if (len + add > SCHFS_MAX_INLINE)
        return -3;                   /* too many objects share this value */

    struct schfs_index_rec ir;
    memset(&ir, 0, sizeof(ir));
    ir.objectid = ino;
    ir.valuelen = vallen;
    memmove(v->dirbuf + len, &ir, sizeof(ir));
    memmove(v->dirbuf + len + sizeof(ir), val, vallen);
    // The padding is zeroed rather than left as whatever was in the buffer, so
    // two identical filesystems produce byte-identical images.
    memset(v->dirbuf + len + sizeof(ir) + vallen, 0,
           add - sizeof(ir) - vallen);

    return schfs_btree_insert(&v->tree, &v->ctx, &k, v->dirbuf, len + add);
}

// Remove one object from the index for (attribute name, value).
static int
index_del(struct schfs_vol *v, const char *name, unsigned namelen,
          const void *val, unsigned vallen, uint64_t ino)
{
    struct schfs_key k;
    schfs_mkkey(&k, schfs_namehash(name, namelen), SCHFS_KEY_INDEX,
                schfs_namehash((const char *)val, vallen));

    unsigned len = SCHFS_MAX_INLINE;
    int r = schfs_btree_lookup(&v->tree, &v->ctx, &k, v->dirbuf, &len);
    if (r == -1)
        return 0;
    if (r != 0)
        return r;

    unsigned off = 0;
    while (off + sizeof(struct schfs_index_rec) <= len) {
        struct schfs_index_rec *ir = (struct schfs_index_rec *)(v->dirbuf + off);
        unsigned step = SCHFS_INDEX_LEN(ir->valuelen);
        if (off + step > len)
            return -2;
        if (ir->objectid == ino) {
            memmove(v->dirbuf + off, v->dirbuf + off + step, len - off - step);
            len -= step;
            if (len == 0)
                return schfs_btree_delete(&v->tree, &v->ctx, &k);
            return schfs_btree_insert(&v->tree, &v->ctx, &k, v->dirbuf, len);
        }
        off += step;
    }
    return 0;
}

/* ------------------------------------------------------------------ */
/* Writing                                                             */
/* ------------------------------------------------------------------ */

// Set an attribute, replacing any previous value of the same name.
//
// The index is kept in step here rather than by a separate pass, because an
// index that can be out of date is worse than none: a query would answer
// confidently and wrongly.
int
schfs_attr_set(struct schfs_vol *v, uint64_t ino, const char *name,
               unsigned namelen, const void *val, unsigned vallen,
               uint16_t type)
{
    if (namelen == 0 || namelen > SCHFS_ATTR_NAME_MAX)
        return -1;
    if (vallen > SCHFS_ATTR_VALUE_MAX)
        return -1;

    schfs_cache_invalidate();

    // The old value has to come out of the index before the new one goes in,
    // and it has to be read before the record is overwritten.
    int indexed = index_wanted(v, name, namelen);
    char oldval[SCHFS_ATTR_VALUE_MAX];
    unsigned oldlen = 0;
    int had_old = 0;
    if (indexed) {
        unsigned vl = 0;
        int g = schfs_attr_get(v, ino, name, namelen, oldval, sizeof(oldval),
                               &vl, 0);
        if (g == 1) { had_old = 1; oldlen = vl; }
    }

    struct schfs_key k;
    schfs_mkkey(&k, ino, SCHFS_KEY_ATTR, schfs_namehash(name, namelen));

    unsigned len = SCHFS_MAX_INLINE;
    int r = schfs_btree_lookup(&v->tree, &v->ctx, &k, v->dirbuf, &len);
    if (r == -1)
        len = 0;
    else if (r != 0)
        return r;

    // Drop any record with this name; what remains is everything that merely
    // hashed the same.
    unsigned off = 0;
    while (off + sizeof(struct schfs_attr_rec) <= len) {
        struct schfs_attr_rec *a = (struct schfs_attr_rec *)(v->dirbuf + off);
        unsigned step = SCHFS_ATTR_LEN(a->namelen, a->valuelen);
        if (a->namelen == 0 || off + step > len)
            return -2;
        const char *nm = (const char *)a + sizeof(*a);
        if (a->namelen == namelen && memcmp(nm, name, namelen) == 0) {
            memmove(v->dirbuf + off, v->dirbuf + off + step, len - off - step);
            len -= step;
            break;
        }
        off += step;
    }

    unsigned add = SCHFS_ATTR_LEN(namelen, vallen);
    if (len + add > SCHFS_MAX_INLINE)
        return -3;

    struct schfs_attr_rec a;
    memset(&a, 0, sizeof(a));
    a.namelen  = namelen;
    a.valuelen = vallen;
    a.type     = type;
    memmove(v->dirbuf + len, &a, sizeof(a));
    memmove(v->dirbuf + len + sizeof(a), name, namelen);
    memmove(v->dirbuf + len + sizeof(a) + namelen, val, vallen);
    memset(v->dirbuf + len + sizeof(a) + namelen + vallen, 0,
           add - sizeof(a) - namelen - vallen);

    r = schfs_btree_insert(&v->tree, &v->ctx, &k, v->dirbuf, len + add);
    if (r != 0)
        return r;

    if (indexed) {
        if (had_old) {
            r = index_del(v, name, namelen, oldval, oldlen, ino);
            if (r != 0)
                return r;
        }
        r = index_add(v, name, namelen, val, vallen, ino);
        if (r != 0)
            return r;
    }
    return 0;
}

// Remove an attribute. Absent is not a failure.
int
schfs_attr_del(struct schfs_vol *v, uint64_t ino, const char *name,
               unsigned namelen)
{
    if (namelen == 0 || namelen > SCHFS_ATTR_NAME_MAX)
        return -1;

    schfs_cache_invalidate();

    int indexed = index_wanted(v, name, namelen);
    char oldval[SCHFS_ATTR_VALUE_MAX];
    unsigned oldlen = 0;
    int had_old = 0;
    if (indexed) {
        unsigned vl = 0;
        int g = schfs_attr_get(v, ino, name, namelen, oldval, sizeof(oldval),
                               &vl, 0);
        if (g == 1) { had_old = 1; oldlen = vl; }
    }

    struct schfs_key k;
    schfs_mkkey(&k, ino, SCHFS_KEY_ATTR, schfs_namehash(name, namelen));

    unsigned len = SCHFS_MAX_INLINE;
    int r = schfs_btree_lookup(&v->tree, &v->ctx, &k, v->dirbuf, &len);
    if (r == -1)
        return 0;
    if (r != 0)
        return r;

    unsigned off = 0;
    int found = 0;
    while (off + sizeof(struct schfs_attr_rec) <= len) {
        struct schfs_attr_rec *a = (struct schfs_attr_rec *)(v->dirbuf + off);
        unsigned step = SCHFS_ATTR_LEN(a->namelen, a->valuelen);
        if (a->namelen == 0 || off + step > len)
            return -2;
        const char *nm = (const char *)a + sizeof(*a);
        if (a->namelen == namelen && memcmp(nm, name, namelen) == 0) {
            memmove(v->dirbuf + off, v->dirbuf + off + step, len - off - step);
            len -= step;
            found = 1;
            break;
        }
        off += step;
    }
    if (!found)
        return 0;

    r = (len == 0) ? schfs_btree_delete(&v->tree, &v->ctx, &k)
                   : schfs_btree_insert(&v->tree, &v->ctx, &k, v->dirbuf, len);
    if (r != 0)
        return r;

    if (indexed && had_old)
        return index_del(v, name, namelen, oldval, oldlen, ino);
    return 0;
}

/* ------------------------------------------------------------------ */
/* Declaring an index, and querying it                                 */
/* ------------------------------------------------------------------ */

// Start indexing an attribute.
//
// Only affects values written *after* this: there is no reindexing pass, and
// pretending otherwise would be the kind of half-truth that makes a query
// answer confidently and wrongly. A caller that wants existing objects in the
// index rewrites their attributes.
int
schfs_index_create(struct schfs_vol *v, const char *name, unsigned namelen)
{
    if (namelen == 0 || namelen > SCHFS_ATTR_NAME_MAX)
        return -1;

    schfs_cache_invalidate();

    if (index_wanted(v, name, namelen))
        return 0;                    /* already indexed */

    if (v->idxnames_len + 1 + namelen > sizeof(v->idxnames))
        return -3;

    v->idxnames[v->idxnames_len] = (char)namelen;
    memmove(v->idxnames + v->idxnames_len + 1, name, namelen);
    v->idxnames_len += 1 + namelen;

    struct schfs_key k;
    schfs_mkkey(&k, SCHFS_INDEX_OBJECTID, SCHFS_KEY_INDEX, SCHFS_INDEX_LIST_OFF);
    return schfs_btree_insert(&v->tree, &v->ctx, &k, v->idxnames,
                              v->idxnames_len);
}

// Every object whose `name` attribute equals `val`.
//
// One tree descent, whatever the size of the filesystem -- which is the entire
// point. Returns how many object ids were written to `out`, or negative on
// error. A caller with a full buffer gets `cap` and no way to ask for more,
// which is a limit worth having rather than an unbounded kernel allocation.
int
schfs_query_equal(struct schfs_vol *v, const char *name, unsigned namelen,
                  const void *val, unsigned vallen, uint64_t *out, int cap)
{
    if (namelen == 0 || namelen > SCHFS_ATTR_NAME_MAX)
        return -1;
    if (!index_wanted(v, name, namelen))
        return -1;                   /* not indexed: there is nothing to scan */

    struct schfs_key k;
    schfs_mkkey(&k, schfs_namehash(name, namelen), SCHFS_KEY_INDEX,
                schfs_namehash((const char *)val, vallen));

    unsigned len = SCHFS_MAX_INLINE;
    int r = schfs_btree_lookup(&v->tree, &v->ctx, &k, v->dirbuf, &len);
    if (r == -1)
        return 0;                    /* no object carries that value */
    if (r != 0)
        return r;

    int n = 0;
    unsigned off = 0;
    while (off + sizeof(struct schfs_index_rec) <= len && n < cap) {
        const struct schfs_index_rec *ir =
            (const struct schfs_index_rec *)(v->dirbuf + off);
        unsigned step = SCHFS_INDEX_LEN(ir->valuelen);
        if (off + step > len)
            return -2;

        // The value is stored, not just its hash, and this is why: two values
        // can hash alike, and a query that returned the wrong file would be a
        // far worse failure than a slow one.
        if (ir->valuelen == vallen &&
            memcmp((const char *)ir + sizeof(*ir), val, vallen) == 0)
            out[n++] = ir->objectid;

        off += step;
    }
    return n;
}