diff --git a/src/main.zig b/src/main.zig
index 7ab545c..6e5dfa2 100644
--- a/src/main.zig
+++ b/src/main.zig
@@ -585,16 +585,150 @@ pub fn main() !void {
         return;
     }
 
+    if (std.mem.eql(u8, command, "remote-select")) {
+        // Reuses the <dir> slot as a host:port to connect to, exactly like
+        // `verify` reuses it as the directory under test - `remote-select`
+        // never opens a local Db at all.
+        try runRemoteSelect(allocator, stdout, stderr, dir, rest);
+        return;
+    }
+
+    if (std.mem.eql(u8, command, "remote-join")) {
+        // Same "<dir> slot is really host:port" convention as
+        // `remote-select` - never opens a local Db.
+        try runRemoteJoin(allocator, stdout, stderr, dir, rest);
+        return;
+    }
+
+    if (std.mem.eql(u8, command, "remote-aggregate")) {
+        // Same "<dir> slot is really host:port" convention as
+        // `remote-select` - never opens a local Db.
+        try runRemoteAggregate(allocator, stdout, stderr, dir, rest);
+        return;
+    }
+
+    if (std.mem.eql(u8, command, "remote-window")) {
+        // Same "<dir> slot is really host:port" convention as
+        // `remote-select` - never opens a local Db.
+        try runRemoteWindow(allocator, stdout, stderr, dir, rest);
+        return;
+    }
+
+    if (std.mem.eql(u8, command, "remote-publish")) {
+        // Same "<dir> slot is really host:port" convention as
+        // `remote-select` - never opens a local Db.
+        try runRemotePublish(allocator, stdout, stderr, dir, rest);
+        return;
+    }
+
+    if (std.mem.eql(u8, command, "remote-subscribe")) {
+        // Same "<dir> slot is really host:port" convention as
+        // `remote-select` - never opens a local Db.
+        try runRemoteSubscribe(allocator, stdout, stderr, dir, rest);
+        return;
+    }
+
+    if (std.mem.eql(u8, command, "remote-unsubscribe")) {
+        // Same "<dir> slot is really host:port" convention as
+        // `remote-select` - never opens a local Db.
+        try runRemoteUnsubscribe(allocator, stdout, stderr, dir, rest);
+        return;
+    }
+
+    if (std.mem.eql(u8, command, "remote-explain")) {
+        // Same "<dir> slot is really host:port" convention as
+        // `remote-select` - never opens a local Db.
+        try runRemoteExplain(allocator, stdout, stderr, dir, rest);
+        return;
+    }
+
+    if (std.mem.eql(u8, command, "remote-update")) {
+        // Same "<dir> slot is really host:port" convention as
+        // `remote-select` - never opens a local Db.
+        try runRemoteUpdate(allocator, stdout, stderr, dir, rest);
+        return;
+    }
+
+    if (std.mem.eql(u8, command, "remote-delete")) {
+        // Same "<dir> slot is really host:port" convention as
+        // `remote-select` - never opens a local Db.
+        try runRemoteDelete(allocator, stdout, stderr, dir, rest);
+        return;
+    }
+
+    if (std.mem.eql(u8, command, "remote-export")) {
+        // Same "<dir> slot is really host:port" convention as
+        // `remote-select` - never opens a local Db.
+        try runRemoteExport(allocator, stdout, stderr, dir, rest);
+        return;
+    }
+
+    if (std.mem.eql(u8, command, "remote-import")) {
+        // Same "<dir> slot is really host:port" convention as
+        // `remote-select` - never opens a local Db.
+        try runRemoteImport(allocator, stdout, stderr, dir, rest);
+        return;
+    }
+
+    if (std.mem.eql(u8, command, "remote-user")) {
+        // Same "<dir> slot is really host:port" convention as
+        // `remote-select` - never opens a local Db.
+        try runRemoteUser(allocator, stdout, stderr, dir, rest);
+        return;
+    }
+
+    if (std.mem.eql(u8, command, "remote-stats")) {
+        // Same "<dir> slot is really host:port" convention as
+        // `remote-select`/`remote-user` - never opens a local Db.
+        try runRemoteStats(allocator, stdout, stderr, dir, rest);
+        return;
+    }
+
+    if (std.mem.eql(u8, command, "remote-ttl")) {
+        // Same "<dir> slot is really host:port" convention as
+        // `remote-select`/`remote-user`/`remote-stats` - never opens a
+        // local Db.
+        try runRemoteTtl(allocator, stdout, stderr, dir, rest);
+        return;
+    }
+
+    if (std.mem.eql(u8, command, "remote-coll-ttl")) {
+        // Same "<dir> slot is really host:port" convention as
+        // `remote-ttl` - never opens a local Db.
+        try runRemoteCollTtl(allocator, stdout, stderr, dir, rest);
+        return;
+    }
+
+    if (std.mem.eql(u8, command, "remote-coll")) {
+        // Same "<dir> slot is really host:port" convention as
+        // `remote-coll-ttl` - never opens a local Db.
+        try runRemoteColl(allocator, stdout, stderr, dir, rest);
+        return;
+    }
+
+    if (std.mem.eql(u8, command, "remote-txn")) {
+        // Same "<dir> slot is really host:port" convention as every other
+        // remote-* command - never opens a local Db.
+        try runRemoteTxn(allocator, stdout, stderr, dir, rest);
+        return;
+    }
+
     var db = try cobalt.Db.open(allocator, dir, .{});
     defer db.close();
 
     if (std.mem.eql(u8, command, "put")) {
         if (rest.len < 2) return fail(stderr, "put needs <key> <value>");
         try db.put(rest[0], rest[1]);
+        if (rest.len >= 4 and std.mem.eql(u8, rest[2], "--ttl")) {
+            const seconds = std.fmt.parseInt(u64, rest[3], 10) catch return fail(stderr, "--ttl needs a whole number of seconds");
+            try cobalt.ttl.setExpireSeconds(&db, allocator, rest[0], std.time.nanoTimestamp(), seconds);
+        } else if (rest.len > 2) {
+            return fail(stderr, "put takes <key> <value> [--ttl <seconds>]");
+        }
         try db.maybeCompact();
     } else if (std.mem.eql(u8, command, "get")) {
         if (rest.len < 1) return fail(stderr, "get needs <key>");
-        if (try db.get(allocator, rest[0])) |value| {
+        if (try cobalt.ttl.getLive(&db, allocator, rest[0], std.time.nanoTimestamp())) |value| {
             defer allocator.free(value);
             try stdout.print("{s}\n", .{value});
         } else {
@@ -603,7 +737,27 @@ pub fn main() !void {
     } else if (std.mem.eql(u8, command, "del")) {
         if (rest.len < 1) return fail(stderr, "del needs <key>");
         try db.delete(rest[0]);
+        try cobalt.ttl.clearExpire(&db, allocator, rest[0]);
         try db.maybeCompact();
+    } else if (std.mem.eql(u8, command, "expire")) {
+        if (rest.len < 2) return fail(stderr, "expire needs <key> <seconds>");
+        const seconds = std.fmt.parseInt(u64, rest[1], 10) catch return fail(stderr, "expire needs a whole number of seconds");
+        const existing = try db.get(allocator, rest[0]) orelse return fail(stderr, "no such key");
+        allocator.free(existing);
+        try cobalt.ttl.setExpireSeconds(&db, allocator, rest[0], std.time.nanoTimestamp(), seconds);
+    } else if (std.mem.eql(u8, command, "persist")) {
+        if (rest.len < 1) return fail(stderr, "persist needs <key>");
+        try cobalt.ttl.clearExpire(&db, allocator, rest[0]);
+    } else if (std.mem.eql(u8, command, "ttl")) {
+        if (rest.len < 1) return fail(stderr, "ttl needs <key>");
+        switch (try cobalt.ttl.remaining(&db, allocator, rest[0], std.time.nanoTimestamp())) {
+            .noKey => try stdout.writeAll("-2\n"),
+            .noExpiry => try stdout.writeAll("-1\n"),
+            .seconds => |s| try stdout.print("{d}\n", .{s}),
+        }
+    } else if (std.mem.eql(u8, command, "expire-sweep")) {
+        const reaped = try cobalt.ttl.sweepExpired(&db, allocator, std.time.nanoTimestamp());
+        try stdout.print("reaped: {d}\n", .{reaped});
     } else if (std.mem.eql(u8, command, "scan")) {
         const start: ?[]const u8 = if (rest.len >= 1) rest[0] else null;
         const end: ?[]const u8 = if (rest.len >= 2) rest[1] else null;
@@ -625,7 +779,47 @@ pub fn main() !void {
         try db.compact();
         try stdout.print("sstables after compaction: {d}\n", .{db.sstables.items.len});
     } else if (std.mem.eql(u8, command, "stats")) {
-        try db.stats().print(stdout);
+        if (rest.len >= 1 and std.mem.eql(u8, rest[0], "--prom")) {
+            try cobalt.metrics.formatDbStats(stdout, "cobalt", db.stats());
+        } else {
+            try db.stats().print(stdout);
+        }
+    } else if (std.mem.eql(u8, command, "serve")) {
+        const bind = if (rest.len >= 1) rest[0] else "127.0.0.1:6380";
+        const serveRest = if (rest.len >= 1) rest[1..] else rest;
+        try runServe(allocator, stdout, &db, bind, serveRest);
+    } else if (std.mem.eql(u8, command, "backup")) {
+        if (rest.len < 1) return fail(stderr, "backup needs <file>");
+        const snap = try db.snapshot();
+        const count = try cobalt.backup.dumpToFile(allocator, snap, rest[0]);
+        try stdout.print("backup written: {d} entries, seq={d}, file={s}\n", .{ count, snap.seq, rest[0] });
+    } else if (std.mem.eql(u8, command, "restore")) {
+        if (rest.len < 1) return fail(stderr, "restore needs <file>");
+        const info = cobalt.backup.restoreFromFile(allocator, &db, rest[0], 500) catch |err| {
+            try stderr.print("restore failed: {s}\n", .{@errorName(err)});
+            std.process.exit(1);
+        };
+        try stdout.print("restore applied: {d} entries from seq={d}\n", .{ info.entryCount, info.seq });
+    } else if (std.mem.eql(u8, command, "coll")) {
+        try runColl(allocator, stdout, stderr, &db, rest);
+    } else if (std.mem.eql(u8, command, "table")) {
+        try runTable(allocator, stdout, stderr, &db, rest);
+    } else if (std.mem.eql(u8, command, "user")) {
+        try runUser(allocator, stdout, stderr, &db, rest);
+    } else if (std.mem.eql(u8, command, "txn")) {
+        try runTxn(allocator, stdout, stderr, &db, rest);
+    } else if (std.mem.eql(u8, command, "replicate")) {
+        if (rest.len < 1) return fail(stderr, "replicate needs <replica-dir>");
+        try runReplicate(allocator, stdout, &db, rest[0], rest[1..]);
+    } else if (std.mem.eql(u8, command, "repl")) {
+        try runRepl(allocator, stdout, stderr, &db, rest);
+    } else if (std.mem.eql(u8, command, "backup-info")) {
+        if (rest.len < 1) return fail(stderr, "backup-info needs <file>");
+        const info = cobalt.backup.inspectFile(allocator, rest[0]) catch |err| {
+            try stderr.print("invalid backup: {s}\n", .{@errorName(err)});
+            std.process.exit(1);
+        };
+        try stdout.print("entries={d} seq={d}\n", .{ info.entryCount, info.seq });
     } else {
         try stderr.print("unknown command: {s}\n", .{command});
         try stderr.writeAll(usage);