Connected Device control to event bus

muki119 · muki119 · commit d2d791bd1cdf · 2026-04-06T09:39:45.000+01:00
diff --git a/services/common/config/event_bus.ts b/services/common/config/event_bus.ts
@@ -21,6 +21,10 @@ export const STREAMS = { // streams instead of actions because certain services
         DOMAIN_CREATED: "DOMAIN_SERVICE.DOMAIN_CREATED",
         DOMAIN_DELETED: "DOMAIN_SERVICE.DOMAIN_DELETED",
         DOMAIN_UPDATED: "DOMAIN_SERVICE.DOMAIN_UPDATED",
+        DOMAIN_USER_ADDED: "DOMAIN_SERVICE.DOMAIN_USER_ADDED", // FOR Roles - sends roles
+        DOMAIN_USER_ROLE_UPDATED: "DOMAIN_SERVICE.DOMAIN_USER_ROLE_UPDATED", // sends patch
+        DOMAIN_USER_UPDATED: "DOMAIN_SERVICE.DOMAIN_USER_UPDATED",// sends patch - no need for it yet
+        DOMAIN_USER_REMOVED: "DOMAIN_SERVICE.DOMAIN_USER_REMOVED", // sends user id
     },
     DEVICE_SERVICE: {
         DEVICE_CREATED: "DEVICE_SERVICE.DEVICE_CREATED",
diff --git a/services/devicecontrol/src/bus/device_control_service_worker.ts b/services/devicecontrol/src/bus/device_control_service_worker.ts
@@ -0,0 +1,23 @@
+
+
+
+import { BaseWorker } from "@services/eventbus";
+import { STREAMS } from "@services/common/config";
+import { DomainUserAddedHandler, DomainUserRemovedHandler, DomainUserRoleUpdatedHandler, UserDeletedHandler } from "./handlers";
+import logger from "../config/logger";
+export class DeviceControlServiceWorker extends BaseWorker {
+    onCreate() {
+        this.handler(STREAMS.DOMAIN_SERVICE.DOMAIN_CREATED, DomainUserAddedHandler) // only has to create a user role for domain owner
+        this.handler(STREAMS.DOMAIN_SERVICE.DOMAIN_USER_ADDED, DomainUserAddedHandler) // should create a user role with the information 
+        this.handler(STREAMS.DOMAIN_SERVICE.DOMAIN_USER_REMOVED, DomainUserRemovedHandler) // should delete the user role 
+        this.handler(STREAMS.DOMAIN_SERVICE.DOMAIN_USER_ROLE_UPDATED, DomainUserRoleUpdatedHandler) // should update the user role with the new information
+        this.handler(STREAMS.AUTH_SERVICE.USER_DELETED, UserDeletedHandler) // should delete the user role 
+        this.errorHandler((error, payload) => {
+            logger.error({ error, payload }, "Error in DeviceControlServiceWorker:");
+        })
+    }
+
+}
+
+const worker = new DeviceControlServiceWorker();
+worker.start(); // this is the only way for the parent process to start the worker and begin communicating
diff --git a/services/devicecontrol/src/bus/handlers/domain_created_handler.ts b/services/devicecontrol/src/bus/handlers/domain_created_handler.ts
@@ -0,0 +1,16 @@
+import type { EventPayload } from "@services/eventbus";
+import { UserRoleModelInstance } from "@services/domain/src/models";
+import type { Role } from "@services/common/types";
+
+export async function DomainCreatedHandler(message: EventPayload) {
+    try {
+        // get role info from payload
+        const userRole = message?.message;
+        if (!userRole || typeof userRole === "string" || !userRole.userId || !userRole.domainId || !userRole.role) {
+            throw new Error("No user role information in payload");
+        }
+        await UserRoleModelInstance.create({ userId: userRole.userId, domainId: userRole.domainId, role: userRole.role as Role }) // create the new role in the database
+    } catch (error) {
+        throw new Error("Failed to process domain created event", { cause: error });
+    }
+}
diff --git a/services/devicecontrol/src/bus/handlers/domain_user_added_handler.ts b/services/devicecontrol/src/bus/handlers/domain_user_added_handler.ts
@@ -0,0 +1,16 @@
+import type { EventPayload } from "@services/eventbus";
+import { UserRoleModelInstance } from "@services/domain/src/models";
+import type { Role } from "@services/common/types";
+
+export async function DomainUserAddedHandler(message: EventPayload) {
+    try {
+        // get role info from payload
+        const userRole = message?.message;
+        if (!userRole || typeof userRole === "string" || !userRole.userId || !userRole.domainId || !userRole.role) {
+            throw new Error("No user role information in payload");
+        }
+        await UserRoleModelInstance.create({ userId: userRole.userId, domainId: userRole.domainId, role: userRole.role as Role }) // create the new role in the database
+    } catch (error) {
+        throw new Error("Failed to process domain user added event", { cause: error });
+    }
+}
diff --git a/services/devicecontrol/src/bus/handlers/domain_user_removed_handler.ts b/services/devicecontrol/src/bus/handlers/domain_user_removed_handler.ts
@@ -0,0 +1,13 @@
+import type { EventPayload } from "@services/eventbus";
+import { UserRoleModelInstance } from "@services/domain/src/models";
+export async function DomainUserRemovedHandler(message: EventPayload) {
+    try {
+        const userDomainInfo = message?.message;
+        if (!userDomainInfo || typeof userDomainInfo === "string" || !userDomainInfo.userId || !userDomainInfo.domainId) {
+            throw new Error("No user domain information in payload");
+        }
+        await UserRoleModelInstance.delete(userDomainInfo.userId, userDomainInfo.domainId) // delete the role associated with the deleted user in the database - this is necessary to ensure that if the user is re-added to the domain later, they don't have any lingering permissions from their previous membership
+    } catch (error) {
+        throw new Error("Failed to process domain user removed event", { cause: error });
+    }
+}
diff --git a/services/devicecontrol/src/bus/handlers/domain_user_role_updated_handler.ts b/services/devicecontrol/src/bus/handlers/domain_user_role_updated_handler.ts
@@ -0,0 +1,18 @@
+import type { EventPayload } from "@services/eventbus";
+import { UserRoleModelInstance } from "../../models";
+import type { Role } from "@services/common/types";
+
+export async function DomainUserRoleUpdatedHandler(message: EventPayload) {
+    try {
+        // get role info from payload
+        const userRoleUpdate = message?.message;
+        if (!userRoleUpdate || typeof userRoleUpdate === "string" || !userRoleUpdate.userId || !userRoleUpdate.domainId || !userRoleUpdate.role) {
+            throw new Error("No user role update information in payload");
+        }
+        // find the existing role in the database and update it with the new information
+        // this is a bit redundant since the domain service already updates the role, but it ensures that the device control service has the most up-to-date information and can react to it if needed (for example, if a user is demoted from admin to user, we might want to immediately revoke their access to certain devices or features)
+        await UserRoleModelInstance.updateRole(userRoleUpdate.userId, userRoleUpdate.domainId, userRoleUpdate.role.toUpperCase() as Role)
+    } catch (error) {
+        throw new Error("Failed to process domain user role updated event", { cause: error });
+    }
+}
diff --git a/services/devicecontrol/src/bus/handlers/index.ts b/services/devicecontrol/src/bus/handlers/index.ts
@@ -0,0 +1,4 @@
+export * from "./domain_user_added_handler";
+export * from "./domain_user_role_updated_handler";
+export * from "./domain_user_removed_handler";
+export * from "./user_deleted_handler";
diff --git a/services/devicecontrol/src/bus/handlers/user_deleted_handler.ts b/services/devicecontrol/src/bus/handlers/user_deleted_handler.ts
@@ -0,0 +1,17 @@
+import { type EventPayload } from "@services/eventbus";
+import { UserRoleModelInstance } from "../../models";
+
+export async function UserDeletedHandler(message: EventPayload): Promise<void> {
+    try {
+        const { id } = message!.message;
+        if (!id) {
+            throw new Error("Missing required fields in message");
+        }
+        const [_, err] = await UserRoleModelInstance.deleteByUserId(id);
+        if (err) {
+            throw new Error("Error deleting user roles: ", { cause: err });
+        }
+    } catch (err) {
+        throw new Error("Failed to process user deleted event: ", { cause: err });
+    }
+}
diff --git a/services/devicecontrol/src/models/user_roles_model.ts b/services/devicecontrol/src/models/user_roles_model.ts
@@ -91,6 +91,15 @@ export class UserRoleModel extends MongoAssociationModel<IUserRole> {
             return [true, null]
         }, externalSession)
     }
+    async deleteByUserId(userId: string, externalSession?: ClientSession): Promise<Result<boolean>> {
+        return await this.transactionWrap(async (session) => {
+            const result = await this.model.updateMany({ userId, deletedAt: null }, { deletedAt: new Date() }, { session }).exec()
+            if (result.modifiedCount === 0) {
+                return [null, new Error("No user roles found for deletion")]
+            }
+            return [true, null]
+        }, externalSession)
+    }
 
 }
 
diff --git a/services/domain/src/bus/handlers/user_deleted_handler.ts b/services/domain/src/bus/handlers/user_deleted_handler.ts
@@ -1,5 +1,5 @@
 import { type EventPayload } from "@services/eventbus";
-import { ProfileModelInstance, DomainModelInstance } from "../../models";
+import { ProfileModelInstance, DomainModelInstance, UserRoleModelInstance } from "../../models";
 
 export default async function HandleUserDeleted(message: EventPayload): Promise<void> {
     try {
@@ -10,6 +10,7 @@ export default async function HandleUserDeleted(message: EventPayload): Promise<
         await DomainModelInstance.multiTableTransaction(async (conn) => {
             await DomainModelInstance.deleteByOwnerId(id, conn)
             await ProfileModelInstance.delete(id, conn)
+            await UserRoleModelInstance.deleteByUserId(id, conn)
         });
     } catch (err) {
         throw new Error("Error handling USER_DELETED event: ", { cause: err });
diff --git a/services/domain/src/models/user_roles_model.ts b/services/domain/src/models/user_roles_model.ts
@@ -72,7 +72,7 @@ export class UserRoleModel extends PostgresAssociationModel<IUserRole> {
         return this.transactionWrap(async (conn) => {
             try {
                 if (!(newRole in ROLES)) {
-                    throw new Error("Invalid role specified")
+                    return [null, new Error("Invalid role specified")]
                 }
                 const updateQuery = `
                     UPDATE user_roles
@@ -110,6 +110,22 @@ export class UserRoleModel extends PostgresAssociationModel<IUserRole> {
             }
         }, externalConn)
     }
+
+    public async deleteByUserId(userId: string, externalConn?: PoolClient): Promise<void> {
+        return this.transactionWrap(async (conn) => {
+            try {
+                const deleteQuery = `
+                    UPDATE user_roles
+                    SET deleted_at = NOW()
+                    WHERE user_id = $1 AND deleted_at IS NULL
+            `
+                const values = [userId]
+                await conn.query(deleteQuery, values)
+            } catch (error) {
+                throw new Error("Failed to delete user role associations by user id: ", { cause: error })
+            }
+        }, externalConn)
+    }
 }
 
 export const UserRoleModelInstance = new UserRoleModel(PostgresPool)
diff --git a/services/domain/src/services/domain_services/add_user_service.ts b/services/domain/src/services/domain_services/add_user_service.ts
@@ -1,5 +1,10 @@
-import { UserDomainModelInstance, UserRoleModelInstance, ROLES } from "../../models";
-import { type ServiceResult } from "@services/common/types";
+import { UserDomainModelInstance, UserRoleModelInstance } from "../../models";
+import { ROLES } from "@services/common/constants";
+import { type Role, type ServiceResult } from "@services/common/types";
+import EventBusInstance from "../../config/event_bus";
+import { STREAMS } from "@services/common/config"
+
+
 
 export default async function AddUserService(inviter: string, invitee: string, domainId: string, role: string): Promise<ServiceResult<boolean>> {
     // user domain model entry
@@ -19,10 +24,15 @@ export default async function AddUserService(inviter: string, invitee: string, d
         if (!inviterIsAllowed.canManageUsers) {
             return [null, new Error("Inviter is not allowed to manage users in this domain")]
         }
-        await UserDomainModelInstance.multiTableTransaction(async (conn) => {
+        const result = await UserDomainModelInstance.multiTableTransaction(async (conn) => {
             await UserDomainModelInstance.create({ userId: invitee, domainId }, conn)
-            await UserRoleModelInstance.create({ userId: invitee, domainId, role }, conn)
+            const userRole = await UserRoleModelInstance.create({ userId: invitee, domainId, role: role.toUpperCase() as Role }, conn)
+            if (!userRole) {
+                throw new Error("Failed to create user role")
+            }
+            return userRole
         })
+        await EventBusInstance.send(STREAMS.DOMAIN_SERVICE.DOMAIN_USER_ADDED, result) // send the new user role to the event bus - flat enough to not cause problems or need stringifying
         return [true, null]
     } catch (error) {
         throw new Error("Failed to add user to domain: ", { cause: error })
diff --git a/services/domain/src/services/domain_services/create_domain_service.ts b/services/domain/src/services/domain_services/create_domain_service.ts
@@ -1,16 +1,20 @@
 import { type ServiceResult } from "@services/common/types";
-import { DomainModelInstance, UserDomainModelInstance, UserRoleModelInstance, ROLES, type IDomain, type IUserDomain, } from "../../models";
+import { DomainModelInstance, UserDomainModelInstance, UserRoleModelInstance, type IDomain, type IUserDomain, } from "../../models";
+import EventBusInstance from "../../config/event_bus";
+import { ROLES } from "@services/common/constants";
+import { STREAMS } from "@services/common/config"
 export default async function CreateDomainService(domainName: string, userID: string): Promise<ServiceResult<{ domain: IDomain, userDomain: IUserDomain }>> {
     try {
         const result = await DomainModelInstance.multiTableTransaction(async (conn) => {
             const newDomain = await DomainModelInstance.create({ name: domainName, ownerId: userID }, conn) // create a new domain
             const userDomain = await UserDomainModelInstance.create({ userId: userID, domainId: newDomain.id }, conn) // create a new user-domain relationship
-            await UserRoleModelInstance.create({ userId: userID, domainId: newDomain.id, role: ROLES.OWNER }, conn) // create a new user-role relationship with the role of owner
-            return { domain: newDomain, userDomain }
+            const ownerRole = await UserRoleModelInstance.create({ userId: userID, domainId: newDomain.id, role: ROLES.OWNER }, conn) // create a new user-role relationship with the role of owner
+            if (!newDomain || !userDomain || !ownerRole) {
+                throw new Error("Failed to create domain or user-domain relationship or user-role relationship");
+            }
+            return { domain: newDomain, userDomain, ownerRole }
         })
-        if (!result || !result.domain || !result.userDomain) {
-            throw new Error("Failed to create domain") // SHOULD NOT EVER HAPPEN
-        }
+        await EventBusInstance.send(STREAMS.DOMAIN_SERVICE.DOMAIN_CREATED, result.ownerRole) // owner role is flat enough to not cause problems or need stringifying
         return [result, null]
     } catch (error) {
         throw new Error("Error in CreateDomainService", { cause: error })
diff --git a/services/domain/src/services/domain_services/delete_user_service.ts b/services/domain/src/services/domain_services/delete_user_service.ts
@@ -1,5 +1,7 @@
 import { UserDomainModelInstance, UserRoleModelInstance } from "../../models"
 import { type ServiceResult } from "@services/common/types";
+import EventBusInstance from "../../config/event_bus";
+import { STREAMS } from "@services/common/config"
 
 export default async function DeleteUserService(userId: string, domainId: string, userToDelete: string): Promise<ServiceResult<boolean>> {
     try {
@@ -23,7 +25,9 @@ export default async function DeleteUserService(userId: string, domainId: string
         await UserDomainModelInstance.multiTableTransaction(async (conn) => {
             await UserDomainModelInstance.delete(userToDelete, domainId, conn)
             await UserRoleModelInstance.delete(userToDelete, domainId, conn)
+
         })
+        await EventBusInstance.send(STREAMS.DOMAIN_SERVICE.DOMAIN_USER_REMOVED, { userId: userToDelete, domainId }) // send the deleted user id and domain id to the event bus
         return [true, null];
     }
     catch (error) {
diff --git a/services/domain/src/services/domain_services/leave_domain_service.ts b/services/domain/src/services/domain_services/leave_domain_service.ts
@@ -1,6 +1,7 @@
 import { UserDomainModelInstance, UserRoleModelInstance } from "../../models"
 import { type ServiceResult } from "@services/common/types";
-
+import EventBusInstance from "../../config/event_bus";
+import { STREAMS } from "@services/common/config"
 export default async function LeaveDomainService(userId: string, domainId: string): Promise<ServiceResult<boolean>> {
     try {
         const isUserMember = await UserRoleModelInstance.userPermissions(userId, domainId)
@@ -14,6 +15,7 @@ export default async function LeaveDomainService(userId: string, domainId: strin
             await UserDomainModelInstance.delete(userId, domainId, conn)
             await UserRoleModelInstance.delete(userId, domainId, conn)
         })
+        await EventBusInstance.send(STREAMS.DOMAIN_SERVICE.DOMAIN_USER_REMOVED, { userId, domainId }) // send the deleted user id and domain id to the event bus
         return [true, null];
     } catch (error) {
         throw new Error("Failed to leave domain", { cause: error });
diff --git a/services/domain/src/services/domain_services/update_user_role_service.ts b/services/domain/src/services/domain_services/update_user_role_service.ts
@@ -1,14 +1,17 @@
-import type { ServiceResult, IUserRole } from "@services/common/types";
-import { UserRoleModelInstance, ROLES } from "../../models";
-
+import type { ServiceResult, IUserRole, Role } from "@services/common/types";
+import { UserRoleModelInstance } from "../../models";
+import { ROLES } from "@services/common/constants";
+import EventBusInstance from "../../config/event_bus";
+import { STREAMS } from "@services/common/config"
 
 export default async function UpdateUserRoleService(updater: string, updatee: string, newRole: string, domainId: string): Promise<ServiceResult<IUserRole>> {
     try {
 
+        newRole = newRole.toUpperCase();
         if (updatee === updater) {
             return [null, new Error("Cannot change your own role")];
         }
-        if (newRole.toUpperCase() === ROLES.OWNER) {
+        if (newRole === ROLES.OWNER) {
             return [null, new Error("Cannot assign owner role , must be done through transfer ownership")];
         }
         const updaterPermissions = await UserRoleModelInstance.userPermissions(updater, domainId);
@@ -27,16 +30,17 @@ export default async function UpdateUserRoleService(updater: string, updatee: st
         if (updateePermissions?.isOwner) { // if they are an owner - reject
             return [null, new Error("Cannot change the role of an owner")];
         }
-        if (updateePermissions.role === newRole.toUpperCase()) {
+        if (updateePermissions.role === newRole) {
             return [null, new Error("New role is the same as the current role")];
         }
-        const [updatedRole, error] = await UserRoleModelInstance.updateRole(updatee, domainId, newRole.toUpperCase());
+        const [updatedRole, error] = await UserRoleModelInstance.updateRole(updatee, domainId, ROLES[newRole as Role]);
         if (error) {
             return [null, error];
         }
-        if (!updatedRole) {
-            return [null, new Error("Role updated but updated role was not returned")];
-        }
+
+        // send the patch to the event bus 
+        await EventBusInstance.send(STREAMS.DOMAIN_SERVICE.DOMAIN_USER_ROLE_UPDATED, { userId: updatee, domainId, role: newRole });
+
         return [updatedRole, null];
     } catch (error) {
         throw new Error("Failed to update role", { cause: error });

Original file line number	Diff line number	Diff line change
`@@ -91,6 +91,15 @@ export class UserRoleModel extends MongoAssociationModel<IUserRole> {`
`91`	`91`	`return [true, null]`
`92`	`92`	`}, externalSession)`
`93`	`93`	`}`
	`94`	`+ async deleteByUserId(userId: string, externalSession?: ClientSession): Promise<Result<boolean>> {`
	`95`	`+ return await this.transactionWrap(async (session) => {`
	`96`	`+ const result = await this.model.updateMany({ userId, deletedAt: null }, { deletedAt: new Date() }, { session }).exec()`
	`97`	`+ if (result.modifiedCount === 0) {`
	`98`	`+ return [null, new Error("No user roles found for deletion")]`
	`99`	`+ }`
	`100`	`+ return [true, null]`
	`101`	`+ }, externalSession)`
	`102`	`+ }`
`94`	`103`
`95`	`104`	`}`
`96`	`105`