Fluent builder API (#51)

* WIP: New fluent api

* Test fixes

Author: Dig-Doug

Cargo.toml

@@ -8,7 +8,7 @@ members = [
 ]
 
 [workspace.package]
-version = "0.0.13"
+version = "0.0.14"
 edition = "2021"
 license = "AGPL-3.0-only"
 repository = "https://github.com/Dig-Doug/observation-tools-client"
@@ -33,10 +33,10 @@ nom = "7"
 minijinja-autoreload = "2.12.0"
 minijinja-embed = "2.12.0"
 object_store = "0.12.4"
-observation-tools = { path = "crates/observation-tools-client", version = "0.0.13" }
-observation-tools-macros = { path = "crates/observation-tools-macros", version = "0.0.13" }
-observation-tools-server = { path = "crates/observation-tools-server", version = "0.0.13" }
-observation-tools-shared = { path = "crates/observation-tools-shared", version = "0.0.13" }
+observation-tools = { path = "crates/observation-tools-client", version = "0.0.14" }
+observation-tools-macros = { path = "crates/observation-tools-macros", version = "0.0.14" }
+observation-tools-server = { path = "crates/observation-tools-server", version = "0.0.14" }
+observation-tools-shared = { path = "crates/observation-tools-shared", version = "0.0.14" }
 openapiv3 = "2.2"
 prettyplease = "0.2"
 progenitor = "0.11.2"

crates/observation-tools-client/index.d.ts

@@ -33,10 +33,13 @@ export declare class ExecutionHandle {
 }
 
 /**
- * Builder for creating observations (without payload set yet)
+ * Builder for creating observations
  *
- * Call `.payload()` or `.custom_payload()` to get an
- * `ObservationBuilderWithPayload` that can be built.
+ * Use the `observe!` macro or `ObservationBuilder::new()` to create a builder,
+ * then chain methods to configure and send the observation.
+ *
+ * Payload methods (`.serde()`, `.debug()`, `.payload()`) send the observation
+ * immediately and return `SendObservation` for optional waiting.
  */
 export declare class ObservationBuilder {
   /** Create a new observation builder with the given name */
@@ -54,30 +57,17 @@ export declare class ObservationBuilder {
   metadata(key: string, value: string): this
   /** Set the source info for the observation */
   source(file: string, line: number): this
-  /** Set the payload as JSON data */
-  jsonPayload(jsonString: string): ObservationBuilderWithPayload
-  /** Set the payload with custom data and MIME type */
-  rawPayload(data: string, mimeType: string): ObservationBuilderWithPayload
-  /** Set the payload as markdown content */
-  markdownPayload(content: string): ObservationBuilderWithPayload
+  /** Set the payload as JSON data, returning a builder that can be sent with an execution */
+  jsonPayload(jsonString: string): ObservationBuilderWithPayloadNapi
+  /** Set the payload with custom data and MIME type, returning a builder that can be sent */
+  rawPayload(data: string, mimeType: string): ObservationBuilderWithPayloadNapi
+  /** Set the payload as markdown content, returning a builder that can be sent */
+  markdownPayload(content: string): ObservationBuilderWithPayloadNapi
 }
 
-/**
- * Builder for creating observations (with payload set)
- *
- * This struct is returned by `ObservationBuilder::payload()` and
- * `ObservationBuilder::custom_payload()`. It has the `build()` methods
- * since a payload is required.
- */
-export declare class ObservationBuilderWithPayload {
-  /**
-   * Build and send the observation
-   *
-   * Returns a SendObservation which allows you to wait for the upload to
-   * complete or get the ObservationHandle immediately.
-   *
-   * If sending fails, returns a stub that will fail on `wait_for_upload()`.
-   */
+/** Intermediate NAPI type that holds a builder and payload, allowing `.send(exe)` pattern */
+export declare class ObservationBuilderWithPayloadNapi {
+  /** Send the observation using the provided execution handle */
   send(execution: ExecutionHandle): SendObservation
 }
 

crates/observation-tools-client/index.js

@@ -580,7 +580,7 @@ module.exports.Client = nativeBinding.Client
 module.exports.ClientBuilder = nativeBinding.ClientBuilder
 module.exports.ExecutionHandle = nativeBinding.ExecutionHandle
 module.exports.ObservationBuilder = nativeBinding.ObservationBuilder
-module.exports.ObservationBuilderWithPayload = nativeBinding.ObservationBuilderWithPayload
+module.exports.ObservationBuilderWithPayloadNapi = nativeBinding.ObservationBuilderWithPayloadNapi
 module.exports.ObservationHandle = nativeBinding.ObservationHandle
 module.exports.SendObservation = nativeBinding.SendObservation
 module.exports.generateExecutionId = nativeBinding.generateExecutionId

crates/observation-tools-client/src/axum/execution_layer.rs

@@ -10,7 +10,8 @@ use std::task::Poll;
 use tower::Layer;
 use tower::Service;
 
-/// A filter function that determines whether an execution should be created for a request.
+/// A filter function that determines whether an execution should be created for
+/// a request.
 ///
 /// Returns `true` to create an execution, `false` to skip.
 pub type RequestFilter = Arc<dyn Fn(&Request) -> bool + Send + Sync>;
@@ -31,10 +32,11 @@ impl ExecutionLayer {
     }
   }
 
-  /// Set a filter function that determines whether an execution should be created.
+  /// Set a filter function that determines whether an execution should be
+  /// created.
   ///
-  /// The filter receives a reference to the incoming request and returns `true` to
-  /// create an execution context, or `false` to skip execution creation.
+  /// The filter receives a reference to the incoming request and returns `true`
+  /// to create an execution context, or `false` to skip execution creation.
   /// When skipped, the request proceeds without an execution context.
   ///
   /// # Example

crates/observation-tools-client/src/axum/request_observer.rs

@@ -4,7 +4,8 @@ use crate::observation::ObservationBuilder;
 use axum::body::Body;
 use axum::extract::Request;
 use axum::response::Response;
-use bytes::{Bytes, BytesMut};
+use bytes::Bytes;
+use bytes::BytesMut;
 use http::header::HeaderMap;
 use http::header::HeaderName;
 use http::header::AUTHORIZATION;
@@ -27,8 +28,9 @@ use tower::Layer;
 use tower::Service;
 
 /// State shared between the streaming body and the observation emitter.
-/// This is used to collect data as it streams and emit the observation when complete.
-/// The observation is emitted in the Drop implementation when the body is finished.
+/// This is used to collect data as it streams and emit the observation when
+/// complete. The observation is emitted in the Drop implementation when the
+/// body is finished.
 struct StreamingObserverState {
   buffer: BytesMut,
   content_type: String,
@@ -71,14 +73,12 @@ impl Drop for StreamingObserverState {
       size: bytes.len(),
     };
 
-    let mut response_body_builder = ObservationBuilder::new("http/response/body");
-    response_body_builder
+    ObservationBuilder::new("http/response/body")
       .label("http/response")
       .label("http/response/body")
       .metadata("status", &self.status_code.to_string())
       .log_level(self.log_level)
-      .payload(payload)
-      .build_with_execution(&self.execution);
+      .payload_with_execution(payload, &self.execution);
   }
 }
 
@@ -246,31 +246,27 @@ where
       };
 
       let (parts, body) = req.into_parts();
-      let mut request_headers_builder = ObservationBuilder::new("http/request/headers");
-      request_headers_builder
+      ObservationBuilder::new("http/request/headers")
         .label("http/request")
         .label("http/request/headers")
         .metadata("method", parts.method.to_string())
         .metadata("uri", parts.uri.to_string())
         .serde(&json!(filter_headers(
           &parts.headers,
           &config.excluded_headers
-        )))
-        .build();
+        )));
 
       let request_body_bytes = body
         .collect()
         .await
         .map(|collected| collected.to_bytes())
         .unwrap_or_else(|_| Bytes::new());
-      let mut request_body_builder = ObservationBuilder::new("http/request/body");
-      request_body_builder
+      ObservationBuilder::new("http/request/body")
         .label("http/request")
         .label("http/request/body")
         .metadata("method", parts.method.to_string())
         .metadata("uri", parts.uri.to_string())
-        .payload(bytes_to_payload(&request_body_bytes, &parts.headers))
-        .build();
+        .payload(bytes_to_payload(&request_body_bytes, &parts.headers));
 
       let response = inner
         .call(Request::from_parts(parts, Body::from(request_body_bytes)))
@@ -283,28 +279,31 @@ where
         500..=599 => LogLevel::Error,
         _ => LogLevel::Info,
       };
-      let mut response_headers_builder = ObservationBuilder::new("http/response/headers");
-      response_headers_builder
+      ObservationBuilder::new("http/response/headers")
         .label("http/response")
         .label("http/response/headers")
         .metadata("status", &parts.status.as_u16().to_string())
         .log_level(log_level)
         .serde(&json!(filter_headers(
           &parts.headers,
           &config.excluded_headers
-        )))
-        .build();
+        )));
 
-      // Wrap the response body in a streaming observer that captures data as it flows through
-      // and emits the observation when the stream completes
+      // Wrap the response body in a streaming observer that captures data as it flows
+      // through and emits the observation when the stream completes
       let content_type = parts
         .headers
         .get(CONTENT_TYPE)
         .and_then(|v| v.to_str().ok())
         .unwrap_or("application/octet-stream")
         .to_string();
-      let streaming_body =
-        StreamingObserverBody::new(body, content_type, log_level, parts.status.as_u16(), execution);
+      let streaming_body = StreamingObserverBody::new(
+        body,
+        content_type,
+        log_level,
+        parts.status.as_u16(),
+        execution,
+      );
 
       Ok(Response::from_parts(parts, Body::new(streaming_body)))
     })

crates/observation-tools-client/src/context.rs

@@ -71,11 +71,11 @@ pub(crate) fn clear_global_execution() {
 /// // Run two tasks concurrently with different execution contexts
 /// let (result1, result2) = tokio::join!(
 ///   with_execution(execution1, async {
-///     observe!("observation-1", "data from task 1")?;
+///     observe!("observation-1").serde(&"data from task 1");
 ///     Ok::<_, Error>(())
 ///   }),
 ///   with_execution(execution2, async {
-///     observe!("observation-2", "data from task 2")?;
+///     observe!("observation-2").serde(&"data from task 2");
 ///     Ok::<_, Error>(())
 ///   })
 /// );
@@ -110,7 +110,7 @@ pub(crate) fn get_current_tracing_span_id() -> Option<String> {
 ///
 /// // The spawned task will inherit the current execution context
 /// tokio::spawn(async move {
-///     observe!("spawned-task", "data from spawned task")?;
+///     observe!("spawned-task").serde(&"data from spawned task");
 ///     Ok::<_, Error>(())
 /// }.with_observations());
 /// ```
@@ -126,9 +126,7 @@ pub trait WithObservations: Future + Sized {
 impl<F: Future + Send + 'static> WithObservations for F {
   fn with_observations(self) -> WithObservationsFuture<Self::Output> {
     match get_current_execution() {
-      Some(execution) => {
-        WithObservationsFuture(Box::pin(TASK_EXECUTION.scope(execution, self)))
-      }
+      Some(execution) => WithObservationsFuture(Box::pin(TASK_EXECUTION.scope(execution, self))),
       None => WithObservationsFuture(Box::pin(self)),
     }
   }

crates/observation-tools-client/src/lib.rs

@@ -28,7 +28,6 @@ pub use execution::BeginExecution;
 pub use execution::ExecutionHandle;
 pub use logger::ObservationLogger;
 pub use observation::ObservationBuilder;
-pub use observation::ObservationBuilderWithPayload;
 pub use observation_handle::ObservationHandle;
 pub use observation_handle::SendObservation;
 // Re-export procedural macro

crates/observation-tools-client/src/logger.rs

@@ -37,15 +37,16 @@ impl Log for ObservationLogger {
       return;
     }
 
-    let mut builder = ObservationBuilder::new("ObservationLogger");
-    builder
+    let builder = ObservationBuilder::new("ObservationLogger")
       .observation_type(ObservationType::LogEntry)
       .log_level(record.level().into())
       .label(format!("log/{}", record.target()));
-    if let (Some(file), Some(line)) = (record.file(), record.line()) {
-      builder.source(file, line);
-    }
-    let _ = builder.payload(&format!("{}", record.args())).build();
+    let builder = if let (Some(file), Some(line)) = (record.file(), record.line()) {
+      builder.source(file, line)
+    } else {
+      builder
+    };
+    let _ = builder.payload(format!("{}", record.args()));
   }
 
   fn flush(&self) {