package blockstore

import (
	"bufio"
	"bytes"
	"compress/zlib"
	"context"
	"fmt"
	"math"
	"os"
	"path/filepath"
	"time"

	"0chain.net/chaincore/block"
	"0chain.net/core/datastore"
	"0chain.net/core/viper"
	"github.com/0chain/common/core/logging"
	"go.uber.org/zap"
	"golang.org/x/sys/unix"
)

const (
	averageBlockSize = 15 * KB
	// minimumInodesRequired is minimum inodes requirements of a disk.
	/// Here 3 is number of years and 80M is expected maximum number of block generation
	expectedTotalBlocksIn3Years = 3 * 80000000
	extension                   = "dat.zlib"
	// subDirs will determine the number of subdirs that should be created to store a block.
	// For example if a block hash is `e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855`
	// and subDirs is 5, then block's path will be:
	//
	// BasePath/e/3/b/0/c/44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855.dat.zlib
	// subDirs 5 will create total of 16^5 + 16^4 + ..+ 16 =  1118480 directories which will store all the finalized
	// blocks.
	subDirs = 5
	// CacheWriteTimeOut Cancel writing to cache if timeout is reached
	CacheWriteTimeOut = 5 * time.Second
)

var (
	store BlockStoreI
)

func hasEnoughInodesAndSize(p string) error {
	var diskStat unix.Statfs_t
	err := unix.Statfs(p, &diskStat)
	if err != nil {
		return err
	}

	availableSize := diskStat.Bavail * uint64(diskStat.Bsize)
	freeInodes := diskStat.Ffree
	minimumInodesRequired := uint64(expectedTotalBlocksIn3Years)
	for i := 0; i < subDirs; i++ {
		minimumInodesRequired += uint64(math.Pow(16, float64(i+1)))
	}

	if freeInodes < minimumInodesRequired {
		return fmt.Errorf("insufficient inodes. Required %d, available %d",
			minimumInodesRequired, freeInodes)
	}

	requiredAvgSize := minimumInodesRequired * averageBlockSize

	if availableSize < uint64(requiredAvgSize) {
		return fmt.Errorf("insufficient disk space. Required %d, available %d",
			requiredAvgSize, availableSize)
	}
	return nil
}

func getBlockFilePath(hash string) (string, error) {
	if len(hash) < subDirs {
		return "", fmt.Errorf("invalid block hash: %s", hash)
	}
	var s string
	for i := 0; i < subDirs; i++ {
		s += string(hash[i]) + string(os.PathSeparator)
	}
	return filepath.Join(s, fmt.Sprintf("%s.%s", hash[subDirs:], extension)), nil
}

type BlockStore struct {
	// basePath is the path to directory where blocks will be permanently stored.
	// It can be any mounted drive but to make it both efficient and cost effective
	// its better to have basePath to be mounted HDD drive and cache as SSD drive.
	// If sharder wants to provide basePath to be mounted SSD drive then they don't
	// need to provide cache.
	basePath              string
	blockMetadataProvider datastore.EntityMetadata
	cache                 cacher
}

func (bStore *BlockStore) writeToDisk(hash string, b *block.Block) error {
	bp, err := getBlockFilePath(hash)
	if err != nil {
		return err
	}
	bPath := filepath.Join(bStore.basePath, bp)
	err = os.MkdirAll(filepath.Dir(bPath), 0700)
	if err != nil {
		return err
	}

	f, err := os.Create(bPath)
	if err != nil {
		return err
	}
	defer f.Close()

	bf := bufio.NewWriterSize(f, 64*1024)
	w, err := zlib.NewWriterLevel(bf, zlib.BestCompression)
	if err != nil {
		return err
	}
	if err := datastore.WriteMsgpack(w, b); err != nil {
		return err
	}
	if err = w.Close(); err != nil {
		return err
	}
	return bf.Flush()
}

func (bStore *BlockStore) write(hash string, b *block.Block) error {
	err := bStore.writeToDisk(hash, b)
	if err != nil {
		return err
	}

	go func() {
		ctx, ctxCncl := context.WithTimeout(context.TODO(), CacheWriteTimeOut)
		defer ctxCncl()
		if err := bStore.cache.Write(ctx, hash, b); err != nil {
			logging.Logger.Error(err.Error())
		}
	}()
	return nil
}

func (bStore *BlockStore) Write(b *block.Block) error {
	err := bStore.write(b.Hash, b)
	if err != nil {
		return err
	}

	if b.MagicBlock != nil && b.Round == b.MagicBlock.StartingRound {
		logging.Logger.Debug("save magic block",
			zap.Int64("round", b.Round),
			zap.String("mb hash", b.MagicBlock.Hash),
		)
		return bStore.write(b.MagicBlock.Hash, b)
	}
	return nil
}

func (bStore *BlockStore) Read(hash string) (*block.Block, error) {
	b := bStore.blockMetadataProvider.Instance().(*block.Block)
	data, err := bStore.cache.Read(hash)
	if data != nil && err == nil {
		r := bytes.NewReader(data)
		err = datastore.ReadMsgpack(r, b)
		if err == nil {
			return b, nil
		}
	}

	b, err = bStore.readFromDisk(hash)
	if err != nil {
		return nil, err
	}

	go func() {
		ctx, ctxCncl := context.WithTimeout(context.TODO(), CacheWriteTimeOut)
		defer ctxCncl()
		err := bStore.cache.Write(ctx, b.Hash, b)
		if err != nil {
			logging.Logger.Error(err.Error())
		}
	}()

	return b, nil
}

func (bStore *BlockStore) readFromDisk(hash string) (*block.Block, error) {
	bp, err := getBlockFilePath(hash)
	if err != nil {
		return nil, err
	}
	bPath := filepath.Join(bStore.basePath, bp)
	f, err := os.Open(bPath)
	if err != nil {
		return nil, err
	}
	defer f.Close()

	r, err := zlib.NewReader(f)
	if err != nil {
		return nil, err
	}
	defer r.Close()
	b := bStore.blockMetadataProvider.Instance().(*block.Block)
	err = datastore.ReadMsgpack(r, b)
	if err != nil {
		return nil, err
	}
	return b, nil
}

// ReadWithBlockSummary - read the block given the block summary
func (bStore *BlockStore) ReadWithBlockSummary(bs *block.BlockSummary) (*block.Block, error) {
	return bStore.Read(bs.Hash)
}

// Init checks for minimum disk size, inodes requirement and assigns
// block storer to a variable. If any error occurs during initialization
// it will panic.
func Init(workDir string, sViper *viper.Viper) {
	logging.Logger.Info("Initializing storage")

	basePath := filepath.Join(workDir, "data", "blocks")
	err := hasEnoughInodesAndSize(basePath)
	if err != nil {
		// TODO comment out for build integration tests.
		// panic(err)
		// only for development phase
		// remove block below
		err = os.MkdirAll(basePath, 0777)
		if err != nil {
			panic(err)
		}
	}

	bStore := &BlockStore{
		cache:                 noOpCache{},
		blockMetadataProvider: datastore.GetEntityMetadata("block"),
		basePath:              basePath,
	}

	if sViper != nil {
		cViper := sViper.Sub("cache")
		if cViper != nil {
			bStore.cache = initCache(cViper)
		}
	}
	SetupStore(bStore)
}
