delivery

type ClassifiedError struct {
	OriginalError error
	ErrorType     ErrorType
	RetryAfter    *time.Duration // for rate-limit errors
	IsRetryable   bool
	Message       string
}

type Config struct {
	// Name uniquely identifies the pipeline (leader-election class + job/service
	// name + log label).
	Name string
	// SyncInterval is how often the discovery loop scans for due records.
	SyncInterval time.Duration
	// MaxRetries caps retry attempts (0 = strategy default).
	MaxRetries int
	// Elector gates the discovery loop to a single leader.
	Elector leaderElection.Elector
	// Queue, when non-nil, distributes processing via Pub/Sub. Nil = poll-only
	// (the leader processes inline).
	Queue QueueClient
	// OnPermanentFailure, when set, is called after a record exhausts retries
	// (e.g. to raise an alert). Optional.
	OnPermanentFailure func(d Delivery, err error)
}

type Delivery interface {
	GetID() string
	DeliveryState() *State
}

type Engine[D Delivery] struct {
	*negotiate.SyncJob[D]
	// contains filtered or unexported fields
}

type ErrorEntry struct {
	AttemptTimestamp string `json:"attemptTimestamp"`
	Message          string `json:"message"`
	// Detail is the underlying error exactly as the Sender returned it. Message is
	// only a classification summary and is deliberately generic ("Authentication or
	// authorization failed"); Detail is what actually went wrong, and is the field
	// to read when diagnosing a failed delivery.
	Detail        string `json:"detail,omitempty"`
	AttemptNumber int    `json:"attemptNumber"`
	ErrorType     string `json:"errorType"`
}

type ErrorType string

type GormStore[D Delivery] struct {
	// contains filtered or unexported fields
}

type QueueClient = pubsub.Client[TaskMessage]

type RetryStrategy struct {
	MaxRetries     int
	BaseDelay      time.Duration
	MaxDelay       time.Duration
	Multiplier     float64
	JitterFraction float64 // 0.0..1.0
}

type Sender[D Delivery] interface {
	Send(ctx context.Context, d D) error
}

type State struct {
	Status       string `gorm:"type:text;not null;default:'pending';column:status" json:"status"`
	AttemptCount int    `gorm:"type:integer;not null;default:0;column:attempt_count" json:"attemptCount"`
	// PushedAt records the first time this record was submitted into the delivery
	// system (published to the queue, or handed to the leader's inline processor
	// in poll-only mode). It is nil for a freshly-created outbox row that the
	// initiating action committed but that has not yet reached the delivery
	// system. The leader-elected discovery loop is responsible for pushing every
	// unpushed row and stamping this column, so a crash between the initiating
	// commit and the push can never lose the message.
	PushedAt      *time.Time   `gorm:"type:timestamptz;column:pushed_at" json:"pushedAt,omitempty"`
	DeliveredAt   *time.Time   `gorm:"type:timestamptz;column:delivered_at" json:"deliveredAt,omitempty"`
	NextRetryAt   *time.Time   `gorm:"type:timestamptz;column:next_retry_at" json:"nextRetryAt,omitempty"`
	LastAttemptAt *time.Time   `gorm:"type:timestamptz;column:last_attempt_at" json:"lastAttemptAt,omitempty"`
	LastErrorType *string      `gorm:"type:text;column:last_error_type" json:"lastErrorType,omitempty"`
	ErrorMessages []ErrorEntry `gorm:"type:jsonb;column:error_messages;serializer:json" json:"errorMessages,omitempty"`
	CreatedAt     time.Time    `gorm:"type:timestamptz;not null;default:now();column:created_at" json:"createdAt"`
	UpdatedAt     time.Time    `gorm:"type:timestamptz;not null;default:now();column:updated_at" json:"updatedAt"`
}

type Store[D Delivery] interface {
	GetByID(ctx context.Context, id string) (D, error)
	Save(ctx context.Context, d D) error
	// MarkPushed stamps pushed_at (once) to record that the record has been
	// submitted into the delivery system. It is a targeted column update, NOT a
	// full Save, so it never clobbers a concurrent consumer's status write; and
	// it only sets the column when still null, so retry re-pushes are no-ops.
	MarkPushed(ctx context.Context, id string) error
	// ListDue returns records that need pushing or (re)processing as of now:
	// unpushed outbox rows, due retries, pushed-but-never-consumed rows, and
	// stuck in_progress rows.
	ListDue(ctx context.Context, now time.Time) ([]D, error)
}

type TaskMessage struct {
	DeliveryID  string    `json:"deliveryId"`
	PublishedAt time.Time `json:"publishedAt"`
}