Merge pull request juju#16613 from SimonRichardson/objectstore-state

juju#16613

Implementation of object store domain state layer. The concept of
the state layer ensures that you can have multiple paths pointing
to the same hash. As long as the metadata hash and size are the same
on insertion then you can add them.

As we're going to be storing the hash as the file path for some of
the backend, they need to be unique. It's possible we could use
an alias instead of the hash to make it readable, but we can do that
later if required.

I had to change the schema to push the size into the metadata as it's
possible, but unlikely that a size could be different for a hash. In
that scenario, the hash function is broken and we can't accept the
metadata.

## Checklist

- [x] Code style: imports ordered, good names, simple structure, etc
- [x] Comments saying why design decisions were made
- [x] Go unit tests, with comments saying what you're testing

## QA steps

```sh
$ TEST_PACKAGES="./domain/objectstore/..." make run-go-tests
```

## Links


**Jira card:** JUJU-4971

core/objectstore/metadata.go

@@ -0,0 +1,14 @@
+// Copyright 2023 Canonical Ltd.
+// Licensed under the AGPLv3, see LICENCE file for details.
+
+package objectstore
+
+// Metadata represents the metadata for an object.
+type Metadata struct {
+	// Hash is the hash of the object.
+	Hash string
+	// Path is the path to the object.
+	Path string
+	// Size is the size of the object.
+	Size int64
+}

domain/errors.go

@@ -4,6 +4,8 @@
 package domain
 
 import (
+	"fmt"
+
 	"github.com/juju/errors"
 
 	"github.com/juju/juju/internal/database"
@@ -21,10 +23,20 @@ const (
 func CoerceError(err error) error {
 	cause := errors.Cause(err)
 	if database.IsErrConstraintUnique(cause) {
-		return errors.Wrap(err, ErrDuplicate)
+		return fmt.Errorf("%w%w", maskError{error: err}, ErrDuplicate)
 	}
 	if database.IsErrNotFound(cause) {
-		return errors.Wrap(err, ErrNoRecord)
+		return fmt.Errorf("%w%w", maskError{error: err}, ErrNoRecord)
 	}
 	return errors.Trace(err)
 }
+
+// maskError is used to mask the error message, yet still allow the
+// error to be identified.
+type maskError struct {
+	error
+}
+
+func (e maskError) Error() string {
+	return ""
+}

domain/externalcontroller/state/state.go

@@ -13,21 +13,24 @@ import (
 	"github.com/juju/utils/v3"
 
 	"github.com/juju/juju/core/crossmodel"
-	coreDB "github.com/juju/juju/core/database"
+	coredatabase "github.com/juju/juju/core/database"
 	"github.com/juju/juju/domain"
 	"github.com/juju/juju/internal/database"
 )
 
+// State implements the domain external controller state.
 type State struct {
 	*domain.StateBase
 }
 
-func NewState(factory coreDB.TxnRunnerFactory) *State {
+// NewState returns a new State instance.
+func NewState(factory coredatabase.TxnRunnerFactory) *State {
 	return &State{
 		StateBase: domain.NewStateBase(factory),
 	}
 }
 
+// Controller returns the external controller with the given UUID.
 func (st *State) Controller(
 	ctx context.Context,
 	controllerUUID string,
@@ -68,6 +71,8 @@ WHERE  ctrl.uuid = $M.id`
 	return &rows.ToControllerInfo()[0], nil
 }
 
+// ControllersForModels returns the external controllers for the given model
+// UUIDs. If no model UUIDs are provided, then no controllers are returned.
 func (st *State) ControllersForModels(ctx context.Context, modelUUIDs ...string) ([]crossmodel.ControllerInfo, error) {
 	db, err := st.DB()
 	if err != nil {
@@ -119,6 +124,7 @@ WHERE  ctrl.uuid = (
 	return resultControllerInfos.ToControllerInfo(), nil
 }
 
+// UpdateExternalController updates the external controller information.
 func (st *State) UpdateExternalController(
 	ctx context.Context,
 	ci crossmodel.ControllerInfo,
@@ -214,6 +220,8 @@ VALUES (?, ?)
 	return nil
 }
 
+// ModelsForController returns the model UUIDs associated with the given
+// controller UUID.
 func (st *State) ModelsForController(
 	ctx context.Context,
 	controllerUUID string,

domain/externalcontroller/state/state_test.go

@@ -388,21 +388,14 @@ func (s *stateSuite) TestControllersForModels(c *gc.C) {
 			ModelUUIDs:    []string{"model2", "model3"},
 		},
 	}
-	// Sort the returning controlelrs which are not order-guaranteed before
+	// Sort the returning controllers which are not order-guaranteed before
 	// deep equals assert
 	sort.Slice(controllers, func(i, j int) bool { return controllers[i].ControllerTag.Id() < controllers[j].ControllerTag.Id() })
 	// Also sort addresses.
 	sort.Slice(controllers[0].Addrs, func(i, j int) bool { return controllers[0].Addrs[i] < controllers[0].Addrs[j] })
 	// Also sort models.
 	sort.Slice(controllers[1].ModelUUIDs, func(i, j int) bool { return controllers[1].ModelUUIDs[i] < controllers[1].ModelUUIDs[j] })
 	c.Assert(controllers, gc.DeepEquals, expectedControllers)
-	// c.Assert(controllers, jc.SameContents, expectedControllers)
-	// c.Assert(controllers, jc.Contains, expectedControllers[0])
-	// c.Assert(controllers, jc.Contains, expectedControllers[1])
-	// c.Assert(controllers[0].Addrs, jc.SameContents, []string{"192.168.1.1", "10.0.0.1", "10.0.0.2"})
-	// c.Assert(controllers[0].ModelUUIDs, jc.SameContents, []string{"model1"})
-	// c.Assert(controllers[1].Addrs, jc.SameContents, []string{"10.0.0.1"})
-	// c.Assert(controllers[1].ModelUUIDs, jc.SameContents, []string{"model2", "model3"})
 }
 
 func (s *stateSuite) TestControllersForModelsOneSingleModel(c *gc.C) {
@@ -437,3 +430,35 @@ func (s *stateSuite) TestControllersForModelsOneSingleModel(c *gc.C) {
 	c.Assert(controllers[0].Addrs, jc.SameContents, []string{"192.168.1.1", "10.0.0.1", "10.0.0.2"})
 	c.Assert(controllers[0].ModelUUIDs, jc.SameContents, []string{"model1", "model2", "model3"})
 }
+
+func (s *stateSuite) TestControllersForModelsWithoutModelUUIDs(c *gc.C) {
+	st := NewState(s.TxnRunnerFactory())
+	db := s.DB()
+
+	// Insert an external controller with one model.
+	_, err := db.Exec(`INSERT INTO external_controller VALUES
+("ctrl1", NULL, "test-cert1")`)
+	c.Assert(err, jc.ErrorIsNil)
+	_, err = db.Exec(`INSERT INTO external_controller_address VALUES
+("addr1", "ctrl1", "192.168.1.1")`)
+	c.Assert(err, jc.ErrorIsNil)
+	_, err = db.Exec(`INSERT INTO external_controller_address VALUES
+("addr2", "ctrl1", "10.0.0.1")`)
+	c.Assert(err, jc.ErrorIsNil)
+	_, err = db.Exec(`INSERT INTO external_controller_address VALUES
+("addr3", "ctrl1", "10.0.0.2")`)
+	c.Assert(err, jc.ErrorIsNil)
+	_, err = db.Exec(`INSERT INTO external_model VALUES
+("model1", "ctrl1")`)
+	c.Assert(err, jc.ErrorIsNil)
+	_, err = db.Exec(`INSERT INTO external_model VALUES
+("model2", "ctrl1")`)
+	c.Assert(err, jc.ErrorIsNil)
+	_, err = db.Exec(`INSERT INTO external_model VALUES
+("model3", "ctrl1")`)
+	c.Assert(err, jc.ErrorIsNil)
+
+	controllers, err := st.ControllersForModels(ctx.Background())
+	c.Assert(err, jc.ErrorIsNil)
+	c.Assert(controllers, gc.HasLen, 0)
+}

domain/objectstore/errors.go

@@ -0,0 +1,11 @@
+// Copyright 2023 Canonical Ltd.
+// Licensed under the AGPLv3, see LICENCE file for details.
+
+package objectstore
+
+import "github.com/juju/errors"
+
+// ErrHashAndSizeAlreadyExists is returned when a hash already exists, but
+// the associated size is different. This should never happen, it means that
+// there is a collision in the hash function.
+const ErrHashAndSizeAlreadyExists = errors.ConstError("hash exists for different file size")

domain/objectstore/package_test.go

@@ -0,0 +1,14 @@
+// Copyright 2023 Canonical Ltd.
+// Licensed under the AGPLv3, see LICENCE file for details.
+
+package objectstore
+
+import (
+	"testing"
+
+	gc "gopkg.in/check.v1"
+)
+
+func TestPackage(t *testing.T) {
+	gc.TestingT(t)
+}

domain/objectstore/service/package_test.go

@@ -0,0 +1,16 @@
+// Copyright 2023 Canonical Ltd.
+// Licensed under the AGPLv3, see LICENCE file for details.
+
+package service
+
+import (
+	"testing"
+
+	gc "gopkg.in/check.v1"
+)
+
+//go:generate go run go.uber.org/mock/mockgen -package service -destination state_mock_test.go github.com/juju/juju/domain/objectstore/service State,WatcherFactory
+
+func TestPackage(t *testing.T) {
+	gc.TestingT(t)
+}

domain/objectstore/service/service.go

@@ -0,0 +1,103 @@
+// Copyright 2023 Canonical Ltd.
+// Licensed under the AGPLv3, see LICENCE file for details.
+
+package service
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/juju/utils/v3"
+
+	"github.com/juju/juju/core/changestream"
+	coreobjectstore "github.com/juju/juju/core/objectstore"
+	"github.com/juju/juju/core/watcher"
+	"github.com/juju/juju/domain"
+	"github.com/juju/juju/domain/objectstore"
+)
+
+// State describes retrieval and persistence methods for the coreobjectstore.
+type State interface {
+	// GetMetadata returns the persistence metadata for the specified path.
+	GetMetadata(ctx context.Context, path string) (objectstore.Metadata, error)
+	// PutMetadata adds a new specified path for the persistence metadata.
+	PutMetadata(ctx context.Context, metadata objectstore.Metadata) error
+	// RemoveMetadata removes the specified path for the persistence metadata.
+	RemoveMetadata(ctx context.Context, path string) error
+	// InitialWatchStatement returns the table and the initial watch statement
+	// for the persistence metadata.
+	InitialWatchStatement() (string, string)
+}
+
+// WatcherFactory describes methods for creating watchers.
+type WatcherFactory interface {
+	// NewNamespaceWatcher returns a new namespace watcher
+	// for events based on the input change mask.
+	NewNamespaceWatcher(string, changestream.ChangeType, string) (watcher.StringsWatcher, error)
+}
+
+// Service provides the API for working with the coreobjectstore.
+type Service struct {
+	st             State
+	watcherFactory WatcherFactory
+}
+
+// NewService returns a new service reference wrapping the input state.
+func NewService(st State, watcherFactory WatcherFactory) *Service {
+	return &Service{
+		st:             st,
+		watcherFactory: watcherFactory,
+	}
+}
+
+// GetMetadata returns the persistence metadata for the specified path.
+func (s *Service) GetMetadata(ctx context.Context, path string) (coreobjectstore.Metadata, error) {
+	metadata, err := s.st.GetMetadata(ctx, path)
+	if err != nil {
+		return coreobjectstore.Metadata{}, fmt.Errorf("retrieving metadata %s: %w", path, domain.CoerceError(err))
+	}
+	return coreobjectstore.Metadata{
+		Path: metadata.Path,
+		Hash: metadata.Hash,
+		Size: metadata.Size,
+	}, nil
+}
+
+// PutMetadata adds a new specified path for the persistence metadata.
+func (s *Service) PutMetadata(ctx context.Context, metadata coreobjectstore.Metadata) error {
+	uuid, err := utils.NewUUID()
+	if err != nil {
+		return err
+	}
+
+	err = s.st.PutMetadata(ctx, objectstore.Metadata{
+		UUID: uuid.String(),
+		Hash: metadata.Hash,
+		Path: metadata.Path,
+		Size: metadata.Size,
+	})
+	if err != nil {
+		return fmt.Errorf("adding path %s: %w", metadata.Path, err)
+	}
+	return nil
+}
+
+// RemoveMetadata removes the specified path for the persistence metadata.
+func (s *Service) RemoveMetadata(ctx context.Context, path string) error {
+	err := s.st.RemoveMetadata(ctx, path)
+	if err != nil {
+		return fmt.Errorf("removing path %s: %w", path, err)
+	}
+	return nil
+}
+
+// Watch returns a watcher that emits the path changes that either have been
+// added or removed.
+func (s *Service) Watch() (watcher.StringsWatcher, error) {
+	table, stmt := s.st.InitialWatchStatement()
+	return s.watcherFactory.NewNamespaceWatcher(
+		table,
+		changestream.All,
+		stmt,
+	)
+}