manager

type ChannelLock ¶

type ChannelLock interface {
	// Key returns the key associated with this lock.
	Key() string

	// Release releases the lock.
	// It does not accept a context parameter because releasing a lock is a commitment
	// that must complete regardless of the caller's context state. Each implementation
	// is responsible for managing its own timeouts internally.
	Release() error
}

ChannelLock represents a single lock on a channel.

type ChannelLocker ¶

type ChannelLocker interface {
	// Obtain tries to obtain a lock for the given key (channel ID) with the specified TTL (time-to-live).
	// It will retry obtaining the lock until the maxWait duration is reached.
	// If the lock is successfully obtained, it returns a ChannelLock instance.
	// If the lock cannot be obtained within the maxWait duration, it returns ErrChannelLockUnavailable.
	Obtain(ctx context.Context, key string, ttl time.Duration, maxWait time.Duration) (ChannelLock, error)
}

ChannelLocker is an interface for obtaining and releasing distributed locks on channels. It is used to ensure that only one process can perform operations on a channel at a time. This is particularly useful in a distributed environment where multiple instances of the manager may be running. Use the RedisChannelLocker implementation for a production-ready solution, using Redis as the backing store. Use the NoopChannelLocker for testing or when locking is not required (i.e in a single instance setup).

type FifoQueue ¶

type FifoQueue interface {
	// Name returns the name of the queue.
	Name() string

	// Send sends a single message to the queue.
	//
	// slackChannelID is the Slack channel to which the message belongs.
	// A queue implementation should use this value to partition the queue (i.e. group ID in an AWS SQS Fifo queue),
	// but it is not required.
	//
	// dedupID is a unique identifier for the message.
	// A queue implementation should use this value to deduplicate messages, but it is not required.
	//
	// body is the json formatted message body.
	Send(ctx context.Context, slackChannelID, dedupID, body string) error

	// Receive receives messages from the queue, until the context is cancelled.
	// Messages are sent to the provided channel.
	// The channel must be closed by the implementation before returning, typically when the context is cancelled or a fatal error occurs.
	Receive(ctx context.Context, sinkCh chan<- *types.FifoQueueItem) error
}

FifoQueue is an interface for interacting with a fifo queue.

type HTTPWebhookHandler ¶

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

HTTPWebhookHandler is an implementation of the WebhookHandler interface that sends webhooks via HTTP. This is the default webhook handler when no other webhook handlers are configured.

func NewHTTPWebhookHandler(logger types.Logger) *HTTPWebhookHandler

func (h *HTTPWebhookHandler) HandleWebhook(ctx context.Context, target string, data *types.WebhookCallback, logger types.Logger) error

func (h *HTTPWebhookHandler) ShouldHandleWebhook(_ context.Context, target string) bool

func (h *HTTPWebhookHandler) WithRequestTimeout(timeout time.Duration) *HTTPWebhookHandler

type LocalRateLimitGate ¶ added in v0.4.0

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

LocalRateLimitGate is an in-process RateLimitGate implementation suitable for single-instance deployments. For multi-instance deployments use RedisRateLimitGate so that all instances respect the same rate-limit window.

func NewLocalRateLimitGate(logger types.Logger, maxDrainWait time.Duration) *LocalRateLimitGate

func (g *LocalRateLimitGate) IsBlocked(_ context.Context) (bool, error)

func (g *LocalRateLimitGate) SetReadyCheck(fn func() bool)

func (g *LocalRateLimitGate) Signal(_ context.Context, until time.Time) error

func (g *LocalRateLimitGate) Wait(ctx context.Context) error

type Manager ¶

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

Manager processes alerts from an alert queue and manages the lifecycle of Slack issues. It connects to Slack via Socket Mode, coordinates per-channel workers, and persists issue state through the injected database.

Create a Manager with New and start it with Manager.Run. Settings can be updated at runtime via Manager.UpdateSettings without restarting the service.

func New(db types.DB, alertQueue FifoQueue, commandQueue FifoQueue, logger types.Logger, cfg *config.ManagerConfig) *Manager

func (m *Manager) RegisterWebhookHandler(handler WebhookHandler) *Manager

func (m *Manager) Run(ctx context.Context) error

func (m *Manager) UpdateSettings(settings *config.ManagerSettings) error

func (m *Manager) WithCacheStore(cacheStore store.StoreInterface) *Manager

func (m *Manager) WithLocker(locker ChannelLocker) *Manager

func (m *Manager) WithMetrics(metrics types.Metrics) *Manager

func (m *Manager) WithSettings(settings *config.ManagerSettings) *Manager

type NoopChannelLock ¶

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

NoopChannelLock is a no-op implementation of the ChannelLock interface.

func (n *NoopChannelLock) Key() string

func (n *NoopChannelLock) Release() error

type NoopChannelLocker ¶

type NoopChannelLocker struct{}

NoopChannelLocker is a no-op implementation of the ChannelLocker interface. It does not perform any locking and is used for testing or when locking is not required. Locking is typically always required when running in a distributed environment, such as in Kubernetes.

func (n *NoopChannelLocker) Obtain(_ context.Context, key string, _ time.Duration, _ time.Duration) (ChannelLock, error)

type NoopRateLimitGate ¶ added in v0.4.0

type NoopRateLimitGate struct{}

NoopRateLimitGate is a no-op RateLimitGate for tests and single-instance deployments that do not need distributed rate-limit coordination.

func (n *NoopRateLimitGate) IsBlocked(_ context.Context) (bool, error)

func (n *NoopRateLimitGate) SetReadyCheck(_ func() bool)

func (n *NoopRateLimitGate) Signal(_ context.Context, _ time.Time) error

func (n *NoopRateLimitGate) Wait(_ context.Context) error

type RateLimitGate ¶ added in v0.4.0

type RateLimitGate interface {
	// Signal records that a rate limit was hit and that callers should block
	// until `until` has passed. Implementations extend the window if `until`
	// is later than any previously recorded deadline.
	Signal(ctx context.Context, until time.Time) error

	// Wait blocks until the gate is open and Socket Mode is quiet. It returns
	// immediately when the gate is not blocked. Context cancellation is
	// propagated as an error.
	Wait(ctx context.Context) error

	// IsBlocked reports whether the gate is currently blocking callers.
	IsBlocked(ctx context.Context) (bool, error)

	// SetReadyCheck registers a function that returns true when Socket Mode is
	// quiet (no in-flight event handlers). Must be called once from [Manager.Run]
	// after the Slack client is connected.
	SetReadyCheck(fn func() bool)
}

RateLimitGate coordinates a global pause across all channel managers when a Slack 429 is detected. Channel managers call [RateLimitGate.Wait] before processing each work item; the gate blocks until the rate-limit window has expired and the Socket Mode handler is quiet (no in-flight event handlers).

type RedisChannelLock ¶

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

RedisChannelLock is an implementation of the ChannelLock interface that uses Redis locks. It represents a lock on a specific channel. It is returned by RedisChannelLocker when a lock is successfully obtained. The lock can be released using RedisChannelLock.Release, which will remove the lock from Redis.

func (l *RedisChannelLock) Key() string

func (l *RedisChannelLock) Release() error

type RedisChannelLocker ¶

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

RedisChannelLocker is an implementation of the ChannelLocker interface that uses Redis for distributed locking. It allows multiple instances of the manager to coordinate access to channels, ensuring that only one instance can perform operations on a channel at a time. The locker uses a key prefix to avoid conflicts with other locks in Redis. It also supports retry backoff for obtaining locks, allowing for configurable retry strategies. The default retry backoff is 2 seconds, but it can be configured using RedisChannelLocker.WithRetryBackoff.

func NewRedisChannelLocker(client *redis.Client) *RedisChannelLocker

func (r *RedisChannelLocker) Obtain(ctx context.Context, key string, ttl time.Duration, maxWait time.Duration) (ChannelLock, error)

func (r *RedisChannelLocker) WithKeyPrefix(prefix string) *RedisChannelLocker

func (r *RedisChannelLocker) WithMaxDrainWait(d time.Duration) *RedisChannelLocker

func (r *RedisChannelLocker) WithRetryBackoff(backoff time.Duration) *RedisChannelLocker

type RedisFifoQueue ¶

type RedisFifoQueue struct {
	*RedisFifoQueueProducer // Embedded for Send() and Name()
	// contains filtered or unexported fields
}

RedisFifoQueue implements the FifoQueue interface using Redis Streams. It uses one stream per Slack channel to ensure message ordering within a channel, mimicking the behavior of SQS FIFO message groups.

Multiple instances can consume from the same queue using Redis consumer groups, providing at-least-once delivery semantics. Messages for different channels can be processed in parallel across instances.

Strict per-channel ordering is enforced using distributed locks. When a consumer reads from a stream, it acquires a lock for that channel. Other consumers cannot read from that stream until the lock is released (after ack/nack). This ensures that messages for a single channel are processed strictly in order, even across multiple instances.

For write-only usage (e.g., API server), consider using RedisFifoQueueProducer instead.

func NewRedisFifoQueue(client redis.UniversalClient, locker ChannelLocker, name string, logger types.Logger, opts ...RedisFifoQueueOption) *RedisFifoQueue

func (q *RedisFifoQueue) Init() (*RedisFifoQueue, error)

func (q *RedisFifoQueue) Receive(ctx context.Context, sinkCh chan<- *types.FifoQueueItem) error

func (q *RedisFifoQueue) Send(ctx context.Context, slackChannelID, dedupID, body string) error

type RedisFifoQueueOption ¶

type RedisFifoQueueOption func(*RedisFifoQueueOptions)

RedisFifoQueueOption is a function that configures RedisFifoQueueOptions.

func WithClaimMinIdleTime(duration time.Duration) RedisFifoQueueOption

func WithConsumerGroup(group string) RedisFifoQueueOption

func WithKeyPrefix(prefix string) RedisFifoQueueOption

func WithLockTTL(ttl time.Duration) RedisFifoQueueOption

func WithMaxStreamLength(length int64) RedisFifoQueueOption

func WithPollInterval(interval time.Duration) RedisFifoQueueOption

func WithStreamInactivityTimeout(timeout time.Duration) RedisFifoQueueOption

func WithStreamRefreshInterval(interval time.Duration) RedisFifoQueueOption

type RedisFifoQueueOptions ¶

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

RedisFifoQueueOptions holds configuration for RedisFifoQueue.

type RedisFifoQueueProducer ¶

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

RedisFifoQueueProducer is a lightweight write-only queue producer. Use this when the service only needs to send messages (e.g., API server in a separate service). For full queue functionality including Receive(), use RedisFifoQueue instead.

This producer has minimal overhead: no consumer groups, no distributed locks, no in-process notification mechanism. It simply writes messages to Redis streams.

func NewRedisFifoQueueProducer(client redis.UniversalClient, name string, logger types.Logger, opts ...RedisFifoQueueOption) *RedisFifoQueueProducer

func (p *RedisFifoQueueProducer) Init() (*RedisFifoQueueProducer, error)

func (p *RedisFifoQueueProducer) Name() string

func (p *RedisFifoQueueProducer) Send(ctx context.Context, slackChannelID, _, body string) error

type RedisRateLimitGate ¶ added in v0.4.0

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

RedisRateLimitGate is a distributed RateLimitGate backed by Redis. It is suitable for multi-instance deployments where every instance must respect the same rate-limit window.

func NewRedisRateLimitGate(client redis.UniversalClient, logger types.Logger, keyPrefix string, maxDrainWait time.Duration) *RedisRateLimitGate

func (g *RedisRateLimitGate) IsBlocked(ctx context.Context) (bool, error)

func (g *RedisRateLimitGate) SetReadyCheck(fn func() bool)

func (g *RedisRateLimitGate) Signal(ctx context.Context, until time.Time) error

func (g *RedisRateLimitGate) Wait(ctx context.Context) error

type SlackClient ¶

type SlackClient interface {
	// GetChannelName returns the human-readable name of the given Slack channel, or an
	// empty string if the name cannot be resolved.
	GetChannelName(ctx context.Context, channelID string) string

	// IsAlertChannel reports whether channelID is a valid target for alerts.
	// Returns the channel name and any lookup error alongside the boolean result.
	IsAlertChannel(ctx context.Context, channelID string) (bool, string, error)

	// Update rebuilds and posts the consolidated issue summary message for all open
	// issues in the given channel.
	Update(ctx context.Context, channelID string, allChannelIssues []*models.Issue) error

	// UpdateSingleIssueWithThrottling updates the Slack message for one issue, applying
	// per-channel throttling when many issues are active to avoid flooding the API.
	UpdateSingleIssueWithThrottling(ctx context.Context, issue *models.Issue, reason string, issuesInChannel int) error

	// UpdateSingleIssue updates the Slack message for one issue without throttling.
	// Use [SlackClient.UpdateSingleIssueWithThrottling] when processing incoming alerts.
	UpdateSingleIssue(ctx context.Context, issue *models.Issue, reason string) error

	// Delete removes an issue's Slack message. When updateIfMessageHasReplies is true
	// and the post already has thread replies, the message is replaced with a tombstone
	// instead of being deleted outright. To delete an arbitrary post use [SlackClient.DeletePost].
	Delete(ctx context.Context, issue *models.Issue, reason string, updateIfMessageHasReplies bool) error

	// DeletePost deletes an arbitrary Slack message identified by channelID and timestamp.
	DeletePost(ctx context.Context, channelID, ts string) error
}

SlackClient is the internal Slack API interface used by the coordinator and channel managers to query channel state and post or update Slack messages. It is implemented by the internal slack.Client and is exposed here to allow injection of test doubles.

type WebhookHandler ¶

type WebhookHandler interface {
	// ShouldHandleWebhook returns true if the handler should handle the specified webhook target.
	ShouldHandleWebhook(ctx context.Context, target string) bool

	// HandleWebhook handles the webhook target, with the specified callback data.
	// If successful, it should DEBUG log the send action and return nil.
	// If unsuccessful, it should return the error (without logging it).
	HandleWebhook(ctx context.Context, target string, data *types.WebhookCallback, logger types.Logger) error
}

WebhookHandler is an interface for handling alert webhooks (i.e. issue callbacks). The webhook may or may not be http based. More than one handler can be registered with a Manager. The first handler that returns true from ShouldHandleWebhook will be used to handle a given webhook.