UBERF-9240 Report session and egress stats to billing service (#7798)

services/love/src/billing.ts

@@ -0,0 +1,143 @@
+//
+// Copyright © 2025 Hardcore Engineering Inc.
+//
+// Licensed under the Eclipse Public License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License. You may
+// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+import { concatLink, MeasureContext, systemAccountEmail } from '@hcengineering/core'
+import { generateToken } from '@hcengineering/server-token'
+import { AccessToken, EgressInfo } from 'livekit-server-sdk'
+import config from './config'
+
+// Example:
+//  {
+//    "roomId": "RM_ROOM_ID",
+//    "roomName": "w-room-name",
+//    "numParticipants": 2,
+//    "bandwidth": "1000",
+//    "connectionMinutes": "120",
+//    "startTime": "2025-01-01T12:00:00Z",
+//    "endTime": "2025-01-01T13:00:00Z",
+//    "participants": [ ... ]
+//  }
+interface LiveKitSession {
+  roomId: string
+  roomName: string
+  startTime: string
+  endTime: string
+  bandwidth: string
+  connectionMinutes: string
+}
+
+async function getLiveKitSession (ctx: MeasureContext, sessionId: string): Promise<LiveKitSession> {
+  const token = await createAnalyticsToken()
+  const endpoint = `https://cloud-api.livekit.io/api/project/${config.LiveKitProject}/sessions/${sessionId}?v=2`
+  try {
+    const response = await fetch(endpoint, {
+      method: 'GET',
+      headers: {
+        Authorization: `Bearer ${token}`,
+        'Content-Type': 'application/json'
+      }
+    })
+
+    if (!response.ok) {
+      ctx.error('failed to get session analytics', { session: sessionId, status: response.status })
+      throw new Error(`HTTP error: ${response.status} ${response.statusText}`)
+    }
+
+    return (await response.json()) as LiveKitSession
+  } catch (error) {
+    throw new Error('Failed to get session analytics')
+  }
+}
+
+const createAnalyticsToken = async (): Promise<string> => {
+  const at = new AccessToken(config.ApiKey, config.ApiSecret, { ttl: '10m' })
+  at.addGrant({ roomList: true })
+
+  return await at.toJwt()
+}
+
+export async function saveLiveKitSessionBilling (ctx: MeasureContext, sessionId: string): Promise<void> {
+  if (config.BillingUrl === '' || config.LiveKitProject === '') {
+    return
+  }
+
+  const session = await getLiveKitSession(ctx, sessionId)
+  const workspace = session.roomName.split('_')[0]
+  const endpoint = concatLink(config.BillingUrl, `/api/v1/billing/${workspace}/livekit/session`)
+
+  const token = generateToken(systemAccountEmail, { name: workspace })
+
+  try {
+    const res = await fetch(endpoint, {
+      method: 'POST',
+      headers: {
+        Authorization: `Bearer ${token}`,
+        'Content-Type': 'application/json'
+      },
+      body: JSON.stringify({
+        sessionId,
+        sessionStart: session.startTime,
+        sessionEnd: session.endTime,
+        bandwidth: Number(session.bandwidth),
+        minutes: Number(session.connectionMinutes),
+        room: session.roomName
+      })
+    })
+    if (!res.ok) {
+      throw new Error(await res.text())
+    }
+  } catch (err: any) {
+    ctx.error('failed to save session billing', { workspace, session, err })
+    throw new Error('Failed to save session billing: ' + err)
+  }
+}
+
+export async function saveLiveKitEgressBilling (ctx: MeasureContext, egress: EgressInfo): Promise<void> {
+  if (config.BillingUrl === '') {
+    return
+  }
+
+  const egressStart = Number(egress.startedAt) / 1000 / 1000
+  const egressEnd = Number(egress.endedAt) / 1000 / 1000
+  const duration = (egressEnd - egressStart) / 1000
+
+  const workspace = egress.roomName.split('_')[0]
+  const endpoint = concatLink(config.BillingUrl, `/api/v1/billing/${workspace}/livekit/egress`)
+
+  const token = generateToken(systemAccountEmail, { name: workspace })
+
+  try {
+    const res = await fetch(endpoint, {
+      method: 'POST',
+      headers: {
+        Authorization: `Bearer ${token}`,
+        'Content-Type': 'application/json'
+      },
+      body: JSON.stringify({
+        room: egress.roomName,
+        egressId: egress.egressId,
+        egressStart: new Date(egressStart).toISOString(),
+        egressEnd: new Date(egressEnd).toISOString(),
+        duration
+      })
+    })
+    if (!res.ok) {
+      throw new Error(await res.text())
+    }
+  } catch (err: any) {
+    ctx.error('failed to save egress billing', { workspace, egress, err })
+    throw new Error('Failed to save egress billing: ' + err)
+  }
+}

services/love/src/config.ts

@@ -18,6 +18,7 @@ interface Config {
   Port: number
   ServiceID: string
 
+  LiveKitProject: string
   LiveKitHost: string
   ApiKey: string
   ApiSecret: string
@@ -28,12 +29,14 @@ interface Config {
   Secret: string
 
   MongoUrl: string
+  BillingUrl: string
 }
 
 const envMap: { [key in keyof Config]: string } = {
   AccountsURL: 'ACCOUNTS_URL',
   Port: 'PORT',
 
+  LiveKitProject: 'LIVEKIT_PROJECT',
   LiveKitHost: 'LIVEKIT_HOST',
   ApiKey: 'LIVEKIT_API_KEY',
   ApiSecret: 'LIVEKIT_API_SECRET',
@@ -43,7 +46,8 @@ const envMap: { [key in keyof Config]: string } = {
   S3StorageConfig: 'S3_STORAGE_CONFIG',
   Secret: 'SECRET',
   ServiceID: 'SERVICE_ID',
-  MongoUrl: 'MONGO_URL'
+  MongoUrl: 'MONGO_URL',
+  BillingUrl: 'BILLING_URL'
 }
 
 const parseNumber = (str: string | undefined): number | undefined => (str !== undefined ? Number(str) : undefined)
@@ -52,6 +56,7 @@ const config: Config = (() => {
   const params: Partial<Config> = {
     AccountsURL: process.env[envMap.AccountsURL],
     Port: parseNumber(process.env[envMap.Port]) ?? 8096,
+    LiveKitProject: process.env[envMap.LiveKitProject] ?? '',
     LiveKitHost: process.env[envMap.LiveKitHost],
     ApiKey: process.env[envMap.ApiKey],
     ApiSecret: process.env[envMap.ApiSecret],
@@ -60,10 +65,11 @@ const config: Config = (() => {
     S3StorageConfig: process.env[envMap.S3StorageConfig],
     Secret: process.env[envMap.Secret],
     ServiceID: process.env[envMap.ServiceID] ?? 'love-service',
-    MongoUrl: process.env[envMap.MongoUrl]
+    MongoUrl: process.env[envMap.MongoUrl],
+    BillingUrl: process.env[envMap.BillingUrl] ?? ''
   }
 
-  const optional = ['StorageConfig', 'S3StorageConfig']
+  const optional = ['StorageConfig', 'S3StorageConfig', 'LiveKitProject', 'BillingUrl']
 
   const missingEnv = (Object.keys(params) as Array<keyof Config>)
     .filter((key) => !optional.includes(key))

services/love/src/main.ts

@@ -33,6 +33,7 @@ import {
   WebhookReceiver
 } from 'livekit-server-sdk'
 import config from './config'
+import { saveLiveKitEgressBilling, saveLiveKitSessionBilling } from './billing'
 import { getS3UploadParams, saveFile } from './storage'
 import { WorkspaceClient } from './workspaceClient'
 
@@ -96,6 +97,13 @@ export const main = async (): Promise<void> => {
             console.log('no data found for', res.filename)
           }
         }
+
+        await saveLiveKitEgressBilling(ctx, event.egressInfo)
+
+        res.send()
+        return
+      } else if (event.event === 'room_finished' && event.room !== undefined) {
+        await saveLiveKitSessionBilling(ctx, event.room.sid)
         res.send()
         return
       }