messages

package
v1.41.2 Latest Latest
Warning

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

Go to latest
Published: May 14, 2026 License: Apache-2.0 Imports: 29 Imported by: 0

Documentation

Index

Constants

View Source
const (
	ProcessingStatePending   ProcessingState = "Pending"
	ProcessingStateProcessed ProcessingState = "Processed"
	ProcessingStateSent      ProcessingState = "Sent"
	ProcessingStateDelivered ProcessingState = "Delivered"
	ProcessingStateFailed    ProcessingState = "Failed"

	MessageTypeText MessageType = "Text"
	MessageTypeData MessageType = "Data"
)

Variables

View Source
var (
	ErrMessageAlreadyExists  = errors.New("duplicate id")
	ErrMessageNotFound       = errors.New("message not found")
	ErrMultipleMessagesFound = errors.New("multiple messages found")
	ErrNoContent             = errors.New("no text or data content")

	ErrQueueLimitExceeded = errors.New("queue limits exceeded")
)

Functions

func Migrate added in v1.24.0

func Migrate(db *gorm.DB) error

func Module

func Module() fx.Option

Types

type Config

type Config struct {
	HashingInterval time.Duration
	CacheTTL        time.Duration

	Queue QueueConfig
}

type DataMessageContent added in v1.24.0

type DataMessageContent = smsgateway.DataMessage

type EnqueueOptions

type EnqueueOptions struct {
	SkipPhoneValidation bool
}

type HashedMessageContent added in v1.38.0

type HashedMessageContent = smsgateway.HashedMessage

type Limiter added in v1.41.0

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

func NewLimiter added in v1.41.0

func NewLimiter(
	config QueueConfig,
	messages *Repository,
	storage cache.Cache,
	metrics *metrics,
	logger *zap.Logger,
) *Limiter

func (*Limiter) Check added in v1.41.0

func (l *Limiter) Check(ctx context.Context, deviceID string) error

func (*Limiter) Refresh added in v1.41.0

func (l *Limiter) Refresh(_ context.Context, deviceID string) error

func (*Limiter) Run added in v1.41.0

func (l *Limiter) Run(ctx context.Context)

type Message added in v1.24.0

type Message struct {
	MessageInput

	CreatedAt time.Time
}

type MessageContent added in v1.38.0

type MessageContent struct {
	TextContent *TextMessageContent `json:"textContent,omitempty"`
	DataContent *DataMessageContent `json:"dataContent,omitempty"`
}

type MessageInput added in v1.38.0

type MessageInput struct {
	MessageContent

	ID string

	PhoneNumbers []string
	IsEncrypted  bool

	SimNumber          *uint8
	WithDeliveryReport *bool
	TTL                *uint64
	ValidUntil         *time.Time
	ScheduleAt         *time.Time
	Priority           smsgateway.MessagePriority
}

type MessageState added in v1.24.0

type MessageState struct {
	MessageStateInput
	MessageStateContent

	DeviceID    string `json:"deviceId"`    // Device ID
	IsHashed    bool   `json:"isHashed"`    // Hashed
	IsEncrypted bool   `json:"isEncrypted"` // Encrypted
}

type MessageStateContent added in v1.38.0

type MessageStateContent struct {
	MessageContent

	HashedContent *HashedMessageContent `json:"hashedContent,omitempty"`
}

type MessageStateInput added in v1.38.0

type MessageStateInput struct {
	ID         string                      `json:"id"`         // Message ID
	State      ProcessingState             `json:"state"`      // State
	Recipients []smsgateway.RecipientState `json:"recipients"` // Recipients states
	States     map[string]time.Time        `json:"states"`     // History of states
}

type MessageType added in v1.24.0

type MessageType string

type Order added in v1.34.3

type Order string

Order defines supported ordering for message selection. Valid values: "lifo" (default), "fifo".

const (
	// MessagesOrderLIFO orders messages newest-first within the same priority (default).
	MessagesOrderLIFO Order = "lifo"
	// MessagesOrderFIFO orders messages oldest-first within the same priority.
	MessagesOrderFIFO Order = "fifo"
)

type ProcessingState added in v1.24.0

type ProcessingState string

type QueueConfig added in v1.41.0

type QueueConfig struct {
	MaxPending    int64
	MaxPendingAge time.Duration
	MaxFailed     int
	MaxFailedAge  time.Duration

	StatsRefreshInterval time.Duration
	StatsCacheTTL        time.Duration
}

func (QueueConfig) IsEmpty added in v1.41.0

func (c QueueConfig) IsEmpty() bool

type Repository added in v1.34.0

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

func NewRepository added in v1.34.0

func NewRepository(db *gorm.DB) *Repository

func (*Repository) Cleanup added in v1.34.0

func (r *Repository) Cleanup(ctx context.Context, until time.Time) (int64, error)

func (*Repository) CountPending added in v1.41.0

func (r *Repository) CountPending(ctx context.Context, deviceID string) (int64, error)

func (*Repository) GetOldestPendingTime added in v1.41.0

func (r *Repository) GetOldestPendingTime(ctx context.Context, deviceID string) (*time.Time, error)

func (*Repository) GetStatesInTimeWindow added in v1.41.0

func (r *Repository) GetStatesInTimeWindow(
	ctx context.Context,
	deviceID string,
	since time.Time,
	limit int,
) ([]ProcessingState, error)

func (*Repository) HashProcessed added in v1.34.0

func (r *Repository) HashProcessed(ctx context.Context, ids []uint64) (int64, error)

func (*Repository) Insert added in v1.34.0

func (r *Repository) Insert(message *messageModel) error

func (*Repository) UpdateState added in v1.34.0

func (r *Repository) UpdateState(message *messageModel) error

type SelectFilter added in v1.34.3

type SelectFilter struct {
	ExtID     string
	UserID    string
	DeviceID  string
	StartDate time.Time
	EndDate   time.Time
	State     ProcessingState
}

func (*SelectFilter) WithDateRange added in v1.34.3

func (f *SelectFilter) WithDateRange(start, end time.Time) *SelectFilter

func (*SelectFilter) WithDeviceID added in v1.34.3

func (f *SelectFilter) WithDeviceID(deviceID string) *SelectFilter

func (*SelectFilter) WithExtID added in v1.34.3

func (f *SelectFilter) WithExtID(extID string) *SelectFilter

func (*SelectFilter) WithState added in v1.34.3

func (f *SelectFilter) WithState(state ProcessingState) *SelectFilter

func (*SelectFilter) WithUserID added in v1.34.3

func (f *SelectFilter) WithUserID(userID string) *SelectFilter

type SelectOptions added in v1.34.3

type SelectOptions struct {
	WithRecipients bool
	WithDevice     bool
	WithStates     bool
	WithContent    bool

	// OrderBy sets the retrieval order for pending messages.
	// Empty (zero) value defaults to "lifo".
	OrderBy Order

	Limit  int
	Offset int
}

func (*SelectOptions) IncludeContent added in v1.38.0

func (o *SelectOptions) IncludeContent() *SelectOptions

func (*SelectOptions) IncludeDevice added in v1.34.3

func (o *SelectOptions) IncludeDevice() *SelectOptions

func (*SelectOptions) IncludeRecipients added in v1.34.3

func (o *SelectOptions) IncludeRecipients() *SelectOptions

func (*SelectOptions) IncludeStates added in v1.34.3

func (o *SelectOptions) IncludeStates() *SelectOptions

func (*SelectOptions) WithLimit added in v1.34.3

func (o *SelectOptions) WithLimit(limit int) *SelectOptions

func (*SelectOptions) WithOffset added in v1.34.3

func (o *SelectOptions) WithOffset(offset int) *SelectOptions

func (*SelectOptions) WithOrderBy added in v1.34.3

func (o *SelectOptions) WithOrderBy(order Order) *SelectOptions

type Service

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

func NewService

func NewService(
	config Config,

	limiter *Limiter,
	messages *Repository,
	eventsSvc *events.Service,

	metrics *metrics,
	cache *stateCache,
	hashingTask *hashingWorker,

	logger *zap.Logger,
	idgen db.IDGen,
) *Service

func (*Service) Enqueue added in v1.20.0

func (s *Service) Enqueue(
	ctx context.Context,
	device models.Device,
	message MessageInput,
	opts EnqueueOptions,
) (*MessageState, error)

func (*Service) GetState

func (s *Service) GetState(userID string, id string) (*MessageState, error)

func (*Service) RunBackgroundTasks

func (s *Service) RunBackgroundTasks(ctx context.Context, wg *sync.WaitGroup)

func (*Service) SelectPending

func (s *Service) SelectPending(deviceID string, order Order) ([]Message, error)

func (*Service) SelectStates added in v1.27.0

func (s *Service) SelectStates(
	userID string,
	filter SelectFilter,
	options SelectOptions,
) ([]MessageState, int64, error)

func (*Service) UpdateState

func (s *Service) UpdateState(device *models.Device, message MessageStateInput) error

type TextMessageContent added in v1.24.0

type TextMessageContent = smsgateway.TextMessage

type ValidationError added in v1.34.3

type ValidationError string

func (ValidationError) Error added in v1.34.3

func (e ValidationError) Error() string

Jump to

Keyboard shortcuts

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