const std = @import("std");
const clock = @import("../core/clock.zig");
const config = @import("../core/config.zig");
const log = @import("../core/log.zig");
const http_api = @import("../http/api.zig");
const eventlog = @import("../storage/eventlog.zig");
const store = @import("../storage/store.zig");
const zat = @import("zat");
const httpz = @import("httpz");

const http = std.http;

const max_subscribe_repos_connections = 32;

/// Backfill reconnect limiter: a consumer that reconnects with a historical
/// cursor more than `max_backfill_connects_per_window` times inside the window
/// is re-streaming history in a loop (a consumer that never persists its
/// cursor drains the whole CPU burst budget on a small machine). Live-tail
/// connects are never limited.
const backfill_window_seconds: u32 = 300;
const max_backfill_connects_per_window: u32 = 4;
const BackfillSlot = struct { ip_hash: u64 = 0, window_start_s: u32 = 0, count: u32 = 0 };
var backfill_slots: [64]BackfillSlot = @splat(.{});
var backfill_slots_lock: std.atomic.Value(bool) = .init(false);

fn backfillReconnectAllowed(ip: []const u8, now_s: u32) bool {
    if (ip.len == 0) return true;
    const hash = std.hash.Wyhash.hash(0, ip);
    while (backfill_slots_lock.cmpxchgWeak(false, true, .acquire, .monotonic) != null) {
        std.atomic.spinLoopHint();
    }
    defer backfill_slots_lock.store(false, .release);
    var free_index: ?usize = null;
    var oldest_index: usize = 0;
    for (&backfill_slots, 0..) |*slot, i| {
        if (slot.ip_hash == hash) {
            if (now_s -| slot.window_start_s >= backfill_window_seconds) {
                slot.window_start_s = now_s;
                slot.count = 1;
                return true;
            }
            slot.count += 1;
            return slot.count <= max_backfill_connects_per_window;
        }
        if (slot.ip_hash == 0 and free_index == null) free_index = i;
        if (slot.window_start_s < backfill_slots[oldest_index].window_start_s) oldest_index = i;
    }
    const index = free_index orelse oldest_index;
    backfill_slots[index] = .{ .ip_hash = hash, .window_start_s = now_s, .count = 1 };
    return true;
}
const max_subscribe_repos_backfills = 4;
const notify_threshold_s = 20 * 60;

var subscribe_repos_connections: usize = 0;
var subscribe_repos_backfills: usize = 0;
var get_repo_connections: usize = 0;
var crawler_last_notified_s: std.atomic.Value(u32) = .init(0);
var crawler_notify_in_flight: std.atomic.Value(bool) = .init(false);

pub const SubscribeReposClient = struct {
    state: *StreamState,

    pub const Context = struct {
        cursor: u64,
        backfill: bool,
        peer: []const u8,
    };

    pub fn init(conn: *httpz.websocket.Conn, ctx: *const Context) !SubscribeReposClient {
        const state = try std.heap.smp_allocator.create(StreamState);
        state.* = .{
            .conn = conn,
            .cursor = ctx.cursor,
            .backfill = ctx.backfill,
            .start_cursor = ctx.cursor,
            .started_s = nowSeconds(),
        };
        state.peer_len = @min(ctx.peer.len, state.peer.len);
        @memcpy(state.peer[0..state.peer_len], ctx.peer[0..state.peer_len]);
        return .{ .state = state };
    }

    pub fn afterInit(self: *SubscribeReposClient) !void {
        self.state.retain();
        const thread = std.Thread.spawn(.{}, streamEvents, .{self.state}) catch |err| {
            self.state.release();
            log.err("sync subscribeRepos spawn failed cursor={d} err={s}\n", .{ self.state.cursor, @errorName(err) });
            return err;
        };
        thread.detach();
    }

    pub fn clientMessage(_: *SubscribeReposClient, _: []const u8) !void {}

    pub fn close(self: *SubscribeReposClient) void {
        @atomicStore(bool, &self.state.closed, true, .release);
        eventlog.wakeWaiters();
        self.state.finish();
        self.state.release();
    }

    const StreamState = struct {
        conn: *httpz.websocket.Conn,
        cursor: u64,
        backfill: bool,
        closed: bool = false,
        counted: bool = true,
        refs: usize = 1,
        start_cursor: u64 = 0,
        started_s: u32 = 0,
        frames: u64 = 0,
        bytes: u64 = 0,
        peer: [64]u8 = @splat(0),
        peer_len: usize = 0,

        fn peerText(self: *const StreamState) []const u8 {
            return self.peer[0..self.peer_len];
        }

        fn retain(self: *StreamState) void {
            _ = @atomicRmw(usize, &self.refs, .Add, 1, .monotonic);
        }

        fn release(self: *StreamState) void {
            if (@atomicRmw(usize, &self.refs, .Sub, 1, .acq_rel) == 1) {
                std.heap.smp_allocator.destroy(self);
            }
        }

        fn finish(self: *StreamState) void {
            if (@atomicRmw(bool, &self.counted, .Xchg, false, .acq_rel)) {
                _ = @atomicRmw(usize, &subscribe_repos_connections, .Sub, 1, .monotonic);
                self.releaseBackfillSlot();
                log.info("sync subscribeRepos end peer={s} start_cursor={d} end_cursor={d} frames={d} bytes={d} duration_s={d}\n", .{
                    self.peerText(),
                    self.start_cursor,
                    self.cursor,
                    self.frames,
                    self.bytes,
                    nowSeconds() -| self.started_s,
                });
            }
        }

        /// A backfill slot bounds concurrent event-history reads, not
        /// connection lifetime; it is released the moment the subscriber
        /// catches up to live, or on teardown if it never does.
        fn releaseBackfillSlot(self: *StreamState) void {
            if (@atomicRmw(bool, &self.backfill, .Xchg, false, .acq_rel)) {
                _ = @atomicRmw(usize, &subscribe_repos_backfills, .Sub, 1, .monotonic);
            }
        }

        fn isClosed(self: *const StreamState) bool {
            return @atomicLoad(bool, &self.closed, .acquire);
        }
    };

    fn streamEvents(state: *StreamState) void {
        defer state.release();
        defer state.finish();

        var observed = eventlog.currentGeneration();
        while (!state.isClosed()) {
            var sent = false;
            {
                var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
                defer arena.deinit();
                const events = store.listSeqEvents(arena.allocator(), state.cursor, 100) catch |err| {
                    log.err("sync subscribeRepos list failed cursor={d} err={s}\n", .{ state.cursor, @errorName(err) });
                    return;
                };
                for (events) |event| {
                    if (state.isClosed()) return;
                    writeFirehoseFrame(state.conn, event.frame) catch |err| {
                        switch (err) {
                            error.BrokenPipe, error.ConnectionResetByPeer => return,
                            else => log.err("sync subscribeRepos write failed cursor={d} seq={d} bytes={d} err={s}\n", .{ state.cursor, event.seq, event.frame.len, @errorName(err) }),
                        }
                        return;
                    };
                    state.cursor = event.seq;
                    state.frames += 1;
                    state.bytes += event.frame.len;
                    sent = true;
                }
            }
            if (sent) {
                observed = eventlog.currentGeneration();
                continue;
            }
            state.releaseBackfillSlot();
            const next = (eventlog.waitForChange(observed, &state.closed) catch return) orelse return;
            if (next.generation != observed or next.latest_seq > state.cursor) {
                observed = next.generation;
            }
        }
    }

    fn writeFirehoseFrame(conn: *httpz.websocket.Conn, payload: []const u8) !void {
        var header: [10]u8 = undefined;
        const header_len = websocketBinaryHeader(&header, payload.len);

        conn.lock.lockUncancelable(conn.io);
        defer conn.lock.unlock(conn.io);

        const socket = conn.stream.socket.handle;
        try writeSocketAll(socket, header[0..header_len]);
        try writeSocketAll(socket, payload);
    }

    fn websocketBinaryHeader(header: *[10]u8, payload_len: usize) usize {
        header[0] = 0x82;
        if (payload_len < 126) {
            header[1] = @intCast(payload_len);
            return 2;
        }
        if (payload_len <= std.math.maxInt(u16)) {
            header[1] = 126;
            std.mem.writeInt(u16, header[2..4], @intCast(payload_len), .big);
            return 4;
        }
        header[1] = 127;
        std.mem.writeInt(u64, header[2..10], @intCast(payload_len), .big);
        return 10;
    }

    fn writeSocketAll(socket: std.posix.socket_t, data: []const u8) !void {
        var remaining = data;
        while (remaining.len > 0) {
            const written = std.c.write(socket, remaining.ptr, remaining.len);
            if (written > 0) {
                remaining = remaining[@intCast(written)..];
                continue;
            }
            if (written == 0) return error.WriteZero;
            switch (std.posix.errno(written)) {
                .AGAIN => {
                    std.Thread.yield() catch {};
                    continue;
                },
                .INTR => continue,
                .PIPE => return error.BrokenPipe,
                .CONNRESET => return error.ConnectionResetByPeer,
                else => return error.WriteFailed,
            }
        }
    }
};

pub fn getBlob(request: *http_api.Request) !void {
    var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
    defer arena.deinit();
    const allocator = arena.allocator();

    var did_buf: [256]u8 = undefined;
    const did = http_api.queryParam(request.url.raw, "did", &did_buf) orelse {
        return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing did");
    };
    var cid_buf: [256]u8 = undefined;
    const cid = http_api.queryParam(request.url.raw, "cid", &cid_buf) orelse {
        return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing cid");
    };
    try requirePublicRepoAvailable(request, did);
    const blob = store.getPublicBlob(allocator, did, cid) orelse {
        return http_api.xrpcError(request, .not_found, "BlobNotFound", "Blob not found");
    };
    const disposition = try std.fmt.allocPrint(allocator, "attachment; filename=\"{s}\"", .{cid});
    const headers = [_]http.Header{
        .{ .name = "content-type", .value = blob.mime_type },
        .{ .name = "cache-control", .value = "public, max-age=31536000, immutable" },
        .{ .name = "x-content-type-options", .value = "nosniff" },
        .{ .name = "content-disposition", .value = disposition },
        .{ .name = "content-security-policy", .value = "default-src 'none'; sandbox" },
        .{ .name = "access-control-allow-origin", .value = "*" },
        .{ .name = "access-control-allow-private-network", .value = "true" },
        .{ .name = "connection", .value = "close" },
    };
    try http_api.respondNowClose(request, .ok, if (request.method == .HEAD) "" else blob.data, &headers);
}

pub fn getRepo(request: *http_api.Request) !void {
    var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
    defer arena.deinit();
    const allocator = arena.allocator();

    var did_buf: [256]u8 = undefined;
    const did = http_api.queryParam(request.url.raw, "did", &did_buf) orelse {
        return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing did");
    };
    var since_buf: [256]u8 = undefined;
    const since = http_api.queryParam(request.url.raw, "since", &since_buf);
    try requirePublicRepoAvailable(request, did);

    const full_export = since == null;
    if (full_export) {
        const active = @atomicRmw(usize, &get_repo_connections, .Add, 1, .monotonic);
        const max_connections = config.maxConcurrentRepoExports();
        if (active >= max_connections) {
            _ = @atomicRmw(usize, &get_repo_connections, .Sub, 1, .monotonic);
            log.debug("sync getRepo rejected too_many_connections active={d} max={d} did={s}\n", .{ active + 1, max_connections, did });
            return http_api.xrpcError(request, .too_many_requests, "RateLimitExceeded", "too many full getRepo exports");
        }
    }
    defer {
        if (full_export) _ = @atomicRmw(usize, &get_repo_connections, .Sub, 1, .monotonic);
    }

    const body = store.writeRepoCarSince(allocator, did, since) catch {
        return http_api.xrpcError(request, .not_found, "RepoNotFound", "Repo not found");
    };
    const headers = [_]http.Header{
        .{ .name = "content-type", .value = "application/vnd.ipld.car" },
        .{ .name = "access-control-allow-origin", .value = "*" },
        .{ .name = "access-control-allow-private-network", .value = "true" },
        .{ .name = "connection", .value = "close" },
    };
    if (request.method == .HEAD) {
        return http_api.respondNowClose(request, .ok, "", &headers);
    }
    try http_api.respond(request, .ok, body, &headers);
}

pub fn getRecord(request: *http_api.Request) !void {
    var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
    defer arena.deinit();
    const allocator = arena.allocator();

    var did_buf: [256]u8 = undefined;
    const did = http_api.queryParam(request.url.raw, "did", &did_buf) orelse {
        return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing did");
    };
    var collection_buf: [256]u8 = undefined;
    const collection = http_api.queryParam(request.url.raw, "collection", &collection_buf) orelse {
        return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing collection");
    };
    var rkey_buf: [512]u8 = undefined;
    const rkey = http_api.queryParam(request.url.raw, "rkey", &rkey_buf) orelse {
        return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing rkey");
    };
    try requirePublicRepoAvailable(request, did);

    const body = store.writeRecordProofCar(allocator, did, collection, rkey) catch |err| switch (err) {
        error.RepoNotFound => return http_api.xrpcError(request, .not_found, "RepoNotFound", "Repo not found"),
        else => {
            log.err("sync getRecord failed did={s} collection={s} rkey={s} err={s}\n", .{ did, collection, rkey, @errorName(err) });
            return http_api.xrpcError(request, .internal_server_error, "InternalServerError", "Could not build record proof");
        },
    };
    const headers = [_]http.Header{
        .{ .name = "content-type", .value = "application/vnd.ipld.car" },
        .{ .name = "access-control-allow-origin", .value = "*" },
        .{ .name = "access-control-allow-private-network", .value = "true" },
        .{ .name = "connection", .value = "close" },
    };
    if (request.method == .HEAD) {
        return http_api.respondNowClose(request, .ok, "", &headers);
    }
    try http_api.respond(request, .ok, body, &headers);
}

pub fn listBlobs(request: *http_api.Request) !void {
    var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
    defer arena.deinit();
    const allocator = arena.allocator();

    var did_buf: [256]u8 = undefined;
    const did = http_api.queryParam(request.url.raw, "did", &did_buf) orelse {
        return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing did");
    };
    var since_buf: [256]u8 = undefined;
    var cursor_buf: [256]u8 = undefined;
    try requirePublicRepoAvailable(request, did);
    const body = try store.writeBlobListJson(
        allocator,
        did,
        http_api.queryParam(request.url.raw, "since", &since_buf),
        http_api.queryParam(request.url.raw, "cursor", &cursor_buf),
        @min(http_api.queryLimit(request.url.raw, 500), 1000),
    );
    return http_api.json(request, .ok, body);
}

pub fn getLatestCommit(request: *http_api.Request) !void {
    var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
    defer arena.deinit();
    const allocator = arena.allocator();

    var did_buf: [256]u8 = undefined;
    const did = http_api.queryParam(request.url.raw, "did", &did_buf) orelse {
        return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing did");
    };
    try requirePublicRepoAvailable(request, did);
    const body = store.writeLatestCommitJson(allocator, did) catch |err| switch (err) {
        error.RepoNotFound => return http_api.xrpcError(request, .not_found, "RepoNotFound", "Repo not found"),
        else => return err,
    };
    return http_api.json(request, .ok, body);
}

pub fn listRepos(request: *http_api.Request) !void {
    var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
    defer arena.deinit();
    const allocator = arena.allocator();

    var cursor_buf: [512]u8 = undefined;
    const body = store.writeRepoListJson(allocator, http_api.queryParam(request.url.raw, "cursor", &cursor_buf), http_api.queryLimit(request.url.raw, 500)) catch |err| switch (err) {
        error.InvalidCursor => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Malformed cursor"),
        else => return err,
    };
    return http_api.json(request, .ok, body);
}

pub fn subscribeRepos(request: *http_api.Request) !void {
    var cursor_buf: [32]u8 = undefined;
    const cursor_param = http_api.queryParam(request.url.raw, "cursor", &cursor_buf);
    const current_seq = store.latestSequence();
    const start = subscriptionStart(cursor_param, current_seq) catch
        return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Malformed cursor");
    const cursor = start.cursor;
    const backfill = start.backfill;
    const peer = http_api.headerValue(request, "fly-client-ip") orelse
        http_api.headerValue(request, "x-forwarded-for") orelse "";
    const user_agent = http_api.headerValue(request, "user-agent") orelse "";
    log.info("sync subscribeRepos connect peer={s} ua={s} cursor={?s} start={d} current={d} backfill={}\n", .{
        peer, user_agent, cursor_param, cursor, current_seq, backfill,
    });

    if (backfill and !backfillReconnectAllowed(peer, nowSeconds())) {
        log.info("sync subscribeRepos rejected backfill_reconnect_loop peer={s} ua={s} cursor={d}\n", .{ peer, user_agent, cursor });
        return http_api.xrpcError(request, .too_many_requests, "RateLimitExceeded", "too many backfill reconnects from this address; persist your cursor");
    }

    const active = @atomicRmw(usize, &subscribe_repos_connections, .Add, 1, .monotonic);
    if (active >= max_subscribe_repos_connections) {
        _ = @atomicRmw(usize, &subscribe_repos_connections, .Sub, 1, .monotonic);
        log.debug("sync subscribeRepos rejected too_many_connections active={d} max={d}\n", .{ active + 1, max_subscribe_repos_connections });
        return http_api.xrpcError(request, .too_many_requests, "RateLimitExceeded", "too many subscribeRepos connections");
    }
    if (backfill) {
        const active_backfills = @atomicRmw(usize, &subscribe_repos_backfills, .Add, 1, .monotonic);
        if (active_backfills >= max_subscribe_repos_backfills) {
            _ = @atomicRmw(usize, &subscribe_repos_backfills, .Sub, 1, .monotonic);
            _ = @atomicRmw(usize, &subscribe_repos_connections, .Sub, 1, .monotonic);
            log.debug("sync subscribeRepos rejected too_many_backfills active={d} max={d}\n", .{ active_backfills + 1, max_subscribe_repos_backfills });
            return http_api.xrpcError(request, .too_many_requests, "RateLimitExceeded", "too many subscribeRepos backfills");
        }
    }
    var upgraded = false;
    defer if (!upgraded) {
        _ = @atomicRmw(usize, &subscribe_repos_connections, .Sub, 1, .monotonic);
        if (backfill) _ = @atomicRmw(usize, &subscribe_repos_backfills, .Sub, 1, .monotonic);
    };

    const ctx = SubscribeReposClient.Context{ .cursor = cursor, .backfill = backfill, .peer = peer };
    upgraded = try http_api.upgradeWebsocket(SubscribeReposClient, request, &ctx);
    if (!upgraded) {
        return http_api.xrpcError(request, .upgrade_required, "InvalidRequest", "Expected WebSocket upgrade");
    }
}

const SubscriptionStart = struct {
    cursor: u64,
    backfill: bool,
};

fn subscriptionStart(cursor_param: ?[]const u8, current_seq: u64) !SubscriptionStart {
    const cursor = if (cursor_param) |raw| try std.fmt.parseInt(u64, raw, 10) else current_seq;
    return .{ .cursor = cursor, .backfill = cursor < current_seq };
}

test "backfill reconnect limiter allows a burst then rejects within the window" {
    backfill_slots = @splat(.{});
    const ip = "203.0.113.7";
    var i: u32 = 0;
    while (i < max_backfill_connects_per_window) : (i += 1) {
        try std.testing.expect(backfillReconnectAllowed(ip, 1000));
    }
    try std.testing.expect(!backfillReconnectAllowed(ip, 1000));
    try std.testing.expect(!backfillReconnectAllowed(ip, 1000 + backfill_window_seconds - 1));
    try std.testing.expect(backfillReconnectAllowed(ip, 1000 + backfill_window_seconds));
    try std.testing.expect(backfillReconnectAllowed("203.0.113.8", 1000));
    try std.testing.expect(backfillReconnectAllowed("", 1000));
    backfill_slots = @splat(.{});
}

test "subscribeRepos starts live unless a historical cursor is explicit" {
    try std.testing.expectEqual(SubscriptionStart{ .cursor = 42, .backfill = false }, try subscriptionStart(null, 42));
    try std.testing.expectEqual(SubscriptionStart{ .cursor = 0, .backfill = true }, try subscriptionStart("0", 42));
    try std.testing.expectEqual(SubscriptionStart{ .cursor = 42, .backfill = false }, try subscriptionStart("42", 42));
    try std.testing.expectError(error.InvalidCharacter, subscriptionStart("nope", 42));
}

pub fn getRepoStatus(request: *http_api.Request) !void {
    var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
    defer arena.deinit();
    const allocator = arena.allocator();

    var did_buf: [256]u8 = undefined;
    const did = http_api.queryParam(request.url.raw, "did", &did_buf) orelse {
        return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing did");
    };
    const body = store.writeRepoStatusJson(allocator, did) catch |err| switch (err) {
        error.RepoNotFound => return http_api.xrpcError(request, .not_found, "RepoNotFound", "Repo not found"),
        else => return err,
    };
    return http_api.json(request, .ok, body);
}

pub fn notifyOfUpdate(request: *http_api.Request) !void {
    return http_api.json(request, .ok, "{}");
}

pub fn requestCrawl(request: *http_api.Request) !void {
    return http_api.json(request, .ok, "{}");
}

fn requirePublicRepoAvailable(request: *http_api.Request, did: []const u8) !void {
    const status = store.accountStatus(did) catch |err| switch (err) {
        error.RepoNotFound => return http_api.xrpcError(request, .not_found, "RepoNotFound", "Repo not found"),
        else => return err,
    };
    switch (status) {
        .active => return,
        .takendown => return http_api.xrpcError(request, .bad_request, "RepoTakendown", "Repo has been taken down"),
        .suspended => return http_api.xrpcError(request, .bad_request, "RepoSuspended", "Repo is suspended"),
        .deactivated => return http_api.xrpcError(request, .bad_request, "RepoDeactivated", "Repo is deactivated"),
        .deleted => return http_api.xrpcError(request, .not_found, "RepoNotFound", "Repo not found"),
    }
}

pub fn notifyCrawlers(force: bool) void {
    const now = nowSeconds();
    if (!force) {
        const last = crawler_last_notified_s.load(.acquire);
        if (last != 0 and now -% last < notify_threshold_s) return;
    }

    if (crawler_notify_in_flight.cmpxchgStrong(false, true, .acq_rel, .acquire) != null) return;
    crawler_last_notified_s.store(now, .release);
    const thread = std.Thread.spawn(.{}, notifyCrawlersThread, .{}) catch |err| {
        crawler_notify_in_flight.store(false, .release);
        log.err("sync request_crawl spawn failed err={s}\n", .{@errorName(err)});
        return;
    };
    thread.detach();
}

fn nowSeconds() u32 {
    return @truncate(@as(u64, @intCast(clock.nowSeconds())));
}

fn notifyCrawlersThread() void {
    defer crawler_notify_in_flight.store(false, .release);
    var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
    defer arena.deinit();
    requestConfiguredCrawls(arena.allocator());
}

fn requestConfiguredCrawls(allocator: std.mem.Allocator) void {
    const host = publicHostname(config.publicUrl()) orelse {
        log.err("sync request_crawl skipped invalid_public_url url={s}\n", .{config.publicUrl()});
        return;
    };
    var crawlers = std.mem.splitScalar(u8, config.crawlers(), ',');
    while (crawlers.next()) |crawler_raw| {
        const crawler = std.mem.trim(u8, crawler_raw, " \t\r\n/");
        if (crawler.len == 0) continue;
        requestCrawler(allocator, crawler, host) catch |err| {
            log.err("sync request_crawl failed crawler={s} host={s} err={s}\n", .{ crawler, host, @errorName(err) });
            continue;
        };
    }
}

fn requestCrawler(allocator: std.mem.Allocator, crawler: []const u8, host: []const u8) !void {
    const url = try std.fmt.allocPrint(allocator, "{s}/xrpc/com.atproto.sync.requestCrawl", .{crawler});
    const payload = try std.fmt.allocPrint(allocator, "{{\"hostname\":{f}}}", .{std.json.fmt(host, .{})});
    var transport = zat.HttpTransport.init(store.currentIo(), allocator);
    defer transport.deinit();
    const result = try transport.fetch(.{
        .url = url,
        .method = .POST,
        .payload = payload,
        .content_type = "application/json",
        .max_response_size = 64 * 1024,
    });
    if (@backingInt(result.status) < 200 or @backingInt(result.status) >= 300) return error.CrawlerRejected;
    log.info("sync request_crawl ok crawler={s} host={s}\n", .{ crawler, host });
}

fn publicHostname(url: []const u8) ?[]const u8 {
    const scheme = std.mem.indexOf(u8, url, "://") orelse return null;
    var rest = url[scheme + 3 ..];
    if (std.mem.indexOfScalar(u8, rest, '/')) |slash| rest = rest[0..slash];
    if (std.mem.indexOfScalar(u8, rest, '@')) |at| rest = rest[at + 1 ..];
    if (rest.len == 0) return null;
    if (std.mem.startsWith(u8, rest, "[")) {
        const close = std.mem.indexOfScalar(u8, rest, ']') orelse return null;
        return rest[0 .. close + 1];
    }
    if (std.mem.indexOfScalar(u8, rest, ':')) |colon| return rest[0..colon];
    return rest;
}