The IoT Ingestion & Dispatch service is a Go service in the GridX platform responsible for consuming IoT data from Kafka, validating and processing incoming records, persisting durable telemetry and registry information to TimescaleDB, maintaining latest-state projections in Redis, and exposing internal query functionality through gRPC.
iot-ingestion/
├── cmd/
│ ├── iot-ingestion/
│ │ └── main.go # parse config, build app, handle signals
│ └── migrate/
│ └── main.go # single-owner migration command/job
├── internal/
│ ├── app/
│ │ ├── app.go # dependency wiring and lifecycle
│ │ └── shutdown.go # readiness off, drain, bounded stop
│ ├── config/
│ │ ├── config.go # typed runtime configuration
│ │ └── validate.go # fail-fast cross-field validation
│ ├── domain/
│ │ ├── telemetry.go # internal domain types and invariants
│ │ ├── heartbeat.go
│ │ └── errors.go # typed domain/application errors
│ ├── ingestion/
│ │ ├── consumer.go # poll loop, partition workers, commits
│ │ ├── router.go # strict topic-to-handler routing
│ │ ├── decoder.go # private JSON wire structs to domain
│ │ ├── meter_handler.go # ordered meter workflow
│ │ ├── heartbeat_handler.go # ordered heartbeat workflow
│ │ ├── retry.go # transient/permanent policy
│ │ └── failure_recorder.go # failure application port
│ ├── admission/
│ │ ├── registry.go # immutable in-memory snapshot
│ │ └── refresher.go # bootstrap and atomic refresh
│ ├── query/
│ │ ├── service.go # storage-neutral query use cases
│ │ └── pagination.go # opaque keyset-token logic
│ ├── store/
│ │ ├── postgres/
│ │ │ ├── pool.go
│ │ │ ├── telemetry_writer.go
│ │ │ ├── heartbeat_writer.go
│ │ │ ├── query_repository.go
│ │ │ ├── grid_loader.go
│ │ │ └── failures.go
│ │ └── redis/
│ │ ├── client.go
│ │ ├── latest.go
│ │ ├── heartbeat.go
│ │ ├── keys.go
│ │ └── scripts/
│ │ ├── set_latest_if_newer.lua
│ │ └── update_heartbeat.lua
│ ├── transport/
│ │ ├── grpc/
│ │ │ ├── server.go # options, registration, lifecycle
│ │ │ ├── telemetry_handler.go # generated API to query service
│ │ │ ├── mapper.go
│ │ │ ├── errors.go # typed errors to gRPC status
│ │ │ └── interceptors.go
│ │ └── httphealth/
│ │ └── server.go # /healthz, /readyz only
│ └── observability/
│ ├── logging.go
│ ├── metrics.go
│ └── tracing.go
├── migrations/
│ ├── embed.go # embed.FS exposed to migrate command
│ └── 00001_initial_schema.sql
├── test/
│ ├── integration/ # real Kafka/Redis/TimescaleDB
│ ├── contract/ # generated-client interoperability
│ ├── endtoend/ # simulator -> query API smoke path
│ ├── load/ # reproducible ingest/query scenarios
│ ├── fixtures/
│ └── helpers/
├── docs/
│ └── runbooks/
│ ├── consumer-lag.md
│ ├── dependency-outage.md
│ ├── failure-replay.md
│ └── redis-rebuild.md
├── .github/workflows/ci.yml
├── .dockerignore
├── .env.example # names and safe examples, no secrets
├── .gitignore # must exclude .env and local credentials
├── Dockerfile # pinned multi-stage, non-root runtime
├── docker-compose.yaml # local development only
├── go.mod
├── go.sum
└── README.md
When working inside the GridX workspace:
cd gridx-workspaceThe repository should exist at:
gridx-workspace/iot-ingestion-dispatch
From the iot-ingestion-dispatch directory:
go mod downloadThen verify and normalize the module dependencies:
go mod tidyThe IoT service depends on infrastructure provided by gridx-infra.
From the parent workspace:
cd gridx-workspaceStart the GridX environment:
go-task upOr
task upDatabase migrations are managed using Goose.
Migration SQL files are stored in:
migrations/
Example:
migrations/
├── embed.go
├── 00001_create_migration_smoke_test.sql
├── 00002_example.sql
└── ...
The files are embedded into the migration executable using Go's embed package.
This means the migration executable and its SQL files are built together.
Ensure TimescaleDB is running first.
Then configure POSTGRES_URL and DOCKER_POSTGRES_URL
The POSTGRES_URL is for local development without docker and DOCKER_POSTGRES_URL is needed for docker deployment.
Apply all pending migrations:
go run ./cmd/migrate upgo run ./cmd/migrate statusgo run ./cmd/migrate versiongo run ./cmd/migrate downThen reapply it with:
go run ./cmd/migrate upThere are two main ways to run the service locally.
This is normally the easiest setup while developing Go code.
From gridx-workspace:
go-task infra-upOr
task infra-upcd iot-ingestion-dispatchFor host execution, use:
Kafka → localhost:9092
Redis → localhost:6379
TimescaleDB → localhost:5433
go run ./cmd/migrate upgo run ./cmd/iot-ingestionNOTE: Refer gridx-workspace docs.
Start the full service containers (or you can start them after building the go containers):
go-task upOr
task upBuild and start the service image from grid-workspace:
go-task build -- iot-ingestion-dispatchOr
task build -- iot-ingestion-dispatchIn the normal GridX development environment, prefer using the parent workspace Compose configuration so the service is connected to the same Docker network as Kafka, Redis, and TimescaleDB.
Run all tests:
go test ./...Run tests with verbose output:
go test -v ./...Run tests with the Go race detector:
go test -race ./...For example:
go test -v ./internal/configFor example:
go test -v ./internal/config -run TestConfigValidateFormat the project:
gofmt -w .Check compilation and common issues:
go vet ./...Run tests:
go test ./...Run with race detection:
go test -race ./...If installed, run Staticcheck:
staticcheck ./...If installed, run Go vulnerability scanning:
govulncheck ./...A useful local verification sequence is:
gofmt -w .
go vet ./...
go test ./...
go test -race ./...Build the main application:
go build -o bin/iot-ingestion ./cmd/iot-ingestionBuild the migration executable:
go build -o bin/migrate ./cmd/migrateThe resulting layout is:
bin/
├── iot-ingestion
└── migrate
Run the application:
./bin/iot-ingestionRun migrations:
./bin/migrate up