445 lines
16 KiB
TypeScript
445 lines
16 KiB
TypeScript
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 crypto from 'crypto';
|
|
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();
|
|
}
|
|
|
|
async register(): Promise<void> {
|
|
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;
|
|
|
|
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}`);
|
|
|
|
const registrationKey = process.env.REGISTRATION_KEY || '';
|
|
|
|
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',
|
|
...(registrationKey ? { 'x-registration-key': registrationKey } : {}),
|
|
},
|
|
body: JSON.stringify(manifest),
|
|
});
|
|
|
|
if (response.ok) {
|
|
const resBody = await response.json();
|
|
const data = resBody && resBody.data !== undefined ? resBody.data : resBody;
|
|
|
|
this.logger.log(`Plugin registered successfully!`);
|
|
|
|
if (data && data.dbCredentials) {
|
|
const creds = data.dbCredentials;
|
|
process.env.PLUGIN_DB_NAME = creds.dbName;
|
|
if (creds.host && !process.env.DB_HOST) process.env.PLUGIN_DB_HOST = creds.host;
|
|
if (creds.port && !process.env.DB_PORT) process.env.PLUGIN_DB_PORT = String(creds.port);
|
|
process.env.PLUGIN_DB_USER = creds.username;
|
|
process.env.PLUGIN_DB_PASSWORD = creds.password;
|
|
this.logger.log(`Database credentials received for plugin ${manifest.id}`);
|
|
}
|
|
|
|
registered = true;
|
|
this.startHeartbeatLoop(manifest.id, backendUrl);
|
|
} else {
|
|
const errText = await response.text();
|
|
this.logger.error(`Registration failed (${response.status}): ${errText}`);
|
|
await new Promise(resolve => setTimeout(resolve, 5000));
|
|
}
|
|
} catch (error) {
|
|
this.logger.error(`Registration error: ${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);
|
|
}
|
|
|
|
async callPluginApi<T = any>(pluginId: string, path: string, body?: any): Promise<T> {
|
|
return this.fetchCore(`/api/plugins/rpc/${pluginId}/${path.replace(/^\//, '')}`, {
|
|
method: 'POST',
|
|
body: body !== undefined ? JSON.stringify(body) : undefined,
|
|
});
|
|
}
|
|
|
|
async callAsService<T = any>(pluginId: string, path: string, body?: any): Promise<T> {
|
|
const registrationKey = process.env.REGISTRATION_KEY || '';
|
|
if (!registrationKey) {
|
|
throw new Error('REGISTRATION_KEY not configured — cannot make service-to-service call');
|
|
}
|
|
const payload = JSON.stringify({
|
|
pluginId,
|
|
path: path.replace(/^\//, ''),
|
|
timestamp: Date.now(),
|
|
body: body !== undefined ? body : null,
|
|
});
|
|
const signature = crypto.createHmac('sha256', registrationKey).update(payload).digest('hex');
|
|
const encoded = Buffer.from(payload).toString('base64');
|
|
const serviceAuth = `v1.${encoded}.${signature}`;
|
|
|
|
const coreUrl = process.env.CORE_BACKEND_URL || 'http://localhost:3001';
|
|
const response = await fetch(`${coreUrl}/api/plugins/rpc/${pluginId}/${path.replace(/^\//, '')}`, {
|
|
method: 'POST',
|
|
headers: { 'Content-Type': 'application/json', 'X-Service-Auth': serviceAuth },
|
|
body: body !== undefined ? JSON.stringify(body) : undefined,
|
|
});
|
|
if (!response.ok) {
|
|
const text = await response.text().catch(() => response.statusText);
|
|
throw new Error(`Service call to ${pluginId}/${path} failed: ${response.status} ${text}`);
|
|
}
|
|
const text = await response.text();
|
|
try { return JSON.parse(text); } catch { return text as any; }
|
|
}
|
|
|
|
async discoverPlugins(): Promise<any[]> {
|
|
try {
|
|
const response = await this.fetchCore('/api/plugins/discovery');
|
|
if (Array.isArray(response)) return response;
|
|
if (response && Array.isArray(response.data)) return response.data;
|
|
return [];
|
|
} catch {
|
|
return [];
|
|
}
|
|
}
|
|
|
|
async findPluginsByEvent(eventName: string): Promise<any[]> {
|
|
const plugins = await this.discoverPlugins();
|
|
return plugins.filter((p) =>
|
|
p.events?.some((e: { name: string }) => e.name === eventName),
|
|
);
|
|
}
|
|
|
|
async emitToPlugin(pluginId: string, pattern: string, payload: any): Promise<void> {
|
|
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}`);
|
|
}
|
|
}
|
|
|
|
async requestPlugin<T = any>(
|
|
pluginId: string,
|
|
pattern: string,
|
|
payload: any,
|
|
timeoutMs: number = 5000,
|
|
): Promise<T | null> {
|
|
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<T | null>((resolve) => {
|
|
const timer = setTimeout(() => {
|
|
this.logger.warn(`[requestPlugin] Timeout for ${routingKey}`);
|
|
if (errorFallback) clearTimeout(errorFallback);
|
|
resolve(null);
|
|
}, timeoutMs);
|
|
|
|
let errorFallback: ReturnType<typeof setTimeout> | null = null;
|
|
|
|
ch.consume(replyQueue.queue, (rmqMsg) => {
|
|
if (rmqMsg && rmqMsg.properties.correlationId === correlationId) {
|
|
try {
|
|
const parsed = JSON.parse(rmqMsg.content.toString());
|
|
const response = parsed.response ?? parsed;
|
|
|
|
// Skip "no matching handler" errors from other services
|
|
// and wait for the correct response (up to timeoutMs - 1s)
|
|
if (response && response.err) {
|
|
if (!errorFallback) {
|
|
errorFallback = setTimeout(() => {
|
|
clearTimeout(timer);
|
|
resolve(response);
|
|
}, Math.min(timeoutMs - 1000, 15000));
|
|
}
|
|
return;
|
|
}
|
|
|
|
if (errorFallback) clearTimeout(errorFallback);
|
|
clearTimeout(timer);
|
|
resolve(response);
|
|
} catch {
|
|
clearTimeout(timer);
|
|
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<T>(pattern: string, payload: T): Promise<void> {
|
|
const event: SystemEvent<T> = {
|
|
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<void> {
|
|
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<T = string>(key: string): Promise<T | null> {
|
|
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<void> {
|
|
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<void> {
|
|
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<void> {
|
|
await this.emit('system.notification.send', {
|
|
userId: payload.userId,
|
|
title: payload.title,
|
|
message: payload.message,
|
|
icon: payload.icon,
|
|
image: payload.image,
|
|
details: payload.details,
|
|
});
|
|
}
|
|
|
|
getDbName(): string {
|
|
return process.env.PLUGIN_DB_NAME || process.env.DB_DATABASE || 'ultimate_crm';
|
|
}
|
|
|
|
static async setupDatabase(options: { pluginId: string }): Promise<string> {
|
|
const coreUrl = process.env.CORE_BACKEND_URL || 'http://localhost:3001';
|
|
const registrationKey = process.env.REGISTRATION_KEY || '';
|
|
|
|
const headers: Record<string, string> = { 'Content-Type': 'application/json' };
|
|
if (registrationKey) {
|
|
headers['x-registration-key'] = registrationKey;
|
|
}
|
|
|
|
const response = await fetch(`${coreUrl}/api/db/setup`, {
|
|
method: 'POST',
|
|
headers,
|
|
body: JSON.stringify({ pluginId: options.pluginId }),
|
|
});
|
|
if (!response.ok) {
|
|
throw new Error(`Database setup failed: ${response.status} ${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.DB_HOST) process.env.PLUGIN_DB_HOST = data.host;
|
|
if (data.port && !process.env.DB_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;
|
|
}
|
|
}
|