<commit_before>package data

import (
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"strconv"

	"github.com/h2oai/steamY/master/auth"
	"github.com/h2oai/steamY/master/az"
	"github.com/lib/pq"
)

//
// -----  Privilege / Sharing rules -----
//
// --------------------------------------
// Entity               Own  Edit View
// --------------------------------------
// Role
//   Read               x    x    x
//   Update             x    x
//   Assign Permission  x    x
//   Delete             x
//   Share              x
//
// Workgroup
//   Read               x    x    x
//   Update             x    x
//   Delete             x
//   Share              x
//
// Identity
//   Read               x    x    x
//   Assign Role        x    x
//   Assign Workgroup   x    x
//   Update             x    x
//   Delete             x
//   Share              x
//
// Cluster
//   Read               x    x    x
//   Start/Stop         x
//
// Project
//   Read               x    x    x
//   Assign Model       x    x
//   Update             x    x
//   Delete             x
//   Share              x
//
// Engine, Datasource, Dataset, Model
//   Read               x    x    x
//   Update             x    x
//   Delete             x
//   Share              x
//
// --------------------------------------

const (
	View = 1 << iota
	Edit // 2
	Own  // 4
)

const (
	Version = "1"

	SuperuserRoleName = "Superuser"

	CanView = "view"
	CanEdit = "edit"
	Owns    = "own"

	ForIdentity  = "identity"
	ForWorkgroup = "workgroup"

	RoleEntity       = "role"
	WorkgroupEntity  = "workgroup"
	IdentityEntity   = "identity"
	EngineEntity     = "engine"
	ClusterEntity    = "cluster"
	ProjectEntity    = "project"
	DatasourceEntity = "datasource"
	DatasetEntity    = "dataset"
	ModelEntity      = "model"
	ServiceEntity    = "service"

	ClusterExternal = "external"
	ClusterYarn     = "yarn"
)

const (
	IdleState         = "idle"
	StartingState     = "starting"
	StartedState      = "started"
	SuspendingState   = "suspending"
	SuspendedState    = "suspended"
	StoppingState     = "stopping"
	StoppedState      = "stopped"
	BlockedState      = "blocked"
	DisconnectedState = "disconnected"
	FailedState       = "failed"
	CompletedState    = "completed"
)

const (
	ManageRole       = "ManageRole"
	ViewRole         = "ViewRole"
	ManageWorkgroup  = "ManageWorkgroup"
	ViewWorkgroup    = "ViewWorkgroup"
	ManageIdentity   = "ManageIdentity"
	ViewIdentity     = "ViewIdentity"
	ManageEngine     = "ManageEngine"
	ViewEngine       = "ViewEngine"
	ManageCluster    = "ManageCluster"
	ViewCluster      = "ViewCluster"
	ManageProject    = "ManageProject"
	ViewProject      = "ViewProject"
	ManageDatasource = "ManageDatasource"
	ViewDatasource   = "ViewDatasource"
	ManageDataset    = "ManageDataset"
	ViewDataset      = "ViewDataset"
	ManageModel      = "ManageModel"
	ViewModel        = "ViewModel"
	ManageService    = "ManageService"
	ViewService      = "ViewService"
)

var (
	Privileges   map[string]int
	Permissions  []Permission
	EntityTypes  []EntityType
	ClusterTypes []ClusterType
)

func init() {
	Privileges = map[string]int{
		CanView: View,
		CanEdit: Edit,
		Owns:    Own,
	}

	Permissions = []Permission{
		{0, ManageRole, "Manage roles"},
		{0, ViewRole, "View roles"},
		{0, ManageWorkgroup, "Manage workgroups"},
		{0, ViewWorkgroup, "View workgroups"},
		{0, ManageIdentity, "Manage identities"},
		{0, ViewIdentity, "View identities"},
		{0, ManageEngine, "Manage engines"},
		{0, ViewEngine, "View engines"},
		{0, ManageCluster, "Manage clusters"},
		{0, ViewCluster, "View clusters"},
		{0, ManageProject, "Manage projects"},
		{0, ViewProject, "View projects"},
		{0, ManageDatasource, "Manage datasources"},
		{0, ViewDatasource, "View datasources"},
		{0, ManageDataset, "Manage datasets"},
		{0, ViewDataset, "View datasets"},
		{0, ManageModel, "Manage models"},
		{0, ViewModel, "View models"},
		{0, ManageService, "Manage services"},
		{0, ViewService, "View services"},
	}

	EntityTypes = []EntityType{
		{0, RoleEntity},
		{0, WorkgroupEntity},
		{0, IdentityEntity},
		{0, EngineEntity},
		{0, ClusterEntity},
		{0, ProjectEntity},
		{0, DatasourceEntity},
		{0, DatasetEntity},
		{0, ModelEntity},
		{0, ServiceEntity},
	}

	ClusterTypes = []ClusterType{
		{0, ClusterExternal},
		{0, ClusterYarn},
	}
}

type metadata map[string]string

type PermissionKeys struct {
	ManageRole       int64
	ViewRole         int64
	ManageWorkgroup  int64
	ViewWorkgroup    int64
	ManageIdentity   int64
	ViewIdentity     int64
	ManageEngine     int64
	ViewEngine       int64
	ManageCluster    int64
	ViewCluster      int64
	ManageProject    int64
	ViewProject      int64
	ManageDatasource int64
	ViewDatasource   int64
	ManageDataset    int64
	ViewDataset      int64
	ManageModel      int64
	ViewModel        int64
	ManageService    int64
	ViewService      int64
}

type EntityTypeKeys struct {
	Role       int64
	Workgroup  int64
	Identity   int64
	Engine     int64
	Cluster    int64
	Project    int64
	Datasource int64
	Dataset    int64
	Model      int64
	Service    int64
}

type ClusterTypeKeys struct {
	External int64
	Yarn     int64
}

func toPermissionKeys(permissions []Permission) *PermissionKeys {
	m := make(map[string]int64)
	for _, p := range permissions {
		m[p.Code] = p.Id
	}

	return &PermissionKeys{
		m[ManageRole],
		m[ViewRole],
		m[ManageWorkgroup],
		m[ViewWorkgroup],
		m[ManageIdentity],
		m[ViewIdentity],
		m[ManageEngine],
		m[ViewEngine],
		m[ManageCluster],
		m[ViewCluster],
		m[ManageProject],
		m[ViewProject],
		m[ManageDatasource],
		m[ViewDatasource],
		m[ManageDataset],
		m[ViewDataset],
		m[ManageModel],
		m[ViewModel],
		m[ManageService],
		m[ViewService],
	}
}

func toEntityTypeKeys(entityTypes []EntityType) *EntityTypeKeys {
	m := make(map[string]int64)
	for _, et := range entityTypes {
		m[et.Name] = et.Id
	}

	return &EntityTypeKeys{
		m[RoleEntity],
		m[WorkgroupEntity],
		m[IdentityEntity],
		m[EngineEntity],
		m[ClusterEntity],
		m[ProjectEntity],
		m[DatasourceEntity],
		m[DatasetEntity],
		m[ModelEntity],
		m[ServiceEntity],
	}
}

func toClusterTypeKeys(clusterTypes []ClusterType) *ClusterTypeKeys {
	m := make(map[string]int64)
	for _, ct := range clusterTypes {
		m[ct.Name] = ct.Id
	}
	return &ClusterTypeKeys{
		m[ClusterExternal],
		m[ClusterYarn],
	}
}

type Datastore struct {
	db                *sql.DB // Singleton; doesn't actually connect until used, and is pooled internally.
	metadata          metadata
	permissions       []Permission
	permissionMap     map[int64]Permission
	Permissions       *PermissionKeys
	entityTypes       []EntityType
	entityTypeMap     map[int64]EntityType
	EntityTypes       *EntityTypeKeys
	clusterTypes      []ClusterType
	clusterTypeMap    map[int64]ClusterType
	ClusterTypes      *ClusterTypeKeys
	ViewPermissions   map[int64]int64
	ManagePermissions map[int64]int64
}

func Create(name, username, sslmode, suname, supass string) (*Datastore, error) {
	db, err := connect(username, name, sslmode)
	if err != nil {
		return nil, fmt.Errorf("Failed connecting to database %s as user %s (SSL=%s): %s\n", name, username, sslmode, err)
	}

	primed, err := isPrimed(db)
	if err != nil {
		return nil, fmt.Errorf("Failed database version check:", err)
	}

	if !primed {
		if suname == "" || supass == "" {
			return nil, fmt.Errorf("Starting Steam for the first time requires both --superuser-name and --superuser-password arguments to \"steam serve master\".")
		}

		if err := auth.ValidateUsername(suname); err != nil {
			return nil, fmt.Errorf("Invalid superuser username: %s", err)
		}

		if err := auth.ValidatePassword(supass); err != nil {
			return nil, fmt.Errorf("Invalid superuser password: %s", err)
		}

		if err := prime(db); err != nil {
			return nil, fmt.Errorf("Failed priming database: %s", err)
		}
	}

	ds, err := newDatastore(db)
	if err != nil {
		return nil, fmt.Errorf("Failed initializing from database: %s", err)
	}

	if !primed {
		passwordHash, err := auth.HashPassword(supass)
		if err != nil {
			return nil, fmt.Errorf("Failed hashing superuser password: %s", err)
		}

		if _, _, err := ds.CreateSuperuser(suname, passwordHash); err != nil {
			return nil, fmt.Errorf("Failed superuser identity setup: %s", err)
		}

		_, err = ds.Lookup(suname)
		if err != nil {
			return nil, fmt.Errorf("Failed reading superuser principal: %s", err)
		}
	}

	return ds, nil
}

func Destroy(name, username, sslmode string) error {
	db, err := connect(username, name, sslmode)
	if err != nil {
		return fmt.Errorf("Failed connecting to database %s as user %s (SSL=%s): %s\n", name, username, sslmode, err)
	}
	return truncate(db)
}

func connect(username, dbname, sslmode string) (*sql.DB, error) {

	log.Println("Connecting to database: user =", username, "db =", dbname, "SSL=", sslmode, "...")

	// Open connection
	db, err := sql.Open("postgres", fmt.Sprintf("user=%s dbname=%s sslmode=%s", username, dbname, sslmode))
	if err != nil {
		return nil, fmt.Errorf("Database connection failed: %s", err)
	}

	// TODO can use db.SetMaxOpenConns() and db.SetMaxIdleConns() to configure further.

	// Verify connection
	if err := db.Ping(); err != nil {
		return nil, fmt.Errorf("Database ping failed: %s", err)
	}

	return db, nil
}

// newDatastore creates a new instance of a data access object.
//
// Valid values for sslmode are:
//   disable - No SSL
//   require - Always SSL (skip verification)
//   verify-ca - Always SSL (verify that the certificate presented by the server was signed by a trusted CA)
//   verify-full - Always SSL (verify that the certification presented by the server was signed by a
//     trusted CA and the server host name matches the one in the certificate)
func newDatastore(db *sql.DB) (*Datastore, error) {

	// Read meta information

	metadata, err := readMetadata(db)
	if err != nil {
		return nil, err
	}

	version, ok := metadata["version"]
	if !ok {
		return nil, fmt.Errorf("Failed reading schema version")
	}

	log.Println("Using schema version:", version)

	if err := upgrade(db, version); err != nil {
		return nil, err
	}

	permissions, err := readAllPermissions(db)
	if err != nil {
		return nil, err
	}

	permissionMap := make(map[int64]Permission)
	for _, permission := range permissions {
		permissionMap[permission.Id] = permission
	}

	permissionKeys := toPermissionKeys(permissions)

	entityTypes, err := readEntityTypes(db)
	if err != nil {
		return nil, err
	}

	entityTypeMap := make(map[int64]EntityType)
	for _, et := range entityTypes {
		entityTypeMap[et.Id] = et
	}

	entityTypeKeys := toEntityTypeKeys(entityTypes)

	clusterTypes, err := readClusterTypes(db)
	if err != nil {
		return nil, err
	}

	clusterTypeMap := make(map[int64]ClusterType)
	for _, ct := range clusterTypes {
		clusterTypeMap[ct.Id] = ct
	}

	viewPermissions := map[int64]int64{
		entityTypeKeys.Engine:     permissionKeys.ViewEngine,
		entityTypeKeys.Cluster:    permissionKeys.ViewCluster,
		entityTypeKeys.Project:    permissionKeys.ViewProject,
		entityTypeKeys.Datasource: permissionKeys.ViewDatasource,
		entityTypeKeys.Dataset:    permissionKeys.ViewDataset,
		entityTypeKeys.Model:      permissionKeys.ViewModel,
		entityTypeKeys.Service:    permissionKeys.ViewService,
		entityTypeKeys.Identity:   permissionKeys.ViewIdentity,
		entityTypeKeys.Role:       permissionKeys.ViewRole,
		entityTypeKeys.Workgroup:  permissionKeys.ViewWorkgroup,
	}

	managePermissions := map[int64]int64{
		entityTypeKeys.Engine:     permissionKeys.ManageEngine,
		entityTypeKeys.Cluster:    permissionKeys.ManageCluster,
		entityTypeKeys.Project:    permissionKeys.ManageProject,
		entityTypeKeys.Datasource: permissionKeys.ManageDatasource,
		entityTypeKeys.Dataset:    permissionKeys.ManageDataset,
		entityTypeKeys.Model:      permissionKeys.ManageModel,
		entityTypeKeys.Service:    permissionKeys.ManageService,
		entityTypeKeys.Identity:   permissionKeys.ManageIdentity,
		entityTypeKeys.Role:       permissionKeys.ManageRole,
		entityTypeKeys.Workgroup:  permissionKeys.ManageWorkgroup,
	}

	return &Datastore{
		db,
		metadata,
		permissions,
		permissionMap,
		permissionKeys,
		entityTypes,
		entityTypeMap,
		entityTypeKeys,
		clusterTypes,
		clusterTypeMap,
		toClusterTypeKeys(clusterTypes),
		viewPermissions,
		managePermissions,
	}, nil
}

func isPrimed(db *sql.DB) (bool, error) {
	row := db.QueryRow(`
		SELECT 
			count(1)
		FROM
			meta
		`)
	count, err := scanInt(row)
	if err != nil {
		return false, err
	}
	return count > 0, err
}

func prime(db *sql.DB) error {
	log.Println("Priming database for first time use...")
	if err := createMetadata(db, "version", "1"); err != nil {
		return err
	}
	if err := primePermissions(db, Permissions); err != nil {
		return err
	}
	if err := primeEntityTypes(db, EntityTypes); err != nil {
		return err
	}
	if err := primeClusterTypes(db, ClusterTypes); err != nil {
		return err
	}

	return nil
}

func bulkInsert(db *sql.DB, table string, columns []string, f func(*sql.Stmt) error) error {
	tx, err := db.Begin()
	if err != nil {
		return err
	}

	stmt, err := tx.Prepare(pq.CopyIn(table, columns...))
	if err != nil {
		return err
	}

	if err := f(stmt); err != nil { // buffer
		return err
	}

	if _, err := stmt.Exec(); err != nil { // flush
		return err
	}

	if err := stmt.Close(); err != nil {
		return err
	}

	if err := tx.Commit(); err != nil {
		return err
	}

	return nil
}

func createMetadata(db *sql.DB, key, value string) error {
	_, err := db.Exec(`
		INSERT INTO
			meta
			(key, value)
		VALUES
			($1,  $2)
		`, key, value)
	return err
}

func primeEntityTypes(db *sql.DB, entityTypes []EntityType) error {
	return bulkInsert(db, "entity_type", []string{"name"}, func(stmt *sql.Stmt) error {
		for _, entityType := range entityTypes {
			_, err := stmt.Exec(entityType.Name)
			if err != nil {
				return err
			}
		}
		return nil
	})
}

func primeClusterTypes(db *sql.DB, clusterTypes []ClusterType) error {
	return bulkInsert(db, "cluster_type", []string{"name"}, func(stmt *sql.Stmt) error {
		for _, clusterType := range clusterTypes {
			_, err := stmt.Exec(clusterType.Name)
			if err != nil {
				return err
			}
		}
		return nil
	})
}

func primePermissions(db *sql.DB, permissions []Permission) error {
	return bulkInsert(db, "permission", []string{"code", "description"}, func(stmt *sql.Stmt) error {
		for _, permission := range permissions {
			_, err := stmt.Exec(permission.Code, permission.Description)
			if err != nil {
				return err
			}
		}
		return nil
	})
}

func upgrade(db *sql.DB, currentVersion string) error {
	if currentVersion == Version {
		return nil
	}

	// TODO add stepwise upgrades

	return nil
}

func truncate(db *sql.DB) error {
	log.Println("Truncating database...")
	return executeTransaction(db, func(tx *sql.Tx) error {
		tables := []string{
			"history",
			"privilege",
			"role_permission",
			"identity_role",
			"identity_workgroup",
			"identity",
			"workgroup",
			"role",
			"permission",<commit_msg>STEAM-282: temporarily remove db logging for testing<commit_after>package data

import (
	"database/sql"
	"encoding/json"
	"fmt"
	"strconv"

	"github.com/h2oai/steamY/master/auth"
	"github.com/h2oai/steamY/master/az"
	"github.com/lib/pq"
)

//
// -----  Privilege / Sharing rules -----
//
// --------------------------------------
// Entity               Own  Edit View
// --------------------------------------
// Role
//   Read               x    x    x
//   Update             x    x
//   Assign Permission  x    x
//   Delete             x
//   Share              x
//
// Workgroup
//   Read               x    x    x
//   Update             x    x
//   Delete             x
//   Share              x
//
// Identity
//   Read               x    x    x
//   Assign Role        x    x
//   Assign Workgroup   x    x
//   Update             x    x
//   Delete             x
//   Share              x
//
// Cluster
//   Read               x    x    x
//   Start/Stop         x
//
// Project
//   Read               x    x    x
//   Assign Model       x    x
//   Update             x    x
//   Delete             x
//   Share              x
//
// Engine, Datasource, Dataset, Model
//   Read               x    x    x
//   Update             x    x
//   Delete             x
//   Share              x
//
// --------------------------------------

const (
	View = 1 << iota
	Edit // 2
	Own  // 4
)

const (
	Version = "1"

	SuperuserRoleName = "Superuser"

	CanView = "view"
	CanEdit = "edit"
	Owns    = "own"

	ForIdentity  = "identity"
	ForWorkgroup = "workgroup"

	RoleEntity       = "role"
	WorkgroupEntity  = "workgroup"
	IdentityEntity   = "identity"
	EngineEntity     = "engine"
	ClusterEntity    = "cluster"
	ProjectEntity    = "project"
	DatasourceEntity = "datasource"
	DatasetEntity    = "dataset"
	ModelEntity      = "model"
	ServiceEntity    = "service"

	ClusterExternal = "external"
	ClusterYarn     = "yarn"
)

const (
	IdleState         = "idle"
	StartingState     = "starting"
	StartedState      = "started"
	SuspendingState   = "suspending"
	SuspendedState    = "suspended"
	StoppingState     = "stopping"
	StoppedState      = "stopped"
	BlockedState      = "blocked"
	DisconnectedState = "disconnected"
	FailedState       = "failed"
	CompletedState    = "completed"
)

const (
	ManageRole       = "ManageRole"
	ViewRole         = "ViewRole"
	ManageWorkgroup  = "ManageWorkgroup"
	ViewWorkgroup    = "ViewWorkgroup"
	ManageIdentity   = "ManageIdentity"
	ViewIdentity     = "ViewIdentity"
	ManageEngine     = "ManageEngine"
	ViewEngine       = "ViewEngine"
	ManageCluster    = "ManageCluster"
	ViewCluster      = "ViewCluster"
	ManageProject    = "ManageProject"
	ViewProject      = "ViewProject"
	ManageDatasource = "ManageDatasource"
	ViewDatasource   = "ViewDatasource"
	ManageDataset    = "ManageDataset"
	ViewDataset      = "ViewDataset"
	ManageModel      = "ManageModel"
	ViewModel        = "ViewModel"
	ManageService    = "ManageService"
	ViewService      = "ViewService"
)

var (
	Privileges   map[string]int
	Permissions  []Permission
	EntityTypes  []EntityType
	ClusterTypes []ClusterType
)

func init() {
	Privileges = map[string]int{
		CanView: View,
		CanEdit: Edit,
		Owns:    Own,
	}

	Permissions = []Permission{
		{0, ManageRole, "Manage roles"},
		{0, ViewRole, "View roles"},
		{0, ManageWorkgroup, "Manage workgroups"},
		{0, ViewWorkgroup, "View workgroups"},
		{0, ManageIdentity, "Manage identities"},
		{0, ViewIdentity, "View identities"},
		{0, ManageEngine, "Manage engines"},
		{0, ViewEngine, "View engines"},
		{0, ManageCluster, "Manage clusters"},
		{0, ViewCluster, "View clusters"},
		{0, ManageProject, "Manage projects"},
		{0, ViewProject, "View projects"},
		{0, ManageDatasource, "Manage datasources"},
		{0, ViewDatasource, "View datasources"},
		{0, ManageDataset, "Manage datasets"},
		{0, ViewDataset, "View datasets"},
		{0, ManageModel, "Manage models"},
		{0, ViewModel, "View models"},
		{0, ManageService, "Manage services"},
		{0, ViewService, "View services"},
	}

	EntityTypes = []EntityType{
		{0, RoleEntity},
		{0, WorkgroupEntity},
		{0, IdentityEntity},
		{0, EngineEntity},
		{0, ClusterEntity},
		{0, ProjectEntity},
		{0, DatasourceEntity},
		{0, DatasetEntity},
		{0, ModelEntity},
		{0, ServiceEntity},
	}

	ClusterTypes = []ClusterType{
		{0, ClusterExternal},
		{0, ClusterYarn},
	}
}

type metadata map[string]string

type PermissionKeys struct {
	ManageRole       int64
	ViewRole         int64
	ManageWorkgroup  int64
	ViewWorkgroup    int64
	ManageIdentity   int64
	ViewIdentity     int64
	ManageEngine     int64
	ViewEngine       int64
	ManageCluster    int64
	ViewCluster      int64
	ManageProject    int64
	ViewProject      int64
	ManageDatasource int64
	ViewDatasource   int64
	ManageDataset    int64
	ViewDataset      int64
	ManageModel      int64
	ViewModel        int64
	ManageService    int64
	ViewService      int64
}

type EntityTypeKeys struct {
	Role       int64
	Workgroup  int64
	Identity   int64
	Engine     int64
	Cluster    int64
	Project    int64
	Datasource int64
	Dataset    int64
	Model      int64
	Service    int64
}

type ClusterTypeKeys struct {
	External int64
	Yarn     int64
}

func toPermissionKeys(permissions []Permission) *PermissionKeys {
	m := make(map[string]int64)
	for _, p := range permissions {
		m[p.Code] = p.Id
	}

	return &PermissionKeys{
		m[ManageRole],
		m[ViewRole],
		m[ManageWorkgroup],
		m[ViewWorkgroup],
		m[ManageIdentity],
		m[ViewIdentity],
		m[ManageEngine],
		m[ViewEngine],
		m[ManageCluster],
		m[ViewCluster],
		m[ManageProject],
		m[ViewProject],
		m[ManageDatasource],
		m[ViewDatasource],
		m[ManageDataset],
		m[ViewDataset],
		m[ManageModel],
		m[ViewModel],
		m[ManageService],
		m[ViewService],
	}
}

func toEntityTypeKeys(entityTypes []EntityType) *EntityTypeKeys {
	m := make(map[string]int64)
	for _, et := range entityTypes {
		m[et.Name] = et.Id
	}

	return &EntityTypeKeys{
		m[RoleEntity],
		m[WorkgroupEntity],
		m[IdentityEntity],
		m[EngineEntity],
		m[ClusterEntity],
		m[ProjectEntity],
		m[DatasourceEntity],
		m[DatasetEntity],
		m[ModelEntity],
		m[ServiceEntity],
	}
}

func toClusterTypeKeys(clusterTypes []ClusterType) *ClusterTypeKeys {
	m := make(map[string]int64)
	for _, ct := range clusterTypes {
		m[ct.Name] = ct.Id
	}
	return &ClusterTypeKeys{
		m[ClusterExternal],
		m[ClusterYarn],
	}
}

type Datastore struct {
	db                *sql.DB // Singleton; doesn't actually connect until used, and is pooled internally.
	metadata          metadata
	permissions       []Permission
	permissionMap     map[int64]Permission
	Permissions       *PermissionKeys
	entityTypes       []EntityType
	entityTypeMap     map[int64]EntityType
	EntityTypes       *EntityTypeKeys
	clusterTypes      []ClusterType
	clusterTypeMap    map[int64]ClusterType
	ClusterTypes      *ClusterTypeKeys
	ViewPermissions   map[int64]int64
	ManagePermissions map[int64]int64
}

func Create(name, username, sslmode, suname, supass string) (*Datastore, error) {
	db, err := connect(username, name, sslmode)
	if err != nil {
		return nil, fmt.Errorf("Failed connecting to database %s as user %s (SSL=%s): %s\n", name, username, sslmode, err)
	}

	primed, err := isPrimed(db)
	if err != nil {
		return nil, fmt.Errorf("Failed database version check:", err)
	}

	if !primed {
		if suname == "" || supass == "" {
			return nil, fmt.Errorf("Starting Steam for the first time requires both --superuser-name and --superuser-password arguments to \"steam serve master\".")
		}

		if err := auth.ValidateUsername(suname); err != nil {
			return nil, fmt.Errorf("Invalid superuser username: %s", err)
		}

		if err := auth.ValidatePassword(supass); err != nil {
			return nil, fmt.Errorf("Invalid superuser password: %s", err)
		}

		if err := prime(db); err != nil {
			return nil, fmt.Errorf("Failed priming database: %s", err)
		}
	}

	ds, err := newDatastore(db)
	if err != nil {
		return nil, fmt.Errorf("Failed initializing from database: %s", err)
	}

	if !primed {
		passwordHash, err := auth.HashPassword(supass)
		if err != nil {
			return nil, fmt.Errorf("Failed hashing superuser password: %s", err)
		}

		if _, _, err := ds.CreateSuperuser(suname, passwordHash); err != nil {
			return nil, fmt.Errorf("Failed superuser identity setup: %s", err)
		}

		_, err = ds.Lookup(suname)
		if err != nil {
			return nil, fmt.Errorf("Failed reading superuser principal: %s", err)
		}
	}

	return ds, nil
}

func Destroy(name, username, sslmode string) error {
	db, err := connect(username, name, sslmode)
	if err != nil {
		return fmt.Errorf("Failed connecting to database %s as user %s (SSL=%s): %s\n", name, username, sslmode, err)
	}
	return truncate(db)
}

func connect(username, dbname, sslmode string) (*sql.DB, error) {

	// FIXME logging need to be handled for testing
	// log.Println("Connecting to database: user =", username, "db =", dbname, "SSL=", sslmode, "...")

	// Open connection
	db, err := sql.Open("postgres", fmt.Sprintf("user=%s dbname=%s sslmode=%s", username, dbname, sslmode))
	if err != nil {
		return nil, fmt.Errorf("Database connection failed: %s", err)
	}

	// TODO can use db.SetMaxOpenConns() and db.SetMaxIdleConns() to configure further.

	// Verify connection
	if err := db.Ping(); err != nil {
		return nil, fmt.Errorf("Database ping failed: %s", err)
	}

	return db, nil
}

// newDatastore creates a new instance of a data access object.
//
// Valid values for sslmode are:
//   disable - No SSL
//   require - Always SSL (skip verification)
//   verify-ca - Always SSL (verify that the certificate presented by the server was signed by a trusted CA)
//   verify-full - Always SSL (verify that the certification presented by the server was signed by a
//     trusted CA and the server host name matches the one in the certificate)
func newDatastore(db *sql.DB) (*Datastore, error) {

	// Read meta information

	metadata, err := readMetadata(db)
	if err != nil {
		return nil, err
	}

	version, ok := metadata["version"]
	if !ok {
		return nil, fmt.Errorf("Failed reading schema version")
	}

	// FIXME logging needs to be handled for testing
	// log.Println("Using schema version:", version)

	if err := upgrade(db, version); err != nil {
		return nil, err
	}

	permissions, err := readAllPermissions(db)
	if err != nil {
		return nil, err
	}

	permissionMap := make(map[int64]Permission)
	for _, permission := range permissions {
		permissionMap[permission.Id] = permission
	}

	permissionKeys := toPermissionKeys(permissions)

	entityTypes, err := readEntityTypes(db)
	if err != nil {
		return nil, err
	}

	entityTypeMap := make(map[int64]EntityType)
	for _, et := range entityTypes {
		entityTypeMap[et.Id] = et
	}

	entityTypeKeys := toEntityTypeKeys(entityTypes)

	clusterTypes, err := readClusterTypes(db)
	if err != nil {
		return nil, err
	}

	clusterTypeMap := make(map[int64]ClusterType)
	for _, ct := range clusterTypes {
		clusterTypeMap[ct.Id] = ct
	}

	viewPermissions := map[int64]int64{
		entityTypeKeys.Engine:     permissionKeys.ViewEngine,
		entityTypeKeys.Cluster:    permissionKeys.ViewCluster,
		entityTypeKeys.Project:    permissionKeys.ViewProject,
		entityTypeKeys.Datasource: permissionKeys.ViewDatasource,
		entityTypeKeys.Dataset:    permissionKeys.ViewDataset,
		entityTypeKeys.Model:      permissionKeys.ViewModel,
		entityTypeKeys.Service:    permissionKeys.ViewService,
		entityTypeKeys.Identity:   permissionKeys.ViewIdentity,
		entityTypeKeys.Role:       permissionKeys.ViewRole,
		entityTypeKeys.Workgroup:  permissionKeys.ViewWorkgroup,
	}

	managePermissions := map[int64]int64{
		entityTypeKeys.Engine:     permissionKeys.ManageEngine,
		entityTypeKeys.Cluster:    permissionKeys.ManageCluster,
		entityTypeKeys.Project:    permissionKeys.ManageProject,
		entityTypeKeys.Datasource: permissionKeys.ManageDatasource,
		entityTypeKeys.Dataset:    permissionKeys.ManageDataset,
		entityTypeKeys.Model:      permissionKeys.ManageModel,
		entityTypeKeys.Service:    permissionKeys.ManageService,
		entityTypeKeys.Identity:   permissionKeys.ManageIdentity,
		entityTypeKeys.Role:       permissionKeys.ManageRole,
		entityTypeKeys.Workgroup:  permissionKeys.ManageWorkgroup,
	}

	return &Datastore{
		db,
		metadata,
		permissions,
		permissionMap,
		permissionKeys,
		entityTypes,
		entityTypeMap,
		entityTypeKeys,
		clusterTypes,
		clusterTypeMap,
		toClusterTypeKeys(clusterTypes),
		viewPermissions,
		managePermissions,
	}, nil
}

func isPrimed(db *sql.DB) (bool, error) {
	row := db.QueryRow(`
		SELECT 
			count(1)
		FROM
			meta
		`)
	count, err := scanInt(row)
	if err != nil {
		return false, err
	}
	return count > 0, err
}

func prime(db *sql.DB) error {
	// FIXME logging needs to be handled for testing
	// log.Println("Priming database for first time use...")
	if err := createMetadata(db, "version", "1"); err != nil {
		return err
	}
	if err := primePermissions(db, Permissions); err != nil {
		return err
	}
	if err := primeEntityTypes(db, EntityTypes); err != nil {
		return err
	}
	if err := primeClusterTypes(db, ClusterTypes); err != nil {
		return err
	}

	return nil
}

func bulkInsert(db *sql.DB, table string, columns []string, f func(*sql.Stmt) error) error {
	tx, err := db.Begin()
	if err != nil {
		return err
	}

	stmt, err := tx.Prepare(pq.CopyIn(table, columns...))
	if err != nil {
		return err
	}

	if err := f(stmt); err != nil { // buffer
		return err
	}

	if _, err := stmt.Exec(); err != nil { // flush
		return err
	}

	if err := stmt.Close(); err != nil {
		return err
	}

	if err := tx.Commit(); err != nil {
		return err
	}

	return nil
}

func createMetadata(db *sql.DB, key, value string) error {
	_, err := db.Exec(`
		INSERT INTO
			meta
			(key, value)
		VALUES
			($1,  $2)
		`, key, value)
	return err
}

func primeEntityTypes(db *sql.DB, entityTypes []EntityType) error {
	return bulkInsert(db, "entity_type", []string{"name"}, func(stmt *sql.Stmt) error {
		for _, entityType := range entityTypes {
			_, err := stmt.Exec(entityType.Name)
			if err != nil {
				return err
			}
		}
		return nil
	})
}

func primeClusterTypes(db *sql.DB, clusterTypes []ClusterType) error {
	return bulkInsert(db, "cluster_type", []string{"name"}, func(stmt *sql.Stmt) error {
		for _, clusterType := range clusterTypes {
			_, err := stmt.Exec(clusterType.Name)
			if err != nil {
				return err
			}
		}
		return nil
	})
}

func primePermissions(db *sql.DB, permissions []Permission) error {
	return bulkInsert(db, "permission", []string{"code", "description"}, func(stmt *sql.Stmt) error {
		for _, permission := range permissions {
			_, err := stmt.Exec(permission.Code, permission.Description)
			if err != nil {
				return err
			}
		}
		return nil
	})
}

func upgrade(db *sql.DB, currentVersion string) error {
	if currentVersion == Version {
		return nil
	}

	// TODO add stepwise upgrades

	return nil
}

func truncate(db *sql.DB) error {
	// FIXME logging needs to be handled for testing
	// log.Println("Truncating database...")
	return executeTransaction(db, func(tx *sql.Tx) error {
		tables := []string{
			"history",
			"privilege",
			"role_permission",
			"identity_role",
			"identity_workgroup",
			"identity",
			"workgroup",
			"role",
			"permission",
{'fraction_non_alphanumeric': 0.1271576715369579, 'fraction_numerical': 0.009442401351754299, 'mean_word_length': 3.0406961178045515, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1021814', 'n_tokens_mistral': 10641, 'n_tokens_neox': 9959, 'n_words': 2606}