jkoberg
@@ -0,0 +1,65 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
|
||||
"github.com/cs3org/reva/v2/pkg/events"
|
||||
"github.com/cs3org/reva/v2/pkg/rgrpc/todo/pool"
|
||||
"github.com/owncloud/ocis/v2/ocis-pkg/log"
|
||||
"github.com/owncloud/ocis/v2/services/clientlog/pkg/config"
|
||||
"go.opentelemetry.io/otel/trace"
|
||||
)
|
||||
|
||||
// Option for the clientlog service
|
||||
type Option func(*Options)
|
||||
|
||||
// Options for the clientlog service
|
||||
type Options struct {
|
||||
Logger log.Logger
|
||||
Stream events.Stream
|
||||
Config *config.Config
|
||||
GatewaySelector pool.Selectable[gateway.GatewayAPIClient]
|
||||
RegisteredEvents []events.Unmarshaller
|
||||
TraceProvider trace.TracerProvider
|
||||
}
|
||||
|
||||
// Logger configures a logger for the clientlog service
|
||||
func Logger(log log.Logger) Option {
|
||||
return func(o *Options) {
|
||||
o.Logger = log
|
||||
}
|
||||
}
|
||||
|
||||
// Stream configures an event stream for the clientlog service
|
||||
func Stream(s events.Stream) Option {
|
||||
return func(o *Options) {
|
||||
o.Stream = s
|
||||
}
|
||||
}
|
||||
|
||||
// Config adds the config for the clientlog service
|
||||
func Config(c *config.Config) Option {
|
||||
return func(o *Options) {
|
||||
o.Config = c
|
||||
}
|
||||
}
|
||||
|
||||
// GatewaySelector adds a grpc client selector for the gateway service
|
||||
func GatewaySelector(gatewaySelector pool.Selectable[gateway.GatewayAPIClient]) Option {
|
||||
return func(o *Options) {
|
||||
o.GatewaySelector = gatewaySelector
|
||||
}
|
||||
}
|
||||
|
||||
// RegisteredEvents registers the events the service should listen to
|
||||
func RegisteredEvents(e []events.Unmarshaller) Option {
|
||||
return func(o *Options) {
|
||||
o.RegisteredEvents = e
|
||||
}
|
||||
}
|
||||
|
||||
// TraceProvider adds a tracer provider for the clientlog service
|
||||
func TraceProvider(tp trace.TracerProvider) Option {
|
||||
return func(o *Options) {
|
||||
o.TraceProvider = tp
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,134 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"reflect"
|
||||
|
||||
gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
|
||||
"github.com/cs3org/reva/v2/pkg/events"
|
||||
"github.com/cs3org/reva/v2/pkg/rgrpc/todo/pool"
|
||||
"github.com/cs3org/reva/v2/pkg/utils"
|
||||
"github.com/owncloud/ocis/v2/ocis-pkg/log"
|
||||
"github.com/owncloud/ocis/v2/services/clientlog/pkg/config"
|
||||
"go.opentelemetry.io/otel/trace"
|
||||
)
|
||||
|
||||
// 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 gatway client")
|
||||
return
|
||||
}
|
||||
|
||||
ctx, err := utils.GetServiceUserContext(cl.cfg.ServiceAccount.ServiceAccountID, gwc, cl.cfg.ServiceAccount.ServiceAccountSecret)
|
||||
if err != nil {
|
||||
cl.log.Error().Err(err).Interface("event", event).Msg("error authenticating service user")
|
||||
return
|
||||
}
|
||||
|
||||
var users []string
|
||||
switch e := event.Event.(type) {
|
||||
default:
|
||||
err = errors.New("unhandled event")
|
||||
case events.UploadReady:
|
||||
info, err := utils.GetResource(ctx, e.FileRef, gwc)
|
||||
if err != nil {
|
||||
cl.log.Error().Err(err).Interface("event", event).Msg("error getting resource")
|
||||
return
|
||||
}
|
||||
|
||||
users, err = utils.GetSpaceMembers(ctx, info.GetSpace().GetId().GetOpaqueId(), gwc, utils.ViewerRole)
|
||||
if err != nil {
|
||||
cl.log.Error().Err(err).Interface("event", event).Msg("error getting space members")
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
cl.log.Info().Err(err).Interface("event", event).Msg("error gathering members for event")
|
||||
return
|
||||
}
|
||||
|
||||
// II) instruct sse service to send the information
|
||||
for _, id := range users {
|
||||
if err := cl.sendSSE(id, event); err != nil {
|
||||
cl.log.Error().Err(err).Str("userID", id).Str("eventid", event.ID).Msg("failed to store event for user")
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (cl *ClientlogService) sendSSE(userid string, event events.Event) error {
|
||||
// TODO: convert event
|
||||
ev := event
|
||||
|
||||
b, err := json.Marshal(ev)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return events.Publish(context.Background(), cl.publisher, events.SendSSE{
|
||||
UserID: userid,
|
||||
Type: "clientlog-notification",
|
||||
Message: b,
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
package service_test
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/owncloud/ocis/v2/ocis-pkg/registry"
|
||||
mRegistry "go-micro.dev/v4/registry"
|
||||
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
func init() {
|
||||
registry.Configure("memory")
|
||||
r := registry.GetRegistry()
|
||||
service := registry.BuildGRPCService("com.owncloud.api.gateway", "", "", "")
|
||||
service.Nodes = []*mRegistry.Node{{
|
||||
Address: "any",
|
||||
}}
|
||||
|
||||
_ = r.Register(service)
|
||||
}
|
||||
func TestSearch(t *testing.T) {
|
||||
RegisterFailHandler(Fail)
|
||||
RunSpecs(t, "Userlog service Suite")
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
package service_test
|
||||
|
||||
// TODO: TEST!
|
||||
Reference in New Issue
Block a user