Merge PR #31: feat(backend): read history from InfluxDB 1.x

backend/src/main/java/home/backend/adapters/influx/InfluxHistoryRepository.java

@@ -0,0 +1,78 @@
+package home.backend.adapters.influx;
+
+import home.backend.application.HistoryRepository;
+import home.backend.domain.HistoryView.Point;
+import home.backend.domain.Series;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URLEncoder;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+
+/** History from InfluxDB 1.x over its HTTP {@code /query} endpoint (ADR-005, ADR-008). */
+public final class InfluxHistoryRepository implements HistoryRepository {
+
+    private final InfluxSettings settings;
+    private final HttpClient http;
+
+    public InfluxHistoryRepository(InfluxSettings settings) {
+        this.settings = settings;
+        this.http = HttpClient.newBuilder().connectTimeout(settings.requestTimeout()).build();
+    }
+
+    @Override
+    public List<Point> history(Series series, Instant from, Instant to, int maxPoints) throws IOException {
+        if (maxPoints < 1 || !from.isBefore(to)) {
+            throw new IllegalArgumentException("need maxPoints >= 1 and from < to");
+        }
+        long stepMinutes = stepMinutes(from, to, maxPoints);
+        Map<Long, Double> summed = new TreeMap<>();
+        for (Source source : SeriesSources.of(series)) {
+            String query = source.query(from.toEpochMilli(), to.toEpochMilli(), stepMinutes);
+            // A part without data is absent, not zero, so a bucket is kept if any part has it.
+            InfluxResponse.parse(fetch(source.database(), query)).forEach((t, v) -> summed.merge(t, v, Double::sum));
+        }
+        return summed.entrySet().stream()
+                .map(e -> new Point(Instant.ofEpochMilli(e.getKey()), e.getValue()))
+                .toList();
+    }
+
+    private static long stepMinutes(Instant from, Instant to, int maxPoints) {
+        long stepMillis = Math.ceilDiv(Duration.between(from, to).toMillis(), maxPoints);
+        return Math.max(1, Math.ceilDiv(stepMillis, Duration.ofMinutes(1).toMillis()));
+    }
+
+    private String fetch(String database, String query) throws IOException {
+        HttpRequest request = HttpRequest.newBuilder(queryUri(database, query))
+                .timeout(settings.requestTimeout())
+                .GET()
+                .build();
+        try {
+            HttpResponse<String> response = http.send(request, HttpResponse.BodyHandlers.ofString());
+            if (response.statusCode() != 200) {
+                throw new IOException("InfluxDB answered HTTP " + response.statusCode());
+            }
+            return response.body();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new IOException("InfluxDB request interrupted", e);
+        }
+    }
+
+    private URI queryUri(String database, String query) {
+        String base = settings.baseUri().toString().replaceAll("/+$", "");
+        return URI.create(base + "/query?db=" + encode(database) + "&epoch=ms&q=" + encode(query));
+    }
+
+    private static String encode(String value) {
+        return URLEncoder.encode(value, StandardCharsets.UTF_8);
+    }
+}

backend/src/main/java/home/backend/adapters/influx/InfluxResponse.java

@@ -0,0 +1,50 @@
+package home.backend.adapters.influx;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.TreeMap;
+
+/** The {@code time, mean} rows of an InfluxDB 1.x query response. */
+final class InfluxResponse {
+
+    private static final ObjectMapper JSON = new ObjectMapper();
+
+    private InfluxResponse() {
+    }
+
+    /** Mean per bucket start in epoch milliseconds, ascending; empty when there is no data. */
+    static Map<Long, Double> parse(String body) throws IOException {
+        JsonNode root = read(body);
+        failOnError(root);
+        Map<Long, Double> means = new TreeMap<>();
+        for (JsonNode result : root.path("results")) {
+            failOnError(result);
+            for (JsonNode series : result.path("series")) {
+                for (JsonNode row : series.path("values")) {
+                    if (row.size() >= 2 && row.get(1).isNumber()) {
+                        means.put(row.get(0).asLong(), row.get(1).asDouble());
+                    }
+                }
+            }
+        }
+        return means;
+    }
+
+    private static JsonNode read(String body) throws IOException {
+        try {
+            return JSON.readTree(body);
+        } catch (JsonProcessingException e) {
+            throw new IOException("InfluxDB returned invalid JSON: " + e.getOriginalMessage(), e);
+        }
+    }
+
+    private static void failOnError(JsonNode node) throws IOException {
+        if (node.hasNonNull("error")) {
+            throw new IOException("InfluxDB query failed: " + node.get("error").asText());
+        }
+    }
+}

backend/src/main/java/home/backend/adapters/influx/InfluxSettings.java

@@ -0,0 +1,7 @@
+package home.backend.adapters.influx;
+
+import java.net.URI;
+import java.time.Duration;
+
+public record InfluxSettings(URI baseUri, Duration requestTimeout) {
+}

backend/src/main/java/home/backend/adapters/influx/SeriesSources.java

@@ -0,0 +1,44 @@
+package home.backend.adapters.influx;
+
+import home.backend.domain.Series;
+
+import java.util.List;
+import java.util.Map;
+
+/** Where each series lives in InfluxDB. A fixed table: no query is built from user input. */
+final class SeriesSources {
+
+    private static final String LEGACY = "Dom_Pompa";
+    private static final String ELECTRICITY = "electricity";
+
+    private static final Map<Series, List<Source>> SOURCES = Map.ofEntries(
+            Map.entry(Series.OUTDOOR_TEMP, List.of(pompa("External temperature"))),
+            Map.entry(Series.INDOOR_TEMP, List.of(pompa("Temp wew"))),
+            Map.entry(Series.CO_BUFFER_TEMP, List.of(pompa("Temp co"))),
+            Map.entry(Series.CWU_TEMP, List.of(pompa("Temp cwu"))),
+            Map.entry(Series.GROUND_IN_TEMP, List.of(pompa("Temp in"))),
+            Map.entry(Series.GROUND_OUT_TEMP, List.of(pompa("Temp out"))),
+            Map.entry(Series.PUMP_OUTLET_TEMP, List.of(pompa("Temp pc out"))),
+            Map.entry(Series.GRID_POWER, List.of(electricity("PRAD2", "total_power_active"))),
+            Map.entry(Series.PV_POWER, List.of(
+                    electricity("deye", "pv1_power_w"),
+                    electricity("deye", "pv2_power_w"),
+                    electricity("fronius", "pv_power_w"))),
+            Map.entry(Series.LOAD_POWER, List.of(electricity("deye", "load_power_w"))),
+            Map.entry(Series.BATTERY_SOC, List.of(electricity("deye", "battery_soc_pct"))));
+
+    private SeriesSources() {
+    }
+
+    static List<Source> of(Series series) {
+        return SOURCES.get(series);
+    }
+
+    private static Source pompa(String field) {
+        return new Source(LEGACY, "POMPA", field);
+    }
+
+    private static Source electricity(String measurement, String field) {
+        return new Source(ELECTRICITY, measurement, field);
+    }
+}

backend/src/main/java/home/backend/adapters/influx/Source.java

@@ -0,0 +1,15 @@
+package home.backend.adapters.influx;
+
+/** One field of one measurement; the parts of a series are summed. */
+record Source(String database, String measurement, String field) {
+
+    String query(long fromMs, long toMs, long stepMinutes) {
+        return "SELECT mean(" + quote(field) + ") FROM " + quote(measurement)
+                + " WHERE time >= " + fromMs + "ms AND time < " + toMs + "ms"
+                + " GROUP BY time(" + stepMinutes + "m) fill(none)";
+    }
+
+    private static String quote(String identifier) {
+        return '"' + identifier.replace("\\", "\\\\").replace("\"", "\\\"") + '"';
+    }
+}

backend/src/test/java/home/backend/adapters/influx/InfluxHistoryRepositoryTest.java

@@ -0,0 +1,208 @@
+package home.backend.adapters.influx;
+
+import com.sun.net.httpserver.HttpServer;
+import home.backend.domain.HistoryView.Point;
+import home.backend.domain.Series;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class InfluxHistoryRepositoryTest {
+
+    private static final Instant FROM = Instant.parse("2026-10-24T00:00:00Z");
+    private static final Instant TO = Instant.parse("2026-10-24T06:00:00Z");
+    private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+    private HttpServer server;
+    private final List<Map<String, String>> requests = new ArrayList<>();
+    private final Map<String, String> bodyByField = new HashMap<>();
+    private int status = 200;
+    private String defaultBody = "{\"results\":[{\"statement_id\":0}]}";
+
+    @BeforeEach
+    void startServer() throws IOException {
+        server = HttpServer.create(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 0);
+        server.createContext("/query", exchange -> {
+            Map<String, String> params = params(exchange.getRequestURI().getRawQuery());
+            requests.add(params);
+            String body = bodyByField.entrySet().stream()
+                    .filter(e -> params.get("q").contains("\"" + e.getKey() + "\""))
+                    .map(Map.Entry::getValue).findFirst().orElse(defaultBody);
+            byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
+            exchange.sendResponseHeaders(status, bytes.length);
+            exchange.getResponseBody().write(bytes);
+            exchange.close();
+        });
+        server.start();
+    }
+
+    @AfterEach
+    void stopServer() {
+        server.stop(0);
+    }
+
+    @Test
+    void asksTheLegacyDatabaseForATemperatureWithQuotedFieldName() throws IOException {
+        defaultBody = fixture("influx-query-cwu-6h.json");
+
+        List<Point> points = repository(TIMEOUT).history(Series.CWU_TEMP, FROM, TO, 360);
+
+        assertEquals(1, requests.size());
+        assertEquals("Dom_Pompa", requests.get(0).get("db"));
+        assertEquals("ms", requests.get(0).get("epoch"));
+        assertEquals("SELECT mean(\"Temp cwu\") FROM \"POMPA\" WHERE time >= 1792800000000ms"
+                + " AND time < 1792821600000ms GROUP BY time(1m) fill(none)", requests.get(0).get("q"));
+        assertEquals(7, points.size());
+        assertEquals(new Point(Instant.ofEpochMilli(1792771200000L), 43.76666666666667), points.get(0));
+    }
+
+    @Test
+    void asksTheElectricityDatabaseForGridPower() throws IOException {
+        repository(TIMEOUT).history(Series.GRID_POWER, FROM, TO, 72);
+
+        assertEquals("electricity", requests.get(0).get("db"));
+        assertEquals("SELECT mean(\"total_power_active\") FROM \"PRAD2\" WHERE time >= 1792800000000ms"
+                + " AND time < 1792821600000ms GROUP BY time(5m) fill(none)", requests.get(0).get("q"));
+    }
+
+    @Test
+    void roundsTheStepUpToWholeMinutes() throws IOException {
+        repository(TIMEOUT).history(Series.LOAD_POWER, FROM, TO, 100);
+
+        assertTrue(requests.get(0).get("q").contains("GROUP BY time(4m)"), requests.get(0).get("q"));
+    }
+
+    @Test
+    void sumsPvPowerAcrossMeasurementsAndKeepsBucketsWithOnePart() throws IOException {
+        bodyByField.put("pv1_power_w", series(1000, 100.0, 2000, 200.0));
+        bodyByField.put("pv2_power_w", series(1000, 50.0));
+        bodyByField.put("pv_power_w", series(2000, 5.0, 3000, 7.0));
+
+        List<Point> points = repository(TIMEOUT).history(Series.PV_POWER, FROM, TO, 72);
+
+        assertEquals(3, requests.size());
+        assertEquals(List.of(
+                new Point(Instant.ofEpochMilli(1000), 150.0),
+                new Point(Instant.ofEpochMilli(2000), 205.0),
+                new Point(Instant.ofEpochMilli(3000), 7.0)), points);
+    }
+
+    @Test
+    void sortsPointsByTime() throws IOException {
+        defaultBody = series(3000, 3.0, 1000, 1.0, 2000, 2.0);
+
+        List<Point> points = repository(TIMEOUT).history(Series.BATTERY_SOC, FROM, TO, 72);
+
+        assertEquals(List.of(1000L, 2000L, 3000L), points.stream().map(p -> p.t().toEpochMilli()).toList());
+    }
+
+    @Test
+    void noDataGivesAnEmptyList() throws IOException {
+        defaultBody = fixture("influx-query-empty.json");
+
+        assertEquals(List.of(), repository(TIMEOUT).history(Series.PV_POWER, FROM, TO, 72));
+    }
+
+    @Test
+    void topLevelErrorIsAnIOException() throws IOException {
+        defaultBody = fixture("influx-query-error.json");
+
+        IOException e = assertThrows(IOException.class,
+                () -> repository(TIMEOUT).history(Series.CWU_TEMP, FROM, TO, 72));
+        assertTrue(e.getMessage().contains("error parsing query"), e.getMessage());
+    }
+
+    @Test
+    void perResultErrorIsAnIOException() {
+        defaultBody = "{\"results\":[{\"statement_id\":0,\"error\":\"database not found: electricity\"}]}";
+
+        IOException e = assertThrows(IOException.class,
+                () -> repository(TIMEOUT).history(Series.GRID_POWER, FROM, TO, 72));
+        assertTrue(e.getMessage().contains("database not found"), e.getMessage());
+    }
+
+    @Test
+    void httpErrorIsAnIOException() {
+        status = 503;
+
+        IOException e = assertThrows(IOException.class,
+                () -> repository(TIMEOUT).history(Series.CWU_TEMP, FROM, TO, 72));
+        assertTrue(e.getMessage().contains("503"), e.getMessage());
+    }
+
+    @Test
+    void badJsonIsAnIOException() {
+        defaultBody = "<html>not json";
+
+        assertThrows(IOException.class, () -> repository(TIMEOUT).history(Series.CWU_TEMP, FROM, TO, 72));
+    }
+
+    @Test
+    void slowServerTimesOut() {
+        server.removeContext("/query");
+        server.createContext("/query", exchange -> {
+            try {
+                Thread.sleep(1000);
+            } catch (InterruptedException ignored) {
+                Thread.currentThread().interrupt();
+            }
+            exchange.close();
+        });
+
+        assertThrows(IOException.class,
+                () -> repository(Duration.ofMillis(200)).history(Series.CWU_TEMP, FROM, TO, 72));
+    }
+
+    @Test
+    void rejectsAnEmptyRange() {
+        assertThrows(IllegalArgumentException.class,
+                () -> repository(TIMEOUT).history(Series.CWU_TEMP, FROM, FROM, 72));
+    }
+
+    private InfluxHistoryRepository repository(Duration timeout) {
+        URI base = URI.create("http://127.0.0.1:" + server.getAddress().getPort());
+        return new InfluxHistoryRepository(new InfluxSettings(base, timeout));
+    }
+
+    private static String series(Object... timeAndValue) {
+        StringBuilder rows = new StringBuilder();
+        for (int i = 0; i < timeAndValue.length; i += 2) {
+            rows.append(i == 0 ? "" : ",").append('[').append(timeAndValue[i]).append(',')
+                    .append(timeAndValue[i + 1]).append(']');
+        }
+        return "{\"results\":[{\"statement_id\":0,\"series\":[{\"name\":\"m\",\"columns\":[\"time\",\"mean\"],"
+                + "\"values\":[" + rows + "]}]}]}";
+    }
+
+    private static String fixture(String name) throws IOException {
+        return Files.readString(Path.of("src/test/resources/influx", name));
+    }
+
+    private static Map<String, String> params(String rawQuery) {
+        Map<String, String> params = new HashMap<>();
+        for (String pair : rawQuery.split("&")) {
+            String[] kv = pair.split("=", 2);
+            params.put(kv[0], URLDecoder.decode(kv[1], StandardCharsets.UTF_8));
+        }
+        return params;
+    }
+}

backend/src/test/resources/influx/influx-query-cwu-6h.json

@@ -0,0 +1 @@
+{"results":[{"statement_id":0,"series":[{"name":"POMPA","columns":["time","mean"],"values":[[1792771200000,43.76666666666667],[1792774800000,42.40000000000001],[1792778400000,26.733333333333334],[1792782000000,19.15833333333333],[1792785600000,13.799999999999997],[1792789200000,13.58333333333333],[1792792800000,13.65]]}]}]}

backend/src/test/resources/influx/influx-query-empty.json

@@ -0,0 +1 @@
+{"results":[{"statement_id":0}]}

backend/src/test/resources/influx/influx-query-error.json

@@ -0,0 +1 @@
+{"error":"error parsing query: found SELEC, expected SELECT, DELETE, SHOW, CREATE, DROP, EXPLAIN, GRANT, REVOKE, ALTER, SET, KILL at line 1, char 1"}