openmeter/streaming/clickhouse/event_query_v2.go (1)

100-136: Count query unchanged (OK), but remember to include customer filters if/when added to v2

No change needed now. If v2 later gains customer filters, consider echoing a coarse customer constraint into the count to improve ETA accuracy.

openmeter/meterevent/service.go (2)

37-38: Nit: comment should be plural to match the field (CustomerIDs)

Doc reads “The event customer ID.” but the field holds multiple IDs.

Apply this diff:

-	// The event customer ID.
-	CustomerIDs *[]string
+	// The customer IDs to filter by.
+	CustomerIDs *[]string

---

79-122: Consider basic validation for CustomerIDs to avoid empty IN lists and bad inputs

Currently CustomerIDs is unchecked. Two practical guards:

• Reject an explicitly provided empty list to avoid generating “IN ()”.
• Reject empty-string IDs early; optionally validate ULID format if that’s the canonical ID.

Keep nil meaning “no filter”.

Here’s a minimal, dependency-free validation you can add:

 func (i ListEventsParams) Validate() error {
   var errs []error
@@
   if i.ClientID != nil && *i.ClientID == "" {
     errs = append(errs, errors.New("client id cannot be empty"))
   }
+
+  if i.CustomerIDs != nil {
+    if len(*i.CustomerIDs) == 0 {
+      errs = append(errs, errors.New("customerIds must not be empty when provided"))
+    } else {
+      for idx, id := range *i.CustomerIDs {
+        if id == "" {
+          errs = append(errs, fmt.Errorf("customerIds[%d] cannot be empty", idx))
+        }
+      }
+    }
+  }

If you already have a ULID validation helper, you can tighten this further; happy to wire it in.

api/spec/src/events.tsp (1)

308-312: Adding customerId to IngestedEvent is aligned; consider including it in the example, too

Helps SDK/gen consumers see it in docs quickly.

You can extend the existing IngestedEvent example near Line 289 like so (outside this hunk):

@example(#{
  event: #{
    id: "5c10fade-1c9e-4d6c-8275-c52c36731d3c",
    source: "service-name",
    specversion: "1.0",
    type: "prompt",
    subject: "customer-id",
    time: DateTime.fromISO("2023-01-01T01:01:01.001Z"),
  },
  customerId: "01HZXABCD3M6E7K8J9PQRS1TUV",
  ingestedAt: DateTime.fromISO("2023-01-01T01:01:01.001Z"),
  storedAt: DateTime.fromISO("2023-01-01T01:01:02.001Z"),
})

openmeter/meterevent/adapter/event.go (4)

74-80: N+1 customer lookups during enrichment; consider batching unique subjects.

The per-event GetCustomerByUsageAttribution call can become costly. You already left a FIXME; recommend collecting unique subject keys first and introducing a bulk lookup in customerService to cut DB round-trips.

Happy to sketch an interface and a batching implementation if you want to pursue this in a follow-up.

---

127-152: Nit: avoid unnecessary allocation for ValidationErrors in mapped events.

You can omit initializing ValidationErrors to an empty slice; the zero value (nil) is sufficient and avoids per-event allocations until you actually add an error.

-           ValidationErrors: make([]error, 0),
+           // ValidationErrors left nil until validation populates it

---

155-195: Reduce validation complexity by indexing meters by event type.

Current validation is O(#events × #meters). Pre-index meters by EventType and validate only relevant ones.

 func (a *adapter) validateEvents(ctx context.Context, namespace string, events []meterevent.Event) ([]meterevent.Event, error) {
   // Get all meters
   meterList, err := a.meterService.ListMeters(ctx, meter.ListMetersParams{
     Namespace: namespace,
   })
   if err != nil {
     return nil, fmt.Errorf("get meters: %w", err)
   }

-  // Validate events against meters
+  // Index meters by event type
+  metersByType := make(map[string][]meter.Meter, len(meterList.Items))
+  for _, m := range meterList.Items {
+    metersByType[m.EventType] = append(metersByType[m.EventType], m)
+  }
+
+  // Validate events against meters
   validatedEvents := make([]meterevent.Event, 0, len(events))

   for _, event := range events {
-    meterMatch := false
-    validationErrors := []error{}
-
-    // Check if the event type matches a meter
-    for _, m := range meterList.Items {
-      // Check if the event type matches the meter event type
-      if event.Type == m.EventType {
-        meterMatch = true
-
-        // Validate the event data against the meter
-        _, err = meter.ParseEventString(m, event.Data)
-        if err != nil {
-          validationErrors = append(validationErrors, err)
-        }
-      }
-    }
+    meterMatch := false
+    var validationErrors []error
+    if ms, ok := metersByType[event.Type]; ok {
+      meterMatch = true
+      for _, m := range ms {
+        if _, err = meter.ParseEventString(m, event.Data); err != nil {
+          validationErrors = append(validationErrors, err)
+        }
+      }
+    }

     // If no meter matches the event type, add an error to the event
     if !meterMatch {
       validationErrors = append(validationErrors, fmt.Errorf("no meter found for event type: %s", event.Type))
     }

     event.ValidationErrors = validationErrors
     validatedEvents = append(validatedEvents, event)
   }

   return validatedEvents, nil
 }

---

197-240: Avoid shadowing the imported customer package with a local variable.

Using the identifier “customer” for the local result from GetCustomerByUsageAttribution shadows the imported package name, hurting readability.

-   customer, err := a.customerService.GetCustomerByUsageAttribution(ctx, customer.GetCustomerByUsageAttributionInput{
+   cust, err := a.customerService.GetCustomerByUsageAttribution(ctx, customer.GetCustomerByUsageAttributionInput{
       Namespace:  namespace,
       SubjectKey: event.Subject,
     })
     if err != nil {
       if models.IsGenericNotFoundError(err) {
         eventsWithCustomerID = append(eventsWithCustomerID, event)
         continue
       }
       return nil, fmt.Errorf("get customer by usage attribution: %w", err)
     }

     // Add the customer ID to the cache
-    cache[event.Subject] = customer.ID
+    cache[event.Subject] = cust.ID

     // Add the event to the list
-    event.CustomerID = &customer.ID
+    event.CustomerID = &cust.ID

openmeter/streaming/eventparams.go (1)

36-61: Add limit validation to mirror v2 and fail fast on bad inputs.

ListEventsParams.Validate should reject non-positive limits (the handler typically sets a sane default, but catching programmer error here helps).

 func (i ListEventsParams) Validate() error {
   var errs []error
   ...
   if i.IngestedAtFrom != nil && i.IngestedAtTo != nil && i.IngestedAtTo.Before(*i.IngestedAtFrom) {
     errs = append(errs, fmt.Errorf("ingestedAtTo date is before ingestedAtFrom date: %s < %s", i.IngestedAtTo.Format(time.RFC3339), i.IngestedAtFrom.Format(time.RFC3339)))
   }
 
+  if i.Limit < 1 {
+    errs = append(errs, errors.New("limit must be greater than 0"))
+  }
+
   return errors.Join(errs...)
 }

openmeter/meterevent/adapter/adapter.go (1)

24-28: New field added to adapter; consider nil-safety expectations.

If customerService is required for correctness, consider validating non-nil in New and documenting the requirement. Otherwise, ensure downstream code defensively handles a nil customerService.

Here’s a minimal guard if you want it:

 func New(
   streamingConnector streaming.Connector,
   customerService customer.Service,
   meterService meter.Service,
 ) meterevent.Service {
+  if customerService == nil {
+    // Option A: panic fast if this must be provided
+    // panic("customerService must not be nil")
+    // Option B: leave as-is but log for visibility (requires a logger in scope)
+  }
   return &adapter{
     streamingConnector: streamingConnector,
     customerService:    customerService,
     meterService:       meterService,
   }
 }

api/client/javascript/src/zod/index.ts (1)

8566-8568: Simplify ULID regex & extract shared constant

The current pattern is correct (26 chars, Crockford base32 without I/L/O/U, first char 0–7). For readability you can switch to a RegExp literal with /i and, optionally, extract it into a shared constant:

-export const listEventsQueryCustomerIdItemRegExp = new RegExp(
-  '^[0-7][0-9A-HJKMNP-TV-Za-hjkmnp-tv-z]{25}$'
-)
+export const listEventsQueryCustomerIdItemRegExp =
+  /^[0-7][0-9A-HJKMNP-TV-Z]{25}$/i

Example shared constant (e.g. zod/ulid.ts):

export const ULID_REGEX = /^[0-7][0-9A-HJKMNP-TV-Z]{25}$/i

then

import {ULID_REGEX} from './ulid'
… .regex(ULID_REGEX)

Other occurrences of the same pattern you may want to align:

• tools/migrate/entitlement_test.go
• api/spec/src/types.tsp
• api/openapi.yaml (multiple schema entries)

openmeter/streaming/testutils/streaming.go (1)

82-88: Mock signature parity with streaming.Connector looks good

The mock implements the updated interface correctly. It’s fine to return empty slices here.

If you want stronger tests around customer filtering later, consider capturing incoming params (including Customers/Subject) for assertions or providing a way to pre-seed RawEvents with CustomerID for integration-style tests.

api/client/javascript/src/client/schemas.ts (1)

17612-17613: Update description to reflect array semantics and repeated query-parameter encoding

Verified in openapi.yaml that no explicit style/explode is set, so arrays use the default form style with explode=true (i.e. ?customerId=a&customerId=b).

• File: api/client/javascript/src/client/schemas.ts (around lines 17612–17613)
• Apply this doc tweak (ideally in the OpenAPI source so the generator emits it):

-        /** @description The event customer ID. */
+        /** @description Filter by customer IDs (ULID). Accepts multiple values (repeated query parameters). */

openmeter/server/server_test.go (3)

604-618: Mock signature update LGTM; consider asserting interface compliance

The shift to ListEvents(ctx, namespace, params streaming.ListEventsParams) matches the new interface. To keep the mock future-proof, add a compile-time assertion to ensure it stays in sync with streaming.Connector.

Add outside the method (anywhere after type declaration):

var _ streaming.Connector = (*MockStreamingConnector)(nil)

---

604-618: Optional: Populate customer_id in mocked events to increase coverage

Given the PR adds customerId to event list APIs, having the mock set CustomerID can help exercise response serialization paths earlier.

 func (c *MockStreamingConnector) ListEvents(ctx context.Context, namespace string, params streaming.ListEventsParams) ([]streaming.RawEvent, error) {
   events := []streaming.RawEvent{
     {
       ID:         mockEvent.ID(),
       Type:       mockEvent.Type(),
       Source:     mockEvent.Source(),
       Subject:    mockEvent.Subject(),
       Time:       mockEvent.Time(),
       Data:       string(mockEvent.Data()),
       IngestedAt: time.Time{},
       StoredAt:   time.Time{},
+      CustomerID: "customer-1",
     },
   }
   return events, nil
 }

---

620-634: V2 mock signature LGTM; mirror the V1 mock’s event shape

Signature aligns with streaming.ListEventsV2Params. Consider mirroring the V1 suggestion to set CustomerID for parity.

 func (c *MockStreamingConnector) ListEventsV2(ctx context.Context, params streaming.ListEventsV2Params) ([]streaming.RawEvent, error) {
   events := []streaming.RawEvent{
     {
       ID:         mockEvent.ID(),
       Type:       mockEvent.Type(),
       Source:     mockEvent.Source(),
       Subject:    mockEvent.Subject(),
       Time:       mockEvent.Time(),
       Data:       string(mockEvent.Data()),
       IngestedAt: time.Time{},
       StoredAt:   time.Time{},
+      CustomerID: "customer-1",
     },
   }
   return events, nil
 }

openmeter/streaming/clickhouse/queryhelper.go (1)

35-41: Consider NULL instead of empty string for unknown customer_id

Returning '' can conflate unknown vs. intentionally empty values. Using NULL may be semantically clearer for downstream consumers.

- caseBuilder.WriteString("ELSE '' END AS customer_id")
+ caseBuilder.WriteString("ELSE NULL END AS customer_id")

If you change this, confirm downstream JSON marshaling keeps the intended shape (omitempty).

api/openapi.cloud.yaml (1)

16422-16426: Prefer referencing CustomerId schema; consider marking readOnly if server-enriched

Avoid duplicating the ULID pattern by reusing the shared schema. If this field is only present in responses (enriched/derived), add readOnly: true.

Apply this diff:

-        customerId:
-          type: string
-          pattern: ^[0-7][0-9A-HJKMNP-TV-Za-hjkmnp-tv-z]{25}$
-          description: The customer ID if the event is associated with a customer.
-          example: 01G65Z755AFWAKHE12NY0CQ9FH
+        customerId:
+          allOf:
+            - $ref: '#/components/schemas/CustomerId'
+          description: The customer ID if the event is associated with a customer.
+          example: 01G65Z755AFWAKHE12NY0CQ9FH
+          # Uncomment if this is set by the system and returned in responses only:
+          # readOnly: true

Note: If other parts of the API allow ULIDOrExternalKey for customer identification, confirm whether this property should accept that union as well for consistency.

Would you like me to scan the spec and flag any places where customer identifiers allow ULIDOrExternalKey so we can align this property?

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled
• Linear integration is disabled

You can enable these sources in your CodeRabbit configuration.

Learnt from: GAlexIHU
PR: openmeterio/openmeter#2383
File: openmeter/entitlement/metered/lateevents_test.go:37-45
Timestamp: 2025-03-07T12:17:43.129Z
Learning: In the OpenMeter codebase, test files like `openmeter/entitlement/metered/lateevents_test.go` may use variables like `meterSlug` and `namespace` without explicit declarations visible in the same file. This appears to be an accepted pattern in their test structure.

openmeter/server/server_test.go (2)

605-619: Updated ListEvents signature: capture params in mock to assert customerId filter wiring.

Now that v1 supports a customerId filter, it’s valuable to assert that the query param is passed through to streaming.ListEventsParams in tests. Capture the last params on the mock to enable targeted assertions.

Apply this diff to record the incoming params:

-type MockStreamingConnector struct{}
+type MockStreamingConnector struct{
+    LastListEventsParams  streaming.ListEventsParams
+    LastListEventsV2Params streaming.ListEventsV2Params
+}

 func (c *MockStreamingConnector) ListEvents(ctx context.Context, namespace string, params streaming.ListEventsParams) ([]streaming.RawEvent, error) {
+    c.LastListEventsParams = params
     events := []streaming.RawEvent{
         {
             ID:         mockEvent.ID(),
             Type:       mockEvent.Type(),
             Source:     mockEvent.Source(),
             Subject:    mockEvent.Subject(),
             Time:       mockEvent.Time(),
             Data:       string(mockEvent.Data()),
             IngestedAt: time.Time{},
             StoredAt:   time.Time{},
         },
     }
     return events, nil
 }

Follow-up: add a test (e.g., GET /api/v1/events?customerId=foo) and assert c.LastListEventsParams.CustomerID (or equivalent field) is set to "foo". I can draft that if you want.

---

621-635: ListEventsV2 signature updated: consider adding a v2 route test to assert CustomerId in responses.

Per PR scope, v2 exposes CustomerId on responses (even if filter is only on v1). Add a simple /api/v2/events test to assert the presence/shape of CustomerId in the returned payload.

I can add a minimal test case mirroring the v1 “query events” case for v2.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled
• Linear integration is disabled

You can enable these sources in your CodeRabbit configuration.

Learnt from: GAlexIHU
PR: openmeterio/openmeter#2383
File: openmeter/entitlement/metered/lateevents_test.go:37-45
Timestamp: 2025-03-07T12:17:43.129Z
Learning: In the OpenMeter codebase, test files like `openmeter/entitlement/metered/lateevents_test.go` may use variables like `meterSlug` and `namespace` without explicit declarations visible in the same file. This appears to be an accepted pattern in their test structure.

openmeter/meterevent/adapter/event.go (1)

24-33: Pass through IngestedAt filters to streaming (regression).

You’re reconstructing streaming.ListEventsParams but omitting IngestedAtFrom/IngestedAtTo. This drops API-level ingestedAt filters for v1.

Apply this diff:

 listParams := streaming.ListEventsParams{
   Namespace: params.Namespace,
   ClientID:  params.ClientID,
+  IngestedAtFrom: params.IngestedAtFrom,
+  IngestedAtTo:   params.IngestedAtTo,
   From:      params.From,
   To:        params.To,
   ID:        params.ID,
   Subject:   params.Subject,
   Limit:     params.Limit,
 }

openmeter/meterevent/adapter/event.go (3)

35-57: Confirm semantics: empty CustomerIDs resolution returns no events.

If CustomerIDs is provided but resolve to 0 customers, you early-return an empty event list. Confirm this is the intended UX (vs. 4xx on invalid IDs, or querying events with an always-false filter).

---

195-237: Avoid shadowing the imported customer package name.

The local variable named customer shadows the imported package customer, which is easy to trip over in future edits. Rename the local to improve clarity.

Apply this diff:

-		customer, err := a.customerService.GetCustomerByUsageAttribution(ctx, customer.GetCustomerByUsageAttributionInput{
+		cust, err := a.customerService.GetCustomerByUsageAttribution(ctx, customer.GetCustomerByUsageAttributionInput{
 			Namespace:  namespace,
 			SubjectKey: event.Subject,
 		})
 		if err != nil {
 			if models.IsGenericNotFoundError(err) {
 				eventsWithCustomerID = append(eventsWithCustomerID, event)
 				continue
 			}
 
 			return nil, fmt.Errorf("get customer by usage attribution: %w", err)
 		}
 
 		// Add the customer ID to the cache
-		cache[event.Subject] = customer.ID
+		cache[event.Subject] = cust.ID
 
 		// Add the event to the list
-		event.CustomerID = &customer.ID
+		event.CustomerID = &cust.ID
 		eventsWithCustomerID = append(eventsWithCustomerID, event)

---

213-219: Optional: batch attribution lookups.

You already cache by subject to avoid duplicate lookups, which is good. If this hot path becomes heavy, consider adding a customer service method to batch-resolve usage attribution for multiple subject keys in one call to reduce DB round trips.

openmeter/server/server_test.go (1)

138-186: Consider adding tests to pin new behaviors (customer filter + ingestedAt pass-through).

• Add a server test that supplies ?customerId=... and asserts the connector received non-nil Customers in ListEvents params (e.g., by extending MockStreamingConnector to capture last params).
• Add a server test for ingestedAtFrom/ingestedAtTo to ensure v1 path forwards them to streaming. This would have caught the current omission in adapter.ListEvents.

If helpful, I can sketch the mock capture and assertions.

Also applies to: 606-621

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled
• Linear integration is disabled

You can enable these sources in your CodeRabbit configuration.

Learnt from: GAlexIHU
PR: openmeterio/openmeter#2383
File: openmeter/entitlement/metered/lateevents_test.go:37-45
Timestamp: 2025-03-07T12:17:43.129Z
Learning: In the OpenMeter codebase, test files like `openmeter/entitlement/metered/lateevents_test.go` may use variables like `meterSlug` and `namespace` without explicit declarations visible in the same file. This appears to be an accepted pattern in their test structure.

openmeter/meterevent/adapter/event.go (1)

24-35: Restored ingestedAt filters pass-through in v1 list — regression fixed

Adding IngestedAtFrom/IngestedAtTo to the streaming params resolves the earlier regression. LGTM.

openmeter/meterevent/adapter/event.go (1)

154-194: Optional: break after first meter match to avoid redundant validations

Assuming event type uniquely identifies a meter, short-circuit the inner loop to reduce work. If multiple meters can legitimately share an event type, ignore this.

Apply this diff:

-		for _, m := range meterList.Items {
-			// Check if the event type matches the meter event type
-			if event.Type == m.EventType {
-				meterMatch = true
-
-				// Validate the event data against the meter
-				_, err = meter.ParseEventString(m, event.Data)
-				if err != nil {
-					validationErrors = append(validationErrors, err)
-				}
-			}
-		}
+		for _, m := range meterList.Items {
+			if event.Type == m.EventType {
+				meterMatch = true
+				if _, e := meter.ParseEventString(m, event.Data); e != nil {
+					validationErrors = append(validationErrors, e)
+				}
+				break
+			}
+		}

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled
• Linear integration is disabled

You can enable these sources in your CodeRabbit configuration.

openmeter/streaming/clickhouse/queryhelper.go (2)

36-43: Good fix: avoids invalid CASE when there are no mappings

Returning a constant projection when no WHENs were added prevents constructing an invalid “CASE … ELSE … END” without any WHENs. This addresses the earlier concern.

---

61-70: Avoid calling Or() with zero args when no subject keys are collected

If customers exist but none has SubjectKeys, query.Or(...) receives zero args. Depending on go-sqlbuilder behavior this can yield an empty/invalid predicate. Guard with a length check.

 	if len(customers) > 0 {
 		var customerSubjects []string
 
 		for _, customer := range customers {
 			customerSubjects = append(customerSubjects, customer.GetUsageAttribution().SubjectKeys...)
 		}
 
-		query = query.Where(query.Or(slicesx.Map(customerSubjects, mapFunc)...))
+		if len(customerSubjects) > 0 {
+			query = query.Where(query.Or(slicesx.Map(customerSubjects, mapFunc)...))
+		}
 	}

openmeter/streaming/clickhouse/queryhelper.go (3)

13-14: Docstring mismatch with function signature (nit)

The comment says “returns the select columns” (plural), but the function returns a single string. Please align the comment for clarity.

-// customerIdSelect returns the select columns for the customer ID.
+// selectCustomerIdColumns returns a single SELECT projection for customer_id.

---

22-34: Guard against duplicate subject keys across customers

If a subject key appears under multiple customers, this will emit multiple WHEN clauses for the same condition with different THEN values; CASE will take the first match, leading to silent, order-dependent selection. If upstream guarantees global uniqueness of subject keys across customers, we’re good—otherwise, dedupe here to avoid ambiguity and reduce query size.

Example dedup inline (keeps first occurrence):

-	// Add the case statements for each subject to customer ID mapping
+	// Add the case statements for each subject to customer ID mapping
+	seen := make(map[string]struct{})
 	for _, customer := range customers {
 		for _, subjectKey := range customer.GetUsageAttribution().SubjectKeys {
+			if _, dup := seen[subjectKey]; dup {
+				continue
+			}
+			seen[subjectKey] = struct{}{}
 			str := fmt.Sprintf(
 				"WHEN %s = '%s' THEN '%s' ",
 				subjectColumn,
 				sqlbuilder.Escape(subjectKey),
 				sqlbuilder.Escape(customer.GetUsageAttribution().ID),
 			)
 			caseBuilder.WriteString(str)
 			added = true
 		}
 	}

---

72-77: Optional: use IN() instead of large OR chains for better readability/plan size

For long subject lists, IN(...) is typically more compact than many OR-ed equality predicates. If go-sqlbuilder supports it for ClickHouse, consider switching to query.In(subjectColumn, sqlbuilder.List(...)).

openmeter/streaming/clickhouse/event_query.go (1)

146-161: DRY opportunity: factor customers/subjects extraction used in both toCountRowSQL and toSQL

Both methods replicate the logic to materialize customers and subjects before calling subjectWhere. Consider extracting a small helper (e.g., func (d queryEventsTable) materializeFilters() (custs []streaming.Customer, subjects []string)) to reduce duplication and avoid drift.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled
• Linear integration is disabled

You can enable these sources in your CodeRabbit configuration.

openmeter/meterevent/adapter/event.go (1)

24-35: Fixed: v1 ingestedAt filter pass-through is now correctly wired.

Including IngestedAtFrom/IngestedAtTo in streaming.ListEventsParams resolves the earlier regression in v1 list. LGTM.

openmeter/meterevent/adapter/event.go (3)

37-59: Honor explicit empty CustomerIDs filter by short-circuiting to no results.

If CustomerIDs is provided but empty, the current code treats it as “no filter,” returning all events. For explicit empty filters, it’s safer to return zero events to avoid unexpectedly broad queries and high load. Also keep the current path for non-empty lists.

Suggested change:

Please confirm desired API semantics: should an explicit empty filter return zero results?

-	// Resolve customer IDs to customers if provided
-	if params.CustomerIDs != nil && len(*params.CustomerIDs) > 0 {
+	// Resolve customer IDs to customers if provided
+	if params.CustomerIDs != nil {
+		// Explicitly provided empty filter => return no events
+		if len(*params.CustomerIDs) == 0 {
+			return []meterevent.Event{}, nil
+		}
 		customerList, err := a.customerService.ListCustomers(ctx, customer.ListCustomersInput{
 			Namespace:   params.Namespace,
 			CustomerIDs: *params.CustomerIDs,
 		})
 		if err != nil {
 			return nil, fmt.Errorf("list customers: %w", err)
 		}
 
-		customers := make([]streaming.Customer, 0, len(customerList.Items))
-
-		for _, c := range customerList.Items {
-			customers = append(customers, c)
-		}
+		customers := make([]streaming.Customer, 0, len(customerList.Items))
+		for _, c := range customerList.Items {
+			customers = append(customers, c)
+		}
 
 		// If no customers are found, return an empty list
 		if len(customers) == 0 {
 			return []meterevent.Event{}, nil
 		}
 
 		listParams.Customers = &customers
 	}

---

158-203: Reduce O(events × meters) validation to O(events) with an index by event type.

Current loop scans all meters per event. If meters or events are large, this will be costly. Pre-index meters by EventType once, then validate per event only against matching meters.

 func (a *adapter) validateEvents(ctx context.Context, namespace string, events []meterevent.Event) ([]meterevent.Event, error) {
   // Get all meters
   meterList, err := a.meterService.ListMeters(ctx, meter.ListMetersParams{
     Namespace: namespace,
   })
   if err != nil {
     return nil, fmt.Errorf("get meters: %w", err)
   }
 
+  // Index meters by event type to avoid scanning for every event
+  metersByType := make(map[string][]meter.Meter, len(meterList.Items))
+  for _, m := range meterList.Items {
+    metersByType[m.EventType] = append(metersByType[m.EventType], m)
+  }
+
   // Validate events against meters
   validatedEvents := make([]meterevent.Event, 0, len(events))
 
   for _, event := range events {
-    meterMatch := false
-    validationErrors := []error{}
-
-    // Check if the event type matches a meter
-    for _, m := range meterList.Items {
-      // Check if the event type matches the meter event type
-      if event.Type == m.EventType {
-        meterMatch = true
-
-        // Validate the event data against the meter
-        _, err = meter.ParseEventString(m, event.Data)
-        if err != nil {
-          validationErrors = append(validationErrors, err)
-        }
-      }
-    }
+    validationErrors := []error{}
+    ms := metersByType[event.Type]
+    meterMatch := len(ms) > 0
+    for _, m := range ms {
+      if _, perr := meter.ParseEventString(m, event.Data); perr != nil {
+        validationErrors = append(validationErrors, perr)
+      }
+    }
 
     // If no meter matches the event type, add an error to the event
     if !meterMatch {
       validationErrors = append(validationErrors, fmt.Errorf("no meter found for event type: %s", event.Type))
     }
 
     // If the event does not have a customer ID, add an error to the event
     if event.CustomerID == nil {
       validationErrors = append(validationErrors, fmt.Errorf("no customer found for event subject: %s", event.Subject))
     }
 
     event.ValidationErrors = validationErrors
     validatedEvents = append(validatedEvents, event)
   }
 
   return validatedEvents, nil
 }

Also, if EventType is guaranteed to be unique across meters, you can break after the first match to further reduce work.

---

205-251: Add negative caching to avoid repeated lookups on “not found”; consider batching later.

Nice touch with per-call positive cache. However, when a subject has no associated customer, we re-query on every occurrence. Add a small negative cache to skip redundant lookups. The FIXME about batching is valid; batching by unique subjects would further reduce DB load.

 func (a *adapter) enrichEventsWithCustomerID(ctx context.Context, namespace string, events []meterevent.Event) ([]meterevent.Event, error) {
   eventsWithCustomerID := make([]meterevent.Event, 0, len(events))
-  cache := make(map[string]string)
+  cache := make(map[string]string)          // positive cache: subject -> customerID
+  notFound := make(map[string]struct{})     // negative cache: subject with no customer
 
   for _, event := range events {
     // If the event already has a customer ID, add it to the list
     if event.CustomerID != nil {
       eventsWithCustomerID = append(eventsWithCustomerID, event)
       continue
     }
 
     // Check if the customer ID for the subject is in the cache
     if customerID, ok := cache[event.Subject]; ok {
       // Create a stable copy to take address of
       id := customerID
       event.CustomerID = &id
       eventsWithCustomerID = append(eventsWithCustomerID, event)
       continue
     }
 
+    // Skip lookup if we already know there is no customer for this subject
+    if _, nf := notFound[event.Subject]; nf {
+      eventsWithCustomerID = append(eventsWithCustomerID, event)
+      continue
+    }
+
     // FIXME: do this in a batches to avoid hitting the database for each event
     // Get the customer by usage attribution subject key
     cust, err := a.customerService.GetCustomerByUsageAttribution(ctx, customer.GetCustomerByUsageAttributionInput{
       Namespace:  namespace,
       SubjectKey: event.Subject,
     })
     if err != nil {
       if models.IsGenericNotFoundError(err) {
+        notFound[event.Subject] = struct{}{}
         eventsWithCustomerID = append(eventsWithCustomerID, event)
         continue
       }
 
       return nil, fmt.Errorf("get customer by usage attribution: %w", err)
     }
 
     // Add the customer ID to the cache
     cache[event.Subject] = cust.ID
 
     // Add the event to the list (use stable copy for pointer)
     customerID := cust.ID
     event.CustomerID = &customerID
     eventsWithCustomerID = append(eventsWithCustomerID, event)
   }
 
   return eventsWithCustomerID, nil
 }

If you want, I can sketch a batched resolver that groups by unique subject and calls a future ListByUsageAttribution API once per request.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled
• Linear integration is disabled

You can enable these sources in your CodeRabbit configuration.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled
• Linear integration is disabled

You can enable these sources in your CodeRabbit configuration.

openmeter/meterevent/adapter/event.go (2)

25-36: Regression fixed: ingestedAt filters are now passed through.

Good catch adding IngestedAtFrom/IngestedAtTo to the streaming params. This addresses the earlier v1 regression where ingestedAt filters were ignored.

---

38-60: Bug: explicit empty CustomerIDs broadens the v1 query to all events. Short-circuit to empty result.

If the client supplies an explicit empty array (customerId=[]), the current check skips filtering entirely and returns all events. That’s surprising and can be expensive. Treat an explicit empty filter as “no results.”

Apply this diff:

-	// Resolve customer IDs to customers if provided
-	if params.CustomerIDs != nil && len(*params.CustomerIDs) > 0 {
+	// Resolve customer IDs to customers if provided
+	if params.CustomerIDs != nil {
+		// Explicitly provided empty filter => no events
+		if len(*params.CustomerIDs) == 0 {
+			return []meterevent.Event{}, nil
+		}
 		customerList, err := a.customerService.ListCustomers(ctx, customer.ListCustomersInput{
 			Namespace:   params.Namespace,
 			CustomerIDs: *params.CustomerIDs,
 		})
 		if err != nil {
 			return nil, fmt.Errorf("list customers: %w", err)
 		}
 
 		customers := make([]streaming.Customer, 0, len(customerList.Items))
 
 		for _, c := range customerList.Items {
 			customers = append(customers, c)
 		}
 
 		// If no customers are found, return an empty list
 		if len(customers) == 0 {
 			return []meterevent.Event{}, nil
 		}
 
 		listParams.Customers = &customers
 	}

openmeter/meterevent/service.go (1)

71-77: Cursor implementation added on Event.

This satisfies pagination requirements and fixes the earlier interface gap.

openmeter/streaming/clickhouse/event_query.go (1)

120-123: Guard against empty customer mappings — thank you for adding this

The additional len(*d.Customers) > 0 guard ensures we only project customer_id when mappings exist, addressing the earlier review note.

openmeter/meterevent/service.go (1)

37-45: Public API: ListEventsParams gains CustomerIDs. Consider input semantics for empty slice.

Adding CustomerIDs is fine, but an explicitly empty slice meaning “no results” vs “no filter” should be clarified and enforced consistently across layers. Current adapter v1 path needs a short-circuit (see adapter comment).

pkg/filter/filter_test.go (1)

470-559: Add a test for empty In = [] to lock in semantics.

We should define whether an explicit empty list is treated as empty filter or “no results.” Recommend asserting “no results” behavior at higher layers, but at the filter level it’s worth deciding if SelectWhereExpr should be empty or generate a safe-false predicate. Add a unit test to prevent regressions.

Would you like me to open a follow-up to add:

• Validate() behavior for empty In (either error or treat as empty)
• SelectWhereExpr behavior for empty In (emit no expr or a false condition)?

I can draft both alternatives with benchmarks.

openmeter/streaming/clickhouse/event_query_v2_test.go (1)

68-82: Great addition: subject IN filter coverage (customerId mapping)

• Solid test for IN semantics with a slice argument.
• Consider adding parity tests:
  • Subject.Nin (NOT IN).
  • Guard/behavior for empty lists (should be rejected upstream; see notes on FilterStringExact below).

api/spec/src/filter.tsp (1)

91-109: New FilterStringExact type is a good fit for exact/in-only filters

The constrained surface (only $eq and $in) is appropriate for identifiers like customerId.

• Suggest documenting the mutual exclusivity explicitly in the schema comments to match backend validation.
• Optional: enforce min/max items on $in here (uncomment @minItems/@Maxitems) since this model intentionally excludes nullable types.

Apply this diff to clarify exclusivity:

 @friendlyName("FilterStringExact")
 model FilterStringExact {
+  /**
+   * Note: $eq and $in are mutually exclusive. Only one may be set.
+   */
   /**
    * The field must be equal to the provided value.
    */
   $eq?: string | null;

Optionally enforce list bounds:

-  // @minItems(1)
-  // @maxItems(100)
+  @minItems(1)
+  @maxItems(100)
   $in?: string[] | null;

api/spec/src/events.tsp (3)

68-75: Remove redundant @query decorator

You have both @query and @query(#{ explode: true }) on the same parameter. One is enough; keep the exploded form.

-    @query
-    @query(#{ explode: true })
+    @query(#{ explode: true })
     @example(#["01HZXABCD3M6E7K8J9PQRS1TUV", "01HZXABCDE3M6E7K8J9PQRSTUV"])
     customerId?: ULID[],

---

181-188: Include customerId in the filter example for discoverability

Adding customerId to the example helps clients discover the new capability quickly.

     @example(#{
       id: #{ $eq: "my-event-id" },
       source: #{ $eq: "my-event-source" },
       subject: #{ $eq: "my-event-subject" },
+      customerId: #{ $in: #["01HZXABCD3M6E7K8J9PQRS1TUV", "01HZXABCDE3M6E7K8J9PQRSTUV"] },
       type: #{ $eq: "my-event-type" },
       time: #{

---

310-314: Add customerId to the IngestedEvent example

Since the field is new and user-facing, show it in the example payload.

 @example(#{
   event: #{
     id: "5c10fade-1c9e-4d6c-8275-c52c36731d3c",
     source: "service-name",
     specversion: "1.0",
     type: "prompt",
     subject: "customer-id",
     time: DateTime.fromISO("2023-01-01T01:01:01.001Z"),
   },
+  customerId: "01HZXABCD3M6E7K8J9PQRS1TUV",
   ingestedAt: DateTime.fromISO("2023-01-01T01:01:01.001Z"),
   storedAt: DateTime.fromISO("2023-01-01T01:01:02.001Z"),
 })

pkg/filter/filter.go (2)

189-198: Defensive guard in SelectWhereExpr for empty $in

Even with validation, keeping SelectWhereExpr resilient is cheap. Emit no predicate if $in is empty.

 func (f FilterStringExact) SelectWhereExpr(field string, q *sqlbuilder.SelectBuilder) string {
   switch {
   case f.Eq != nil:
     return q.EQ(field, *f.Eq)
-  case f.In != nil:
+  case f.In != nil && len(*f.In) > 0:
     return q.In(field, *f.In)
   default:
     return ""
   }
 }

---

160-166: Add compile-time assertion that FilterStringExact implements Filter

Matches the pattern used for other filters and helps catch regressions.

Add near the existing assertions (around Lines 23-29):

var (
    // ...
    _ Filter = (*FilterStringExact)(nil)
)

api/client/javascript/src/client/schemas.ts (2)

5285-5291: Clarify FilterStringExact semantics; consider stricter schema if exclusivity is desired

The new schema looks good. Small doc nit: the description can be clearer about how $eq and $in interact. If both can be provided, say so. If they’re mutually exclusive, consider encoding that in OpenAPI (oneOf) so clients generate a discriminated union.

Apply this minimal doc tweak here to avoid ambiguity:

-    /** @description A filter for a string field allowing only equality or inclusion. */
+    /** @description A filter for a string field allowing exact match ($eq) and/or set membership ($in). */

If you want exclusivity, update the OpenAPI instead (generator-friendly), e.g.:

components:
  schemas:
    FilterStringExact:
      oneOf:
        - type: object
          properties:
            $eq:
              type: string
          required: [$eq]
        - type: object
          properties:
            $in:
              type: array
              items: { type: string }
          required: [$in]

---

5471-5475: Mention ULID in the customerId documentation for IngestedEvent

The field is a ULID per the spec elsewhere; reflecting that here helps client-side consumers.

Apply this doc-only diff:

-      /**
-       * @description The customer ID if the event is associated with a customer.
-       * @example 01G65Z755AFWAKHE12NY0CQ9FH
-       */
+      /**
+       * @description The customer ID (ULID) if the event is associated with a customer.
+       * @example 01G65Z755AFWAKHE12NY0CQ9FH
+       */

Note: If this file is generated, please make this change in the OpenAPI source so it propagates here.

api/openapi.yaml (2)

5005-5017: Tighten query param semantics for array input and examples.

Good addition. Recommend minor spec hardening for better client generation and validation.

Apply this diff to clarify CSV semantics, enforce non-empty/unique lists, and provide a concrete multi-value example:

         - name: customerId
           in: query
           required: false
-          description: The event customer ID.
+          description: The event customer ID(s). With style=form and explode=false, provide a comma-separated list (CSV).
           schema:
             type: array
+            minItems: 1
+            uniqueItems: true
             items:
               type: string
               pattern: ^[0-7][0-9A-HJKMNP-TV-Za-hjkmnp-tv-z]{25}$
               description: ULID (Universally Unique Lexicographically Sortable Identifier).
               example: 01G65Z755AFWAKHE12NY0CQ9FH
           explode: false
           style: form
+          example: 01G65Z755AFWAKHE12NY0CQ9FH,01H0F2K5B1ABCDEF234567890A

Two nits to consider:

• Parameter name is singular but accepts an array. If this diverges from your conventions, consider renaming to customerIds before release to avoid future breaking changes.
• If the backend imposes an upper bound, add maxItems (e.g., 100) to prevent abuse.

---

16798-16802: Event ingestion payload: clarify nullability and reuse constraints.

If customerId can be absent (no mapping), mark it nullable to disambiguate “omitted vs. empty”, and tighten the ULID constraints. Also consider reusing a shared ULID format across the spec to avoid pattern duplication.

Apply this diff for minimal hardening:

         customerId:
           type: string
+          nullable: true
+          minLength: 26
+          maxLength: 26
+          format: ulid
           pattern: ^[0-7][0-9A-HJKMNP-TV-Za-hjkmnp-tv-z]{25}$
           description: The customer ID if the event is associated with a customer.
           example: 01G65Z755AFWAKHE12NY0CQ9FH

Follow-up checks:

• If the core Event schema already contains customerId, avoid duplicating it at the wrapper level unless this field represents a post-validation/enrichment result. In that case, clarify it in the description (e.g., “Resolved via usage-attribution”).
• Consider introducing components/schemas/Ulid and referencing it (and an array variant) wherever ULIDs appear.

app/common/meterevent.go (1)

17-21: All NewMeterEventService call sites updated
Verification shows that every invocation of NewMeterEventService now passes three arguments—connector, customerService, service—and the generated Wire file (cmd/server/wire_gen.go:453) has been updated accordingly. There are no remaining two-argument call sites in the codebase, and Wire providers correctly include a concrete customer.Service.

Optional: for clarity, consider adding the suggested doc comment above the constructor:

+// NewMeterEventService constructs a meterevent.Service wired with streaming, customer, and meter services.
+// Note: customerService is required to enable customer attribution in event listing APIs.
 func NewMeterEventService(
   streamingConnector streaming.Connector,
   customerService customer.Service,
   meterService meter.Service,
 ) meterevent.Service {

openmeter/streaming/clickhouse/event_query.go (3)

61-63: Clarify presence vs. intent semantics for Customers pointer

Using a pointer to []streaming.Customer is good to distinguish “unset” vs “set but empty”. Consider adding a short comment on the struct field specifying:

• nil: no customer-based filtering or projection
• empty slice: same as nil (no filter/projection)

This prevents future regressions in helpers that may treat empty as “match nothing.”

---

146-161: Duplicate “derive customers/subjects then subjectWhere” logic — consider a tiny helper

The blocks that derive customers/subjects and then call subjectWhere are duplicated in both toCountRowSQL and toSQL. Extracting into a small private helper will reduce drift risk.

Example signature:

• func (d queryEventsTable) applySubjectFilters(q *sqlbuilder.SelectBuilder) *sqlbuilder.SelectBuilder

---

121-123: append usage is correct; helper returns a single string
The function

func selectCustomerIdColumns(eventsTableName string, customers []streaming.Customer) string

returns a single string, so

- selectColumns = append(selectColumns, selectCustomerIdColumns(d.EventsTableName, *d.Customers))
+ // no change needed: append accepts a string element
  selectColumns = append(selectColumns, selectCustomerIdColumns(d.EventsTableName, *d.Customers))

is already correct.

Nit (optional): consider renaming the helper to a singular form—e.g. selectCustomerIdColumn—to better reflect that it returns one string rather than a slice.

api/openapi.cloud.yaml (4)

5004-5016: Clarify serialization, add array constraints, and provide a parameter-level example for customerId query param.

With style: form and explode: false, clients must send a comma-separated list (e.g., ?customerId=01G...,01H...). If that’s the intended contract, let’s:

• make this explicit in the description,
• add minItems and uniqueItems to the array schema,
• include a parameter-level example that matches the chosen serialization.

If, instead, you want repeated keys (e.g., ?customerId=01G...&customerId=01H...), switch explode to true. Please confirm the expected client behavior across SDKs before merging.

Proposed in-place diff:

         - name: customerId
           in: query
           required: false
-          description: The event customer ID.
+          description: Filter events by customer ID(s). With style=form and explode=false, provide a comma-separated list (e.g., customerId=01G65Z755AFWAKHE12NY0CQ9FH,01H8F6W6Z6V0V1NX5X3S9B1ABC).
           schema:
             type: array
+            minItems: 1
+            uniqueItems: true
             items:
               type: string
               pattern: ^[0-7][0-9A-HJKMNP-TV-Za-hjkmnp-tv-z]{25}$
               description: ULID (Universally Unique Lexicographically Sortable Identifier).
               example: 01G65Z755AFWAKHE12NY0CQ9FH
           explode: false
           style: form
+          example: 01G65Z755AFWAKHE12NY0CQ9FH,01H8F6W6Z6V0V1NX5X3S9B1ABC

To reduce duplication of the ULID pattern across the spec, consider introducing a reusable schema and referencing it here (see below in a separate block).

---

5004-5016: DRY the ULID pattern via a reusable component.

You repeat the ULID regex here and elsewhere. Define a single ULID schema under components/schemas and reference it to avoid drift.

Add this component once (outside this hunk):

components:
  schemas:
    ULID:
      type: string
      pattern: '^[0-7][0-9A-HJKMNP-TV-Za-hjkmnp-tv-z]{25}$'
      description: ULID (Universally Unique Lexicographically Sortable Identifier).
      example: 01G65Z755AFWAKHE12NY0CQ9FH

Then replace the inline items schema here with:

-            items:
-              type: string
-              pattern: ^[0-7][0-9A-HJKMNP-TV-Za-hjkmnp-tv-z]{25}$
-              description: ULID (Universally Unique Lexicographically Sortable Identifier).
-              example: 01G65Z755AFWAKHE12NY0CQ9FH
+            items:
+              $ref: '#/components/schemas/ULID'

---

11316-11317: Optional: Attach a description without breaking $ref (wrap via allOf).

OpenAPI tooling often ignores siblings next to $ref. If you want a property-level description here, wrap the reference to keep compatibility.

Proposed diff:

-                  customerId:
-                    $ref: '#/components/schemas/FilterStringExact'
+                  customerId:
+                    allOf:
+                      - $ref: '#/components/schemas/FilterStringExact'
+                    description: Filter by exact customerId using $eq or by a set using $in.

---

16438-16442: Propose reusing ULID schema for customerId and marking it read-only

After verifying, the Event schema in both api/openapi.cloud.yaml (line 15832+) and api/openapi.yaml (line 16098+) does not define customerId, so there’s no duplication to worry about. You can safely refactor the standalone customerId field to reference the shared ULID component and mark it as read-only.

• File: api/openapi.cloud.yaml (lines 16438–16442)

-        customerId:
-          type: string
-          pattern: ^[0-7][0-9A-HJKMNP-TV-Za-hjkmnp-tv-z]{25}$
-          description: The customer ID if the event is associated with a customer.
-          example: 01G65Z755AFWAKHE12NY0CQ9FH
+        customerId:
+          allOf:
+            - $ref: '#/components/schemas/ULID'
+          readOnly: true
+          description: The customer ID if the event is associated with a customer.
+          example: 01G65Z755AFWAKHE12NY0CQ9FH

openmeter/streaming/eventparams.go (2)

12-33: Namespace duplication in v1 params vs. method signature causes ambiguity

ListEventsParams includes Namespace string, but streaming.Connector.ListEvents already takes namespace string separately and the ClickHouse code uses the method arg (not the struct field). This invites mismatches.

• Option A (preferred): remove Namespace from ListEventsParams (breaking change).
• Option B: keep field but enforce equality in the connector and/or mark as deprecated to prevent misuse.

If you keep it, consider marking it as deprecated in the comment:

 type ListEventsParams struct {
-  // The namespace.
-  Namespace string
+  // DEPRECATED: Ignored by streaming.Connector.ListEvents. Use the method's `namespace` argument.
+  // Kept for compatibility with higher layers.
+  Namespace string

And see connector.go suggestion to assert equality at runtime.

---

36-61: Add safe guard for negative limit in v1 params

V1 Limit is an int with no validation. If zero is meaningful (e.g., “use default”), that’s fine; but negative should be rejected.

Apply this diff:

 func (i ListEventsParams) Validate() error {
   var errs []error
@@
   if i.IngestedAtFrom != nil && i.IngestedAtTo != nil && i.IngestedAtTo.Before(*i.IngestedAtFrom) {
     errs = append(errs, fmt.Errorf("ingestedAtTo date is before ingestedAtFrom date: %s < %s", i.IngestedAtTo.Format(time.RFC3339), i.IngestedAtFrom.Format(time.RFC3339)))
   }
 
+  if i.Limit < 0 {
+    errs = append(errs, errors.New("limit cannot be negative"))
+  }
+
   return errors.Join(errs...)
 }

If zero must be rejected too, change the guard to < 1 and ensure callers pass an explicit default.

openmeter/streaming/clickhouse/connector.go (1)

309-324: Confirm Customers param type matches the ClickHouse query builder

You thread params.Customers into queryEventsTable{ Customers: params.Customers }. In streaming params this is *[]Customer, while meterevent/service.go exposes *[]string (ULIDs). Ensure the builder expects the same element type to avoid compile-time or logical mismatches.

Would you like me to align v1 to CustomerIDs *[]string for consistency with the service layer and API? I can propose a full refactor diff if desired.

openmeter/server/server_test.go (3)

168-186: Confirm API semantics for missing customer ID (empty string vs null/omitted)

Test expects CustomerId: "" when not set. If API intends “not available” rather than “empty value,” prefer null (pointer nil) or omit the field to reduce ambiguity and wire compatibility.

If the agreed shape is null, update expected to CustomerId: nil and ensure the mapping returns nil when RawEvent.CustomerID is nil.

---

606-620: Enhance mock to include CustomerID for at least one event to assert mapping

Right now, mocks never set CustomerID, so we only test the “missing customer” path. Add a second event with a non-empty ULID to validate end-to-end propagation in v1.

Apply this diff:

 func (c *MockStreamingConnector) ListEvents(ctx context.Context, namespace string, params streaming.ListEventsParams) ([]streaming.RawEvent, error) {
-  events := []streaming.RawEvent{
-    {
+  events := []streaming.RawEvent{
+    {
       ID:         mockEvent.ID(),
       Type:       mockEvent.Type(),
       Source:     mockEvent.Source(),
       Subject:    mockEvent.Subject(),
       Time:       mockEvent.Time(),
       Data:       string(mockEvent.Data()),
       IngestedAt: time.Time{},
       StoredAt:   time.Time{},
     },
+    {
+      ID:         "evt-2",
+      Type:       mockEvent.Type(),
+      Source:     mockEvent.Source(),
+      Subject:    mockEvent.Subject(),
+      Time:       mockEvent.Time(),
+      Data:       string(mockEvent.Data()),
+      CustomerID: lo.ToPtr("01ARZ3NDEKTSV4RRFFQ69G5FAV"), // sample ULID
+    },
   }
   return events, nil
 }

And add a companion assertion in the “list events v1” test for the second item’s CustomerId.

---

622-636: Add a V2 list-events test case to cover the new customerId filter and response mapping

Given the V2 path now validates and (after fixes) will support CustomerID, add a test like “list events v2 with customer filter” to ensure both filtering and payload mapping work.

I can draft the test case and adjust the mock to assert that params.CustomerID is honored by returning only matching events. Let me know, and I’ll provide a concrete diff.

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled
• Linear integration is disabled

You can enable these sources in your CodeRabbit configuration.

Learnt from: GAlexIHU
PR: openmeterio/openmeter#2383
File: openmeter/entitlement/metered/lateevents_test.go:37-45
Timestamp: 2025-03-07T12:17:43.129Z
Learning: In the OpenMeter codebase, test files like `openmeter/entitlement/metered/lateevents_test.go` may use variables like `meterSlug` and `namespace` without explicit declarations visible in the same file. This appears to be an accepted pattern in their test structure.

openmeter/meterevent/adapter/event.go (2)

144-164: Post-processing pipeline ordering is sensible.

Mapping → enrichment → validation ensures customer mapping errors surface in ValidationErrors. Looks good.

---

239-284: Enrichment correctness: stable pointer copies and cache usage look good.

• Good fix using stable local copies for CustomerID pointers.
• The per-subject cache avoids redundant lookups; the FIXME about batching is noted.

openmeter/apiconverter/filter.go (1)

13-24: Converters for FilterStringExact added coherently.

The new ConvertStringExact and ConvertStringExactPtr align with existing converter patterns and internal filter types. No issues.

openmeter/apiconverter/filter.gen.go (1)

137-150: Generated mapping for FilterStringExact is correct.

Eq/In are mapped 1:1; pointer wrapper handles nil safely. This mirrors other generated converters.

openmeter/meterevent/service.go (2)

61-69: Event now exposes CustomerID; downstream mapping compatibility looks good.

Matches streaming.RawEvent.CustomerID and handler mapping needs. No concerns.

---

140-148: V2 adds CustomerID filter with validation.

Using FilterStringExact is appropriate here. Validation via ValidateWithComplexity(1) is consistent with other fields.

Also applies to: 182-186

pkg/filter/filter_test.go (2)

360-559: Solid coverage for FilterStringExact Validate and SelectWhereExpr.

Covers nil/eq/in and multiple-operator error paths. Nice.

---

527-559: IsEmpty tests cover both Eq and In cases.

Good to see parity with other filter types.

openmeter/streaming/clickhouse/event_query_v2_test.go (2)

9-9: Import switch to streaming package is correct

Using streaming.ListEventsV2Params here aligns tests with the refactor away from meterevent types. No issues spotted.

---

112-114: Verify epoch‐second consistency in all time filters

Before merging, please confirm that the time and ingested_at filters in your v2 query builder consistently bind Unix‐seconds, just as the cursor predicates do via Cursor.Time.Unix(). In particular:

• In queryEventsTableV2.toSQL(), the cursor clause clearly uses
  q.Params.Cursor.Time.Unix().
  Ensure that q.Params.Time.SelectWhereExpr("time", …) and
  q.Params.IngestedAt.SelectWhereExpr("ingested_at", …) likewise pass epoch‐second values (not raw time.Time), so the driver binds an Int64 matching your ClickHouse schema.
• If SelectWhereExpr doesn’t currently invoke .Unix(), update its implementation (or wrap it) to convert time.Time → int64 before binding.
• Add a brief comment in your table‐creation code (e.g. in createEventsTable.toSQL()) noting that both time and ingested_at columns are stored as epoch seconds, to prevent future confusion.

pkg/filter/filter.go (1)

189-198: Confirm ClickHouse driver slice expansion for IN placeholders

Our unit tests for FilterStringExact.SelectWhereExpr currently assert:

• SQL:

  SELECT * FROM table WHERE test_field IN (?)

• Args:

  []interface{}{[]string{"test1", "test2"}}

However, not all SQL drivers automatically expand a single placeholder with a slice argument into multiple placeholders (e.g. IN (?, ?)). Please verify that the ClickHouse Go driver you’re using correctly expands slice arguments in this scenario, or adjust the code to emit repeated placeholders and flatten the slice into individual args.

Affected locations:

• pkg/filter/filter.go – SelectWhereExpr (lines 189–198)
• pkg/filter/filter_test.go – TestFilterStringExact_SelectWhereExpr (around line 439)

Suggested actions:

• Consult your ClickHouse driver’s documentation or add an integration test to confirm its behavior when binding a slice to a single ?.
• If the driver does not expand the slice, modify SelectWhereExpr to build an IN (?, ?, …) clause manually, for example:

  vals := *f.In
  placeholders := strings.TrimRight(strings.Repeat("?,", len(vals)), ",")
  expr := fmt.Sprintf("%s IN (%s)", field, placeholders)
  args := make([]interface{}, len(vals))
  for i, v := range vals {
    args[i] = v
  }
  q = q.Args(args...)
  return expr

• Alternatively, use or implement a helper that both repeats placeholders and flattens the slice.

api/client/javascript/src/client/schemas.ts (2)

10931-10931: LGTM: Public alias for FilterStringExact is correctly exposed

This keeps parity with the Go client and API types shown in the PR. No further action needed.

---

17620-17621: Tighten JSDoc for customerId and confirm repeated‐key serialization

The JSDoc on line 17620 currently reads

/** @description The event customer ID. */
customerId?: string[]

To clarify its array nature and filtering behavior, please update it to:

- /** @description The event customer ID. */
+ /** @description Filter by event customer ULIDs. Multiple values behave as an IN filter (e.g. ?customerId=…&customerId=…). */

• We’ve verified that the generated client’s default query serializer is configured with explode: true (client/index.ts), so array parameters are emitted as repeated keys (customerId=…&customerId=…) as intended.
• OpenAPI relies on its default style: form and explode: true semantics for array query parameters, which aligns with the client’s serialization.

api/openapi.yaml (2)

11317-11320: Filter operator scope and precedence with the new query param.

Using FilterStringExact for customerId is appropriate. However, the endpoint now supports both the plain query param (customerId[]) and the JSON filter (filters.customerId). Please confirm and document how they combine (AND vs OR, and precedence on duplicates). If it’s an AND, state it in the parameter or endpoint description to avoid client confusion.

Would you like me to add a one-liner to the endpoint description clarifying: “If both customerId and filters.customerId are provided, both must match (logical AND)”?

---

5005-5017: Review update: v2 Events customerId usage differs by design

The v2 /api/v2/events operations no longer expose a standalone customerId query parameter—instead they use a single filter parameter whose JSON schema includes customerId via FilterStringExact. Both api/openapi.yaml and api/openapi.cloud.yaml are consistent in this approach, and there is exactly one definition of FilterStringExact in each spec.

Key points:

• The direct customerId array parameter exists only on v1 endpoints (and matches identically in both YAMLs).
• On v2, filtering—including by customerId—is handled via the filter query parameter (FilterStringExact), and both specs align on that.
• There are no duplicate or conflicting FilterStringExact definitions.
• No component parameter named customerId is defined under components/parameters (aside from its inclusion in v1 via usage in generated BillingProfileListCustomerOverridesParams).

You can safely ignore the original suggestion to add a standalone customerId param on v2 endpoints.

Likely an incorrect or invalid review comment.

app/common/meterevent.go (2)

6-6: Import addition looks correct and necessary

The new customer import is used in the constructor signature; no unused-import issues.

---

22-22: Constructor wiring: argument order matches adapter.New signature

adapter.New(streamingConnector, customerService, meterService) aligns with the updated adapter constructor.

openmeter/streaming/clickhouse/event_query.go (2)

83-101: Count-path now applies subject/customer filters via subjectWhere — good; ensure no-ops on empty inputs

The unified use of subjectWhere in toCountRowSQL keeps estimated row counts consistent with the data query path. Verify subjectWhere is a no-op when both slices are empty to avoid unnecessary CASE/WHERE fragments.

---

162-165: Double-check Desc() chaining order in go-sqlbuilder

Ensure sqlbuilder formats as ORDER BY time DESC with query.Desc().OrderBy("time"). Some builders expect OrderBy(...).Desc(). If formatting differs, swap the call order.

Consider adding/adjusting a unit test that asserts the rendered ORDER BY clause.

api/openapi.cloud.yaml (1)

11316-11317: Confirm semantics when both $eq and $in are provided for customerId filter.

You reference FilterStringExact for customerId. Please confirm whether providing both $eq and $in is allowed and how it’s combined (AND vs OR). If you want to allow either but not necessarily both, encode that in the schema (see proposed constraints added to FilterStringExact in a later comment).

openmeter/server/server_test.go (1)

457-465: LGTM: proper wiring of customerService into meterevent adapter

This matches the new constructor signature and keeps the server assembly coherent.

openmeter/meterevent/adapter/event.go (3)

99-127: v2 customerId filter: guard empties and AND-combine with existing Subject instead of setting a second operator.

• If CustomerID.Eq/In resolves to an empty set, we should return no events rather than broadening the query.
• Setting Subject.In when Subject already has an operator risks multi-operator misuse; combine via AND with the existing Subject filter.

Apply this diff:

 	// Resolve customer IDs to customers if provided
 	if params.CustomerID != nil {
 		customerIDs := make([]string, 0)
 		if params.CustomerID.Eq != nil {
 			customerIDs = append(customerIDs, *params.CustomerID.Eq)
 		}
 		if params.CustomerID.In != nil {
 			customerIDs = append(customerIDs, *params.CustomerID.In...)
 		}
 
+		// Explicit empty filter => no events
+		if len(customerIDs) == 0 {
+			return pagination.NewResult([]meterevent.Event{}), nil
+		}
+
 		customerList, err := a.customerService.ListCustomers(ctx, customer.ListCustomersInput{
 			Namespace:   params.Namespace,
 			CustomerIDs: customerIDs,
 		})
 		if err != nil {
 			return pagination.Result[meterevent.Event]{}, fmt.Errorf("list customers: %w", err)
 		}
 
 		var subjectKeys []string
 		for _, c := range customerList.Items {
 			subjectKeys = append(subjectKeys, c.GetUsageAttribution().SubjectKeys...)
 		}
 
-		if listParams.Subject == nil {
-			listParams.Subject = &filter.FilterString{}
-		}
-
-		listParams.Subject.In = &subjectKeys
+		// Mapping produced no subjects => no events
+		if len(subjectKeys) == 0 {
+			return pagination.NewResult([]meterevent.Event{}), nil
+		}
+
+		// Combine with existing Subject using AND to preserve single-operator semantics
+		if listParams.Subject == nil {
+			listParams.Subject = &filter.FilterString{In: &subjectKeys}
+		} else {
+			existing := *listParams.Subject
+			and := []filter.FilterString{
+				existing,
+				{In: &subjectKeys},
+			}
+			listParams.Subject = &filter.FilterString{And: &and}
+		}
 	}

---

38-60: Treat explicitly empty CustomerIDs as “no results” (avoid broad v1 queries).

If CustomerIDs is provided but empty, the current code skips filtering and returns all events. That’s surprising and potentially expensive; explicit empty filters should yield an empty result set.

Apply this diff:

-	// Resolve customer IDs to customers if provided
-	if params.CustomerIDs != nil && len(*params.CustomerIDs) > 0 {
+	// Resolve customer IDs to customers if provided
+	if params.CustomerIDs != nil {
+		// Explicitly provided empty filter => return no events
+		if len(*params.CustomerIDs) == 0 {
+			return []meterevent.Event{}, nil
+		}
 		customerList, err := a.customerService.ListCustomers(ctx, customer.ListCustomersInput{
 			Namespace:   params.Namespace,
 			CustomerIDs: *params.CustomerIDs,
 		})
 		if err != nil {
 			return nil, fmt.Errorf("list customers: %w", err)
 		}
 
 		customers := make([]streaming.Customer, 0, len(customerList.Items))
 
 		for _, c := range customerList.Items {
 			customers = append(customers, c)
 		}
 
 		// If no customers are found, return an empty list
 		if len(customers) == 0 {
 			return []meterevent.Event{}, nil
 		}
 
 		listParams.Customers = &customers
 	}

---

141-142: meterevent.Event must implement pagination.Item (Cursor method missing)

The call to pagination.NewResult(meterEvents) requires that meterevent.Event satisfy the Item interface (defined in pkg/pagination/v2/pagination.go), which mandates a Cursor() Cursor method. Currently, the Event struct has no such method, so the code will not compile.

Please add a Cursor implementation on the Event type—e.g., in openmeter/meterevent/service.go—so that it returns the appropriate pagination.Cursor. For example:

diff --git a/openmeter/meterevent/service.go b/openmeter/meterevent/service.go
index abc1234..def5678 100644
--- a/openmeter/meterevent/service.go
+++ b/openmeter/meterevent/service.go
@@ type Event struct {
     ID        string
     Timestamp time.Time
     // other fields...
 }
+
+// Cursor implements pagination.Item for Event.
+func (e Event) Cursor() pagination.Cursor {
+    return pagination.Cursor{
+        Time: e.Timestamp,
+        ID:   e.ID,
+    }
+}

• File requiring change:

• openmeter/meterevent/service.go (add Cursor() pagination.Cursor)
  • Why:
• pagination.NewResult[T Item] requires T to implement Cursor() Cursor pkg/pagination/v2/pagination.go
• Without this method, meterevent.Event does not satisfy the Item interface and the call on lines 141–142 of openmeter/meterevent/adapter/event.go will fail.

openmeter/apiconverter/filter.gen.go (1)

63-76: Add focused tests for the new converters

Please add unit tests that assert:

• Eq only → Eq set, In nil
• In only → In set, Eq nil
• Both set → both propagated
• Ptr nil → returns nil

This guards against accidental generator/config changes.

Example test outline (outside this file):

func TestConvertIDExact(t *testing.T) {
  eq := "01J8..." // ULID sample
  in := []string{"01J8...", "01J9..."}
  got := apiconverter.ConvertIDExact(api.FilterIDExact{Eq: &eq, In: &in})
  if got.Eq != &eq { t.Fatalf("Eq mismatch") }
  if got.In != &in { t.Fatalf("In mismatch") }
}

func TestConvertIDExactPtr_Nil(t *testing.T) {
  if apiconverter.ConvertIDExactPtr(nil) != nil { t.Fatalf("expected nil") }
}

openmeter/apiconverter/filter.go (1)

13-17: Confirm intent of // goverter:ignoreMissing for IDExact; consider documenting it inline

With ignoreMissing, if FilterIDExact gains new fields later (e.g., $ne, $nin), they’ll be silently dropped in the conversion, which may hide capability drift. If the intent is to strictly allow only $eq/$in for IDs, it’s fine—recommend an explicit comment to make this future-proof for readers.

Apply this small doc tweak:

-    // goverter:ignoreMissing
+    // goverter:ignoreMissing
+    // NOTE: FilterIDExact intentionally supports only $eq and $in for ID fields.
+    // If the API adds more operators later, they are ignored by design here.
     ConvertIDExact    func(api.FilterIDExact) filter.FilterString
     ConvertIDExactPtr func(*api.FilterIDExact) *filter.FilterString

api/openapi.yaml (2)

5005-5017: Tighten query param validation and clarify CSV semantics for customerId.

Good addition. Recommend hardening and clarifying usage for clients.

Apply this diff:

-        - name: customerId
+        - name: customerId
           in: query
           required: false
-          description: The event customer ID.
+          description: Filter by customer ULIDs. When multiple values are provided, they must be comma-separated (style=form, explode=false).
           schema:
             type: array
+            minItems: 1
+            uniqueItems: true
             items:
               type: string
               pattern: ^[0-7][0-9A-HJKMNP-TV-Za-hjkmnp-tv-z]{25}$
               description: ULID (Universally Unique Lexicographically Sortable Identifier).
               example: 01G65Z755AFWAKHE12NY0CQ9FH
           explode: false
           style: form
+          example: 01G65Z755AFWAKHE12NY0CQ9FH,01H0SZ755AFWAKHE12NY0CQ9FF

---

16803-16807: Event schema does not include customerId; marking as readOnly is safe and recommended

I’ve verified that under components.schemas.Event (lines ~16098–16104) there is no existing customerId property—so this definition isn’t a duplicate. However, if customerId on this response model is derived rather than client-supplied, it’s best practice to mark it as read-only.

• Apply this diff to clearly indicate that clients should not send this field:

         customerId:
           type: string
           pattern: ^[0-7][0-9A-HJKMNP-TV-Za-hjkmnp-tv-z]{25}$
           description: The customer ID if the event is associated with a customer.
           example: 01G65Z755AFWAKHE12NY0CQ9FH
+          readOnly: true

api/client/javascript/src/client/schemas.ts (2)

5254-5263: Remove nullable from optional filter schema properties

The ? modifier already indicates that a property may be omitted, so including | null introduces an extra “null” state and inconsistent encoding. Please drop | null from all optional filter properties in api/client/javascript/src/client/schemas.ts.

• FilterIDExact (lines 5258–5263)
• FilterString (lines 5265–5293)
• FilterTime (lines 5299–5320)

Apply within each filter block, for example:

-    /** @description A filter for an ID (ULID) field allowing only equality or inclusion. */
+    /** @description A filter for an ID (ULID) field allowing only equality or inclusion. */
@@
-      $eq?: string | null
+      $eq?: string
@@
-      $in?: string[] | null
+      $in?: string[]

And similarly remove | null from all of:

• FilterString: $eq, $ne, $in, $nin, $like, $nlike, $ilike, $nilike, $gt, $gte, $lt, $lte, $and, $or
• FilterTime: $gt, $gte, $lt, $lte, $and, $or

Ensure consistency across the file by running:

rg -nP '\$[A-Za-z]+\?: [^|\n]+\| null' api/client/javascript/src/client/schemas.ts

and updating each match to drop the | null union.

---

10933-10933: Normalize “ID” capitalization in exported type aliases

We currently have several exported aliases whose names don’t exactly match the schema’s casing:

• FilterIdExact → components['schemas']['FilterIDExact']
• IdResource → components['schemas']['IDResource']
• UlidOrExternalKey → components['schemas']['ULIDOrExternalKey']

To eliminate this drift, pick one of the following patterns:

Option A: Align aliases to the exact schema names (no deprecated variants):

-export type FilterIdExact = components['schemas']['FilterIDExact']
+export type FilterIDExact = components['schemas']['FilterIDExact']

-export type IdResource = components['schemas']['IDResource']
+export type IDResource = components['schemas']['IDResource']

-export type UlidOrExternalKey = components['schemas']['ULIDOrExternalKey']
+export type ULIDOrExternalKey = components['schemas']['ULIDOrExternalKey']

Option B: Retain the existing Id/Ulid forms for backward-compatibility, but mark them deprecated:

-export type FilterIdExact = components['schemas']['FilterIDExact']
+export type FilterIDExact = components['schemas']['FilterIDExact']
+/** @deprecated Use FilterIDExact */
+export type FilterIdExact = FilterIDExact

-export type IdResource = components['schemas']['IDResource']
+export type IDResource = components['schemas']['IDResource']
+/** @deprecated Use IDResource */
+export type IdResource = IDResource

-export type UlidOrExternalKey = components['schemas']['ULIDOrExternalKey']
+export type ULIDOrExternalKey = components['schemas']['ULIDOrExternalKey']
+/** @deprecated Use ULIDOrExternalKey */
+export type UlidOrExternalKey = ULIDOrExternalKey

Please apply the chosen convention consistently across this file—and any others exporting schema aliases—to reduce surprises and ensure a uniform developer experience.

api/openapi.cloud.yaml (1)

5004-5016: Tighten v1 customerId array param: require at least one item and uniqueness; add a CSV example.

Prevents empty/duplicate ULIDs and clarifies CSV serialization for clients.

         - name: customerId
           in: query
           required: false
           description: The event customer ID.
           schema:
             type: array
+            minItems: 1
+            uniqueItems: true
             items:
               type: string
               pattern: ^[0-7][0-9A-HJKMNP-TV-Za-hjkmnp-tv-z]{25}$
               description: ULID (Universally Unique Lexicographically Sortable Identifier).
               example: 01G65Z755AFWAKHE12NY0CQ9FH
           explode: false
           style: form
+          example: 01G65Z755AFWAKHE12NY0CQ9FH,01H8F6W6Z6V0V1NX5X3S9B1ABC

openmeter/meterevent/adapter/event.go (2)

191-236: Validation is clear; minor perf consideration if EventType is unique.

If EventType is guaranteed unique per namespace, you can break after the first match to avoid extra Parse calls. If not guaranteed, keep as is.

---

239-284: Good: stable-pointer fix and subject cache; consider batching lookups.

• Using a stable local copy for CustomerID avoids pointer aliasing bugs. Nice.
• Optional: pre-deduplicate subjects and perform a batched lookup to reduce N calls; you already noted the FIXME.

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled
• Linear integration is disabled

You can enable these sources in your CodeRabbit configuration.

openmeter/apiconverter/filter.gen.go (1)

63-76: IDExact converters map Eq/In correctly and are nil-safe — LGTM

The generated ConvertIDExact/ConvertIDExactPtr mirror the pattern used by other converters, correctly wiring Eq and In and preserving pointer semantics. No functional or performance concerns here.

openmeter/apiconverter/filter.go (1)

13-17: ConvertIDExactPtr correctly wired for CustomerId filter mapping

The HTTP handler in openmeter/meterevent/httphandler/mapping.go already invokes the new converter for customerId:

• openmeter/meterevent/httphandler/mapping.go:34 – p.CustomerID = apiconverter.ConvertIDExactPtr(params.Filter.CustomerId)

No further changes are required.

api/openapi.yaml (1)

11317-11318: LGTM: customerId filter is correctly wired to a dedicated exact/inclusion schema.

api/client/javascript/src/client/schemas.ts (1)

5474-5478: Next Steps: Verify Event.customerId Definition

Before updating docs on IngestedEvent.customerId, let’s confirm whether components['schemas']['Event'] also declares a customerId field and where it lives. Please run:

# Show around the suspected `Event` definition (near line 3843)
sed -n '3800,3860p' api/client/javascript/src/client/schemas.ts | sed -n '1,20p'

Once we know if—and in what context—Event exposes customerId, we can decide:

• If both Event and IngestedEvent have customerId but with different semantics, call out both in the doc (e.g. “Event’s customerId comes directly from payload; IngestedEvent’s customerId is derived server-side”).
• If only IngestedEvent has it, update the doc as suggested.
• If the fields are identical, note that and leave a single unified description.

Please share the snippet around the Event interface so we can finalize the doc tweak.

api/openapi.cloud.yaml (2)

11316-11317: LGTM: v2 filter wires customerId to a dedicated ULID-exact schema.

Referencing FilterIDExact for customerId in v2 keeps format/semantics consistent and reusable across endpoints.

---

16443-16447: No duplicate customerId in Event schema – no action needed.

I verified that components.schemas.Event in api/openapi.cloud.yaml does not define a customerId property, so there’s no redundancy or conflict to resolve. You can safely disregard the original suggestion.

Likely an incorrect or invalid review comment.

openmeter/meterevent/adapter/event.go (3)

25-36: Good: v1 now passes through IngestedAt filters (regression fixed).

The addition of IngestedAtFrom/IngestedAtTo to streaming.ListEventsParams restores v1 ingestedAt filtering. Looks correct.

---

86-97: v2 params mapping looks consistent with streaming schema.

Namespace, ClientID, Cursor, Limit, ID/Source/Subject/Type/Time/IngestedAt are correctly forwarded.

---

166-188: Mapping RawEvent → Event looks correct and preserves nullable customerId.

Field-by-field copy, including CustomerID pointer and timestamps, is sound.

openmeter/streaming/clickhouse/event_query_v2_test.go (1)

33-36: Migration to streaming.ListEventsV2Params looks consistent here

All test cases now construct params via streaming.ListEventsV2Params. Good alignment with the refactor.

Also applies to: 45-51, 60-66, 75-81, 90-97, 107-113, 178-181, 190-196, 205-210, 220-226, 235-251

api/spec/src/events.tsp (2)

309-313: Include customerId in IngestedEvent example to showcase the new field

You introduced customerId?: ULID; but it’s missing from the @example above. Adding it helps users see the end-to-end mapping result.

Suggested example tweak (outside this hunk):

@@
 @example(#{
   event: #{
     id: "5c10fade-1c9e-4d6c-8275-c52c36731d3c",
     source: "service-name",
     specversion: "1.0",
     type: "prompt",
     subject: "customer-id",
     time: DateTime.fromISO("2023-01-01T01:01:01.001Z"),
   },
+  customerId: "01HZXABCD3M6E7K8J9PQRS1TVW",
   ingestedAt: DateTime.fromISO("2023-01-01T01:01:01.001Z"),
   storedAt: DateTime.fromISO("2023-01-01T01:01:02.001Z"),
 })

Optional: In the field doc, you might clarify when it’s omitted (e.g., “omitted if no customer mapping found”).

---

183-187: Add customerId to the filter example and clarify default AND semantics

To help clients adopt the new V2 filters, please mirror the customerId field in the sample and call out how top-level fields are combined:

• In api/spec/src/events.tsp, update the @example block to include a valid ULID for customerId.
• Add a brief note above the example stating that top-level filter properties are AND-combined by default, while $and/$or operators allow grouping for more complex logic.

Example diff:

@@ @example(#{
   subject: #{ $eq: "my-event-subject" },
+  customerId: #{ $eq: "01HZXABCD3M6E7K8J9PQRS1TVW" },
   type: #{ $eq: "my-event-type" },
@@

No changes are needed in api/spec/src/filter.tsp: the FilterIDExact doc (around line 93) already specifies support for equality ($eq) and inclusion ($in).

openmeter/streaming/clickhouse/queryhelper.go (1)

83-88: Nit: minor allocation reduction

Passing subjects directly is fine, but if this runs hot, consider deduping subjects to shrink IN lists when both Customers and Subjects are applied upstream (two INs). Low priority.

openmeter/streaming/eventparams.go (2)

63-87: Consider exposing CustomerID filter at the streaming layer (parity with API v2)

Today, ListEventsV2Params accepts Customers (materialized set) rather than the API’s CustomerId filter. It works because the adapter resolves IDs → Customers, but it couples two layers. Consider adding CustomerID (exact) here and optionally keeping Customers for pre-resolved cases.

---

89-148: Optional: mirror v1 ClientID validation in v2

v1 Validate rejects empty ClientID when provided; v2 currently doesn’t. Aligning behavior avoids surprises.

 if p.Cursor != nil {
   if err := p.Cursor.Validate(); err != nil {
     errs = append(errs, fmt.Errorf("cursor: %w", err))
   }
 }
+if p.ClientID != nil && *p.ClientID == "" {
+  errs = append(errs, errors.New("client id cannot be empty"))
+}

openmeter/meterevent/adapter/event.go (3)

40-53: Treat explicitly empty v1 CustomerIDs as “no results”

If CustomerIDs is set but empty, the current code skips setting a filter and returns all events. Consider short-circuiting to return an empty list to match user intent and prevent heavy scans.

- // Resolve customer IDs to customers if provided
- if params.CustomerIDs != nil && len(*params.CustomerIDs) > 0 {
+ // Resolve customer IDs to customers if provided
+ if params.CustomerIDs != nil {
+   if len(*params.CustomerIDs) == 0 {
+     return []meterevent.Event{}, nil
+   }
    customers, err := a.listCustomers(ctx, params.Namespace, *params.CustomerIDs)
    if err != nil {
      return nil, fmt.Errorf("list customers: %w", err)
    }
    // If no customers are found, return an empty list
    if len(customers) == 0 {
      return []meterevent.Event{}, nil
    }
    listParams.Customers = &customers
- }
+ }

---

256-302: Batch attribution lookups to reduce N+1 calls

enrichEventsWithCustomerID issues one GetCustomerByUsageAttribution per event. Consider batching by unique subjects (or adding a ListCustomersBySubjects API) to cut latency and load. Cache helps, but batching first occurrence of each subject would still be a big win.

I can draft a batched path that collects unique subjects, queries once, populates the cache, then assigns CustomerID.

---

122-160: listCustomers: tighten “not found” detection and avoid O(n²)

Length check is unnecessary; you already scan inputs to find misses. Build a set of returned IDs and compute differences in O(n).

- if len(customerList.Items) != len(customerIDs) {
-   var notFoundCustomerIDs []string
-   for _, c := range customerIDs {
-     _, found := lo.Find(customerList.Items, func(item customer.Customer) bool {
-       return item.ID == c
-     })
-     if !found {
-       notFoundCustomerIDs = append(notFoundCustomerIDs, c)
-     }
-   }
-   if len(notFoundCustomerIDs) > 0 {
-     return nil, models.NewGenericValidationError(
-       fmt.Errorf("customers not found in namespace %s: %v", namespace, strings.Join(notFoundCustomerIDs, ", ")),
-     )
-   }
- }
+ // Verify all requested IDs were found
+ if len(customerList.Items) > 0 {
+   found := make(map[string]struct{}, len(customerList.Items))
+   for _, it := range customerList.Items {
+     found[it.ID] = struct{}{}
+   }
+   var missing []string
+   for _, id := range customerIDs {
+     if _, ok := found[id]; !ok {
+       missing = append(missing, id)
+     }
+   }
+   if len(missing) > 0 {
+     return nil, models.NewGenericValidationError(
+       fmt.Errorf("customers not found in namespace %s: %s", namespace, strings.Join(missing, ", ")),
+     )
+   }
+ }

openmeter/streaming/clickhouse/meter_query_test.go (1)

28-48: Add tests for empty-mapping edges

• Customer with no SubjectKeys → expect WHERE 1 = 0 (or no rows).
• Customers slice provided but empty → expect no rows (adapter-level contract).
• Duplicate subject keys across customers → ensure CASE and IN don’t duplicate args or break grouping.

I can add table-driven cases mirroring these scenarios.

Also applies to: 236-256

openmeter/streaming/clickhouse/event_query_v2.go (1)

101-111: Cursor predicate: prefer the canonical form for readability

Current form is correct but reads oddly. Canonical “seek” predicate for DESC(time), DESC(id) is:
time < cursorTime OR (time = cursorTime AND id < cursorID)

- query.Where(
-   query.LessEqualThan(timeColumn, q.Params.Cursor.Time.Unix()),
-   query.Or(
-     query.LessThan(timeColumn, q.Params.Cursor.Time.Unix()),
-     query.LessThan("id", q.Params.Cursor.ID),
-   ),
- )
+ query.Where(
+   query.Or(
+     query.LessThan(timeColumn, q.Params.Cursor.Time.Unix()),
+     query.And(
+       query.Equal(timeColumn, q.Params.Cursor.Time.Unix()),
+       query.LessThan("id", q.Params.Cursor.ID),
+     ),
+   ),
+ )

openmeter/streaming/clickhouse/event_query_v2_test.go (1)

90-101: Cursor time arg uses Unix seconds while other time filters pass time.Time

Here the cursor binds cursorTime.Unix(), elsewhere time filters bind time.Time. This can be fine if the builder consistently casts on the SQL side, but it’s easy to drift. Consider normalizing the bound type (either always Unix or always time.Time) to reduce surprises.

Double-check toSQL(): if it expects int64 for cursor comparisons, keep it consistent and document it; otherwise bind time.Time for cursor as well.