import { Injectable, Inject, Logger } from '@nestjs/common'; import { ClientProxy } from '@nestjs/microservices'; import { firstValueFrom, timeout as rxTimeout } from 'rxjs'; import { SystemEvent } from '../interfaces/communication'; import { v4 as uuidv4 } from 'uuid'; import * as fs from 'fs'; import * as path from 'path'; import * as amqp from 'amqplib'; import { requestContext } from '../interceptors/request-context.interceptor'; @Injectable() export class CoreService { private readonly logger = new Logger(CoreService.name); private readonly serviceName = process.env.SERVICE_NAME || 'plugin'; private heartbeatInterval: any = null; constructor( @Inject('RABBITMQ_CLIENT') private readonly client: ClientProxy, ) {} /** * Спеціальний метод для виконання запитів до Ядра з автоматичним прокиданням авторизації */ async fetchCore(endpoint: string, options: RequestInit = {}) { const coreUrl = process.env.CORE_BACKEND_URL || 'http://localhost:3001'; const store = requestContext.getStore(); const authHeader = store?.get('authorization'); const headers = { 'Content-Type': 'application/json', ...(authHeader ? { 'Authorization': authHeader } : {}), ...((options.headers as any) || {}), }; this.logger.log(`[CoreRequest] Sending request to ${endpoint} (Auth: ${!!authHeader})`); const response = await fetch(`${coreUrl}${endpoint}`, { ...options, headers, }); if (!response.ok) { this.logger.error(`[CoreRequest] Request to ${endpoint} failed with status ${response.status}`); } return response.json(); } // ... (решта методів register, emit, request залишаються без змін) async register(): Promise { const coreUrl = process.env.CORE_BACKEND_URL || 'http://localhost:3001'; const possiblePaths = [ path.resolve(process.cwd(), '..', 'manifest.json'), path.resolve(process.cwd(), 'manifest.json'), ]; let manifestPath = ''; for (const p of possiblePaths) { if (fs.existsSync(p)) { manifestPath = p; break; } } if (!manifestPath) return; // Dynamically calculate the backendUrl const port = process.env.PORT || '3002'; const serviceName = process.env.SERVICE_NAME || 'plugin'; const isDocker = coreUrl.includes('core-backend') || (!coreUrl.includes('localhost') && !coreUrl.includes('127.0.0.1')); const backendUrl = process.env.PLUGIN_BACKEND_URL || (isDocker ? `http://${serviceName}:${port}` : `http://localhost:${port}`); let registered = false; while (!registered) { try { const manifest = JSON.parse(fs.readFileSync(manifestPath, 'utf8')); if (process.env.PUBLIC_URL) { let p = '/remoteEntry.js'; if (manifest.remoteEntry) { try { if (manifest.remoteEntry.startsWith('http://') || manifest.remoteEntry.startsWith('https://')) { p = new URL(manifest.remoteEntry).pathname; } else { p = manifest.remoteEntry; } } catch (e) { p = manifest.remoteEntry; } } const publicUrl = process.env.PUBLIC_URL.endsWith('/') ? process.env.PUBLIC_URL.slice(0, -1) : process.env.PUBLIC_URL; const pathname = p.startsWith('/') ? p : '/' + p; manifest.remoteEntry = `${publicUrl}${pathname}`; } manifest.backendUrl = backendUrl; const response = await fetch(`${coreUrl}/api/plugins/register`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(manifest), }); if (response.ok) { this.logger.log(`Plugin registered successfully!`); registered = true; this.startHeartbeatLoop(manifest.id, backendUrl); } else { await new Promise(resolve => setTimeout(resolve, 5000)); } } catch (error) { await new Promise(resolve => setTimeout(resolve, 5000)); } } } private startHeartbeatLoop(pluginId: string, backendUrl: string) { if (this.heartbeatInterval) { clearInterval(this.heartbeatInterval); } this.heartbeatInterval = setInterval(async () => { try { const coreUrl = process.env.CORE_BACKEND_URL || 'http://localhost:3001'; this.logger.log(`[Heartbeat] Sending heartbeat for ${pluginId}`); const response = await fetch(`${coreUrl}/api/plugins/heartbeat`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ id: pluginId, backendUrl }), }); if (response.ok) { const resBody = await response.json(); const data = resBody && resBody.data !== undefined ? resBody.data : resBody; if (data && data.registered === false) { this.logger.warn(`[Heartbeat] Core does not recognize plugin ${pluginId}. Re-registering.`); this.register(); } else { this.logger.log(`[Heartbeat] Heartbeat acknowledged by core.`); } } else { this.logger.error(`[Heartbeat] Failed to send heartbeat. Status: ${response.status}`); } } catch (error: any) { this.logger.error(`[Heartbeat] Error sending heartbeat:`, error.message || error); } }, 30 * 1000); // 30 seconds } /** * Викликати API іншого плагіна через core RPC-проксі. * Запит проксується ядром на backendUrl цільового плагіна методом POST. */ async callPluginApi(pluginId: string, path: string, body?: any): Promise { return this.fetchCore(`/api/plugins/rpc/${pluginId}/${path.replace(/^\//, '')}`, { method: 'POST', body: body !== undefined ? JSON.stringify(body) : undefined, }); } /** * Отримати список усіх зареєстрованих плагінів з їхніми events, objectTypes, backendUrl тощо. * Корисно для discovery — дізнатись, які плагіни доступні та що вони вміють. */ async discoverPlugins(): Promise { const response = await this.fetchCore('/api/plugins/discovery'); return response.data || response || []; } /** * Знайти плагіни, які оголосили певну подію у своєму manifest.json (поле events). */ async findPluginsByEvent(eventName: string): Promise { const plugins = await this.discoverPlugins(); return plugins.filter((p) => p.events?.some((e: { name: string }) => e.name === eventName), ); } /** * Надіслати подію конкретному плагіну через RabbitMQ. * Публікує в system_events з routingKey: plugin... * Цільовий плагін отримує подію через @EventPattern('plugin..') * або більш загальний @EventPattern('plugin.*'). * * Інші плагіни теж отримують подію (через # binding), але за конвенцією * мають ігнорувати події з префіксом plugin.<чужойId>. * * IMPORTANT: Uses raw amqplib to publish to system_events exchange (not ClientProxy), * because NestJS 10 ClientRMQ ignores the exchange option and sends directly to a queue. */ async emitToPlugin(pluginId: string, pattern: string, payload: any): Promise { const routingKey = `plugin.${pluginId}.${pattern}`; const event: SystemEvent = { eventId: uuidv4(), pattern: routingKey, payload, source: this.serviceName, timestamp: Date.now(), }; try { const conn = await amqp.connect(this.getRMQUrl()); const ch = await conn.createChannel(); ch.publish('system_events', routingKey, Buffer.from(JSON.stringify(event)), { persistent: true, contentType: 'application/json', }); await ch.close(); await conn.close(); this.logger.log(`[emitToPlugin] Published to ${routingKey}`); } catch (err: any) { this.logger.error(`[emitToPlugin] Failed for ${routingKey}: ${err.message}`); } } /** * RPC-виклик до іншого плагіна з очікуванням відповіді. * Використовує NestJS RabbitMQ RPC (send / @MessagePattern). * Публікує в system_events exchange (raw amqplib, not ClientProxy). * Цільовий плагін має обробити запит через @MessagePattern('plugin..'). * * @param pluginId — ID цільового плагіна * @param pattern — назва команди * @param payload — тіло запиту * @param timeoutMs — таймаут очікування (за замовчуванням 5000 мс) * @returns відповідь від цільового плагіна або null по таймауту */ async requestPlugin( pluginId: string, pattern: string, payload: any, timeoutMs: number = 5000, ): Promise { const routingKey = `plugin.${pluginId}.${pattern}`; const correlationId = uuidv4(); try { const conn = await amqp.connect(this.getRMQUrl()); const ch = await conn.createChannel(); const replyQueue = await ch.assertQueue('', { exclusive: true, autoDelete: true }); const msg = { id: correlationId, pattern: routingKey, data: payload, }; ch.publish('system_events', routingKey, Buffer.from(JSON.stringify(msg)), { correlationId, replyTo: replyQueue.queue, persistent: true, contentType: 'application/json', }); const result = await new Promise((resolve) => { const timer = setTimeout(() => { this.logger.warn(`[requestPlugin] Timeout for ${routingKey}`); resolve(null); }, timeoutMs); ch.consume(replyQueue.queue, (rmqMsg) => { if (rmqMsg && rmqMsg.properties.correlationId === correlationId) { clearTimeout(timer); try { const parsed = JSON.parse(rmqMsg.content.toString()); resolve(parsed.response ?? parsed); } catch { resolve(null); } } }, { noAck: true }); }); await ch.close(); await conn.close(); return result; } catch (err: any) { this.logger.error(`[requestPlugin] Failed for ${routingKey}: ${err.message}`); return null; } } private getRMQUrl(): string { return process.env.RABBITMQ_URL || 'amqp://localhost:5672'; } async emit(pattern: string, payload: T): Promise { const event: SystemEvent = { eventId: uuidv4(), pattern, payload, source: this.serviceName, timestamp: Date.now(), }; this.client.emit(pattern, event).subscribe({ error: (err) => this.logger.error(`Failed to emit event ${pattern}:`, err), }); } async auditLog(action: string, resource: string, details?: any): Promise { const store = requestContext.getStore(); const authHeader = store?.get('authorization'); const ipAddress = store?.get('ip'); const userAgent = store?.get('userAgent'); let userId: number | undefined; if (authHeader && authHeader.startsWith('Bearer ')) { try { const token = authHeader.split(' ')[1]; const base64Url = token.split('.')[1]; const base64 = base64Url.replace(/-/g, '+').replace(/_/g, '/'); const jsonPayload = Buffer.from(base64, 'base64').toString(); const payload = JSON.parse(jsonPayload); userId = payload.sub || payload.userId || payload.id; } catch (e) { this.logger.error('Failed to decode JWT for audit log', e); } } await this.emit('system.audit.log', { userId, action, resource, details, ipAddress, userAgent, }); } async getGlobalData(key: string): Promise { const response = await this.fetchCore(`/api/store/${key}`, { headers: { 'x-plugin-id': this.serviceName, }, }); if (response.success) { if (response.data === null || response.data === undefined) { return null; } try { return JSON.parse(response.data) as T; } catch (e) { return response.data as unknown as T; } } return null; } async setGlobalData(key: string, value: any): Promise { const stringValue = typeof value === 'string' ? value : JSON.stringify(value); await this.fetchCore(`/api/store/${key}`, { method: 'POST', body: JSON.stringify({ value: stringValue }), headers: { 'x-plugin-id': this.serviceName, }, }); } async deleteGlobalData(key: string): Promise { await this.fetchCore(`/api/store/${key}`, { method: 'DELETE', headers: { 'x-plugin-id': this.serviceName, }, }); } async sendNotification(payload: { userId: number | null; title: string; message: string; icon?: string; image?: string; details?: any }): Promise { await this.emit('system.notification.send', { userId: payload.userId, title: payload.title, message: payload.message, icon: payload.icon, image: payload.image, details: payload.details, }); } /** * Отримати назву БД, наданої ядром (має сенс після виклику setupDatabase) */ getDbName(): string { return process.env.PLUGIN_DB_NAME || process.env.DB_DATABASE || 'ultimate_crm'; } /** * Зареєструвати БД плагіна в ядрі (створює БД p_, якщо ще не існує). * Викликати ДО створення NestJS app, щоб TypeORM міг підхопити PLUGIN_DB_NAME. */ static async setupDatabase(options: { pluginId: string }): Promise { const coreUrl = process.env.CORE_BACKEND_URL || 'http://localhost:3001'; const response = await fetch(`${coreUrl}/api/db/setup`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ pluginId: options.pluginId }), }); if (!response.ok) { throw new Error(`Database setup failed: ${response.statusText}`); } const result = await response.json(); const data = result.data || result; const dbName = data.dbName; if (!dbName) throw new Error('Database name not returned from core'); process.env.PLUGIN_DB_NAME = dbName; if (data.host) process.env.PLUGIN_DB_HOST = data.host; if (data.port) process.env.PLUGIN_DB_PORT = String(data.port); if (data.username) process.env.PLUGIN_DB_USER = data.username; if (data.password) process.env.PLUGIN_DB_PASSWORD = data.password; return dbName; } }