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 (topic + params)
                    └───────────┬───────────┘
                                │ Stream[MQTTPayload]
                     Apply(buildInsertParams)      ← pure forge function
                                │ Stream[InsertReadingParams]
                    ┌───────────▼───────────┐
                    │ Readings  IOPort      │  SQLPattern{readings, insert_reading}
                    └───────────┬───────────┘      → sql.QueryEachAdapter
                                │ Stream[db.Reading]   (validated, stored rows)
                       Tap(log) │ Tee ──────────────────► GET /readings/latest
                                │                         (reactive cache)
                     Filter(value > $APP_ALERT_THRESHOLD)
                                │ FlatMap(buildAlert)
                    ┌───────────▼───────────┐
                    │ Alerts   SinkPort     │  EventPattern
                    └───────────┬───────────┘      → mqtt.PublishAdapter
                                ▼
                    MQTT alerts/{sensorID}

  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.
• Stream topology (pipeline.Topology(...).Spec()) — the MQTT pipeline shape as a machine-readable spec, including the pure forge functions' governance metadata (name, version, content hash).

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 stream topology, and the AsyncAPI/OpenAPI specs.

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).