mdb/testdb_test.go

package mdb

import (
	"errors"
	"net/mail"
	"strings"
	"time"
)

// ----------------------------------------------------------------------------
// Validate errors.
// ----------------------------------------------------------------------------

var ErrInvalidName = errors.New("invalid name")

// ----------------------------------------------------------------------------
// User Collection
// ----------------------------------------------------------------------------

type User struct {
	ID    uint64
	Email string
	Name  string
	ExtID string
}

type Users struct {
	c          *Collection[User]
	emailMap   *MapIndex[string, User] // Full map index.
	emailBTree *BTreeIndex[User]       // Full btree index.
	nameBTree  *BTreeIndex[User]       // Full btree with duplicates.
	extIDMap   *MapIndex[string, User] // Partial map index.
	extIDBTree *BTreeIndex[User]       // Partial btree index.
}

func userGetID(u *User) uint64 { return u.ID }

func userSanitize(u *User) {
	u.Name = strings.TrimSpace(u.Name)
	e, err := mail.ParseAddress(strings.ToLower(strings.TrimSpace(u.Email)))
	if err == nil {
		u.Email = e.Address
	}
}

func userValidate(u *User) error {
	if len(u.Name) == 0 {
		return ErrInvalidName
	}
	return nil
}

// ----------------------------------------------------------------------------
// Account Collection
// ----------------------------------------------------------------------------

type Account struct {
	ID   uint64
	Name string
}

type Accounts struct {
	c       *Collection[Account]
	nameMap *MapIndex[string, Account]
}

func accountGetID(a *Account) uint64 { return a.ID }

// ----------------------------------------------------------------------------
// Database
// ----------------------------------------------------------------------------

type DB struct {
	*Database
	root     string
	Users    Users
	Accounts Accounts
}

func OpenDB(root string, primary bool) *DB {
	db := &DB{root: root}
	if primary {
		db.Database = NewPrimary(root)
	} else {
		db.Database = NewSecondary(root)
	}

	db.Users = Users{}
	db.Users.c = NewCollection(db.Database, "users", userGetID)
	db.Users.c.SetSanitize(userSanitize)
	db.Users.c.SetValidate(userValidate)

	db.Users.emailMap = NewMapIndex(
		db.Users.c,
		"email",
		func(u *User) string { return u.Email },
		nil)

	db.Users.emailBTree = NewBTreeIndex(
		db.Users.c,
		func(lhs, rhs *User) bool { return lhs.Email < rhs.Email },
		nil)

	db.Users.nameBTree = NewBTreeIndex(
		db.Users.c,
		func(lhs, rhs *User) bool {
			if lhs.Name != rhs.Name {
				return lhs.Name < rhs.Name
			}
			return lhs.ID < rhs.ID
		},
		nil)

	db.Users.extIDMap = NewMapIndex(
		db.Users.c,
		"extID",
		func(u *User) string { return u.ExtID },
		func(u *User) bool { return u.ExtID != "" })

	db.Users.extIDBTree = NewBTreeIndex(
		db.Users.c,
		func(lhs, rhs *User) bool { return lhs.ExtID < rhs.ExtID },
		func(u *User) bool { return u.ExtID != "" })

	db.Accounts = Accounts{}
	db.Accounts.c = NewCollection(db.Database, "accounts", accountGetID)

	return db
}

func (db *DB) Equals(rhs *DB) error {
	db.WaitForSync(rhs)

	// Users: itemMap.
	if err := db.Users.c.items.Equals(rhs.Users.c.items); err != nil {
		return err
	}

	// Users: emailMap
	if err := db.Users.emailMap.Equals(rhs.Users.emailMap); err != nil {
		return err
	}

	// Users: emailBTree
	if err := db.Users.emailBTree.Equals(rhs.Users.emailBTree); err != nil {
		return err
	}

	// Users: nameBTree
	if err := db.Users.nameBTree.Equals(rhs.Users.nameBTree); err != nil {
		return err
	}

	// Users: extIDMap
	if err := db.Users.extIDMap.Equals(rhs.Users.extIDMap); err != nil {
		return err
	}

	// Users: extIDBTree
	if err := db.Users.extIDBTree.Equals(rhs.Users.extIDBTree); err != nil {
		return err
	}

	// Accounts: itemMap
	if err := db.Accounts.c.items.Equals(rhs.Accounts.c.items); err != nil {
		return err
	}

	// Accounts: nameMap
	if err := db.Accounts.nameMap.Equals(rhs.Accounts.nameMap); err != nil {
		return err
	}

	return nil
}

// Wait for two databases to become synchronized.
func (db *DB) WaitForSync(rhs *DB) {
	for {
		s1 := db.WALStatus()
		s2 := rhs.WALStatus()
		if s1 == s2 && s1.MaxSeqNumKV == s1.MaxSeqNumWAL {
			return
		}
		time.Sleep(100 * time.Millisecond)
	}
}