feat(billing): billing service (#325)

* feat(billing): billing service

* better destroy

* fix

* better billing

packages/server/src/app.ts

@@ -1,4 +1,5 @@
 import { Engine } from '@botpress/messaging-engine'
+import { BillingService } from './billing/service'
 import { ChannelService } from './channels/service'
 import { ClientTokenService } from './client-tokens/service'
 import { ClientService } from './clients/service'
@@ -36,6 +37,7 @@ export class App extends Engine {
   syncs: SyncService
   health: HealthService
   sockets: SocketService
+  billing: BillingService
 
   constructor() {
     super()
@@ -94,6 +96,7 @@ export class App extends Engine {
       this.instances
     )
     this.sockets = new SocketService(this.caching, this.users)
+    this.billing = new BillingService(this.logger, this.conversations, this.messages)
   }
 
   async setup() {
@@ -124,10 +127,12 @@ export class App extends Engine {
   async monitor() {
     await this.syncs.setup()
     await this.instances.monitor()
+    await this.billing.setup()
   }
 
   async destroy() {
     await this.batching?.destroy()
+    await this.billing?.destroy()
     await this.instances?.destroy()
     await this.distributed?.destroy()
     await this.database?.destroy()

packages/server/src/billing/service.ts

@@ -0,0 +1,98 @@
+import { Logger, LoggerService, Service } from '@botpress/messaging-engine'
+import axios from 'axios'
+import clc from 'cli-color'
+import ms from 'ms'
+import { ConversationService } from '../conversations/service'
+import { MessageCreatedEvent, MessageEvents } from '../messages/events'
+import { MessageService } from '../messages/service'
+
+export class BillingService extends Service {
+  private logger: Logger
+  private timeout?: NodeJS.Timeout
+  private stats: { [clientId: string]: { received: number; sent: number } } = {}
+
+  constructor(loggers: LoggerService, private conversations: ConversationService, private messages: MessageService) {
+    super()
+    this.logger = loggers.root.sub('billing')
+  }
+
+  async setup() {
+    if (process.env.BILLING_ENDPOINT?.length) {
+      this.messages.events.on(MessageEvents.Created, this.handleMessageCreated.bind(this))
+
+      void this.tickBilling()
+    }
+  }
+
+  async destroy() {
+    try {
+      if (this.timeout) {
+        clearTimeout(this.timeout)
+      }
+
+      await this.flushBilling()
+    } catch (e) {
+      this.logger.error(e, 'Failed to destroy billing')
+    }
+  }
+
+  private async tickBilling() {
+    try {
+      await this.flushBilling()
+    } catch (e) {
+      this.logger.error(e, 'Error occurred in billing')
+    } finally {
+      this.timeout = setTimeout(this.tickBilling.bind(this), ms('3s'))
+    }
+  }
+
+  private async handleMessageCreated({ message }: MessageCreatedEvent) {
+    const conversation = await this.conversations.get(message.conversationId)
+    const stat = this.stats[conversation.clientId] || { sent: 0, received: 0 }
+
+    if (message.authorId) {
+      stat.received++
+    } else {
+      stat.sent++
+    }
+
+    this.stats[conversation.clientId] = stat
+  }
+
+  private async flushBilling() {
+    const entries = [...Object.entries(this.stats)]
+
+    for (const [clientId, stat] of entries) {
+      const sentStats = { ...stat }
+      const timestamp = new Date().toISOString()
+
+      stat.sent = 0
+      stat.received = 0
+
+      await axios.post(process.env.BILLING_ENDPOINT!, {
+        meta: {
+          timestamp,
+          sender: 'messaging',
+          type: 'messages_processed',
+          schema_version: '1.0.0'
+        },
+        schema_version: '1.0.0',
+        records: [
+          {
+            client_id: clientId,
+            messages: sentStats,
+            timestamp
+          }
+        ]
+      })
+
+      // it's possible that new stats have been entered while the post request
+      // was waiting since it's async
+      if (stat.received === 0 && stat.sent === 0) {
+        delete this.stats[clientId]
+      }
+
+      this.logger.info(`${clc.blackBright(`[${clientId}]`)} ${clc.cyan('stats')}`, sentStats)
+    }
+  }
+}

packages/server/src/global.ts

@@ -13,6 +13,7 @@ export type MessagingEnv = EngineEnv & {
   NO_LOGO?: string
   APM_ENABLED?: string
   TWILIO_TESTING?: string
+  BILLING_ENDPOINT?: string
 }
 
 declare global {