Migrate po service to new repo (#557)

* Migrate documents package to use new errors

* typeerror -> typederror

* Fix failing tests

* Migrate invoice service to new repo

* Fix fmt errors

* Migrate new repository to new errors

* Fix test

* Fix test

* Fix lint issue

* Fix error name

* Fix error name

* Migrate PO service to new repository

documents/purchaseorder/bootstrapper.go

@@ -11,8 +11,6 @@ import (
 	"github.com/centrifuge/go-centrifuge/documents"
 	"github.com/centrifuge/go-centrifuge/identity"
 	"github.com/centrifuge/go-centrifuge/p2p"
-	"github.com/centrifuge/go-centrifuge/storage"
-	"github.com/syndtr/goleveldb/leveldb"
 )
 
 // Bootstrapper implements bootstrap.Bootstrapper.
@@ -25,11 +23,6 @@ func (Bootstrapper) Bootstrap(ctx map[string]interface{}) error {
 	}
 	cfg := ctx[config.BootstrappedConfig].(config.Configuration)
 
-	ldb, ok := ctx[storage.BootstrappedLevelDB].(*leveldb.DB)
-	if !ok {
-		return errors.New("initializing LevelDB repository failed")
-	}
-
 	p2pClient, ok := ctx[p2p.BootstrappedP2PClient].(p2p.Client)
 	if !ok {
 		return fmt.Errorf("p2p client not initialised")
@@ -50,8 +43,14 @@ func (Bootstrapper) Bootstrap(ctx map[string]interface{}) error {
 		return fmt.Errorf("identity service not initialised")
 	}
 
+	repo, ok := ctx[documents.BootstrappedDocumentRepository].(documents.Repository)
+	if !ok {
+		return fmt.Errorf("document db repository not initialised")
+	}
+	repo.Register(&PurchaseOrder{})
+
 	// register service
-	srv := DefaultService(cfg, getRepository(ldb), coredocument.DefaultProcessor(idService, p2pClient, anchorRepo, cfg), anchorRepo, idService)
+	srv := DefaultService(cfg, repo, coredocument.DefaultProcessor(idService, p2pClient, anchorRepo, cfg), anchorRepo, idService)
 	err := registry.Register(documenttypes.PurchaseOrderDataTypeUrl, srv)
 	if err != nil {
 		return fmt.Errorf("failed to register purchase order service")

documents/purchaseorder/service.go

@@ -52,16 +52,17 @@ type Service interface {
 // service implements Service and handles all purchase order related persistence and validations
 // service always returns errors of type `errors.Error` or `errors.TypedError`
 type service struct {
-	repo             documents.LegacyRepository
+	config           documents.Config
+	repo             documents.Repository
 	coreDocProcessor coredocument.Processor
 	notifier         notification.Sender
 	anchorRepository anchors.AnchorRepository
 	identityService  identity.Service
 }
 
 // DefaultService returns the default implementation of the service
-func DefaultService(config config.Configuration, repo documents.LegacyRepository, processor coredocument.Processor, anchorRepository anchors.AnchorRepository, identityService identity.Service) Service {
-	return service{repo: repo, coreDocProcessor: processor, notifier: notification.NewWebhookSender(config), anchorRepository: anchorRepository, identityService: identityService}
+func DefaultService(config config.Configuration, repo documents.Repository, processor coredocument.Processor, anchorRepository anchors.AnchorRepository, identityService identity.Service) Service {
+	return service{config: config, repo: repo, coreDocProcessor: processor, notifier: notification.NewWebhookSender(config), anchorRepository: anchorRepository, identityService: identityService}
 }
 
 // DeriveFromCoreDocument takes a core document and returns a purchase order
@@ -94,8 +95,14 @@ func (s service) calculateDataRoot(old, new documents.Model, validator documents
 		return nil, errors.NewTypedError(documents.ErrDocumentInvalid, err)
 	}
 
+	// get tenant ID
+	tenantID, err := s.config.GetIdentityID()
+	if err != nil {
+		return nil, errors.NewTypedError(documents.ErrDocumentConfigTenantID, err)
+	}
+
 	// we use CurrentVersion as the id since that will be unique across multiple versions of the same document
-	err = s.repo.Create(po.CoreDocument.CurrentVersion, po)
+	err = s.repo.Create(tenantID, po.CoreDocument.CurrentVersion, po)
 	if err != nil {
 		return nil, errors.NewTypedError(documents.ErrDocumentPersistence, err)
 	}
@@ -110,14 +117,24 @@ func (s service) Create(ctx *header.ContextHeader, po documents.Model) (document
 		return nil, err
 	}
 
-	po, err = documents.AnchorDocument(ctx, po, s.coreDocProcessor, s.repo.Update)
+	po, err = documents.AnchorDocument(ctx, po, s.coreDocProcessor, s.updater)
 	if err != nil {
 		return nil, err
 	}
 
 	return po, nil
 }
 
+// updater wraps logic related to updating documents so that it can be executed as a closure
+func (s service) updater(id []byte, model documents.Model) error {
+	// get tenant ID
+	tenantID, err := s.config.GetIdentityID()
+	if err != nil {
+		return errors.NewTypedError(documents.ErrDocumentConfigTenantID, err)
+	}
+	return s.repo.Update(tenantID, id, model)
+}
+
 // Update validates, persists, and anchors a new version of purchase order
 func (s service) Update(ctx *header.ContextHeader, po documents.Model) (documents.Model, error) {
 	cd, err := po.PackCoreDocument()
@@ -135,7 +152,7 @@ func (s service) Update(ctx *header.ContextHeader, po documents.Model) (document
 		return nil, err
 	}
 
-	po, err = documents.AnchorDocument(ctx, po, s.coreDocProcessor, s.repo.Update)
+	po, err = documents.AnchorDocument(ctx, po, s.coreDocProcessor, s.updater)
 	if err != nil {
 		return nil, err
 	}
@@ -241,8 +258,12 @@ func (s service) DerivePurchaseOrderResponse(doc documents.Model) (*clientpopb.P
 }
 
 func (s service) getPurchaseOrderVersion(documentID, version []byte) (model *PurchaseOrder, err error) {
-	var doc documents.Model = new(PurchaseOrder)
-	err = s.repo.LoadByID(version, doc)
+	// get tenant ID
+	tenantID, err := s.config.GetIdentityID()
+	if err != nil {
+		return nil, errors.NewTypedError(documents.ErrDocumentConfigTenantID, err)
+	}
+	doc, err := s.repo.Get(tenantID, version)
 	if err != nil {
 		return nil, errors.NewTypedError(documents.ErrDocumentVersionNotFound, err)
 	}
@@ -351,15 +372,21 @@ func (s service) RequestDocumentSignature(contextHeader *header.ContextHeader, m
 		return nil, errors.NewTypedError(documents.ErrDocumentUnPackingCoreDocument, err)
 	}
 
+	// get tenant ID
+	tenantID, err := s.config.GetIdentityID()
+	if err != nil {
+		return nil, errors.NewTypedError(documents.ErrDocumentConfigTenantID, err)
+	}
+
 	// Logic for receiving version n (n > 1) of the document for the first time
-	if !s.repo.Exists(cd.DocumentIdentifier) && !utils.IsSameByteSlice(cd.DocumentIdentifier, cd.CurrentVersion) {
-		err = s.repo.Create(cd.DocumentIdentifier, model)
+	if !s.repo.Exists(tenantID, cd.DocumentIdentifier) && !utils.IsSameByteSlice(cd.DocumentIdentifier, cd.CurrentVersion) {
+		err = s.repo.Create(tenantID, cd.DocumentIdentifier, model)
 		if err != nil {
 			return nil, errors.NewTypedError(documents.ErrDocumentPersistence, err)
 		}
 	}
 
-	err = s.repo.Create(cd.CurrentVersion, model)
+	err = s.repo.Create(tenantID, cd.CurrentVersion, model)
 	if err != nil {
 		return nil, errors.NewTypedError(documents.ErrDocumentPersistence, err)
 	}
@@ -381,7 +408,13 @@ func (s service) ReceiveAnchoredDocument(model documents.Model, headers *p2ppb.C
 		return errors.NewTypedError(documents.ErrDocumentPackingCoreDocument, err)
 	}
 
-	err = s.repo.Update(doc.CurrentVersion, model)
+	// get tenant ID
+	tenantID, err := s.config.GetIdentityID()
+	if err != nil {
+		return errors.NewTypedError(documents.ErrDocumentConfigTenantID, err)
+	}
+
+	err = s.repo.Update(tenantID, doc.CurrentVersion, model)
 	if err != nil {
 		return errors.NewTypedError(documents.ErrDocumentPersistence, err)
 	}
@@ -403,5 +436,10 @@ func (s service) ReceiveAnchoredDocument(model documents.Model, headers *p2ppb.C
 
 // Exists checks if an purchase order exists
 func (s service) Exists(documentID []byte) bool {
-	return s.repo.Exists(documentID)
+	// get tenant ID
+	tenantID, err := s.config.GetIdentityID()
+	if err != nil {
+		return false
+	}
+	return s.repo.Exists(tenantID, documentID)
 }