package fetchconf

import (
	"archive/tar"
	"fmt"
	"io"
	"os"
	"sync"

	"github.com/DataDog/zstd"
	"github.com/pingcap/log"
	"go.uber.org/zap"

	"github.com/pingcap-inc/clinic/pkg/storage/iface"
)

const FetchConfFileSubffix = "-conf"

type FetchConfSink struct {
	clusterID string

	fileUUID string

	file *os.File

	tw          *tar.Writer
	fileStorage iface.IStorage

	// for test
	disablePersistent bool
	zstdFilePath      string

	lock sync.Mutex
}

func NewFetchConfSink(clusterID, fileUUID string, fileStorage iface.IStorage) (*FetchConfSink, error) {
	disablePersistent := false
	if fileStorage == nil {
		log.Warn("istorage is nil")
		disablePersistent = true
	}
	file, err := os.CreateTemp("", "")
	if err != nil {
		return nil, err
	}

	tw := tar.NewWriter(file)

	return &FetchConfSink{
		clusterID:         clusterID,
		fileUUID:          fileUUID,
		fileStorage:       fileStorage,
		file:              file,
		tw:                tw,
		lock:              sync.Mutex{},
		disablePersistent: disablePersistent,
	}, nil
}

func (fs *FetchConfSink) Sink(fh *tar.Header, reader io.Reader) {
	fs.lock.Lock()
	defer func() {
		fs.lock.Unlock()
	}()

	if err := fs.tw.WriteHeader(fh); err != nil {
		log.Error("write header failed", zap.String("path", fh.Name), zap.Error(err))
		return
	}

	if _, err := io.Copy(fs.tw, reader); err != nil {
		log.Error("write data failed", zap.String("path", fh.Name), zap.Error(err))
	}
}

func (fs *FetchConfSink) Close() {
	if err := fs.tw.Close(); err != nil {
		log.Error("FetchConfSink tar writer close failed", zap.Error(err))
	}

	defer func() {
		if err := fs.file.Close(); err != nil {
			log.Error("FetchConfSink file close failed", zap.Error(err))
		}
		if err := os.Remove(fs.file.Name()); err != nil {
			log.Error("FetchConfSink file remove failed", zap.Error(err))
		}
	}()

	if _, err := fs.file.Seek(0, 0); err != nil {
		log.Error("seek conf file failed", zap.String("fileUUID", fs.fileUUID), zap.Error(err))
		return
	}

	zstdFile, err := compressFile(fs.file)
	if err != nil {
		log.Error("compress file to zstd format failed", zap.String("fileUUID", fs.fileUUID), zap.Error(err))
		return
	}

	fs.zstdFilePath = zstdFile.Name()

	defer func() {
		if zstdFile != nil {
			if err := zstdFile.Close(); err != nil {
				log.Error("zstd file close failed", zap.Error(err))
			}
		}
		if !fs.disablePersistent {
			if err := os.Remove(zstdFile.Name()); err != nil {
				log.Error("file remove failed", zap.Error(err))
			}
		}
	}()

	if fs.disablePersistent {
		return
	}

	if err := fs.fileStorage.Persist(fmt.Sprintf("%s%s", fs.fileUUID, FetchConfFileSubffix), fs.clusterID, zstdFile); err != nil {
		log.Error("persist conf file failed", zap.String("fileUUID", fs.fileUUID), zap.Error(err))
	}
}

func compressFile(originalFile *os.File) (*os.File, error) {
	zstdFile, err := os.CreateTemp("", "")
	if err != nil {
		return nil, err
	}

	zw := zstd.NewWriter(zstdFile)
	if _, err := io.Copy(zw, originalFile); err != nil {
		return zstdFile, err
	}

	if err := zw.Flush(); err != nil {
		log.Error("zstd writer flush failed", zap.Error(err))
	}
	if err := zw.Close(); err != nil {
		log.Error("zstd writer close failed", zap.Error(err))
	}

	if _, err := zstdFile.Seek(0, 0); err != nil {
		return zstdFile, err
	}

	return zstdFile, nil
}