sqlstore

package

v0.1.0 Latest Latest Go to latest Published: Oct 4, 2026 License: MIT Imports: 10 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/Ali127Dev/xoutbox

Links

Open Source Insights

Documentation ¶

Overview ¶

Package sqlstore holds the dialect-independent SQL store implementation.

Index ¶

Constants
func Placeholders(n int) string
func Scan(rows *sql.Rows) ([]xoutbox.Event, error)
type Base
- func New(db *sql.DB, d Dialect, opts []Option) *Base
type Dialect
type Option

Constants ¶

View Source

const Cols = "seq, id, topic, partition_key, event_type, payload, headers, attempts, max_attempts, {created}"

Cols is the column list returned by claims, matching Scan.

Variables ¶

This section is empty.

Functions ¶

func Placeholders ¶

func Placeholders(n int) string

Placeholders returns "?, ?, ..." with n entries.

func Scan ¶

func Scan(rows *sql.Rows) ([]xoutbox.Event, error)

Scan reads Cols rows sorted by seq.

Types ¶

type Base ¶

type Base struct {
	DB          *sql.DB
	Table       string
	Ordered     bool
	MaxAttempts int
	D           Dialect
}

Base implements everything except Claim.

func New ¶

func New(db *sql.DB, d Dialect, opts []Option) *Base

func (*Base) Cleanup ¶

func (b *Base) Cleanup(ctx context.Context, retention time.Duration) (int64, error)

func (*Base) Filter ¶

func (b *Base) Filter() string

Filter is the WHERE condition selecting claimable rows (table alias o).

func (*Base) Insert ¶

func (b *Base) Insert(ctx context.Context, ex xoutbox.Execer, events ...xoutbox.Event) error

Insert writes events using ex, which should be your business transaction.

func (*Base) MarkFailed ¶

func (b *Base) MarkFailed(ctx context.Context, owner string, f xoutbox.Failure) error

func (*Base) MarkPublished ¶

func (b *Base) MarkPublished(ctx context.Context, ids []string) error

func (*Base) Migrate ¶

func (b *Base) Migrate(ctx context.Context) error

Migrate creates the table and indexes if they do not exist.

func (*Base) Q ¶

func (b *Base) Q(s string) string

Q expands {t} {n} {now} {dur} {created} and rebinds placeholders.

func (*Base) Requeue ¶

func (b *Base) Requeue(ctx context.Context, ids ...string) (int64, error)

Requeue moves dead events back to pending with a fresh attempt budget. With no ids, every dead event is requeued.

func (*Base) SchemaSQL ¶

func (b *Base) SchemaSQL() string

SchemaSQL returns the DDL for the configured table.

func (*Base) Stats ¶

func (b *Base) Stats(ctx context.Context) (xoutbox.Stats, error)

Stats returns backlog counters, cheap enough to scrape as metrics.

type Dialect ¶

type Dialect struct {
	Rebind  func(string) string
	Now     string                  // current timestamp expression
	Dur     string                  // duration expression with exactly one placeholder
	DurArg  func(time.Duration) any // value bound to Dur
	Created string                  // created_at as unix microseconds
	Age     string                  // seconds since MIN(created_at)
	Delete  string                  // deletes up to 1000 published rows older than {dur}
	Schema  string
}

Dialect captures the SQL differences between databases.

type Option ¶

type Option func(*Base)

Option configures a Base.

func WithMaxAttempts ¶

func WithMaxAttempts(n int) Option

func WithOrdering ¶

func WithOrdering() Option

func WithTable ¶

func WithTable(name string) Option

Source Files ¶

View all Source files

sqlstore.go

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL