path: root/src/server/main.zig

const std = @import("std");
const Message = @import("./message_parser.zig").Message;
pub const ServerInfo = Message.ServerInfo;

const ClientState = @import("./client.zig").ClientState;
const Server = @This();

const Subscription = struct {
    subject: []const u8,
    client_id: usize,
    sid: []const u8,
};

info: ServerInfo,
clients: std.AutoHashMapUnmanaged(usize, *ClientState) = .empty,

subs_lock: std.Io.Mutex = .init,
subscriptions: std.ArrayList(Subscription) = .empty,

var keep_running = std.atomic.Value(bool).init(true);

fn handleSigInt(sig: std.os.linux.SIG) callconv(.c) void {
    _ = sig;
    keep_running.store(false, .monotonic);
}

pub fn main(gpa: std.mem.Allocator, server_config: ServerInfo) !void {
    // Configure the signal action
    // const act = std.posix.Sigaction{
    //     .handler = .{ .handler = handleSigInt },
    //     .mask = std.posix.sigemptyset(),
    //     .flags = 0,
    // };

    // // Register the handler for SIGINT (Ctrl+C)
    // std.posix.sigaction(std.posix.SIG.INT, &act, null);

    var server: Server = .{
        .info = server_config,
    };

    var threaded: std.Io.Threaded = .init(gpa, .{});
    defer threaded.deinit();
    const io = threaded.io();


    var tcp_server = try std.Io.net.IpAddress.listen(try std.Io.net.IpAddress.parse(
        server.info.host,
        server.info.port,
    ), io, .{ .reuse_address = true });
    defer tcp_server.deinit(io);

    var id: usize = 0;
    // Run until SIGINT is handled, then exit gracefully
    while (keep_running.load(.monotonic)) : (id +%= 1) {
        std.debug.print("in server loop\n", .{});
        if (server.clients.contains(id)) continue;
        const stream = try tcp_server.accept(io);
        std.debug.print("accepted connection\n", .{});
        _ = io.concurrent(handleConnection, .{ &server, gpa, io, id, stream }) catch {
            std.debug.print("could not start concurrent handler for {d}\n", .{id});
            stream.close(io);
        };
    }

    std.debug.print("Exiting gracefully\n", .{});
}

fn processMsgs(server: *Server, io: std.Io, alloc: std.mem.Allocator) void {
    while (true) {
        const msg = server.msg_queue.getOne(io) catch break;
        defer msg.deinit(alloc);

        for (server.subscriptions.items) |subscription| {
            if (subjectMatches(subscription.subject, msg.subject)) {
                const client = server.clients.get(subscription.client_id) orelse {
                    std.debug.print("trying to publish to a client that no longer exists: {d}", .{subscription.client_id});
                    continue;
                };
                client.send(io, .{ .msg = .{
                    .subject = msg.subject,
                    .sid = subscription.sid,
                    .reply_to = msg.reply_to,
                    .payload = msg.payload,
                } }) catch continue;
            }
        }
    }
}

fn addClient(server: *Server, allocator: std.mem.Allocator, id: usize, client: *ClientState) !void {
    // server.clients.lockPointers();
    try server.clients.put(allocator, id, client);
    // server.clients.unlockPointers();
}

fn removeClient(server: *Server, io: std.Io, allocator: std.mem.Allocator, id: usize) void {
    server.subs_lock.lockUncancelable(io);
    defer server.subs_lock.unlock(io);
    _ = server.clients.remove(id);
    const len = server.subscriptions.items.len;
    for (0..len) |i| {
        const sub = server.subscriptions.items[len - i - 1];
        if (sub.client_id == id) {
            allocator.free(sub.sid);
            allocator.free(sub.subject);
            _ = server.subscriptions.swapRemove(i);
        }
    }
}

fn handleConnection(
    server: *Server,
    server_allocator: std.mem.Allocator,
    io: std.Io,
    id: usize,
    stream: std.Io.net.Stream,
) !void {
    var client_allocator: std.heap.DebugAllocator(.{}) = .init;
    client_allocator.backing_allocator = server_allocator;
    defer _ = client_allocator.deinit();

    const allocator = client_allocator.allocator();
    defer stream.close(io);
    const w_buffer: []u8 = try allocator.alloc(u8, 1024);
    defer allocator.free(w_buffer);
    var writer = stream.writer(io, w_buffer);
    const out = &writer.interface;

    const r_buffer: []u8 = try allocator.alloc(u8, 1024);
    defer allocator.free(r_buffer);
    var reader = stream.reader(io, r_buffer);
    const in = &reader.interface;

    var client_state: ClientState = .init(null, in, out);
    try client_state.send(io, .{ .info = server.info });

    var connect_arena: std.heap.ArenaAllocator = .init(allocator);
    defer connect_arena.deinit();
    client_state.connect = (Message.next(connect_arena.allocator(), in) catch return).connect;

    try server.addClient(server_allocator, id, &client_state);
    defer server.removeClient(io, server_allocator, id);

    // Messages are owned by the server after they are received from the client
    while (client_state.next(server_allocator)) |msg| {
        switch (msg) {
            .ping => {
                // Respond to ping with pong.
                try client_state.send(io, .pong);
            },
            .@"pub" => |pb| {
                _ = io.async(publishMessage, .{ server, io, server_allocator, &client_state, pb });
            },
            .sub => |sub| {
                try server.subscribe(io, server_allocator, id, sub);
            },
            .unsub => |unsub| {
                try server.unsubscribe(io, server_allocator, id, unsub);
            },
            else => |e| {
                std.debug.panic("Unimplemented message: {any}\n", .{e});
            },
        }
    } else |err| {
        // This is probably going to be normal on disconnect
        std.debug.print("Ran into error in client process loop: {}\n", .{err});
    }

    // client_state.task.await(io);
}

// // Result is owned by the caller
// fn subscribers(server: *Server, gpa: std.mem.Allocator, subject: []const u8) []ClientState {
//     var acc: std.ArrayList(ClientState) = .empty;

//     return acc.toOwnedSlice();
// }

fn subjectMatches(expected: []const u8, actual: []const u8) bool {
    return std.mem.eql(u8, expected, actual);
}

fn publishMessage(server: *Server, io: std.Io, gpa: std.mem.Allocator, source_client: *ClientState, msg: Message.Pub) !void {
    errdefer {
        if (source_client.connect) |c| {
            if (c.verbose) {
                source_client.send(io, .@"-err") catch {};
            }
        }
    }
    defer msg.deinit(gpa);
    for (server.subscriptions.items) |subscription| {
        if (subjectMatches(subscription.subject, msg.subject)) {
            const client = server.clients.get(subscription.client_id) orelse {
                std.debug.print("trying to publish to a client that no longer exists: {d}", .{subscription.client_id});
                continue;
            };
            client.send(io, .{ .msg = .{
                .subject = msg.subject,
                .sid = subscription.sid,
                .reply_to = msg.reply_to,
                .payload = msg.payload,
            } }) catch continue;
        }
    }
    if (source_client.connect) |c| {
        if (c.verbose) {
            source_client.send(io, .@"+ok") catch {};
        }
    }
}

fn subscribe(server: *Server, io: std.Io, gpa: std.mem.Allocator, id: usize, msg: Message.Sub) !void {
    try server.subs_lock.lock(io);
    defer server.subs_lock.unlock(io);
    try server.subscriptions.append(gpa, .{
        .subject = msg.subject,
        .client_id = id,
        .sid = msg.sid,
    });
}

fn unsubscribe(server: *Server, io: std.Io, gpa: std.mem.Allocator, id: usize, msg: Message.Unsub) !void {
    try server.subs_lock.lock(io);
    defer server.subs_lock.unlock(io);
    const len = server.subscriptions.items.len;
    for (0..len) |i| {
        const sub = server.subscriptions.items[len - i - 1];
        if (sub.client_id == id and std.mem.eql(u8, sub.sid, msg.sid)) {
            gpa.free(sub.sid);
            gpa.free(sub.subject);
            _ = server.subscriptions.swapRemove(i);
        }
    }
}

pub fn createId() []const u8 {
    return "SERVERID";
}

pub fn createName() []const u8 {
    return "SERVERNAME";
}

// TESTING

// fn initTestServer() Server {
//     return .{
//         .info = .{
//             .server_id = "ABCD",
//             .server_name = "test server",
//             .version = "0.1.2",
//             .max_payload = 1234,
//         },
//     };
// }

// fn initTestClient(
//     io: std.Io,
//     allocator: std.mem.Allocator,
//     id: usize,
//     data_from: []const u8,
// ) !struct {
//     Client,
//     *std.Io.Reader,
//     *std.Io.Writer,
// } {
//     return .init(io, allocator, id, .{}, in, out);
// }

// test {
//     const gpa = std.testing.allocator;
//     const io = std.testing.io;

//     const server = initTestServer();
//     const client: Client = .init(
//         io,
//         gpa,
//         1,
//         .{},
//     );
// }