feat(collector): app-owner vs M2M customer identity

README.md

@@ -223,26 +223,46 @@ Three layers — each owns a different piece of the identity story:
 | Layer | What it defines | Role of `client_id:usage_subject` |
 | --- | --- | --- |
 | **go-livepeer** | Remote-signer webhook wire protocol ([PR #3897](https://github.com/livepeer/go-livepeer/pull/3897)): webhook returns an opaque `auth_id` string; signer stores it in payment state and copies it into Kafka `create_signed_ticket` events. | go-livepeer treats `auth_id` as an opaque string — it does not parse tenant vs end user. |
-| **Clearinghouse** | Multi-tenant usage identity: `client_id` = tenant (developer app), `usage_subject` = end user within that tenant. The identity webhook joins them as `auth_id = "{client_id}:{usage_subject}"` ([`protocol.mjs`](identity-webhook/protocol.mjs)). | This compound format is a clearinghouse convention so one shared signer can attribute usage across many platform apps and their users. |
-| **OpenMeter / Konnect** | CloudEvent `subject` is the customer attribution key; each customer has `subject_keys` that must match incoming events exactly. | Bootstrap provisions customers keyed by the same compound id (e.g. `demo-client:demo-user`). The collector sets CloudEvent `subject = auth_id` so usage lands on the right customer subscription. |
+| **Clearinghouse** | Multi-tenant usage identity: `client_id` = tenant (developer app), `usage_subject` = end user within that tenant. The identity webhook joins them as `auth_id = "{client_id}:{usage_subject}"` ([`protocol.mjs`](identity-webhook/protocol.mjs)). App owners use wire `usage_subject = owner:{users.id}` as a transport marker. | This compound format is a clearinghouse convention so one shared signer can attribute usage across many platform apps and their users. |
+| **OpenMeter / Konnect** | CloudEvent `subject` is the customer attribution key; each customer has `subject_keys` that must match incoming events exactly. | M2M customers are keyed by compound id (e.g. `demo-client:demo-user`). App owners share one customer keyed by bare `{users.id}`. The collector sets CloudEvent `subject` accordingly. |
 
 Upstream Kafka events carry `auth_id` unchanged (`webhook → go-livepeer state → Kafka`). The collector parses it once (first-colon split) and emits normalized CloudEvents ([`collector.yaml`](openmeter-collector/collector.yaml)):
 
-- `subject` = compound `auth_id` (`client_id:usage_subject`) — **must** match the OpenMeter customer `subject_key`; a bare `usage_subject` will not attribute
+- **M2M / managed user:** `subject` = compound `auth_id` (`client_id:usage_subject`)
+- **App owner:** wire `auth_id = client_id:owner:{users.id}` → `subject` = bare `{users.id}` (shared wallet); `usage_subject_type = app_owner`
 - `data.client_id` = tenant (parsed from `auth_id`)
-- `data.usage_subject` = end user (parsed from `auth_id`)
-- `data.auth_id` retained for compatibility; `data.external_user_id` mirrors `usage_subject` for meter `groupBy`
+- `data.usage_subject` / `data.external_user_id` = end user id, or bare `{users.id}` for owners
+- `data.auth_id` retained for compatibility; `data.openmeter_customer_key` = billing wallet key
+- `data.eth_usd_price` = ETH/USD oracle rate used for that event's Wei → USD micros conversion
 
-Example egress event for `auth_id = demo-client:demo-user`:
+Example egress event for M2M `auth_id = demo-client:demo-user`:
 
 ```json
 {
   "subject": "demo-client:demo-user",
   "data": {
     "client_id": "demo-client",
     "usage_subject": "demo-user",
+    "usage_subject_type": "external_user_id",
     "external_user_id": "demo-user",
-    "auth_id": "demo-client:demo-user"
+    "auth_id": "demo-client:demo-user",
+    "openmeter_customer_key": "demo-client:demo-user"
+  }
+}
+```
+
+Example egress event for app owner `auth_id = app_abc:owner:2e51154b-d296-4015-990c-02d5f16ecf1e`:
+
+```json
+{
+  "subject": "2e51154b-d296-4015-990c-02d5f16ecf1e",
+  "data": {
+    "client_id": "app_abc",
+    "usage_subject": "2e51154b-d296-4015-990c-02d5f16ecf1e",
+    "usage_subject_type": "app_owner",
+    "external_user_id": "2e51154b-d296-4015-990c-02d5f16ecf1e",
+    "auth_id": "app_abc:owner:2e51154b-d296-4015-990c-02d5f16ecf1e",
+    "openmeter_customer_key": "2e51154b-d296-4015-990c-02d5f16ecf1e"
   }
 }
 ```

openmeter-collector/collector.yaml

@@ -4,7 +4,8 @@
 # sequence blocks until
 # warm succeeds; collector never reads Kafka until cache is populated). Refreshed on
 # PRICE_ORACLE_REFRESH; refresh errors keep the last cached value. Runtime events read
-# from cache only (no static default).
+# from cache only (no static default). Each egress CloudEvent includes data.eth_usd_price
+# (the cached rate used for that event's Wei → USD micros conversion).
 #
 # Oracle JSON (any one shape):
 #   {"price": 3456.78}
@@ -16,13 +17,12 @@
 #
 # Identity (go-livepeer Kafka wire -> normalized CloudEvent egress):
 # - Upstream auth_id stays client_id:usage_subject (webhook -> go-livepeer state -> Kafka).
-# - Collector parses auth_id once (first-colon split), then emits:
-#   subject = auth_id (compound client_id:usage_subject) so it matches the OpenMeter
-#   customer subject_key; data.client_id = tenant, data.usage_subject = end user.
-# - subject MUST be compound: OpenMeter attributes usage by exact subject match and
-#   forbids changing a customer's subject_keys once it has an active subscription, so a
-#   bare usage_subject would never attribute. Group/break down by data.* dimensions instead.
-# - data.auth_id and data.external_user_id kept for compatibility (meters group by external_user_id).
+# - For app owners, webhook maps JWT sub (bare user id) to usage_subject owner:{users.id}
+#   so auth_id is app_…:owner:{id}. The collector strips the owner: prefix so CloudEvent
+#   subject / openmeter_customer_key = bare {users.id} (canonical shared Konnect customer).
+#   data.client_id retains the app id; data.usage_subject / external_user_id are bare ids.
+# - M2M / managed end-users keep CloudEvent subject = full compound auth_id.
+# - data.client_id / data.usage_subject always retain the parsed tenant + subject.
 
 cache_resources:
   - label: eth_usd
@@ -122,13 +122,26 @@ pipeline:
         let is_compound = $colon > 0 && $colon < $auth_id.length() - 1
         let client_id = if $is_compound { $auth_id.slice(0, $colon) } else { $auth_id }
         let usage_subject_parsed = if $is_compound { $auth_id.slice($colon + 1) } else { $auth_id }
-        let usage_subject_type = $data.usage_subject_type.or("external_user_id").string().trim()
+        # Owner wallet: wire usage_subject is owner:{users.id}; CE subject = bare id.
+        # M2M end-users: meter subject = full compound auth_id.
+        let is_owner_subject = $usage_subject_parsed.index_of("owner:") == 0
+        let usage_subject_type = if $is_owner_subject { "app_owner" } else { "external_user_id" }
+        let owner_bare_id = if $is_owner_subject {
+          $usage_subject_parsed.slice("owner:".length())
+        } else {
+          $usage_subject_parsed
+        }
+        let openmeter_subject = if $is_owner_subject { $owner_bare_id } else { $auth_id }
+        let openmeter_customer_key = $openmeter_subject
+        let usage_subject_out = if $is_owner_subject { $owner_bare_id } else { $usage_subject_parsed }
 
         let eth_usd = meta("eth_usd_price").number()
         let _ = if $eth_usd == null || $eth_usd <= 0 { throw("eth_usd price cache miss or non-positive") }
 
+        # USD micros = fee_wei * eth_usd / 1e12 (1 USD = 1e6 micros).
+        # Ceil so sub-micro live-runner fees still count as at least 1 micro.
         let fee_wei = $data.computed_fee.number().or(0)
-        let fee_usd_micros = ($fee_wei * $eth_usd / 1000000000000).round()
+        let fee_usd_micros = ($fee_wei * $eth_usd / 1000000000000).ceil()
 
         root = if $auth_id == "" {
           deleted()
@@ -138,13 +151,13 @@ pipeline:
             "type": "create_signed_ticket",
             "id": $data.request_id.or(uuid_v4()),
             "source": "go-livepeer-remote-signer",
-            "subject": $auth_id,
+            "subject": $openmeter_subject,
             "time": $data.current_time.or(now()),
             "data": {
               "client_id": $client_id,
-              "usage_subject": $usage_subject_parsed,
+              "usage_subject": $usage_subject_out,
               "usage_subject_type": $usage_subject_type,
-              "external_user_id": $usage_subject_parsed,
+              "external_user_id": $usage_subject_out,
               "network_fee_usd_micros": $fee_usd_micros,
               # Interim passthrough: billable == network fee until phase-2 markup
               # rules are applied. Present so the billable_usd_micros meter validates.
@@ -155,6 +168,8 @@ pipeline:
               "fee_wei": $data.computed_fee.or("0"),
               "gateway_request_id": $data.request_id,
               "auth_id": $auth_id,
+              "openmeter_customer_key": $openmeter_customer_key,
+              "eth_usd_price": $eth_usd.string(),
             }
           }
         }

openmeter-collector/identity/customer_key.go

@@ -0,0 +1,56 @@
+// Package identity defines the OpenMeter customer / CloudEvent subject key contract
+// shared by the collector mapping and bootstrap provisioning.
+package identity
+
+import "strings"
+
+// OwnerWirePrefix is the transport marker webhook → go-livepeer stamps on
+// app-owner usage subjects (auth_id = app_…:owner:{users.id}). The collector
+// strips this before writing the CloudEvent subject; Konnect customer keys use
+// the bare platform user id.
+const OwnerWirePrefix = "owner:"
+
+// CustomerKey returns the deterministic OpenMeter customer / CloudEvent subject key.
+//
+// M2M / managed users: compound clientID:externalUserID (unchanged).
+// App owners: bare {users.id} when externalUserID is the owner: wire subject.
+func CustomerKey(clientID, externalUserID string) string {
+	clientID = strings.TrimSpace(clientID)
+	externalUserID = strings.TrimSpace(externalUserID)
+	if ownerID, ok := ParseOwnerWireSubject(externalUserID); ok {
+		return ownerID
+	}
+	return clientID + ":" + externalUserID
+}
+
+// ParseOwnerWireSubject returns the bare user id when s is owner:{id}.
+func ParseOwnerWireSubject(s string) (string, bool) {
+	s = strings.TrimSpace(s)
+	if !strings.HasPrefix(s, OwnerWirePrefix) {
+		return "", false
+	}
+	id := strings.TrimPrefix(s, OwnerWirePrefix)
+	if id == "" {
+		return "", false
+	}
+	return id, true
+}
+
+// IsOwnerWireSubject reports whether s is the owner:{id} transport marker.
+func IsOwnerWireSubject(s string) bool {
+	_, ok := ParseOwnerWireSubject(s)
+	return ok
+}
+
+// CloudEventSubject is the OpenMeter attribution subject for a Kafka auth_id.
+// Equivalent to CustomerKey after splitting auth_id on the first colon.
+func CloudEventSubject(authID string) string {
+	authID = strings.TrimSpace(authID)
+	colon := strings.Index(authID, ":")
+	if colon <= 0 || colon >= len(authID)-1 {
+		return authID
+	}
+	clientID := authID[:colon]
+	usageSubject := authID[colon+1:]
+	return CustomerKey(clientID, usageSubject)
+}

openmeter-collector/identity/customer_key_test.go

@@ -0,0 +1,57 @@
+package identity_test
+
+import (
+	"testing"
+
+	"github.com/livepeer/clearinghouse/openmeter-collector/identity"
+)
+
+func TestCustomerKeyM2MUnchanged(t *testing.T) {
+	t.Parallel()
+	got := identity.CustomerKey(" pub-client ", " demo-user ")
+	if got != "pub-client:demo-user" {
+		t.Fatalf("CustomerKey() = %q", got)
+	}
+}
+
+func TestCustomerKeyOwnerWireSubject(t *testing.T) {
+	t.Parallel()
+	ownerID := "2e51154b-d296-4015-990c-02d5f16ecf1e"
+	got := identity.CustomerKey("app_abc", "owner:"+ownerID)
+	if got != ownerID {
+		t.Fatalf("owner CustomerKey() = %q, want bare id %q", got, ownerID)
+	}
+}
+
+func TestParseOwnerWireSubject(t *testing.T) {
+	t.Parallel()
+	id, ok := identity.ParseOwnerWireSubject("owner:uuid-1")
+	if !ok || id != "uuid-1" {
+		t.Fatalf("ParseOwnerWireSubject = (%q, %v)", id, ok)
+	}
+	if identity.IsOwnerWireSubject("app_abc:user-1") {
+		t.Fatal("compound key should not be owner wire subject")
+	}
+	if _, ok := identity.ParseOwnerWireSubject("owner:"); ok {
+		t.Fatal("empty owner id should not parse")
+	}
+}
+
+func TestCloudEventSubject(t *testing.T) {
+	t.Parallel()
+	ownerID := "2e51154b-d296-4015-990c-02d5f16ecf1e"
+
+	cases := []struct {
+		authID string
+		want   string
+	}{
+		{"demo-client:demo-user", "demo-client:demo-user"},
+		{"app_abc:owner:" + ownerID, ownerID},
+		{"bare-subject", "bare-subject"},
+	}
+	for _, tc := range cases {
+		if got := identity.CloudEventSubject(tc.authID); got != tc.want {
+			t.Fatalf("CloudEventSubject(%q) = %q, want %q", tc.authID, got, tc.want)
+		}
+	}
+}

openmeter-collector/identity/go.mod

@@ -0,0 +1,3 @@
+module github.com/livepeer/clearinghouse/openmeter-collector/identity
+
+go 1.22

openmeter-collector/provision/README.md

@@ -45,9 +45,12 @@ cd openmeter-collector/provision
 # Catalog only — ensure meters, features, and the active plan.
 ./bootstrap.sh catalog
 
-# Provision one tenant customer (key = client_id:external_user_id).
+# Provision one M2M / managed-user customer (key = client_id:external_user_id).
 ./bootstrap.sh customer demo-client demo-user "Demo User"
 
+# Provision a shared app-owner customer (key = bare {users.id}).
+./bootstrap.sh owner 2e51154b-d296-4015-990c-02d5f16ecf1e "App Owner"
+
 # Catalog + customer in one run; --subscribe also ensures a plan subscription.
 ./bootstrap.sh all demo-client demo-user "Demo User" --subscribe
 ```
@@ -57,6 +60,7 @@ Windows (PowerShell):
 ```powershell
 .\bootstrap.ps1 catalog
 .\bootstrap.ps1 customer demo-client demo-user "Demo User"
+.\bootstrap.ps1 owner 2e51154b-d296-4015-990c-02d5f16ecf1e "App Owner"
 .\bootstrap.ps1 all demo-client demo-user "Demo User" -Subscribe
 ```
 
@@ -78,14 +82,21 @@ From [`catalog.json`](catalog.json):
 
 ## Identity contract (important)
 
-The CloudEvent **`subject` is the compound `client_id:external_user_id`** (e.g.
-`demo-client:demo-user`), which is also the customer key and its single `subject_key`.
-OpenMeter attributes usage by exact subject match, and **forbids changing a customer's
-`subject_keys` once it has an active subscription** — so the subject must be compound and
-correct from creation. Break usage down per-tenant/user with the `client_id` / `external_user_id`
-meter dimensions, not by changing the subject. The scripts therefore never mutate
-`subject_keys` on existing customers; they warn if an existing customer is missing the
-expected compound key.
+OpenMeter attributes usage by exact CloudEvent `subject` match, and **forbids changing a
+customer's `subject_keys` once it has an active subscription** — so the key must be
+correct from creation. Two customer shapes:
+
+| Caller | Wire `auth_id` | CloudEvent `subject` / customer key |
+| --- | --- | --- |
+| **M2M / managed user** | `client_id:external_user_id` | compound `client_id:external_user_id` |
+| **App owner** | `client_id:owner:{users.id}` | bare `{users.id}` (shared wallet) |
+
+The webhook stamps `owner:` as a **transport marker**; the collector strips it before
+egress. Break usage down per-tenant/user with the `client_id` / `external_user_id` meter
+dimensions, not by changing the subject. The scripts never mutate `subject_keys` on
+existing customers; they warn if an existing customer is missing the expected key.
+
+See also [`../identity`](../identity) (Go helpers + unit tests for this contract).
 
 ## Limitations
 

openmeter-collector/provision/bootstrap.ps1

@@ -14,12 +14,14 @@
   .\bootstrap.ps1 catalog
 .EXAMPLE
   .\bootstrap.ps1 customer demo-client demo-user "Demo User"
+.EXAMPLE
+  .\bootstrap.ps1 owner 2e51154b-d296-4015-990c-02d5f16ecf1e "App Owner"
 .EXAMPLE
   .\bootstrap.ps1 all demo-client demo-user "Demo User" -Subscribe
 #>
 [CmdletBinding()]
 param(
-  [ValidateSet('catalog', 'customer', 'all')]
+  [ValidateSet('catalog', 'customer', 'owner', 'all')]
   [string]$Command = 'catalog',
   [string]$ClientId,
   [string]$ExternalUserId,
@@ -256,30 +258,46 @@ function Invoke-Catalog {
 }
 
 # --- customer --------------------------------------------------------------
-function Ensure-Customer($clientId, $externalUserId, $display, $subscribe) {
-  if (-not $clientId -or -not $externalUserId) { Die 'customer requires <client_id> <external_user_id>' }
-  # CloudEvent subject = compound client_id:external_user_id = customer key = its only
-  # subject_key. OpenMeter forbids changing subject_keys on subscribed customers, so we
-  # set it at creation and never mutate it.
-  $compound = "$clientId`:$externalUserId"
-  if (-not $display) { $display = $compound }
+function Ensure-CustomerKey($key, $display, $subscribe, $label) {
+  if (-not $key) { Die 'customer key is required' }
+  if (-not $display) { $display = $key }
+  if (-not $label) { $label = $key }
 
-  $cust = Items (Kapi-Get '/customers') | Where-Object { $_.key -eq $compound } | Select-Object -First 1
+  $cust = Items (Kapi-Get '/customers') | Where-Object { $_.key -eq $key } | Select-Object -First 1
   if (-not $cust) {
-    $body = [ordered]@{ key = $compound; name = $display; usage_attribution = @{ subject_keys = @($compound) } }
+    $body = [ordered]@{ key = $key; name = $display; usage_attribution = @{ subject_keys = @($key) } }
     $created = Kapi-Send 'post' '/customers' ($body | ConvertTo-Json -Depth 5 -Compress)
     $id = $created.id
-    Info "customer $compound - created (subject: $compound)"
+    Info "customer $label - created (subject: $key)"
   }
   else {
     $id = $cust.id
-    if ($cust.usage_attribution.subject_keys -contains $compound) { Info "customer $compound - up to date" }
+    if ($cust.usage_attribution.subject_keys -contains $key) { Info "customer $label - up to date" }
     else {
-      Warn "customer $compound exists but its subject_keys do not include '$compound'"
+      Warn "customer $label exists but its subject_keys do not include '$key'"
       Warn "  (OpenMeter blocks subject_key changes on subscribed customers - reconcile manually)"
     }
   }
-  if ($subscribe) { Ensure-Subscription $id $compound }
+  if ($subscribe) { Ensure-Subscription $id $key }
+}
+
+function Ensure-OwnerCustomer($userId, $display, $subscribe) {
+  if (-not $userId) { Die 'owner requires <user_id>' }
+  if ($userId.StartsWith('owner:')) { $userId = $userId.Substring('owner:'.Length) }
+  if (-not $userId) { Die 'owner requires a non-empty <user_id>' }
+  Ensure-CustomerKey $userId $display $subscribe "owner:$userId"
+}
+
+function Ensure-Customer($clientId, $externalUserId, $display, $subscribe) {
+  if (-not $clientId -or -not $externalUserId) { Die 'customer requires <client_id> <external_user_id>' }
+  # App-owner wire subjects share one bare-{users.id} customer.
+  if ($externalUserId.StartsWith('owner:')) {
+    Ensure-OwnerCustomer $externalUserId $display $subscribe
+    return
+  }
+  # M2M / managed user: CloudEvent subject = compound client_id:external_user_id.
+  $compound = "$clientId`:$externalUserId"
+  Ensure-CustomerKey $compound $display $subscribe $compound
 }
 
 function Ensure-Subscription($customerId, $label) {
@@ -304,6 +322,12 @@ function Ensure-Subscription($customerId, $label) {
 switch ($Command) {
   'catalog' { Invoke-Catalog }
   'customer' { Ensure-Customer $ClientId $ExternalUserId $DisplayName $Subscribe.IsPresent }
+  'owner' {
+    # Positional: .\bootstrap.ps1 owner <user_id> ["Display Name"]
+    $uid = $ClientId
+    $display = if ($DisplayName) { $DisplayName } elseif ($ExternalUserId) { $ExternalUserId } else { $uid }
+    Ensure-OwnerCustomer $uid $display $Subscribe.IsPresent
+  }
   'all' { Invoke-Catalog; Ensure-Customer $ClientId $ExternalUserId $DisplayName $Subscribe.IsPresent }
 }
 Info 'done.'