Make the write buffer duration configurable

.woodpecker.star

@@ -1886,6 +1886,7 @@ def opencloudServer(storage = "decomposed", accounts_hash_difficulty = 4, depend
         "OC_JWT_SECRET": "some-opencloud-jwt-secret",
         "EVENTHISTORY_STORE": "memory",
         "OC_TRANSLATION_PATH": "%s/tests/config/translations" % dirs["base"],
+        "ACTIVITYLOG_WRITE_BUFFER_DURATION": "0",  # Disable write buffer so that test expectations are met in time
         # debug addresses required for running services health tests
         "ACTIVITYLOG_DEBUG_ADDR": "0.0.0.0:9197",
         "APP_PROVIDER_DEBUG_ADDR": "0.0.0.0:9165",

services/activitylog/pkg/config/config.go

@@ -32,6 +32,8 @@ type Config struct {
 	ServiceAccount ServiceAccount `yaml:"service_account"`
 
 	Context context.Context `yaml:"-"`
+
+	WriteBufferDuration time.Duration `yaml:"write_buffer_duration" env:"ACTIVITYLOG_WRITE_BUFFER_DURATION" desc:"The duration to wait before flushing the write buffer. This is used to reduce the number of writes to the store." introductionVersion:"%%NEXT%%"`
 }
 
 // Events combines the configuration options for the event bus.

services/activitylog/pkg/config/defaults/defaultconfig.go

@@ -1,6 +1,8 @@
 package defaults
 
 import (
+	"time"
+
 	"github.com/opencloud-eu/opencloud/pkg/shared"
 	"github.com/opencloud-eu/opencloud/pkg/structs"
 	"github.com/opencloud-eu/opencloud/services/activitylog/pkg/config"
@@ -50,6 +52,7 @@ func DefaultConfig() *config.Config {
 				AllowCredentials: true,
 			},
 		},
+		WriteBufferDuration: 10 * time.Second,
 	}
 }
 

services/activitylog/pkg/server/http/server.go

@@ -88,6 +88,7 @@ func Server(opts ...Option) (http.Service, error) {
 		svc.HistoryClient(options.HistoryClient),
 		svc.ValueClient(options.ValueClient),
 		svc.RegisteredEvents(options.RegisteredEvents),
+		svc.WriteBufferDuration(options.Config.WriteBufferDuration),
 	)
 	if err != nil {
 		return http.Service{}, err

services/activitylog/pkg/service/options.go

@@ -1,6 +1,8 @@
 package service
 
 import (
+	"time"
+
 	gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
 	"github.com/go-chi/chi/v5"
 	"github.com/opencloud-eu/opencloud/pkg/log"
@@ -18,16 +20,17 @@ type Option func(*Options)
 
 // Options for the activitylog service
 type Options struct {
-	Logger           log.Logger
-	Config           *config.Config
-	TraceProvider    trace.TracerProvider
-	Stream           events.Stream
-	RegisteredEvents []events.Unmarshaller
-	Store            microstore.Store
-	GatewaySelector  pool.Selectable[gateway.GatewayAPIClient]
-	Mux              *chi.Mux
-	HistoryClient    ehsvc.EventHistoryService
-	ValueClient      settingssvc.ValueService
+	Logger              log.Logger
+	Config              *config.Config
+	TraceProvider       trace.TracerProvider
+	Stream              events.Stream
+	RegisteredEvents    []events.Unmarshaller
+	Store               microstore.Store
+	GatewaySelector     pool.Selectable[gateway.GatewayAPIClient]
+	Mux                 *chi.Mux
+	HistoryClient       ehsvc.EventHistoryService
+	ValueClient         settingssvc.ValueService
+	WriteBufferDuration time.Duration
 }
 
 // Logger configures a logger for the activitylog service
@@ -99,3 +102,10 @@ func ValueClient(vs settingssvc.ValueService) Option {
 		o.ValueClient = vs
 	}
 }
+
+// WriteBufferDuration sets the write buffer duration
+func WriteBufferDuration(d time.Duration) Option {
+	return func(o *Options) {
+		o.WriteBufferDuration = d
+	}
+}

services/activitylog/pkg/service/service.go

@@ -143,7 +143,7 @@ func New(opts ...Option) (*ActivitylogService, error) {
 	}
 
 	cache := ttlcache.NewCache()
-	err = cache.SetTTL(10 * time.Second)
+	err = cache.SetTTL(30 * time.Second)
 	if err != nil {
 		return nil, err
 	}
@@ -163,7 +163,7 @@ func New(opts ...Option) (*ActivitylogService, error) {
 		tracer:           o.TraceProvider.Tracer("github.com/opencloud-eu/opencloud/services/activitylog/pkg/service"),
 		parentIdCache:    cache,
 	}
-	s.debouncer = NewDebouncer(10*time.Second, s.storeActivity)
+	s.debouncer = NewDebouncer(o.WriteBufferDuration, s.storeActivity)
 
 	s.mux.Get("/graph/v1beta1/extensions/org.libregraph/activities", s.HandleGetItemActivities)