feat(activitylog): add a mutex to the store

Signed-off-by: jkoberg <[email protected]>

changelog/6.0.0_2024-06-19/activity-service.md

@@ -2,5 +2,4 @@ Enhancement: Activitylog Service
 
 Adds a new service `activitylog` which stores events (activities) per resource. This data can be retrieved by clients to show item activities
 
-https://github.com/owncloud/ocis/pull/9360
 https://github.com/owncloud/ocis/pull/9327

changelog/unreleased/activity-api.md

@@ -0,0 +1,5 @@
+Enhancement: Activitylog API
+
+Adds an api to the `activitylog` service  which allows retrieving data by clients to show item activities
+
+https://github.com/owncloud/ocis/pull/9361

services/activitylog/pkg/config/config.go

@@ -49,7 +49,7 @@ type Store struct {
 	Database     string        `yaml:"database" env:"ACTIVITYLOG_STORE_DATABASE" desc:"The database name the configured store should use." introductionVersion:"pre5.0"`
 	Table        string        `yaml:"table" env:"ACTIVITYLOG_STORE_TABLE" desc:"The database table the store should use." introductionVersion:"pre5.0"`
 	TTL          time.Duration `yaml:"ttl" env:"OCIS_PERSISTENT_STORE_TTL;ACTIVITYLOG_STORE_TTL" desc:"Time to live for events in the store. See the Environment Variable Types description for more details." introductionVersion:"pre5.0"`
-	Size         int           `yaml:"size" env:"OCIS_PERSISTENT_STORE_SIZE;ACTIVITYLOG_STORE_SIZE" desc:"The maximum quantity of items in the store. Only applies when store type 'ocmem' is configured. Defaults to 512 which is derived from the ocmem package though not exclicitly set as default." introductionVersion:"pre5.0"`
+	Size         int           `yaml:"size" env:"OCIS_PERSISTENT_STORE_SIZE;ACTIVITYLOG_STORE_SIZE" desc:"The maximum quantity of items in the store. Only applies when store type 'ocmem' is configured. Defaults to 512 which is derived from the ocmem package though not explicitly set as default." introductionVersion:"pre5.0"`
 	AuthUsername string        `yaml:"username" env:"OCIS_PERSISTENT_STORE_AUTH_USERNAME;ACTIVITYLOG_STORE_AUTH_USERNAME" desc:"The username to authenticate with the store. Only applies when store type 'nats-js-kv' is configured." introductionVersion:"5.0"`
 	AuthPassword string        `yaml:"password" env:"OCIS_PERSISTENT_STORE_AUTH_PASSWORD;ACTIVITYLOG_STORE_AUTH_PASSWORD" desc:"The password to authenticate with the store. Only applies when store type 'nats-js-kv' is configured." introductionVersion:"5.0"`
 }

services/activitylog/pkg/service/service.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"path/filepath"
 	"reflect"
+	"sync"
 	"time"
 
 	gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
@@ -40,6 +41,7 @@ type ActivitylogService struct {
 	mux        *chi.Mux
 	evHistory  ehsvc.EventHistoryService
 	valService settingssvc.ValueService
+	lock       sync.RWMutex
 
 	registeredEvents map[string]events.Unmarshaller
 }
@@ -73,6 +75,7 @@ func New(opts ...Option) (*ActivitylogService, error) {
 		mux:              o.Mux,
 		evHistory:        o.HistoryClient,
 		valService:       o.ValueClient,
+		lock:             sync.RWMutex{},
 		registeredEvents: make(map[string]events.Unmarshaller),
 	}
 
@@ -184,28 +187,18 @@ func (a *ActivitylogService) AddSpaceActivity(spaceID *provider.StorageSpaceId,
 
 // Activities returns the activities for the given resource
 func (a *ActivitylogService) Activities(rid *provider.ResourceId) ([]RawActivity, error) {
-	resourceID := storagespace.FormatResourceID(*rid)
-
-	records, err := a.store.Read(resourceID)
-	if err != nil && err != microstore.ErrNotFound {
-		return nil, fmt.Errorf("could not read activities: %w", err)
-	}
-
-	if len(records) == 0 {
-		return []RawActivity{}, nil
-	}
+	a.lock.RLock()
+	defer a.lock.RUnlock()
 
-	var activities []RawActivity
-	if err := json.Unmarshal(records[0].Value, &activities); err != nil {
-		return nil, fmt.Errorf("could not unmarshal activities: %w", err)
-	}
-
-	return activities, nil
+	return a.activities(rid)
 }
 
 // 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)
+	a.lock.Lock()
+	defer a.lock.Unlock()
+
+	curActivities, err := a.activities(rid)
 	if err != nil {
 		return err
 	}
@@ -228,6 +221,26 @@ func (a *ActivitylogService) RemoveActivities(rid *provider.ResourceId, toDelete
 	})
 }
 
+func (a *ActivitylogService) activities(rid *provider.ResourceId) ([]RawActivity, error) {
+	resourceID := storagespace.FormatResourceID(*rid)
+
+	records, err := a.store.Read(resourceID)
+	if err != nil && err != microstore.ErrNotFound {
+		return nil, fmt.Errorf("could not read activities: %w", err)
+	}
+
+	if len(records) == 0 {
+		return []RawActivity{}, nil
+	}
+
+	var activities []RawActivity
+	if err := json.Unmarshal(records[0].Value, &activities); err != nil {
+		return nil, fmt.Errorf("could not unmarshal activities: %w", err)
+	}
+
+	return activities, nil
+}
+
 // 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 (
@@ -256,6 +269,9 @@ func (a *ActivitylogService) addActivity(initRef *provider.Reference, eventID st
 }
 
 func (a *ActivitylogService) storeActivity(resourceID string, eventID string, depth int, timestamp time.Time) error {
+	a.lock.Lock()
+	defer a.lock.Unlock()
+
 	records, err := a.store.Read(resourceID)
 	if err != nil && err != microstore.ErrNotFound {
 		return err