// Delete removes a value with given key.
func (db *LevelDB) Delete(ctx storage.Context, tk storage.TKey) error {
	if db == nil {
		return fmt.Errorf("Can't call Delete on nil LevelDB")
	}
	if ctx == nil {
		return fmt.Errorf("Received nil context in Delete()")
	}
	wo := db.options.WriteOptions

	var err error
	key := ctx.ConstructKey(tk)
	if !ctx.Versioned() {
		dvid.StartCgo()
		err = db.ldb.Delete(wo, key)
		dvid.StopCgo()
	} else {
		vctx, ok := ctx.(storage.VersionedCtx)
		if !ok {
			return fmt.Errorf("Non-versioned context that says it's versioned received in Delete(): %v", ctx)
		}
		tombstoneKey := vctx.TombstoneKey(tk)
		batch := db.NewBatch(vctx).(*goBatch)
		batch.WriteBatch.Delete(key)
		batch.WriteBatch.Put(tombstoneKey, dvid.EmptyValue())
		if err = batch.Commit(); err != nil {
			dvid.Criticalf("Error on batch commit of Delete: %v\n", err)
			err = fmt.Errorf("Error on batch commit of Delete: %v", err)
		}
	}

	return err
}

// Put writes a value with given key.
func (db *LevelDB) Put(ctx storage.Context, tk storage.TKey, v []byte) error {
	if ctx == nil {
		return fmt.Errorf("Received nil context in Put()")
	}
	wo := db.options.WriteOptions

	var err error
	key := ctx.ConstructKey(tk)
	if !ctx.Versioned() {
		dvid.StartCgo()
		err = db.ldb.Put(wo, key, v)
		dvid.StopCgo()
	} else {
		vctx, ok := ctx.(storage.VersionedCtx)
		if !ok {
			return fmt.Errorf("Non-versioned context that says it's versioned received in Put(): %v", ctx)
		}
		tombstoneKey := vctx.TombstoneKey(tk)
		batch := db.NewBatch(vctx).(*goBatch)
		batch.WriteBatch.Delete(tombstoneKey)
		batch.WriteBatch.Put(key, v)
		if err = batch.Commit(); err != nil {
			batch.Close()
			err = fmt.Errorf("Error on PUT: %v\n", err)
		}
	}

	storage.StoreKeyBytesWritten <- len(key)
	storage.StoreValueBytesWritten <- len(v)
	return err
}

// PutRange puts key-value pairs that have been sorted in sequential key order.
func (db *LMDB) PutRange(values []KeyValue) error {
	if db == nil || db.env == nil {
		return fmt.Errorf("Cannot run PutRange() on invalid database.")
	}
	dvid.StartCgo()
	defer dvid.StopCgo()

	txn, err := db.env.BeginTxn(nil, 0)
	if err != nil {
		return err
	}
	defer txn.Commit()

	for _, kv := range values {
		kBytes := kv.K.Bytes()
		v := kv.V
		if v == nil || len(v) == 0 {
			v = []byte{0}
		}
		if err := txn.Put(db.dbi, kBytes, v, 0); err != nil {
			return err
		}
		StoreKeyBytesRead <- len(kBytes)
		StoreValueBytesRead <- len(v)
	}
	return nil
}

// Get returns a value given a key.
func (db *LevelDB) Get(ctx storage.Context, tk storage.TKey) ([]byte, error) {
	if ctx == nil {
		return nil, fmt.Errorf("Received nil context in Get()")
	}
	if ctx.Versioned() {
		vctx, ok := ctx.(storage.VersionedCtx)
		if !ok {
			return nil, fmt.Errorf("Bad Get(): context is versioned but doesn't fulfill interface: %v", ctx)
		}

		// Get all versions of this key and return the most recent
		// log.Printf("  basholeveldb versioned get of key %v\n", k)
		values, err := db.getSingleKeyVersions(vctx, tk)
		// log.Printf("            got back %v\n", values)
		if err != nil {
			return nil, err
		}
		kv, err := vctx.VersionedKeyValue(values)
		// log.Printf("  after deversioning: %v\n", kv)
		if kv != nil {
			return kv.V, err
		}
		return nil, err
	} else {
		key := ctx.ConstructKey(tk)
		ro := db.options.ReadOptions
		// log.Printf("  basholeveldb unversioned get of key %v\n", key)
		dvid.StartCgo()
		v, err := db.ldb.Get(ro, key)
		dvid.StopCgo()
		storage.StoreValueBytesRead <- len(v)
		return v, err
	}
}

func (db *LevelDB) metadataExists() (bool, error) {
	var ctx storage.MetadataContext
	keyBeg, keyEnd := ctx.KeyRange()
	dvid.StartCgo()
	ro := levigo.NewReadOptions()
	it := db.ldb.NewIterator(ro)
	defer func() {
		it.Close()
		dvid.StopCgo()
	}()

	it.Seek(keyBeg)
	for {
		if it.Valid() {
			// Did we pass the final key?
			if bytes.Compare(it.Key(), keyEnd) > 0 {
				break
			}
			return true, nil
		}
		break
	}
	if err := it.GetError(); err != nil {
		return false, err
	}
	dvid.Infof("No metadata found for %s...\n", db)
	return false, nil
}

func (batch *goBatch) Commit() error {
	dvid.StartCgo()
	defer dvid.StopCgo()

	err := batch.ldb.Write(batch.wo, batch.WriteBatch)
	batch.WriteBatch.Close()
	return err
}

func (b *batch) Commit() error {
	if b == nil {
		return fmt.Errorf("Illegal Commit() on a nil batch")
	}
	dvid.StartCgo()
	defer dvid.StopCgo()
	return b.txn.Commit()
}

// RawDelete is a low-level function.  It deletes a key-value pair using full keys
// without any context.  This can be used in conjunction with RawRangeQuery.
func (db *LevelDB) RawDelete(k storage.Key) error {
	if db == nil {
		return fmt.Errorf("Can't call RawDelete on nil LevelDB")
	}
	wo := db.options.WriteOptions
	dvid.StartCgo()
	defer dvid.StopCgo()
	return db.ldb.Delete(wo, k)
}

func (b *batch) Delete(k Key) {
	if b != nil {
		dvid.StartCgo()
		defer dvid.StopCgo()
		if err := b.txn.Del(b.dbi, k.Bytes(), nil); err != nil {
			dvid.Error("Error in batch Delete: %s", err.Error())
		}
	}
}

// newLevelDB returns a leveldb backend, creating leveldb
// at the path if it doesn't already exist.
func (e Engine) newLevelDB(config dvid.StoreConfig) (*LevelDB, bool, error) {
	path, _, err := parseConfig(config)
	if err != nil {
		return nil, false, err
	}

	// Is there a database already at this path?  If not, create.
	var created bool
	if _, err := os.Stat(path); os.IsNotExist(err) {
		dvid.Infof("Database not already at path (%s). Creating directory...\n", path)
		created = true
		// Make a directory at the path.
		if err := os.MkdirAll(path, 0744); err != nil {
			return nil, true, fmt.Errorf("Can't make directory at %s: %v", path, err)
		}
	} else {
		dvid.Infof("Found directory at %s (err = %v)\n", path, err)
	}

	// Open the database
	dvid.StartCgo()
	defer dvid.StopCgo()

	opt, err := getOptions(config.Config)
	if err != nil {
		return nil, false, err
	}

	leveldb := &LevelDB{
		directory: path,
		config:    config,
		options:   opt,
	}

	dvid.Infof("Opening basholeveldb @ path %s\n", path)
	ldb, err := levigo.Open(path, opt.Options)
	if err != nil {
		return nil, false, err
	}
	leveldb.ldb = ldb

	// if we know it's newly created, just return.
	if created {
		return leveldb, created, nil
	}

	// otherwise, check if there's been any metadata or we need to initialize it.
	metadataExists, err := leveldb.metadataExists()
	if err != nil {
		leveldb.Close()
		return nil, false, err
	}

	return leveldb, !metadataExists, nil
}

func (batch *goBatch) Commit() error {
	if batch == nil {
		return fmt.Errorf("Received nil batch in batch.Commit()\n")
	}
	dvid.StartCgo()
	defer dvid.StopCgo()

	err := batch.ldb.Write(batch.wo, batch.WriteBatch)
	batch.WriteBatch.Close()
	return err
}

// OpenFUSE mounts the given directory as a FUSE file system.
// The FUSE system is a singleton with only one FUSE server operable.
func OpenFUSE(dir string, data Mountable, vinfo VersionInfo) error {
	fuseServer.mutex.Lock()
	defer fuseServer.mutex.Unlock()

	// Make sure we haven't switched mount directory.
	if len(fuseServer.dir) > 0 {
		if fuseServer.dir != dir {
			return fmt.Errorf("Cannot open more than one FUSE directory.  Currently open: %s\n",
				fuseServer.dir)
		}
	}

	// Make sure our mount directory is present and a directory.
	finfo, err := os.Stat(dir)
	if err != nil {
		if os.IsNotExist(err) {
			if err = os.MkdirAll(dir, 0744); err != nil {
				return fmt.Errorf("Cannot create mount directory: %s (%s)\n",
					dir, err.Error())
			}
		} else {
			return fmt.Errorf("Cannot access given mount directory: %s\n", dir)
		}
	} else if !finfo.IsDir() {
		return fmt.Errorf("Given mount point (%s) is not a directory\n", dir)
	}

	// Check if data is already mounted at this version.
	mount, found := fuseServer.mounts[vinfo.uuid]
	if found {
		mount.AddData(data, vinfo)
		return nil
	}

	fuseServer.mounts[vinfo.uuid] = Mount{Data{data}, vinfo}

	// Mount and serve if not already served.
	if fuseServer.dir == "" {
		fuseServer.dir = dir
		conn, err := fuse.Mount(dir)
		if err != nil {
			return err
		}

		// Run FUSE system in gothread.
		go func() {
			dvid.StartCgo()
			fs.Serve(conn, fuseServer)
			dvid.StopCgo()
		}()
	}
	return nil
}

// unversionedRange sends a range of key-value pairs down a channel.
func (db *LevelDB) unversionedRange(ctx storage.Context, begTKey, endTKey storage.TKey, ch chan errorableKV, done <-chan struct{}, keysOnly bool) {
	dvid.StartCgo()
	ro := levigo.NewReadOptions()
	it := db.ldb.NewIterator(ro)
	defer func() {
		it.Close()
		dvid.StopCgo()
	}()

	// Apply context if applicable
	begKey := ctx.ConstructKey(begTKey)
	endKey := ctx.ConstructKey(endTKey)

	// fmt.Printf("unversionedRange():\n")
	// fmt.Printf("    index beg: %v\n", kStart)
	// fmt.Printf("    index end: %v\n", kEnd)
	// fmt.Printf("    key start: %v\n", keyBeg)
	// fmt.Printf("      key end: %v\n", keyEnd)

	var itValue []byte
	it.Seek(begKey)
	for {
		if it.Valid() {
			// fmt.Printf("unversioned found key %v, %d bytes value\n", it.Key(), len(it.Value()))
			if !keysOnly {
				itValue = it.Value()
				storage.StoreValueBytesRead <- len(itValue)
			}
			itKey := it.Key()
			storage.StoreKeyBytesRead <- len(itKey)
			// Did we pass the final key?
			if bytes.Compare(itKey, endKey) > 0 {
				break
			}
			select {
			case <-done:
				ch <- errorableKV{nil, nil}
				return
			case ch <- errorableKV{&storage.KeyValue{K: itKey, V: itValue}, nil}:
				it.Next()
			}
		} else {
			break
		}
	}
	if err := it.GetError(); err != nil {
		ch <- errorableKV{nil, err}
	} else {
		ch <- errorableKV{nil, nil}
	}
	return
}

// RawPut is a low-level function that puts a key-value pair using full keys.
// This can be used in conjunction with RawRangeQuery.
func (db *LevelDB) RawPut(k storage.Key, v []byte) error {
	wo := db.options.WriteOptions
	dvid.StartCgo()
	defer dvid.StopCgo()

	if err := db.ldb.Put(wo, k, v); err != nil {
		return err
	}

	storage.StoreKeyBytesWritten <- len(k)
	storage.StoreValueBytesWritten <- len(v)
	return nil
}

// ProcessRange sends a range of key-value pairs to chunk handlers.
func (db *LMDB) ProcessRange(kStart, kEnd Key, op *ChunkOp, f func(*Chunk)) error {
	if db == nil || db.env == nil {
		return fmt.Errorf("Cannot ProcessRange() on invalid database.")
	}
	dvid.StartCgo()
	defer dvid.StopCgo()

	txn, err := db.env.BeginTxn(nil, lmdb.RDONLY)
	if err != nil {
		return err
	}
	defer txn.Abort()
	cursor, err := txn.CursorOpen(db.dbi)
	if err != nil {
		return err
	}
	defer cursor.Close()

	seekKey := kStart.Bytes()
	endBytes := kEnd.Bytes()
	var cursorOp uint = lmdb.SET_RANGE
	for {
		k, v, rc := cursor.Get(seekKey, cursorOp)
		if rc != nil {
			break
		}
		seekKey = nil
		cursorOp = lmdb.NEXT
		StoreKeyBytesRead <- len(k)
		StoreValueBytesRead <- len(v)
		if k == nil || bytes.Compare(k, endBytes) > 0 {
			break
		}
		// Convert byte representation of key to storage.Key
		var key Key
		key, err = kStart.BytesToKey(k)
		if err != nil {
			return err
		}
		if op.Wg != nil {
			op.Wg.Add(1)
		}
		chunk := &Chunk{
			op,
			KeyValue{key, v},
		}
		f(chunk)
	}
	return nil
}

// NewKeyValueStore returns a lmdb backend.
func NewKeyValueStore(path string, create bool, config dvid.Config) (Engine, error) {
	// Create the directory if it doesn't exist.
	if _, err := os.Stat(path); os.IsNotExist(err) {
		if err = os.MkdirAll(path, 0770); err != nil {
			return nil, fmt.Errorf("Datastore (%s) doesn't exist and couldn't be created: %s", err.Error())
		}
	}
	dvid.StartCgo()
	defer dvid.StopCgo()

	opt, err := GetOptions(create, config)
	if err != nil {
		return nil, err
	}

	env, err := lmdb.NewEnv()
	if err != nil {
		return nil, err
	}

	if err = env.SetMapSize(uint64(opt.GBytes * dvid.Giga)); err != nil {
		return nil, err
	}

	if err = env.Open(path, lmdb.NOSYNC|lmdb.WRITEMAP, 0664); err != nil {
		return nil, err
	}

	txn, err := env.BeginTxn(nil, 0)
	if err != nil {
		return nil, fmt.Errorf("Cannot begin transaction: %s", err.Error())
	}

	dbi, err := txn.DBIOpen(nil, 0)
	if err != nil {
		return nil, fmt.Errorf("Cannot create DBI: %s", err.Error())
	}

	db := &LMDB{
		path:    path,
		config:  config,
		options: opt,
		env:     env,
		dbi:     dbi,
	}
	txn.Abort()

	return db, nil
}

// RepairStore tries to repair a damaged leveldb
func RepairStore(path string, config dvid.Config) error {
	dvid.StartCgo()
	defer dvid.StopCgo()

	opt, err := GetOptions(false, config)
	if err != nil {
		return err
	}

	err = levigo.RepairDatabase(path, opt.Options)
	if err != nil {
		return err
	}
	return nil
}

// Repair tries to repair a damaged leveldb.  Requires "path" string.  Implements
// the RepairableEngine interface.
func (e Engine) Repair(path string) error {
	dvid.StartCgo()
	defer dvid.StopCgo()

	opt, err := getOptions(dvid.Config{})
	if err != nil {
		return err
	}

	err = levigo.RepairDatabase(path, opt.Options)
	if err != nil {
		return err
	}
	return nil
}

func (b *batch) Put(k Key, v []byte) {
	if b != nil {
		dvid.StartCgo()
		defer dvid.StopCgo()
		kBytes := k.Bytes()
		if v == nil || len(v) == 0 {
			v = []byte{0}
		}
		if err := b.txn.Put(b.dbi, kBytes, v, 0); err != nil {
			dvid.Error("Error in batch Put: %s", err.Error())
			return
		}
		StoreKeyBytesWritten <- len(kBytes)
		StoreValueBytesWritten <- len(v)
	}
}

// NewBatch returns an implementation that allows batch writes
func (db *LevelDB) NewBatch(ctx storage.Context) storage.Batch {
	if ctx == nil {
		dvid.Criticalf("Received nil context in NewBatch()")
		return nil
	}
	dvid.StartCgo()
	defer dvid.StopCgo()

	var vctx storage.VersionedCtx
	var ok bool
	vctx, ok = ctx.(storage.VersionedCtx)
	if !ok {
		vctx = nil
	}
	return &goBatch{ctx, vctx, levigo.NewWriteBatch(), db.options.WriteOptions, db.ldb}
}