Ralf Haferkamp
ff6674f97b
reva tries to avoid copying proto messages now. This changed some calls to take pointers now (mainly e.g. FormatResourceId())
299 lines
9.5 KiB
Go
299 lines
9.5 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"path/filepath"
|
|
"reflect"
|
|
|
|
gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
|
|
group "github.com/cs3org/go-cs3apis/cs3/identity/group/v1beta1"
|
|
user "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
|
|
rpc "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
|
|
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
|
|
"go.opentelemetry.io/otel/trace"
|
|
|
|
"github.com/cs3org/reva/v2/pkg/events"
|
|
"github.com/cs3org/reva/v2/pkg/rgrpc/todo/pool"
|
|
"github.com/cs3org/reva/v2/pkg/storagespace"
|
|
"github.com/cs3org/reva/v2/pkg/utils"
|
|
|
|
"github.com/owncloud/ocis/v2/ocis-pkg/log"
|
|
"github.com/owncloud/ocis/v2/services/clientlog/pkg/config"
|
|
)
|
|
|
|
// ClientlogService is the service responsible for user activities
|
|
type ClientlogService struct {
|
|
log log.Logger
|
|
cfg *config.Config
|
|
gatewaySelector pool.Selectable[gateway.GatewayAPIClient]
|
|
registeredEvents map[string]events.Unmarshaller // ?
|
|
tp trace.TracerProvider
|
|
tracer trace.Tracer
|
|
publisher events.Publisher
|
|
ch <-chan events.Event
|
|
}
|
|
|
|
// NewClientlogService returns a clientlog service
|
|
func NewClientlogService(opts ...Option) (*ClientlogService, error) {
|
|
o := &Options{}
|
|
for _, opt := range opts {
|
|
opt(o)
|
|
}
|
|
|
|
if o.Stream == nil {
|
|
return nil, fmt.Errorf("need non nil stream (%v) to work properly", o.Stream)
|
|
}
|
|
|
|
ch, err := events.Consume(o.Stream, "clientlog", o.RegisteredEvents...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
cl := &ClientlogService{
|
|
log: o.Logger,
|
|
cfg: o.Config,
|
|
gatewaySelector: o.GatewaySelector,
|
|
registeredEvents: make(map[string]events.Unmarshaller),
|
|
tp: o.TraceProvider,
|
|
tracer: o.TraceProvider.Tracer("github.com/owncloud/ocis/services/clientlog/pkg/service"),
|
|
publisher: o.Stream,
|
|
ch: ch,
|
|
}
|
|
|
|
for _, e := range o.RegisteredEvents {
|
|
typ := reflect.TypeOf(e)
|
|
cl.registeredEvents[typ.String()] = e
|
|
}
|
|
|
|
return cl, nil
|
|
}
|
|
|
|
// Run runs the service
|
|
func (cl *ClientlogService) Run() error {
|
|
for event := range cl.ch {
|
|
cl.processEvent(event)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (cl *ClientlogService) processEvent(event events.Event) {
|
|
gwc, err := cl.gatewaySelector.Next()
|
|
if err != nil {
|
|
cl.log.Error().Err(err).Interface("event", event).Msg("error getting gateway client")
|
|
return
|
|
}
|
|
|
|
ctx, err := utils.GetServiceUserContextWithContext(context.Background(), gwc, cl.cfg.ServiceAccount.ServiceAccountID, cl.cfg.ServiceAccount.ServiceAccountSecret)
|
|
if err != nil {
|
|
cl.log.Error().Err(err).Interface("event", event).Msg("error authenticating service user")
|
|
return
|
|
}
|
|
|
|
var (
|
|
users []string
|
|
evType string
|
|
data interface{}
|
|
)
|
|
|
|
fileEv := func(typ string, ref *provider.Reference) {
|
|
evType = typ
|
|
users, data, err = processFileEvent(ctx, ref, gwc, event.InitiatorID)
|
|
}
|
|
|
|
shareEv := func(typ string, ref *provider.Reference, uid *user.UserId, gid *group.GroupId) {
|
|
evType = typ
|
|
users, data, err = processShareEvent(ctx, ref, gwc, event.InitiatorID, uid, gid)
|
|
}
|
|
|
|
switch e := event.Event.(type) {
|
|
default:
|
|
err = errors.New("unhandled event")
|
|
case events.UploadReady:
|
|
if e.Failed {
|
|
// we don't inform about failed uploads yet
|
|
return
|
|
}
|
|
fileEv("postprocessing-finished", e.FileRef)
|
|
case events.ItemTrashed:
|
|
evType = "item-trashed"
|
|
users, data, err = processItemTrashedEvent(ctx, e.Ref, gwc, event.InitiatorID, e.ID)
|
|
case events.ItemRestored:
|
|
fileEv("item-restored", e.Ref)
|
|
case events.ContainerCreated:
|
|
fileEv("folder-created", e.Ref)
|
|
case events.ItemMoved:
|
|
// we send a dedicated event in case the item was only renamed
|
|
if isRename(e.OldReference, e.Ref) {
|
|
fileEv("item-renamed", e.Ref)
|
|
} else {
|
|
fileEv("item-moved", e.Ref)
|
|
}
|
|
case events.FileLocked:
|
|
fileEv("file-locked", e.Ref)
|
|
case events.FileUnlocked:
|
|
fileEv("file-unlocked", e.Ref)
|
|
case events.FileTouched:
|
|
fileEv("file-touched", e.Ref)
|
|
case events.SpaceShared:
|
|
r, _ := storagespace.ParseReference(e.ID.GetOpaqueId())
|
|
shareEv("space-member-added", &r, e.GranteeUserID, e.GranteeGroupID)
|
|
case events.SpaceShareUpdated:
|
|
r, _ := storagespace.ParseReference(e.ID.GetOpaqueId())
|
|
shareEv("space-share-updated", &r, e.GranteeUserID, e.GranteeGroupID)
|
|
case events.SpaceUnshared:
|
|
r, _ := storagespace.ParseReference(e.ID.GetOpaqueId())
|
|
shareEv("space-member-removed", &r, e.GranteeUserID, e.GranteeGroupID)
|
|
case events.ShareCreated:
|
|
shareEv("share-created", &provider.Reference{ResourceId: e.ItemID}, e.GranteeUserID, e.GranteeGroupID)
|
|
case events.ShareUpdated:
|
|
shareEv("share-updated", &provider.Reference{ResourceId: e.ItemID}, e.GranteeUserID, e.GranteeGroupID)
|
|
case events.ShareRemoved:
|
|
shareEv("share-removed", &provider.Reference{ResourceId: e.ItemID}, e.GranteeUserID, e.GranteeGroupID)
|
|
case events.LinkCreated:
|
|
fileEv("link-created", &provider.Reference{ResourceId: e.ItemID})
|
|
case events.LinkUpdated:
|
|
fileEv("link-updated", &provider.Reference{ResourceId: e.ItemID})
|
|
case events.LinkRemoved:
|
|
fileEv("link-removed", &provider.Reference{ResourceId: e.ItemID})
|
|
case events.BackchannelLogout:
|
|
evType, users, data = backchannelLogoutEvent(e)
|
|
}
|
|
|
|
if err != nil {
|
|
cl.log.Error().Err(err).Interface("event", event).Msg("error gathering members for event")
|
|
return
|
|
}
|
|
|
|
// II) instruct sse service to send the information
|
|
if err := cl.sendSSE(users, evType, data); err != nil {
|
|
cl.log.Error().Err(err).Interface("userIDs", users).Str("eventid", event.ID).Msg("failed to store event for user")
|
|
return
|
|
}
|
|
}
|
|
|
|
func (cl *ClientlogService) sendSSE(userIDs []string, evType string, data interface{}) error {
|
|
b, err := json.Marshal(data)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return events.Publish(context.Background(), cl.publisher, events.SendSSE{
|
|
UserIDs: userIDs,
|
|
Type: evType,
|
|
Message: b,
|
|
})
|
|
}
|
|
|
|
// process file related events
|
|
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
|
|
}
|
|
|
|
data := FileEvent{
|
|
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)
|
|
return users, data, err
|
|
}
|
|
|
|
// process share related events
|
|
func processShareEvent(ctx context.Context, ref *provider.Reference, gwc gateway.GatewayAPIClient, initiatorid string, shareeID *user.UserId, shareeGroupID *group.GroupId) ([]string, FileEvent, error) {
|
|
users, data, err := processFileEvent(ctx, ref, gwc, initiatorid)
|
|
if err != nil {
|
|
return users, data, err
|
|
}
|
|
|
|
return addShareeData(ctx, gwc, data, users, shareeID, shareeGroupID)
|
|
}
|
|
|
|
// custom logic for item trashed event
|
|
func processItemTrashedEvent(ctx context.Context, ref *provider.Reference, gwc gateway.GatewayAPIClient, initiatorid string, itemID *provider.ResourceId) ([]string, FileEvent, error) {
|
|
resp, err := gwc.ListRecycle(ctx, &provider.ListRecycleRequest{
|
|
Ref: ref,
|
|
})
|
|
if err != nil {
|
|
return nil, FileEvent{}, err
|
|
}
|
|
if resp.GetStatus().GetCode() != rpc.Code_CODE_OK {
|
|
return nil, FileEvent{}, fmt.Errorf("error listing recycle: %s", resp.GetStatus().GetMessage())
|
|
}
|
|
|
|
for _, item := range resp.GetRecycleItems() {
|
|
if item.GetKey() == itemID.GetOpaqueId() {
|
|
|
|
data := FileEvent{
|
|
ItemID: storagespace.FormatResourceID(itemID),
|
|
// TODO: check with web if parentID is needed
|
|
// ParentItemID: storagespace.FormatResourceID(*item.GetRef().GetResourceId()),
|
|
SpaceID: storagespace.FormatStorageID(itemID.GetStorageId(), itemID.GetSpaceId()),
|
|
InitiatorID: initiatorid,
|
|
}
|
|
|
|
users, err := utils.GetSpaceMembers(ctx, itemID.GetSpaceId(), gwc, utils.ViewerRole)
|
|
return users, data, err
|
|
}
|
|
}
|
|
return nil, FileEvent{}, errors.New("item not found in recycle bin")
|
|
}
|
|
|
|
// adds share related data to the FileEvent
|
|
func addShareeData(ctx context.Context, gwc gateway.GatewayAPIClient, fe FileEvent, users []string, shareeID *user.UserId, shareeGroupID *group.GroupId) ([]string, FileEvent, error) {
|
|
us, err := resolveID(ctx, gwc, shareeID, shareeGroupID)
|
|
if err != nil {
|
|
return users, fe, err
|
|
}
|
|
|
|
fe.AffectedUserIDs = us
|
|
|
|
// TODO: this list can get long. Should we add a limit? If yes, how big?
|
|
for _, u := range us {
|
|
users = appendUnique(users, u)
|
|
}
|
|
return users, fe, nil
|
|
}
|
|
|
|
// returns the user or the members of the affected group
|
|
func resolveID(ctx context.Context, gwc gateway.GatewayAPIClient, uid *user.UserId, gid *group.GroupId) ([]string, error) {
|
|
if uid != nil {
|
|
return []string{uid.GetOpaqueId()}, nil
|
|
}
|
|
return utils.GetGroupMembers(ctx, gid.GetOpaqueId(), gwc)
|
|
}
|
|
|
|
// returns users or append(users, user)
|
|
func appendUnique(users []string, user string) []string {
|
|
for _, u := range users {
|
|
if u == user {
|
|
return users
|
|
}
|
|
}
|
|
return append(users, user)
|
|
}
|
|
|
|
// returns true if this is just a rename
|
|
func isRename(o, n *provider.Reference) bool {
|
|
// if resourceids are different we assume it is a move
|
|
if !utils.ResourceIDEqual(o.GetResourceId(), n.GetResourceId()) {
|
|
return false
|
|
}
|
|
return filepath.Base(o.GetPath()) != filepath.Base(n.GetPath())
|
|
}
|
|
|
|
func backchannelLogoutEvent(e events.BackchannelLogout) (string, []string, BackchannelLogout) {
|
|
return "backchannel-logout", []string{e.Executant.GetOpaqueId()}, BackchannelLogout{
|
|
UserID: e.Executant.GetOpaqueId(),
|
|
Timestamp: e.Timestamp.String(),
|
|
}
|
|
}
|