Why I chose PGMQ over Redis for IoT sensor ingestion

In my last post, I introduced CultivatorsLedger — a local-first cultivation telemetry platform built on Next.js, PostgreSQL, and Docker.

One of the first architectural decisions I made was how to handle sensor data ingestion. In a grow environment, WiFi is not reliable. Network drops happen. And when they do, standard HTTP posts lose data.

I needed a queue.

The Contenders

I looked at three options:

• 
  Kafka — powerful, but way too heavy for a homelab setup. I don't need a distributed log system to track VPD in a 4x4 tent.


• 
  Redis — lightweight and fast, but it meant adding another service to the stack. More moving parts, more things to break.


• 
  MQTT — great for IoT, but it's a protocol, not a queue. I'd still need to handle reliability and persistence myself.

Why PGMQ

PGMQ is a PostgreSQL extension that turns Postgres into a message queue. It's simple, reliable, and already built into the database I'm using.

Here's the pattern:

1. Sensor data comes in via HTTP POST


2. The API route sends it to the queue with pgmq.send()
   


3. A worker picks it up with pgmq.read()
   


4. The worker processes the data and inserts it into the timeseries tables


5. The worker calls pgmq.delete() to remove the message

Why this matters

If a sensor sends data and the worker is busy, the message stays in the queue. If the network drops, the message stays in the queue. If the server restarts, the message stays in the queue.

No data loss. No complex broker setup. Just Postgres.

The Tradeoff

PGMQ isn't as fast as Kafka at massive scale. But for a home grow or small facility, it's more than enough. And keeping everything in one database means fewer services to monitor and maintain.

The Code

The consumer worker is a small Python script that runs continuously:

import psycopg2

from pgmq import PGMQ



queue = PGMQ(conn)

messages = queue.read(1, vt=60)



for msg in messages:

    sensor_data = msg['message']

    save_to_database(sensor_data)

    queue.delete(msg['id'])

That's it. The worker reads one message at a time, processes it, and deletes it. If the processing fails, the message stays in the queue until the visibility timeout expires.

What's Next

The architecture is stable. Now I'm working on the dashboard UI, Home Assistant integration, and CSV import.

If you're building something similar or just want to follow along, check out the repo:

Try the Live Demo

I've deployed an interactive demo so you can see what the dashboard looks like without installing anything.

👉 Live Demo

Play with the VPD sliders, track dry-back progress, and test the feeding calculator. All data is simulated, but the experience is real.