Merge pull request #181 from harmony-one/add_payment_log

Add logs and DAU metrics

package.json

@@ -57,6 +57,7 @@
     "lokijs": "^1.5.12",
     "lru-cache": "^10.0.0",
     "moment": "^2.29.4",
+    "moment-timezone": "^0.5.43",
     "node-cron": "^3.0.2",
     "openai": "^4.0.1",
     "otpauth": "^9.1.3",

src/bot.ts

@@ -29,15 +29,14 @@ import { BotPayments } from "./modules/payment";
 import { BotSchedule } from "./modules/schedule";
 import config from "./config";
 import { commandsHelpText, TERMS, SUPPORT, FEEDBACK, LOVE } from "./constants";
-import prometheusRegister from "./metrics/prometheus";
+import prometheusRegister, {PrometheusMetrics} from "./metrics/prometheus";
 
 import { chatService, statsService } from "./database/services";
 import { AppDataSource } from "./database/datasource";
-import { text } from "stream/consumers";
 import { autoRetry } from "@grammyjs/auto-retry";
 import {run} from "@grammyjs/runner";
 import {runBotHeartBit} from "./monitoring/monitoring";
-
+import {BotPaymentLog} from "./database/stats.service";
 const logger = pino({
   name: "bot",
   transport: {
@@ -185,6 +184,30 @@ bot.use((ctx, next) => {
   return next();
 });
 
+const writeCommandLog = async (ctx: OnMessageContext, isSupportedCommand = true) => {
+  const { from, text = '', chat } = ctx.update.message
+
+  try {
+    const accountId = payments.getAccountId(ctx);
+    const [command] = text?.split(' ')
+
+    const log: BotPaymentLog = {
+      tgUserId: from.id,
+      accountId,
+      command,
+      groupId: chat.id,
+      isPrivate: chat.type === 'private',
+      message: text,
+      isSupportedCommand,
+      amountCredits: 0,
+      amountOne: 0,
+    }
+    await statsService.writeLog(log)
+  } catch (e) {
+    logger.error(`Cannot write unsupported command log: ${(e as Error).message}`)
+  }
+}
+
 const onMessage = async (ctx: OnMessageContext) => {
   try {
     await assignFreeCredits(ctx);
@@ -280,16 +303,21 @@ const onMessage = async (ctx: OnMessageContext) => {
     }
     // if (ctx.update.message.text && ctx.update.message.text.startsWith("/", 0)) {
     //  const command = ctx.update.message.text.split(' ')[0].slice(1)
-    // onlfy for private chats
+    // only for private chats
     if (ctx.update.message.chat && ctx.chat.type === "private") {
-      await ctx.reply(`Unsupported, type */help* for commands.`, {
-        parse_mode: "Markdown",
-      });
+      await ctx.reply(
+          `Unsupported, type */help* for commands.`,
+          {
+            parse_mode: "Markdown",
+          }
+      );
+      await writeCommandLog(ctx, false)
       return;
     }
     if (ctx.update.message.chat) {
       logger.info(`Received message in chat id: ${ctx.update.message.chat.id}`);
     }
+    await writeCommandLog(ctx, false)
   } catch (ex: any) {
     console.error("onMessage error", ex);
   }
@@ -325,6 +353,8 @@ bot.command(["start", "help", "menu"], async (ctx) => {
     return false;
   }
 
+  await writeCommandLog(ctx as OnMessageContext)
+
   const addressBalance = await payments.getAddressBalance(account.address);
   const credits = await chatService.getBalance(accountId);
   const balance = addressBalance.plus(credits);
@@ -341,34 +371,39 @@ bot.command(["start", "help", "menu"], async (ctx) => {
 });
 
 bot.command("more", async (ctx) => {
+  writeCommandLog(ctx as OnMessageContext)
   return ctx.reply(commandsHelpText.more, {
     parse_mode: "Markdown",
     disable_web_page_preview: true,
   });
 });
 
 bot.command("terms", (ctx) => {
+  writeCommandLog(ctx as OnMessageContext)
   return ctx.reply(TERMS.text, {
     parse_mode: "Markdown",
     disable_web_page_preview: true,
   });
 });
 
 bot.command("support", (ctx) => {
+  writeCommandLog(ctx as OnMessageContext)
   return ctx.reply(SUPPORT.text, {
     parse_mode: "Markdown",
     disable_web_page_preview: true,
   });
 });
 
 bot.command("feedback", (ctx) => {
+  writeCommandLog(ctx as OnMessageContext)
   return ctx.reply(FEEDBACK.text, {
     parse_mode: "Markdown",
     disable_web_page_preview: true,
   });
 });
 
 bot.command("love", (ctx) => {
+  writeCommandLog(ctx as OnMessageContext)
   return ctx.reply(LOVE.text, {
     parse_mode: "Markdown",
     disable_web_page_preview: true,
@@ -444,7 +479,12 @@ const stopRunner = () => {
 process.once("SIGINT", stopRunner);
 process.once("SIGTERM", stopRunner);
 
-AppDataSource.initialize();
+AppDataSource.initialize().then(() => {
+  const prometheusMetrics = new PrometheusMetrics()
+  prometheusMetrics.bootstrap()
+}).catch((e) => {
+  logger.error(`Error during DB initialization: ${(e as Error).message}`)
+})
 
 if (config.betteruptime.botHeartBitId) {
   const task = runBotHeartBit(runner, config.betteruptime.botHeartBitId);

src/database/datasource.ts

@@ -4,12 +4,13 @@ import { Chat } from "./entities/Chat"
 import { User } from "./entities/User";
 import { StatBotCommand } from "./entities/StatBotCommand";
 import config from "../config"
+import {BotLog} from "./entities/Log";
 
 export const AppDataSource = new DataSource({
   type: "postgres",
   url: config.db.url,
-  entities: [Chat, User, StatBotCommand],
-  synchronize: false,
+  entities: [Chat, User, StatBotCommand, BotLog],
+  synchronize: true,
   migrations: ['./src/database/migrations/**/*.{.ts,.js}'],
   logging: false,
 })

src/database/entities/Log.ts

@@ -0,0 +1,37 @@
+import {Entity, Column, PrimaryGeneratedColumn, CreateDateColumn} from 'typeorm';
+
+@Entity({ name: 'logs' })
+export class BotLog {
+  @PrimaryGeneratedColumn()
+  id!: number;
+
+  @Column({type: "bigint"})
+  tgUserId: number;
+
+  @Column({type: "bigint"})
+  accountId: number;
+
+  @Column({type: "bigint"})
+  groupId: number;
+
+  @Column({ default: false })
+  isPrivate: boolean;
+
+  @Column()
+  command: string;
+
+  @Column()
+  message: string;
+
+  @Column({ default: false })
+  isSupportedCommand: boolean;
+
+  @Column({ type: 'decimal', precision: 10, scale: 8, default: 0 })
+  amountOne: number;
+
+  @Column({ type: 'decimal', precision: 10, scale: 8, default: 0 })
+  amountCredits: number;
+
+  @CreateDateColumn()
+  createdAt: Date;
+}

src/database/migrations/1692905335026-AddLogs.ts

@@ -0,0 +1,14 @@
+import { MigrationInterface, QueryRunner } from "typeorm";
+
+export class AddLogs1692905335026 implements MigrationInterface {
+    name = 'AddLogs1692905335026'
+
+    public async up(queryRunner: QueryRunner): Promise<void> {
+        await queryRunner.query(`CREATE TABLE "logs" ("id" SERIAL NOT NULL, "tgUserId" bigint NOT NULL, "accountId" bigint NOT NULL, "groupId" bigint NOT NULL, "isPrivate" boolean NOT NULL DEFAULT false, "command" character varying NOT NULL, "message" character varying NOT NULL, "isSupportedCommand" boolean NOT NULL DEFAULT false, "amountOne" numeric(10,8) NOT NULL DEFAULT '0', "amountCredits" numeric(10,8) NOT NULL DEFAULT '0', "createdAt" TIMESTAMP NOT NULL DEFAULT now(), CONSTRAINT "PK_fb1b805f2f7795de79fa69340ba" PRIMARY KEY ("id"))`);
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<void> {
+        await queryRunner.query(`DROP TABLE "logs"`);
+    }
+
+}

src/database/stats.service.ts

@@ -1,10 +1,80 @@
 import {AppDataSource} from "./datasource";
 import {StatBotCommand} from "./entities/StatBotCommand";
-import moment from "moment/moment";
+import moment from "moment-timezone";
+import {BotLog} from "./entities/Log";
 
 const statBotCommandRepository = AppDataSource.getRepository(StatBotCommand);
+const logRepository = AppDataSource.getRepository(BotLog);
+
+export interface BotPaymentLog {
+  tgUserId: number
+  accountId: number
+  groupId: number
+  isPrivate: boolean
+  command: string
+  message: string
+  isSupportedCommand: boolean
+  amountOne: number
+  amountCredits: number
+}
 
 export class StatsService {
+  public writeLog(log: BotPaymentLog) {
+    let paymentLog = new BotLog()
+    paymentLog = {
+      ...paymentLog,
+      ...log,
+      message: (log.message || '').trim().slice(0, 1024)
+    }
+    return logRepository.save(paymentLog);
+  }
+
+  async getTotalONE() {
+    const [result] = await logRepository.query(`select sum("amountOne") from logs`)
+    if(result) {
+      return +result.sum
+    }
+    return 0
+  }
+
+  async getTotalFreeCredits() {
+    const [result] = await logRepository.query(`select sum("amountCredits") from logs`)
+    if(result) {
+      return +result.sum
+    }
+    return 0
+  }
+
+  async getUniqueUsersCount() {
+    const [result] = await logRepository.query(`select count(distinct("tgUserId")) from logs`)
+    if(result) {
+      return +result.count
+    }
+    return 0
+  }
+
+  public async getDAUFromLogs() {
+    const currentTime = moment();
+    const dateStart = moment()
+      .tz('America/Los_Angeles')
+      .set({ hour: 11, minute: 59, second: 0 })
+      .unix()
+
+    const dateEnd = currentTime.unix();
+
+    const rows = await logRepository.query(`
+        select count(logs."tgUserId") from logs
+        where logs."createdAt" BETWEEN TO_TIMESTAMP($1) and TO_TIMESTAMP($2)
+        group by logs."tgUserId"
+      `,
+      [dateStart, dateEnd])
+
+    if(rows.length > 0) {
+      return +rows[0].count
+    }
+    return 0
+  }
+
   public addCommandStat({tgUserId, rawMessage, command}: {tgUserId: number, rawMessage: string, command: string}) {
     const stat = new StatBotCommand();
 
@@ -16,9 +86,12 @@ export class StatsService {
   }
 
   public async getDAU() {
-    const currentTime = moment(); // Get the current time
+    const currentTime = moment();
     const eightPm = moment().set({ hour: 20, minute: 0, second: 0 });
-    let dateStart = moment().set({ hour: 20, minute: 0, second: 0 }).subtract(1, 'days').unix();
+    let dateStart = moment()
+      .set({ hour: 20, minute: 0, second: 0 })
+      .subtract(1, 'days')
+      .unix();
 
     if (currentTime.isAfter(eightPm)) {
       dateStart = eightPm.unix();
@@ -48,4 +121,4 @@ export class StatsService {
 
     return rows.length;
   }
-}
+}

src/metrics/prometheus.ts

@@ -1,4 +1,6 @@
 import client from 'prom-client'
+import {statsService} from "../database/services";
+
 
 const register = new client.Registry()
 
@@ -8,23 +10,60 @@ register.setDefaultLabels({
 
 client.collectDefaultMetrics({register})
 
-
-export const usdFeeCounter = new client.Counter({
-  name: 'fee_usd',
-  help: 'fee_help'
-})
 export const oneTokenFeeCounter = new client.Counter({
   name: 'fee_one',
-  help: 'fee_help'
+  help: 'Fees earned by the bot in ONE tokens'
 })
 
-export const creditsFeeCounter = new client.Counter({
+export const freeCreditsFeeCounter = new client.Counter({
   name: 'free_credits_fee',
-  help: 'free_credits_fee_help'
+  help: 'Free credits spent by users'
+})
+
+export const uniqueUsersCounter = new client.Counter({
+  name: 'total_unique_users',
+  help: 'Number of unique users'
+})
+
+export const dauCounter = new client.Counter({
+  name: 'daily_active_users',
+  help: 'Number of daily active users'
 })
 
-register.registerMetric(usdFeeCounter)
 register.registerMetric(oneTokenFeeCounter)
-register.registerMetric(creditsFeeCounter)
+register.registerMetric(freeCreditsFeeCounter)
+register.registerMetric(uniqueUsersCounter)
+register.registerMetric(dauCounter)
+
+export class PrometheusMetrics {
+  constructor() {
+    setInterval(() => this.updateDau(), 30 * 60 * 1000)
+  }
+
+  public async bootstrap() {
+    try {
+      const totalOne = await statsService.getTotalONE()
+      const freeCredits = await statsService.getTotalFreeCredits()
+      const uniqueUsersCount = await statsService.getUniqueUsersCount()
+      const dauValue = await statsService.getDAUFromLogs()
+      oneTokenFeeCounter.inc(totalOne)
+      freeCreditsFeeCounter.inc(freeCredits)
+      uniqueUsersCounter.inc(uniqueUsersCount)
+      dauCounter.inc(dauValue)
+      console.log(`Prometheus bootstrap: total ONE: ${totalOne}, freeCredits: ${freeCredits}, uniqueUsersCount: ${uniqueUsersCount}, dau: ${dauValue}`)
+    } catch (e) {
+      console.log('Prometheus bootstrap error:', (e as Error).message)
+    }
+  }
+
+  async updateDau() {
+    try {
+      const dauValue = await statsService.getDAUFromLogs()
+      dauCounter.inc(dauValue)
+    } catch (e) {
+      console.log('Prometheus interval update error:', (e as Error).message)
+    }
+  }
+}
 
 export default register