Add local Mosquitto/TimescaleDB/Bento ingestion sandbox with Bento unit tests

.gitignore

@@ -0,0 +1,3 @@
+# Bento runtime state: dead-letter file + dedupe cache, regenerated locally, never commit
+bento/failed/*
+!bento/failed/.gitkeep

Makefile

@@ -0,0 +1,13 @@
+BENTO_IMAGE := ghcr.io/warpstreamlabs/bento:1.20.0
+
+# On Windows/Git Bash, MSYS rewrites the container-side "/bento.yaml" path
+# into a Windows path before it reaches docker. Harmless no-op on Linux/macOS.
+export MSYS_NO_PATHCONV := 1
+
+.PHONY: test-bento lint
+
+test-bento:
+	docker run --rm -v "$(CURDIR)/bento/fertloops.yaml:/bento.yaml" $(BENTO_IMAGE) test /bento.yaml
+
+lint:
+	docker run --rm -v "$(CURDIR)/bento/fertloops.yaml:/bento.yaml" $(BENTO_IMAGE) lint /bento.yaml

bento/fertloops.yaml

@@ -0,0 +1,200 @@
+input:
+  mqtt:
+    urls: ["tcp://mosquitto:1883"]
+    client_id: "bento-fertloops"
+    topics: ["fertloops/#"]
+    qos: 1
+    # A stable client_id + clean_session:false asks Mosquitto to keep a
+    # persistent session for this consumer, so readings published while
+    # Bento is down/restarting are queued by the broker instead of lost.
+    clean_session: false
+
+pipeline:
+  processors:
+    # Pulled out as a named resource (instead of inline here) so the unit
+    # tests below can target just the validate/map/reject logic, without
+    # dedupe's cache dependency in the way. See processor_resources below.
+    - resource: validate_and_map
+    - dedupe:
+        # MQTT QoS 1 is at-least-once: the same frame can be redelivered
+        # (reconnect, unacked replay). TimescaleDB can't enforce a unique
+        # index here without including the hypertable's ingestion-time
+        # partitioning column, which differs per redelivery -- so dedupe
+        # happens here instead, keyed on dev_id + the device's own Timestamp.
+        # File-backed (not memory): confirmed by testing that a plain memory
+        # cache forgets everything on a Bento restart -- republishing an
+        # already-inserted reading right after a restart produced a second
+        # row. Persisting to /bento_data survives restarts/redeploys.
+        # Hashed: the file cache uses the key as a filename, and the device's
+        # raw Timestamp contains "/" -- confirmed by testing that using it
+        # unhashed makes Bento try (and fail) to create subdirectories.
+        # drop_on_err:false is deliberate: confirmed by testing that the
+        # default (true) makes a cache hiccup (e.g. the directory below
+        # missing) silently DROP the reading instead of just skipping dedup.
+        # Losing a real sensor reading is worse than risking an occasional
+        # duplicate, so a broken cache must fail open, not closed.
+        cache: dedupe_cache
+        key: '${! (json("dev_id") + "|" + json("raw.Timestamp")).hash("sha256").encode("hex") }'
+        drop_on_err: false
+
+processor_resources:
+  - label: validate_and_map
+    processors:
+      - try:
+          - mapping: |
+              # fertloops/# is a wildcard: anything published under it reaches
+              # this pipeline, not just sensor readings. Bloblang tolerates
+              # missing nested fields (returns null) instead of erroring, so
+              # a wrong-shaped message would otherwise sail through mapping
+              # with a null dev_id -- confirmed by testing that this then
+              # blows up two steps later inside dedupe's key interpolation,
+              # logged only at DEBUG (invisible in production) with the
+              # message silently dropped: no DB row, no fallback file, no
+              # visible error. Reject it here instead, at the boundary,
+              # where it's loud. See the unit tests at the bottom of this
+              # file (bento test bento/fertloops.yaml) for the regression
+              # coverage on this exact failure mode.
+              root = if this.devID == null || this.Data == null {
+                  throw("not a reading frame: missing devID or Data")
+                } else { this }
+
+              root.time = now().ts_format("2006-01-02T15:04:05Z07:00", "UTC")
+              root.dev_id = this.devID
+              root.ph = this.Data.pH
+              root.ce = this.Data.CE
+              root.solar = this.Data.Solar
+              root.volume = this.Data.Volume
+              root.soil_temp = this.Data.THC.T
+              root.soil_humidity = this.Data.THC.H
+              root.soil_conduct = this.Data.THC.C
+              root.air_temp = this.Data.TH.T
+              root.air_humidity = this.Data.TH.H
+              root.err_adc = this.Data.Errors.ADC
+              root.err_pulses = this.Data.Errors.Pulses
+              root.err_i2c = this.Data.Errors.I2C
+              root.err_inverter = this.Data.Errors.Inverter
+              root.err_inverter_state = this.Data.Errors.Inverter_State
+              root.valve = this.Control.Valve
+              root.inv_on = this.Control.Inv.On
+              root.inv_freq = this.Control.Inv.Freq
+              root.raw = this
+      - catch:
+          # A frame that isn't valid JSON (truncated/corrupted UART read) or
+          # doesn't look like a reading lands here instead of nacking back to
+          # MQTT forever: log once and drop it, so one bad frame can't loop
+          # endlessly or bury good frames in the log.
+          - log:
+              level: ERROR
+              message: "dropping unparseable reading: ${! error() }"
+          - mapping: "root = deleted()"
+
+cache_resources:
+  - label: dedupe_cache
+    file:
+      # Bento does NOT create this directory itself -- confirmed by testing
+      # that a fresh subdirectory here fails every cache read/write with
+      # "no such file or directory" until it exists. Pointing at the mount's
+      # root (already created by the bind mount itself, see docker-compose.yml)
+      # avoids depending on a nested folder nobody remembers to create.
+      directory: /bento_data
+
+output:
+  fallback:
+    # Primary path: insert into TimescaleDB.
+    - sql_insert:
+        driver: postgres
+        dsn: "postgres://fertloops:fertloops@timescaledb:5432/fertloops?sslmode=disable"
+        table: readings
+        columns:
+          - time
+          - dev_id
+          - ph
+          - ce
+          - solar
+          - volume
+          - soil_temp
+          - soil_humidity
+          - soil_conduct
+          - air_temp
+          - air_humidity
+          - err_adc
+          - err_pulses
+          - err_i2c
+          - err_inverter
+          - err_inverter_state
+          - valve
+          - inv_on
+          - inv_freq
+          - raw
+        args_mapping: |
+          root = [
+            this.time, this.dev_id, this.ph, this.ce, this.solar, this.volume,
+            this.soil_temp, this.soil_humidity, this.soil_conduct, this.air_temp, this.air_humidity,
+            this.err_adc, this.err_pulses, this.err_i2c, this.err_inverter, this.err_inverter_state,
+            this.valve, this.inv_on, this.inv_freq, this.raw.format_json()
+          ]
+    # Fallback: if the DB insert fails (e.g. TimescaleDB briefly down), park
+    # the reading on disk instead of nacking forever, so it isn't lost or
+    # stuck retrying indefinitely.
+    - file:
+        path: /bento_data/failed_readings.jsonl
+        codec: lines
+
+logger:
+  level: INFO
+
+# Run with: bento test bento/fertloops.yaml
+# Targets /pipeline/processors/0 (the validate_and_map resource) on purpose,
+# skipping dedupe -- dedupe needs a real cache directory to behave (see
+# cache_resources above), and none of these cases are about deduplication.
+tests:
+  - name: valid reading frame maps every field
+    target_processors: '/pipeline/processors/0'
+    input_batch:
+      - content: '{"devID":"A4:CF:12:34:56:78","Timestamp":"27/07/2026 10:52:55","Control":{"Valve":56,"Inv":{"On":1,"Freq":50}},"Data":{"pH":7,"CE":12345.67,"Solar":845.65,"Volume":20,"THC":{"T":22.55,"H":57.89,"C":1.85},"TH":{"T":22.23,"H":64.25},"Errors":{"ADC":0,"Pulses":1,"I2C":0,"Inverter":0,"Inverter_State":0}}}'
+    output_batches:
+      -
+        - json_contains: |
+            {
+              "dev_id": "A4:CF:12:34:56:78",
+              "ph": 7, "ce": 12345.67, "solar": 845.65, "volume": 20,
+              "soil_temp": 22.55, "soil_humidity": 57.89, "soil_conduct": 1.85,
+              "air_temp": 22.23, "air_humidity": 64.25,
+              "err_adc": 0, "err_pulses": 1, "err_i2c": 0, "err_inverter": 0, "err_inverter_state": 0,
+              "valve": 56, "inv_on": 1, "inv_freq": 50,
+              "raw": {"devID": "A4:CF:12:34:56:78", "Timestamp": "27/07/2026 10:52:55"}
+            }
+
+  - name: message missing devID and Data is dropped, not silently lost
+    # Regression test: this used to sail through mapping with a null dev_id
+    # (Bloblang tolerates missing nested fields) and only blow up two steps
+    # later inside dedupe's key interpolation, logged at DEBUG (invisible in
+    # production), with the message vanishing -- no DB row, no fallback file.
+    target_processors: '/pipeline/processors/0'
+    input_batch:
+      - content: '{"hello":"world"}'
+    output_batches: []
+
+  - name: message missing only Data is dropped
+    target_processors: '/pipeline/processors/0'
+    input_batch:
+      - content: '{"devID":"A4:CF:12:34:56:78"}'
+    output_batches: []
+
+  - name: non-JSON payload (corrupted UART read) is dropped, not looped
+    target_processors: '/pipeline/processors/0'
+    input_batch:
+      - content: 'not valid json {{{'
+    output_batches: []
+
+  - name: literal null payload is dropped
+    target_processors: '/pipeline/processors/0'
+    input_batch:
+      - content: 'null'
+    output_batches: []
+
+  - name: empty payload is dropped
+    target_processors: '/pipeline/processors/0'
+    input_batch:
+      - content: ''
+    output_batches: []

docker-compose.yml

@@ -0,0 +1,59 @@
+services:
+  mosquitto:
+    # eclipse-mosquitto has no granular patch-version tag (2.1.2) on Docker
+    # Hub, only rolling "2" -- pinned by digest instead so this doesn't
+    # silently change later. Currently resolves to mosquitto 2.1.2.
+    image: eclipse-mosquitto@sha256:9cfdd46ad59f3e3e5f592f6baf57ab23e1ad00605509d0f5c1e9b179c5314d87
+    container_name: fertloops-mosquitto
+    restart: unless-stopped
+    ports:
+      - "1883:1883"
+    volumes:
+      - ./mosquitto/config:/mosquitto/config
+      - mosquitto-data:/mosquitto/data
+      - mosquitto-log:/mosquitto/log
+    healthcheck:
+      test: ["CMD-SHELL", "mosquitto_sub -t '$$SYS/#' -C 1 -i healthcheck -W 3 || exit 1"]
+      interval: 5s
+      timeout: 5s
+      retries: 5
+
+  timescaledb:
+    image: timescale/timescaledb:2.29.0-pg16
+    container_name: fertloops-timescaledb
+    restart: unless-stopped
+    environment:
+      POSTGRES_USER: fertloops
+      POSTGRES_PASSWORD: fertloops
+      POSTGRES_DB: fertloops
+    ports:
+      - "5433:5432"
+    volumes:
+      - ./timescaledb/init:/docker-entrypoint-initdb.d
+      - timescaledb-data:/var/lib/postgresql/data
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U fertloops -d fertloops"]
+      interval: 5s
+      timeout: 5s
+      retries: 5
+
+  bento:
+    image: ghcr.io/warpstreamlabs/bento:1.20.0
+    container_name: fertloops-bento
+    restart: unless-stopped
+    depends_on:
+      mosquitto:
+        condition: service_healthy
+      timescaledb:
+        condition: service_healthy
+    volumes:
+      - ./bento/fertloops.yaml:/bento.yaml
+      - ./bento/failed:/bento_data
+    command: ["-c", "/bento.yaml"]
+    ports:
+      - "4195:4195"
+
+volumes:
+  mosquitto-data:
+  mosquitto-log:
+  timescaledb-data:

docs/fertloops-propuesta.md

@@ -30,7 +30,7 @@ El objetivo principal del prototipo que se pretende construir es cerrar el bucle
 
 A nivel global, la agricultura consume aproximadamente el 85% del agua dulce disponible (FAO, 2022) debido a la mayor demanda de alimentos por el crecimiento constante de la población se hace necesario el uso de estrategias de riego de precisión y para garantizar la seguridad alimentaria y promover el ahorro de agua. El sistema tradicional de gestión del riego presenta problemas como una baja eficiencia en el uso del agua lo que se traduce en una productividad limitada. Además, las condiciones variables del entorno requieren un enfoque adaptativo, utilizando sistemas de riego de precisión que incorporen tecnologías como sensores para medir la conductividad eléctrica del suelo, la evapotranspiración de las plantas o el nivel de clorofilas entre otros. Aunque en un invernadero se puede controlar el ambiente, estos factores son más difíciles de gestionar en un cultivo al aire libre. Actualmente se está haciendo uso de nuevas tecnologías como los gemelos digitales para llevar a cabo un riego inteligente, así como la fertirrigación, donde los nutrientes se mezclan con el agua de riego. Esto hace necesario una programación óptima del uso de fertilizantes para administrar la dosis correcta en el momento en que se requiera. Para ello es imprescindible tener en cuenta un volumen alto de datos tanto ambientales, de características del suelo o datos fisiológicos del cultivo en cuestión.
 
-Ferloops, se basa en un sistema de fertirrigación de bucle cerrado para economizar el uso del agua y hacer una utilización eficiente de los fertilizantes de manera que se facilite el manejo de explotaciones agrarias de cultivos de interés agrícola. La Figura 1 presenta el sistema de control y dosificación de fertilizante que se pretende construir.
+Fertloops, se basa en un sistema de fertirrigación de bucle cerrado para economizar el uso del agua y hacer una utilización eficiente de los fertilizantes de manera que se facilite el manejo de explotaciones agrarias de cultivos de interés agrícola. La Figura 1 presenta el sistema de control y dosificación de fertilizante que se pretende construir.
 
 Fertloops no solo pretende mejorar la productividad, sino que también busca alinearse con los Objetivos del Desarrollo Sostenible (ODS) y con la Estrategia de Especialización Inteligente 2021-2027 RIS3 de la Junta de Castilla y León para la mejorar y economizar el uso de agua y fertilizantes. Uno de los beneficios del desarrollo de esta nueva agricultura sería la reducción de la lixiviación de nitratos al suelo, que es uno de los principales problemas de la agricultura intensiva. La implementación de este sistema de fertirrigación se podrá establecer un sistema de monitorización sólido que permita la captación de datos a tiempo real de las condiciones ambientales y de sustrato para, posteriormente, poder caracterizar el valor óptimo de esos parámetros en el desarrollo de cultivos.
 

mosquitto/config/mosquitto.conf

@@ -0,0 +1,16 @@
+listener 1883
+allow_anonymous true
+
+persistence true
+persistence_location /mosquitto/data/
+log_dest file /mosquitto/log/mosquitto.log
+log_dest stdout
+
+# Mosquitto's default (1000 messages per offline durable subscriber) silently
+# drops newly-arriving messages once the cap is hit -- confirmed locally by
+# setting this to 5 and watching "Outgoing messages are being dropped" in the
+# log after only 10 publishes. Unbounded here for the sandbox; the real number
+# for the VPS broker is a sizing decision for issue #4/#6, not a default to
+# inherit by accident.
+max_queued_messages 0
+max_queued_bytes 0

scripts/publish_fake_reading.sh

@@ -0,0 +1,38 @@
+#!/usr/bin/env bash
+# Publishes a fake ESP32 reading frame to Mosquitto for local testing.
+# Usage: ./scripts/publish_fake_reading.sh [devID]
+set -euo pipefail
+
+DEV_ID="${1:-A4:CF:12:34:56:78}"
+TIMESTAMP="$(date '+%d/%m/%Y %H:%M:%S')"
+
+PAYLOAD=$(cat <<EOF
+{
+  "devID": "${DEV_ID}",
+  "Timestamp": "${TIMESTAMP}",
+  "Control": {
+    "Sample_per_minute": 6,
+    "Inv": { "On": 1, "Freq": 50 },
+    "Restart": 0,
+    "Valve": 56
+  },
+  "Data": {
+    "pH": 6.8,
+    "CE": 1850.3,
+    "Solar": 645.2,
+    "Volume": 20,
+    "THC": { "T": 22.55, "H": 57.89, "C": 1.85 },
+    "TH": { "T": 22.23, "H": 64.25 },
+    "Errors": { "ADC": 0, "Pulses": 0, "I2C": 0, "Inverter": 0, "Inverter_State": 0 }
+  }
+}
+EOF
+)
+
+# QoS 1: the broker only queues messages for an offline durable subscriber
+# (Bento's clean_session:false session) when they're published at QoS >= 1.
+# Whatever publishes real readings later must do the same or store-and-forward
+# silently does nothing during a Bento restart/outage.
+docker exec fertloops-mosquitto mosquitto_pub -q 1 -t "fertloops/${DEV_ID}/reading" -m "${PAYLOAD}"
+echo "Published to fertloops/${DEV_ID}/reading:"
+echo "${PAYLOAD}"

timescaledb/init/01_schema.sql

@@ -0,0 +1,43 @@
+-- Sandbox schema for local experimentation (Mosquitto + TimescaleDB).
+-- Mirrors the ESP32 frame from docs/trama-de-datos-riego.md as one row per frame.
+-- NOT the canonical measurement model (that's decided in GitHub issue #7) -- just enough to
+-- get data flowing end-to-end for the #3/#4 research tickets.
+
+CREATE EXTENSION IF NOT EXISTS timescaledb;
+
+CREATE TABLE readings (
+    time            TIMESTAMPTZ       NOT NULL,
+    dev_id          TEXT              NOT NULL,
+    ph              DOUBLE PRECISION,
+    ce              DOUBLE PRECISION,
+    solar           DOUBLE PRECISION,
+    volume          DOUBLE PRECISION,
+    soil_temp       DOUBLE PRECISION,
+    soil_humidity   DOUBLE PRECISION,
+    soil_conduct    DOUBLE PRECISION,
+    air_temp        DOUBLE PRECISION,
+    air_humidity    DOUBLE PRECISION,
+    err_adc         SMALLINT,
+    err_pulses      SMALLINT,
+    err_i2c         SMALLINT,
+    err_inverter    SMALLINT,
+    err_inverter_state SMALLINT,
+    valve           SMALLINT,
+    inv_on          SMALLINT,
+    inv_freq        DOUBLE PRECISION,
+    raw             JSONB             NOT NULL
+);
+
+SELECT create_hypertable('readings', 'time');
+
+CREATE INDEX ON readings (dev_id, time DESC);
+
+-- MQTT QoS 1 is at-least-once: the same frame can be redelivered (broker
+-- reconnect, unacked message replayed, etc). `time` is ingestion time so it
+-- differs per redelivery -- a plain unique index on (dev_id, device Timestamp)
+-- can't dedupe it here: TimescaleDB refuses a unique index on a hypertable
+-- unless the partitioning column ("time") is part of it, which would defeat
+-- the purpose (confirmed by testing -- this failed with "cannot create a
+-- unique index without the column "time" (used in partitioning)" on a fresh
+-- database). Redelivery dedup happens upstream instead, in Bento's pipeline
+-- (see bento/fertloops.yaml), keyed on dev_id + the device's own Timestamp.