add real-time activity tracking and domain types · zzstoatzz.io/leaflet-search@81e5fae

+46

backend/src/activity.zig

··· 1 + const std = @import("std"); 2 + 3 + /// Ring buffer - 60 slots at 100ms each = 6 second window 4 + const SLOTS = 60; 5 + const TICK_MS = 100; 6 + 7 + var counts: [SLOTS]u16 = .{0} ** SLOTS; 8 + var current_slot: usize = 0; 9 + var mutex: std.Thread.Mutex = .{}; 10 + var tick_thread: ?std.Thread = null; 11 + 12 + /// Start the background tick thread 13 + pub fn init() void { 14 + tick_thread = std.Thread.spawn(.{}, tickLoop, .{}) catch null; 15 + } 16 + 17 + /// Record a search event 18 + pub fn record() void { 19 + mutex.lock(); 20 + defer mutex.unlock(); 21 + counts[current_slot] +|= 1; 22 + } 23 + 24 + /// Get activity counts (oldest to newest) 25 + pub fn getCounts() [SLOTS]u16 { 26 + mutex.lock(); 27 + defer mutex.unlock(); 28 + 29 + var result: [SLOTS]u16 = undefined; 30 + for (0..SLOTS) |i| { 31 + const idx = (current_slot + 1 + i) % SLOTS; 32 + result[i] = counts[idx]; 33 + } 34 + return result; 35 + } 36 + 37 + /// Background thread - advances slot every 100ms 38 + fn tickLoop() void { 39 + while (true) { 40 + std.Thread.sleep(TICK_MS * std.time.ns_per_ms); 41 + mutex.lock(); 42 + current_slot = (current_slot + 1) % SLOTS; 43 + counts[current_slot] = 0; 44 + mutex.unlock(); 45 + } 46 + }

+28 -6

backend/src/dashboard.zig

··· 172 172 \\ } 173 173 \\ .metrics { 174 174 \\ display: flex; 175 - \\ gap: 2rem; 176 - \\ margin-bottom: 1.5rem; 175 + \\ gap: 1.5rem; 176 + \\ margin-bottom: 1rem; 177 177 \\ } 178 178 \\ .metric-value { 179 - \\ font-size: 24px; 180 - \\ color: #fff; 179 + \\ font-size: 16px; 180 + \\ color: #888; 181 + \\ font-weight: normal; 181 182 \\ } 182 183 \\ .metric-label { 183 - \\ font-size: 11px; 184 - \\ color: #555; 184 + \\ font-size: 10px; 185 + \\ color: #444; 186 + \\ text-transform: uppercase; 187 + \\ letter-spacing: 0.5px; 185 188 \\ } 186 189 \\ .chart-box { 187 190 \\ background: #111; ··· 247 250 \\ color: #aaa; 248 251 \\ } 249 252 \\ .tag .n { color: #444; margin-left: 4px; } 253 + \\ .live { font-size: 11px; color: #555; } 254 + \\ .live span { color: #4ade80; } 250 255 \\ footer { 251 256 \\ margin-top: 2rem; 252 257 \\ padding-top: 1rem; ··· 287 292 \\ <div class="metric-label">publications</div> 288 293 \\ </div> 289 294 \\ </div> 295 + \\ <div class="live" id="live"></div> 290 296 \\ </section> 291 297 \\ 292 298 \\ <section> ··· 395 401 \\ '<a class="tag" href="https://leaflet-search.pages.dev/?tag=' + encodeURIComponent(t.tag) + '">' + 396 402 \\ t.tag + '<span class="n">' + t.count + '</span></a>' 397 403 \\ ).join(''); 404 + \\ 405 + \\ // live activity - just a number 406 + \\ const liveEl = document.getElementById('live'); 407 + \\ let lastN = -1; 408 + \\ async function pollLive() { 409 + \\ try { 410 + \\ const r = await fetch('/activity'); 411 + \\ const c = await r.json(); 412 + \\ const n = c.reduce((a,b) => a+b, 0); 413 + \\ if (n !== lastN) { 414 + \\ liveEl.innerHTML = n > 0 ? '<span>' + n + '</span> req/6s' : ''; 415 + \\ lastN = n; 416 + \\ } 417 + \\ } catch(e) {} 418 + \\ } 419 + \\ pollLive(); setInterval(pollLive, 1000); 398 420 \\ </script> 399 421 \\</body> 400 422 \\</html>

+2

backend/src/db/mod.zig

··· 6 6 const Client = @import("Client.zig"); 7 7 const schema = @import("schema.zig"); 8 8 const result = @import("result.zig"); 9 + const activity = @import("../activity.zig"); 9 10 10 11 pub const Row = result.Row; 11 12 pub const BatchResult = result.BatchResult; ··· 328 329 } 329 330 330 331 pub fn recordSearch(query: []const u8) void { 332 + activity.record(); // track for real-time sparkline 331 333 var c = &(client orelse return); 332 334 c.exec("UPDATE stats SET total_searches = total_searches + 1 WHERE id = 1", &.{}) catch {}; 333 335

+4

backend/src/main.zig

··· 5 5 const db = @import("db/mod.zig"); 6 6 const server = @import("server.zig"); 7 7 const tap = @import("tap.zig"); 8 + const activity = @import("activity.zig"); 8 9 9 10 const MAX_HTTP_WORKERS = 16; 10 11 const SOCKET_TIMEOUT_SECS = 30; ··· 16 17 17 18 // init turso 18 19 try db.init(); 20 + 21 + // start activity tracker 22 + activity.init(); 19 23 20 24 // start tap consumer in background 21 25 const tap_thread = try Thread.spawn(.{}, tap.consumer, .{allocator});

+21

backend/src/server.zig

··· 4 4 const mem = std.mem; 5 5 const db = @import("db/mod.zig"); 6 6 const dashboard = @import("dashboard.zig"); 7 + const activity = @import("activity.zig"); 7 8 8 9 const HTTP_BUF_SIZE = 8192; 9 10 const QUERY_PARAM_BUF_SIZE = 64; ··· 58 59 try handleDashboard(request); 59 60 } else if (mem.startsWith(u8, target, "/similar")) { 60 61 try handleSimilar(request, target); 62 + } else if (mem.eql(u8, target, "/activity")) { 63 + try handleActivity(request); 61 64 } else { 62 65 try sendNotFound(request); 63 66 } ··· 213 216 214 217 try sendJson(request, results); 215 218 } 219 + 220 + fn handleActivity(request: *http.Server.Request) !void { 221 + const counts = activity.getCounts(); 222 + 223 + // format as JSON array 224 + var buf: [512]u8 = undefined; 225 + var stream = std.io.fixedBufferStream(&buf); 226 + const writer = stream.writer(); 227 + 228 + writer.writeByte('[') catch return; 229 + for (counts, 0..) |c, i| { 230 + if (i > 0) writer.writeByte(',') catch return; 231 + writer.print("{d}", .{c}) catch return; 232 + } 233 + writer.writeByte(']') catch return; 234 + 235 + try sendJson(request, stream.getWritten()); 236 + }

+20 -16

backend/src/tap.zig

··· 5 5 const Allocator = mem.Allocator; 6 6 const websocket = @import("websocket"); 7 7 const db = @import("db/mod.zig"); 8 + const types = @import("types.zig"); 9 + const Did = types.Did; 10 + const AtUri = types.AtUri; 8 11 9 12 const DOCUMENT_COLLECTION = "pub.leaflet.document"; 10 13 const PUBLICATION_COLLECTION = "pub.leaflet.publication"; ··· 118 121 const action = rec.get("action") orelse return; 119 122 if (action != .string) return; 120 123 121 - const did = rec.get("did") orelse return; 122 - if (did != .string) return; 124 + const did_val = rec.get("did") orelse return; 125 + if (did_val != .string) return; 126 + const did = Did.parse(did_val.string) orelse return; 123 127 124 128 const rkey = rec.get("rkey") orelse return; 125 129 if (rkey != .string) return; 126 130 127 - const uri_str = try std.fmt.allocPrint(allocator, "at://{s}/{s}/{s}", .{ did.string, collection.string, rkey.string }); 128 - defer allocator.free(uri_str); 131 + const uri = AtUri.build(allocator, did, collection.string, rkey.string) catch return; 132 + defer allocator.free(uri.raw); 129 133 130 134 if (mem.eql(u8, action.string, "create") or mem.eql(u8, action.string, "update")) { 131 135 const record = rec.get("record") orelse return; 132 136 if (record != .object) return; 133 137 134 138 if (mem.eql(u8, collection.string, DOCUMENT_COLLECTION)) { 135 - processDocument(allocator, uri_str, did.string, rkey.string, record.object) catch |err| { 139 + processDocument(allocator, uri, record.object) catch |err| { 136 140 std.debug.print("document processing error: {}\n", .{err}); 137 141 }; 138 142 } else if (mem.eql(u8, collection.string, PUBLICATION_COLLECTION)) { 139 - processPublication(uri_str, did.string, rkey.string, record.object) catch |err| { 143 + processPublication(uri, record.object) catch |err| { 140 144 std.debug.print("publication processing error: {}\n", .{err}); 141 145 }; 142 146 } 143 147 } else if (mem.eql(u8, action.string, "delete")) { 144 148 if (mem.eql(u8, collection.string, DOCUMENT_COLLECTION)) { 145 - db.deleteDocument(uri_str); 146 - std.debug.print("deleted document: {s}\n", .{uri_str}); 149 + db.deleteDocument(uri.str()); 150 + std.debug.print("deleted document: {s}\n", .{uri.str()}); 147 151 } else if (mem.eql(u8, collection.string, PUBLICATION_COLLECTION)) { 148 - db.deletePublication(uri_str); 149 - std.debug.print("deleted publication: {s}\n", .{uri_str}); 152 + db.deletePublication(uri.str()); 153 + std.debug.print("deleted publication: {s}\n", .{uri.str()}); 150 154 } 151 155 } 152 156 } 153 157 154 - fn processDocument(allocator: Allocator, uri: []const u8, did: []const u8, rkey: []const u8, record: json.ObjectMap) !void { 158 + fn processDocument(allocator: Allocator, uri: AtUri, record: json.ObjectMap) !void { 155 159 // get title 156 160 const title_val = record.get("title") orelse return; 157 161 if (title_val != .string) return; ··· 214 218 return; 215 219 } 216 220 217 - try db.insertDocument(uri, did, rkey, title, content_buf.items, created_at, publication_uri, tags_list.items); 218 - std.debug.print("indexed document: {s} ({} chars, {} tags)\n", .{ uri, content_buf.items.len, tags_list.items.len }); 221 + try db.insertDocument(uri.str(), uri.did().str(), uri.rkey(), title, content_buf.items, created_at, publication_uri, tags_list.items); 222 + std.debug.print("indexed document: {s} ({} chars, {} tags)\n", .{ uri.str(), content_buf.items.len, tags_list.items.len }); 219 223 } 220 224 221 225 fn extractPlaintextFromPage(allocator: Allocator, buf: *std.ArrayList(u8), page: json.ObjectMap) !void { ··· 299 303 } 300 304 } 301 305 302 - fn processPublication(uri: []const u8, did: []const u8, rkey: []const u8, record: json.ObjectMap) !void { 306 + fn processPublication(uri: AtUri, record: json.ObjectMap) !void { 303 307 const name_val = record.get("name") orelse return; 304 308 if (name_val != .string) return; 305 309 const name = name_val.string; ··· 318 322 break :blk null; 319 323 }; 320 324 321 - try db.insertPublication(uri, did, rkey, name, description, base_path); 322 - std.debug.print("indexed publication: {s} (base_path: {s})\n", .{ uri, base_path orelse "none" }); 325 + try db.insertPublication(uri.str(), uri.did().str(), uri.rkey(), name, description, base_path); 326 + std.debug.print("indexed publication: {s} (base_path: {s})\n", .{ uri.str(), base_path orelse "none" }); 323 327 }

+129

backend/src/types.zig

··· 1 + const std = @import("std"); 2 + const mem = std.mem; 3 + 4 + /// decentralized identifier - did:plc:xxx or did:web:xxx 5 + pub const Did = struct { 6 + raw: []const u8, 7 + 8 + pub fn parse(s: []const u8) ?Did { 9 + if (!mem.startsWith(u8, s, "did:")) return null; 10 + const rest = s[4..]; 11 + // must have method:identifier 12 + const colon = mem.indexOf(u8, rest, ":") orelse return null; 13 + if (colon == 0 or colon == rest.len - 1) return null; 14 + return .{ .raw = s }; 15 + } 16 + 17 + pub fn format(self: Did, comptime _: []const u8, _: std.fmt.FormatOptions, writer: anytype) !void { 18 + try writer.writeAll(self.raw); 19 + } 20 + 21 + pub fn eql(self: Did, other: Did) bool { 22 + return mem.eql(u8, self.raw, other.raw); 23 + } 24 + 25 + pub fn str(self: Did) []const u8 { 26 + return self.raw; 27 + } 28 + }; 29 + 30 + /// at-uri - at://did/collection/rkey 31 + pub const AtUri = struct { 32 + raw: []const u8, 33 + did_end: usize, 34 + collection_end: usize, 35 + 36 + pub fn parse(s: []const u8) ?AtUri { 37 + if (!mem.startsWith(u8, s, "at://")) return null; 38 + const rest = s[5..]; 39 + 40 + // find did end (first slash after did) 41 + const did_end = mem.indexOf(u8, rest, "/") orelse return null; 42 + if (did_end == 0) return null; 43 + 44 + // validate did portion 45 + const did_str = rest[0..did_end]; 46 + if (Did.parse(did_str) == null) return null; 47 + 48 + // find collection end (second slash) 49 + const after_did = rest[did_end + 1 ..]; 50 + const collection_end = mem.indexOf(u8, after_did, "/") orelse return null; 51 + if (collection_end == 0) return null; 52 + 53 + // rkey must exist 54 + const rkey_part = after_did[collection_end + 1 ..]; 55 + if (rkey_part.len == 0) return null; 56 + 57 + return .{ 58 + .raw = s, 59 + .did_end = 5 + did_end, 60 + .collection_end = 5 + did_end + 1 + collection_end, 61 + }; 62 + } 63 + 64 + pub fn build(allocator: mem.Allocator, d: Did, coll: []const u8, rk: []const u8) !AtUri { 65 + const raw = try std.fmt.allocPrint(allocator, "at://{s}/{s}/{s}", .{ d.raw, coll, rk }); 66 + return .{ 67 + .raw = raw, 68 + .did_end = 5 + d.raw.len, 69 + .collection_end = 5 + d.raw.len + 1 + coll.len, 70 + }; 71 + } 72 + 73 + pub fn did(self: AtUri) Did { 74 + return .{ .raw = self.raw[5..self.did_end] }; 75 + } 76 + 77 + pub fn collection(self: AtUri) []const u8 { 78 + return self.raw[self.did_end + 1 .. self.collection_end]; 79 + } 80 + 81 + pub fn rkey(self: AtUri) []const u8 { 82 + return self.raw[self.collection_end + 1 ..]; 83 + } 84 + 85 + pub fn format(self: AtUri, comptime _: []const u8, _: std.fmt.FormatOptions, writer: anytype) !void { 86 + try writer.writeAll(self.raw); 87 + } 88 + 89 + pub fn str(self: AtUri) []const u8 { 90 + return self.raw; 91 + } 92 + }; 93 + 94 + test "Did.parse valid" { 95 + const d = Did.parse("did:plc:abc123").?; 96 + try std.testing.expectEqualStrings("did:plc:abc123", d.raw); 97 + } 98 + 99 + test "Did.parse invalid" { 100 + try std.testing.expect(Did.parse("notadid") == null); 101 + try std.testing.expect(Did.parse("did:") == null); 102 + try std.testing.expect(Did.parse("did:plc") == null); 103 + try std.testing.expect(Did.parse("did::abc") == null); 104 + } 105 + 106 + test "AtUri.parse valid" { 107 + const u = AtUri.parse("at://did:plc:abc/app.bsky.post/123").?; 108 + try std.testing.expectEqualStrings("did:plc:abc", u.did().raw); 109 + try std.testing.expectEqualStrings("app.bsky.post", u.collection()); 110 + try std.testing.expectEqualStrings("123", u.rkey()); 111 + } 112 + 113 + test "AtUri.parse invalid" { 114 + try std.testing.expect(AtUri.parse("https://example.com") == null); 115 + try std.testing.expect(AtUri.parse("at://") == null); 116 + try std.testing.expect(AtUri.parse("at://did:plc:abc") == null); 117 + try std.testing.expect(AtUri.parse("at://did:plc:abc/collection") == null); 118 + try std.testing.expect(AtUri.parse("at://did:plc:abc/collection/") == null); 119 + } 120 + 121 + test "AtUri.build" { 122 + const d = Did.parse("did:plc:xyz").?; 123 + const u = try AtUri.build(std.testing.allocator, d, "pub.leaflet.document", "abc"); 124 + defer std.testing.allocator.free(u.raw); 125 + try std.testing.expectEqualStrings("at://did:plc:xyz/pub.leaflet.document/abc", u.raw); 126 + try std.testing.expectEqualStrings("did:plc:xyz", u.did().raw); 127 + try std.testing.expectEqualStrings("pub.leaflet.document", u.collection()); 128 + try std.testing.expectEqualStrings("abc", u.rkey()); 129 + }

Configure Feed

Configure Feed