MG-2358 - Add Domain to Events (#2510)

Signed-off-by: nyagamunene <stevenyaga2014@gmail.com>
Signed-off-by: Felix Gateru <felix.gateru@gmail.com>
Co-authored-by: Felix Gateru <felix.gateru@gmail.com>

apidocs/openapi/journal.yml

@@ -197,7 +197,7 @@ components:
 
     entity_type:
       name: entityType
-      description: Type of entity, e.g. user, group, client, etc.entityType
+      description: Type of entity, e.g. group, client, channel.
       in: path
       schema:
         type: string
@@ -220,7 +220,7 @@ components:
 
     id:
       name: id
-      description: Unique identifier for an entity, e.g. group, channel or thing.  Used together with entity_type.
+      description: Unique identifier for an entity, e.g. group, channel or client. Used together with entity_type.
       in: path
       schema:
         type: string

channels/events/events.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/absmach/supermq/channels"
+	"github.com/absmach/supermq/pkg/authn"
 	"github.com/absmach/supermq/pkg/connections"
 	"github.com/absmach/supermq/pkg/events"
 	"github.com/absmach/supermq/pkg/roles"
@@ -40,6 +41,7 @@ var (
 type createChannelEvent struct {
 	channels.Channel
 	rolesProvisioned []roles.RoleProvision
+	authn.Session
 }
 
 func (cce createChannelEvent) Encode() (map[string]interface{}, error) {
@@ -49,6 +51,10 @@ func (cce createChannelEvent) Encode() (map[string]interface{}, error) {
 		"roles_provisioned": cce.rolesProvisioned,
 		"status":            cce.Status.String(),
 		"created_at":        cce.CreatedAt,
+		"domain":            cce.DomainID,
+		"user_id":           cce.UserID,
+		"token_type":        cce.Type.String(),
+		"super_admin":       cce.SuperAdmin,
 	}
 
 	if cce.Name != "" {
@@ -57,9 +63,6 @@ func (cce createChannelEvent) Encode() (map[string]interface{}, error) {
 	if len(cce.Tags) > 0 {
 		val["tags"] = cce.Tags
 	}
-	if cce.Domain != "" {
-		val["domain"] = cce.Domain
-	}
 	if cce.Metadata != nil {
 		val["metadata"] = cce.Metadata
 	}
@@ -70,13 +73,18 @@ func (cce createChannelEvent) Encode() (map[string]interface{}, error) {
 type updateChannelEvent struct {
 	channels.Channel
 	operation string
+	authn.Session
 }
 
 func (uce updateChannelEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation":  channelUpdate,
-		"updated_at": uce.UpdatedAt,
-		"updated_by": uce.UpdatedBy,
+		"operation":   channelUpdate,
+		"updated_at":  uce.UpdatedAt,
+		"updated_by":  uce.UpdatedBy,
+		"domain":      uce.DomainID,
+		"user_id":     uce.UserID,
+		"token_type":  uce.Type.String(),
+		"super_admin": uce.SuperAdmin,
 	}
 	if uce.operation != "" {
 		val["operation"] = channelUpdate + "_" + uce.operation
@@ -91,9 +99,6 @@ func (uce updateChannelEvent) Encode() (map[string]interface{}, error) {
 	if len(uce.Tags) > 0 {
 		val["tags"] = uce.Tags
 	}
-	if uce.Domain != "" {
-		val["domain"] = uce.Domain
-	}
 	if uce.Metadata != nil {
 		val["metadata"] = uce.Metadata
 	}
@@ -112,26 +117,36 @@ type changeStatusChannelEvent struct {
 	status    string
 	updatedAt time.Time
 	updatedBy string
+	authn.Session
 }
 
 func (rce changeStatusChannelEvent) Encode() (map[string]interface{}, error) {
 	return map[string]interface{}{
-		"operation":  channelChangeStatus,
-		"id":         rce.id,
-		"status":     rce.status,
-		"updated_at": rce.updatedAt,
-		"updated_by": rce.updatedBy,
+		"operation":   channelChangeStatus,
+		"id":          rce.id,
+		"status":      rce.status,
+		"updated_at":  rce.updatedAt,
+		"updated_by":  rce.updatedBy,
+		"domain":      rce.DomainID,
+		"user_id":     rce.UserID,
+		"token_type":  rce.Type.String(),
+		"super_admin": rce.SuperAdmin,
 	}, nil
 }
 
 type viewChannelEvent struct {
 	channels.Channel
+	authn.Session
 }
 
 func (vce viewChannelEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation": channelView,
-		"id":        vce.ID,
+		"operation":   channelView,
+		"id":          vce.ID,
+		"domain":      vce.DomainID,
+		"user_id":     vce.UserID,
+		"token_type":  vce.Type.String(),
+		"super_admin": vce.SuperAdmin,
 	}
 
 	if vce.Name != "" {
@@ -140,9 +155,6 @@ func (vce viewChannelEvent) Encode() (map[string]interface{}, error) {
 	if len(vce.Tags) > 0 {
 		val["tags"] = vce.Tags
 	}
-	if vce.Domain != "" {
-		val["domain"] = vce.Domain
-	}
 	if vce.Metadata != nil {
 		val["metadata"] = vce.Metadata
 	}
@@ -164,14 +176,19 @@ func (vce viewChannelEvent) Encode() (map[string]interface{}, error) {
 
 type listChannelEvent struct {
 	channels.PageMetadata
+	authn.Session
 }
 
 func (lce listChannelEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation": channelList,
-		"total":     lce.Total,
-		"offset":    lce.Offset,
-		"limit":     lce.Limit,
+		"operation":   channelList,
+		"total":       lce.Total,
+		"offset":      lce.Offset,
+		"limit":       lce.Limit,
+		"domain":      lce.DomainID,
+		"user_id":     lce.UserID,
+		"token_type":  lce.Type.String(),
+		"super_admin": lce.SuperAdmin,
 	}
 
 	if lce.Name != "" {
@@ -186,9 +203,6 @@ func (lce listChannelEvent) Encode() (map[string]interface{}, error) {
 	if lce.Metadata != nil {
 		val["metadata"] = lce.Metadata
 	}
-	if lce.Domain != "" {
-		val["domain"] = lce.Domain
-	}
 	if lce.Tag != "" {
 		val["tag"] = lce.Tag
 	}
@@ -208,15 +222,20 @@ func (lce listChannelEvent) Encode() (map[string]interface{}, error) {
 type listChannelByClientEvent struct {
 	clientID string
 	channels.PageMetadata
+	authn.Session
 }
 
 func (lcte listChannelByClientEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation": channelList,
-		"client_id": lcte.clientID,
-		"total":     lcte.Total,
-		"offset":    lcte.Offset,
-		"limit":     lcte.Limit,
+		"operation":   channelList,
+		"client_id":   lcte.clientID,
+		"total":       lcte.Total,
+		"offset":      lcte.Offset,
+		"limit":       lcte.Limit,
+		"domain":      lcte.DomainID,
+		"user_id":     lcte.UserID,
+		"token_type":  lcte.Type.String(),
+		"super_admin": lcte.SuperAdmin,
 	}
 
 	if lcte.Name != "" {
@@ -231,9 +250,6 @@ func (lcte listChannelByClientEvent) Encode() (map[string]interface{}, error) {
 	if lcte.Metadata != nil {
 		val["metadata"] = lcte.Metadata
 	}
-	if lcte.Domain != "" {
-		val["domain"] = lcte.Domain
-	}
 	if lcte.Tag != "" {
 		val["tag"] = lcte.Tag
 	}
@@ -252,12 +268,17 @@ func (lcte listChannelByClientEvent) Encode() (map[string]interface{}, error) {
 
 type removeChannelEvent struct {
 	id string
+	authn.Session
 }
 
 func (dce removeChannelEvent) Encode() (map[string]interface{}, error) {
 	return map[string]interface{}{
-		"operation": channelRemove,
-		"id":        dce.id,
+		"operation":   channelRemove,
+		"id":          dce.id,
+		"domain":      dce.DomainID,
+		"user_id":     dce.UserID,
+		"token_type":  dce.Type.String(),
+		"super_admin": dce.SuperAdmin,
 	}, nil
 }
 
@@ -265,6 +286,7 @@ type connectEvent struct {
 	chIDs []string
 	thIDs []string
 	types []connections.ConnType
+	authn.Session
 }
 
 func (ce connectEvent) Encode() (map[string]interface{}, error) {
@@ -273,6 +295,10 @@ func (ce connectEvent) Encode() (map[string]interface{}, error) {
 		"client_ids":  ce.thIDs,
 		"channel_ids": ce.chIDs,
 		"types":       ce.types,
+		"domain":      ce.DomainID,
+		"user_id":     ce.UserID,
+		"token_type":  ce.Type.String(),
+		"super_admin": ce.SuperAdmin,
 	}, nil
 }
 
@@ -280,6 +306,7 @@ type disconnectEvent struct {
 	chIDs []string
 	thIDs []string
 	types []connections.ConnType
+	authn.Session
 }
 
 func (de disconnectEvent) Encode() (map[string]interface{}, error) {
@@ -288,12 +315,17 @@ func (de disconnectEvent) Encode() (map[string]interface{}, error) {
 		"client_ids":  de.thIDs,
 		"channel_ids": de.chIDs,
 		"types":       de.types,
+		"domain":      de.DomainID,
+		"user_id":     de.UserID,
+		"token_type":  de.Type.String(),
+		"super_admin": de.SuperAdmin,
 	}, nil
 }
 
 type setParentGroupEvent struct {
 	id            string
 	parentGroupID string
+	authn.Session
 }
 
 func (spge setParentGroupEvent) Encode() (map[string]interface{}, error) {
@@ -301,16 +333,25 @@ func (spge setParentGroupEvent) Encode() (map[string]interface{}, error) {
 		"operation":       channelSetParent,
 		"id":              spge.id,
 		"parent_group_id": spge.parentGroupID,
+		"domain":          spge.DomainID,
+		"user_id":         spge.UserID,
+		"token_type":      spge.Type.String(),
+		"super_admin":     spge.SuperAdmin,
 	}, nil
 }
 
 type removeParentGroupEvent struct {
 	id string
+	authn.Session
 }
 
 func (rpge removeParentGroupEvent) Encode() (map[string]interface{}, error) {
 	return map[string]interface{}{
-		"operation": channelRemoveParent,
-		"id":        rpge.id,
+		"operation":   channelRemoveParent,
+		"id":          rpge.id,
+		"domain":      rpge.DomainID,
+		"user_id":     rpge.UserID,
+		"token_type":  rpge.Type.String(),
+		"super_admin": rpge.SuperAdmin,
 	}, nil
 }

channels/events/streams.go

@@ -51,6 +51,7 @@ func (es *eventStore) CreateChannels(ctx context.Context, session authn.Session,
 		event := createChannelEvent{
 			Channel:          ch,
 			rolesProvisioned: rps,
+			Session:          session,
 		}
 		if err := es.Publish(ctx, event); err != nil {
 			return chs, rps, err
@@ -66,7 +67,7 @@ func (es *eventStore) UpdateChannel(ctx context.Context, session authn.Session,
 		return chann, err
 	}
 
-	return es.update(ctx, "", chann)
+	return es.update(ctx, "", session, chann)
 }
 
 func (es *eventStore) UpdateChannelTags(ctx context.Context, session authn.Session, ch channels.Channel) (channels.Channel, error) {
@@ -75,12 +76,14 @@ func (es *eventStore) UpdateChannelTags(ctx context.Context, session authn.Sessi
 		return chann, err
 	}
 
-	return es.update(ctx, "tags", chann)
+	return es.update(ctx, "tags", session, chann)
 }
 
-func (es *eventStore) update(ctx context.Context, operation string, ch channels.Channel) (channels.Channel, error) {
+func (es *eventStore) update(ctx context.Context, operation string, session authn.Session, ch channels.Channel) (channels.Channel, error) {
 	event := updateChannelEvent{
-		ch, operation,
+		Channel:   ch,
+		operation: operation,
+		Session:   session,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -97,7 +100,8 @@ func (es *eventStore) ViewChannel(ctx context.Context, session authn.Session, id
 	}
 
 	event := viewChannelEvent{
-		chann,
+		Channel: chann,
+		Session: session,
 	}
 	if err := es.Publish(ctx, event); err != nil {
 		return chann, err
@@ -112,7 +116,8 @@ func (es *eventStore) ListChannels(ctx context.Context, session authn.Session, p
 		return cp, err
 	}
 	event := listChannelEvent{
-		pm,
+		PageMetadata: pm,
+		Session:      session,
 	}
 	if err := es.Publish(ctx, event); err != nil {
 		return cp, err
@@ -127,8 +132,9 @@ func (es *eventStore) ListChannelsByClient(ctx context.Context, session authn.Se
 		return cp, err
 	}
 	event := listChannelByClientEvent{
-		clientID,
-		pm,
+		clientID:     clientID,
+		PageMetadata: pm,
+		Session:      session,
 	}
 	if err := es.Publish(ctx, event); err != nil {
 		return cp, err
@@ -143,7 +149,7 @@ func (es *eventStore) EnableChannel(ctx context.Context, session authn.Session,
 		return cli, err
 	}
 
-	return es.changeStatus(ctx, cli)
+	return es.changeStatus(ctx, session, cli)
 }
 
 func (es *eventStore) DisableChannel(ctx context.Context, session authn.Session, id string) (channels.Channel, error) {
@@ -152,15 +158,16 @@ func (es *eventStore) DisableChannel(ctx context.Context, session authn.Session,
 		return cli, err
 	}
 
-	return es.changeStatus(ctx, cli)
+	return es.changeStatus(ctx, session, cli)
 }
 
-func (es *eventStore) changeStatus(ctx context.Context, ch channels.Channel) (channels.Channel, error) {
+func (es *eventStore) changeStatus(ctx context.Context, session authn.Session, ch channels.Channel) (channels.Channel, error) {
 	event := changeStatusChannelEvent{
 		id:        ch.ID,
 		updatedAt: ch.UpdatedAt,
 		updatedBy: ch.UpdatedBy,
 		status:    ch.Status.String(),
+		Session:   session,
 	}
 	if err := es.Publish(ctx, event); err != nil {
 		return ch, err
@@ -174,7 +181,10 @@ func (es *eventStore) RemoveChannel(ctx context.Context, session authn.Session,
 		return err
 	}
 
-	event := removeChannelEvent{id}
+	event := removeChannelEvent{
+		id:      id,
+		Session: session,
+	}
 
 	if err := es.Publish(ctx, event); err != nil {
 		return err
@@ -188,7 +198,12 @@ func (es *eventStore) Connect(ctx context.Context, session authn.Session, chIDs,
 		return err
 	}
 
-	event := connectEvent{chIDs, thIDs, connTypes}
+	event := connectEvent{
+		chIDs:   chIDs,
+		thIDs:   thIDs,
+		types:   connTypes,
+		Session: session,
+	}
 
 	if err := es.Publish(ctx, event); err != nil {
 		return err
@@ -202,7 +217,12 @@ func (es *eventStore) Disconnect(ctx context.Context, session authn.Session, chI
 		return err
 	}
 
-	event := disconnectEvent{chIDs, thIDs, connTypes}
+	event := disconnectEvent{
+		chIDs:   chIDs,
+		thIDs:   thIDs,
+		types:   connTypes,
+		Session: session,
+	}
 
 	if err := es.Publish(ctx, event); err != nil {
 		return err
@@ -216,7 +236,11 @@ func (es *eventStore) SetParentGroup(ctx context.Context, session authn.Session,
 		return err
 	}
 
-	event := setParentGroupEvent{parentGroupID: parentGroupID, id: id}
+	event := setParentGroupEvent{
+		parentGroupID: parentGroupID,
+		id:            id,
+		Session:       session,
+	}
 
 	if err := es.Publish(ctx, event); err != nil {
 		return err
@@ -230,7 +254,10 @@ func (es *eventStore) RemoveParentGroup(ctx context.Context, session authn.Sessi
 		return err
 	}
 
-	event := removeParentGroupEvent{id: id}
+	event := removeParentGroupEvent{
+		id:      id,
+		Session: session,
+	}
 
 	if err := es.Publish(ctx, event); err != nil {
 		return err

channels/middleware/authorization.go

@@ -85,7 +85,7 @@ func (am *authorizationMiddleware) CreateChannels(ctx context.Context, session a
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainChannelsScope,
@@ -126,7 +126,7 @@ func (am *authorizationMiddleware) ViewChannel(ctx context.Context, session auth
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainChannelsScope,
@@ -153,7 +153,7 @@ func (am *authorizationMiddleware) ListChannels(ctx context.Context, session aut
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainChannelsScope,
@@ -174,7 +174,7 @@ func (am *authorizationMiddleware) ListChannelsByClient(ctx context.Context, ses
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainChannelsScope,
@@ -191,7 +191,7 @@ func (am *authorizationMiddleware) UpdateChannel(ctx context.Context, session au
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainChannelsScope,
@@ -218,7 +218,7 @@ func (am *authorizationMiddleware) UpdateChannelTags(ctx context.Context, sessio
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainChannelsScope,
@@ -245,7 +245,7 @@ func (am *authorizationMiddleware) EnableChannel(ctx context.Context, session au
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainChannelsScope,
@@ -272,7 +272,7 @@ func (am *authorizationMiddleware) DisableChannel(ctx context.Context, session a
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainChannelsScope,
@@ -299,7 +299,7 @@ func (am *authorizationMiddleware) RemoveChannel(ctx context.Context, session au
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainChannelsScope,
@@ -326,7 +326,7 @@ func (am *authorizationMiddleware) Connect(ctx context.Context, session authn.Se
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainChannelsScope,
@@ -338,7 +338,7 @@ func (am *authorizationMiddleware) Connect(ctx context.Context, session authn.Se
 
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainClientsScope,
@@ -378,7 +378,7 @@ func (am *authorizationMiddleware) Disconnect(ctx context.Context, session authn
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainChannelsScope,
@@ -390,7 +390,7 @@ func (am *authorizationMiddleware) Disconnect(ctx context.Context, session authn
 
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainClientsScope,
@@ -431,7 +431,7 @@ func (am *authorizationMiddleware) SetParentGroup(ctx context.Context, session a
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainChannelsScope,
@@ -468,7 +468,7 @@ func (am *authorizationMiddleware) RemoveParentGroup(ctx context.Context, sessio
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainChannelsScope,

clients/events/events.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/absmach/supermq/clients"
+	"github.com/absmach/supermq/pkg/authn"
 	"github.com/absmach/supermq/pkg/events"
 	"github.com/absmach/supermq/pkg/roles"
 )
@@ -44,6 +45,7 @@ var (
 type createClientEvent struct {
 	clients.Client
 	rolesProvisioned []roles.RoleProvision
+	authn.Session
 }
 
 func (cce createClientEvent) Encode() (map[string]interface{}, error) {
@@ -53,6 +55,10 @@ func (cce createClientEvent) Encode() (map[string]interface{}, error) {
 		"roles_provisioned": cce.rolesProvisioned,
 		"status":            cce.Status.String(),
 		"created_at":        cce.CreatedAt,
+		"domain":            cce.DomainID,
+		"user_id":           cce.UserID,
+		"token_type":        cce.Type.String(),
+		"super_admin":       cce.SuperAdmin,
 	}
 
 	if cce.Name != "" {
@@ -61,9 +67,6 @@ func (cce createClientEvent) Encode() (map[string]interface{}, error) {
 	if len(cce.Tags) > 0 {
 		val["tags"] = cce.Tags
 	}
-	if cce.Domain != "" {
-		val["domain"] = cce.Domain
-	}
 	if cce.Metadata != nil {
 		val["metadata"] = cce.Metadata
 	}
@@ -77,13 +80,18 @@ func (cce createClientEvent) Encode() (map[string]interface{}, error) {
 type updateClientEvent struct {
 	clients.Client
 	operation string
+	authn.Session
 }
 
 func (uce updateClientEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation":  clientUpdate,
-		"updated_at": uce.UpdatedAt,
-		"updated_by": uce.UpdatedBy,
+		"operation":   clientUpdate,
+		"updated_at":  uce.UpdatedAt,
+		"updated_by":  uce.UpdatedBy,
+		"domain":      uce.DomainID,
+		"user_id":     uce.UserID,
+		"token_type":  uce.Type.String(),
+		"super_admin": uce.SuperAdmin,
 	}
 	if uce.operation != "" {
 		val["operation"] = clientUpdate + "_" + uce.operation
@@ -98,9 +106,6 @@ func (uce updateClientEvent) Encode() (map[string]interface{}, error) {
 	if len(uce.Tags) > 0 {
 		val["tags"] = uce.Tags
 	}
-	if uce.Domain != "" {
-		val["domain"] = uce.Domain
-	}
 	if uce.Credentials.Identity != "" {
 		val["identity"] = uce.Credentials.Identity
 	}
@@ -122,26 +127,36 @@ type changeStatusClientEvent struct {
 	status    string
 	updatedAt time.Time
 	updatedBy string
+	authn.Session
 }
 
 func (rce changeStatusClientEvent) Encode() (map[string]interface{}, error) {
 	return map[string]interface{}{
-		"operation":  clientChangeStatus,
-		"id":         rce.id,
-		"status":     rce.status,
-		"updated_at": rce.updatedAt,
-		"updated_by": rce.updatedBy,
+		"operation":   clientChangeStatus,
+		"id":          rce.id,
+		"status":      rce.status,
+		"updated_at":  rce.updatedAt,
+		"updated_by":  rce.updatedBy,
+		"domain":      rce.DomainID,
+		"user_id":     rce.UserID,
+		"token_type":  rce.Type.String(),
+		"super_admin": rce.SuperAdmin,
 	}, nil
 }
 
 type viewClientEvent struct {
 	clients.Client
+	authn.Session
 }
 
 func (vce viewClientEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation": clientView,
-		"id":        vce.ID,
+		"operation":   clientView,
+		"id":          vce.ID,
+		"domain":      vce.DomainID,
+		"user_id":     vce.UserID,
+		"token_type":  vce.Type.String(),
+		"super_admin": vce.SuperAdmin,
 	}
 
 	if vce.Name != "" {
@@ -150,9 +165,6 @@ func (vce viewClientEvent) Encode() (map[string]interface{}, error) {
 	if len(vce.Tags) > 0 {
 		val["tags"] = vce.Tags
 	}
-	if vce.Domain != "" {
-		val["domain"] = vce.Domain
-	}
 	if vce.Credentials.Identity != "" {
 		val["identity"] = vce.Credentials.Identity
 	}
@@ -177,12 +189,17 @@ func (vce viewClientEvent) Encode() (map[string]interface{}, error) {
 
 type viewClientPermsEvent struct {
 	permissions []string
+	authn.Session
 }
 
 func (vcpe viewClientPermsEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
 		"operation":   clientViewPerms,
 		"permissions": vcpe.permissions,
+		"domain":      vcpe.DomainID,
+		"user_id":     vcpe.UserID,
+		"token_type":  vcpe.Type.String(),
+		"super_admin": vcpe.SuperAdmin,
 	}
 	return val, nil
 }
@@ -190,15 +207,20 @@ func (vcpe viewClientPermsEvent) Encode() (map[string]interface{}, error) {
 type listClientEvent struct {
 	reqUserID string
 	clients.Page
+	authn.Session
 }
 
 func (lce listClientEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation": clientList,
-		"reqUserID": lce.reqUserID,
-		"total":     lce.Total,
-		"offset":    lce.Offset,
-		"limit":     lce.Limit,
+		"operation":   clientList,
+		"reqUserID":   lce.reqUserID,
+		"total":       lce.Total,
+		"offset":      lce.Offset,
+		"limit":       lce.Limit,
+		"domain":      lce.DomainID,
+		"user_id":     lce.UserID,
+		"token_type":  lce.Type.String(),
+		"super_admin": lce.SuperAdmin,
 	}
 
 	if lce.Name != "" {
@@ -213,9 +235,6 @@ func (lce listClientEvent) Encode() (map[string]interface{}, error) {
 	if lce.Metadata != nil {
 		val["metadata"] = lce.Metadata
 	}
-	if lce.Domain != "" {
-		val["domain"] = lce.Domain
-	}
 	if lce.Tag != "" {
 		val["tag"] = lce.Tag
 	}
@@ -238,15 +257,20 @@ func (lce listClientEvent) Encode() (map[string]interface{}, error) {
 type listClientByGroupEvent struct {
 	clients.Page
 	channelID string
+	authn.Session
 }
 
 func (lcge listClientByGroupEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation":  clientListByGroup,
-		"total":      lcge.Total,
-		"offset":     lcge.Offset,
-		"limit":      lcge.Limit,
-		"channel_id": lcge.channelID,
+		"operation":   clientListByGroup,
+		"total":       lcge.Total,
+		"offset":      lcge.Offset,
+		"limit":       lcge.Limit,
+		"channel_id":  lcge.channelID,
+		"domain":      lcge.DomainID,
+		"user_id":     lcge.UserID,
+		"token_type":  lcge.Type.String(),
+		"super_admin": lcge.SuperAdmin,
 	}
 
 	if lcge.Name != "" {
@@ -261,9 +285,6 @@ func (lcge listClientByGroupEvent) Encode() (map[string]interface{}, error) {
 	if lcge.Metadata != nil {
 		val["metadata"] = lcge.Metadata
 	}
-	if lcge.Domain != "" {
-		val["domain"] = lcge.Domain
-	}
 	if lcge.Tag != "" {
 		val["tag"] = lcge.Tag
 	}
@@ -282,12 +303,17 @@ func (lcge listClientByGroupEvent) Encode() (map[string]interface{}, error) {
 
 type identifyClientEvent struct {
 	clientID string
+	authn.Session
 }
 
 func (ice identifyClientEvent) Encode() (map[string]interface{}, error) {
 	return map[string]interface{}{
-		"operation": clientIdentify,
-		"id":        ice.clientID,
+		"operation":   clientIdentify,
+		"id":          ice.clientID,
+		"domain":      ice.DomainID,
+		"user_id":     ice.UserID,
+		"token_type":  ice.Type.String(),
+		"super_admin": ice.SuperAdmin,
 	}, nil
 }
 
@@ -295,12 +321,17 @@ type authorizeClientEvent struct {
 	clientID   string
 	channelID  string
 	permission string
+	authn.Session
 }
 
 func (ice authorizeClientEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation": clientAuthorize,
-		"id":        ice.clientID,
+		"operation":   clientAuthorize,
+		"id":          ice.clientID,
+		"domain":      ice.DomainID,
+		"user_id":     ice.UserID,
+		"token_type":  ice.Type.String(),
+		"super_admin": ice.SuperAdmin,
 	}
 
 	if ice.permission != "" {
@@ -318,31 +349,42 @@ type shareClientEvent struct {
 	id       string
 	relation string
 	userIDs  []string
+	authn.Session
 }
 
 func (sce shareClientEvent) Encode() (map[string]interface{}, error) {
 	return map[string]interface{}{
-		"operation": clientPrefix + sce.action,
-		"id":        sce.id,
-		"relation":  sce.relation,
-		"user_ids":  sce.userIDs,
+		"operation":   clientPrefix + sce.action,
+		"id":          sce.id,
+		"relation":    sce.relation,
+		"user_ids":    sce.userIDs,
+		"domain":      sce.DomainID,
+		"user_id":     sce.UserID,
+		"token_type":  sce.Type.String(),
+		"super_admin": sce.SuperAdmin,
 	}, nil
 }
 
 type removeClientEvent struct {
 	id string
+	authn.Session
 }
 
 func (dce removeClientEvent) Encode() (map[string]interface{}, error) {
 	return map[string]interface{}{
-		"operation": clientRemove,
-		"id":        dce.id,
+		"operation":   clientRemove,
+		"id":          dce.id,
+		"domain":      dce.DomainID,
+		"user_id":     dce.UserID,
+		"token_type":  dce.Type.String(),
+		"super_admin": dce.SuperAdmin,
 	}, nil
 }
 
 type setParentGroupEvent struct {
 	id            string
 	parentGroupID string
+	authn.Session
 }
 
 func (spge setParentGroupEvent) Encode() (map[string]interface{}, error) {
@@ -350,16 +392,25 @@ func (spge setParentGroupEvent) Encode() (map[string]interface{}, error) {
 		"operation":       clientSetParent,
 		"id":              spge.id,
 		"parent_group_id": spge.parentGroupID,
+		"domain":          spge.DomainID,
+		"user_id":         spge.UserID,
+		"token_type":      spge.Type.String(),
+		"super_admin":     spge.SuperAdmin,
 	}, nil
 }
 
 type removeParentGroupEvent struct {
 	id string
+	authn.Session
 }
 
 func (rpge removeParentGroupEvent) Encode() (map[string]interface{}, error) {
 	return map[string]interface{}{
-		"operation": clientRemoveParent,
-		"id":        rpge.id,
+		"operation":   clientRemoveParent,
+		"id":          rpge.id,
+		"domain":      rpge.DomainID,
+		"user_id":     rpge.UserID,
+		"token_type":  rpge.Type.String(),
+		"super_admin": rpge.SuperAdmin,
 	}, nil
 }

clients/events/streams.go

@@ -50,6 +50,7 @@ func (es *eventStore) CreateClients(ctx context.Context, session authn.Session,
 		event := createClientEvent{
 			Client:           cli,
 			rolesProvisioned: rps,
+			Session:          session,
 		}
 		if err := es.Publish(ctx, event); err != nil {
 			return clis, rps, err
@@ -65,7 +66,7 @@ func (es *eventStore) Update(ctx context.Context, session authn.Session, client
 		return cli, err
 	}
 
-	return es.update(ctx, "", cli)
+	return es.update(ctx, session, "", cli)
 }
 
 func (es *eventStore) UpdateTags(ctx context.Context, session authn.Session, client clients.Client) (clients.Client, error) {
@@ -74,7 +75,7 @@ func (es *eventStore) UpdateTags(ctx context.Context, session authn.Session, cli
 		return cli, err
 	}
 
-	return es.update(ctx, "tags", cli)
+	return es.update(ctx, session, "tags", cli)
 }
 
 func (es *eventStore) UpdateSecret(ctx context.Context, session authn.Session, id, key string) (clients.Client, error) {
@@ -83,12 +84,14 @@ func (es *eventStore) UpdateSecret(ctx context.Context, session authn.Session, i
 		return cli, err
 	}
 
-	return es.update(ctx, "secret", cli)
+	return es.update(ctx, session, "secret", cli)
 }
 
-func (es *eventStore) update(ctx context.Context, operation string, client clients.Client) (clients.Client, error) {
+func (es *eventStore) update(ctx context.Context, session authn.Session, operation string, client clients.Client) (clients.Client, error) {
 	event := updateClientEvent{
-		client, operation,
+		Client:    client,
+		operation: operation,
+		Session:   session,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -105,7 +108,8 @@ func (es *eventStore) View(ctx context.Context, session authn.Session, id string
 	}
 
 	event := viewClientEvent{
-		cli,
+		Client:  cli,
+		Session: session,
 	}
 	if err := es.Publish(ctx, event); err != nil {
 		return cli, err
@@ -120,8 +124,9 @@ func (es *eventStore) ListClients(ctx context.Context, session authn.Session, re
 		return cp, err
 	}
 	event := listClientEvent{
-		reqUserID,
-		pm,
+		reqUserID: reqUserID,
+		Page:      pm,
+		Session:   session,
 	}
 	if err := es.Publish(ctx, event); err != nil {
 		return cp, err
@@ -136,7 +141,7 @@ func (es *eventStore) Enable(ctx context.Context, session authn.Session, id stri
 		return cli, err
 	}
 
-	return es.changeStatus(ctx, cli)
+	return es.changeStatus(ctx, session, cli)
 }
 
 func (es *eventStore) Disable(ctx context.Context, session authn.Session, id string) (clients.Client, error) {
@@ -145,15 +150,16 @@ func (es *eventStore) Disable(ctx context.Context, session authn.Session, id str
 		return cli, err
 	}
 
-	return es.changeStatus(ctx, cli)
+	return es.changeStatus(ctx, session, cli)
 }
 
-func (es *eventStore) changeStatus(ctx context.Context, cli clients.Client) (clients.Client, error) {
+func (es *eventStore) changeStatus(ctx context.Context, session authn.Session, cli clients.Client) (clients.Client, error) {
 	event := changeStatusClientEvent{
 		id:        cli.ID,
 		updatedAt: cli.UpdatedAt,
 		updatedBy: cli.UpdatedBy,
 		status:    cli.Status.String(),
+		Session:   session,
 	}
 	if err := es.Publish(ctx, event); err != nil {
 		return cli, err
@@ -167,7 +173,10 @@ func (es *eventStore) Delete(ctx context.Context, session authn.Session, id stri
 		return err
 	}
 
-	event := removeClientEvent{id}
+	event := removeClientEvent{
+		id:      id,
+		Session: session,
+	}
 
 	if err := es.Publish(ctx, event); err != nil {
 		return err
@@ -181,7 +190,11 @@ func (es *eventStore) SetParentGroup(ctx context.Context, session authn.Session,
 		return err
 	}
 
-	event := setParentGroupEvent{parentGroupID: parentGroupID, id: id}
+	event := setParentGroupEvent{
+		parentGroupID: parentGroupID,
+		id:            id,
+		Session:       session,
+	}
 
 	if err := es.Publish(ctx, event); err != nil {
 		return err
@@ -195,7 +208,10 @@ func (es *eventStore) RemoveParentGroup(ctx context.Context, session authn.Sessi
 		return err
 	}
 
-	event := removeParentGroupEvent{id: id}
+	event := removeParentGroupEvent{
+		id:      id,
+		Session: session,
+	}
 
 	if err := es.Publish(ctx, event); err != nil {
 		return err

clients/middleware/authorization.go

@@ -78,7 +78,7 @@ func (am *authorizationMiddleware) CreateClients(ctx context.Context, session au
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainClientsScope,
@@ -106,7 +106,7 @@ func (am *authorizationMiddleware) View(ctx context.Context, session authn.Sessi
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainClientsScope,
@@ -133,7 +133,7 @@ func (am *authorizationMiddleware) ListClients(ctx context.Context, session auth
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainClientsScope,
@@ -155,7 +155,7 @@ func (am *authorizationMiddleware) Update(ctx context.Context, session authn.Ses
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainClientsScope,
@@ -183,7 +183,7 @@ func (am *authorizationMiddleware) UpdateTags(ctx context.Context, session authn
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainClientsScope,
@@ -211,7 +211,7 @@ func (am *authorizationMiddleware) UpdateSecret(ctx context.Context, session aut
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainClientsScope,
@@ -238,7 +238,7 @@ func (am *authorizationMiddleware) Enable(ctx context.Context, session authn.Ses
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainClientsScope,
@@ -266,7 +266,7 @@ func (am *authorizationMiddleware) Disable(ctx context.Context, session authn.Se
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainClientsScope,
@@ -293,7 +293,7 @@ func (am *authorizationMiddleware) Delete(ctx context.Context, session authn.Ses
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainClientsScope,
@@ -320,7 +320,7 @@ func (am *authorizationMiddleware) SetParentGroup(ctx context.Context, session a
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainGroupsScope,
@@ -357,7 +357,7 @@ func (am *authorizationMiddleware) RemoveParentGroup(ctx context.Context, sessio
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainGroupsScope,

domains/api/http/endpoint_test.go

@@ -816,7 +816,6 @@ func TestUpdateDomain(t *testing.T) {
 				contentType: tc.contentType,
 				token:       tc.token,
 			}
-			fmt.Println("req url", req.url)
 
 			if tc.token == validToken {
 				tc.session = authn.Session{UserID: userID, DomainID: tc.domainID, DomainUserID: tc.domainID + "_" + userID}

domains/events/events.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/absmach/supermq/domains"
+	"github.com/absmach/supermq/pkg/authn"
 	"github.com/absmach/supermq/pkg/events"
 	"github.com/absmach/supermq/pkg/roles"
 )
@@ -38,6 +39,7 @@ var (
 type createDomainEvent struct {
 	domains.Domain
 	rolesProvisioned []roles.RoleProvision
+	authn.Session
 }
 
 func (cde createDomainEvent) Encode() (map[string]interface{}, error) {
@@ -49,6 +51,9 @@ func (cde createDomainEvent) Encode() (map[string]interface{}, error) {
 		"created_at":        cde.CreatedAt,
 		"created_by":        cde.CreatedBy,
 		"roles_provisioned": cde.rolesProvisioned,
+		"user_id":           cde.UserID,
+		"token_type":        cde.Type.String(),
+		"super_admin":       cde.SuperAdmin,
 	}
 
 	if cde.Name != "" {
@@ -66,15 +71,19 @@ func (cde createDomainEvent) Encode() (map[string]interface{}, error) {
 
 type retrieveDomainEvent struct {
 	domains.Domain
+	authn.Session
 }
 
 func (rde retrieveDomainEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation":  domainRetrieve,
-		"id":         rde.ID,
-		"alias":      rde.Alias,
-		"status":     rde.Status.String(),
-		"created_at": rde.CreatedAt,
+		"operation":   domainRetrieve,
+		"id":          rde.ID,
+		"alias":       rde.Alias,
+		"status":      rde.Status.String(),
+		"created_at":  rde.CreatedAt,
+		"user_id":     rde.UserID,
+		"token_type":  rde.Type.String(),
+		"super_admin": rde.SuperAdmin,
 	}
 
 	if rde.Name != "" {
@@ -99,13 +108,17 @@ func (rde retrieveDomainEvent) Encode() (map[string]interface{}, error) {
 type retrieveDomainStatusEvent struct {
 	id     string
 	status domains.Status
+	authn.Session
 }
 
 func (rdse retrieveDomainStatusEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation": domainRetrieve,
-		"id":        rdse.id,
-		"status":    rdse.status.String(),
+		"operation":   domainRetrieve,
+		"id":          rdse.id,
+		"status":      rdse.status.String(),
+		"user_id":     rdse.UserID,
+		"token_type":  rdse.Type.String(),
+		"super_admin": rdse.SuperAdmin,
 	}
 
 	return val, nil
@@ -113,18 +126,22 @@ func (rdse retrieveDomainStatusEvent) Encode() (map[string]interface{}, error) {
 
 type updateDomainEvent struct {
 	domains.Domain
+	authn.Session
 }
 
 func (ude updateDomainEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation":  domainUpdate,
-		"id":         ude.ID,
-		"alias":      ude.Alias,
-		"status":     ude.Status.String(),
-		"created_at": ude.CreatedAt,
-		"created_by": ude.CreatedBy,
-		"updated_at": ude.UpdatedAt,
-		"updated_by": ude.UpdatedBy,
+		"operation":   domainUpdate,
+		"id":          ude.ID,
+		"alias":       ude.Alias,
+		"status":      ude.Status.String(),
+		"created_at":  ude.CreatedAt,
+		"created_by":  ude.CreatedBy,
+		"updated_at":  ude.UpdatedAt,
+		"updated_by":  ude.UpdatedBy,
+		"user_id":     ude.UserID,
+		"token_type":  ude.Type.String(),
+		"super_admin": ude.SuperAdmin,
 	}
 
 	if ude.Name != "" {
@@ -144,14 +161,18 @@ type enableDomainEvent struct {
 	domainID  string
 	updatedAt time.Time
 	updatedBy string
+	authn.Session
 }
 
 func (cdse enableDomainEvent) Encode() (map[string]interface{}, error) {
 	return map[string]interface{}{
-		"operation":  domainEnable,
-		"id":         cdse.domainID,
-		"updated_at": cdse.updatedAt,
-		"updated_by": cdse.updatedBy,
+		"operation":   domainEnable,
+		"id":          cdse.domainID,
+		"updated_at":  cdse.updatedAt,
+		"updated_by":  cdse.updatedBy,
+		"user_id":     cdse.UserID,
+		"token_type":  cdse.Type.String(),
+		"super_admin": cdse.SuperAdmin,
 	}, nil
 }
 
@@ -159,14 +180,18 @@ type disableDomainEvent struct {
 	domainID  string
 	updatedAt time.Time
 	updatedBy string
+	authn.Session
 }
 
 func (cdse disableDomainEvent) Encode() (map[string]interface{}, error) {
 	return map[string]interface{}{
-		"operation":  domainDisable,
-		"id":         cdse.domainID,
-		"updated_at": cdse.updatedAt,
-		"updated_by": cdse.updatedBy,
+		"operation":   domainDisable,
+		"id":          cdse.domainID,
+		"updated_at":  cdse.updatedAt,
+		"updated_by":  cdse.updatedBy,
+		"user_id":     cdse.UserID,
+		"token_type":  cdse.Type.String(),
+		"super_admin": cdse.SuperAdmin,
 	}, nil
 }
 
@@ -174,28 +199,38 @@ type freezeDomainEvent struct {
 	domainID  string
 	updatedAt time.Time
 	updatedBy string
+	authn.Session
 }
 
 func (cdse freezeDomainEvent) Encode() (map[string]interface{}, error) {
 	return map[string]interface{}{
-		"operation":  domainFreeze,
-		"id":         cdse.domainID,
-		"updated_at": cdse.updatedAt,
-		"updated_by": cdse.updatedBy,
+		"operation":   domainFreeze,
+		"id":          cdse.domainID,
+		"updated_at":  cdse.updatedAt,
+		"updated_by":  cdse.updatedBy,
+		"user_id":     cdse.UserID,
+		"token_type":  cdse.Type.String(),
+		"super_admin": cdse.SuperAdmin,
 	}, nil
 }
 
 type listDomainsEvent struct {
 	domains.Page
-	total uint64
+	total      uint64
+	userID     string
+	tokenType  string
+	superAdmin bool
 }
 
 func (lde listDomainsEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation": domainList,
-		"total":     lde.total,
-		"offset":    lde.Offset,
-		"limit":     lde.Limit,
+		"operation":   domainList,
+		"total":       lde.total,
+		"offset":      lde.Offset,
+		"limit":       lde.Limit,
+		"user_id":     lde.userID,
+		"token_type":  lde.tokenType,
+		"super_admin": lde.superAdmin,
 	}
 
 	if lde.Name != "" {

domains/events/streams.go

@@ -50,6 +50,7 @@ func (es *eventStore) CreateDomain(ctx context.Context, session authn.Session, d
 	event := createDomainEvent{
 		Domain:           domain,
 		rolesProvisioned: rps,
+		Session:          session,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -67,6 +68,7 @@ func (es *eventStore) RetrieveDomain(ctx context.Context, session authn.Session,
 
 	event := retrieveDomainEvent{
 		domain,
+		session,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -84,6 +86,7 @@ func (es *eventStore) UpdateDomain(ctx context.Context, session authn.Session, i
 
 	event := updateDomainEvent{
 		domain,
+		session,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -103,6 +106,7 @@ func (es *eventStore) EnableDomain(ctx context.Context, session authn.Session, i
 		domainID:  id,
 		updatedAt: domain.UpdatedAt,
 		updatedBy: domain.UpdatedBy,
+		Session:   session,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -122,6 +126,7 @@ func (es *eventStore) DisableDomain(ctx context.Context, session authn.Session,
 		domainID:  id,
 		updatedAt: domain.UpdatedAt,
 		updatedBy: domain.UpdatedBy,
+		Session:   session,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -141,6 +146,7 @@ func (es *eventStore) FreezeDomain(ctx context.Context, session authn.Session, i
 		domainID:  id,
 		updatedAt: domain.UpdatedAt,
 		updatedBy: domain.UpdatedBy,
+		Session:   session,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -157,7 +163,11 @@ func (es *eventStore) ListDomains(ctx context.Context, session authn.Session, p
 	}
 
 	event := listDomainsEvent{
-		p, dp.Total,
+		Page:       p,
+		total:      dp.Total,
+		userID:     session.UserID,
+		tokenType:  session.Type.String(),
+		superAdmin: session.SuperAdmin,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {

groups/events/events.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	groups "github.com/absmach/supermq/groups"
+	"github.com/absmach/supermq/pkg/authn"
 	"github.com/absmach/supermq/pkg/events"
 	"github.com/absmach/supermq/pkg/roles"
 )
@@ -51,6 +52,7 @@ var (
 type createGroupEvent struct {
 	groups.Group
 	rolesProvisioned []roles.RoleProvision
+	authn.Session
 }
 
 func (cge createGroupEvent) Encode() (map[string]interface{}, error) {
@@ -60,11 +62,12 @@ func (cge createGroupEvent) Encode() (map[string]interface{}, error) {
 		"roles_provisioned": cge.rolesProvisioned,
 		"status":            cge.Status.String(),
 		"created_at":        cge.CreatedAt,
+		"domain":            cge.DomainID,
+		"user_id":           cge.UserID,
+		"token_type":        cge.Type.String(),
+		"super_admin":       cge.SuperAdmin,
 	}
 
-	if cge.Domain != "" {
-		val["domain"] = cge.Domain
-	}
 	if cge.Parent != "" {
 		val["parent"] = cge.Parent
 	}
@@ -86,21 +89,23 @@ func (cge createGroupEvent) Encode() (map[string]interface{}, error) {
 
 type updateGroupEvent struct {
 	groups.Group
+	authn.Session
 }
 
 func (uge updateGroupEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation":  groupUpdate,
-		"updated_at": uge.UpdatedAt,
-		"updated_by": uge.UpdatedBy,
+		"operation":   groupUpdate,
+		"updated_at":  uge.UpdatedAt,
+		"updated_by":  uge.UpdatedBy,
+		"domain":      uge.DomainID,
+		"user_id":     uge.UserID,
+		"token_type":  uge.Type.String(),
+		"super_admin": uge.SuperAdmin,
 	}
 
 	if uge.ID != "" {
 		val["id"] = uge.ID
 	}
-	if uge.Domain != "" {
-		val["domain"] = uge.Domain
-	}
 	if uge.Parent != "" {
 		val["parent"] = uge.Parent
 	}
@@ -128,31 +133,38 @@ type changeStatusGroupEvent struct {
 	status    string
 	updatedAt time.Time
 	updatedBy string
+	authn.Session
 }
 
 func (rge changeStatusGroupEvent) Encode() (map[string]interface{}, error) {
 	return map[string]interface{}{
-		"operation":  groupChangeStatus,
-		"id":         rge.id,
-		"status":     rge.status,
-		"updated_at": rge.updatedAt,
-		"updated_by": rge.updatedBy,
+		"operation":   groupChangeStatus,
+		"id":          rge.id,
+		"status":      rge.status,
+		"updated_at":  rge.updatedAt,
+		"updated_by":  rge.updatedBy,
+		"domain":      rge.DomainID,
+		"user_id":     rge.UserID,
+		"token_type":  rge.Type.String(),
+		"super_admin": rge.SuperAdmin,
 	}, nil
 }
 
 type viewGroupEvent struct {
 	groups.Group
+	authn.Session
 }
 
 func (vge viewGroupEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation": groupView,
-		"id":        vge.ID,
+		"operation":   groupView,
+		"id":          vge.ID,
+		"domain":      vge.DomainID,
+		"user_id":     vge.UserID,
+		"token_type":  vge.Type.String(),
+		"super_admin": vge.SuperAdmin,
 	}
 
-	if vge.Domain != "" {
-		val["domain"] = vge.Domain
-	}
 	if vge.Parent != "" {
 		val["parent"] = vge.Parent
 	}
@@ -183,22 +195,27 @@ func (vge viewGroupEvent) Encode() (map[string]interface{}, error) {
 
 type listGroupEvent struct {
 	groups.PageMeta
+	domainID   string
+	userID     string
+	tokenType  string
+	superAdmin bool
 }
 
 func (lge listGroupEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation": groupList,
-		"total":     lge.Total,
-		"offset":    lge.Offset,
-		"limit":     lge.Limit,
+		"operation":   groupList,
+		"total":       lge.Total,
+		"offset":      lge.Offset,
+		"limit":       lge.Limit,
+		"domain":      lge.domainID,
+		"user_id":     lge.userID,
+		"token_type":  lge.tokenType,
+		"super_admin": lge.superAdmin,
 	}
 
 	if lge.Name != "" {
 		val["name"] = lge.Name
 	}
-	if lge.DomainID != "" {
-		val["domain_id"] = lge.DomainID
-	}
 	if lge.Tag != "" {
 		val["tag"] = lge.Tag
 	}
@@ -213,25 +230,28 @@ func (lge listGroupEvent) Encode() (map[string]interface{}, error) {
 }
 
 type listUserGroupEvent struct {
-	userID string
+	userID   string
+	domainID string
 	groups.PageMeta
+	tokenType  string
+	superAdmin bool
 }
 
 func (luge listUserGroupEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation": groupListUserGroups,
-		"user_id":   luge.userID,
-		"total":     luge.Total,
-		"offset":    luge.Offset,
-		"limit":     luge.Limit,
+		"operation":   groupListUserGroups,
+		"user_id":     luge.userID,
+		"domain":      luge.domainID,
+		"total":       luge.Total,
+		"offset":      luge.Offset,
+		"limit":       luge.Limit,
+		"token_type":  luge.tokenType,
+		"super_admin": luge.superAdmin,
 	}
 
 	if luge.Name != "" {
 		val["name"] = luge.Name
 	}
-	if luge.DomainID != "" {
-		val["domain_id"] = luge.DomainID
-	}
 	if luge.Tag != "" {
 		val["tag"] = luge.Tag
 	}
@@ -247,27 +267,37 @@ func (luge listUserGroupEvent) Encode() (map[string]interface{}, error) {
 
 type deleteGroupEvent struct {
 	id string
+	authn.Session
 }
 
 func (rge deleteGroupEvent) Encode() (map[string]interface{}, error) {
 	return map[string]interface{}{
-		"operation": groupRemove,
-		"id":        rge.id,
+		"operation":   groupRemove,
+		"id":          rge.id,
+		"domain":      rge.DomainID,
+		"user_id":     rge.UserID,
+		"token_type":  rge.Type.String(),
+		"super_admin": rge.SuperAdmin,
 	}, nil
 }
 
 type retrieveGroupHierarchyEvent struct {
 	id string
 	groups.HierarchyPageMeta
+	authn.Session
 }
 
 func (vcge retrieveGroupHierarchyEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation": groupRetrieveGroupHierarchy,
-		"id":        vcge.id,
-		"level":     vcge.Level,
-		"direction": vcge.Direction,
-		"tree":      vcge.Tree,
+		"operation":   groupRetrieveGroupHierarchy,
+		"id":          vcge.id,
+		"level":       vcge.Level,
+		"direction":   vcge.Direction,
+		"tree":        vcge.Tree,
+		"domain":      vcge.DomainID,
+		"user_id":     vcge.UserID,
+		"token_type":  vcge.Type.String(),
+		"super_admin": vcge.SuperAdmin,
 	}
 	return val, nil
 }
@@ -275,41 +305,54 @@ func (vcge retrieveGroupHierarchyEvent) Encode() (map[string]interface{}, error)
 type addParentGroupEvent struct {
 	id       string
 	parentID string
+	authn.Session
 }
 
 func (apge addParentGroupEvent) Encode() (map[string]interface{}, error) {
 	return map[string]interface{}{
-		"operation": groupAddParentGroup,
-		"id":        apge.id,
-		"parent_id": apge.parentID,
+		"operation":   groupAddParentGroup,
+		"id":          apge.id,
+		"parent_id":   apge.parentID,
+		"domain":      apge.DomainID,
+		"user_id":     apge.UserID,
+		"token_type":  apge.Type.String(),
+		"super_admin": apge.SuperAdmin,
 	}, nil
 }
 
 type removeParentGroupEvent struct {
 	id string
+	authn.Session
 }
 
 func (rpge removeParentGroupEvent) Encode() (map[string]interface{}, error) {
 	return map[string]interface{}{
-		"operation": groupRemoveParentGroup,
-		"id":        rpge.id,
+		"operation":   groupRemoveParentGroup,
+		"id":          rpge.id,
+		"domain":      rpge.DomainID,
+		"user_id":     rpge.UserID,
+		"token_type":  rpge.Type.String(),
+		"super_admin": rpge.SuperAdmin,
 	}, nil
 }
 
 type viewParentGroupEvent struct {
-	id string
+	id       string
+	domainID string
 }
 
 func (vpge viewParentGroupEvent) Encode() (map[string]interface{}, error) {
 	return map[string]interface{}{
 		"operation": groupViewParentGroup,
 		"id":        vpge.id,
+		"domain":    vpge.domainID,
 	}, nil
 }
 
 type addChildrenGroupsEvent struct {
 	id          string
 	childrenIDs []string
+	authn.Session
 }
 
 func (acge addChildrenGroupsEvent) Encode() (map[string]interface{}, error) {
@@ -317,12 +360,17 @@ func (acge addChildrenGroupsEvent) Encode() (map[string]interface{}, error) {
 		"operation":   groupAddChildrenGroups,
 		"id":          acge.id,
 		"childre_ids": acge.childrenIDs,
+		"domain":      acge.DomainID,
+		"user_id":     acge.UserID,
+		"token_type":  acge.Type.String(),
+		"super_admin": acge.SuperAdmin,
 	}, nil
 }
 
 type removeChildrenGroupsEvent struct {
 	id          string
 	childrenIDs []string
+	authn.Session
 }
 
 func (rcge removeChildrenGroupsEvent) Encode() (map[string]interface{}, error) {
@@ -330,17 +378,26 @@ func (rcge removeChildrenGroupsEvent) Encode() (map[string]interface{}, error) {
 		"operation":    groupRemoveChildrenGroups,
 		"id":           rcge.id,
 		"children_ids": rcge.childrenIDs,
+		"domain":       rcge.DomainID,
+		"user_id":      rcge.UserID,
+		"token_type":   rcge.Type.String(),
+		"super_admin":  rcge.SuperAdmin,
 	}, nil
 }
 
 type removeAllChildrenGroupsEvent struct {
 	id string
+	authn.Session
 }
 
 func (racge removeAllChildrenGroupsEvent) Encode() (map[string]interface{}, error) {
 	return map[string]interface{}{
-		"operation": groupRemoveAllChildrenGroups,
-		"id":        racge.id,
+		"operation":   groupRemoveAllChildrenGroups,
+		"id":          racge.id,
+		"domain":      racge.DomainID,
+		"user_id":     racge.UserID,
+		"token_type":  racge.Type.String(),
+		"super_admin": racge.SuperAdmin,
 	}, nil
 }
 
@@ -349,6 +406,10 @@ type listChildrenGroupsEvent struct {
 	startLevel int64
 	endLevel   int64
 	groups.PageMeta
+	domainID   string
+	userID     string
+	tokenType  string
+	superAdmin bool
 }
 
 func (vcge listChildrenGroupsEvent) Encode() (map[string]interface{}, error) {
@@ -360,13 +421,14 @@ func (vcge listChildrenGroupsEvent) Encode() (map[string]interface{}, error) {
 		"total":       vcge.Total,
 		"offset":      vcge.Offset,
 		"limit":       vcge.Limit,
+		"domain":      vcge.domainID,
+		"user_id":     vcge.userID,
+		"token_type":  vcge.tokenType,
+		"super_admin": vcge.superAdmin,
 	}
 	if vcge.Name != "" {
 		val["name"] = vcge.Name
 	}
-	if vcge.DomainID != "" {
-		val["domain_id"] = vcge.DomainID
-	}
 	if vcge.Tag != "" {
 		val["tag"] = vcge.Tag
 	}

groups/events/streams.go

@@ -49,6 +49,7 @@ func (es eventStore) CreateGroup(ctx context.Context, session authn.Session, gro
 	event := createGroupEvent{
 		Group:            group,
 		rolesProvisioned: rps,
+		Session:          session,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -66,6 +67,7 @@ func (es eventStore) UpdateGroup(ctx context.Context, session authn.Session, gro
 
 	event := updateGroupEvent{
 		group,
+		session,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -82,6 +84,7 @@ func (es eventStore) ViewGroup(ctx context.Context, session authn.Session, id st
 	}
 	event := viewGroupEvent{
 		group,
+		session,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -97,7 +100,11 @@ func (es eventStore) ListGroups(ctx context.Context, session authn.Session, pm g
 		return gp, err
 	}
 	event := listGroupEvent{
-		pm,
+		PageMeta:   pm,
+		domainID:   session.DomainID,
+		userID:     session.UserID,
+		tokenType:  session.Type.String(),
+		superAdmin: session.SuperAdmin,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -113,8 +120,11 @@ func (es eventStore) ListUserGroups(ctx context.Context, session authn.Session,
 		return gp, err
 	}
 	event := listUserGroupEvent{
-		userID:   userID,
-		PageMeta: pm,
+		userID:     userID,
+		PageMeta:   pm,
+		domainID:   session.DomainID,
+		tokenType:  session.Type.String(),
+		superAdmin: session.SuperAdmin,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -130,7 +140,7 @@ func (es eventStore) EnableGroup(ctx context.Context, session authn.Session, id
 		return group, err
 	}
 
-	return es.changeStatus(ctx, group)
+	return es.changeStatus(ctx, session, group)
 }
 
 func (es eventStore) DisableGroup(ctx context.Context, session authn.Session, id string) (groups.Group, error) {
@@ -139,15 +149,16 @@ func (es eventStore) DisableGroup(ctx context.Context, session authn.Session, id
 		return group, err
 	}
 
-	return es.changeStatus(ctx, group)
+	return es.changeStatus(ctx, session, group)
 }
 
-func (es eventStore) changeStatus(ctx context.Context, group groups.Group) (groups.Group, error) {
+func (es eventStore) changeStatus(ctx context.Context, session authn.Session, group groups.Group) (groups.Group, error) {
 	event := changeStatusGroupEvent{
 		id:        group.ID,
 		updatedAt: group.UpdatedAt,
 		updatedBy: group.UpdatedBy,
 		status:    group.Status.String(),
+		Session:   session,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -161,7 +172,10 @@ func (es eventStore) DeleteGroup(ctx context.Context, session authn.Session, id
 	if err := es.svc.DeleteGroup(ctx, session, id); err != nil {
 		return err
 	}
-	if err := es.Publish(ctx, deleteGroupEvent{id}); err != nil {
+	if err := es.Publish(ctx, deleteGroupEvent{
+		id:      id,
+		Session: session,
+	}); err != nil {
 		return err
 	}
 	return nil
@@ -172,7 +186,7 @@ func (es eventStore) RetrieveGroupHierarchy(ctx context.Context, session authn.S
 	if err != nil {
 		return g, err
 	}
-	if err := es.Publish(ctx, retrieveGroupHierarchyEvent{id, hm}); err != nil {
+	if err := es.Publish(ctx, retrieveGroupHierarchyEvent{id: id, Session: session, HierarchyPageMeta: hm}); err != nil {
 		return g, err
 	}
 	return g, nil
@@ -182,7 +196,7 @@ func (es eventStore) AddParentGroup(ctx context.Context, session authn.Session,
 	if err := es.svc.AddParentGroup(ctx, session, id, parentID); err != nil {
 		return err
 	}
-	if err := es.Publish(ctx, addParentGroupEvent{id, parentID}); err != nil {
+	if err := es.Publish(ctx, addParentGroupEvent{id: id, parentID: parentID, Session: session}); err != nil {
 		return err
 	}
 	return nil
@@ -192,7 +206,7 @@ func (es eventStore) RemoveParentGroup(ctx context.Context, session authn.Sessio
 	if err := es.svc.RemoveParentGroup(ctx, session, id); err != nil {
 		return err
 	}
-	if err := es.Publish(ctx, removeParentGroupEvent{id}); err != nil {
+	if err := es.Publish(ctx, removeParentGroupEvent{id: id, Session: session}); err != nil {
 		return err
 	}
 	return nil
@@ -202,7 +216,7 @@ func (es eventStore) AddChildrenGroups(ctx context.Context, session authn.Sessio
 	if err := es.svc.AddChildrenGroups(ctx, session, id, childrenGroupIDs); err != nil {
 		return err
 	}
-	if err := es.Publish(ctx, addChildrenGroupsEvent{id, childrenGroupIDs}); err != nil {
+	if err := es.Publish(ctx, addChildrenGroupsEvent{id: id, Session: session, childrenIDs: childrenGroupIDs}); err != nil {
 		return err
 	}
 	return nil
@@ -212,7 +226,7 @@ func (es eventStore) RemoveChildrenGroups(ctx context.Context, session authn.Ses
 	if err := es.svc.RemoveChildrenGroups(ctx, session, id, childrenGroupIDs); err != nil {
 		return err
 	}
-	if err := es.Publish(ctx, removeChildrenGroupsEvent{id, childrenGroupIDs}); err != nil {
+	if err := es.Publish(ctx, removeChildrenGroupsEvent{id: id, Session: session, childrenIDs: childrenGroupIDs}); err != nil {
 		return err
 	}
 
@@ -223,7 +237,7 @@ func (es eventStore) RemoveAllChildrenGroups(ctx context.Context, session authn.
 	if err := es.svc.RemoveAllChildrenGroups(ctx, session, id); err != nil {
 		return err
 	}
-	if err := es.Publish(ctx, removeAllChildrenGroupsEvent{id}); err != nil {
+	if err := es.Publish(ctx, removeAllChildrenGroupsEvent{id: id, Session: session}); err != nil {
 		return err
 	}
 	return nil
@@ -234,7 +248,16 @@ func (es eventStore) ListChildrenGroups(ctx context.Context, session authn.Sessi
 	if err != nil {
 		return g, err
 	}
-	if err := es.Publish(ctx, listChildrenGroupsEvent{id, startLevel, endLevel, pm}); err != nil {
+	if err := es.Publish(ctx, listChildrenGroupsEvent{
+		id:         id,
+		domainID:   session.DomainID,
+		startLevel: startLevel,
+		endLevel:   endLevel,
+		PageMeta:   pm,
+		userID:     session.UserID,
+		tokenType:  session.Type.String(),
+		superAdmin: session.SuperAdmin,
+	}); err != nil {
 		return g, err
 	}
 	return g, nil

groups/middleware/authorization.go

@@ -84,7 +84,7 @@ func (am *authorizationMiddleware) CreateGroup(ctx context.Context, session auth
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainGroupsScope,
@@ -125,7 +125,7 @@ func (am *authorizationMiddleware) UpdateGroup(ctx context.Context, session auth
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainGroupsScope,
@@ -154,7 +154,7 @@ func (am *authorizationMiddleware) ViewGroup(ctx context.Context, session authn.
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainGroupsScope,
@@ -183,7 +183,7 @@ func (am *authorizationMiddleware) ListGroups(ctx context.Context, session authn
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainGroupsScope,
@@ -235,7 +235,7 @@ func (am *authorizationMiddleware) EnableGroup(ctx context.Context, session auth
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainGroupsScope,
@@ -263,7 +263,7 @@ func (am *authorizationMiddleware) DisableGroup(ctx context.Context, session aut
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainGroupsScope,
@@ -291,7 +291,7 @@ func (am *authorizationMiddleware) DeleteGroup(ctx context.Context, session auth
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainGroupsScope,
@@ -319,7 +319,7 @@ func (am *authorizationMiddleware) RetrieveGroupHierarchy(ctx context.Context, s
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainGroupsScope,
@@ -346,7 +346,7 @@ func (am *authorizationMiddleware) AddParentGroup(ctx context.Context, session a
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainGroupsScope,
@@ -383,7 +383,7 @@ func (am *authorizationMiddleware) RemoveParentGroup(ctx context.Context, sessio
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainGroupsScope,
@@ -427,7 +427,7 @@ func (am *authorizationMiddleware) AddChildrenGroups(ctx context.Context, sessio
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainGroupsScope,
@@ -467,7 +467,7 @@ func (am *authorizationMiddleware) RemoveChildrenGroups(ctx context.Context, ses
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainGroupsScope,
@@ -495,7 +495,7 @@ func (am *authorizationMiddleware) RemoveAllChildrenGroups(ctx context.Context,
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainGroupsScope,
@@ -523,7 +523,7 @@ func (am *authorizationMiddleware) ListChildrenGroups(ctx context.Context, sessi
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       auth.PlatformDomainsScope,
 			OptionalDomainID:         session.DomainID,
 			OptionalDomainEntityType: auth.DomainGroupsScope,

http/handler_test.go

@@ -248,7 +248,7 @@ func TestPublish(t *testing.T) {
 			err:       svcerr.ErrAuthentication,
 		},
 		{
-			desc:      "publish with thing key and failed to authenticate",
+			desc:      "publish with client key and failed to authenticate",
 			topic:     &topic,
 			payload:   &payload,
 			password:  clientKey,
@@ -260,7 +260,7 @@ func TestPublish(t *testing.T) {
 			err:       svcerr.ErrAuthentication,
 		},
 		{
-			desc:      "publish with thing key and failed to authenticate with error",
+			desc:      "publish with client key and failed to authenticate with error",
 			topic:     &topic,
 			payload:   &payload,
 			password:  clientKey,

journal/journal.go

@@ -93,6 +93,7 @@ func (e EntityType) Query() string {
 // Journal represents an event journal that occurred in the system.
 type Journal struct {
 	ID         string                 `json:"id,omitempty" db:"id"`
+	Domain     string                 `json:"domain,omitempty" db:"domain"`
 	Operation  string                 `json:"operation,omitempty" db:"operation,omitempty"`
 	OccurredAt time.Time              `json:"occurred_at,omitempty" db:"occurred_at,omitempty"`
 	Attributes map[string]interface{} `json:"attributes,omitempty" db:"attributes,omitempty"` // This is extra information about the journal for example client_id, user_id, group_id etc.

journal/middleware/authorization.go

@@ -14,6 +14,8 @@ import (
 
 var _ journal.Service = (*authorizationMiddleware)(nil)
 
+var readPermission = "read_permission"
+
 type authorizationMiddleware struct {
 	svc   journal.Service
 	authz smqauthz.Authorization
@@ -32,7 +34,7 @@ func (am *authorizationMiddleware) Save(ctx context.Context, journal journal.Jou
 }
 
 func (am *authorizationMiddleware) RetrieveAll(ctx context.Context, session smqauthn.Session, page journal.Page) (journal.JournalsPage, error) {
-	permission := policies.ViewPermission
+	permission := readPermission
 	objectType := page.EntityType.AuthString()
 	object := page.EntityID
 	subject := session.DomainUserID

journal/postgres/init.go

@@ -17,6 +17,7 @@ func Migration() *migrate.MemoryMigrationSource {
 					`CREATE TABLE IF NOT EXISTS journal (
 						id          VARCHAR(36) PRIMARY KEY,
 						operation	VARCHAR NOT NULL,
+						domain		VARCHAR,
 						occurred_at	TIMESTAMP NOT NULL,
 						attributes	JSONB NOT NULL,
 						metadata	JSONB,

journal/postgres/journal.go

@@ -25,8 +25,19 @@ func NewRepository(db postgres.Database) journal.Repository {
 }
 
 func (repo *repository) Save(ctx context.Context, j journal.Journal) (err error) {
-	q := `INSERT INTO journal (id, operation, occurred_at, attributes, metadata)
-		VALUES (:id, :operation, :occurred_at, :attributes, :metadata);`
+	domain, ok := j.Attributes["domain"].(string)
+	if ok {
+		j.Domain = domain
+	}
+	if strings.HasPrefix(j.Operation, "domain.") {
+		domain, ok := j.Attributes["id"].(string)
+		if ok {
+			j.Domain = domain
+		}
+	}
+
+	q := `INSERT INTO journal (id, operation, occurred_at, attributes, metadata, domain)
+		VALUES (:id, :operation, :occurred_at, :attributes, :metadata, :domain);`
 
 	dbJournal, err := toDBJournal(j)
 	if err != nil {
@@ -43,7 +54,7 @@ func (repo *repository) Save(ctx context.Context, j journal.Journal) (err error)
 func (repo *repository) RetrieveAll(ctx context.Context, page journal.Page) (journal.JournalsPage, error) {
 	query := pageQuery(page)
 
-	sq := "operation, occurred_at"
+	sq := "operation, occurred_at, domain"
 	if page.WithAttributes {
 		sq += ", attributes"
 	}
@@ -117,6 +128,7 @@ func pageQuery(pm journal.Page) string {
 type dbJournal struct {
 	ID         string    `db:"id"`
 	Operation  string    `db:"operation"`
+	Domain     string    `db:"domain"`
 	OccurredAt time.Time `db:"occurred_at"`
 	Attributes []byte    `db:"attributes"`
 	Metadata   []byte    `db:"metadata"`
@@ -148,6 +160,7 @@ func toDBJournal(j journal.Journal) (dbJournal, error) {
 	return dbJournal{
 		ID:         j.ID,
 		Operation:  j.Operation,
+		Domain:     j.Domain,
 		OccurredAt: j.OccurredAt,
 		Attributes: attributes,
 		Metadata:   metadata,
@@ -171,6 +184,7 @@ func toJournal(dbj dbJournal) (journal.Journal, error) {
 
 	return journal.Journal{
 		Operation:  dbj.Operation,
+		Domain:     dbj.Domain,
 		OccurredAt: dbj.OccurredAt,
 		Attributes: attributes,
 		Metadata:   metadata,

journal/postgres/journal_test.go

@@ -43,6 +43,7 @@ var (
 	}
 
 	entityID           = testsutil.GenerateUUID(&testing.T{})
+	domain             = testsutil.GenerateUUID(&testing.T{})
 	clientOperation    = "client.create"
 	clientAttributesV1 = map[string]interface{}{
 		"id":         entityID,
@@ -50,7 +51,7 @@ var (
 		"created_at": time.Now().Add(-time.Hour),
 		"name":       "client",
 		"tags":       []interface{}{"tag1", "tag2"},
-		"domain":     testsutil.GenerateUUID(&testing.T{}),
+		"domain":     domain,
 		"metadata":   payload,
 		"identity":   testsutil.GenerateUUID(&testing.T{}),
 	}
@@ -64,7 +65,7 @@ var (
 		"created_at": time.Now().Add(-time.Hour),
 		"name":       "user",
 		"tags":       []interface{}{"tag1", "tag2"},
-		"domain":     testsutil.GenerateUUID(&testing.T{}),
+		"domain":     domain,
 		"metadata":   payload,
 		"identity":   testsutil.GenerateUUID(&testing.T{}),
 	}
@@ -294,6 +295,7 @@ func TestJournalRetrieveAll(t *testing.T) {
 	for i := 0; i < num; i++ {
 		j := journal.Journal{
 			ID:         testsutil.GenerateUUID(t),
+			Domain:     domain,
 			Operation:  fmt.Sprintf("%s-%d", operation, i),
 			OccurredAt: time.Now().UTC().Truncate(time.Microsecond),
 			Attributes: userAttributesV1,

pkg/authn/authn.go

@@ -29,7 +29,7 @@ func (t TokenType) String() string {
 
 type Session struct {
 	Type         TokenType
-	ID           string
+	PatID        string
 	DomainUserID string
 	UserID       string
 	DomainID     string

pkg/authn/authsvc/authn.go

@@ -47,7 +47,7 @@ func (a authentication) Authenticate(ctx context.Context, token string) (authn.S
 			return authn.Session{}, errors.Wrap(errors.ErrAuthentication, err)
 		}
 
-		return authn.Session{Type: authn.PersonalAccessToken, ID: res.GetId(), UserID: res.GetUserId()}, nil
+		return authn.Session{Type: authn.PersonalAccessToken, PatID: res.GetId(), UserID: res.GetUserId()}, nil
 	}
 	res, err := a.authSvcClient.Authenticate(ctx, &grpcAuthV1.AuthNReq{Token: token})
 	if err != nil {

pkg/sdk/channels_test.go

@@ -1335,7 +1335,6 @@ func TestDisableChannel(t *testing.T) {
 			authCall := auth.On("Authenticate", mock.Anything, tc.token).Return(tc.session, tc.authenticateErr)
 			svcCall := gsvc.On("DisableChannel", mock.Anything, tc.session, tc.channelID).Return(tc.svcRes, tc.svcErr)
 			resp, err := mgsdk.DisableChannel(tc.channelID, tc.domainID, tc.token)
-			fmt.Println(resp)
 			assert.Equal(t, tc.err, err)
 			assert.Equal(t, tc.response, resp)
 			if tc.err == nil {
@@ -1535,7 +1534,6 @@ func TestConnect(t *testing.T) {
 			authCall := auth.On("Authenticate", mock.Anything, tc.token).Return(tc.session, tc.authenticateErr)
 			svcCall := gsvc.On("Connect", mock.Anything, tc.session, tc.connection.ChannelIDs, tc.connection.ClientIDs, connTypes).Return(tc.svcErr)
 			err := mgsdk.Connect(tc.connection, tc.domainID, tc.token)
-			fmt.Println(err)
 			assert.Equal(t, tc.err, err)
 			if tc.err == nil {
 				ok := svcCall.Parent.AssertCalled(t, "Connect", mock.Anything, tc.session, tc.connection.ChannelIDs, tc.connection.ClientIDs, connTypes)

pkg/sdk/sdk.go

@@ -1319,7 +1319,7 @@ type SDK interface {
 	// Journal returns a list of journal logs.
 	//
 	// For example:
-	//  journals, _ := sdk.Journal("client", "clientID","domainID", PageMetadata{Offset: 0, Limit: 10, Operation: "thing.create"}, "token")
+	//  journals, _ := sdk.Journal("client", "clientID","domainID", PageMetadata{Offset: 0, Limit: 10, Operation: "client.create"}, "token")
 	//  fmt.Println(journals)
 	Journal(entityType, entityID, domainID string, pm PageMetadata, token string) (journal JournalsPage, err error)
 }

users/events/events.go

@@ -6,6 +6,7 @@ package events
 import (
 	"time"
 
+	"github.com/absmach/supermq/pkg/authn"
 	"github.com/absmach/supermq/pkg/events"
 	"github.com/absmach/supermq/users"
 )
@@ -57,14 +58,17 @@ var (
 
 type createUserEvent struct {
 	users.User
+	authn.Session
 }
 
 func (uce createUserEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation":  userCreate,
-		"id":         uce.ID,
-		"status":     uce.Status.String(),
-		"created_at": uce.CreatedAt,
+		"operation":   userCreate,
+		"id":          uce.ID,
+		"status":      uce.Status.String(),
+		"created_at":  uce.CreatedAt,
+		"token_type":  uce.Type.String(),
+		"super_admin": uce.SuperAdmin,
 	}
 
 	if uce.FirstName != "" {
@@ -92,13 +96,16 @@ func (uce createUserEvent) Encode() (map[string]interface{}, error) {
 type updateUserEvent struct {
 	users.User
 	operation string
+	authn.Session
 }
 
 func (uce updateUserEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation":  userUpdate,
-		"updated_at": uce.UpdatedAt,
-		"updated_by": uce.UpdatedBy,
+		"operation":   userUpdate,
+		"updated_at":  uce.UpdatedAt,
+		"updated_by":  uce.UpdatedBy,
+		"token_type":  uce.Type.String(),
+		"super_admin": uce.SuperAdmin,
 	}
 	if uce.operation != "" {
 		val["operation"] = userUpdate + "_" + uce.operation
@@ -137,13 +144,16 @@ func (uce updateUserEvent) Encode() (map[string]interface{}, error) {
 
 type updateUsernameEvent struct {
 	users.User
+	authn.Session
 }
 
 func (une updateUsernameEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation":  userUpdateUsername,
-		"updated_at": une.UpdatedAt,
-		"updated_by": une.UpdatedBy,
+		"operation":   userUpdateUsername,
+		"updated_at":  une.UpdatedAt,
+		"updated_by":  une.UpdatedBy,
+		"token_type":  une.Type.String(),
+		"super_admin": une.SuperAdmin,
 	}
 
 	if une.ID != "" {
@@ -164,13 +174,16 @@ func (une updateUsernameEvent) Encode() (map[string]interface{}, error) {
 
 type updateProfilePictureEvent struct {
 	users.User
+	authn.Session
 }
 
 func (uppe updateProfilePictureEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation":  userUpdateProfilePicture,
-		"updated_at": uppe.UpdatedAt,
-		"updated_by": uppe.UpdatedBy,
+		"operation":   userUpdateProfilePicture,
+		"updated_at":  uppe.UpdatedAt,
+		"updated_by":  uppe.UpdatedBy,
+		"token_type":  uppe.Type.String(),
+		"super_admin": uppe.SuperAdmin,
 	}
 
 	if uppe.ID != "" {
@@ -188,26 +201,32 @@ type removeUserEvent struct {
 	status    string
 	updatedAt time.Time
 	updatedBy string
+	authn.Session
 }
 
 func (rce removeUserEvent) Encode() (map[string]interface{}, error) {
 	return map[string]interface{}{
-		"operation":  userRemove,
-		"id":         rce.id,
-		"status":     rce.status,
-		"updated_at": rce.updatedAt,
-		"updated_by": rce.updatedBy,
+		"operation":   userRemove,
+		"id":          rce.id,
+		"status":      rce.status,
+		"updated_at":  rce.updatedAt,
+		"updated_by":  rce.updatedBy,
+		"token_type":  rce.Type.String(),
+		"super_admin": rce.SuperAdmin,
 	}, nil
 }
 
 type viewUserEvent struct {
 	users.User
+	authn.Session
 }
 
 func (vue viewUserEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation": userView,
-		"id":        vue.ID,
+		"operation":   userView,
+		"id":          vue.ID,
+		"token_type":  vue.Type.String(),
+		"super_admin": vue.SuperAdmin,
 	}
 
 	if vue.LastName != "" {
@@ -246,12 +265,15 @@ func (vue viewUserEvent) Encode() (map[string]interface{}, error) {
 
 type viewProfileEvent struct {
 	users.User
+	authn.Session
 }
 
 func (vpe viewProfileEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation": profileView,
-		"id":        vpe.ID,
+		"operation":   profileView,
+		"id":          vpe.ID,
+		"token_type":  vpe.Type.String(),
+		"super_admin": vpe.SuperAdmin,
 	}
 
 	if vpe.FirstName != "" {
@@ -287,14 +309,17 @@ func (vpe viewProfileEvent) Encode() (map[string]interface{}, error) {
 
 type listUserEvent struct {
 	users.Page
+	authn.Session
 }
 
 func (lue listUserEvent) Encode() (map[string]interface{}, error) {
 	val := map[string]interface{}{
-		"operation": userList,
-		"total":     lue.Total,
-		"offset":    lue.Offset,
-		"limit":     lue.Limit,
+		"operation":   userList,
+		"total":       lue.Total,
+		"offset":      lue.Offset,
+		"limit":       lue.Limit,
+		"token_type":  lue.Type.String(),
+		"super_admin": lue.SuperAdmin,
 	}
 
 	if lue.FirstName != "" {
@@ -338,6 +363,7 @@ type listUserByGroupEvent struct {
 	users.Page
 	objectKind string
 	objectID   string
+	authn.Session
 }
 
 func (lcge listUserByGroupEvent) Encode() (map[string]interface{}, error) {
@@ -348,6 +374,9 @@ func (lcge listUserByGroupEvent) Encode() (map[string]interface{}, error) {
 		"limit":       lcge.Limit,
 		"object_kind": lcge.objectKind,
 		"object_id":   lcge.objectID,
+		"domain":      lcge.DomainID,
+		"token_type":  lcge.Type.String(),
+		"super_admin": lcge.SuperAdmin,
 	}
 
 	if lcge.Username != "" {
@@ -496,24 +525,30 @@ func (oce oauthCallbackEvent) Encode() (map[string]interface{}, error) {
 
 type deleteUserEvent struct {
 	id string
+	authn.Session
 }
 
 func (dce deleteUserEvent) Encode() (map[string]interface{}, error) {
 	return map[string]interface{}{
-		"operation": deleteUser,
-		"id":        dce.id,
+		"operation":   deleteUser,
+		"id":          dce.id,
+		"token_type":  dce.Type.String(),
+		"super_admin": dce.SuperAdmin,
 	}, nil
 }
 
 type addUserPolicyEvent struct {
 	id   string
 	role string
+	authn.Session
 }
 
 func (acpe addUserPolicyEvent) Encode() (map[string]interface{}, error) {
 	return map[string]interface{}{
-		"operation": addClientPolicy,
-		"id":        acpe.id,
-		"role":      acpe.role,
+		"operation":   addClientPolicy,
+		"id":          acpe.id,
+		"role":        acpe.role,
+		"token_type":  acpe.Type.String(),
+		"super_admin": acpe.SuperAdmin,
 	}, nil
 }

users/events/streams.go

@@ -44,6 +44,7 @@ func (es *eventStore) Register(ctx context.Context, session authn.Session, user
 
 	event := createUserEvent{
 		user,
+		session,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -59,7 +60,7 @@ func (es *eventStore) Update(ctx context.Context, session authn.Session, user us
 		return user, err
 	}
 
-	return es.update(ctx, "", user)
+	return es.update(ctx, session, "", user)
 }
 
 func (es *eventStore) UpdateRole(ctx context.Context, session authn.Session, user users.User) (users.User, error) {
@@ -68,7 +69,7 @@ func (es *eventStore) UpdateRole(ctx context.Context, session authn.Session, use
 		return user, err
 	}
 
-	return es.update(ctx, "role", user)
+	return es.update(ctx, session, "role", user)
 }
 
 func (es *eventStore) UpdateTags(ctx context.Context, session authn.Session, user users.User) (users.User, error) {
@@ -77,7 +78,7 @@ func (es *eventStore) UpdateTags(ctx context.Context, session authn.Session, use
 		return user, err
 	}
 
-	return es.update(ctx, "tags", user)
+	return es.update(ctx, session, "tags", user)
 }
 
 func (es *eventStore) UpdateSecret(ctx context.Context, session authn.Session, oldSecret, newSecret string) (users.User, error) {
@@ -86,7 +87,7 @@ func (es *eventStore) UpdateSecret(ctx context.Context, session authn.Session, o
 		return user, err
 	}
 
-	return es.update(ctx, "secret", user)
+	return es.update(ctx, session, "secret", user)
 }
 
 func (es *eventStore) UpdateUsername(ctx context.Context, session authn.Session, id, username string) (users.User, error) {
@@ -97,6 +98,7 @@ func (es *eventStore) UpdateUsername(ctx context.Context, session authn.Session,
 
 	event := updateUsernameEvent{
 		user,
+		session,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -114,6 +116,7 @@ func (es *eventStore) UpdateProfilePicture(ctx context.Context, session authn.Se
 
 	event := updateProfilePictureEvent{
 		user,
+		session,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -129,12 +132,12 @@ func (es *eventStore) UpdateEmail(ctx context.Context, session authn.Session, id
 		return user, err
 	}
 
-	return es.update(ctx, "email", user)
+	return es.update(ctx, session, "email", user)
 }
 
-func (es *eventStore) update(ctx context.Context, operation string, user users.User) (users.User, error) {
+func (es *eventStore) update(ctx context.Context, session authn.Session, operation string, user users.User) (users.User, error) {
 	event := updateUserEvent{
-		user, operation,
+		user, operation, session,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -152,6 +155,7 @@ func (es *eventStore) View(ctx context.Context, session authn.Session, id string
 
 	event := viewUserEvent{
 		user,
+		session,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -169,6 +173,7 @@ func (es *eventStore) ViewProfile(ctx context.Context, session authn.Session) (u
 
 	event := viewProfileEvent{
 		user,
+		session,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -185,6 +190,7 @@ func (es *eventStore) ListUsers(ctx context.Context, session authn.Session, pm u
 	}
 	event := listUserEvent{
 		pm,
+		session,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -216,7 +222,10 @@ func (es *eventStore) ListMembers(ctx context.Context, session authn.Session, ob
 		return mp, err
 	}
 	event := listUserByGroupEvent{
-		pm, objectKind, objectID,
+		Page:       pm,
+		objectKind: objectKind,
+		objectID:   objectID,
+		Session:    session,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -232,7 +241,7 @@ func (es *eventStore) Enable(ctx context.Context, session authn.Session, id stri
 		return user, err
 	}
 
-	return es.delete(ctx, user)
+	return es.delete(ctx, session, user)
 }
 
 func (es *eventStore) Disable(ctx context.Context, session authn.Session, id string) (users.User, error) {
@@ -241,15 +250,16 @@ func (es *eventStore) Disable(ctx context.Context, session authn.Session, id str
 		return user, err
 	}
 
-	return es.delete(ctx, user)
+	return es.delete(ctx, session, user)
 }
 
-func (es *eventStore) delete(ctx context.Context, user users.User) (users.User, error) {
+func (es *eventStore) delete(ctx context.Context, session authn.Session, user users.User) (users.User, error) {
 	event := removeUserEvent{
 		id:        user.ID,
 		updatedAt: user.UpdatedAt,
 		updatedBy: user.UpdatedBy,
 		status:    user.Status.String(),
+		Session:   session,
 	}
 
 	if err := es.Publish(ctx, event); err != nil {
@@ -369,7 +379,8 @@ func (es *eventStore) Delete(ctx context.Context, session authn.Session, id stri
 	}
 
 	event := deleteUserEvent{
-		id: id,
+		id:      id,
+		Session: session,
 	}
 
 	return es.Publish(ctx, event)

users/middleware/authorization.go

@@ -47,7 +47,7 @@ func (am *authorizationMiddleware) View(ctx context.Context, session authn.Sessi
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       smqauth.PlatformUsersScope,
 			OptionalDomainEntityType: smqauth.DomainNullScope,
 			Operation:                smqauth.ReadOp,
@@ -68,7 +68,7 @@ func (am *authorizationMiddleware) ViewProfile(ctx context.Context, session auth
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       smqauth.PlatformUsersScope,
 			OptionalDomainEntityType: smqauth.DomainNullScope,
 			Operation:                smqauth.ReadOp,
@@ -84,7 +84,7 @@ func (am *authorizationMiddleware) ListUsers(ctx context.Context, session authn.
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       smqauth.PlatformUsersScope,
 			OptionalDomainEntityType: smqauth.DomainNullScope,
 			Operation:                smqauth.ListOp,
@@ -106,7 +106,7 @@ func (am *authorizationMiddleware) ListMembers(ctx context.Context, session auth
 		case policies.GroupsKind:
 			if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 				UserID:                   session.UserID,
-				PatID:                    session.ID,
+				PatID:                    session.PatID,
 				OptionalDomainID:         session.DomainID,
 				PlatformEntityType:       smqauth.PlatformUsersScope,
 				OptionalDomainEntityType: smqauth.DomainGroupsScope,
@@ -118,7 +118,7 @@ func (am *authorizationMiddleware) ListMembers(ctx context.Context, session auth
 		case policies.DomainsKind:
 			if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 				UserID:                   session.UserID,
-				PatID:                    session.ID,
+				PatID:                    session.PatID,
 				OptionalDomainID:         session.DomainID,
 				PlatformEntityType:       smqauth.PlatformUsersScope,
 				OptionalDomainEntityType: smqauth.DomainManagementScope,
@@ -130,7 +130,7 @@ func (am *authorizationMiddleware) ListMembers(ctx context.Context, session auth
 		case policies.ClientsKind:
 			if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 				UserID:                   session.UserID,
-				PatID:                    session.ID,
+				PatID:                    session.PatID,
 				OptionalDomainID:         session.DomainID,
 				PlatformEntityType:       smqauth.PlatformUsersScope,
 				OptionalDomainEntityType: smqauth.DomainClientsScope,
@@ -175,7 +175,7 @@ func (am *authorizationMiddleware) Update(ctx context.Context, session authn.Ses
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       smqauth.PlatformUsersScope,
 			OptionalDomainEntityType: smqauth.DomainNullScope,
 			Operation:                smqauth.UpdateOp,
@@ -196,7 +196,7 @@ func (am *authorizationMiddleware) UpdateTags(ctx context.Context, session authn
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       smqauth.PlatformUsersScope,
 			OptionalDomainEntityType: smqauth.DomainNullScope,
 			Operation:                smqauth.UpdateOp,
@@ -217,7 +217,7 @@ func (am *authorizationMiddleware) UpdateEmail(ctx context.Context, session auth
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       smqauth.PlatformUsersScope,
 			OptionalDomainEntityType: smqauth.DomainNullScope,
 			Operation:                smqauth.UpdateOp,
@@ -237,7 +237,7 @@ func (am *authorizationMiddleware) UpdateUsername(ctx context.Context, session a
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       smqauth.PlatformUsersScope,
 			OptionalDomainEntityType: smqauth.DomainNullScope,
 			Operation:                smqauth.UpdateOp,
@@ -258,7 +258,7 @@ func (am *authorizationMiddleware) UpdateProfilePicture(ctx context.Context, ses
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       smqauth.PlatformUsersScope,
 			OptionalDomainEntityType: smqauth.DomainNullScope,
 			Operation:                smqauth.UpdateOp,
@@ -283,7 +283,7 @@ func (am *authorizationMiddleware) UpdateSecret(ctx context.Context, session aut
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       smqauth.PlatformUsersScope,
 			OptionalDomainEntityType: smqauth.DomainNullScope,
 			Operation:                smqauth.UpdateOp,
@@ -308,7 +308,7 @@ func (am *authorizationMiddleware) UpdateRole(ctx context.Context, session authn
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       smqauth.PlatformUsersScope,
 			OptionalDomainEntityType: smqauth.DomainNullScope,
 			Operation:                smqauth.UpdateOp,
@@ -333,7 +333,7 @@ func (am *authorizationMiddleware) Enable(ctx context.Context, session authn.Ses
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       smqauth.PlatformUsersScope,
 			OptionalDomainEntityType: smqauth.DomainNullScope,
 			Operation:                smqauth.UpdateOp,
@@ -354,7 +354,7 @@ func (am *authorizationMiddleware) Disable(ctx context.Context, session authn.Se
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       smqauth.PlatformUsersScope,
 			OptionalDomainEntityType: smqauth.DomainNullScope,
 			Operation:                smqauth.UpdateOp,
@@ -375,7 +375,7 @@ func (am *authorizationMiddleware) Delete(ctx context.Context, session authn.Ses
 	if session.Type == authn.PersonalAccessToken {
 		if err := am.authz.AuthorizePAT(ctx, smqauthz.PatReq{
 			UserID:                   session.UserID,
-			PatID:                    session.ID,
+			PatID:                    session.PatID,
 			PlatformEntityType:       smqauth.PlatformUsersScope,
 			OptionalDomainEntityType: smqauth.DomainNullScope,
 			Operation:                smqauth.DeleteOp,