WIP message-delivery-method-web-push

services/message-delivery-method-web-push/PouchDB.js

@@ -1,4 +0,0 @@
-import PouchDB from "pouchdb-core";
-import PouchDBAdapterHttp from "pouchdb-adapter-http";
-
-export default PouchDB.plugin(PouchDBAdapterHttp);

services/message-delivery-method-web-push/package.json

@@ -2,7 +2,7 @@
   "name": "@hereconnect/message-delivery-method-web-push",
   "version": "1.0.0",
   "description": "",
-  "main": "index.js",
+  "main": "src/server.ts",
   "type": "module",
   "license": "@hereconnect/license",
   "private": true,
@@ -24,13 +24,16 @@
   },
   "contributors": [],
   "scripts": {
-    "generate-keys": "node ./generate-keys.js",
-    "test-send-notification": "node ./test-send-notification.js",
-    "test-send-notification-cancel": "node ./test-send-notification-cancel.js"
+    "generate-keys": "node ./src/generate-keys.js",
+    "test-send-notification": "node ./src/test-send-notification.js",
+    "test-send-notification-cancel": "node ./src/test-send-notification-cancel.js"
   },
   "dependencies": {
+    "@hereconnect/types": "^1.0.0",
+    "couchdb-changes-stream": "^1.0.1",
     "pouchdb-adapter-http": "^9.0.0",
     "pouchdb-core": "^9.0.0",
+    "pouchdb-find": "^9.0.0",
     "web-push": "^3.6.7"
   }
 }

services/message-delivery-method-web-push/src/PouchDB.ts

@@ -0,0 +1,7 @@
+import PouchDB from "pouchdb-core";
+import PouchDBAdapterHttp from "pouchdb-adapter-http";
+import PouchDBFindPlugin from 'pouchdb-find'
+
+export default PouchDB
+  .plugin(PouchDBAdapterHttp)
+  .plugin(PouchDBFindPlugin);

services/message-delivery-method-web-push/src/dbs.ts

@@ -0,0 +1,38 @@
+import PouchDB from "./PouchDB";
+import type {
+  MessageDocument,
+  QRCodeDocument,
+  ServerSettingsDocument,
+} from "@hereconnect/types";
+
+export interface ServiceDoc {
+  _id: string;
+  lastSeq: string | number;
+}
+
+if (!process.env.API_URL) {
+  throw new Error("Environment variable API_URL is not set");
+}
+
+export const messagesDbUrl = `${process.env.API_URL}/messages`;
+
+export const serverSettingsDb = new PouchDB<ServerSettingsDocument>(
+  `${process.env.API_URL}/server_settings`,
+  {
+    skip_setup: true,
+  },
+);
+export const qrDb = new PouchDB<QRCodeDocument>(`${process.env.API_URL}/qr`, {
+  skip_setup: true,
+});
+
+export const { messagesDb, serviceDb } = (() => {
+  const messagesDB = new PouchDB(messagesDbUrl, {
+    skip_setup: true,
+  });
+
+  return {
+    messagesDb: messagesDB as PouchDB.Database<MessageDocument>,
+    serviceDb: messagesDB as PouchDB.Database<ServiceDoc>,
+  } as const;
+})();

services/message-delivery-method-web-push/src/server.ts

@@ -0,0 +1,81 @@
+import webpush from "web-push";
+
+import { CouchDBChangesStream } from "couchdb-changes-stream";
+import { qrDb, messagesDbUrl } from "./dbs";
+import { setupVapidDetails } from "./setupVapidDetails";
+import type { MessageDocument } from "@hereconnect/types";
+import { ServiceSeqTracker } from "./serviceSeqTracker";
+
+const serviceSeqTracker = new ServiceSeqTracker();
+
+const start = async () => {
+  await setupVapidDetails();
+  const since = await serviceSeqTracker.getSince();
+
+  const changesStream = new CouchDBChangesStream<MessageDocument>(
+    messagesDbUrl,
+    {
+      since,
+      feed: "continuous",
+      include_docs: true,
+      heartbeat: 1000,
+      filter: "_selector",
+      selector: {
+        type: "message",
+      },
+    },
+  );
+
+  // Gracefully handle SIGINT
+  process.on("SIGINT", async () => {
+    changesStream.stop();
+    process.stdout.write("\n");
+    process.exit(2);
+  });
+
+  console.info("Service started from sequence", since);
+
+  for await (const change of changesStream) {
+    const { seq, doc } = change;
+
+    if (!doc) {
+      console.error("Change missing document:", change);
+      continue;
+    }
+
+    serviceSeqTracker.checkNextSeq(seq);
+
+    try {
+      if (doc.from === "guest") {
+        const qrDoc = await qrDb.get(`qr:${doc.qr_code_uri}`);
+        if (qrDoc.messageDeliveryMethod !== "webPush") {
+          console.log(`Skip !webPush`);
+          continue;
+        }
+
+        console.log("FROM GUEST");
+        console.log(qrDoc.user_uuid);
+        console.log(qrDoc.browser_uuid);
+      } else if (doc.from === "owner") {
+        console.log("FROM OWNER");
+      } else {
+        continue;
+      }
+
+      console.log(
+        `Processed change: ${JSON.stringify({ _id: change.id, ref: doc._rev, seq: change.seq })}`,
+      );
+    } catch (error) {
+      console.error(
+        `Failed processing: ${JSON.stringify(error)} change: ${JSON.stringify({ id: doc._id, rev: doc._rev, seq: seq })}`,
+      );
+    } finally {
+      await serviceSeqTracker.saveSeq(seq);
+    }
+  }
+};
+
+start().catch((error) => {
+  console.error("error", error);
+  process.exit(1);
+});

services/message-delivery-method-web-push/src/serviceSeqTracker.ts

@@ -0,0 +1,38 @@
+import { serviceDb, type ServiceDoc } from "./dbs";
+
+const SERVICE_DOC_ID = "_local/service:message-delivery-method-web-push";
+
+export class ServiceSeqTracker {
+  private serviceDoc: ServiceDoc;
+
+  async getSince() {
+    this.serviceDoc = await serviceDb.get(SERVICE_DOC_ID).catch((error) => {
+      if (error.name === "not_found") {
+        return {
+          _id: SERVICE_DOC_ID,
+          lastSeq: "0",
+        };
+      }
+      throw error; // Re-throw other errors
+    });
+
+    this.serviceDoc.lastSeq = 0;
+
+    return this.serviceDoc.lastSeq;
+  }
+
+  checkNextSeq(seq: number | string) {
+    const next = parseInt(seq as string, 10);
+    const current = parseInt(this.serviceDoc.lastSeq as string, 10);
+    if (next < current) {
+      throw new Error(
+        `Sequence number is too old: ${seq} (${next} < ${current})`,
+      );
+    }
+  }
+
+  saveSeq(seq: number | string) {
+    this.serviceDoc.lastSeq = seq;
+    return serviceDb.put(this.serviceDoc);
+  }
+}

services/message-delivery-method-web-push/src/setupVapidDetails.ts

@@ -0,0 +1,77 @@
+import webpush from "web-push";
+import { serverSettingsDb } from "./dbs";
+import { ServerSettingsDocument } from "@hereconnect/types";
+
+function isType<N extends ServerSettingsDocument["name"]>(
+  doc: ServerSettingsDocument,
+  name: N,
+): doc is ServerSettingsDocument & { name: N } {
+  return doc.name === name;
+}
+
+export const setupVapidDetails = async () => {
+  const changes = serverSettingsDb.changes({
+    since: "now",
+    live: true,
+    include_docs: true,
+    doc_ids: [
+      "server_settings:message_delivery_method_web_push_vapid",
+      "public_settings:message_delivery_method_web_push_gcm_api_key",
+    ],
+  });
+
+  const {
+    rows: [vapidDetailsRow, gcmApiRow],
+  } = await serverSettingsDb.allDocs({
+    keys: [
+      "server_settings:message_delivery_method_web_push_vapid",
+      "public_settings:message_delivery_method_web_push_gcm_api_key",
+    ],
+    include_docs: true,
+  });
+
+  if (
+    !("doc" in vapidDetailsRow) ||
+    !isType(vapidDetailsRow.doc, "message_delivery_method_web_push_vapid")
+  ) {
+    throw new Error(
+      `VAPID details ${"error" in vapidDetailsRow ? vapidDetailsRow.error : "not exists"}`,
+    );
+  }
+
+  if (
+    "doc" in gcmApiRow &&
+    isType(gcmApiRow.doc, "message_delivery_method_web_push_gcm_api_key")
+  ) {
+    webpush.setGCMAPIKey(gcmApiRow.doc.value);
+  }
+
+  webpush.setVapidDetails(
+    vapidDetailsRow.doc.value.subject,
+    vapidDetailsRow.doc.value.keys.publicKey,
+    vapidDetailsRow.doc.value.keys.privateKey,
+  );
+
+  changes.on("change", (change) => {
+    if (!("doc" in change)) {
+      return;
+    }
+    if (isType(change.doc, "message_delivery_method_web_push_vapid")) {
+      console.log("update vapid details");
+      webpush.setVapidDetails(
+        change.doc.value.subject,
+        change.doc.value.keys.publicKey,
+        change.doc.value.keys.privateKey,
+      );
+      return;
+    }
+
+    if (
+      "doc" in change &&
+      isType(change.doc, "message_delivery_method_web_push_gcm_api_key")
+    ) {
+      console.log("update gcm api key");
+      webpush.setGCMAPIKey(change.doc.value);
+    }
+  });
+};