/core/src/connection.zig at 7120e027fa60e17d7d1d65cc60efcdf21c56d92c

// Copyright 2025 Shota FUJI
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// SPDX-License-Identifier: Apache-2.0

const std = @import("std");

const moo = @import("moo");
const websocket = @import("websocket");

const Arc = @import("./Arc.zig");
const discovery = @import("./discovery.zig");
const extension = @import("./extension.zig").extension;
const ping = @import("./services/ping.zig");
const registry = @import("./services/registry.zig");
const TransportService = @import("./services/transport.zig").TransportService;
const transport = @import("./transport.zig");

pub const ConnectionError = enum(c_int) {
    unknown = 0,
    closed_by_server = 1,
    out_of_memory = 2,
    unexpected_response = 3,
    network_unavailable = 4,
    network_error = 5,
};

pub const ConnectionErrorEvent = extern struct {
    const cname = "plac_connection_connection_error_event";
    const allocator = std.heap.c_allocator;

    internal: *Internal,
    code: ConnectionError,

    const Internal = struct {
        arc: Arc = .{},
    };

    pub fn make(code: ConnectionError) std.mem.Allocator.Error!*ConnectionErrorEvent {
        const internal = try allocator.create(Internal);
        errdefer allocator.destroy(internal);
        internal.* = .{};

        const self = try allocator.create(ConnectionErrorEvent);
        errdefer allocator.destroy(self);

        self.* = .{
            .internal = internal,
            .code = code,
        };

        return self;
    }

    pub fn retain(ptr: ?*ConnectionErrorEvent) callconv(.C) *ConnectionErrorEvent {
        var self = ptr orelse @panic(
            std.fmt.comptimePrint("Received null pointer on {s}_{s}", .{ cname, @src().fn_name }),
        );
        self.internal.arc.ref();
        return self;
    }

    pub fn release(ptr: ?*ConnectionErrorEvent) callconv(.C) void {
        var self = ptr orelse @panic(
            std.fmt.comptimePrint("Received null pointer on {s}_{s}", .{ cname, @src().fn_name }),
        );
        if (self.internal.arc.unref()) {
            std.log.debug("Releasing {*}...", .{self});

            allocator.destroy(self.internal);
            allocator.destroy(self);
        }
    }

    pub fn export_capi() void {
        @export(&retain, .{ .name = std.fmt.comptimePrint("{s}_retain", .{cname}) });
        @export(&release, .{ .name = std.fmt.comptimePrint("{s}_release", .{cname}) });
    }
};

pub const ConnectedEvent = extern struct {
    const cname = "plac_connection_connected_event";
    const allocator = std.heap.c_allocator;

    internal: *Internal,
    token: [*:0]const u8,

    pub const Internal = struct {
        token: [:0]const u8,
        arc: Arc = .{},
    };

    /// This function takes ownership of `token`.
    pub fn make(token: [:0]const u8) std.mem.Allocator.Error!*ConnectedEvent {
        const internal = try allocator.create(Internal);
        errdefer allocator.destroy(internal);

        internal.* = .{
            .token = token,
        };

        const self = try allocator.create(ConnectedEvent);
        errdefer allocator.destroy(self);

        self.* = .{
            .internal = internal,
            .token = token.ptr,
        };

        return self;
    }

    pub fn retain(ptr: ?*ConnectedEvent) callconv(.C) *ConnectedEvent {
        var self = ptr orelse @panic(
            std.fmt.comptimePrint("Received null pointer on {s}_{s}", .{ cname, @src().fn_name }),
        );
        self.internal.arc.ref();
        return self;
    }

    pub fn release(ptr: ?*ConnectedEvent) callconv(.C) void {
        var self = ptr orelse @panic(
            std.fmt.comptimePrint("Received null pointer on {s}_{s}", .{ cname, @src().fn_name }),
        );
        if (self.internal.arc.unref()) {
            std.log.debug("Releasing {*}...", .{self});

            allocator.free(self.internal.token);
            allocator.destroy(self.internal);
            allocator.destroy(self);
        }
    }

    pub fn export_capi() void {
        @export(&retain, .{ .name = std.fmt.comptimePrint("{s}_retain", .{cname}) });
        @export(&release, .{ .name = std.fmt.comptimePrint("{s}_release", .{cname}) });
    }
};

pub const Event = extern struct {
    const cname = "plac_connection_event";
    const allocator = std.heap.c_allocator;

    pub const Kind = enum(c_int) {
        connection_error = 0,
        connected = 1,
        zone_list = 10,
    };

    internal: *Internal,
    kind: Kind,

    pub const Internal = struct {
        const Payload = union(Kind) {
            connection_error: *ConnectionErrorEvent,
            connected: *ConnectedEvent,
            zone_list: *transport.ZoneListEvent,
        };

        payload: Payload,

        arc: Arc = .{},
    };

    pub fn makeConnectionError(err: anyerror) std.mem.Allocator.Error!*Event {
        const internal = try allocator.create(Internal);
        errdefer allocator.destroy(internal);

        const connection_error = try ConnectionErrorEvent.make(switch (err) {
            error.OutOfMemory => .out_of_memory,
            error.NetworkUnavailable => .network_unavailable,
            error.SocketError, error.SocketPermissionDenied => .network_error,
            error.RequestIdMismatch => .unexpected_response,
            error.ClosedByServer, error.ReadClosedConnection => .closed_by_server,
            else => .unknown,
        });
        errdefer connection_error.release();

        internal.* = .{
            .payload = .{ .connection_error = connection_error.retain() },
        };

        const self = try allocator.create(Event);
        errdefer allocator.destroy(self);

        self.* = .{
            .internal = internal,
            .kind = .connection_error,
        };

        return self;
    }

    /// This function takes ownership of `token`.
    pub fn makeConnected(token: [:0]const u8) std.mem.Allocator.Error!*Event {
        const internal = try allocator.create(Internal);
        errdefer allocator.destroy(internal);

        const connected = try ConnectedEvent.make(token);
        errdefer connected.release();

        internal.* = .{
            .payload = .{ .connected = connected.retain() },
        };

        const self = try allocator.create(Event);
        errdefer allocator.destroy(self);

        self.* = .{
            .internal = internal,
            .kind = .connected,
        };

        return self;
    }

    pub fn makeZoneListFromInitial(
        res: *const TransportService.SubscribeZoneChanges.InitialResponse,
    ) std.mem.Allocator.Error!*Event {
        const internal = try allocator.create(Internal);
        errdefer allocator.destroy(internal);

        const zone_list = try transport.ZoneListEvent.makeFromInitial(res);
        errdefer zone_list.release();

        internal.* = .{
            .payload = .{ .zone_list = zone_list.retain() },
        };

        const self = try allocator.create(Event);
        errdefer allocator.destroy(self);

        self.* = .{
            .internal = internal,
            .kind = .zone_list,
        };

        return self;
    }

    pub fn makeZoneListFromChanges(
        res: *const TransportService.SubscribeZoneChanges.Response,
    ) std.mem.Allocator.Error!*Event {
        const internal = try allocator.create(Internal);
        errdefer allocator.destroy(internal);

        const zone_list = try transport.ZoneListEvent.makeFromChanges(res);
        errdefer zone_list.release();

        internal.* = .{
            .payload = .{ .zone_list = zone_list.retain() },
        };

        const self = try allocator.create(Event);
        errdefer allocator.destroy(self);

        self.* = .{
            .internal = internal,
            .kind = .zone_list,
        };

        return self;
    }

    pub fn retain(ptr: ?*Event) callconv(.C) *Event {
        var self = ptr orelse @panic(
            std.fmt.comptimePrint("Received null pointer on {s}_{s}", .{ cname, @src().fn_name }),
        );
        self.internal.arc.ref();
        return self;
    }

    pub fn release(ptr: ?*Event) callconv(.C) void {
        var self = ptr orelse @panic(
            std.fmt.comptimePrint("Received null pointer on {s}_{s}", .{ cname, @src().fn_name }),
        );
        if (self.internal.arc.unref()) {
            std.log.debug("Releasing {*}...", .{self});

            switch (self.internal.payload) {
                .connection_error => |ev| ev.release(),
                .connected => |ev| ev.release(),
                .zone_list => |ev| ev.release(),
            }

            allocator.destroy(self.internal);
            allocator.destroy(self);
        }
    }

    fn exportEventCastFunction(comptime TargetEvent: type, comptime kind: Kind) void {
        const Caster = struct {
            pub const fn_name = std.fmt.comptimePrint("{s}_get_{s}_event", .{ cname, @tagName(kind) });

            pub fn cast(ptr: ?*Event) callconv(.C) *TargetEvent {
                const self = ptr orelse @panic(
                    std.fmt.comptimePrint("Received null pointer on {s}", .{fn_name}),
                );

                if (self.internal.payload == kind) {
                    return @field(self.internal.payload, @tagName(kind)).retain();
                }

                std.log.err("{s} called on {s}", .{ fn_name, @tagName(self.internal.payload) });
                unreachable;
            }
        };

        @export(&Caster.cast, .{ .name = Caster.fn_name });
    }

    pub fn export_capi() void {
        @export(&retain, .{ .name = std.fmt.comptimePrint("{s}_retain", .{cname}) });
        @export(&release, .{ .name = std.fmt.comptimePrint("{s}_release", .{cname}) });

        Event.exportEventCastFunction(ConnectionErrorEvent, .connection_error);
        Event.exportEventCastFunction(ConnectedEvent, .connected);
        Event.exportEventCastFunction(transport.ZoneListEvent, .zone_list);
    }
};

pub const Connection = extern struct {
    const cname = "plac_connection";
    const allocator = std.heap.c_allocator;

    internal: *Internal,

    pub const Internal = struct {
        server: *discovery.Server,
        ws: ?websocket.Client = null,
        request_id: i64 = 0,
        subscription_id: u64 = 0,
        host: []const u8,
        zone_subscription_request_id: ?i64 = null,
        arc: Arc = .{},

        fn init(server: *discovery.Server) !Internal {
            var addr = std.ArrayList(u8).init(allocator);
            defer addr.deinit();

            try addr.writer().print("{}", .{server.internal.address});

            const addr_string = try addr.toOwnedSlice();
            defer allocator.free(addr_string);

            // Zig std always prints "<addr>:<port>" for IPv4 and IPv6
            const port_start = std.mem.lastIndexOfScalar(u8, addr_string, ':') orelse {
                unreachable;
            };

            const host = try allocator.dupe(u8, addr_string[0..port_start]);
            errdefer allocator.free(host);

            return .{
                .server = server.retain(),
                .host = host,
            };
        }

        fn deinit(self: *Internal) void {
            if (self.ws) |*ws| {
                ws.deinit();
            }

            allocator.free(self.host);
            self.server.release();
        }
    };

    pub fn make(server_ptr: ?*discovery.Server) callconv(.C) ?*Connection {
        const server = server_ptr orelse @panic(
            std.fmt.comptimePrint("Received null pointer on {s}_{s}", .{ cname, @src().fn_name }),
        );

        const self = allocator.create(Connection) catch return null;

        const internal = allocator.create(Internal) catch {
            allocator.destroy(self);
            return null;
        };

        internal.* = Internal.init(server) catch |err| {
            std.log.err("Unable to establish connection to Roon Server: {s}", .{@errorName(err)});
            allocator.destroy(internal);
            allocator.destroy(self);
            return null;
        };

        self.* = .{ .internal = internal };

        internal.arc.ref();

        return self;
    }

    pub fn retain(ptr: ?*Connection) callconv(.C) *Connection {
        var self = ptr orelse @panic(
            std.fmt.comptimePrint("Received null pointer on {s}_{s}", .{ cname, @src().fn_name }),
        );
        self.internal.arc.ref();
        return self;
    }

    pub fn release(ptr: ?*Connection) callconv(.C) void {
        var self = ptr orelse @panic(
            std.fmt.comptimePrint("Received null pointer on {s}_{s}", .{ cname, @src().fn_name }),
        );
        if (self.internal.arc.unref()) {
            std.log.debug("Releasing {*}...", .{self});

            self.internal.deinit();
            allocator.destroy(self.internal);
            allocator.destroy(self);
        }
    }

    pub fn getEvent(ptr: ?*Connection) callconv(.C) ?*Event {
        var self = ptr orelse @panic(
            std.fmt.comptimePrint("Received null pointer on {s}_{s}", .{ cname, @src().fn_name }),
        );

        var ws = self.internal.ws orelse {
            // TODO: Pass saved token
            const new_token = self.connect(null) catch |err| {
                std.log.err("Unable to connect: {s}", .{@errorName(err)});
                const event = Event.makeConnectionError(err) catch |make_err| {
                    std.log.err("Unable to compose {s} event: {s}", .{
                        @tagName(Event.Kind.connection_error),
                        @errorName(make_err),
                    });
                    return null;
                };
                return event.retain();
            };

            const event = Event.makeConnected(new_token) catch |err| {
                std.log.err("Unable to compose {s} event: {s}", .{
                    @tagName(Event.Kind.connected),
                    @errorName(err),
                });
                return null;
            };
            return event.retain();
        };

        while (true) {
            const meta, const header_ctx, const msg = readMessage(&ws) catch |err| {
                std.log.err("Failed to read a message: {s}", .{@errorName(err)});
                const event = Event.makeConnectionError(err) catch |make_err| {
                    std.log.err("Unable to compose {s} event: {s}", .{
                        @tagName(Event.Kind.connection_error),
                        @errorName(make_err),
                    });
                    return null;
                };
                return event.retain();
            };
            defer ws.done(msg);

            const header: moo.NoBodyHeaders, _ = moo.NoBodyHeaders.parse(msg.data, header_ctx) catch |err| {
                std.log.err("Invalid MOO message header: {s}", .{@errorName(err)});
                continue;
            };

            if (self.internal.zone_subscription_request_id) |req_id| {
                if (header.request_id == req_id) {
                    if (std.mem.eql(u8, meta.service, "Subscribed")) {
                        const res = TransportService.SubscribeZoneChanges.initialResponse(
                            allocator,
                            header_ctx,
                            msg.data,
                        ) catch |err| {
                            std.log.err(
                                "Received unexpected zone subscription response: {s}",
                                .{@errorName(err)},
                            );
                            continue;
                        };
                        defer res.deinit();

                        const event = Event.makeZoneListFromInitial(res.value) catch |make_err| {
                            std.log.err("Unable to compose {s} event: {s}", .{
                                @tagName(Event.Kind.zone_list),
                                @errorName(make_err),
                            });
                            return null;
                        };
                        return event.retain();
                    }

                    if (std.mem.eql(u8, meta.service, "Changed")) {
                        const res = TransportService.SubscribeZoneChanges.response(
                            allocator,
                            header_ctx,
                            msg.data,
                        ) catch |err| {
                            std.log.err(
                                "Received unexpected zone change response: {s}",
                                .{@errorName(err)},
                            );
                            continue;
                        };
                        defer res.deinit();

                        const event = Event.makeZoneListFromChanges(res.value) catch |err| {
                            std.log.err("Unable to compose {s} event: {s}", .{
                                @tagName(Event.Kind.zone_list),
                                @errorName(err),
                            });
                            return null;
                        };
                        return event.retain();
                    }

                    if (std.mem.eql(u8, meta.service, "Unsubscribed")) {
                        self.internal.zone_subscription_request_id = null;
                        // TODO: Return unsubscribed event
                        continue;
                    }

                    std.log.warn("Unknown response received for zone subscription: {s}", .{
                        meta.service,
                    });
                    continue;
                }
            }

            std.log.debug("Got message on {s}", .{meta.service});

            // TODO: Return an event
        }
    }

    const ReadError = error{
        ClosedByServer,
        ReadClosedConnection,
        RequestIdMismatch,
    };

    /// Returns a token. Caller owns the returned memory.
    fn connect(self: *Connection, token: ?[]const u8) ![:0]const u8 {
        std.log.debug(
            "Establishing WebSocket connection to {}...",
            .{self.internal.server.internal.address},
        );

        var ws = try websocket.Client.init(allocator, .{
            .host = self.internal.host,
            .port = self.internal.server.internal.address.getPort(),
        });
        errdefer ws.deinit();

        std.log.debug("Performing WebSocket handshake...", .{});
        try ws.handshake("/api", .{ .timeout_ms = 1_000 });

        var request_id: i64 = 0;

        {
            {
                std.log.debug("Checking server status...", .{});
                const req = try registry.RegistryService.Info.request(allocator, request_id);
                defer allocator.free(req);
                try ws.writeBin(req);
            }

            _, const header_ctx, const msg = try readMessage(&ws);
            defer ws.done(msg);

            const header: moo.NoBodyHeaders, _ = try moo.NoBodyHeaders.parse(msg.data, header_ctx);
            if (header.request_id != request_id) {
                return ReadError.RequestIdMismatch;
            }

            const res = try registry.RegistryService.Info.response(
                allocator,
                header_ctx,
                msg.data,
            );
            defer res.deinit();

            request_id += 1;
        }

        const new_token = new_token: {
            {
                std.log.debug("Registering extension...", .{});
                const req = try registry.RegistryService.Register.request(
                    allocator,
                    request_id,
                    &extension,
                    token,
                );
                defer allocator.free(req);
                try ws.writeBin(req);
            }

            const meta, const header_ctx, const msg = try readMessage(&ws);
            defer ws.done(msg);

            const header: moo.NoBodyHeaders, _ = try moo.NoBodyHeaders.parse(msg.data, header_ctx);
            if (header.request_id != request_id) {
                return ReadError.RequestIdMismatch;
            }

            const res = try registry.RegistryService.Register.response(
                allocator,
                &meta,
                header_ctx,
                msg.data,
            );
            defer res.deinit();

            request_id += 1;

            break :new_token try allocator.dupeZ(u8, res.value.token);
        };

        self.internal.request_id = request_id;
        self.internal.ws = ws;

        return new_token;
    }

    /// Caller is responsible for closing message by calling `ws.done()`.
    fn readMessage(ws: *websocket.Client) !struct {
        moo.Metadata,
        moo.HeaderParsingContext,
        websocket.Message,
    } {
        while (true) {
            const msg = ws.read() catch |err| {
                switch (err) {
                    error.Closed => return ReadError.ReadClosedConnection,
                    else => {
                        std.log.warn("Unable to read WebSocket message: {s}", .{@errorName(err)});
                        continue;
                    },
                }
            } orelse unreachable;

            switch (msg.type) {
                // NOTE: roon-node-api does not check whether message is binaryType.
                .text, .binary => {
                    const meta, const header_ctx = moo.Metadata.parse(msg.data) catch |err| {
                        std.log.warn("Failed to parse MOO metadata: {s}", .{@errorName(err)});
                        continue;
                    };

                    if (std.mem.eql(u8, ping.PingService.ping_id, meta.service)) {
                        writePong(ws, header_ctx, msg.data) catch |err| {
                            std.log.warn(
                                "Failed to respond to ping request: {s}",
                                .{@errorName(err)},
                            );
                        };
                        continue;
                    }

                    return .{ meta, header_ctx, msg };
                },
                .ping => ws.writePong(msg.data) catch |err| {
                    std.log.warn("Failed to respond to ping: {s}", .{@errorName(err)});
                },
                .pong => {},
                .close => return ReadError.ClosedByServer,
            }
        }
    }

    fn writePong(
        ws: *websocket.Client,
        header_ctx: moo.HeaderParsingContext,
        message: []const u8,
    ) !void {
        var buffer = std.ArrayList(u8).init(allocator);
        defer buffer.deinit();

        try ping.PingService.ping(buffer.writer(), header_ctx, message);

        try ws.writeBin(buffer.items);
    }

    pub fn subscribeZones(ptr: ?*Connection) callconv(.C) void {
        var self = ptr orelse @panic(
            std.fmt.comptimePrint("Received null pointer on {s}_{s}", .{ cname, @src().fn_name }),
        );

        var ws = self.internal.ws orelse {
            std.log.err("{s}_{s} called, but WebSocket connection is not ready", .{ cname, @src().fn_name });
            return;
        };

        const req_id = self.internal.request_id;
        self.internal.request_id += 1;

        const sub_id = self.internal.subscription_id;
        self.internal.subscription_id += 1;

        {
            std.log.debug("Subscribing to zone changes...", .{});
            const req = TransportService.SubscribeZoneChanges.request(
                allocator,
                req_id,
                sub_id,
            ) catch |err| {
                std.log.err("Unable to compose zone subscription request: {s}", .{@errorName(err)});
                return;
            };
            defer allocator.free(req);

            ws.writeBin(req) catch |err| {
                std.log.err("Unable to write subscription request: {s}", .{@errorName(err)});
                return;
            };
        }

        self.internal.zone_subscription_request_id = req_id;
    }

    pub fn export_capi() void {
        @export(&make, .{ .name = std.fmt.comptimePrint("{s}_make", .{cname}) });
        @export(&retain, .{ .name = std.fmt.comptimePrint("{s}_retain", .{cname}) });
        @export(&release, .{ .name = std.fmt.comptimePrint("{s}_release", .{cname}) });
        @export(&getEvent, .{ .name = std.fmt.comptimePrint("{s}_get_event", .{cname}) });
        @export(&subscribeZones, .{ .name = std.fmt.comptimePrint("{s}_subscribe_zones", .{cname}) });
    }
};

pub fn export_capi() void {
    ConnectionErrorEvent.export_capi();
    ConnectedEvent.export_capi();
    Event.export_capi();
    Connection.export_capi();
}