Building an Event-Driven System in Go with Redis Pub/Sub and PostgreSQL
Introduction to Event-Driven Architecture
Event-driven architecture (EDA) is a system design approach where components communicate through events rather than direct calls. Instead of service A directly invoking a method on service B, it publishes an event — "something happened" — and service B (or multiple services) reacts to it independently.
The benefits of this approach are clear to any experienced developer:
- Loose coupling: the publisher knows nothing about its subscribers and doesn't depend on their availability.
- Horizontal scalability: subscribers scale independently from publishers.
- Fault tolerance: temporary unavailability of one component doesn't block the others.
- Auditability and reproducibility: stored events allow you to reconstruct system state at any point in time.
In 2026, event-driven architecture in Go has become the de facto standard for high-load microservice systems. In this article, we'll build a fully functional system from scratch using Redis Pub/Sub as the event bus and PostgreSQL for persistence.
Tool Overview: Go, Redis Pub/Sub, PostgreSQL
Each component in our stack has a strictly defined role.
Go — the system core
Go is an ideal fit for event-driven systems thanks to its native concurrency support via goroutines and channels, minimalist standard library, and low memory footprint. Its compiled nature, strong typing, and fast startup make it an excellent choice for implementing both publishers and subscribers.
Redis Pub/Sub — the event bus
Redis in Pub/Sub mode provides a fire-and-forget mechanism with minimal latency. It does not persist messages — if a subscriber is unavailable at the time of publishing, the message is lost. This is a key limitation we'll account for in our architecture.
PostgreSQL — the event store
PostgreSQL serves as a reliable persistent store: we write every event to an audit log table before publishing it to Redis. This gives us the ability to recover missed events and provide delivery guarantees.
Event Design: Structure, Schema, and Versioning
An event is an immutable fact that occurred in the system. A well-designed event contains enough data for processing without requiring additional queries.
Basic event structure in Go:
package events
import (
"time"
"github.com/google/uuid"
)
// EventType defines the type of event
type EventType string
const (
UserCreated EventType = "user.created"
UserUpdated EventType = "user.updated"
OrderPlaced EventType = "order.placed"
PaymentFailed EventType = "payment.failed"
)
// BaseEvent — a common wrapper for all events
type BaseEvent struct {
ID string `json:"id"` // unique event identifier
Type EventType `json:"type"` // event type
Version int `json:"version"` // schema version
OccurredAt time.Time `json:"occurred_at"` // time of occurrence
Source string `json:"source"` // source service
Payload interface{} `json:"payload"` // event data
}
// NewEvent creates a new event with populated metadata
func NewEvent(eventType EventType, source string, version int, payload interface{}) BaseEvent {
return BaseEvent{
ID: uuid.New().String(),
Type: eventType,
Version: version,
OccurredAt: time.Now().UTC(),
Source: source,
Payload: payload,
}
}
// UserCreatedPayload — payload for the user creation event
type UserCreatedPayload struct {
UserID string `json:"user_id"`
Email string `json:"email"`
Name string `json:"name"`
CreatedAt string `json:"created_at"`
}Event versioning is critical for long-lived systems. Use the version field and handle schema migrations explicitly in your handlers. It's recommended to follow the additive changes principle: add new fields without removing existing ones.
Implementing the Publisher in Go
The publisher is responsible for publishing events to a Redis channel. An important pattern here is to first save the event to PostgreSQL (outbox pattern) and then publish it to Redis. This ensures durability.
package publisher
import (
"context"
"encoding/json"
"fmt"
"log"
"github.com/go-redis/redis/v9"
"github.com/jmoiron/sqlx"
"myapp/events"
)
// Publisher publishes events to Redis and persists them in PostgreSQL
type Publisher struct {
redis *redis.Client
db *sqlx.DB
}
func NewPublisher(redisClient *redis.Client, db *sqlx.DB) *Publisher {
return &Publisher{
redis: redisClient,
db: db,
}
}
// Publish saves the event to the DB and publishes it to Redis
func (p *Publisher) Publish(ctx context.Context, event events.BaseEvent) error {
// 1. Serialize the event
payload, err := json.Marshal(event)
if err != nil {
return fmt.Errorf("marshal event: %w", err)
}
// 2. Save to PostgreSQL (outbox / audit log)
if err := p.saveEventToDB(ctx, event, payload); err != nil {
return fmt.Errorf("save event to db: %w", err)
}
// 3. Publish to Redis Pub/Sub
channel := string(event.Type)
if err := p.redis.Publish(ctx, channel, payload).Err(); err != nil {
// Not fatal — event is already in the DB, relay worker will retry
log.Printf("WARN: failed to publish to Redis channel %s: %v", channel, err)
return nil
}
log.Printf("INFO: published event %s (id=%s) to channel %s", event.Type, event.ID, channel)
return nil
}
// saveEventToDB saves the event to the event_outbox table
func (p *Publisher) saveEventToDB(ctx context.Context, event events.BaseEvent, payload []byte) error {
query := `
INSERT INTO event_outbox (
id, event_type, version, source, occurred_at, payload, published
) VALUES (
$1, $2, $3, $4, $5, $6, false
)
`
_, err := p.db.ExecContext(ctx, query,
event.ID,
string(event.Type),
event.Version,
event.Source,
event.OccurredAt,
payload,
)
return err
}Relay Worker: from outbox to Redis
To ensure that events that didn't make it to Redis due to a failure are still delivered, we implement a background worker that polls the outbox and retransmits unpublished events:
package publisher
import (
"context"
"log"
"time"
)
// RelayWorker retransmits events from the outbox to Redis
func (p *Publisher) RelayWorker(ctx context.Context) {
ticker := time.NewTicker(5 * time.Second)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
if err := p.relayPendingEvents(ctx); err != nil {
log.Printf("ERROR: relay worker: %v", err)
}
}
}
}
func (p *Publisher) relayPendingEvents(ctx context.Context) error {
rows, err := p.db.QueryContext(ctx, `
SELECT id, event_type, payload
FROM event_outbox
WHERE published = false
ORDER BY occurred_at
LIMIT 100
FOR UPDATE SKIP LOCKED
`)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var id, eventType string
var payload []byte
if err := rows.Scan(&id, &eventType, &payload); err != nil {
continue
}
if err := p.redis.Publish(ctx, eventType, payload).Err(); err != nil {
log.Printf("WARN: relay failed for event %s: %v", id, err)
continue
}
// Mark as published
_, _ = p.db.ExecContext(ctx,
`UPDATE event_outbox SET published = true, published_at = NOW() WHERE id = $1`,
id,
)
}
return rows.Err()
}Implementing the Subscriber and Event Handlers
The subscriber listens to Redis channels and dispatches events to registered handlers. We use the registry pattern for flexible handler registration:
package subscriber
import (
"context"
"encoding/json"
"fmt"
"log"
"github.com/go-redis/redis/v9"
"myapp/events"
)
// HandlerFunc — the type of an event handler function
type HandlerFunc func(ctx context.Context, event events.BaseEvent) error
// Subscriber manages event subscriptions
type Subscriber struct {
redis *redis.Client
handlers map[events.EventType][]HandlerFunc
}
func NewSubscriber(redisClient *redis.Client) *Subscriber {
return &Subscriber{
redis: redisClient,
handlers: make(map[events.EventType][]HandlerFunc),
}
}
// Register registers a handler for a specific event type
func (s *Subscriber) Register(eventType events.EventType, handler HandlerFunc) {
s.handlers[eventType] = append(s.handlers[eventType], handler)
}
// Listen starts listening to Redis channels
func (s *Subscriber) Listen(ctx context.Context) error {
channels := make([]string, 0, len(s.handlers))
for eventType := range s.handlers {
channels = append(channels, string(eventType))
}
if len(channels) == 0 {
return fmt.Errorf("no channels registered")
}
pubsub := s.redis.Subscribe(ctx, channels...)
defer pubsub.Close()
log.Printf("INFO: subscribed to channels: %v", channels)
msgCh := pubsub.Channel()
for {
select {
case <-ctx.Done():
log.Println("INFO: subscriber shutting down")
return nil
case msg, ok := <-msgCh:
if !ok {
return fmt.Errorf("subscription channel closed")
}
go s.dispatch(ctx, msg.Channel, []byte(msg.Payload))
}
}
}
// dispatch deserializes the event and calls the appropriate handlers
func (s *Subscriber) dispatch(ctx context.Context, channel string, payload []byte) {
var event events.BaseEvent
if err := json.Unmarshal(payload, &event); err != nil {
log.Printf("ERROR: unmarshal event on channel %s: %v", channel, err)
return
}
handlers, ok := s.handlers[event.Type]
if !ok {
log.Printf("WARN: no handlers for event type %s", event.Type)
return
}
for _, handler := range handlers {
if err := handler(ctx, event); err != nil {
log.Printf("ERROR: handler for %s failed: %v", event.Type, err)
}
}
}Example of a concrete handler
package handlers
import (
"context"
"encoding/json"
"log"
"myapp/events"
)
// EmailHandler sends a welcome email when a user is created
type EmailHandler struct {
emailService EmailService
}
func NewEmailHandler(es EmailService) *EmailHandler {
return &EmailHandler{emailService: es}
}
func (h *EmailHandler) Handle(ctx context.Context, event events.BaseEvent) error {
var payload events.UserCreatedPayload
if err := remarshal(event.Payload, &payload); err != nil {
return fmt.Errorf("parse UserCreatedPayload: %w", err)
}
log.Printf("INFO: sending welcome email to %s", payload.Email)
return h.emailService.SendWelcome(ctx, payload.Email, payload.Name)
}
// remarshal converts an interface{} payload into a concrete type
func remarshal(src interface{}, dst interface{}) error {
data, err := json.Marshal(src)
if err != nil {
return err
}
return json.Unmarshal(data, dst)
}Delivery Guarantees: At-Least-Once and Idempotency
Redis Pub/Sub provides at-most-once guarantees: a message is delivered zero or one time. To achieve at-least-once delivery, we use the combination of the outbox pattern and relay worker described above. However, retries can lead to duplicate events, so handlers must be idempotent.
The idempotency pattern — a processed events table:
-- Table for deduplication
CREATE TABLE IF NOT EXISTS processed_events (
event_id UUID PRIMARY KEY,
handler VARCHAR(255) NOT NULL,
processed_at TIMESTAMPTZ DEFAULT NOW()
);
-- Index for fast lookups
CREATE INDEX IF NOT EXISTS idx_processed_events_handler
ON processed_events (handler, event_id);package handlers
import (
"context"
"database/sql"
"errors"
"fmt"
"github.com/jmoiron/sqlx"
"myapp/events"
)
// IdempotentHandler wraps a handler to ensure idempotency
type IdempotentHandler struct {
db *sqlx.DB
handlerName string
inner func(ctx context.Context, event events.BaseEvent) error
}
func NewIdempotentHandler(
db *sqlx.DB,
name string,
inner func(ctx context.Context, event events.BaseEvent) error,
) *IdempotentHandler {
return &IdempotentHandler{db: db, handlerName: name, inner: inner}
}
func (h *IdempotentHandler) Handle(ctx context.Context, event events.BaseEvent) error {
// Check whether the event has already been processed
var exists bool
err := h.db.QueryRowContext(ctx,
`SELECT EXISTS(SELECT 1 FROM processed_events WHERE event_id = $1 AND handler = $2)`,
event.ID, h.handlerName,
).Scan(&exists)
if err != nil && !errors.Is(err, sql.ErrNoRows) {
return fmt.Errorf("check idempotency: %w", err)
}
if exists {
// Event already processed — skip
return nil
}
// Execute business logic
if err := h.inner(ctx, event); err != nil {
return err
}
// Mark as processed
_, err = h.db.ExecContext(ctx,
`INSERT INTO processed_events (event_id, handler) VALUES ($1, $2) ON CONFLICT DO NOTHING`,
event.ID, h.handlerName,
)
return err
}Persisting Events in PostgreSQL as an Audit Log
Table schema for the outbox and audit log:
-- Table for the Outbox Pattern
CREATE TABLE IF NOT EXISTS event_outbox (
id UUID PRIMARY KEY,
event_type VARCHAR(255) NOT NULL,
version INTEGER NOT NULL DEFAULT 1,
source VARCHAR(255) NOT NULL,
occurred_at TIMESTAMPTZ NOT NULL,
payload JSONB NOT NULL,
published BOOLEAN NOT NULL DEFAULT false,
published_at TIMESTAMPTZ,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_event_outbox_unpublished
ON event_outbox (occurred_at)
WHERE published = false;
-- Audit log for analytics and state recovery
CREATE TABLE IF NOT EXISTS event_audit_log (
id BIGSERIAL PRIMARY KEY,
event_id UUID NOT NULL,
event_type VARCHAR(255) NOT NULL,
version INTEGER NOT NULL,
source VARCHAR(255) NOT NULL,
occurred_at TIMESTAMPTZ NOT NULL,
payload JSONB NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_event_audit_log_event_type
ON event_audit_log (event_type, occurred_at DESC);
CREATE INDEX idx_event_audit_log_event_id
ON event_audit_log (event_id);The function that writes to the audit log is called from the subscriber after successful event processing:
func SaveToAuditLog(ctx context.Context, db *sqlx.DB, event events.BaseEvent) error {
payload, err := json.Marshal(event.Payload)
if err != nil {
return err
}
_, err = db.ExecContext(ctx, `
INSERT INTO event_audit_log
(event_id, event_type, version, source, occurred_at, payload)
VALUES ($1, $2, $3, $4, $5, $6)
ON CONFLICT DO NOTHING
`, event.ID, event.Type, event.Version, event.Source, event.OccurredAt, payload)
return err
}Running and Orchestrating with Docker Compose
Docker Compose lets you spin up the entire environment with a single command. Below is a complete file for local development:
version: '3.9'
services:
postgres:
image: postgres:16-alpine
container_name: eda_postgres
environment:
POSTGRES_USER: eda_user
POSTGRES_PASSWORD: eda_secret
POSTGRES_DB: eda_db
ports:
- "5432:5432"
volumes:
- postgres_data:/var/lib/postgresql/data
- ./migrations:/docker-entrypoint-initdb.d
healthcheck:
test: ["CMD-SHELL", "pg_isready -U eda_user -d eda_db"]
interval: 5s
timeout: 5s
retries: 5
redis:
image: redis:7-alpine
container_name: eda_redis
ports:
- "6379:6379"
command: redis-server --save 60 1 --loglevel warning
volumes:
- redis_data:/data
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 5s
timeout: 3s
retries: 5
publisher:
build:
context: .
dockerfile: ./cmd/publisher/Dockerfile
container_name: eda_publisher
environment:
DATABASE_URL: postgres://eda_user:eda_secret@postgres:5432/eda_db?sslmode=disable
REDIS_URL: redis:6379
SERVICE_NAME: publisher-service
depends_on:
postgres:
condition: service_healthy
redis:
condition: service_healthy
subscriber:
build:
context: .
dockerfile: ./cmd/subscriber/Dockerfile
container_name: eda_subscriber
environment:
DATABASE_URL: postgres://eda_user:eda_secret@postgres:5432/eda_db?sslmode=disable
REDIS_URL: redis:6379
SERVICE_NAME: subscriber-service
depends_on:
postgres:
condition: service_healthy
redis:
condition: service_healthy
deploy:
replicas: 2
volumes:
postgres_data:
redis_data:Testing an Event-Driven System
Testing EDA requires a special approach. We split tests into three levels: unit tests for handlers, integration tests with real Redis and PostgreSQL, and end-to-end tests for event flows.
Unit tests for handlers
package handlers_test
import (
"context"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"myapp/events"
"myapp/handlers"
)
type MockEmailService struct {
mock.Mock
}
func (m *MockEmailService) SendWelcome(ctx context.Context, email, name string) error {
args := m.Called(ctx, email, name)
return args.Error(0)
}
func TestEmailHandler_Handle(t *testing.T) {
mockES := new(MockEmailService)
mockES.On("SendWelcome", mock.Anything, "user@example.com", "Alice").Return(nil)
handler := handlers.NewEmailHandler(mockES)
payload := events.UserCreatedPayload{
UserID: "123",
Email: "user@example.com",
Name: "Alice",
}
event := events.NewEvent(events.UserCreated, "user-service", 1, payload)
err := handler.Handle(context.Background(), event)
assert.NoError(t, err)
mockES.AssertExpectations(t)
}
func TestIdempotentHandler_SkipsDuplicate(t *testing.T) {
// Integration test with testcontainers-go or sqlmock
// First call — processes the event
// Second call with the same event.ID — skips it
// Detailed implementation depends on the test infrastructure
t.Log("See integration tests for full idempotency coverage")
}Integration tests with testcontainers-go
package integration_test
import (
"context"
"testing"
"time"
"github.com/stretchr/testify/require"
"github.com/testcontainers/testcontainers-go"
"github.com/testcontainers/testcontainers-go/modules/redis"
"github.com/testcontainers/testcontainers-go/modules/postgres"
)
func TestPublishSubscribeFlow(t *testing.T) {
ctx := context.Background()
// Start a Redis container
redisContainer, err := redis.RunContainer(ctx,
testcontainers.WithImage("redis:7-alpine"),
)
require.NoError(t, err)
defer redisContainer.Terminate(ctx)
// Start a PostgreSQL container
pgContainer, err := postgres.RunContainer(ctx,
testcontainers.WithImage("postgres:16-alpine"),
postgres.WithDatabase("test_db"),
postgres.WithUsername("test"),
postgres.WithPassword("test"),
)
require.NoError(t, err)
defer pgContainer.Terminate(ctx)
// ... initialize publisher and subscriber
// ... publish a test event
// ... verify processing via channel or WaitGroup with timeout
received := make(chan events.BaseEvent, 1)
// subscriber.Register(events.UserCreated, func(ctx context.Context, e events.BaseEvent) error {
// received <- e
// return nil
// })
select {
case event := <-received:
require.Equal(t, events.UserCreated, event.Type)
case <-time.After(5 * time.Second):
t.Fatal("timeout waiting for event")
}
}Redis Pub/Sub Limitations and When to Consider Kafka
Redis Pub/Sub is a great tool for certain scenarios, but it has fundamental limitations that you need to understand.
- No message persistence: if a subscriber is not connected at the time of publishing, the message is lost permanently. That's exactly why we use the outbox pattern.
- No consumer groups: all subscribers on a channel receive a copy of every message. It's impossible to distribute load across multiple instances of the same service without additional logic.
- No replay: you can't replay events from a specific offset the way you can with Kafka.
- Limited throughput: at very high loads (millions of messages per second), Redis becomes a bottleneck.
Use Redis Streams (XADD/XREADGROUP) if you need consumer groups and basic persistence while staying within the Redis ecosystem. Switch to Apache Kafka when you require guaranteed delivery, event replay, processing millions of messages per second, or long-term event log retention.
In a microservice architecture, Redis Pub/Sub works great for: low-latency internal notifications, cache invalidation, real-time notifications, and cases where losing an occasional message is acceptable given outbox-based compensation. For financial transactions, critical business events, and compliance-driven systems — choose Kafka or another guaranteed message broker.
Conclusion
We've built a complete event-driven system in Go, combining Redis Pub/Sub as a lightweight event bus with PostgreSQL for reliable persistence via the outbox pattern. Key patterns we applied:
- Outbox Pattern — saves the event to the DB before publishing to Redis, ensuring at-least-once delivery.
- Idempotent Handlers — a processed events table protects against duplication on retry.
- Event Registry — flexible registration of multiple handlers for a single event type.
- Relay Worker — a background process that retransmits events when Redis recovers.
Asynchronous event processing in Go in 2026 is not an exotic technique — it's a practical necessity for scalable microservice systems. Start with Redis Pub/Sub + PostgreSQL and migrate to Kafka only when you genuinely hit its limitations.
Technologies
Tags
Ruslan Ismailov
Senior Web / Backend Developer. Senior web/backend developer with 9 years of experience. Stack: PHP, Laravel, PostgreSQL, Redis, Docker, Kubernetes, REST, microservices, CI/CD. More about me →