From 22d4467770260234120ac78376c4470c7fb0867f Mon Sep 17 00:00:00 2001 From: John | Elite Encoder Date: Fri, 17 Jul 2026 13:39:45 -0400 Subject: [PATCH] feat(collector): app-owner vs M2M customer identity Port the pymthouse collector identity contract: strip owner: wire subjects to bare {users.id} CloudEvent subjects, keep M2M compound keys unchanged, and teach bootstrap to provision shared owner customers. --- README.md | 34 ++++++++-- openmeter-collector/collector.yaml | 41 ++++++++---- openmeter-collector/identity/customer_key.go | 56 ++++++++++++++++ .../identity/customer_key_test.go | 57 ++++++++++++++++ openmeter-collector/identity/go.mod | 3 + openmeter-collector/provision/README.md | 29 +++++--- openmeter-collector/provision/bootstrap.ps1 | 52 +++++++++++---- openmeter-collector/provision/bootstrap.sh | 66 ++++++++++++++----- 8 files changed, 278 insertions(+), 60 deletions(-) create mode 100644 openmeter-collector/identity/customer_key.go create mode 100644 openmeter-collector/identity/customer_key_test.go create mode 100644 openmeter-collector/identity/go.mod diff --git a/README.md b/README.md index 6c2a1a8..ee9895f 100644 --- a/README.md +++ b/README.md @@ -223,17 +223,19 @@ 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 { @@ -241,8 +243,26 @@ Example egress event for `auth_id = 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" } } ``` diff --git a/openmeter-collector/collector.yaml b/openmeter-collector/collector.yaml index 17dfdec..7159e20 100644 --- a/openmeter-collector/collector.yaml +++ b/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(), } } } diff --git a/openmeter-collector/identity/customer_key.go b/openmeter-collector/identity/customer_key.go new file mode 100644 index 0000000..b8bcb4f --- /dev/null +++ b/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) +} diff --git a/openmeter-collector/identity/customer_key_test.go b/openmeter-collector/identity/customer_key_test.go new file mode 100644 index 0000000..dd8b3be --- /dev/null +++ b/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) + } + } +} diff --git a/openmeter-collector/identity/go.mod b/openmeter-collector/identity/go.mod new file mode 100644 index 0000000..cb51a4d --- /dev/null +++ b/openmeter-collector/identity/go.mod @@ -0,0 +1,3 @@ +module github.com/livepeer/clearinghouse/openmeter-collector/identity + +go 1.22 diff --git a/openmeter-collector/provision/README.md b/openmeter-collector/provision/README.md index 3fb0483..eae1397 100644 --- a/openmeter-collector/provision/README.md +++ b/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 diff --git a/openmeter-collector/provision/bootstrap.ps1 b/openmeter-collector/provision/bootstrap.ps1 index 9ed4639..bacdf10 100644 --- a/openmeter-collector/provision/bootstrap.ps1 +++ b/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 ' } - # 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 ' } + if ($userId.StartsWith('owner:')) { $userId = $userId.Substring('owner:'.Length) } + if (-not $userId) { Die 'owner requires a non-empty ' } + Ensure-CustomerKey $userId $display $subscribe "owner:$userId" +} + +function Ensure-Customer($clientId, $externalUserId, $display, $subscribe) { + if (-not $clientId -or -not $externalUserId) { Die 'customer requires ' } + # 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 ["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.' diff --git a/openmeter-collector/provision/bootstrap.sh b/openmeter-collector/provision/bootstrap.sh index 6bcae6a..a4cdfdf 100755 --- a/openmeter-collector/provision/bootstrap.sh +++ b/openmeter-collector/provision/bootstrap.sh @@ -17,8 +17,14 @@ # ./bootstrap.sh customer [display_name] [--subscribe] # Ensure an OpenMeter customer keyed : exists with # subject_keys = [:] (matches the CloudEvent subject). +# M2M / managed users only. If external_user_id starts with owner:, delegates to +# the owner command (bare {users.id} customer key). # --subscribe also ensures a subscription on the catalog plan (best-effort). # +# ./bootstrap.sh owner [display_name] [--subscribe] +# Ensure a shared app-owner customer keyed by bare {users.id} (CloudEvent subject +# after the collector strips the owner: wire prefix from auth_id). +# # ./bootstrap.sh all [display_name] [--subscribe] # catalog + customer in one run. # @@ -304,35 +310,58 @@ find_customer() { kapi_get "/customers" | jq -c --arg k "$1" '(.data // .)[] | select(.key == $k)' | head -n 1 } -ensure_customer() { - local client_id="$1" external_user_id="$2" display="$3" subscribe="$4" - [ -n "$client_id" ] && [ -n "$external_user_id" ] || die "customer requires " - # The CloudEvent subject is the compound client_id:external_user_id (globally - # unique, tenant-scoped). It is also the customer key and its only subject_key. - # OpenMeter forbids changing subject_keys once a customer has an active - # subscription, so we set it correctly at creation and never mutate it. - local compound="$client_id:$external_user_id" - [ -n "$display" ] || display="$compound" +# Create/ensure a customer with exact key + subject_keys = [key]. Never mutates +# subject_keys on existing records (OpenMeter forbids changes once subscribed). +ensure_customer_key() { + local key="$1" display="$2" subscribe="$3" label="${4:-$1}" + [ -n "$key" ] || die "customer key is required" + [ -n "$display" ] || display="$key" - local cust; cust="$(find_customer "$compound")" + local cust; cust="$(find_customer "$key")" local id if [ -z "$cust" ]; then local body - body="$(jq -n --arg key "$compound" --arg name "$display" \ + body="$(jq -n --arg key "$key" --arg name "$display" \ '{key:$key, name:$name, usage_attribution:{subject_keys:[$key]}}')" id="$(printf '%s' "$body" | kapi_post /customers | jq -r '.id')" - info "customer $compound — created (subject: $compound)" + info "customer $label — created (subject: $key)" else id="$(jq -r '.id' <<<"$cust")" - if jq -e --arg c "$compound" '(.usage_attribution.subject_keys // []) | index($c)' <<<"$cust" >/dev/null; then - info "customer $compound — up to date" + if jq -e --arg c "$key" '(.usage_attribution.subject_keys // []) | index($c)' <<<"$cust" >/dev/null; then + 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)" fi fi - [ "$subscribe" = "1" ] && ensure_subscription "$id" "$compound" || true + [ "$subscribe" = "1" ] && ensure_subscription "$id" "$key" || true +} + +ensure_owner_customer() { + local user_id="$1" display="$2" subscribe="$3" + [ -n "$user_id" ] || die "owner requires " + # Strip accidental owner: prefix so the key is always the bare {users.id}. + case "$user_id" in + owner:*) user_id="${user_id#owner:}" ;; + esac + [ -n "$user_id" ] || die "owner requires a non-empty " + ensure_customer_key "$user_id" "$display" "$subscribe" "owner:$user_id" +} + +ensure_customer() { + local client_id="$1" external_user_id="$2" display="$3" subscribe="$4" + [ -n "$client_id" ] && [ -n "$external_user_id" ] || die "customer requires " + # App-owner wire subjects (owner:{users.id}) share one bare-id customer. + case "$external_user_id" in + owner:*) + ensure_owner_customer "$external_user_id" "$display" "$subscribe" + return + ;; + esac + # M2M / managed user: CloudEvent subject = compound client_id:external_user_id. + local compound="$client_id:$external_user_id" + ensure_customer_key "$compound" "$display" "$subscribe" "$compound" } # Best-effort subscription on the catalog plan. Skips if the customer already has one. @@ -377,12 +406,15 @@ case "$cmd" in customer) ensure_customer "${1:-}" "${2:-}" "${3:-}" "$SUBSCRIBE" ;; + owner) + ensure_owner_customer "${1:-}" "${2:-}" "$SUBSCRIBE" + ;; all) cmd_catalog ensure_customer "${1:-}" "${2:-}" "${3:-}" "$SUBSCRIBE" ;; *) - die "unknown command '$cmd' (expected: catalog | customer | all)" + die "unknown command '$cmd' (expected: catalog | customer | owner | all)" ;; esac info "done."