expand command surface: strings + sets + lists + zrange/zcount + stream extras · zzstoatzz.io/burner-redis@dcfca15

+1372

2 changed files

Expand all

src

scripting.zig

store.zig

+300

src/scripting.zig

View file

··· 272 272 return 1; 273 273 } 274 274 275 + // --- Strings ----------------------------------------------------------- 276 + if (std.mem.eql(u8, cmd, "SET")) { 277 + if (args.len < 2) return errorReply(lua, pcall, "ERR wrong number of arguments for 'set'"); 278 + var flags: Store.SetFlags = .{}; 279 + var i: usize = 2; 280 + while (i < args.len) { 281 + if (std.ascii.eqlIgnoreCase(args[i], "NX")) { flags.nx = true; i += 1; } 282 + else if (std.ascii.eqlIgnoreCase(args[i], "XX")) { flags.xx = true; i += 1; } 283 + else if (std.ascii.eqlIgnoreCase(args[i], "EX")) { 284 + if (i + 1 >= args.len) return errorReply(lua, pcall, "ERR syntax error"); 285 + const sec = std.fmt.parseInt(i64, args[i + 1], 10) catch 286 + return errorReply(lua, pcall, "ERR value is not an integer"); 287 + flags.ttl_ns = @as(i128, sec) * std.time.ns_per_s; 288 + i += 2; 289 + } else if (std.ascii.eqlIgnoreCase(args[i], "PX")) { 290 + if (i + 1 >= args.len) return errorReply(lua, pcall, "ERR syntax error"); 291 + const ms = std.fmt.parseInt(i64, args[i + 1], 10) catch 292 + return errorReply(lua, pcall, "ERR value is not an integer"); 293 + flags.ttl_ns = @as(i128, ms) * std.time.ns_per_ms; 294 + i += 2; 295 + } else break; 296 + } 297 + const ok = store.set(args[0], args[1], flags) catch |err| 298 + return errorReply(lua, pcall, redisErrorMsg(err)); 299 + if (ok) { 300 + lua.newTable(); 301 + _ = lua.pushString("OK"); 302 + lua.setField(-2, "ok"); 303 + } else { 304 + lua.pushBoolean(false); 305 + } 306 + return 1; 307 + } 308 + 309 + if (std.mem.eql(u8, cmd, "GET")) { 310 + if (args.len != 1) return errorReply(lua, pcall, "ERR wrong number of arguments for 'get'"); 311 + const v = store.get(scratch, args[0]) catch |err| 312 + return errorReply(lua, pcall, redisErrorMsg(err)); 313 + if (v) |b| { 314 + _ = lua.pushString(b); 315 + scratch.free(b); 316 + } else lua.pushBoolean(false); 317 + return 1; 318 + } 319 + 320 + if (std.mem.eql(u8, cmd, "MGET")) { 321 + if (args.len == 0) return errorReply(lua, pcall, "ERR wrong number of arguments for 'mget'"); 322 + const res = store.mget(scratch, args) catch |err| 323 + return errorReply(lua, pcall, redisErrorMsg(err)); 324 + defer Store.freeMGetResult(scratch, res); 325 + lua.newTable(); 326 + for (res, 0..) |opt, i| { 327 + if (opt) |b| _ = lua.pushString(b) else lua.pushBoolean(false); 328 + lua.setIndexRaw(-2, @intCast(i + 1)); 329 + } 330 + return 1; 331 + } 332 + 333 + if (std.mem.eql(u8, cmd, "KEYS")) { 334 + if (args.len != 1) return errorReply(lua, pcall, "ERR wrong number of arguments for 'keys'"); 335 + const out = store.keysGlob(scratch, args[0]) catch |err| 336 + return errorReply(lua, pcall, redisErrorMsg(err)); 337 + defer Store.freeKeys(scratch, out); 338 + lua.newTable(); 339 + for (out, 0..) |k, i| { 340 + _ = lua.pushString(k); 341 + lua.setIndexRaw(-2, @intCast(i + 1)); 342 + } 343 + return 1; 344 + } 345 + 346 + if (std.mem.eql(u8, cmd, "TTL")) { 347 + if (args.len != 1) return errorReply(lua, pcall, "ERR wrong number of arguments for 'ttl'"); 348 + lua.pushInteger(store.ttl(args[0])); 349 + return 1; 350 + } 351 + 352 + if (std.mem.eql(u8, cmd, "EXPIRE")) { 353 + if (args.len != 2) return errorReply(lua, pcall, "ERR wrong number of arguments for 'expire'"); 354 + const sec = std.fmt.parseInt(u64, args[1], 10) catch 355 + return errorReply(lua, pcall, "ERR value is not an integer"); 356 + lua.pushInteger(if (store.expire(args[0], sec)) @as(i64, 1) else 0); 357 + return 1; 358 + } 359 + 360 + if (std.mem.eql(u8, cmd, "INCR") or std.mem.eql(u8, cmd, "INCRBY") or 361 + std.mem.eql(u8, cmd, "DECR") or std.mem.eql(u8, cmd, "DECRBY")) { 362 + const is_decr = std.ascii.startsWithIgnoreCase(cmd, "DECR"); 363 + const has_arg = std.mem.eql(u8, cmd, "INCRBY") or std.mem.eql(u8, cmd, "DECRBY"); 364 + const arg_count: usize = if (has_arg) 2 else 1; 365 + if (args.len != arg_count) return errorReply(lua, pcall, "ERR wrong number of arguments"); 366 + const delta: i64 = if (has_arg) (std.fmt.parseInt(i64, args[1], 10) catch 367 + return errorReply(lua, pcall, "ERR value is not an integer")) else 1; 368 + const adjusted = if (is_decr) -delta else delta; 369 + const n = store.incrby(args[0], adjusted) catch |err| 370 + return errorReply(lua, pcall, redisErrorMsg(err)); 371 + lua.pushInteger(n); 372 + return 1; 373 + } 374 + 375 + // --- Sets -------------------------------------------------------------- 376 + if (std.mem.eql(u8, cmd, "SADD")) { 377 + if (args.len < 2) return errorReply(lua, pcall, "ERR wrong number of arguments for 'sadd'"); 378 + const n = store.sadd(args[0], args[1..]) catch |err| 379 + return errorReply(lua, pcall, redisErrorMsg(err)); 380 + lua.pushInteger(n); 381 + return 1; 382 + } 383 + 384 + if (std.mem.eql(u8, cmd, "SMEMBERS")) { 385 + if (args.len != 1) return errorReply(lua, pcall, "ERR wrong number of arguments for 'smembers'"); 386 + const out = store.smembers(scratch, args[0]) catch |err| 387 + return errorReply(lua, pcall, redisErrorMsg(err)); 388 + defer Store.freeKeys(scratch, out); 389 + lua.newTable(); 390 + for (out, 0..) |m, i| { 391 + _ = lua.pushString(m); 392 + lua.setIndexRaw(-2, @intCast(i + 1)); 393 + } 394 + return 1; 395 + } 396 + 397 + if (std.mem.eql(u8, cmd, "SISMEMBER")) { 398 + if (args.len != 2) return errorReply(lua, pcall, "ERR wrong number of arguments for 'sismember'"); 399 + const r = store.sismember(args[0], args[1]) catch |err| 400 + return errorReply(lua, pcall, redisErrorMsg(err)); 401 + lua.pushInteger(r); 402 + return 1; 403 + } 404 + 405 + if (std.mem.eql(u8, cmd, "SREM")) { 406 + if (args.len < 2) return errorReply(lua, pcall, "ERR wrong number of arguments for 'srem'"); 407 + const n = store.srem(args[0], args[1..]) catch |err| 408 + return errorReply(lua, pcall, redisErrorMsg(err)); 409 + lua.pushInteger(n); 410 + return 1; 411 + } 412 + 413 + if (std.mem.eql(u8, cmd, "SCARD")) { 414 + if (args.len != 1) return errorReply(lua, pcall, "ERR wrong number of arguments for 'scard'"); 415 + const n = store.scard(args[0]) catch |err| 416 + return errorReply(lua, pcall, redisErrorMsg(err)); 417 + lua.pushInteger(n); 418 + return 1; 419 + } 420 + 421 + // --- Lists ------------------------------------------------------------- 422 + if (std.mem.eql(u8, cmd, "LPUSH") or std.mem.eql(u8, cmd, "RPUSH")) { 423 + if (args.len < 2) return errorReply(lua, pcall, "ERR wrong number of arguments"); 424 + const n = if (std.mem.eql(u8, cmd, "LPUSH")) 425 + (store.lpush(args[0], args[1..]) catch |err| return errorReply(lua, pcall, redisErrorMsg(err))) 426 + else 427 + (store.rpush(args[0], args[1..]) catch |err| return errorReply(lua, pcall, redisErrorMsg(err))); 428 + lua.pushInteger(n); 429 + return 1; 430 + } 431 + 432 + if (std.mem.eql(u8, cmd, "LLEN")) { 433 + if (args.len != 1) return errorReply(lua, pcall, "ERR wrong number of arguments for 'llen'"); 434 + const n = store.llen(args[0]) catch |err| 435 + return errorReply(lua, pcall, redisErrorMsg(err)); 436 + lua.pushInteger(n); 437 + return 1; 438 + } 439 + 440 + if (std.mem.eql(u8, cmd, "LINDEX")) { 441 + if (args.len != 2) return errorReply(lua, pcall, "ERR wrong number of arguments for 'lindex'"); 442 + const idx = std.fmt.parseInt(i64, args[1], 10) catch 443 + return errorReply(lua, pcall, "ERR value is not an integer"); 444 + const v = store.lindex(scratch, args[0], idx) catch |err| 445 + return errorReply(lua, pcall, redisErrorMsg(err)); 446 + if (v) |b| { 447 + _ = lua.pushString(b); 448 + scratch.free(b); 449 + } else lua.pushBoolean(false); 450 + return 1; 451 + } 452 + 453 + if (std.mem.eql(u8, cmd, "LRANGE")) { 454 + if (args.len != 3) return errorReply(lua, pcall, "ERR wrong number of arguments for 'lrange'"); 455 + const start = std.fmt.parseInt(i64, args[1], 10) catch 456 + return errorReply(lua, pcall, "ERR not integer"); 457 + const stop = std.fmt.parseInt(i64, args[2], 10) catch 458 + return errorReply(lua, pcall, "ERR not integer"); 459 + const out = store.lrange(scratch, args[0], start, stop) catch |err| 460 + return errorReply(lua, pcall, redisErrorMsg(err)); 461 + defer Store.freeKeys(scratch, out); 462 + lua.newTable(); 463 + for (out, 0..) |v, i| { 464 + _ = lua.pushString(v); 465 + lua.setIndexRaw(-2, @intCast(i + 1)); 466 + } 467 + return 1; 468 + } 469 + 470 + if (std.mem.eql(u8, cmd, "LPOP") or std.mem.eql(u8, cmd, "RPOP")) { 471 + if (args.len < 1 or args.len > 2) return errorReply(lua, pcall, "ERR wrong number of arguments"); 472 + const count: ?usize = if (args.len == 2) 473 + (std.fmt.parseInt(usize, args[1], 10) catch return errorReply(lua, pcall, "ERR not integer")) 474 + else 475 + null; 476 + const result = if (std.mem.eql(u8, cmd, "LPOP")) 477 + (store.lpop(scratch, args[0], count) catch |err| return errorReply(lua, pcall, redisErrorMsg(err))) 478 + else 479 + (store.rpop(scratch, args[0], count) catch |err| return errorReply(lua, pcall, redisErrorMsg(err))); 480 + defer Store.freeLPopResult(scratch, result); 481 + switch (result) { 482 + .nil => lua.pushBoolean(false), 483 + .single => |s| _ = lua.pushString(s), 484 + .array => |arr| { 485 + lua.newTable(); 486 + for (arr, 0..) |v, i| { 487 + _ = lua.pushString(v); 488 + lua.setIndexRaw(-2, @intCast(i + 1)); 489 + } 490 + }, 491 + } 492 + return 1; 493 + } 494 + 495 + if (std.mem.eql(u8, cmd, "ZRANGE")) { 496 + if (args.len < 3) return errorReply(lua, pcall, "ERR wrong number of arguments for 'zrange'"); 497 + const start = std.fmt.parseInt(i64, args[1], 10) catch 498 + return errorReply(lua, pcall, "ERR not integer"); 499 + const stop = std.fmt.parseInt(i64, args[2], 10) catch 500 + return errorReply(lua, pcall, "ERR not integer"); 501 + var with_scores = false; 502 + var i: usize = 3; 503 + while (i < args.len) : (i += 1) { 504 + if (std.ascii.eqlIgnoreCase(args[i], "WITHSCORES")) with_scores = true; 505 + } 506 + const sms = store.zrange(scratch, args[0], start, stop, with_scores) catch |err| 507 + return errorReply(lua, pcall, redisErrorMsg(err)); 508 + defer Store.freeScoredMembers(scratch, sms); 509 + lua.newTable(); 510 + var idx: i32 = 1; 511 + for (sms) |sm| { 512 + _ = lua.pushString(sm.member); 513 + lua.setIndexRaw(-2, idx); 514 + idx += 1; 515 + if (with_scores) { 516 + var buf: [64]u8 = undefined; 517 + const txt = std.fmt.bufPrint(&buf, "{d}", .{sm.score}) catch "0"; 518 + _ = lua.pushString(txt); 519 + lua.setIndexRaw(-2, idx); 520 + idx += 1; 521 + } 522 + } 523 + return 1; 524 + } 525 + 526 + if (std.mem.eql(u8, cmd, "ZCOUNT")) { 527 + if (args.len != 3) return errorReply(lua, pcall, "ERR wrong number of arguments for 'zcount'"); 528 + const min = parseScoreBound(args[1]) catch 529 + return errorReply(lua, pcall, "ERR min is not a valid float"); 530 + const max = parseScoreBound(args[2]) catch 531 + return errorReply(lua, pcall, "ERR max is not a valid float"); 532 + const n = store.zcount(args[0], min, max) catch |err| 533 + return errorReply(lua, pcall, redisErrorMsg(err)); 534 + lua.pushInteger(n); 535 + return 1; 536 + } 537 + 275 538 if (std.mem.eql(u8, cmd, "EXISTS")) { 276 539 if (args.len == 0) return errorReply(lua, pcall, "ERR wrong number of arguments"); 277 540 lua.pushInteger(store.exists(args)); ··· 771 1034 defer v.deinit(std.testing.allocator); 772 1035 try std.testing.expect(v == .bulk_string); 773 1036 try std.testing.expectEqualStrings("v1", v.bulk_string); 1037 + } 1038 + 1039 + test "EVAL: SET/GET via redis.call" { 1040 + var threaded = std.Io.Threaded.init(std.testing.allocator, .{}); 1041 + defer threaded.deinit(); 1042 + var store = Store.init(std.testing.allocator, threaded.io()); 1043 + defer store.deinit(); 1044 + var engine = LuaEngine.init(std.testing.allocator, &store); 1045 + defer engine.deinit(); 1046 + 1047 + const script = 1048 + \\redis.call('SET', KEYS[1], 'hello') 1049 + \\return redis.call('GET', KEYS[1]) 1050 + ; 1051 + const v = try engine.eval(std.testing.allocator, script, &.{"k"}, &.{}); 1052 + defer v.deinit(std.testing.allocator); 1053 + try std.testing.expect(v == .bulk_string); 1054 + try std.testing.expectEqualStrings("hello", v.bulk_string); 1055 + } 1056 + 1057 + test "EVAL: list LPUSH/RPUSH/LRANGE" { 1058 + var threaded = std.Io.Threaded.init(std.testing.allocator, .{}); 1059 + defer threaded.deinit(); 1060 + var store = Store.init(std.testing.allocator, threaded.io()); 1061 + defer store.deinit(); 1062 + var engine = LuaEngine.init(std.testing.allocator, &store); 1063 + defer engine.deinit(); 1064 + 1065 + const script = 1066 + \\redis.call('RPUSH', KEYS[1], 'a', 'b', 'c') 1067 + \\local r = redis.call('LRANGE', KEYS[1], '0', '-1') 1068 + \\return #r 1069 + ; 1070 + const v = try engine.eval(std.testing.allocator, script, &.{"L"}, &.{}); 1071 + defer v.deinit(std.testing.allocator); 1072 + try std.testing.expect(v == .integer); 1073 + try std.testing.expectEqual(@as(i64, 3), v.integer); 774 1074 } 775 1075 776 1076 test "EVAL: full XGROUP + XADD + XREADGROUP cycle (docket-critical)" {

+1072

src/store.zig

View file

··· 1164 1164 } 1165 1165 return count; 1166 1166 } 1167 + 1168 + // ------------------------------------------------------------------------ 1169 + // Strings — SET, GET, MGET, KEYS, TTL, EXPIRE, INCRBY, DECRBY 1170 + // ------------------------------------------------------------------------ 1171 + 1172 + pub const SetFlags = struct { 1173 + nx: bool = false, 1174 + xx: bool = false, 1175 + /// TTL in nanoseconds from now. null = no expiration. 1176 + ttl_ns: ?i128 = null, 1177 + }; 1178 + 1179 + pub fn set(self: *Store, key: []const u8, value: []const u8, flags: SetFlags) StoreError!bool { 1180 + self.mutex.lockUncancelable(self.io); 1181 + defer self.mutex.unlock(self.io); 1182 + 1183 + _ = self.expireIfDue(key); 1184 + const key_exists = self.data.contains(key); 1185 + if (flags.nx and key_exists) return false; 1186 + if (flags.xx and !key_exists) return false; 1187 + 1188 + if (key_exists) { 1189 + const old = self.data.getPtr(key).?; 1190 + self.freeValueData(&old.data); 1191 + } 1192 + const owned_value = try self.allocator.dupe(u8, value); 1193 + errdefer self.allocator.free(owned_value); 1194 + const exp = if (flags.ttl_ns) |t| self.nowNs() + t else null; 1195 + 1196 + if (key_exists) { 1197 + self.data.getPtr(key).?.* = .{ 1198 + .data = .{ .string = owned_value }, 1199 + .expires_at_ns = exp, 1200 + }; 1201 + } else { 1202 + const owned_key = try self.allocator.dupe(u8, key); 1203 + errdefer self.allocator.free(owned_key); 1204 + try self.data.put(owned_key, .{ 1205 + .data = .{ .string = owned_value }, 1206 + .expires_at_ns = exp, 1207 + }); 1208 + } 1209 + return true; 1210 + } 1211 + 1212 + pub fn get(self: *Store, out_alloc: Allocator, key: []const u8) StoreError!?[]u8 { 1213 + self.mutex.lockUncancelable(self.io); 1214 + defer self.mutex.unlock(self.io); 1215 + 1216 + const entry = self.liveEntry(key) orelse return null; 1217 + if (entry.data != .string) return StoreError.WrongType; 1218 + return try out_alloc.dupe(u8, entry.data.string); 1219 + } 1220 + 1221 + pub fn mget(self: *Store, out_alloc: Allocator, keys_in: []const []const u8) StoreError![]?[]u8 { 1222 + self.mutex.lockUncancelable(self.io); 1223 + defer self.mutex.unlock(self.io); 1224 + 1225 + const now = self.nowNs(); 1226 + const out = try out_alloc.alloc(?[]u8, keys_in.len); 1227 + errdefer out_alloc.free(out); 1228 + 1229 + for (keys_in, 0..) |k, i| { 1230 + const e = self.data.getPtr(k); 1231 + if (e == null or e.?.isExpired(now) or e.?.data != .string) { 1232 + out[i] = null; 1233 + continue; 1234 + } 1235 + out[i] = try out_alloc.dupe(u8, e.?.data.string); 1236 + } 1237 + return out; 1238 + } 1239 + 1240 + pub fn freeMGetResult(alloc: Allocator, result: []?[]u8) void { 1241 + for (result) |opt| if (opt) |s| alloc.free(s); 1242 + alloc.free(result); 1243 + } 1244 + 1245 + pub fn keysGlob(self: *Store, out_alloc: Allocator, pattern: []const u8) StoreError![][]u8 { 1246 + self.mutex.lockUncancelable(self.io); 1247 + defer self.mutex.unlock(self.io); 1248 + 1249 + const now = self.nowNs(); 1250 + var out: std.ArrayList([]u8) = .empty; 1251 + errdefer { 1252 + for (out.items) |k| out_alloc.free(k); 1253 + out.deinit(out_alloc); 1254 + } 1255 + var it = self.data.iterator(); 1256 + while (it.next()) |kv| { 1257 + if (kv.value_ptr.isExpired(now)) continue; 1258 + if (globMatch(pattern, kv.key_ptr.*)) { 1259 + try out.append(out_alloc, try out_alloc.dupe(u8, kv.key_ptr.*)); 1260 + } 1261 + } 1262 + return out.toOwnedSlice(out_alloc); 1263 + } 1264 + 1265 + pub fn freeKeys(alloc: Allocator, list: [][]u8) void { 1266 + for (list) |k| alloc.free(k); 1267 + alloc.free(list); 1268 + } 1269 + 1270 + /// TTL: remaining seconds. -2 missing/expired, -1 no TTL, >=0 remaining. 1271 + pub fn ttl(self: *Store, key: []const u8) i64 { 1272 + self.mutex.lockUncancelable(self.io); 1273 + defer self.mutex.unlock(self.io); 1274 + 1275 + if (self.expireIfDue(key)) return -2; 1276 + const e = self.data.getPtr(key) orelse return -2; 1277 + const exp_ns = e.expires_at_ns orelse return -1; 1278 + const now = self.nowNs(); 1279 + if (exp_ns <= now) return -2; 1280 + return @intCast(@divFloor(exp_ns - now, std.time.ns_per_s)); 1281 + } 1282 + 1283 + pub fn expire(self: *Store, key: []const u8, seconds: u64) bool { 1284 + self.mutex.lockUncancelable(self.io); 1285 + defer self.mutex.unlock(self.io); 1286 + 1287 + if (self.expireIfDue(key)) return false; 1288 + const e = self.data.getPtr(key) orelse return false; 1289 + e.expires_at_ns = self.nowNs() + @as(i128, seconds) * std.time.ns_per_s; 1290 + return true; 1291 + } 1292 + 1293 + pub fn incrby(self: *Store, key: []const u8, delta: i64) StoreError!i64 { 1294 + self.mutex.lockUncancelable(self.io); 1295 + defer self.mutex.unlock(self.io); 1296 + 1297 + _ = self.expireIfDue(key); 1298 + 1299 + var cur: i64 = 0; 1300 + if (self.data.getPtr(key)) |e| { 1301 + if (e.data != .string) return StoreError.WrongType; 1302 + cur = std.fmt.parseInt(i64, e.data.string, 10) catch return StoreError.Syntax; 1303 + } 1304 + const new_val = cur + delta; 1305 + var buf: [32]u8 = undefined; 1306 + const txt = std.fmt.bufPrint(&buf, "{d}", .{new_val}) catch return StoreError.Syntax; 1307 + const owned = try self.allocator.dupe(u8, txt); 1308 + errdefer self.allocator.free(owned); 1309 + 1310 + if (self.data.getPtr(key)) |e| { 1311 + self.freeValueData(&e.data); 1312 + e.data = .{ .string = owned }; 1313 + } else { 1314 + const owned_key = try self.allocator.dupe(u8, key); 1315 + errdefer self.allocator.free(owned_key); 1316 + try self.data.put(owned_key, .{ 1317 + .data = .{ .string = owned }, 1318 + .expires_at_ns = null, 1319 + }); 1320 + } 1321 + return new_val; 1322 + } 1323 + 1324 + pub fn decrby(self: *Store, key: []const u8, delta: i64) StoreError!i64 { 1325 + return self.incrby(key, -delta); 1326 + } 1327 + 1328 + // ------------------------------------------------------------------------ 1329 + // Sets — SADD, SMEMBERS, SISMEMBER, SREM, SCARD 1330 + // ------------------------------------------------------------------------ 1331 + 1332 + fn initSet(alloc: Allocator) ValueData { 1333 + return .{ .set = BytesHashSet.init(alloc) }; 1334 + } 1335 + 1336 + pub fn sadd(self: *Store, key: []const u8, members: []const []const u8) StoreError!i64 { 1337 + if (members.len == 0) return StoreError.Syntax; 1338 + self.mutex.lockUncancelable(self.io); 1339 + defer self.mutex.unlock(self.io); 1340 + 1341 + const entry = try self.getOrCreate(key, .set, initSet); 1342 + var added: i64 = 0; 1343 + for (members) |m| { 1344 + const gop = try entry.data.set.getOrPut(m); 1345 + if (!gop.found_existing) { 1346 + gop.key_ptr.* = try self.allocator.dupe(u8, m); 1347 + added += 1; 1348 + } 1349 + } 1350 + return added; 1351 + } 1352 + 1353 + pub fn smembers(self: *Store, out_alloc: Allocator, key: []const u8) StoreError![][]u8 { 1354 + self.mutex.lockUncancelable(self.io); 1355 + defer self.mutex.unlock(self.io); 1356 + 1357 + const entry = self.liveEntry(key) orelse return &[_][]u8{}; 1358 + if (entry.data != .set) return StoreError.WrongType; 1359 + 1360 + var out: std.ArrayList([]u8) = .empty; 1361 + errdefer { 1362 + for (out.items) |k| out_alloc.free(k); 1363 + out.deinit(out_alloc); 1364 + } 1365 + var it = entry.data.set.keyIterator(); 1366 + while (it.next()) |m| try out.append(out_alloc, try out_alloc.dupe(u8, m.*)); 1367 + return out.toOwnedSlice(out_alloc); 1368 + } 1369 + 1370 + pub fn sismember(self: *Store, key: []const u8, member: []const u8) StoreError!i64 { 1371 + self.mutex.lockUncancelable(self.io); 1372 + defer self.mutex.unlock(self.io); 1373 + 1374 + const entry = self.liveEntry(key) orelse return 0; 1375 + if (entry.data != .set) return StoreError.WrongType; 1376 + return if (entry.data.set.contains(member)) 1 else 0; 1377 + } 1378 + 1379 + pub fn srem(self: *Store, key: []const u8, members: []const []const u8) StoreError!i64 { 1380 + self.mutex.lockUncancelable(self.io); 1381 + defer self.mutex.unlock(self.io); 1382 + 1383 + const entry = self.liveEntry(key) orelse return 0; 1384 + if (entry.data != .set) return StoreError.WrongType; 1385 + 1386 + var count: i64 = 0; 1387 + for (members) |m| { 1388 + const kv = entry.data.set.fetchRemove(m) orelse continue; 1389 + self.allocator.free(kv.key); 1390 + count += 1; 1391 + } 1392 + if (entry.data.set.count() == 0) _ = self.removeKey(key); 1393 + return count; 1394 + } 1395 + 1396 + pub fn scard(self: *Store, key: []const u8) StoreError!i64 { 1397 + self.mutex.lockUncancelable(self.io); 1398 + defer self.mutex.unlock(self.io); 1399 + const entry = self.liveEntry(key) orelse return 0; 1400 + if (entry.data != .set) return StoreError.WrongType; 1401 + return @intCast(entry.data.set.count()); 1402 + } 1403 + 1404 + // ------------------------------------------------------------------------ 1405 + // Lists — LPUSH, RPUSH, LLEN, LINDEX, LRANGE, LPOP, RPOP, LREM, LSET, LTRIM 1406 + // ------------------------------------------------------------------------ 1407 + 1408 + fn initList(_: Allocator) ValueData { 1409 + return .{ .list = .{} }; 1410 + } 1411 + 1412 + pub fn lpush(self: *Store, key: []const u8, values: []const []const u8) StoreError!i64 { 1413 + if (values.len == 0) return StoreError.Syntax; 1414 + self.mutex.lockUncancelable(self.io); 1415 + defer self.mutex.unlock(self.io); 1416 + 1417 + const entry = try self.getOrCreate(key, .list, initList); 1418 + for (values) |v| { 1419 + const node = try self.allocator.create(ListNode); 1420 + errdefer self.allocator.destroy(node); 1421 + node.* = .{ .data = try self.allocator.dupe(u8, v) }; 1422 + entry.data.list.prepend(&node.node); 1423 + } 1424 + return @intCast(listLen(&entry.data.list)); 1425 + } 1426 + 1427 + pub fn rpush(self: *Store, key: []const u8, values: []const []const u8) StoreError!i64 { 1428 + if (values.len == 0) return StoreError.Syntax; 1429 + self.mutex.lockUncancelable(self.io); 1430 + defer self.mutex.unlock(self.io); 1431 + 1432 + const entry = try self.getOrCreate(key, .list, initList); 1433 + for (values) |v| { 1434 + const node = try self.allocator.create(ListNode); 1435 + errdefer self.allocator.destroy(node); 1436 + node.* = .{ .data = try self.allocator.dupe(u8, v) }; 1437 + entry.data.list.append(&node.node); 1438 + } 1439 + return @intCast(listLen(&entry.data.list)); 1440 + } 1441 + 1442 + pub fn llen(self: *Store, key: []const u8) StoreError!i64 { 1443 + self.mutex.lockUncancelable(self.io); 1444 + defer self.mutex.unlock(self.io); 1445 + const entry = self.liveEntry(key) orelse return 0; 1446 + if (entry.data != .list) return StoreError.WrongType; 1447 + return @intCast(listLen(&entry.data.list)); 1448 + } 1449 + 1450 + pub fn lindex(self: *Store, out_alloc: Allocator, key: []const u8, index: i64) StoreError!?[]u8 { 1451 + self.mutex.lockUncancelable(self.io); 1452 + defer self.mutex.unlock(self.io); 1453 + 1454 + const entry = self.liveEntry(key) orelse return null; 1455 + if (entry.data != .list) return StoreError.WrongType; 1456 + const n: i64 = @intCast(listLen(&entry.data.list)); 1457 + const actual = if (index < 0) index + n else index; 1458 + if (actual < 0 or actual >= n) return null; 1459 + const node = listAt(&entry.data.list, @intCast(actual)) orelse return null; 1460 + const ln: *ListNode = @fieldParentPtr("node", node); 1461 + return try out_alloc.dupe(u8, ln.data); 1462 + } 1463 + 1464 + pub fn lrange(self: *Store, out_alloc: Allocator, key: []const u8, start: i64, stop: i64) StoreError![][]u8 { 1465 + self.mutex.lockUncancelable(self.io); 1466 + defer self.mutex.unlock(self.io); 1467 + 1468 + const entry = self.liveEntry(key) orelse return &[_][]u8{}; 1469 + if (entry.data != .list) return StoreError.WrongType; 1470 + const n: usize = listLen(&entry.data.list); 1471 + const r = normalizeRange(start, stop, n) orelse return &[_][]u8{}; 1472 + 1473 + var out: std.ArrayList([]u8) = .empty; 1474 + errdefer { 1475 + for (out.items) |k| out_alloc.free(k); 1476 + out.deinit(out_alloc); 1477 + } 1478 + var node = entry.data.list.first; 1479 + var i: usize = 0; 1480 + while (node) |nd| : ({ 1481 + node = nd.next; 1482 + i += 1; 1483 + }) { 1484 + if (i < r.start) continue; 1485 + if (i > r.stop) break; 1486 + const ln: *ListNode = @fieldParentPtr("node", nd); 1487 + try out.append(out_alloc, try out_alloc.dupe(u8, ln.data)); 1488 + } 1489 + return out.toOwnedSlice(out_alloc); 1490 + } 1491 + 1492 + pub const LPopResult = union(enum) { 1493 + nil, 1494 + single: []u8, 1495 + array: [][]u8, 1496 + }; 1497 + 1498 + pub fn lpop(self: *Store, out_alloc: Allocator, key: []const u8, count: ?usize) StoreError!LPopResult { 1499 + return self.popImpl(out_alloc, key, count, true); 1500 + } 1501 + 1502 + pub fn rpop(self: *Store, out_alloc: Allocator, key: []const u8, count: ?usize) StoreError!LPopResult { 1503 + return self.popImpl(out_alloc, key, count, false); 1504 + } 1505 + 1506 + fn popImpl(self: *Store, out_alloc: Allocator, key: []const u8, count: ?usize, head: bool) StoreError!LPopResult { 1507 + self.mutex.lockUncancelable(self.io); 1508 + defer self.mutex.unlock(self.io); 1509 + 1510 + const entry = self.liveEntry(key) orelse return .nil; 1511 + if (entry.data != .list) return StoreError.WrongType; 1512 + if (count) |c| if (c == 0) return .{ .array = try out_alloc.alloc([]u8, 0) }; 1513 + 1514 + const list = &entry.data.list; 1515 + if (count == null) { 1516 + const node = if (head) list.popFirst() else list.pop(); 1517 + if (node) |nd| { 1518 + const ln: *ListNode = @fieldParentPtr("node", nd); 1519 + const dup = try out_alloc.dupe(u8, ln.data); 1520 + self.allocator.free(ln.data); 1521 + self.allocator.destroy(ln); 1522 + if (listLen(list) == 0) _ = self.removeKey(key); 1523 + return .{ .single = dup }; 1524 + } 1525 + return .nil; 1526 + } 1527 + 1528 + const want = count.?; 1529 + var popped: std.ArrayList([]u8) = .empty; 1530 + errdefer { 1531 + for (popped.items) |s| out_alloc.free(s); 1532 + popped.deinit(out_alloc); 1533 + } 1534 + var i: usize = 0; 1535 + while (i < want) : (i += 1) { 1536 + const node = if (head) list.popFirst() else list.pop(); 1537 + if (node == null) break; 1538 + const ln: *ListNode = @fieldParentPtr("node", node.?); 1539 + try popped.append(out_alloc, try out_alloc.dupe(u8, ln.data)); 1540 + self.allocator.free(ln.data); 1541 + self.allocator.destroy(ln); 1542 + } 1543 + if (listLen(list) == 0) _ = self.removeKey(key); 1544 + return .{ .array = try popped.toOwnedSlice(out_alloc) }; 1545 + } 1546 + 1547 + pub fn freeLPopResult(alloc: Allocator, r: LPopResult) void { 1548 + switch (r) { 1549 + .nil => {}, 1550 + .single => |s| alloc.free(s), 1551 + .array => |arr| { 1552 + for (arr) |s| alloc.free(s); 1553 + alloc.free(arr); 1554 + }, 1555 + } 1556 + } 1557 + 1558 + pub fn lrem(self: *Store, key: []const u8, count: i64, value: []const u8) StoreError!i64 { 1559 + self.mutex.lockUncancelable(self.io); 1560 + defer self.mutex.unlock(self.io); 1561 + 1562 + const entry = self.liveEntry(key) orelse return 0; 1563 + if (entry.data != .list) return StoreError.WrongType; 1564 + const list = &entry.data.list; 1565 + 1566 + var removed: i64 = 0; 1567 + const target_abs: usize = if (count == 0) std.math.maxInt(usize) else @abs(count); 1568 + 1569 + if (count >= 0) { 1570 + var node = list.first; 1571 + while (node) |nd| { 1572 + const next = nd.next; 1573 + const ln: *ListNode = @fieldParentPtr("node", nd); 1574 + if (std.mem.eql(u8, ln.data, value)) { 1575 + list.remove(nd); 1576 + self.allocator.free(ln.data); 1577 + self.allocator.destroy(ln); 1578 + removed += 1; 1579 + if (@as(usize, @intCast(removed)) >= target_abs) break; 1580 + } 1581 + node = next; 1582 + } 1583 + } else { 1584 + var node = list.last; 1585 + while (node) |nd| { 1586 + const prev = nd.prev; 1587 + const ln: *ListNode = @fieldParentPtr("node", nd); 1588 + if (std.mem.eql(u8, ln.data, value)) { 1589 + list.remove(nd); 1590 + self.allocator.free(ln.data); 1591 + self.allocator.destroy(ln); 1592 + removed += 1; 1593 + if (@as(usize, @intCast(removed)) >= target_abs) break; 1594 + } 1595 + node = prev; 1596 + } 1597 + } 1598 + if (listLen(list) == 0) _ = self.removeKey(key); 1599 + return removed; 1600 + } 1601 + 1602 + pub fn lset(self: *Store, key: []const u8, index: i64, value: []const u8) StoreError!void { 1603 + self.mutex.lockUncancelable(self.io); 1604 + defer self.mutex.unlock(self.io); 1605 + 1606 + const entry = self.liveEntry(key) orelse return StoreError.NoSuchKey; 1607 + if (entry.data != .list) return StoreError.WrongType; 1608 + const n: i64 = @intCast(listLen(&entry.data.list)); 1609 + const actual = if (index < 0) index + n else index; 1610 + if (actual < 0 or actual >= n) return StoreError.Syntax; 1611 + const node = listAt(&entry.data.list, @intCast(actual)) orelse return StoreError.Syntax; 1612 + const ln: *ListNode = @fieldParentPtr("node", node); 1613 + const new_val = try self.allocator.dupe(u8, value); 1614 + self.allocator.free(ln.data); 1615 + ln.data = new_val; 1616 + } 1617 + 1618 + pub fn ltrim(self: *Store, key: []const u8, start: i64, stop: i64) StoreError!void { 1619 + self.mutex.lockUncancelable(self.io); 1620 + defer self.mutex.unlock(self.io); 1621 + 1622 + const entry = self.liveEntry(key) orelse return; 1623 + if (entry.data != .list) return StoreError.WrongType; 1624 + const list = &entry.data.list; 1625 + const n: usize = listLen(list); 1626 + const r = normalizeRange(start, stop, n) orelse { 1627 + _ = self.removeKey(key); 1628 + return; 1629 + }; 1630 + 1631 + var dropped: usize = 0; 1632 + while (dropped < r.start) : (dropped += 1) { 1633 + const nd = list.popFirst() orelse break; 1634 + const ln: *ListNode = @fieldParentPtr("node", nd); 1635 + self.allocator.free(ln.data); 1636 + self.allocator.destroy(ln); 1637 + } 1638 + const keep = r.stop - r.start + 1; 1639 + while (listLen(list) > keep) { 1640 + const nd = list.pop() orelse break; 1641 + const ln: *ListNode = @fieldParentPtr("node", nd); 1642 + self.allocator.free(ln.data); 1643 + self.allocator.destroy(ln); 1644 + } 1645 + if (listLen(list) == 0) _ = self.removeKey(key); 1646 + } 1647 + 1648 + // ------------------------------------------------------------------------ 1649 + // More sorted-set — ZRANGE, ZCOUNT 1650 + // ------------------------------------------------------------------------ 1651 + 1652 + pub fn zrange( 1653 + self: *Store, 1654 + out_alloc: Allocator, 1655 + key: []const u8, 1656 + start: i64, 1657 + stop: i64, 1658 + withscores: bool, 1659 + ) StoreError![]ScoredMember { 1660 + self.mutex.lockUncancelable(self.io); 1661 + defer self.mutex.unlock(self.io); 1662 + 1663 + const entry = self.liveEntry(key) orelse return &[_]ScoredMember{}; 1664 + if (entry.data != .sorted_set) return StoreError.WrongType; 1665 + const zset = &entry.data.sorted_set; 1666 + const n = zset.by_score.items.len; 1667 + const r = normalizeRange(start, stop, n) orelse return &[_]ScoredMember{}; 1668 + 1669 + var out: std.ArrayList(ScoredMember) = .empty; 1670 + errdefer { 1671 + for (out.items) |sm| out_alloc.free(sm.member); 1672 + out.deinit(out_alloc); 1673 + } 1674 + var i: usize = r.start; 1675 + while (i <= r.stop) : (i += 1) { 1676 + const sm = zset.by_score.items[i]; 1677 + try out.append(out_alloc, .{ 1678 + .score = if (withscores) sm.score else 0, 1679 + .member = try out_alloc.dupe(u8, sm.member), 1680 + }); 1681 + } 1682 + return out.toOwnedSlice(out_alloc); 1683 + } 1684 + 1685 + pub fn zcount(self: *Store, key: []const u8, min: f64, max: f64) StoreError!i64 { 1686 + self.mutex.lockUncancelable(self.io); 1687 + defer self.mutex.unlock(self.io); 1688 + const entry = self.liveEntry(key) orelse return 0; 1689 + if (entry.data != .sorted_set) return StoreError.WrongType; 1690 + var count: i64 = 0; 1691 + for (entry.data.sorted_set.by_score.items) |sm| { 1692 + if (sm.score < min) continue; 1693 + if (sm.score > max) break; 1694 + count += 1; 1695 + } 1696 + return count; 1697 + } 1698 + 1699 + // ------------------------------------------------------------------------ 1700 + // More streams — XREAD, XRANGE, XTRIM, XGROUP DESTROY, stream_last_id, 1701 + // sweepExpired 1702 + // ------------------------------------------------------------------------ 1703 + 1704 + pub const ReadResult = struct { 1705 + key: []u8, 1706 + entries: []DeliveredEntry, 1707 + }; 1708 + 1709 + pub fn xread( 1710 + self: *Store, 1711 + out_alloc: Allocator, 1712 + stream_keys: []const []const u8, 1713 + start_ids: []const StreamId, 1714 + count: ?usize, 1715 + ) StoreError![]ReadResult { 1716 + if (stream_keys.len != start_ids.len) return StoreError.Syntax; 1717 + self.mutex.lockUncancelable(self.io); 1718 + defer self.mutex.unlock(self.io); 1719 + 1720 + var out: std.ArrayList(ReadResult) = .empty; 1721 + errdefer { 1722 + for (out.items) |r| freeReadResult(out_alloc, r); 1723 + out.deinit(out_alloc); 1724 + } 1725 + 1726 + for (stream_keys, start_ids) |k, start| { 1727 + const entry = self.liveEntry(k) orelse continue; 1728 + if (entry.data != .stream) return StoreError.WrongType; 1729 + const s = &entry.data.stream; 1730 + 1731 + const max_n = count orelse std.math.maxInt(usize); 1732 + var ents: std.ArrayList(DeliveredEntry) = .empty; 1733 + errdefer { 1734 + for (ents.items) |de| freeDeliveredEntryLocal(out_alloc, de); 1735 + ents.deinit(out_alloc); 1736 + } 1737 + 1738 + for (s.entries.items) |se| { 1739 + if (ents.items.len >= max_n) break; 1740 + if (StreamId.cmp(se.id, start) != .gt) continue; 1741 + try ents.append(out_alloc, .{ 1742 + .id = se.id, 1743 + .fields = try dupeStreamFields(out_alloc, se.fields), 1744 + }); 1745 + } 1746 + if (ents.items.len == 0) continue; 1747 + try out.append(out_alloc, .{ 1748 + .key = try out_alloc.dupe(u8, k), 1749 + .entries = try ents.toOwnedSlice(out_alloc), 1750 + }); 1751 + } 1752 + return out.toOwnedSlice(out_alloc); 1753 + } 1754 + 1755 + pub fn xrange( 1756 + self: *Store, 1757 + out_alloc: Allocator, 1758 + key: []const u8, 1759 + min: StreamId, 1760 + max: StreamId, 1761 + count: ?usize, 1762 + ) StoreError![]DeliveredEntry { 1763 + self.mutex.lockUncancelable(self.io); 1764 + defer self.mutex.unlock(self.io); 1765 + 1766 + const entry = self.liveEntry(key) orelse return &[_]DeliveredEntry{}; 1767 + if (entry.data != .stream) return StoreError.WrongType; 1768 + const s = &entry.data.stream; 1769 + 1770 + const max_n = count orelse std.math.maxInt(usize); 1771 + var out: std.ArrayList(DeliveredEntry) = .empty; 1772 + errdefer { 1773 + for (out.items) |de| freeDeliveredEntryLocal(out_alloc, de); 1774 + out.deinit(out_alloc); 1775 + } 1776 + for (s.entries.items) |se| { 1777 + if (out.items.len >= max_n) break; 1778 + if (StreamId.cmp(se.id, min) == .lt) continue; 1779 + if (StreamId.cmp(se.id, max) == .gt) break; 1780 + try out.append(out_alloc, .{ 1781 + .id = se.id, 1782 + .fields = try dupeStreamFields(out_alloc, se.fields), 1783 + }); 1784 + } 1785 + return out.toOwnedSlice(out_alloc); 1786 + } 1787 + 1788 + pub fn xtrimMaxlen(self: *Store, key: []const u8, maxlen: usize) StoreError!i64 { 1789 + self.mutex.lockUncancelable(self.io); 1790 + defer self.mutex.unlock(self.io); 1791 + 1792 + const entry = self.liveEntry(key) orelse return 0; 1793 + if (entry.data != .stream) return StoreError.WrongType; 1794 + const s = &entry.data.stream; 1795 + 1796 + var removed: i64 = 0; 1797 + while (s.entries.items.len > maxlen) { 1798 + var e = s.entries.orderedRemove(0); 1799 + var it = e.fields.iterator(); 1800 + while (it.next()) |kv| { 1801 + self.allocator.free(kv.key_ptr.*); 1802 + self.allocator.free(kv.value_ptr.*); 1803 + } 1804 + e.fields.deinit(); 1805 + removed += 1; 1806 + } 1807 + return removed; 1808 + } 1809 + 1810 + pub fn streamLastId(self: *Store, key: []const u8) ?StreamId { 1811 + self.mutex.lockUncancelable(self.io); 1812 + defer self.mutex.unlock(self.io); 1813 + const entry = self.liveEntry(key) orelse return null; 1814 + if (entry.data != .stream) return null; 1815 + return entry.data.stream.last_id; 1816 + } 1817 + 1818 + pub fn xgroupDestroy(self: *Store, key: []const u8, group: []const u8) StoreError!bool { 1819 + self.mutex.lockUncancelable(self.io); 1820 + defer self.mutex.unlock(self.io); 1821 + 1822 + const entry = self.liveEntry(key) orelse return false; 1823 + if (entry.data != .stream) return StoreError.WrongType; 1824 + const s = &entry.data.stream; 1825 + const kv = s.groups.fetchRemove(group) orelse return false; 1826 + self.allocator.free(kv.key); 1827 + var cg = kv.value; 1828 + cg.deinit(self.allocator); 1829 + return true; 1830 + } 1831 + 1832 + /// Sweep up to `budget` expired keys. Returns count removed. 1833 + pub fn sweepExpired(self: *Store, budget: usize) usize { 1834 + self.mutex.lockUncancelable(self.io); 1835 + defer self.mutex.unlock(self.io); 1836 + 1837 + const now = self.nowNs(); 1838 + var to_remove: std.ArrayList([]const u8) = .empty; 1839 + defer to_remove.deinit(self.allocator); 1840 + 1841 + var checked: usize = 0; 1842 + var it = self.data.iterator(); 1843 + while (it.next()) |kv| { 1844 + if (checked >= budget) break; 1845 + if (kv.value_ptr.expires_at_ns != null) { 1846 + checked += 1; 1847 + if (kv.value_ptr.isExpired(now)) { 1848 + to_remove.append(self.allocator, kv.key_ptr.*) catch break; 1849 + } 1850 + } 1851 + } 1852 + const count = to_remove.items.len; 1853 + for (to_remove.items) |k| _ = self.removeKey(k); 1854 + return count; 1855 + } 1167 1856 }; 1168 1857 1169 1858 // ---------------------------------------------------------------------------- ··· 1202 1891 alloc.free(entries); 1203 1892 } 1204 1893 1894 + /// Used by Store.xread/xrange to free results that escape via out_alloc. 1895 + fn freeDeliveredEntryLocal(alloc: Allocator, de: Store.DeliveredEntry) void { 1896 + Store.freeHashPairs(alloc, de.fields); 1897 + } 1898 + 1899 + pub fn freeReadResult(alloc: Allocator, r: Store.ReadResult) void { 1900 + alloc.free(r.key); 1901 + for (r.entries) |de| freeDeliveredEntryLocal(alloc, de); 1902 + alloc.free(r.entries); 1903 + } 1904 + 1905 + pub fn freeReadResults(alloc: Allocator, results: []Store.ReadResult) void { 1906 + for (results) |r| freeReadResult(alloc, r); 1907 + alloc.free(results); 1908 + } 1909 + 1910 + /// Linked-list length. Could cache this on the value but lists are 1911 + /// expected small for docket / Redis embedded use. 1912 + fn listLen(list: *std.DoublyLinkedList) usize { 1913 + var n: usize = 0; 1914 + var nd = list.first; 1915 + while (nd) |x| : (nd = x.next) n += 1; 1916 + return n; 1917 + } 1918 + 1919 + /// Walk to the nth node (0-indexed). Returns null on out-of-range. 1920 + fn listAt(list: *std.DoublyLinkedList, index: usize) ?*std.DoublyLinkedList.Node { 1921 + var nd = list.first; 1922 + var i: usize = 0; 1923 + while (nd) |x| : ({ 1924 + nd = x.next; 1925 + i += 1; 1926 + }) if (i == index) return x; 1927 + return null; 1928 + } 1929 + 1930 + /// Inclusive index range after normalising negative + clamping. Returns 1931 + /// null when the result is empty. 1932 + pub const NormalizedRange = struct { start: usize, stop: usize }; 1933 + 1934 + fn normalizeRange(start: i64, stop: i64, len: usize) ?NormalizedRange { 1935 + if (len == 0) return null; 1936 + const n: i64 = @intCast(len); 1937 + var s = if (start < 0) start + n else start; 1938 + var e = if (stop < 0) stop + n else stop; 1939 + if (s < 0) s = 0; 1940 + if (e >= n) e = n - 1; 1941 + if (s > e or s >= n) return null; 1942 + return .{ .start = @intCast(s), .stop = @intCast(e) }; 1943 + } 1944 + 1945 + /// Redis glob match: `*`, `?`, character classes `[...]`, escape with `\`. 1946 + /// Subset suitable for KEYS / PUBSUB NUMPAT — no `**` recursion. 1947 + fn globMatch(pattern: []const u8, str: []const u8) bool { 1948 + return globMatchInner(pattern, 0, str, 0); 1949 + } 1950 + 1951 + fn globMatchInner(p: []const u8, pi_in: usize, s: []const u8, si_in: usize) bool { 1952 + var pi = pi_in; 1953 + var si = si_in; 1954 + while (pi < p.len) { 1955 + const c = p[pi]; 1956 + switch (c) { 1957 + '*' => { 1958 + // Collapse runs of *. 1959 + while (pi < p.len and p[pi] == '*') pi += 1; 1960 + if (pi == p.len) return true; 1961 + while (si <= s.len) : (si += 1) { 1962 + if (globMatchInner(p, pi, s, si)) return true; 1963 + } 1964 + return false; 1965 + }, 1966 + '?' => { 1967 + if (si >= s.len) return false; 1968 + pi += 1; 1969 + si += 1; 1970 + }, 1971 + '[' => { 1972 + if (si >= s.len) return false; 1973 + pi += 1; 1974 + var negate = false; 1975 + if (pi < p.len and p[pi] == '^') { 1976 + negate = true; 1977 + pi += 1; 1978 + } 1979 + var matched = false; 1980 + while (pi < p.len and p[pi] != ']') { 1981 + if (pi + 2 < p.len and p[pi + 1] == '-') { 1982 + const lo = p[pi]; 1983 + const hi = p[pi + 2]; 1984 + if (s[si] >= lo and s[si] <= hi) matched = true; 1985 + pi += 3; 1986 + } else { 1987 + if (s[si] == p[pi]) matched = true; 1988 + pi += 1; 1989 + } 1990 + } 1991 + if (pi < p.len) pi += 1; // skip ']' 1992 + if (matched == negate) return false; 1993 + si += 1; 1994 + }, 1995 + '\\' => { 1996 + pi += 1; 1997 + if (pi >= p.len) return false; 1998 + if (si >= s.len or s[si] != p[pi]) return false; 1999 + pi += 1; 2000 + si += 1; 2001 + }, 2002 + else => { 2003 + if (si >= s.len or s[si] != c) return false; 2004 + pi += 1; 2005 + si += 1; 2006 + }, 2007 + } 2008 + } 2009 + return si == s.len; 2010 + } 2011 + 1205 2012 // ---------------------------------------------------------------------------- 1206 2013 // Tests 1207 2014 // ---------------------------------------------------------------------------- ··· 1482 2289 try std.testing.expectEqual(@as(i64, 3), s.delete(&.{ "h", "z", "st" })); 1483 2290 try std.testing.expectEqual(@as(i64, 0), s.exists(&.{ "h", "z", "st" })); 1484 2291 } 2292 + 2293 + // --- Strings (SET/GET/MGET/KEYS/TTL/EXPIRE/INCRBY/DECRBY) ------------------- 2294 + 2295 + test "SET / GET roundtrip + NX/XX semantics" { 2296 + var threaded = Io.Threaded.init(std.testing.allocator, .{}); 2297 + defer threaded.deinit(); 2298 + var s = Store.init(std.testing.allocator, threaded.io()); 2299 + defer s.deinit(); 2300 + 2301 + try std.testing.expect(try s.set("k", "v1", .{})); 2302 + const v = try s.get(std.testing.allocator, "k"); 2303 + try std.testing.expectEqualStrings("v1", v.?); 2304 + std.testing.allocator.free(v.?); 2305 + 2306 + try std.testing.expect(!try s.set("k", "v2", .{ .nx = true })); 2307 + try std.testing.expect(try s.set("k", "v3", .{ .xx = true })); 2308 + const v3 = try s.get(std.testing.allocator, "k"); 2309 + try std.testing.expectEqualStrings("v3", v3.?); 2310 + std.testing.allocator.free(v3.?); 2311 + 2312 + try std.testing.expect(!try s.set("missing", "x", .{ .xx = true })); 2313 + } 2314 + 2315 + test "MGET returns null for missing or non-string keys" { 2316 + var threaded = Io.Threaded.init(std.testing.allocator, .{}); 2317 + defer threaded.deinit(); 2318 + var s = Store.init(std.testing.allocator, threaded.io()); 2319 + defer s.deinit(); 2320 + 2321 + _ = try s.set("a", "1", .{}); 2322 + _ = try s.hset("h", &.{ "f", "v" }); 2323 + 2324 + const res = try s.mget(std.testing.allocator, &.{ "a", "missing", "h" }); 2325 + defer Store.freeMGetResult(std.testing.allocator, res); 2326 + 2327 + try std.testing.expectEqualStrings("1", res[0].?); 2328 + try std.testing.expect(res[1] == null); 2329 + try std.testing.expect(res[2] == null); 2330 + } 2331 + 2332 + test "INCRBY initializes and accumulates; WRONGTYPE on non-string" { 2333 + var threaded = Io.Threaded.init(std.testing.allocator, .{}); 2334 + defer threaded.deinit(); 2335 + var s = Store.init(std.testing.allocator, threaded.io()); 2336 + defer s.deinit(); 2337 + 2338 + try std.testing.expectEqual(@as(i64, 5), try s.incrby("ctr", 5)); 2339 + try std.testing.expectEqual(@as(i64, 8), try s.incrby("ctr", 3)); 2340 + try std.testing.expectEqual(@as(i64, 6), try s.decrby("ctr", 2)); 2341 + 2342 + _ = try s.hset("h", &.{ "f", "v" }); 2343 + try std.testing.expectError(StoreError.WrongType, s.incrby("h", 1)); 2344 + } 2345 + 2346 + test "EXPIRE + TTL reflect remaining lifetime" { 2347 + var threaded = Io.Threaded.init(std.testing.allocator, .{}); 2348 + defer threaded.deinit(); 2349 + var s = Store.init(std.testing.allocator, threaded.io()); 2350 + defer s.deinit(); 2351 + 2352 + _ = try s.set("k", "v", .{}); 2353 + try std.testing.expectEqual(@as(i64, -1), s.ttl("k")); 2354 + try std.testing.expect(s.expire("k", 60)); 2355 + const t = s.ttl("k"); 2356 + try std.testing.expect(t >= 0 and t <= 60); 2357 + try std.testing.expectEqual(@as(i64, -2), s.ttl("missing")); 2358 + } 2359 + 2360 + test "KEYS glob matches prefix and ?" { 2361 + var threaded = Io.Threaded.init(std.testing.allocator, .{}); 2362 + defer threaded.deinit(); 2363 + var s = Store.init(std.testing.allocator, threaded.io()); 2364 + defer s.deinit(); 2365 + 2366 + _ = try s.set("user:1", "a", .{}); 2367 + _ = try s.set("user:2", "b", .{}); 2368 + _ = try s.set("session:x", "c", .{}); 2369 + 2370 + const matched = try s.keysGlob(std.testing.allocator, "user:?"); 2371 + defer Store.freeKeys(std.testing.allocator, matched); 2372 + try std.testing.expectEqual(@as(usize, 2), matched.len); 2373 + } 2374 + 2375 + // --- Sets -------------------------------------------------------------------- 2376 + 2377 + test "SADD / SMEMBERS / SISMEMBER / SREM / SCARD" { 2378 + var threaded = Io.Threaded.init(std.testing.allocator, .{}); 2379 + defer threaded.deinit(); 2380 + var s = Store.init(std.testing.allocator, threaded.io()); 2381 + defer s.deinit(); 2382 + 2383 + try std.testing.expectEqual(@as(i64, 3), try s.sadd("g", &.{ "a", "b", "c" })); 2384 + try std.testing.expectEqual(@as(i64, 0), try s.sadd("g", &.{"a"})); 2385 + try std.testing.expectEqual(@as(i64, 3), try s.scard("g")); 2386 + try std.testing.expectEqual(@as(i64, 1), try s.sismember("g", "b")); 2387 + try std.testing.expectEqual(@as(i64, 0), try s.sismember("g", "z")); 2388 + 2389 + const members = try s.smembers(std.testing.allocator, "g"); 2390 + defer Store.freeKeys(std.testing.allocator, members); 2391 + try std.testing.expectEqual(@as(usize, 3), members.len); 2392 + 2393 + try std.testing.expectEqual(@as(i64, 2), try s.srem("g", &.{ "a", "b", "missing" })); 2394 + try std.testing.expectEqual(@as(i64, 1), try s.scard("g")); 2395 + } 2396 + 2397 + // --- Lists ------------------------------------------------------------------- 2398 + 2399 + test "LPUSH / RPUSH / LLEN / LINDEX / LRANGE" { 2400 + var threaded = Io.Threaded.init(std.testing.allocator, .{}); 2401 + defer threaded.deinit(); 2402 + var s = Store.init(std.testing.allocator, threaded.io()); 2403 + defer s.deinit(); 2404 + 2405 + try std.testing.expectEqual(@as(i64, 3), try s.rpush("l", &.{ "a", "b", "c" })); 2406 + try std.testing.expectEqual(@as(i64, 4), try s.lpush("l", &.{"z"})); 2407 + try std.testing.expectEqual(@as(i64, 4), try s.llen("l")); 2408 + 2409 + const v0 = try s.lindex(std.testing.allocator, "l", 0); 2410 + try std.testing.expectEqualStrings("z", v0.?); 2411 + std.testing.allocator.free(v0.?); 2412 + 2413 + const tail = try s.lindex(std.testing.allocator, "l", -1); 2414 + try std.testing.expectEqualStrings("c", tail.?); 2415 + std.testing.allocator.free(tail.?); 2416 + 2417 + const range = try s.lrange(std.testing.allocator, "l", 0, -1); 2418 + defer Store.freeKeys(std.testing.allocator, range); 2419 + try std.testing.expectEqual(@as(usize, 4), range.len); 2420 + try std.testing.expectEqualStrings("z", range[0]); 2421 + try std.testing.expectEqualStrings("c", range[3]); 2422 + } 2423 + 2424 + test "LPOP / RPOP single + count" { 2425 + var threaded = Io.Threaded.init(std.testing.allocator, .{}); 2426 + defer threaded.deinit(); 2427 + var s = Store.init(std.testing.allocator, threaded.io()); 2428 + defer s.deinit(); 2429 + 2430 + _ = try s.rpush("l", &.{ "a", "b", "c", "d" }); 2431 + 2432 + const r1 = try s.lpop(std.testing.allocator, "l", null); 2433 + defer Store.freeLPopResult(std.testing.allocator, r1); 2434 + try std.testing.expectEqualStrings("a", r1.single); 2435 + 2436 + const r2 = try s.rpop(std.testing.allocator, "l", 2); 2437 + defer Store.freeLPopResult(std.testing.allocator, r2); 2438 + try std.testing.expectEqual(@as(usize, 2), r2.array.len); 2439 + try std.testing.expectEqualStrings("d", r2.array[0]); 2440 + try std.testing.expectEqualStrings("c", r2.array[1]); 2441 + 2442 + try std.testing.expectEqual(@as(i64, 1), try s.llen("l")); 2443 + } 2444 + 2445 + test "LREM head / tail / all" { 2446 + var threaded = Io.Threaded.init(std.testing.allocator, .{}); 2447 + defer threaded.deinit(); 2448 + var s = Store.init(std.testing.allocator, threaded.io()); 2449 + defer s.deinit(); 2450 + 2451 + _ = try s.rpush("l", &.{ "x", "y", "x", "y", "x" }); 2452 + try std.testing.expectEqual(@as(i64, 2), try s.lrem("l", 2, "x")); 2453 + try std.testing.expectEqual(@as(i64, 3), try s.llen("l")); 2454 + // From the tail this time. 2455 + try std.testing.expectEqual(@as(i64, 1), try s.lrem("l", -1, "x")); 2456 + // Remove all remaining y. 2457 + try std.testing.expectEqual(@as(i64, 2), try s.lrem("l", 0, "y")); 2458 + try std.testing.expectEqual(@as(i64, 0), s.exists(&.{"l"})); 2459 + } 2460 + 2461 + // --- ZRANGE / ZCOUNT --------------------------------------------------------- 2462 + 2463 + test "ZRANGE returns inclusive index range; ZCOUNT counts score range" { 2464 + var threaded = Io.Threaded.init(std.testing.allocator, .{}); 2465 + defer threaded.deinit(); 2466 + var s = Store.init(std.testing.allocator, threaded.io()); 2467 + defer s.deinit(); 2468 + 2469 + _ = try s.zadd("z", &.{ 2470 + .{ .score = 1, .member = "a" }, 2471 + .{ .score = 2, .member = "b" }, 2472 + .{ .score = 3, .member = "c" }, 2473 + .{ .score = 4, .member = "d" }, 2474 + }, .{}); 2475 + 2476 + const r = try s.zrange(std.testing.allocator, "z", 1, 2, false); 2477 + defer Store.freeScoredMembers(std.testing.allocator, r); 2478 + try std.testing.expectEqual(@as(usize, 2), r.len); 2479 + try std.testing.expectEqualStrings("b", r[0].member); 2480 + try std.testing.expectEqualStrings("c", r[1].member); 2481 + 2482 + try std.testing.expectEqual(@as(i64, 2), try s.zcount("z", 2, 3)); 2483 + try std.testing.expectEqual(@as(i64, 4), try s.zcount("z", -std.math.inf(f64), std.math.inf(f64))); 2484 + } 2485 + 2486 + // --- Stream extras: XREAD / XRANGE / XTRIM / streamLastId / xgroupDestroy --- 2487 + 2488 + test "XREAD returns entries strictly after start_id" { 2489 + var threaded = Io.Threaded.init(std.testing.allocator, .{}); 2490 + defer threaded.deinit(); 2491 + var s = Store.init(std.testing.allocator, threaded.io()); 2492 + defer s.deinit(); 2493 + 2494 + _ = try s.xadd("st", .{ .ms = 1, .seq = 0 }, &.{ "k", "v1" }); 2495 + _ = try s.xadd("st", .{ .ms = 2, .seq = 0 }, &.{ "k", "v2" }); 2496 + 2497 + const res = try s.xread(std.testing.allocator, &.{"st"}, &.{StreamId.zero()}, null); 2498 + defer freeReadResults(std.testing.allocator, res); 2499 + try std.testing.expectEqual(@as(usize, 1), res.len); 2500 + try std.testing.expectEqual(@as(usize, 2), res[0].entries.len); 2501 + } 2502 + 2503 + test "XRANGE inclusive on both ends" { 2504 + var threaded = Io.Threaded.init(std.testing.allocator, .{}); 2505 + defer threaded.deinit(); 2506 + var s = Store.init(std.testing.allocator, threaded.io()); 2507 + defer s.deinit(); 2508 + 2509 + _ = try s.xadd("st", .{ .ms = 1, .seq = 0 }, &.{ "k", "v" }); 2510 + _ = try s.xadd("st", .{ .ms = 2, .seq = 0 }, &.{ "k", "v" }); 2511 + _ = try s.xadd("st", .{ .ms = 3, .seq = 0 }, &.{ "k", "v" }); 2512 + 2513 + const res = try s.xrange(std.testing.allocator, "st", .{ .ms = 1, .seq = 0 }, .{ .ms = 2, .seq = 0 }, null); 2514 + defer freeDeliveredEntries(std.testing.allocator, res); 2515 + try std.testing.expectEqual(@as(usize, 2), res.len); 2516 + } 2517 + 2518 + test "XTRIM MAXLEN drops the oldest entries" { 2519 + var threaded = Io.Threaded.init(std.testing.allocator, .{}); 2520 + defer threaded.deinit(); 2521 + var s = Store.init(std.testing.allocator, threaded.io()); 2522 + defer s.deinit(); 2523 + 2524 + var i: u64 = 0; 2525 + while (i < 5) : (i += 1) { 2526 + _ = try s.xadd("st", .{ .ms = i + 1, .seq = 0 }, &.{ "k", "v" }); 2527 + } 2528 + try std.testing.expectEqual(@as(i64, 3), try s.xtrimMaxlen("st", 2)); 2529 + try std.testing.expectEqual(@as(i64, 2), try s.xlen("st")); 2530 + } 2531 + 2532 + test "XGROUP DESTROY removes the group" { 2533 + var threaded = Io.Threaded.init(std.testing.allocator, .{}); 2534 + defer threaded.deinit(); 2535 + var s = Store.init(std.testing.allocator, threaded.io()); 2536 + defer s.deinit(); 2537 + 2538 + _ = try s.xadd("st", null, &.{ "k", "v" }); 2539 + try s.xgroupCreate("st", "g1", .{ .id = StreamId.zero() }); 2540 + try std.testing.expect(try s.xgroupDestroy("st", "g1")); 2541 + try std.testing.expect(!try s.xgroupDestroy("st", "g1")); 2542 + } 2543 + 2544 + // --- glob match -------------------------------------------------------------- 2545 + 2546 + test "globMatch covers the common patterns" { 2547 + try std.testing.expect(globMatch("*", "hello")); 2548 + try std.testing.expect(globMatch("hel*", "hello")); 2549 + try std.testing.expect(globMatch("h?llo", "hello")); 2550 + try std.testing.expect(!globMatch("h?llo", "helllo")); 2551 + try std.testing.expect(globMatch("h[ae]llo", "hallo")); 2552 + try std.testing.expect(!globMatch("h[ae]llo", "hbllo")); 2553 + try std.testing.expect(globMatch("[a-c]at", "bat")); 2554 + try std.testing.expect(globMatch("user:?", "user:1")); 2555 + try std.testing.expect(!globMatch("user:?", "user:11")); 2556 + }

Configure Feed

Configure Feed