Integrate indexer / remove bleve [WIP]

accounts/pkg/indexer/indexer.go

@@ -6,6 +6,8 @@ import (
 	"github.com/owncloud/ocis/accounts/pkg/config"
 	"github.com/owncloud/ocis/accounts/pkg/indexer/errors"
 	"github.com/owncloud/ocis/accounts/pkg/indexer/index"
+	_ "github.com/owncloud/ocis/accounts/pkg/indexer/index/cs3"
+	_ "github.com/owncloud/ocis/accounts/pkg/indexer/index/disk"
 	"github.com/owncloud/ocis/accounts/pkg/indexer/option"
 	"github.com/owncloud/ocis/accounts/pkg/indexer/registry"
 	"path"

accounts/pkg/service/v0/accounts.go

@@ -3,19 +3,18 @@ package service
 import (
 	"context"
 	"fmt"
+	"github.com/CiscoM31/godata"
 	"path/filepath"
 	"regexp"
 	"sync"
 	"time"
 
-	"github.com/CiscoM31/godata"
 	"github.com/blevesearch/bleve"
 	"github.com/gofrs/uuid"
 	"github.com/golang/protobuf/ptypes/empty"
 	fieldmask_utils "github.com/mennanov/fieldmask-utils"
 	merrors "github.com/micro/go-micro/v2/errors"
 	"github.com/owncloud/ocis/accounts/pkg/proto/v0"
-	"github.com/owncloud/ocis/accounts/pkg/provider"
 	"github.com/owncloud/ocis/accounts/pkg/storage"
 	"github.com/owncloud/ocis/ocis-pkg/roles"
 	settings "github.com/owncloud/ocis/settings/pkg/proto/v0"
@@ -36,15 +35,14 @@ import (
 var accLock sync.Mutex
 
 func (s Service) indexAccount(id string) error {
-	a := &proto.BleveAccount{
-		BleveType: "account",
-	}
-	if err := s.repo.LoadAccount(context.Background(), id, &a.Account); err != nil {
+	a := &proto.Account{}
+
+	if err := s.repo.LoadAccount(context.Background(), id, a); err != nil {
 		s.log.Error().Err(err).Str("account", id).Msg("could not load account")
 		return err
 	}
 	s.log.Debug().Interface("account", a).Msg("found account")
-	if err := s.index.Index(a.Id, a); err != nil {
+	if err := s.index.Add(a); err != nil {
 		s.log.Error().Err(err).Interface("account", a).Msg("could not index account")
 		return err
 	}
@@ -104,15 +102,12 @@ func (s Service) hasAccountManagementPermissions(ctx context.Context) bool {
 // serviceUserToIndex temporarily adds a service user to the index, which is supposed to be removed before the lock on the handler function is released
 func (s Service) serviceUserToIndex() (teardownServiceUser func()) {
 	if s.Config.ServiceUser.Username != "" && s.Config.ServiceUser.UUID != "" {
-		err := s.index.Index(s.Config.ServiceUser.UUID, &proto.BleveAccount{
-			BleveType: "account",
-			Account:   s.getInMemoryServiceUser(),
-		})
+		err := s.index.Add(s.getInMemoryServiceUser())
 		if err != nil {
 			s.log.Logger.Err(err).Msg("service user was configured but failed to be added to the index")
 		} else {
 			return func() {
-				_ = s.index.Delete(s.Config.ServiceUser.UUID)
+				_ = s.index.Delete(s.getInMemoryServiceUser())
 			}
 		}
 	}
@@ -159,8 +154,9 @@ func (s Service) ListAccounts(ctx context.Context, in *proto.ListAccountsRequest
 	tq := bleve.NewTermQuery("account")
 	tq.SetField("bleve_type")
 
-	query := bleve.NewConjunctionQuery(tq)
-
+	/*
+		query := bleve.NewConjunctionQuery(tq)
+	*/
 	if in.Query != "" {
 		// parse the query like an odata filter
 		var q *godata.GoDataFilterQuery
@@ -169,65 +165,76 @@ func (s Service) ListAccounts(ctx context.Context, in *proto.ListAccountsRequest
 			return merrors.InternalServerError(s.id, "could not parse query: %v", err.Error())
 		}
 
-		// convert to bleve query
-		bq, err := provider.BuildBleveQuery(q)
+		fmt.Print(q)
+	}
+	/*
+
+
+			// convert to bleve query
+			bq, err := provider.BuildBleveQuery(q)
+			if err != nil {
+				s.log.Error().Err(err).Msg("could not build bleve query")
+				return merrors.InternalServerError(s.id, "could not build bleve query: %v", err.Error())
+			}
+			query.AddQuery(bq)
+		}
+
+		s.log.Debug().Interface("query", query).Msg("using query")
+
+		searchRequest := bleve.NewSearchRequest(query)
+		var searchResult *bleve.SearchResult
+		searchResult, err = s.index.Search(searchRequest)
+
+
 		if err != nil {
-			s.log.Error().Err(err).Msg("could not build bleve query")
-			return merrors.InternalServerError(s.id, "could not build bleve query: %v", err.Error())
-		}
-		query.AddQuery(bq)
-	}
-
-	s.log.Debug().Interface("query", query).Msg("using query")
-
-	searchRequest := bleve.NewSearchRequest(query)
-	var searchResult *bleve.SearchResult
-	searchResult, err = s.index.Search(searchRequest)
-	if err != nil {
-		s.log.Error().Err(err).Msg("could not execute bleve search")
-		return merrors.InternalServerError(s.id, "could not execute bleve search: %v", err.Error())
-	}
-
-	s.log.Debug().Interface("result", searchResult).Msg("result")
-
-	out.Accounts = make([]*proto.Account, 0)
-
-	for _, hit := range searchResult.Hits {
-		a := &proto.Account{}
-		if hit.ID == s.Config.ServiceUser.UUID {
-			acc := s.getInMemoryServiceUser()
-			a = &acc
-		} else if err = s.repo.LoadAccount(ctx, hit.ID, a); err != nil {
-			s.log.Error().Err(err).Str("account", hit.ID).Msg("could not load account, skipping")
-			continue
-		}
-		var currentHash string
-		if a.PasswordProfile != nil {
-			currentHash = a.PasswordProfile.Password
+			s.log.Error().Err(err).Msg("could not execute bleve search")
+			return merrors.InternalServerError(s.id, "could not execute bleve search: %v", err.Error())
 		}
 
-		s.debugLogAccount(a).Msg("found account")
+		s.log.Debug().Interface("result", searchResult).Msg("result")
 
-		if password != "" {
-			if a.PasswordProfile == nil {
-				s.debugLogAccount(a).Msg("no password profile")
-				return merrors.Unauthorized(s.id, "invalid password")
+		out.Accounts = make([]*proto.Account, 0)
+
+		for _, hit := range searchResult.Hits {
+			a := &proto.Account{}
+			if hit.ID == s.Config.ServiceUser.UUID {
+				acc := s.getInMemoryServiceUser()
+				a = &acc
+			} else if err = s.repo.LoadAccount(ctx, hit.ID, a); err != nil {
+				s.log.Error().Err(err).Str("account", hit.ID).Msg("could not load account, skipping")
+				continue
 			}
-			if !s.passwordIsValid(currentHash, password) {
-				return merrors.Unauthorized(s.id, "invalid password")
+			var currentHash string
+			if a.PasswordProfile != nil {
+				currentHash = a.PasswordProfile.Password
 			}
-		}
-		// TODO add groups if requested
-		// if in.FieldMask ...
-		s.expandMemberOf(a)
 
-		// remove password before returning
-		if a.PasswordProfile != nil {
-			a.PasswordProfile.Password = ""
+			s.debugLogAccount(a).Msg("found account")
+
+			if password != "" {
+				if a.PasswordProfile == nil {
+					s.debugLogAccount(a).Msg("no password profile")
+					return merrors.Unauthorized(s.id, "invalid password")
+				}
+				if !s.passwordIsValid(currentHash, password) {
+					return merrors.Unauthorized(s.id, "invalid password")
+				}
+			}
+
+
+			// TODO add groups if requested
+			// if in.FieldMask ...
+			s.expandMemberOf(a)
+
+			// remove password before returning
+			if a.PasswordProfile != nil {
+				a.PasswordProfile.Password = ""
+			}
+
+			out.Accounts = append(out.Accounts, a)
 		}
 
-		out.Accounts = append(out.Accounts, a)
-	}
+	*/
 
 	return
 }
@@ -319,7 +326,7 @@ func (s Service) CreateAccount(ctx context.Context, in *proto.CreateAccountReque
 		s.debugLogAccount(acc).Msg("could not persist new account")
 		return merrors.InternalServerError(s.id, "could not persist new account: %v", err.Error())
 	}
-	if err = s.indexAccount(acc.Id); err != nil {
+	if err = s.index.Add(acc); err != nil {
 		return merrors.InternalServerError(s.id, "could not index new account: %v", err.Error())
 	}
 	s.log.Debug().Interface("account", acc).Msg("account after indexing")
@@ -434,12 +441,19 @@ func (s Service) UpdateAccount(ctx context.Context, in *proto.UpdateAccountReque
 		out.ExternalUserStateChangeDateTime = tsnow
 	}
 
+	// We need to reload the old account state to be able to compute the update
+	old := &proto.Account{}
+	if err = s.repo.LoadAccount(ctx, id, old); err != nil {
+		s.log.Error().Err(err).Str("id", out.Id).Msg("could not load old account representation during update, maybe the account got deleted meanwhile?")
+		return merrors.InternalServerError(s.id, "could not load current account for update: %v", err.Error())
+	}
+
 	if err = s.repo.WriteAccount(ctx, out); err != nil {
 		s.log.Error().Err(err).Str("id", out.Id).Msg("could not persist updated account")
 		return merrors.InternalServerError(s.id, "could not persist updated account: %v", err.Error())
 	}
 
-	if err = s.indexAccount(id); err != nil {
+	if err = s.index.Update(old, out); err != nil {
 		s.log.Error().Err(err).Str("id", id).Str("path", path).Msg("could not index new account")
 		return merrors.InternalServerError(s.id, "could not index updated account: %v", err.Error())
 	}

accounts/pkg/service/v0/groups.go

@@ -14,22 +14,6 @@ import (
 	"github.com/owncloud/ocis/accounts/pkg/provider"
 )
 
-func (s Service) indexGroup(id string) error {
-	g := &proto.BleveGroup{
-		BleveType: "group",
-	}
-	if err := s.repo.LoadGroup(context.Background(), id, &g.Group); err != nil {
-		s.log.Error().Err(err).Str("group", id).Msg("could not load group")
-		return err
-	}
-	s.log.Debug().Interface("group", g).Msg("found group")
-	if err := s.index.Index(g.Id, g); err != nil {
-		s.log.Error().Err(err).Interface("group", g).Msg("could not index group")
-		return err
-	}
-	return nil
-}
-
 func (s Service) expandMembers(g *proto.Group) {
 	if g == nil {
 		return
@@ -93,9 +77,10 @@ func (s Service) ListGroups(c context.Context, in *proto.ListGroupsRequest, out
 
 	s.log.Debug().Interface("query", query).Msg("using query")
 
-	searchRequest := bleve.NewSearchRequest(query)
+	//searchRequest := bleve.NewSearchRequest(query)
 	var searchResult *bleve.SearchResult
-	searchResult, err = s.index.Search(searchRequest)
+	//searchResult, err = s.index.Search(searchRequest)
+	searchResult = &bleve.SearchResult{}
 	if err != nil {
 		s.log.Error().Err(err).Msg("could not execute bleve search")
 		return merrors.InternalServerError(s.id, "could not execute bleve search: %v", err.Error())
@@ -150,7 +135,7 @@ func (s Service) GetGroup(c context.Context, in *proto.GetGroupRequest, out *pro
 
 // CreateGroup implements the GroupsServiceHandler interface
 func (s Service) CreateGroup(c context.Context, in *proto.CreateGroupRequest, out *proto.Group) (err error) {
-	var id string
+	var _ string
 	if in.Group == nil {
 		return merrors.BadRequest(s.id, "account missing")
 	}
@@ -158,7 +143,9 @@ func (s Service) CreateGroup(c context.Context, in *proto.CreateGroupRequest, ou
 		in.Group.Id = uuid.Must(uuid.NewV4()).String()
 	}
 
-	if id, err = cleanupID(in.Group.Id); err != nil {
+	_ = in.Group.Id
+
+	if _, err = cleanupID(in.Group.Id); err != nil {
 		return merrors.InternalServerError(s.id, "could not clean up account id: %v", err.Error())
 	}
 
@@ -170,7 +157,7 @@ func (s Service) CreateGroup(c context.Context, in *proto.CreateGroupRequest, ou
 		return merrors.InternalServerError(s.id, "could not persist new group: %v", err.Error())
 	}
 
-	if err = s.indexGroup(id); err != nil {
+	if err = s.index.Add(in.Group); err != nil {
 		return merrors.InternalServerError(s.id, "could not index new group: %v", err.Error())
 	}
 

accounts/pkg/service/v0/service.go

@@ -3,21 +3,15 @@ package service
 import (
 	"context"
 	"errors"
+	"github.com/owncloud/ocis/accounts/pkg/indexer"
 	"github.com/owncloud/ocis/accounts/pkg/storage"
-	"os"
 	"path/filepath"
 	"strings"
 	"time"
 
-	"github.com/blevesearch/bleve"
-	"github.com/blevesearch/bleve/analysis/analyzer/custom"
-	"github.com/blevesearch/bleve/analysis/analyzer/keyword"
-	"github.com/blevesearch/bleve/analysis/analyzer/simple"
-	"github.com/blevesearch/bleve/analysis/analyzer/standard"
-	"github.com/blevesearch/bleve/analysis/token/lowercase"
-	"github.com/blevesearch/bleve/analysis/tokenizer/unicode"
 	mclient "github.com/micro/go-micro/v2/client"
 	"github.com/owncloud/ocis/accounts/pkg/config"
+	idxerrs "github.com/owncloud/ocis/accounts/pkg/indexer/errors"
 	"github.com/owncloud/ocis/accounts/pkg/proto/v0"
 	"github.com/owncloud/ocis/ocis-pkg/log"
 	"github.com/owncloud/ocis/ocis-pkg/roles"
@@ -72,6 +66,41 @@ func New(opts ...Option) (s *Service, err error) {
 	return
 }
 
+func (s Service) buildIndex() (*indexer.Indexer, error) {
+	s.Config.Repo.CS3.JWTSecret = "Pive-Fumkiu4"
+	idx := indexer.CreateIndexer(s.Config)
+
+	// Accounts
+
+	if err := idx.AddIndex(&proto.Account{}, "DisplayName", "Id", "accounts", "unique"); err != nil {
+		return nil, err
+	}
+	if err := idx.AddIndex(&proto.Account{}, "Mail", "Id", "accounts", "unique"); err != nil {
+		return nil, err
+	}
+
+	if err := idx.AddIndex(&proto.Account{}, "OnPremisesSamAccountName", "Id", "accounts", "non_unique"); err != nil {
+		return nil, err
+	}
+
+	if err := idx.AddIndex(&proto.Account{}, "PreferredName", "Id", "accounts", "unique"); err != nil {
+		return nil, err
+	}
+
+	// Groups
+	if err := idx.AddIndex(&proto.Group{}, "OnPremisesSamAccountName", "Id", "groups", "unique"); err != nil {
+		return nil, err
+	}
+
+	if err := idx.AddIndex(&proto.Group{}, "DisplayName", "Id", "accounts", "non_unique"); err != nil {
+		return nil, err
+	}
+
+	return idx, nil
+
+}
+
+/*
 func (s Service) buildIndex() (index bleve.Index, err error) {
 	indexDir := filepath.Join(s.Config.Server.AccountsDataPath, "index.bleve")
 	if index, err = bleve.Open(indexDir); err != nil {
@@ -160,6 +189,8 @@ func (s Service) buildIndex() (index bleve.Index, err error) {
 	return
 }
 
+*/
+
 func (s Service) createDefaultAccounts() (err error) {
 	accounts := []proto.Account{
 		{
@@ -276,7 +307,11 @@ func (s Service) createDefaultAccounts() (err error) {
 		}
 
 		if err := s.indexAccount(accounts[i].Id); err != nil {
-			return err
+			if idxerrs.IsAlreadyExistsErr(err) {
+				continue
+			} else {
+				return err
+			}
 		}
 	}
 
@@ -337,8 +372,12 @@ func (s Service) createDefaultGroups() (err error) {
 			return err
 		}
 
-		if err := s.indexGroup(groups[i].Id); err != nil {
-			return err
+		if err := s.index.Add(&groups[i]); err != nil {
+			if idxerrs.IsAlreadyExistsErr(err) {
+				continue
+			} else {
+				return err
+			}
 		}
 	}
 	return nil
@@ -374,7 +413,7 @@ type Service struct {
 	id          string
 	log         log.Logger
 	Config      *config.Config
-	index       bleve.Index
+	index       *indexer.Indexer
 	RoleService settings.RoleService
 	RoleManager *roles.Manager
 	repo        storage.Repo