diff --git a/README.md b/README.md index a5ccd4d..0e0acf0 100644 --- a/README.md +++ b/README.md @@ -235,6 +235,7 @@ tracer.TrackTool(raindrop.TrackToolOptions{ - `WithWriteKey(string)`: Sets the Raindrop write key. When empty and no local Workshop URL resolves, the client becomes a no-op. - `WithEndpoint(string)`: Overrides the base API endpoint. Defaults to `https://api.raindrop.ai/v1/`. +- `WithProjectID(string)`: Scopes telemetry to a Raindrop project by attaching the `X-Raindrop-Project-Id` header to every outbound request. The value is trimmed and validated against `^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$`. When unset, no header is sent and the backend routes events to the org's default project; an invalid value is ignored with a logged warning (no header is sent) so a misconfiguration never breaks ingestion. - `WithLocalWorkshopURL(string)`: Pins the local Workshop daemon URL, suppressing env vars and the auto-detect probe. Pass an empty string to revert to inherit-from-env behavior. - `WithDisableLocalWorkshop()`: Opts out of the local mirror entirely, even if `RAINDROP_LOCAL_DEBUGGER` / `RAINDROP_WORKSHOP` is set or a daemon is listening on the default port. - `WithDebug(bool)`: Enables debug logging. @@ -250,6 +251,27 @@ tracer.TrackTool(raindrop.TrackToolOptions{ - `WithCloseTimeout(time.Duration)`: Sets the hard deadline for `Close()`'s final flush; once it passes, in-flight sends are aborted and remaining payloads are dropped. Defaults to `10s`. Non-positive values are ignored. - `WithLogger(*slog.Logger)`: Uses a custom structured logger. +## Routing To A Project + +By default, telemetry lands in your org's `default` project. Pass +`WithProjectID` to route every event, signal, identify call, and trace to a +named project instead: + +```go +client, err := raindrop.New( + raindrop.WithWriteKey("rk_..."), + raindrop.WithProjectID("checkout-bot"), +) +``` + +When set to a valid slug, the `X-Raindrop-Project-Id` header is attached to +every outbound request (including the local Workshop mirror). When unset, no +header is sent and the backend falls back to the default project, so existing +callers are unaffected. The slug is trimmed and validated against +`^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$`; an invalid value is ignored with a +logged warning and no header is sent, so a misconfigured project ID can never +break telemetry shipping. + ## Local Workshop Mirror When a local Workshop daemon URL resolves, every cloud-bound POST is also mirrored to the local URL so events show up in a local Workshop instance during development. @@ -272,6 +294,7 @@ The local POST uses a 2s timeout, no retries, and errors are logged at `debug` l - Signal payloads go to `/signals/track`. - User identify payloads go to `/users/identify`. - Traces are sent as OTLP JSON to `/traces`. +- When `WithProjectID` is set to a valid slug, every outbound request carries the `X-Raindrop-Project-Id` header; otherwise the header is omitted. - `Begin()`/`Finish()` is the recommended flow for new code. - `ResumeInteraction()` is only for recovering an active interaction handle in the same process. - Empty `writeKey` with no local Workshop URL resolved disables all shipping without raising errors. diff --git a/http.go b/http.go index 126d67d..037b68b 100644 --- a/http.go +++ b/http.go @@ -9,6 +9,7 @@ import ( "log/slog" "math/rand" "net/http" + "regexp" "strconv" "strings" "time" @@ -26,12 +27,41 @@ const ( // maxRetryAfterDelay caps how long a server-provided Retry-After header // can delay the next attempt. maxRetryAfterDelay = 30 * time.Second + + // projectIDHeader routes telemetry to a specific Raindrop project when set. + projectIDHeader = "X-Raindrop-Project-Id" ) +// projectIDSlugPattern bounds a project_id to a DNS-label-style slug. +var projectIDSlugPattern = regexp.MustCompile(`^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$`) + +// normalizeProjectID trims and validates a configured project_id. An empty or +// whitespace-only value yields "" (no header is sent). An invalid value is +// dropped with a warning rather than risking an ingest-time HTTP 400, so a +// misconfigured project_id can never break telemetry shipping. +func normalizeProjectID(raw string, logger *slog.Logger) string { + trimmed := strings.TrimSpace(raw) + if trimmed == "" { + return "" + } + if !projectIDSlugPattern.MatchString(trimmed) { + if logger != nil { + logger.Warn( + "raindrop: ignoring invalid project_id; no X-Raindrop-Project-Id header will be sent", + "project_id", trimmed, + "pattern", projectIDSlugPattern.String(), + ) + } + return "" + } + return trimmed +} + type retryingHTTPClient struct { baseURL string localBaseURL string writeKey string + projectID string client *http.Client localClient *http.Client debug bool @@ -67,6 +97,7 @@ func newRetryingHTTPClient(cfg config, localBaseURL string) *retryingHTTPClient baseURL: cfg.endpoint, localBaseURL: localBaseURL, writeKey: cfg.writeKey, + projectID: cfg.projectID, client: cfg.httpClient, localClient: localClient, debug: cfg.debug, @@ -142,6 +173,7 @@ func (c *retryingHTTPClient) postOnce(ctx context.Context, url string, payload [ } req.Header.Set("Authorization", "Bearer "+c.writeKey) req.Header.Set("Content-Type", "application/json") + c.setProjectIDHeader(req) resp, err := c.client.Do(req) if err != nil { @@ -207,6 +239,7 @@ func (c *retryingHTTPClient) postLocalMirror(ctx context.Context, path string, p req.Header.Set("Authorization", "Bearer "+c.writeKey) } req.Header.Set("Content-Type", "application/json") + c.setProjectIDHeader(req) resp, err := c.localClient.Do(req) if err != nil { c.debugMirror("local mirror POST failed", "error", err, "url", url) @@ -225,6 +258,15 @@ func (c *retryingHTTPClient) debugMirror(msg string, args ...any) { c.logger.Debug(msg, args...) } +// setProjectIDHeader attaches the project routing header when a project_id is +// set. Values are trimmed and validated once at New(), so projectID is empty +// here for blank or invalid input and no header is sent. +func (c *retryingHTTPClient) setProjectIDHeader(req *http.Request) { + if c.projectID != "" { + req.Header.Set(projectIDHeader, c.projectID) + } +} + func (c *retryingHTTPClient) retryDelay(retryNumber int, previous error) time.Duration { if statusErr, ok := previous.(*httpStatusError); ok && statusErr.RetryAfter > 0 { // Clamp server-controlled values: an arbitrary Retry-After (hours, diff --git a/options.go b/options.go index ba0d57a..565fc6e 100644 --- a/options.go +++ b/options.go @@ -12,6 +12,7 @@ type Option func(*config) error type config struct { writeKey string endpoint string + projectID string localWorkshop LocalWorkshopConfig autoDetectLocal bool debug bool @@ -69,6 +70,20 @@ func WithEndpoint(endpoint string) Option { } } +// WithProjectID scopes all telemetry to a Raindrop project. When set to a +// valid slug, every outbound request carries the X-Raindrop-Project-Id +// header so the ingest boundary routes events to the named project; when +// unset, no header is sent and the backend falls back to the org's default +// project (fully backward compatible). The value is trimmed and validated +// against ^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$ at New(); an invalid value +// is ignored with a warning rather than risking an ingest-time rejection. +func WithProjectID(projectID string) Option { + return func(cfg *config) error { + cfg.projectID = projectID + return nil + } +} + // WithLocalWorkshopURL pins the local Workshop daemon URL, suppressing env // vars and the auto-detect probe. Pass an empty string to revert to the // inherit-from-env default behavior. diff --git a/project_id_test.go b/project_id_test.go new file mode 100644 index 0000000..082338d --- /dev/null +++ b/project_id_test.go @@ -0,0 +1,266 @@ +package raindrop + +import ( + "bytes" + "context" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" +) + +type recordedRequest struct { + path string + projectID string + hasProject bool +} + +// newProjectIDRecordingServer records the request path and the +// X-Raindrop-Project-Id header (presence + value) for every inbound request. +func newProjectIDRecordingServer(t *testing.T) (*httptest.Server, func() []recordedRequest) { + t.Helper() + var mu sync.Mutex + var records []recordedRequest + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.ReadAll(r.Body) + values := r.Header.Values(projectIDHeader) + rec := recordedRequest{path: r.URL.Path, hasProject: len(values) > 0} + if rec.hasProject { + rec.projectID = values[0] + } + mu.Lock() + records = append(records, rec) + mu.Unlock() + w.WriteHeader(http.StatusNoContent) + })) + snapshot := func() []recordedRequest { + mu.Lock() + defer mu.Unlock() + out := make([]recordedRequest, len(records)) + copy(out, records) + return out + } + return server, snapshot +} + +// exerciseEveryRequestSite drives each outbound request path the SDK can emit: +// events/track_partial, signals/track, users/identify, and the OTLP /traces +// export. This is the coverage that proves the project-id header is attached +// everywhere, not just on one endpoint. +func exerciseEveryRequestSite(t *testing.T, client *Client) { + t.Helper() + ctx := context.Background() + + if err := client.TrackAI(ctx, AIEvent{ + EventID: "evt_project", + UserID: "user-123", + Event: "chat_message", + Input: "hello", + Output: "hi", + }); err != nil { + t.Fatalf("track ai: %v", err) + } + if err := client.TrackSignal(ctx, Signal{EventID: "evt_project", Name: "thumbs_up", Type: "feedback"}); err != nil { + t.Fatalf("track signal: %v", err) + } + if err := client.Identify(ctx, User{UserID: "user-123", Traits: map[string]any{"plan": "paid"}}); err != nil { + t.Fatalf("identify: %v", err) + } + + span := client.StartSpan(ctx, SpanOptions{Name: "llm_call", EventID: "evt_project"}) + span.SetAttributes(StringAttr("ai.model.id", "gpt-4o")) + span.End() + if err := client.Flush(ctx); err != nil { + t.Fatalf("flush: %v", err) + } +} + +func TestNormalizeProjectID(t *testing.T) { + maxLenSlug := strings.Repeat("a", 63) // 63 chars is the longest valid slug + + tests := []struct { + name string + in string + want string + }{ + {"plain", "default", "default"}, + {"hyphenated", "my-project", "my-project"}, + {"single-char", "a", "a"}, + {"alphanumeric", "abc123", "abc123"}, + {"interior-hyphens", "a-b-c", "a-b-c"}, + {"single-digit", "0", "0"}, + {"letter-digit", "z9", "z9"}, + {"max-length-63", maxLenSlug, maxLenSlug}, + {"trims-surrounding-whitespace", " my-project ", "my-project"}, + {"empty", "", ""}, + {"whitespace-only", " ", ""}, + {"leading-hyphen", "-leading", ""}, + {"trailing-hyphen", "trailing-", ""}, + {"uppercase", "UPPER", ""}, + {"underscore", "with_underscore", ""}, + {"space", "with space", ""}, + {"unicode", "ünicode", ""}, + {"too-long-64", strings.Repeat("a", 64), ""}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := normalizeProjectID(tc.in, nil); got != tc.want { + t.Fatalf("normalizeProjectID(%q) = %q, want %q", tc.in, got, tc.want) + } + }) + } +} + +func TestProjectIDHeaderAttachedToEveryRequestWhenValid(t *testing.T) { + server, snapshot := newProjectIDRecordingServer(t) + defer server.Close() + + client := newTestClient(t, server.URL+"/", WithProjectID("my-project")) + defer func() { _ = client.Close() }() + + exerciseEveryRequestSite(t, client) + + records := snapshot() + wantPaths := map[string]bool{ + "/events/track_partial": false, + "/signals/track": false, + "/users/identify": false, + "/traces": false, + } + for _, rec := range records { + if _, ok := wantPaths[rec.path]; ok { + wantPaths[rec.path] = true + } + if !rec.hasProject || rec.projectID != "my-project" { + t.Fatalf("request to %s missing project header: %#v", rec.path, rec) + } + } + for path, seen := range wantPaths { + if !seen { + t.Fatalf("expected a request to %s, got %#v", path, records) + } + } +} + +func TestProjectIDHeaderOmittedWhenUnset(t *testing.T) { + server, snapshot := newProjectIDRecordingServer(t) + defer server.Close() + + client := newTestClient(t, server.URL+"/") + defer func() { _ = client.Close() }() + + exerciseEveryRequestSite(t, client) + + records := snapshot() + if len(records) == 0 { + t.Fatalf("expected at least one request") + } + for _, rec := range records { + if rec.hasProject { + t.Fatalf("request to %s unexpectedly carried project header %q", rec.path, rec.projectID) + } + } +} + +func TestProjectIDHeaderTrimsSurroundingWhitespace(t *testing.T) { + server, snapshot := newProjectIDRecordingServer(t) + defer server.Close() + + client := newTestClient(t, server.URL+"/", WithProjectID(" trimmed-project ")) + defer func() { _ = client.Close() }() + + if err := client.TrackSignal(context.Background(), Signal{EventID: "evt", Name: "thumbs_up"}); err != nil { + t.Fatalf("track signal: %v", err) + } + + records := snapshot() + if len(records) != 1 { + t.Fatalf("expected 1 request, got %d", len(records)) + } + if !records[0].hasProject || records[0].projectID != "trimmed-project" { + t.Fatalf("expected trimmed project header, got %#v", records[0]) + } +} + +func TestProjectIDHeaderOmittedAndWarnedWhenInvalid(t *testing.T) { + server, snapshot := newProjectIDRecordingServer(t) + defer server.Close() + + var logBuf bytes.Buffer + logger := slog.New(slog.NewTextHandler(&logBuf, &slog.HandlerOptions{Level: slog.LevelWarn})) + + client := newTestClient(t, server.URL+"/", + WithProjectID("Bad Slug"), + WithLogger(logger), + ) + defer func() { _ = client.Close() }() + + if err := client.Identify(context.Background(), User{UserID: "user-123"}); err != nil { + t.Fatalf("identify: %v", err) + } + + records := snapshot() + if len(records) != 1 { + t.Fatalf("expected 1 request, got %d", len(records)) + } + if records[0].hasProject { + t.Fatalf("invalid project_id must omit the header, got %#v", records[0]) + } + + logged := logBuf.String() + if !strings.Contains(logged, "invalid project_id") { + t.Fatalf("expected warning about invalid project_id, got %q", logged) + } + if !strings.Contains(logged, "Bad Slug") { + t.Fatalf("expected warning to include the rejected value, got %q", logged) + } +} + +func TestProjectIDHeaderAttachedToLocalMirror(t *testing.T) { + cloud, cloudSnapshot := newProjectIDRecordingServer(t) + defer cloud.Close() + local, localSnapshot := newProjectIDRecordingServer(t) + defer local.Close() + + t.Setenv(LocalDebuggerEnvVar, "") + t.Setenv(WorkshopEnvVar, "") + + client, err := New( + WithWriteKey("rk_test"), + WithEndpoint(cloud.URL+"/"), + WithLocalWorkshopURL(local.URL+"/"), + WithProjectID("mirror-project"), + WithDebug(false), + WithPartialFlushInterval(0), + WithTraceFlushInterval(0), + WithLogger(slog.New(slog.NewTextHandler(io.Discard, nil))), + ) + if err != nil { + t.Fatalf("new client: %v", err) + } + defer func() { _ = client.Close() }() + + if err := client.Identify(context.Background(), User{UserID: "user-123"}); err != nil { + t.Fatalf("identify: %v", err) + } + + for _, tc := range []struct { + name string + snapshot func() []recordedRequest + }{ + {"cloud", cloudSnapshot}, + {"local mirror", localSnapshot}, + } { + records := tc.snapshot() + if len(records) != 1 { + t.Fatalf("%s: expected 1 request, got %d", tc.name, len(records)) + } + if !records[0].hasProject || records[0].projectID != "mirror-project" { + t.Fatalf("%s: expected project header, got %#v", tc.name, records[0]) + } + } +} diff --git a/raindrop.go b/raindrop.go index a6c49bf..9d72505 100644 --- a/raindrop.go +++ b/raindrop.go @@ -56,6 +56,7 @@ func New(opts ...Option) (*Client, error) { cfg.writeKey = strings.TrimSpace(cfg.writeKey) cfg.endpoint = formatEndpoint(cfg.endpoint) + cfg.projectID = normalizeProjectID(cfg.projectID, cfg.logger) resolvedLocal := ResolveLocalWorkshopURL(cfg.localWorkshop, cfg.autoDetectLocal) client := &Client{ diff --git a/version.go b/version.go index 986a0a2..55c2889 100644 --- a/version.go +++ b/version.go @@ -1,3 +1,3 @@ package raindrop -const Version = "0.1.4" +const Version = "0.1.5"