func (c *Cluster) SchemaVersion() int {
	return len(c.schemaProvider.Updates())
}
SchemaVersion returns the underlying schema version for the cluster
package cluster
import (
	"context"
	"database/sql/driver"
	"fmt"
	"path/filepath"
	"sync"
	"sync/atomic"
	dqlite "github.com/CanonicalLtd/go-dqlite"
	"github.com/pkg/errors"
	"github.com/spoke-d/thermionic/internal/clock"
	"github.com/spoke-d/thermionic/internal/db/database"
	"github.com/spoke-d/thermionic/internal/db/query"
	"github.com/spoke-d/thermionic/internal/db/schema"
	"github.com/spoke-d/thermionic/internal/fsys"
	"github.com/spoke-d/thermionic/internal/version"
)
var StmtInitialNode = `
INSERT INTO nodes 
            (id, 
             name, 
             address, 
             schema, 
             api_extensions) 
VALUES     (1, 
            'none', 
            '0.0.0.0', 
            ?, 
            ?) 
`
type DatabaseRegistrar interface {
	Register(name string, driver driver.Driver)
	Drivers() []string
}
type DatabaseOpener interface {
	Open(driverName, dataSourceName string) (database.DB, error)
}
type DatabaseDriver interface {
	Create(ServerStore, ...dqlite.DriverOption) (driver.Driver, error)
}
type DatabaseIO interface {
	DatabaseDriver
	DatabaseRegistrar
	DatabaseOpener
}
type NameProvider interface {
	DriverName() string
}
type APIExtensions interface {
	Count() int
}
type SchemaProvider interface {
	Schema() Schema
	Updates() []schema.Update
}
type Schema interface {
	Fresh(string)
	File(string)
	Check(schema.Check)
	Hook(schema.Hook)
	Ensure(database.DB) (int, error)
}
type ServerInfo struct {
	ID      uint64
	Address string
}
type ServerStore interface {
	Get(context.Context) ([]ServerInfo, error)
	Set(context.Context, []ServerInfo) error
}
type Cluster struct {
	database       database.DB
	databaseIO     DatabaseIO
	nameProvider   NameProvider
	schemaProvider SchemaProvider
	apiExtensions  APIExtensions
	fileSystem     fsys.FileSystem
	sleeper        clock.Sleeper
	once           sync.Once
}
func New(apiExtensions APIExtensions, schemaProvider SchemaProvider, options ...Option) *Cluster {
	opts := newOptions()
	opts.apiExtensions = apiExtensions
	opts.schemaProvider = schemaProvider
	for _, option := range options {
		option(opts)
	}
	return &Cluster{
		database:       opts.database,
		databaseIO:     opts.databaseIO,
		nameProvider:   opts.nameProvider,
		schemaProvider: opts.schemaProvider,
		apiExtensions:  opts.apiExtensions,
		fileSystem:     opts.fileSystem,
		sleeper:        opts.sleeper,
	}
}
func (c *Cluster) Open(store ServerStore, options ...dqlite.DriverOption) error {
	driver, err := c.databaseIO.Create(store, options...)
	if err != nil {
		return errors.WithStack(err)
	}
	driverName := c.nameProvider.DriverName()
	c.register(driverName, driver)
	db, err := c.databaseIO.Open(driverName, "db.bin?_foreign_keys=1")
	c.database = db
	return errors.Wrap(err, "cannot open cluster database")
}
func (c *Cluster) EnsureSchema(address string, dir string) (bool, error) {
	ctx := &hookContext{}
	numAPIExtensions := c.apiExtensions.Count()
	numSchemaUpdates := len(c.schemaProvider.Updates())
	schema := c.schemaProvider.Schema()
	schema.File(filepath.Join(dir, "patch.global.sql"))
	schema.Check(func(version int, tx database.Tx) error {
		err := check(ctx, address, version, numSchemaUpdates, numAPIExtensions, tx)
		return errors.WithStack(err)
	})
	schema.Hook(func(version int, tx database.Tx) error {
		err := hook(ctx, c.fileSystem, dir, version, tx)
		return errors.WithStack(err)
	})
	var initial int
	err := query.Retry(c.sleeper, func() error {
		var err error
		initial, err = schema.Ensure(c.database)
		return err
	})
	if ctx.someNodesAreBehind {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	if initial == 0 {
		tx, err := c.database.Begin()
		if err != nil {
			return false, errors.WithStack(err)
		}
		numAPIExtensions := c.apiExtensions.Count()
		if _, err := tx.Exec(StmtInitialNode, numSchemaUpdates, numAPIExtensions); err != nil {
			if rollbackErr := tx.Rollback(); rollbackErr != nil {
				return false, errors.Wrap(err, rollbackErr.Error())
			}
			return false, errors.WithStack(err)
		}
		if err := tx.Commit(); err != nil {
			return false, errors.WithStack(err)
		}
	}
	return true, nil
}
func (c *Cluster) DB() database.DB {
	return c.database
}