feat(collector): implement customer upsert functionality in OpenMeter collector

.gitignore

@@ -1,4 +1,10 @@
 kafka/.env
 remote-signer/.env
 openmeter-collector/.env
+identity-webhook/.env
 remote-signer/data/
+node_modules/
+
+# Editor / OS
+.vscode/
+.DS_Store

README.md

@@ -1,12 +1,13 @@
 # clearinghouse
 
 Docker Compose stack for the clearinghouse runtime:
-**Redpanda → go-livepeer remote signer → OpenMeter/Benthos collector → Konnect metering**.
+**identity-webhook → Redpanda → go-livepeer remote signer → OpenMeter/Benthos collector → Konnect metering**.
 
 ## Components
 
 | Service | Role | Docs |
 | --- | --- | --- |
+| **identity-webhook** (`identity-webhook`) | Resolves end-user API keys to `auth_id` for go-livepeer's `/authorize` hook. Uses builder-sdk's API-key provider. | [builder-sdk](https://github.com/pymthouse/builder-sdk) |
 | **Redpanda** (`kafka`) | Kafka-compatible event bus. The signer publishes gateway events; the collector consumes them. | [Redpanda docs](https://docs.redpanda.com/) |
 | **go-livepeer remote signer** (`remote-signer`) | Signs Livepeer payment tickets and emits `create_signed_ticket` events to Kafka. | [go-livepeer](https://github.com/livepeer/go-livepeer) |
 | **OpenMeter collector** (`openmeter-collector`) | Benthos pipeline: filters Kafka events, converts fees to USD micros, POSTs CloudEvents to OpenMeter ingest. | [OpenMeter collector](https://openmeter.io/docs/collectors) |
@@ -26,30 +27,43 @@ Signer HTTP request
 
 **Redpanda over Apache Kafka.** The stack uses Redpanda as the Kafka-compatible broker. Redpanda runs as a single-binary dev container with no ZooKeeper dependency and faster local startup.
 
-**Identity & auth.** The signer container runs `go-livepeer` directly. In the normal path, every signing request is authorized by go-livepeer's `-remoteSignerWebhookUrl` hook, which calls your `/authorize` endpoint with `Authorization: Bearer <WEBHOOK_SECRET>` — no reverse proxy or gateway in front of the signer. For local alive checks only, leave `REMOTE_SIGNER_WEBHOOK_URL` empty to omit the webhook hook.
+**Identity & auth.** The in-compose **identity-webhook** uses builder-sdk's API-key provider. The signer container runs `go-livepeer` directly; every signing request is authorized by go-livepeer's `-remoteSignerWebhookUrl` hook, which calls `/authorize` with `Authorization: Bearer <WEBHOOK_SECRET>`. End users present `Authorization: Bearer sk_…` to the signer; the webhook resolves the key to `auth_id = "{client_id}:{usage_subject}"`. For local alive checks only, leave `REMOTE_SIGNER_WEBHOOK_URL` empty to omit the webhook hook.
 
 **CLI port not exposed.** go-livepeer's `-cliAddr` (admin/RPC) is bound to `127.0.0.1:4935` inside the container and is never published or mapped to the host. Only the signing HTTP port (`8081`) is exposed.
 
-**Per-service configuration.** Each service reads a local `.env` file mounted at `/service/.env` and sourced by its entrypoint. Copy the `.env.example` in each service directory before starting the stack.
+**Per-service configuration.** Each service has a local `.env` file (copy from `.env.example` before starting). Kafka, remote-signer, and openmeter-collector mount theirs at `/service/.env` and source it in the entrypoint. identity-webhook reads its `.env` via Compose `env_file`.
 
 ## Local stack
 
-### 1. Quick check — Kafka + signer
+### 1. Quick check — Kafka + identity webhook + signer
 
-Start here before wiring identity or metering. This runs only the Kafka broker and remote signer so you can confirm the core path is alive.
+Start here before wiring metering. This runs the broker, identity webhook, and remote signer.
 
 ```bash
 cp kafka/.env.example kafka/.env
+cp identity-webhook/.env.example identity-webhook/.env
 cp remote-signer/.env.example remote-signer/.env
-$EDITOR remote-signer/.env
+$EDITOR identity-webhook/.env remote-signer/.env
+# WEBHOOK_SECRET must match in both files (`.env.example` ships a local dev value).
 # For a local alive check without an identity webhook:
 #   REMOTE_SIGNER_WEBHOOK_URL=
 #   WEBHOOK_SECRET=
 
-docker compose up -d --build kafka remote-signer
+docker compose up -d --build kafka identity-webhook remote-signer
 docker compose logs -f remote-signer
 ```
 
+Verify the identity webhook (simulates go-livepeer calling `/authorize`; secret matches `.env.example`):
+
+```bash
+docker compose exec identity-webhook \
+  curl -sS -X POST http://localhost:8090/authorize \
+    -H "Authorization: Bearer dev-webhook-secret-change-me" \
+    -H "Content-Type: application/json" \
+    -d '{"headers":{"Authorization":["Bearer sk_demo_local_key"]}}'
+# expected: "status":200, "auth_id":"demo-client:demo-user"
+```
+
 Expected result: `remote-signer` starts cleanly, connects to Kafka, and serves the signing HTTP port.
 
 Verify CLI port is not published:
@@ -80,9 +94,10 @@ Each service documents its variables in its own `.env.example`:
 
 | Service | Config file | Key variables |
 | --- | --- | --- |
+| `identity-webhook` | [`identity-webhook/.env.example`](identity-webhook/.env.example) | `WEBHOOK_SECRET`, `IDENTITY_ISSUER`, `DEMO_API_KEY`, `DEMO_CLIENT_ID`, `DEMO_USER_ID`, `API_KEY_PREFIX` (optional, default `sk_`) |
 | `kafka` | [`kafka/.env.example`](kafka/.env.example) | `KAFKA_ADVERTISED_ADDR` |
 | `remote-signer` | [`remote-signer/.env.example`](remote-signer/.env.example) | `REMOTE_SIGNER_WEBHOOK_URL`, `WEBHOOK_SECRET`, `SIGNER_*`, `KAFKA_BROKERS`, `KAFKA_GATEWAY_TOPIC` |
-| `openmeter-collector` | [`openmeter-collector/.env.example`](openmeter-collector/.env.example) | `KAFKA_BROKERS`, `KAFKA_GATEWAY_TOPIC`, `OPENMETER_INGEST_URL`, `OPENMETER_API_KEY`, `ETH_USD_PRICE` |
+| `openmeter-collector` | [`openmeter-collector/.env.example`](openmeter-collector/.env.example) | `KAFKA_BROKERS`, `KAFKA_GATEWAY_TOPIC`, `OPENMETER_URL`, `OPENMETER_INGEST_URL`, `OPENMETER_API_KEY`, `OPENMETER_DEFAULT_PLAN_KEY`, `ETH_USD_PRICE` |
 
 Signer state (keystore, `.eth-password`, chain DB) is stored under [`remote-signer/data/`](remote-signer/data/), bind-mounted to `/data` in the container.
 
@@ -129,10 +144,36 @@ Signer computed_fee (wei)
 
 Markup rules are defined in the bootstrap CLI catalog. Collector
 pipeline config: [`openmeter-collector/collector.yaml`](openmeter-collector/collector.yaml).
-The collector does not yet emit `billable_usd_micros` (phase 2); until then the billable meter
-stays empty while the catalog is ready.
+`billable_usd_micros` is initially set equal to `network_fee_usd_micros`; markup rules will diverge in a later phase.
+
+### Customer upsert (collector self-heal)
+
+The collector runs a local Go provision sidecar (`openmeter-collector/provision`, Kong `sdk-konnect-go`) that holds the
+OpenMeter admin credentials — the identity webhook does **not** need them.
+
+For each `create_signed_ticket` event:
+
+1. Benthos maps the CloudEvent (including `billable_usd_micros`, initially equal to `network_fee_usd_micros`).
+2. `POST http://127.0.0.1:8091/ensure` idempotently creates customer + subscription (`OPENMETER_DEFAULT_PLAN_KEY`).
+3. Event is ingested to Konnect.
+4. On ingest failure (e.g. `no customer found for event subject`), the collector ensures again and retries once.
+
+### Future admin/query boundary (OAuth later)
+
+When an admin/query API is added, introduce a small internal **billing-gateway** service:
+
+- Move ensure-customer and usage-query logic behind that gateway.
+- Protect caller-to-gateway with OAuth (client credentials / service-to-service).
+- Keep gateway-to-OpenMeter on backend machine credentials (`kpat_…`).
+
+The collector provision sidecar is the thin local equivalent until that gateway exists.
 
 ### Identity contract (collector)
 
 The collector expects Kafka `auth_id` as `client_id:external_user_id` (first-colon split).
 Konnect customer key matches that compound id (e.g. `demo-client:demo-user`).
+
+Demo API key defaults: `sk_demo_local_key` → `demo-client:demo-user` (configured in `identity-webhook/.env`).
+
+Customer upsert is handled by the collector provision sidecar (see above). The Go bootstrap CLI
+still provisions meters/features/plans; per-event customer+subscription ensure runs in the collector.

docker-compose.yml

@@ -1,4 +1,4 @@
-# Clearinghouse stack: Redpanda + go-livepeer remote signer + OpenMeter collector.
+# Clearinghouse stack: identity webhook + Redpanda + remote signer + OpenMeter collector.
 #
 # Redpanda is the Kafka-compatible broker. The signer container runs go-livepeer directly —
 # only its signing HTTP port (8081) is published to the host. The CLI/admin port (-cliAddr)
@@ -9,10 +9,26 @@
 # Full stack:
 #   docker compose up -d --build
 #
-# Kafka + signer only (no metering):
-#   docker compose up -d --build kafka remote-signer
+# Kafka + identity webhook + signer only (no metering):
+#   docker compose up -d --build kafka identity-webhook remote-signer
 
 services:
+  identity-webhook:
+    build:
+      context: .
+      dockerfile: identity-webhook/Dockerfile
+    restart: unless-stopped
+    env_file:
+      - ./identity-webhook/.env
+    environment:
+      PORT: "8090"
+    healthcheck:
+      test: ["CMD", "curl", "-fsS", "http://localhost:8090/health"]
+      interval: 10s
+      timeout: 5s
+      retries: 6
+      start_period: 10s
+
   kafka:
     build:
       context: .
@@ -35,6 +51,8 @@ services:
     depends_on:
       kafka:
         condition: service_healthy
+      identity-webhook:
+        condition: service_healthy
     extra_hosts:
       - "host.docker.internal:host-gateway"
     ports:

identity-webhook/.env.example

@@ -0,0 +1,15 @@
+# Identity webhook service env — copied to identity-webhook/.env at runtime.
+#
+#   cp identity-webhook/.env.example identity-webhook/.env
+
+# Local dev only — change before any shared or production use.
+WEBHOOK_SECRET=dev-webhook-secret-change-me
+IDENTITY_ISSUER=http://identity-webhook:8090
+
+# Demo API key resolved by the webhook (must match remote-signer end-user Bearer token).
+DEMO_API_KEY=sk_demo_local_key
+DEMO_CLIENT_ID=demo-client
+DEMO_USER_ID=demo-user
+USAGE_SUBJECT_TYPE=api_key_user
+# API_KEY_PREFIX=sk_
+# DEMO_API_KEYS={"sk_other":{"clientId":"app-b","userId":"user-b"}}

identity-webhook/Dockerfile

@@ -0,0 +1,19 @@
+FROM node:22-alpine
+
+RUN apk add --no-cache curl
+
+WORKDIR /app
+
+COPY identity-webhook/package.json identity-webhook/package-lock.json ./
+RUN npm ci --omit=dev
+
+COPY identity-webhook/keys.mjs identity-webhook/server.mjs ./
+
+ENV PORT=8090
+
+EXPOSE 8090
+
+HEALTHCHECK --interval=10s --timeout=3s --retries=6 --start-period=5s \
+  CMD curl -fsS http://localhost:8090/health || exit 1
+
+CMD ["node", "server.mjs"]

identity-webhook/keys.mjs

@@ -0,0 +1,54 @@
+/**
+ * Load demo API keys from env for the in-compose identity webhook.
+ *
+ * DEMO_API_KEY + DEMO_CLIENT_ID + DEMO_USER_ID define one key.
+ * Optional DEMO_API_KEYS JSON map for multiple keys:
+ *   {"sk_other":{"clientId":"app-b","userId":"user-b"}}
+ */
+export function loadApiKeyStore(env) {
+  const store = new Map();
+
+  const primaryKey = env.DEMO_API_KEY?.trim();
+  if (primaryKey) {
+    store.set(primaryKey, {
+      clientId: env.DEMO_CLIENT_ID?.trim() || "demo-client",
+      userId: env.DEMO_USER_ID?.trim() || "demo-user",
+      usageSubjectType: env.USAGE_SUBJECT_TYPE?.trim() || "api_key_user",
+    });
+  }
+
+  const extra = env.DEMO_API_KEYS?.trim();
+  if (extra) {
+    let parsed;
+    try {
+      parsed = JSON.parse(extra);
+    } catch {
+      throw new Error("DEMO_API_KEYS must be valid JSON");
+    }
+    if (parsed && typeof parsed === "object") {
+      for (const [apiKey, entry] of Object.entries(parsed)) {
+        if (!apiKey?.trim() || !entry || typeof entry !== "object") {
+          continue;
+        }
+        const userId = entry.userId?.trim() || entry.user_id?.trim();
+        if (!userId) {
+          continue;
+        }
+        store.set(apiKey.trim(), {
+          clientId: entry.clientId?.trim() || entry.client_id?.trim() || "demo-client",
+          userId,
+          usageSubjectType:
+            entry.usageSubjectType?.trim() ||
+            entry.usage_subject_type?.trim() ||
+            "api_key_user",
+        });
+      }
+    }
+  }
+
+  if (store.size === 0) {
+    throw new Error("DEMO_API_KEY or DEMO_API_KEYS is required");
+  }
+
+  return store;
+}

identity-webhook/keys.test.mjs

@@ -0,0 +1,62 @@
+import assert from "node:assert/strict";
+import { describe, it } from "node:test";
+import { loadApiKeyStore } from "./keys.mjs";
+
+describe("loadApiKeyStore", () => {
+  it("loads a primary demo API key", () => {
+    const store = loadApiKeyStore({
+      DEMO_API_KEY: "sk_demo_local_key",
+      DEMO_CLIENT_ID: "demo-client",
+      DEMO_USER_ID: "demo-user",
+      USAGE_SUBJECT_TYPE: "api_key_user",
+    });
+
+    assert.equal(store.size, 1);
+    assert.deepEqual(store.get("sk_demo_local_key"), {
+      clientId: "demo-client",
+      userId: "demo-user",
+      usageSubjectType: "api_key_user",
+    });
+  });
+
+  it("loads extra keys from DEMO_API_KEYS with snake_case fields", () => {
+    const store = loadApiKeyStore({
+      DEMO_API_KEYS: JSON.stringify({
+        sk_other: { client_id: "app-b", user_id: "user-b" },
+      }),
+    });
+
+    assert.equal(store.size, 1);
+    assert.deepEqual(store.get("sk_other"), {
+      clientId: "app-b",
+      userId: "user-b",
+      usageSubjectType: "api_key_user",
+    });
+  });
+
+  it("rejects invalid DEMO_API_KEYS JSON", () => {
+    assert.throws(
+      () => loadApiKeyStore({ DEMO_API_KEYS: "not-json" }),
+      /DEMO_API_KEYS must be valid JSON/,
+    );
+  });
+
+  it("requires at least one configured key", () => {
+    assert.throws(
+      () => loadApiKeyStore({}),
+      /DEMO_API_KEY or DEMO_API_KEYS is required/,
+    );
+  });
+
+  it("skips entries missing userId", () => {
+    assert.throws(
+      () =>
+        loadApiKeyStore({
+          DEMO_API_KEYS: JSON.stringify({
+            sk_invalid: { clientId: "app-b" },
+          }),
+        }),
+      /DEMO_API_KEY or DEMO_API_KEYS is required/,
+    );
+  });
+});

identity-webhook/package.json

@@ -0,0 +1,16 @@
+{
+  "name": "@livepeer/clearinghouse-identity-webhook",
+  "version": "0.1.0",
+  "private": true,
+  "type": "module",
+  "description": "API-key identity webhook for go-livepeer remote signer (builder-sdk)",
+  "engines": {
+    "node": ">=20"
+  },
+  "scripts": {
+    "test": "node --test keys.test.mjs"
+  },
+  "dependencies": {
+    "@pymthouse/builder-sdk": "0.4.6"
+  }
+}