remove user online status

This commit is contained in:
Anton SHEPILOV
2024-02-03 14:25:42 +01:00
parent b35687829c
commit ceee21b5eb
18 changed files with 4695 additions and 9195 deletions
@@ -80,7 +80,6 @@ export default class RoomManager implements RealtimeRoomManager {
try {
//Public rooms we just check the user is logged in
if (
joinEvent.name.startsWith("/users/online") || //User online room
joinEvent.name.startsWith("/users/") || //User update room
joinEvent.name === "/ping"
) {
@@ -25,7 +25,6 @@ import { ConsoleServiceImpl } from "./console/service";
import { DocumentsService } from "./documents/services";
import { DocumentsEngine } from "./documents/services/engine";
import { FileServiceImpl } from "./files/services";
import OnlineServiceImpl from "./online/service";
import { PreviewProcessService } from "./previews/services/files/processing/service";
import { StatisticsServiceImpl } from "./statistics/service";
import { TagsService } from "./tags/services/tags";
@@ -68,7 +67,6 @@ type TdriveServices = {
hooks: ApplicationHooksService;
};
files: FileServiceImpl;
online: OnlineServiceImpl;
documents: {
documents: DocumentsService;
engine: DocumentsEngine;
@@ -132,7 +130,6 @@ class GlobalResolver {
hooks: await new ApplicationHooksService().init(),
},
files: await new FileServiceImpl().init(),
online: await new OnlineServiceImpl().init(),
documents: {
documents: await new DocumentsService().init(),
engine: await new DocumentsEngine().init(),
@@ -1,25 +0,0 @@
import { TdriveServiceProvider } from "../../core/platform/framework";
import UserOnline, { UserOnlinePrimaryKey } from "./entities/user-online";
export type OnlineEvent = {
company_id: string;
user_id: string;
is_online: boolean;
};
export type UsersOnlineMessage = Array<OnlineEvent>;
export type OnlineGetRequest = {
/* Array of ids to get status */
data: Array<string>;
};
export type OnlineGetResponse = {
data: Array<[string, boolean]>;
};
export interface OnlineServiceAPI extends TdriveServiceProvider {
isOnline(userId: string): Promise<boolean>;
setLastSeenOnline(userIds: Array<string>, lastSeen: number, is_connected: boolean): Promise<void>;
get(userId: UserOnlinePrimaryKey): Promise<UserOnline>;
}
@@ -1,2 +0,0 @@
export const DISCONNECTED_DELAY = 900000;
export const ONLINE_TOPIC = "user:online";
@@ -1,32 +0,0 @@
import { Type } from "class-transformer";
import { merge } from "lodash";
import { Column, Entity } from "../../../core/platform/services/database/services/orm/decorators";
export const TYPE = "user_online";
@Entity(TYPE, {
primaryKey: ["user_id"],
type: TYPE,
})
export default class UserOnline {
@Type(() => String)
@Column("user_id", "timeuuid", { generator: "timeuuid" })
user_id: string;
@Type(() => Boolean)
@Column("is_connected", "boolean")
is_connected: boolean;
/**
* Save the date the user has been seen connected
*/
@Type(() => Number)
@Column("last_seen", "number")
last_seen: number;
}
export type UserOnlinePrimaryKey = { user_id: string };
export function getInstance(element: Partial<UserOnline>): UserOnline {
return merge(new UserOnline(), element);
}
@@ -1,15 +0,0 @@
import { Prefix, TdriveService } from "../../core/platform/framework";
@Prefix("/internal/services/online/v1")
export default class OnlineService extends TdriveService<undefined> {
version = "1";
name = "online";
api(): undefined {
return undefined;
}
public async doInit(): Promise<this> {
return this;
}
}
@@ -1,29 +0,0 @@
import { getLogger, Initializable, TdriveLogger } from "../../../core/platform/framework";
import { UsersOnlineMessage } from "../api";
import { ONLINE_TOPIC } from "../constants";
import { UserOnlineProcessor } from "./processor";
import gr from "../../global-resolver";
export class OnlinePubsubService implements Initializable {
private logger: TdriveLogger;
constructor() {
this.logger = getLogger("online.pubsub.OnlinePubsubService");
}
async init(): Promise<this> {
gr.platformServices.messageQueue.processor.addHandler(new UserOnlineProcessor());
return this;
}
async broadcastOnline(online: UsersOnlineMessage = []): Promise<void> {
if (!online.length) {
return;
}
this.logger.debug(`Publishing online users ${online.map(u => u.user_id).join(",")}`);
return gr.platformServices.messageQueue.publish<UsersOnlineMessage>(ONLINE_TOPIC, {
data: online,
});
}
}
@@ -1,52 +0,0 @@
import _ from "lodash";
import { getLogger, TdriveLogger } from "../../../core/platform/framework";
import { MessageQueueHandler } from "../../../core/platform/services/message-queue/api";
import { websocketEventBus } from "../../../core/platform/services/realtime/bus";
import {
RealtimeEntityActionType,
ResourcePath,
} from "../../../core/platform/services/realtime/types";
import { UsersOnlineMessage } from "../api";
import { ONLINE_TOPIC } from "../constants";
export class UserOnlineProcessor implements MessageQueueHandler<UsersOnlineMessage, void> {
private logger: TdriveLogger;
readonly topics = {
in: ONLINE_TOPIC,
};
readonly options = {
unique: false,
ack: true,
};
readonly name = "UserOnlineProcessor";
constructor() {
this.logger = getLogger(`online.pubsub.${this.name}`);
}
validate(message: UsersOnlineMessage): boolean {
return !!(message && message.length);
}
async process(message: UsersOnlineMessage): Promise<void> {
this.logger.debug(
`Pushing user online status for users ${message.map(u => u.user_id).join(",")}`,
);
const grouped_data = _.groupBy(message, "company_id");
Object.values(grouped_data).forEach((messagePerCompany: UsersOnlineMessage) => {
websocketEventBus.publish(RealtimeEntityActionType.Event, {
type: "user:online",
room: ResourcePath.get(`/users/online/${messagePerCompany[0].company_id}`),
entity: { online: messagePerCompany },
resourcePath: null,
result: null,
});
});
return;
}
}
@@ -1,174 +0,0 @@
import Repository from "../../../core/platform/services/database/services/orm/repository/repository";
import { OnlineGetRequest, OnlineGetResponse, OnlineServiceAPI } from "../api";
import { OnlinePubsubService } from "../pubsub";
import { DISCONNECTED_DELAY } from "../constants";
import UserOnline, {
getInstance,
TYPE as ONLINE_TYPE,
UserOnlinePrimaryKey,
} from "../entities/user-online";
import gr from "../../global-resolver";
import { getLogger, TdriveLogger, TdriveServiceProvider } from "../../../core/platform/framework";
import { getUserRoom } from "../../../services/user/realtime";
import User from "../../../services/user/entities/user";
import { WebsocketUserEvent } from "../../../core/platform/services/websocket/types";
import { ExecutionContext } from "../../../core/platform/framework/api/crud-service";
export default class OnlineServiceImpl implements TdriveServiceProvider {
version = "1";
service: OnlineServiceAPI;
private pubsubService: OnlinePubsubService;
onlineRepository: Repository<UserOnline>;
private logger: TdriveLogger;
constructor() {
this.logger = getLogger("online.service");
}
public async init(): Promise<this> {
this.onlineRepository = await gr.database.getRepository(ONLINE_TYPE, UserOnline);
this.pubsubService = new OnlinePubsubService();
await this.pubsubService.init();
gr.platformServices.websocket.onUserConnected(async event => {
const user: User = await gr.services.users.get({
id: event.user.id,
});
const companies = user?.cache?.companies;
this.logger.info("User connected", event.user.id);
// save the last connection date
this.setLastSeenOnline([event.user.id], Date.now(), true);
// broadcast to global pubsub so that everyone can publish to websockets
this.broadcastOnline(event, companies);
event.socket.on(
"online:get",
async (request: OnlineGetRequest, ack: (response: OnlineGetResponse) => void) => {
this.logger.debug(`Got an online:get request for ${(request.data || []).length} users`);
ack({ data: await this.getOnlineStatuses(request.data) });
},
);
event.socket.on(
"online:set",
async (request: OnlineGetRequest, ack: () => void): Promise<void> => {
this.logger.debug(`Got an online:set request for ${(request.data || []).length} users`);
this.broadcastOnline(event, companies);
this.setLastSeenOnline([event.user.id], Date.now(), true);
ack();
},
);
});
gr.platformServices.websocket.onUserDisconnected(async event => {
this.logger.info("User disconnected", event.user.id);
// Since the user can be connected on several nodes, we cannot directly set it status to offline
const room = getUserRoom(event.user.id);
const userSockets = await event.socket.in(room).allSockets();
if (userSockets.size === 0) {
this.pubsubService.broadcastOnline([
{
company_id: "toBeFetched",
user_id: event.user.id,
is_online: false,
},
]);
this.setLastSeenOnline([event.user.id], Date.now(), false);
}
});
return this;
}
private async getOnlineStatuses(ids: Array<string>): Promise<Array<[string, boolean]>> {
console.log("socket getOnlineStatuses", ids);
return this.areOnline(ids);
}
async setLastSeenOnline(
userIds: Array<string> = [],
date: number,
is_connected: boolean,
): Promise<void> {
this.logger.debug(`setLastSeenOnline ${userIds.join(",")}`);
if (!userIds.length) {
return;
}
const last_seen = date || Date.now();
const uniqueIds = new Set<string>(userIds);
this.logger.info(`Update last active state for users ${userIds.join(",")}`);
const onlineUsers: UserOnline[] = Array.from(uniqueIds.values()).map(user_id =>
getInstance({ user_id, last_seen, is_connected }),
);
await this.onlineRepository.saveAll(onlineUsers);
//Send websocket event
onlineUsers.forEach(u => {
gr.services.users.publishPublicUserRealtime(u.user_id);
});
}
async isOnline(userId: string, context?: ExecutionContext): Promise<boolean> {
const user = await this.onlineRepository.findOne({ user_id: userId }, {}, context);
if (!user) {
return false;
}
return Date.now() - user.last_seen < DISCONNECTED_DELAY;
}
private async areOnline(
ids: Array<string> = [],
context?: ExecutionContext,
): Promise<Array<[string, boolean]>> {
const users = [];
//This foreach is needed for $in operators https://github.com/linagora/Tdrive/issues/1246
for (let i = 0; i < ids.length; i += 100) {
users.push(
...(
await this.onlineRepository.find(
{},
{ $in: [["user_id", ids.slice(i, i + 100)]] },
context,
)
).getEntities(),
);
}
return users.map(user => [
user.user_id,
this.isStillConnected(user.last_seen, user.is_connected),
]);
}
/**
* let's say that a user is connected when its last connection is more than some delay ago
*/
private isStillConnected(date: number, is_connected: boolean): boolean {
return Date.now() - date < DISCONNECTED_DELAY && is_connected;
}
private broadcastOnline(event: WebsocketUserEvent, companies: Array<string>): void {
(companies || []).forEach(company => {
this.pubsubService.broadcastOnline([
{
company_id: company,
user_id: event.user.id,
is_online: true,
},
]);
});
}
async get(pk: UserOnlinePrimaryKey): Promise<UserOnline> {
return await this.onlineRepository.findOne(pk);
}
}
-12
View File
@@ -35,18 +35,6 @@ export async function formatUser(
cache: { companies: user.cache?.companies || [] },
} as UserObject;
const userOnline = await gr.services.online.get({ user_id: user.id });
if (userOnline) {
const { last_seen, is_connected } = userOnline;
resUser = {
...resUser,
last_seen,
is_connected,
last_activity: last_seen,
};
}
if (options?.includeCompanies) {
const userCompanies = await gr.services.users.getUserCompanies({ id: user.id });