feat(activitylog): allow filtering activities

Signed-off-by: jkoberg <jkoberg@owncloud.com>

services/activitylog/pkg/config/debug.go

@@ -2,8 +2,8 @@ package config
 
 // Debug defines the available debug configuration.
 type Debug struct {
-	Addr   string `yaml:"addr" env:"CLIENTLOG_DEBUG_ADDR" desc:"Bind address of the debug server, where metrics, health, config and debug endpoints will be exposed." introductionVersion:"5.0"`
-	Token  string `yaml:"token" env:"CLIENTLOG_DEBUG_TOKEN" desc:"Token to secure the metrics endpoint." introductionVersion:"5.0"`
-	Pprof  bool   `yaml:"pprof" env:"CLIENTLOG_DEBUG_PPROF" desc:"Enables pprof, which can be used for profiling." introductionVersion:"5.0"`
-	Zpages bool   `yaml:"zpages" env:"CLIENTLOG_DEBUG_ZPAGES" desc:"Enables zpages, which can be used for collecting and viewing in-memory traces." introductionVersion:"5.0"`
+	Addr   string `yaml:"addr" env:"ACTIVITYLOG_DEBUG_ADDR" desc:"Bind address of the debug server, where metrics, health, config and debug endpoints will be exposed." introductionVersion:"5.0"`
+	Token  string `yaml:"token" env:"ACTIVITYLOG_DEBUG_TOKEN" desc:"Token to secure the metrics endpoint." introductionVersion:"5.0"`
+	Pprof  bool   `yaml:"pprof" env:"ACTIVITYLOG_DEBUG_PPROF" desc:"Enables pprof, which can be used for profiling." introductionVersion:"5.0"`
+	Zpages bool   `yaml:"zpages" env:"ACTIVITYLOG_DEBUG_ZPAGES" desc:"Enables zpages, which can be used for collecting and viewing in-memory traces." introductionVersion:"5.0"`
 }

services/activitylog/pkg/config/log.go

@@ -2,8 +2,8 @@ package config
 
 // Log defines the available log configuration.
 type Log struct {
-	Level  string `mapstructure:"level" env:"OCIS_LOG_LEVEL;CLIENTLOG_USERLOG_LOG_LEVEL" desc:"The log level. Valid values are: 'panic', 'fatal', 'error', 'warn', 'info', 'debug', 'trace'." introductionVersion:"5.0"`
-	Pretty bool   `mapstructure:"pretty" env:"OCIS_LOG_PRETTY;CLIENTLOG_USERLOG_LOG_PRETTY" desc:"Activates pretty log output." introductionVersion:"5.0"`
-	Color  bool   `mapstructure:"color" env:"OCIS_LOG_COLOR;CLIENTLOG_USERLOG_LOG_COLOR" desc:"Activates colorized log output." introductionVersion:"5.0"`
-	File   string `mapstructure:"file" env:"OCIS_LOG_FILE;CLIENTLOG_USERLOG_LOG_FILE" desc:"The path to the log file. Activates logging to this file if set." introductionVersion:"5.0"`
+	Level  string `mapstructure:"level" env:"OCIS_LOG_LEVEL;ACTIVITYLOG_USERLOG_LOG_LEVEL" desc:"The log level. Valid values are: 'panic', 'fatal', 'error', 'warn', 'info', 'debug', 'trace'." introductionVersion:"5.0"`
+	Pretty bool   `mapstructure:"pretty" env:"OCIS_LOG_PRETTY;ACTIVITYLOG_USERLOG_LOG_PRETTY" desc:"Activates pretty log output." introductionVersion:"5.0"`
+	Color  bool   `mapstructure:"color" env:"OCIS_LOG_COLOR;ACTIVITYLOG_USERLOG_LOG_COLOR" desc:"Activates colorized log output." introductionVersion:"5.0"`
+	File   string `mapstructure:"file" env:"OCIS_LOG_FILE;ACTIVITYLOG_USERLOG_LOG_FILE" desc:"The path to the log file. Activates logging to this file if set." introductionVersion:"5.0"`
 }

services/activitylog/pkg/config/tracing.go

@@ -4,10 +4,10 @@ import "github.com/owncloud/ocis/v2/ocis-pkg/tracing"
 
 // Tracing defines the available tracing configuration.
 type Tracing struct {
-	Enabled   bool   `yaml:"enabled" env:"OCIS_TRACING_ENABLED;CLIENTLOG_TRACING_ENABLED" desc:"Activates tracing." introductionVersion:"5.0"`
-	Type      string `yaml:"type" env:"OCIS_TRACING_TYPE;CLIENTLOG_TRACING_TYPE" desc:"The type of tracing. Defaults to '', which is the same as 'jaeger'. Allowed tracing types are 'jaeger' and '' as of now." introductionVersion:"5.0"`
-	Endpoint  string `yaml:"endpoint" env:"OCIS_TRACING_ENDPOINT;CLIENTLOG_TRACING_ENDPOINT" desc:"The endpoint of the tracing agent." introductionVersion:"5.0"`
-	Collector string `yaml:"collector" env:"OCIS_TRACING_COLLECTOR;CLIENTLOG_TRACING_COLLECTOR" desc:"The HTTP endpoint for sending spans directly to a collector, i.e. http://jaeger-collector:14268/api/traces. Only used if the tracing endpoint is unset." introductionVersion:"5.0"`
+	Enabled   bool   `yaml:"enabled" env:"OCIS_TRACING_ENABLED;ACTIVITYLOG_TRACING_ENABLED" desc:"Activates tracing." introductionVersion:"5.0"`
+	Type      string `yaml:"type" env:"OCIS_TRACING_TYPE;ACTIVITYLOG_TRACING_TYPE" desc:"The type of tracing. Defaults to '', which is the same as 'jaeger'. Allowed tracing types are 'jaeger' and '' as of now." introductionVersion:"5.0"`
+	Endpoint  string `yaml:"endpoint" env:"OCIS_TRACING_ENDPOINT;ACTIVITYLOG_TRACING_ENDPOINT" desc:"The endpoint of the tracing agent." introductionVersion:"5.0"`
+	Collector string `yaml:"collector" env:"OCIS_TRACING_COLLECTOR;ACTIVITYLOG_TRACING_COLLECTOR" desc:"The HTTP endpoint for sending spans directly to a collector, i.e. http://jaeger-collector:14268/api/traces. Only used if the tracing endpoint is unset." introductionVersion:"5.0"`
 }
 
 // Convert Tracing to the tracing package's Config struct.

services/activitylog/pkg/service/http.go

@@ -3,8 +3,9 @@ package service
 import (
 	"embed"
 	"encoding/json"
+	"errors"
 	"net/http"
-	"net/url"
+	"strconv"
 	"strings"
 
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
@@ -12,13 +13,11 @@ import (
 	"github.com/cs3org/reva/v2/pkg/events"
 	"github.com/cs3org/reva/v2/pkg/storagespace"
 	"github.com/cs3org/reva/v2/pkg/utils"
-	"github.com/go-chi/chi/v5"
 	libregraph "github.com/owncloud/libre-graph-api-go"
 
 	"github.com/owncloud/ocis/v2/ocis-pkg/l10n"
 	ehmsg "github.com/owncloud/ocis/v2/protogen/gen/ocis/messages/eventhistory/v0"
 	ehsvc "github.com/owncloud/ocis/v2/protogen/gen/ocis/services/eventhistory/v0"
-	"github.com/owncloud/ocis/v2/services/graph/pkg/errorcode"
 	"github.com/owncloud/ocis/v2/services/search/pkg/query/ast"
 	"github.com/owncloud/ocis/v2/services/search/pkg/query/kql"
 )
@@ -47,40 +46,15 @@ func (s *ActivitylogService) HandleGetItemActivities(w http.ResponseWriter, r *h
 		return
 	}
 
-	qraw := r.URL.Query().Get("kql")
-	if qraw == "" {
-		w.WriteHeader(http.StatusBadRequest)
-	}
-
-	qBuilder := kql.Builder{}
-	qast, err := qBuilder.Build(qraw)
+	rid, limit, rawActivityAccepted, activityAccepted, err := s.getFilters(r.URL.Query().Get("kql"))
 	if err != nil {
-		w.WriteHeader(http.StatusBadRequest)
-	}
-
-	var itemID string
-
-	for _, n := range qast.Nodes {
-		v, ok := n.(*ast.StringNode)
-		if !ok {
-			continue
-		}
-
-		if strings.ToLower(v.Key) != "itemid" {
-			continue
-		}
-
-		itemID = v.Value
-	}
-
-	rid, err := storagespace.ParseID(itemID)
-	if err != nil {
-		s.log.Info().Err(err).Msg("invalid resource id")
+		s.log.Info().Str("query", r.URL.Query().Get("kql")).Err(err).Msg("error getting filters")
+		_, _ = w.Write([]byte(err.Error()))
 		w.WriteHeader(http.StatusBadRequest)
 		return
 	}
 
-	raw, err := s.Activities(&rid)
+	raw, err := s.Activities(rid)
 	if err != nil {
 		s.log.Error().Err(err).Msg("error getting activities")
 		w.WriteHeader(http.StatusInternalServerError)
@@ -88,9 +62,13 @@ func (s *ActivitylogService) HandleGetItemActivities(w http.ResponseWriter, r *h
 	}
 
 	ids := make([]string, 0, len(raw))
+	toDelete := make(map[string]struct{}, len(raw))
 	for _, a := range raw {
-		// TODO: Filter by depth and timestamp
+		if !rawActivityAccepted(a) {
+			continue
+		}
 		ids = append(ids, a.EventID)
+		toDelete[a.EventID] = struct{}{}
 	}
 
 	evRes, err := s.evHistory.GetEvents(r.Context(), &ehsvc.GetEventsRequest{Ids: ids})
@@ -102,9 +80,15 @@ func (s *ActivitylogService) HandleGetItemActivities(w http.ResponseWriter, r *h
 
 	var resp GetActivitiesResponse
 	for _, e := range evRes.GetEvents() {
-		// TODO: compare returned events with initial list and remove missing ones
+		delete(toDelete, e.GetId())
 
-		// FIXME: Should all users get all events? If not we can filter here
+		if limit != 0 && len(resp.Activities) >= limit {
+			continue
+		}
+
+		if !activityAccepted(e) {
+			continue
+		}
 
 		var (
 			message string
@@ -169,13 +153,23 @@ func (s *ActivitylogService) HandleGetItemActivities(w http.ResponseWriter, r *h
 			continue
 		}
 
-		// todo: configurable default locale?
+		// FIXME: configurable default locale?
 		loc := l10n.MustGetUserLocale(r.Context(), activeUser.GetId().GetOpaqueId(), r.Header.Get(l10n.HeaderAcceptLanguage), s.valService)
 		t := l10n.NewTranslatorFromCommonConfig("en", _domain, "", _localeFS, _localeSubPath)
 
 		resp.Activities = append(resp.Activities, NewActivity(t.Translate(message, loc), res, act, ts, e.GetId()))
 	}
 
+	// delete activities in separate go routine
+	if len(toDelete) > 0 {
+		go func() {
+			err := s.RemoveActivities(rid, toDelete)
+			if err != nil {
+				s.log.Error().Err(err).Msg("error removing activities")
+			}
+		}()
+	}
+
 	b, err := json.Marshal(resp)
 	if err != nil {
 		s.log.Error().Err(err).Msg("error marshalling activities")
@@ -207,16 +201,80 @@ func (s *ActivitylogService) unwrapEvent(e *ehmsg.Event) interface{} {
 	return einterface
 }
 
-// TODO: I found this on graph service. We should move it to `utils` pkg so both services can use it.
-func parseIDParam(r *http.Request, param string) (provider.ResourceId, error) {
-	driveID, err := url.PathUnescape(chi.URLParam(r, param))
+func (s *ActivitylogService) getFilters(query string) (*provider.ResourceId, int, func(RawActivity) bool, func(*ehmsg.Event) bool, error) {
+	qast, err := kql.Builder{}.Build(query)
 	if err != nil {
-		return provider.ResourceId{}, errorcode.New(errorcode.InvalidRequest, err.Error())
+		return nil, 0, nil, nil, err
 	}
 
-	id, err := storagespace.ParseID(driveID)
-	if err != nil {
-		return provider.ResourceId{}, errorcode.New(errorcode.InvalidRequest, err.Error())
+	prefilters := make([]func(RawActivity) bool, 0)
+	postfilters := make([]func(*ehmsg.Event) bool, 0)
+
+	var (
+		itemID string
+		limit  int
+	)
+
+	for _, n := range qast.Nodes {
+		switch v := n.(type) {
+		case *ast.StringNode:
+			switch strings.ToLower(v.Key) {
+			case "itemid":
+				itemID = v.Value
+			case "depth":
+				depth, err := strconv.Atoi(v.Value)
+				if err != nil {
+					return nil, limit, nil, nil, err
+				}
+
+				prefilters = append(prefilters, func(a RawActivity) bool {
+					return a.Depth <= depth
+				})
+			case "limit":
+				l, err := strconv.Atoi(v.Value)
+				if err != nil {
+					return nil, limit, nil, nil, err
+				}
+
+				limit = l
+			}
+		case *ast.DateTimeNode:
+			switch v.Operator.Value {
+			case "<", "<=":
+				prefilters = append(prefilters, func(a RawActivity) bool {
+					return a.Timestamp.Before(v.Value)
+				})
+			case ">", ">=":
+				prefilters = append(prefilters, func(a RawActivity) bool {
+					return a.Timestamp.After(v.Value)
+				})
+			}
+		case *ast.OperatorNode:
+			if v.Value != "AND" {
+				return nil, limit, nil, nil, errors.New("only AND operator is supported")
+			}
+		}
 	}
-	return id, nil
+
+	rid, err := storagespace.ParseID(itemID)
+	if err != nil {
+		return nil, limit, nil, nil, err
+	}
+	pref := func(a RawActivity) bool {
+		for _, f := range prefilters {
+			if !f(a) {
+				return false
+			}
+		}
+		return true
+	}
+	postf := func(e *ehmsg.Event) bool {
+		for _, f := range postfilters {
+			if !f(e) {
+				return false
+			}
+		}
+		return true
+	}
+	return &rid, limit, pref, postf, nil
 }

services/activitylog/pkg/service/service.go

@@ -89,7 +89,7 @@ func New(opts ...Option) (*ActivitylogService, error) {
 }
 
 // Run runs the service
-func (a *ActivitylogService) Run() error {
+func (a *ActivitylogService) Run() {
 	for e := range a.events {
 		var err error
 		switch ev := e.Event.(type) {
@@ -129,7 +129,6 @@ func (a *ActivitylogService) Run() error {
 			a.log.Error().Err(err).Interface("event", e).Msg("could not process event")
 		}
 	}
-	return nil
 }
 
 // AddActivity adds the activity to the given resource and all its parents
@@ -198,6 +197,31 @@ func (a *ActivitylogService) Activities(rid *provider.ResourceId) ([]RawActivity
 	return activities, nil
 }
 
+// RemoveActivities removes the activities from the given resource
+func (a *ActivitylogService) RemoveActivities(rid *provider.ResourceId, toDelete map[string]struct{}) error {
+	curActivities, err := a.Activities(rid)
+	if err != nil {
+		return err
+	}
+
+	var acts []RawActivity
+	for _, a := range curActivities {
+		if _, ok := toDelete[a.EventID]; !ok {
+			acts = append(acts, a)
+		}
+	}
+
+	b, err := json.Marshal(acts)
+	if err != nil {
+		return err
+	}
+
+	return a.store.Write(&microstore.Record{
+		Key:   storagespace.FormatResourceID(*rid),
+		Value: b,
+	})
+}
+
 // note: getResource is abstracted to allow unit testing, in general this will just be utils.GetResource
 func (a *ActivitylogService) addActivity(initRef *provider.Reference, eventID string, timestamp time.Time, getResource func(*provider.Reference) (*provider.ResourceInfo, error)) error {
 	var (