package bolt

import (
	"os"

	"git.orcatech.org/sdks/golang/log/v2"
	bolt "github.com/coreos/bbolt"
)

// borrowed and editted from: https://github.com/glycerine/hnatsd/blob/master/peer/compact.go
// Compact does these steps to reduce the
// space fragmentation that happens in a bolt database.
//
// 0) open a fresh .compressed bolt db file.
// 1) read each object from the bolt db and write it to the fresh -> .compressed bolt db file.
// 2) close the both files.
// 3) rename the .compressed file to be the original db file name.
//    os.Rename is atomic.
// 4) re-open the newly compact-ed db file
//
// The routines below were adapted from the compaction code in
// https://github.com/boltdb/bolt/blob/master/cmd/bolt/main.go
// which is used under the following MIT license.
/*
The MIT License (MIT)
Copyright (c) 2013 Ben Johnson
Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
// INVAR: b.db must be already open.

const compactTxMaxSizeBytes = 1024 * 1024 * 4

func (d *DB) compactAndClose() error {
	log.Test("Compacting bolt db")
	d.mutex.Lock()
	defer d.mutex.Unlock()
	defer log.Test("Unlocking mutex")

	dstPath := d.options.Database() + ".compressed"
	os.Remove(dstPath)

	// Open destination database.
	log.Test("Opening bolt db %s", dstPath)
	dst, err := bolt.Open(dstPath, 0600, bolt.DefaultOptions)
	if err != nil {
		return err
	}
	defer dst.Close()

	// Run compaction.
	log.Test("Compacting bolt db %s", dstPath)
	if err = compactInBatches(dst, d.db); err != nil {
		return err
	}

	log.Test("Closing: %s", dst)
	if err := dst.Close(); err != nil {
		log.Warn(err.Error())
	}
	log.Test("Closing: %s", d.db)
	if err := d.db.Close(); err != nil {
		log.Warn(err.Error())
	}

	// now move into place the compacted file.
	log.Test("Moving files: %s -> %s", dstPath, d.options.Database())
	return os.Rename(dstPath, d.options.Database())
}

func compactInBatches(dst, src *bolt.DB) error {
	// commit regularly, or we'll run out of memory for large datasets if using one transaction.
	var size int64
	tx, err := dst.Begin(true)
	if err != nil {
		return err
	}

	err = walk(src, func(keys [][]byte, k, v []byte, seq uint64) error {
		// On each key/value, check if we have exceeded tx size.
		sz := int64(len(k) + len(v))
		if size+sz > compactTxMaxSizeBytes && compactTxMaxSizeBytes != 0 {
			log.Verbose("Size is max, committing")

			// Commit previous transaction.
			if err = tx.Commit(); err != nil {
				log.Verbose("Error committing: %s", err.Error())
				return err
			}

			log.Verbose("Committed")

			// Start new transaction.
			tx, err = dst.Begin(true)
			if err != nil {
				log.Verbose("Error starting new transaction: %s", err.Error())
				return err
			}
			size = 0
		}
		size += sz

		// Create bucket on the root transaction if this is the first level.
		nk := len(keys)
		if nk == 0 {
			log.Verbose("Creating bucket")
			bkt, bucketErr := tx.CreateBucket(k)
			if bucketErr != nil {
				return bucketErr
			}
			if err = bkt.SetSequence(seq); err != nil {
				return err
			}
			return nil
		}

		// Create buckets on subsequent levels, if necessary.
		b := tx.Bucket(keys[0])
		if nk > 1 {
			for _, k := range keys[1:] {
				b = b.Bucket(k)
			}
		}

		// If there is no value then this is a bucket call.
		if v == nil {
			log.Verbose("Creating another bucket")
			bkt, bucketErr := b.CreateBucket(k)
			if bucketErr != nil {
				return bucketErr
			}
			if err = bkt.SetSequence(seq); err != nil {
				return err
			}
			return nil
		}

		// Otherwise treat it as a key/value pair.
		return b.Put(k, v)
	})

	commitErr := tx.Commit()
	if err != nil {
		return err
	}
	log.Verbose("Done compacting")

	return commitErr
}

// walkFunc is the type of the function called for keys (buckets and "normal"
// values) discovered by Walk. keys is the list of keys to descend to the bucket
// owning the discovered key/value pair k/v.
type walkFunc func(keys [][]byte, k, v []byte, seq uint64) error

// walk walks recursively the bolt database db, calling walkFn for each key it finds.
func walk(db *bolt.DB, walkFn walkFunc) error {
	return db.View(func(tx *bolt.Tx) error {
		return tx.ForEach(func(name []byte, buck *bolt.Bucket) error {
			return walkBucket(buck, nil, name, nil, buck.Sequence(), walkFn)
		})
	})
}

func walkBucket(buck *bolt.Bucket, keypath [][]byte, k, v []byte, seq uint64, fn walkFunc) error {
	// Execute callback.
	if err := fn(keypath, k, v, seq); err != nil {
		return err
	}

	// If this is not a bucket then stop.
	if v != nil {
		return nil
	}

	// Iterate over each child key/value.
	keypath = append(keypath, k)
	return buck.ForEach(func(k, v []byte) error {
		if v == nil {
			bkt := buck.Bucket(k)
			return walkBucket(bkt, keypath, k, nil, bkt.Sequence(), fn)
		}
		return walkBucket(buck, keypath, k, v, buck.Sequence(), fn)
	})
}
