From ef46fc849ab2078432430e23ee4c1fc2ad69927f Mon Sep 17 00:00:00 2001 From: Domagoj Zecevic Date: Mon, 12 Oct 2026 15:18:56 +0200 Subject: [PATCH] feat(web): refresh laundry status on MQTT messages via server-sent events --- .ai/HANDOFF.md | 27 +++++ .ai/PLAN.md | 108 +++++++---------- .ai/REVIEW.md | 19 +++ .ai/TASKS.md | 1 + README.md | 41 ++++++- ROADMAP.md | 18 +++ internal/sensor/hub.go | 41 +++++++ internal/sensor/hub_test.go | 75 ++++++++++++ internal/sensor/sampler.go | 10 +- internal/sensor/sampler_test.go | 35 ++++++ internal/web/handler.go | 78 +++++++++++- internal/web/handler_test.go | 202 +++++++++++++++++++++++++++++++- internal/web/static/index.html | 9 ++ main.go | 27 +++-- main_test.go | 44 +++++++ 15 files changed, 649 insertions(+), 86 deletions(-) create mode 100644 internal/sensor/hub.go create mode 100644 internal/sensor/hub_test.go diff --git a/.ai/HANDOFF.md b/.ai/HANDOFF.md index cf5e2c1..f558cf8 100644 --- a/.ai/HANDOFF.md +++ b/.ai/HANDOFF.md @@ -60,3 +60,30 @@ Append-only role handoff log. Each role adds one entry when its step is complete - Status: done. Commit message read from the T-003 next_task entry. - All three tasks completed and reviewed; one separate Conventional Commit per task. + +### T-004 — plan — 2026-10-12T12:06:52Z + +| Field | Value | +|-------|-------| +| Agent | claude | +| Summary | Planned SSE-based instant page refresh on each valid MQTT message (Hub, Sampler.OnReceive, /api/events, EventSource with polling fallback) including README and comment updates | +| Files Changed | .ai/PLAN.md, .ai/TASKS.md, .ai/HANDOFF.md, ROADMAP.md | +| Next Role | implement | + +### T-004 — next_task (dev/all_task) — 2026-10-12T13:18:18Z + +| Field | Value | +|-------|-------| +| Agent | codex | +| Summary | Added MQTT-triggered SSE refresh, non-blocking broadcasts, keep-alives, stream shutdown, polling fallback, tests, and endpoint documentation | +| Validation | fmt/vet/test/race PASS; real Mosquitto/Chrome SSE smoke PASS; page fallback/coalescing checks PASS; review PASS after one recorder synchronization retry | +| Commit | feat(web): refresh laundry status on MQTT messages via server-sent events | +| Next Role | dev commit | + +- Includes the existing T-004 roadmap and planning artifacts. No new dependencies. +- PR limitation: `aide pr --dry-run` refuses main; repository instructions require staying on the current branch. + +### T-004 — commit_task (all_task) — 2026-10-12T13:18:29Z + +- Status: done. Commit message read from the T-004 next_task entry. +- All four board tasks are done; T-004 is the only task completed in this invocation and receives its own Conventional Commit. diff --git a/.ai/PLAN.md b/.ai/PLAN.md index e3ccef3..17accde 100644 --- a/.ai/PLAN.md +++ b/.ai/PLAN.md @@ -2,79 +2,51 @@ Status: **ready_for_implement** -Goal: implement the scope defined in `ROADMAP.md` — a phone-first webpage, served by a small Go backend in Docker, that shows whether the laundry is done based on the last 3 one-minute power samples from `tele/tasmota_washingmashine_C21183/SENSOR`. +Goal: implement `ROADMAP.md` Priority 2 — the page refreshes itself whenever a new valid MQTT power message arrives, using Server-Sent Events (SSE). T-001..T-003 (Priority 1) are done and stay unchanged in behavior. ## Scope -- Go module `mqtt_washingmachine` (Go 1.26), standard library plus `github.com/eclipse/paho.mqtt.golang` for MQTT. -- Layout: - - `main.go`: config from env, wiring, graceful shutdown. - - `internal/status/`: pure logic (sample window, status evaluation). - - `internal/sensor/`: Tasmota payload parsing, MQTT subscription, per-minute sampler. - - `internal/web/`: HTTP handlers and embedded static page (`go:embed`). -- Docker packaging and documentation. -- Out of scope: persistence, TLS/auth, history beyond 3 samples, multiple devices. - -## Design decisions (from ROADMAP.md) -- Threshold: 3.00 W. A sample is "idle" only if `power < 3.0`; `>= 3.0` is "active". -- Sampling: the MQTT handler stores only the latest `ENERGY.Power` value and its receive time. A ticker fires every `SAMPLE_INTERVAL` (default 60s). At each tick: if a message was received since the previous tick, record its power as a sample; otherwise record a missing sample. Then clear the "received" flag. -- Window: ring of the last 3 samples (oldest first). Each sample is `{Power float64, Valid bool}`. -- Status evaluation (first match wins): - 1. fewer than 3 samples recorded since start -> `waiting`, "Waiting for data...", grey - 2. any of the last 3 samples missing -> `incomplete`, "Incomplete data", grey - 3. any sample `>= 3.0` -> `running`, "Laundry not yet done", red - 4. otherwise (all 3 `< 3.0`) -> `done`, "Laundry done", default text color -- JSON contract for `GET /api/status`: - ```json - {"state":"running","message":"Laundry not yet done","readings":[{"power":47.7,"valid":true},{"power":0,"valid":false}]} - ``` - `readings` has 0–3 entries, oldest first. The page renders the display text, color class, and `47.70W` / `--` formatting. -- Config env vars: `MQTT_BROKER` (default `tcp://192.168.1.12:1883`), `MQTT_TOPIC` (default `tele/tasmota_washingmashine_C21183/SENSOR`), `MQTT_USERNAME`, `MQTT_PASSWORD` (optional), `HTTP_ADDR` (default `:8080`), `SAMPLE_INTERVAL` (Go duration, default `60s`). -- MQTT client uses auto-reconnect and re-subscribes in the `OnConnect` handler. +- Standard library only; no new dependencies. +- Out of scope: live-reading line, WebSocket, changes to the status rule, JSON contract of `/api/status`, or 1-minute sampling. + +## Design decisions (from ROADMAP.md Priority 2) +- Event source: a valid payload accepted by `Sampler.Receive` (about every 10 s). Invalid payloads return an error and never notify. +- Hub: new `internal/sensor/hub.go` with a concurrency-safe `Hub` (zero value usable): + - `Subscribe() (events <-chan struct{}, cancel func())`: each subscriber gets a channel with buffer 1. + - `Notify()`: non-blocking send to every subscriber (drop if the buffer is full, so a slow client never blocks MQTT handling and bursts coalesce). + - `cancel` removes the subscriber; safe to call more than once. +- `Sampler` gets an optional exported field `OnReceive func()`, set before use (documented as such), called after a valid payload is stored and the mutex is released. Zero value stays ready to use. `main.go` sets `sampler.OnReceive = hub.Notify`. +- Endpoint `GET /api/events` (`text/event-stream`, `Cache-Control: no-store`): + - On connect: flush headers and send an initial `event: update` so the client syncs immediately (covers reconnects). + - Per notification: write `event: update\ndata: {}\n\n` and flush. + - Keep-alive: SSE comment line `: keep-alive\n\n` every 25 s (interval a package-level variable so tests can shorten it). + - Clear the write deadline for this stream with `http.NewResponseController(w).SetWriteDeadline(time.Time{})` so the server-wide 10 s `WriteTimeout` does not cut it; keep `WriteTimeout` as is for other routes. If the ResponseWriter does not support flushing, return 500. + - Exit on `r.Context().Done()` (client gone or server shutdown) and call `cancel`. +- Graceful shutdown: `server.Shutdown` waits for active handlers; SSE handlers must also stop on shutdown. Register `server.RegisterOnShutdown(cancelStreams)` with a context the handler selects on (e.g. a base context passed to `NewHandler`), so shutdown is not delayed to the 5 s timeout. +- `NewHandler` signature becomes `NewHandler(snapshot func() status.Result, events *sensor.Hub)`; `web` imports `sensor` only for the Hub type (or accept an interface `Subscribe() (<-chan struct{}, func())` to avoid the import; prefer the interface). +- Page (`index.html`): + - Open `new EventSource('/api/events')`; on `update` call `refresh()`. `EventSource` reconnects automatically; the `open` event after a reconnect triggers a refresh via the initial `update`. + - Keep `setInterval(refresh, 60000)` as a fallback. Coalesce overlapping refreshes (skip or reuse if one is in flight) so bursts do not stack requests. + - If `EventSource` is unavailable, rely on polling only. ## Acceptance Criteria -- `GET /` returns a white-background page with a viewport meta tag, a very large status text, and the last 3 readings below it in normal-size black text formatted like `47.70W`; missing readings show `--`. -- The page fetches `/api/status` on load and every 60 s; no horizontal scroll at 360 px width. -- Status states and colors match the table in `ROADMAP.md` exactly, including `>= 3.00 W` = not done. -- The backend collects samples continuously with no page open; a page load immediately reflects the current state. -- A minute with no MQTT message produces a missing sample and `incomplete` state. -- Malformed payloads or payloads without `ENERGY.Power` are ignored (logged), without crashing or changing the latest value. -- `docker build` succeeds and the container runs with env-var configuration; README documents setup. -- `go fmt ./...`, `go vet ./...`, `go test ./...` all pass. +- A valid MQTT message triggers an SSE `update` event to all connected clients within the same moment; the page then re-fetches `/api/status`. +- Invalid payloads produce no event. +- A new connection receives an immediate `update`; idle connections receive a `: keep-alive` comment at the configured interval. +- The stream survives longer than the 10 s `WriteTimeout`; other routes keep it. +- A slow or stuck subscriber never blocks `Receive`/MQTT handling or other subscribers. +- Disconnected clients are removed from the hub (no goroutine or subscriber leak); shutdown closes streams promptly. +- `/api/status` JSON, status rule, and 60 s polling fallback unchanged. +- `go fmt ./...`, `go vet ./...`, `go test ./...` (also `-race` for sensor and web) pass. ## Tasks -### T-001 — Status logic and Tasmota payload parsing -Tests first, then implementation. -- `internal/status/status_test.go`, `internal/status/status.go`: window of 3, `Evaluate`, boundary tests (2.99 -> idle, 3.00 -> active, 0 W all three -> done, mixed -> running, gap -> incomplete, <3 samples -> waiting). -- `internal/sensor/payload_test.go`, `internal/sensor/payload.go`: parse the sample payload from the roadmap (`Power` 139.60 etc.), missing `ENERGY`, missing `Power`, invalid JSON. -- `go.mod` (module init, no external deps yet in this task). -- Files to change: `go.mod`, `internal/status/*`, `internal/sensor/payload*.go`. -- Documentation: package doc comments; no README change yet. - -### T-002 — Sampler, MQTT subscription, HTTP server and page +### T-004 — SSE push so the page refreshes on each MQTT message Tests first, then implementation. -- `internal/sensor/sampler_test.go`, `internal/sensor/sampler.go`: latest-value store plus ticker-driven sampling into the window, with an injectable tick channel/clock for deterministic tests (message between ticks -> valid sample, none -> missing). -- `internal/sensor/mqtt.go`: paho client, auto-reconnect, subscribe in `OnConnect`, feeds payloads to the parser then the sampler (thin glue, not unit-tested against a real broker). -- `internal/web/handler_test.go`, `internal/web/handler.go`, `internal/web/static/index.html`: `/` serves the embedded page, `/api/status` returns the JSON contract (test with `httptest`, including state/message/readings per state). Page styling: white background, `font-size` about 3rem+ bold for status via `clamp()`, readings in black at normal size, red/grey classes, `meta viewport`, 60 s `setInterval` fetch, graceful display if the fetch fails. -- `main.go`: env config with defaults, wiring, signal-based shutdown. -- `go.mod`, `go.sum`: add `github.com/eclipse/paho.mqtt.golang`. -- Documentation: code comments; README deferred to T-003. - -### T-003 — Docker packaging and README -- `Dockerfile` (multi-stage: golang builder with `CGO_ENABLED=0`, final `scratch` or `distroless/static` image, `EXPOSE 8080`). -- `docker-compose.yml` example with the env vars from this plan. -- `.dockerignore`. -- `README.md`: replace the project TODOs (Overview, Getting Started) with what it does, status rules, env var table, `docker compose up` instructions, the Tasmota `TelePeriod` note (10 s, any value works since sampling is per minute), and limitations (in-memory state, LAN only, no auth). Keep the existing AI Workflow section untouched. English only. -- Files to change: `Dockerfile`, `docker-compose.yml`, `.dockerignore`, `README.md`. -- Validation extra: `docker build .` if Docker is available; otherwise note it in the handoff. - -## Validation -- `go fmt ./...` -- `go vet ./...` -- `go test ./...` -- Manual check after T-002: run with `HTTP_ADDR=:8080`, publish test payloads with `mosquitto_pub` (if available), open `/` at a phone-width viewport. - -## Commit subjects (suggested) -- T-001: `feat(status): evaluate laundry state from the last three power samples` -- T-002: `feat(web): show live laundry status page fed by MQTT power readings` -- T-003: `feat(docker): package laundry status service with Docker and README` +- `internal/sensor/hub_test.go`, `internal/sensor/hub.go`: subscribe/notify delivery to multiple subscribers, coalescing when buffer is full (Notify never blocks), cancel removes subscriber and is idempotent, concurrent use under `-race`. +- `internal/sensor/sampler_test.go`, `internal/sensor/sampler.go`: `OnReceive` called once per valid payload, not called for invalid payload, nil `OnReceive` is safe. +- `internal/web/handler_test.go`, `internal/web/handler.go`: `/api/events` headers (`text/event-stream`, `no-store`), initial `update`, `update` after a notification, keep-alive with a shortened interval, handler returns and cancels on client context cancel, non-flushable writer -> 500; existing tests updated for the new `NewHandler` signature; page test asserts `EventSource` and `/api/events` are present and `60000` fallback remains. +- `internal/web/static/index.html`: EventSource wiring, in-flight refresh coalescing, polling fallback. +- `main.go`, `main_test.go`: create the hub, set `sampler.OnReceive`, pass to `NewHandler`, shutdown handling for streams; keep `WriteTimeout`; test that shutdown completes promptly with an open stream if feasible. +- Documentation (explicit scope): `README.md` — describe `GET /api/events` (event format, keep-alive), that the page refreshes on each MQTT message with 60 s polling as fallback, and update the statements at lines ~13 and ~76 that say the page only refreshes every 60 s; note that reverse proxies must not buffer `text/event-stream`. Update code comments in `main.go` (`loadConfig` comment says the page always polls every 60 s) and package docs for `Hub` and `OnReceive`. `ROADMAP.md` already updated. +- Files to change: `internal/sensor/hub.go`, `internal/sensor/hub_test.go`, `internal/sensor/sampler.go`, `internal/sensor/sampler_test.go`, `internal/web/handler.go`, `internal/web/handler_test.go`, `internal/web/static/index.html`, `main.go`, `main_test.go`, `README.md`. Stage new files with `git add`. +- Suggested commit: `feat(web): refresh the laundry page instantly on each MQTT message via server-sent events` diff --git a/.ai/REVIEW.md b/.ai/REVIEW.md index a5bbff9..6de460e 100644 --- a/.ai/REVIEW.md +++ b/.ai/REVIEW.md @@ -35,3 +35,22 @@ - Environment retry during implementation: Docker build metadata was read-only under the sandbox; approved rerun built successfully. No review validation retries or weakened assertions. - Workflow limitation: `aide pr --dry-run` rejected the current `main` branch. AGENTS.md requires working in the current branch, so no branch change or PR creation was performed. - Verdict: PASS, approved for the per-task commit under `all_task`. + +## Task: T-004 + +### Review Round 1 — 2026-10-12T13:17:23Z + +- Mechanical retry 1 (`go test -race ./...`): the test recorder's body read before advancing synthetic time was not synchronized with the subsequent keep-alive write. Synchronize the recorder fixture's writes and body reads; retain all assertions and expected behavior. No product race was reported. +- Retry 1 result: fmt/vet/test/race PASS after synchronizing the test recorder; no assertions weakened or removed. + +### Final verification — 2026-10-12T13:18:18Z + +- Findings: no required fixes or semantic concerns. Reviewed the staged and unstaged diff against T-004; status rules, JSON contract, and sampling remain unchanged. Hub broadcasts coalesce without blocking; callbacks execute after storage outside the sampler lock; SSE subscribers and blocked writes clean up on cancellation. +- Tests first: Go tests failed on missing Hub/callback/handler symbols before implementation; temporary page behavior checks failed on missing EventSource before page changes. Shutdown completion was fixed during implementation without changing its assertion. +- Validation: `go fmt ./...`, `go vet ./...`, `go test ./...`, `go test -race ./...`, and staged/unstaged diff checks PASS. +- E2E: NOT_CONFIGURED as a repository command. Temporary `node /tmp/mqtt-washingmachine-t004-smoke.mjs` PASS with isolated Mosquitto and real Chrome: initial events, multiple subscribers, invalid payload suppression, disconnect/reconnect, streams surviving the actual 10-second timeout, actual 25-second keep-alives, browser refresh on MQTT, sampled readings at 360px, and prompt clean SIGTERM shutdown with active streams. +- Page behavior: temporary `node /tmp/mqtt-washingmachine-t004-page-test.mjs` PASS for burst coalescing, 60-second polling, no EventSource support, and failed-fetch recovery. Screenshot inspected: `/tmp/mqtt-washingmachine-t004-mobile.png`. +- Documentation: README covers SSE format, refresh behavior, sampling, keep-alives, proxy buffering, and manual checks; AI Workflow section is byte-for-byte unchanged. +- Environment retries: Go formatting/testing needed writable build-cache access; staging needed writable Git-index access. Approved reruns passed. +- Workflow limitation: `aide pr --dry-run` refuses the current `main` branch. Kept the current branch as instructed; no PR created. +- Verdict: PASS, approved for the per-task commit under `all_task`. diff --git a/.ai/TASKS.md b/.ai/TASKS.md index 633cdbb..77bfa43 100644 --- a/.ai/TASKS.md +++ b/.ai/TASKS.md @@ -23,3 +23,4 @@ Command expectations: | T-001 | Status evaluation logic (3-sample window, 3.00 W threshold, waiting/incomplete/running/done) and Tasmota payload parsing, with tests first | done | All state/boundary tests pass (2.99 idle, 3.00 active, gap -> incomplete, <3 samples -> waiting); payload parser handles valid, missing ENERGY/Power, and invalid JSON; fmt/vet/test pass | 2026-10-12T11:03:20Z: tests first, fmt/vet/test PASS; review PASS | none | | T-002 | Per-minute sampler, MQTT subscription, HTTP `/api/status` and phone-first page, `main.go` env config | done | Sampler records valid/missing samples per tick; `/api/status` matches JSON contract; page has white background, large status, `47.70W` readings, viewport meta, 60 s refresh; fmt/vet/test pass | 2026-10-12T11:17:06Z: fmt/vet/test/race PASS; MQTT and 360px Chrome smoke PASS | none | | T-003 | Dockerfile, docker-compose example, .dockerignore, README documentation | done | Image builds (or noted if Docker unavailable); README documents rules, env vars, deployment, limitations; existing AI Workflow section untouched | 2026-10-12T11:22:18Z: fmt/vet/test PASS; Docker build/runtime PASS; README workflow preserved | none | +| T-004 | SSE push (`GET /api/events`) so the page refreshes on each valid MQTT message: sensor Hub, `Sampler.OnReceive`, SSE handler with keep-alive and write-deadline override, page EventSource with 60 s polling fallback, README/comment updates, tests first | done | Valid message -> `update` event to all clients; invalid -> none; immediate `update` on connect; keep-alive sent; stream outlives 10 s WriteTimeout; slow subscriber never blocks Receive; disconnects/shutdown clean up; `/api/status` unchanged; README documents endpoint and updated refresh behavior; fmt/vet/test (+race) pass | 2026-10-12T13:18:18Z: fmt/vet/test/race PASS; MQTT/SSE/Chrome smoke PASS; review PASS | none | diff --git a/README.md b/README.md index e57e6a8..183b9e7 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,9 @@ readings appear as `--`. The backend runs continuously, even with no page open. Every 60 seconds it records the latest valid message received since the previous tick, or a missing sample if none arrived. It keeps only the last three samples in memory, oldest first. The -page fetches the current status on load and every 60 seconds. +page fetches the current status on load and whenever a valid MQTT power message +arrives, using Server-Sent Events (SSE). It also polls every 60 seconds as a fallback. +The readings still show the last three interval samples, not a live measurement. ### Status rules @@ -73,7 +75,8 @@ port in the Compose `ports` mapping. To change only the published host port, edi the left side of `8080:8080` and leave `HTTP_ADDR` unchanged. Changing `SAMPLE_INTERVAL` affects backend sampling and the three-sample window; -the browser still refreshes every 60 seconds. Invalid or nonpositive intervals +the browser still refreshes on valid MQTT messages with a 60-second polling +fallback. Invalid or nonpositive intervals stop startup with an error. The service retries an unavailable broker in the background and subscribes again after reconnecting. HTTP remains available during broker outages; intervals without valid messages become missing data. @@ -121,8 +124,10 @@ mosquitto_pub -h -t -m '{"ENERGY":{"Power":2.99}}' After three valid samples, `/api/status` should report `done`. Publish `3.00` to get `running`, then stop publishing for a sample interval to get `incomplete`. -Reload the page to inspect each result immediately. Check the page at 360 px width -for readable text and no horizontal scrolling. +Keep the page open: valid messages trigger status fetches without a reload, while +missing intervals appear on the next fallback poll when no messages arrive. +Readings only change after a sampling tick. Check the page at 360 px width for +readable text and no horizontal scrolling. ### Status API @@ -136,6 +141,34 @@ readings ordered oldest first. Responses are not cached. For example: Use `valid` to distinguish a missing sample from a real zero-watt reading. Startup returns `{"state":"waiting","message":"Waiting for data...","readings":[]}`. +### Status events + +`GET /api/events` streams `text/event-stream` with `Cache-Control: no-store`. +It sends an immediate event on connection (including reconnection), then an +`update` notification after each valid MQTT power message: + +```text +event: update +data: {} + +``` + +Each event ends with a blank line. Events contain no measurements; fetch +`/api/status` for the current sampled state. Invalid payloads do not notify +clients. Bursts may coalesce for slow subscribers without delaying MQTT handling +or other clients. An idle connection receives `: keep-alive` followed by a blank +line every 25 seconds. Streams outlive the ordinary HTTP write timeout and close +when the client disconnects or the service shuts down. + +The page uses `EventSource`, which reconnects automatically. It refreshes on each +`update`, skips overlapping status fetches, and retains the 60-second poll even +when streaming is unavailable or the browser has no `EventSource` support. +If using a reverse proxy, disable buffering for `text/event-stream` and allow +idle connections longer than the 25-second keep-alive interval. + +To inspect events manually, run `curl -N http://localhost:8080/api/events` while +publishing valid and invalid payloads to the test topic. + ### Limitations - In-memory state only: a restart clears the window. With the default interval, diff --git a/ROADMAP.md b/ROADMAP.md index a44a34a..6dcc657 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -36,3 +36,21 @@ Objective: show the laundry status from the last 3 one-minute power samples of t ### Notes - State is in memory only. After a backend restart the page shows "Waiting for data..." until 3 samples are collected (about 3 minutes). + +## Priority 2 + +Objective: the page refreshes itself whenever a new MQTT power message arrives, instead of waiting for the 60 s poll. + +### Decided + +- Mechanism: Server-Sent Events. New endpoint `GET /api/events` (name adjustable during planning) streams an event each time the backend receives a valid `ENERGY.Power` message (about every 10 s). +- The event is only a notification. The page reacts by re-fetching `/api/status`, so the status rule, JSON contract and 1-minute sampling stay unchanged. +- Readings still show the last 3 one-minute samples; they visibly change once per minute. No live-reading line is added. +- Page uses `EventSource` (auto-reconnect). The 60 s poll stays as a fallback if the stream is down; on reconnect the page refreshes once. +- Backend: the sampler (or MQTT handler) broadcasts to all connected SSE clients without blocking on slow clients. The HTTP `WriteTimeout` of 10 s must not kill the stream (per-route override or periodic keep-alive). +- Invalid payloads do not trigger an event. +- Docs: README documents the new endpoint and behavior. + +### Open + +- None. diff --git a/internal/sensor/hub.go b/internal/sensor/hub.go new file mode 100644 index 0000000..8636436 --- /dev/null +++ b/internal/sensor/hub.go @@ -0,0 +1,41 @@ +package sensor + +import "sync" + +// Hub broadcasts refresh notifications. Its zero value is ready to use, and all +// methods are concurrency-safe. Bursts coalesce so slow subscribers never block +// the sender or other subscribers. +type Hub struct { + mu sync.Mutex + subscribers map[chan struct{}]struct{} +} + +// Subscribe registers a buffered notification channel. The caller must cancel +// when finished; cancellation is idempotent and leaves the channel open. +func (h *Hub) Subscribe() (<-chan struct{}, func()) { + events := make(chan struct{}, 1) + h.mu.Lock() + if h.subscribers == nil { + h.subscribers = make(map[chan struct{}]struct{}) + } + h.subscribers[events] = struct{}{} + h.mu.Unlock() + return events, func() { + h.mu.Lock() + delete(h.subscribers, events) + h.mu.Unlock() + } +} + +// Notify queues at most one pending notification per subscriber without waiting +// for readers. Notifications carry no data; clients fetch the current snapshot. +func (h *Hub) Notify() { + h.mu.Lock() + defer h.mu.Unlock() + for events := range h.subscribers { + select { + case events <- struct{}{}: + default: + } + } +} diff --git a/internal/sensor/hub_test.go b/internal/sensor/hub_test.go new file mode 100644 index 0000000..6c9ca82 --- /dev/null +++ b/internal/sensor/hub_test.go @@ -0,0 +1,75 @@ +package sensor + +import ( + "sync" + "testing" + "time" +) + +func TestHubDeliveryCoalescingAndCancel(t *testing.T) { + var hub Hub + slow, cancelSlow := hub.Subscribe() + fast, cancelFast := hub.Subscribe() + defer cancelFast() + done := make(chan struct{}) + go func() { + for range 1000 { + hub.Notify() + } + close(done) + }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("slow subscribers blocked notification") + } + for _, events := range []<-chan struct{}{slow, fast} { + select { + case <-events: + default: + t.Fatal("subscriber missed notification") + } + select { + case <-events: + t.Fatal("burst was not coalesced") + default: + } + } + cancelSlow() + cancelSlow() + hub.Notify() + select { + case _, ok := <-slow: + if ok { + t.Fatal("cancelled subscriber received a new event") + } + default: + } + select { + case <-fast: + default: + t.Fatal("remaining subscriber missed notification") + } +} + +func TestHubConcurrentUse(t *testing.T) { + var hub Hub + hub.Notify() // The zero value also supports notification without subscribers. + var workers sync.WaitGroup + for range 10 { + workers.Go(func() { + for range 100 { + events, cancel := hub.Subscribe() + hub.Notify() + select { + case <-events: + default: + t.Error("subscriber missed notification") + } + cancel() + cancel() + } + }) + } + workers.Wait() +} diff --git a/internal/sensor/sampler.go b/internal/sensor/sampler.go index 7f91bee..9136986 100644 --- a/internal/sensor/sampler.go +++ b/internal/sensor/sampler.go @@ -11,6 +11,11 @@ import ( // Sampler collects the latest valid message in each interval, independently of // HTTP requests. Its zero value is ready to use; all methods are concurrency-safe. type Sampler struct { + // OnReceive, if set, runs after each valid message is stored and the lock + // is released. Set it before using the sampler; it must be concurrency-safe + // and return promptly so it does not delay MQTT handling. + OnReceive func() + mu sync.Mutex window status.Window power float64 @@ -25,8 +30,11 @@ func (s *Sampler) Receive(payload []byte) error { return err } s.mu.Lock() - defer s.mu.Unlock() s.power, s.receivedAt, s.received = power, time.Now(), true + s.mu.Unlock() + if s.OnReceive != nil { + s.OnReceive() + } return nil } diff --git a/internal/sensor/sampler_test.go b/internal/sensor/sampler_test.go index f249c6e..2cb530e 100644 --- a/internal/sensor/sampler_test.go +++ b/internal/sensor/sampler_test.go @@ -83,6 +83,41 @@ func TestSamplerClosedTicks(t *testing.T) { } } +func TestSamplerOnReceive(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + var sampler Sampler + calls := 0 + ticks := make(chan time.Time) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + go sampler.Run(ctx, ticks) + sampler.OnReceive = func() { + calls++ + // Both operations acquire the sampler lock: the callback must run + // after storage and outside that lock. + ticks <- time.Now() + synctest.Wait() + got := sampler.Status().Readings + if got[len(got)-1] != (status.Sample{Power: 47.7, Valid: true}) { + t.Errorf("callback ran before storage: %+v", got) + } + } + for range 2 { + if err := sampler.Receive([]byte(`{"ENERGY":{"Power":47.7}}`)); err != nil { + t.Fatal(err) + } + } + for _, payload := range []string{`broken`, `{}`, `{"ENERGY":{}}`, `{"ENERGY":{"Power":"bad"}}`} { + if err := sampler.Receive([]byte(payload)); err == nil { + t.Fatalf("accepted %s", payload) + } + } + if calls != 2 { + t.Fatalf("callback calls = %d; want 2", calls) + } + }) +} + func TestSamplerConcurrentAccess(t *testing.T) { var sampler Sampler ctx, cancel := context.WithCancel(t.Context()) diff --git a/internal/web/handler.go b/internal/web/handler.go index c80eb28..2b943e4 100644 --- a/internal/web/handler.go +++ b/internal/web/handler.go @@ -2,9 +2,12 @@ package web import ( + "context" _ "embed" "encoding/json" + "errors" "net/http" + "time" "mqtt_washingmachine/internal/status" ) @@ -12,9 +15,17 @@ import ( //go:embed static/index.html var indexHTML []byte +var keepAliveInterval = 25 * time.Second + +// EventSource provides coalesced refresh notifications and subscriber cleanup. +type EventSource interface { + Subscribe() (<-chan struct{}, func()) +} + // NewHandler fetches a fresh snapshot for each API request. Disabling caching -// keeps reloads and polling from showing a previous laundry state. -func NewHandler(snapshot func() status.Result) http.Handler { +// keeps reloads, push refreshes, and polling from showing a previous state. +// The server must cancel request contexts on shutdown to close active streams. +func NewHandler(snapshot func() status.Result, events EventSource) http.Handler { mux := http.NewServeMux() mux.HandleFunc("GET /{$}", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/html; charset=utf-8") @@ -26,5 +37,68 @@ func NewHandler(snapshot func() status.Result) http.Handler { w.Header().Set("Cache-Control", "no-store") json.NewEncoder(w).Encode(snapshot()) }) + mux.HandleFunc("GET /api/events", func(w http.ResponseWriter, r *http.Request) { + serveEvents(w, r, events) + }) return mux } + +func serveEvents(w http.ResponseWriter, r *http.Request, source EventSource) { + if _, ok := w.(http.Flusher); !ok { + http.Error(w, "Streaming is not supported", http.StatusInternalServerError) + return + } + controller := http.NewResponseController(w) + // Streaming outlives the server's ordinary request write timeout. Test + // recorders may support flushing without implementing write deadlines. + if err := controller.SetWriteDeadline(time.Time{}); err != nil && !errors.Is(err, http.ErrNotSupported) { + http.Error(w, "Cannot start stream", http.StatusInternalServerError) + return + } + // Cancellation must also interrupt a write blocked by an unresponsive + // client. Wait for this callback before returning so it cannot use the + // ResponseWriter after the handler finishes. + interrupted := make(chan struct{}) + stop := context.AfterFunc(r.Context(), func() { + controller.SetWriteDeadline(time.Now()) + close(interrupted) + }) + defer func() { + if !stop() { + <-interrupted + // Let net/http finish an idle stream's chunked response normally. + // A failed blocked write already carries its error in the writer. + controller.SetWriteDeadline(time.Time{}) + } + }() + events, cancel := source.Subscribe() + defer cancel() + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-store") + write := func(frame string) bool { + if _, err := w.Write([]byte(frame)); err != nil { + return false + } + return controller.Flush() == nil + } + const update = "event: update\ndata: {}\n\n" + if !write(update) { + return + } + ticker := time.NewTicker(keepAliveInterval) + defer ticker.Stop() + for { + select { + case <-r.Context().Done(): + return + case <-events: + if !write(update) { + return + } + case <-ticker.C: + if !write(": keep-alive\n\n") { + return + } + } + } +} diff --git a/internal/web/handler_test.go b/internal/web/handler_test.go index 3d63710..6eaa8ba 100644 --- a/internal/web/handler_test.go +++ b/internal/web/handler_test.go @@ -1,13 +1,21 @@ package web import ( + "bufio" + "context" "encoding/json" + "errors" + "io" "net/http" "net/http/httptest" "reflect" "strings" + "sync" "testing" + "testing/synctest" + "time" + "mqtt_washingmachine/internal/sensor" "mqtt_washingmachine/internal/status" ) @@ -23,7 +31,7 @@ func TestStatusAPI(t *testing.T) { {"done", "done", "Laundry done", []status.Sample{{Power: 2.99, Valid: true}, {Valid: true}, {Valid: true}}}, } { t.Run(tt.name, func(t *testing.T) { - handler := NewHandler(func() status.Result { return status.Evaluate(tt.readings) }) + handler := NewHandler(func() status.Result { return status.Evaluate(tt.readings) }, &sensor.Hub{}) response := httptest.NewRecorder() handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/api/status", nil)) if response.Code != http.StatusOK { @@ -48,13 +56,13 @@ func TestStatusAPI(t *testing.T) { } func TestPageAndRouting(t *testing.T) { - handler := NewHandler(func() status.Result { return status.Evaluate(nil) }) + handler := NewHandler(func() status.Result { return status.Evaluate(nil) }, &sensor.Hub{}) page := httptest.NewRecorder() handler.ServeHTTP(page, httptest.NewRequest(http.MethodGet, "/", nil)) if page.Code != http.StatusOK || !strings.HasPrefix(page.Header().Get("Content-Type"), "text/html") { t.Fatalf("page response: %+v", page.Result()) } - for _, text := range []string{`name="viewport"`, "Waiting for data...", "/api/status", "60000", "toFixed(2)", "--", "clamp("} { + for _, text := range []string{`name="viewport"`, "Waiting for data...", "/api/status", "60000", "EventSource", "/api/events", "toFixed(2)", "--", "clamp("} { if !strings.Contains(page.Body.String(), text) { t.Errorf("page missing %q", text) } @@ -65,6 +73,7 @@ func TestPageAndRouting(t *testing.T) { }{ {http.MethodGet, "/missing", http.StatusNotFound}, {http.MethodPost, "/api/status", http.StatusMethodNotAllowed}, + {http.MethodPost, "/api/events", http.StatusMethodNotAllowed}, {http.MethodPost, "/", http.StatusMethodNotAllowed}, } { response := httptest.NewRecorder() @@ -77,7 +86,7 @@ func TestPageAndRouting(t *testing.T) { func TestStatusAPIReadsCurrentSnapshot(t *testing.T) { var window status.Window - handler := NewHandler(func() status.Result { return status.Evaluate(window.Readings()) }) + handler := NewHandler(func() status.Result { return status.Evaluate(window.Readings()) }, &sensor.Hub{}) for range 3 { window.Add(status.Sample{Valid: true}) response := httptest.NewRecorder() @@ -91,3 +100,188 @@ func TestStatusAPIReadsCurrentSnapshot(t *testing.T) { } } } + +type trackedEvents struct { + events chan struct{} + subscribed bool + cancelled bool +} + +func (e *trackedEvents) Subscribe() (<-chan struct{}, func()) { + e.subscribed = true + return e.events, func() { e.cancelled = true } +} + +// Synthetic time orders timer wakeups but does not establish a race-detector +// happens-before edge from a body inspection to the next timed write. +type streamRecorder struct { + *httptest.ResponseRecorder + mu sync.Mutex +} + +func (w *streamRecorder) Write(p []byte) (int, error) { + w.mu.Lock() + defer w.mu.Unlock() + return w.ResponseRecorder.Write(p) +} + +func (w *streamRecorder) body() string { + w.mu.Lock() + defer w.mu.Unlock() + return w.Body.String() +} + +func TestEventsInitialNotificationKeepAliveAndCleanup(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + oldInterval := keepAliveInterval + keepAliveInterval = time.Second + defer func() { keepAliveInterval = oldInterval }() + events := &trackedEvents{events: make(chan struct{}, 1)} + handler := NewHandler(func() status.Result { return status.Evaluate(nil) }, events) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + request := httptest.NewRequest(http.MethodGet, "/api/events", nil).WithContext(ctx) + response := &streamRecorder{ResponseRecorder: httptest.NewRecorder()} + done := make(chan struct{}) + go func() { handler.ServeHTTP(response, request); close(done) }() + synctest.Wait() + if !events.subscribed || !response.Flushed || response.Code != http.StatusOK { + t.Fatal("stream did not subscribe and flush immediately") + } + if response.Header().Get("Content-Type") != "text/event-stream" || response.Header().Get("Cache-Control") != "no-store" { + t.Fatalf("stream headers: %v", response.Header()) + } + if got := response.body(); got != "event: update\ndata: {}\n\n" { + t.Fatalf("initial event = %q", got) + } + events.events <- struct{}{} + synctest.Wait() + if got := strings.Count(response.body(), "event: update\n"); got != 2 { + t.Fatalf("update count = %d; want 2", got) + } + time.Sleep(time.Second) + synctest.Wait() + if !strings.HasSuffix(response.body(), ": keep-alive\n\n") { + t.Fatalf("missing keep-alive: %q", response.body()) + } + cancel() + synctest.Wait() + select { + case <-done: + default: + t.Fatal("handler did not exit after disconnect") + } + if !events.cancelled { + t.Fatal("disconnected subscriber was not removed") + } + }) +} + +type nonFlushingWriter struct{ http.ResponseWriter } + +func TestEventsRejectNonFlushingWriter(t *testing.T) { + events := &trackedEvents{events: make(chan struct{}, 1)} + handler := NewHandler(func() status.Result { return status.Evaluate(nil) }, events) + response := httptest.NewRecorder() + handler.ServeHTTP(nonFlushingWriter{response}, httptest.NewRequest(http.MethodGet, "/api/events", nil)) + if response.Code != http.StatusInternalServerError || events.subscribed { + t.Fatalf("unsupported stream: status %d, subscribed %v", response.Code, events.subscribed) + } +} + +type blockedStreamWriter struct { + *httptest.ResponseRecorder + started chan struct{} + unblock chan struct{} + once sync.Once + cleared bool +} + +func (w *blockedStreamWriter) Write(p []byte) (int, error) { + close(w.started) + <-w.unblock + return 0, errors.New("write deadline exceeded") +} + +func (w *blockedStreamWriter) SetWriteDeadline(deadline time.Time) error { + if deadline.IsZero() { + w.cleared = true + } else { + w.once.Do(func() { close(w.unblock) }) + } + return nil +} + +func TestEventsCancellationInterruptsBlockedWrite(t *testing.T) { + events := &trackedEvents{events: make(chan struct{}, 1)} + handler := NewHandler(func() status.Result { return status.Evaluate(nil) }, events) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + response := &blockedStreamWriter{ResponseRecorder: httptest.NewRecorder(), started: make(chan struct{}), unblock: make(chan struct{})} + done := make(chan struct{}) + go func() { + handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/api/events", nil).WithContext(ctx)) + close(done) + }() + select { + case <-response.started: + case <-time.After(time.Second): + t.Fatal("stream did not start writing") + } + cancel() + select { + case <-done: + case <-time.After(time.Second): + response.once.Do(func() { close(response.unblock) }) + <-done + t.Fatal("cancellation did not interrupt the blocked write") + } + if !events.cancelled || !response.cleared { + t.Fatalf("cancelled=%v, write deadline cleared=%v", events.cancelled, response.cleared) + } +} + +func TestEventsSurviveWriteTimeoutAndBroadcastValidMessages(t *testing.T) { + var hub sensor.Hub + sampler := &sensor.Sampler{OnReceive: hub.Notify} + server := httptest.NewUnstartedServer(NewHandler(sampler.Status, &hub)) + server.Config.WriteTimeout = 50 * time.Millisecond + server.Start() + defer server.Close() + client := &http.Client{Timeout: 3 * time.Second} + readers := make([]*bufio.Reader, 2) + for i := range readers { + response, err := client.Get(server.URL + "/api/events") + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + readers[i] = bufio.NewReader(response.Body) + assertUpdate(t, readers[i]) + } + time.Sleep(100 * time.Millisecond) + if err := sampler.Receive([]byte(`broken`)); err == nil { + t.Fatal("invalid payload accepted") + } + if err := sampler.Receive([]byte(`{"ENERGY":{"Power":47.7}}`)); err != nil { + t.Fatal(err) + } + for _, reader := range readers { + assertUpdate(t, reader) + } + if server.Config.WriteTimeout != 50*time.Millisecond { + t.Fatal("global write timeout changed") + } +} + +func assertUpdate(t *testing.T, reader io.Reader) { + t.Helper() + want := "event: update\ndata: {}\n\n" + got := make([]byte, len(want)) + if _, err := io.ReadFull(reader, got); err != nil { + t.Fatal(err) + } + if string(got) != want { + t.Fatalf("event = %q; want %q", got, want) + } +} diff --git a/internal/web/static/index.html b/internal/web/static/index.html index 49b819f..10248b4 100644 --- a/internal/web/static/index.html +++ b/internal/web/static/index.html @@ -29,8 +29,11 @@ diff --git a/main.go b/main.go index 91a0659..de1456c 100644 --- a/main.go +++ b/main.go @@ -24,7 +24,8 @@ type config struct { } // Empty variables use defaults. SAMPLE_INTERVAL accepts positive Go durations; -// it controls backend sampling only, while the page always polls every 60 s. +// it controls backend sampling only. The page refreshes on valid MQTT messages, +// with a fixed 60 s polling fallback. func loadConfig() (config, error) { env := func(key, fallback string) string { if value := os.Getenv(key); value != "" { @@ -58,6 +59,8 @@ func run(ctx context.Context, cfg config) error { ticker := time.NewTicker(cfg.SampleInterval) defer ticker.Stop() var sampler sensor.Sampler + var hub sensor.Hub + sampler.OnReceive = hub.Notify done := make(chan struct{}) go func() { defer close(done) @@ -66,12 +69,7 @@ func run(ctx context.Context, cfg config) error { defer func() { cancel(); <-done }() client := sensor.StartMQTT(cfg.MQTTConfig, &sampler, slog.Default()) defer client.Disconnect(250) - server := &http.Server{ - Handler: web.NewHandler(sampler.Status), - ReadHeaderTimeout: 5 * time.Second, - WriteTimeout: 10 * time.Second, - IdleTimeout: 60 * time.Second, - } + server := newHTTPServer(ctx, web.NewHandler(sampler.Status, &hub)) errorsCh := make(chan error, 1) go func() { errorsCh <- server.Serve(listener) }() slog.Info("Laundry status server listening", "address", listener.Addr().String()) @@ -89,6 +87,21 @@ func run(ctx context.Context, cfg config) error { } } +// newHTTPServer cancels stream contexts when shutdown starts, allowing active +// SSE handlers (including blocked writers) to exit before Shutdown waits on them. +func newHTTPServer(ctx context.Context, handler http.Handler) *http.Server { + streams, cancelStreams := context.WithCancel(ctx) + server := &http.Server{ + Handler: handler, + BaseContext: func(net.Listener) context.Context { return streams }, + ReadHeaderTimeout: 5 * time.Second, + WriteTimeout: 10 * time.Second, + IdleTimeout: 60 * time.Second, + } + server.RegisterOnShutdown(cancelStreams) + return server +} + func main() { cfg, err := loadConfig() if err == nil { diff --git a/main_test.go b/main_test.go index f5b5ef9..f99ebab 100644 --- a/main_test.go +++ b/main_test.go @@ -1,8 +1,15 @@ package main import ( + "context" + "io" + "net" + "net/http" "testing" "time" + + "mqtt_washingmachine/internal/sensor" + "mqtt_washingmachine/internal/web" ) func TestLoadConfigDefaults(t *testing.T) { @@ -18,6 +25,43 @@ func TestLoadConfigDefaults(t *testing.T) { } } +func TestHTTPServerShutdownWithOpenStream(t *testing.T) { + var sampler sensor.Sampler + var hub sensor.Hub + server := newHTTPServer(t.Context(), web.NewHandler(sampler.Status, &hub)) + if server.WriteTimeout != 10*time.Second { + t.Fatalf("write timeout = %v", server.WriteTimeout) + } + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer server.Close() + served := make(chan error, 1) + go func() { served <- server.Serve(listener) }() + client := &http.Client{Timeout: 3 * time.Second} + response, err := client.Get("http://" + listener.Addr().String() + "/api/events") + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + initial := make([]byte, len("event: update\ndata: {}\n\n")) + if _, err := io.ReadFull(response.Body, initial); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + if err := server.Shutdown(ctx); err != nil { + t.Fatalf("open stream delayed shutdown: %v", err) + } + if body, err := io.ReadAll(response.Body); err != nil || len(body) != 0 { + t.Fatalf("stream did not close cleanly: %q, %v", body, err) + } + if err := <-served; err != http.ErrServerClosed { + t.Fatalf("Serve = %v", err) + } +} + func TestLoadConfigOverrides(t *testing.T) { for key, value := range map[string]string{"MQTT_BROKER": "tcp://localhost:1884", "MQTT_TOPIC": "test/power", "MQTT_USERNAME": "tester", "MQTT_PASSWORD": "test-password", "HTTP_ADDR": "127.0.0.1:9090", "SAMPLE_INTERVAL": "250ms"} { t.Setenv(key, value)