A pub/sub log is different from a job queue. A queue hands each work item to one worker and removes it after processing. A pub/sub log keeps ordered messages so multiple independent consumer groups can read the same stream at their own pace. Kafka, Redpanda, and Pulsar are purpose-built around this abstraction, but the same core mechanics can be built on PostgreSQL when the workload is moderate and the team values operational simplicity over maximum broker throughput.
The PostgreSQL-backed design uses ordinary tables as topic partitions, a counter row to reserve monotonically increasing offsets, and a consumer-offset table to track each group’s read position. Writers batch inserts into one partition and atomically reserve a contiguous offset range. Readers poll each partition, claim a range by advancing their group offset in a transaction, then read and process the claimed records. The result is a Kafka-like pull log with strict ordering per partition, fanout through consumer groups, and at-least-once processing semantics.
This pattern is not a universal replacement for Kafka. PostgreSQL is not optimized as a distributed append-only broker, and a database-backed log will hit limits around write amplification, vacuum pressure, retention, rebalancing, and replica lag. It is useful when the application already depends on PostgreSQL, the stream volume is measured in megabytes per second rather than gigabytes per second, and teams need a debuggable event stream before they need a specialized streaming platform.
A topic is split into several independent logs. In a broker, those logs are partitions; in the PostgreSQL implementation, each partition can be represented by a table such as topicpartition0, topicpartition1, and so on. Each row stores a unique offset, a payload, and metadata such as creation time. Partitioning prevents one global table from becoming the only contention point and lets writers distribute load across several logs.
Offsets must be gapless and monotonically increasing inside a partition. A separate log_counter row stores the next offset for each partition. A writer updates that counter and inserts its payload batch in the same transaction, so the reserved offset range and inserted rows commit together. This is the core invariant that makes later consumer progress tracking meaningful.
Fanout comes from storing read progress per group and partition. A group has one logical cursor for each partition; different groups advance independently, so analytics, notification, and indexing consumers can all read the same source log. Within one group, only one worker should claim a given partition range at a time, which the design enforces with row locks on the consumer-offset row.
A durable pub/sub log needs two categories of state: the message records and the consumer cursors. Keeping both in PostgreSQL makes recovery straightforward: after a crash, each group resumes from its stored next_offset. The cost is that every read path also writes metadata, so high fanout increases database write load even when message production is unchanged.
The schema has one counter table, one table per log partition, and one cursor table keyed by consumer group and partition. The counter row is the serialization point for offset assignment inside a partition.
CREATE TABLE log_counter (
id INT PRIMARY KEY,
next_offset BIGINT NOT NULL
);
CREATE TABLE topicpartition0 (
id BIGSERIAL PRIMARY KEY,
c_offset BIGINT UNIQUE NOT NULL,
payload BYTEA NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE consumer_offsets (
group_id TEXT NOT NULL,
topic_id INT NOT NULL,
next_offset BIGINT NOT NULL DEFAULT 1,
PRIMARY KEY (group_id, topic_id)
);Writers reserve offsets by incrementing log_counter.next_offset and returning the first reserved value. The payload batch is inserted with ordinal numbers added to that first offset. If the transaction aborts, neither the counter update nor the message rows become visible.
WITH reserve AS (
UPDATE log_counter
SET next_offset = next_offset + $1
WHERE id = $3::int
RETURNING (next_offset - $1) AS first_off
)
INSERT INTO topicpartition0(c_offset, payload)
SELECT r.first_off + p.ord - 1, p.payload
FROM reserve r,
unnest($2::bytea[]) WITH ORDINALITY AS p(payload, ord);A reader claims work by locking its group cursor row, computing the available range up to the current committed tip, and advancing next_offset. The reader then selects exactly that claimed offset range. If processing and commit succeed, progress is durable. If the reader crashes before commit, the cursor does not advance and the records are retried.
BEGIN;
WITH counter_tip AS (
SELECT (next_offset - 1) AS highest_committed_offset
FROM log_counter
WHERE id = $3::int
),
to_claim AS (
SELECT
c.group_id,
c.next_offset AS n0,
LEAST(
$2::bigint,
GREATEST(0, (SELECT highest_committed_offset FROM counter_tip) - c.next_offset + 1)
) AS delta
FROM consumer_offsets c
WHERE c.group_id = $1::text AND c.topic_id = $3::int
FOR UPDATE
),
upd AS (
UPDATE consumer_offsets c
SET next_offset = c.next_offset + t.delta
FROM to_claim t
WHERE c.group_id = t.group_id AND c.topic_id = $3::int
RETURNING t.n0 AS start_offset, (c.next_offset - 1) AS end_offset
)
SELECT start_offset, end_offset FROM upd;
SELECT c_offset, payload, created_at
FROM topicpartition0
WHERE c_offset BETWEEN $1 AND $2
ORDER BY c_offset;
COMMIT;Product events often need to feed several independent consumers: audit trails, billing materialization, notification delivery, search indexing, and analytics. A PostgreSQL-backed log can support this fanout when the event volume is moderate and each consumer group needs a durable cursor rather than a best-effort notification.
Teams frequently adopt Kafka before they have Kafka-sized problems. If the system already runs PostgreSQL with backups, monitoring, and operational expertise, a database-backed log can be a smaller first step. It gives engineers SQL-based inspection, transactional writes with local application data, and a clear migration boundary if volume later justifies a dedicated broker.
Because records are retained with offsets, downstream materializers can rebuild derived state by resetting a consumer group cursor or creating a new group. This is useful for rebuilding search indexes, regenerating projections, or testing a new consumer version against historical events without changing the producer.
Keep the first implementation simple: assign each reader in a consumer group to one partition, and run exactly one active claimant for each (group_id, topic_id) pair. That avoids distributed rebalancing logic. If a process fails, a supervisor can restart it or a lightweight lease table can move the partition to another worker.
type PartitionReader struct {
GroupID string
TopicID int
BatchSize int
DB *sql.DB
}
func (r *PartitionReader) Poll(ctx context.Context) error {
tx, err := r.DB.BeginTx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback()
start, end, err := claimRange(ctx, tx, r.GroupID, r.TopicID, r.BatchSize)
if err != nil {
return err
}
if end < start {
return tx.Commit()
}
rows, err := fetchMessages(ctx, tx, r.TopicID, start, end)
if err != nil {
return err
}
if err := processRows(rows); err != nil {
return err
}
return tx.Commit()
}Retention must consider all consumer groups, not only the fastest one. A safe cleanup job computes the minimum committed offset across active groups for a partition, then deletes rows older than that point after applying a retention floor. Teams should also support explicit group expiration so abandoned experiments do not pin the log forever.
LISTEN/NOTIFY can reduce idle latency, but it should be treated as a wake-up signal, not as the durable transport. The polling query remains the source of truth because it can recover after disconnects, missed notifications, and process restarts.