hereconnect/services/tg-bot/old/service.ts

import { startHealthServer } from "@hereconnect/lib-service-health-checkserver";
import { ServiceSeqTracker } from "@hereconnect/lib-service-seq-tracker";
import { type AbortSignal } from "grammy/out/shim.node";
import { setupWebHook } from "setup/setupWebHook";

const heartbeat = Number(process.env.HEARTBEAT_INTERVAL ?? 30_000);

import { CouchDBChangesStream } from "@hereconnect/couchdb-changes-stream";
import { TG_BOT_DB_URL, serviceSeqTrackerDb, checkDatabases } from "dbs";
import type { TgUpdateDocument } from "types";
import { init, bot } from "old_bot";

const abortController = new AbortController();

const start = async () => {
  const checkedDatabases = await checkDatabases();

  await init(abortController.signal as AbortSignal);
  await setupWebHook();

  const serviceSeqTracker = await ServiceSeqTracker.Init(
    "_local/service:tg-bot",
    serviceSeqTrackerDb,
  );

  const changesStream = new CouchDBChangesStream<TgUpdateDocument>(
    TG_BOT_DB_URL,
    {
      abortController,
      since: serviceSeqTracker.lastSeq,
      feed: "continuous",
      include_docs: true,
      heartbeat,
      filter: "_selector",
      selector: {
        type: "tg_update" satisfies TgUpdateDocument["type"],
      },
    },
  );

  // Gracefully handle SIGINT
  process.on("SIGINT", async () => {
    changesStream?.stop();
    setTimeout(() => process.exit(1), 5_000).unref();
  });

  console.info("Service started from sequence", serviceSeqTracker.lastSeq);
  if (
    parseInt(serviceSeqTracker.lastSeq.toString(), 10) ===
    parseInt(checkedDatabases.tgBotDb.update_seq.toString(), 10)
  ) {
    console.log("Since is the last seq");
  } else {
    console.log("Last seq is", checkedDatabases.tgBotDb.update_seq);
  }

  await startHealthServer(abortController);

  for await (const change of changesStream) {
    const { seq, doc: changeDoc } = change;

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

    serviceSeqTracker.checkNextSeq(seq);
    await bot.handleUpdate(changeDoc.data);

    // const ctx = new Context(changeDoc.data, bot.api, bot.me);
    // if (ctx.hasCommand("start")) {
    // }

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

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