Merge pull request #3345 from aduffeck/merge-edge-into-experimental

[tests-only] Merge edge into experimental

changelog/unreleased/add-spaceowner-to-events.md

@@ -0,0 +1,6 @@
+Enhancement: Add SpaceOwner to some event
+
+We added a SpaceOwner field to some of the events which can be used by
+consumers to gain access to the affected space.
+
+https://github.com/cs3org/reva/pull/3340

internal/grpc/interceptors/eventsmiddleware/conversion.go

@@ -28,10 +28,11 @@ import (
 )
 
 // ContainerCreated converts the response to an event
-func ContainerCreated(r *provider.CreateContainerResponse, req *provider.CreateContainerRequest, executant *user.UserId) events.ContainerCreated {
+func ContainerCreated(r *provider.CreateContainerResponse, req *provider.CreateContainerRequest, spaceOwner, executant *user.UserId) events.ContainerCreated {
  return events.ContainerCreated{
- Executant: executant,
- Ref: req.Ref,
+ SpaceOwner: spaceOwner,
+ Executant: executant,
+ Ref: req.Ref,
  }
 }
 
@@ -166,18 +167,20 @@ func LinkRemoved(r *link.RemovePublicShareResponse, req *link.RemovePublicShareR
 }
 
 // FileTouched converts the response to an event
-func FileTouched(r *provider.TouchFileResponse, req *provider.TouchFileRequest, executant *user.UserId) events.FileTouched {
+func FileTouched(r *provider.TouchFileResponse, req *provider.TouchFileRequest, spaceOwner, executant *user.UserId) events.FileTouched {
  return events.FileTouched{
- Executant: executant,
- Ref: req.Ref,
+ SpaceOwner: spaceOwner,
+ Executant: executant,
+ Ref: req.Ref,
  }
 }
 
 // FileUploaded converts the response to an event
-func FileUploaded(r *provider.InitiateFileUploadResponse, req *provider.InitiateFileUploadRequest, executant *user.UserId) events.FileUploaded {
+func FileUploaded(r *provider.InitiateFileUploadResponse, req *provider.InitiateFileUploadRequest, spaceOwner, executant *user.UserId) events.FileUploaded {
  return events.FileUploaded{
- Executant: executant,
- Ref: req.Ref,
+ SpaceOwner: spaceOwner,
+ Executant: executant,
+ Ref: req.Ref,
  }
 }
 
@@ -190,11 +193,12 @@ func FileDownloaded(r *provider.InitiateFileDownloadResponse, req *provider.Init
 }
 
 // ItemTrashed converts the response to an event
-func ItemTrashed(r *provider.DeleteResponse, req *provider.DeleteRequest, executant *user.UserId) events.ItemTrashed {
+func ItemTrashed(r *provider.DeleteResponse, req *provider.DeleteRequest, spaceOwner, executant *user.UserId) events.ItemTrashed {
  opaqueID := utils.ReadPlainFromOpaque(r.Opaque, "opaque_id")
  return events.ItemTrashed{
- Executant: executant,
- Ref: req.Ref,
+ SpaceOwner: spaceOwner,
+ Executant: executant,
+ Ref: req.Ref,
  ID: &provider.ResourceId{
  StorageId: req.Ref.GetResourceId().GetStorageId(),
  SpaceId: req.Ref.GetResourceId().GetSpaceId(),
@@ -204,8 +208,9 @@ func ItemTrashed(r *provider.DeleteResponse, req *provider.DeleteRequest, execut
 }
 
 // ItemMoved converts the response to an event
-func ItemMoved(r *provider.MoveResponse, req *provider.MoveRequest, executant *user.UserId) events.ItemMoved {
+func ItemMoved(r *provider.MoveResponse, req *provider.MoveRequest, spaceOwner, executant *user.UserId) events.ItemMoved {
  return events.ItemMoved{
+ SpaceOwner: spaceOwner,
  Executant: executant,
  Ref: req.Destination,
  OldReference: req.Source,
@@ -221,12 +226,13 @@ func ItemPurged(r *provider.PurgeRecycleResponse, req *provider.PurgeRecycleRequ
 }
 
 // ItemRestored converts the response to an event
-func ItemRestored(r *provider.RestoreRecycleItemResponse, req *provider.RestoreRecycleItemRequest, executant *user.UserId) events.ItemRestored {
+func ItemRestored(r *provider.RestoreRecycleItemResponse, req *provider.RestoreRecycleItemRequest, spaceOwner, executant *user.UserId) events.ItemRestored {
  ref := req.Ref
  if req.RestoreRef != nil {
  ref = req.RestoreRef
  }
  return events.ItemRestored{
+ SpaceOwner: spaceOwner,
  Executant: executant,
  Ref: ref,
  OldReference: req.Ref,
@@ -235,11 +241,12 @@ func ItemRestored(r *provider.RestoreRecycleItemResponse, req *provider.RestoreR
 }
 
 // FileVersionRestored converts the response to an event
-func FileVersionRestored(r *provider.RestoreFileVersionResponse, req *provider.RestoreFileVersionRequest, executant *user.UserId) events.FileVersionRestored {
+func FileVersionRestored(r *provider.RestoreFileVersionResponse, req *provider.RestoreFileVersionRequest, spaceOwner, executant *user.UserId) events.FileVersionRestored {
  return events.FileVersionRestored{
- Executant: executant,
- Ref: req.Ref,
- Key: req.Key,
+ SpaceOwner: spaceOwner,
+ Executant: executant,
+ Ref: req.Ref,
+ Key: req.Key,
  }
 }
 

internal/grpc/interceptors/eventsmiddleware/events.go

@@ -40,6 +40,7 @@ import (
  "github.com/cs3org/reva/v2/pkg/events"
  "github.com/cs3org/reva/v2/pkg/events/server"
  "github.com/cs3org/reva/v2/pkg/rgrpc"
+ "github.com/cs3org/reva/v2/pkg/storagespace"
  "github.com/cs3org/reva/v2/pkg/utils"
  "github.com/go-micro/plugins/v4/events/natsjs"
 )
@@ -63,11 +64,22 @@ func NewUnary(m map[string]interface{}) (grpc.UnaryServerInterceptor, int, error
  }
 
  interceptor := func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+ // Register a channel in the context to receive the space owner id from the handler(s) further down the stack
+ var ownerID *user.UserId
+ sendOwnerChan := make(chan *user.UserId)
+ ctx = storagespace.ContextRegisterSendOwnerChan(ctx, sendOwnerChan)
+
  res, err := handler(ctx, req)
  if err != nil {
  return res, err
  }
 
+ // Read the space owner id from the channel
+ select {
+ case ownerID = <-sendOwnerChan:
+ default:
+ }
+
  var executantID *user.UserId
  u, ok := revactx.ContextGetUser(ctx)
  if ok {
@@ -120,31 +132,31 @@ func NewUnary(m map[string]interface{}) (grpc.UnaryServerInterceptor, int, error
  }
  case *provider.CreateContainerResponse:
  if isSuccess(v) {
- ev = ContainerCreated(v, req.(*provider.CreateContainerRequest), executantID)
+ ev = ContainerCreated(v, req.(*provider.CreateContainerRequest), ownerID, executantID)
  }
  case *provider.InitiateFileDownloadResponse:
  if isSuccess(v) {
  ev = FileDownloaded(v, req.(*provider.InitiateFileDownloadRequest), executantID)
  }
  case *provider.DeleteResponse:
  if isSuccess(v) {
- ev = ItemTrashed(v, req.(*provider.DeleteRequest), executantID)
+ ev = ItemTrashed(v, req.(*provider.DeleteRequest), ownerID, executantID)
  }
  case *provider.MoveResponse:
  if isSuccess(v) {
- ev = ItemMoved(v, req.(*provider.MoveRequest), executantID)
+ ev = ItemMoved(v, req.(*provider.MoveRequest), ownerID, executantID)
  }
  case *provider.PurgeRecycleResponse:
  if isSuccess(v) {
  ev = ItemPurged(v, req.(*provider.PurgeRecycleRequest), executantID)
  }
  case *provider.RestoreRecycleItemResponse:
  if isSuccess(v) {
- ev = ItemRestored(v, req.(*provider.RestoreRecycleItemRequest), executantID)
+ ev = ItemRestored(v, req.(*provider.RestoreRecycleItemRequest), ownerID, executantID)
  }
  case *provider.RestoreFileVersionResponse:
  if isSuccess(v) {
- ev = FileVersionRestored(v, req.(*provider.RestoreFileVersionRequest), executantID)
+ ev = FileVersionRestored(v, req.(*provider.RestoreFileVersionRequest), ownerID, executantID)
  }
  case *provider.CreateStorageSpaceResponse:
  if isSuccess(v) && v.StorageSpace != nil { // TODO: Why are there CreateStorageSpaceResponses with nil StorageSpace?
@@ -172,7 +184,7 @@ func NewUnary(m map[string]interface{}) (grpc.UnaryServerInterceptor, int, error
  }
  case *provider.TouchFileResponse:
  if isSuccess(v) {
- ev = FileTouched(v, req.(*provider.TouchFileRequest), executantID)
+ ev = FileTouched(v, req.(*provider.TouchFileRequest), ownerID, executantID)
  }
  }
 

pkg/events/files.go

@@ -27,9 +27,10 @@ import (
 
 // ContainerCreated is emitted when a directory has been created
 type ContainerCreated struct {
- Executant *user.UserId
- Ref *provider.Reference
- Owner *user.UserId
+ SpaceOwner *user.UserId
+ Executant *user.UserId
+ Ref *provider.Reference
+ Owner *user.UserId
 }
 
 // Unmarshal to fulfill umarshaller interface
@@ -41,9 +42,10 @@ func (ContainerCreated) Unmarshal(v []byte) (interface{}, error) {
 
 // FileUploaded is emitted when a file is uploaded
 type FileUploaded struct {
- Executant *user.UserId
- Ref *provider.Reference
- Owner *user.UserId
+ SpaceOwner *user.UserId
+ Executant *user.UserId
+ Ref *provider.Reference
+ Owner *user.UserId
 }
 
 // Unmarshal to fulfill umarshaller interface
@@ -55,8 +57,9 @@ func (FileUploaded) Unmarshal(v []byte) (interface{}, error) {
 
 // FileTouched is emitted when a file is uploaded
 type FileTouched struct {
- Executant *user.UserId
- Ref *provider.Reference
+ SpaceOwner *user.UserId
+ Executant *user.UserId
+ Ref *provider.Reference
 }
 
 // Unmarshal to fulfill umarshaller interface
@@ -82,10 +85,11 @@ func (FileDownloaded) Unmarshal(v []byte) (interface{}, error) {
 
 // ItemTrashed is emitted when a file or folder is trashed
 type ItemTrashed struct {
- Executant *user.UserId
- ID *provider.ResourceId
- Ref *provider.Reference
- Owner *user.UserId
+ SpaceOwner *user.UserId
+ Executant *user.UserId
+ ID *provider.ResourceId
+ Ref *provider.Reference
+ Owner *user.UserId
 }
 
 // Unmarshal to fulfill umarshaller interface
@@ -97,6 +101,7 @@ func (ItemTrashed) Unmarshal(v []byte) (interface{}, error) {
 
 // ItemMoved is emitted when a file or folder is moved
 type ItemMoved struct {
+ SpaceOwner *user.UserId
  Executant *user.UserId
  Ref *provider.Reference
  Owner *user.UserId
@@ -127,6 +132,7 @@ func (ItemPurged) Unmarshal(v []byte) (interface{}, error) {
 
 // ItemRestored is emitted when a file or folder is restored from trashbin
 type ItemRestored struct {
+ SpaceOwner *user.UserId
  Executant *user.UserId
  ID *provider.ResourceId
  Ref *provider.Reference
@@ -144,10 +150,11 @@ func (ItemRestored) Unmarshal(v []byte) (interface{}, error) {
 
 // FileVersionRestored is emitted when a file version is restored
 type FileVersionRestored struct {
- Executant *user.UserId
- Ref *provider.Reference
- Owner *user.UserId
- Key string
+ SpaceOwner *user.UserId
+ Executant *user.UserId
+ Ref *provider.Reference
+ Owner *user.UserId
+ Key string
 }
 
 // Unmarshal to fulfill umarshaller interface

pkg/rhttp/datatx/datatx.go

@@ -36,15 +36,16 @@ type DataTX interface {
 }
 
 // EmitFileUploadedEvent is a helper function which publishes a FileUploaded event
-func EmitFileUploadedEvent(owner *userv1beta1.UserId, ref *provider.Reference, publisher events.Publisher) error {
+func EmitFileUploadedEvent(spaceOwnerOrManager, executant *userv1beta1.UserId, ref *provider.Reference, publisher events.Publisher) error {
  if ref == nil || publisher == nil {
  return nil
  }
 
  uploadedEv := events.FileUploaded{
- Owner: owner,
- Executant: owner,
- Ref: ref,
+ SpaceOwner: spaceOwnerOrManager,
+ Owner: spaceOwnerOrManager,
+ Executant: executant,
+ Ref: ref,
  }
 
  return events.Publish(publisher, uploadedEv)

pkg/rhttp/datatx/manager/simple/simple.go

@@ -93,9 +93,9 @@ func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
  defer r.Body.Close()
 
  ref := &provider.Reference{Path: fn}
- info, err := fs.Upload(ctx, ref, r.Body, func(owner *userpb.UserId, ref *provider.Reference) {
+ info, err := fs.Upload(ctx, ref, r.Body, func(spaceOwner, owner *userpb.UserId, ref *provider.Reference) {
  datatx.InvalidateCache(owner, ref, m.statCache)
- if err := datatx.EmitFileUploadedEvent(owner, ref, m.publisher); err != nil {
+ if err := datatx.EmitFileUploadedEvent(spaceOwner, owner, ref, m.publisher); err != nil {
  sublog.Error().Err(err).Msg("failed to publish FileUploaded event")
  }
  })

pkg/rhttp/datatx/manager/spaces/spaces.go

@@ -107,9 +107,9 @@ func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
  Path: fn,
  }
  var info provider.ResourceInfo
- info, err = fs.Upload(ctx, ref, r.Body, func(owner *userpb.UserId, ref *provider.Reference) {
+ info, err = fs.Upload(ctx, ref, r.Body, func(spaceOwner, owner *userpb.UserId, ref *provider.Reference) {
  datatx.InvalidateCache(owner, ref, m.statCache)
- if err := datatx.EmitFileUploadedEvent(owner, ref, m.publisher); err != nil {
+ if err := datatx.EmitFileUploadedEvent(spaceOwner, owner, ref, m.publisher); err != nil {
  sublog.Error().Err(err).Msg("failed to publish FileUploaded event")
  }
  })

pkg/rhttp/datatx/manager/tus/tus.go

@@ -109,6 +109,9 @@ func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
  for {
  ev := <-handler.CompleteUploads
  info := ev.Upload
+ spaceOwner := &userv1beta1.UserId{
+ OpaqueId: info.Storage["SpaceOwnerOrManager"],
+ }
  owner := &userv1beta1.UserId{
  Idp: info.Storage["Idp"],
  OpaqueId: info.Storage["UserId"],
@@ -123,7 +126,7 @@ func (m *manager) Handler(fs storage.FS) (http.Handler, error) {
  }
  datatx.InvalidateCache(owner, ref, m.statCache)
  if m.publisher != nil {
- if err := datatx.EmitFileUploadedEvent(owner, ref, m.publisher); err != nil {
+ if err := datatx.EmitFileUploadedEvent(spaceOwner, owner, ref, m.publisher); err != nil {
  appctx.GetLogger(context.Background()).Error().Err(err).Msg("failed to publish FileUploaded event")
  }
  }

pkg/storage/fs/owncloudsql/upload.go

@@ -103,7 +103,10 @@ func (fs *owncloudsqlfs) Upload(ctx context.Context, ref *provider.Reference, r
  if !ok {
  return provider.ResourceInfo{}, errtypes.PreconditionFailed("error getting user from uploadinfo context")
  }
- uff(owner.Id, uploadRef)
+ // spaces support in localfs needs to be revisited:
+ // * info.Storage["SpaceRoot"] is never set
+ // * there is no space owner or manager that could be passed to the UploadFinishedFunc
+ uff(owner.Id, owner.Id, uploadRef)
  }
 
  ri := provider.ResourceInfo{

pkg/storage/storage.go

@@ -31,7 +31,7 @@ import (
 )
 
 // UploadFinishedFunc is a callback function used in storage drivers to indicate that an upload has finished
-type UploadFinishedFunc func(owner *userpb.UserId, ref *provider.Reference)
+type UploadFinishedFunc func(spaceOwner, owner *userpb.UserId, ref *provider.Reference)
 
 // FS is the interface to implement access to the storage.
 type FS interface {