story/internal/outbox/store.go

package outbox

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

var ErrDeduplicationConflict = errors.New("outbox deduplication key was reused with different payload")

type Event struct {
	ID               string
	TenantID         *string
	AggregateType    string
	AggregateID      string
	EventType        string
	DeduplicationKey *string
	Payload          json.RawMessage
	AttemptCount     int
}

type NewEvent struct {
	TenantID         *string
	AggregateType    string
	AggregateID      string
	EventType        string
	DeduplicationKey *string
	Payload          json.RawMessage
}

type Store struct{ pool *pgxpool.Pool }

func NewStore(pool *pgxpool.Pool) *Store { return &Store{pool: pool} }

func (s *Store) Enqueue(ctx context.Context, tx pgx.Tx, event NewEvent) (string, error) {
	if event.AggregateType == "" || event.AggregateID == "" || event.EventType == "" || !json.Valid(event.Payload) {
		return "", errors.New("aggregate, event type and valid JSON payload are required")
	}
	payloadHash := hashPayload(event.Payload)
	var eventID string
	err := tx.QueryRow(ctx, `
		INSERT INTO outbox_event (
			tenant_id, aggregate_type, aggregate_id, event_type,
			deduplication_key, payload, payload_hash
		) VALUES ($1::uuid, $2, $3::uuid, $4, $5, $6::jsonb, $7)
		ON CONFLICT DO NOTHING
		RETURNING id::text`,
		event.TenantID, event.AggregateType, event.AggregateID, event.EventType,
		event.DeduplicationKey, event.Payload, payloadHash,
	).Scan(&eventID)
	if err == nil {
		return eventID, nil
	}
	if !errors.Is(err, pgx.ErrNoRows) || event.DeduplicationKey == nil {
		return "", fmt.Errorf("enqueue outbox event: %w", err)
	}
	var storedHash string
	err = tx.QueryRow(ctx, `
		SELECT id::text, payload_hash
		FROM outbox_event
		WHERE COALESCE(tenant_id, '00000000-0000-0000-0000-000000000000'::uuid) =
			COALESCE($1::uuid, '00000000-0000-0000-0000-000000000000'::uuid)
			AND event_type = $2 AND deduplication_key = $3
		FOR UPDATE`, event.TenantID, event.EventType, event.DeduplicationKey,
	).Scan(&eventID, &storedHash)
	if err != nil {
		return "", fmt.Errorf("read deduplicated outbox event: %w", err)
	}
	if storedHash != payloadHash {
		return "", ErrDeduplicationConflict
	}
	return eventID, nil
}

func (s *Store) ClaimBatch(ctx context.Context, workerID string, batchSize int, lockTimeout time.Duration) ([]Event, error) {
	rows, err := s.pool.Query(ctx, `
		WITH candidates AS (
			SELECT id
			FROM outbox_event
			WHERE (status = 'pending' AND available_at <= now())
			   OR (status = 'processing' AND locked_at < now() - $3::interval)
			ORDER BY available_at, created_at
			FOR UPDATE SKIP LOCKED
			LIMIT $2
		)
		UPDATE outbox_event event
		SET status = 'processing', locked_at = now(), locked_by = $1,
			attempt_count = event.attempt_count + 1, updated_at = now()
		FROM candidates
		WHERE event.id = candidates.id
		RETURNING event.id::text, event.tenant_id::text, event.aggregate_type,
			event.aggregate_id::text, event.event_type, event.deduplication_key,
			event.payload, event.attempt_count`,
		workerID, batchSize, postgresInterval(lockTimeout),
	)
	if err != nil {
		return nil, fmt.Errorf("claim outbox events: %w", err)
	}
	defer rows.Close()
	events := make([]Event, 0, batchSize)
	for rows.Next() {
		var event Event
		if err := rows.Scan(
			&event.ID, &event.TenantID, &event.AggregateType, &event.AggregateID,
			&event.EventType, &event.DeduplicationKey, &event.Payload, &event.AttemptCount,
		); err != nil {
			return nil, fmt.Errorf("scan outbox event: %w", err)
		}
		events = append(events, event)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate outbox events: %w", err)
	}
	return events, nil
}

func (s *Store) MarkCompleted(ctx context.Context, eventID, workerID string) error {
	tag, err := s.pool.Exec(ctx, `
		UPDATE outbox_event
		SET status = 'completed', completed_at = now(), locked_at = NULL,
			locked_by = NULL, last_error_code = NULL, updated_at = now()
		WHERE id = $1::uuid AND status = 'processing' AND locked_by = $2`, eventID, workerID)
	if err != nil {
		return fmt.Errorf("complete outbox event: %w", err)
	}
	if tag.RowsAffected() != 1 {
		return errors.New("outbox event lock was lost before completion")
	}
	return nil
}

func (s *Store) MarkFailed(
	ctx context.Context,
	event Event,
	workerID, errorCode string,
	maxAttempts int,
	nextAttempt time.Time,
) error {
	status := "pending"
	if event.AttemptCount >= maxAttempts {
		status = "dead"
	}
	tag, err := s.pool.Exec(ctx, `
		UPDATE outbox_event
		SET status = $3::varchar, available_at = $4, locked_at = NULL, locked_by = NULL,
			last_error_code = $5, dead_at = CASE WHEN $3::varchar = 'dead' THEN now() ELSE NULL END,
			updated_at = now()
		WHERE id = $1::uuid AND status = 'processing' AND locked_by = $2`,
		event.ID, workerID, status, nextAttempt, errorCode,
	)
	if err != nil {
		return fmt.Errorf("fail outbox event: %w", err)
	}
	if tag.RowsAffected() != 1 {
		return errors.New("outbox event lock was lost before failure update")
	}
	return nil
}

func hashPayload(payload []byte) string {
	digest := sha256.Sum256(payload)
	return hex.EncodeToString(digest[:])
}

func postgresInterval(duration time.Duration) string {
	return fmt.Sprintf("%f seconds", duration.Seconds())
}