go_message_dispatcher

A Go service that drains a message queue held in PostgreSQL and sends the messages to an SMS provider. It polls every 2 minutes, takes 2 messages per batch in FIFO order, and marks them sent. Several instances can run at once without sending anything twice.

What this demonstrates

• Safe concurrent queue draining. The repository claims rows with SELECT ... FOR UPDATE SKIP LOCKED, so two instances polling the same table at the same moment take different rows instead of blocking or double-sending.
• One active processor via a Redis lock. With DISTRIBUTED_LOCK_ENABLED=true an instance acquires a Redis lock before each cycle, and only the holder processes that cycle. The lock is released after the batch and carries a TTL, so if a holder dies mid-cycle the TTL lets another instance take the next one.
• Graceful shutdown. On SIGINT or SIGTERM the scheduler lets the batch it is holding finish before the process exits, so a message is never left half-processed.
• Partial failure handling. If one message in a batch sends and the other fails, the successful one stays marked sent and only the failure is retried.
• Redis as a cache, not a dependency. Delivery metadata is cached for the read API. If Redis is unavailable, sending continues and reads fall back to PostgreSQL.

Run it

docker compose up -d

That brings up PostgreSQL, Redis, a mock SMS provider, and the dispatcher. The database migration seeds ten sample messages, and the dispatcher starts processing on startup: two messages every two minutes, in FIFO order. Watch them drain:

curl localhost:8080/api/messages/sent

Two are sent almost at once, then two more each interval. The loop is running by default; stop pauses it and start resumes it:

curl -X POST localhost:8080/api/messaging/stop
curl -X POST localhost:8080/api/messaging/start

Swagger UI is at /swagger/index.html. Pre-built binaries for Linux, macOS and Windows are on the releases page, and images are at h4mid2019/message-dispatcher on Docker Hub.

API

Method	Path	Notes
POST	/api/messaging/start	Start the background processor
POST	/api/messaging/stop	Stop it
GET	/api/messages/sent	Sent messages, enriched from the Redis cache
GET	/health	Checks PostgreSQL and Redis
GET	/version	Build version, time and commit

Schema

CREATE TABLE messages (
    id           SERIAL PRIMARY KEY,
    phone_number VARCHAR(20) NOT NULL,
    content      VARCHAR(160) NOT NULL,
    sent         BOOLEAN DEFAULT FALSE,
    created_at   TIMESTAMP DEFAULT NOW()
);

content is VARCHAR(160), so the database rejects an over-length insert, and the application checks the length again before sending. The migration also creates the (sent, created_at) index the claim query orders by, and seeds ten sample messages so the queue is not empty on a fresh start.

Configuration

Set via environment variables. Every one has a default, so the service starts with none set, reaching localhost for both PostgreSQL and Redis with database messages_db and user postgres. The compose file overrides the connection values to point at its own containers; set them yourself to run the binary against your own infrastructure.

Variable	Default
DB_HOST	localhost
DB_PORT	5432
DB_NAME	messages_db
DB_USER	postgres
DB_PASSWORD	password
REDIS_HOST	localhost
REDIS_PORT	6379
SERVER_PORT	8080
SMS_API_URL	http://localhost:3001/send
SMS_API_TOKEN	mock-token
PROCESSING_INTERVAL	2m
BATCH_SIZE	2
MAX_CONTENT_LENGTH	160
DISTRIBUTED_LOCK_ENABLED	false
DISTRIBUTED_LOCK_TTL	3m
LOG_LEVEL	info

To run several instances, set DISTRIBUTED_LOCK_ENABLED=true on all of them and keep DISTRIBUTED_LOCK_TTL longer than PROCESSING_INTERVAL. See TIER2_IMPLEMENTATION.md.

Layout

cmd/server/         HTTP server
cmd/migrate/        migration tool
cmd/mock-api/       stand-in SMS provider for local runs
internal/domain/    entities and validation
internal/repository/ PostgreSQL access, including the SKIP LOCKED claim query
internal/service/   sending logic and cache handling
internal/scheduler/ the polling loop
internal/lock/      Redis distributed lock
internal/handler/   HTTP handlers

Tests

go test ./...
go test -race -coverprofile=coverage.out ./...

Covered today: internal/domain (validation), internal/lock (acquire, extend, expiry), internal/scheduler (lifecycle, interval, graceful shutdown), internal/service (normal flow, empty queue, partial batch failure, Redis outage tolerance, cache fallback).

Not covered yet, in the order I would add them:

1. internal/repository. This is the gap that matters. The FOR UPDATE SKIP LOCKED claim query is the thing standing between this service and sending a message twice, and there is no test that two concurrent claims return disjoint rows. It needs a real PostgreSQL, so it belongs in an integration test with a service container in CI.
2. internal/handler. The HTTP layer has no tests. httptest plus a fake service would cover status codes and the start/stop transitions.
3. internal/config. Defaulting and required-variable errors are untested.

License

MIT