Stage 7 · Master
Phase 16 — Kafka Integration
Consumers
Give notification-service its own inbox — modeled directly on the Payment module's InboxStore — so the duplicate deliveries Lesson 2 accepts as normal never turn into a duplicate notification job.
The First Question Every Handler Answers Is 'Have I Seen This Already?'
Lesson 2 established that a relay crash between a successful Kafka write and its own commit republishes the same batch. That is not a rare edge case to defensively code around later — it is a normal, expected occurrence this consumer will see in production. Before notification-service's consumer does anything else with a payment.completed message, it must answer one question: has this consumer group already turned this exact event_id into a notification job? payment-service already proved out the right idiom for that question on its own inbound side (payment_webhook_inbox's InboxStore, Phase 9) — this lesson mirrors it, on notification-service's own database, for its own consumer.
consumed_events Is Keyed by (consumer_group, event_id), Not event_id Alone
CREATE TABLE consumed_events (
consumer_group text NOT NULL,
event_id uuid NOT NULL,
topic text NOT NULL,
consumed_at timestamptz NOT NULL DEFAULT now(),
PRIMARY KEY (consumer_group, event_id)
);The primary key is a pair, not event_id alone, because two independent consumer groups reading the same topic for two different purposes must each get their own independent answer to "have I seen this." If a future reporting consumer subscribes to payment.completed for its own purpose, it must not be silently blocked from processing an event this consumer group already claimed.
Same Inbox Claim as the Payment Webhook, Keyed by Consumer Group
package consumer
import (
"context"
"errors"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
)
type InboxStore struct{}
func NewInboxStore() *InboxStore {
return &InboxStore{}
}
// Claim reports whether this is the first time consumerGroup has ever seen
// eventID. A false, nil result means an earlier delivery — or this exact
// redelivery — already claimed it; the caller must skip straight to
// committing the Kafka offset without creating a second job row.
func (s *InboxStore) Claim(ctx context.Context, tx pgx.Tx, consumerGroup, topic string, eventID uuid.UUID) (bool, error) {
var claimed uuid.UUID
err := tx.QueryRow(ctx, `
INSERT INTO consumed_events (consumer_group, event_id, topic)
VALUES ($1, $2, $3)
ON CONFLICT (consumer_group, event_id) DO NOTHING
RETURNING event_id`,
consumerGroup, eventID, topic,
).Scan(&claimed)
if errors.Is(err, pgx.ErrNoRows) {
return false, nil
}
if err != nil {
return false, err
}
return true, nil
}
One Handler, One Place Where org_id Becomes organization_id
payment-service's envelope names the tenant field org_id because that is payment-service's own column name. notification_jobs (Phase 14) names the same fact organization_id. This handler is the single, explicit place that translation happens — not a coincidence of two services independently choosing different names, but a deliberate boundary: each service's schema uses its own vocabulary, and the consumer that crosses the boundary is where the mapping is written down once, in one line of code, instead of leaking one service's column name into another's migration.
package consumer
import (
"context"
"encoding/json"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgxpool"
)
type paymentCompletedEnvelope struct {
EventID uuid.UUID `json:"event_id"`
EventType string `json:"event_type"`
EventVersion int `json:"event_version"`
OrgID uuid.UUID `json:"org_id"`
AggregateID uuid.UUID `json:"aggregate_id"`
OccurredAt string `json:"occurred_at"`
Payload json.RawMessage `json:"payload"`
}
const consumerGroup = "notification-service"
type Handler struct {
pool *pgxpool.Pool
inbox *InboxStore
}
func NewHandler(pool *pgxpool.Pool, inbox *InboxStore) *Handler {
return &Handler{pool: pool, inbox: inbox}
}
// HandlePaymentCompleted does not go through Phase 14's JobRepository —
// that repository's Claim method manages its own transaction for the
// worker's read side. This insert must share one transaction with the
// inbox claim above it, so it writes directly, the same way Phase 9's
// InboxStore writes directly rather than through a separate repository.
func (h *Handler) HandlePaymentCompleted(ctx context.Context, topic string, raw []byte) error {
var env paymentCompletedEnvelope
if err := json.Unmarshal(raw, &env); err != nil {
return err // Lesson 5 classifies this as a poison event, not a retryable one
}
tx, err := h.pool.Begin(ctx)
if err != nil {
return err
}
defer tx.Rollback(ctx)
claimed, err := h.inbox.Claim(ctx, tx, consumerGroup, topic, env.EventID)
if err != nil {
return err
}
if !claimed {
return nil // duplicate delivery; an earlier attempt already created the job row
}
// organization_id is notification_jobs' own column name (Phase 14);
// org_id above is payment-service's. This is the one place the two
// vocabularies meet.
if _, err := tx.Exec(ctx, `
INSERT INTO notification_jobs (organization_id, channel, payload)
VALUES ($1, 'email', $2)`,
env.OrgID, env.Payload,
); err != nil {
return err
}
return tx.Commit(ctx)
}
Prove a Redelivered Event Produces Exactly One Job Row
Applied exercise
Prove two consumer replicas cannot both claim the same event
Phase 14's worker proved two replicas can't double-claim the same job. Prove the same property holds for this consumer's inbox, using a real duplicate delivery instead of a database row race.
- Start two instances of the notification-service consumer against the same NOTIFICATION_SERVICE_DB_URL.
- Publish the same event_id to payment.completed exactly once.
- Confirm only one of the two consumer processes' logs shows a job being created — the other should log nothing, not an error.
- Explain, in one paragraph, why InboxStore.Claim being a single INSERT ... ON CONFLICT statement makes a race between the two processes impossible, unlike a SELECT-then-INSERT approach would be.
Deliverable
A one-paragraph explanation plus the observed log output from both consumer processes.
Completion checks
- Exactly one job row is created for the single published event, regardless of which replica happened to receive it.
- The explanation specifically contrasts INSERT ... ON CONFLICT's atomicity against a SELECT-then-INSERT race.
Why is consumed_events keyed on (consumer_group, event_id) instead of event_id alone?
Mark this lesson complete to store local progress and unlock a cleaner resume path the next time you visit.