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

import webpush from "web-push";

import { CouchDBChangesStream } from "couchdb-changes-stream";
import { qrDb, messagesDbUrl, webPushSubscriptionsDb } 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: msgDoc } = change;

    if (!msgDoc) {
      console.error("Change missing document:", change);
      continue;
    }

    serviceSeqTracker.checkNextSeq(seq);

    try {
      if (msgDoc.from === "guest") {
        // send message to owner
        const qrDoc = await qrDb.get(`qr_code:${msgDoc.qr_code_uri}`);
        if (qrDoc.messageDeliveryMethod !== "webPush") {
          console.log(`Skip !webPush`);
          continue;
        }

        const { rows } = await webPushSubscriptionsDb.allDocs({
          startkey: `web_push_subscription:${qrDoc.user_uuid}:`,
          endkey: `web_push_subscription:${qrDoc.user_uuid}:\uffff`,
          include_docs: true,
        });

        if (rows.length === 0) {
          console.log("NO SUBSCRIPTIONS");
        } else {
          console.log(`sending to ${rows.length} devices`);
        }

        for (const row of rows) {
          try {
            const subscription = row.doc.subscription;
            const topic = `chat-${msgDoc.qr_code_uri}-${msgDoc.chat}`;
            const tag = `${topic}-${(new Date(msgDoc.created_at).getTime() - new Date(qrDoc.created_at).getTime()).toString(36)}`;

            const options: NotificationOptions = {
              icon: "/images/qr-code-logo-200x200.jpg",
              body: msgDoc.body,
              tag,
              requireInteraction: true,
              data: {
                url: `/manage/${msgDoc.qr_code_uri}/chat/${msgDoc.chat}`,
                payload: { tag },
              },
            } as const;

            await webpush.sendNotification(
              subscription,
              JSON.stringify({
                type: "notification",
                notification: {
                  title: qrDoc.name,
                  options,
                },
              }),
              {
                urgency: "high",
                topic,
              },
            );
          } catch (error) {
            console.error(
              `Error sending notification to ${row.id} - ${JSON.stringify(error)}`,
            );
          }
        }
      } else if (msgDoc.from === "owner") {
        console.log("FROM OWNER");
      } else {
        continue;
      }

      console.log(
        `Processed change: ${JSON.stringify({ _id: change.id, ref: msgDoc._rev, seq: change.seq })}`,
      );
    } catch (error) {
      console.error(
        `Failed processing: ${JSON.stringify(error)} change: ${JSON.stringify({ id: msgDoc._id, rev: msgDoc._rev, seq: seq })}`,
      );
    } finally {
      await serviceSeqTracker.saveSeq(seq);
    }
  }
};

start().catch((error) => {
  console.error("error", error);
  process.exit(1);
});