sensor-service

sensor-service

The go-codex flagship example: a small but complete sensor-readings service, structured like a real project. It tells one coherent story and shows how protocol-agnostic ports make every IO boundary of a pipeline composable — the pipeline code never imports an adapter; swapping MQTT → ZeroMQ or SQL → HTTP changes only main.go.

The use case

1. Ingest — sensors publish readings over MQTT.
2. Persist — every reading is validated and written to the database.
3. Alert — readings above a threshold (configured via env var) publish an MQTT alert.
4. Query — the time series of one sensor is served over REST, queried from the database.
5. Export — a REST call triggers an export of all readings, written to a typed JSON file.

Data flow — every hop is a port

                    MQTT sensors/{sensorID}/data
                                │
                    ┌───────────▼───────────┐
                    │ Sensors  SourcePort   │  EventPattern plugged in at wiring
                    └───────────┬───────────┘  time (PluginEventPattern) →
                                │              mqtt.SubscribeAdapter
                                │ Chain(buildParams.Apply, pure)
                    ┌───────────▼───────────┐
                    │ Params   PipePort     │  pipeline-internal stage, no adapters
                    └───────────┬───────────┘
                                │ ChainStream(Persist.Connect + Tap(log))
                    ┌───────────▼───────────┐
                    │ Saved    PipePort     │  the SQL persistence hop is its OWN
                    └───────────┬───────────┘  real edge (Params → Saved) — see
                                │              ioports.Readings' IOPort below
                                │ Stream() ──────────────► GET /readings/latest
                                │                          (reactive cache)
                     ChainStream(Filter(threshold) + FlatMap(buildAlert))
                    ┌───────────▼───────────┐
                    │ Alerts     SinkPort   │  EventPattern plugged in at wiring
                    └───────────┬───────────┘  time (PluginEventPattern) →
                                ▼              mqtt.PublishAdapter
                    MQTT alerts/{sensorID}

  Readings  IOPort      SQLPattern{readings, insert_reading} → sql.QueryEachAdapter
  (Connected INSIDE the Params→Saved ChainStream transform above — the
  pipeline never hides an IO hop inside a bigger stage's closure; every
  boundary, including this one, is a real, named port edge)

  GET /sensors/{sensorID}/readings          POST /export
  ┌────────────────────────┐                ┌────────────────────────┐
  │ HistoryTool  ToolPort  │ RESTPattern    │ ExportTool  ToolPort   │ RESTPattern
  └───────────┬────────────┘                └───────────┬────────────┘
              │ Single(SensorQuery)                     │ Single(ExportRequest)
  ┌───────────▼────────────┐                ┌───────────▼────────────┐
  │ History  IOPort        │ SQLPattern     │ ExportQuery  IOPort    │ SQLPattern
  └───────────┬────────────┘                └───────────┬────────────┘
              │ Stream[TimeSeries]                      │ Stream[ExportSnapshot]
              ▼                                Tap ─────┼──► Exports SinkPort
        200 TimeSeries                                  │    FilePattern
                                       Apply(buildExportResult)  {exportID}.json
                                                        ▼         → file adapter
                                          201 ExportResult{File, Count}

Package layout

Import direction is strictly acyclic; domain imports nothing internal but db.

Ports declared in ioports/

Alongside the ports, ioports declares the three classic REST routes and registers them against the same shared builder at declaration time:

Spec generation — the declarations are the spec

The ioports declarations are the single source of truth; the demo renders three artifacts from them, without any separate spec-authoring step:

• AsyncAPI (EventsBuilder.AsyncAPISpec()) — both event ports appear as channels + receive/send operations with full payload schemas and topic parameters. Each EventPattern was registered internally by its port's constructor against the shared EventsBuilder (which also enforces the topic-format constraints at construction time — an invalid topic fails port construction, not spec rendering).
• OpenAPI (RESTBuilder.OpenAPISpec()) — covers all five HTTP endpoints with request/response schemas: the two RESTPattern-based tool ports register internally at port construction, and the three classic routes register explicitly next to their declarations (CreateHandle = codex.Must(CreateRoute.Register(RESTBuilder))). Header fields declared with codecs ride along: ExportTool's rest.HeaderParam{Name: "X-Api-Key", Required: true}.WithCodec(domain.APIKeyCodec) is enforced by the adapter before the pipeline runs (400 + rest.HeaderParamError, observer location "header") and appears in the spec as an in: header parameter — one declaration, both behaviors.
• Pipeline spec (ports.PipelineSpec("Sensor Service MQTT Pipeline", "1.0.0", ioports.Sensors, pipeline.Params, pipeline.Saved, ioports.Alerts)) — the MQTT pipeline shape derived directly from the real port/PipePort wiring, not hand-typed: pipe/port names, buffer sizes, bound adapter identities, and every Chain/ChainStream edge (including each transform's real Go function identity via reflection — e.g. buildInsertParams for the pure map, honestly closure-opaque for the SQL-persistence and filter+alert transforms). No separate pipeline.Topology function to keep in sync with the code — see docs/features/ports.md.

The same declare-once principle covers the file boundary: the export response's file path comes from the same FilePattern declaration that writes the file (FileHandle.BuildPath).

HTTP endpoints

Configuration

Run

go run ./examples/sensor-service

# raise the alert threshold — the 87.3 °C reading no longer alerts:
APP_ALERT_THRESHOLD=90 go run ./examples/sensor-service

The demo runs the full story in-process (mock MQTT client, in-memory SQLite, httptest server) and prints each scene, the observer summary, the derived pipeline spec, and the AsyncAPI/OpenAPI specs. Lifecycle is managed by app.New: the demo ends with a.Shutdown() (ordered LIFO teardown); a real service would call a.Run(ctx) and get SIGINT/SIGTERM handling on the same teardown path.

After a run, the exported snapshot sits right here in exports/ ({exportID}.json, per the FilePattern declaration) — open it to see the codec-shaped file the Exports port wrote. The directory is wiped and recreated on each run and is gitignored.

Regenerating the database layer

cd examples/sensor-service
sqlc generate   # query/readings.sql → db/

Migrations in migrations/ are applied at startup via sqladapter.NewMigrator (goose).