// Package leveldb implements a logpattern store using a LevelDB backend
// database.
package leveldb

import (
	"bytes"
	"context"
	"fmt"
	"io"

	"github.com/pingcap-inc/alioth/storage/keyvalue"
	"github.com/syndtr/goleveldb/leveldb"
	goIterator "github.com/syndtr/goleveldb/leveldb/iterator"
	"github.com/syndtr/goleveldb/leveldb/opt"
	"github.com/syndtr/goleveldb/leveldb/util"
)

// Options for customizing a LevelDB backend.
type Options struct {
	// CacheCapacity is the caching capacity (in bytes) used for the LevelDB.
	CacheCapacity int

	// CacheLargeReads determines whether to use the cache for large reads. This
	// is usually discouraged but may be useful when the entire LevelDB is known
	// to fit into the cache.
	CacheLargeReads bool

	// WriteBufferSize is the number of bytes the database will build up in memory
	// (backed by a disk log) before writing to the on-disk table.
	WriteBufferSize int

	// MustExist ensures that the given database exists before opening it.  If
	// false and the database does not exist, it will be created.
	MustExist bool
}

// levelDB is a wrapper around a levigo.DB that implements keyvalue.DB
type goLevelDB struct {
	db *leveldb.DB
}

var DefaultGoLevelDBOptions = &opt.Options{
	BlockCacheCapacity:            512 * opt.MiB,
	BlockRestartInterval:          16,
	BlockSize:                     4 * opt.KiB,
	CompactionL0Trigger:           8,
	CompactionTableSize:           64 * opt.MiB,
	CompactionTotalSize:           512 * opt.MiB,
	CompactionTotalSizeMultiplier: 8,
	WriteBuffer:                   64 * opt.MiB,
	WriteL0PauseTrigger:           24,
	WriteL0SlowdownTrigger:        17,
}

// OpenGoLevelDB returns a keyvalue DB backed by a GoLevelDB database at the given
// filepath.  If opts==nil, the DefaultOptions are used.
func OpenGoLevelDB(path string, opts *Options) (keyvalue.DB, error) {
	var goLevelDBOpts *opt.Options
	if opts == nil {
		goLevelDBOpts = DefaultGoLevelDBOptions
	}

	db, err := leveldb.OpenFile(path, goLevelDBOpts)
	if err != nil {
		return nil, fmt.Errorf("could not open LevelDB at %q: %v", path, err)
	}
	return &goLevelDB{
		db: db,
	}, nil
}

// Close will close the underlying LevelDB database.
func (s *goLevelDB) Close(_ context.Context) error {
	return s.db.Close()
}

// Writer implements part of the keyvalue.DB interface.
func (s *goLevelDB) Writer(_ context.Context) (keyvalue.Writer, error) {
	return &goLevelDBWriter{s, new(leveldb.Batch)}, nil
}

// Get implements part of the keyvalue.DB interface.
func (s *goLevelDB) Get(_ context.Context, key []byte, opts *keyvalue.Options) ([]byte, error) {
	v, err := s.db.Get(key, nil)
	if err != nil {
		return nil, err
	} else if v == nil {
		return nil, io.EOF
	}
	return v, nil
}

// ScanPrefix implements part of the keyvalue.DB interface.
func (s *goLevelDB) ScanPrefix(_ context.Context, prefix []byte, opts *keyvalue.Options) (keyvalue.Iterator, error) {
	iter := s.db.NewIterator(util.BytesPrefix(prefix), nil)
	return &goLevelDBIterator{iter, prefix, nil}, nil
}

// ScanRange implements part of the keyvalue.DB interface.
func (s *goLevelDB) ScanRange(_ context.Context, r *keyvalue.Range, opts *keyvalue.Options) (keyvalue.Iterator, error) {
	iter := s.db.NewIterator(&util.Range{
		Start: r.Start,
	}, nil)
	return &goLevelDBIterator{iter, nil, r}, nil
}

type goLevelDBWriter struct {
	s *goLevelDB
	*leveldb.Batch
}

// Write implements part of the keyvalue.Writer interface.
func (w *goLevelDBWriter) Write(key, val []byte) error {
	w.Put(key, val)
	return nil
}

// Close implements part of the keyvalue.Writer interface.
func (w *goLevelDBWriter) Close() error {
	if err := w.s.db.Write(w.Batch, nil); err != nil {
		return err
	}
	w.Batch.Reset()
	return nil
}

type goLevelDBIterator struct {
	it     goIterator.Iterator
	prefix []byte
	r      *keyvalue.Range
}

// Close implements part of the keyvalue.Iterator interface.
func (i *goLevelDBIterator) Close() error {
	i.it.Release()
	return i.it.Error()
}

// Next implements part of the keyvalue.Iterator interface.
func (i *goLevelDBIterator) Next() ([]byte, []byte, error) {
	if !i.it.Next() {
		if err := i.it.Error(); err != nil {
			return nil, nil, err
		}
		return nil, nil, io.EOF
	}

	key, val := i.it.Key(), i.it.Value()
	if (i.r == nil && !bytes.HasPrefix(key, i.prefix)) || (i.r != nil && len(i.r.End) > 0 && bytes.Compare(key, i.r.End) >= 0) {
		return nil, nil, io.EOF
	}
	return key, val, nil
}