add prometheus metrics endpoint

- Add @effect/opentelemetry with PrometheusExporter
- Expose /metrics on port 9464
- Track messages processed, list operations, cursor position
- Track memory (heap, RSS) every 30 seconds
- Fix BufferOverflowError class, index exports, Cursor double-write

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: kristojorg

bun.lockb

@@ -47,8 +47,18 @@
   "dependencies": {
     "@atcute/cbor": "^2.1.1",
     "@atproto/api": "^0.14.21",
+    "@effect/opentelemetry": "^0.60.0",
     "@effect/platform": "^0.72.2",
     "@effect/platform-bun": "^0.52.2",
+    "@opentelemetry/api": "^1.9.0",
+    "@opentelemetry/core": "^2.4.0",
+    "@opentelemetry/exporter-prometheus": "^0.210.0",
+    "@opentelemetry/resources": "^2.4.0",
+    "@opentelemetry/sdk-logs": "^0.210.0",
+    "@opentelemetry/sdk-metrics": "^2.4.0",
+    "@opentelemetry/sdk-trace-base": "^2.4.0",
+    "@opentelemetry/sdk-trace-node": "^2.4.0",
+    "@opentelemetry/semantic-conventions": "^1.39.0",
     "dotenv": "^16.4.7",
     "effect": "^3.12.2",
     "semver": "^7.6.3"

package.json

@@ -79,11 +79,6 @@ const writeCursor = (env: Env, fs: FileSystem.FileSystem) => (cursor: number) =>
   Effect.gen(function*() {
     yield* Effect.log("Writing cursor to file: ", cursor)
     const { labelerCursorFilepath } = env
-    yield* Effect.tryPromise({
-      try: () => Bun.write(labelerCursorFilepath, cursor.toString()),
-      catch: (cause) =>
-        new CursorError({ message: "Failed to write cursor", cause }),
-    })
     yield* fs.writeFileString(labelerCursorFilepath, cursor.toString())
   }).pipe(Effect.catchAllCause(Effect.logError))
 

src/Cursor.ts

@@ -1,7 +1,8 @@
 import { AtpListAccountAgent } from "@/AtpAgent";
 import { Env } from "@/Environment";
+import { cursorGauge, listOperations, messagesProcessed } from "@/Metrics";
 import { decodeFirst } from "@atcute/cbor";
-import { Data, Effect, Layer, Schema, Stream } from "effect";
+import { Data, Effect, Layer, Metric, Schema, Stream } from "effect";
 import { Cursor } from "./Cursor";
 import { RetryingSocket } from "./RetryingSocket";
 import { MessageLabels, parseSubscribeLabelsMessage } from "./schema";
@@ -67,15 +68,22 @@ export const LabelWatcherLive = Layer.scopedDiscard(run).pipe(
  */
 const handleLabel = (agent: AtpListAccountAgent) => (label: MessageLabels) =>
   Effect.gen(function* () {
+    yield* Metric.increment(messagesProcessed);
+
     const labels = label.body.labels;
     for (const label of labels) {
       if (label.neg) {
         yield* agent.removeUserFromList(label.uri, label.val);
+        yield* Metric.increment(Metric.tagged(listOperations, "op", "remove"));
         continue;
       }
       yield* agent.addUserToList(label.uri, label.val);
+      yield* Metric.increment(Metric.tagged(listOperations, "op", "add"));
     }
-    return label.body.seq;
+
+    const seq = label.body.seq;
+    yield* Metric.set(cursorGauge, seq);
+    return seq;
   });
 
 const parseMessage = (u: Uint8Array) =>

src/LabelWatcher.ts

@@ -0,0 +1,63 @@
+import * as NodeSdk from "@effect/opentelemetry/NodeSdk"
+import { PrometheusExporter } from "@opentelemetry/exporter-prometheus"
+import { Config, Effect, Layer, Metric, Schedule } from "effect"
+
+// App metrics
+export const messagesProcessed = Metric.counter("messages_processed_total", {
+  description: "Total websocket messages processed",
+})
+
+export const cursorGauge = Metric.gauge("cursor_value", {
+  description: "Current cursor position",
+})
+
+export const listOperations = Metric.counter("list_operations_total", {
+  description: "List add/remove operations",
+})
+
+export const wsReconnects = Metric.counter("websocket_reconnects_total", {
+  description: "WebSocket reconnection attempts",
+})
+
+// Memory metrics
+export const heapUsed = Metric.gauge("nodejs_heap_used_bytes", {
+  description: "Node.js heap used in bytes",
+})
+
+export const rss = Metric.gauge("nodejs_rss_bytes", {
+  description: "Node.js resident set size in bytes",
+})
+
+// Record memory metrics every 30 seconds
+export const recordMemoryMetrics = Effect.gen(function* () {
+  const mem = process.memoryUsage()
+  yield* Metric.set(heapUsed, mem.heapUsed)
+  yield* Metric.set(rss, mem.rss)
+  yield* Effect.log(
+    `Memory: heap=${Math.round(mem.heapUsed / 1024 / 1024)}MB, rss=${Math.round(mem.rss / 1024 / 1024)}MB`
+  )
+}).pipe(Effect.repeat(Schedule.spaced("30 seconds")), Effect.forkScoped)
+
+// Prometheus exporter config
+const MetricsConfig = Config.integer("METRICS_PORT").pipe(
+  Config.withDefault(9464)
+)
+
+// NodeSdk layer with Prometheus exporter
+export const MetricsLive = Layer.unwrapEffect(
+  Effect.gen(function* () {
+    const port = yield* MetricsConfig
+    yield* Effect.log(`Starting Prometheus metrics exporter on port ${port}`)
+
+    return NodeSdk.layer(() => ({
+      resource: {
+        serviceName: "bsky-label-watcher",
+        serviceVersion: "1.0.0",
+      },
+      metricReader: new PrometheusExporter({ port }),
+    }))
+  })
+)
+
+// Layer that starts memory recording
+export const MemoryMetricsLive = Layer.scopedDiscard(recordMemoryMetrics)

src/Metrics.ts

@@ -1,6 +1,10 @@
 import { Socket } from "@effect/platform";
 import type { SocketError } from "@effect/platform/Socket";
-import { Cause, Effect, Schedule, Stream } from "effect";
+import { Data, Effect, Schedule, Stream } from "effect";
+
+class BufferOverflowError extends Data.TaggedError("BufferOverflowError")<{
+  message: string;
+}> {}
 
 /**
  * A stream that will reconnect to the websocket on error.

src/RetryingSocket.ts

@@ -1,7 +1,7 @@
 // Core services and layers
 export { LabelWatcherLive } from "./LabelWatcher"
 export { Env } from "./Environment"
-export { AtpAgent } from "./AtpAgent"
+export { AtpListAccountAgent, LabelerInfo, make as makeAtpAgent } from "./AtpAgent"
 export { ListService } from "./ListService"
 export { Cursor } from "./Cursor"
 

src/index.ts

@@ -1,6 +1,7 @@
 import { ApiLive } from "@/HttpApi"
 import { LabelWatcherLive } from "@/LabelWatcher"
 import { LoggerLive } from "@/logger"
+import { MemoryMetricsLive, MetricsLive } from "@/Metrics"
 import { Layer } from "effect"
 import "dotenv/config"
 import { BunRuntime } from "@effect/platform-bun"
@@ -31,8 +32,10 @@ import { Env } from "@/Environment"
 export const MainLiveLayer = Layer.mergeAll(
   LabelWatcherLive.pipe(Layer.provide(Env.Default)),
   ApiLive,
+  MemoryMetricsLive,
 ).pipe(
   Layer.provide(LoggerLive),
+  Layer.provide(MetricsLive),
 )
 
 Layer.launch(MainLiveLayer).pipe(BunRuntime.runMain)