web-terminal/control-plane/src/metering/collect.ts

/**
 * T12 — billing-metering hooks (EXPLORE §6): meter PAIRED HOSTS + CONCURRENT VIEWERS (not
 * bandwidth). The sample's `nodeId` is the caller's authenticated mTLS identity (INV3-analog) and
 * its `accountId` is re-derived SERVER-SIDE from the host registry (INV1) — a node-supplied account
 * is never trusted. Samples are append-only immutable rows (INV8); zero terminal payload (INV10).
 */
import { z } from 'zod'
import type { MeteringSampleRow } from '../model/records.js'
import type { MeteringStore } from '../store/ports.js'
import type { HostRegistry } from '../registry/hosts.js'
import type { NodeIdentity } from '../node-auth/identity.js'

export interface MeteringSample {
  readonly hostId: string
  readonly concurrentViewers: number
  readonly sampledAt: string
  // NO client/node-supplied accountId or nodeId — both are derived server-side.
}

const MeteringSampleSchema = z
  .object({
    hostId: z.string().uuid(),
    concurrentViewers: z.number().int().nonnegative(),
    sampledAt: z.string().datetime({ offset: true }),
  })
  .strict()

export class MeteringError extends Error {}

export interface UsageRollup {
  readonly pairedHostPeak: number
  readonly viewerPeak: number
  readonly viewerHours: number
}

export interface MeteringCollector {
  ingestSample(caller: NodeIdentity, sample: MeteringSample): Promise<void>
  pairedHostCount(accountId: string): Promise<number>
  rollupUsage(accountId: string, from: string, to: string): Promise<UsageRollup>
}

export interface MeteringDeps {
  readonly metering: MeteringStore
  readonly hosts: HostRegistry
}

export function createMeteringCollector(deps: MeteringDeps): MeteringCollector {
  return {
    async ingestSample(caller, sample) {
      const parsed = MeteringSampleSchema.safeParse(sample)
      if (!parsed.success) throw new MeteringError(`invalid metering sample: ${parsed.error.issues[0]?.message}`)
      const host = await deps.hosts.getHost(parsed.data.hostId)
      // Attribution derived from ownership — a hostId the caller can't substantiate is rejected (INV1).
      if (host === null) throw new MeteringError('unknown hostId — cannot attribute usage')
      const row: MeteringSampleRow = {
        hostId: parsed.data.hostId,
        accountId: host.accountId, // server-derived, never node-asserted
        nodeId: caller.nodeId, // authenticated identity, never a body field
        concurrentViewers: parsed.data.concurrentViewers,
        sampledAt: parsed.data.sampledAt,
      }
      await deps.metering.append(row) // append-only (INV8)
    },
    async pairedHostCount(accountId) {
      const hosts = await deps.hosts.listHosts(accountId)
      return hosts.filter((h) => h.status !== 'revoked').length
    },
    async rollupUsage(accountId, from, to) {
      const samples = [...(await deps.metering.query(accountId, from, to))].sort(
        (a, b) => Date.parse(a.sampledAt) - Date.parse(b.sampledAt),
      )
      const viewerPeak = samples.reduce((m, s) => Math.max(m, s.concurrentViewers), 0)
      const distinctHosts = new Set(samples.map((s) => s.hostId))
      // Step-function integral of concurrent viewers over the window → viewer-hours.
      const toMs = Date.parse(to)
      let viewerHours = 0
      for (let i = 0; i < samples.length; i++) {
        const cur = samples[i] as MeteringSampleRow
        const nextMs = i + 1 < samples.length ? Date.parse((samples[i + 1] as MeteringSampleRow).sampledAt) : toMs
        const spanHours = Math.max(0, nextMs - Date.parse(cur.sampledAt)) / 3_600_000
        viewerHours += cur.concurrentViewers * spanHours
      }
      return { pairedHostPeak: distinctHosts.size, viewerPeak, viewerHours }
    },
  }
}