Producers

CREATE TABLE payment_outbox (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    org_id UUID NOT NULL,
    aggregate_id UUID NOT NULL REFERENCES payments(id) ON DELETE CASCADE,
    topic TEXT NOT NULL,
    event_type TEXT NOT NULL,
    event_version SMALLINT NOT NULL DEFAULT 1,
    partition_key TEXT NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    published_at TIMESTAMPTZ
);

CREATE INDEX payment_outbox_unpublished_idx
    ON payment_outbox (created_at)
    WHERE published_at IS NULL;

DROP TABLE IF EXISTS payment_outbox;

package payment

import (
	"context"
	"encoding/json"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
)

type OutboxWriter struct{}

func NewOutboxWriter() *OutboxWriter {
	return &OutboxWriter{}
}

type CompletedPayload struct {
	AmountPaise       int64  `json:"amount_paise"`
	Currency          string `json:"currency"`
	ProviderReference string `json:"provider_reference"`
}

type envelope 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   time.Time        `json:"occurred_at"`
	Payload      CompletedPayload `json:"payload"`
}

// WritePaymentCompleted must be called with the same *pgx.Tx the payment's
// captured transition is committing through. This line only ever runs if
// that transition also commits — there is no code path where one succeeds
// without the other.
func (w *OutboxWriter) WritePaymentCompleted(ctx context.Context, tx pgx.Tx, orgID, paymentID uuid.UUID, payload CompletedPayload) error {
	body, err := json.Marshal(envelope{
		EventID:      uuid.New(),
		EventType:    "payment.completed",
		EventVersion: 1,
		OrgID:        orgID,
		AggregateID:  paymentID,
		OccurredAt:   time.Now().UTC(),
		Payload:      payload,
	})
	if err != nil {
		return err
	}

	_, err = tx.Exec(ctx, `
		INSERT INTO payment_outbox (org_id, aggregate_id, topic, event_type, event_version, partition_key, payload)
		VALUES ($1, $2, $3, $4, $5, $6, $7)`,
		orgID, paymentID, "payment.completed", "payment.completed", 1, paymentID.String(), body,
	)
	return err
}

	if err := tx.updatePaymentStatus(ctx, tx.pgx, paymentID, StatusCaptured); err != nil {
		return err
	}

	if err := outbox.WritePaymentCompleted(ctx, tx.pgx, orgID, paymentID, CompletedPayload{
		AmountPaise:       amountPaise,
		Currency:          currency,
		ProviderReference: providerReference,
	}); err != nil {
		return err
	}

	return tx.pgx.Commit(ctx)

package payment

import (
	"context"
	"time"

	kafkago "github.com/segmentio/kafka-go"
	"github.com/jackc/pgx/v5/pgxpool"
)

type Relay struct {
	pool   *pgxpool.Pool
	writer *kafkago.Writer
}

func NewRelay(pool *pgxpool.Pool, brokers []string) *Relay {
	return &Relay{
		pool: pool,
		writer: &kafkago.Writer{
			Addr:         kafkago.TCP(brokers...),
			Topic:        "payment.completed",
			Balancer:     &kafkago.Hash{}, // partition_key drives placement, never round robin
			RequiredAcks: kafkago.RequireAll,
		},
	}
}

type outboxRow struct {
	id           string
	partitionKey string
	payload      []byte
}

// RelayOnce claims up to batchSize unpublished rows, writes them to Kafka,
// and only then marks them published — all inside one transaction, so a
// crash at any point leaves every claimed row either fully published or
// still unpublished, never half-done.
func (r *Relay) RelayOnce(ctx context.Context, batchSize int) (int, error) {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return 0, err
	}
	defer tx.Rollback(ctx)

	rows, err := tx.Query(ctx, `
		SELECT id, partition_key, payload
		FROM payment_outbox
		WHERE published_at IS NULL
		ORDER BY created_at
		LIMIT $1
		FOR UPDATE SKIP LOCKED`, batchSize)
	if err != nil {
		return 0, err
	}

	var claimed []outboxRow
	for rows.Next() {
		var o outboxRow
		if err := rows.Scan(&o.id, &o.partitionKey, &o.payload); err != nil {
			rows.Close()
			return 0, err
		}
		claimed = append(claimed, o)
	}
	rows.Close()
	if len(claimed) == 0 {
		return 0, nil
	}

	messages := make([]kafkago.Message, len(claimed))
	for i, o := range claimed {
		messages[i] = kafkago.Message{Key: []byte(o.partitionKey), Value: o.payload}
	}
	if err := r.writer.WriteMessages(ctx, messages...); err != nil {
		return 0, err // the transaction rolls back; every claimed row stays unpublished for the next call
	}

	ids := make([]string, len(claimed))
	for i, o := range claimed {
		ids[i] = o.id
	}
	if _, err := tx.Exec(ctx, `UPDATE payment_outbox SET published_at = $2 WHERE id = ANY($1)`, ids, time.Now().UTC()); err != nil {
		return 0, err
	}
	return len(claimed), tx.Commit(ctx)
}