package ddbtoredis

import (
	"context"
	"fmt"
	"hash/maphash"
	"sort"
	"time"

	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/trace"
	"go.uber.org/zap"
)

type EventType string

const (
	Add         EventType = "ADD"
	Modify      EventType = "MODIFY"
	Delete      EventType = "DELETE"
	AntiEntropy EventType = "ANTI_ENTROPY"
)

const otelName = "syncer"

var seed maphash.Seed

func init() {
	seed = maphash.MakeSeed()
}

type CacheItem struct {
	publicKey           string
	encryptedPrivateKey string
	rateLimitRpm        int
	expiredTs           int
}

func (ci CacheItem) hash() uint64 {
	return maphash.String(seed, fmt.Sprintf("%s:%s:%d", ci.publicKey, ci.encryptedPrivateKey, ci.rateLimitRpm))
}

func (ci CacheItem) Diff(item CacheItem) bool {
	return ci.publicKey != item.publicKey || ci.expiredTs != item.expiredTs || ci.hash() != item.hash()
}

type Cache []CacheItem

func (c Cache) Len() int           { return len(c) }
func (c Cache) Swap(i, j int)      { c[i], c[j] = c[j], c[i] }
func (c Cache) Less(i, j int) bool { return c[i].publicKey < c[j].publicKey }

type Event struct {
	Type    EventType
	TraceID trace.TraceID
	Key     KeyDynamodb
}

type Syncer struct {
	ddbReader            DynamodbReader
	ddbStreamsSubscriber DynamodbStreamsSubscriber
	redisRw              RedisReaderWriter
	eventChan            chan Event
	errorChan            chan error
}

func NewSyncer(redisRw RedisReaderWriter, ddbReader DynamodbReader, ddbStreamsSubscriber DynamodbStreamsSubscriber) (*Syncer, error) {
	return &Syncer{
		redisRw:              redisRw,
		ddbReader:            ddbReader,
		ddbStreamsSubscriber: ddbStreamsSubscriber,
		eventChan:            make(chan Event, 10),
		errorChan:            make(chan error, 1),
	}, nil
}

func (s *Syncer) Run() error {
	s.ddbStreamsSubscriber.Subscribe(s.eventChan, s.errorChan)

	go func() {
		s.triggerAntiEntopy()
		tick := time.NewTicker(time.Minute * 5)
		for {
			<-tick.C
			s.triggerAntiEntopy()
		}
	}()

	for {
		select {
		case event := <-s.eventChan:
			ctx := trace.ContextWithRemoteSpanContext(
				context.TODO(),
				trace.NewSpanContext(trace.SpanContextConfig{TraceID: event.TraceID}),
			)
			switch event.Type {
			case AntiEntropy:
				if err := s.AntiEntropy(ctx); err != nil {
					go func() {
						time.Sleep(10 * time.Second)
						s.triggerAntiEntopy()
					}()
				}
			case Add:
				if err := s.redisRw.SetAPIKey(ctx, event.Key.PublicKey, event.Key.EncryptedPrivateKey, event.Key.RateLimitRpm, event.Key.ExpiredTs); err != nil {
					tilog.Error(ctx, "set key failed", zap.String("publicKey", event.Key.PublicKey), zap.Error(err))
					s.triggerAntiEntopy()
				}
			case Modify:
				if err := s.redisRw.SetAPIKey(ctx, event.Key.PublicKey, event.Key.EncryptedPrivateKey, event.Key.RateLimitRpm, event.Key.ExpiredTs); err != nil {
					tilog.Error(ctx, "set key failed", zap.String("publicKey", event.Key.PublicKey), zap.Error(err))
					s.triggerAntiEntopy()
				}
			case Delete:
				if err := s.redisRw.DeleteAPIKey(ctx, event.Key.PublicKey); err != nil {
					tilog.Error(ctx, "delete key failed", zap.String("publicKey", event.Key.PublicKey), zap.Error(err))
					s.triggerAntiEntopy()
				}
			}
		case err := <-s.errorChan:
			tilog.PlainFatal("received from error chan", zap.Error(err))
		}
	}
}

func (s *Syncer) triggerAntiEntopy() {
	ctx, span := otel.Tracer(otelName).Start(context.TODO(), "AntiEntropy")
	defer span.End()
	tilog.Info(ctx, "tigger anti-entropy")
	spanCtx := trace.SpanContextFromContext(ctx)
	event := Event{Type: AntiEntropy, TraceID: spanCtx.TraceID()}
	s.eventChan <- event
}

func (s *Syncer) loadRedis(ctx context.Context) (*Cache, error) {
	apiKeys, err := s.redisRw.ListAPIKeys(ctx)
	if err != nil {
		return nil, err
	}
	var cache Cache
	for pk, apiKey := range apiKeys {
		cache = append(cache, CacheItem{
			publicKey:           pk,
			encryptedPrivateKey: apiKey.PrivateKey,
			rateLimitRpm:        apiKey.RateLimitRpm,
			expiredTs:           apiKey.ExpiredTs,
		})
	}
	sort.Sort(cache)
	return &cache, nil
}

func (s *Syncer) loadDynamodb(ctx context.Context) (*Cache, error) {
	records, err := s.ddbReader.ScanRecords(ctx)
	if err != nil {
		return nil, err
	}
	var cache Cache
	for _, record := range records {
		cache = append(cache, CacheItem{
			publicKey:           record.PublicKey,
			encryptedPrivateKey: record.EncryptedPrivateKey,
			rateLimitRpm:        record.RateLimitRpm,
			expiredTs:           record.ExpiredTs,
		})
	}
	sort.Sort(cache)
	return &cache, nil
}

func (s *Syncer) AntiEntropy(c context.Context) error {
	ctx, span := otel.Tracer(otelName).Start(c, "AntiEntropy")
	defer span.End()

	tilog.Info(ctx, "start anti-entropy")
	defer func(ctx context.Context) {
		tilog.Info(ctx, "finish anti-entropy")
	}(ctx)

	tilog.Info(ctx, "start load redis")
	cacheR, err := s.loadRedis(ctx)
	if err != nil {
		tilog.Error(ctx, "load redis failed", zap.Error(err))
		return err
	}
	tilog.Info(ctx, "start load dynamodb")
	cacheD, err := s.loadDynamodb(ctx)
	if err != nil {
		tilog.Error(ctx, "load dynamodb failed", zap.Error(err))
		return err
	}

	// make data in redis same as dynamodb
	r, d, R, D := 0, 0, len(*cacheR), len(*cacheD)
	for d < D && r < R {
		itemD := (*cacheD)[d]
		itemR := (*cacheR)[r]
		if itemD.Diff(itemR) {
			if itemD.publicKey == itemR.publicKey {
				if err := s.redisRw.SetAPIKey(ctx, itemD.publicKey, itemD.encryptedPrivateKey, itemD.rateLimitRpm, itemD.expiredTs); err != nil {
					tilog.Error(ctx, "set key failed", zap.String("publicKey", itemD.publicKey), zap.Error(err))
					return err
				}
			} else if itemD.publicKey < itemR.publicKey {
				if err := s.redisRw.SetAPIKey(ctx, itemD.publicKey, itemD.encryptedPrivateKey, itemD.rateLimitRpm, itemD.expiredTs); err != nil {
					tilog.Error(ctx, "set key failed", zap.String("publicKey", itemD.publicKey), zap.Error(err))
					return err
				}
				d++
				continue
			} else {
				if err := s.redisRw.DeleteAPIKey(ctx, itemR.publicKey); err != nil {
					tilog.Error(ctx, "delete key failed", zap.String("publicKey", itemR.publicKey), zap.Error(err))
					return err
				}
				r++
				continue
			}
		}
		r++
		d++
	}
	for d < D {
		itemD := (*cacheD)[d]
		if err := s.redisRw.SetAPIKey(ctx, itemD.publicKey, itemD.encryptedPrivateKey, itemD.rateLimitRpm, itemD.expiredTs); err != nil {
			tilog.Error(ctx, "set key failed", zap.String("publicKey", itemD.publicKey), zap.Error(err))
			return err
		}
		d++
	}
	for r < R {
		itemR := (*cacheR)[r]
		if err := s.redisRw.DeleteAPIKey(ctx, itemR.publicKey); err != nil {
			tilog.Error(ctx, "delete key failed", zap.String("publicKey", itemR.publicKey), zap.Error(err))
			return err
		}
		r++
	}

	return nil
}