Mirrors/colanode
1
0
Fork 0
mirror of https://github.com/colanode/colanode.git synced 2025-12-28 16:06:37 +01:00
Code Issues Packages Projects Releases Wiki Activity

Refactor events

Browse Source
This commit is contained in:
Hakan Shehu
2024-10-27 19:00:41 +01:00
parent 902c09b809
commit 7b699b2b1e
1 changed files with 37 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
server/src/queues/events.ts
View File

@@ -57,9 +57,27 @@ const handleEventJob = async (job: Job) => {
const handleNodeCreatedEvent = async (
event: NodeCreatedEvent,
): Promise<void> => {
await createNodeUserStates(event);
await publishChange(event.id, event.workspaceId, 'node_create');
};
const handleNodeUpdatedEvent = async (
event: NodeUpdatedEvent,
): Promise<void> => {
await checkForCollaboratorsChange(event);
await publishChange(event.id, event.workspaceId, 'node_update');
};
const handleNodeDeletedEvent = async (
event: NodeDeletedEvent,
): Promise<void> => {
await publishChange(event.id, event.workspaceId, 'node_delete');
};
const createNodeUserStates = async (event: NodeCreatedEvent): Promise<void> => {
const userStatesToCreate: CreateNodeUserState[] = [];
if (event.attributes.type === NodeTypes.User) {
const userIds = await fetchWorkspaceUsers(event.workspaceId);
const userStatesToCreate: CreateNodeUserState[] = [];
for (const userId of userIds) {
userStatesToCreate.push({
@@ -92,46 +110,33 @@ const handleNodeCreatedEvent = async (
updated_at: null,
});
}
if (userStatesToCreate.length > 0) {
await database
.insertInto('node_user_states')
.values(userStatesToCreate)
.onConflict((cb) => cb.doNothing())
.execute();
}
} else {
const collaborators = await fetchNodeCollaborators(event.id);
const collaboratorIds = collaborators.map((c) => c.collaboratorId);
for (const collaboratorId of collaboratorIds) {
userStatesToCreate.push({
user_id: collaboratorId,
node_id: event.id,
last_seen_version_id: null,
workspace_id: event.workspaceId,
last_seen_at: null,
mentions_count: 0,
created_at: new Date(),
access_removed_at: null,
version_id: generateId(IdType.Version),
updated_at: null,
});
}
}
if (userStatesToCreate.length > 0) {
await database
.insertInto('node_user_states')
.values(
collaboratorIds.map((collaboratorId) => ({
user_id: collaboratorId,
node_id: event.id,
last_seen_version_id: null,
workspace_id: event.workspaceId,
last_seen_at: null,
mentions_count: 0,
created_at: new Date(),
access_removed_at: null,
version_id: generateId(IdType.Version),
updated_at: null,
})),
)
.values(userStatesToCreate)
.onConflict((cb) => cb.doNothing())
.execute();
}
await publishChange(event.id, event.workspaceId, 'node_create');
};
const handleNodeUpdatedEvent = async (
event: NodeUpdatedEvent,
): Promise<void> => {
await checkForCollaboratorsChange(event);
await publishChange(event.id, event.workspaceId, 'node_update');
};
const checkForCollaboratorsChange = async (
@@ -242,12 +247,6 @@ const extractCollaboratorIds = (collaborators: ServerNodeAttributes) => {
return Object.keys(collaborators.collaborators).sort();
};
const handleNodeDeletedEvent = async (
event: NodeDeletedEvent,
): Promise<void> => {
await publishChange(event.id, event.workspaceId, 'node_delete');
};
const publishChange = async (
nodeId: string,
workspaceId: string,