diff --git a/src/server.zig b/src/server.zig index 4d77d81..c75f118 100644 --- a/src/server.zig +++ b/src/server.zig @@ -619,3 +619,161 @@ pub const Server = struct { } } + /// Answers the `coll_ttl_op` opcode: the network counterpart of the + /// CLI's local `coll expire|ttl|persist `/`coll + /// expire-sweep` (main.zig), all backed by the same `collttl.zig` + /// functions the local commands call directly - same shape as + /// `handleTtlOp` above, but `expire`'s existence check is + /// `collections.collectionExists` (a whole namespace) instead of + /// `db.get` (one key), matching the CLI's own `coll expire` + /// check. + fn handleCollTtlOp(self: *Server, payload: []const u8, writer: anytype) !void { + const args = protocol.decodeCollTtlOp(payload) catch { + return writeErr(writer, .bad_request, "malformed coll ttl payload"); + }; + const now = std.time.nanoTimestamp(); + switch (args.action) { + .expire => { + const exists = collmod.collectionExists(self.db, self.allocator, args.kind, args.name) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + if (!exists) return writeErr(writer, .not_found, "no such collection"); + collttlmod.setExpireSeconds(self.db, self.allocator, args.kind, args.name, now, args.seconds) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), ""); + }, + .persist => { + collttlmod.clearExpire(self.db, self.allocator, args.kind, args.name) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), ""); + }, + .query => { + const result = collttlmod.remaining(self.db, self.allocator, args.kind, args.name, 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 = collttlmod.sweepExpired(self.db, self.allocator, now) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + if (reaped > 0) self.collTtlSweptTotal.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); + }, + } + } + + /// Answers the `coll_op` opcode: the network counterpart of the CLI's + /// local `coll ...` (main.zig's `runColl`, backed by + /// `collections.zig`) - collections were previously reachable only from + /// a process holding the `Db` directly, the same gap `ttl_op`/ + /// `coll_ttl_op` already closed for TTL management. Applies the same + /// lazy-expiry check `runColl` performs locally before every dispatch + /// (`collttl.expireIfDue` on the target name, plus `expireIfDueMany` on + /// the extra names of the three multi-set actions) so a collection + /// whose TTL is already due never appears alive over the wire either. + /// The actual per-action work lives in `dispatchCollOp` below; this + /// function only handles decode and the shared expiry check. + fn handleCollOp(self: *Server, payload: []const u8, writer: anytype) !void { + const args = protocol.decodeCollOp(self.allocator, payload) catch { + return writeErr(writer, .bad_request, "malformed coll payload"); + }; + defer self.allocator.free(args.args); + const now = std.time.nanoTimestamp(); + _ = collttlmod.expireIfDue(self.db, self.allocator, args.action.kind(), args.name, now) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + switch (args.action) { + .set_inter, .set_union, .set_diff => _ = collttlmod.expireIfDueMany(self.db, self.allocator, .set, args.args, now) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }, + else => {}, + } + // Thirty actions each wrapping their own `db` call in a `catch |err| + // return writeErr(...)` (the pattern every other handler in this + // file uses) would triple the size of this function for no benefit + // over one shared boundary: `dispatchCollOp` uses plain `try` and + // two sentinel errors for malformed arguments, and this one catch + // maps both cases to the right status - the same information either + // way, just not repeated thirty times. + self.dispatchCollOp(args, writer) catch |err| return switch (err) { + error.InvalidArgCount, error.InvalidNumber => writeErr(writer, .bad_request, @errorName(err)), + else => writeErr(writer, .internal, @errorName(err)), + }; + } + + fn writeCollReply(self: *Server, writer: anytype, reply: protocol.CollReply) !void { + const respPayload = protocol.encodeCollReply(self.allocator, reply) catch |err| { + return writeErr(writer, .internal, @errorName(err)); + }; + defer self.allocator.free(respPayload); + try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), respPayload); + } + + /// One arm per `protocol.CollAction`, each translating straight from + /// `args.args` (the CLI-text-form string arguments) into the matching + /// `collections.zig` call and back into a `protocol.CollReply` - the + /// exact same translation `main.zig`'s `runColl` already does for the + /// local CLI, just producing a wire reply instead of a printed line. + fn dispatchCollOp(self: *Server, args: protocol.CollOpArgs, writer: anytype) !void { + const a = args.args; + switch (args.action) { + .counter_get => { + const v = try collmod.Counter.get(self.db, self.allocator, args.name); + try self.writeCollReply(writer, .{ .int = v }); + }, + .counter_set => { + if (a.len < 1) return error.InvalidArgCount; + const v = std.fmt.parseInt(i64, a[0], 10) catch return error.InvalidNumber; + try collmod.Counter.set(self.db, self.allocator, args.name, v); + try self.writeCollReply(writer, .ok); + }, + .counter_incrby => { + if (a.len < 1) return error.InvalidArgCount; + const delta = std.fmt.parseInt(i64, a[0], 10) catch return error.InvalidNumber; + const v = try collmod.Counter.incrBy(self.db, self.allocator, args.name, delta); + try self.writeCollReply(writer, .{ .int = v }); + }, + .hash_set => { + if (a.len < 2) return error.InvalidArgCount; + const created = try collmod.Hash.set(self.db, self.allocator, args.name, a[0], a[1]); + try self.writeCollReply(writer, .{ .int = if (created) 1 else 0 }); + }, + .hash_get => { + if (a.len < 1) return error.InvalidArgCount; + if (try collmod.Hash.get(self.db, self.allocator, args.name, a[0])) |v| { + defer self.allocator.free(v); + try self.writeCollReply(writer, .{ .bytes = v }); + } else { + try self.writeCollReply(writer, .nil); + } + }, + .hash_del => { + if (a.len < 1) return error.InvalidArgCount; + const existed = try collmod.Hash.del(self.db, self.allocator, args.name, a[0]); + try self.writeCollReply(writer, .{ .int = if (existed) 1 else 0 }); + }, + .hash_len => { + const n = try collmod.Hash.len(self.db, self.allocator, args.name); + try self.writeCollReply(writer, .{ .int = @intCast(n) }); + }, + .hash_getall => { + const fields = try collmod.Hash.getAll(self.db, self.allocator, args.name); + defer { + for (fields) |f| f.free(self.allocator); + self.allocator.free(fields); + } + const pairs = try self.allocator.alloc(protocol.CollPair, fields.len); + defer self.allocator.free(pairs); + for (fields, 0..) |f, i| pairs[i] = .{ .field = f.field, .value = f.value }; + try self.writeCollReply(writer, .{ .pairList = pairs });