func (c *Collector) Flush(addresses ...swarm.Address) error {
	var (
		mErr  error
		dirty []string
		batch = new(leveldb.Batch)
	)
	for _, addr := range addresses {
		val, ok := c.counters.Load(addr.ByteString())
		if !ok {
			continue
		}
		cs := val.(*Counters)
		if err := cs.flush(c.db, batch); err != nil {
			mErr = multierror.Append(mErr, fmt.Errorf("unable to batch the counters of peer %q for flash: %w", addr, err))
			continue
		}
		dirty = append(dirty, addr.ByteString())
	}
	if len(addresses) == 0 {
		c.counters.Range(func(_, val interface{}) bool {
			cs := val.(*Counters)
			if err := cs.flush(c.db, batch); err != nil {
				mErr = multierror.Append(mErr, fmt.Errorf("unable to batch the counters of peer %q for flash: %w", cs.peer, err))
				return true
			}
			dirty = append(dirty, cs.peer.ByteString())
			return true
		})
	}
	if batch.Len() == 0 {
		return mErr
	}
	if err := c.db.WriteBatch(batch); err != nil {
		mErr = multierror.Append(mErr, fmt.Errorf("unable to persist counters in batch: %w", err))
	}
	for _, addr := range dirty {
		val, ok := c.counters.Load(addr)
		if !ok {
			continue
		}
		cs := val.(*Counters)
		cs.dirty.CAS(1, 0)
	}
	return mErr
}
Flush sync the dirty in memory counters by flushing their values to the
underlying storage. If an address or a set of addresses is specified then
only counters related to them will be flushed, otherwise counters for all
peers will be flushed.
package metrics
import (
	"fmt"
	"sync"
	"time"
	"github.com/ethsana/sana/pkg/shed"
	"github.com/ethsana/sana/pkg/swarm"
	"github.com/hashicorp/go-multierror"
	"github.com/syndtr/goleveldb/leveldb"
	"go.uber.org/atomic"
)
const (
	peerLastSeenTimestamp       string = "peer-last-seen-timestamp"
	peerTotalConnectionDuration string = "peer-total-connection-duration"
)
type PeerConnectionDirection string
const (
	PeerConnectionDirectionInbound  PeerConnectionDirection = "inbound"
	PeerConnectionDirectionOutbound PeerConnectionDirection = "outbound"
)
type peerKey struct {
	prefix  string
	address string
}
func (pk peerKey) String() string {
	return fmt.Sprintf("%s-%s", pk.prefix, pk.address)
}
func newPeerKey(p, a string) *peerKey {
	return &peerKey{
		prefix:  p,
		address: a,
	}
}
type RecordOp func(*Counters)
func PeerLogIn(t time.Time, dir PeerConnectionDirection) RecordOp {
	return func(cs *Counters) {
		cs.Lock()
		defer cs.Unlock()
		if cs.loggedIn {
			return  
		}
		cs.loggedIn = true
		ls := t.UnixNano()
		if ls < 0 {
			panic(fmt.Errorf("time before unix epoch: %s", t))
		}
		cs.sessionConnDirection = dir
		cs.lastSeenTimestamp = ls
		cs.dirty.Store(3)
	}
}
func PeerLogOut(t time.Time) RecordOp {
	return func(cs *Counters) {
		cs.Lock()
		defer cs.Unlock()
		if !cs.loggedIn {
			return  
		}
		cs.loggedIn = false
		curLs := cs.lastSeenTimestamp
		newLs := t.UnixNano()
		if newLs < 0 {
			panic(fmt.Errorf("time before unix epoch: %s", t))
		}
		cs.sessionConnDuration = time.Duration(newLs - curLs)
		cs.connTotalDuration += cs.sessionConnDuration
		cs.lastSeenTimestamp = newLs
		cs.dirty.Store(3)
	}
}
func IncSessionConnectionRetry() RecordOp {
	return func(cs *Counters) {
		cs.Lock()
		defer cs.Unlock()
		cs.sessionConnRetry++
	}
}
type Snapshot struct {
	LastSeenTimestamp          int64
	SessionConnectionRetry     uint64
	ConnectionTotalDuration    time.Duration
	SessionConnectionDuration  time.Duration
	SessionConnectionDirection PeerConnectionDirection
}
func (ss *Snapshot) HasAtMaxOneConnectionAttempt() bool {
	return ss.LastSeenTimestamp == 0 && ss.SessionConnectionRetry <= 1
}
type Counters struct {
	sync.Mutex
	peer     *swarm.Address
	loggedIn bool
	dirty atomic.Int32
	lastSeenTimestamp    int64
	connTotalDuration    time.Duration
	sessionConnRetry     uint64
	sessionConnDuration  time.Duration
	sessionConnDirection PeerConnectionDirection
	persistentLastSeenTimestamp atomic.Value
	persistentConnTotalDuration atomic.Value
}
func (cs *Counters) flush(db *shed.DB, batch *leveldb.Batch) error {
	if cs.dirty.Load() > 1 {
		return nil
	}
	cs.dirty.CAS(3, 2)
	cs.Lock()
	var (
		key                             = cs.peer.String()
		lastSeenTimestampSnapshot       = cs.lastSeenTimestamp
		connectionTotalDurationSnapshot = cs.connTotalDuration
	)
	cs.Unlock()
	ls, ok := cs.persistentLastSeenTimestamp.Load().(*shed.Uint64Field)
	if !ok {
		mk := newPeerKey(peerLastSeenTimestamp, key)
		field, err := db.NewUint64Field(mk.String())
		if err != nil {
			return fmt.Errorf("field initialization for %q failed: %w", mk, err)
		}
		ls = &field
		cs.persistentLastSeenTimestamp.Store(ls)
	}
	cd, ok := cs.persistentConnTotalDuration.Load().(*shed.Uint64Field)
	if !ok {
		mk := newPeerKey(peerTotalConnectionDuration, key)
		field, err := db.NewUint64Field(mk.String())
		if err != nil {
			return fmt.Errorf("field initialization for %q failed: %w", mk, err)
		}
		cd = &field
		cs.persistentConnTotalDuration.Store(cd)
	}
	ls.PutInBatch(batch, uint64(lastSeenTimestampSnapshot))
	cd.PutInBatch(batch, uint64(connectionTotalDurationSnapshot))
	cs.dirty.CAS(2, 1)
	return nil
}
func (cs *Counters) snapshot(t time.Time) *Snapshot {
	cs.Lock()
	defer cs.Unlock()
	connTotalDuration := cs.connTotalDuration
	sessionConnDuration := cs.sessionConnDuration
	if cs.loggedIn {
		sessionConnDuration = t.Sub(time.Unix(0, cs.lastSeenTimestamp))
		connTotalDuration += sessionConnDuration
	}
	return &Snapshot{
		LastSeenTimestamp:          cs.lastSeenTimestamp,
		SessionConnectionRetry:     cs.sessionConnRetry,
		ConnectionTotalDuration:    connTotalDuration,
		SessionConnectionDuration:  sessionConnDuration,
		SessionConnectionDirection: cs.sessionConnDirection,
	}
}
func NewCollector(db *shed.DB) *Collector {
	return &Collector{db: db}
}
type Collector struct {
	db       *shed.DB
	counters sync.Map
}
func (c *Collector) Record(addr swarm.Address, rop ...RecordOp) {
	val, _ := c.counters.LoadOrStore(addr.ByteString(), &Counters{peer: &addr})
	for _, op := range rop {
		op(val.(*Counters))
	}
}
func (c *Collector) Snapshot(t time.Time, addresses ...swarm.Address) map[string]*Snapshot {
	snapshot := make(map[string]*Snapshot)
	for _, addr := range addresses {
		val, ok := c.counters.Load(addr.ByteString())
		if !ok {
			continue
		}
		cs := val.(*Counters)
		snapshot[addr.ByteString()] = cs.snapshot(t)
	}
	if len(addresses) == 0 {
		c.counters.Range(func(key, val interface{}) bool {
			cs := val.(*Counters)
			snapshot[cs.peer.ByteString()] = cs.snapshot(t)
			return true
		})
	}
	return snapshot
}
func (c *Collector) Inspect(addr swarm.Address, fn func(ss *Snapshot)) {
	snapshots := c.Snapshot(time.Now(), addr)
	fn(snapshots[addr.ByteString()])
}