//! XRPC endpoint handlers for the AT Protocol sync API.
//!
//! implements com.atproto.sync.* lexicon endpoints:
//!   listRepos, getRepo, getRepoStatus, getLatestCommit, listReposByCollection,
//!   listHosts, getHostStatus, requestCrawl
//!
//! DB-accessing handlers use DbRequest + DbRequestQueue to route queries through
//! pool_io workers.

const std = @import("std");
const Io = std.Io;
const pg = @import("pg");
const h = @import("http.zig");
const router = @import("router.zig");
const event_log_mod = @import("../event_log.zig");
const collection_index_mod = @import("../collection_index/index.zig");
const slurper_mod = @import("../slurper.zig");

const Allocator = std.mem.Allocator;
const HttpContext = router.HttpContext;
const DbRequest = event_log_mod.DbRequest;
const DiskPersist = event_log_mod.DiskPersist;
const log = std.log.scoped(.relay);

// --- listRepos ---

const RepoListEncoder = struct {
    output: Io.Writer.Allocating,
    count: i64 = 0,
    last_uid: i64 = 0,

    fn init(allocator: Allocator) !RepoListEncoder {
        var self: RepoListEncoder = .{ .output = .init(allocator) };
        errdefer self.output.deinit();
        try self.output.writer.writeAll("{\"repos\":[");
        return self;
    }

    fn append(self: *RepoListEncoder, uid: i64, did: []const u8, head: []const u8, rev: []const u8, local_status: []const u8, upstream_status: []const u8) !void {
        if (head.len == 0 or rev.len == 0) return error.MissingRepoCommit;
        const w = &self.output.writer;
        if (self.count > 0) try w.writeByte(',');
        const local_ok = std.mem.eql(u8, local_status, "active");
        const active = local_ok and std.mem.eql(u8, upstream_status, "active");
        const status = if (!local_ok) local_status else upstream_status;
        try w.writeAll("{\"did\":\"");
        try w.writeAll(did);
        try w.writeAll("\",\"head\":\"");
        try w.writeAll(head);
        try w.writeAll("\",\"rev\":\"");
        try w.writeAll(rev);
        try w.writeAll("\"");
        if (active) {
            try w.writeAll(",\"active\":true");
        } else {
            try w.writeAll(",\"active\":false,\"status\":\"");
            try w.writeAll(status);
            try w.writeByte('"');
        }
        try w.writeByte('}');
        self.last_uid = uid;
        self.count += 1;
    }

    fn finish(self: *RepoListEncoder, limit: i64) ![]u8 {
        const w = &self.output.writer;
        try w.writeByte(']');
        if (self.count >= limit) try w.print(",\"cursor\":\"{d}\"", .{self.last_uid});
        try w.writeByte('}');
        return self.output.toOwnedSlice();
    }
};

const ListReposReq = struct {
    base: DbRequest = .{ .callback = &execute },
    cursor_val: i64,
    limit: i64,
    body: ?[]u8 = null,
    db_err: bool = false,

    fn execute(b: *DbRequest, dp: *DiskPersist) void {
        const self: *@This() = @fieldParentPtr("base", b);
        self.body = self.render(dp) catch {
            self.db_err = true;
            return;
        };
    }

    fn render(self: *ListReposReq, dp: *DiskPersist) ![]u8 {
        var output = try RepoListEncoder.init(std.heap.c_allocator);
        defer output.output.deinit();
        var result = try dp.db.query(
            \\SELECT a.uid, a.did, a.status, a.upstream_status, COALESCE(r.rev, ''), COALESCE(r.commit_cid, '')
            \\FROM account a INNER JOIN account_repo r ON a.uid = r.uid
            \\WHERE a.uid > $1 AND r.uid > $1 AND r.rev <> '' AND r.commit_cid <> ''
            \\ORDER BY a.uid ASC LIMIT $2
        , .{ self.cursor_val, self.limit });
        defer result.deinit();
        while (try result.nextUnsafe()) |row| {
            try output.append(row.get(i64, 0), row.get([]const u8, 1), row.get([]const u8, 5), row.get([]const u8, 4), row.get([]const u8, 2), row.get([]const u8, 3));
        }
        return output.finish(self.limit);
    }
};

pub fn handleListRepos(conn: *h.Conn, query: []const u8, ctx: *HttpContext) void {
    const cursor_str = h.queryParam(query, "cursor") orelse "0";
    const limit_str = h.queryParam(query, "limit") orelse "500";

    const cursor_val = std.fmt.parseInt(i64, cursor_str, 10) catch {
        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid cursor\"}");
        return;
    };
    if (cursor_val < 0) {
        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"cursor must be >= 0\"}");
        return;
    }

    const limit = std.fmt.parseInt(i64, limit_str, 10) catch {
        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid limit\"}");
        return;
    };
    if (limit < 1 or limit > 1000) {
        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"limit must be 1..1000\"}");
        return;
    }

    var req: ListReposReq = .{ .cursor_val = cursor_val, .limit = limit };
    defer if (req.body) |body| std.heap.c_allocator.free(body);
    ctx.db_queue.push(&req.base);
    req.base.wait(ctx.io, ctx.shutdown);

    if (req.base.err != null) {
        h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}");
        return;
    }
    if (req.db_err) {
        h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query or response encoding failed\"}");
        return;
    }

    h.respondJson(conn, .ok, req.body.?);
}

test "listRepos renders full default and maximum pages beyond 64 KiB" {
    const allocator = std.testing.allocator;
    for ([_]i64{ 500, 1000 }) |limit| {
        var output = try RepoListEncoder.init(allocator);
        defer output.output.deinit();
        for (0..@intCast(limit)) |i| {
            try output.append(@intCast(i + 1), "did:plc:2222cxntd4c3qttyd7smldsp", "bafyreigvbgb2nn5d7zgiiq3ziot25wfsximdjax2oetzbpgzmobnrjqh7u", "3my46tkfps52u", "active", if (i == 0) "deactivated" else "active");
        }
        const body = try output.finish(limit);
        defer allocator.free(body);
        try std.testing.expect(body.len > 65536);
        const parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{});
        defer parsed.deinit();
        const repos = parsed.value.object.get("repos").?.array.items;
        try std.testing.expectEqual(@as(usize, @intCast(limit)), repos.len);
        try std.testing.expect(!repos[0].object.get("active").?.bool);
        try std.testing.expectEqualStrings("deactivated", repos[0].object.get("status").?.string);
        try std.testing.expect(repos[1].object.get("active").?.bool);
        try std.testing.expectEqual(limit, try std.fmt.parseInt(i64, parsed.value.object.get("cursor").?.string, 10));
    }
}

test "listRepos keeps a cursor for a one-repo page and omits it at end" {
    const allocator = std.testing.allocator;
    var output = try RepoListEncoder.init(allocator);
    defer output.output.deinit();
    try output.append(42, "did:plc:2222cxntd4c3qttyd7smldsp", "bafyreigvbgb2nn5d7zgiiq3ziot25wfsximdjax2oetzbpgzmobnrjqh7u", "3my46tkfps52u", "taken_down", "active");
    const body = try output.finish(1);
    defer allocator.free(body);
    const parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{});
    defer parsed.deinit();
    try std.testing.expectEqualStrings("42", parsed.value.object.get("cursor").?.string);
    try std.testing.expectEqualStrings("taken_down", parsed.value.object.get("repos").?.array.items[0].object.get("status").?.string);
    var empty = try RepoListEncoder.init(allocator);
    defer empty.output.deinit();
    const end = try empty.finish(1);
    defer allocator.free(end);
    try std.testing.expectEqualStrings("{\"repos\":[]}", end);
}

fn encodeRepoPageWithAllocationFailures(allocator: Allocator) !void {
    encodeRepoPage(allocator) catch |err| switch (err) {
        error.WriteFailed => return error.OutOfMemory,
        else => return err,
    };
}

test "listRepos rejects incomplete repository commits without emitting a row" {
    var output = try RepoListEncoder.init(std.testing.allocator);
    defer output.output.deinit();
    try std.testing.expectError(error.MissingRepoCommit, output.append(1, "did:plc:2222cxntd4c3qttyd7smldsp", "", "3my46tkfps52u", "active", "active"));
    try std.testing.expectError(error.MissingRepoCommit, output.append(2, "did:plc:2222cxntd4c3qttyd7smldsp", "bafyreigvbgb2nn5d7zgiiq3ziot25wfsximdjax2oetzbpgzmobnrjqh7u", "", "active", "active"));
    const body = try output.finish(500);
    defer std.testing.allocator.free(body);
    try std.testing.expectEqualStrings("{\"repos\":[]}", body);
}

fn encodeRepoPage(allocator: Allocator) !void {
    var output = try RepoListEncoder.init(allocator);
    defer output.output.deinit();
    for (0..1000) |i| try output.append(@intCast(i + 1), "did:plc:2222cxntd4c3qttyd7smldsp", "bafyreigvbgb2nn5d7zgiiq3ziot25wfsximdjax2oetzbpgzmobnrjqh7u", "3my46tkfps52u", "active", "active");
    const body = try output.finish(1000);
    defer allocator.free(body);
}

test "listRepos releases partial responses on allocation failure" {
    try std.testing.checkAllAllocationFailures(std.testing.allocator, encodeRepoPageWithAllocationFailures, .{});
}

// --- getRepoStatus ---

const GetRepoStatusReq = struct {
    base: DbRequest = .{ .callback = &execute },
    did_buf: [256]u8 = undefined,
    did_len: usize = 0,
    json_buf: [4096]u8 = undefined,
    json_len: usize = 0,
    not_found: bool = false,
    db_err: bool = false,

    fn execute(b: *DbRequest, dp: *DiskPersist) void {
        const self: *@This() = @fieldParentPtr("base", b);
        const did = self.did_buf[0..self.did_len];

        var row = (dp.db.rowUnsafe(
            "SELECT a.uid, a.status, a.upstream_status, COALESCE(r.rev, '') FROM account a LEFT JOIN account_repo r ON a.uid = r.uid WHERE a.did = $1",
            .{did},
        ) catch {
            self.db_err = true;
            return;
        }) orelse {
            self.not_found = true;
            return;
        };
        defer row.deinit() catch {};

        const local_status = row.get([]const u8, 1);
        const upstream_status = row.get([]const u8, 2);
        const rev = row.get([]const u8, 3);
        const local_ok = std.mem.eql(u8, local_status, "active");
        const upstream_ok = std.mem.eql(u8, upstream_status, "active");
        const active = local_ok and upstream_ok;
        const status = if (!local_ok) local_status else upstream_status;

        var w: Io.Writer = .fixed(&self.json_buf);
        w.writeAll("{\"did\":\"") catch return;
        w.writeAll(did) catch return;
        w.writeAll("\"") catch return;

        if (active) {
            w.writeAll(",\"active\":true") catch return;
        } else {
            w.writeAll(",\"active\":false,\"status\":\"") catch return;
            w.writeAll(status) catch return;
            w.writeAll("\"") catch return;
        }

        if (rev.len > 0) {
            w.writeAll(",\"rev\":\"") catch return;
            w.writeAll(rev) catch return;
            w.writeAll("\"") catch return;
        }

        w.writeByte('}') catch return;
        self.json_len = w.end;
    }
};

pub fn handleGetRepoStatus(conn: *h.Conn, query: []const u8, ctx: *HttpContext) void {
    var did_buf: [256]u8 = undefined;
    const did = h.queryParamDecoded(query, "did", &did_buf) orelse {
        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"did parameter required\"}");
        return;
    };
    if (!std.mem.startsWith(u8, did, "did:")) {
        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid DID\"}");
        return;
    }

    var req: GetRepoStatusReq = .{};
    @memcpy(req.did_buf[0..did.len], did);
    req.did_len = did.len;
    ctx.db_queue.push(&req.base);
    req.base.wait(ctx.io, ctx.shutdown);

    if (req.base.err != null) {
        h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}");
        return;
    }
    if (req.db_err) {
        h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}");
        return;
    }
    if (req.not_found) {
        h.respondJson(conn, .not_found, "{\"error\":\"RepoNotFound\",\"message\":\"account not found\"}");
        return;
    }

    h.respondJson(conn, .ok, req.json_buf[0..req.json_len]);
}

// --- getRepo ---

const GetRepoReq = struct {
    base: DbRequest = .{ .callback = &execute },
    did_buf: [256]u8 = undefined,
    did_len: usize = 0,
    url_buf: [512]u8 = undefined,
    url_len: usize = 0,
    not_found: bool = false,
    db_err: bool = false,

    fn execute(b: *DbRequest, dp: *DiskPersist) void {
        const self: *@This() = @fieldParentPtr("base", b);
        const did = self.did_buf[0..self.did_len];

        var row = (dp.db.rowUnsafe(
            "SELECT h.hostname FROM account a JOIN host h ON a.host_id = h.id WHERE a.did = $1 AND a.host_id > 0",
            .{did},
        ) catch {
            self.db_err = true;
            return;
        }) orelse {
            self.not_found = true;
            return;
        };
        defer row.deinit() catch {};

        const hostname = row.get([]const u8, 0);
        const url = std.fmt.bufPrint(&self.url_buf, "https://{s}/xrpc/com.atproto.sync.getRepo?did={s}", .{ hostname, did }) catch return;
        self.url_len = url.len;
    }
};

pub fn handleGetRepo(conn: *h.Conn, query: []const u8, ctx: *HttpContext) void {
    var did_buf: [256]u8 = undefined;
    const did = h.queryParamDecoded(query, "did", &did_buf) orelse {
        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"did parameter required\"}");
        return;
    };
    if (!std.mem.startsWith(u8, did, "did:")) {
        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid DID\"}");
        return;
    }

    var req: GetRepoReq = .{};
    @memcpy(req.did_buf[0..did.len], did);
    req.did_len = did.len;
    ctx.db_queue.push(&req.base);
    req.base.wait(ctx.io, ctx.shutdown);

    if (req.base.err != null) {
        h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}");
        return;
    }
    if (req.db_err) {
        h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}");
        return;
    }
    if (req.not_found) {
        h.respondJson(conn, .not_found, "{\"error\":\"RepoNotFound\",\"message\":\"account not found\"}");
        return;
    }

    h.respondRedirect(conn, req.url_buf[0..req.url_len]);
}

// --- getLatestCommit ---

const GetLatestCommitReq = struct {
    base: DbRequest = .{ .callback = &execute },
    did_buf: [256]u8 = undefined,
    did_len: usize = 0,
    json_buf: [4096]u8 = undefined,
    json_len: usize = 0,
    not_found: bool = false,
    db_err: bool = false,
    status_err: ?[]const u8 = null,

    fn execute(b: *DbRequest, dp: *DiskPersist) void {
        const self: *@This() = @fieldParentPtr("base", b);
        const did = self.did_buf[0..self.did_len];

        var row = (dp.db.rowUnsafe(
            "SELECT a.status, a.upstream_status, COALESCE(r.rev, ''), COALESCE(r.commit_cid, '') FROM account a LEFT JOIN account_repo r ON a.uid = r.uid WHERE a.did = $1",
            .{did},
        ) catch {
            self.db_err = true;
            return;
        }) orelse {
            self.not_found = true;
            return;
        };
        defer row.deinit() catch {};

        const local_status = row.get([]const u8, 0);
        const upstream_status = row.get([]const u8, 1);
        const rev = row.get([]const u8, 2);
        const cid = row.get([]const u8, 3);

        const status = if (!std.mem.eql(u8, local_status, "active")) local_status else upstream_status;

        if (std.mem.eql(u8, status, "takendown") or std.mem.eql(u8, status, "suspended")) {
            self.status_err = "{\"error\":\"RepoTakendown\",\"message\":\"account has been taken down\"}";
            return;
        } else if (std.mem.eql(u8, status, "deactivated")) {
            self.status_err = "{\"error\":\"RepoDeactivated\",\"message\":\"account is deactivated\"}";
            return;
        } else if (std.mem.eql(u8, status, "deleted")) {
            self.status_err = "{\"error\":\"RepoDeleted\",\"message\":\"account is deleted\"}";
            return;
        } else if (!std.mem.eql(u8, status, "active")) {
            self.status_err = "{\"error\":\"RepoInactive\",\"message\":\"account is not active\"}";
            return;
        }

        if (rev.len == 0 or cid.len == 0) {
            self.not_found = true;
            return;
        }

        var w: Io.Writer = .fixed(&self.json_buf);
        w.writeAll("{\"cid\":\"") catch return;
        w.writeAll(cid) catch return;
        w.writeAll("\",\"rev\":\"") catch return;
        w.writeAll(rev) catch return;
        w.writeAll("\"}") catch return;
        self.json_len = w.end;
    }
};

pub fn handleGetLatestCommit(conn: *h.Conn, query: []const u8, ctx: *HttpContext) void {
    var did_buf: [256]u8 = undefined;
    const did = h.queryParamDecoded(query, "did", &did_buf) orelse {
        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"did parameter required\"}");
        return;
    };
    if (!std.mem.startsWith(u8, did, "did:")) {
        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid DID\"}");
        return;
    }

    var req: GetLatestCommitReq = .{};
    @memcpy(req.did_buf[0..did.len], did);
    req.did_len = did.len;
    ctx.db_queue.push(&req.base);
    req.base.wait(ctx.io, ctx.shutdown);

    if (req.base.err != null) {
        h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}");
        return;
    }
    if (req.db_err) {
        h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}");
        return;
    }
    if (req.status_err) |err_body| {
        h.respondJson(conn, .forbidden, err_body);
        return;
    }
    if (req.not_found) {
        if (req.json_len == 0) {
            h.respondJson(conn, .not_found, "{\"error\":\"RepoNotSynchronized\",\"message\":\"relay has no repo data for this account\"}");
        } else {
            h.respondJson(conn, .not_found, "{\"error\":\"RepoNotFound\",\"message\":\"account not found\"}");
        }
        return;
    }

    h.respondJson(conn, .ok, req.json_buf[0..req.json_len]);
}

// --- listReposByCollection (no DB, uses collection index directly) ---

pub fn handleListReposByCollection(conn: *h.Conn, query: []const u8, ci: *collection_index_mod.CollectionIndex) void {
    const collection = h.queryParam(query, "collection") orelse {
        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"collection parameter required\"}");
        return;
    };

    if (collection.len == 0 or !std.mem.containsAtLeast(u8, collection, 1, ".")) {
        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid collection NSID\"}");
        return;
    }

    const limit_str = h.queryParam(query, "limit") orelse "500";
    const limit = std.fmt.parseInt(usize, limit_str, 10) catch {
        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid limit\"}");
        return;
    };
    if (limit < 1 or limit > 2000) {
        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"limit must be 1..2000\"}");
        return;
    }

    var cursor_buf: [256]u8 = undefined;
    const cursor_did = h.queryParamDecoded(query, "cursor", &cursor_buf);

    var did_buf: [65536]u8 = undefined;
    var did_offsets: [2000]collection_index_mod.Span = undefined;
    const ci_result = ci.listReposByCollection(collection, limit, cursor_did, &did_buf, &did_offsets) catch {
        h.respondJson(conn, .internal_server_error, "{\"error\":\"InternalError\",\"message\":\"index scan failed\"}");
        return;
    };

    const body = renderListReposByCollection(std.heap.c_allocator, ci_result, limit) catch {
        h.respondJson(conn, .internal_server_error, "{\"error\":\"InternalError\",\"message\":\"response encoding failed\"}");
        return;
    };
    defer std.heap.c_allocator.free(body);
    h.respondJson(conn, .ok, body);
}

fn renderListReposByCollection(allocator: Allocator, result: collection_index_mod.ListResult, limit: usize) ![]u8 {
    var aw: Io.Writer.Allocating = .init(allocator);
    errdefer aw.deinit();
    const w = &aw.writer;

    try w.writeAll("{\"repos\":[");
    for (0..result.count) |i| {
        if (i > 0) try w.writeByte(',');
        try w.writeAll("{\"did\":\"");
        try w.writeAll(result.getDid(i));
        try w.writeAll("\"}");
    }
    try w.writeByte(']');

    if (result.last_did) |last| {
        if (result.count >= limit) {
            try w.writeAll(",\"cursor\":\"");
            try w.writeAll(last);
            try w.writeByte('"');
        }
    }

    try w.writeByte('}');
    return aw.toOwnedSlice();
}

test "listReposByCollection renders paginated collection index results" {
    const alloc = std.testing.allocator;
    var dir = std.testing.tmpDir(.{});
    defer dir.cleanup();
    const path = try dir.dir.realPathFileAlloc(std.testing.io, ".", alloc);
    defer alloc.free(path);
    var ci = try collection_index_mod.CollectionIndex.open(alloc, std.testing.io, path);
    defer ci.deinit();
    const collection = "app.bsky.feed.post";
    const dids = [_][]const u8{ "did:plc:2222bxcsksmtrafdqh7hgvxo", "did:web:example.com", "did:web:example.org:users:alice" };
    for (dids) |did| try ci.addCollection(did, collection);
    try ci.addCollection("did:web:other.example", "app.bsky.feed.like");

    for ([_]usize{ 1, 2, 2000 }) |limit| {
        var cursor: ?[]const u8 = null;
        var seen: usize = 0;
        var cursor_buf: [256]u8 = undefined;
        while (true) {
            var buf: [65536]u8 = undefined;
            var offsets: [2000]collection_index_mod.Span = undefined;
            const result = try @call(.never_inline, collection_index_mod.CollectionIndex.listReposByCollection, .{ &ci, collection, limit, cursor, &buf, &offsets });
            var other_buf: [256]u8 = undefined;
            var other_offsets: [1]collection_index_mod.Span = undefined;
            const other = try @call(.never_inline, collection_index_mod.CollectionIndex.listReposByCollection, .{ &ci, "app.bsky.feed.like", 1, null, &other_buf, &other_offsets });
            try std.testing.expectEqual(@as(usize, 1), other.count);
            const body = try renderListReposByCollection(alloc, result, limit);
            defer alloc.free(body);
            const parsed = try std.json.parseFromSlice(std.json.Value, alloc, body, .{});
            defer parsed.deinit();
            for (parsed.value.object.get("repos").?.array.items) |repo| {
                try std.testing.expect(seen < dids.len);
                try std.testing.expectEqualStrings(dids[seen], repo.object.get("did").?.string);
                seen += 1;
            }
            if (parsed.value.object.get("cursor")) |next| {
                @memcpy(cursor_buf[0..next.string.len], next.string);
                cursor = cursor_buf[0..next.string.len];
            } else break;
        }
        try std.testing.expectEqual(dids.len, seen);
    }
}

test "listReposByCollection renders a full page at the 2000 limit" {
    const alloc = std.testing.allocator;
    const did = "did:plc:2222cxntd4c3qttyd7smldsp";
    const limit = 2000;

    const did_buf = try alloc.alloc(u8, did.len * limit);
    defer alloc.free(did_buf);
    var offsets: [limit]collection_index_mod.Span = undefined;
    for (0..limit) |i| {
        @memcpy(did_buf[i * did.len ..][0..did.len], did);
        offsets[i] = .{ .start = i * did.len, .len = did.len };
    }
    const result: collection_index_mod.ListResult = .{
        .count = limit,
        .offsets = &offsets,
        .buf = did_buf,
        .last_did = did_buf[(limit - 1) * did.len ..][0..did.len],
    };

    const body = try renderListReposByCollection(alloc, result, limit);
    defer alloc.free(body);

    try std.testing.expect(body.len > 65536);
    const parsed = try std.json.parseFromSlice(std.json.Value, alloc, body, .{});
    defer parsed.deinit();
    try std.testing.expectEqual(@as(usize, limit), parsed.value.object.get("repos").?.array.items.len);
    try std.testing.expectEqualStrings(did, parsed.value.object.get("cursor").?.string);
}

// --- listHosts ---

const ListHostsReq = struct {
    base: DbRequest = .{ .callback = &execute },
    cursor_val: i64,
    limit: i64,
    json_buf: [65536]u8 = undefined,
    json_len: usize = 0,
    db_err: bool = false,

    fn execute(b: *DbRequest, dp: *DiskPersist) void {
        const self: *@This() = @fieldParentPtr("base", b);
        var w: Io.Writer = .fixed(&self.json_buf);

        var result = dp.db.query(
            "SELECT id, hostname, status, last_seq FROM host WHERE id > $1 AND last_seq > 0 ORDER BY id ASC LIMIT $2",
            .{ self.cursor_val, self.limit },
        ) catch {
            self.db_err = true;
            return;
        };
        defer result.deinit();

        var count: i64 = 0;
        var last_id: i64 = 0;

        w.writeAll("{\"hosts\":[") catch return;

        while (result.nextUnsafe() catch null) |row| {
            if (count > 0) w.writeByte(',') catch return;

            const id = row.get(i64, 0);
            const hostname = row.get([]const u8, 1);
            const status = row.get([]const u8, 2);
            const seq = row.get(i64, 3);

            w.writeAll("{\"hostname\":\"") catch return;
            w.writeAll(hostname) catch return;
            w.writeAll("\"") catch return;
            w.print(",\"seq\":{d}", .{seq}) catch return;
            w.writeAll(",\"status\":\"") catch return;
            w.writeAll(status) catch return;
            w.writeAll("\"}") catch return;

            last_id = id;
            count += 1;
        }

        w.writeByte(']') catch return;
        if (count >= self.limit and count > 1) {
            w.print(",\"cursor\":\"{d}\"", .{last_id}) catch return;
        }
        w.writeByte('}') catch return;
        self.json_len = w.end;
    }
};

pub fn handleListHosts(conn: *h.Conn, query: []const u8, ctx: *HttpContext) void {
    const cursor_str = h.queryParam(query, "cursor") orelse "0";
    const limit_str = h.queryParam(query, "limit") orelse "200";

    const cursor_val = std.fmt.parseInt(i64, cursor_str, 10) catch {
        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid cursor\"}");
        return;
    };
    if (cursor_val < 0) {
        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"cursor must be >= 0\"}");
        return;
    }

    const limit = std.fmt.parseInt(i64, limit_str, 10) catch {
        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid limit\"}");
        return;
    };
    if (limit < 1 or limit > 1000) {
        h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"limit must be 1..1000\"}");
        return;
    }

    var req: ListHostsReq = .{ .cursor_val = cursor_val, .limit = limit };
    ctx.db_queue.push(&req.base);
    req.base.wait(ctx.io, ctx.shutdown);

    if (req.base.err != null) {
        h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}");
        return;
    }
    if (req.db_err) {
        h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}");
        return;
    }

    h.respondJson(conn, .ok, req.json_buf[0..req.json_len]);
}

// --- getHostStatus ---

const GetHostStatusReq = struct {
    base: DbRequest = .{ .callback = &execute },
    hostname_buf: [256]u8 = undefined,
    hostname_len: usize = 0,
    json_buf: [4096]u8 = undefined,
    json_len: usize = 0,
    not_found: bool = false,
    db_err: bool = false,

    fn execute(b: *DbRequest, dp: *DiskPersist) void {
        const self: *@This() = @fieldParentPtr("base", b);
        const hostname = self.hostname_buf[0..self.hostname_len];

        var row = (dp.db.rowUnsafe(
            "SELECT id, hostname, status, last_seq FROM host WHERE hostname = $1",
            .{hostname},
        ) catch {
            self.db_err = true;
            return;
        }) orelse {
            self.not_found = true;
            return;
        };
        defer row.deinit() catch {};

        const host_id = row.get(i64, 0);
        const host_name = row.get([]const u8, 1);
        const raw_status = row.get([]const u8, 2);
        const seq = row.get(i64, 3);

        const status = if (std.mem.eql(u8, raw_status, "blocked"))
            "banned"
        else if (std.mem.eql(u8, raw_status, "exhausted"))
            "offline"
        else
            raw_status;

        // count accounts on this host
        const account_count: i64 = if (dp.db.rowUnsafe(
            "SELECT COUNT(*) FROM account WHERE host_id = $1",
            .{host_id},
        ) catch null) |cnt_row| blk: {
            var r = cnt_row;
            defer r.deinit() catch {};
            break :blk r.get(i64, 0);
        } else 0;

        var w: Io.Writer = .fixed(&self.json_buf);
        w.writeAll("{\"hostname\":\"") catch return;
        w.writeAll(host_name) catch return;
        w.writeAll("\"") catch return;
        w.print(",\"seq\":{d},\"accountCount\":{d}", .{ seq, account_count }) catch return;
        w.writeAll(",\"status\":\"") catch return;
        w.writeAll(status) catch return;
        w.writeAll("\"}") catch return;
        self.json_len = w.end;
    }
};

pub fn handleGetHostStatus(conn: *h.Conn, query: []const u8, ctx: *HttpContext) void {
    var hostname_buf: [256]u8 = undefined;
    const hostname = h.queryParamDecoded(query, "hostname", &hostname_buf) orelse {
        h.respondJson(conn, .bad_request, "{\"error\":\"InvalidRequest\",\"message\":\"hostname parameter required\"}");
        return;
    };

    var req: GetHostStatusReq = .{};
    @memcpy(req.hostname_buf[0..hostname.len], hostname);
    req.hostname_len = hostname.len;
    ctx.db_queue.push(&req.base);
    req.base.wait(ctx.io, ctx.shutdown);

    if (req.base.err != null) {
        h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}");
        return;
    }
    if (req.db_err) {
        h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}");
        return;
    }
    if (req.not_found) {
        h.respondJson(conn, .not_found, "{\"error\":\"HostNotFound\",\"message\":\"host not found\"}");
        return;
    }

    h.respondJson(conn, .ok, req.json_buf[0..req.json_len]);
}

// --- requestCrawl ---

pub fn handleRequestCrawl(conn: *h.Conn, body: []const u8, ctx: *HttpContext) void {
    // log on entry so we can tell from logs whether requests are being received
    // even if downstream waits wedge. body is small (~32 bytes) so this is cheap.
    log.info("requestCrawl received (body_len={d})", .{body.len});

    const parsed = std.json.parseFromSlice(struct { hostname: []const u8 }, ctx.persist.allocator, body, .{ .ignore_unknown_fields = true }) catch {
        h.respondJson(conn, .bad_request, "{\"error\":\"InvalidRequest\",\"message\":\"invalid JSON, expected {\\\"hostname\\\":\\\"...\\\"}\"}");
        return;
    };
    defer parsed.deinit();

    // fast validation: hostname format
    const hostname = slurper_mod.validateHostname(ctx.persist.allocator, parsed.value.hostname) catch |err| {
        log.warn("requestCrawl rejected '{s}': {s}", .{ parsed.value.hostname, @errorName(err) });
        h.respondJson(conn, .bad_request, switch (err) {
            error.EmptyHostname => "{\"error\":\"InvalidRequest\",\"message\":\"empty hostname\"}",
            error.InvalidCharacter => "{\"error\":\"InvalidRequest\",\"message\":\"hostname contains invalid characters\"}",
            error.InvalidLabel => "{\"error\":\"InvalidRequest\",\"message\":\"hostname has invalid label\"}",
            error.TooFewLabels => "{\"error\":\"InvalidRequest\",\"message\":\"hostname must have at least two labels (e.g. pds.example.com)\"}",
            error.LooksLikeIpAddress => "{\"error\":\"InvalidRequest\",\"message\":\"IP addresses not allowed, use a hostname\"}",
            error.PortNotAllowed => "{\"error\":\"InvalidRequest\",\"message\":\"port numbers not allowed\"}",
            error.LocalhostNotAllowed => "{\"error\":\"InvalidRequest\",\"message\":\"localhost not allowed\"}",
            else => "{\"error\":\"InvalidRequest\",\"message\":\"invalid hostname\"}",
        });
        return;
    };
    defer ctx.persist.allocator.free(hostname);

    // domain ban check via DbRequestQueue
    const DomainBanReq = struct {
        base_req: DbRequest = .{ .callback = &execute },
        hostname_buf: [256]u8 = undefined,
        hostname_len: usize = 0,
        banned: bool = false,

        fn execute(b: *DbRequest, dp: *DiskPersist) void {
            const self: *@This() = @fieldParentPtr("base_req", b);
            self.banned = dp.isDomainBanned(self.hostname_buf[0..self.hostname_len]);
        }
    };
    var ban_req: DomainBanReq = .{};
    const copy_len = @min(hostname.len, ban_req.hostname_buf.len);
    @memcpy(ban_req.hostname_buf[0..copy_len], hostname[0..copy_len]);
    ban_req.hostname_len = copy_len;
    ctx.db_queue.push(&ban_req.base_req);
    ban_req.base_req.wait(ctx.io, ctx.shutdown);

    if (ban_req.banned) {
        log.warn("requestCrawl rejected '{s}': domain banned", .{hostname});
        h.respondJson(conn, .bad_request, "{\"error\":\"InvalidRequest\",\"message\":\"domain is banned\"}");
        return;
    }

    // enqueue for async processing
    ctx.slurper.addCrawlRequest(hostname) catch {
        h.respondJson(conn, .internal_server_error, "{\"error\":\"failed to store crawl request\"}");
        return;
    };

    log.info("crawl requested: {s}", .{hostname});
    h.respondJson(conn, .ok, "{\"success\":true}");
}

test "repo endpoints return the signed commit CID and omit legacy roots" {
    const base_url = @import("../util/util.zig").getenv("DATABASE_URL") orelse return error.SkipZigTest;
    var tdb = try event_log_mod.TestDb.create(base_url);
    defer tdb.destroy();
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    var path_buf: [Io.Dir.max_path_bytes]u8 = undefined;
    const path_len = try tmp.dir.realPath(std.testing.io, &path_buf);
    var dp = try DiskPersist.init(std.testing.allocator, path_buf[0..path_len], tdb.url, 5, std.testing.io);
    defer dp.deinit();
    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
    defer arena.deinit();
    const a = arena.allocator();
    const zat = @import("zat");
    const did = "did:plc:2222cxntd4c3qttyd7smldsp";
    const rev = "3my46tkfps52u";
    const kp = try zat.Keypair.fromSecretKey(.p256, @splat(1));
    var mst = zat.mst.Mst.init(a);
    const root = try mst.rootCid();
    const signed = try zat.signCommit(a, .{ .did = did, .rev = rev, .data = root }, &kp);
    const car = try zat.car.writeAlloc(a, .{ .roots = &.{signed.cid}, .blocks = &.{.{ .cid_raw = signed.cid.raw, .data = signed.bytes }} });
    const verified = try zat.verifyCommitCar(a, car, .{ .key_type = .p256, .raw = &(try kp.publicKey()) }, .{ .verify_mst = false, .expected_did = did });
    const commit_cid = try zat.multibase.encode(a, .base32lower, verified.commit_cid);
    const data_cid = try zat.multibase.encode(a, .base32lower, root.raw);
    try std.testing.expect(!std.mem.eql(u8, commit_cid, data_cid));

    const legacy_did = "did:plc:legacy";
    const legacy_uid = try dp.uidForDid(legacy_did);
    _ = try dp.db.exec("INSERT INTO account_repo (uid, rev, commit_data_cid) VALUES ($1, $2, $3)", .{ @as(i64, @intCast(legacy_uid)), rev, data_cid });
    const uid = try dp.uidForDid(did);
    try std.testing.expect(try dp.updateAccountState(uid, rev, data_cid, commit_cid));

    var list: ListReposReq = .{ .cursor_val = 0, .limit = 1 };
    const body = try list.render(&dp);
    defer std.heap.c_allocator.free(body);
    const parsed = try std.json.parseFromSlice(std.json.Value, a, body, .{});
    const repos = parsed.value.object.get("repos").?.array.items;
    try std.testing.expectEqual(@as(usize, 1), repos.len);
    try std.testing.expectEqualStrings(did, repos[0].object.get("did").?.string);
    try std.testing.expectEqualStrings(commit_cid, repos[0].object.get("head").?.string);
    try std.testing.expectEqualStrings(rev, repos[0].object.get("rev").?.string);
    list.cursor_val = @intCast(uid);
    const end = try list.render(&dp);
    defer std.heap.c_allocator.free(end);
    try std.testing.expectEqualStrings("{\"repos\":[]}", end);

    var latest: GetLatestCommitReq = .{};
    @memcpy(latest.did_buf[0..did.len], did);
    latest.did_len = did.len;
    GetLatestCommitReq.execute(&latest.base, &dp);
    try std.testing.expect(!latest.db_err and !latest.not_found);
    const head = try std.json.parseFromSlice(std.json.Value, a, latest.json_buf[0..latest.json_len], .{});
    try std.testing.expectEqualStrings(commit_cid, head.value.object.get("cid").?.string);
    try std.testing.expectEqualStrings(rev, head.value.object.get("rev").?.string);

    latest = .{};
    @memcpy(latest.did_buf[0..legacy_did.len], legacy_did);
    latest.did_len = legacy_did.len;
    GetLatestCommitReq.execute(&latest.base, &dp);
    try std.testing.expect(latest.not_found and !latest.db_err);
}