build service

services/message-delivery-method-web-push/Dockerfile

@@ -0,0 +1,8 @@
+FROM oven/bun:1.1.38-alpine
+
+COPY . ./
+
+HEALTHCHECK --interval=30s --timeout=10s --start-period=10s --retries=1 \
+	CMD curl -f http://127.0.0.1:8000/health || exit 1
+
+CMD ["bun", "dist/server.js"]

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

@@ -2,7 +2,7 @@
   "name": "@hereconnect/service-message-delivery-method-web-push",
   "version": "1.0.0",
   "description": "",
-  "main": "src/server.ts",
+  "main": "dist/server.ts",
   "type": "module",
   "license": "@hereconnect/license",
   "private": true,
@@ -15,6 +15,9 @@
     "type": "git",
     "url": ""
   },
+  "files": [
+    "dist/**/*"
+  ],
   "bugs": "",
   "keywords": [],
   "author": {
@@ -32,18 +35,18 @@
   },
   "dependencies": {
     "@hereconnect/couchdb-changes-stream": "workspace:^",
-    "@hereconnect/types": "^1.0.0",
     "pouchdb-adapter-http": "^9.0.0",
     "pouchdb-core": "^9.0.0",
     "pouchdb-find": "^9.0.0",
-    "typescript": "~5.6.3",
     "web-push": "^3.6.7"
   },
   "devDependencies": {
+    "@hereconnect/types": "^1.0.0",
     "@types/node": "^20.17.6",
     "@types/pouchdb-adapter-http": "^6.1.6",
     "@types/pouchdb-core": "^7.0.15",
     "@types/pouchdb-find": "^7.3.3",
-    "@types/web-push": "^3.6.4"
+    "@types/web-push": "^3.6.4",
+    "typescript": "~5.6.3"
   }
 }

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

@@ -8,8 +8,7 @@ import type {
 } from "@hereconnect/types";
 import { PushSubscription } from "web-push";
 
-export interface ServiceDoc {
-  _id: string;
+export interface ServiceSeqTrackerDocument {
   lastSeq: string | number;
 }
 
@@ -35,14 +34,15 @@ export const webPushSubscriptionsDb = new PouchDB<
   skip_setup: true,
 });
 
-export const { messagesDb, serviceDb } = (() => {
+export const { messagesDb, serviceSeqTrackerDb } = (() => {
   const messagesDB = new PouchDB(messagesDbUrl, {
     skip_setup: true,
   });
 
   return {
     messagesDb: messagesDB as PouchDB.Database<MessageDocument>,
-    serviceDb: messagesDB as PouchDB.Database<ServiceDoc>,
+    serviceSeqTrackerDb:
+      messagesDB as PouchDB.Database<ServiceSeqTrackerDocument>,
   } as const;
 })();
 
@@ -52,3 +52,33 @@ export const userChatSettingsDb = new PouchDB<UserChatSettingsDocument>(
     skip_setup: true,
   },
 );
+
+export const checkDatabases = async () => {
+  const entries = await Promise.all(
+    Object.entries({
+      serverSettingsDb,
+      qrDb,
+      webPushSubscriptionsDb,
+      messagesDb,
+      userChatSettingsDb,
+    } as const).map(
+      async ([name, db]): Promise<
+        [string, Awaited<ReturnType<typeof db.info>>]
+      > => {
+        const info = await db.info();
+        return [name, info];
+      },
+    ),
+  );
+
+  for (const [name, info] of entries) {
+    if ("error" in info && "reason" in info) {
+      throw new Error(`[${name}] ${info.error}: ${info.reason}`);
+    }
+    if ("error" in info) {
+      throw new Error(`[${name}] ${info.error}`);
+    }
+  }
+
+  return Object.fromEntries(entries);
+};

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

@@ -10,7 +10,7 @@ export function startHealthServer(
   controller: AbortController,
   defaultPort: number = 8000,
   defaultHostName: string = "127.0.0.1",
-): void {
+) {
   const port = parseInt(process.env.HEALTH_PORT || `${defaultPort}`, 10);
   const hostname = process.env.HEALTH_HOST || defaultHostName;
 
@@ -43,7 +43,11 @@ export function startHealthServer(
     }
   });
 
-  server.listen(port, hostname, () => {
+  const { promise, resolve, reject } = Promise.withResolvers<void>();
+  server.once("error", reject);
+  server.listen(port, hostname, resolve);
+
+  promise.then(() => {
     console.log(
       `Healthcheck server running on http://${hostname}:${port}/health`,
     );
@@ -52,6 +56,9 @@ export function startHealthServer(
   // Закрываем сервер при активации AbortController
   controller.signal.addEventListener("abort", () => {
     console.log("Healthcheck server shutting down due to abort signal.");
+    reject();
     server.close();
   });
+
+  return promise;
 }

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

@@ -1,11 +1,14 @@
 import webpush from "web-push";
 
+const heartbeat = Number(process.env.HEARTBEAT_INTERVAL ?? 30_000);
+
 import { CouchDBChangesStream } from "@hereconnect/couchdb-changes-stream";
 import {
   qrDb,
   messagesDbUrl,
   webPushSubscriptionsDb,
   userChatSettingsDb,
+  checkDatabases,
 } from "./dbs";
 import { setupVapidDetails } from "./setupVapidDetails";
 import type { MessageDocument } from "@hereconnect/types";
@@ -14,8 +17,6 @@ import { startHealthServer } from "./health-server";
 
 const abortController = new AbortController();
 
-startHealthServer(abortController);
-
 function limitStringLengthWithEllipsis(str: string, maxLength: number): string {
   return str.length > maxLength ? str.substring(0, maxLength) + "…" : str;
 }
@@ -33,10 +34,13 @@ const start = async () => {
       since: serviceSeqTracker.lastSeq,
       feed: "continuous",
       include_docs: true,
-      heartbeat: 1000,
+      heartbeat,
       filter: "_selector",
       selector: {
-        type: "message",
+        $or: [
+          { type: "message" },
+          { type: "service", service: "message-delivery-method-web-push" },
+        ],
       },
     },
   );
@@ -50,6 +54,20 @@ const start = async () => {
 
   console.info("Service started from sequence", serviceSeqTracker.lastSeq);
 
+  // check dbs
+  await checkDatabases().then(({ messagesDb }) => {
+    if (
+      parseInt(serviceSeqTracker.lastSeq.toString(), 10) ===
+      parseInt(messagesDb.update_seq.toString(), 10)
+    ) {
+      console.log("Since is the last seq");
+    } else {
+      console.log("Last seq is", messagesDb.update_seq);
+    }
+  });
+
+  await startHealthServer(abortController);
+
   for await (const change of changesStream) {
     const { seq, doc: msgDoc } = change;
 

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

@@ -1,4 +1,4 @@
-import { serviceDb, type ServiceDoc } from "./dbs";
+import { serviceSeqTrackerDb, type ServiceSeqTrackerDocument } from "./dbs";
 
 export class ServiceSeqTracker {
   /**
@@ -6,7 +6,7 @@ export class ServiceSeqTracker {
    * @constructor
    */
   static async Init(_id: string) {
-    const serviceDoc = await serviceDb.get(_id).catch((error) => {
+    const doc = await serviceSeqTrackerDb.get(_id).catch((error) => {
       if (error.name === "not_found") {
         return {
           _id,
@@ -15,20 +15,20 @@ export class ServiceSeqTracker {
       }
       throw error; // Re-throw other errors
     });
-    return new ServiceSeqTracker(serviceDoc);
+    return new ServiceSeqTracker(doc);
   }
 
-  private constructor(private serviceDoc: ServiceDoc) {}
+  private constructor(private doc: ServiceSeqTrackerDocument) {}
 
   get lastSeq() {
-    return this.serviceDoc.lastSeq;
+    return this.doc.lastSeq;
   }
 
   checkNextSeq(seq: number | string) {
-    if (!this.serviceDoc) {
+    if (!this.doc) {
     }
     const next = parseInt(seq as string, 10);
-    const current = parseInt(this.serviceDoc.lastSeq as string, 10);
+    const current = parseInt(this.doc.lastSeq as string, 10);
     if (next < current) {
       throw new Error(
         `Sequence number is too old: ${seq} (${next} < ${current})`,
@@ -37,7 +37,7 @@ export class ServiceSeqTracker {
   }
 
   saveSeq(seq: number | string) {
-    this.serviceDoc.lastSeq = seq;
-    return serviceDb.put(this.serviceDoc);
+    this.doc.lastSeq = seq;
+    return serviceSeqTrackerDb.put(this.doc);
   }
 }