Add metrics to redirect service

bskylink/src/config.ts

@@ -15,6 +15,7 @@ export type ServiceConfig = {
   safelinkPdsUrl?: string
   safelinkAgentIdentifier?: string
   safelinkAgentPass?: string
+  metricsApiHost?: string
 }
 
 export type DbConfig = {
@@ -45,6 +46,7 @@ export type Environment = {
   safelinkPdsUrl?: string
   safelinkAgentIdentifier?: string
   safelinkAgentPass?: string
+  metricsApiHost?: string
 }
 
 export const readEnv = (): Environment => {
@@ -65,6 +67,7 @@ export const readEnv = (): Environment => {
     safelinkPdsUrl: envStr('LINK_SAFELINK_PDS_URL'),
     safelinkAgentIdentifier: envStr('LINK_SAFELINK_AGENT_IDENTIFIER'),
     safelinkAgentPass: envStr('LINK_SAFELINK_AGENT_PASS'),
+    metricsApiHost: envStr('LINK_METRICS_API_HOST'),
   }
 }
 
@@ -79,6 +82,7 @@ export const envToCfg = (env: Environment): Config => {
     safelinkPdsUrl: env.safelinkPdsUrl,
     safelinkAgentIdentifier: env.safelinkAgentIdentifier,
     safelinkAgentPass: env.safelinkAgentPass,
+    metricsApiHost: env.metricsApiHost,
   }
   if (!env.dbPostgresUrl) {
     throw new Error('Must configure postgres url (LINK_DB_POSTGRES_URL)')

bskylink/src/context.ts

@@ -1,6 +1,7 @@
 import {SafelinkClient} from './cache/safelinkClient.js'
 import {type Config} from './config.js'
 import Database from './db/index.js'
+import {MetricsClient} from './metrics.js'
 
 export type AppContextOptions = {
   cfg: Config
@@ -12,6 +13,7 @@ export class AppContext {
   db: Database
   safelinkClient: SafelinkClient
   abortController = new AbortController()
+  metrics: MetricsClient
 
   constructor(private opts: AppContextOptions) {
     this.cfg = this.opts.cfg
@@ -20,6 +22,9 @@ export class AppContext {
       cfg: this.opts.cfg.service,
       db: this.opts.db,
     })
+    this.metrics = new MetricsClient({
+      trackingEndpoint: this.opts.cfg.service.metricsApiHost,
+    })
   }
 
   static async fromConfig(cfg: Config, overrides?: Partial<AppContextOptions>) {

bskylink/src/index.ts

@@ -36,6 +36,7 @@ export class LinkService {
   }
 
   async start() {
+    this.ctx.metrics.start()
     this.server = this.app.listen(this.ctx.cfg.service.port)
     this.server.keepAliveTimeout = 90000
     this.terminator = createHttpTerminator({server: this.server})
@@ -46,5 +47,6 @@ export class LinkService {
     this.ctx.abortController.abort()
     await this.terminator?.terminate()
     await this.ctx.db.close()
+    this.ctx.metrics.stop()
   }
 }

bskylink/src/metrics.ts

@@ -0,0 +1,136 @@
+import crypto from 'node:crypto'
+
+import {httpLogger} from './logger.js'
+
+/**
+ * New metrics events should be added here
+ */
+type Events = {
+  redirect: {
+    link: string
+    whitelisted: 'unknown' | 'yes'
+    blocked: boolean
+    warned: boolean
+  }
+  invalid_redirect: {
+    link: string
+  }
+}
+
+type Event<M extends Record<string, any>> = {
+  time: number
+  event: keyof M
+  payload: M[keyof M]
+  metadata: Record<string, any>
+}
+
+export type Config = {
+  trackingEndpoint?: string
+}
+
+/**
+ * This MetricsClient is duplicated from both `social-app` and `atproto`
+ * codebases.
+ */
+export class MetricsClient<M extends Record<string, any> = Events> {
+  maxBatchSize = 100
+
+  private disabled: boolean = false
+  private started: boolean = false
+  private queue: Event<M>[] = []
+  private flushInterval: NodeJS.Timeout | null = null
+  constructor(private config: Config) {
+    this.disabled = !config.trackingEndpoint
+  }
+
+  start() {
+    if (this.disabled) return
+    if (this.started) return
+    this.started = true
+    this.flushInterval = setInterval(() => {
+      this.flush()
+    }, 10_000)
+  }
+
+  stop() {
+    if (this.flushInterval) {
+      clearInterval(this.flushInterval)
+      this.flushInterval = null
+    }
+    this.flush()
+  }
+
+  track<E extends keyof M>(event: E, payload: M[E]) {
+    if (this.disabled) return
+
+    this.start()
+
+    /**
+     * deviceId is required for sharding events in Middleman. To avoid a hot
+     * shard, we generate a random anonymous IDs for this client.
+     *
+     * @see https://github.com/bluesky-social/tango/blob/d5819cde419d13e0d2cf837f4b30d48529d64060/middleman/handlers_tracking.go#L195
+     */
+    const anonId = `anon-${crypto.randomUUID()}`
+
+    /**
+     * Event structure is like this to ensure compat with Middleman, which
+     * receives events like this from other codebases, including `social-app`.
+     */
+    const e = {
+      source: 'blink',
+      time: Date.now(),
+      event,
+      payload,
+      metadata: {
+        base: {
+          deviceId: anonId,
+          sessionId: anonId,
+        },
+        session: {
+          did: undefined,
+        },
+      },
+    }
+    this.queue.push(e)
+
+    if (this.queue.length > this.maxBatchSize) {
+      this.flush()
+    }
+  }
+
+  flush() {
+    if (this.disabled) return
+    if (!this.queue.length) return
+    const events = this.queue.splice(0, this.queue.length)
+    this.sendBatch(events)
+  }
+
+  private async sendBatch(events: Event<M>[]) {
+    if (this.disabled || !this.config.trackingEndpoint) return
+
+    try {
+      const res = await fetch(this.config.trackingEndpoint, {
+        method: 'POST',
+        headers: {
+          'Content-Type': 'application/json',
+        },
+        body: JSON.stringify({events}),
+        keepalive: true,
+      })
+
+      if (!res.ok) {
+        const errorText = await res.text().catch(() => 'Unknown error')
+        httpLogger.error(
+          {err: new Error(`${res.status} Failed to fetch - ${errorText}`)},
+          'Failed to send metrics',
+        )
+      } else {
+        // Drain response body to allow connection reuse.
+        await res.text().catch(() => {})
+      }
+    } catch (err) {
+      httpLogger.error({err}, 'Failed to send metrics')
+    }
+  }
+}

bskylink/src/routes/redirect.ts

@@ -37,6 +37,7 @@ export default function (ctx: AppContext, app: Express) {
           url.pathname === '/redirect') || // is a redirect loop
         INTERNAL_IP_REGEX.test(url.hostname) // isn't directing to an internal location
       ) {
+        ctx.metrics.track('invalid_redirect', {link})
         res.setHeader('Cache-Control', 'no-store')
         res.setHeader('Location', `https://${ctx.cfg.service.appHostname}`)
         return res.status(302).end()
@@ -48,6 +49,9 @@ export default function (ctx: AppContext, app: Express) {
       res.type('html')
 
       let html: string | undefined
+      let whitelisted: 'unknown' | 'yes' = 'unknown'
+      let blocked: boolean = false
+      let warned: boolean = false
 
       if (ctx.cfg.service.safelinkEnabled) {
         const rule = await ctx.safelinkClient.tryFindRule(link)
@@ -55,6 +59,7 @@ export default function (ctx: AppContext, app: Express) {
           switch (rule.action) {
             case 'whitelist':
               redirectLogger.info({rule}, 'Whitelist rule matched')
+              whitelisted = 'yes'
               break
             case 'block':
               html = linkWarningLayout(
@@ -66,6 +71,7 @@ export default function (ctx: AppContext, app: Express) {
               )
               res.setHeader('Cache-Control', 'no-store')
               redirectLogger.info({rule}, 'Block rule matched')
+              blocked = true
               break
             case 'warn':
               html = linkWarningLayout(
@@ -77,6 +83,7 @@ export default function (ctx: AppContext, app: Express) {
               )
               res.setHeader('Cache-Control', 'no-store')
               redirectLogger.info({rule}, 'Warn rule matched')
+              warned = true
               break
             default:
               redirectLogger.warn({rule}, 'Unknown rule matched')
@@ -89,6 +96,13 @@ export default function (ctx: AppContext, app: Express) {
         html = linkRedirectContents(url.href)
       }
 
+      ctx.metrics.track('redirect', {
+        link,
+        whitelisted,
+        blocked,
+        warned,
+      })
+
       return res.end(html)
     }),
   )