package redis

import (
	"context"
	"fmt"
	"log"
	"net/url"
	"os"
	"sync"

	"github.com/go-redis/redis/v8"
)

var (
	once        sync.Once
	redisClient *RedisClient
)

type RedisClientInterface interface {
	Set(ctx context.Context, key string, value interface{}) error
	Get(ctx context.Context, key string) (string, error)
	PublishMessage(channel string, message string) error
	SubscribeToChannel(channel string) <-chan *redis.Message
	SaveRedisList(listName string, songs []string) error
	FetchRedisList(listName string) ([]string, error)
	FetchRedisKey(key string) (string, error)
	AddToRedisList(listName string, song string) error
	DeleteFromRedisList(listName string, song string, count int) error
	AddToRedisSet(setName string, song string) error
	RemoveFromRedisSet(setName string, song string) error
	FetchRedisSet(setName string) ([]string, error)
}

type RedisClient struct {
	client *redis.Client
}

func newRedisClient(redisURL string) (*RedisClient, error) {
	parsedURL, err := url.Parse(redisURL)
	if err != nil {
		log.Fatalf("Invalid REDIS_URL: %v", err)
	}

	redisClient := redis.NewClient(&redis.Options{
		Addr: parsedURL.Host,
	})
	if err := redisClient.Ping(context.Background()).Err(); err != nil {
		log.Fatalf("Failed to connect to Redis: %v", err)
	}
	return &RedisClient{client: redisClient}, nil
}

func GetRedisClient() *RedisClient {
	once.Do(func() {
		redisURL := os.Getenv("REDIS_URL")
		client, err := newRedisClient(redisURL)
		if err != nil {
			log.Fatalf("Failed to create Redis client: %v", err)
		}
		redisClient = client
	})
	return redisClient
}

// PublishMessage publishes a message to a Redis channel.
func (c *RedisClient) PublishMessage(channel string, message string) error {
	err := c.client.Publish(context.Background(), channel, message).Err()
	if err != nil {
		fmt.Printf("Error publishing message to channel %s: %v\n", channel, err)
		return err
	}
	fmt.Printf("Message published to channel %s: %s\n", channel, message)
	return nil
}

func (c *RedisClient) Subscribe(channel string) *redis.PubSub {
	return c.client.Subscribe(context.Background(), channel)
}

func (c *RedisClient) SubscribeToChannel(channel string) <-chan *redis.Message {
	pubsub := c.client.Subscribe(context.Background(), channel)
	_, err := pubsub.Receive(context.Background())
	if err != nil {
		panic(err)
	}
	return pubsub.Channel()
}

func (c *RedisClient) SaveRedisList(listName string, songs []string) error {
	for _, song := range songs {
		err := c.client.RPush(context.Background(), listName, song).Err()
		if err != nil {
			return err
		}
	}
	return nil
}

func (c *RedisClient) FetchRedisList(listName string) ([]string, error) {
	songs, err := c.client.LRange(context.Background(), listName, 0, -1).Result()
	if err != nil {
		return nil, err
	}
	return songs, nil
}

func (c *RedisClient) FetchRedisKey(key string) (string, error) {
	value, err := c.client.Get(context.Background(), key).Result()
	if err != nil {
		return "", err
	}
	return value, nil
}

func (c *RedisClient) AddToRedisList(listName string, song string) error {
	err := c.client.RPush(context.Background(), listName, song).Err()
	if err != nil {
		return err
	}
	return nil
}

func (c *RedisClient) DeleteFromRedisList(listName string, song string, count int) error {
	err := c.client.LRem(context.Background(), listName, int64(count), song).Err()
	if err != nil {
		return err
	}
	return nil
}

func (c *RedisClient) AddToRedisSet(setName string, song string) error {
	err := c.client.SAdd(context.Background(), setName, song).Err()
	if err != nil {
		return err
	}
	return nil
}

func (c *RedisClient) RemoveFromRedisSet(setName string, song string) error {
	err := c.client.SRem(context.Background(), setName, song).Err()
	if err != nil {
		return err
	}
	return nil
}

func (c *RedisClient) FetchRedisSet(setName string) ([]string, error) {
	songs, err := c.client.SMembers(context.Background(), setName).Result()
	if err != nil {
		return nil, err
	}
	return songs, nil
}

func (c *RedisClient) XGroupCreateMkStream(ctx context.Context, streamName string, groupName string, start string) error {
	err := c.client.XGroupCreateMkStream(ctx, streamName, groupName, start).Err()
	if err != nil {
		return err
	}
	return nil
}

func (c *RedisClient) XReadGroup(ctx context.Context, args *redis.XReadGroupArgs) ([]redis.XStream, error) {
	streams, err := c.client.XReadGroup(ctx, args).Result()
	if err != nil {
		return nil, err
	}
	return streams, nil
}

func (c *RedisClient) XAck(ctx context.Context, streamName string, groupName string, ids ...string) error {
	err := c.client.XAck(ctx, streamName, groupName, ids...).Err()
	if err != nil {
		return err
	}
	return nil
}

func (c *RedisClient) XDel(ctx context.Context, streamName string, ids ...string) error {
	err := c.client.XDel(ctx, streamName, ids...).Err()
	if err != nil {
		return err
	}
	return nil
}

func (c *RedisClient) XTrimMaxLen(ctx context.Context, streamName string, maxLen int64) error {
	err := c.client.XTrimMaxLen(ctx, streamName, maxLen).Err()
	if err != nil {
		return err
	}
	return nil
}

func (c *RedisClient) UnregisterConsumer(ctx context.Context, streamName string, groupName string, consumerName string) error {
	err := c.client.XGroupDelConsumer(ctx, streamName, groupName, consumerName).Err()
	if err != nil {
		return err
	}
	log.Printf("Successfully unregistered consumer %s from group %s on stream %s", consumerName, groupName, streamName)
	return nil
}

// merge function to combine multiple channels into one
func Merge(ctx context.Context, cs ...<-chan *redis.Message) <-chan *redis.Message {
	out := make(chan *redis.Message)
	var wg sync.WaitGroup

	for _, c := range cs {
		wg.Add(1)
		go func(c <-chan *redis.Message) {
			defer wg.Done()
			for {
				select {
				case msg, ok := <-c:
					if !ok {
						return
					}
					select {
					case out <- msg:
					case <-ctx.Done():
						return
					}
				case <-ctx.Done():
					return
				}
			}
		}(c)
	}

	// Goroutine to close the out channel once all input channels are processed
	go func() {
		wg.Wait()
		close(out)
	}()

	return out
}

// mergeChannelsDynamically merges channels dynamically as they are added
func MergeChannelsDynamically(newChan <-chan <-chan *redis.Message) <-chan *redis.Message {
	out := make(chan *redis.Message)
	var wg sync.WaitGroup

	go func() {
		for c := range newChan {
			fmt.Println("Received new channel")
			wg.Add(1)
			go func(c <-chan *redis.Message) {
				defer wg.Done()
				for msg := range c {
					fmt.Println("Received message:", msg)
					out <- msg
				}
			}(c)
		}
		wg.Wait()
		fmt.Println("Closing out channel")
		close(out)
	}()

	return out
}