Split code

src/socket/index.ts

@@ -1,21 +1,26 @@
 import { WebSocketServer } from "ws";
 import { buildMessageStreammingWebSocket } from "./message-streaming";
 import { authenticateWebSocket } from "./authentication";
-import { syncEntriesViaSocket } from "./sync";
+import { buildSyncingWebSocket, syncEntries } from "./sync";
 
 export function setupWebsocketServer(server: WebSocketServer) {
   server.on("connection", async (socket, request) => {
     // Enable message streaming
     const messageSocket = buildMessageStreammingWebSocket(socket);
 
-    // Authenticate user and node
+    // Authenticate user
     const authenticatedSocket = authenticateWebSocket(messageSocket, request);
     if (!authenticatedSocket) {
       messageSocket.close();
       return;
     }
 
     // Sync entries
-    syncEntriesViaSocket(authenticatedSocket);
+    try {
+      const syncingSocket = buildSyncingWebSocket(authenticatedSocket);
+      await syncEntries(syncingSocket);
+    } finally {
+      socket.close();
+    }
   });
 }

src/socket/sync-error.ts

@@ -0,0 +1,3 @@
+import { ServerError } from "../error";
+
+export class MalformedMessageError extends ServerError {}

src/socket/sync-service.ts

@@ -0,0 +1,23 @@
+import { EntityManager } from "typeorm";
+import { getManager } from "../db";
+import { UserService } from "../service/user";
+import Container from "typedi";
+import { HistoryService } from "../service/history";
+import { NodeService } from "../service/node";
+import { EntryService } from "../service/entry";
+
+export class SyncService {
+  public manager: EntityManager;
+  public user: UserService;
+  public history: HistoryService;
+  public node: NodeService;
+  public entry: EntryService;
+
+  public constructor() {
+    this.manager = getManager();
+    this.user = Container.get(UserService);
+    this.history = Container.get(HistoryService);
+    this.node = Container.get(NodeService);
+    this.entry = Container.get(EntryService);
+  }
+}

src/socket/sync-utils.ts

@@ -0,0 +1,63 @@
+import { ControlMessage, Message } from "./message";
+import { BadParameterError } from "../error";
+import { RawData } from "ws";
+import { MalformedMessageError } from "./sync-error";
+import { SyncingWebSocket } from "./sync";
+
+function tryParseMessageJSON(socket: SyncingWebSocket, data: RawData): Message {
+  let message;
+
+  try {
+    message = JSON.parse(data.toString());
+  } catch (err) {
+    if (err instanceof SyntaxError) {
+      socket.sendMessage(
+        new ControlMessage([new BadParameterError("Bad JSON syntax.")])
+      );
+      socket.close();
+    }
+
+    socket.log(err);
+    throw err;
+  }
+
+  return message;
+}
+function checkMessageIntegrity(
+  socket: SyncingWebSocket,
+  message: Message
+): void {
+  if (typeof message.session !== "string" || message.session === "") {
+    socket.sendMessage(
+      new ControlMessage([
+        new BadParameterError(
+          "Messsage should contain an non-empty session of string type."
+        ),
+      ])
+    );
+    socket.close();
+    throw new MalformedMessageError();
+  }
+
+  if (!message.type) {
+    socket.replyMessage(
+      message,
+      new ControlMessage([
+        new BadParameterError(
+          "Each message must contain a valid `type` field."
+        ),
+      ])
+    );
+    throw new MalformedMessageError();
+  }
+}
+
+export function parseMessage(
+  socket: SyncingWebSocket,
+  data: RawData
+): Message | null {
+  const message = tryParseMessageJSON(socket, data);
+  checkMessageIntegrity(socket, message);
+
+  return message;
+}