[FIX] MAU when using micro services (#24204)
Diego Sampaio
3 years ago
committed by
GitHub
parent
56f892401f
commit
37714424bf
@ -1,394 +0,0 @@ |
||||
import { Meteor } from 'meteor/meteor'; |
||||
import { Accounts } from 'meteor/accounts-base'; |
||||
import { SyncedCron } from 'meteor/littledata:synced-cron'; |
||||
import UAParser from 'ua-parser-js'; |
||||
|
||||
import { UAParserMobile, UAParserDesktop } from './UAParserCustom'; |
||||
import { Sessions } from '../../../models/server/raw'; |
||||
import { aggregates } from '../../../models/server/raw/Sessions'; |
||||
import { Logger } from '../../../logger'; |
||||
import { getMostImportantRole } from './getMostImportantRole'; |
||||
|
||||
const getDateObj = (dateTime = new Date()) => ({ |
||||
day: dateTime.getDate(), |
||||
month: dateTime.getMonth() + 1, |
||||
year: dateTime.getFullYear(), |
||||
}); |
||||
|
||||
const isSameDateObj = (oldest, newest) => oldest.year === newest.year && oldest.month === newest.month && oldest.day === newest.day; |
||||
|
||||
const logger = new Logger('SAUMonitor'); |
||||
|
||||
/** |
||||
* Server Session Monitor for SAU(Simultaneously Active Users) based on Meteor server sessions |
||||
*/ |
||||
export class SAUMonitorClass { |
||||
constructor() { |
||||
this._started = false; |
||||
this._monitorTime = 60000; |
||||
this._timer = null; |
||||
this._today = getDateObj(); |
||||
this._instanceId = null; |
||||
this._jobName = 'aggregate-sessions'; |
||||
} |
||||
|
||||
async start(instanceId) { |
||||
if (this.isRunning()) { |
||||
return; |
||||
} |
||||
|
||||
this._instanceId = instanceId; |
||||
|
||||
if (!this._instanceId) { |
||||
logger.debug('[start] - InstanceId is not defined.'); |
||||
return; |
||||
} |
||||
|
||||
await this._startMonitoring(() => { |
||||
this._started = true; |
||||
logger.debug(`[start] - InstanceId: ${this._instanceId}`); |
||||
}); |
||||
} |
||||
|
||||
stop() { |
||||
if (!this.isRunning()) { |
||||
return; |
||||
} |
||||
|
||||
this._started = false; |
||||
|
||||
if (this._timer) { |
||||
Meteor.clearInterval(this._timer); |
||||
} |
||||
|
||||
SyncedCron.remove(this._jobName); |
||||
|
||||
logger.debug(`[stop] - InstanceId: ${this._instanceId}`); |
||||
} |
||||
|
||||
isRunning() { |
||||
return this._started === true; |
||||
} |
||||
|
||||
async _startMonitoring(callback) { |
||||
try { |
||||
this._handleAccountEvents(); |
||||
this._handleOnConnection(); |
||||
this._startSessionControl(); |
||||
await this._initActiveServerSessions(); |
||||
this._startAggregation(); |
||||
if (callback) { |
||||
callback(); |
||||
} |
||||
} catch (err) { |
||||
throw new Meteor.Error(err); |
||||
} |
||||
} |
||||
|
||||
_startSessionControl() { |
||||
if (this.isRunning()) { |
||||
return; |
||||
} |
||||
|
||||
if (this._monitorTime < 0) { |
||||
return; |
||||
} |
||||
|
||||
this._timer = Meteor.setInterval(async () => { |
||||
await this._updateActiveSessions(); |
||||
}, this._monitorTime); |
||||
} |
||||
|
||||
_handleOnConnection() { |
||||
if (this.isRunning()) { |
||||
return; |
||||
} |
||||
|
||||
Meteor.onConnection((connection) => { |
||||
if (!this.isRunning()) { |
||||
return; |
||||
} |
||||
// this._handleSession(connection, getDateObj());
|
||||
|
||||
connection.onClose(async () => { |
||||
await Sessions.closeByInstanceIdAndSessionId(this._instanceId, connection.id); |
||||
}); |
||||
}); |
||||
} |
||||
|
||||
_handleAccountEvents() { |
||||
if (this.isRunning()) { |
||||
return; |
||||
} |
||||
|
||||
Accounts.onLogin(async (info) => { |
||||
if (!this.isRunning()) { |
||||
return; |
||||
} |
||||
|
||||
const { roles, _id: userId } = info.user; |
||||
|
||||
const mostImportantRole = getMostImportantRole(roles); |
||||
|
||||
const loginAt = new Date(); |
||||
const params = { userId, roles, mostImportantRole, loginAt, ...getDateObj() }; |
||||
await this._handleSession(info.connection, params); |
||||
this._updateConnectionInfo(info.connection.id, { loginAt }); |
||||
}); |
||||
|
||||
Accounts.onLogout(async (info) => { |
||||
if (!this.isRunning()) { |
||||
return; |
||||
} |
||||
|
||||
const sessionId = info.connection.id; |
||||
if (info.user) { |
||||
const userId = info.user._id; |
||||
await Sessions.logoutByInstanceIdAndSessionIdAndUserId(this._instanceId, sessionId, userId); |
||||
} |
||||
}); |
||||
} |
||||
|
||||
async _handleSession(connection, params) { |
||||
const data = this._getConnectionInfo(connection, params); |
||||
await Sessions.createOrUpdate(data); |
||||
} |
||||
|
||||
async _updateActiveSessions() { |
||||
if (!this.isRunning()) { |
||||
return; |
||||
} |
||||
|
||||
const { year, month, day } = this._today; |
||||
const currentDateTime = new Date(); |
||||
const currentDay = getDateObj(currentDateTime); |
||||
|
||||
if (!isSameDateObj(this._today, currentDay)) { |
||||
const beforeDateTime = new Date(this._today.year, this._today.month - 1, this._today.day, 23, 59, 59, 999); |
||||
const nextDateTime = new Date(currentDay.year, currentDay.month - 1, currentDay.day); |
||||
|
||||
const createSessions = async (objects, ids) => { |
||||
await Sessions.createBatch(objects); |
||||
|
||||
Meteor.defer(() => { |
||||
Sessions.updateActiveSessionsByDateAndInstanceIdAndIds({ year, month, day }, this._instanceId, ids, { |
||||
lastActivityAt: beforeDateTime, |
||||
}); |
||||
}); |
||||
}; |
||||
this._applyAllServerSessionsBatch(createSessions, { |
||||
createdAt: nextDateTime, |
||||
lastActivityAt: nextDateTime, |
||||
...currentDay, |
||||
}); |
||||
this._today = currentDay; |
||||
return; |
||||
} |
||||
|
||||
// Otherwise, just update the lastActivityAt field
|
||||
await this._applyAllServerSessionsIds(async (sessions) => { |
||||
await Sessions.updateActiveSessionsByDateAndInstanceIdAndIds({ year, month, day }, this._instanceId, sessions, { |
||||
lastActivityAt: currentDateTime, |
||||
}); |
||||
}); |
||||
} |
||||
|
||||
_getConnectionInfo(connection, params = {}) { |
||||
if (!connection) { |
||||
return; |
||||
} |
||||
|
||||
const ip = connection.httpHeaders |
||||
? connection.httpHeaders['x-real-ip'] || connection.httpHeaders['x-forwarded-for'] |
||||
: connection.clientAddress; |
||||
const host = connection.httpHeaders && connection.httpHeaders.host; |
||||
const info = { |
||||
type: 'session', |
||||
sessionId: connection.id, |
||||
instanceId: this._instanceId, |
||||
ip, |
||||
host, |
||||
...this._getUserAgentInfo(connection), |
||||
...params, |
||||
}; |
||||
|
||||
if (connection.loginAt) { |
||||
info.loginAt = connection.loginAt; |
||||
} |
||||
|
||||
return info; |
||||
} |
||||
|
||||
_getUserAgentInfo(connection) { |
||||
if (!(connection && connection.httpHeaders && connection.httpHeaders['user-agent'])) { |
||||
return; |
||||
} |
||||
|
||||
const uaString = connection.httpHeaders['user-agent']; |
||||
let result; |
||||
|
||||
if (UAParserMobile.isMobileApp(uaString)) { |
||||
result = UAParserMobile.uaObject(uaString); |
||||
} else if (UAParserDesktop.isDesktopApp(uaString)) { |
||||
result = UAParserDesktop.uaObject(uaString); |
||||
} else { |
||||
const ua = new UAParser(uaString); |
||||
result = ua.getResult(); |
||||
} |
||||
|
||||
const info = { |
||||
type: 'other', |
||||
}; |
||||
|
||||
const removeEmptyProps = (obj) => { |
||||
Object.keys(obj).forEach((p) => (!obj[p] || obj[p] === undefined) && delete obj[p]); |
||||
return obj; |
||||
}; |
||||
|
||||
if (result.browser && result.browser.name) { |
||||
info.type = 'browser'; |
||||
info.name = result.browser.name; |
||||
info.longVersion = result.browser.version; |
||||
} |
||||
|
||||
if (result.os && result.os.name) { |
||||
info.os = removeEmptyProps(result.os); |
||||
} |
||||
|
||||
if (result.device && (result.device.type || result.device.model)) { |
||||
info.type = result.device.type; |
||||
|
||||
if (result.app && result.app.name) { |
||||
info.name = result.app.name; |
||||
info.longVersion = result.app.version; |
||||
if (result.app.bundle) { |
||||
info.longVersion += ` ${result.app.bundle}`; |
||||
} |
||||
} |
||||
} |
||||
|
||||
if (typeof info.longVersion === 'string') { |
||||
info.version = info.longVersion.match(/(\d+\.){0,2}\d+/)[0]; |
||||
} |
||||
|
||||
return { |
||||
device: info, |
||||
}; |
||||
} |
||||
|
||||
async _initActiveServerSessions() { |
||||
await this._applyAllServerSessions(async (connectionHandle) => { |
||||
await this._handleSession(connectionHandle, getDateObj()); |
||||
}); |
||||
} |
||||
|
||||
async _applyAllServerSessions(callback) { |
||||
if (!callback || typeof callback !== 'function') { |
||||
return; |
||||
} |
||||
|
||||
const sessions = Object.values(Meteor.server.sessions).filter((session) => session.userId); |
||||
for await (const session of sessions) { |
||||
await callback(session.connectionHandle); |
||||
} |
||||
} |
||||
|
||||
async recursive(callback, sessionIds) { |
||||
await callback(sessionIds.splice(0, 500)); |
||||
|
||||
if (sessionIds.length) { |
||||
await this.recursive(callback, sessionIds); |
||||
} |
||||
} |
||||
|
||||
async _applyAllServerSessionsIds(callback) { |
||||
if (!callback || typeof callback !== 'function') { |
||||
return; |
||||
} |
||||
|
||||
const sessionIds = Object.values(Meteor.server.sessions) |
||||
.filter((session) => session.userId) |
||||
.map((s) => s.id); |
||||
await this.recursive(callback, sessionIds); |
||||
} |
||||
|
||||
_updateConnectionInfo(sessionId, data = {}) { |
||||
if (!sessionId) { |
||||
return; |
||||
} |
||||
const session = Meteor.server.sessions.get(sessionId); |
||||
if (session) { |
||||
Object.keys(data).forEach((p) => { |
||||
session.connectionHandle = Object.assign({}, session.connectionHandle, { [p]: data[p] }); |
||||
}); |
||||
} |
||||
} |
||||
|
||||
_applyAllServerSessionsBatch(callback, params) { |
||||
const batch = (arr, limit) => { |
||||
if (!arr.length) { |
||||
return Promise.resolve(); |
||||
} |
||||
const ids = []; |
||||
return Promise.all( |
||||
arr.splice(0, limit).map((item) => { |
||||
ids.push(item.id); |
||||
return this._getConnectionInfo(item.connectionHandle, params); |
||||
}), |
||||
) |
||||
.then(async (data) => { |
||||
await callback(data, ids); |
||||
return batch(arr, limit); |
||||
}) |
||||
.catch((e) => { |
||||
logger.debug(`Error: ${e.message}`); |
||||
}); |
||||
}; |
||||
|
||||
const sessions = Object.values(Meteor.server.sessions).filter((session) => session.userId); |
||||
batch(sessions, 500); |
||||
} |
||||
|
||||
_startAggregation() { |
||||
logger.info('[aggregate] - Start Cron.'); |
||||
|
||||
SyncedCron.add({ |
||||
name: this._jobName, |
||||
schedule: (parser) => parser.text('at 2:00 am'), |
||||
job: async () => { |
||||
await this.aggregate(); |
||||
}, |
||||
}); |
||||
} |
||||
|
||||
async aggregate() { |
||||
if (!this.isRunning()) { |
||||
return; |
||||
} |
||||
|
||||
logger.info('[aggregate] - Aggregating data.'); |
||||
|
||||
const date = new Date(); |
||||
date.setDate(date.getDate() - 0); // yesterday
|
||||
const yesterday = getDateObj(date); |
||||
|
||||
const match = { |
||||
type: 'session', |
||||
year: { $lte: yesterday.year }, |
||||
month: { $lte: yesterday.month }, |
||||
day: { $lte: yesterday.day }, |
||||
}; |
||||
|
||||
await aggregates.dailySessionsOfYesterday(Sessions.col, yesterday).forEach(async (record) => { |
||||
record._id = `${record.userId}-${record.year}-${record.month}-${record.day}`; |
||||
await Sessions.updateOne({ _id: record._id }, { $set: record }, { upsert: true }); |
||||
}); |
||||
|
||||
await Sessions.updateMany(match, { |
||||
$set: { |
||||
type: 'computed-session', |
||||
_computedAt: new Date(), |
||||
}, |
||||
}); |
||||
} |
||||
} |
||||
@ -0,0 +1,342 @@ |
||||
import { Meteor } from 'meteor/meteor'; |
||||
import { SyncedCron } from 'meteor/littledata:synced-cron'; |
||||
import UAParser from 'ua-parser-js'; |
||||
import mem from 'mem'; |
||||
|
||||
import { UAParserMobile, UAParserDesktop } from './UAParserCustom'; |
||||
import { Sessions, Users } from '../../../models/server/raw'; |
||||
import { aggregates } from '../../../models/server/raw/Sessions'; |
||||
import { Logger } from '../../../../server/lib/logger/Logger'; |
||||
import { getMostImportantRole } from './getMostImportantRole'; |
||||
import { sauEvents } from '../../../../server/services/sauMonitor/events'; |
||||
import { ISession, ISessionDevice } from '../../../../definition/ISession'; |
||||
import { ISocketConnection } from '../../../../definition/ISocketConnection'; |
||||
import { IUser } from '../../../../definition/IUser'; |
||||
|
||||
type DateObj = { day: number; month: number; year: number }; |
||||
|
||||
const getDateObj = (dateTime = new Date()): DateObj => ({ |
||||
day: dateTime.getDate(), |
||||
month: dateTime.getMonth() + 1, |
||||
year: dateTime.getFullYear(), |
||||
}); |
||||
|
||||
const logger = new Logger('SAUMonitor'); |
||||
|
||||
const getUserRoles = mem( |
||||
async (userId: string): Promise<string[]> => { |
||||
const user = await Users.findOneById<IUser>(userId, { projection: { roles: 1 } }); |
||||
|
||||
return user?.roles || []; |
||||
}, |
||||
{ maxAge: 5000 }, |
||||
); |
||||
|
||||
/** |
||||
* Server Session Monitor for SAU(Simultaneously Active Users) based on Meteor server sessions |
||||
*/ |
||||
export class SAUMonitorClass { |
||||
private _started: boolean; |
||||
|
||||
private _dailyComputeJobName: string; |
||||
|
||||
private _dailyFinishSessionsJobName: string; |
||||
|
||||
constructor() { |
||||
this._started = false; |
||||
this._dailyComputeJobName = 'aggregate-sessions'; |
||||
this._dailyFinishSessionsJobName = 'aggregate-sessions'; |
||||
} |
||||
|
||||
async start(): Promise<void> { |
||||
if (this.isRunning()) { |
||||
return; |
||||
} |
||||
|
||||
await this._startMonitoring(); |
||||
|
||||
this._started = true; |
||||
logger.debug('[start]'); |
||||
} |
||||
|
||||
stop(): void { |
||||
if (!this.isRunning()) { |
||||
return; |
||||
} |
||||
|
||||
this._started = false; |
||||
|
||||
SyncedCron.remove(this._dailyComputeJobName); |
||||
SyncedCron.remove(this._dailyFinishSessionsJobName); |
||||
|
||||
logger.debug('[stop]'); |
||||
} |
||||
|
||||
isRunning(): boolean { |
||||
return this._started === true; |
||||
} |
||||
|
||||
async _startMonitoring(): Promise<void> { |
||||
try { |
||||
this._handleAccountEvents(); |
||||
this._handleOnConnection(); |
||||
this._startCronjobs(); |
||||
} catch (err: any) { |
||||
throw new Meteor.Error(err); |
||||
} |
||||
} |
||||
|
||||
private _handleOnConnection(): void { |
||||
if (this.isRunning()) { |
||||
return; |
||||
} |
||||
|
||||
sauEvents.on('socket.disconnected', async ({ id, instanceId }) => { |
||||
if (!this.isRunning()) { |
||||
return; |
||||
} |
||||
|
||||
await Sessions.closeByInstanceIdAndSessionId(instanceId, id); |
||||
}); |
||||
} |
||||
|
||||
private _handleAccountEvents(): void { |
||||
if (this.isRunning()) { |
||||
return; |
||||
} |
||||
|
||||
sauEvents.on('accounts.login', async ({ userId, connection }) => { |
||||
if (!this.isRunning()) { |
||||
return; |
||||
} |
||||
|
||||
const roles = await getUserRoles(userId); |
||||
|
||||
const mostImportantRole = getMostImportantRole(roles); |
||||
|
||||
const loginAt = new Date(); |
||||
const params = { userId, roles, mostImportantRole, loginAt, ...getDateObj() }; |
||||
await this._handleSession(connection, params); |
||||
}); |
||||
|
||||
sauEvents.on('accounts.logout', async ({ userId, connection }) => { |
||||
if (!this.isRunning()) { |
||||
return; |
||||
} |
||||
|
||||
await Sessions.logoutByInstanceIdAndSessionIdAndUserId(connection.instanceId, connection.id, userId); |
||||
}); |
||||
} |
||||
|
||||
private async _handleSession( |
||||
connection: ISocketConnection, |
||||
params: Pick<ISession, 'userId' | 'mostImportantRole' | 'loginAt' | 'day' | 'month' | 'year' | 'roles'>, |
||||
): Promise<void> { |
||||
const data = this._getConnectionInfo(connection, params); |
||||
if (!data) { |
||||
return; |
||||
} |
||||
await Sessions.createOrUpdate(data); |
||||
} |
||||
|
||||
private async _finishSessionsFromDate(yesterday: Date, today: Date): Promise<void> { |
||||
if (!this.isRunning()) { |
||||
return; |
||||
} |
||||
|
||||
const { day, month, year } = getDateObj(yesterday); |
||||
const beforeDateTime = new Date(year, month - 1, day, 23, 59, 59, 999); |
||||
|
||||
const currentDate = getDateObj(today); |
||||
const nextDateTime = new Date(currentDate.year, currentDate.month - 1, currentDate.day); |
||||
|
||||
const cursor = Sessions.findSessionsNotClosedByDateWithoutLastActivity({ year, month, day }); |
||||
|
||||
const batch = []; |
||||
|
||||
for await (const session of cursor) { |
||||
// create a new session for the current day
|
||||
batch.push({ |
||||
...session, |
||||
...currentDate, |
||||
createdAt: nextDateTime, |
||||
}); |
||||
|
||||
if (batch.length === 500) { |
||||
await Sessions.createBatch(batch); |
||||
batch.length = 0; |
||||
} |
||||
} |
||||
|
||||
if (batch.length > 0) { |
||||
await Sessions.createBatch(batch); |
||||
} |
||||
|
||||
// close all sessions from current 'date'
|
||||
await Sessions.updateActiveSessionsByDate( |
||||
{ year, month, day }, |
||||
{ |
||||
lastActivityAt: beforeDateTime, |
||||
}, |
||||
); |
||||
|
||||
// TODO missing an action to perform on dangling sessions (for example remove sessions not closed one month ago)
|
||||
} |
||||
|
||||
private _getConnectionInfo( |
||||
connection: ISocketConnection, |
||||
params: Pick<ISession, 'userId' | 'mostImportantRole' | 'loginAt' | 'day' | 'month' | 'year' | 'roles'>, |
||||
): Omit<ISession, '_id' | '_updatedAt' | 'createdAt'> | undefined { |
||||
if (!connection) { |
||||
return; |
||||
} |
||||
|
||||
const ip = connection.clientAddress || connection.httpHeaders?.['x-real-ip'] || connection.httpHeaders?.['x-forwarded-for']; |
||||
|
||||
const host = connection.httpHeaders?.host || ''; |
||||
|
||||
return { |
||||
type: 'session', |
||||
sessionId: connection.id, |
||||
instanceId: connection.instanceId, |
||||
ip: (Array.isArray(ip) ? ip[0] : ip) || '', |
||||
host, |
||||
...this._getUserAgentInfo(connection), |
||||
...params, |
||||
}; |
||||
} |
||||
|
||||
private _getUserAgentInfo(connection: ISocketConnection): { device: ISessionDevice } | undefined { |
||||
if (!connection?.httpHeaders?.['user-agent']) { |
||||
return; |
||||
} |
||||
|
||||
const uaString = connection.httpHeaders['user-agent']; |
||||
|
||||
// TODO define a type for "result" below
|
||||
// | UAParser.IResult
|
||||
// | { device: { type: string; model?: string }; browser: undefined; os: undefined; app: { name: string; version: string } }
|
||||
// | {
|
||||
// device: { type: string; model?: string };
|
||||
// browser: undefined;
|
||||
// os: string;
|
||||
// app: { name: string; version: string };
|
||||
// }
|
||||
|
||||
const result = ((): any => { |
||||
if (UAParserMobile.isMobileApp(uaString)) { |
||||
return UAParserMobile.uaObject(uaString); |
||||
} |
||||
|
||||
if (UAParserDesktop.isDesktopApp(uaString)) { |
||||
return UAParserDesktop.uaObject(uaString); |
||||
} |
||||
|
||||
const ua = new UAParser(uaString); |
||||
return ua.getResult(); |
||||
})(); |
||||
|
||||
const info: ISessionDevice = { |
||||
type: 'other', |
||||
name: '', |
||||
longVersion: '', |
||||
os: { |
||||
name: '', |
||||
version: '', |
||||
}, |
||||
version: '', |
||||
}; |
||||
|
||||
const removeEmptyProps = (obj: any): any => { |
||||
Object.keys(obj).forEach((p) => (!obj[p] || obj[p] === undefined) && delete obj[p]); |
||||
return obj; |
||||
}; |
||||
|
||||
if (result.browser && result.browser.name) { |
||||
info.type = 'browser'; |
||||
info.name = result.browser.name; |
||||
info.longVersion = result.browser.version || ''; |
||||
} |
||||
|
||||
if (typeof result.os !== 'string' && result.os?.name) { |
||||
info.os = removeEmptyProps(result.os) || ''; |
||||
} |
||||
|
||||
if (result.device && (result.device.type || result.device.model)) { |
||||
info.type = result.device.type || ''; |
||||
|
||||
if (result.hasOwnProperty('app') && result.app?.name) { |
||||
info.name = result.app.name; |
||||
info.longVersion = result.app.version; |
||||
if (result.app.bundle) { |
||||
info.longVersion += ` ${result.app.bundle}`; |
||||
} |
||||
} |
||||
} |
||||
|
||||
if (typeof info.longVersion === 'string') { |
||||
info.version = info.longVersion.match(/(\d+\.){0,2}\d+/)?.[0] || ''; |
||||
} |
||||
|
||||
return { |
||||
device: info, |
||||
}; |
||||
} |
||||
|
||||
private _startCronjobs(): void { |
||||
logger.info('[aggregate] - Start Cron.'); |
||||
|
||||
SyncedCron.add({ |
||||
name: this._dailyComputeJobName, |
||||
schedule: (parser: any) => parser.text('at 2:00 am'), |
||||
job: async () => { |
||||
await this._aggregate(); |
||||
}, |
||||
}); |
||||
|
||||
SyncedCron.add({ |
||||
name: this._dailyFinishSessionsJobName, |
||||
schedule: (parser: any) => parser.text('at 1:05 am'), |
||||
job: async () => { |
||||
const yesterday = new Date(); |
||||
yesterday.setDate(yesterday.getDate() - 1); |
||||
|
||||
await this._finishSessionsFromDate(yesterday, new Date()); |
||||
}, |
||||
}); |
||||
} |
||||
|
||||
private async _aggregate(): Promise<void> { |
||||
if (!this.isRunning()) { |
||||
return; |
||||
} |
||||
|
||||
logger.info('[aggregate] - Aggregating data.'); |
||||
|
||||
const date = new Date(); |
||||
date.setDate(date.getDate() - 0); // yesterday
|
||||
const yesterday = getDateObj(date); |
||||
|
||||
const match = { |
||||
type: 'session', |
||||
year: { $lte: yesterday.year }, |
||||
month: { $lte: yesterday.month }, |
||||
day: { $lte: yesterday.day }, |
||||
}; |
||||
|
||||
for await (const record of aggregates.dailySessionsOfYesterday(Sessions.col, yesterday)) { |
||||
await Sessions.updateOne( |
||||
{ _id: `${record.userId}-${record.year}-${record.month}-${record.day}` }, |
||||
{ $set: record }, |
||||
{ upsert: true }, |
||||
); |
||||
} |
||||
|
||||
await Sessions.updateMany(match, { |
||||
$set: { |
||||
type: 'computed-session', |
||||
_computedAt: new Date(), |
||||
}, |
||||
}); |
||||
} |
||||
} |
||||
@ -0,0 +1,10 @@ |
||||
import type { IncomingHttpHeaders } from 'http'; |
||||
|
||||
export interface ISocketConnection { |
||||
id: string; |
||||
instanceId: string; |
||||
livechatToken?: string; |
||||
onClose(fn: (...args: any[]) => void): void; |
||||
clientAddress: string | undefined; |
||||
httpHeaders: IncomingHttpHeaders; |
||||
} |
||||
@ -0,0 +1,5 @@ |
||||
declare module 'meteor/konecty:multiple-instances-status' { |
||||
namespace InstanceStatus { |
||||
function id(): string; |
||||
} |
||||
} |
||||
@ -0,0 +1 @@ |
||||
import './sauMonitorHooks'; |
||||
@ -0,0 +1,42 @@ |
||||
import type { IncomingHttpHeaders } from 'http'; |
||||
|
||||
import { InstanceStatus } from 'meteor/konecty:multiple-instances-status'; |
||||
import { Accounts } from 'meteor/accounts-base'; |
||||
import { Meteor } from 'meteor/meteor'; |
||||
|
||||
import { sauEvents } from '../services/sauMonitor/events'; |
||||
|
||||
Accounts.onLogin((info: { user: Meteor.User; connection: Meteor.Connection }) => { |
||||
const { httpHeaders } = info.connection; |
||||
|
||||
sauEvents.emit('accounts.login', { |
||||
userId: info.user._id, |
||||
connection: { instanceId: InstanceStatus.id(), ...info.connection, httpHeaders: httpHeaders as IncomingHttpHeaders }, |
||||
}); |
||||
}); |
||||
|
||||
Accounts.onLogout((info: { user: Meteor.User; connection: Meteor.Connection }) => { |
||||
const { httpHeaders } = info.connection; |
||||
|
||||
sauEvents.emit('accounts.logout', { |
||||
userId: info.user._id, |
||||
connection: { instanceId: InstanceStatus.id(), ...info.connection, httpHeaders: httpHeaders as IncomingHttpHeaders }, |
||||
}); |
||||
}); |
||||
|
||||
Meteor.onConnection((connection) => { |
||||
connection.onClose(async () => { |
||||
const { httpHeaders } = connection; |
||||
sauEvents.emit('socket.disconnected', { |
||||
instanceId: InstanceStatus.id(), |
||||
...connection, |
||||
httpHeaders: httpHeaders as IncomingHttpHeaders, |
||||
}); |
||||
}); |
||||
}); |
||||
|
||||
Meteor.onConnection((connection) => { |
||||
const { httpHeaders } = connection; |
||||
|
||||
sauEvents.emit('socket.connected', { instanceId: InstanceStatus.id(), ...connection, httpHeaders: httpHeaders as IncomingHttpHeaders }); |
||||
}); |
||||
@ -0,0 +1,3 @@ |
||||
import { IServiceClass } from './ServiceClass'; |
||||
|
||||
export type ISAUMonitorService = IServiceClass; |
||||
@ -0,0 +1,10 @@ |
||||
import { Emitter } from '@rocket.chat/emitter'; |
||||
|
||||
import { ISocketConnection } from '../../../definition/ISocketConnection'; |
||||
|
||||
export const sauEvents = new Emitter<{ |
||||
'accounts.login': { userId: string; connection: ISocketConnection }; |
||||
'accounts.logout': { userId: string; connection: ISocketConnection }; |
||||
'socket.connected': ISocketConnection; |
||||
'socket.disconnected': ISocketConnection; |
||||
}>(); |
||||
@ -0,0 +1,31 @@ |
||||
// import type { Db } from 'mongodb';
|
||||
|
||||
import { ServiceClass } from '../../sdk/types/ServiceClass'; |
||||
import { ISAUMonitorService } from '../../sdk/types/ISAUMonitorService'; |
||||
import { sauEvents } from './events'; |
||||
|
||||
export class SAUMonitorService extends ServiceClass implements ISAUMonitorService { |
||||
protected name = 'sau-monitor'; |
||||
|
||||
constructor() { |
||||
super(); |
||||
|
||||
this.onEvent('accounts.login', async (data) => { |
||||
sauEvents.emit('accounts.login', data); |
||||
}); |
||||
|
||||
this.onEvent('accounts.logout', async (data) => { |
||||
sauEvents.emit('accounts.logout', data); |
||||
}); |
||||
|
||||
this.onEvent('socket.disconnected', async (data) => { |
||||
// console.log('socket.disconnected', data);
|
||||
sauEvents.emit('socket.disconnected', data); |
||||
}); |
||||
|
||||
this.onEvent('socket.connected', async (data) => { |
||||
// console.log('socket.connected', data);
|
||||
sauEvents.emit('socket.connected', data); |
||||
}); |
||||
} |
||||
} |
||||
Loading…
Reference in new issue