//! page through every row of a namespace in id order, the way the export
//! guide describes: rank by id ascending, then filter `id > last_id`.
//!
//! ```zig
//! var pages = ns.exporter(alloc, .{ .attributes = .only(&.{"uri"}) });
//! while (try pages.next()) |page| {
//!     defer page.deinit();
//!     var rows = page.rows();
//!     while (rows.next()) |row| ...
//! }
//! ```
//!
//! rows written while the export runs may or may not appear.

const Export = @This();

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

pub const Options = struct {
    /// rows per request; the api caps `top_k` at 10,000.
    page_size: u32 = 10_000,
    attributes: query.Attributes = .all,
    filters: ?json.Value = null,
    diagnostics: ?*@import("Client.zig").Diagnostics = null,
};

namespace: Namespace,
alloc: Allocator,
options: Options,
cursor: Cursor = .none,
done: bool = false,

const Cursor = union(enum) {
    none,
    string: struct { buf: [64]u8, len: usize },
    integer: u64,

    fn id(self: *const Cursor) ?query.Id {
        return switch (self.*) {
            .none => null,
            .string => |*s| .{ .string = s.buf[0..s.len] },
            .integer => |i| .{ .integer = i },
        };
    }

    fn from(value: query.Id) Cursor {
        return switch (value) {
            .integer => |i| .{ .integer = i },
            .string => |s| blk: {
                var c: Cursor = .{ .string = .{ .buf = undefined, .len = @min(s.len, 64) } };
                @memcpy(c.string.buf[0..c.string.len], s[0..c.string.len]);
                break :blk c;
            },
        };
    }
};

pub fn init(namespace: Namespace, alloc: Allocator, options: Options) Export {
    return .{ .namespace = namespace, .alloc = alloc, .options = options };
}

const IdFilter = struct { []const u8, []const u8, query.Id };

const Request = struct {
    rank_by: struct { []const u8, []const u8 } = query.rank.asc("id"),
    top_k: u32,
    include_attributes: query.Attributes,
    filters: ?Filters = null,

    const Filters = union(enum) {
        cursor: IdFilter,
        both: struct { []const u8, struct { IdFilter, json.Value } },
        user: json.Value,

        pub fn jsonStringify(self: Filters, jw: *json.Stringify) !void {
            switch (self) {
                .cursor => |f| try jw.write(f),
                .both => |f| try jw.write(f),
                .user => |v| try jw.write(v),
            }
        }
    };
};

fn buildRequest(self: *const Export) Request {
    const cursor_filter: ?IdFilter = if (self.cursor.id()) |last| query.filter.gt("id", last) else null;
    const filters: ?Request.Filters = if (cursor_filter) |c|
        (if (self.options.filters) |user| .{ .both = .{ "And", .{ c, user } } } else .{ .cursor = c })
    else if (self.options.filters) |user| .{ .user = user } else null;
    return .{
        .top_k = self.options.page_size,
        .include_attributes = self.options.attributes,
        .filters = filters,
    };
}

/// the next non-empty page, or null when the namespace is exhausted.
pub fn next(self: *Export) !?query.Result {
    if (self.done) return null;
    const result = try self.namespace.query(self.alloc, self.buildRequest(), .{ .diagnostics = self.options.diagnostics });
    errdefer result.deinit();

    const rows = result.rows();
    if (rows.len() < self.options.page_size) self.done = true;
    if (rows.len() == 0) {
        result.deinit();
        return null;
    }
    const last = rows.at(rows.len() - 1).?.id() orelse return error.UnexpectedResponse;
    self.cursor = .from(last);
    return result;
}

fn expectRequest(expected: []const u8, req: Request) !void {
    const body = try json.Stringify.valueAlloc(std.testing.allocator, req, .{ .emit_null_optional_fields = false });
    defer std.testing.allocator.free(body);
    try std.testing.expectEqualStrings(expected, body);
}

test "first page has no cursor filter" {
    var transport: zttp.Client = .init(std.testing.io, std.testing.allocator);
    defer transport.deinit();
    const client: @import("Client.zig") = .init(&transport, .{ .api_key = "k" });
    var ex: Export = .init(try client.namespace("ns"), std.testing.allocator, .{ .page_size = 500 });
    try expectRequest("{\"rank_by\":[\"id\",\"asc\"],\"top_k\":500,\"include_attributes\":true}", ex.buildRequest());

    ex.cursor = .from(.{ .string = "abc" });
    try expectRequest(
        "{\"rank_by\":[\"id\",\"asc\"],\"top_k\":500,\"include_attributes\":true,\"filters\":[\"id\",\"Gt\",\"abc\"]}",
        ex.buildRequest(),
    );

    ex.cursor = .from(.{ .integer = 9 });
    ex.options.attributes = .only(&.{"uri"});
    try expectRequest(
        "{\"rank_by\":[\"id\",\"asc\"],\"top_k\":500,\"include_attributes\":[\"uri\"],\"filters\":[\"id\",\"Gt\",9]}",
        ex.buildRequest(),
    );
}

test "user filters are And-ed with the cursor" {
    var transport: zttp.Client = .init(std.testing.io, std.testing.allocator);
    defer transport.deinit();
    const client: @import("Client.zig") = .init(&transport, .{ .api_key = "k" });
    const user = try json.parseFromSlice(json.Value, std.testing.allocator, "[\"kind\",\"Eq\",\"post\"]", .{});
    defer user.deinit();
    var ex: Export = .init(try client.namespace("ns"), std.testing.allocator, .{ .page_size = 10, .filters = user.value });
    try expectRequest(
        "{\"rank_by\":[\"id\",\"asc\"],\"top_k\":10,\"include_attributes\":true,\"filters\":[\"kind\",\"Eq\",\"post\"]}",
        ex.buildRequest(),
    );
    ex.cursor = .from(.{ .integer = 1 });
    try expectRequest(
        "{\"rank_by\":[\"id\",\"asc\"],\"top_k\":10,\"include_attributes\":true,\"filters\":[\"And\",[[\"id\",\"Gt\",1],[\"kind\",\"Eq\",\"post\"]]]}",
        ex.buildRequest(),
    );
}