Merge branch 'develop'

1 files changed, 211 insertions, 0 deletions

diff --git a/search_controller.go b/search_controller.go
new file mode 100644
index 0000000..5ed762a
--- /dev/null
+++ b/search_controller.go
@@ -0,0 +1,211 @@
+/*
+ * This file is part of caronte (https://github.com/eciavatta/caronte).
+ * Copyright (c) 2020 Emiliano Ciavatta.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, version 3.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package main
+
+import (
+	"context"
+	log "github.com/sirupsen/logrus"
+	"strings"
+	"sync"
+	"time"
+)
+
+const (
+	secondsToNano     = 1000 * 1000 * 1000
+	maxSearchTimeout  = 10 * secondsToNano
+	maxRecentSearches = 200
+)
+
+type PerformedSearch struct {
+	ID                       RowID         `bson:"_id" json:"id"`
+	SearchOptions            SearchOptions `bson:"search_options" json:"search_options"`
+	AffectedConnections      []RowID       `bson:"affected_connections" json:"-"`
+	AffectedConnectionsCount int           `bson:"affected_connections_count" json:"affected_connections_count"`
+	StartedAt                time.Time     `bson:"started_at" json:"started_at"`
+	FinishedAt               time.Time     `bson:"finished_at" json:"finished_at"`
+	UpdatedAt                time.Time     `bson:"updated_at" json:"updated_at"`
+	Timeout                  time.Duration `bson:"timeout" json:"timeout"`
+}
+
+type SearchOptions struct {
+	TextSearch  TextSearch    `bson:"text_search" json:"text_search"`
+	RegexSearch RegexSearch   `bson:"regex_search" json:"regex_search"`
+	Timeout     time.Duration `bson:"timeout" json:"timeout" binding:"max=60"`
+}
+
+type TextSearch struct {
+	Terms         []string `bson:"terms" json:"terms" binding:"isdefault|min=1,dive,min=3"`
+	ExcludedTerms []string `bson:"excluded_terms" json:"excluded_terms" binding:"isdefault|min=1,dive,min=3"`
+	ExactPhrase   string   `bson:"exact_phrase" json:"exact_phrase" binding:"isdefault|min=3"`
+	CaseSensitive bool     `bson:"case_sensitive" json:"case_sensitive"`
+}
+
+type RegexSearch struct {
+	Pattern           string `bson:"pattern" json:"pattern" binding:"isdefault|min=3"`
+	NotPattern        string `bson:"not_pattern" json:"not_pattern" binding:"isdefault|min=3"`
+	CaseInsensitive   bool   `bson:"case_insensitive" json:"case_insensitive"`
+	MultiLine         bool   `bson:"multi_line" json:"multi_line"`
+	IgnoreWhitespaces bool   `bson:"ignore_whitespaces" json:"ignore_whitespaces"`
+	DotCharacter      bool   `bson:"dot_character" json:"dot_character"`
+}
+
+type SearchController struct {
+	storage           Storage
+	performedSearches []PerformedSearch
+	mutex             sync.Mutex
+}
+
+func NewSearchController(storage Storage) *SearchController {
+	var searches []PerformedSearch
+	if err := storage.Find(Searches).Limit(maxRecentSearches).All(&searches); err != nil {
+		log.WithError(err).Panic("failed to retrieve performed searches")
+	}
+
+	if searches == nil {
+		searches = []PerformedSearch{}
+	}
+
+	return &SearchController{
+		storage:           storage,
+		performedSearches: searches,
+	}
+}
+
+func (sc *SearchController) GetPerformedSearches() []PerformedSearch {
+	sc.mutex.Lock()
+	defer sc.mutex.Unlock()
+
+	return sc.performedSearches
+}
+
+func (sc *SearchController) GetPerformedSearch(id RowID) PerformedSearch {
+	sc.mutex.Lock()
+	defer sc.mutex.Unlock()
+
+	var performedSearch PerformedSearch
+	for _, search := range sc.performedSearches {
+		if search.ID == id {
+			performedSearch = search
+		}
+	}
+
+	return performedSearch
+}
+
+func (sc *SearchController) PerformSearch(c context.Context, options SearchOptions) PerformedSearch {
+	findQuery := sc.storage.Find(ConnectionStreams).Projection(OrderedDocument{{"connection_id", 1}}).Context(c)
+	timeout := options.Timeout * secondsToNano
+	if timeout <= 0 || timeout > maxSearchTimeout {
+		timeout = maxSearchTimeout
+	}
+	findQuery = findQuery.MaxTime(timeout)
+
+	if !options.TextSearch.isZero() {
+		var text string
+		if options.TextSearch.ExactPhrase != "" {
+			text = "\"" + options.TextSearch.ExactPhrase + "\""
+		} else {
+			text = strings.Join(options.TextSearch.Terms, " ")
+			if options.TextSearch.ExcludedTerms != nil {
+				text += " -" + strings.Join(options.TextSearch.ExcludedTerms, " -")
+			}
+		}
+
+		findQuery = findQuery.Filter(OrderedDocument{{"$text", UnorderedDocument{
+			"$search":             text,
+			"$language":           "none",
+			"$caseSensitive":      options.TextSearch.CaseSensitive,
+			"$diacriticSensitive": false,
+		}}})
+	} else {
+		var regexOptions string
+		if options.RegexSearch.CaseInsensitive {
+			regexOptions += "i"
+		}
+		if options.RegexSearch.MultiLine {
+			regexOptions += "m"
+		}
+		if options.RegexSearch.IgnoreWhitespaces {
+			regexOptions += "x"
+		}
+		if options.RegexSearch.DotCharacter {
+			regexOptions += "s"
+		}
+
+		var regex UnorderedDocument
+		if options.RegexSearch.Pattern != "" {
+			regex = UnorderedDocument{"$regex": options.RegexSearch.Pattern, "$options": regexOptions}
+		} else {
+			regex = UnorderedDocument{"$not":
+			UnorderedDocument{"$regex": options.RegexSearch.NotPattern, "$options": regexOptions}}
+		}
+
+		findQuery = findQuery.Filter(OrderedDocument{{"payload_string", regex}})
+	}
+
+	var connections []ConnectionStream
+	startedAt := time.Now()
+	if err := findQuery.All(&connections); err != nil {
+		log.WithError(err).Error("oh no")
+	}
+	affectedConnections := uniqueConnectionIds(connections)
+
+	finishedAt := time.Now()
+	performedSearch := PerformedSearch{
+		ID:                       NewRowID(),
+		SearchOptions:            options,
+		AffectedConnections:      affectedConnections,
+		AffectedConnectionsCount: len(affectedConnections),
+		StartedAt:                startedAt,
+		FinishedAt:               finishedAt,
+		UpdatedAt:                finishedAt,
+		Timeout:                  options.Timeout,
+	}
+	if _, err := sc.storage.Insert(Searches).Context(c).One(performedSearch); err != nil {
+		log.WithError(err).Panic("failed to insert a new performed search")
+	}
+
+	sc.mutex.Lock()
+	sc.performedSearches = append([]PerformedSearch{performedSearch}, sc.performedSearches...)
+	if len(sc.performedSearches) > maxRecentSearches {
+		sc.performedSearches = sc.performedSearches[:200]
+	}
+	sc.mutex.Unlock()
+
+	return performedSearch
+}
+
+func (sc TextSearch) isZero() bool {
+	return sc.Terms == nil && sc.ExcludedTerms == nil && sc.ExactPhrase == ""
+}
+
+func (sc RegexSearch) isZero() bool {
+	return RegexSearch{} == sc
+}
+
+func uniqueConnectionIds(connections []ConnectionStream) []RowID {
+	keys := make(map[RowID]bool)
+	var out []RowID
+	for _, entry := range connections {
+		if _, value := keys[entry.ConnectionID]; !value {
+			keys[entry.ConnectionID] = true
+			out = append(out, entry.ConnectionID)
+		}
+	}
+	return out
+}