[full-ci] InitiatorIDs

changelog/unreleased/bump-reva.md

@@ -2,4 +2,5 @@ Enhancement: Bump Reva
 
 bumps reva version
 
+https://github.com/owncloud/ocis/pull/8701
 https://github.com/owncloud/ocis/pull/8606

changelog/unreleased/initiator-ids.md

@@ -0,0 +1,5 @@
+Enhancement: Initiator-IDs
+
+Allows sending a header `Initiator-ID` on http requests. This id will be added to sse events so clients can figure out if their particular instance was triggering the event. Additionally this adds the etag of the file/folder to all sse events.
+
+https://github.com/owncloud/ocis/pull/8701

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/cenkalti/backoff v2.2.1+incompatible
 	github.com/coreos/go-oidc/v3 v3.10.0
 	github.com/cs3org/go-cs3apis v0.0.0-20231023073225-7748710e0781
-	github.com/cs3org/reva/v2 v2.19.2-0.20240320153915-890d63097350
+	github.com/cs3org/reva/v2 v2.19.2-0.20240322140620-cbb501a7ae3a
 	github.com/dhowden/tag v0.0.0-20230630033851-978a0926ee25
 	github.com/disintegration/imaging v1.6.2
 	github.com/dutchcoders/go-clamd v0.0.0-20170520113014-b970184f4d9e

go.sum

@@ -1018,8 +1018,8 @@ github.com/crewjam/saml v0.4.14 h1:g9FBNx62osKusnFzs3QTN5L9CVA/Egfgm+stJShzw/c=
 github.com/crewjam/saml v0.4.14/go.mod h1:UVSZCf18jJkk6GpWNVqcyQJMD5HsRugBPf4I1nl2mME=
 github.com/cs3org/go-cs3apis v0.0.0-20231023073225-7748710e0781 h1:BUdwkIlf8IS2FasrrPg8gGPHQPOrQ18MS1Oew2tmGtY=
 github.com/cs3org/go-cs3apis v0.0.0-20231023073225-7748710e0781/go.mod h1:UXha4TguuB52H14EMoSsCqDj7k8a/t7g4gVP+bgY5LY=
-github.com/cs3org/reva/v2 v2.19.2-0.20240320153915-890d63097350 h1:Y0P9eve8cAwj3i3rDDeUusoW7yUs6AZFkU+ANygkrKo=
-github.com/cs3org/reva/v2 v2.19.2-0.20240320153915-890d63097350/go.mod h1:GRUrOp5HbFVwZTgR9bVrMZ/MvVy+Jhxw1PdMmhhKP9E=
+github.com/cs3org/reva/v2 v2.19.2-0.20240322140620-cbb501a7ae3a h1:dqIqhnxiRYfmDHhlgtEAeTbOJxQ2nca4O/Gius/TnxQ=
+github.com/cs3org/reva/v2 v2.19.2-0.20240322140620-cbb501a7ae3a/go.mod h1:GRUrOp5HbFVwZTgR9bVrMZ/MvVy+Jhxw1PdMmhhKP9E=
 github.com/cyberdelia/templates v0.0.0-20141128023046-ca7fffd4298c/go.mod h1:GyV+0YP4qX0UQ7r2MoYZ+AvYDp12OF5yg4q8rGnyNh4=
 github.com/cyphar/filepath-securejoin v0.2.4 h1:Ugdm7cg7i6ZK6x3xDF1oEu1nfkyfH53EtKeQYTC3kyg=
 github.com/cyphar/filepath-securejoin v0.2.4/go.mod h1:aPGpWjXOXUn2NCNjFvBE6aRxGGx79pTxQpKOJNYHHl4=

services/clientlog/pkg/service/events.go

@@ -5,4 +5,6 @@ type FileEvent struct {
 	ParentItemID string `json:"parentitemid"`
 	ItemID       string `json:"itemid"`
 	SpaceID      string `json:"spaceid"`
+	InitiatorID  string `json:"initiatorid"`
+	Etag         string `json:"etag"`
 }

services/clientlog/pkg/service/service.go

@@ -95,12 +95,17 @@ func (cl *ClientlogService) processEvent(event events.Event) {
 		evType string
 		data   interface{}
 	)
+
+	p := func(typ string, ref *provider.Reference) {
+		evType = typ
+		users, data, err = processFileEvent(ctx, ref, gwc, event.InitiatorID)
+	}
+
 	switch e := event.Event.(type) {
 	default:
 		err = errors.New("unhandled event")
 	case events.UploadReady:
-		evType = "postprocessing-finished"
-		users, data, err = processFileEvent(ctx, e.FileRef, gwc)
+		p("postprocessing-finished", e.FileRef)
 	case events.ItemTrashed:
 		evType = "item-trashed"
 
@@ -120,32 +125,28 @@ func (cl *ClientlogService) processEvent(event events.Event) {
 					ItemID: storagespace.FormatResourceID(*e.ID),
 					// TODO: check with web if parentID is needed
 					// ParentItemID: storagespace.FormatResourceID(*item.GetRef().GetResourceId()),
-					SpaceID: storagespace.FormatStorageID(e.ID.GetStorageId(), e.ID.GetSpaceId()),
+					SpaceID:     storagespace.FormatStorageID(e.ID.GetStorageId(), e.ID.GetSpaceId()),
+					InitiatorID: event.InitiatorID,
 				}
 
 				users, err = utils.GetSpaceMembers(ctx, e.ID.GetSpaceId(), gwc, utils.ViewerRole)
 				break
 			}
 		}
 	case events.ItemRestored:
-		evType = "item-restored"
-		users, data, err = processFileEvent(ctx, e.Ref, gwc)
+		p("item-restored", e.Ref)
 	case events.ContainerCreated:
-		evType = "folder-created"
-		users, data, err = processFileEvent(ctx, e.Ref, gwc)
+		p("folder-created", e.Ref)
 	case events.ItemMoved:
 		// we are only interested in the rename case
 		if !utils.ResourceIDEqual(e.OldReference.GetResourceId(), e.Ref.GetResourceId()) || e.Ref.GetPath() == e.OldReference.GetPath() {
 			return
 		}
-		evType = "item-renamed"
-		users, data, err = processFileEvent(ctx, e.Ref, gwc)
+		p("item-renamed", e.Ref)
 	case events.FileLocked:
-		evType = "file-locked"
-		users, data, err = processFileEvent(ctx, e.Ref, gwc)
+		p("file-locked", e.Ref)
 	case events.FileUnlocked:
-		evType = "file-unlocked"
-		users, data, err = processFileEvent(ctx, e.Ref, gwc)
+		p("file-unlocked", e.Ref)
 	}
 
 	if err != nil {
@@ -173,7 +174,7 @@ func (cl *ClientlogService) sendSSE(userIDs []string, evType string, data interf
 	})
 }
 
-func processFileEvent(ctx context.Context, ref *provider.Reference, gwc gateway.GatewayAPIClient) ([]string, FileEvent, error) {
+func processFileEvent(ctx context.Context, ref *provider.Reference, gwc gateway.GatewayAPIClient, initiatorid string) ([]string, FileEvent, error) {
 	info, err := utils.GetResource(ctx, ref, gwc)
 	if err != nil {
 		return nil, FileEvent{}, err
@@ -183,6 +184,8 @@ func processFileEvent(ctx context.Context, ref *provider.Reference, gwc gateway.
 		ParentItemID: storagespace.FormatResourceID(*info.GetParentId()),
 		ItemID:       storagespace.FormatResourceID(*info.GetId()),
 		SpaceID:      storagespace.FormatStorageID(info.GetSpace().GetRoot().GetStorageId(), info.GetSpace().GetRoot().GetSpaceId()),
+		InitiatorID:  initiatorid,
+		Etag:         info.GetEtag(),
 	}
 
 	users, err := utils.GetSpaceMembers(ctx, info.GetSpace().GetId().GetOpaqueId(), gwc, utils.ViewerRole)

services/postprocessing/pkg/postprocessing/postprocessing.go

@@ -12,15 +12,16 @@ import (
 
 // Postprocessing handles postprocessing of a file
 type Postprocessing struct {
-	ID         string
-	URL        string
-	User       *user.User
-	Filename   string
-	Filesize   uint64
-	ResourceID *provider.ResourceId
-	Steps      []events.Postprocessingstep
-	Status     Status
-	Failures   int
+	ID          string
+	URL         string
+	User        *user.User
+	Filename    string
+	Filesize    uint64
+	ResourceID  *provider.ResourceId
+	Steps       []events.Postprocessingstep
+	Status      Status
+	Failures    int
+	InitiatorID string
 
 	config config.Postprocessing
 }

services/postprocessing/pkg/service/service.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"time"
 
+	ctxpkg "github.com/cs3org/reva/v2/pkg/ctx"
 	"github.com/cs3org/reva/v2/pkg/events"
 	"github.com/cs3org/reva/v2/pkg/utils"
 	"github.com/owncloud/ocis/v2/ocis-pkg/log"
@@ -92,13 +93,14 @@ func (pps *PostprocessingService) processEvent(e events.Event) error {
 	switch ev := e.Event.(type) {
 	case events.BytesReceived:
 		pp = &postprocessing.Postprocessing{
-			ID:         ev.UploadID,
-			URL:        ev.URL,
-			User:       ev.ExecutingUser,
-			Filename:   ev.Filename,
-			Filesize:   ev.Filesize,
-			ResourceID: ev.ResourceID,
-			Steps:      pps.steps,
+			ID:          ev.UploadID,
+			URL:         ev.URL,
+			User:        ev.ExecutingUser,
+			Filename:    ev.Filename,
+			Filesize:    ev.Filesize,
+			ResourceID:  ev.ResourceID,
+			Steps:       pps.steps,
+			InitiatorID: e.InitiatorID,
 		}
 		next = pp.Init(ev)
 	case events.PostprocessingStepFinished:
@@ -160,11 +162,14 @@ func (pps *PostprocessingService) processEvent(e events.Event) error {
 	}
 
 	if pp != nil {
+		ctx = ctxpkg.ContextSetInitiator(ctx, pp.InitiatorID)
+
 		if err := storePP(pps.store, pp); err != nil {
 			pps.log.Error().Str("uploadID", pp.ID).Err(err).Msg("cannot store upload")
 			return fmt.Errorf("%w: cannot store upload", errEvent)
 		}
 	}
+
 	if next != nil {
 		if err := events.Publish(ctx, pps.pub, next); err != nil {
 			pps.log.Error().Err(err).Msg("unable to publish event")