gorm_outbox

type Config ¶

type Config struct {
	SubjectPrefix string
	Source        string
}

Config carries the relay's publishing identity. SubjectPrefix prefixes every published subject; Source identifies the publishing service in the event envelope (typically the app slug).

type Message ¶

type Message struct {
	ID          int64          `gorm:"column:id;type:bigserial;primaryKey;autoIncrement"`
	MessageID   uuid.UUID      `gorm:"column:message_id;type:char(36);not null;uniqueIndex"`
	Subject     string         `gorm:"column:subject;type:varchar(255);not null"`
	Version     string         `gorm:"column:version;type:varchar(20);not null"`
	Payload     datatypes.JSON `gorm:"column:payload;type:jsonb;not null"`
	OccurredAt  time.Time      `gorm:"column:occurred_at;type:timestamp;not null"`
	PublishedAt *time.Time     `gorm:"column:published_at;type:timestamp"`
	Error       *string        `gorm:"column:error;type:text"`
	FailedAt    *time.Time     `gorm:"column:failed_at;type:timestamp"`
}

Message is one row in the outbox_messages table. It holds a pending outbound message — an integration event or a queue task — until the Relay publishes it to the broker and marks it published.

FailedAt is set only for permanent failures (e.g. marshal errors). Transient broker failures never touch the row — the message is simply retried on the next poll tick. Error records the reason for ops inspection and replay.

func (*Message) TableName() string

type Relay ¶

type Relay struct {
	// contains filtered or unexported fields
}

Relay polls the outbox for unpublished rows, publishes each to NATS JetStream, then marks the row published. It runs as a single goroutine; no per-row concurrency keeps order-per-subject roughly stable.

Loop behavior:

• Empty outbox → sleep pollInterval, then poll again.
• Full batch → loop back immediately (more messages likely queued).
• Broker/DB error → sleep errorBackoff, then retry.

Two failure modes for individual rows:

• Permanent (marshal failure): quarantined immediately, does not block others.
• Transient (NATS unavailable): batch stopped, retried after errorBackoff.

func NewRelay(
	l logger.Logger,
	c clock.Clock,
	db *gorm.DB,
	js natslib.JetStreamContext,
	cfg Config,
) (*Relay, error)

func (r *Relay) Start(ctx context.Context) error

func (r *Relay) Stop(ctx context.Context) error

type Storage ¶

type Storage struct {
	// contains filtered or unexported fields
}

Storage implements the Outbox port by writing each message to the outbox_messages table in the caller's transaction (Transactional Outbox). The Relay picks up unpublished rows and ships them to the central message bus.

func NewStorage(c clock.Clock, db *gorm.DB) *Storage

func (p *Storage) Store(ctx context.Context, m outbox.Message) error