Full text search

Author: vdimir

backend/app/cmd/server.go

@@ -18,6 +18,7 @@ import (
 	"github.com/go-pkgz/lcw/eventbus"
 	log "github.com/go-pkgz/lgr"
 	ntf "github.com/go-pkgz/notify"
+	"github.com/go-pkgz/syncs"
 	"github.com/golang-jwt/jwt"
 	"github.com/kyokomi/emoji/v2"
 	bolt "go.etcd.io/bbolt"
@@ -38,6 +39,7 @@ import (
 	"github.com/umputun/remark42/backend/app/store/admin"
 	"github.com/umputun/remark42/backend/app/store/engine"
 	"github.com/umputun/remark42/backend/app/store/image"
+	"github.com/umputun/remark42/backend/app/store/search"
 	"github.com/umputun/remark42/backend/app/store/service"
 	"github.com/umputun/remark42/backend/app/templates"
 )
@@ -57,6 +59,7 @@ type ServerCommand struct {
 	Image      ImageGroup      `group:"image" namespace:"image" env-namespace:"IMAGE"`
 	SSL        SSLGroup        `group:"ssl" namespace:"ssl" env-namespace:"SSL"`
 	ImageProxy ImageProxyGroup `group:"image-proxy" namespace:"image-proxy" env-namespace:"IMAGE_PROXY"`
+	Search     SearchGroup     `group:"search" namespace:"search" env-namespace:"SEARCH"`
 
 	Sites            []string      `long:"site" env:"SITE" default:"remark" description:"site names" env-delim:","`
 	AnonymousVote    bool          `long:"anon-vote" env:"ANON_VOTE" description:"enable anonymous votes (works only with VOTES_IP enabled)"`
@@ -280,6 +283,13 @@ type AdminRPCGroup struct {
 	SecretPerSite bool `long:"secret_per_site" env:"SECRET_PER_SITE" description:"enable JWT secret retrieval per aud, which is site_id in this case"`
 }
 
+// SearchGroup defines options group for search engine
+type SearchGroup struct {
+	Enable    bool   `long:"enable" env:"ENABLE" description:"enable search engine"`
+	IndexPath string `long:"index_path" env:"INDEX_PATH" description:"search index location" default:"./var/search_index"`
+	Analyzer  string `long:"analyzer" env:"ANALYZER" description:"text analyzer type, set language-specific one to improve search quality" choice:"standard" choice:"ar" choice:"de" choice:"en" choice:"es" choice:"fi" choice:"fr" choice:"it" choice:"ru" default:"standard"` //nolint
+}
+
 // LoadingCache defines interface for caching
 type LoadingCache interface {
 	Get(key cache.Key, fn func() ([]byte, error)) (data []byte, err error) // load from cache if found or put to cache and return
@@ -574,6 +584,12 @@ func (s *ServerCommand) newServerApp(ctx context.Context) (*serverApp, error) {
 		return nil, fmt.Errorf("failed to make config of ssl server params: %w", err)
 	}
 
+	dataService.SearchService, err = s.makeSearchService()
+	if err != nil {
+		_ = dataService.Close()
+		return nil, fmt.Errorf("failed to create search service: %w", err)
+	}
+
 	srv := &api.Rest{
 		Version:             s.Revision,
 		DataService:         dataService,
@@ -656,6 +672,20 @@ func (a *serverApp) run(ctx context.Context) error {
 		log.Printf("[WARN] failed to resubmit comments with staging images, %s", e)
 	}
 
+	numWorkersForIndexing := 8
+	grp := syncs.NewErrSizedGroup(numWorkersForIndexing)
+	for _, siteID := range a.Sites {
+		a.dataService.RunSiteIndexers(ctx, siteID, grp) // index comments for the first time run
+	}
+
+	// don't lock here, wait in background to log error if any
+	go func() {
+		err := grp.Wait()
+		if err != nil {
+			log.Printf("[WARN] background task for indexing existing comments failed: %s", err)
+		}
+	}()
+
 	go a.imageService.Cleanup(ctx) // pictures cleanup for staging images
 
 	a.restSrv.Run(a.Address, a.Port)
@@ -1194,6 +1224,22 @@ func (s *ServerCommand) getAuthenticator(ds *service.DataStore, avas avatar.Stor
 	})
 }
 
+func (s *ServerCommand) makeSearchService() (*search.Service, error) {
+	if !s.Search.Enable {
+		return nil, nil
+	}
+	log.Printf("[INFO] creating search service")
+
+	if s.Search.IndexPath == "" {
+		return nil, fmt.Errorf("search index path is not set")
+	}
+
+	return search.NewService(s.Sites, search.ServiceParams{
+		IndexPath: s.Search.IndexPath,
+		Analyzer:  s.Search.Analyzer,
+	})
+}
+
 func (s *ServerCommand) parseSameSite(ss string) http.SameSite {
 	switch strings.ToLower(ss) {
 	case "default":

backend/app/cmd/server_test.go

@@ -828,5 +828,11 @@ func createAppFromCmd(t *testing.T, cmd ServerCommand) (*serverApp, context.Cont
 
 func TestMain(m *testing.M) {
 	// ignore is added only for GitHub Actions, can't reproduce locally
-	goleak.VerifyTestMain(m, goleak.IgnoreTopFunction("net/http.(*Server).Shutdown"))
+	goleak.VerifyTestMain(m,
+		goleak.IgnoreTopFunction("net/http.(*Server).Shutdown"),
+
+		// we should call bleve.Config.Shutdown() to close all the workers,
+		// but we don't do it for each server instance because it's global per application
+		goleak.IgnoreTopFunction("github.com/blevesearch/bleve_index_api.AnalysisWorker"),
+	)
 }

backend/app/main_test.go

@@ -157,5 +157,10 @@ func TestMain(m *testing.M) {
 		m,
 		goleak.IgnoreTopFunction("github.com/umputun/remark42/backend/app.init.0.func1"),
 		goleak.IgnoreTopFunction("net/http.(*Server).Shutdown"),
+
+		// we should call bleve.Config.Shutdown() to close all the workers,
+		// but it's global per application and cannot be reinitialized multiple times
+		// so we do not terminate them, ignore the leak
+		goleak.IgnoreTopFunction("github.com/blevesearch/bleve_index_api.AnalysisWorker"),
 	)
 }

backend/app/rest/api/rest.go

@@ -91,12 +91,20 @@ type LoadingCache interface {
 const hardBodyLimit = 1024 * 64 // limit size of body
 
 const lastCommentsScope = "last"
+const searchScope = "search"
 
 type commentsWithInfo struct {
 	Comments []store.Comment `json:"comments"`
 	Info     store.PostInfo  `json:"info,omitempty"`
 }
 
+type commentsWithTotal struct {
+	// List of comments matching the query and selected by `limit` and `skip` parameters.
+	Comments []store.Comment `json:"comments"`
+	// Total number of comments in the index matching query.
+	Total uint64 `json:"total"`
+}
+
 // Run the lister and request's router, activate rest server
 func (s *Rest) Run(address string, port int) {
 	if address == "*" {
@@ -263,6 +271,7 @@ func (s *Rest) routes() chi.Router {
 			ropen.Get("/list", s.pubRest.listCtrl)
 			ropen.Get("/info", s.pubRest.infoCtrl)
 			ropen.Get("/img", s.ImageProxy.Handler)
+			ropen.Get("/search", s.pubRest.searchQueryCtrl)
 
 			ropen.Route("/rss", func(rrss chi.Router) {
 				rrss.Get("/post", s.rssRest.postCommentsCtrl)

backend/app/rest/api/rest_public.go

@@ -42,6 +42,7 @@ type pubStore interface {
 	Count(locator store.Locator) (int, error)
 	List(siteID string, limit, skip int) ([]store.PostInfo, error)
 	Info(locator store.Locator, readonlyAge int) (store.PostInfo, error)
+	Search(siteID, query, sortBy string, limit, skip int) ([]store.Comment, uint64, error)
 
 	ValidateComment(c *store.Comment) error
 	IsReadOnly(locator store.Locator) bool
@@ -386,6 +387,73 @@ func (s *public) telegramQrCtrl(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+// GET /search?site=siteID&query=query&limit=20&skip=10&sort=[-]field - search documents
+// site - site ID
+// query - user-provided search query
+// limit - number of documents to return, default is 20, maximum is 100
+// skip - number of documents to skip, default is 0 (do not skip anything), maximum is 1000
+// sort - sort by specified field, can be prefixed with "-" to reverse the order
+func (s *public) searchQueryCtrl(w http.ResponseWriter, r *http.Request) {
+	getIntParamInRange := func(min, max, defaultVal int, name string) (int, error) {
+		s := r.URL.Query().Get(name)
+		if s == "" {
+			return defaultVal, nil
+		}
+
+		value, err := strconv.Atoi(s)
+		if err != nil {
+			return 0, fmt.Errorf("%s parameter should be an integer", name)
+		}
+
+		if min <= value && value <= max {
+			return value, nil
+		}
+		return 0, fmt.Errorf("%s parameter should be between %d and %d", name, min, max)
+	}
+
+	var err error
+
+	var limit int
+	if limit, err = getIntParamInRange(1, 100, 20, "limit"); err != nil {
+		rest.SendErrorJSON(w, r, http.StatusBadRequest, fmt.Errorf("wrong param"), err.Error(), rest.ErrInternal)
+		return
+	}
+
+	var skip int
+	if skip, err = getIntParamInRange(0, 1000, 0, "skip"); err != nil {
+		rest.SendErrorJSON(w, r, http.StatusBadRequest, fmt.Errorf("wrong param"), err.Error(), rest.ErrInternal)
+		return
+	}
+
+	siteID := r.URL.Query().Get("site")
+	query := r.URL.Query().Get("query")
+	sortBy := r.URL.Query().Get("sort")
+
+	if siteID == "" || query == "" {
+		rest.SendErrorJSON(w, r, http.StatusBadRequest, fmt.Errorf("missing param"), "site and query parameters are required", rest.ErrInternal)
+		return
+	}
+
+	key := cache.NewKey(siteID).ID(URLKey(r)).Scopes(siteID, searchScope)
+	data, err := s.cache.Get(key, func() ([]byte, error) {
+		comments, total, searchErr := s.dataService.Search(siteID, query, sortBy, limit, skip)
+		if searchErr != nil {
+			return nil, searchErr
+		}
+
+		return encodeJSONWithHTML(commentsWithTotal{Comments: comments, Total: total})
+	})
+
+	if err != nil {
+		rest.SendErrorJSON(w, r, http.StatusBadRequest, err, "can't perform search request", rest.ErrInternal)
+		return
+	}
+
+	if err = R.RenderJSONFromBytes(w, r, data); err != nil {
+		log.Printf("[WARN] can't render search results for site %s", siteID)
+	}
+}
+
 func (s *public) applyView(comments []store.Comment, view string) []store.Comment {
 	if strings.EqualFold(view, "user") {
 		projection := make([]store.Comment, 0, len(comments))

backend/app/rest/api/rest_test.go

@@ -10,6 +10,7 @@ import (
 	"net"
 	"net/http"
 	"net/http/httptest"
+	urlpkg "net/url"
 	"os"
 	"strconv"
 	"strings"
@@ -34,6 +35,7 @@ import (
 	adminstore "github.com/umputun/remark42/backend/app/store/admin"
 	"github.com/umputun/remark42/backend/app/store/engine"
 	"github.com/umputun/remark42/backend/app/store/image"
+	"github.com/umputun/remark42/backend/app/store/search"
 	"github.com/umputun/remark42/backend/app/store/service"
 )
 
@@ -212,6 +214,94 @@ func TestRest_rejectAnonUser(t *testing.T) {
 	assert.Equal(t, http.StatusOK, resp.StatusCode, "real user")
 }
 
+func TestRest_Search(t *testing.T) {
+	var searchIndexPath string
+	ts, _, teardown := startupT(t, func(srv *Rest) {
+		var err error
+		searchIndexPath, err = randomPath(srv.WebRoot, "test-search-remark", "")
+		require.NoError(t, err)
+		srv.DataService.SearchService, err = search.NewService([]string{"remark42"}, search.ServiceParams{
+			IndexPath: searchIndexPath,
+			Analyzer:  "standard",
+		})
+		require.NoError(t, err)
+	})
+	defer teardown()
+	defer func() { _ = os.RemoveAll(searchIndexPath) }()
+
+	t0 := time.Date(2018, 12, 20, 15, 18, 22, 0, time.Local)
+
+	id1 := addComment(t, store.Comment{Text: "test test", Timestamp: t0.Add(1 * time.Minute), Locator: store.Locator{SiteID: "remark42", URL: "https://radio-t.com/abc"}}, ts)
+	id2 := addComment(t, store.Comment{Text: "test hello", Timestamp: t0.Add(2 * time.Minute), Locator: store.Locator{SiteID: "remark42", URL: "https://radio-t.com/blah"}}, ts)
+	id3 := addComment(t, store.Comment{Text: "hello world", Timestamp: t0.Add(3 * time.Minute), Locator: store.Locator{SiteID: "remark42", URL: "https://radio-t.com/blah"}}, ts)
+	id4 := addComment(t, store.Comment{Text: "# title\n\nok\n", Timestamp: t0.Add(4 * time.Minute), Locator: store.Locator{SiteID: "remark42", URL: "https://radio-t.com/blah"}}, ts)
+
+	idToPos := map[string]int{id1: 1, id2: 2, id3: 3, id4: 4}
+	tbl := []struct {
+		query       string
+		extraParams string
+		ids         []int
+		status      int
+	}{
+		{"blah", "", []int{}, http.StatusOK},
+		{"h1", "", []int{}, http.StatusOK},
+		{"world", "", []int{3}, http.StatusOK},
+		{"title", "", []int{4}, http.StatusOK},
+		{"\"hello world\"", "", []int{3}, http.StatusOK},
+		{"test", "&sort=time", []int{1, 2}, http.StatusOK},
+		{"hello", "&sort=-time", []int{3, 2}, http.StatusOK},
+		{"hello world", "&sort=time", []int{2, 3}, http.StatusOK},
+		{"hello world test", "&sort=time", []int{1, 2, 3}, http.StatusOK},
+		{"hello world test", "&sort=time&skip=1", []int{2, 3}, http.StatusOK},
+		{"hello world test", "&sort=time&skip=1&limit=1", []int{2}, http.StatusOK},
+		{"", "", []int{}, http.StatusBadRequest},
+		{"test", "&sort=text", []int{}, http.StatusBadRequest},
+		{"test", "&skip=-1", []int{}, http.StatusBadRequest},
+		{"test", "&limit=999999", []int{}, http.StatusBadRequest},
+	}
+
+	cnt := 0
+	for i, tt := range tbl {
+		tt := tt
+
+		cnt++
+		if (cnt % 10) == 0 {
+			// wait for rate limiter
+			time.Sleep(1 * time.Second)
+		}
+		t.Run(strconv.Itoa(i), func(t *testing.T) {
+			resp, err := http.Get(fmt.Sprintf("%s/api/v1/search?site=remark42&query=%s%s",
+				ts.URL, urlpkg.QueryEscape(tt.query), tt.extraParams))
+			require.NoError(t, err)
+
+			defer resp.Body.Close()
+			body, err := io.ReadAll(resp.Body)
+			require.NoError(t, err)
+			assert.Equal(t, tt.status, resp.StatusCode)
+
+			result := commentsWithTotal{}
+			err = json.Unmarshal(body, &result)
+			require.NoError(t, err)
+
+			foundIds := make([]int, len(result.Comments))
+			for i, c := range result.Comments {
+				foundIds[i] = idToPos[c.ID]
+			}
+			require.Equal(t, tt.ids, foundIds)
+		})
+	}
+}
+
+func TestRest_SearchDisabledFeature(t *testing.T) {
+	ts, _, teardown := startupT(t)
+	defer teardown()
+
+	resp, err := http.Get(ts.URL + "/api/v1/search?site=remark42&query=test")
+	require.NoError(t, err)
+	require.NoError(t, resp.Body.Close())
+	assert.Equal(t, http.StatusBadRequest, resp.StatusCode)
+}
+
 func Test_URLKey(t *testing.T) {
 	tbl := []struct {
 		url  string
@@ -633,5 +723,9 @@ func waitForHTTPSServerStart(port int) {
 }
 
 func TestMain(m *testing.M) {
-	goleak.VerifyTestMain(m)
+	goleak.VerifyTestMain(m,
+		// we should call bleve.Config.Shutdown() to close all the workers,
+		// but we don't do it for each server instance because it's global per application
+		goleak.IgnoreTopFunction("github.com/blevesearch/bleve_index_api.AnalysisWorker"),
+	)
 }