perf(sse+tasks): индекс SSE-соединений, getTaskTree на recursive CTE, батчи N+1
Шаги 1.4 и 1.5 плана production-готовности: - SseConnectionIndex (byOrg/byUser), один heartbeat-таймер, протокол SSE не тронут - getTaskTree/getTaskParentChain: WITH RECURSIVE CTE (было 1+2N запросов) - sync initial/delta: Promise.all по формам - embedding queue/reindex: батч-предзагрузки вместо поштучных запросов - 8 новых тестов (96/96)
This commit is contained in:
@@ -6,6 +6,7 @@ import { db } from '../db';
|
||||
import { formatUserName } from '../utils/formatUserName';
|
||||
import { tasks, forms, roles } from '@shared/schema';
|
||||
import { eq, and, inArray } from 'drizzle-orm';
|
||||
import { SseConnectionIndex } from '../utils/sse-connection-index';
|
||||
|
||||
// ── Rate-limit helpers ───────────────────────────────────────────────────────
|
||||
// In offices with a single public IP and many users, IP-only rate limiting
|
||||
@@ -61,7 +62,6 @@ export interface SSEConnection {
|
||||
userId: number;
|
||||
organizationId: number;
|
||||
res: any; // Express Response object
|
||||
heartbeat?: NodeJS.Timeout;
|
||||
}
|
||||
|
||||
export interface Event {
|
||||
@@ -78,8 +78,10 @@ export interface BufferedEvent extends Event {
|
||||
}
|
||||
|
||||
export class EventBus {
|
||||
private connections: Map<string, SSEConnection> = new Map();
|
||||
// Индекс соединений по организации/пользователю — publishEvent не перебирает все соединения
|
||||
private index = new SseConnectionIndex<SSEConnection>();
|
||||
private heartbeatInterval = 30000; // 30 seconds
|
||||
private heartbeatTimer: NodeJS.Timeout | null = null;
|
||||
private eventIdCounter = 0;
|
||||
// Буферы последних событий для восстановления после reconnect
|
||||
private userBuffers = new Map<number, BufferedEvent[]>();
|
||||
@@ -87,32 +89,39 @@ export class EventBus {
|
||||
private readonly BUFFER_SIZE = 500;
|
||||
|
||||
addConnection(connection: SSEConnection) {
|
||||
this.connections.set(connection.id, connection);
|
||||
|
||||
// Heartbeat для поддержания соединения
|
||||
connection.heartbeat = setInterval(() => {
|
||||
try {
|
||||
connection.res.write(`: heartbeat\n\n`);
|
||||
const resAny = connection.res as any;
|
||||
if (typeof resAny.flush === 'function') {
|
||||
resAny.flush();
|
||||
}
|
||||
} catch (error) {
|
||||
this.removeConnection(connection.id);
|
||||
}
|
||||
}, this.heartbeatInterval);
|
||||
this.index.add(connection);
|
||||
this.ensureHeartbeatTimer();
|
||||
}
|
||||
|
||||
removeConnection(connectionId: string) {
|
||||
const connection = this.connections.get(connectionId);
|
||||
if (connection) {
|
||||
if (connection.heartbeat) {
|
||||
clearInterval(connection.heartbeat);
|
||||
}
|
||||
this.connections.delete(connectionId);
|
||||
this.index.remove(connectionId);
|
||||
// Когда соединений нет, останавливаем heartbeat-таймер, чтобы не будить event loop впустую
|
||||
if (this.index.size === 0 && this.heartbeatTimer) {
|
||||
clearInterval(this.heartbeatTimer);
|
||||
this.heartbeatTimer = null;
|
||||
}
|
||||
}
|
||||
|
||||
// Один глобальный heartbeat-таймер на все соединения (вместо таймера на каждое соединение)
|
||||
private ensureHeartbeatTimer() {
|
||||
if (this.heartbeatTimer) return;
|
||||
this.heartbeatTimer = setInterval(() => {
|
||||
this.index.forEach((connection) => {
|
||||
try {
|
||||
connection.res.write(`: heartbeat\n\n`);
|
||||
const resAny = connection.res as any;
|
||||
if (typeof resAny.flush === 'function') {
|
||||
resAny.flush();
|
||||
}
|
||||
} catch (error) {
|
||||
this.removeConnection(connection.id);
|
||||
}
|
||||
});
|
||||
}, this.heartbeatInterval);
|
||||
// unref, чтобы таймер не удерживал процесс (и не мешал тестам завершаться)
|
||||
this.heartbeatTimer.unref?.();
|
||||
}
|
||||
|
||||
private generateEventId(): string {
|
||||
return `${Date.now()}-${++this.eventIdCounter}`;
|
||||
}
|
||||
@@ -163,15 +172,6 @@ export class EventBus {
|
||||
timestamp: Date.now(),
|
||||
};
|
||||
|
||||
// Логируем chat-события для диагностики задержек/пропусков
|
||||
const isChatEvent = event.type === 'conv_message_created' || event.type === 'conv_message_updated' ||
|
||||
event.type === 'conv_message_deleted' || event.type === 'conv_read' || event.type === 'message_created' ||
|
||||
event.type === 'message_read';
|
||||
if (isChatEvent) {
|
||||
const target = event.userId ? `user=${event.userId}` : `org=${event.organizationId}`;
|
||||
console.log(`[SSE] publish ${event.type} ${target} conn=${this.connections.size} id=${bufferedEvent.id}`);
|
||||
}
|
||||
|
||||
// Буферизуем событие для восстановления после reconnect
|
||||
if (event.userId) {
|
||||
this.pushToBuffer(this.userBuffers, event.userId, bufferedEvent);
|
||||
@@ -179,24 +179,9 @@ export class EventBus {
|
||||
this.pushToBuffer(this.orgBuffers, event.organizationId, bufferedEvent);
|
||||
}
|
||||
|
||||
let delivered = 0;
|
||||
this.connections.forEach((connection, id) => {
|
||||
// Проверяем tenant isolation
|
||||
if (event.organizationId && connection.organizationId !== event.organizationId) {
|
||||
return; // Пропускаем соединения из других организаций
|
||||
}
|
||||
|
||||
// Если событие для конкретного пользователя
|
||||
if (event.userId && connection.userId !== event.userId) {
|
||||
return;
|
||||
}
|
||||
|
||||
const ok = this.writeEventToConnection(connection, bufferedEvent);
|
||||
if (ok) delivered++;
|
||||
});
|
||||
|
||||
if (isChatEvent) {
|
||||
console.log(`[SSE] delivered ${event.type} to ${delivered}/${this.connections.size} connection(s)`);
|
||||
// Рассылка только соединениям нужной организации/пользователя (tenant isolation в индексе)
|
||||
for (const connection of this.index.getForEventTarget(event.organizationId, event.userId)) {
|
||||
this.writeEventToConnection(connection, bufferedEvent);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -237,14 +222,11 @@ export class EventBus {
|
||||
}
|
||||
|
||||
getActiveConnections() {
|
||||
return this.connections.size;
|
||||
return this.index.size;
|
||||
}
|
||||
|
||||
isUserConnected(userId: number, organizationId: number): boolean {
|
||||
for (const conn of this.connections.values()) {
|
||||
if (conn.userId === userId && conn.organizationId === organizationId) return true;
|
||||
}
|
||||
return false;
|
||||
return this.index.isUserConnected(userId, organizationId);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user