sapphire-backend/websocket.c

/*
 * Sapphire backend
 *
 * Copyright (C) 2018 Alyssa Rosenzweig
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02111-1301  USA
 *
 */

#include <stdio.h>
#include <purple.h>

#include <gio/gio.h>
#include <gio/gunixsocketaddress.h>
#include <glib/gstdio.h>

#include "core.h"
#include "websocket.h"
#include "push.h"

#define WS_PORT 7070

/* This module is responsible for networking via UNIX sockets, translated to
 * WebSockets at the proxy level. */

/* List of all authenticated connections. These connections will be broadcasted
 * to by broadcast_raw_packet. */

GSList *authenticated_connections = NULL;

/* Returns TRUE if there are connected clients (such that we don't need push
 * notifications) */

gboolean
sapphire_any_connected_clients(void){
	return authenticated_connections != NULL;
}

void
sapphire_send_raw_packet(Connection *conn, const char *frame)
{
	if (!g_socket_connection_is_connected(conn->connection)) {
		printf("Tried to send %s to closed connection, ignoring\n", frame);
		return;
	}

	GError *gerror = NULL;
	GOutputStream *ostream = g_io_stream_get_output_stream(G_IO_STREAM(conn->connection));

	g_output_stream_write_all(ostream, frame, strlen(frame), NULL, NULL, &gerror);

	if (gerror) {
		printf("Ahh gerror %s!\n", gerror->message);
		gerror = NULL;
		return;
	}

	char end = '\n';
	g_output_stream_write(ostream, &end, 1, NULL, &gerror);

	if (gerror) {
		printf("Ahhh gerror v2!\n");
	}
}

/* Sends the packet to any connection, or save it for the next connection */

GSList *queued_messages = NULL;

void sapphire_send_any_or_save(char *packet)
{
	if (authenticated_connections) {
		Connection *first_conn = (Connection *) authenticated_connections->data;
		sapphire_send_raw_packet(first_conn, packet);
		g_free(packet);
	} else {
		queued_messages = g_slist_prepend(queued_messages, packet);
	}
}

/* Dequeue saved, send and free */

static void sapphire_dequeue_saved_messages(Connection *conn)
{
	for (GSList *it = queued_messages; it != NULL; it = it->next) {
		gchar *msg = (gchar *) it->data;
		sapphire_send_raw_packet(conn, msg);
		g_free(msg);
	}

	g_slist_free(queued_messages);
	queued_messages = NULL;
}

/* Broadcast a packet to all currently connected clients. */
void
sapphire_broadcast_raw_packet(const char *packet)
{
	/* Iterate the connection list and send message everywhere */

	for (GSList *l = authenticated_connections; l != NULL; l = l->next) {
		Connection *connection = (Connection *) l->data;
		sapphire_send_raw_packet(connection, packet);
	}
}

/* XXX copypasted from proxy.c */

static void sapphire_got_line(GObject *source_object, GAsyncResult *res, gpointer user_data);

static void
sapphire_read_line(Connection *conn)
{
	g_data_input_stream_read_line_async(conn->distream, G_PRIORITY_DEFAULT, NULL, sapphire_got_line, conn);
}

static void
sapphire_got_line(GObject *source_object,
                        GAsyncResult *res,
                        gpointer user_data)
{
	Connection *conn = (Connection *) user_data;

	GError *err = NULL;
	gsize len;
	char *data = g_data_input_stream_read_line_finish_utf8(G_DATA_INPUT_STREAM(source_object), res, &len, &err);

	if (err || !data) {
		/* Borp, error -- disconnect */

		/* Free the connection */
		g_hash_table_remove_all(conn->subscribed_ids);

		/* Splice the socket out of the authenticated list, so we no longer
		 * attempt to broadcast to it */

		authenticated_connections = g_slist_remove(authenticated_connections, conn);

		g_free(conn);

		/* Disconnect accounts if needed */
		sapphire_enable_accounts_by_connections();

		return;
	}

	/* The message should be interpreted as JSON, decode that here */

	JsonParser *parser = json_parser_new();

	if (!json_parser_load_from_data(parser, data, -1, NULL)) {
		fprintf(stderr, "Error parsing response: %s\n", data);
		fprintf(stderr, "^ Couldn't do it\n");
		goto refresh;
	}

	JsonNode *root = json_parser_get_root(parser);

	if (root == NULL) {
		printf("NULL root, ignoring\n");
		goto refresh;
	}

	/* Delegate off */

	JsonObject *obj = json_node_get_object(root);
	sapphire_process_message(conn, obj);

refresh:
	g_free(data);
	sapphire_read_line(conn);
}


gboolean
incoming_callback  (GSocketService *service,
                    GSocketConnection *connection,
                    GObject *source_object,
                    gpointer user_data)
{
	g_print("Received Connection from client!\n");

	/* Allocate a connection object for us and fill it in */
	Connection *conn = g_new0(Connection, 1);

	conn->is_authenticated = FALSE;

	/* Save the connection.
	 * IMPORTANT: Reference counting is necessary to keep the connection
	 * alive. No idea why this isn't documented anywhere, xxx
	 */

	conn->connection = g_object_ref(connection);

	/* Subscribe to incoming data */

	GInputStream *istream = g_io_stream_get_input_stream (G_IO_STREAM (conn->connection));
	conn->distream = g_data_input_stream_new(istream);
	sapphire_read_line(conn);

	/* Consider ourselves authenticated */
	conn->is_authenticated = TRUE;
	authenticated_connections = g_slist_append(authenticated_connections, conn);

	/* Dequeue messsages */
	sapphire_dequeue_saved_messages(conn);

	/* Flag the new connection in the push notification module */
	sapphire_push_connected();

	/* Setup accounts if needed */
	sapphire_enable_accounts_by_connections();

	/* Now that we're authenticated, it's necessary to bring the client up-to-date on what's popping */
	sapphire_send_world(conn);

	return FALSE;
}

void
sapphire_init_websocket(void)
{
	const gchar *filename = "./sockpuppet";
	g_unlink(filename);

	GSocketService *service = g_socket_service_new();
	GSocketAddress *addr = g_unix_socket_address_new (filename);

	g_socket_listener_add_address(G_SOCKET_LISTENER(service), addr, G_SOCKET_TYPE_STREAM, G_SOCKET_PROTOCOL_DEFAULT, NULL, NULL, NULL);
	g_signal_connect (service, "incoming", G_CALLBACK (incoming_callback), NULL);
}