NOISSUE - Implementation of Group removal (absmach#161)

* add: group delete flow Signed-off-by: Arvindh <[email protected]> * sync with master Signed-off-by: Arvindh <[email protected]> * improved grpc error handling Signed-off-by: Arvindh <[email protected]> * gofumpt -ed Signed-off-by: Arvindh <[email protected]> * changed database unassign parent group id Signed-off-by: Arvindh <[email protected]> * seperate event for delete group Signed-off-by: Arvindh <[email protected]> * change group event name Signed-off-by: Arvindh <[email protected]> * update channel remove events Signed-off-by: Arvindh <[email protected]> * fix: channels event Signed-off-by: Arvindh <[email protected]> * add: sdk, sdk_tet, cli Signed-off-by: Arvindh <[email protected]> * fix: sdk mock Signed-off-by: Arvindh <[email protected]> * mockery Signed-off-by: Arvindh <[email protected]> * convert to inline case Signed-off-by: Arvindh <[email protected]> * change in group delete flow Signed-off-by: Arvindh <[email protected]> * change return error in delete Signed-off-by: Arvindh <[email protected]> * add: openapi Signed-off-by: Arvindh <[email protected]> * rename events: from channel delete to channel remove Signed-off-by: Arvindh <[email protected]> --------- Signed-off-by: Arvindh <[email protected]>
shree-0403 · Dec 19, 2023 · 6486550 · 6486550
1 parent 0fa80ba
commit 6486550
Show file tree

Hide file tree

Showing 35 changed files with 440 additions and 74 deletions.
diff --git a/api/openapi/things.yml b/api/openapi/things.yml
@@ -466,6 +466,26 @@ paths:
           description: Missing or invalid content type.
         "500":
           $ref: "#/components/responses/ServiceError"
+    delete:
+      summary: Delete channel for given channel id.
+      description: |
+        Delete channel remove given channel id from repo
+        and removes all the policies related to channel.
+      tags:
+        - Channels
+      parameters:
+        - $ref: "#/components/parameters/chanID"
+      security:
+        - bearerAuth: []
+      responses:
+        "204":
+          $ref: "#/components/responses/ChannelDeleteRes"
+        "401":
+          description: Missing or invalid access token provided.
+        "403":
+          description: Unauthorized access to thing id.
+        "500":
+          $ref: "#/components/responses/ServiceError"
 
   /channels/{chanID}/enable:
     post:
@@ -1751,6 +1771,9 @@ components:
           schema:
             $ref: "#/components/schemas/Channel"
 
+    ChannelDeleteRes:
+      description: Channel Deleted.
+
     ChannelPageRes:
       description: Data retrieved.
       content:

diff --git a/auth/api/grpc/client.go b/auth/api/grpc/client.go
@@ -175,7 +175,7 @@ func (client grpcClient) Issue(ctx context.Context, req *magistrala.IssueReq, _
 
 	res, err := client.issue(ctx, issueReq{userID: req.GetUserId(), domainID: req.GetDomainId(), keyType: auth.KeyType(req.Type)})
 	if err != nil {
-		return nil, decodeError(err)
+		return &magistrala.Token{}, decodeError(err)
 	}
 	return res.(*magistrala.Token), nil
 }
@@ -195,7 +195,7 @@ func (client grpcClient) Refresh(ctx context.Context, req *magistrala.RefreshReq
 
 	res, err := client.refresh(ctx, refreshReq{refreshToken: req.GetRefreshToken(), domainID: req.GetDomainId()})
 	if err != nil {
-		return nil, decodeError(err)
+		return &magistrala.Token{}, decodeError(err)
 	}
 	return res.(*magistrala.Token), nil
 }
@@ -215,7 +215,7 @@ func (client grpcClient) Identify(ctx context.Context, token *magistrala.Identit
 
 	res, err := client.identify(ctx, identityReq{token: token.GetToken()})
 	if err != nil {
-		return nil, err
+		return &magistrala.IdentityRes{}, decodeError(err)
 	}
 	ir := res.(identityRes)
 	return &magistrala.IdentityRes{Id: ir.id, UserId: ir.userID, DomainId: ir.domainID}, nil
@@ -246,11 +246,11 @@ func (client grpcClient) Authorize(ctx context.Context, req *magistrala.Authoriz
 		Object:      req.GetObject(),
 	})
 	if err != nil {
-		return &magistrala.AuthorizeRes{}, err
+		return &magistrala.AuthorizeRes{}, decodeError(err)
 	}
 
 	ar := res.(authorizeRes)
-	return &magistrala.AuthorizeRes{Authorized: ar.authorized, Id: ar.id}, err
+	return &magistrala.AuthorizeRes{Authorized: ar.authorized, Id: ar.id}, nil
 }
 
 func decodeAuthorizeResponse(_ context.Context, grpcRes interface{}) (interface{}, error) {
@@ -288,11 +288,11 @@ func (client grpcClient) AddPolicy(ctx context.Context, in *magistrala.AddPolicy
 		Object:      in.GetObject(),
 	})
 	if err != nil {
-		return &magistrala.AddPolicyRes{}, err
+		return &magistrala.AddPolicyRes{}, decodeError(err)
 	}
 
 	apr := res.(addPolicyRes)
-	return &magistrala.AddPolicyRes{Authorized: apr.authorized}, err
+	return &magistrala.AddPolicyRes{Authorized: apr.authorized}, nil
 }
 
 func decodeAddPolicyResponse(_ context.Context, grpcRes interface{}) (interface{}, error) {
@@ -337,11 +337,11 @@ func (client grpcClient) AddPolicies(ctx context.Context, in *magistrala.AddPoli
 
 	res, err := client.addPolicies(ctx, r)
 	if err != nil {
-		return &magistrala.AddPoliciesRes{}, err
+		return &magistrala.AddPoliciesRes{}, decodeError(err)
 	}
 
 	apr := res.(addPoliciesRes)
-	return &magistrala.AddPoliciesRes{Authorized: apr.authorized}, err
+	return &magistrala.AddPoliciesRes{Authorized: apr.authorized}, nil
 }
 
 func decodeAddPoliciesResponse(_ context.Context, grpcRes interface{}) (interface{}, error) {
@@ -386,11 +386,11 @@ func (client grpcClient) DeletePolicy(ctx context.Context, in *magistrala.Delete
 		Object:      in.GetObject(),
 	})
 	if err != nil {
-		return &magistrala.DeletePolicyRes{}, err
+		return &magistrala.DeletePolicyRes{}, decodeError(err)
 	}
 
 	dpr := res.(deletePolicyRes)
-	return &magistrala.DeletePolicyRes{Deleted: dpr.deleted}, err
+	return &magistrala.DeletePolicyRes{Deleted: dpr.deleted}, nil
 }
 
 func decodeDeletePolicyResponse(_ context.Context, grpcRes interface{}) (interface{}, error) {
@@ -435,11 +435,11 @@ func (client grpcClient) DeletePolicies(ctx context.Context, in *magistrala.Dele
 	}
 	res, err := client.deletePolicies(ctx, r)
 	if err != nil {
-		return &magistrala.DeletePoliciesRes{}, err
+		return &magistrala.DeletePoliciesRes{}, decodeError(err)
 	}
 
 	dpr := res.(deletePoliciesRes)
-	return &magistrala.DeletePoliciesRes{Deleted: dpr.deleted}, err
+	return &magistrala.DeletePoliciesRes{Deleted: dpr.deleted}, nil
 }
 
 func decodeDeletePoliciesResponse(_ context.Context, grpcRes interface{}) (interface{}, error) {
@@ -482,11 +482,11 @@ func (client grpcClient) ListObjects(ctx context.Context, in *magistrala.ListObj
 		Object:      in.GetObject(),
 	})
 	if err != nil {
-		return &magistrala.ListObjectsRes{}, err
+		return &magistrala.ListObjectsRes{}, decodeError(err)
 	}
 
 	lpr := res.(listObjectsRes)
-	return &magistrala.ListObjectsRes{Policies: lpr.policies}, err
+	return &magistrala.ListObjectsRes{Policies: lpr.policies}, nil
 }
 
 func decodeListObjectsResponse(_ context.Context, grpcRes interface{}) (interface{}, error) {
@@ -521,11 +521,11 @@ func (client grpcClient) ListAllObjects(ctx context.Context, in *magistrala.List
 		Object:      in.GetObject(),
 	})
 	if err != nil {
-		return &magistrala.ListObjectsRes{}, err
+		return &magistrala.ListObjectsRes{}, decodeError(err)
 	}
 
 	lpr := res.(listObjectsRes)
-	return &magistrala.ListObjectsRes{Policies: lpr.policies}, err
+	return &magistrala.ListObjectsRes{Policies: lpr.policies}, nil
 }
 
 func (client grpcClient) CountObjects(ctx context.Context, in *magistrala.CountObjectsReq, opts ...grpc.CallOption) (*magistrala.CountObjectsRes, error) {
@@ -542,11 +542,11 @@ func (client grpcClient) CountObjects(ctx context.Context, in *magistrala.CountO
 		Object:      in.GetObject(),
 	})
 	if err != nil {
-		return &magistrala.CountObjectsRes{}, err
+		return &magistrala.CountObjectsRes{}, decodeError(err)
 	}
 
 	cp := res.(countObjectsRes)
-	return &magistrala.CountObjectsRes{Count: int64(cp.count)}, err
+	return &magistrala.CountObjectsRes{Count: int64(cp.count)}, nil
 }
 
 func decodeCountObjectsResponse(_ context.Context, grpcRes interface{}) (interface{}, error) {
@@ -582,11 +582,11 @@ func (client grpcClient) ListSubjects(ctx context.Context, in *magistrala.ListSu
 		NextPageToken: in.GetNextPageToken(),
 	})
 	if err != nil {
-		return &magistrala.ListSubjectsRes{}, err
+		return &magistrala.ListSubjectsRes{}, decodeError(err)
 	}
 
 	lpr := res.(listSubjectsRes)
-	return &magistrala.ListSubjectsRes{Policies: lpr.policies}, err
+	return &magistrala.ListSubjectsRes{Policies: lpr.policies}, nil
 }
 
 func decodeListSubjectsResponse(_ context.Context, grpcRes interface{}) (interface{}, error) {
@@ -621,11 +621,11 @@ func (client grpcClient) ListAllSubjects(ctx context.Context, in *magistrala.Lis
 		Object:      in.GetObject(),
 	})
 	if err != nil {
-		return &magistrala.ListSubjectsRes{}, err
+		return &magistrala.ListSubjectsRes{}, decodeError(err)
 	}
 
 	lpr := res.(listSubjectsRes)
-	return &magistrala.ListSubjectsRes{Policies: lpr.policies}, err
+	return &magistrala.ListSubjectsRes{Policies: lpr.policies}, nil
 }
 
 func (client grpcClient) CountSubjects(ctx context.Context, in *magistrala.CountSubjectsReq, opts ...grpc.CallOption) (*magistrala.CountSubjectsRes, error) {
@@ -681,7 +681,7 @@ func (client grpcClient) ListPermissions(ctx context.Context, in *magistrala.Lis
 		FilterPermissions: in.GetFilterPermissions(),
 	})
 	if err != nil {
-		return &magistrala.ListPermissionsRes{}, err
+		return &magistrala.ListPermissionsRes{}, decodeError(err)
 	}
 
 	lp := res.(listPermissionsRes)

diff --git a/auth/api/grpc/requests.go b/auth/api/grpc/requests.go
@@ -94,18 +94,6 @@ type policyReq struct {
 }
 
 func (req policyReq) validate() error {
-	if req.Subject == "" || req.SubjectType == "" {
-		return apiutil.ErrMissingPolicySub
-	}
-
-	if req.Object == "" || req.ObjectType == "" {
-		return apiutil.ErrMissingPolicyObj
-	}
-
-	if req.Relation == "" && req.Permission == "" {
-		return apiutil.ErrMalformedPolicyRel
-	}
-
 	return nil
 }
 

diff --git a/auth/api/grpc/server.go b/auth/api/grpc/server.go
@@ -540,11 +540,13 @@ func encodeError(err error) error {
 		err == apiutil.ErrMalformedPolicyAct:
 		return status.Error(codes.InvalidArgument, err.Error())
 	case errors.Contains(err, errors.ErrAuthentication),
+		errors.Contains(err, svcerr.ErrAuthentication),
 		errors.Contains(err, auth.ErrKeyExpired),
 		err == apiutil.ErrMissingEmail,
 		err == apiutil.ErrBearerToken:
 		return status.Error(codes.Unauthenticated, err.Error())
 	case errors.Contains(err, errors.ErrAuthorization),
+		errors.Contains(err, svcerr.ErrAuthorization),
 		errors.Contains(err, errors.ErrDomainAuthorization):
 		return status.Error(codes.PermissionDenied, err.Error())
 	case errors.Contains(err, errors.ErrNotFound):

diff --git a/auth/service.go b/auth/service.go
@@ -153,7 +153,7 @@ func (svc service) Identify(ctx context.Context, token string) (Key, error) {
 		return Key{}, errors.Wrap(ErrAPIKeyExpired, err)
 	}
 	if err != nil {
-		return Key{}, errors.Wrap(errIdentify, err)
+		return Key{}, errors.Wrap(svcerr.ErrAuthentication, errors.Wrap(errIdentify, err))
 	}
 
 	switch key.Type {

diff --git a/bootstrap/events/consumer/streams.go b/bootstrap/events/consumer/streams.go
@@ -17,7 +17,7 @@ const (
 	thingRemove     = "thing.remove"
 	thingDisconnect = "policy.delete"
 
-	channelPrefix = "channel."
+	channelPrefix = "group."
 	channelUpdate = channelPrefix + "update"
 	channelRemove = channelPrefix + "remove"
 )
@@ -95,20 +95,8 @@ func decodeUpdateChannel(event map[string]interface{}) updateChannelEvent {
 }
 
 func decodeRemoveChannel(event map[string]interface{}) removeEvent {
-	status := read(event, "status", "")
-	st, err := clients.ToStatus(status)
-	if err != nil {
-		return removeEvent{}
-	}
-	switch st {
-	case clients.EnabledStatus:
-		return removeEvent{}
-	case clients.DisabledStatus:
-		return removeEvent{
-			id: read(event, "id", ""),
-		}
-	default:
-		return removeEvent{}
+	return removeEvent{
+		id: read(event, "id", ""),
 	}
 }
 

diff --git a/bootstrap/events/producer/events.go b/bootstrap/events/producer/events.go
@@ -25,7 +25,7 @@ const (
 	thingUpdateConnections = thingPrefix + "update_connections"
 	thingDisconnect        = thingPrefix + "disconnect"
 
-	channelPrefix        = "channel."
+	channelPrefix        = "group."
 	channelHandlerRemove = channelPrefix + "remove_handler"
 	channelUpdateHandler = channelPrefix + "update_handler"
 

diff --git a/bootstrap/events/producer/streams_test.go b/bootstrap/events/producer/streams_test.go
@@ -48,7 +48,7 @@ const (
 	thingUpdateConnections = thingPrefix + "update_connections"
 	thingDisconnect        = thingPrefix + "disconnect"
 
-	channelPrefix        = "channel."
+	channelPrefix        = "group."
 	channelHandlerRemove = channelPrefix + "remove_handler"
 	channelUpdateHandler = channelPrefix + "update_handler"
 

diff --git a/cli/channels.go b/cli/channels.go
@@ -81,6 +81,24 @@ var cmdChannels = []cobra.Command{
 			logJSON(c)
 		},
 	},
+	{
+		Use:   "delete <channel_id> <user_auth_token>",
+		Short: "Delete channel",
+		Long: "Delete channel by id.\n" +
+			"Usage:\n" +
+			"\tmagistrala-cli channels delete <channel_id> $USERTOKEN - delete the given channel ID\n",
+		Run: func(cmd *cobra.Command, args []string) {
+			if len(args) != 2 {
+				logUsage(cmd.Use)
+				return
+			}
+			if err := sdk.DeleteChannel(args[0], args[1]); err != nil {
+				logError(err)
+				return
+			}
+			logOK()
+		},
+	},
 	{
 		Use:   "update <channel_id> <JSON_string> <user_auth_token>",
 		Short: "Update channel",

diff --git a/cli/groups.go b/cli/groups.go
@@ -141,6 +141,24 @@ var cmdGroups = []cobra.Command{
 			logJSON(t)
 		},
 	},
+	{
+		Use:   "delete <group_id> <user_auth_token>",
+		Short: "Delete group",
+		Long: "Delete group by id.\n" +
+			"Usage:\n" +
+			"\tmagistrala-cli groups delete <group_id> $USERTOKEN - delete the given group ID\n",
+		Run: func(cmd *cobra.Command, args []string) {
+			if len(args) != 2 {
+				logUsage(cmd.Use)
+				return
+			}
+			if err := sdk.DeleteGroup(args[0], args[1]); err != nil {
+				logError(err)
+				return
+			}
+			logOK()
+		},
+	},
 	{
 		Use:   "assign user <relation> <user_ids> <group_id> <user_auth_token>",
 		Short: "Assign user",

diff --git a/cmd/things/main.go b/cmd/things/main.go
@@ -19,6 +19,7 @@ import (
 	redisclient "github.com/absmach/magistrala/internal/clients/redis"
 	mggroups "github.com/absmach/magistrala/internal/groups"
 	gapi "github.com/absmach/magistrala/internal/groups/api"
+	gevents "github.com/absmach/magistrala/internal/groups/events"
 	gpostgres "github.com/absmach/magistrala/internal/groups/postgres"
 	gtracing "github.com/absmach/magistrala/internal/groups/tracing"
 	"github.com/absmach/magistrala/internal/postgres"
@@ -58,6 +59,8 @@ const (
 	defDB              = "things"
 	defSvcHTTPPort     = "9000"
 	defSvcAuthGRPCPort = "7000"
+
+	streamID = "magistrala.things"
 )
 
 type config struct {
@@ -232,6 +235,11 @@ func newService(ctx context.Context, db *sqlx.DB, dbConfig pgclient.Config, auth
 		return nil, nil, err
 	}
 
+	gsvc, err = gevents.NewEventStoreMiddleware(ctx, gsvc, esURL, streamID)
+	if err != nil {
+		return nil, nil, err
+	}
+
 	csvc = ctracing.New(csvc, tracer)
 	csvc = api.LoggingMiddleware(csvc, logger)
 	counter, latency := internal.MakeMetrics(svcName, "api")