Something went wrong. Try again.
GET /xrpc/tech.waow.typeahead.searchActors typeahead.waow.tech
Something went wrong. Try again.
Zig
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568//! Read-only Turso HTTP API client for sync//! Uses hrana v2 pipeline protocol
const std = @import("std");const io = std.Options.debug_io;const http = std.http;const json = std.json;const mem = std.mem;const Allocator = mem.Allocator;
const log = std.log.scoped(.turso);
fn cGetenv(name: [*:0]const u8) ?[]const u8 { if (std.c.getenv(name)) |p| return mem.span(p); return null;}
const TursoClient = @This();
const Value = struct { type: []const u8 = "text", value: []const u8 };const Stmt = struct { sql: []const u8, args: ?[]const Value = null };const ExecuteReq = struct { type: []const u8 = "execute", stmt: Stmt };const CloseReq = struct { type: []const u8 = "close" };
allocator: Allocator,url: []const u8, // host only (no protocol prefix)token: []const u8,http_client: http.Client,mutex: std.Io.Mutex = std.Io.Mutex.init,
pub fn init(allocator: Allocator) !TursoClient { const url = cGetenv("TURSO_URL") orelse { log.err("TURSO_URL not set", .{}); return error.MissingEnv; }; const token = cGetenv("TURSO_AUTH_TOKEN") orelse { log.err("TURSO_AUTH_TOKEN not set", .{}); return error.MissingEnv; };
const libsql_prefix = "libsql://"; const host = if (mem.startsWith(u8, url, libsql_prefix)) url[libsql_prefix.len..] else url;
log.info("turso client → {s}", .{host});
return .{ .allocator = allocator, .url = host, .token = token, .http_client = .{ .allocator = allocator, .io = io }, };}
pub fn deinit(self: *TursoClient) void { self.http_client.deinit();}
pub const Row = struct { columns: []const json.Value,
pub fn text(self: Row, index: usize) []const u8 { if (index >= self.columns.len) return ""; return extractText(self.columns[index]); }
pub fn int(self: Row, index: usize) i64 { if (index >= self.columns.len) return 0; return extractInt(self.columns[index]); }
pub fn float(self: Row, index: usize) f64 { if (index >= self.columns.len) return 0; return extractFloat(self.columns[index]); }};
pub const Result = struct { allocator: Allocator, parsed: ?json.Parsed(json.Value), rows: []const Row,
pub fn deinit(self: *Result) void { self.allocator.free(self.rows); if (self.parsed) |*p| p.deinit(); }};
pub fn query(self: *TursoClient, sql: []const u8, args: []const []const u8) !Result { const response = try self.executeRaw(sql, args); defer self.allocator.free(response); return parseResult(self.allocator, response);}
/// Typed bind value for the runtime-typed exec/query path. Hrana wants/// per-arg type tags; emitting the right one matters for INTEGER columns/// where text-encoded numbers would be coerced unpredictably.////// `null` is included for future-proofing — zug binds only `text` and/// `int` today.pub const RuntimeValue = union(enum) { text: []const u8, int: i64, null,};
/// Execute a statement that returns no rows (INSERT/UPDATE/CREATE/etc.)/// using runtime-typed bind values. Required for callers that can't/// express their args as `[]const []const u8` (e.g. zug binds an i64 for/// the dirty flag).pub fn execRuntime(self: *TursoClient, sql: []const u8, args: []const RuntimeValue) !void { const body = try self.buildTypedRequestBody(sql, args); defer self.allocator.free(body); const response = try self.doRequest(body); self.allocator.free(response);}
/// Same as `query` but takes runtime-typed bind values.pub fn queryRuntime(self: *TursoClient, sql: []const u8, args: []const RuntimeValue) !Result { const body = try self.buildTypedRequestBody(sql, args); defer self.allocator.free(body); const response = try self.doRequest(body); defer self.allocator.free(response); return parseResult(self.allocator, response);}
fn executeRaw(self: *TursoClient, sql: []const u8, args: []const []const u8) ![]const u8 { const body = try self.buildRequestBody(sql, args); defer self.allocator.free(body); return self.doRequest(body);}
/// Read deadline for one turso round trip. 2026-08-04 kernel-stack capture/// proved a fetch can block in tcp_recvmsg indefinitely when the flow stalls/// mid-response (stdlib http has no timeouts); the sync loop then wedges for/// a full watchdog cycle. Generous vs the observed p99 (~15s cold) so it only/// fires on genuinely dead flows.////// The concurrent-task + deadline + `Future.cancel` shape below is worth/// copying, not just reading: any zig service using stdlib http against a/// remote has this exposure. pub-search's `backend/src/db/Client.zig` and its/// builder still do (its 2026-08-03 `verify_remote` hang was this class,/// bounded today only by watchdogs). If a third consumer shows up, this/// belongs in a shared zat helper.const FETCH_DEADLINE_MS: u64 = 60_000;
const FetchOut = struct { done: std.atomic.Value(bool) = .init(false), result: anyerror![]const u8 = error.HttpError,};
fn fetchTask(self: *TursoClient, url: []const u8, auth: []const u8, payload: []const u8, out: *FetchOut) void { out.result = self.rawFetch(url, auth, payload); out.done.store(true, .release);}
/// POST a fully-built hrana v2 pipeline JSON body. Shared between the/// string-only `executeRaw` and the typed `execRuntime`/`queryRuntime`/// paths so the HTTP / mutex / error handling stays one place.////// The fetch runs in a concurrent task so this caller can enforce/// FETCH_DEADLINE_MS: on expiry, Future.cancel interrupts the task at its/// next cancelation point — including a recv blocked on a stalled flow —/// and the caller returns error.FetchTimeout (retryable; the next attempt/// opens a fresh connection since keep_alive=false).fn doRequest(self: *TursoClient, body: []const u8) ![]const u8 { self.mutex.lockUncancelable(io); defer self.mutex.unlock(io);
var url_buf: [512]u8 = undefined; const url = std.fmt.bufPrint(&url_buf, "https://{s}/v2/pipeline", .{self.url}) catch return error.UrlTooLong;
var auth_buf: [512]u8 = undefined; const auth = std.fmt.bufPrint(&auth_buf, "Bearer {s}", .{self.token}) catch return error.AuthTooLong;
var out: FetchOut = .{}; var future = std.Io.concurrent(io, fetchTask, .{ self, url, auth, body, &out }) catch |err| { // pool exhaustion is OUR resource problem — fail the query (callers // retry with backoff) rather than quietly degrading to an undeadlined // fetch that can block forever, which is the exact hang this deadline // exists to prevent. log.err("turso fetch: no concurrency available ({s}); failing for retry", .{@errorName(err)}); return error.ConcurrencyUnavailable; };
var waited_ms: u64 = 0; while (!out.done.load(.acquire) and waited_ms < FETCH_DEADLINE_MS) { io.sleep(.{ .nanoseconds = 100 * std.time.ns_per_ms }, .real) catch |err| { future.cancel(io); if (out.result) |bytes| self.allocator.free(bytes) else |_| {} return err; }; waited_ms += 100; }
if (!out.done.load(.acquire)) { log.err("turso fetch exceeded {d}ms deadline; canceling stalled request", .{FETCH_DEADLINE_MS}); future.cancel(io); if (!out.done.load(.acquire)) return error.FetchTimeout; return out.result catch error.FetchTimeout; }
future.await(io); return out.result;}
fn rawFetch(self: *TursoClient, url: []const u8, auth: []const u8, payload: []const u8) ![]const u8 { var response_body: std.Io.Writer.Allocating = .init(self.allocator); errdefer response_body.deinit();
const res = self.http_client.fetch(.{ .location = .{ .url = url }, .method = .POST, .headers = .{ .content_type = .{ .override = "application/json" }, .authorization = .{ .override = auth }, }, .payload = payload, .response_writer = &response_body.writer, .keep_alive = false, }) catch |err| { log.err("http failed: {s}", .{@errorName(err)}); return error.HttpError; };
if (res.status != .ok) { const resp_text = response_body.toOwnedSlice() catch ""; defer if (resp_text.len > 0) self.allocator.free(resp_text); const preview = if (resp_text.len > 200) resp_text[0..200] else resp_text; log.err("turso error: {} | {s}", .{ res.status, preview }); return error.TursoError; }
return try response_body.toOwnedSlice();}
fn buildRequestBody(self: *TursoClient, sql: []const u8, args: []const []const u8) ![]const u8 { var body: std.Io.Writer.Allocating = .init(self.allocator); errdefer body.deinit(); var jw: json.Stringify = .{ .writer = &body.writer, .options = .{ .emit_null_optional_fields = false } };
var values: []const Value = &.{}; defer if (values.len > 0) self.allocator.free(values);
if (args.len > 0) { const v = try self.allocator.alloc(Value, args.len); for (args, 0..) |arg, i| { v[i] = .{ .value = arg }; } values = v; }
try jw.beginObject(); try jw.objectField("requests"); try jw.beginArray(); try jw.write(ExecuteReq{ .stmt = .{ .sql = sql, .args = if (values.len > 0) values else null }, }); try jw.write(CloseReq{}); try jw.endArray(); try jw.endObject();
return try body.toOwnedSlice();}
/// Hrana v2 pipeline body for a single execute with per-arg typed bind/// values. Written manually (rather than via a serde-style struct) because/// the hrana value shape varies per type — `text` and `integer` carry a/// `value`, `null` doesn't — and integers serialize as a JSON string/// (per hrana, to preserve i64 precision through JSON).fn buildTypedRequestBody(self: *TursoClient, sql: []const u8, args: []const RuntimeValue) ![]const u8 { var body: std.Io.Writer.Allocating = .init(self.allocator); errdefer body.deinit(); var jw: json.Stringify = .{ .writer = &body.writer, .options = .{ .emit_null_optional_fields = false } };
try jw.beginObject(); try jw.objectField("requests"); try jw.beginArray();
// execute request try jw.beginObject(); try jw.objectField("type"); try jw.write("execute"); try jw.objectField("stmt"); try jw.beginObject(); try jw.objectField("sql"); try jw.write(sql); if (args.len > 0) { try jw.objectField("args"); try jw.beginArray(); var int_buf: [24]u8 = undefined; // i64 max width is 20 chars (incl. sign), 24 is plenty for (args) |arg| { try jw.beginObject(); switch (arg) { .text => |s| { try jw.objectField("type"); try jw.write("text"); try jw.objectField("value"); try jw.write(s); }, .int => |n| { try jw.objectField("type"); try jw.write("integer"); try jw.objectField("value"); const s = std.fmt.bufPrint(&int_buf, "{d}", .{n}) catch return error.IntFormat; try jw.write(s); }, .null => { try jw.objectField("type"); try jw.write("null"); }, } try jw.endObject(); } try jw.endArray(); } try jw.endObject(); try jw.endObject();
// close request try jw.write(CloseReq{});
try jw.endArray(); try jw.endObject();
return try body.toOwnedSlice();}
fn parseResult(allocator: Allocator, response: []const u8) !Result { const parsed = json.parseFromSlice(json.Value, allocator, response, .{}) catch { return .{ .allocator = allocator, .parsed = null, .rows = &.{} }; };
const json_rows = getRowsFromParsed(parsed.value) orelse { return .{ .allocator = allocator, .parsed = parsed, .rows = &.{} }; };
var rows: std.ArrayList(Row) = .empty; errdefer rows.deinit(allocator);
for (json_rows.items) |item| { if (item == .array) { try rows.append(allocator, .{ .columns = item.array.items }); } }
return .{ .allocator = allocator, .parsed = parsed, .rows = try rows.toOwnedSlice(allocator), };}
fn getRowsFromResult(item: json.Value) ?json.Array { if (item != .object) return null; const resp = item.object.get("response") orelse return null; if (resp != .object) return null; const res = resp.object.get("result") orelse return null; if (res != .object) return null; const rows = res.object.get("rows") orelse return null; if (rows != .array) return null; return rows.array;}
fn getRowsFromParsed(value: json.Value) ?json.Array { const results = value.object.get("results") orelse return null; if (results != .array or results.array.items.len == 0) return null; return getRowsFromResult(results.array.items[0]);}
fn extractText(val: json.Value) []const u8 { return switch (val) { .string => |s| s, .object => |obj| { const v = obj.get("value") orelse return ""; return if (v == .string) v.string else ""; }, else => "", };}
// Recursive, exhaustive-by-construction. Hrana wraps INTEGER columns as// `{"type":"integer","value":"<digits>"}` (value is a quoted string to// preserve i64 precision). The original implementation matched that exact// shape but used a non-recursive inner switch with an `else => 0`, which// silently dropped values that arrived in any other shape (e.g. number// type without a value field, or future server-side encoding changes).// When that happened to the rowid/updated_at trailing column, the keyset// cursor in sync.zig never advanced and the loop re-fetched the same// 2000-row batch forever, hammering Turso. Recursing on the inner Value// makes this resilient to any of hrana's documented value shapes plus// any reasonable bare JSON forms.fn extractInt(val: json.Value) i64 { return switch (val) { .integer => |i| i, .float => |f| @as(i64, @intFromFloat(f)), .number_string => |s| std.fmt.parseInt(i64, s, 10) catch 0, .string => |s| std.fmt.parseInt(i64, s, 10) catch 0, .object => |obj| blk: { const v = obj.get("value") orelse { // log once with the keys we saw so we can diagnose future drift logExtractIntMiss(obj); break :blk 0; }; break :blk extractInt(v); }, .bool => |b| if (b) 1 else 0, else => 0, };}
// Same shape-tolerance as extractInt — hrana wraps REAL columns as// `{"type":"float","value":<number>}` (bare JSON number, no string quoting).// Falls through to integer/string forms so we don't silently lose data// if the server-side encoding ever shifts.fn extractFloat(val: json.Value) f64 { return switch (val) { .float => |f| f, .integer => |i| @floatFromInt(i), .number_string => |s| std.fmt.parseFloat(f64, s) catch 0, .string => |s| std.fmt.parseFloat(f64, s) catch 0, .object => |obj| blk: { const v = obj.get("value") orelse break :blk 0; break :blk extractFloat(v); }, .bool => |b| if (b) 1 else 0, else => 0, };}
// ─── database export (GET /info + GET /export/{generation}) ──────────────//// The fast bulk-read path. `turso db export` does exactly this under the// hood: GET /info for the current generation, then GET /export/{gen} which// streams a standalone, internally-consistent SQLite checkpoint file.//// We deliberately SKIP the CLI's /sync WAL-frame reassembly. The /export// body is a complete committed database (a real checkpoint boundary, never// torn); the only thing the WAL frames add is the handful of writes// committed after that checkpoint. Our index tolerates minutes-staleness// (rebuild cadence is hours; the overlay layer covers recent changes), so// the main DB file alone is the right snapshot. source_watermark comes// from MAX(updated_at) in the exported file — the watermark of THAT// snapshot, exactly.//// Measured ~60× faster wire throughput than /v3/cursor for bulk reads// (export: 234s for 5.9M rows; cursor: ~4h).
/// GET /info → current_generation. Small JSON response, buffered.fn fetchCurrentGeneration(self: *TursoClient) !i64 { self.mutex.lockUncancelable(io); defer self.mutex.unlock(io);
var url_buf: [512]u8 = undefined; const url = std.fmt.bufPrint(&url_buf, "https://{s}/info", .{self.url}) catch return error.UrlTooLong; var auth_buf: [512]u8 = undefined; const auth = std.fmt.bufPrint(&auth_buf, "Bearer {s}", .{self.token}) catch return error.AuthTooLong;
var response_body: std.Io.Writer.Allocating = .init(self.allocator); defer response_body.deinit();
const res = self.http_client.fetch(.{ .location = .{ .url = url }, .method = .GET, .headers = .{ .authorization = .{ .override = auth } }, .response_writer = &response_body.writer, .keep_alive = false, }) catch |err| { log.err("/info http failed: {s}", .{@errorName(err)}); return error.HttpError; }; if (res.status != .ok) { log.err("/info status: {}", .{res.status}); return error.TursoError; }
const parsed = json.parseFromSlice(json.Value, self.allocator, response_body.written(), .{}) catch return error.TursoError; defer parsed.deinit(); if (parsed.value != .object) return error.TursoError; const gen_v = parsed.value.object.get("current_generation") orelse return error.TursoError; return switch (gen_v) { .integer => |n| n, else => error.TursoError, };}
/// Export the database to a local SQLite file. Two GETs: /info for the/// generation, then /export/{gen} streamed straight to disk. Writes to/// `<output_path>.part` first, then renames into place so a partial/// download is never mistaken for a complete file. `output_path` must/// be absolute.pub fn exportToFile(self: *TursoClient, output_path: []const u8) !void { const gen = try self.fetchCurrentGeneration(); log.info("turso export: current_generation={d}", .{gen});
var part_buf: [320]u8 = undefined; const part_path = std.fmt.bufPrint(&part_buf, "{s}.part", .{output_path}) catch return error.PathTooLong;
self.mutex.lockUncancelable(io); defer self.mutex.unlock(io);
var url_buf: [512]u8 = undefined; const url = std.fmt.bufPrint(&url_buf, "https://{s}/export/{d}", .{ self.url, gen }) catch return error.UrlTooLong; var auth_buf: [512]u8 = undefined; const auth = std.fmt.bufPrint(&auth_buf, "Bearer {s}", .{self.token}) catch return error.AuthTooLong;
const file = std.Io.Dir.createFileAbsolute(io, part_path, .{ .truncate = true }) catch |err| { log.err("export create {s} failed: {s}", .{ part_path, @errorName(err) }); return error.ExportFileCreateFailed; }; // 64KB write buffer so the file writer batches fd writes; the HTTP // body streams through here, never buffered whole in memory. var write_buf: [64 * 1024]u8 = undefined; var fw = std.Io.File.Writer.init(file, io, &write_buf);
const res = self.http_client.fetch(.{ .location = .{ .url = url }, .method = .GET, .headers = .{ .authorization = .{ .override = auth } }, .response_writer = &fw.interface, .keep_alive = false, }) catch |err| { log.err("/export http failed: {s}", .{@errorName(err)}); file.close(io); return error.HttpError; };
fw.interface.flush() catch |err| { log.err("export flush failed: {s}", .{@errorName(err)}); file.close(io); return error.ExportFlushFailed; }; file.close(io);
if (res.status != .ok) { log.err("/export status: {}", .{res.status}); return error.TursoError; }
std.Io.Dir.renameAbsolute(part_path, output_path, io) catch |err| { log.err("export rename {s} → {s} failed: {s}", .{ part_path, output_path, @errorName(err) }); return error.ExportRenameFailed; };}
var logged_miss: bool = false;fn logExtractIntMiss(obj: anytype) void { if (logged_miss) return; logged_miss = true; var it = obj.iterator(); var buf: [256]u8 = undefined; const w = std.fmt.bufPrint(&buf, "extractInt missed: keys=", .{}) catch return; var used = w.len; while (it.next()) |entry| { const k = entry.key_ptr.*; const remain = buf.len - used; if (remain < k.len + 2) break; const part = std.fmt.bufPrint(buf[used..], "{s},", .{k}) catch break; used += part.len; } log.warn("{s}", .{buf[0..used]});}