const std = @import("std");
const zat = @import("zat");
const Owned = @import("owned.zig").Owned;
const rate = @import("rate.zig");

pub const Profile = struct { did: []const u8, handle: []const u8, displayName: []const u8 = "" };
pub const Session = struct { did: []const u8, handle: []const u8, accessJwt: []const u8, refreshJwt: []const u8 };
pub const Param = zat.XrpcClient.QueryParam;

pub const Client = struct {
    io: std.Io,
    allocator: std.mem.Allocator,
    rpc: zat.XrpcClient,
    session: ?Session = null,
    session_owner: ?std.json.Parsed(Session) = null,
    fault_owner: ?std.json.Parsed(Fault) = null,
    appview: ?[]const u8 = null,
    governor: rate.Governor = .{},
    shared_governor: ?*rate.Shared = null,
    last_response: ?Observation = null,
    last_fault: ?Fault = null,

    pub const Fault = struct { @"error": []const u8, message: ?[]const u8 = null };

    pub const Observation = struct {
        host: []const u8,
        appview: ?[]const u8,
        authenticated: bool,
        method: []const u8,
        status: u16,
        rate_limit: rate.Headers,
        observed_at: i64,
    };

    fn observe(self: *Client, method: []const u8, status: std.http.Status, headers: rate.Headers, retry: u8) void {
        const now = std.Io.Clock.real.now(self.io).toSeconds();
        self.governor.observe(headers, status == .too_many_requests, retry, std.Io.Clock.awake.now(self.io).toMilliseconds(), now);
        if (self.shared_governor) |shared| shared.observe(self.io, headers, status == .too_many_requests, retry);
        self.last_response = .{ .host = self.rpc.host, .appview = if (std.mem.startsWith(u8, method, "app.bsky.")) self.appview else null, .authenticated = self.session != null, .method = method, .status = @backingInt(status), .rate_limit = headers, .observed_at = now };
    }

    fn wait(self: *Client) !void {
        if (self.shared_governor) |shared| return shared.wait(self.io);
        return self.governor.wait(self.io);
    }

    pub fn init(io: std.Io, allocator: std.mem.Allocator, pds: []const u8) Client {
        return .{ .io = io, .allocator = allocator, .rpc = zat.XrpcClient.initWithUserAgent(io, allocator, pds, "tend/0.1") };
    }

    pub fn deinit(self: *Client) void {
        if (self.session_owner) |owner| owner.deinit();
        if (self.fault_owner) |owner| owner.deinit();
        self.rpc.deinit();
    }

    pub fn login(self: *Client, expected_did: []const u8, password: []const u8) !void {
        if (!std.mem.startsWith(u8, self.rpc.host, "https://")) return error.InsecurePds;
        const body = try std.json.Stringify.valueAlloc(self.allocator, .{ .identifier = expected_did, .password = password }, .{});
        defer self.allocator.free(body);
        try self.wait();
        var response = try self.rpc.procedure(zat.Nsid.parse("com.atproto.server.createSession").?, body);
        self.observe("com.atproto.server.createSession", response.status, response.rate_limit, 0);
        defer response.deinit();
        if (!response.ok()) return error.LoginFailed;
        const parsed = try std.json.parseFromSlice(Session, self.allocator, response.body, .{ .ignore_unknown_fields = true, .allocate = .alloc_always });
        errdefer parsed.deinit();
        const session = parsed.value;
        if (!std.mem.eql(u8, expected_did, session.did)) return error.IdentityMismatch;
        if (self.session_owner) |owner| owner.deinit();
        self.session_owner = parsed;
        self.session = session;
    }

    fn refresh(self: *Client) !void {
        const old = self.session orelse return error.NotAuthenticated;
        self.rpc.setAuth(old.refreshJwt);
        defer self.rpc.access_token = null;
        try self.wait();
        var response = try self.rpc.procedure(zat.Nsid.parse("com.atproto.server.refreshSession").?, null);
        self.observe("com.atproto.server.refreshSession", response.status, response.rate_limit, 0);
        defer response.deinit();
        if (!response.ok()) return error.RefreshFailed;
        const parsed = try std.json.parseFromSlice(Session, self.allocator, response.body, .{ .ignore_unknown_fields = true, .allocate = .alloc_always });
        errdefer parsed.deinit();
        const session = parsed.value;
        if (!std.mem.eql(u8, old.did, session.did)) return error.IdentityMismatch;
        if (self.session_owner) |owner| owner.deinit();
        self.session_owner = parsed;
        self.session = session;
    }

    pub fn query(self: *Client, comptime T: type, method: []const u8, params: []const Param) !std.json.Parsed(T) {
        if (self.fault_owner) |owner| owner.deinit();
        self.fault_owner = null;
        self.last_fault = null;
        if (zat.Nsid.parse(method) == null) return error.InvalidMethod;
        const url = try queryUrl(self.allocator, self.rpc.host, method, params);
        defer self.allocator.free(url);
        const bsky = std.mem.startsWith(u8, method, "app.bsky.");
        if (bsky and self.session != null) try validateAppview(self.appview orelse return error.MissingAppview);
        var refreshed = false;
        var throttles: u8 = 0;
        while (true) {
            try self.wait();
            const authorization = if (self.session) |s| try std.fmt.allocPrint(self.allocator, "Bearer {s}", .{s.accessJwt}) else null;
            defer if (authorization) |header| self.allocator.free(header);
            var response = try self.rpc.transport.fetch(.{
                .url = url,
                .authorization = authorization,
                .extra_headers = if (bsky and self.session != null) &.{.{ .name = "atproto-proxy", .value = self.appview.? }} else null,
                .deadline_ns = 30 * std.time.ns_per_s,
            });
            defer response.deinit(self.allocator);
            self.observe(method, response.status, response.rate_limit, throttles);
            if (response.status == .unauthorized and self.session != null and !refreshed) {
                const fault = std.json.parseFromSlice(struct { @"error": []const u8 }, self.allocator, response.body, .{ .ignore_unknown_fields = true }) catch return error.Unauthorized;
                defer fault.deinit();
                if (!std.mem.eql(u8, fault.value.@"error", "ExpiredToken")) return error.Unauthorized;
                refreshed = true;
                try self.refresh();
                continue;
            }
            if (response.status == .too_many_requests) {
                if (throttles >= 2) return error.RateLimited;
                throttles += 1;
                continue;
            }
            if (response.status != .ok) {
                self.fault_owner = std.json.parseFromSlice(Fault, self.allocator, response.body, .{ .ignore_unknown_fields = true, .allocate = .alloc_always }) catch null;
                self.last_fault = if (self.fault_owner) |owner| owner.value else null;
                return error.RequestFailed;
            }
            return std.json.parseFromSlice(T, self.allocator, response.body, .{ .ignore_unknown_fields = true, .allocate = .alloc_always });
        }
    }

    pub fn follows(self: *Client) !Owned([]const Profile) {
        return self.profiles("app.bsky.graph.getFollows", "follows", true);
    }

    pub fn download(self: *Client, method: []const u8, params: []const Param, accept: []const u8, sink: *std.Io.Writer, max_bytes: usize) !u64 {
        if (self.session != null) return error.AuthenticatedArchiveClient;
        if (zat.Nsid.parse(method) == null) return error.InvalidMethod;
        const url = try queryUrl(self.allocator, self.rpc.host, method, params);
        defer self.allocator.free(url);
        var throttles: u8 = 0;
        while (true) {
            try self.wait();
            var response = try self.rpc.transport.open(.{ .url = url, .accept = accept, .accept_encoding = "gzip", .deadline_ns = 60 * std.time.ns_per_s });
            defer response.deinit();
            self.observe(method, response.status, response.rate_limit, throttles);
            if (response.status == .too_many_requests) {
                if (throttles >= 2) return error.RateLimited;
                throttles += 1;
                continue;
            }
            if (response.status != .ok) return error.RequestFailed;
            return response.stream(sink, .{ .max_bytes = max_bytes });
        }
    }

    pub fn mutes(self: *Client) !Owned([]const Profile) {
        return self.profiles("app.bsky.graph.getMutes", "mutes", false);
    }

    fn profiles(self: *Client, comptime method: []const u8, comptime field: []const u8, actor: bool) !Owned([]const Profile) {
        const session = self.session orelse return error.NotAuthenticated;
        var arena = std.heap.ArenaAllocator.init(self.allocator);
        errdefer arena.deinit();
        const a = arena.allocator();
        const actor_did = try a.dupe(u8, session.did);
        var all: std.ArrayList(Profile) = .empty;
        var cursor: ?[]const u8 = null;
        var seen = std.StringHashMap(void).init(a);
        for (0..1000) |_| {
            var params: std.ArrayList(Param) = .empty;
            try params.append(a, .{ .name = "limit", .value = "100" });
            if (actor) try params.append(a, .{ .name = "actor", .value = actor_did });
            if (cursor) |value| try params.append(a, .{ .name = "cursor", .value = value });
            const Page = if (comptime std.mem.eql(u8, field, "follows")) struct { follows: []Profile, cursor: ?[]const u8 = null } else struct { mutes: []Profile, cursor: ?[]const u8 = null };
            const page = try self.query(Page, method, params.items);
            defer page.deinit();
            for (@field(page.value, field)) |account| try all.append(a, try copyProfile(a, account));
            cursor = if (page.value.cursor) |next| try a.dupe(u8, next) else null;
            if (cursor) |next| {
                if (seen.contains(next)) return error.RepeatedCursor;
                try seen.put(next, {});
            } else return .{ .value = try all.toOwnedSlice(a), .arena = arena };
        }
        return error.PageLimitExceeded;
    }
};

pub const Identity = struct { did: []const u8, pds: []const u8 };
pub fn resolve(io: std.Io, allocator: std.mem.Allocator, identifier: []const u8) !Owned(Identity) {
    var arena = std.heap.ArenaAllocator.init(allocator);
    errdefer arena.deinit();
    const a = arena.allocator();
    var handles = zat.HandleResolver.init(io, a);
    defer handles.deinit();
    const did = if (zat.Did.parse(identifier) != null) try a.dupe(u8, identifier) else try handles.resolve(zat.Handle.parse(identifier) orelse return error.InvalidHandle);
    var identities = zat.DidResolver.init(io, a);
    defer identities.deinit();
    var doc = try identities.resolve(zat.Did.parse(did) orelse return error.InvalidDid);
    defer doc.deinit();
    return .{ .arena = arena, .value = .{ .did = did, .pds = try a.dupe(u8, doc.pdsEndpoint() orelse return error.NoPdsEndpoint) } };
}

pub fn copyProfile(a: std.mem.Allocator, profile: Profile) !Profile {
    const did = try a.dupe(u8, profile.did);
    errdefer a.free(did);
    const handle = try a.dupe(u8, profile.handle);
    errdefer a.free(handle);
    return .{ .did = did, .handle = handle, .displayName = try a.dupe(u8, profile.displayName) };
}

fn queryUrl(allocator: std.mem.Allocator, host: []const u8, method: []const u8, params: []const Param) ![]const u8 {
    var url: std.ArrayList(u8) = .empty;
    errdefer url.deinit(allocator);
    try url.print(allocator, "{s}/xrpc/{s}", .{ host, method });
    for (params, 0..) |param, i| {
        try url.print(allocator, "{c}{s}=", .{ @as(u8, if (i == 0) '?' else '&'), param.name });
        for (param.value) |c| {
            if (std.ascii.isAlphanumeric(c) or c == '-' or c == '_' or c == '.' or c == '~') try url.append(allocator, c) else try url.print(allocator, "%{X:0>2}", .{c});
        }
    }
    return url.toOwnedSlice(allocator);
}

test "opaque cursors cannot inject query parameters" {
    const url = try queryUrl(std.testing.allocator, "https://pds.example", "app.bsky.graph.getMutes", &.{.{ .name = "cursor", .value = "a&actor=other?x#" }});
    defer std.testing.allocator.free(url);
    try std.testing.expectEqualStrings("https://pds.example/xrpc/app.bsky.graph.getMutes?cursor=a%26actor%3Dother%3Fx%23", url);
}

pub fn validateAppview(service: []const u8) !void {
    const split = std.mem.indexOfScalar(u8, service, '#') orelse return error.InvalidAppview;
    if (zat.Did.parse(service[0..split]) == null or split + 1 == service.len) return error.InvalidAppview;
    for (service[split + 1 ..]) |char| {
        if (!std.ascii.isAlphanumeric(char) and char != '_' and char != '-' and char != '.') return error.InvalidAppview;
    }
}

test "appview must identify a service, not an HTTP endpoint" {
    try validateAppview("did:web:api.bsky.app#bsky_appview");
    try validateAppview("did:web:alternative.example#microblog");
    try std.testing.expectError(error.InvalidAppview, validateAppview("https://public.api.bsky.app"));
    try std.testing.expectError(error.InvalidAppview, validateAppview("did:web:api.bsky.app"));
    try std.testing.expectError(error.InvalidAppview, validateAppview("did:web:api.bsky.app#"));
}

test "download retries a real 429 before writing successful response bytes" {
    const origin_module = @import("test_origin.zig");
    const io = std.testing.io;
    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
    defer arena.deinit();
    const a = arena.allocator();
    const replies: []const []const u8 = &.{
        "HTTP/1.1 429 Too Many Requests\r\nRetry-After: 0\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
        try origin_module.response(a, "archive"),
    };
    var origin = try origin_module.Origin.init(io, replies);
    defer origin.deinit();
    var serving = try io.concurrent(origin_module.Origin.serve, .{&origin});
    defer _ = serving.cancel(io) catch {};
    var client = Client.init(io, a, try origin.url(a));
    defer client.deinit();
    var sink: std.Io.Writer.Allocating = .init(a);
    const length = try client.download("com.atproto.sync.getRepo", &.{}, "application/octet-stream", &sink.writer, 100);
    try serving.await(io);
    try std.testing.expectEqual(@as(u64, 7), length);
    try std.testing.expectEqualStrings("archive", sink.written());
    try std.testing.expectEqual(@as(usize, 2), origin.requests);
}

test "a long-lived client releases request storage and results outlive later requests" {
    const origin_module = @import("test_origin.zig");
    const io = std.testing.io;
    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
    defer arena.deinit();
    const a = arena.allocator();
    const replies: []const []const u8 = &.{
        try origin_module.response(a, "{\"name\":\"first\"}"),
        "HTTP/1.1 400 Bad Request\r\nContent-Length: 15\r\nConnection: close\r\n\r\n{\"error\":\"Bad\"}",
        try origin_module.response(a, "{\"name\":\"third\"}"),
    };
    var origin = try origin_module.Origin.init(io, replies);
    defer origin.deinit();
    var serving = try io.concurrent(origin_module.Origin.serve, .{&origin});
    defer _ = serving.cancel(io) catch {};
    var rpc = Client.init(io, std.testing.allocator, try origin.url(a));
    defer rpc.deinit();
    const T = struct { name: []const u8 };
    const first = try rpc.query(T, "app.bsky.actor.getProfile", &.{});
    defer first.deinit();
    try std.testing.expectError(error.RequestFailed, rpc.query(T, "app.bsky.actor.getProfile", &.{}));
    try std.testing.expectEqualStrings("Bad", rpc.last_fault.?.@"error");
    const third = try rpc.query(T, "app.bsky.actor.getProfile", &.{});
    defer third.deinit();
    try std.testing.expect(rpc.last_fault == null);
    try std.testing.expectEqualStrings("first", first.value.name);
    try std.testing.expectEqualStrings("third", third.value.name);
    try serving.await(io);
}

test "relationship pages own their profiles after the client is closed" {
    const origin_module = @import("test_origin.zig");
    const io = std.testing.io;
    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
    defer arena.deinit();
    const a = arena.allocator();
    const replies: []const []const u8 = &.{
        try origin_module.response(a, "{\"follows\":[{\"did\":\"did:plc:one\",\"handle\":\"one.test\"}],\"cursor\":\"next\"}"),
        try origin_module.response(a, "{\"follows\":[{\"did\":\"did:plc:two\",\"handle\":\"two.test\"}]}"),
    };
    var origin = try origin_module.Origin.init(io, replies);
    defer origin.deinit();
    var serving = try io.concurrent(origin_module.Origin.serve, .{&origin});
    defer _ = serving.cancel(io) catch {};
    var result = block: {
        var rpc = Client.init(io, std.testing.allocator, try origin.url(a));
        defer rpc.deinit();
        rpc.session = .{ .did = "did:plc:viewer", .handle = "viewer.test", .accessJwt = "test-only", .refreshJwt = "test-only" };
        rpc.appview = "did:web:appview.test#bsky_appview";
        break :block try rpc.follows();
    };
    defer result.deinit();
    try std.testing.expectEqual(@as(usize, 2), result.value.len);
    try std.testing.expectEqualStrings("one.test", result.value[0].handle);
    try std.testing.expectEqualStrings("two.test", result.value[1].handle);
    try serving.await(io);
}