refactor(usage): extract buffering into a standalone Accumulator

README.md

@@ -79,8 +79,13 @@ Event-specific columns (see `Metric::EVENT_COLUMNS`): `path`, `method`, `status`
 `deviceModel`.
 
 ```php
+use Utopia\Usage\Accumulator;
+
+// Buffer metrics in memory and flush them in batch
+$accumulator = new Accumulator($adapter);
+
 // Collect events — values accumulate in-memory buffer (summed per metric)
-$usage->collect('bandwidth', 5000, Usage::TYPE_EVENT, [
+$accumulator->collect('bandwidth', 5000, Usage::TYPE_EVENT, [
     'path' => '/v1/storage/files',
     'method' => 'POST',
     'status' => '201',
@@ -115,8 +120,8 @@ Gauge-specific columns (see `Metric::GAUGE_COLUMNS`): `teamId`,
 
 ```php
 // Collect gauges — last value wins per metric in buffer
-$usage->collect('users', 1500, Usage::TYPE_GAUGE);
-$usage->collect('storage.size', 1048576, Usage::TYPE_GAUGE, [
+$accumulator->collect('users', 1500, Usage::TYPE_GAUGE);
+$accumulator->collect('storage.size', 1048576, Usage::TYPE_GAUGE, [
     'teamId' => 'team_x',
     'teamInternalId' => '7',
     'resourceId' => 'abc123',
@@ -126,15 +131,15 @@ $usage->collect('storage.size', 1048576, Usage::TYPE_GAUGE, [
 
 ### Flushing
 
+The accumulator exposes raw signals — `count()` (buffered entries) and
+`elapsedSeconds()` (seconds since last flush) — and leaves the flush policy to
+the caller.
+
 ```php
-// Check if flush is recommended (threshold or interval reached)
-if ($usage->shouldFlush()) {
-    $usage->flush(); // Writes events to events table, gauges to gauges table
+// Flush when the buffer grows large or enough time has passed
+if ($accumulator->count() >= 5000 || $accumulator->elapsedSeconds() >= 10) {
+    $accumulator->flush(); // Writes events to events table, gauges to gauges table
 }
-
-// Configure thresholds
-$usage->setFlushThreshold(5000);  // Flush after 5000 collect() calls (default: 10,000)
-$usage->setFlushInterval(10);     // Flush after 10 seconds (default: 20)
 ```
 
 ### Batch Writes

src/Usage/Accumulator.php

@@ -0,0 +1,169 @@
+<?php
+
+namespace Utopia\Usage;
+
+/**
+ * In-memory metric accumulator.
+ *
+ * Buffers collect() calls and flushes them to an adapter in batches.
+ * Events are summed per metric+tags; gauges use last-write-wins.
+ *
+ * The accumulator exposes raw signals — count() and elapsedSeconds() — so
+ * callers can decide when to flush rather than baking a policy in here.
+ */
+class Accumulator
+{
+    private Adapter $adapter;
+
+    /**
+     * In-memory buffer for metrics.
+     * Keyed by "{metric}:{type}:{tagsHash}" — events are summed, gauges use last-write-wins.
+     *
+     * @var array<string, array{metric: string, value: int, type: string, tags: array<string, mixed>}>
+     */
+    private array $buffer = [];
+
+    /** @var float Timestamp of the last flush */
+    private float $flushedAt;
+
+    /**
+     * @param  Adapter  $adapter  The adapter to flush buffered metrics to
+     */
+    public function __construct(Adapter $adapter)
+    {
+        $this->adapter = $adapter;
+        $this->flushedAt = microtime(true);
+    }
+
+    /**
+     * Collect a metric into the in-memory buffer for deferred flushing.
+     *
+     * For event type: multiple collect() calls for the same metric are summed.
+     * For gauge type: last-write-wins semantics.
+     * No period fan-out — raw timestamps are used.
+     *
+     * @param string $metric Metric name
+     * @param int $value Value
+     * @param string $type Metric type: 'event' or 'gauge'
+     * @param array<string,mixed> $tags Optional tags
+     * @return self
+     */
+    public function collect(string $metric, int $value, string $type, array $tags = []): self
+    {
+        if (empty($metric)) {
+            throw new \InvalidArgumentException('Metric name cannot be empty');
+        }
+        if ($value < 0) {
+            throw new \InvalidArgumentException('Value cannot be negative');
+        }
+        if ($type !== Usage::TYPE_EVENT && $type !== Usage::TYPE_GAUGE) {
+            throw new \InvalidArgumentException("Invalid metric type '{$type}'. Allowed: " . Usage::TYPE_EVENT . ', ' . Usage::TYPE_GAUGE);
+        }
+
+        $tagsHash = !empty($tags) ? md5(json_encode($tags, JSON_THROW_ON_ERROR)) : '';
+        $key = $metric . ':' . $type . ':' . $tagsHash;
+
+        if ($type === Usage::TYPE_EVENT && isset($this->buffer[$key])) {
+            // Additive: sum values for the same metric + tags combination
+            $this->buffer[$key]['value'] += $value;
+        } else {
+            // New event entry, or gauge (last-write-wins)
+            $this->buffer[$key] = [
+                'metric' => $metric,
+                'value' => $value,
+                'type' => $type,
+                'tags' => $tags,
+            ];
+        }
+
+        return $this;
+    }
+
+    /**
+     * Flush the in-memory buffer to storage.
+     *
+     * Separates buffered metrics into events and gauges, then writes each batch
+     * to the appropriate table via addBatch(). Only entries whose batch write
+     * succeeds are removed from the buffer, so a partial failure preserves
+     * the unwritten metrics for retry on the next flush.
+     *
+     * If addBatch() throws mid-flush, any earlier successful batches have
+     * already been cleared from the buffer — the exception is allowed to
+     * propagate so the caller can observe the failure.
+     *
+     * @return bool True if all batches succeeded (or buffer was empty)
+     * @throws \Exception
+     */
+    public function flush(): bool
+    {
+        if (empty($this->buffer)) {
+            $this->flushedAt = microtime(true);
+            return true;
+        }
+
+        // Separate events and gauges; keep track of buffer keys so we can
+        // selectively unset only the entries whose write succeeded.
+        $eventKeys = [];
+        $gaugeKeys = [];
+        $events = [];
+        $gauges = [];
+
+        foreach ($this->buffer as $key => $entry) {
+            if ($entry['type'] === Usage::TYPE_EVENT) {
+                $events[] = $entry;
+                $eventKeys[] = $key;
+            } else {
+                $gauges[] = $entry;
+                $gaugeKeys[] = $key;
+            }
+        }
+
+        $overallResult = true;
+
+        // Flush events — clear buffer entries only on success.
+        if (!empty($events)) {
+            if ($this->adapter->addBatch($events, Usage::TYPE_EVENT)) {
+                foreach ($eventKeys as $key) {
+                    unset($this->buffer[$key]);
+                }
+            } else {
+                $overallResult = false;
+            }
+        }
+
+        // Flush gauges — clear buffer entries only on success.
+        if (!empty($gauges)) {
+            if ($this->adapter->addBatch($gauges, Usage::TYPE_GAUGE)) {
+                foreach ($gaugeKeys as $key) {
+                    unset($this->buffer[$key]);
+                }
+            } else {
+                $overallResult = false;
+            }
+        }
+
+        $this->flushedAt = microtime(true);
+
+        return $overallResult;
+    }
+
+    /**
+     * Number of unique metric entries currently buffered.
+     *
+     * @return int
+     */
+    public function count(): int
+    {
+        return count($this->buffer);
+    }
+
+    /**
+     * Seconds elapsed since the last flush.
+     *
+     * @return float
+     */
+    public function elapsedSeconds(): float
+    {
+        return microtime(true) - $this->flushedAt;
+    }
+}