domain

Documentation

Index

• Variables
• func NewLogger(logLevel string) *slog.Logger
• type Config
  • func NewConfig() *Config
• type HealthCheck
• type HealthCheckRepository
• type HealthCheckService
• type Message
  • func (m *Message) Ack(now time.Time)
  • func (m *Message) DeliverySetup(queue *Queue, now time.Time)
  • func (m *Message) Enqueue(queue *Queue, now time.Time)
  • func (m *Message) Nack(now time.Time, visibilityTimeoutSeconds uint)
  • func (m Message) Validate() error
• type MessageRepository
• type MessageService
• type Queue
  • func (q Queue) Validate() error
• type QueueRepository
• type QueueService
• type QueueStats
• type Subscription
  • func (s *Subscription) ShouldCreateMessage(message *Message) bool
  • func (s Subscription) Validate() error
• type SubscriptionRepository
• type SubscriptionService
• type Topic
  • func (t Topic) Validate() error
• type TopicRepository
• type TopicService

Variables

var (
	// ErrQueueAlreadyExists is returned when the queue already exists.
	ErrQueueAlreadyExists = errors.New("queue already exists")
	// ErrQueueNotFound is returned when the queue is not found.
	ErrQueueNotFound = errors.New("queue not found")
	// ErrMessageAlreadyExists is returned when the message already exists.
	ErrMessageAlreadyExists = errors.New("message already exists")
	// ErrMessageNotFound is returned when the message is not found.
	ErrMessageNotFound = errors.New("message not found")
	// ErrTopicAlreadyExists is returned when the topic already exists.
	ErrTopicAlreadyExists = errors.New("topic already exists")
	// ErrTopicNotFound is returned when the topic is not found.
	ErrTopicNotFound = errors.New("topic not found")
	// ErrSubscriptionAlreadyExists is returned when the subscription already exists.
	ErrSubscriptionAlreadyExists = errors.New("subscription already exists")
	// ErrSubscriptionNotFound is returned when the subscription is not found.
	ErrSubscriptionNotFound = errors.New("subscription not found")
)

Functions

func NewLogger

func NewLogger(logLevel string) *slog.Logger

NewLogger returns a configured JSON logger.

Types

type Config

type Config struct {
	Testing                        bool
	LogLevel                       string
	ServerHost                     string
	ServerPort                     uint
	ServerReadHeaderTimeoutSeconds uint
	MetricsHost                    string
	MetricsPort                    uint
	DatabaseURL                    string
	TestDatabaseURL                string
	DatabaseMinConns               uint
	DatabaseMaxConns               uint
	QueueMaxNumberOfMessages       uint
}

Config holds all application configuration data.

func NewConfig

func NewConfig() *Config

NewConfig returns a Config with values loaded from environment variables.

type HealthCheck

type HealthCheck struct {
	Success bool `json:"success"`
}

HealthCheck entity.

type HealthCheckRepository

type HealthCheckRepository interface {
	Check(ctx context.Context) (*HealthCheck, error)
}

HealthCheckRepository is the repository interface for the HealthCheck entity.

type HealthCheckService

type HealthCheckService interface {
	Check(ctx context.Context) (*HealthCheck, error)
}

HealthCheckService is the service interface for the HealthCheck entity.

type Message

type Message struct {
	ID               string            `json:"id" db:"id"`
	QueueID          string            `json:"queue_id" db:"queue_id"`
	Label            *string           `json:"label" db:"label" form:"label"`
	Body             string            `json:"body" db:"body" form:"body"`
	Attributes       map[string]string `json:"attributes" db:"attributes" form:"attributes"`
	DeliveryAttempts uint              `json:"delivery_attempts" db:"delivery_attempts"`
	ExpiredAt        time.Time         `json:"-" db:"expired_at"`
	ScheduledAt      time.Time         `json:"-" db:"scheduled_at"`
	CreatedAt        time.Time         `json:"created_at" db:"created_at"`
	UpdatedAt        time.Time         `json:"-" db:"updated_at"`
}

Message entity.

func (*Message) Ack

func (m *Message) Ack(now time.Time)

func (*Message) DeliverySetup

func (m *Message) DeliverySetup(queue *Queue, now time.Time)

func (*Message) Enqueue

func (m *Message) Enqueue(queue *Queue, now time.Time)

func (*Message) Nack

func (m *Message) Nack(now time.Time, visibilityTimeoutSeconds uint)

func (Message) Validate

func (m Message) Validate() error

type MessageRepository

type MessageRepository interface {
	CreateMany(ctx context.Context, messages []*Message) error
	Create(ctx context.Context, message *Message) error
	Get(ctx context.Context, id string) (*Message, error)
	List(ctx context.Context, queue *Queue, label *string, limit uint) ([]*Message, error)
	Ack(ctx context.Context, id string) error
	Nack(ctx context.Context, id string, visibilityTimeoutSeconds uint) error
}

MessageRepository is the repository interface for the Message entity.

type MessageService

type MessageService interface {
	Create(ctx context.Context, message *Message) error
	List(ctx context.Context, queueID string, label *string, limit uint) ([]*Message, error)
	Ack(ctx context.Context, id string) error
	Nack(ctx context.Context, id string, visibilityTimeoutSeconds uint) error
}

MessageService is the service interface for the Message entity.

type Queue

type Queue struct {
	ID                      string    `json:"id" db:"id" form:"id"`
	AckDeadlineSeconds      uint      `json:"ack_deadline_seconds" db:"ack_deadline_seconds" form:"ack_deadline_seconds"`
	MessageRetentionSeconds uint      `json:"message_retention_seconds" db:"message_retention_seconds" form:"message_retention_seconds"`
	DeliveryDelaySeconds    uint      `json:"delivery_delay_seconds" db:"delivery_delay_seconds" form:"delivery_delay_seconds"`
	CreatedAt               time.Time `json:"created_at" db:"created_at"`
	UpdatedAt               time.Time `json:"updated_at" db:"updated_at"`
}

Queue entity.

func (Queue) Validate

func (q Queue) Validate() error

type QueueRepository

type QueueRepository interface {
	Create(ctx context.Context, queue *Queue) error
	Update(ctx context.Context, queue *Queue) error
	Get(ctx context.Context, id string) (*Queue, error)
	List(ctx context.Context, offset, limit uint) ([]*Queue, error)
	Delete(ctx context.Context, id string) error
	Stats(ctx context.Context, id string) (*QueueStats, error)
	Purge(ctx context.Context, id string) error
	Cleanup(ctx context.Context, id string) error
}

QueueRepository is the repository interface for the Queue entity.

type QueueService

type QueueService interface {
	Create(ctx context.Context, queue *Queue) error
	Update(ctx context.Context, queue *Queue) error
	Get(ctx context.Context, id string) (*Queue, error)
	List(ctx context.Context, offset, limit uint) ([]*Queue, error)
	Delete(ctx context.Context, id string) error
	Stats(ctx context.Context, id string) (*QueueStats, error)
	Purge(ctx context.Context, id string) error
	Cleanup(ctx context.Context, id string) error
}

QueueService is the service interface for the Queue entity.

type QueueStats

type QueueStats struct {
	NumUndeliveredMessages         uint `json:"num_undelivered_messages"`
	OldestUnackedMessageAgeSeconds uint `json:"oldest_unacked_message_age_seconds"`
}

QueueStats entity.

type Subscription

type Subscription struct {
	ID             string              `json:"id" db:"id" form:"id"`
	TopicID        string              `json:"topic_id" db:"topic_id" form:"topic_id"`
	QueueID        string              `json:"queue_id" db:"queue_id" form:"queue_id"`
	MessageFilters map[string][]string `json:"message_filters" db:"message_filters" form:"message_filters"`
	CreatedAt      time.Time           `json:"created_at" db:"created_at"`
}

Subscription entity.

func (*Subscription) ShouldCreateMessage

func (s *Subscription) ShouldCreateMessage(message *Message) bool

func (Subscription) Validate

func (s Subscription) Validate() error

type SubscriptionRepository

type SubscriptionRepository interface {
	Create(ctx context.Context, subscription *Subscription) error
	Get(ctx context.Context, id string) (*Subscription, error)
	List(ctx context.Context, offset, limit uint) ([]*Subscription, error)
	ListByTopic(ctx context.Context, topicID string, offset, limit uint) ([]*Subscription, error)
	Delete(ctx context.Context, id string) error
}

SubscriptionRepository is the repository interface for the Subscription entity.

type SubscriptionService

type SubscriptionService interface {
	Create(ctx context.Context, subscription *Subscription) error
	Get(ctx context.Context, id string) (*Subscription, error)
	List(ctx context.Context, offset, limit uint) ([]*Subscription, error)
	Delete(ctx context.Context, id string) error
}

SubscriptionService is the service interface for the Subscription entity.

type Topic

type Topic struct {
	ID        string    `json:"id" db:"id" form:"id"`
	CreatedAt time.Time `json:"created_at" db:"created_at"`
}

Topic entity.

func (Topic) Validate

func (t Topic) Validate() error

type TopicRepository

type TopicRepository interface {
	Create(ctx context.Context, topic *Topic) error
	Get(ctx context.Context, id string) (*Topic, error)
	List(ctx context.Context, offset, limit uint) ([]*Topic, error)
	Delete(ctx context.Context, id string) error
}

TopicRepository is the repository interface for the Topic entity.

type TopicService

type TopicService interface {
	Create(ctx context.Context, topic *Topic) error
	Get(ctx context.Context, id string) (*Topic, error)
	List(ctx context.Context, offset, limit uint) ([]*Topic, error)
	Delete(ctx context.Context, id string) error
	CreateMessage(ctx context.Context, topicID string, message *Message) error
}

TopicService is the service interface for the Topic entity.