package store

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"github.com/pingcap/log"
	dto "github.com/prometheus/client_model/go"
	"github.com/tidbcloud/serverless-metering/pkg/config"
	"github.com/tidbcloud/serverless-metering/pkg/monitor"
	"go.uber.org/zap"
)

type StorageImpl struct {
	localStorage  LocalStorage
	remoteStorage RemoteStorage
	uploadCursor  int64
	hostname      string
}

func NewStorage(hostname string, localStorage LocalStorage, remoteStorage RemoteStorage) (Storage, error) {
	cfg := config.Global().Store
	s := &StorageImpl{
		hostname:      hostname,
		localStorage:  localStorage,
		remoteStorage: remoteStorage,
	}

	checkpoint, err := s.GetCheckPoint(context.TODO(), GetCheckPointName(s.hostname))
	if err != nil {
		if strings.Contains(err.Error(), "NoSuchKey") {
			s.uploadCursor = 0
		} else {
			return nil, err
		}
	} else {
		s.uploadCursor = checkpoint.Value
	}
	log.Info("init storage with uploadCursor", zap.Int64("uploadCursor", s.uploadCursor))

	go func() {
		if cfg.TickSeconds <= 0 {
			log.Info("disable storage ticker", zap.Int("tickSeconds", cfg.TickSeconds))
			return
		}
		ticker := time.NewTicker(time.Second * time.Duration(cfg.TickSeconds))
		for {
			<-ticker.C
			s.tick()
		}
	}()

	return s, nil
}

func (s *StorageImpl) Log(labels map[string]string, metricFamilies map[string]*dto.MetricFamily) error {
	return s.localStorage.Log(labels, metricFamilies)
}

func (s *StorageImpl) BuildRemoteScanner(ctx context.Context, ts int64) (scanner RemoteScanner, err error) {
	return s.remoteStorage.BuildScanner(ctx, ts)
}

func (s *StorageImpl) PutCheckPoint(ctx context.Context, name string, checkpoint *CheckPoint) error {
	key := fmt.Sprintf("checkpoints/%s", name)
	data, err := json.Marshal(checkpoint)
	if err != nil {
		return fmt.Errorf("failed to marshal check point: %s", err)
	}

	return s.remoteStorage.PutData(ctx, key, data)
}

func (s *StorageImpl) GetCheckPoint(ctx context.Context, name string) (*CheckPoint, error) {
	key := fmt.Sprintf("checkpoints/%s", name)
	data, err := s.remoteStorage.GetData(ctx, key)
	if err != nil {
		return nil, err
	}

	var checkpoint CheckPoint
	err = json.Unmarshal(data, &checkpoint)
	if err != nil {
		return nil, fmt.Errorf("failed to unmarshal checkpoint: %s", err)
	}

	return &checkpoint, nil
}

func (s *StorageImpl) PutIntermediateResult(ctx context.Context, ts int64, digest string, data []byte) error {
	key := fmt.Sprintf("intermediate/%d/%s/%s", ts, s.hostname, digest)
	return s.remoteStorage.PutData(ctx, key, data)
}

func (s *StorageImpl) GetIntermediateResult(ctx context.Context, ts int64, digest string) ([]byte, error) {
	key := fmt.Sprintf("intermediate/%d/%s/%s", ts, s.hostname, digest)
	return s.remoteStorage.GetData(ctx, key)
}

func (s *StorageImpl) tick() {
	log.Debug("storage tick")
	files, err := s.localStorage.LoadFiles(s.uploadCursor)
	if err != nil {
		log.Error("load files from local storage failed", zap.Error(err))
		return
	}

	count := len(files)
	if count > 0 {
		log.Info("load files success", zap.Int("count", count))
	}

	ctx := context.TODO()
	for _, file := range files {
		err := s.remoteStorage.Upload(ctx, file)
		if err != nil {
			log.Error("update file to remote storage failed", zap.Error(err))
			return
		}
		log.Info("upload file success", zap.String("filepath", file.Path))
		err = s.PutCheckPoint(ctx, GetCheckPointName(s.hostname), &CheckPoint{Value: file.End})
		if err != nil {
			log.Error("failed to commit checkpoint", zap.Error(err))
			return
		}
		s.uploadCursor = file.End
	}

	if count > 0 {
		log.Info("all files uploaded", zap.Int64("checkpoint", s.uploadCursor))
	}
	monitor.StoreUploadProgressGauge.Set(float64(s.uploadCursor))
}