IoT Telemetry Ingestion Backend on AWS

A telemetry backend starts as one endpoint that accepts readings and inserts them into a table, and it holds up until a fleet of devices reconnects after an outage and sends six hours of buffered data in the same minute. In that burst, inserts time out, gateways retry and create duplicates, the dashboard query that reads the latest value per device scans millions of rows, and the history table keeps growing with no rollups.

This guide designs an ingestion backend for that job on AWS: accept batches from gateways or AWS IoT Core rules, buffer them on a queue, store them idempotently in Postgres, keep the latest state per device in a cache and roll up history on a schedule, first by hand and then in Go with Encore.

The ingestion pipeline at a glance

How do you build it by hand on AWS?

1. Device connection: Constrained devices connect to AWS IoT Core over MQTT with an X.509 certificate each, and an IoT rule with a SQL statement such as SELECT * FROM 'telemetry/+' forwards messages to SQS, Lambda or an HTTPS endpoint. Gateways that already batch can skip IoT Core and post to an HTTPS API with a per-device token.
2. Ingest: An ALB in front of an ECS service, or API Gateway with Lambda, validates each batch and sends it to SQS. Validation here matters, since a device with a broken clock sending readings from 1970 will otherwise land in the oldest partition.
3. Buffer: An SQS queue with a dead-letter queue absorbs bursts, and the worker's visibility timeout covers the slowest batch insert. Kinesis Data Streams is the alternative when ordering per device and replay of the stream are required.
4. Storage: RDS for PostgreSQL with a readings table partitioned by day or month using declarative partitioning, a unique key on (device_id, recorded_at) so retries are harmless, and a BRIN index on recorded_at, which stays small on append-only time data.
5. Latest state and rollups: The worker writes each device's latest reading to ElastiCache, and a scheduled task aggregates the previous hour into a readings_hourly table and drops partitions past the retention window.

Each stage needs its own IAM role, security group rules, environment variables for queue URLs and endpoints, and alarms on queue age and the dead-letter queue.

How do you build it with Encore?

Encore combines an open source infrastructure SDK with a platform that provisions and deploys into your own AWS or GCP account. The ingest endpoint, the topic that buffers readings, the database, the cache for latest state and the rollup schedule are declared in one Go service, and deploying it creates SNS and SQS with a dead-letter queue, RDS, ElastiCache and the IAM role for the service in your account, as listed in the AWS infrastructure docs.

Set up the app

-- telemetry/migrations/1_create_readings.up.sql
CREATE TABLE readings (
    device_id TEXT NOT NULL,
    recorded_at TIMESTAMPTZ NOT NULL,
    temperature_c DOUBLE PRECISION,
    battery_pct SMALLINT,
    PRIMARY KEY (device_id, recorded_at)
) PARTITION BY RANGE (recorded_at);

CREATE TABLE readings_2026_11 PARTITION OF readings
    FOR VALUES FROM ('2026-11-01') TO ('2026-12-01');
CREATE TABLE readings_2026_12 PARTITION OF readings
    FOR VALUES FROM ('2026-12-01') TO ('2027-01-01');

CREATE INDEX readings_recorded_brin ON readings USING BRIN (recorded_at);

CREATE TABLE readings_hourly (
    device_id TEXT NOT NULL,
    hour TIMESTAMPTZ NOT NULL,
    avg_temperature_c DOUBLE PRECISION,
    min_battery_pct SMALLINT,
    samples INT NOT NULL,
    PRIMARY KEY (device_id, hour)
);

The ingest endpoint validates a batch and publishes it, so the request returns as soon as the batch is on the queue:

// telemetry/ingest.go
package telemetry

import (
	"context"
	"time"

	"encore.dev/beta/errs"
	"encore.dev/pubsub"
)

type Reading struct {
	RecordedAt   time.Time `json:"recorded_at"`
	TemperatureC float64   `json:"temperature_c"`
	BatteryPct   int16     `json:"battery_pct"`
}

type Batch struct {
	DeviceID string    `json:"device_id"`
	Readings []Reading `json:"readings"`
}

// SNS and SQS on AWS, absorbs bursts when a fleet reconnects.
var Batches = pubsub.NewTopic[*Batch]("telemetry-batches", pubsub.TopicConfig{
	DeliveryGuarantee: pubsub.AtLeastOnce,
})

type IngestParams struct {
	Readings []Reading `json:"readings"`
}

// Ingest accepts up to 500 readings per call from a gateway.
//
//encore:api auth method=POST path=/devices/:deviceID/readings
func Ingest(ctx context.Context, deviceID string, p *IngestParams) error {
	if len(p.Readings) == 0 || len(p.Readings) > 500 {
		return &errs.Error{Code: errs.InvalidArgument, Message: "send 1 to 500 readings"}
	}
	floor := time.Now().Add(-7 * 24 * time.Hour)
	ceiling := time.Now().Add(5 * time.Minute)
	for _, r := range p.Readings {
		if r.RecordedAt.Before(floor) || r.RecordedAt.After(ceiling) {
			return &errs.Error{Code: errs.InvalidArgument, Message: "recorded_at out of range"}
		}
	}
	_, err := Batches.Publish(ctx, &Batch{DeviceID: deviceID, Readings: p.Readings})
	return err
}

A subscription stores each batch idempotently and updates the latest state per device, and an hourly cron job rolls up the previous hour:

// telemetry/store.go
package telemetry

import (
	"context"
	"time"

	"encore.dev/cron"
	"encore.dev/pubsub"
	"encore.dev/storage/cache"
	"encore.dev/storage/sqldb"
)

var db = sqldb.NewDatabase("telemetry", sqldb.DatabaseConfig{Migrations: "./migrations"})

var cluster = cache.NewCluster("telemetry", cache.ClusterConfig{EvictionPolicy: cache.AllKeysLRU})

// Latest reading per device, for dashboards and alerts.
var latest = cache.NewStructKeyspace[string, Reading](cluster, cache.KeyspaceConfig{
	KeyPattern:    "latest/:key",
	DefaultExpiry: cache.ExpireIn(24 * time.Hour),
})

var _ = pubsub.NewSubscription(Batches, "store-readings", pubsub.SubscriptionConfig[*Batch]{
	Handler:        storeBatch,
	MaxConcurrency: 10,
	RetryPolicy:    &pubsub.RetryPolicy{MinBackoff: 5 * time.Second, MaxRetries: 10},
})

func storeBatch(ctx context.Context, b *Batch) error {
	tx, err := db.Begin(ctx)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	var newest Reading
	for _, r := range b.Readings {
		// The primary key makes a redelivered batch a no-op.
		if _, err := tx.Exec(ctx, `
			INSERT INTO readings (device_id, recorded_at, temperature_c, battery_pct)
			VALUES ($1, $2, $3, $4)
			ON CONFLICT DO NOTHING
		`, b.DeviceID, r.RecordedAt, r.TemperatureC, r.BatteryPct); err != nil {
			return err
		}
		if r.RecordedAt.After(newest.RecordedAt) {
			newest = r
		}
	}
	if err := tx.Commit(); err != nil {
		return err
	}
	return latest.Set(ctx, b.DeviceID, newest)
}

var _ = cron.NewJob("hourly-rollup", cron.JobConfig{
	Title:    "Roll up the previous hour of readings",
	Schedule: "5 * * * *",
	Endpoint: Rollup,
})

//encore:api private
func Rollup(ctx context.Context) error {
	_, err := db.Exec(ctx, `
		INSERT INTO readings_hourly (device_id, hour, avg_temperature_c, min_battery_pct, samples)
		SELECT device_id, date_trunc('hour', recorded_at), avg(temperature_c), min(battery_pct), count(*)
		FROM readings
		WHERE recorded_at >= date_trunc('hour', now()) - interval '1 hour'
		  AND recorded_at < date_trunc('hour', now())
		GROUP BY 1, 2
		ON CONFLICT (device_id, hour) DO UPDATE
		SET avg_temperature_c = EXCLUDED.avg_temperature_c,
		    min_battery_pct = EXCLUDED.min_battery_pct,
		    samples = EXCLUDED.samples
	`)
	return err
}

The auth tag on Ingest requires an auth handler in the app that validates the gateway's token, for example by looking up a hashed per-device key. When devices use AWS IoT Core instead, an IoT rule with an HTTPS action can post to the same endpoint. If a reconnecting fleet publishes faster than the database can write, the backlog waits in SQS instead of failing requests, and MaxConcurrency caps how many batches write at once so RDS connections stay bounded. Retries after a crash land on the primary key and change nothing, and messages that keep failing move to the subscription's dead-letter queue, as the Pub/Sub docs describe.

encore run starts Postgres, an in-memory cache and the topic locally, so a test script that replays a burst of readings exercises the whole path on a laptop. Cron jobs do not run locally or in preview environments, so the rollup is tested by calling Rollup from the development dashboard. A coding agent asked to add humidity readings changes the struct, the migration and the rollup query in one diff and replays the test burst before review. Partitions for future months can be created by a further migration or by a monthly cron job that runs CREATE TABLE IF NOT EXISTS ... PARTITION OF readings.

encore test gives each test its own Postgres and Pub/Sub instances, per the test infrastructure docs, so a replayed burst in a test never touches another test's rows, and the cron docs cover the rollup schedule syntax.

Best fit: Teams building device and fleet backends on their own AWS account who want ingest, buffering, storage, latest state and rollups in one Go service, with the queues, database and cache provisioned from it.

Things to decide early

• Clock trust. Device clocks drift, so store both the device's recorded_at and a server received_at if you need to detect bad clocks later.
• Retention. Dropping a monthly partition is instant, while DELETE on a large table is not, so partition by the unit you expire.
• Batch size. Gateways that send 500 readings per call cut request overhead sharply compared with one request per reading, and the limit in Ingest protects the queue's message size.

Related guides

• Shipment tracking event pipeline on AWS and GCP
• SNS vs SQS
• Amazon ElastiCache with Terraform
• Background jobs in Go
• How to host a webhook receiver and worker on AWS