inbox

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

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

type HoldRow struct {
	Consumer   string
	Stream     string
	Offset     int64
	TenantID   string
	Data       []byte
	Properties []byte
	HeldAt     time.Time
}

type HoldStats struct {
	Tenants         int64
	Rows            int64
	OldestHeldSince time.Time
}

type HoldStore interface {
	// Park inserts the row and marks its tenant held, in tx. It is idempotent on
	// the row's identity: a redelivery of an already-parked offset reports
	// inserted=false rather than failing.
	Park(ctx context.Context, tx dbtypes.Tx, row *HoldRow) (inserted bool, err error)

	// HeldTenants lists the tenants currently held for a consumer.
	HeldTenants(ctx context.Context, db dbtypes.Interface, consumer string) ([]string, error)

	// ListTenants returns every held tenant's full drain state, oldest first.
	ListTenants(ctx context.Context, db dbtypes.Interface, consumer string) ([]HoldTenant, error)

	// DueTenants lists held tenants whose next attempt has come and whose lease is
	// free, oldest first.
	DueTenants(ctx context.Context, db dbtypes.Interface, consumer string, limit int) ([]HoldTenant, error)

	// AcquireLease takes or renews the drain lease for one tenant, reporting false
	// when another owner holds a live one.
	AcquireLease(ctx context.Context, db dbtypes.Interface, consumer, tenant, owner string, lease time.Duration) (bool, error)

	// ReleaseLease drops a lease this owner holds.
	ReleaseLease(ctx context.Context, db dbtypes.Interface, consumer, tenant, owner string) error

	// NextRows returns the tenant's rows in (stream, offset) order.
	NextRows(ctx context.Context, db dbtypes.Interface, consumer, tenant string, limit int) ([]HoldRow, error)

	// DeleteRow removes one replayed row. Fenced by the lease: a write affecting no
	// rows means the lease was lost, and the caller discards the replay's outcome.
	DeleteRow(ctx context.Context, db dbtypes.Interface, consumer, stream string, offset int64, tenant, owner string) (deleted bool, err error)

	// Defer records a failed replay: one more attempt, the next one backed off,
	// the error bounded, the lease cleared. Fenced by the lease.
	Defer(ctx context.Context, db dbtypes.Interface, consumer, tenant, owner string, backoff time.Duration, lastErr string) (updated bool, err error)

	// Release deletes the tenant's marker, and only when no rows remain. Fenced by
	// the lease.
	Release(ctx context.Context, db dbtypes.Interface, consumer, tenant, owner string) (released bool, err error)

	// Stats reports what the gauges publish for one consumer.
	Stats(ctx context.Context, db dbtypes.Interface, consumer string) (HoldStats, error)

	// CreateTable creates both tables and their indexes if they do not exist.
	CreateTable(ctx context.Context, db dbtypes.Interface) error
}

type HoldTenant struct {
	Consumer      string
	TenantID      string
	HeldSince     time.Time
	Attempts      int
	NextAttemptAt time.Time
	LastError     string
}

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

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

type Record struct {
	TenantID    string
	EventID     string
	ProcessedAt time.Time
}

type Store interface {
	// MarkProcessed records (tenant_id, event_id) within the given transaction.
	// It returns inserted=true the first time an id is seen and inserted=false on
	// a duplicate (the id was already processed).
	MarkProcessed(ctx context.Context, tx dbtypes.Tx, rec Record) (inserted bool, err error)

	// DeleteProcessed removes ledger rows processed before the given time.
	// Returns the number of rows deleted.
	DeleteProcessed(ctx context.Context, db dbtypes.Interface, before time.Time) (int64, error)

	// CreateTable creates the inbox table and its index if they do not exist.
	// Used for auto-migration when inbox.autocreatetable is true.
	CreateTable(ctx context.Context, db dbtypes.Interface) error
}