Continue integration tests

TODO: rework reconnect subscription stuff.

spec/integration.spec.ts

@@ -2,6 +2,7 @@ import * as crypto from "crypto";
 
 import { createPushManager } from "../src";
 import { deriveKeyAndNonce, generateEcKeys, randomBytes } from "../src/crypto";
+import { ClientAck, ClientAckCodes } from "../src/messages/message";
 import { PushManager } from "../src/push-manager";
 import { GenericPushSubscription } from "../src/push-subscription";
 import {
@@ -10,10 +11,15 @@ import {
   fromUtf8ToBuffer,
 } from "../src/string-manipulation";
 
-import { applicationPublicKey } from "./constants";
+import {
+  applicationPrivateKey,
+  applicationPublicKey,
+  applicationPublicKeyX,
+  applicationPublicKeyY,
+} from "./constants";
 import { TestLogger } from "./test-logger";
 import { TestBackingStore } from "./test-storage";
-import { defaultUaid, TestWebSocketServer } from "./test-websocket-server";
+import { defaultUaid, helloHandlerWithUaid, TestWebSocketServer } from "./test-websocket-server";
 
 const port = 1234;
 const url = "ws://localhost:" + port;
@@ -39,10 +45,110 @@ describe("end to end", () => {
 
   afterEach(async () => {
     await pushManager?.destroy();
+    // ensure the server is using the default handlers
+    server.useDefaultHandlers();
     // ensure we don't leak connections between tests
     server.closeClients();
   });
 
+  describe("reconnection", () => {
+    beforeEach(async () => {
+      pushManager = await createPushManager(storage, logger, {
+        autopushUrl: url,
+        // Set reconnect to occur after 10ms
+        reconnectDelay: () => new Promise((resolve) => setTimeout(resolve, 10)),
+      });
+    });
+
+    async function closeWebSocket() {
+      const client = server.clients[0];
+      client.ws.close();
+      await new Promise<void>((resolve) => {
+        client.ws.on("close", resolve);
+      });
+      return client;
+    }
+
+    it("reconnects when the connection is closed", async () => {
+      const previousClient = await closeWebSocket();
+
+      // TODO: better await for reconnect
+      await new Promise<void>((resolve) => {
+        setTimeout(() => {
+          resolve();
+        }, 1000);
+      });
+
+      expect(server.clients).toHaveLength(1);
+      expect(server.clients[0]).not.toBe(previousClient);
+    });
+
+    it("maintains event subscriptions after reconnect", async () => {
+      const sub = await pushManager.subscribe({
+        userVisibleOnly: true,
+        applicationServerKey: applicationPublicKey,
+      });
+
+      const notificationSpy = jest.fn();
+      const notificationPromise = new Promise<void>((resolve, reject) => {
+        sub.addEventListener("notification", (d) => {
+          notificationSpy(d);
+          resolve();
+        });
+        setTimeout(reject, 1000);
+      });
+
+      await closeWebSocket();
+
+      // TODO: better await for reconnect
+      await new Promise<void>((resolve) => {
+        setTimeout(() => {
+          resolve();
+        }, 500);
+      });
+
+      server.sendNotification(sub.channelID);
+
+      await notificationPromise;
+      expect(notificationSpy).toHaveBeenCalled();
+    });
+
+    it("maintains event subscriptions after reconnect and a new uaid", async () => {
+      const newUaid = "new-uaid";
+      server.helloHandler = helloHandlerWithUaid(newUaid);
+
+      const sub = await pushManager.subscribe({
+        userVisibleOnly: true,
+        applicationServerKey: applicationPublicKey,
+      });
+
+      const notificationSpy = jest.fn();
+      const notificationPromise = new Promise<void>((resolve, reject) => {
+        sub.addEventListener("notification", (d) => {
+          notificationSpy(d);
+          resolve();
+        });
+        setTimeout(reject, 1000);
+      });
+
+      await closeWebSocket();
+
+      // TODO: better await for reconnect
+      await new Promise<void>((resolve) => {
+        setTimeout(() => {
+          resolve();
+        }, 500);
+      });
+
+      expect(server.identifiedClients[0].uaid).toEqual(newUaid);
+
+      server.sendNotification(sub.channelID);
+
+      await notificationPromise;
+      expect(notificationSpy).toHaveBeenCalled();
+    });
+  });
+
   describe("Hello", () => {
     it("connects to the server", async () => {
       pushManager = await createPushManager(storage, logger, { autopushUrl: url });
@@ -84,21 +190,78 @@ describe("end to end", () => {
         use_webpush: true,
       });
     });
+
+    describe("existing subscriptions", () => {
+      beforeEach(async () => {
+        // Set up existing storage
+        await storage.write("channelIDs", JSON.stringify(["f2ca74ee-d688-4cb2-8ae1-9deb4805be29"]));
+        await storage.write(
+          "f2ca74ee-d688-4cb2-8ae1-9deb4805be29:endpoint",
+          JSON.stringify("https://example.com/push//f2ca74ee-d688-4cb2-8ae1-9deb4805be29"),
+        );
+        await storage.write(
+          "f2ca74ee-d688-4cb2-8ae1-9deb4805be29:options",
+          JSON.stringify({
+            userVisibleOnly: true,
+            applicationServerKey: applicationPublicKey,
+          }),
+        );
+        await storage.write(
+          "f2ca74ee-d688-4cb2-8ae1-9deb4805be29:auth",
+          JSON.stringify("kKZ96yjFVbvnUa458DDWNg"),
+        );
+        await storage.write(
+          "f2ca74ee-d688-4cb2-8ae1-9deb4805be29:privateEncKey",
+          JSON.stringify({
+            key_ops: ["deriveKey", "deriveBits"],
+            ext: true,
+            kty: "EC",
+            x: applicationPublicKeyX,
+            y: applicationPublicKeyY,
+            crv: "P-256",
+            d: applicationPrivateKey,
+          }),
+        );
+      });
+
+      it("reconnects existing channels", async () => {
+        // Same Uaid as response
+        await storage.write("uaid", JSON.stringify(defaultUaid));
+
+        pushManager = await createPushManager(storage, logger, { autopushUrl: url });
+        const client = server.clients[0];
+        expect(client).toHaveReceived({
+          messageType: "hello",
+          uaid: "5f0774ac-09a3-45d9-91e4-f4aaebaeec72",
+          channelIDs: ["f2ca74ee-d688-4cb2-8ae1-9deb4805be29"],
+          use_webpush: true,
+        });
+      });
+    });
   });
 
   describe("Notification", () => {
-    it("sends a notification", async () => {
-      pushManager = await createPushManager(storage, logger, { autopushUrl: url });
-      const sub = await pushManager.subscribe({
+    let sub: GenericPushSubscription;
+
+    beforeEach(async () => {
+      pushManager = await createPushManager(storage, logger, {
+        autopushUrl: url,
+        ackIntervalMs: 100,
+      });
+      sub = await pushManager.subscribe({
         userVisibleOnly: true,
         applicationServerKey: applicationPublicKey,
       });
+    });
+
+    it("sends a notification", async () => {
       const notifiedSpy = jest.fn();
-      const notifiedCalled = new Promise<void>((resolve) => {
+      const notifiedCalled = new Promise<void>((resolve, reject) => {
         sub.addEventListener("notification", (data) => {
           notifiedSpy(data);
           resolve();
         });
+        setTimeout(() => reject(), 1000);
       });
 
       server.sendNotification(sub.channelID);
@@ -108,11 +271,6 @@ describe("end to end", () => {
     });
 
     it("sends a notification message", async () => {
-      pushManager = await createPushManager(storage, logger, { autopushUrl: url });
-      const sub = await pushManager.subscribe({
-        userVisibleOnly: true,
-        applicationServerKey: applicationPublicKey,
-      });
       const notifiedSpy = jest.fn();
       const notifiedCalled = new Promise<void>((resolve, reject) => {
         sub.addEventListener("notification", (data) => {
@@ -135,6 +293,109 @@ describe("end to end", () => {
 
       expect(notifiedSpy).toHaveBeenCalledWith("some data");
     });
+
+    it("sends acks when notifications are received", async () => {
+      const ackPromise = new Promise<void>((resolve, reject) => {
+        server.ackHandler = () => resolve();
+        setTimeout(() => reject(), 1000);
+      });
+
+      const version = server.sendNotification(sub.channelID);
+
+      const expectedAck: ClientAck = {
+        messageType: "ack",
+        updates: [{ channelID: sub.channelID, version, code: ClientAckCodes.SUCCESS }],
+      };
+
+      await expect(ackPromise).resolves.toBeUndefined();
+
+      const client = server.identifiedClientFor(sub.channelID);
+      if (!client) {
+        fail("Client not found");
+      }
+      expect(client).toHaveReceived(expectedAck);
+    });
+
+    it("acks decryption errors", async () => {
+      const ackPromise = new Promise<void>((resolve, reject) => {
+        server.ackHandler = () => resolve();
+        setTimeout(() => reject(), 1000);
+      });
+
+      const version = server.sendNotification(sub.channelID, "This should have been encrypted", {
+        encoding: "aes128gcm",
+      });
+
+      const expectedAck: ClientAck = {
+        messageType: "ack",
+        updates: [{ channelID: sub.channelID, version, code: ClientAckCodes.DECRYPT_FAIL }],
+      };
+
+      await expect(ackPromise).resolves.toBeUndefined();
+
+      const client = server.identifiedClientFor(sub.channelID);
+      if (!client) {
+        fail("Client not found");
+      }
+      expect(client).toHaveReceived(expectedAck);
+    });
+
+    it("groups acks together", async () => {
+      const ackPromise = new Promise<void>((resolve, reject) => {
+        server.ackHandler = () => resolve();
+        setTimeout(() => reject(), 1000);
+      });
+
+      const version1 = server.sendNotification(sub.channelID);
+      const version2 = server.sendNotification(sub.channelID);
+
+      const expectedAck: ClientAck = {
+        messageType: "ack",
+        updates: [
+          { channelID: sub.channelID, version: version1, code: ClientAckCodes.SUCCESS },
+          { channelID: sub.channelID, version: version2, code: ClientAckCodes.SUCCESS },
+        ],
+      };
+
+      await expect(ackPromise).resolves.toBeUndefined();
+
+      const client = server.identifiedClientFor(sub.channelID);
+      if (!client) {
+        fail("Client not found");
+      }
+      expect(client).toHaveReceived(expectedAck);
+    });
+
+    it("groups acks togher across subscriptions", async () => {
+      const sub2 = await pushManager.subscribe({
+        userVisibleOnly: true,
+        applicationServerKey: applicationPublicKey,
+      });
+
+      const ackPromise = new Promise<void>((resolve, reject) => {
+        server.ackHandler = () => resolve();
+        setTimeout(() => reject(), 1000);
+      });
+
+      const version1 = server.sendNotification(sub.channelID);
+      const version2 = server.sendNotification(sub2.channelID);
+
+      const expectedAck: ClientAck = {
+        messageType: "ack",
+        updates: [
+          { channelID: sub.channelID, version: version1, code: ClientAckCodes.SUCCESS },
+          { channelID: sub2.channelID, version: version2, code: ClientAckCodes.SUCCESS },
+        ],
+      };
+
+      await expect(ackPromise).resolves.toBeUndefined();
+
+      const client = server.identifiedClientFor(sub.channelID);
+      if (!client) {
+        fail("Client not found");
+      }
+      expect(client).toHaveReceived(expectedAck);
+    });
   });
 });