aggregate telemetry over the last 7 days (#470)

* aggregate telemetry over the last 7 days

* cover missing unit test vectors

web/lib/potato_mesh/application/queries.rb

@@ -18,6 +18,36 @@ module PotatoMesh
   module App
     module Queries
       MAX_QUERY_LIMIT = 1000
+      DEFAULT_TELEMETRY_WINDOW_SECONDS = 86_400
+      DEFAULT_TELEMETRY_BUCKET_SECONDS = 300
+      TELEMETRY_AGGREGATE_COLUMNS =
+        %w[
+          battery_level
+          voltage
+          channel_utilization
+          air_util_tx
+          temperature
+          relative_humidity
+          barometric_pressure
+          gas_resistance
+          current
+          iaq
+          distance
+          lux
+          white_lux
+          ir_lux
+          uv_lux
+          wind_direction
+          wind_speed
+          wind_gust
+          wind_lull
+          weight
+          radiation
+          rainfall_1h
+          rainfall_24h
+          soil_moisture
+          soil_temperature
+        ].freeze
 
       # Remove nil or empty values from an API response hash to reduce payload size
       # while preserving legitimate zero-valued measurements.
@@ -465,6 +495,82 @@ module PotatoMesh
         db&.close
       end
 
+      def query_telemetry_buckets(window_seconds:, bucket_seconds:)
+        window = coerce_integer(window_seconds) || DEFAULT_TELEMETRY_WINDOW_SECONDS
+        window = DEFAULT_TELEMETRY_WINDOW_SECONDS if window <= 0
+        bucket = coerce_integer(bucket_seconds) || DEFAULT_TELEMETRY_BUCKET_SECONDS
+        bucket = DEFAULT_TELEMETRY_BUCKET_SECONDS if bucket <= 0
+
+        db = open_database(readonly: true)
+        db.results_as_hash = true
+        now = Time.now.to_i
+        min_timestamp = now - window
+        bucket_expression = "((COALESCE(rx_time, telemetry_time) / ?) * ?)"
+        select_clauses = [
+          "#{bucket_expression} AS bucket_start",
+          "COUNT(*) AS sample_count",
+          "MIN(COALESCE(rx_time, telemetry_time)) AS first_timestamp",
+          "MAX(COALESCE(rx_time, telemetry_time)) AS last_timestamp",
+        ]
+
+        TELEMETRY_AGGREGATE_COLUMNS.each do |column|
+          select_clauses << "AVG(#{column}) AS #{column}_avg"
+          select_clauses << "MIN(#{column}) AS #{column}_min"
+          select_clauses << "MAX(#{column}) AS #{column}_max"
+        end
+
+        sql = <<~SQL
+          SELECT
+            #{select_clauses.join(",\n            ")}
+          FROM telemetry
+          WHERE COALESCE(rx_time, telemetry_time) IS NOT NULL
+            AND COALESCE(rx_time, telemetry_time, 0) >= ?
+          GROUP BY bucket_start
+          ORDER BY bucket_start ASC
+          LIMIT ?
+        SQL
+        params = [bucket, bucket, min_timestamp, MAX_QUERY_LIMIT]
+        rows = db.execute(sql, params)
+        rows.map do |row|
+          bucket_start = coerce_integer(row["bucket_start"])
+          bucket_end = bucket_start ? bucket_start + bucket : nil
+          first_timestamp = coerce_integer(row["first_timestamp"])
+          last_timestamp = coerce_integer(row["last_timestamp"])
+
+          aggregates = {}
+          TELEMETRY_AGGREGATE_COLUMNS.each do |column|
+            avg = coerce_float(row["#{column}_avg"])
+            min_value = coerce_float(row["#{column}_min"])
+            max_value = coerce_float(row["#{column}_max"])
+
+            metrics = {}
+            metrics["avg"] = avg unless avg.nil?
+            metrics["min"] = min_value unless min_value.nil?
+            metrics["max"] = max_value unless max_value.nil?
+            aggregates[column] = metrics unless metrics.empty?
+          end
+
+          bucket_response = {
+            "bucket_start" => bucket_start,
+            "bucket_start_iso" => bucket_start ? Time.at(bucket_start).utc.iso8601 : nil,
+            "bucket_end" => bucket_end,
+            "bucket_end_iso" => bucket_end ? Time.at(bucket_end).utc.iso8601 : nil,
+            "bucket_seconds" => bucket,
+            "sample_count" => coerce_integer(row["sample_count"]),
+            "first_timestamp" => first_timestamp,
+            "first_timestamp_iso" => first_timestamp ? Time.at(first_timestamp).utc.iso8601 : nil,
+            "last_timestamp" => last_timestamp,
+            "last_timestamp_iso" => last_timestamp ? Time.at(last_timestamp).utc.iso8601 : nil,
+            "aggregates" => aggregates,
+          }
+          bucket_response["timestamp"] = bucket_start if bucket_start
+          bucket_response["timestamp_iso"] = bucket_response["bucket_start_iso"] if bucket_response["bucket_start_iso"]
+          compact_api_row(bucket_response)
+        end
+      ensure
+        db&.close
+      end
+
       def query_traces(limit, node_ref: nil)
         limit = coerce_query_limit(limit)
         db = open_database(readonly: true)

web/lib/potato_mesh/application/routes/api.rb

@@ -127,6 +127,37 @@ module PotatoMesh
             query_telemetry(limit).to_json
           end
 
+          app.get "/api/telemetry/aggregated" do
+            content_type :json
+            default_window = PotatoMesh::App::Queries::DEFAULT_TELEMETRY_WINDOW_SECONDS
+            default_bucket = PotatoMesh::App::Queries::DEFAULT_TELEMETRY_BUCKET_SECONDS
+
+            window_seconds = if params.key?("windowSeconds")
+                coerce_integer(params["windowSeconds"])
+              else
+                default_window
+              end
+            bucket_seconds = if params.key?("bucketSeconds")
+                coerce_integer(params["bucketSeconds"])
+              else
+                default_bucket
+              end
+
+            if window_seconds.nil? || window_seconds <= 0
+              halt 400, { error: "windowSeconds must be positive" }.to_json
+            end
+            if bucket_seconds.nil? || bucket_seconds <= 0
+              halt 400, { error: "bucketSeconds must be positive" }.to_json
+            end
+
+            bucket_count = (window_seconds.to_f / bucket_seconds).ceil
+            if bucket_count > PotatoMesh::App::Queries::MAX_QUERY_LIMIT
+              halt 400, { error: "bucketSeconds too small for requested window" }.to_json
+            end
+
+            query_telemetry_buckets(window_seconds: window_seconds, bucket_seconds: bucket_seconds).to_json
+          end
+
           app.get "/api/telemetry/:id" do
             content_type :json
             node_ref = string_or_nil(params["id"])

web/public/assets/js/app/__tests__/charts-page.test.js

@@ -33,23 +33,35 @@ function createResponse(status, body) {
   };
 }
 
-test('fetchAggregatedTelemetry requests the latest 1000 telemetry entries', async () => {
+test('fetchAggregatedTelemetry requests aggregated telemetry buckets and normalizes stats', async () => {
   const requests = [];
   const fetchImpl = async url => {
     requests.push(url);
-    return createResponse(200, [{ rx_time: 1_700_000_000, node_id: '!demo' }]);
+    return createResponse(200, [
+      {
+        bucket_start: 1_700_000_000,
+        bucket_seconds: 300,
+        sample_count: 4,
+        aggregates: {
+          battery_level: { avg: 85.2, min: 80, max: 90 },
+          temperature: { avg: 22.5 },
+        },
+      },
+    ]);
   };
-  const snapshots = await fetchAggregatedTelemetry({ fetchImpl });
+  const snapshots = await fetchAggregatedTelemetry({ fetchImpl, windowMs: 604_800_000, bucketSeconds: 900 });
   assert.equal(requests.length, 1);
-  assert.equal(requests[0], '/api/telemetry?limit=1000');
+  assert.equal(requests[0], '/api/telemetry/aggregated?windowSeconds=604800&bucketSeconds=900');
   assert.equal(Array.isArray(snapshots), true);
-  assert.equal(snapshots[0].node_id, '!demo');
+  assert.equal(snapshots[0].battery_level, 85.2);
+  assert.equal(snapshots[0].battery_level_max, 90);
+  assert.equal(snapshots[0].temperature, 22.5);
 });
 
 test('fetchAggregatedTelemetry validates fetch availability and response codes', async () => {
   await assert.rejects(() => fetchAggregatedTelemetry({ fetchImpl: null }), /fetch implementation/i);
   const fetchImpl = async () => createResponse(503, []);
-  await assert.rejects(() => fetchAggregatedTelemetry({ fetchImpl }), /Failed to fetch telemetry/);
+  await assert.rejects(() => fetchAggregatedTelemetry({ fetchImpl }), /Failed to fetch aggregated telemetry/);
 });
 
 test('initializeChartsPage renders the telemetry charts when snapshots are available', async () => {
@@ -59,7 +71,14 @@ test('initializeChartsPage renders the telemetry charts when snapshots are avail
       return id === 'chartsPage' ? container : null;
     },
   };
-  const fetchImpl = async () => createResponse(200, [{ rx_time: 1_700_000_000, temperature: 22.5 }]);
+  const fetchImpl = async () => createResponse(200, [
+    {
+      bucket_start: 1_700_000_000,
+      bucket_seconds: 300,
+      sample_count: 3,
+      aggregates: { temperature: { avg: 22.5 } },
+    },
+  ]);
   let receivedOptions = null;
   const renderCharts = (node, options) => {
     receivedOptions = options;
@@ -69,8 +88,10 @@ test('initializeChartsPage renders the telemetry charts when snapshots are avail
   assert.equal(result, true);
   assert.equal(container.innerHTML.includes('node-detail__charts'), true);
   assert.ok(receivedOptions);
-  assert.equal(receivedOptions.chartOptions.windowMs, 86_400_000);
+  assert.equal(receivedOptions.chartOptions.windowMs, 604_800_000);
   assert.equal(typeof receivedOptions.chartOptions.lineReducer, 'function');
+  assert.equal(typeof receivedOptions.chartOptions.xAxisTickBuilder, 'function');
+  assert.equal(typeof receivedOptions.chartOptions.xAxisTickFormatter, 'function');
   const average = receivedOptions.chartOptions.lineReducer(
     [
       { timestamp: 0, value: 0 },
@@ -79,6 +100,12 @@ test('initializeChartsPage renders the telemetry charts when snapshots are avail
     ],
   );
   assert.equal(Array.isArray(average), true);
+  const nowMs = Date.UTC(2025, 8, 16); // September 16, 2025
+  const ticks = receivedOptions.chartOptions.xAxisTickBuilder(nowMs, 604_800_000);
+  assert.equal(Array.isArray(ticks), true);
+  assert.equal(new Date(ticks[0]).getHours(), 0);
+  const label = receivedOptions.chartOptions.xAxisTickFormatter(ticks[0]);
+  assert.equal(/^\d{2}$/.test(label), true);
 });
 
 test('initializeChartsPage shows an error message when fetching fails', async () => {
@@ -122,7 +149,12 @@ test('initializeChartsPage shows a status when rendering produces no markup', as
       return container;
     },
   };
-  const fetchImpl = async () => createResponse(200, [{ rx_time: 1_700_000_000 }]);
+  const fetchImpl = async () => createResponse(200, [
+    {
+      bucket_start: 1_700_000_000,
+      aggregates: { voltage: { avg: 3.9 } },
+    },
+  ]);
   const renderCharts = () => '';
   const result = await initializeChartsPage({ document: documentStub, fetchImpl, renderCharts });
   assert.equal(result, true);

web/public/assets/js/app/charts-page.js

@@ -16,9 +16,10 @@
 
 import { renderTelemetryCharts } from './node-page.js';
 
-const TELEMETRY_AGGREGATE_LIMIT = 1000;
+const TELEMETRY_BUCKET_SECONDS = 60 * 60;
 const HOUR_MS = 60 * 60 * 1000;
 const DAY_MS = 24 * HOUR_MS;
+const CHART_WINDOW_MS = 7 * DAY_MS;
 
 function escapeHtml(value) {
   return String(value ?? '')
@@ -40,22 +41,98 @@ function padTwo(value) {
   return num < 10 ? `0${Math.trunc(num)}` : String(Math.trunc(num));
 }
 
-function buildHourlyTickList(nowMs, windowMs = DAY_MS) {
+function buildMidnightTickList(nowMs, windowMs = CHART_WINDOW_MS) {
   const ticks = [];
-  const safeWindow = Number.isFinite(windowMs) && windowMs > 0 ? windowMs : DAY_MS;
+  const safeWindow = Number.isFinite(windowMs) && windowMs > 0 ? windowMs : CHART_WINDOW_MS;
   const domainStart = nowMs - safeWindow;
   const cursor = new Date(nowMs);
-  cursor.setMinutes(0, 0, 0);
-  for (let ts = cursor.getTime(); ts >= domainStart; ts -= HOUR_MS) {
+  cursor.setHours(0, 0, 0, 0);
+  for (let ts = cursor.getTime(); ts >= domainStart; ts -= DAY_MS) {
     ticks.push(ts);
   }
   return ticks.reverse();
 }
 
-function formatHourLabel(timestampMs) {
+function formatDayOfMonthLabel(timestampMs) {
   const date = new Date(timestampMs);
   if (Number.isNaN(date.getTime())) return '';
-  return padTwo(date.getHours());
+  return padTwo(date.getDate());
+}
+
+function normalizeAggregatedSnapshot(bucket) {
+  if (!bucket || typeof bucket !== 'object') {
+    return null;
+  }
+  const timestamp =
+    Number.isFinite(bucket.timestamp) ? bucket.timestamp
+      : Number.isFinite(bucket.bucket_start) ? bucket.bucket_start
+        : Number.isFinite(bucket.bucketStart) ? bucket.bucketStart
+          : null;
+  if (!Number.isFinite(timestamp)) {
+    return null;
+  }
+  const bucketSecondsCandidate =
+    Number.isFinite(bucket.bucket_seconds) ? bucket.bucket_seconds
+      : Number.isFinite(bucket.bucketSeconds) ? bucket.bucketSeconds
+        : TELEMETRY_BUCKET_SECONDS;
+  const bucketSeconds = bucketSecondsCandidate > 0 ? bucketSecondsCandidate : TELEMETRY_BUCKET_SECONDS;
+  const timestampIso =
+    typeof bucket.timestamp_iso === 'string' ? bucket.timestamp_iso
+      : typeof bucket.timestampIso === 'string' ? bucket.timestampIso
+        : typeof bucket.bucket_start_iso === 'string' ? bucket.bucket_start_iso
+          : typeof bucket.bucketStartIso === 'string' ? bucket.bucketStartIso
+            : null;
+  const snapshot = {
+    rx_time: Number.isFinite(bucket.rx_time) ? bucket.rx_time : timestamp,
+    telemetry_time: Number.isFinite(bucket.telemetry_time) ? bucket.telemetry_time : timestamp,
+    rx_iso: typeof bucket.rx_iso === 'string' ? bucket.rx_iso : timestampIso,
+    telemetry_time_iso:
+      typeof bucket.telemetry_time_iso === 'string'
+        ? bucket.telemetry_time_iso
+        : timestampIso,
+    timestamp,
+    timestampIso,
+    bucket_seconds: bucketSeconds,
+    bucket_start: Number.isFinite(bucket.bucket_start) ? bucket.bucket_start : timestamp,
+    bucket_start_iso:
+      typeof bucket.bucket_start_iso === 'string'
+        ? bucket.bucket_start_iso
+        : timestampIso,
+    bucket_end: Number.isFinite(bucket.bucket_end) ? bucket.bucket_end : timestamp + bucketSeconds,
+    bucket_end_iso: typeof bucket.bucket_end_iso === 'string' ? bucket.bucket_end_iso : null,
+    sample_count:
+      Number.isFinite(bucket.sample_count)
+        ? bucket.sample_count
+        : Number.isFinite(bucket.sampleCount)
+          ? bucket.sampleCount
+          : null,
+  };
+  const aggregates = bucket.aggregates && typeof bucket.aggregates === 'object' ? bucket.aggregates : null;
+  if (aggregates) {
+    snapshot.aggregates = aggregates;
+    for (const [field, stats] of Object.entries(aggregates)) {
+      if (!stats || typeof stats !== 'object') {
+        continue;
+      }
+      const avg = Number.isFinite(stats.avg) ? stats.avg : null;
+      const min = Number.isFinite(stats.min) ? stats.min : null;
+      const max = Number.isFinite(stats.max) ? stats.max : null;
+      if (avg != null) {
+        snapshot[field] = avg;
+        snapshot[`${field}_avg`] = avg;
+      }
+      if (min != null) {
+        snapshot[`${field}_min`] = min;
+      }
+      if (max != null) {
+        snapshot[`${field}_max`] = max;
+      }
+    }
+  }
+  if (!snapshot.bucket_end_iso && Number.isFinite(snapshot.bucket_end)) {
+    snapshot.bucket_end_iso = new Date(snapshot.bucket_end * 1000).toISOString();
+  }
+  return snapshot;
 }
 
 export function buildMovingAverageSeries(points, windowMs = HOUR_MS) {
@@ -86,18 +163,34 @@ export function buildMovingAverageSeries(points, windowMs = HOUR_MS) {
   return averages;
 }
 
-export async function fetchAggregatedTelemetry({ fetchImpl = globalThis.fetch, limit = TELEMETRY_AGGREGATE_LIMIT } = {}) {
+export async function fetchAggregatedTelemetry({
+  fetchImpl = globalThis.fetch,
+  windowMs = CHART_WINDOW_MS,
+  bucketSeconds = TELEMETRY_BUCKET_SECONDS,
+} = {}) {
   const fetchFn = typeof fetchImpl === 'function' ? fetchImpl : null;
   if (!fetchFn) {
     throw new TypeError('A fetch implementation is required to load telemetry');
   }
-  const effectiveLimit = Number.isFinite(limit) && limit > 0 ? Math.floor(limit) : TELEMETRY_AGGREGATE_LIMIT;
-  const response = await fetchFn(`/api/telemetry?limit=${effectiveLimit}`, { cache: 'no-store' });
+  const windowSecondsCandidate =
+    Number.isFinite(windowMs) ? Math.floor(windowMs / 1000) : Math.floor(CHART_WINDOW_MS / 1000);
+  const windowSeconds = windowSecondsCandidate > 0 ? windowSecondsCandidate : Math.floor(CHART_WINDOW_MS / 1000);
+  const bucketSecondsCandidate = Number.isFinite(bucketSeconds) ? Math.floor(bucketSeconds) : TELEMETRY_BUCKET_SECONDS;
+  const bucketSecondsSafe = bucketSecondsCandidate > 0 ? bucketSecondsCandidate : TELEMETRY_BUCKET_SECONDS;
+  const response = await fetchFn(
+    `/api/telemetry/aggregated?windowSeconds=${windowSeconds}&bucketSeconds=${bucketSecondsSafe}`,
+    { cache: 'no-store' },
+  );
   if (!response.ok) {
-    throw new Error(`Failed to fetch telemetry (HTTP ${response.status})`);
+    throw new Error(`Failed to fetch aggregated telemetry (HTTP ${response.status})`);
   }
   const payload = await response.json();
-  return Array.isArray(payload) ? payload : [];
+  if (!Array.isArray(payload)) {
+    return [];
+  }
+  return payload
+    .map(bucket => normalizeAggregatedSnapshot(bucket))
+    .filter(snapshot => snapshot != null);
 }
 
 export async function initializeChartsPage(options = {}) {
@@ -113,12 +206,13 @@ export async function initializeChartsPage(options = {}) {
 
   const renderCharts = typeof options.renderCharts === 'function' ? options.renderCharts : renderTelemetryCharts;
   const fetchImpl = options.fetchImpl ?? globalThis.fetch;
-  const limit = options.limit ?? TELEMETRY_AGGREGATE_LIMIT;
+  const bucketSeconds = options.bucketSeconds ?? TELEMETRY_BUCKET_SECONDS;
+  const windowMs = options.windowMs ?? CHART_WINDOW_MS;
 
   container.innerHTML = renderStatus('Loading aggregated telemetry charts…');
 
   try {
-    const snapshots = await fetchAggregatedTelemetry({ fetchImpl, limit });
+    const snapshots = await fetchAggregatedTelemetry({ fetchImpl, bucketSeconds, windowMs });
     if (!Array.isArray(snapshots) || snapshots.length === 0) {
       container.innerHTML = renderStatus('Telemetry snapshots are unavailable.');
       return true;
@@ -127,10 +221,10 @@ export async function initializeChartsPage(options = {}) {
     const chartsHtml = renderCharts(node, {
       nowMs: Date.now(),
       chartOptions: {
-        windowMs: DAY_MS,
-        timeRangeLabel: 'Last 24 hours',
-        xAxisTickBuilder: buildHourlyTickList,
-        xAxisTickFormatter: formatHourLabel,
+        windowMs,
+        timeRangeLabel: 'Last 7 days',
+        xAxisTickBuilder: buildMidnightTickList,
+        xAxisTickFormatter: formatDayOfMonthLabel,
         lineReducer: points => buildMovingAverageSeries(points, HOUR_MS),
       },
     });

web/spec/app_spec.rb

@@ -4350,6 +4350,54 @@ RSpec.describe "Potato Mesh Sinatra app" do
     end
   end
 
+  describe "GET /api/telemetry/aggregated" do
+    it "returns aggregated telemetry buckets for the requested interval" do
+      post "/api/telemetry", telemetry_fixture.to_json, auth_headers
+      expect(last_response).to be_ok
+
+      get "/api/telemetry/aggregated?windowSeconds=86400&bucketSeconds=300"
+
+      expect(last_response).to be_ok
+      buckets = JSON.parse(last_response.body)
+      expect(buckets).not_to be_empty
+      a_bucket = buckets.first
+      expect(a_bucket["bucket_seconds"]).to eq(300)
+      expect(a_bucket["sample_count"]).to be >= 1
+      expect(a_bucket["bucket_start"]).to be_a(Integer)
+      expect(a_bucket["bucket_end"]).to be_a(Integer)
+      expect(a_bucket["aggregates"]).to be_a(Hash)
+      expect(a_bucket["aggregates"]).to have_key("battery_level")
+      expect(a_bucket["aggregates"]["battery_level"]).to include("avg")
+      expect(a_bucket).not_to have_key("device_metrics")
+    end
+
+    it "applies default window and bucket sizes when parameters are omitted" do
+      post "/api/telemetry", telemetry_fixture.to_json, auth_headers
+      expect(last_response).to be_ok
+
+      get "/api/telemetry/aggregated"
+
+      expect(last_response).to be_ok
+      buckets = JSON.parse(last_response.body)
+      expect(buckets.length).to be >= 1
+      expect(buckets.first["bucket_seconds"]).to eq(PotatoMesh::App::Queries::DEFAULT_TELEMETRY_BUCKET_SECONDS)
+    end
+
+    it "rejects invalid bucket and window parameters" do
+      get "/api/telemetry/aggregated?windowSeconds=0&bucketSeconds=300"
+      expect(last_response.status).to eq(400)
+      expect(JSON.parse(last_response.body)).to eq("error" => "windowSeconds must be positive")
+
+      get "/api/telemetry/aggregated?windowSeconds=86400&bucketSeconds=0"
+      expect(last_response.status).to eq(400)
+      expect(JSON.parse(last_response.body)).to eq("error" => "bucketSeconds must be positive")
+
+      get "/api/telemetry/aggregated?windowSeconds=86400&bucketSeconds=1"
+      expect(last_response.status).to eq(400)
+      expect(JSON.parse(last_response.body)).to eq("error" => "bucketSeconds too small for requested window")
+    end
+  end
+
   describe "GET /api/traces" do
     it "returns stored traces ordered by receive time" do
       clear_database