diff --git a/src/server.zig b/src/server.zig index 982c7d6..4d77d81 100644 --- a/src/server.zig +++ b/src/server.zig @@ -470,3 +470,152 @@ pub const Server = struct { // unconditionally rather than ever consulting a session store. .auth_token => try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), ""), + .get => { + const value = self.db.get(self.allocator, frame.payload) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + if (value) |v| { + defer self.allocator.free(v); + try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), v); + } else { + try protocol.writeFrame(writer, @intFromEnum(protocol.Status.not_found), ""); + } + }, + + .set => { + const args = protocol.decodeSet(frame.payload) catch { + return writeErr(writer, .bad_request, "malformed set payload"); + }; + self.db.put(args.key, args.value) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), ""); + }, + + .del => { + self.db.delete(frame.payload) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), ""); + }, + + .scan => try self.handleScan(frame.payload, writer), + .prefix => try self.handlePrefix(frame.payload, writer), + + .stats => { + const s = self.db.stats(); + const payload = protocol.encodeStats(self.allocator, s) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + defer self.allocator.free(payload); + try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), payload); + }, + + .flush => { + self.db.flush() catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), ""); + }, + + .compact => { + self.db.compact() catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), ""); + }, + + .batch => try self.handleBatch(frame.payload, writer), + + .table_select => try self.handleTableSelect(frame.payload, writer), + + .table_join => try self.handleTableJoin(frame.payload, writer), + + .table_aggregate => try self.handleTableAggregate(frame.payload, writer), + .table_window => try self.handleTableWindow(frame.payload, writer), + + // All three already handled and returned from above, before + // `self.mutex` was ever taken - never reached here. + .subscribe, .publish, .unsubscribe => unreachable, + + .table_explain => try self.handleTableExplain(frame.payload, writer), + + .table_update => try self.handleTableUpdate(frame.payload, writer), + + .table_delete => try self.handleTableDelete(frame.payload, writer), + + .table_export => try self.handleTableExport(frame.payload, writer), + + .table_import => try self.handleTableImport(frame.payload, writer), + + .user_admin => try self.handleUserAdmin(frame.payload, writer), + + .metrics => try self.handleMetrics(writer), + + .ttl_op => try self.handleTtlOp(frame.payload, writer), + + .coll_ttl_op => try self.handleCollTtlOp(frame.payload, writer), + + .coll_op => try self.handleCollOp(frame.payload, writer), + + .txn_op => try self.handleTxnOp(frame.payload, writer), + } + } + + /// Answers the `ttl_op` opcode: the network counterpart of the CLI's + /// local `expire`/`persist`/`ttl`/`expire-sweep` (main.zig), all backed + /// by the same `ttl.zig` functions the local commands call directly - + /// `query`/`sweep` were previously reachable only from a process that + /// had the `Db` open itself. `expire` mirrors the CLI's Redis-EXPIRE- + /// strict existence check (main.zig checks `db.get` before calling + /// `ttl.setExpireSeconds`; here the same check happens before the same + /// call) and answers `not_found` rather than `ok` when the key does not + /// currently exist, instead of the looser PEXPIREAT-like + /// `ttl.setExpireAt` that would silently accept it. + fn handleTtlOp(self: *Server, payload: []const u8, writer: anytype) !void { + const args = protocol.decodeTtlOp(payload) catch { + return writeErr(writer, .bad_request, "malformed ttl payload"); + }; + const now = std.time.nanoTimestamp(); + switch (args.action) { + .expire => { + const existing = self.db.get(self.allocator, args.key) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + if (existing == null) return writeErr(writer, .not_found, "no such key"); + self.allocator.free(existing.?); + ttlmod.setExpireSeconds(self.db, self.allocator, args.key, now, args.seconds) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), ""); + }, + .persist => { + ttlmod.clearExpire(self.db, self.allocator, args.key) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), ""); + }, + .query => { + const result = ttlmod.remaining(self.db, self.allocator, args.key, now) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + const respPayload = protocol.encodeTtlResult(self.allocator, result) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + defer self.allocator.free(respPayload); + try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), respPayload); + }, + .sweep => { + const reaped = ttlmod.sweepExpired(self.db, self.allocator, now) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + if (reaped > 0) self.ttlSweptTotal.incrBy(@intCast(reaped)); + const respPayload = protocol.encodeTtlSweepCount(self.allocator, @intCast(reaped)) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + defer self.allocator.free(respPayload); + try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), respPayload); + }, + } + } +