mirror of
https://github.com/opencloud-eu/opencloud.git
synced 2026-01-05 19:59:37 -06:00
300 lines
9.6 KiB
Go
300 lines
9.6 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,
|
|
Key: itemID.GetOpaqueId(),
|
|
})
|
|
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(),
|
|
}
|
|
}
|