lib

package
v0.3.0-20260803204450-... Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Aug 3, 2026 License: Apache-2.0 Imports: 9 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func FormatJSON

func FormatJSON(w io.Writer, v any) error

FormatJSON marshals v as indented JSON and writes it to w.

func FormatMillis

func FormatMillis(ms int64) string

FormatMillis converts an epoch-millisecond timestamp to a human-readable string. Returns "-" for zero values (no timestamp set).

func FormatTable

func FormatTable(w io.Writer, headers []string, rows [][]string)

FormatTable writes rows as an aligned text table to w. headers defines column names. rows is a slice of slices where each inner slice corresponds to one row's cell values (pre-formatted as strings).

Types

type AdminStore

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

AdminStore provides read-only inspection and targeted admin mutations for the MySQL queue tables.

func NewAdminStore

func NewAdminStore(db *sql.DB) *AdminStore

NewAdminStore creates a new AdminStore backed by the given database connection.

func (*AdminStore) ConsumerLag

func (s *AdminStore) ConsumerLag(ctx context.Context, topic string) ([]LagInfo, error)

ConsumerLag returns per-partition lag for each consumer group on a topic. Lag = max message offset in partition - consumer group's acked offset.

func (*AdminStore) DeleteMessage

func (s *AdminStore) DeleteMessage(ctx context.Context, topic string, messageID string) (int64, error)

DeleteMessage deletes a specific message by topic and ID.

func (*AdminStore) GetTopicStats

func (s *AdminStore) GetTopicStats(ctx context.Context, topic string, dlqSuffix string) (TopicStats, error)

GetTopicStats returns detailed statistics for a topic.

func (*AdminStore) InspectMessage

func (s *AdminStore) InspectMessage(ctx context.Context, topic string, messageID string) (MessageDetail, bool, error)

InspectMessage returns full message details including payload and DLQ fields.

func (*AdminStore) ListLeases

func (s *AdminStore) ListLeases(ctx context.Context) ([]LeaseInfo, error)

ListLeases returns all partition leases.

func (*AdminStore) ListMessages

func (s *AdminStore) ListMessages(ctx context.Context, topic string, partition string, limit int) ([]MessageSummary, error)

ListMessages returns messages for a topic, optionally filtered by partition.

func (*AdminStore) ListOffsets

func (s *AdminStore) ListOffsets(ctx context.Context, consumerGroup string) ([]OffsetInfo, error)

ListOffsets returns consumer group offsets, optionally filtered by group.

func (*AdminStore) ListTopics

func (s *AdminStore) ListTopics(ctx context.Context) ([]TopicInfo, error)

ListTopics returns all topics with their message counts.

func (*AdminStore) PurgeTopic

func (s *AdminStore) PurgeTopic(ctx context.Context, topic string) (int64, error)

PurgeTopic deletes all messages for a topic.

func (*AdminStore) ReleaseLease

func (s *AdminStore) ReleaseLease(ctx context.Context, consumerGroup, topic, partition string) (int64, error)

ReleaseLease force-releases a partition lease.

func (*AdminStore) RequeueDLQ

func (s *AdminStore) RequeueDLQ(ctx context.Context, topic string, messageID string, dlqSuffix string) error

RequeueDLQ moves a message from the DLQ topic back to its original topic. This is done transactionally: read from DLQ, insert into original topic, delete from DLQ.

func (*AdminStore) ResetOffset

func (s *AdminStore) ResetOffset(ctx context.Context, consumerGroup, topic, partition string, offset int64) (int64, error)

ResetOffset updates the acked offset for a consumer group/topic/partition.

func (*AdminStore) StaleLeases

func (s *AdminStore) StaleLeases(ctx context.Context, thresholdMs int64) ([]LeaseInfo, error)

StaleLeases returns leases whose lease_renewed_at is older than the threshold. thresholdMs is the staleness threshold in milliseconds — leases not renewed within this duration from now are considered stale.

type LagInfo

type LagInfo struct {
	// ConsumerGroup is the consumer group name
	ConsumerGroup string
	// Topic is the topic being consumed
	Topic string
	// PartitionKey is the partition being consumed
	PartitionKey string
	// LatestOffset is the highest message offset in this partition
	LatestOffset int64
	// AckedOffset is the last acked offset for this consumer group
	AckedOffset int64
	// Lag is the number of unprocessed messages (LatestOffset - AckedOffset)
	Lag int64
}

LagInfo contains consumer lag information for a single partition.

type LeaseInfo

type LeaseInfo struct {
	// ConsumerGroup is the consumer group name
	ConsumerGroup string
	// Topic is the topic being consumed
	Topic string
	// PartitionKey is the partition that is leased
	PartitionKey string
	// LeasedBy is the worker that owns the lease
	LeasedBy string
	// LeasedAt is the epoch milliseconds when the lease was acquired
	LeasedAt int64
	// LeaseRenewedAt is the epoch milliseconds of the last renewal
	LeaseRenewedAt int64
}

LeaseInfo contains partition lease information.

type MessageDetail

type MessageDetail struct {
	MessageSummary
	// Payload is the message body
	Payload []byte
	// Metadata contains key-value pairs for message attributes
	Metadata map[string]string
	// FailedAt is epoch milliseconds when the message failed (0 for normal)
	FailedAt int64
	// FailureCount is total failures before DLQ move (0 for normal)
	FailureCount int
	// LastError is the error message from final failure
	LastError string
	// OriginalTopic is where the message originally failed
	OriginalTopic string
}

MessageDetail contains all message fields including payload and DLQ info.

type MessageSummary

type MessageSummary struct {
	// Offset is the auto-incrementing sequence number
	Offset int64
	// ID is the unique message identifier
	ID string
	// Topic identifies the queue type
	Topic string
	// PartitionKey determines message distribution
	PartitionKey string
	// CreatedAt is the epoch milliseconds when the message was created
	CreatedAt int64
	// PublishedAt is the epoch milliseconds when the message was published
	PublishedAt int64
}

MessageSummary contains a subset of message fields for listing.

type OffsetInfo

type OffsetInfo struct {
	// ConsumerGroup is the consumer group name
	ConsumerGroup string
	// Topic is the topic being consumed
	Topic string
	// PartitionKey is the partition being consumed
	PartitionKey string
	// OffsetAcked is the last successfully acked offset
	OffsetAcked int64
	// UpdatedAt is the epoch milliseconds of the last update
	UpdatedAt int64
}

OffsetInfo contains consumer group offset information.

type TopicInfo

type TopicInfo struct {
	// Topic is the queue topic name
	Topic string
	// MessageCount is the number of messages in this topic
	MessageCount int64
}

TopicInfo contains a topic name and its message count.

type TopicStats

type TopicStats struct {
	// Topic is the queue topic name
	Topic string
	// TotalMessages is the total number of messages
	TotalMessages int64
	// DLQCount is the number of messages in the DLQ for this topic
	DLQCount int64
	// PartitionCount is the number of distinct partitions
	PartitionCount int64
	// ConsumerGroupCount is the number of consumer groups consuming this topic
	ConsumerGroupCount int64
}

TopicStats contains detailed statistics for a topic.

Jump to

Keyboard shortcuts

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