From a8a53c5ad7395e10ec7e189791dc65e2ae49c2fe Mon Sep 17 00:00:00 2001
From: jbiskur <julius@flowcore.io>
Date: Wed, 8 Apr 2026 11:12:34 +0100
Subject: [PATCH] feat: configurable log levels for pulse success and failure

PulseEmitter now accepts successLogLevel and failureLogLevel options
and logs structured events on both success and failure paths:
- success: logs { pathwayId, flowType, timeBucket, isLive, bufferDepth }
  (defaults to 'debug' level)
- failure: logs { error, url, pathwayId, flowType }
  (defaults to 'warn' level)

Plumbed through FlowcoreDataPumpOptions.pulse for downstream consumers.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---
 src/data-pump/data-pump.ts |  6 ++++++
 src/data-pump/pulse.ts     | 29 ++++++++++++++++++++++++++++-
 2 files changed, 34 insertions(+), 1 deletion(-)

diff --git a/src/data-pump/data-pump.ts b/src/data-pump/data-pump.ts
index f56a6c0..dbc1bb7 100644
--- a/src/data-pump/data-pump.ts
+++ b/src/data-pump/data-pump.ts
@@ -56,6 +56,10 @@ export interface FlowcoreDataPumpOptions {
     url: string
     intervalMs?: number
     pathwayId: string
+    /** Log level for successful pulses. Defaults to 'debug'. */
+    successLogLevel?: "debug" | "info" | "warn" | "error"
+    /** Log level for pulse failures. Defaults to 'warn'. */
+    failureLogLevel?: "debug" | "info" | "warn" | "error"
   }
 }
 
@@ -179,6 +183,8 @@ export class FlowcoreDataPump {
           intervalMs: options.pulse.intervalMs,
           auth: options.auth,
           logger: options.logger,
+          successLogLevel: options.pulse.successLogLevel,
+          failureLogLevel: options.pulse.failureLogLevel,
         },
         () => {
           const snapshot = pump.getSnapshot()
diff --git a/src/data-pump/pulse.ts b/src/data-pump/pulse.ts
index e637815..a252d1a 100644
--- a/src/data-pump/pulse.ts
+++ b/src/data-pump/pulse.ts
@@ -2,11 +2,20 @@ import { SendPumpPulseCommand } from "@flowcore/sdk"
 import type { FlowcoreDataPumpAuth, FlowcoreLogger } from "./types.ts"
 import { getFlowcoreClient } from "./flowcore-client.ts"
 
+/**
+ * Log level for pulse emitter events. Corresponds to FlowcoreLogger methods.
+ */
+export type PulseLogLevel = "debug" | "info" | "warn" | "error"
+
 export interface PulseEmitterOptions {
   url: string
   intervalMs?: number
   auth: FlowcoreDataPumpAuth
   logger?: FlowcoreLogger
+  /** Log level for successful pulses. Defaults to 'debug'. */
+  successLogLevel?: PulseLogLevel
+  /** Log level for pulse failures. Defaults to 'warn'. */
+  failureLogLevel?: PulseLogLevel
 }
 
 export interface PulseSnapshot {
@@ -29,6 +38,8 @@ export class PulseEmitter {
   private startTimeout: ReturnType<typeof setTimeout> | null = null
   private readonly intervalMs: number
   private readonly logger?: FlowcoreLogger
+  private readonly successLogLevel: PulseLogLevel
+  private readonly failureLogLevel: PulseLogLevel
 
   constructor(
     private readonly options: PulseEmitterOptions,
@@ -36,6 +47,8 @@ export class PulseEmitter {
   ) {
     this.intervalMs = options.intervalMs ?? 30_000
     this.logger = options.logger
+    this.successLogLevel = options.successLogLevel ?? "debug"
+    this.failureLogLevel = options.failureLogLevel ?? "warn"
   }
 
   start(): void {
@@ -63,8 +76,14 @@ export class PulseEmitter {
 
   private emitSafe(): void {
     this.emit().catch((err) => {
+      const snapshot = this.getSnapshot()
       const msg = err instanceof Error ? err.message : String(err)
-      this.logger?.warn?.(`Pulse emission failed: ${msg}`)
+      this.logger?.[this.failureLogLevel]?.("Pulse emission failed", {
+        error: msg,
+        url: this.options.url,
+        pathwayId: snapshot?.pathwayId,
+        flowType: snapshot?.flowType,
+      })
     })
   }
 
@@ -94,5 +113,13 @@ export class PulseEmitter {
         uptimeMs: snapshot.uptimeMs,
       }),
     )
+
+    this.logger?.[this.successLogLevel]?.("Pulse sent", {
+      pathwayId: snapshot.pathwayId,
+      flowType: snapshot.flowType,
+      timeBucket: snapshot.timeBucket,
+      isLive: snapshot.isLive,
+      bufferDepth: snapshot.bufferDepth,
+    })
   }
 }