read tg_bot_updates from db

services/tg-bot/src/bot/middleware/forwardQrCodeMessageMiddleware.ts

@@ -91,7 +91,7 @@ export const forwardQrCodeMessageMiddleware = async (
 
   const hashTag = `#qr${qr_code_uri}_c${chat}`;
   const manageMessageUrl = `${env.APP_BASE_URL}/manage/qr/${qr_code_uri}/chat/${chat}#message-${encodeURIComponent(created_at)}`;
-  const text = `<i>сообщение <a href="${escapeHtml(manageMessageUrl)}">диалог QR-кода <b>«${escapeHtml(qrCodeDoc.name)}»</b></a> ${escapeHtml(hashTag)}</i>`;
+  const text = `<i>сообщение отправлено <a href="${escapeHtml(manageMessageUrl)}">в диалог QR-кода <b>«${escapeHtml(qrCodeDoc.name)}»</b></a> ${escapeHtml(hashTag)}</i>`;
 
   // await ctx.react("⚡");
   await ctx.reply(text, {

services/tg-bot/src/config/env.ts

@@ -14,6 +14,10 @@ if (!process.env.API_URL) {
   throw new Error("Environment variable API_URL is not set");
 }
 
+if (!process.env.PUBLIC_API_URL) {
+  throw new Error("Environment variable PUBLIC_API_URL is not set");
+}
+
 if (!process.env.HEARTBEAT_INTERVAL) {
   throw new Error("Environment variable HEARTBEAT_INTERVAL is not set");
 }
@@ -23,5 +27,6 @@ export const env = {
   READ_BASE_URL: process.env.READ_BASE_URL,
   APP_BASE_URL: process.env.APP_BASE_URL,
   API_URL: process.env.API_URL,
+  PUBLIC_API_URL: process.env.PUBLIC_API_URL,
   HEARTBEAT_INTERVAL: parseInt(process.env.HEARTBEAT_INTERVAL, 10),
 } as const;

services/tg-bot/src/dbs.ts

@@ -11,55 +11,55 @@ import type {
 
 import type { ServiceSeqTrackerDocument } from "@hereconnect/lib-service-seq-tracker";
 
-import type { TelegramWebhookDesignDocument } from "@hereconnect/tg-bot-webhook-ddoc";
+import type { TgBotUpdateWebhookDesignDocument } from "@hereconnect/tg-bot-update-webhook-ddoc";
 import { CouchDBSessionDocument } from "./bot/storageAdapters/CouchDBStorageAdapter";
 import type { SessionData } from "./types/myContext";
+import { TgBotUpdateDocument } from "./types/TgBotUpdateDocument";
 import { env } from "./config/env";
 
 if (!env.API_URL) {
   throw new Error("Environment variable API_URL is not set");
 }
 
-export const TG_BOT_DB_NAME = "tg_bot";
+export const TG_BOT_UPDATES_DB_NAME = "tg_bot_updates";
 export const MESSAGES_DB_NAME = "messages";
-export const TG_BOT_DB_URL = `${env.API_URL}/${TG_BOT_DB_NAME}`;
+export const TG_BOT_UPDATES_DB_URL = `${env.API_URL}/${TG_BOT_UPDATES_DB_NAME}`;
 export const MESSAGES_DB_URL = `${env.API_URL}/${MESSAGES_DB_NAME}`;
 
 export const messagesDb = new PouchDB<MessageDocument>(MESSAGES_DB_URL, {
   skip_setup: true,
 });
 export const messagesServiceSeqTrackerDb =
-  messagesDb satisfies PouchDB.Database as unknown as PouchDB.Database<ServiceSeqTrackerDocument>;
+  messagesDb satisfies PouchDB.Database as PouchDB.Database as PouchDB.Database<ServiceSeqTrackerDocument>;
 
 export const publicSettingsDb = new PouchDB<PublicSettingsDocument>(
   `${env.API_URL}/public_settings`,
-  {
-    skip_setup: true,
-  },
+  { skip_setup: true },
 );
 
 export const qrDb = new PouchDB<QRCodeDocument>(`${env.API_URL}/qr`, {
   skip_setup: true,
 });
 
-export const tgBotDb = new PouchDB(TG_BOT_DB_URL, {
-  skip_setup: true,
-});
-
-export const serviceSeqTrackerDb = tgBotDb as ServiceSeqTrackerDatabase;
-
 export const tgBotSessionsDb = new PouchDB<CouchDBSessionDocument<SessionData>>(
   `${env.API_URL}/tg_bot_sessions`,
-  {
-    skip_setup: true,
-  },
+  { skip_setup: true },
 );
 
+export const tgBotUpdatesDb = new PouchDB<TgBotUpdateDocument>(
+  `${TG_BOT_UPDATES_DB_URL}`,
+  { skip_setup: true },
+);
+
+export const tgBotUpdatesDesignDocumentsDb =
+  tgBotUpdatesDb satisfies PouchDB.Database as PouchDB.Database as PouchDB.Database<TgBotUpdateWebhookDesignDocument>;
+
+export const tgBotUpdatesServiceSeqTrackerDb =
+  tgBotUpdatesDb satisfies PouchDB.Database as PouchDB.Database as ServiceSeqTrackerDatabase;
+
 export const userChatSettingsDb = new PouchDB<UserChatSettingsDocument>(
   `${env.API_URL}/user_chat_settings`,
-  {
-    skip_setup: true,
-  },
+  { skip_setup: true },
 );
 
 export const usersDb = new PouchDB<UserDocument>(`${env.API_URL}/_users`, {
@@ -71,11 +71,11 @@ const checkDatabases = async () => {
     publicSettingsDb,
     qrDb,
     usersDb,
-    serviceSeqTrackerDb,
     userChatSettingsDb,
-    tgBotDb,
+    tgBotUpdatesDb,
     tgBotSessionsDb,
     messagesServiceSeqTrackerDb,
+    messagesDb,
   } as const;
 
   const databasesEntries = Object.entries(databases);

services/tg-bot/src/service.ts

@@ -1,31 +1,45 @@
 import { createHealthServer } from "@hereconnect/lib-service-health-checkserver";
 import { createBot, setupCommands } from "./bot";
 import { createServiceChangesMessagesDb } from "./services/serviceChangesMessagesDb";
+import { createServiceChangesTgUpdatesDb } from "./services/serviceChangesTgUpdatesDb";
 import { iterateMessagesChanges } from "./services/iterateMessagesChanges";
+import { setupWebHook } from "./services/setupWebHook";
 
 const abortController = new AbortController();
 
 (async () => {
   const healthServer = createHealthServer(abortController);
   const bot = createBot({ abortController });
-  const { changesStream, serviceSeqTracker } =
-    await createServiceChangesMessagesDb({
-      abortController,
-    });
+  const messagesChanges = await createServiceChangesMessagesDb({
+    abortController,
+  });
+  const tgUpdatesChanges = await createServiceChangesTgUpdatesDb({
+    abortController,
+  });
 
   process.on("SIGINT", async () => {
-    changesStream.stop();
+    messagesChanges.changesStream.stop();
+    tgUpdatesChanges.changesStream.stop();
     bot.stop();
   });
 
   await healthServer.start();
 
   await bot.init();
-  // Устанавливаем команд
+
+  await setupWebHook(bot);
   await setupCommands(bot);
 
-  bot.start();
+  // bot.start();
   console.log("Бот запущен!");
 
-  iterateMessagesChanges({ changesStream, serviceSeqTracker, bot });
+  iterateMessagesChanges({ bot, ...messagesChanges });
+
+  for await (const change of tgUpdatesChanges.changesStream) {
+    tgUpdatesChanges.serviceSeqTracker.checkNextSeq(change.seq);
+    if (change.doc) {
+      await bot.handleUpdate(change.doc.data);
+    }
+    await tgUpdatesChanges.serviceSeqTracker.saveSeq(change.seq);
+  }
 })();

services/tg-bot/src/services/serviceChangesTgUpdatesDb.ts

@@ -0,0 +1,55 @@
+import { ServiceSeqTracker } from "@hereconnect/lib-service-seq-tracker";
+import { CouchDBChangesStream } from "@hereconnect/couchdb-changes-stream";
+import type { TgBotUpdateDocument } from "../types/TgBotUpdateDocument";
+import {
+  TG_BOT_UPDATES_DB_URL,
+  tgBotUpdatesServiceSeqTrackerDb,
+  checkDatabasesPromise,
+} from "../dbs";
+import { env } from "../config/env";
+
+export const createServiceChangesTgUpdatesDb = async ({
+  abortController,
+}: {
+  abortController: AbortController;
+}) => {
+  const {
+    tgBotUpdatesDb: { update_seq: lastSeq },
+  } = await checkDatabasesPromise;
+
+  const serviceSeqTracker = await ServiceSeqTracker.Init(
+    "_local/service:tg-bot",
+    tgBotUpdatesServiceSeqTrackerDb,
+  );
+
+  const changesStream = new CouchDBChangesStream<TgBotUpdateDocument>(
+    TG_BOT_UPDATES_DB_URL,
+    {
+      abortController,
+      since: serviceSeqTracker.lastSeq,
+      feed: "continuous",
+      include_docs: true,
+      heartbeat: env.HEARTBEAT_INTERVAL,
+      filter: "_selector",
+      selector: {
+        type: "tg_bot_update",
+      } satisfies Partial<TgBotUpdateDocument>,
+    },
+  );
+
+  console.info(
+    "Service started listening changes tg_bot_update from sequence",
+    serviceSeqTracker.lastSeq,
+  );
+
+  if (
+    parseInt(serviceSeqTracker.lastSeq.toString(), 10) ===
+    parseInt(lastSeq.toString(), 10)
+  ) {
+    console.log("Since is the last seq");
+  } else {
+    console.log("Last seq is", lastSeq);
+  }
+
+  return { changesStream, serviceSeqTracker };
+};

services/tg-bot/src/services/setupWebHook.ts

@@ -0,0 +1,20 @@
+import { setupWebhookDesignDocument } from "./setupWebhookDesignDocument";
+import { TG_BOT_UPDATES_DB_NAME, tgBotUpdatesDesignDocumentsDb } from "../dbs";
+import { env } from "../config/env";
+import { Bot } from "grammy";
+import { MyContext } from "../types/myContext";
+
+export const setupWebHook = async (bot: Bot<MyContext>) => {
+  const designDocument = await setupWebhookDesignDocument();
+  const update: keyof typeof designDocument.updates = "saveUpdate";
+  const webhookUrl = `${env.PUBLIC_API_URL}/${TG_BOT_UPDATES_DB_NAME}/${designDocument._id}/_update/${update}`;
+  if (designDocument.constants.telegram_webhook_url !== webhookUrl) {
+    await bot.api.setWebhook(webhookUrl, {
+      secret_token: designDocument.constants.telegram_secret_token,
+    });
+    designDocument.constants.telegram_webhook_url = webhookUrl;
+    const { rev } = await tgBotUpdatesDesignDocumentsDb.put(designDocument);
+    designDocument._rev = rev;
+  }
+  return designDocument;
+};

services/tg-bot/src/services/setupWebhookDesignDocument.ts

@@ -0,0 +1,71 @@
+import { tgBotUpdatesDesignDocumentsDb } from "../dbs";
+import { saveUpdateString as saveUpdate } from "@hereconnect/tg-bot-update-webhook-ddoc";
+import { type TgBotUpdateWebhookDesignDocument } from "@hereconnect/tg-bot-update-webhook-ddoc";
+
+function generateTelegramSecretToken(length = 256) {
+  const characters =
+    "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789_-";
+  let token = "";
+  for (let i = 0; i < length; i++) {
+    const randomIndex = Math.floor(Math.random() * characters.length);
+    token += characters[randomIndex];
+  }
+  return token;
+}
+
+export const setupWebhookDesignDocument = async () => {
+  const _id = "_design/tg_bot_webhook";
+
+  const result: TgBotUpdateWebhookDesignDocument = {
+    _id,
+    type: _id,
+    created_at: "",
+    updated_at: "",
+    updates: {
+      saveUpdate,
+    },
+    constants: {
+      telegram_secret_token: "",
+    },
+  };
+
+  const { rev } = await tgBotUpdatesDesignDocumentsDb.upsert(_id, (ddoc) => {
+    const isEmptyTelegramSecretToken =
+      !ddoc.constants?.telegram_secret_token?.length;
+    const needToSetSaveUpdateFunction =
+      ddoc.updates?.saveUpdate !== result.updates.saveUpdate;
+
+    if (!needToSetSaveUpdateFunction && !isEmptyTelegramSecretToken) {
+      Object.assign(result, ddoc);
+      return false;
+    }
+
+    ddoc.updated_at = new Date().toISOString();
+    if (!ddoc.created_at) {
+      ddoc.created_at = ddoc.updated_at;
+    }
+
+    if (isEmptyTelegramSecretToken) {
+      if (!ddoc.constants) {
+        ddoc.constants = result.constants;
+      }
+      ddoc.constants.telegram_secret_token = generateTelegramSecretToken();
+    }
+
+    if (needToSetSaveUpdateFunction) {
+      if (ddoc.updates) {
+        ddoc.updates.saveUpdate = result.updates.saveUpdate;
+      } else {
+        ddoc.updates = result.updates;
+      }
+    }
+
+    return Object.assign(
+      result,
+      ddoc,
+    ) satisfies TgBotUpdateWebhookDesignDocument;
+  });
+
+  result._rev = rev;
+  return result;
+};

services/tg-bot/src/types.ts

@@ -1,27 +0,0 @@
-import type { BaseDocument } from "@hereconnect/types";
-import type { Update } from "@grammyjs/types";
-
-/**
- * Интерфейс для документа CouchDB, представляющего обновление Telegram
- */
-export interface TgUpdateDocument extends BaseDocument {
-  /**
-   * Уникальный идентификатор документа, например, "tg_update:<update_id>"
-   */
-  _id: string;
-
-  /**
-   * Ревизия документа (если документ уже существует, автоматически добавляется CouchDB)
-   */
-  _rev?: string;
-
-  /**
-   * Тип документа для идентификации (фиксированное значение "tg_update")
-   */
-  type: "tg_update";
-
-  /**
-   * Данные обновления Telegram
-   */
-  data: Update;
-}

services/tg-bot/src/types/TgBotUpdateDocument.ts

@@ -0,0 +1,12 @@
+import type { Update } from "@grammyjs/types";
+import type { TgBotUpdateDocument as BaseTgBotUpdateDocument } from "@hereconnect/tg-bot-update-webhook-ddoc";
+
+/**
+ * Интерфейс для документа CouchDB, представляющего обновление Telegram
+ */
+export interface TgBotUpdateDocument extends BaseTgBotUpdateDocument {
+  /**
+   * Данные обновления Telegram
+   */
+  data: Update;
+}