POC: Cross-team recent search (#20027)

Just a quick POC to move fast :P We use a search pointer to keep track of the next row to inesrt to. For every new search we increment the pointer and do modulo 5. This means that the value will always remain between 0-4. And that way, we will always overwrite the oldest entry on every search. And while getting the results, we get all results for that user. The search parameters are json marshalled and stored as a JSON blob. This is because there is no need to search/filter them in the DB. Pending items: Tests obviously. To improve: The client needs to send the channel ids instead of channel names.
2022-05-16 13:16:11 +05:30
--- a/store/opentracinglayer/opentracinglayer.go
+++ b/store/opentracinglayer/opentracinglayer.go
@@ -5862,6 +5862,24 @@ func (s *OpenTracingLayerPostStore) GetPostsSinceForSync(options model.GetPostsS
 	return result, resultVar1, err
 }

+func (s *OpenTracingLayerPostStore) GetRecentSearchesForUser(userID string) ([]*model.SearchParams, error) {
+	origCtx := s.Root.Store.Context()
+	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "PostStore.GetRecentSearchesForUser")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	result, err := s.PostStore.GetRecentSearchesForUser(userID)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return result, err
+}
+
 func (s *OpenTracingLayerPostStore) GetRepliesForExport(parentID string) ([]*model.ReplyForExport, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "PostStore.GetRepliesForExport")
@@ -5929,6 +5947,24 @@ func (s *OpenTracingLayerPostStore) InvalidateLastPostTimeCache(channelID string

 }

+func (s *OpenTracingLayerPostStore) LogRecentSearch(userID string, searchQuery []byte, createAt int64) error {
+	origCtx := s.Root.Store.Context()
+	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "PostStore.LogRecentSearch")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	err := s.PostStore.LogRecentSearch(userID, searchQuery, createAt)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return err
+}
+
 func (s *OpenTracingLayerPostStore) Overwrite(post *model.Post) (*model.Post, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "PostStore.Overwrite")
--- a/store/retrylayer/retrylayer.go
+++ b/store/retrylayer/retrylayer.go
@@ -6637,6 +6637,27 @@ func (s *RetryLayerPostStore) GetPostsSinceForSync(options model.GetPostsSinceFo

 }

+func (s *RetryLayerPostStore) GetRecentSearchesForUser(userID string) ([]*model.SearchParams, error) {
+
+	tries := 0
+	for {
+		result, err := s.PostStore.GetRecentSearchesForUser(userID)
+		if err == nil {
+			return result, nil
+		}
+		if !isRepeatableError(err) {
+			return result, err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return result, err
+		}
+		timepkg.Sleep(100 * timepkg.Millisecond)
+	}
+
+}
+
 func (s *RetryLayerPostStore) GetRepliesForExport(parentID string) ([]*model.ReplyForExport, error) {

 	tries := 0
@@ -6706,6 +6727,27 @@ func (s *RetryLayerPostStore) InvalidateLastPostTimeCache(channelID string) {

 }

+func (s *RetryLayerPostStore) LogRecentSearch(userID string, searchQuery []byte, createAt int64) error {
+
+	tries := 0
+	for {
+		err := s.PostStore.LogRecentSearch(userID, searchQuery, createAt)
+		if err == nil {
+			return nil
+		}
+		if !isRepeatableError(err) {
+			return err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return err
+		}
+		timepkg.Sleep(100 * timepkg.Millisecond)
+	}
+
+}
+
 func (s *RetryLayerPostStore) Overwrite(post *model.Post) (*model.Post, error) {

 	tries := 0
--- a/store/sqlstore/channel_store.go
+++ b/store/sqlstore/channel_store.go
@@ -1522,11 +1522,19 @@ func (s SqlChannelStore) GetByNameIncludeDeleted(teamId string, name string, all
 }

 func (s SqlChannelStore) getByName(teamId string, name string, includeDeleted bool, allowFromCache bool) (*model.Channel, error) {
-	var query string
-	if includeDeleted {
-		query = "SELECT * FROM Channels WHERE (TeamId = ? OR TeamId = '') AND Name = ?"
-	} else {
-		query = "SELECT * FROM Channels WHERE (TeamId = ? OR TeamId = '') AND Name = ? AND DeleteAt = 0"
+	query := s.getQueryBuilder().
+		Select("*").
+		From("Channels").
+		Where(sq.Eq{"Name": name})
+
+	if !includeDeleted {
+		query = query.Where(sq.Eq{"DeleteAt": 0})
+	}
+	if teamId != "" {
+		query = query.Where(sq.Or{
+			sq.Eq{"TeamId": teamId},
+			sq.Eq{"TeamId": ""},
+		})
 	}
 	channel := model.Channel{}

@@ -1543,7 +1551,12 @@ func (s SqlChannelStore) getByName(teamId string, name string, includeDeleted bo
 		}
 	}

-	if err := s.GetReplicaX().Get(&channel, query, teamId, name); err != nil {
+	queryStr, args, err := query.ToSql()
+	if err != nil {
+		return nil, errors.Wrapf(err, "getByName_tosql")
+	}
+
+	if err := s.GetReplicaX().Get(&channel, queryStr, args...); err != nil {
 		if err == sql.ErrNoRows {
 			return nil, store.NewErrNotFound("Channel", fmt.Sprintf("TeamId=%s&Name=%s", teamId, name))
 		}
--- a/store/sqlstore/file_info_store.go
+++ b/store/sqlstore/file_info_store.go
@@ -5,6 +5,7 @@ package sqlstore

 import (
 	"database/sql"
+	"encoding/json"
 	"fmt"
 	"regexp"
 	"strconv"
@@ -507,12 +508,33 @@ func (fs SqlFileInfoStore) Search(paramsList []*model.SearchParams, userId, team
 		LeftJoin("Posts as P ON FileInfo.PostId=P.Id").
 		LeftJoin("Channels as C ON C.Id=P.ChannelId").
 		LeftJoin("ChannelMembers as CM ON C.Id=CM.ChannelId").
-		Where(sq.Or{sq.Eq{"C.TeamId": teamId}, sq.Eq{"C.TeamId": ""}}).
 		Where(sq.Eq{"FileInfo.DeleteAt": 0}).
 		OrderBy("FileInfo.CreateAt DESC").
 		Limit(100)

+	if teamId != "" {
+		query = query.Where(sq.Or{
+			sq.Eq{"C.TeamId": teamId},
+			sq.Eq{"C.TeamId": ""},
+		})
+	}
+
+	now := model.GetMillis()
 	for _, params := range paramsList {
+		if params.Modifier == model.ModifierFiles {
+			// Deliberately keeping non-alphanumeric characters to
+			// prevent surprises in UI.
+			buf, err := json.Marshal(params)
+			if err != nil {
+				return nil, err
+			}
+
+			err = fs.stores.post.LogRecentSearch(userId, buf, now)
+			if err != nil {
+				return nil, err
+			}
+		}
+
 		params.Terms = removeNonAlphaNumericUnquotedTerms(params.Terms, " ")

 		if !params.IncludeDeletedChannels {
--- a/store/sqlstore/post_store.go
+++ b/store/sqlstore/post_store.go
@@ -6,9 +6,11 @@ package sqlstore
 import (
 	"context"
 	"database/sql"
+	"encoding/json"
 	"fmt"
 	"reflect"
 	"regexp"
+	"strconv"
 	"strings"
 	"sync"

@@ -1802,11 +1804,13 @@ func (s *SqlPostStore) buildSearchPostFilterClause(teamID string, fromUsers []st
 	}

 	// Sub-query builder.
-	sb := s.getSubQueryBuilder().Select("Id").From("Users, TeamMembers").Where(
-		sq.And{
-			sq.Eq{"TeamMembers.TeamId": teamID},
-			sq.Expr("Users.Id = TeamMembers.UserId"),
-		})
+	sb := s.getSubQueryBuilder().
+		Select("Id").
+		From("Users, TeamMembers").
+		Where(sq.Expr("Users.Id = TeamMembers.UserId"))
+	if teamID != "" {
+		sb = sb.Where(sq.Eq{"TeamMembers.TeamId": teamID})
+	}
 	sb = s.buildSearchUserFilterClause(fromUsers, false, userByUsername, sb)
 	sb = s.buildSearchUserFilterClause(excludedUsers, true, userByUsername, sb)
 	subQuery, subQueryArgs, err := sb.ToSql()
@@ -2522,7 +2526,7 @@ func (s *SqlPostStore) GetDirectPostParentsForExportAfter(limit int, afterId str
 }

 //nolint:unparam
-func (s *SqlPostStore) SearchPostsForUser(paramsList []*model.SearchParams, userId, teamId string, page, perPage int) (*model.PostSearchResults, error) {
+func (s *SqlPostStore) SearchPostsForUser(paramsList []*model.SearchParams, userID, teamId string, page, perPage int) (*model.PostSearchResults, error) {
 	// Since we don't support paging for DB search, we just return nothing for later pages
 	if page > 0 {
 		return model.MakePostSearchResults(model.NewPostList(), nil), nil
@@ -2532,11 +2536,22 @@ func (s *SqlPostStore) SearchPostsForUser(paramsList []*model.SearchParams, user
 		return nil, err
 	}

-	var wg sync.WaitGroup
-
+	now := model.GetMillis()
 	pchan := make(chan store.StoreResult, len(paramsList))

+	var wg sync.WaitGroup
 	for _, params := range paramsList {
+		// Deliberately keeping non-alphanumeric characters to
+		// prevent surprises in UI.
+		buf, err := json.Marshal(params)
+		if err != nil {
+			return nil, err
+		}
+		err = s.LogRecentSearch(userID, buf, now)
+		if err != nil {
+			return nil, err
+		}
+
 		// remove any unquoted term that contains only non-alphanumeric chars
 		// ex: abcd "**" && abc     >>     abcd "**" abc
 		params.Terms = removeNonAlphaNumericUnquotedTerms(params.Terms, " ")
@@ -2545,7 +2560,7 @@ func (s *SqlPostStore) SearchPostsForUser(paramsList []*model.SearchParams, user

 		go func(params *model.SearchParams) {
 			defer wg.Done()
-			postList, err := s.search(teamId, userId, params, false, false)
+			postList, err := s.search(teamId, userID, params, false, false)
 			pchan <- store.StoreResult{Data: postList, NErr: err}
 		}(params)
 	}
@@ -2568,6 +2583,103 @@ func (s *SqlPostStore) SearchPostsForUser(paramsList []*model.SearchParams, user
 	return model.MakePostSearchResults(posts, nil), nil
 }

+const lastSearchesLimit = 5
+
+func (s *SqlPostStore) LogRecentSearch(userID string, searchQuery []byte, createAt int64) error {
+	transaction, err := s.GetMasterX().Beginx()
+	if err != nil {
+		return errors.Wrap(err, "begin_transaction")
+	}
+
+	defer finalizeTransactionX(transaction)
+
+	var lastSearchPointer int
+	var queryStr string
+	// get search_pointer
+	// We coalesce to -1 because we want to start from 0
+	if s.DriverName() == model.DatabaseDriverPostgres {
+		queryStr = `SELECT COALESCE((props->>'last_search_pointer')::integer, -1)
+			FROM Users
+			WHERE Id=?`
+	} else {
+		queryStr = `SELECT COALESCE(CAST(JSON_EXTRACT(Props, '$.last_search_pointer') as unsigned), -1)
+			FROM Users
+			WHERE Id=?`
+	}
+	err = transaction.Get(&lastSearchPointer, queryStr, userID)
+	if err != nil {
+		return errors.Wrapf(err, "failed to find last_search_pointer for user=%s", userID)
+	}
+
+	// (ptr+1)%lastSearchesLimit
+	lastSearchPointer = (lastSearchPointer + 1) % lastSearchesLimit
+
+	if s.IsBinaryParamEnabled() {
+		searchQuery = AppendBinaryFlag(searchQuery)
+	}
+
+	// insert at pointer
+	query := s.getQueryBuilder().
+		Insert("RecentSearches").
+		Columns("UserId", "SearchPointer", "Query", "CreateAt").
+		Values(userID, lastSearchPointer, searchQuery, createAt)
+
+	if s.DriverName() == model.DatabaseDriverPostgres {
+		query = query.SuffixExpr(sq.Expr("ON CONFLICT (userid, searchpointer) DO UPDATE SET Query = ?, CreateAt = ?", searchQuery, createAt))
+	} else {
+		query = query.SuffixExpr(sq.Expr("ON DUPLICATE KEY UPDATE Query = ?, CreateAt = ?", searchQuery, createAt))
+	}
+
+	queryString, args, err := query.ToSql()
+	if err != nil {
+		return errors.Wrap(err, "log_recent_search_tosql")
+	}
+
+	if _, err2 := transaction.Exec(queryString, args...); err2 != nil {
+		return errors.Wrapf(err2, "failed to upsert recent_search for user=%s", userID)
+	}
+
+	// write ptr on users prop
+	if s.DriverName() == model.DatabaseDriverPostgres {
+		_, err = transaction.Exec(`UPDATE Users
+			SET Props = jsonb_set(Props, $1, $2)
+			WHERE Id = $3`, jsonKeyPath("last_search_pointer"), jsonStringVal(strconv.Itoa(lastSearchPointer)), userID)
+	} else {
+		_, err = transaction.Exec(`UPDATE Users
+			SET Props = JSON_SET(Props, ?, ?)
+			WHERE Id = ?`, "$.last_search_pointer", strconv.Itoa(lastSearchPointer), userID)
+	}
+	if err != nil {
+		return errors.Wrapf(err, "failed to update last_search_pointer for user=%s", userID)
+	}
+
+	if err2 := transaction.Commit(); err2 != nil {
+		return errors.Wrap(err2, "commit_transaction")
+	}
+
+	return nil
+}
+
+func (s *SqlPostStore) GetRecentSearchesForUser(userID string) ([]*model.SearchParams, error) {
+	params := [][]byte{}
+	err := s.GetReplicaX().Select(&params, `SELECT query
+		FROM RecentSearches
+		WHERE UserId=?
+		ORDER BY CreateAt DESC`, userID)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to get recent searches for user=%s", userID)
+	}
+
+	res := make([]*model.SearchParams, len(params))
+	for i, param := range params {
+		err = json.Unmarshal(param, &res[i])
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed to unmarshal recent search query for user=%s", userID)
+		}
+	}
+	return res, nil
+}
+
 func (s *SqlPostStore) GetOldestEntityCreationTime() (int64, error) {
 	query := s.getQueryBuilder().Select("MIN(min_createat) min_createat").
 		Suffix(`FROM (
--- a/store/store.go
+++ b/store/store.go
@@ -366,6 +366,8 @@ type PostStore interface {
 	GetRepliesForExport(parentID string) ([]*model.ReplyForExport, error)
 	GetDirectPostParentsForExportAfter(limit int, afterID string) ([]*model.DirectPostForExport, error)
 	SearchPostsForUser(paramsList []*model.SearchParams, userID, teamID string, page, perPage int) (*model.PostSearchResults, error)
+	GetRecentSearchesForUser(userID string) ([]*model.SearchParams, error)
+	LogRecentSearch(userID string, searchQuery []byte, createAt int64) error
 	GetOldestEntityCreationTime() (int64, error)
 	HasAutoResponsePostByUserSince(options model.GetPostsSinceOptions, userId string) (bool, error)
 	GetPostsSinceForSync(options model.GetPostsSinceForSyncOptions, cursor model.GetPostsSinceForSyncCursor, limit int) ([]*model.Post, model.GetPostsSinceForSyncCursor, error)
--- a/store/storetest/mocks/PostStore.go
+++ b/store/storetest/mocks/PostStore.go
@@ -610,6 +610,29 @@ func (_m *PostStore) GetPostsSinceForSync(options model.GetPostsSinceForSyncOpti
 	return r0, r1, r2
 }

+// GetRecentSearchesForUser provides a mock function with given fields: userID
+func (_m *PostStore) GetRecentSearchesForUser(userID string) ([]*model.SearchParams, error) {
+	ret := _m.Called(userID)
+
+	var r0 []*model.SearchParams
+	if rf, ok := ret.Get(0).(func(string) []*model.SearchParams); ok {
+		r0 = rf(userID)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*model.SearchParams)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(string) error); ok {
+		r1 = rf(userID)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetRepliesForExport provides a mock function with given fields: parentID
 func (_m *PostStore) GetRepliesForExport(parentID string) ([]*model.ReplyForExport, error) {
 	ret := _m.Called(parentID)
@@ -682,6 +705,20 @@ func (_m *PostStore) InvalidateLastPostTimeCache(channelID string) {
 	_m.Called(channelID)
 }

+// LogRecentSearch provides a mock function with given fields: userID, searchQuery, createAt
+func (_m *PostStore) LogRecentSearch(userID string, searchQuery []byte, createAt int64) error {
+	ret := _m.Called(userID, searchQuery, createAt)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(string, []byte, int64) error); ok {
+		r0 = rf(userID, searchQuery, createAt)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
 // Overwrite provides a mock function with given fields: post
 func (_m *PostStore) Overwrite(post *model.Post) (*model.Post, error) {
 	ret := _m.Called(post)
--- a/store/timerlayer/timerlayer.go
+++ b/store/timerlayer/timerlayer.go
@@ -5306,6 +5306,22 @@ func (s *TimerLayerPostStore) GetPostsSinceForSync(options model.GetPostsSinceFo
 	return result, resultVar1, err
 }

+func (s *TimerLayerPostStore) GetRecentSearchesForUser(userID string) ([]*model.SearchParams, error) {
+	start := timemodule.Now()
+
+	result, err := s.PostStore.GetRecentSearchesForUser(userID)
+
+	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("PostStore.GetRecentSearchesForUser", success, elapsed)
+	}
+	return result, err
+}
+
 func (s *TimerLayerPostStore) GetRepliesForExport(parentID string) ([]*model.ReplyForExport, error) {
 	start := timemodule.Now()

@@ -5369,6 +5385,22 @@ func (s *TimerLayerPostStore) InvalidateLastPostTimeCache(channelID string) {
 	}
 }

+func (s *TimerLayerPostStore) LogRecentSearch(userID string, searchQuery []byte, createAt int64) error {
+	start := timemodule.Now()
+
+	err := s.PostStore.LogRecentSearch(userID, searchQuery, createAt)
+
+	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("PostStore.LogRecentSearch", success, elapsed)
+	}
+	return err
+}
+
 func (s *TimerLayerPostStore) Overwrite(post *model.Post) (*model.Post, error) {
 	start := timemodule.Now()