//! Lua scripting engine — translation of upstream src/scripting.rs.
//!
//! Owns a thin wrapper around ziglua (Lua 5.1, the same version Redis uses)
//! and exposes the `redis.call` / `redis.pcall` API that real Redis EVAL
//! scripts rely on. Inside the engine we dispatch on uppercase command
//! names and call into `Store` methods.
//!
//! Like upstream, every `eval` call spins up a fresh Lua VM — no script
//! state leaks across invocations. The Store is shared (locked at the
//! per-method level) so concurrent EVAL on different VMs is safe.
//!
//! Command coverage: the docket-critical subset (DEL, EXISTS, HSET, HGET,
//! HGETALL, HDEL, HEXISTS, ZADD, ZRANGEBYSCORE, ZREM, ZCARD, ZSCORE, XADD,
//! XLEN, XDEL, XGROUP CREATE, XREADGROUP, XACK). Remaining commands land
//! in follow-on translations.

const std = @import("std");
const Allocator = std.mem.Allocator;
const zlua = @import("zlua");
const Lua = zlua.Lua;

const store_mod = @import("store.zig");
const Store = store_mod.Store;
const StoreError = store_mod.StoreError;
const StreamId = store_mod.StreamId;
const ScoredMember = store_mod.ScoredMember;

// ----------------------------------------------------------------------------
// RedisValue — return type from `redis.call` / `eval`
// ----------------------------------------------------------------------------

pub const RedisValue = union(enum) {
    bulk_string: []u8,
    integer: i64,
    array: []RedisValue,
    nil,
    err: []u8,
    status: []u8,

    pub fn deinit(self: RedisValue, alloc: Allocator) void {
        switch (self) {
            .bulk_string, .err, .status => |s| alloc.free(s),
            .array => |items| {
                for (items) |it| it.deinit(alloc);
                alloc.free(items);
            },
            .integer, .nil => {},
        }
    }
};

// ----------------------------------------------------------------------------
// LuaEngine
// ----------------------------------------------------------------------------

pub const LuaEngine = struct {
    allocator: Allocator,
    store: *Store,

    pub fn init(allocator: Allocator, store: *Store) LuaEngine {
        return .{ .allocator = allocator, .store = store };
    }

    pub fn deinit(_: *LuaEngine) void {}

    /// Compute SHA1 hex digest of a script body. Matches upstream
    /// `LuaEngine::sha1_hex`. Returns the lowercased hex into `out_buf`
    /// (must be at least 40 bytes); returns a slice of the prefix used.
    pub fn sha1Hex(out_buf: *[40]u8, script: []const u8) []u8 {
        var hasher = std.crypto.hash.Sha1.init(.{});
        hasher.update(script);
        var raw: [20]u8 = undefined;
        hasher.final(&raw);

        const hex = "0123456789abcdef";
        for (raw, 0..) |b, i| {
            out_buf[i * 2] = hex[b >> 4];
            out_buf[i * 2 + 1] = hex[b & 0x0f];
        }
        return out_buf[0..40];
    }

    /// EVAL a script with KEYS + ARGV. Returns a heap-allocated RedisValue;
    /// caller frees with `value.deinit(out_alloc)`. Errors raised inside Lua
    /// turn into `RedisValue.err`.
    pub fn eval(
        self: *LuaEngine,
        out_alloc: Allocator,
        script: []const u8,
        keys: []const []const u8,
        args: []const []const u8,
    ) !RedisValue {
        const lua = try Lua.init(self.allocator);
        defer lua.deinit();

        // Open just the libs Redis scripts can rely on.
        lua.openBase();
        lua.openString();
        lua.openTable();
        lua.openMath();

        // Globals: KEYS, ARGV.
        try pushStringSliceTable(lua, keys);
        lua.setGlobal("KEYS");
        try pushStringSliceTable(lua, args);
        lua.setGlobal("ARGV");

        // Per-call DispatchCtx — outlives the Lua VM, captured by the
        // closure as a lightuserdata upvalue.
        var ctx: DispatchCtx = .{ .engine = self, .scratch = out_alloc };

        lua.newTable();
        lua.pushLightUserdata(&ctx);
        lua.pushClosure(luaCall, 1);
        lua.setField(-2, "call");
        lua.pushLightUserdata(&ctx);
        lua.pushClosure(luaPcall, 1);
        lua.setField(-2, "pcall");
        lua.setGlobal("redis");

        // ziglua's loadString demands `[:0]const u8` — stash a NUL-terminated copy.
        const z = try self.allocator.allocSentinel(u8, script.len, 0);
        defer self.allocator.free(z);
        @memcpy(z, script);

        lua.loadString(z) catch {
            const msg = lua.toString(-1) catch "load failed";
            const dup = try out_alloc.dupe(u8, msg);
            return .{ .err = dup };
        };
        // Held for the whole script: no other thread's command lands between
        // two of its calls. Taken only now, with the VM built and the script
        // compiled, to keep it short.
        self.store.mutex.lock();
        defer self.store.mutex.unlock();
        lua.protectedCall(.{ .results = 1 }) catch {
            const msg = lua.toString(-1) catch "script error";
            const dup = try out_alloc.dupe(u8, msg);
            return .{ .err = dup };
        };

        return luaToRedis(out_alloc, lua, -1);
    }
};

// ----------------------------------------------------------------------------
// Dispatch context + Lua callbacks
// ----------------------------------------------------------------------------

const DispatchCtx = struct {
    engine: *LuaEngine,
    scratch: Allocator,
};

fn pushStringSliceTable(lua: *Lua, items: []const []const u8) anyerror!void {
    lua.newTable();
    for (items, 0..) |s, i| {
        _ = lua.pushString(s);
        lua.setIndexRaw(-2, @intCast(i + 1));
    }
}

fn readArgs(alloc: Allocator, lua: *Lua) ![]const []const u8 {
    const top: i32 = lua.getTop();
    if (top <= 0) return &.{};

    var out = try alloc.alloc([]const u8, @intCast(top));
    var i: i32 = 1;
    var n: usize = 0;
    while (i <= top) : (i += 1) {
        const s = lua.toString(i) catch {
            if (lua.isNumber(i)) {
                const num = lua.toNumber(i) catch 0;
                var buf: [64]u8 = undefined;
                const txt = std.fmt.bufPrint(&buf, "{d}", .{num}) catch "0";
                const dup = try alloc.dupe(u8, txt);
                out[n] = dup;
                n += 1;
                continue;
            }
            out[n] = "";
            n += 1;
            continue;
        };
        out[n] = s;
        n += 1;
    }
    return out[0..n];
}

/// `redis.call`: raises a Lua error on failure.
fn luaCall(state: ?*zlua.LuaState) callconv(.c) c_int {
    const lua: *Lua = @ptrCast(state.?);
    const ctx = lua.toUserdata(DispatchCtx, Lua.upvalueIndex(1)) catch {
        lua.raiseErrorStr("redis.call: internal context missing", .{});
    };
    return dispatchCall(ctx, lua, false);
}

/// `redis.pcall`: on failure pushes `{err="..."}` instead of raising.
fn luaPcall(state: ?*zlua.LuaState) callconv(.c) c_int {
    const lua: *Lua = @ptrCast(state.?);
    const ctx = lua.toUserdata(DispatchCtx, Lua.upvalueIndex(1)) catch {
        lua.newTable();
        _ = lua.pushString("redis.pcall: internal context missing");
        lua.setField(-2, "err");
        return 1;
    };
    return dispatchCall(ctx, lua, true);
}

fn dispatchCall(ctx: *DispatchCtx, lua: *Lua, pcall: bool) c_int {
    var arena = std.heap.ArenaAllocator.init(ctx.scratch);
    defer arena.deinit();
    const a = arena.allocator();

    const args = readArgs(a, lua) catch {
        return errorReply(lua, pcall, "ERR redis.call: read args failed");
    };
    if (args.len == 0) return errorReply(lua, pcall, "ERR wrong number of arguments for 'redis.call'");

    const cmd_in = args[0];
    var ucmd_buf: [32]u8 = undefined;
    if (cmd_in.len > ucmd_buf.len) return errorReply(lua, pcall, "ERR unknown command");
    var i: usize = 0;
    while (i < cmd_in.len) : (i += 1) {
        ucmd_buf[i] = std.ascii.toUpper(cmd_in[i]);
    }
    const cmd = ucmd_buf[0..cmd_in.len];

    return dispatchInner(ctx, lua, pcall, cmd, args[1..]) catch |err| {
        return errorReply(lua, pcall, @errorName(err));
    };
}

fn errorReply(lua: *Lua, pcall: bool, msg: []const u8) c_int {
    if (pcall) {
        lua.newTable();
        _ = lua.pushString(msg);
        lua.setField(-2, "err");
        return 1;
    } else {
        _ = lua.pushString(msg);
        lua.raiseError();
    }
}

fn redisErrorMsg(err: StoreError) []const u8 {
    return switch (err) {
        StoreError.WrongType => "WRONGTYPE Operation against a key holding the wrong kind of value",
        StoreError.NoGroup => "NOGROUP No such key or consumer group",
        StoreError.BusyGroup => "BUSYGROUP Consumer Group name already exists",
        StoreError.KeyNotFound => "ERR The XGROUP subcommand requires the key to exist",
        StoreError.IntegerOverflow => "ERR increment or decrement would overflow",
        StoreError.Syntax => "ERR syntax error",
        StoreError.NoSuchKey => "ERR no such key",
        StoreError.OutOfMemory => "ERR out of memory",
    };
}

// ----------------------------------------------------------------------------
// Per-command dispatch
// ----------------------------------------------------------------------------

fn dispatchInner(
    ctx: *DispatchCtx,
    lua: *Lua,
    pcall: bool,
    cmd: []const u8,
    args: []const []const u8,
) !c_int {
    const store = ctx.engine.store;
    const scratch = ctx.scratch;

    if (std.mem.eql(u8, cmd, "DEL")) {
        if (args.len == 0) return errorReply(lua, pcall, "ERR wrong number of arguments");
        lua.pushInteger(store.delete(args));
        return 1;
    }

    // --- Strings -----------------------------------------------------------
    if (std.mem.eql(u8, cmd, "SET")) {
        if (args.len < 2) return errorReply(lua, pcall, "ERR wrong number of arguments for 'set'");
        var flags: Store.SetFlags = .{};
        var i: usize = 2;
        while (i < args.len) {
            if (std.ascii.eqlIgnoreCase(args[i], "NX")) {
                flags.nx = true;
                i += 1;
            } else if (std.ascii.eqlIgnoreCase(args[i], "XX")) {
                flags.xx = true;
                i += 1;
            } else if (std.ascii.eqlIgnoreCase(args[i], "EX")) {
                if (i + 1 >= args.len) return errorReply(lua, pcall, "ERR syntax error");
                const sec = std.fmt.parseInt(i64, args[i + 1], 10) catch
                    return errorReply(lua, pcall, "ERR value is not an integer");
                flags.ttl_ns = @as(i128, sec) * std.time.ns_per_s;
                i += 2;
            } else if (std.ascii.eqlIgnoreCase(args[i], "PX")) {
                if (i + 1 >= args.len) return errorReply(lua, pcall, "ERR syntax error");
                const ms = std.fmt.parseInt(i64, args[i + 1], 10) catch
                    return errorReply(lua, pcall, "ERR value is not an integer");
                flags.ttl_ns = @as(i128, ms) * std.time.ns_per_ms;
                i += 2;
            } else break;
        }
        const ok = store.set(args[0], args[1], flags) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        if (ok) {
            lua.newTable();
            _ = lua.pushString("OK");
            lua.setField(-2, "ok");
        } else {
            lua.pushBoolean(false);
        }
        return 1;
    }

    if (std.mem.eql(u8, cmd, "GET")) {
        if (args.len != 1) return errorReply(lua, pcall, "ERR wrong number of arguments for 'get'");
        const v = store.get(scratch, args[0]) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        if (v) |b| {
            _ = lua.pushString(b);
            scratch.free(b);
        } else lua.pushBoolean(false);
        return 1;
    }

    if (std.mem.eql(u8, cmd, "MGET")) {
        if (args.len == 0) return errorReply(lua, pcall, "ERR wrong number of arguments for 'mget'");
        const res = store.mget(scratch, args) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        defer Store.freeMGetResult(scratch, res);
        lua.newTable();
        for (res, 0..) |opt, i| {
            if (opt) |b| _ = lua.pushString(b) else lua.pushBoolean(false);
            lua.setIndexRaw(-2, @intCast(i + 1));
        }
        return 1;
    }

    if (std.mem.eql(u8, cmd, "KEYS")) {
        if (args.len != 1) return errorReply(lua, pcall, "ERR wrong number of arguments for 'keys'");
        const out = store.keysGlob(scratch, args[0]) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        defer Store.freeKeys(scratch, out);
        lua.newTable();
        for (out, 0..) |k, i| {
            _ = lua.pushString(k);
            lua.setIndexRaw(-2, @intCast(i + 1));
        }
        return 1;
    }

    if (std.mem.eql(u8, cmd, "TTL")) {
        if (args.len != 1) return errorReply(lua, pcall, "ERR wrong number of arguments for 'ttl'");
        lua.pushInteger(store.ttl(args[0]));
        return 1;
    }

    if (std.mem.eql(u8, cmd, "EXPIRE")) {
        if (args.len != 2) return errorReply(lua, pcall, "ERR wrong number of arguments for 'expire'");
        const sec = std.fmt.parseInt(u64, args[1], 10) catch
            return errorReply(lua, pcall, "ERR value is not an integer");
        lua.pushInteger(if (store.expire(args[0], sec)) @as(i64, 1) else 0);
        return 1;
    }

    if (std.mem.eql(u8, cmd, "INCR") or std.mem.eql(u8, cmd, "INCRBY") or
        std.mem.eql(u8, cmd, "DECR") or std.mem.eql(u8, cmd, "DECRBY"))
    {
        const is_decr = std.ascii.startsWithIgnoreCase(cmd, "DECR");
        const has_arg = std.mem.eql(u8, cmd, "INCRBY") or std.mem.eql(u8, cmd, "DECRBY");
        const arg_count: usize = if (has_arg) 2 else 1;
        if (args.len != arg_count) return errorReply(lua, pcall, "ERR wrong number of arguments");
        const delta: i64 = if (has_arg) (std.fmt.parseInt(i64, args[1], 10) catch
            return errorReply(lua, pcall, "ERR value is not an integer")) else 1;
        const n = (if (is_decr) store.decrby(args[0], delta) else store.incrby(args[0], delta)) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        lua.pushInteger(n);
        return 1;
    }

    // --- Sets --------------------------------------------------------------
    if (std.mem.eql(u8, cmd, "SADD")) {
        if (args.len < 2) return errorReply(lua, pcall, "ERR wrong number of arguments for 'sadd'");
        const n = store.sadd(args[0], args[1..]) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        lua.pushInteger(n);
        return 1;
    }

    if (std.mem.eql(u8, cmd, "SMEMBERS")) {
        if (args.len != 1) return errorReply(lua, pcall, "ERR wrong number of arguments for 'smembers'");
        const out = store.smembers(scratch, args[0]) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        defer Store.freeKeys(scratch, out);
        lua.newTable();
        for (out, 0..) |m, i| {
            _ = lua.pushString(m);
            lua.setIndexRaw(-2, @intCast(i + 1));
        }
        return 1;
    }

    if (std.mem.eql(u8, cmd, "SISMEMBER")) {
        if (args.len != 2) return errorReply(lua, pcall, "ERR wrong number of arguments for 'sismember'");
        const r = store.sismember(args[0], args[1]) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        lua.pushInteger(r);
        return 1;
    }

    if (std.mem.eql(u8, cmd, "SREM")) {
        if (args.len < 2) return errorReply(lua, pcall, "ERR wrong number of arguments for 'srem'");
        const n = store.srem(args[0], args[1..]) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        lua.pushInteger(n);
        return 1;
    }

    if (std.mem.eql(u8, cmd, "SCARD")) {
        if (args.len != 1) return errorReply(lua, pcall, "ERR wrong number of arguments for 'scard'");
        const n = store.scard(args[0]) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        lua.pushInteger(n);
        return 1;
    }

    // --- Lists -------------------------------------------------------------
    if (std.mem.eql(u8, cmd, "LPUSH") or std.mem.eql(u8, cmd, "RPUSH")) {
        if (args.len < 2) return errorReply(lua, pcall, "ERR wrong number of arguments");
        const n = if (std.mem.eql(u8, cmd, "LPUSH"))
            (store.lpush(args[0], args[1..]) catch |err| return errorReply(lua, pcall, redisErrorMsg(err)))
        else
            (store.rpush(args[0], args[1..]) catch |err| return errorReply(lua, pcall, redisErrorMsg(err)));
        lua.pushInteger(n);
        return 1;
    }

    if (std.mem.eql(u8, cmd, "LLEN")) {
        if (args.len != 1) return errorReply(lua, pcall, "ERR wrong number of arguments for 'llen'");
        const n = store.llen(args[0]) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        lua.pushInteger(n);
        return 1;
    }

    if (std.mem.eql(u8, cmd, "LINDEX")) {
        if (args.len != 2) return errorReply(lua, pcall, "ERR wrong number of arguments for 'lindex'");
        const idx = std.fmt.parseInt(i64, args[1], 10) catch
            return errorReply(lua, pcall, "ERR value is not an integer");
        const v = store.lindex(scratch, args[0], idx) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        if (v) |b| {
            _ = lua.pushString(b);
            scratch.free(b);
        } else lua.pushBoolean(false);
        return 1;
    }

    if (std.mem.eql(u8, cmd, "LRANGE")) {
        if (args.len != 3) return errorReply(lua, pcall, "ERR wrong number of arguments for 'lrange'");
        const start = std.fmt.parseInt(i64, args[1], 10) catch
            return errorReply(lua, pcall, "ERR not integer");
        const stop = std.fmt.parseInt(i64, args[2], 10) catch
            return errorReply(lua, pcall, "ERR not integer");
        const out = store.lrange(scratch, args[0], start, stop) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        defer Store.freeKeys(scratch, out);
        lua.newTable();
        for (out, 0..) |v, i| {
            _ = lua.pushString(v);
            lua.setIndexRaw(-2, @intCast(i + 1));
        }
        return 1;
    }

    if (std.mem.eql(u8, cmd, "LPOP") or std.mem.eql(u8, cmd, "RPOP")) {
        if (args.len < 1 or args.len > 2) return errorReply(lua, pcall, "ERR wrong number of arguments");
        const count: ?usize = if (args.len == 2)
            (std.fmt.parseInt(usize, args[1], 10) catch return errorReply(lua, pcall, "ERR not integer"))
        else
            null;
        const result = if (std.mem.eql(u8, cmd, "LPOP"))
            (store.lpop(scratch, args[0], count) catch |err| return errorReply(lua, pcall, redisErrorMsg(err)))
        else
            (store.rpop(scratch, args[0], count) catch |err| return errorReply(lua, pcall, redisErrorMsg(err)));
        defer Store.freeLPopResult(scratch, result);
        switch (result) {
            .nil => lua.pushBoolean(false),
            .single => |s| _ = lua.pushString(s),
            .array => |arr| {
                lua.newTable();
                for (arr, 0..) |v, i| {
                    _ = lua.pushString(v);
                    lua.setIndexRaw(-2, @intCast(i + 1));
                }
            },
        }
        return 1;
    }

    if (std.mem.eql(u8, cmd, "ZRANGE")) {
        if (args.len < 3) return errorReply(lua, pcall, "ERR wrong number of arguments for 'zrange'");
        const start = std.fmt.parseInt(i64, args[1], 10) catch
            return errorReply(lua, pcall, "ERR not integer");
        const stop = std.fmt.parseInt(i64, args[2], 10) catch
            return errorReply(lua, pcall, "ERR not integer");
        var with_scores = false;
        var i: usize = 3;
        while (i < args.len) : (i += 1) {
            if (std.ascii.eqlIgnoreCase(args[i], "WITHSCORES")) with_scores = true;
        }
        const sms = store.zrange(scratch, args[0], start, stop, with_scores) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        defer Store.freeScoredMembers(scratch, sms);
        lua.newTable();
        var idx: i32 = 1;
        for (sms) |sm| {
            _ = lua.pushString(sm.member);
            lua.setIndexRaw(-2, idx);
            idx += 1;
            if (with_scores) {
                var buf: [64]u8 = undefined;
                const txt = std.fmt.bufPrint(&buf, "{d}", .{sm.score}) catch "0";
                _ = lua.pushString(txt);
                lua.setIndexRaw(-2, idx);
                idx += 1;
            }
        }
        return 1;
    }

    if (std.mem.eql(u8, cmd, "ZCOUNT")) {
        if (args.len != 3) return errorReply(lua, pcall, "ERR wrong number of arguments for 'zcount'");
        const min = parseScoreBound(args[1]) catch
            return errorReply(lua, pcall, "ERR min is not a valid float");
        const max = parseScoreBound(args[2]) catch
            return errorReply(lua, pcall, "ERR max is not a valid float");
        const n = store.zcount(args[0], min, max) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        lua.pushInteger(n);
        return 1;
    }

    if (std.mem.eql(u8, cmd, "EXISTS")) {
        if (args.len == 0) return errorReply(lua, pcall, "ERR wrong number of arguments");
        lua.pushInteger(store.exists(args));
        return 1;
    }

    if (std.mem.eql(u8, cmd, "HSET")) {
        if (args.len < 3 or (args.len - 1) % 2 != 0)
            return errorReply(lua, pcall, "ERR wrong number of arguments for 'hset'");
        const n = store.hset(args[0], args[1..]) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        lua.pushInteger(n);
        return 1;
    }

    if (std.mem.eql(u8, cmd, "HGET")) {
        if (args.len != 2) return errorReply(lua, pcall, "ERR wrong number of arguments for 'hget'");
        const v = store.hget(scratch, args[0], args[1]) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        if (v) |bytes| {
            _ = lua.pushString(bytes);
            scratch.free(bytes);
        } else {
            lua.pushBoolean(false);
        }
        return 1;
    }

    if (std.mem.eql(u8, cmd, "HGETALL")) {
        if (args.len != 1) return errorReply(lua, pcall, "ERR wrong number of arguments for 'hgetall'");
        const pairs = store.hgetall(scratch, args[0]) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        defer Store.freeHashPairs(scratch, pairs);

        lua.newTable();
        var idx: i32 = 1;
        for (pairs) |p| {
            _ = lua.pushString(p.field);
            lua.setIndexRaw(-2, idx);
            idx += 1;
            _ = lua.pushString(p.value);
            lua.setIndexRaw(-2, idx);
            idx += 1;
        }
        return 1;
    }

    if (std.mem.eql(u8, cmd, "HDEL")) {
        if (args.len < 2) return errorReply(lua, pcall, "ERR wrong number of arguments for 'hdel'");
        const n = store.hdel(args[0], args[1..]) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        lua.pushInteger(n);
        return 1;
    }

    if (std.mem.eql(u8, cmd, "HEXISTS")) {
        if (args.len != 2) return errorReply(lua, pcall, "ERR wrong number of arguments for 'hexists'");
        const r = store.hexists(args[0], args[1]) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        lua.pushInteger(r);
        return 1;
    }

    if (std.mem.eql(u8, cmd, "ZADD")) return try doZadd(ctx, lua, pcall, args);
    if (std.mem.eql(u8, cmd, "ZRANGEBYSCORE")) return try doZrangebyscore(ctx, lua, pcall, args);

    if (std.mem.eql(u8, cmd, "ZREM")) {
        if (args.len < 2) return errorReply(lua, pcall, "ERR wrong number of arguments for 'zrem'");
        const n = store.zrem(args[0], args[1..]) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        lua.pushInteger(n);
        return 1;
    }

    if (std.mem.eql(u8, cmd, "ZREMRANGEBYSCORE")) {
        if (args.len != 3) return errorReply(lua, pcall, "ERR wrong number of arguments for 'zremrangebyscore'");
        const min = parseScoreBound(args[1]) catch
            return errorReply(lua, pcall, "ERR min is not a valid float");
        const max = parseScoreBound(args[2]) catch
            return errorReply(lua, pcall, "ERR max is not a valid float");
        const n = store.zremrangebyscore(args[0], min, max) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        lua.pushInteger(n);
        return 1;
    }

    if (std.mem.eql(u8, cmd, "ZCARD")) {
        if (args.len != 1) return errorReply(lua, pcall, "ERR wrong number of arguments for 'zcard'");
        const n = store.zcard(args[0]) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        lua.pushInteger(n);
        return 1;
    }

    if (std.mem.eql(u8, cmd, "ZSCORE")) {
        if (args.len != 2) return errorReply(lua, pcall, "ERR wrong number of arguments for 'zscore'");
        const sc = store.zscore(args[0], args[1]) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        if (sc) |s| {
            var buf: [64]u8 = undefined;
            const txt = std.fmt.bufPrint(&buf, "{d}", .{s}) catch return errorReply(lua, pcall, "ERR format zscore");
            _ = lua.pushString(txt);
        } else {
            lua.pushBoolean(false);
        }
        return 1;
    }

    if (std.mem.eql(u8, cmd, "XADD")) return try doXadd(ctx, lua, pcall, args);

    if (std.mem.eql(u8, cmd, "XLEN")) {
        if (args.len != 1) return errorReply(lua, pcall, "ERR wrong number of arguments for 'xlen'");
        const n = store.xlen(args[0]) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        lua.pushInteger(n);
        return 1;
    }

    if (std.mem.eql(u8, cmd, "XDEL")) {
        if (args.len < 2) return errorReply(lua, pcall, "ERR wrong number of arguments for 'xdel'");
        var ids = try scratch.alloc(StreamId, args.len - 1);
        defer scratch.free(ids);
        for (args[1..], 0..) |arg, i| {
            ids[i] = StreamId.parse(arg) orelse return errorReply(lua, pcall, "ERR invalid stream ID");
        }
        const n = store.xdel(args[0], ids) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        lua.pushInteger(n);
        return 1;
    }

    if (std.mem.eql(u8, cmd, "XGROUP")) return try doXgroup(ctx, lua, pcall, args);
    if (std.mem.eql(u8, cmd, "XREADGROUP")) return try doXreadgroup(ctx, lua, pcall, args);

    if (std.mem.eql(u8, cmd, "XACK")) {
        if (args.len < 3) return errorReply(lua, pcall, "ERR wrong number of arguments for 'xack'");
        var ids = try scratch.alloc(StreamId, args.len - 2);
        defer scratch.free(ids);
        for (args[2..], 0..) |arg, i| {
            ids[i] = StreamId.parse(arg) orelse return errorReply(lua, pcall, "ERR invalid stream ID");
        }
        const n = store.xack(args[0], args[1], ids) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        lua.pushInteger(n);
        return 1;
    }

    std.log.warn("burner-redis: unknown command '{s}' (docket-critical subset only)", .{cmd});
    return errorReply(lua, pcall, "ERR unknown command (docket-critical subset only)");
}

// ---- per-command helpers ----------------------------------------------------

fn doZadd(ctx: *DispatchCtx, lua: *Lua, pcall: bool, args: []const []const u8) !c_int {
    if (args.len < 3) return errorReply(lua, pcall, "ERR wrong number of arguments for 'zadd'");
    const key = args[0];
    var flags: Store.ZAddFlags = .{};
    var i: usize = 1;
    while (i < args.len) {
        const a = args[i];
        if (std.ascii.eqlIgnoreCase(a, "NX")) {
            flags.nx = true;
            i += 1;
        } else if (std.ascii.eqlIgnoreCase(a, "XX")) {
            flags.xx = true;
            i += 1;
        } else if (std.ascii.eqlIgnoreCase(a, "GT")) {
            flags.gt = true;
            i += 1;
        } else if (std.ascii.eqlIgnoreCase(a, "LT")) {
            flags.lt = true;
            i += 1;
        } else if (std.ascii.eqlIgnoreCase(a, "CH")) {
            flags.ch = true;
            i += 1;
        } else break;
    }
    if (i >= args.len or (args.len - i) % 2 != 0) {
        return errorReply(lua, pcall, "ERR wrong number of arguments for 'zadd'");
    }

    var pairs: std.ArrayList(ScoredMember) = .empty;
    defer pairs.deinit(ctx.scratch);
    var j: usize = i;
    while (j < args.len) : (j += 2) {
        const score = std.fmt.parseFloat(f64, args[j]) catch
            return errorReply(lua, pcall, "ERR value is not a valid float");
        try pairs.append(ctx.scratch, .{ .score = score, .member = args[j + 1] });
    }

    const n = ctx.engine.store.zadd(key, pairs.items, flags) catch |err|
        return errorReply(lua, pcall, redisErrorMsg(err));
    lua.pushInteger(n);
    return 1;
}

fn doZrangebyscore(ctx: *DispatchCtx, lua: *Lua, pcall: bool, args: []const []const u8) !c_int {
    if (args.len < 3) return errorReply(lua, pcall, "ERR wrong number of arguments for 'zrangebyscore'");
    const key = args[0];
    const min = parseScoreBound(args[1]) catch
        return errorReply(lua, pcall, "ERR min is not a valid float");
    const max = parseScoreBound(args[2]) catch
        return errorReply(lua, pcall, "ERR max is not a valid float");

    var with_scores = false;
    var i: usize = 3;
    while (i < args.len) : (i += 1) {
        if (std.ascii.eqlIgnoreCase(args[i], "WITHSCORES")) with_scores = true;
    }

    const sms = ctx.engine.store.zrangebyscore(ctx.scratch, key, min, max, with_scores) catch |err|
        return errorReply(lua, pcall, redisErrorMsg(err));
    defer Store.freeScoredMembers(ctx.scratch, sms);

    lua.newTable();
    var idx: i32 = 1;
    for (sms) |sm| {
        _ = lua.pushString(sm.member);
        lua.setIndexRaw(-2, idx);
        idx += 1;
        if (with_scores) {
            var buf: [64]u8 = undefined;
            const txt = std.fmt.bufPrint(&buf, "{d}", .{sm.score}) catch "0";
            _ = lua.pushString(txt);
            lua.setIndexRaw(-2, idx);
            idx += 1;
        }
    }
    return 1;
}

fn parseScoreBound(s: []const u8) !f64 {
    if (std.mem.eql(u8, s, "-inf") or std.mem.eql(u8, s, "-INF")) return -std.math.inf(f64);
    if (std.mem.eql(u8, s, "+inf") or std.mem.eql(u8, s, "+INF") or std.mem.eql(u8, s, "inf"))
        return std.math.inf(f64);
    return std.fmt.parseFloat(f64, s);
}

fn doXadd(ctx: *DispatchCtx, lua: *Lua, pcall: bool, args: []const []const u8) !c_int {
    if (args.len < 4) return errorReply(lua, pcall, "ERR wrong number of arguments for 'xadd'");
    const key = args[0];
    const id_arg = args[1];

    const id: ?StreamId = if (std.mem.eql(u8, id_arg, "*")) null else StreamId.parse(id_arg);
    if (!std.mem.eql(u8, id_arg, "*") and id == null) {
        return errorReply(lua, pcall, "ERR invalid stream ID");
    }

    const fields = args[2..];
    const new_id = ctx.engine.store.xadd(key, id, fields) catch |err|
        return errorReply(lua, pcall, redisErrorMsg(err));

    var buf: [48]u8 = undefined;
    const txt = std.fmt.bufPrint(&buf, "{d}-{d}", .{ new_id.ms, new_id.seq }) catch "0-0";
    _ = lua.pushString(txt);
    return 1;
}

fn doXgroup(ctx: *DispatchCtx, lua: *Lua, pcall: bool, args: []const []const u8) !c_int {
    if (args.len < 1) return errorReply(lua, pcall, "ERR wrong number of arguments for 'xgroup'");
    const sub = args[0];

    if (std.ascii.eqlIgnoreCase(sub, "CREATE")) {
        if (args.len < 4) return errorReply(lua, pcall, "ERR wrong number of arguments for 'xgroup'");
        const key = args[1];
        const group = args[2];
        const id_arg = args[3];

        var mkstream = false;
        for (args[4..]) |opt| {
            if (std.ascii.eqlIgnoreCase(opt, "MKSTREAM")) mkstream = true;
        }

        const id: StreamId = if (std.mem.eql(u8, id_arg, "$") or std.mem.eql(u8, id_arg, "0"))
            StreamId.zero()
        else
            StreamId.parse(id_arg) orelse
                return errorReply(lua, pcall, "ERR invalid stream ID");

        ctx.engine.store.xgroupCreate(key, group, .{ .id = id, .mkstream = mkstream }) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        _ = lua.pushString("OK");
        return 1;
    }

    return errorReply(lua, pcall, "ERR unknown XGROUP subcommand (docket-critical subset only)");
}

fn doXreadgroup(ctx: *DispatchCtx, lua: *Lua, pcall: bool, args: []const []const u8) !c_int {
    // XREADGROUP GROUP <group> <consumer> [COUNT n] [BLOCK ms] [NOACK] STREAMS key [key...] id [id...]
    var i: usize = 0;
    if (i >= args.len or !std.ascii.eqlIgnoreCase(args[i], "GROUP"))
        return errorReply(lua, pcall, "ERR syntax error");
    i += 1;
    if (i + 2 > args.len) return errorReply(lua, pcall, "ERR syntax error");
    const group = args[i];
    i += 1;
    const consumer = args[i];
    i += 1;

    var count: ?usize = null;

    while (i < args.len) {
        if (std.ascii.eqlIgnoreCase(args[i], "COUNT")) {
            if (i + 1 >= args.len) return errorReply(lua, pcall, "ERR syntax error");
            count = std.fmt.parseInt(usize, args[i + 1], 10) catch
                return errorReply(lua, pcall, "ERR not an integer");
            i += 2;
        } else if (std.ascii.eqlIgnoreCase(args[i], "BLOCK")) {
            // skip BLOCK ms — no in-engine blocking yet
            i += 2;
        } else if (std.ascii.eqlIgnoreCase(args[i], "NOACK")) {
            i += 1;
        } else break;
    }

    if (i >= args.len or !std.ascii.eqlIgnoreCase(args[i], "STREAMS"))
        return errorReply(lua, pcall, "ERR syntax error");
    i += 1;

    const rest = args[i..];
    if (rest.len == 0 or rest.len % 2 != 0) return errorReply(lua, pcall, "ERR syntax error");
    const half = rest.len / 2;
    const keys = rest[0..half];
    const ids = rest[half..];

    lua.newTable();
    var top_idx: i32 = 1;

    for (keys, ids) |key, id_str| {
        const id_form: Store.ReadIdForm = if (std.mem.eql(u8, id_str, ">"))
            .new_messages
        else if (StreamId.parse(id_str)) |id|
            .{ .from_pel = id }
        else
            return errorReply(lua, pcall, "ERR invalid stream ID");

        const delivered = ctx.engine.store.xreadgroupOne(ctx.scratch, key, group, consumer, id_form, count) catch |err|
            return errorReply(lua, pcall, redisErrorMsg(err));
        defer store_mod.freeDeliveredEntries(ctx.scratch, delivered);

        if (delivered.len == 0) continue;

        // Per-stream pair: [name, entries-table]
        lua.newTable();
        _ = lua.pushString(key);
        lua.setIndexRaw(-2, 1);

        lua.newTable();
        for (delivered, 0..) |de, j| {
            // entry pair: [id_str, [field, value, ...]]
            lua.newTable();
            var id_buf: [48]u8 = undefined;
            const id_str_out = std.fmt.bufPrint(&id_buf, "{d}-{d}", .{ de.id.ms, de.id.seq }) catch "0-0";
            _ = lua.pushString(id_str_out);
            lua.setIndexRaw(-2, 1);

            lua.newTable();
            var fld_idx: i32 = 1;
            for (de.fields) |p| {
                _ = lua.pushString(p.field);
                lua.setIndexRaw(-2, fld_idx);
                fld_idx += 1;
                _ = lua.pushString(p.value);
                lua.setIndexRaw(-2, fld_idx);
                fld_idx += 1;
            }
            lua.setIndexRaw(-2, 2);
            lua.setIndexRaw(-2, @intCast(j + 1));
        }
        lua.setIndexRaw(-2, 2);

        lua.setIndexRaw(-2, top_idx);
        top_idx += 1;
    }

    if (top_idx == 1) {
        // No streams produced — Redis returns nil for XREADGROUP with no data.
        lua.pop(1);
        lua.pushBoolean(false);
    }
    return 1;
}

// ---- Lua → RedisValue conversion -------------------------------------------

fn luaToRedis(out_alloc: Allocator, lua: *Lua, index: i32) anyerror!RedisValue {
    return switch (lua.typeOf(index)) {
        .nil => .nil,
        .boolean => if (lua.toBoolean(index)) RedisValue{ .integer = 1 } else RedisValue.nil,
        .number => blk: {
            const n = lua.toNumber(index) catch break :blk RedisValue.nil;
            const as_i: i64 = @intFromFloat(n);
            break :blk .{ .integer = as_i };
        },
        .string => blk: {
            const s = lua.toString(index) catch break :blk RedisValue.nil;
            break :blk .{ .bulk_string = try out_alloc.dupe(u8, s) };
        },
        .table => try luaTableToRedis(out_alloc, lua, index),
        else => .nil,
    };
}

fn luaTableToRedis(out_alloc: Allocator, lua: *Lua, index: i32) anyerror!RedisValue {
    const abs = if (index < 0) lua.getTop() + index + 1 else index;

    // {err = "..."} → error
    _ = lua.getField(abs, "err");
    if (lua.typeOf(-1) == .string) {
        const msg = lua.toString(-1) catch unreachable;
        const dup = try out_alloc.dupe(u8, msg);
        lua.pop(1);
        return .{ .err = dup };
    }
    lua.pop(1);

    // {ok = "..."} → status
    _ = lua.getField(abs, "ok");
    if (lua.typeOf(-1) == .string) {
        const msg = lua.toString(-1) catch unreachable;
        const dup = try out_alloc.dupe(u8, msg);
        lua.pop(1);
        return .{ .status = dup };
    }
    lua.pop(1);

    var items: std.ArrayList(RedisValue) = .empty;
    errdefer {
        for (items.items) |it| it.deinit(out_alloc);
        items.deinit(out_alloc);
    }

    var i: i32 = 1;
    while (true) : (i += 1) {
        lua.pushInteger(@intCast(i));
        _ = lua.getTable(abs);
        if (lua.typeOf(-1) == .nil) {
            lua.pop(1);
            break;
        }
        const v = try luaToRedis(out_alloc, lua, -1);
        lua.pop(1);
        try items.append(out_alloc, v);
    }

    return .{ .array = try items.toOwnedSlice(out_alloc) };
}

// ----------------------------------------------------------------------------
// Tests
// ----------------------------------------------------------------------------

test "ziglua boots and runs return-string" {
    var lua = try Lua.init(std.testing.allocator);
    defer lua.deinit();
    try lua.doString("return 'hello from lua'");
    const got = try lua.toString(-1);
    try std.testing.expectEqualStrings("hello from lua", got);
}

test "sha1Hex matches a known vector" {
    var buf: [40]u8 = undefined;
    const h = LuaEngine.sha1Hex(&buf, "");
    try std.testing.expectEqualStrings("da39a3ee5e6b4b0d3255bfef95601890afd80709", h);
}

test "EVAL returning a string" {
    var threaded = std.Io.Threaded.init(std.testing.allocator, .{});
    defer threaded.deinit();
    var store = Store.init(std.testing.allocator, threaded.io());
    defer store.deinit();

    var engine = LuaEngine.init(std.testing.allocator, &store);
    defer engine.deinit();

    const v = try engine.eval(std.testing.allocator, "return 'hello'", &.{}, &.{});
    defer v.deinit(std.testing.allocator);
    try std.testing.expect(v == .bulk_string);
    try std.testing.expectEqualStrings("hello", v.bulk_string);
}

test "EVAL: redis.call HSET + HGET roundtrips" {
    var threaded = std.Io.Threaded.init(std.testing.allocator, .{});
    defer threaded.deinit();
    var store = Store.init(std.testing.allocator, threaded.io());
    defer store.deinit();

    var engine = LuaEngine.init(std.testing.allocator, &store);
    defer engine.deinit();

    const script1 =
        \\redis.call('HSET', KEYS[1], 'f1', 'v1', 'f2', 'v2')
        \\return redis.call('HGET', KEYS[1], 'f1')
    ;
    const v = try engine.eval(std.testing.allocator, script1, &.{"h"}, &.{});
    defer v.deinit(std.testing.allocator);
    try std.testing.expect(v == .bulk_string);
    try std.testing.expectEqualStrings("v1", v.bulk_string);
}

test "EVAL: SET/GET via redis.call" {
    var threaded = std.Io.Threaded.init(std.testing.allocator, .{});
    defer threaded.deinit();
    var store = Store.init(std.testing.allocator, threaded.io());
    defer store.deinit();
    var engine = LuaEngine.init(std.testing.allocator, &store);
    defer engine.deinit();

    const script =
        \\redis.call('SET', KEYS[1], 'hello')
        \\return redis.call('GET', KEYS[1])
    ;
    const v = try engine.eval(std.testing.allocator, script, &.{"k"}, &.{});
    defer v.deinit(std.testing.allocator);
    try std.testing.expect(v == .bulk_string);
    try std.testing.expectEqualStrings("hello", v.bulk_string);
}

test "EVAL: list LPUSH/RPUSH/LRANGE" {
    var threaded = std.Io.Threaded.init(std.testing.allocator, .{});
    defer threaded.deinit();
    var store = Store.init(std.testing.allocator, threaded.io());
    defer store.deinit();
    var engine = LuaEngine.init(std.testing.allocator, &store);
    defer engine.deinit();

    const script =
        \\redis.call('RPUSH', KEYS[1], 'a', 'b', 'c')
        \\local r = redis.call('LRANGE', KEYS[1], '0', '-1')
        \\return #r
    ;
    const v = try engine.eval(std.testing.allocator, script, &.{"L"}, &.{});
    defer v.deinit(std.testing.allocator);
    try std.testing.expect(v == .integer);
    try std.testing.expectEqual(@as(i64, 3), v.integer);
}

test "EVAL: full XGROUP + XADD + XREADGROUP cycle (docket-critical)" {
    var threaded = std.Io.Threaded.init(std.testing.allocator, .{});
    defer threaded.deinit();
    var store = Store.init(std.testing.allocator, threaded.io());
    defer store.deinit();

    var engine = LuaEngine.init(std.testing.allocator, &store);
    defer engine.deinit();

    const script =
        \\redis.call('XGROUP', 'CREATE', KEYS[1], 'g1', '0', 'MKSTREAM')
        \\redis.call('XADD', KEYS[1], '*', 'function', 'tick', 'attempt', '1')
        \\local r = redis.call('XREADGROUP', 'GROUP', 'g1', 'c1', 'COUNT', '1', 'STREAMS', KEYS[1], '>')
        \\if r and #r > 0 then return 'got' else return 'empty' end
    ;
    const v = try engine.eval(std.testing.allocator, script, &.{"st"}, &.{});
    defer v.deinit(std.testing.allocator);
    try std.testing.expect(v == .bulk_string);
    try std.testing.expectEqualStrings("got", v.bulk_string);
}

const Racer = struct {
    store: *Store,
    done: std.atomic.Value(bool) = .init(false),

    /// what another thread sees while the scripts run: `n` is only ever even
    fn readCounter(self: *Racer, odd: *usize) void {
        var digits: [32]u8 = undefined;
        while (!self.done.load(.monotonic)) {
            var buffer: std.heap.FixedBufferAllocator = .init(&digits);
            const got = (self.store.get(buffer.allocator(), "n") catch unreachable) orelse continue;
            if ((std.fmt.parseInt(u64, got, 10) catch unreachable) % 2 == 1) odd.* += 1;
        }
    }

    /// docket's schedule, for a task already due: park the body, then queue the key
    fn schedule(self: *Racer, tasks: usize) void {
        var parked: [32]u8 = undefined;
        for (0..tasks) |n| {
            const key = std.fmt.bufPrint(&parked, "parked:{d}", .{n}) catch unreachable;
            _ = self.store.hset(key, &.{ "function", "f" }) catch unreachable;
            _ = self.store.zadd("queue", &.{.{ .score = 1, .member = key["parked:".len..] }}, .{}) catch unreachable;
        }
        self.done.store(true, .monotonic);
    }
};

test "no other thread sees a script half done" {
    var threaded = std.Io.Threaded.init(std.testing.allocator, .{});
    defer threaded.deinit();
    var store = Store.init(std.testing.allocator, threaded.io());
    defer store.deinit();
    // not the testing allocator: it takes a stack trace for each of the VM's allocations
    var engine = LuaEngine.init(std.heap.smp_allocator, &store);

    var racer: Racer = .{ .store = &store };
    var odd: usize = 0;
    const reader = try std.Thread.spawn(.{}, Racer.readCounter, .{ &racer, &odd });
    var arena = std.heap.ArenaAllocator.init(std.heap.smp_allocator);
    defer arena.deinit();
    for (0..2000) |_| {
        _ = try engine.eval(arena.allocator(), "redis.call('INCR', KEYS[1]) redis.call('INCR', KEYS[1]) return 1", &.{"n"}, &.{});
        _ = arena.reset(.retain_capacity);
    }
    racer.done.store(true, .monotonic);
    reader.join();
    try std.testing.expectEqual(@as(usize, 0), odd);
}

test "a task scheduled while a script promotes the due ones is not stranded" {
    var threaded = std.Io.Threaded.init(std.testing.allocator, .{});
    defer threaded.deinit();
    // twenty thousand tasks: the testing allocator would take a stack trace for each allocation
    var store = Store.init(std.heap.smp_allocator, threaded.io());
    defer store.deinit();
    var engine = LuaEngine.init(std.heap.smp_allocator, &store);

    const tasks = 20_000;
    var racer: Racer = .{ .store = &store };
    const scheduler = try std.Thread.spawn(.{}, Racer.schedule, .{ &racer, tasks });
    var arena = std.heap.ArenaAllocator.init(std.heap.smp_allocator);
    defer arena.deinit();
    // docket's promote_due: read the due range, move each parked body to the
    // stream, then remove the range by score. A task queued in between is
    // removed with the rest, its body still parked.
    const promote_due =
        \\local promoted = 0
        \\for _, task in ipairs(redis.call('ZRANGEBYSCORE', KEYS[1], 0, ARGV[1])) do
        \\    local fields = redis.call('HGETALL', 'parked:' .. task)
        \\    if #fields > 0 then
        \\        redis.call('XADD', KEYS[2], '*', unpack(fields))
        \\        redis.call('DEL', 'parked:' .. task)
        \\        promoted = promoted + 1
        \\    end
        \\end
        \\if promoted > 0 then redis.call('ZREMRANGEBYSCORE', KEYS[1], 0, ARGV[1]) end
        \\return promoted
    ;
    var promoted: i64 = 0;
    var scheduling = true;
    while (scheduling) {
        // read before the pass, so the last pass starts with every task queued
        scheduling = !racer.done.load(.monotonic);
        promoted += (try engine.eval(arena.allocator(), promote_due, &.{ "queue", "stream" }, &.{"2"})).integer;
        _ = arena.reset(.retain_capacity);
    }
    scheduler.join();
    try std.testing.expectEqual(@as(i64, tasks), promoted);
    try std.testing.expectEqual(@as(i64, tasks), try store.xlen("stream"));
}

test "EVAL integer overflow returns errors without changing the key" {
    var threaded = std.Io.Threaded.init(std.testing.allocator, .{});
    defer threaded.deinit();
    var store = Store.init(std.testing.allocator, threaded.io());
    defer store.deinit();
    var engine = LuaEngine.init(std.testing.allocator, &store);
    defer engine.deinit();
    const cases = [_]struct { value: []const u8, command: []const u8 }{
        .{ .value = "9223372036854775807", .command = "'INCR', 'k'" },
        .{ .value = "9223372036854775807", .command = "'INCRBY', 'k', '1'" },
        .{ .value = "-9223372036854775808", .command = "'DECR', 'k'" },
        .{ .value = "-9223372036854775808", .command = "'DECRBY', 'k', '1'" },
        .{ .value = "-1", .command = "'DECRBY', 'k', '-9223372036854775808'" },
    };
    for (cases) |case| {
        for ([_][]const u8{ "call", "pcall" }) |mode| {
            var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
            defer arena.deinit();
            _ = try store.set("k", case.value, .{});
            const script = try std.fmt.allocPrint(arena.allocator(), "return redis.{s}({s})", .{ mode, case.command });
            const result = try engine.eval(arena.allocator(), script, &.{}, &.{});
            try std.testing.expect(result == .err);
            try std.testing.expect(std.mem.indexOf(u8, result.err, "overflow") != null);
            const value = (try store.get(arena.allocator(), "k")).?;
            try std.testing.expectEqualStrings(case.value, value);
        }
    }
}