services

type CleanupService interface {
	CleanupDoneQueue(ctx context.Context, f *models.Flow, sink *models.Sink, doneQueueCleanupDelay time.Duration) (int, error)
	CleanupDeadQueue(ctx context.Context, f *models.Flow, sink *models.Sink, deadQueueCleanupDelay time.Duration) (int, error)
}

type InhooksConfigService interface {
	Load(path string) error
	FindFlowForSource(sourceSlug string) *models.Flow
	GetFlow(flowID string) *models.Flow
	GetFlows() map[string]*models.Flow
	GetTransformDefinition(transformID string) *models.TransformDefinition
}

type MessageBuilder interface {
	FromHttp(flow *models.Flow, r *http.Request, reqID string) ([]*models.Message, error)
}

type MessageEnqueuer interface {
	Enqueue(ctx context.Context, messages []*models.Message) ([]*models.QueuedInfo, error)
}

type MessageFetcher interface {
	GetMessageForProcessing(ctx context.Context, timeout time.Duration, flowID string, sinkID string) (*models.Message, error)
}

type MessageProcessor interface {
	Process(ctx context.Context, sink *models.Sink, m *models.Message) error
}

type MessageTransformer interface {
	Transform(ctx context.Context, transformDefinition *models.TransformDefinition, m *models.Message) error
}

type MessageVerifier interface {
	Verify(flow *models.Flow, m *models.Message) error
}

type ProcessingRecoveryService interface {
	MoveProcessingToReady(ctx context.Context, flow *models.Flow, sink *models.Sink, processingRecoveryInterval time.Duration) ([]string, error)
	AddToCache(mID string, ttl time.Duration)
}

type ProcessingResultsService interface {
	HandleFailed(ctx context.Context, sink *models.Sink, m *models.Message, processingErr error) (*models.QueuedInfo, error)
	HandleOK(ctx context.Context, m *models.Message) error
	RescheduleForRateLimit(ctx context.Context, sink *models.Sink, m *models.Message, delay time.Duration) (*models.QueuedInfo, error)
}

type QueueMetricsService interface {
	UpdateMetrics(ctx context.Context, f *models.Flow, sink *models.Sink) error
}

type RateLimiter interface {
	Allow(ctx context.Context, key string, cfg *models.RateLimitConfig) models.RateLimitDecision
	Peek(ctx context.Context, key string, cfg *models.RateLimitConfig) (float64, error)
}

type RedisStore interface {
	Get(ctx context.Context, messageKey string) ([]byte, error)
	SetAndEnqueue(ctx context.Context, messageKey string, value []byte, queueKey string, messageID string) error
	SetAndZAdd(ctx context.Context, messageKey string, value []byte, queueKey string, messageID string, score float64) error
	SetAndMove(ctx context.Context, messageKey string, value []byte, sourceQueueKey, destQueueKey string, messageID string) error
	SetLRemZAdd(ctx context.Context, messageKey string, value []byte, sourceQueueKey, destQueueKey string, messageID string, score float64) error
	Enqueue(ctx context.Context, key string, value []byte) error
	Dequeue(ctx context.Context, timeout time.Duration, key string) ([]byte, error)
	BLMove(ctx context.Context, timeout time.Duration, sourceQueueKey string, destQueueKey string) ([]byte, error)
	ZRangeBelowScore(ctx context.Context, queueKey string, score float64) ([]string, error)
	ZRemRpush(ctx context.Context, messageIDs []string, sourceQueueKey string, destQueueKey string) error
	LRangeAll(ctx context.Context, queueKey string) ([]string, error)
	LRemRPush(ctx context.Context, sourceQueueKey, destQueueKey string, messageIDs []string) error
	ZRemRangeBelowScore(ctx context.Context, queueKey string, maxScore int) (int, error)
	LRemDel(ctx context.Context, sourceQueueKey string, messageIDs []string, messageKeys []string) error
	MultiGet(ctx context.Context, keys []string) (map[string][]byte, error)
	ZRemDel(ctx context.Context, queueKey string, messageIDs []string, messageKeys []string) error
	LLen(ctx context.Context, queueKey string) (int64, error)
	ZCard(ctx context.Context, queueKey string) (int64, error)
	Eval(ctx context.Context, script string, keys []string, args ...interface{}) ([]interface{}, error)
}

type RetryCalculator interface {
	NextAttemptInterval(attemptsCount int, retryInterval *time.Duration, retryExpMultiplier *float64) time.Duration
}

type SchedulerService interface {
	MoveDueScheduled(ctx context.Context, f *models.Flow, sink *models.Sink) error
}

type TimeService interface {
	Now() time.Time
}