diff --git a/.gitignore b/.gitignore index 7501e7f..a192860 100644 --- a/.gitignore +++ b/.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 diff --git a/README.md b/README.md index 5b6e2cf..529a59c 100644 --- a/README.md +++ b/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 ` — 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 `. 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. diff --git a/docker-compose.yml b/docker-compose.yml index e0969fa..1eba0aa 100644 --- a/docker-compose.yml +++ b/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: diff --git a/identity-webhook/.env.example b/identity-webhook/.env.example new file mode 100644 index 0000000..73de441 --- /dev/null +++ b/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"}} diff --git a/identity-webhook/Dockerfile b/identity-webhook/Dockerfile new file mode 100644 index 0000000..91ec93f --- /dev/null +++ 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"] diff --git a/identity-webhook/keys.mjs b/identity-webhook/keys.mjs new file mode 100644 index 0000000..be5e912 --- /dev/null +++ b/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; +} diff --git a/identity-webhook/keys.test.mjs b/identity-webhook/keys.test.mjs new file mode 100644 index 0000000..fd2049b --- /dev/null +++ b/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/, + ); + }); +}); diff --git a/identity-webhook/package-lock.json b/identity-webhook/package-lock.json new file mode 100644 index 0000000..b644eb2 --- /dev/null +++ b/identity-webhook/package-lock.json @@ -0,0 +1,39 @@ +{ + "name": "@livepeer/clearinghouse-identity-webhook", + "version": "0.1.0", + "lockfileVersion": 3, + "requires": true, + "packages": { + "": { + "name": "@livepeer/clearinghouse-identity-webhook", + "version": "0.1.0", + "dependencies": { + "@pymthouse/builder-sdk": "0.4.6" + }, + "engines": { + "node": ">=20" + } + }, + "node_modules/@pymthouse/builder-sdk": { + "version": "0.4.6", + "resolved": "https://registry.npmjs.org/@pymthouse/builder-sdk/-/builder-sdk-0.4.6.tgz", + "integrity": "sha512-o00iwt4n+5mORHldio/NtDmr05YugFbj7x4Wi73IcybDJRfaHbGjGqde9BbdShljPeP5WgHUhREPopLjpJr7Cg==", + "license": "MIT", + "dependencies": { + "oauth4webapi": "^3.8.5" + }, + "engines": { + "node": ">=20" + } + }, + "node_modules/oauth4webapi": { + "version": "3.8.6", + "resolved": "https://registry.npmjs.org/oauth4webapi/-/oauth4webapi-3.8.6.tgz", + "integrity": "sha512-iwemM91xz8nryHti2yTmg5fhyEMVOkOXwHNqbvcATjyajb5oQxCQzrNOA6uElRHuMhQQTKUyFKV9y/CNyg25BQ==", + "license": "MIT", + "funding": { + "url": "https://github.com/sponsors/panva" + } + } + } +} diff --git a/identity-webhook/package.json b/identity-webhook/package.json new file mode 100644 index 0000000..8b24204 --- /dev/null +++ b/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" + } +} diff --git a/identity-webhook/server.mjs b/identity-webhook/server.mjs new file mode 100644 index 0000000..3daaee6 --- /dev/null +++ b/identity-webhook/server.mjs @@ -0,0 +1,95 @@ +import { createServer } from "node:http"; +import { + createApiKeyEndUserVerifier, + routeRemoteSignerWebhookRequest, +} from "@pymthouse/builder-sdk/signer/webhook"; +import { loadApiKeyStore } from "./keys.mjs"; + +const port = Number(process.env.PORT || 8090); + +function required(name) { + const value = process.env[name]?.trim(); + if (!value) { + throw new Error(`${name} is required`); + } + return value; +} + +const keyStore = loadApiKeyStore(process.env); + +const config = { + webhookSecret: required("WEBHOOK_SECRET"), + endUserAuth: createApiKeyEndUserVerifier({ + issuer: required("IDENTITY_ISSUER"), + apiKeyPrefix: process.env.API_KEY_PREFIX?.trim() || "sk_", + defaultClientId: process.env.DEMO_CLIENT_ID?.trim() || "demo-client", + defaultUsageSubjectType: process.env.USAGE_SUBJECT_TYPE?.trim() || "api_key_user", + resolveApiKey: async (apiKey) => { + const entry = keyStore.get(apiKey); + if (!entry) { + return null; + } + return { + userId: entry.userId, + clientId: entry.clientId, + usageSubjectType: entry.usageSubjectType, + }; + }, + }), +}; + +function readBody(req) { + return new Promise((resolve, reject) => { + const chunks = []; + req.on("data", (chunk) => chunks.push(chunk)); + req.on("end", () => resolve(Buffer.concat(chunks))); + req.on("error", reject); + }); +} + +async function handleRequest(req, res) { + if (req.method === "GET" && req.url === "/health") { + res.writeHead(200, { "Content-Type": "text/plain" }); + res.end("ok"); + return; + } + + const host = req.headers.host || `localhost:${port}`; + const body = + req.method === "GET" || req.method === "HEAD" ? undefined : await readBody(req); + const headers = new Headers(); + for (const [key, value] of Object.entries(req.headers)) { + if (value === undefined) { + continue; + } + headers.set(key, Array.isArray(value) ? value.join(", ") : value); + } + + const request = new Request(`http://${host}${req.url}`, { + method: req.method, + headers, + body: body?.length ? body : undefined, + }); + + const response = await routeRemoteSignerWebhookRequest(request, config); + if (!response) { + res.writeHead(404); + res.end(); + return; + } + + res.writeHead(response.status, Object.fromEntries(response.headers)); + res.end(await response.text()); +} + +createServer((req, res) => { + handleRequest(req, res).catch((err) => { + console.error("identity-webhook error:", err); + if (!res.headersSent) { + res.writeHead(500, { "Content-Type": "text/plain" }); + } + res.end("internal error"); + }); +}).listen(port, "0.0.0.0", () => { + console.log(`identity-webhook (api-key) listening on :${port}`); +}); diff --git a/openmeter-collector/.env.example b/openmeter-collector/.env.example index d57e5ec..834833c 100644 --- a/openmeter-collector/.env.example +++ b/openmeter-collector/.env.example @@ -8,8 +8,11 @@ KAFKA_BROKERS=kafka:9092 KAFKA_GATEWAY_TOPIC=livepeer-gateway-events # --- OpenMeter / Konnect (required) --- +# Konnect base URL for customer upsert (provision sidecar). +OPENMETER_URL=https://us.api.konghq.com/v3/openmeter # Konnect: https://.api.konghq.com/v3/openmeter/events # Self-hosted: https:///api/v1/events OPENMETER_INGEST_URL=https://us.api.konghq.com/v3/openmeter/events OPENMETER_API_KEY= +OPENMETER_DEFAULT_PLAN_KEY=clearinghouse_default_ppu ETH_USD_PRICE=3500 diff --git a/openmeter-collector/Dockerfile b/openmeter-collector/Dockerfile index 878989c..b1502c7 100644 --- a/openmeter-collector/Dockerfile +++ b/openmeter-collector/Dockerfile @@ -1,8 +1,32 @@ # OpenMeter Benthos collector: Kafka create_signed_ticket events -> Konnect/OpenMeter ingest. -FROM ghcr.io/openmeterio/benthos-collector:main-6b60ab6-1782310960 +# Includes a local Go provision sidecar for idempotent customer upsert (Kong sdk-konnect-go). +FROM golang:1.25.11-alpine AS provision-build -COPY openmeter-collector/collector.yaml /config.yaml +WORKDIR /src +COPY openmeter-collector/provision/go.mod openmeter-collector/provision/go.sum ./ +RUN go mod download +COPY openmeter-collector/provision/ ./ +RUN CGO_ENABLED=0 go build -trimpath -ldflags="-s -w" -o /provision-server . + +FROM ghcr.io/openmeterio/benthos-collector:main-6b60ab6-1782310960 AS benthos + +FROM alpine:3.21 + +RUN apk add --no-cache wget ca-certificates + +WORKDIR /app + +COPY --from=provision-build /provision-server /app/provision-server COPY openmeter-collector/entrypoint.sh /entrypoint.sh -RUN chmod +x /entrypoint.sh +COPY openmeter-collector/collector.yaml /config.yaml +COPY --from=benthos /usr/local/bin/benthos /usr/local/bin/benthos + +RUN chmod +x /entrypoint.sh /app/provision-server + +ENV PROVISION_PORT=8091 +ENV OPENMETER_DEFAULT_PLAN_KEY=clearinghouse_default_ppu + +HEALTHCHECK --interval=10s --timeout=3s --retries=6 --start-period=10s \ + CMD wget -q -O- http://127.0.0.1:8091/health || exit 1 ENTRYPOINT ["/entrypoint.sh"] diff --git a/openmeter-collector/collector.yaml b/openmeter-collector/collector.yaml index 0f2ed2f..c7fdbfa 100644 --- a/openmeter-collector/collector.yaml +++ b/openmeter-collector/collector.yaml @@ -6,6 +6,7 @@ # - Uses first-colon split semantics to preserve compatibility with builder-sdk. # - Konnect ingest: POST OPENMETER_INGEST_URL (e.g. https://.api.konghq.com/v3/openmeter/events). # For self-hosted OpenMeter use https:///api/v1/events. +# - Customer upsert: local Go provision sidecar on :8091/ensure (Kong sdk-konnect-go). input: kafka: @@ -49,6 +50,7 @@ pipeline: "client_id": $client_id, "external_user_id": $external_user_id, "network_fee_usd_micros": $fee_usd_micros, + "billable_usd_micros": $fee_usd_micros, "pipeline": if $data.pipeline != "" && $data.pipeline != null { $data.pipeline } else { "unknown" }, "model_id": if $data.model_id != "" && $data.model_id != null { $data.model_id } else { "unknown" }, "pixels": $data.pixels.string().or("0"), @@ -64,14 +66,58 @@ pipeline: message: "signed_ticket mapping failed: ${! error() }" - mapping: root = deleted() + # Proactive idempotent customer ensure before ingest (preserve CloudEvent body). + - branch: + request_map: | + root.client_id = this.data.client_id + root.external_user_id = this.data.external_user_id + root.auth_id = this.subject + processors: + - http: + url: http://127.0.0.1:8091/ensure + verb: POST + headers: + Content-Type: application/json + timeout: 10s + result_map: "" + output: - http_client: - url: ${OPENMETER_INGEST_URL} - verb: POST - headers: - Authorization: "Bearer ${OPENMETER_API_KEY}" - Content-Type: application/cloudevents+json - successful_on: - - 200 - - 202 - - 204 + fallback: + - http_client: + url: ${OPENMETER_INGEST_URL} + verb: POST + headers: + Authorization: "Bearer ${OPENMETER_API_KEY}" + Content-Type: application/cloudevents+json + successful_on: + - 200 + - 202 + - 204 + # Self-heal: on ingest failure (e.g. no customer), ensure then retry once. + - processors: + - log: + level: WARN + message: "ingest failed, ensuring customer then retrying: ${! error() }" + - branch: + request_map: | + root.client_id = this.data.client_id + root.external_user_id = this.data.external_user_id + root.auth_id = this.subject + processors: + - http: + url: http://127.0.0.1:8091/ensure + verb: POST + headers: + Content-Type: application/json + timeout: 10s + result_map: "" + http_client: + url: ${OPENMETER_INGEST_URL} + verb: POST + headers: + Authorization: "Bearer ${OPENMETER_API_KEY}" + Content-Type: application/cloudevents+json + successful_on: + - 200 + - 202 + - 204 diff --git a/openmeter-collector/entrypoint.sh b/openmeter-collector/entrypoint.sh old mode 100644 new mode 100755 index 15517be..b4acb02 --- a/openmeter-collector/entrypoint.sh +++ b/openmeter-collector/entrypoint.sh @@ -8,4 +8,27 @@ if [ -f /service/.env ]; then set +a fi +if [ -z "${OPENMETER_URL:-}" ] || [ -z "${OPENMETER_API_KEY:-}" ]; then + echo "entrypoint: OPENMETER_URL and OPENMETER_API_KEY are required" >&2 + exit 1 +fi + +/app/provision-server & +PROVISION_PID=$! + +cleanup() { + if kill -0 "$PROVISION_PID" 2>/dev/null; then + kill "$PROVISION_PID" 2>/dev/null || true + fi +} +trap cleanup EXIT INT TERM + +# Wait for provision sidecar before Benthos starts posting events. +for _ in $(seq 1 30); do + if wget -q -O- "http://127.0.0.1:${PROVISION_PORT:-8091}/health" >/dev/null 2>&1; then + break + fi + sleep 0.2 +done + exec /usr/local/bin/benthos -c /config.yaml diff --git a/openmeter-collector/provision/go.mod b/openmeter-collector/provision/go.mod new file mode 100644 index 0000000..b4218b6 --- /dev/null +++ b/openmeter-collector/provision/go.mod @@ -0,0 +1,11 @@ +module github.com/livepeer/clearinghouse/openmeter-collector/provision + +go 1.25.10 + +require github.com/Kong/sdk-konnect-go v0.39.0 + +require ( + github.com/itchyny/gojq v0.12.17 // indirect + github.com/itchyny/timefmt-go v0.1.6 // indirect + github.com/spyzhov/ajson v0.8.0 // indirect +) diff --git a/openmeter-collector/provision/go.sum b/openmeter-collector/provision/go.sum new file mode 100644 index 0000000..b34824b --- /dev/null +++ b/openmeter-collector/provision/go.sum @@ -0,0 +1,16 @@ +github.com/Kong/sdk-konnect-go v0.39.0 h1:fQhfBsMOERUYkfxi5/cmXaiKC508ZsT+pWA6u7o56qE= +github.com/Kong/sdk-konnect-go v0.39.0/go.mod h1:rDWJcOR3KBkB4Mmrc+2RThAzlNZmcvGVdGpESfgTK9A= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/itchyny/gojq v0.12.17 h1:8av8eGduDb5+rvEdaOO+zQUjA04MS0m3Ps8HiD+fceg= +github.com/itchyny/gojq v0.12.17/go.mod h1:WBrEMkgAfAGO1LUcGOckBl5O726KPp+OlkKug0I/FEY= +github.com/itchyny/timefmt-go v0.1.6 h1:ia3s54iciXDdzWzwaVKXZPbiXzxxnv1SPGFfM/myJ5Q= +github.com/itchyny/timefmt-go v0.1.6/go.mod h1:RRDZYC5s9ErkjQvTvvU7keJjxUYzIISJGxm9/mAERQg= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/spyzhov/ajson v0.8.0 h1:sFXyMbi4Y/BKjrsfkUZHSjA2JM1184enheSjjoT/zCc= +github.com/spyzhov/ajson v0.8.0/go.mod h1:63V+CGM6f1Bu/p4nLIN8885ojBdt88TbLoSFzyqMuVA= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/openmeter-collector/provision/main.go b/openmeter-collector/provision/main.go new file mode 100644 index 0000000..8f4d5b7 --- /dev/null +++ b/openmeter-collector/provision/main.go @@ -0,0 +1,166 @@ +package main + +import ( + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "os" + "strings" +) + +type ensureRequest struct { + ClientID string `json:"client_id"` + ClientIDAlt string `json:"clientId"` + ExternalUserID string `json:"external_user_id"` + ExternalUserAlt string `json:"externalUserId"` + AuthID string `json:"auth_id"` + AuthIDAlt string `json:"authId"` + Subject string `json:"subject"` + Data map[string]any `json:"data"` +} + +func requiredEnv(name string) (string, error) { + value := strings.TrimSpace(os.Getenv(name)) + if value == "" { + return "", fmt.Errorf("%s is required", name) + } + return value, nil +} + +func stringField(body map[string]any, keys ...string) string { + for _, key := range keys { + if raw, ok := body[key]; ok { + if value, ok := raw.(string); ok { + if trimmed := strings.TrimSpace(value); trimmed != "" { + return trimmed + } + } + } + } + return "" +} + +func parseIdentity(body ensureRequest) (clientID, externalUserID string, ok bool) { + data := body.Data + if data == nil { + data = map[string]any{} + } + + clientID = firstNonEmpty( + body.ClientID, + body.ClientIDAlt, + stringField(data, "client_id", "clientId"), + ) + externalUserID = firstNonEmpty( + body.ExternalUserID, + body.ExternalUserAlt, + stringField(data, "external_user_id", "externalUserId"), + ) + if clientID != "" && externalUserID != "" { + return clientID, externalUserID, true + } + + authID := firstNonEmpty( + body.AuthID, + body.AuthIDAlt, + stringField(data, "auth_id", "authId"), + body.Subject, + ) + colon := strings.Index(authID, ":") + if colon > 0 && colon < len(authID)-1 { + return authID[:colon], authID[colon+1:], true + } + return "", "", false +} + +func firstNonEmpty(values ...string) string { + for _, value := range values { + if trimmed := strings.TrimSpace(value); trimmed != "" { + return trimmed + } + } + return "" +} + +func writeJSON(w http.ResponseWriter, status int, body any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(body) +} + +func main() { + port := strings.TrimSpace(os.Getenv("PROVISION_PORT")) + if port == "" { + port = "8091" + } + planKey := strings.TrimSpace(os.Getenv("OPENMETER_DEFAULT_PLAN_KEY")) + if planKey == "" { + planKey = "clearinghouse_default_ppu" + } + + baseURL, err := requiredEnv("OPENMETER_URL") + if err != nil { + log.Fatal(err) + } + apiKey, err := requiredEnv("OPENMETER_API_KEY") + if err != nil { + log.Fatal(err) + } + + provisioner := NewProvisioner(baseURL, apiKey, planKey) + + mux := http.NewServeMux() + mux.HandleFunc("/health", func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/plain") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ok")) + }) + mux.HandleFunc("/ensure", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.NotFound(w, r) + return + } + + raw, err := io.ReadAll(r.Body) + if err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request body"}) + return + } + + var body ensureRequest + if len(raw) > 0 { + if err := json.Unmarshal(raw, &body); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid request json"}) + return + } + } + + clientID, externalUserID, ok := parseIdentity(body) + if !ok { + writeJSON(w, http.StatusBadRequest, map[string]string{ + "error": "client_id and external_user_id (or auth_id) are required", + }) + return + } + + result, err := provisioner.Ensure(r.Context(), ProvisionInput{ + ClientID: clientID, + ExternalUserID: externalUserID, + DisplayName: fmt.Sprintf("%s:%s", clientID, externalUserID), + }) + if err != nil { + log.Printf("provision-server ensure failed: %v", err) + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) + return + } + writeJSON(w, http.StatusOK, result) + }) + + addr := "127.0.0.1:" + port + log.Printf("provision-server listening on %s plan=%s", addr, planKey) + if err := http.ListenAndServe(addr, mux); err != nil { + log.Fatal(err) + } +} diff --git a/openmeter-collector/provision/provision.go b/openmeter-collector/provision/provision.go new file mode 100644 index 0000000..d6f2d78 --- /dev/null +++ b/openmeter-collector/provision/provision.go @@ -0,0 +1,307 @@ +package main + +import ( + "context" + "fmt" + "strings" + + sdkkonnectgo "github.com/Kong/sdk-konnect-go" + "github.com/Kong/sdk-konnect-go/models/components" + "github.com/Kong/sdk-konnect-go/models/operations" +) + +type Provisioner struct { + sdk *sdkkonnectgo.SDK + planKey string +} + +type ProvisionInput struct { + ClientID string + ExternalUserID string + DisplayName string +} + +type ProvisionResult struct { + CustomerKey string `json:"customerKey"` + CustomerID string `json:"customerId"` + SubscriptionID string `json:"subscriptionId"` + PlanKey string `json:"planKey"` + Status string `json:"status"` + Created struct { + Customer bool `json:"customer"` + Subscription bool `json:"subscription"` + } `json:"created"` +} + +func buildCustomerKey(clientID, externalUserID string) string { + return strings.TrimSpace(clientID) + ":" + strings.TrimSpace(externalUserID) +} + +func normalizeOpenMeterURL(baseURL string) string { + url := strings.TrimSpace(baseURL) + url = strings.TrimSuffix(url, "/") + url = strings.TrimSuffix(url, "/events") + url = strings.TrimSuffix(url, "/openmeter") + url = strings.TrimSuffix(url, "/v3") + return url +} + +func NewProvisioner(baseURL, apiKey, planKey string) *Provisioner { + url := normalizeOpenMeterURL(baseURL) + opts := []sdkkonnectgo.SDKOption{ + sdkkonnectgo.WithSecurity(components.Security{ + PersonalAccessToken: sdkkonnectgo.Pointer(apiKey), + }), + } + if url != "" { + opts = append(opts, sdkkonnectgo.WithServerURL(url)) + } else { + opts = append(opts, sdkkonnectgo.WithServerIndex(1)) + } + return &Provisioner{ + sdk: sdkkonnectgo.New(opts...), + planKey: strings.TrimSpace(planKey), + } +} + +func (p *Provisioner) Ensure(ctx context.Context, input ProvisionInput) (*ProvisionResult, error) { + clientID := strings.TrimSpace(input.ClientID) + externalUserID := strings.TrimSpace(input.ExternalUserID) + if clientID == "" || externalUserID == "" { + return nil, fmt.Errorf("clientId and externalUserId must be non-empty") + } + if p.planKey == "" { + return nil, fmt.Errorf("plan key must be non-empty") + } + + customerKey := buildCustomerKey(clientID, externalUserID) + displayName := strings.TrimSpace(input.DisplayName) + if displayName == "" { + displayName = customerKey + } + + customerID, customerCreated, err := p.ensureCustomer(ctx, customerKey, displayName) + if err != nil { + return nil, err + } + + subscriptionID, status, subscriptionCreated, err := p.ensureSubscription(ctx, customerID) + if err != nil { + return nil, err + } + + result := &ProvisionResult{ + CustomerKey: customerKey, + CustomerID: customerID, + SubscriptionID: subscriptionID, + PlanKey: p.planKey, + Status: status, + } + result.Created.Customer = customerCreated + result.Created.Subscription = subscriptionCreated + return result, nil +} + +func (p *Provisioner) ensureCustomer(ctx context.Context, customerKey, displayName string) (string, bool, error) { + if existing, err := p.findCustomerByKey(ctx, customerKey); err != nil { + return "", false, err + } else if existing != "" { + return existing, false, nil + } + + res, err := p.sdk.OpenMeterCustomers.CreateCustomer(ctx, components.CreateCustomerRequest{ + Key: customerKey, + Name: displayName, + UsageAttribution: &components.UsageAttribution{ + SubjectKeys: []string{customerKey}, + }, + }) + if err != nil { + if existing, findErr := p.findCustomerByKey(ctx, customerKey); findErr == nil && existing != "" { + return existing, false, nil + } + return "", false, fmt.Errorf("creating customer %s: %w", customerKey, err) + } + if res.BillingCustomer == nil || res.BillingCustomer.ID == "" { + return "", false, fmt.Errorf("creating customer %s: empty response", customerKey) + } + if res.StatusCode < 200 || res.StatusCode >= 300 { + return "", false, fmt.Errorf("creating customer %s: status %d", customerKey, res.StatusCode) + } + return res.BillingCustomer.ID, true, nil +} + +func (p *Provisioner) findCustomerByKey(ctx context.Context, customerKey string) (string, error) { + res, err := p.sdk.OpenMeterCustomers.ListCustomers(ctx, operations.ListCustomersRequest{ + Filter: &components.ListCustomersParamsFilter{ + Key: &components.StringFieldFilter{ + Eq: sdkkonnectgo.Pointer(customerKey), + }, + }, + Page: &components.PagePaginationQuery{ + Number: sdkkonnectgo.Pointer(int64(1)), + Size: sdkkonnectgo.Pointer(int64(100)), + }, + }) + if err != nil { + return "", fmt.Errorf("listing customers for key %s: %w", customerKey, err) + } + if res.CustomerPagePaginatedResponse != nil { + for _, customer := range res.CustomerPagePaginatedResponse.Data { + if customer.Key == customerKey && customer.ID != "" { + return customer.ID, nil + } + } + } + + getRes, err := p.sdk.OpenMeterCustomers.GetCustomer(ctx, customerKey) + if err == nil && getRes.BillingCustomer != nil && getRes.BillingCustomer.ID != "" { + return getRes.BillingCustomer.ID, nil + } + if err == nil && (getRes.StatusCode == 404) { + return "", nil + } + if err == nil && getRes.StatusCode >= 200 && getRes.StatusCode < 300 { + return "", nil + } + if err != nil { + return "", nil + } + return "", nil +} + +func subscriptionStatusActive(status components.BillingSubscriptionStatus) bool { + switch status { + case components.BillingSubscriptionStatusActive, + components.BillingSubscriptionStatusScheduled: + return true + default: + return false + } +} + +func planStatusUsable(status components.BillingPlanStatus) bool { + switch status { + case components.BillingPlanStatusActive, + components.BillingPlanStatusScheduled: + return true + default: + return false + } +} + +func (p *Provisioner) resolveActivePlan(ctx context.Context) (components.Plan, error) { + const pageSize = int64(50) + page := int64(1) + var best *components.BillingPlan + + for { + res, err := p.sdk.OpenMeterProductCatalog.ListPlans(ctx, operations.ListPlansRequest{ + Filter: &components.ListPlansParamsFilter{ + Key: &components.StringFieldFilter{ + Eq: sdkkonnectgo.Pointer(p.planKey), + }, + }, + Page: &components.PagePaginationQuery{ + Number: sdkkonnectgo.Pointer(page), + Size: sdkkonnectgo.Pointer(pageSize), + }, + }) + if err != nil { + return components.Plan{}, fmt.Errorf("listing plans for key %s: %w", p.planKey, err) + } + if res.PlanPagePaginatedResponse == nil { + break + } + for i := range res.PlanPagePaginatedResponse.Data { + plan := res.PlanPagePaginatedResponse.Data[i] + if plan.DeletedAt != nil || !planStatusUsable(plan.Status) || plan.ID == "" { + continue + } + if best == nil || planVersion(plan) > planVersion(*best) { + copy := plan + best = © + } + } + if len(res.PlanPagePaginatedResponse.Data) < int(pageSize) { + break + } + page++ + } + + if best == nil { + return components.Plan{}, fmt.Errorf("no active plan found for key %s (latest version may be deleted; re-run clearinghouse-bootstrap --skip-auth0)", p.planKey) + } + + return components.Plan{ + ID: sdkkonnectgo.Pointer(best.ID), + Key: sdkkonnectgo.Pointer(p.planKey), + Version: best.Version, + }, nil +} + +func planVersion(plan components.BillingPlan) int64 { + if plan.Version == nil { + return 0 + } + return *plan.Version +} + +func (p *Provisioner) ensureSubscription(ctx context.Context, customerID string) (string, string, bool, error) { + const pageSize = int64(100) + page := int64(1) + + for { + customerFilter := components.CreateULIDFieldFilterStr(customerID) + res, err := p.sdk.OpenMeterSubscriptions.ListSubscriptions(ctx, operations.ListSubscriptionsRequest{ + Filter: &components.ListSubscriptionsParamsFilter{ + CustomerID: &customerFilter, + PlanKey: &components.StringFieldFilterExact{ + Eq: sdkkonnectgo.Pointer(p.planKey), + }, + }, + Page: &components.PagePaginationQuery{ + Number: sdkkonnectgo.Pointer(page), + Size: sdkkonnectgo.Pointer(pageSize), + }, + }) + if err != nil { + return "", "", false, fmt.Errorf("listing subscriptions for customer %s: %w", customerID, err) + } + if res.SubscriptionPagePaginatedResponse != nil { + for _, sub := range res.SubscriptionPagePaginatedResponse.Data { + if sub.ID != "" && subscriptionStatusActive(sub.Status) { + return sub.ID, string(sub.Status), false, nil + } + } + if len(res.SubscriptionPagePaginatedResponse.Data) < int(pageSize) { + break + } + } else { + break + } + page++ + } + + planRef, err := p.resolveActivePlan(ctx) + if err != nil { + return "", "", false, err + } + + createRes, err := p.sdk.OpenMeterSubscriptions.CreateSubscription(ctx, components.BillingSubscriptionCreate{ + Customer: components.BillingSubscriptionCreateCustomer{ + ID: sdkkonnectgo.Pointer(customerID), + }, + Plan: planRef, + }) + if err != nil { + return "", "", false, fmt.Errorf("creating subscription for customer %s plan %s: %w", customerID, p.planKey, err) + } + if createRes.BillingSubscription == nil || createRes.BillingSubscription.ID == "" { + return "", "", false, fmt.Errorf("creating subscription for customer %s plan %s: empty response", customerID, p.planKey) + } + if createRes.StatusCode < 200 || createRes.StatusCode >= 300 { + return "", "", false, fmt.Errorf("creating subscription for customer %s plan %s: status %d", customerID, p.planKey, createRes.StatusCode) + } + return createRes.BillingSubscription.ID, string(createRes.BillingSubscription.Status), true, nil +} diff --git a/remote-signer/.env.example b/remote-signer/.env.example index 1c1bcc7..58007e2 100644 --- a/remote-signer/.env.example +++ b/remote-signer/.env.example @@ -4,10 +4,12 @@ # $EDITOR remote-signer/.env # --- Identity webhook --- -REMOTE_SIGNER_WEBHOOK_URL=https://your-platform.vercel.app/api/signer/authorize -WEBHOOK_SECRET= +REMOTE_SIGNER_WEBHOOK_URL=http://identity-webhook:8090/authorize +# Must match identity-webhook/.env (local dev value below). +WEBHOOK_SECRET=dev-webhook-secret-change-me # Leave REMOTE_SIGNER_WEBHOOK_URL empty to start signer without webhook authorization. # REMOTE_SIGNER_WEBHOOK_URL should always be set in production. +# WEBHOOK_SECRET must match identity-webhook/.env when using the in-compose webhook. # --- Signer --- # Paths below are inside the container. Host files live under remote-signer/data/