//! what every namespace shares: host resolution, auth, status-level retries,
//! error mapping. one `Client` per process, borrowing a `zttp.Client` so
//! connections are pooled across namespaces and threads and every phase of
//! a request is bounded.

const Client = @This();

const std = @import("std");
const json = std.json;
const Allocator = std.mem.Allocator;
const Io = std.Io;
const zttp = @import("zttp");
const Namespace = @import("Namespace.zig");

const log = std.log.scoped(.tpuf);

pub const version = "0.1.4";

transport: *zttp.Client,
api_key: []const u8,
host: Host,
retry: Retry,

pub const Host = union(enum) {
    /// documented form: `https://{region}.turbopuffer.com`
    region: []const u8,
    /// any base url, no trailing slash
    base_url: []const u8,

    /// the region-less host the production adopters were already using.
    pub const default: Host = .{ .base_url = "https://api.turbopuffer.com" };
};

/// 429 and 5xx responses are retried with exponential backoff and equal
/// jitter (half the step fixed, half random, drawn from the transport's
/// `Io`), or after `Retry-After` when the api sends one. transport failures
/// are not retried here: zttp already replays the ones that provably never
/// took effect, and names the rest so the caller can decide.
///
/// upsert, patch and delete are idempotent, so a retried write is safe.
/// a write carrying a `*_condition` is sent once: `Namespace.write` turns
/// retries off for it at compile time.
pub const Retry = struct {
    max_retries: u8 = 2,
    base_delay_ms: u32 = 250,
    max_delay_ms: u32 = 2_000,
    /// a `Retry-After` beyond this is clamped rather than waited out.
    max_retry_after_ms: u32 = 30_000,

    pub const none: Retry = .{ .max_retries = 0 };

    /// the capped exponential step for `attempt`, before jitter.
    pub fn stepMs(self: Retry, attempt: u8) u32 {
        const shift: u5 = @intCast(@min(attempt, 20));
        const scaled: u64 = @as(u64, self.base_delay_ms) << shift;
        return @intCast(@min(scaled, self.max_delay_ms));
    }

    pub fn delayMs(self: Retry, random: std.Random, attempt: u8, retry_after_s: ?u64) u32 {
        if (retry_after_s) |s| {
            const ms = std.math.mul(u64, s, 1000) catch return self.max_retry_after_ms;
            return @intCast(@min(ms, self.max_retry_after_ms));
        }
        const step = self.stepMs(attempt);
        const half = step / 2;
        return half + random.uintAtMost(u32, step - half);
    }
};

pub const Options = struct {
    api_key: []const u8,
    host: Host = .default,
    retry: Retry = .{},
};

/// errors this client adds over zttp's. a non-2xx status maps to one of
/// these; `Diagnostics` carries the api's message.
pub const Error = error{
    BadRequest,
    Unauthorized,
    Forbidden,
    NotFound,
    Conflict,
    RateLimited,
    ServerError,
    UnexpectedStatus,
    UrlTooLong,
    ApiKeyTooLong,
};

/// optional sink for the api's error body. pass one when the message matters
/// (schema mismatches, conditional-write failures); otherwise the message is
/// logged under the `tpuf` scope and the status is mapped to an error.
pub const Diagnostics = struct {
    status: u16 = 0,
    buf: [512]u8 = undefined,
    len: usize = 0,

    pub fn message(self: *const Diagnostics) []const u8 {
        return self.buf[0..self.len];
    }

    pub fn set(self: *Diagnostics, status: u16, body: []const u8) void {
        self.status = status;
        self.len = (errorMessage(&self.buf, body) orelse copyTruncated(&self.buf, body)).len;
    }
};

pub fn init(transport: *zttp.Client, options: Options) Client {
    return .{
        .transport = transport,
        .api_key = options.api_key,
        .host = options.host,
        .retry = options.retry,
    };
}

pub fn namespace(self: *const Client, name: []const u8) error{InvalidNamespace}!Namespace {
    return Namespace.init(self, name);
}

pub const ListOptions = struct {
    prefix: ?[]const u8 = null,
    page_size: ?u32 = null,
    cursor: ?[]const u8 = null,
    diagnostics: ?*Diagnostics = null,
};

pub const NamespaceList = struct {
    namespaces: []const Entry = &.{},
    next_cursor: ?[]const u8 = null,

    pub const Entry = struct { id: []const u8 };
};

/// one page of namespaces. pass `next_cursor` back as `cursor` for the next.
pub fn listNamespaces(self: *const Client, alloc: Allocator, options: ListOptions) !json.Parsed(NamespaceList) {
    var path_buf: [1024]u8 = undefined;
    var w: std.Io.Writer = .fixed(&path_buf);
    w.writeAll("/v1/namespaces") catch return error.UrlTooLong;
    var sep: u8 = '?';
    if (options.prefix) |prefix| {
        w.print("{c}prefix=", .{sep}) catch return error.UrlTooLong;
        percentEncode(&w, prefix) catch return error.UrlTooLong;
        sep = '&';
    }
    if (options.page_size) |n| {
        w.print("{c}page_size={d}", .{ sep, n }) catch return error.UrlTooLong;
        sep = '&';
    }
    if (options.cursor) |cursor| {
        w.print("{c}cursor=", .{sep}) catch return error.UrlTooLong;
        percentEncode(&w, cursor) catch return error.UrlTooLong;
    }

    const body = try self.request(.GET, w.buffered(), null, .{ .diagnostics = options.diagnostics });
    defer self.free(body);
    return json.parseFromSlice(NamespaceList, alloc, body, .{ .ignore_unknown_fields = true }) catch
        return error.UnexpectedResponse;
}

pub const RequestOptions = struct {
    diagnostics: ?*Diagnostics = null,
    /// off for requests whose replay could apply twice: conditional writes.
    retry: bool = true,
};

/// send one request and return the response body, owned by the transport's
/// allocator: release it with `free`. `path` is absolute
/// (`/v2/namespaces/foo/query`). retries per `self.retry`.
///
/// `accept_encoding: identity` follows the api docs: "for most workloads,
/// disabling compression offers the best performance. The official client
/// libraries disable request and response compression by default."
/// (https://turbopuffer.com/docs/api-overview#compression)
pub fn request(
    self: *const Client,
    method: std.http.Method,
    path: []const u8,
    payload: ?[]const u8,
    options: RequestOptions,
) ![]u8 {
    var url_buf: [1536]u8 = undefined;
    const url = try buildUrl(&url_buf, self.host, path);

    var auth_buf: [256]u8 = undefined;
    const auth = std.fmt.bufPrint(&auth_buf, "Bearer {s}", .{self.api_key}) catch
        return error.ApiKeyTooLong;

    const max_retries: u8 = if (options.retry) self.retry.max_retries else 0;
    const random_source: std.Random.IoSource = .{ .io = self.transport.io };

    var attempt: u8 = 0;
    while (true) : (attempt += 1) {
        var res = try self.transport.fetch(.{
            .url = url,
            .method = method,
            .payload = payload,
            .authorization = auth,
            .accept_encoding = "identity",
            .content_type = if (payload != null) "application/json" else null,
        });

        const status: u16 = @intFromEnum(res.status);
        if (status >= 200 and status < 300) {
            const body = res.body;
            res.body = &.{};
            res.deinit(self.transport.allocator);
            return body;
        }
        defer res.deinit(self.transport.allocator);

        if (isRetryable(status) and attempt < max_retries) {
            const now = Io.Clock.real.now(self.transport.io).toSeconds();
            const ms = self.retry.delayMs(random_source.interface(), attempt, res.rate_limit.delaySeconds(now));
            log.warn("{t} {s} -> {d}, retrying in {d}ms", .{ method, path, status, ms });
            try self.transport.io.sleep(Io.Duration.fromMilliseconds(ms), .awake);
            continue;
        }

        if (options.diagnostics) |d| d.set(status, res.body);
        var msg_buf: [200]u8 = undefined;
        log.err("{t} {s} -> {d}: {s}", .{ method, path, status, errorMessage(&msg_buf, res.body) orelse copyTruncated(&msg_buf, res.body) });
        return statusError(status);
    }
}

pub fn free(self: *const Client, body: []u8) void {
    self.transport.allocator.free(body);
}

pub fn buildUrl(buf: []u8, host: Host, path: []const u8) error{UrlTooLong}![]const u8 {
    return switch (host) {
        .region => |region| std.fmt.bufPrint(buf, "https://{s}.turbopuffer.com{s}", .{ region, path }),
        .base_url => |base| std.fmt.bufPrint(buf, "{s}{s}", .{ base, path }),
    } catch error.UrlTooLong;
}

pub fn isRetryable(status: u16) bool {
    return switch (status) {
        429, 500, 502, 503, 504 => true,
        else => false,
    };
}

pub fn statusError(status: u16) Error {
    return switch (status) {
        400 => error.BadRequest,
        401 => error.Unauthorized,
        403 => error.Forbidden,
        404 => error.NotFound,
        409 => error.Conflict,
        429 => error.RateLimited,
        500...599 => error.ServerError,
        else => error.UnexpectedStatus,
    };
}

const Envelope = struct { @"error": ?[]const u8 = null };

/// the api's error envelope is `{"status":"error","error":"..."}`. the
/// message is copied into `out`, truncated to fit.
pub fn errorMessage(out: []u8, body: []const u8) ?[]const u8 {
    var scratch: [2048]u8 = undefined;
    var fba: std.heap.FixedBufferAllocator = .init(&scratch);
    const envelope = json.parseFromSliceLeaky(Envelope, fba.allocator(), body, .{ .ignore_unknown_fields = true }) catch return null;
    return copyTruncated(out, envelope.@"error" orelse return null);
}

fn copyTruncated(out: []u8, text: []const u8) []const u8 {
    const n = @min(out.len, text.len);
    @memcpy(out[0..n], text[0..n]);
    return out[0..n];
}

/// query-string encoding: unreserved bytes pass through, everything else is %XX.
pub fn percentEncode(w: *std.Io.Writer, s: []const u8) !void {
    for (s) |c| {
        switch (c) {
            'A'...'Z', 'a'...'z', '0'...'9', '-', '_', '.', '~' => try w.writeByte(c),
            else => try w.print("%{X:0>2}", .{c}),
        }
    }
}

test "buildUrl for region and base_url hosts" {
    var buf: [256]u8 = undefined;
    try std.testing.expectEqualStrings(
        "https://gcp-us-central1.turbopuffer.com/v2/namespaces/ns/query",
        try buildUrl(&buf, .{ .region = "gcp-us-central1" }, "/v2/namespaces/ns/query"),
    );
    try std.testing.expectEqualStrings(
        "https://api.turbopuffer.com/v1/namespaces",
        try buildUrl(&buf, .default, "/v1/namespaces"),
    );
    var tiny: [8]u8 = undefined;
    try std.testing.expectError(error.UrlTooLong, buildUrl(&tiny, .default, "/v1/namespaces"));
}

test "status mapping and retryability" {
    try std.testing.expectEqual(error.BadRequest, statusError(400));
    try std.testing.expectEqual(error.Conflict, statusError(409));
    try std.testing.expectEqual(error.RateLimited, statusError(429));
    try std.testing.expectEqual(error.ServerError, statusError(503));
    try std.testing.expectEqual(error.UnexpectedStatus, statusError(302));
    try std.testing.expect(isRetryable(429));
    try std.testing.expect(isRetryable(502));
    try std.testing.expect(!isRetryable(400));
    try std.testing.expect(!isRetryable(409));
}

test "errorMessage reads the api envelope" {
    var out: [64]u8 = undefined;
    try std.testing.expectEqualStrings(
        "attribute \"status\" not found in schema",
        errorMessage(&out, "{\"status\":\"error\",\"error\":\"attribute \\\"status\\\" not found in schema\"}").?,
    );
    try std.testing.expectEqual(@as(?[]const u8, null), errorMessage(&out, "<html>bad gateway</html>"));
    try std.testing.expectEqual(@as(?[]const u8, null), errorMessage(&out, "{\"status\":\"ok\"}"));
    var small: [5]u8 = undefined;
    try std.testing.expectEqualStrings("attri", errorMessage(&small, "{\"error\":\"attribute\"}").?);
}

test "Diagnostics captures status and message" {
    var d: Diagnostics = .{};
    d.set(400, "{\"status\":\"error\",\"error\":\"duplicate id\"}");
    try std.testing.expectEqual(@as(u16, 400), d.status);
    try std.testing.expectEqualStrings("duplicate id", d.message());
    d.set(502, "upstream exploded");
    try std.testing.expectEqualStrings("upstream exploded", d.message());
}

test "Retry step doubles and caps" {
    const r: Retry = .{ .base_delay_ms = 100, .max_delay_ms = 350 };
    try std.testing.expectEqual(@as(u32, 100), r.stepMs(0));
    try std.testing.expectEqual(@as(u32, 200), r.stepMs(1));
    try std.testing.expectEqual(@as(u32, 350), r.stepMs(2));
    try std.testing.expectEqual(@as(u32, 350), r.stepMs(40));
}

test "Retry delay jitters within [step/2, step] and defers to Retry-After" {
    const r: Retry = .{ .base_delay_ms = 100, .max_delay_ms = 350, .max_retry_after_ms = 5_000 };
    var prng: std.Random.DefaultPrng = .init(42);
    const random = prng.random();
    var seen_low = false;
    var seen_high = false;
    for (0..200) |_| {
        const ms = r.delayMs(random, 1, null);
        try std.testing.expect(ms >= 100 and ms <= 200);
        if (ms < 130) seen_low = true;
        if (ms > 170) seen_high = true;
    }
    try std.testing.expect(seen_low and seen_high);
    try std.testing.expectEqual(@as(u32, 3_000), r.delayMs(random, 0, 3));
    try std.testing.expectEqual(@as(u32, 5_000), r.delayMs(random, 0, 3600));
    try std.testing.expectEqual(@as(u32, 5_000), r.delayMs(random, 0, std.math.maxInt(u64)));
}

test "NamespaceList parses the api's shape" {
    const parsed = try json.parseFromSlice(NamespaceList, std.testing.allocator,
        \\{"namespaces":[{"id":"bufos"},{"id":"bufos-v2"}],"next_cursor":null}
    , .{ .ignore_unknown_fields = true });
    defer parsed.deinit();
    try std.testing.expectEqual(@as(usize, 2), parsed.value.namespaces.len);
    try std.testing.expectEqualStrings("bufos-v2", parsed.value.namespaces[1].id);
    try std.testing.expectEqual(@as(?[]const u8, null), parsed.value.next_cursor);

    const paged = try json.parseFromSlice(NamespaceList, std.testing.allocator,
        \\{"namespaces":[{"id":"a"}],"next_cursor":"eyJ"}
    , .{ .ignore_unknown_fields = true });
    defer paged.deinit();
    try std.testing.expectEqualStrings("eyJ", paged.value.next_cursor.?);
}

test "percentEncode" {
    var buf: [64]u8 = undefined;
    var w: std.Io.Writer = .fixed(&buf);
    try percentEncode(&w, "user-a b/c~");
    try std.testing.expectEqualStrings("user-a%20b%2Fc~", w.buffered());
}