SMQ-2740 - Create events streams per service per action type

channels/events/streams.go

@@ -16,7 +16,19 @@
 	"github.com/go-chi/chi/v5/middleware"
 )
 
-const streamID = "supermq.channels"
+const (
+	supermqPrefix      = "supermq."
+	createStream       = supermqPrefix + channelCreate
+	updateStream       = supermqPrefix + channelUpdate
+	changeStatusStream = supermqPrefix + channelChangeStatus
+	removeStream       = supermqPrefix + channelRemove
+	viewStream         = supermqPrefix + channelView
+	listStream         = supermqPrefix + channelList
+	connectStream      = supermqPrefix + channelConnect
+	disconnectStream   = supermqPrefix + channelDisconnect
+	setParentStream    = supermqPrefix + channelSetParent
+	removeParentStream = supermqPrefix + channelRemoveParent
+)
 
 var _ channels.Service = (*eventStore)(nil)
 
@@ -29,7 +41,7 @@
 // NewEventStoreMiddleware returns wrapper around clients service that sends
 // events to event store.
 func NewEventStoreMiddleware(ctx context.Context, svc channels.Service, url string) (channels.Service, error) {
-	publisher, err := store.NewPublisher(ctx, url, streamID)
+	publisher, err := store.NewPublisher(ctx, url)
 	if err != nil {
 		return nil, err
 	}
@@ -55,7 +67,7 @@
 			Session:          session,
 			requestID:        middleware.GetReqID(ctx),
 		}
-		if err := es.Publish(ctx, event); err != nil {
+		if err := es.Publish(ctx, createStream, event); err != nil {
 			return chs, rps, err
 		}
 	}
@@ -89,7 +101,7 @@
 		requestID: middleware.GetReqID(ctx),
 	}
 
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, updateStream, event); err != nil {
 		return ch, err
 	}
 
@@ -107,7 +119,7 @@
 		Session:   session,
 		requestID: middleware.GetReqID(ctx),
 	}
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, viewStream, event); err != nil {
 		return chann, err
 	}
 
@@ -124,7 +136,7 @@
 		Session:      session,
 		requestID:    middleware.GetReqID(ctx),
 	}
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, listStream, event); err != nil {
 		return cp, err
 	}
 
@@ -142,7 +154,7 @@
 		Session:      session,
 		requestID:    middleware.GetReqID(ctx),
 	}
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, listStream, event); err != nil {
 		return cp, err
 	}
 
@@ -176,7 +188,7 @@
 		Session:   session,
 		requestID: middleware.GetReqID(ctx),
 	}
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, changeStatusStream, event); err != nil {
 		return ch, err
 	}
 
@@ -194,7 +206,7 @@
 		requestID: middleware.GetReqID(ctx),
 	}
 
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, removeStream, event); err != nil {
 		return err
 	}
 
@@ -214,7 +226,7 @@
 		requestID: middleware.GetReqID(ctx),
 	}
 
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, connectStream, event); err != nil {
 		return err
 	}
 
@@ -234,7 +246,7 @@
 		requestID: middleware.GetReqID(ctx),
 	}
 
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, disconnectStream, event); err != nil {
 		return err
 	}
 
@@ -253,7 +265,7 @@
 		requestID:     middleware.GetReqID(ctx),
 	}
 
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, setParentStream, event); err != nil {
 		return err
 	}
 
@@ -271,7 +283,7 @@
 		requestID: middleware.GetReqID(ctx),
 	}
 
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, removeParentStream, event); err != nil {
 		return err
 	}
 

clients/events/streams.go

@@ -15,7 +15,20 @@
 	"github.com/go-chi/chi/v5/middleware"
 )
 
-const streamID = "supermq.clients"
+const (
+	supermqPrefix      = "supermq."
+	createStream       = supermqPrefix + clientCreate
+	updateStream       = supermqPrefix + clientUpdate
+	changeStatusStream = supermqPrefix + clientChangeStatus
+	removeStream       = supermqPrefix + clientRemove
+	viewStream         = supermqPrefix + clientView
+	viewPermsStream    = supermqPrefix + clientViewPerms
+	listStream         = supermqPrefix + clientList
+	identifyStream     = supermqPrefix + clientIdentify
+	authorizeStream    = supermqPrefix + clientAuthorize
+	setParentStream    = supermqPrefix + clientSetParent
+	removeParentStream = supermqPrefix + clientRemoveParent
+)
 
 var _ clients.Service = (*eventStore)(nil)
 
@@ -28,7 +41,7 @@
 // NewEventStoreMiddleware returns wrapper around clients service that sends
 // events to event store.
 func NewEventStoreMiddleware(ctx context.Context, svc clients.Service, url string) (clients.Service, error) {
-	publisher, err := store.NewPublisher(ctx, url, streamID)
+	publisher, err := store.NewPublisher(ctx, url)
 	if err != nil {
 		return nil, err
 	}
@@ -54,7 +67,7 @@
 			Session:          session,
 			requestID:        middleware.GetReqID(ctx),
 		}
-		if err := es.Publish(ctx, event); err != nil {
+		if err := es.Publish(ctx, createStream, event); err != nil {
 			return clis, rps, err
 		}
 	}
@@ -97,7 +110,7 @@
 		requestID: middleware.GetReqID(ctx),
 	}
 
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, updateStream, event); err != nil {
 		return client, err
 	}
 
@@ -115,7 +128,7 @@
 		Session:   session,
 		requestID: middleware.GetReqID(ctx),
 	}
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, viewStream, event); err != nil {
 		return cli, err
 	}
 
@@ -132,7 +145,7 @@
 		session,
 		middleware.GetReqID(ctx),
 	}
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, listStream, event); err != nil {
 		return cp, err
 	}
 
@@ -150,7 +163,7 @@
 		session,
 		middleware.GetReqID(ctx),
 	}
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, listStream, event); err != nil {
 		return cp, err
 	}
 
@@ -184,7 +197,7 @@
 		Session:   session,
 		requestID: middleware.GetReqID(ctx),
 	}
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, changeStatusStream, event); err != nil {
 		return cli, err
 	}
 
@@ -202,7 +215,7 @@
 		requestID: middleware.GetReqID(ctx),
 	}
 
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, removeStream, event); err != nil {
 		return err
 	}
 
@@ -221,7 +234,7 @@
 		requestID:     middleware.GetReqID(ctx),
 	}
 
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, setParentStream, event); err != nil {
 		return err
 	}
 
@@ -239,7 +252,7 @@
 		requestID: middleware.GetReqID(ctx),
 	}
 
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, removeParentStream, event); err != nil {
 		return err
 	}
 

domains/events/streams.go

@@ -15,7 +15,22 @@
 	"github.com/go-chi/chi/v5/middleware"
 )
 
-const streamID = "supermq.domains"
+const (
+	supermqPrefix            = "supermq."
+	createStream             = supermqPrefix + domainCreate
+	retrieveStream           = supermqPrefix + domainRetrieve
+	retrieveStatusStream     = supermqPrefix + domainRetrieveStatus
+	updateStream             = supermqPrefix + domainUpdate
+	changeStatusStream       = supermqPrefix + domainPrefix + "change_status"
+	listStream               = supermqPrefix + domainList
+	userDeleteStream         = supermqPrefix + domainUserDelete
+	sendInvitationStream     = supermqPrefix + invitationSend
+	acceptInvitationStream   = supermqPrefix + invitationAccept
+	rejectInvitationStream   = supermqPrefix + invitationReject
+	listInvitationsStream    = supermqPrefix + invitationList
+	retrieveInvitationStream = supermqPrefix + invitationRetrieve
+	deleteInvitationStream   = supermqPrefix + invitationDelete
+)
 
 var _ domains.Service = (*eventStore)(nil)
 
@@ -28,7 +43,7 @@
 // NewEventStoreMiddleware returns wrapper around auth service that sends
 // events to event store.
 func NewEventStoreMiddleware(ctx context.Context, svc domains.Service, url string) (domains.Service, error) {
-	publisher, err := store.NewPublisher(ctx, url, streamID)
+	publisher, err := store.NewPublisher(ctx, url)
 	if err != nil {
 		return nil, err
 	}
@@ -55,7 +70,7 @@
 		requestID:        middleware.GetReqID(ctx),
 	}
 
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, createStream, event); err != nil {
 		return domain, rps, err
 	}
 
@@ -74,7 +89,7 @@
 		middleware.GetReqID(ctx),
 	}
 
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, retrieveStream, event); err != nil {
 		return domain, err
 	}
 
@@ -93,7 +108,7 @@
 		requestID: middleware.GetReqID(ctx),
 	}
 
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, updateStream, event); err != nil {
 		return domain, err
 	}
 
@@ -114,7 +129,7 @@
 		requestID: middleware.GetReqID(ctx),
 	}
 
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, changeStatusStream, event); err != nil {
 		return domain, err
 	}
 
@@ -135,7 +150,7 @@
 		requestID: middleware.GetReqID(ctx),
 	}
 
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, changeStatusStream, event); err != nil {
 		return domain, err
 	}
 
@@ -156,7 +171,7 @@
 		requestID: middleware.GetReqID(ctx),
 	}
 
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, changeStatusStream, event); err != nil {
 		return domain, err
 	}
 
@@ -178,7 +193,7 @@
 		requestID:  middleware.GetReqID(ctx),
 	}
 
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, listStream, event); err != nil {
 		return dp, err
 	}
 
@@ -195,7 +210,7 @@
 		session:    session,
 	}
 
-	return es.Publish(ctx, event)
+	return es.Publish(ctx, sendInvitationStream, event)
 }
 
 func (es *eventStore) ViewInvitation(ctx context.Context, session authn.Session, userID, domainID string) (domains.Invitation, error) {
@@ -212,7 +227,7 @@
 		session:       session,
 	}
 
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, retrieveInvitationStream, event); err != nil {
 		return invitation, err
 	}
 
@@ -230,7 +245,7 @@
 		session:            session,
 	}
 
-	if err := es.Publish(ctx, event); err != nil {
+	if err := es.Publish(ctx, listInvitationsStream, event); err != nil {
 		return ip, err
 	}
 
@@ -247,7 +262,7 @@
 		session:  session,
 	}
 
-	return es.Publish(ctx, event)
+	return es.Publish(ctx, acceptInvitationStream, event)
 }
 
 func (es *eventStore) RejectInvitation(ctx context.Context, session authn.Session, domainID string) error {
@@ -260,7 +275,7 @@
 		session:  session,
 	}
 
-	return es.Publish(ctx, event)
+	return es.Publish(ctx, rejectInvitationStream, event)
 }
 
 func (es *eventStore) DeleteInvitation(ctx context.Context, session authn.Session, inviteeUserID, domainID string) error {
@@ -274,5 +289,5 @@
 		session:       session,
 	}
 
-	return es.Publish(ctx, event)
+	return es.Publish(ctx, deleteInvitationStream, event)
 }

groups/events/events.go

@@ -12,7 +12,7 @@ import (
 	"github.com/absmach/supermq/pkg/roles"
 )
 
-var (
+const (
 	groupPrefix                  = "group."
 	groupCreate                  = groupPrefix + "create"
 	groupUpdate                  = groupPrefix + "update"