sevenhill/QSfera
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

enhancement(graph): resolve review feedback such as, use fmt for error wrapping, make use of store auth options, use a dedicated jetstream package

Browse Source
This commit is contained in:
fschade
2025-10-08 16:45:16 +02:00
committed by Ralf Haferkamp
parent 2ca88e66da
commit c0244fc977
9 changed files with 991 additions and 794 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/graph/.mockery.yaml
+1 -1
View File
@@ -45,7 +45,7 @@ packages:
Client:
config:
filename: ldapclient.go
github.com/nats-io/nats.go:
github.com/nats-io/nats.go/jetstream:
config:
dir: mocks
interfaces:
services/graph/mocks/key_value.go
+939 -748
View File
File diff suppressed because it is too large Load Diff
services/graph/mocks/key_value_entry.go
+7 -7
View File
@@ -7,7 +7,7 @@ package mocks
import (
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
mock "github.com/stretchr/testify/mock"
)
@@ -215,18 +215,18 @@ func (_c *KeyValueEntry_Key_Call) RunAndReturn(run func() string) *KeyValueEntry
}
// Operation provides a mock function for the type KeyValueEntry
func (_mock *KeyValueEntry) Operation() nats.KeyValueOp {
func (_mock *KeyValueEntry) Operation() jetstream.KeyValueOp {
ret := _mock.Called()
if len(ret) == 0 {
panic("no return value specified for Operation")
}
var r0 nats.KeyValueOp
if returnFunc, ok := ret.Get(0).(func() nats.KeyValueOp); ok {
var r0 jetstream.KeyValueOp
if returnFunc, ok := ret.Get(0).(func() jetstream.KeyValueOp); ok {
r0 = returnFunc()
} else {
r0 = ret.Get(0).(nats.KeyValueOp)
r0 = ret.Get(0).(jetstream.KeyValueOp)
}
return r0
}
@@ -248,12 +248,12 @@ func (_c *KeyValueEntry_Operation_Call) Run(run func()) *KeyValueEntry_Operation
return _c
}
func (_c *KeyValueEntry_Operation_Call) Return(keyValueOp nats.KeyValueOp) *KeyValueEntry_Operation_Call {
func (_c *KeyValueEntry_Operation_Call) Return(keyValueOp jetstream.KeyValueOp) *KeyValueEntry_Operation_Call {
_c.Call.Return(keyValueOp)
return _c
}
func (_c *KeyValueEntry_Operation_Call) RunAndReturn(run func() nats.KeyValueOp) *KeyValueEntry_Operation_Call {
func (_c *KeyValueEntry_Operation_Call) RunAndReturn(run func() jetstream.KeyValueOp) *KeyValueEntry_Operation_Call {
_c.Call.Return(run)
return _c
}
services/graph/pkg/command/server.go
+13 -10
View File
@@ -6,6 +6,10 @@ import (
"os/signal"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"github.com/pkg/errors"
"github.com/urfave/cli/v2"
"github.com/opencloud-eu/opencloud/pkg/config/configlog"
"github.com/opencloud-eu/opencloud/pkg/runner"
"github.com/opencloud-eu/opencloud/pkg/tracing"
@@ -16,8 +20,6 @@ import (
"github.com/opencloud-eu/opencloud/services/graph/pkg/metrics"
"github.com/opencloud-eu/opencloud/services/graph/pkg/server/debug"
"github.com/opencloud-eu/opencloud/services/graph/pkg/server/http"
"github.com/pkg/errors"
"github.com/urfave/cli/v2"
)
// Server is the entrypoint for the server command.
@@ -48,29 +50,30 @@ func Server(cfg *config.Config) *cli.Command {
//Connect to NATS servers
natsOptions := nats.Options{
Servers: cfg.Store.Nodes,
Servers: cfg.Store.Nodes,
User: cfg.Store.AuthUsername,
Password: cfg.Store.AuthPassword,
}
conn, err := natsOptions.Connect()
if err != nil {
return err
}
js, err := conn.JetStream()
js, err := jetstream.New(conn)
if err != nil {
return err
}
kv, err := js.KeyValue(cfg.Store.Database)
kv, err := js.KeyValue(ctx, cfg.Store.Database)
if err != nil {
if !errors.Is(err, nats.ErrBucketNotFound) {
return errors.Wrapf(err, "Failed to get bucket (%s)", cfg.Store.Database)
if !errors.Is(err, jetstream.ErrBucketNotFound) {
return fmt.Errorf("failed to get bucket (%s): %w", cfg.Store.Database, err)
}
kv, err = js.CreateKeyValue(&nats.KeyValueConfig{
kv, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{
Bucket: cfg.Store.Database,
})
if err != nil {
return errors.Wrapf(err, "Failed to create bucket (%s)", cfg.Store.Database)
return fmt.Errorf("failed to create bucket (%s): %w", cfg.Store.Database, err)
}
}
if err != nil {
services/graph/pkg/server/http/option.go
+6 -5
View File
@@ -3,12 +3,13 @@ package http
import (
"context"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"github.com/urfave/cli/v2"
"go.opentelemetry.io/otel/trace"
"github.com/opencloud-eu/opencloud/pkg/log"
"github.com/opencloud-eu/opencloud/services/graph/pkg/config"
"github.com/opencloud-eu/opencloud/services/graph/pkg/metrics"
"github.com/urfave/cli/v2"
"go.opentelemetry.io/otel/trace"
)
// Option defines a single option function.
@@ -23,7 +24,7 @@ type Options struct {
Flags []cli.Flag
Namespace string
TraceProvider trace.TracerProvider
NatsKeyValue nats.KeyValue
NatsKeyValue jetstream.KeyValue
}
// newOptions initializes the available default options.
@@ -87,7 +88,7 @@ func TraceProvider(val trace.TracerProvider) Option {
}
// NatsKeyValue provides a function to set the NatsKeyValue option.
func NatsKeyValue(val nats.KeyValue) Option {
func NatsKeyValue(val jetstream.KeyValue) Option {
return func(o *Options) {
o.NatsKeyValue = val
}
services/graph/pkg/service/v0/graph.go
+2 -2
View File
@@ -12,7 +12,7 @@ import (
storageprovider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
"github.com/go-chi/chi/v5"
"github.com/jellydator/ttlcache/v3"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"go-micro.dev/v4/client"
"go.opentelemetry.io/otel/trace"
"google.golang.org/protobuf/types/known/emptypb"
@@ -68,7 +68,7 @@ type Graph struct {
keycloakClient keycloak.Client
historyClient ehsvc.EventHistoryService
traceProvider trace.TracerProvider
natskv nats.KeyValue
natskv jetstream.KeyValue
}
// ServeHTTP implements the Service interface.
services/graph/pkg/service/v0/option.go
+3 -3
View File
@@ -5,7 +5,7 @@ import (
"net/http"
gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"github.com/opencloud-eu/reva/v2/pkg/events"
"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
"go.opentelemetry.io/otel/trace"
@@ -44,7 +44,7 @@ type Options struct {
KeycloakClient keycloak.Client
EventHistoryClient ehsvc.EventHistoryService
TraceProvider trace.TracerProvider
NatsKeyValue nats.KeyValue
NatsKeyValue jetstream.KeyValue
}
// newOptions initializes the available default options.
@@ -115,7 +115,7 @@ func WithIdentityEducationBackend(val identity.EducationBackend) Option {
}
// WithNatsKeyValue provides a function to set the NatsKeyValue option.
func WithNatsKeyValue(val nats.KeyValue) Option {
func WithNatsKeyValue(val jetstream.KeyValue) Option {
return func(o *Options) {
o.NatsKeyValue = val
}
services/graph/pkg/service/v0/users.go
+9 -8
View File
@@ -23,8 +23,13 @@ import (
"github.com/go-chi/chi/v5"
"github.com/go-chi/render"
"github.com/google/uuid"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
libregraph "github.com/opencloud-eu/libre-graph-api-go"
revactx "github.com/opencloud-eu/reva/v2/pkg/ctx"
"github.com/opencloud-eu/reva/v2/pkg/events"
"github.com/opencloud-eu/reva/v2/pkg/rgrpc/status"
"github.com/opencloud-eu/reva/v2/pkg/utils"
settingsmsg "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/messages/settings/v0"
settingssvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/settings/v0"
"github.com/opencloud-eu/opencloud/services/graph/pkg/errorcode"
@@ -33,10 +38,6 @@ import (
"github.com/opencloud-eu/opencloud/services/graph/pkg/userstate"
ocsettingssvc "github.com/opencloud-eu/opencloud/services/settings/pkg/service/v0"
"github.com/opencloud-eu/opencloud/services/settings/pkg/store/defaults"
revactx "github.com/opencloud-eu/reva/v2/pkg/ctx"
"github.com/opencloud-eu/reva/v2/pkg/events"
"github.com/opencloud-eu/reva/v2/pkg/rgrpc/status"
"github.com/opencloud-eu/reva/v2/pkg/utils"
)
// GetMe implements the Service interface.
@@ -1157,9 +1158,9 @@ func (g Graph) getUserStateFromNatsKeyValue(ctx context.Context, userID string)
return userstate.UserState{}, errors.New("nats connection or user state key value store not configured")
}
entry, err := g.natskv.Get(userID)
entry, err := g.natskv.Get(ctx, userID)
if err != nil {
if errors.Is(err, nats.ErrKeyNotFound) {
if errors.Is(err, jetstream.ErrKeyNotFound) {
logger.Debug().Str("userid", userID).Msg("no user state found in nats key value store")
return userstate.UserState{
UserId: userID,
@@ -1200,7 +1201,7 @@ func (g Graph) setUserStateToNatsKeyValue(ctx context.Context, userID string, us
return err
}
if _, err := g.natskv.Put(userID, data); err != nil {
if _, err := g.natskv.Put(ctx, userID, data); err != nil {
logger.Error().Err(err).Str("userid", userID).Msg("error putting user state to nats key value store")
return err
}
services/graph/pkg/service/v0/users_test.go
+11 -10
View File
@@ -16,11 +16,10 @@ import (
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
typesv1beta1 "github.com/cs3org/go-cs3apis/cs3/types/v1beta1"
"github.com/go-chi/chi/v5"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
libregraph "github.com/opencloud-eu/libre-graph-api-go"
"github.com/opencloud-eu/opencloud/services/graph/pkg/userstate"
revactx "github.com/opencloud-eu/reva/v2/pkg/ctx"
"github.com/opencloud-eu/reva/v2/pkg/rgrpc/status"
"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
@@ -29,6 +28,8 @@ import (
"go-micro.dev/v4/client"
"google.golang.org/grpc"
"github.com/opencloud-eu/opencloud/services/graph/pkg/userstate"
"github.com/opencloud-eu/opencloud/pkg/shared"
settingsmsg "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/messages/settings/v0"
settings "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/settings/v0"
@@ -973,7 +974,7 @@ var _ = Describe("Users", func() {
lu.SetId(currentUser.Id.OpaqueId)
identityBackend.On("GetUser", mock.Anything, mock.Anything, mock.Anything).Return(&lu, nil)
natsKeyValueMock.EXPECT().Get(mock.Anything).RunAndReturn(func(key string) (nats.KeyValueEntry, error) {
natsKeyValueMock.EXPECT().Get(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, key string) (jetstream.KeyValueEntry, error) {
byteRep, _ := json.Marshal(userstate.UserState{
UserId: lu.GetId(),
State: userstate.UserStateSoftDeleted,
@@ -988,7 +989,7 @@ var _ = Describe("Users", func() {
return kve, nil
}).Once()
natsKeyValueMock.EXPECT().Put(mock.Anything, mock.Anything).RunAndReturn(func(key string, val []byte) (uint64, error) {
natsKeyValueMock.EXPECT().Put(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, key string, val []byte) (uint64, error) {
return 1, nil
}).Once()
@@ -1028,7 +1029,7 @@ var _ = Describe("Users", func() {
},
}, nil)
natsKeyValueMock.EXPECT().Get(mock.Anything).RunAndReturn(func(key string) (nats.KeyValueEntry, error) {
natsKeyValueMock.EXPECT().Get(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, key string) (jetstream.KeyValueEntry, error) {
byteRep, _ := json.Marshal(userstate.UserState{
UserId: lu.GetId(),
State: userstate.UserStateSoftDeleted,
@@ -1043,7 +1044,7 @@ var _ = Describe("Users", func() {
return kve, nil
}).Once()
natsKeyValueMock.EXPECT().Put(mock.Anything, mock.Anything).RunAndReturn(func(key string, val []byte) (uint64, error) {
natsKeyValueMock.EXPECT().Put(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, key string, val []byte) (uint64, error) {
return 1, nil
}).Once()
@@ -1108,7 +1109,7 @@ var _ = Describe("Users", func() {
},
}, nil)
natsKeyValueMock.EXPECT().Get(mock.Anything).RunAndReturn(func(key string) (nats.KeyValueEntry, error) {
natsKeyValueMock.EXPECT().Get(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, key string) (jetstream.KeyValueEntry, error) {
byteRep, _ := json.Marshal(userstate.UserState{
UserId: lu.GetId(),
State: userstate.UserStateSoftDeleted,
@@ -1123,7 +1124,7 @@ var _ = Describe("Users", func() {
return kve, nil
}).Once()
natsKeyValueMock.EXPECT().Put(mock.Anything, mock.Anything).RunAndReturn(func(key string, val []byte) (uint64, error) {
natsKeyValueMock.EXPECT().Put(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, key string, val []byte) (uint64, error) {
return 1, nil
}).Once()
r := httptest.NewRequest(http.MethodDelete, "/graph/v1.0/users/{userid}", nil)
@@ -1165,7 +1166,7 @@ var _ = Describe("Users", func() {
},
}, nil)
natsKeyValueMock.EXPECT().Get(mock.Anything).RunAndReturn(func(key string) (nats.KeyValueEntry, error) {
natsKeyValueMock.EXPECT().Get(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, key string) (jetstream.KeyValueEntry, error) {
byteRep, _ := json.Marshal(userstate.UserState{
UserId: lu.GetId(),
State: userstate.UserStateSoftDeleted,
@@ -1180,7 +1181,7 @@ var _ = Describe("Users", func() {
return kve, nil
}).Once()
natsKeyValueMock.EXPECT().Put(mock.Anything, mock.Anything).RunAndReturn(func(key string, val []byte) (uint64, error) {
natsKeyValueMock.EXPECT().Put(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, key string, val []byte) (uint64, error) {
return 1, nil
}).Once()