collector.yaml

# OpenMeter Collector: go-livepeer remote signer Kafka -> hosted OpenMeter/Konnect.
#
# Kept in sync with pymthouse deploy/openmeter-collector/collector.yaml (identity +
# fee mapping). Clearinghouse-specific: entrypoint normalizes OPENMETER_URL for
# Konnect/self-hosted ingest; stricter owner compound-auth + empty-id guards;
# billable_usd_micros passthrough for phase-2 catalog meters; non-positive ETH/USD
# rejection. Catalog/bootstrap Starter plans align with pymthouse Starter billing
# (network_spend + credit_then_invoice + discounts.usage; see provision/catalog.json).
#
# ETH/USD: fetched from PRICE_ORACLE_URL at startup (defaults to Coinbase spot endpoint;
# 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). Each egress CloudEvent includes data.eth_usd_price
# (the cached rate used for that event's Wei → USD micros conversion), numeric data.fee_wei
# and data.billable_secs (so OpenMeter SUM valueProperty accumulates), and data.manifest_id
# (fallback: session_id -> request_id -> "unknown"). network_fee_usd_micros is the exact
# fractional fee_wei * eth_usd / 1e12 (no per-ticket ceil; read paths ceil once per
# session/boundary). Meter values that must SUM are numbers (not strings).
#
# Oracle JSON (any one shape):
#   {"price": 3456.78}
#   {"ethereum":{"usd":3456.78}}              # CoinGecko simple price API
#   {"data":{"amount":"3456.78"}}             # Coinbase spot API
#
# Events ingest: OPENMETER_URL is normalized by entrypoint.sh (base URL -> full ingest
# path) and used verbatim by the Benthos http_client output below.
#
# Identity (go-livepeer Kafka wire -> normalized CloudEvent egress):
# - Upstream auth_id stays client_id:usage_subject (webhook -> go-livepeer state -> Kafka).
# - 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
    memory: {}

processor_resources:
  - label: fetch_eth_usd_price
    processors:
      - http:
          url: ${PRICE_ORACLE_URL:https://api.coinbase.com/v2/prices/ETH-USD/spot}
          verb: GET
      - mapping: |
          let price = match {
            this.exists("price") => this.price.number(),
            this.exists("ethereum.usd") => this.ethereum.usd.number(),
            this.exists("data.amount") => this.data.amount.number(),
            _ => null
          }
          root = if $price == null { throw("price oracle response missing price field") }
          root = if $price <= 0 { throw("price oracle returned non-positive price") }
          root = $price.string()
      - cache:
          resource: eth_usd
          operator: set
          key: eth_usd
          value: ${! content() }

input:
  sequence:
    inputs:
      # Startup warm — must succeed before Kafka starts (no catch).
      - label: price_warm
        generate:
          count: 1
          interval: 0s
          mapping: 'root = {}'
        processors:
          - resource: fetch_eth_usd_price
          - mapping: root = deleted()

      # Ongoing: background refresh + Kafka events.
      - label: runtime
        broker:
          inputs:
            - label: price_refresh
              generate:
                count: 0
                interval: ${PRICE_ORACLE_REFRESH:5m}
                mapping: 'root = {}'
              processors:
                - try:
                    - resource: fetch_eth_usd_price
                    - mapping: root = deleted()
                - catch:
                    - log:
                        level: WARN
                        message: "price oracle refresh failed: ${! error() }"
                    - mapping: root = deleted()

            - label: kafka_events
              kafka:
                addresses:
                  - ${KAFKA_BROKERS}
                topics:
                  - ${KAFKA_GATEWAY_TOPIC}
                consumer_group: openmeter-collector
                start_from_oldest: true

pipeline:
  processors:
    - mapping: |
        root = if this.type != "create_signed_ticket" { deleted() }

    - branch:
        request_map: 'root = ""'
        processors:
          - cache:
              resource: eth_usd
              operator: get
              key: eth_usd
        result_map: |
          meta eth_usd_price = content().number()

    - mapping: |
        let data = this.data.or({})
        let auth_id_raw = $data.auth_id.or("").string().trim()
        let client_id_field = $data.client_id.or("").string().trim()
        let usage_subject = $data.usage_subject.or("").string().trim()
        let auth_id = if $auth_id_raw != "" {
          $auth_id_raw
        } else if $client_id_field != "" && $usage_subject != "" {
          $client_id_field + ":" + $usage_subject
        } else {
          ""
        }

        let colon = $auth_id.index_of(":")
        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 }
        # Owner wallet: wire usage_subject is owner:{users.id}; CE subject = bare id.
        # Only compound auth_id (client_id:owner:…) is treated as owner — a bare
        # "owner:…" string is not a valid wire form.
        # M2M end-users: meter subject = full compound auth_id.
        let is_owner_subject = $is_compound && $usage_subject_parsed.index_of("owner:") == 0
        let owner_bare_id = if $is_owner_subject {
          $usage_subject_parsed.slice("owner:".length()).trim()
        } else {
          ""
        }
        # Preserve upstream type (oidc_user / api_key_user) when present; owners always app_owner.
        let upstream_subject_type = $data.usage_subject_type.or("").string().trim()
        let usage_subject_type = if $is_owner_subject {
          "app_owner"
        } else if $upstream_subject_type != "" {
          $upstream_subject_type
        } else {
          "external_user_id"
        }
        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).
        # Exact fractional value at ingest (no per-ticket ceil/round). Per-ticket
        # ceil overbills dense live sessions (every sub-micro ticket becomes
        # >= 1 micro); read paths and session/display boundaries ceil once so
        # sub-micro tickets still accumulate into whole micros.
        let fee_wei = $data.computed_fee.number().or(0)
        let fee_usd_micros = $fee_wei * $eth_usd / 1000000000000
        let billable_secs = $data.billable_secs.number().or(0)
        # Session key: stream manifest -> payment StateID -> request id (BYOC/stateless).
        let manifest_id = if $data.manifest_id != "" && $data.manifest_id != null {
          $data.manifest_id.string().trim()
        } else if $data.session_id != "" && $data.session_id != null {
          $data.session_id.string().trim()
        } else if $data.request_id != "" && $data.request_id != null {
          $data.request_id.string().trim()
        } else {
          "unknown"
        }
        let manifest_id_out = if $manifest_id != "" { $manifest_id } else { "unknown" }

        root = if $auth_id == "" || ($is_owner_subject && $owner_bare_id == "") {
          deleted()
        } else {
          {
            "specversion": "1.0",
            "type": "create_signed_ticket",
            "id": $data.request_id.or(uuid_v4()),
            "source": "go-livepeer-remote-signer",
            "subject": $openmeter_subject,
            "time": $data.current_time.or(now()),
            "data": {
              "client_id": $client_id,
              "usage_subject": $usage_subject_out,
              "usage_subject_type": $usage_subject_type,
              "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.
              "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" },
              "manifest_id": $manifest_id_out,
              # Numbers (not strings) so OpenMeter SUM $.billable_secs / $.fee_wei accumulate.
              "billable_secs": $billable_secs,
              "pixels": $data.pixels.string().or("0"),
              "fee_wei": $fee_wei,
              "gateway_request_id": $data.request_id,
              "auth_id": $auth_id,
              "openmeter_customer_key": $openmeter_customer_key,
              "eth_usd_price": $eth_usd,
            }
          }
        }

    - mapping: |
        meta client_id = this.data.client_id.or("")

    # One Kong OpenMeter organization serves every tenant; customers are
    # separated by the openmeter_customer_key contract above, not by per-tenant
    # ingest credentials. Uses env() rather than a dollar-brace interpolation:
    # Benthos substitutes those in the raw config text before parsing, which
    # makes them lint-required even inside a comment.
    - mapping: |
        meta openmeter_url = env("OPENMETER_URL").or("")
        meta openmeter_token = env("OPENMETER_API_KEY").or("")
        root = if meta("openmeter_url") == "" || meta("openmeter_token") == "" {
          throw("openmeter ingest unresolved: set OPENMETER_URL and OPENMETER_API_KEY")
        } else {
          this
        }

    - catch:
        - log:
            level: ERROR
            message: "signed_ticket mapping failed: ${! error() }"
        - mapping: root = deleted()

output:
  http_client:
    url: '${! meta("openmeter_url") }'
    verb: POST
    headers:
      Authorization: 'Bearer ${! meta("openmeter_token") }'
      Content-Type: application/cloudevents+json
    successful_on:
      - 200
      - 202
      - 204