//! HTTP on siv's event loop: libcurl's multi interface, with its sockets
//! watched by siv.appWatchFd and its timeouts kept by siv.timerStart, so
//! requests run on the app's thread and never block it. Requests to one
//! host share HTTP/2 connections, ten in flight at most, as Google turns
//! away a user's requests past some number at once; the rest wait their
//! turn in curl. A reply that says to slow down is retried, later. A
//! reply that goes on, such as server-sent events, is heard as it comes.
//!
//! libcurl stands in until baskingspot/http and baskingspot/tls build on
//! Zig 0.17: they are sans-I/O too, and would sit behind the same watches.
//! The declarations below are the few parts of curl.h it needs, written
//! out because Zig 0.17 has no @cImport; their values are libcurl's ABI.

const std = @import("std");
const siv = @import("siv");
const ascii = std.ascii;
const mem = std.mem;
const testing = std.testing;
const Allocator = std.mem.Allocator;
const ArrayList = std.ArrayList;
const Io = std.Io;
const log = std.log.scoped(.http);

const Http = @This();

gpa: Allocator,
app: siv.App,
multi: *c.Multi,
/// When curl wants to be called with no socket ready; null when it does
/// not.
timer: ?siv.Timer,
/// Requests in flight, to free if the client goes first.
transfers: ArrayList(*Transfer),

pub const Request = struct {
    method: Method,
    url: []const u8,
    /// Whole header lines, such as "authorization: Bearer …".
    headers: []const []const u8,
    /// What a POST or a PUT sends.
    body: ?[]const u8,
};

pub const Method = enum { get, post, put, delete };

pub const Response = struct {
    status: u16,
    /// Where the reply came from, after any redirects; freed with the
    /// body.
    url: []const u8,
    /// Freed once the callback returns.
    body: []const u8,
};

pub const Error = error{ HttpFailed, OutOfMemory };

/// Hears how a request went, on the app's thread.
pub const Done = *const fn (context: *anyopaque, result: Error!Response) void;

/// Hears what has come of a reply that goes on, on the app's thread.
pub const Heard = *const fn (context: *anyopaque, bytes: []const u8) void;

pub fn init(gpa: Allocator, app: siv.App) !*Http {
    try globalInit();
    const self = try gpa.create(Http);
    errdefer gpa.destroy(self);
    const multi = c.curl_multi_init() orelse return error.HttpFailed;
    self.* = .{ .gpa = gpa, .app = app, .multi = multi, .timer = null, .transfers = .empty };
    _ = c.curl_multi_setopt(multi, c.mopt_socketfunction, &onSocket);
    _ = c.curl_multi_setopt(multi, c.mopt_socketdata, self);
    _ = c.curl_multi_setopt(multi, c.mopt_timerfunction, &onTimer);
    _ = c.curl_multi_setopt(multi, c.mopt_timerdata, self);
    _ = c.curl_multi_setopt(multi, c.mopt_max_host_connections, @as(c_long, 2));
    _ = c.curl_multi_setopt(multi, c.mopt_max_concurrent_streams, @as(c_long, 5));
    return self;
}

/// Drops the requests in flight without calling their callbacks.
pub fn deinit(self: *Http) void {
    if (self.timer) |t| siv.timerCancel(t);
    for (self.transfers.items) |t| {
        if (t.retry) |r| siv.timerCancel(r) else _ = c.curl_multi_remove_handle(self.multi, t.easy);
        t.destroy();
    }
    self.transfers.deinit(self.gpa);
    _ = c.curl_multi_cleanup(self.multi);
    self.gpa.destroy(self);
}

/// Starts a request; `done` hears how it went, never before this returns.
pub fn send(self: *Http, request: Request, context: *anyopaque, done: Done) Error!void {
    const t = try Transfer.create(self.gpa, request);
    errdefer t.destroy();
    try self.start(t, context, done);
}

/// Starts a GET whose reply goes on, such as server-sent events: `heard`
/// hears a 2xx reply's bytes as they come, and `done` how it ended, with
/// any other reply's body. One silent for `silent_s` seconds fails.
pub fn listen(self: *Http, request: Request, silent_s: u32, context: *anyopaque, heard: Heard, done: Done) Error!void {
    const t = try Transfer.create(self.gpa, request);
    errdefer t.destroy();
    t.heard = heard;
    // curl's speed is over the last few seconds, so a byte now and then,
    // such as a ping, keeps it above 1 B/s for long enough to reset this.
    try ok(c.curl_easy_setopt(t.easy, c.opt_low_speed_limit, @as(c_long, 1)));
    try ok(c.curl_easy_setopt(t.easy, c.opt_low_speed_time, @as(c_long, silent_s)));
    // A compressed stream may hold back what has come.
    try ok(c.curl_easy_setopt(t.easy, c.opt_accept_encoding, @as(?[*:0]const u8, null)));
    try self.start(t, context, done);
}

fn start(self: *Http, t: *Transfer, context: *anyopaque, done: Done) Error!void {
    t.context = context;
    t.done = done;
    t.http = self;
    try self.transfers.append(self.gpa, t);
    errdefer _ = self.transfers.pop();
    if (c.curl_multi_add_handle(self.multi, t.easy) != 0) return error.HttpFailed;
}

/// Makes a request and waits for it, for the command line, where there
/// is no event loop. `arena` holds the body.
pub fn fetch(arena: Allocator, request: Request) Error!Response {
    try globalInit();
    const t = try Transfer.create(arena, request);
    defer t.destroy();
    const code = c.curl_easy_perform(t.easy);
    if (code != 0) {
        log.err("{s}: {s}", .{ request.url, t.reason(code) });
        return error.HttpFailed;
    }
    return .{
        .status = t.status(),
        .url = try arena.dupe(u8, t.effectiveUrl()),
        .body = try arena.dupe(u8, t.body.items),
    };
}

fn globalInit() Error!void {
    // curl_global_init is not thread-safe; it is called before any thread
    // makes a request, and again harmlessly.
    if (c.curl_global_init(c.global_default) != 0) return error.HttpFailed;
}

/// The largest reply taken: Gmail's 25 MB attachment limit, in base64
/// and JSON.
const max_body = 40 << 20;

/// How long to wait before each retry of a request told to slow down:
/// about a minute in all, as Google's quotas are per minute.
const backoff_ms = [_]u32{ 1000, 2000, 4000, 8000, 16_000, 32_000 };

/// Whether a reply says to slow down: a 429, or a 403 for a rate limit
/// or a quota, as Google's APIs answer them.
fn slowDown(status: u16, body: []const u8) bool {
    return status == 429 or (status == 403 and
        (ascii.findIgnoreCase(body, "ratelimitexceeded") != null or
            ascii.findIgnoreCase(body, "quota exceeded") != null));
}

/// One request and what has come back of it.
const Transfer = struct {
    gpa: Allocator,
    /// The client sending it; unset for `fetch`.
    http: *Http,
    easy: *c.Easy,
    headers: ?*c.Slist,
    body: ArrayList(u8),
    context: *anyopaque,
    done: Done,
    /// Set for a reply that goes on, which hears its body as it comes.
    heard: ?Heard,
    error_buf: [c.error_size]u8,
    /// Retries so far.
    tries: u8,
    /// Set while it waits to be retried, off curl's books.
    retry: ?siv.Timer,

    fn create(gpa: Allocator, request: Request) Error!*Transfer {
        const t = try gpa.create(Transfer);
        errdefer gpa.destroy(t);
        const easy = c.curl_easy_init() orelse return error.HttpFailed;
        t.* = .{
            .gpa = gpa,
            .http = undefined,
            .easy = easy,
            .headers = null,
            .body = .empty,
            .context = undefined,
            .done = undefined,
            .heard = null,
            .error_buf = @splat(0),
            .tries = 0,
            .retry = null,
        };
        errdefer t.destroy();
        // curl copies strings it is given, but wants them NUL-terminated.
        var z: std.heap.ArenaAllocator = .init(gpa);
        defer z.deinit();
        const url = try z.allocator().dupeSentinel(u8, request.url, 0);
        for (request.headers) |h| {
            t.headers = c.curl_slist_append(t.headers, try z.allocator().dupeSentinel(u8, h, 0)) orelse
                return error.OutOfMemory;
        }
        try ok(c.curl_easy_setopt(easy, c.opt_url, url.ptr));
        try ok(c.curl_easy_setopt(easy, c.opt_private, t));
        try ok(c.curl_easy_setopt(easy, c.opt_writefunction, &onData));
        try ok(c.curl_easy_setopt(easy, c.opt_writedata, t));
        try ok(c.curl_easy_setopt(easy, c.opt_errorbuffer, &t.error_buf));
        try ok(c.curl_easy_setopt(easy, c.opt_nosignal, @as(c_long, 1)));
        try ok(c.curl_easy_setopt(easy, c.opt_accept_encoding, "".ptr));
        try ok(c.curl_easy_setopt(easy, c.opt_useragent, "parcel".ptr));
        try ok(c.curl_easy_setopt(easy, c.opt_connecttimeout_ms, @as(c_long, 15_000)));
        // Wait for a connection that can multiplex rather than open more.
        try ok(c.curl_easy_setopt(easy, c.opt_pipewait, @as(c_long, 1)));
        // Images a message shows come from anywhere, often by way of a
        // redirect: web addresses only, a few hops, and nothing huge.
        try ok(c.curl_easy_setopt(easy, c.opt_protocols_str, "http,https".ptr));
        try ok(c.curl_easy_setopt(easy, c.opt_redir_protocols_str, "http,https".ptr));
        try ok(c.curl_easy_setopt(easy, c.opt_followlocation, @as(c_long, 1)));
        try ok(c.curl_easy_setopt(easy, c.opt_maxredirs, @as(c_long, 5)));
        try ok(c.curl_easy_setopt(easy, c.opt_maxfilesize_large, @as(i64, max_body)));
        if (t.headers) |h| try ok(c.curl_easy_setopt(easy, c.opt_httpheader, h));
        if (request.method == .post or request.method == .put) {
            const body = request.body orelse "";
            try ok(c.curl_easy_setopt(easy, c.opt_postfieldsize, @as(c_long, @intCast(body.len))));
            try ok(c.curl_easy_setopt(easy, c.opt_copypostfields, body.ptr));
        }
        // A POST, under another name.
        if (request.method == .put) try ok(c.curl_easy_setopt(easy, c.opt_customrequest, "PUT".ptr));
        if (request.method == .delete) try ok(c.curl_easy_setopt(easy, c.opt_customrequest, "DELETE".ptr));
        return t;
    }

    fn destroy(t: *Transfer) void {
        c.curl_easy_cleanup(t.easy);
        c.curl_slist_free_all(t.headers);
        t.body.deinit(t.gpa);
        t.gpa.destroy(t);
    }

    /// Hands the request back to curl after a wait, if it has tries left.
    fn retryLater(t: *Transfer) bool {
        if (t.tries == backoff_ms.len) return false;
        const delay = backoff_ms[t.tries];
        t.retry = siv.timerStart(t.http.app, delay, onRetry, null, t) catch |err| {
            log.err("setting a retry's timer: {t}", .{err});
            return false;
        };
        t.tries += 1;
        log.info("asked to slow down; trying again in {d} s", .{delay / 1000});
        return true;
    }

    fn onRetry(user_data: ?*anyopaque) callconv(.c) void {
        const t: *Transfer = @ptrCast(@alignCast(user_data));
        t.retry = null;
        t.body.clearRetainingCapacity();
        if (c.curl_multi_add_handle(t.http.multi, t.easy) == 0) return;
        // As if curl had failed it.
        const http = t.http;
        if (mem.findScalar(*Transfer, http.transfers.items, t)) |i| _ = http.transfers.swapRemove(i);
        defer t.destroy();
        t.done(t.context, error.HttpFailed);
    }

    fn status(t: *Transfer) u16 {
        var code: c_long = 0;
        _ = c.curl_easy_getinfo(t.easy, c.info_response_code, &code);
        return std.math.cast(u16, code) orelse 0;
    }

    /// The URL curl ended at, which curl owns until the transfer goes.
    fn effectiveUrl(t: *Transfer) []const u8 {
        var effective: ?[*:0]const u8 = null;
        _ = c.curl_easy_getinfo(t.easy, c.info_effective_url, &effective);
        return if (effective) |e| mem.span(e) else "";
    }

    fn reason(t: *const Transfer, code: c_int) []const u8 {
        const detail = mem.sliceTo(&t.error_buf, 0);
        return if (detail.len > 0) detail else mem.span(c.curl_easy_strerror(code));
    }

    fn onData(ptr: [*]const u8, size: usize, count: usize, userdata: *anyopaque) callconv(.c) usize {
        const t: *Transfer = @ptrCast(@alignCast(userdata));
        const n = size * count;
        // Fewer bytes than given makes curl fail the transfer.
        if (t.body.items.len + n > max_body) return 0;
        t.body.appendSlice(t.gpa, ptr[0..n]) catch return 0;
        return n;
    }
};

fn ok(code: c_int) Error!void {
    if (code != 0) return error.HttpFailed;
}

/// A socket curl uses, and siv's watch on it.
const Socket = struct {
    http: *Http,
    fd: c_int,
    watch: siv.Watch,

    fn onReady(user_data: ?*anyopaque, _: siv.Watch, events: siv.WatchFlags) callconv(.c) void {
        const s: *Socket = @ptrCast(@alignCast(user_data));
        var mask: c_int = 0;
        if (events.readable) mask |= c.cselect_in;
        if (events.writable) mask |= c.cselect_out;
        if (events.@"error") mask |= c.cselect_err;
        // This may remove the watch, and free `s`: keep what is needed.
        const http = s.http;
        http.act(s.fd, mask);
    }

    fn destroy(user_data: ?*anyopaque) callconv(.c) void {
        const s: *Socket = @ptrCast(@alignCast(user_data));
        s.http.gpa.destroy(s);
    }
};

/// curl says which sockets to watch, and for what.
fn onSocket(_: *c.Easy, fd: c_int, what: c_int, userp: *anyopaque, socketp: ?*anyopaque) callconv(.c) c_int {
    const self: *Http = @ptrCast(@alignCast(userp));
    const existing: ?*Socket = @ptrCast(@alignCast(socketp));
    if (what == c.poll_remove) {
        if (existing) |s| siv.watchDestroy(s.watch);
        return 0;
    }
    const events: siv.WatchFlags = .{ .readable = what & c.poll_in != 0, .writable = what & c.poll_out != 0 };
    if (existing) |s| {
        siv.watchSetEvents(s.watch, events) catch |err| log.err("watching a socket: {t}", .{err});
        return 0;
    }
    const s = self.gpa.create(Socket) catch return -1;
    s.* = .{ .http = self, .fd = fd, .watch = undefined };
    s.watch = siv.appWatchFd(self.app, fd, events, Socket.onReady, Socket.destroy, s) catch |err| {
        log.err("watching a socket: {t}", .{err});
        self.gpa.destroy(s);
        return -1;
    };
    _ = c.curl_multi_assign(self.multi, fd, s);
    return 0;
}

/// curl says when to call it back if no socket is ready first.
fn onTimer(_: *c.Multi, timeout_ms: c_long, userp: *anyopaque) callconv(.c) c_int {
    const self: *Http = @ptrCast(@alignCast(userp));
    if (self.timer) |t| siv.timerCancel(t);
    self.timer = null;
    if (timeout_ms < 0) return 0;
    const delay: u32 = @intCast(@min(timeout_ms, std.math.maxInt(u32)));
    self.timer = siv.timerStart(self.app, delay, onTimeout, null, self) catch |err| {
        log.err("setting curl's timer: {t}", .{err});
        return -1;
    };
    return 0;
}

fn onTimeout(user_data: ?*anyopaque) callconv(.c) void {
    const self: *Http = @ptrCast(@alignCast(user_data));
    self.timer = null;
    self.act(c.socket_timeout, 0);
}

/// Lets curl work on a ready socket, or on its timeout, then hands out
/// the requests that finished.
fn act(self: *Http, fd: c_int, mask: c_int) void {
    var running: c_int = 0;
    _ = c.curl_multi_socket_action(self.multi, fd, mask, &running);
    // Out of curl's callbacks, so what hears it may send requests.
    var at: usize = 0;
    while (at < self.transfers.items.len) : (at += 1) {
        const t = self.transfers.items[at];
        const heard = t.heard orelse continue;
        if (t.body.items.len == 0 or t.status() / 100 != 2) continue;
        heard(t.context, t.body.items);
        t.body.clearRetainingCapacity();
    }
    var left: c_int = 0;
    while (c.curl_multi_info_read(self.multi, &left)) |msg| {
        if (msg.msg != c.msg_done) continue;
        var private: ?*anyopaque = null;
        _ = c.curl_easy_getinfo(msg.easy, c.info_private, &private);
        const t: *Transfer = @ptrCast(@alignCast(private.?));
        const code = msg.data.result;
        _ = c.curl_multi_remove_handle(self.multi, t.easy);
        if (code == 0 and slowDown(t.status(), t.body.items) and t.retryLater()) continue;
        if (mem.findScalar(*Transfer, self.transfers.items, t)) |i| _ = self.transfers.swapRemove(i);
        defer t.destroy();
        if (code != 0) {
            log.warn("{s}", .{t.reason(code)});
            t.done(t.context, error.HttpFailed);
        } else {
            t.done(t.context, .{ .status = t.status(), .url = t.effectiveUrl(), .body = t.body.items });
        }
    }
}

/// The parts of curl.h used here.
const c = struct {
    const Easy = opaque {};
    const Multi = opaque {};
    const Slist = opaque {};
    const Msg = extern struct {
        msg: c_int,
        easy: *Easy,
        data: extern union { whatever: ?*anyopaque, result: c_int },
    };

    const global_default: c_long = 3;
    const error_size = 256;

    const opt_writedata: c_int = 10001;
    const opt_url: c_int = 10002;
    const opt_errorbuffer: c_int = 10010;
    const opt_writefunction: c_int = 20011;
    const opt_useragent: c_int = 10018;
    const opt_httpheader: c_int = 10023;
    const opt_customrequest: c_int = 10036;
    const opt_postfieldsize: c_int = 60;
    const opt_nosignal: c_int = 99;
    const opt_accept_encoding: c_int = 10102;
    const opt_private: c_int = 10103;
    const opt_connecttimeout_ms: c_int = 156;
    const opt_copypostfields: c_int = 10165;
    const opt_pipewait: c_int = 237;
    const opt_followlocation: c_int = 52;
    const opt_maxredirs: c_int = 68;
    const opt_low_speed_limit: c_int = 19;
    const opt_low_speed_time: c_int = 20;
    const opt_maxfilesize_large: c_int = 30117;
    const opt_protocols_str: c_int = 10318;
    const opt_redir_protocols_str: c_int = 10319;

    const mopt_socketfunction: c_int = 20001;
    const mopt_socketdata: c_int = 10002;
    const mopt_timerfunction: c_int = 20004;
    const mopt_timerdata: c_int = 10005;
    const mopt_max_host_connections: c_int = 7;
    const mopt_max_concurrent_streams: c_int = 16;

    const info_response_code: c_int = 0x200002;
    const info_private: c_int = 0x100015;
    const info_effective_url: c_int = 0x100001;

    const socket_timeout: c_int = -1;
    const cselect_in: c_int = 1;
    const cselect_out: c_int = 2;
    const cselect_err: c_int = 4;
    const poll_in: c_int = 1;
    const poll_out: c_int = 2;
    const poll_remove: c_int = 4;
    const msg_done: c_int = 1;

    extern fn curl_global_init(flags: c_long) c_int;
    extern fn curl_easy_init() ?*Easy;
    extern fn curl_easy_cleanup(easy: *Easy) void;
    extern fn curl_easy_setopt(easy: *Easy, option: c_int, ...) c_int;
    extern fn curl_easy_getinfo(easy: *Easy, info: c_int, ...) c_int;
    extern fn curl_easy_perform(easy: *Easy) c_int;
    extern fn curl_easy_strerror(code: c_int) [*:0]const u8;
    extern fn curl_slist_append(list: ?*Slist, string: [*:0]const u8) ?*Slist;
    extern fn curl_slist_free_all(list: ?*Slist) void;
    extern fn curl_multi_init() ?*Multi;
    extern fn curl_multi_cleanup(multi: *Multi) c_int;
    extern fn curl_multi_setopt(multi: *Multi, option: c_int, ...) c_int;
    extern fn curl_multi_add_handle(multi: *Multi, easy: *Easy) c_int;
    extern fn curl_multi_remove_handle(multi: *Multi, easy: *Easy) c_int;
    extern fn curl_multi_assign(multi: *Multi, fd: c_int, data: ?*anyopaque) c_int;
    extern fn curl_multi_socket_action(multi: *Multi, fd: c_int, mask: c_int, running: *c_int) c_int;
    extern fn curl_multi_info_read(multi: *Multi, left: *c_int) ?*Msg;
};

test "a request told to slow down is sent again, and its callback hears only the answer" {
    var address: Io.net.IpAddress = .{ .ip4 = .loopback(0) };
    var server = try address.listen(testing.io, .{});
    defer server.deinit(testing.io);
    var serving = try testing.io.concurrent(slowThenFine, .{ &server, testing.io });
    defer serving.cancel(testing.io);

    const app = try siv.appCreate(.{
        .backend = .headless,
        .flags = .{ .no_system_fonts = true, .no_system_icons = true, .reduced_motion = true },
    });
    defer siv.appDestroy(app);
    const http = try init(testing.allocator, app);
    defer http.deinit();
    var url_buf: [64]u8 = undefined;
    const url = try mem.print(&url_buf, "http://127.0.0.1:{d}/", .{server.socket.address.getPort()});
    const Answer = struct { calls: u32 = 0, status: u16 = 0, body: [16]u8 = undefined, len: usize = 0 };
    var heard: Answer = .{};
    try http.send(.{ .method = .get, .url = url, .headers = &.{}, .body = null }, &heard, struct {
        fn done(context: *anyopaque, result: Error!Response) void {
            const h: *Answer = @ptrCast(@alignCast(context));
            h.calls += 1;
            const r = result catch return;
            h.status = r.status;
            h.len = @min(r.body.len, h.body.len);
            @memcpy(h.body[0..h.len], r.body[0..h.len]);
        }
    }.done);
    var waited: u32 = 0;
    while (heard.calls == 0 and waited < 50) : (waited += 1) try siv.appDispatch(app, 100);
    try testing.expectEqual(1, heard.calls);
    try testing.expectEqual(200, heard.status);
    try testing.expectEqualStrings("fine", heard.body[0..heard.len]);
}

/// Answers its first request with Gmail's quota refusal, then "fine".
fn slowThenFine(server: *Io.net.Server, io: Io) void {
    var answered: u32 = 0;
    while (true) {
        const stream = server.accept(io) catch |err| switch (err) {
            error.Canceled => return,
            else => continue,
        };
        defer stream.close(io);
        var read_buf: [4096]u8 = undefined;
        var write_buf: [4096]u8 = undefined;
        var reader = stream.reader(io, &read_buf);
        var writer = stream.writer(io, &write_buf);
        var server_http: std.http.Server = .init(&reader.interface, &writer.interface);
        while (true) {
            var request = server_http.receiveHead() catch break;
            const refused =
                \\{"error":{"code":403,"message":"Quota exceeded for quota metric 'Total Query Cost'",
                \\"errors":[{"reason":"rateLimitExceeded"}]}}
            ;
            const first = answered == 0;
            answered += 1;
            request.respond(if (first) refused else "fine", .{ .status = if (first) .forbidden else .ok }) catch break;
        }
    }
}