package state

import (
	"context"

	"tangled.org/core/appview/cache"
	"tangled.org/core/appview/config"
	ec "tangled.org/core/eventconsumer"
	"tangled.org/core/eventconsumer/cursor"
	"tangled.org/core/log"
)

func bootstrapStream(
	ctx context.Context,
	name string,
	kind ec.Kind,
	hosts []string,
	redisAddr string,
	streamCfg config.ConsumerConfig,
	dev bool,
	processFn ec.ProcessFunc,
) *ec.Consumer {
	logger := log.SubLogger(log.FromContext(ctx), name)

	srcs := make(map[ec.Source]struct{}, len(hosts))
	for _, h := range hosts {
		srcs[ec.Source{Kind: kind, Host: h}] = struct{}{}
	}

	redisCache := cache.New(redisAddr)
	for _, h := range hosts {
		migrateLegacyCursor(ctx, logger, redisCache, h, string(kind))
	}
	cursorStore := cursor.NewRedisCursorStore(redisCache)

	return ec.NewConsumer(ec.ConsumerConfig{
		Sources:           srcs,
		ProcessFunc:       processFn,
		RetryInterval:     streamCfg.RetryInterval,
		MaxRetryInterval:  streamCfg.MaxRetryInterval,
		ConnectionTimeout: streamCfg.ConnectionTimeout,
		WorkerCount:       streamCfg.WorkerCount,
		QueueSize:         streamCfg.QueueSize,
		Logger:            logger,
		URLFunc:           ec.DefaultURL(dev),
		CursorStore:       &cursorStore,
	})
}
