// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.

package rawdb

import (
	"bytes"
	"errors"
	"fmt"
	"maps"
	"math/big"
	"os"
	"path/filepath"
	"slices"
	"strings"
	"time"

	"github.com/ethereum/go-ethereum/common"
	"github.com/ethereum/go-ethereum/crypto"
	"github.com/ethereum/go-ethereum/ethdb"
	"github.com/ethereum/go-ethereum/ethdb/memorydb"
	"github.com/ethereum/go-ethereum/log"
	"github.com/olekukonko/tablewriter"
)

var ErrDeleteRangeInterrupted = errors.New("safe delete range operation interrupted")

// freezerdb is a database wrapper that enables ancient chain segment freezing.
type freezerdb struct {
	ethdb.KeyValueStore
	*chainFreezer

	readOnly    bool
	ancientRoot string
}

// AncientDatadir returns the path of root ancient directory.
func (frdb *freezerdb) AncientDatadir() (string, error) {
	return frdb.ancientRoot, nil
}

// Close implements io.Closer, closing both the fast key-value store as well as
// the slow ancient tables.
func (frdb *freezerdb) Close() error {
	var errs []error
	if err := frdb.chainFreezer.Close(); err != nil {
		errs = append(errs, err)
	}

	if err := frdb.KeyValueStore.Close(); err != nil {
		errs = append(errs, err)
	}

	if len(errs) != 0 {
		return fmt.Errorf("%v", errs)
	}

	return nil
}

// Freeze is a helper method used for external testing to trigger and block until
// a freeze cycle completes, without having to sleep for a minute to trigger the
// automatic background run.
func (frdb *freezerdb) Freeze() error {
	if frdb.readOnly {
		return errReadOnly
	}
	// Trigger a freeze cycle and block until it's done
	trigger := make(chan struct{}, 1)
	frdb.chainFreezer.trigger <- trigger
	<-trigger

	return nil
}

// nofreezedb is a database wrapper that disables freezer data retrievals.
type nofreezedb struct {
	ethdb.KeyValueStore
}

// HasAncient returns an error as we don't have a backing chain freezer.
func (db *nofreezedb) HasAncient(kind string, number uint64) (bool, error) {
	return false, errNotSupported
}

// Ancient returns an error as we don't have a backing chain freezer.
func (db *nofreezedb) Ancient(kind string, number uint64) ([]byte, error) {
	return nil, errNotSupported
}

// AncientRange returns an error as we don't have a backing chain freezer.
func (db *nofreezedb) AncientRange(kind string, start, max, maxByteSize uint64) ([][]byte, error) {
	return nil, errNotSupported
}

// Ancients returns an error as we don't have a backing chain freezer.
func (db *nofreezedb) Ancients() (uint64, error) {
	return 0, errNotSupported
}

// Tail returns an error as we don't have a backing chain freezer.
func (db *nofreezedb) Tail() (uint64, error) {
	return 0, errNotSupported
}

// AncientSize returns an error as we don't have a backing chain freezer.
func (db *nofreezedb) AncientSize(kind string) (uint64, error) {
	return 0, errNotSupported
}

// ModifyAncients is not supported.
func (db *nofreezedb) ModifyAncients(func(ethdb.AncientWriteOp) error) (int64, error) {
	return 0, errNotSupported
}

// TruncateHead returns an error as we don't have a backing chain freezer.
func (db *nofreezedb) TruncateHead(items uint64) (uint64, error) {
	return 0, errNotSupported
}

// TruncateTail returns an error as we don't have a backing chain freezer.
func (db *nofreezedb) TruncateTail(items uint64) (uint64, error) {
	return 0, errNotSupported
}

// ItemAmountInAncient returns an error as we don't have a backing chain freezer.
func (db *nofreezedb) ItemAmountInAncient() (uint64, error) {
	return 0, errNotSupported
}

// AncientOffSet returns 0 as we don't have a backing chain freezer.
func (db *nofreezedb) AncientOffSet() uint64 {
	return 0
}

// Sync returns an error as we don't have a backing chain freezer.
func (db *nofreezedb) Sync() error {
	return errNotSupported
}

func (db *nofreezedb) ReadAncients(fn func(reader ethdb.AncientReaderOp) error) (err error) {
	// Unlike other ancient-related methods, this method does not return
	// errNotSupported when invoked.
	// The reason for this is that the caller might want to do several things:
	// 1. Check if something is in the freezer,
	// 2. If not, check leveldb.
	//
	// This will work, since the ancient-checks inside 'fn' will return errors,
	// and the leveldb work will continue.
	//
	// If we instead were to return errNotSupported here, then the caller would
	// have to explicitly check for that, having an extra clause to do the
	// non-ancient operations.
	return fn(db)
}

// AncientDatadir returns an error as we don't have a backing chain freezer.
func (db *nofreezedb) AncientDatadir() (string, error) {
	return "", errNotSupported
}

// NewDatabase creates a high level database on top of a given key-value data
// store without a freezer moving immutable chain segments into cold storage.
func NewDatabase(db ethdb.KeyValueStore) ethdb.Database {
	return &nofreezedb{KeyValueStore: db}
}

// ReadOffsetOfCurrentAncientFreezer reads the offset of current ancient freezer
func ReadOffsetOfCurrentAncientFreezer(db ethdb.KeyValueReader) uint64 {
	offset, _ := db.Get(offsetOfCurrentAncientFreezer)
	if offset == nil {
		return 0
	}

	return new(big.Int).SetBytes(offset).Uint64()
}

// ReadOffsetOfLastAncientFreezer reads the offset of last pruned ancient freezer
func ReadOffsetOfLastAncientFreezer(db ethdb.KeyValueReader) uint64 {
	offset, _ := db.Get(offsetOfLastAncientFreezer)
	if offset == nil {
		return 0
	}

	return new(big.Int).SetBytes(offset).Uint64()
}

// WriteOffsetOfCurrentAncientFreezer writes current offset of ancient freezer into ethdb
func WriteOffsetOfCurrentAncientFreezer(db ethdb.KeyValueWriter, offset uint64) {
	if err := db.Put(offsetOfCurrentAncientFreezer, new(big.Int).SetUint64(offset).Bytes()); err != nil {
		log.Crit("Failed to store offSetOfAncientFreezer", "err", err)
	}
}

// WriteOffsetOfLastAncientFreezer writes the last offset of ancient freezer into ethdb
func WriteOffsetOfLastAncientFreezer(db ethdb.KeyValueWriter, offset uint64) {
	if err := db.Put(offsetOfLastAncientFreezer, new(big.Int).SetUint64(offset).Bytes()); err != nil {
		log.Crit("Failed to store offSetOfAncientFreezer", "err", err)
	}
}

// NewDatabaseWithOnlyFreezer create a freezer db without state
func NewDatabaseWithOnlyFreezer(db ethdb.KeyValueStore, frz, namespace string, readonly bool, newOffSet uint64) (*Freezer, error) {
	// Create the idle freezer instance, this operation should be atomic to avoid mismatch between offset and ancientDB.
	frdb, err := NewChainFreezer(frz, namespace, readonly, newOffSet)
	if err != nil {
		return nil, err
	}

	return frdb, nil
}

// resolveChainFreezerDir is a helper function which resolves the absolute path
// of chain freezer by considering backward compatibility.
func resolveChainFreezerDir(ancient string) string {
	// Check if the chain freezer is already present in the specified
	// sub folder, if not then two possibilities:
	// - chain freezer is not initialized
	// - chain freezer exists in legacy location (root ancient folder)
	freezer := filepath.Join(ancient, ChainFreezerName)
	if !common.FileExist(freezer) {
		if !common.FileExist(ancient) {
			// The entire ancient store is not initialized, still use the sub
			// folder for initialization.
		} else {
			// Ancient root is already initialized, then we hold the assumption
			// that chain freezer is also initialized and located in root folder.
			// In this case fallback to legacy location.
			freezer = ancient
			log.Info("Found legacy ancient chain path", "location", ancient)
		}
	}

	return freezer
}

// resolveOffset is a helper function which resolves the value of offset to use
// while opening a chain freezer.
func resolveOffset(db ethdb.KeyValueStore, isLastOffset bool) uint64 {
	// The offset of ancientDB should be handled differently in different scenarios.
	if isLastOffset {
		return ReadOffsetOfLastAncientFreezer(db)
	} else {
		return ReadOffsetOfCurrentAncientFreezer(db)
	}
}

// NewDatabaseWithFreezer creates a high level database on top of a given key-
// value data store with a freezer moving immutable chain segments into cold
// storage.
//
//nolint:gocognit
func NewDatabaseWithFreezer(db ethdb.KeyValueStore, ancient string, namespace string, readonly, disableFreeze, isLastOffset, stateless bool) (ethdb.Database, error) {
	offset := resolveOffset(db, isLastOffset)
	log.Info("Resolving ancient pruner offset", "isLastOffset", isLastOffset, "offset", offset)

	// Create the idle freezer instance. If the given ancient directory is empty,
	// in-memory chain freezer is used (e.g. dev mode); otherwise the regular
	// file-based freezer is created.
	chainFreezerDir := ancient
	if chainFreezerDir != "" {
		chainFreezerDir = resolveChainFreezerDir(chainFreezerDir)
	}

	// Create the idle freezer instance
	frdb, err := newChainFreezer(chainFreezerDir, namespace, readonly, offset)
	if err != nil {
		return nil, err
	}

	// Since the freezer can be stored separately from the user's key-value database,
	// there's a fairly high probability that the user requests invalid combinations
	// of the freezer and database. Ensure that we don't shoot ourselves in the foot
	// by serving up conflicting data, leading to both datastores getting corrupted.
	//
	//   - If both the freezer and key-value store are empty (no genesis), we just
	//     initialized a new empty freezer, so everything's fine.
	//   - If the key-value store is empty, but the freezer is not, we need to make
	//     sure the user's genesis matches the freezer. That will be checked in the
	//     blockchain, since we don't have the genesis block here (nor should we at
	//     this point care, the key-value/freezer combo is valid).
	//   - If neither the key-value store nor the freezer is empty, cross validate
	//     the genesis hashes to make sure they are compatible. If they are, also
	//     ensure that there's no gap between the freezer and subsequently leveldb.
	//   - If the key-value store is not empty, but the freezer is, we might just be
	//     upgrading to the freezer release, or we might have had a small chain and
	//     not frozen anything yet. Ensure that no blocks are missing yet from the
	//     key-value store, since that would mean we already had an old freezer.

	// If the genesis hash is empty, we have a new key-value store, so nothing to
	// validate in this method. If, however, the genesis hash is not nil, compare
	// it to the freezer content.
	//nolint:nestif
	if kvgenesis, _ := db.Get(headerHashKey(0)); len(kvgenesis) > 0 {
		if frozen, _ := frdb.ItemAmountInAncient(); frozen > 0 {
			// If the freezer already contains something, ensure that the genesis blocks
			// match, otherwise we might mix up freezers across chains and destroy both
			// the freezer and the key-value store.
			// Only validate genesis if we have `offset` set to 0, which means ancient db pruning
			// hasn't happened yet. If the pruning would've happened, genesis would have been wiped
			// from ancient db.
			if offset == 0 {
				frgenesis, err := frdb.Ancient(ChainFreezerHashTable, 0)
				if err != nil {
					printChainMetadata(db)
					return nil, fmt.Errorf("failed to retrieve genesis from ancient %v", err)
				} else if !bytes.Equal(kvgenesis, frgenesis) {
					printChainMetadata(db)
					return nil, fmt.Errorf("genesis mismatch: %#x (leveldb) != %#x (ancients)", kvgenesis, frgenesis)
				}
			}

			// Key-value store and freezer belong to the same network. Ensure that they
			// are contiguous, otherwise we might end up with a non-functional freezer.
			//
			// If ancient db pruning has happened, the number of items in ancient db should
			// be less and hence we need to calculate the first block of leveldb by adding
			// the offset to it i.e. start block of leveldb = frozen + offset.
			startBlock := frozen + offset
			var kvhash []byte
			if kvhash, _ = db.Get(headerHashKey(startBlock)); len(kvhash) == 0 {
				// Subsequent header after the freezer limit is missing from the database.
				// Reject startup if the database has a more recent head.
				if head := *ReadHeaderNumber(db, ReadHeadHeaderHash(db)); head > startBlock-1 {
					// Find the smallest block stored in the key-value store
					// in range of [frozen, head]
					var number uint64
					for number = startBlock; number <= head; number++ {
						if present, _ := db.Has(headerHashKey(number)); present {
							break
						}
					}
					// We are about to exit on error. Print database metadata before exiting
					printChainMetadata(db)
					return nil, fmt.Errorf("gap in the chain between ancients [0 - #%d] and leveldb [#%d - #%d]",
						startBlock-1, number, head)
				}
				// Database contains only older data than the freezer, this happens if the
				// state was wiped and reinited from an existing freezer.
			}
			// Otherwise, key-value store continues where the freezer left off, all is fine.
			// We might have duplicate blocks (crash after freezer write but before key-value
			// store deletion, but that's fine). Still, check if the first block of key-value
			// store points to last block in freezer.
			if head := ReadHeaderFromKvStore(db, common.BytesToHash(kvhash), startBlock); head != nil {
				parentHash := head.ParentHash.Bytes()
				ancientParentHash, _ := frdb.Ancient(ChainFreezerHashTable, startBlock-1)
				if ancientParentHash == nil {
					printChainMetadata(db)
					return nil, fmt.Errorf("missing parent hash for block #%d in ancient", startBlock-1)
				}
				if !bytes.Equal(parentHash, ancientParentHash) {
					printChainMetadata(db)
					return nil, fmt.Errorf("broken chain due to parent hash mismatch: %#x (leveldb) != %#x (ancients) for block #%d, please set --datadir.ancient to the correct path", parentHash, ancientParentHash, startBlock-1)
				}
				// First block of key-value store points back to correct parent in ancient
			}
		} else {
			// This case means the freezer is empty. Either nothing is moved from the
			// key-value store or we've pruned all data.

			// No pruning case
			if offset == 0 {
				// If the freezer is empty, ensure nothing was moved yet from the key-value
				// store, otherwise we'll end up missing data. We check block #1 to decide
				// if we froze anything previously or not, but do take care of databases with
				// only the genesis block.
				if ReadHeadHeaderHash(db) != common.BytesToHash(kvgenesis) {
					// Key-value store contains more data than the genesis block, make sure we
					// didn't freeze anything yet.
					// Bor Change: Stateless client skip this check because of fast forward feature, which usually jumps #1 block
					if kvblob, _ := db.Get(headerHashKey(1)); len(kvblob) == 0 && !stateless {
						printChainMetadata(db)
						return nil, errors.New("ancient chain segments already extracted, please set --datadir.ancient to the correct path")
					}
					// Block #1 is still in the database, we're allowed to init a new feezer
				}
				// Otherwise, the head header is still the genesis, we're allowed to init a new
				// freezer.
			} else {
				// Full pruning case. Check if the key-value store isn't missing any block.
				if kvhash, _ := db.Get(headerHashKey(offset)); len(kvhash) == 0 {
					printChainMetadata(db)
					return nil, fmt.Errorf("missing blocks from leveldb post ancientdb pruning, block: %d", offset)
				}
			}
		}
	}
	// Freezer is consistent with the key-value database, permit combining the two
	if !disableFreeze {
		frdb.wg.Add(1)

		go func() {
			frdb.freeze(db)
			frdb.wg.Done()
		}()
	}

	return &freezerdb{
		ancientRoot:   ancient,
		KeyValueStore: db,
		chainFreezer:  frdb,
	}, nil
}

// NewMemoryDatabase creates an ephemeral in-memory key-value database without a
// freezer moving immutable chain segments into cold storage.
func NewMemoryDatabase() ethdb.Database {
	return NewDatabase(memorydb.New())
}

const (
	DBPebble  = "pebble"
	DBLeveldb = "leveldb"
)

// PreexistingDatabase checks the given data directory whether a database is already
// instantiated at that location, and if so, returns the type of database (or the
// empty string).
func PreexistingDatabase(path string) string {
	if _, err := os.Stat(filepath.Join(path, "CURRENT")); err != nil {
		return "" // No pre-existing db
	}

	if matches, err := filepath.Glob(filepath.Join(path, "OPTIONS*")); len(matches) > 0 || err != nil {
		if err != nil {
			panic(err) // only possible if the pattern is malformed
		}
		return DBPebble
	}
	return DBLeveldb
}

type counter uint64

func (c counter) String() string {
	return fmt.Sprintf("%d", c)
}

func (c counter) Percentage(current uint64) string {
	return fmt.Sprintf("%d", current*100/uint64(c))
}

// stat stores sizes and count for a parameter
type stat struct {
	size  common.StorageSize
	count counter
}

// Add size to the stat and increase the counter by 1
func (s *stat) Add(size common.StorageSize) {
	s.size += size
	s.count++
}

func (s *stat) Size() string {
	return s.size.String()
}

func (s *stat) Count() string {
	return s.count.String()
}

// AncientInspect inspects the underlying freezer db and prints stats relevant
// for ancient data pruning. It prints the start and end blocks of freezer db.
func AncientInspect(db ethdb.Database) error {
	offset := counter(ReadOffsetOfCurrentAncientFreezer(db))
	// Get number of ancient rows inside the freezer.
	ancients := counter(0)

	if count, err := db.ItemAmountInAncient(); err != nil {
		log.Error("failed to get the items amount in ancientDB", "err", err)
		return err
	} else {
		ancients = counter(count)
	}

	var endNumber counter

	if offset+ancients <= 0 {
		endNumber = 0
	} else {
		sum := counter(0)
		if ancients != 0 {
			sum = ancients - 1
		}
		endNumber = offset + sum
	}

	stats := [][]string{
		{"Start block number of ancientDB (offset)", offset.String()},
		{"End block number of ancientDB", endNumber.String()},
		{"Remaining items in ancientDB", ancients.String()},
	}
	table := tablewriter.NewWriter(os.Stdout)
	table.SetHeader([]string{"Field", "Items"})
	table.SetFooter([]string{"AncientStore information", ""})
	table.AppendBulk(stats)
	table.Render()

	return nil
}

// InspectDatabase traverses the entire database and checks the size
// of all different categories of data.
func InspectDatabase(db ethdb.Database, keyPrefix, keyStart []byte) error {
	it := db.NewIterator(keyPrefix, keyStart)
	defer it.Release()

	var (
		count  int64
		start  = time.Now()
		logged = time.Now()

		// Key-value store statistics
		headers            stat
		bodies             stat
		receipts           stat
		tds                stat
		numHashPairings    stat
		hashNumPairings    stat
		legacyTries        stat
		stateLookups       stat
		accountTries       stat
		storageTries       stat
		codes              stat
		txLookups          stat
		accountSnaps       stat
		storageSnaps       stat
		preimages          stat
		beaconHeaders      stat
		cliqueSnaps        stat
		bloomBits          stat
		filterMapRows      stat
		filterMapLastBlock stat
		filterMapBlockLV   stat

		// Verkle statistics
		verkleTries        stat
		verkleStateLookups stat

		// Meta- and unaccounted data
		metadata    stat
		unaccounted stat

		// Totals
		total common.StorageSize

		// This map tracks example keys for unaccounted data.
		// For each unique two-byte prefix, the first unaccounted key encountered
		// by the iterator will be stored.
		unaccountedKeys = make(map[[2]byte][]byte)
	)
	// Inspect key-value database first.
	for it.Next() {
		var (
			key  = it.Key()
			size = common.StorageSize(len(key) + len(it.Value()))
		)

		total += size

		switch {
		case bytes.HasPrefix(key, headerPrefix) && len(key) == (len(headerPrefix)+8+common.HashLength):
			headers.Add(size)
		case bytes.HasPrefix(key, blockBodyPrefix) && len(key) == (len(blockBodyPrefix)+8+common.HashLength):
			bodies.Add(size)
		case bytes.HasPrefix(key, blockReceiptsPrefix) && len(key) == (len(blockReceiptsPrefix)+8+common.HashLength):
			receipts.Add(size)
		case bytes.HasPrefix(key, headerPrefix) && bytes.HasSuffix(key, headerTDSuffix):
			tds.Add(size)
		case bytes.HasPrefix(key, headerPrefix) && bytes.HasSuffix(key, headerHashSuffix):
			numHashPairings.Add(size)
		case bytes.HasPrefix(key, headerNumberPrefix) && len(key) == (len(headerNumberPrefix)+common.HashLength):
			hashNumPairings.Add(size)
		case IsLegacyTrieNode(key, it.Value()):
			legacyTries.Add(size)
		case bytes.HasPrefix(key, stateIDPrefix) && len(key) == len(stateIDPrefix)+common.HashLength:
			stateLookups.Add(size)
		case IsAccountTrieNode(key):
			accountTries.Add(size)
		case IsStorageTrieNode(key):
			storageTries.Add(size)
		case bytes.HasPrefix(key, CodePrefix) && len(key) == len(CodePrefix)+common.HashLength:
			codes.Add(size)
		case bytes.HasPrefix(key, txLookupPrefix) && len(key) == (len(txLookupPrefix)+common.HashLength):
			txLookups.Add(size)
		case bytes.HasPrefix(key, SnapshotAccountPrefix) && len(key) == (len(SnapshotAccountPrefix)+common.HashLength):
			accountSnaps.Add(size)
		case bytes.HasPrefix(key, SnapshotStoragePrefix) && len(key) == (len(SnapshotStoragePrefix)+2*common.HashLength):
			storageSnaps.Add(size)
		case bytes.HasPrefix(key, PreimagePrefix) && len(key) == (len(PreimagePrefix)+common.HashLength):
			preimages.Add(size)
		case bytes.HasPrefix(key, configPrefix) && len(key) == (len(configPrefix)+common.HashLength):
			metadata.Add(size)
		case bytes.HasPrefix(key, genesisPrefix) && len(key) == (len(genesisPrefix)+common.HashLength):
			metadata.Add(size)
		case bytes.HasPrefix(key, skeletonHeaderPrefix) && len(key) == (len(skeletonHeaderPrefix)+8):
			beaconHeaders.Add(size)
		case bytes.HasPrefix(key, CliqueSnapshotPrefix) && len(key) == 7+common.HashLength:
			cliqueSnaps.Add(size)

		// new log index
		case bytes.HasPrefix(key, filterMapRowPrefix) && len(key) <= len(filterMapRowPrefix)+9:
			filterMapRows.Add(size)
		case bytes.HasPrefix(key, filterMapLastBlockPrefix) && len(key) == len(filterMapLastBlockPrefix)+4:
			filterMapLastBlock.Add(size)
		case bytes.HasPrefix(key, filterMapBlockLVPrefix) && len(key) == len(filterMapBlockLVPrefix)+8:
			filterMapBlockLV.Add(size)

		// old log index (deprecated)
		case bytes.HasPrefix(key, bloomBitsPrefix) && len(key) == (len(bloomBitsPrefix)+10+common.HashLength):
			bloomBits.Add(size)
		case bytes.HasPrefix(key, bloomBitsMetaPrefix) && len(key) < len(bloomBitsMetaPrefix)+8:
			bloomBits.Add(size)

		// Verkle trie data is detected, determine the sub-category
		case bytes.HasPrefix(key, VerklePrefix):
			remain := key[len(VerklePrefix):]
			switch {
			case IsAccountTrieNode(remain):
				verkleTries.Add(size)
			case bytes.HasPrefix(remain, stateIDPrefix) && len(remain) == len(stateIDPrefix)+common.HashLength:
				verkleStateLookups.Add(size)
			case bytes.Equal(remain, persistentStateIDKey):
				metadata.Add(size)
			case bytes.Equal(remain, trieJournalKey):
				metadata.Add(size)
			case bytes.Equal(remain, snapSyncStatusFlagKey):
				metadata.Add(size)
			default:
				unaccounted.Add(size)
			}

		// Metadata keys
		case slices.ContainsFunc(knownMetadataKeys, func(x []byte) bool { return bytes.Equal(x, key) }):
			metadata.Add(size)

		default:
			unaccounted.Add(size)
			if len(key) >= 2 {
				prefix := [2]byte(key[:2])
				if _, ok := unaccountedKeys[prefix]; !ok {
					unaccountedKeys[prefix] = bytes.Clone(key)
				}
			}
		}

		count++
		if count%1000 == 0 && time.Since(logged) > 8*time.Second {
			log.Info("Inspecting database", "count", count, "elapsed", common.PrettyDuration(time.Since(start)))
			logged = time.Now()
		}
	}
	// Display the database statistic of key-value store.
	stats := [][]string{
		{"Key-Value store", "Headers", headers.Size(), headers.Count()},
		{"Key-Value store", "Bodies", bodies.Size(), bodies.Count()},
		{"Key-Value store", "Receipt lists", receipts.Size(), receipts.Count()},
		{"Key-Value store", "Difficulties", tds.Size(), tds.Count()},
		{"Key-Value store", "Block number->hash", numHashPairings.Size(), numHashPairings.Count()},
		{"Key-Value store", "Block hash->number", hashNumPairings.Size(), hashNumPairings.Count()},
		{"Key-Value store", "Transaction index", txLookups.Size(), txLookups.Count()},
		{"Key-Value store", "Log index filter-map rows", filterMapRows.Size(), filterMapRows.Count()},
		{"Key-Value store", "Log index last-block-of-map", filterMapLastBlock.Size(), filterMapLastBlock.Count()},
		{"Key-Value store", "Log index block-lv", filterMapBlockLV.Size(), filterMapBlockLV.Count()},
		{"Key-Value store", "Log bloombits (deprecated)", bloomBits.Size(), bloomBits.Count()},
		{"Key-Value store", "Contract codes", codes.Size(), codes.Count()},
		{"Key-Value store", "Hash trie nodes", legacyTries.Size(), legacyTries.Count()},
		{"Key-Value store", "Path trie state lookups", stateLookups.Size(), stateLookups.Count()},
		{"Key-Value store", "Path trie account nodes", accountTries.Size(), accountTries.Count()},
		{"Key-Value store", "Path trie storage nodes", storageTries.Size(), storageTries.Count()},
		{"Key-Value store", "Verkle trie nodes", verkleTries.Size(), verkleTries.Count()},
		{"Key-Value store", "Verkle trie state lookups", verkleStateLookups.Size(), verkleStateLookups.Count()},
		{"Key-Value store", "Trie preimages", preimages.Size(), preimages.Count()},
		{"Key-Value store", "Account snapshot", accountSnaps.Size(), accountSnaps.Count()},
		{"Key-Value store", "Storage snapshot", storageSnaps.Size(), storageSnaps.Count()},
		{"Key-Value store", "Beacon sync headers", beaconHeaders.Size(), beaconHeaders.Count()},
		{"Key-Value store", "Clique snapshots", cliqueSnaps.Size(), cliqueSnaps.Count()},
		{"Key-Value store", "Singleton metadata", metadata.Size(), metadata.Count()},
	}
	// Inspect all registered append-only file store then.
	ancients, err := inspectFreezers(db)
	if err != nil {
		return err
	}

	for _, ancient := range ancients {
		for _, table := range ancient.sizes {
			stats = append(stats, []string{
				fmt.Sprintf("Ancient store (%s)",
					//nolint:staticcheck
					strings.Title(ancient.name)),
				//nolint:staticcheck
				strings.Title(table.name),
				table.size.String(),
				fmt.Sprintf("%d", ancient.count()),
			})
		}

		total += ancient.size()
	}

	table := tablewriter.NewWriter(os.Stdout)
	table.SetHeader([]string{"Database", "Category", "Size", "Items"})
	table.SetFooter([]string{"", "Total", total.String(), " "})
	table.AppendBulk(stats)
	table.Render()

	if unaccounted.size > 0 {
		log.Error("Database contains unaccounted data", "size", unaccounted.size, "count", unaccounted.count)
		for _, e := range slices.SortedFunc(maps.Values(unaccountedKeys), bytes.Compare) {
			log.Error(fmt.Sprintf("   example key: %x", e))
		}
	}

	return nil
}

// This is the list of known 'metadata' keys stored in the databasse.
var knownMetadataKeys = [][]byte{
	databaseVersionKey, headHeaderKey, headBlockKey, headFastBlockKey, headFinalizedBlockKey,
	lastPivotKey, fastTrieProgressKey, snapshotDisabledKey, SnapshotRootKey, snapshotJournalKey,
	snapshotGeneratorKey, snapshotRecoveryKey, txIndexTailKey, fastTxLookupLimitKey,
	uncleanShutdownKey, badBlockKey, transitionStatusKey, skeletonSyncStatusKey,
	persistentStateIDKey, trieJournalKey, snapshotSyncStatusKey, snapSyncStatusFlagKey,
	filterMapsRangeKey, bytecodeSyncLastBlockKey,
}

// printChainMetadata prints out chain metadata to stderr.
func printChainMetadata(db ethdb.KeyValueStore) {
	fmt.Fprintf(os.Stderr, "Chain metadata\n")

	for _, v := range ReadChainMetadata(db) {
		fmt.Fprintf(os.Stderr, "  %s\n", strings.Join(v, ": "))
	}

	fmt.Fprintf(os.Stderr, "\n\n")
}

// ReadChainMetadata returns a set of key/value pairs that contains information
// about the database chain status. This can be used for diagnostic purposes
// when investigating the state of the node.
func ReadChainMetadata(db ethdb.KeyValueStore) [][]string {
	pp := func(val *uint64) string {
		if val == nil {
			return "<nil>"
		}

		return fmt.Sprintf("%d (%#x)", *val, *val)
	}

	data := [][]string{
		{"databaseVersion", pp(ReadDatabaseVersion(db))},
		{"headBlockHash", fmt.Sprintf("%v", ReadHeadBlockHash(db))},
		{"headFastBlockHash", fmt.Sprintf("%v", ReadHeadFastBlockHash(db))},
		{"headHeaderHash", fmt.Sprintf("%v", ReadHeadHeaderHash(db))},
		{"lastPivotNumber", pp(ReadLastPivotNumber(db))},
		{"len(snapshotSyncStatus)", fmt.Sprintf("%d bytes", len(ReadSnapshotSyncStatus(db)))},
		{"snapshotDisabled", fmt.Sprintf("%v", ReadSnapshotDisabled(db))},
		{"snapshotJournal", fmt.Sprintf("%d bytes", len(ReadSnapshotJournal(db)))},
		{"snapshotRecoveryNumber", pp(ReadSnapshotRecoveryNumber(db))},
		{"snapshotRoot", fmt.Sprintf("%v", ReadSnapshotRoot(db))},
		{"txIndexTail", pp(ReadTxIndexTail(db))},
	}
	if b := ReadSkeletonSyncStatus(db); b != nil {
		data = append(data, []string{"SkeletonSyncStatus", string(b)})
	}
	if fmr, ok, _ := ReadFilterMapsRange(db); ok {
		data = append(data, []string{"filterMapsRange", fmt.Sprintf("%+v", fmr)})
	}
	return data
}

// SafeDeleteRange deletes all of the keys (and values) in the range
// [start,end) (inclusive on start, exclusive on end).
// If hashScheme is true then it always uses an iterator and skips hashdb trie
// node entries. If it is false and the backing db is pebble db then it uses
// the fast native range delete.
// In case of fallback mode (hashdb or leveldb) the range deletion might be
// very slow depending on the number of entries. In this case stopCallback
// is periodically called and if it returns an error then SafeDeleteRange
// stops and also returns that error. The callback is not called if native
// range delete is used or there are a small number of keys only. The bool
// argument passed to the callback is true if enrties have actually been
// deleted already.
func SafeDeleteRange(db ethdb.KeyValueStore, start, end []byte, hashScheme bool, stopCallback func(bool) bool) error {
	if !hashScheme {
		// delete entire range; use fast native range delete on pebble db
		for {
			switch err := db.DeleteRange(start, end); {
			case err == nil:
				return nil
			case errors.Is(err, ethdb.ErrTooManyKeys):
				if stopCallback(true) {
					return ErrDeleteRangeInterrupted
				}
			default:
				return err
			}
		}
	}

	var (
		count, deleted, skipped int
		buff                    = crypto.NewKeccakState()
		startTime               = time.Now()
	)

	batch := db.NewBatch()
	it := db.NewIterator(nil, start)
	defer func() {
		it.Release() // it might be replaced during the process
		log.Debug("SafeDeleteRange finished", "deleted", deleted, "skipped", skipped, "elapsed", common.PrettyDuration(time.Since(startTime)))
	}()

	for it.Next() && bytes.Compare(end, it.Key()) > 0 {
		// Prevent deletion for trie nodes in hash mode
		if len(it.Key()) != 32 || crypto.HashData(buff, it.Value()) != common.BytesToHash(it.Key()) {
			if err := batch.Delete(it.Key()); err != nil {
				return err
			}
			deleted++
		} else {
			skipped++
		}
		count++
		if count > 10000 { // should not block for more than a second
			if err := batch.Write(); err != nil {
				return err
			}
			if stopCallback(deleted != 0) {
				return ErrDeleteRangeInterrupted
			}
			start = append(bytes.Clone(it.Key()), 0) // appending a zero gives us the next possible key
			it.Release()
			batch = db.NewBatch()
			it = db.NewIterator(nil, start)
			count = 0
		}
	}
	return batch.Write()
}
