feat(usage): peak concurrent connections aggregation (CLO-4542)

[lohanidamodar]

CLO-4542 — Peak concurrent realtime connections (query-side)

realtime.connections is emitted as +1 on connect / -1 on disconnect and stored as a TYPE_EVENT, aggregated with SUM. A plain MAX(value) over ±1 rows is meaningless (returns 1), and a per-pod gauge snapshot can't give a project-wide peak. The ±1 deltas already contain every connect/disconnect across all pods with timestamps, so peak concurrency is exactly max(running_sum(value)) ordered by time — cross-pod correct, no producer change (approach B).

Changes

UsageQuery — new aggregate query hint mirroring groupBy:

• TYPE_AGGREGATE, VALID_AGGREGATES = ['sum','peak'] (sum = current default).
• aggregate(string $function): self (validates), plus isAggregate / extractAggregate / removeAggregate helpers and isMethod() support.

ClickHouse adapter:

• parseQueries() records aggregate=peak and splits the time filters from the non-time filters, capturing the window-start param so the baseline subquery can key on time < start. The sum/absent path is 100% unchanged.
• findFromTable() routes peak to a new findPeakFromTable() (before the SUM-based aggregated path) which builds:

SELECT metric, max(running) as value, toStartOfInterval(time, {interval}) AS bucket
FROM (
  SELECT metric, time,
         ifNull((SELECT sum(value) FROM t WHERE {nonTimeFilters} AND time < {start}), 0)
           + sum(value) OVER (PARTITION BY metric[, tenant][, dims] ORDER BY time ASC
                              ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS running
  FROM t WHERE {nonTimeFilters} AND time >= {start} AND time <= {end}
)
GROUP BY metric, bucket ORDER BY bucket ASC

Without groupByInterval the bucket column is dropped → a single max(running) row per metric. Honours interval bucketing, groupBy dimensions, and limit/offset/orderBy consistently with findAggregatedFromTable. Injection-safe via the existing {name:Type} parameter binding.

Baseline note

The baseline scalar subquery (time < start) adds connections opened before the window (still open at start) so an in-window peak isn't understated. It scans history before the window; for billing windows this is bounded by monthly partition pruning but could be non-trivial on very old projects. It can be bounded later (e.g. time >= start - 1 day) at a small accuracy cost if it proves costly. Default: correct/unbounded baseline.

Tests

Flat and interval peak, pre-window baseline inclusion, interleaved-"pod" sum-before-max, and UsageQuery unit coverage for aggregate() + helpers.

Draft: appwrite/cloud depends on a subsequent tagged release.

---

Update — write-path correctness (opt-in -1, per-second fold)

Two follow-ups so the peak is correct end-to-end (the write path previously dropped -1 disconnects and folded sub-flush bursts away):

1. Opt-in negative deltas (strict by default). Negative values stay rejected by default for every metric, events included, so a buggy negative count/bandwidth is still caught. A caller emitting a genuine signed delta opts in per row:
   • Accumulator::collect(..., ?int $foldSeconds = null, bool $allowNegative = false) — the flag rides onto the buffered entry and is handed to addBatch.
   • ClickHouse::validateMetricData(..., bool $allowNegative = false) gates the guard (if ($value < 0 && !$allowNegative) throw); validateMetricsBatch() reads allowNegative off each row.
   • The flag is validation-only — it is never written as a stored column. The library stays generic; the cloud worker passes allowNegative: true (with foldSeconds: 1) for realtime.connections.
2. Optional per-second fold. When foldSeconds is set, the second bucket floor(ts / foldSeconds) * foldSeconds joins the fold key and becomes the entry time, so events fold only within the same window and intra-flush bursts survive as per-second net rows. When null, behaviour is identical to before.

Tests: negatives rejected by default via both collect() and addBatch; persisted and netted when opted in; gauges still reject; per-second fold groups within / splits across seconds; and aggregate('peak') over per-second net rows returns the true peak (4) for a burst that a single per-flush net row hides (peak 1).