Merge pull request #22 from utopia-php/clo-4542

feat(usage): signed deltas, cross-tenant reads and max aggregation (CLO-4542)

Author: lohanidamodar

src/Usage/Accumulator.php

@@ -16,7 +16,7 @@ class Accumulator
     private Usage $usage;
 
     /**
-     * @var array<string, array{tenant: string, metric: string, value: int, type: string, tags: array<string, mixed>, time?: \DateTime}>
+     * @var array<string, array{tenant: string, metric: string, value: int, type: string, tags: array<string, mixed>, allowNegative: bool, time?: \DateTime}>
      */
     private array $buffer = [];
 
@@ -38,9 +38,16 @@ public function __construct(Usage $usage)
      * Events fold additively; earliest non-null time wins on merge.
      * Gauges use last-write-wins.
      *
+     * Negative values are rejected by default for every metric, so a buggy
+     * negative count/bandwidth is still caught. A caller that emits a genuine
+     * signed delta (e.g. realtime connections `+1`/`-1`) opts in per call with
+     * `$allowNegative = true`. The library stays generic — the decision lives
+     * with the caller, not with any metric-name knowledge here.
+     *
      * @param array<string,mixed> $tags
+     * @param bool $allowNegative Permit a negative value for this metric (default: reject).
      */
-    public function collect(string $tenant, string $metric, int $value, string $type, array $tags = [], ?\DateTime $time = null): self
+    public function collect(string $tenant, string $metric, int $value, string $type, array $tags = [], ?\DateTime $time = null, bool $allowNegative = false): self
     {
         // Compare against '' rather than empty(): the string "0" is a valid
         // tenant/metric id but empty("0") is true in PHP.
@@ -50,7 +57,7 @@ public function collect(string $tenant, string $metric, int $value, string $type
         if ($metric === '') {
             throw new \InvalidArgumentException('Metric name cannot be empty');
         }
-        if ($value < 0) {
+        if ($value < 0 && !$allowNegative) {
             throw new \InvalidArgumentException('Value cannot be negative');
         }
         if ($type !== Usage::TYPE_EVENT && $type !== Usage::TYPE_GAUGE) {
@@ -68,6 +75,15 @@ public function collect(string $tenant, string $metric, int $value, string $type
         if ($type === Usage::TYPE_EVENT && isset($this->buffer[$key])) {
             // earliest time wins on merge
             $this->buffer[$key]['value'] += $value;
+
+            // The opt-in is a property of the folded row, not of whichever
+            // call happened to create it. Folding a signed delta into an entry
+            // opened by a plain positive must not leave the net row looking
+            // unauthorised: it would be rejected at write time, and since a
+            // failed batch keeps its entries buffered, every later flush would
+            // retry the same rejection.
+            $this->buffer[$key]['allowNegative'] = $this->buffer[$key]['allowNegative'] || $allowNegative;
+
             if ($time !== null && (!isset($this->buffer[$key]['time']) || $time < $this->buffer[$key]['time'])) {
                 $this->buffer[$key]['time'] = $time;
             }
@@ -78,6 +94,7 @@ public function collect(string $tenant, string $metric, int $value, string $type
                 'value' => $value,
                 'type' => $type,
                 'tags' => $tags,
+                'allowNegative' => $allowNegative,
             ];
             if ($time !== null) {
                 $entry['time'] = $time;

src/Usage/Adapter.php

@@ -102,6 +102,26 @@ abstract public function purge(string $tenant, array $queries = [], ?string $typ
      */
     abstract public function find(string $tenant, array $queries = [], ?string $type = null): array;
 
+    /**
+     * Find metrics across every tenant in shared-tables mode.
+     *
+     * Deliberately crosses the per-tenant isolation every other read enforces,
+     * so it is reserved for operator-side aggregation jobs that roll many
+     * tenants up in one pass. Never reachable from a tenant-scoped request
+     * path. Pair with `groupBy('tenant')` to keep the rows attributable.
+     *
+     * Adapters that cannot express an unscoped read leave this unsupported.
+     *
+     * @param array<\Utopia\Query\Query> $queries
+     * @param string|null $type Metric type: 'event', 'gauge', or null (query both)
+     * @return array<Metric>
+     * @throws \Exception
+     */
+    public function findAcrossTenants(array $queries = [], ?string $type = null): array
+    {
+        throw new \Exception($this->getName() . ' does not support cross-tenant reads');
+    }
+
     /**
      * Count metrics using Query objects.
      *

src/Usage/Adapter/ClickHouse.php

@@ -1115,6 +1115,13 @@ private function validateGroupByAttribute(string $attribute, string $type): bool
     {
         $allowed = $type === Usage::TYPE_GAUGE ? Metric::GAUGE_COLUMNS : Metric::EVENT_COLUMNS;
 
+        // `tenant` is a real column in shared-tables mode, not a dimension on
+        // Metric. Grouping by it is what makes a cross-tenant read
+        // ({@see findAcrossTenants}) attributable, so allow it there.
+        if ($this->sharedTables) {
+            $allowed[] = 'tenant';
+        }
+
         if (in_array($attribute, $allowed, true)) {
             return true;
         }
@@ -1279,9 +1286,10 @@ private function getColumnCodec(string $id): string
      * @param string $type Metric type ('event' or 'gauge')
      * @param array<string,mixed> $tags Tags
      * @param int|null $metricIndex Index for batch error messages
+     * @param bool $allowNegative Permit a negative value for this row (default: reject)
      * @throws Exception
      */
-    private function validateMetricData(string $metric, int $value, string $type, array $tags, ?int $metricIndex = null): void
+    private function validateMetricData(string $metric, int $value, string $type, array $tags, ?int $metricIndex = null, bool $allowNegative = false): void
     {
         $prefix = $metricIndex !== null ? "Metric #{$metricIndex}: " : '';
 
@@ -1293,7 +1301,11 @@ private function validateMetricData(string $metric, int $value, string $type, ar
             throw new Exception($prefix . 'Metric exceeds maximum size of 255 characters');
         }
 
-        if ($value < 0) {
+        // Negatives are rejected by default so a buggy negative count/bandwidth
+        // is caught. A row opts in with `allowNegative` for genuine signed
+        // deltas (realtime connections emit +1/-1). The library stays generic —
+        // the caller decides which metrics may be negative.
+        if ($value < 0 && !$allowNegative) {
             throw new Exception($prefix . 'Value cannot be negative');
         }
 
@@ -1331,7 +1343,10 @@ private function validateMetricsBatch(array $metrics, string $type): void
 
             /** @var array<string, mixed> */
             $tags = $metricData['tags'] ?? [];
-            $this->validateMetricData($metric, $value, $type, $tags, $index);
+            // `allowNegative` is a validation-only flag carried on the row; it
+            // gates the negative-value guard and is never stored as a column.
+            $allowNegative = (bool) ($metricData['allowNegative'] ?? false);
+            $this->validateMetricData($metric, $value, $type, $tags, $index, $allowNegative);
 
             $hasTenant = array_key_exists('tenant', $metricData);
 
@@ -1452,6 +1467,44 @@ public function find(string $tenant, array $queries = [], ?string $type = null):
     {
         $this->setOperationContext('find()');
 
+        return $this->findScoped($tenant, $queries, $type);
+    }
+
+    /**
+     * Find metrics across every tenant. Applies no tenant filter, so it is
+     * restricted to shared-tables mode and reserved for operator-side
+     * aggregation jobs — see {@see Adapter::findAcrossTenants()}.
+     *
+     * Callers should add `groupBy('tenant')` to keep rows attributable; the
+     * aggregated paths already carry `tenant` through select/group-by in
+     * shared-tables mode.
+     *
+     * @param array<Query> $queries
+     * @param string|null $type
+     * @return array<Metric>
+     * @throws Exception
+     */
+    public function findAcrossTenants(array $queries = [], ?string $type = null): array
+    {
+        $this->setOperationContext('findAcrossTenants()');
+
+        if (!$this->sharedTables) {
+            throw new Exception('findAcrossTenants() requires shared-tables mode; use find() instead');
+        }
+
+        return $this->findScoped(null, $queries, $type);
+    }
+
+    /**
+     * Shared body for find()/findAcrossTenants(). A null $tenant means no
+     * tenant filter (cross-tenant); a string scopes to that tenant.
+     *
+     * @param array<Query> $queries
+     * @return array<Metric>
+     * @throws Exception
+     */
+    private function findScoped(?string $tenant, array $queries, ?string $type): array
+    {
         if ($type !== null) {
             return $this->findFromTable($tenant, $queries, $type);
         }
@@ -1533,12 +1586,15 @@ private function queriesMatchType(array $queries, string $type): bool
      * - Gauges: SELECT metric, argMax(value, time) as value, toStartOfInterval(time, INTERVAL ...) as time
      * Results are grouped by metric and time bucket, ordered by time ASC.
      *
+     * An `aggregate('max')` query overrides the per-type default value
+     * expression — see {@see findAggregatedFromTable()}.
+     *
      * @param array<Query> $queries
      * @param string $type 'event' or 'gauge'
      * @return array<Metric>
      * @throws Exception
      */
-    private function findFromTable(string $tenant, array $queries, string $type): array
+    private function findFromTable(?string $tenant, array $queries, string $type): array
     {
         $tableName = $this->getTableForType($type);
         $fromTable = $this->buildTableReference($tableName);
@@ -1551,9 +1607,17 @@ private function findFromTable(string $tenant, array $queries, string $type): ar
             throw new Exception('Cursor pagination cannot be combined with groupByInterval');
         }
 
-        // Route through the aggregated path whenever any aggregation
-        // hint is present — time bucketing, dimension breakdown, or both.
-        if (isset($parsed['groupByInterval']) || !empty($parsed['groupBy'])) {
+        // Route through the aggregated path whenever any aggregation hint is
+        // present — time bucketing, dimension breakdown, an explicit
+        // aggregate(), or any combination. An aggregate() on its own still
+        // counts: `aggregate('max')` with no interval and no dimensions is the
+        // flat "highest value over this window" shape, and without this it
+        // would fall through and return raw rows instead.
+        if (
+            isset($parsed['groupByInterval'])
+            || !empty($parsed['groupBy'])
+            || isset($parsed['aggregate'])
+        ) {
             return $this->findAggregatedFromTable($parsed, $fromTable, $type);
         }
 
@@ -1614,7 +1678,7 @@ private function findFromTable(string $tenant, array $queries, string $type): ar
      *          toStartOfInterval(time, INTERVAL 1 HOUR) as time
      *   FROM table WHERE ... GROUP BY metric, time ORDER BY time ASC
      *
-     * @param array{filters: array<string>, params: array<string, mixed>, orderBy?: array<string>, limit?: int, offset?: int, groupByInterval?: string, groupBy?: array<int, string>} $parsed Parsed query data from parseQueries()
+     * @param array{filters: array<string>, params: array<string, mixed>, orderBy?: array<string>, limit?: int, offset?: int, groupByInterval?: string, groupBy?: array<int, string>, aggregate?: string} $parsed Parsed query data from parseQueries()
      * @param string $fromTable Fully qualified table reference
      * @param string $type 'event' or 'gauge'
      * @return array<Metric>
@@ -1624,11 +1688,18 @@ private function findAggregatedFromTable(array $parsed, string $fromTable, strin
     {
         $hasInterval = isset($parsed['groupByInterval']);
 
-        // Choose aggregation function based on metric type
+        // Choose aggregation function based on metric type. `aggregate('max')`
+        // overrides both defaults: for gauges it takes the highest reading in
+        // the bucket rather than the latest one (argMax), which is what rolling
+        // a sampled level series up to a coarser interval needs.
         $valueExpr = $type === Usage::TYPE_GAUGE
             ? 'argMax(value, time) as value'
             : 'SUM(value) as value';
 
+        if (($parsed['aggregate'] ?? null) === 'max') {
+            $valueExpr = 'max(value) as value';
+        }
+
         // Bucket column is only emitted when time bucketing is requested.
         // Without it the result is a flat aggregate per (metric, …dims).
         $bucketSelect = '';
@@ -3470,15 +3541,18 @@ private function buildOrderBySql(array $orderAttributes, bool $flip = false): ar
      * @param string $tenant Tenant scope (shared-tables mode)
      * @param array<Query> $queries
      * @param string $type 'event' or 'gauge' — used for attribute validation
-     * @return array{filters: array<int, string>, params: array<string, mixed>, orderBy?: array<string>, orderAttributes?: array<int, array{attribute: string, direction: string}>, limit?: int, offset?: int, groupByInterval?: string, groupBy?: array<int, string>, cursor?: array<string, mixed>, cursorDirection?: string}
+     * @return array{filters: array<int, string>, params: array<string, mixed>, orderBy?: array<string>, orderAttributes?: array<int, array{attribute: string, direction: string}>, limit?: int, offset?: int, groupByInterval?: string, groupBy?: array<int, string>, aggregate?: string, cursor?: array<string, mixed>, cursorDirection?: string}
      * @throws Exception
      */
-    private function parseQueries(string $tenant, array $queries, string $type = 'event'): array
+    private function parseQueries(?string $tenant, array $queries, string $type = 'event'): array
     {
-        if ($this->sharedTables) {
-            // An empty tenant would compile to `tenant = ''` and silently read
-            // an empty scope. Fail fast instead, like the write side. ("0" is
-            // a valid tenant id, so check for '' specifically.)
+        // A null tenant is the explicit cross-tenant read used by operator-side
+        // aggregation ({@see findAcrossTenants}) — no tenant filter is applied.
+        // It is deliberately distinct from '': an empty string would compile to
+        // `tenant = ''` and silently read an empty scope, so that still fails
+        // fast, like the write side. ("0" is a valid tenant id, so check for ''
+        // specifically.)
+        if ($this->sharedTables && $tenant !== null) {
             if ($tenant === '') {
                 throw new Exception('Tenant cannot be empty in shared-tables mode');
             }
@@ -3493,6 +3567,7 @@ private function parseQueries(string $tenant, array $queries, string $type = 'ev
         $offset = null;
         $groupByInterval = null;
         $groupBy = [];
+        $aggregate = null;
         $cursor = null;
         $cursorDirection = null;
         $paramCounter = 0;
@@ -3739,6 +3814,16 @@ private function parseQueries(string $tenant, array $queries, string $type = 'ev
                         $groupBy[] = $attribute;
                     }
                     break;
+
+                case UsageQuery::TYPE_AGGREGATE:
+                    $aggValue = $values[0] ?? null;
+                    if (!is_string($aggValue) || !in_array($aggValue, UsageQuery::VALID_AGGREGATES, true)) {
+                        throw new Exception(
+                            'Invalid aggregate: expected one of ' . implode(', ', UsageQuery::VALID_AGGREGATES)
+                        );
+                    }
+                    $aggregate = $aggValue;
+                    break;
             }
         }
 
@@ -3768,6 +3853,10 @@ private function parseQueries(string $tenant, array $queries, string $type = 'ev
             $result['groupBy'] = $groupBy;
         }
 
+        if ($aggregate !== null) {
+            $result['aggregate'] = $aggregate;
+        }
+
         if ($cursor !== null && $cursorDirection !== null) {
             $result['cursor'] = $cursor;
             $result['cursorDirection'] = $cursorDirection;

src/Usage/Adapter/Database.php

@@ -142,7 +142,7 @@ protected function getColumnDefinition(string $id, string $type = 'event'): stri
      * Each metric carries its own `tenant` (shared-tables mode), so a single
      * batch may span multiple tenants.
      *
-     * @param array<array{tenant: string, metric: string, value: int, tags?: array<string,mixed>}> $metrics
+     * @param array<array{tenant: string, metric: string, value: int, tags?: array<string,mixed>, allowNegative?: bool}> $metrics
      * @param string $type Metric type: 'event' or 'gauge'
      * @param int $batchSize
      * @return bool
@@ -157,7 +157,13 @@ public function addBatch(array $metrics, string $type, int $batchSize = 1000): b
                     throw new \InvalidArgumentException("Invalid type '{$type}'. Allowed: event, gauge");
                 }
 
-                if ($metric['value'] < 0) {
+                // Negatives are rejected by default so a buggy count or
+                // bandwidth figure is still caught. A row carrying a genuine
+                // signed delta opts in, the same contract the accumulator sets
+                // and the ClickHouse adapter honours - an adapter that ignored
+                // it would reject the row on every flush, and a failed batch
+                // keeps its entries buffered.
+                if ($metric['value'] < 0 && !($metric['allowNegative'] ?? false)) {
                     throw new \InvalidArgumentException('Value cannot be negative');
                 }
 

src/Usage/Usage.php

@@ -154,6 +154,24 @@ public function find(string $tenant, array $queries = [], ?string $type = null):
         return $this->adapter->find($tenant, $queries, $type);
     }
 
+    /**
+     * Find metrics across every tenant in shared-tables mode.
+     *
+     * Unlike find(), this applies no tenant filter — it is for operator-side
+     * aggregation jobs that need one pass over all tenants instead of N
+     * per-tenant queries. Keep it out of request-scoped code paths, and add
+     * `groupBy('tenant')` so the returned rows stay attributable.
+     *
+     * @param array<\Utopia\Query\Query> $queries
+     * @param string|null $type Metric type: 'event', 'gauge', or null (query both)
+     * @return array<Metric>
+     * @throws \Exception
+     */
+    public function findAcrossTenants(array $queries = [], ?string $type = null): array
+    {
+        return $this->adapter->findAcrossTenants($queries, $type);
+    }
+
     /**
      * Count metrics using Query objects.
      *