restored backup
This commit is contained in:
andrei-mihnea-cerbu
+120
-160
@@ -1,209 +1,169 @@
|
||||
import { JsonManager } from './json_manager';
|
||||
import { TcpCommunicator } from "./tcp_communicator";
|
||||
import { operationCodes } from '../network/operation_codes';
|
||||
import path from 'path';
|
||||
import fs from 'fs';
|
||||
import crypto from 'crypto';
|
||||
import path from 'path';
|
||||
import { FileEncryptor } from './file_encryptor';
|
||||
import { MemoryManager } from './memory_manager';
|
||||
import { JsonManager } from './json_manager';
|
||||
import { TcpCommunicator } from './tcp_communicator';
|
||||
import { operationCodes } from '../network/operation_codes';
|
||||
import { parentPort } from 'worker_threads';
|
||||
import { ParsedMessage } from "../network/message_handler";
|
||||
|
||||
export class BackupRetrievalWorker {
|
||||
private userConfig: JsonManager;
|
||||
export class BackupManager {
|
||||
private fileEncryptor: FileEncryptor | null = null;
|
||||
private memoryManager: MemoryManager;
|
||||
private applicationInfo: JsonManager;
|
||||
private userConfig: JsonManager;
|
||||
private readonly clientPort: number;
|
||||
private readonly destinationPath: string;
|
||||
private encryptionKey: Buffer | null = null;
|
||||
private iv: Buffer | null = null;
|
||||
private tcpCommunicator: TcpCommunicator | null = null;
|
||||
private isBusy: boolean;
|
||||
private lastProcessedUserIndex: number;
|
||||
private isBusy: boolean = false;
|
||||
|
||||
constructor(userConfigPath: string, applicationInfoPath: string, clientPort: number, destinationPath: string) {
|
||||
this.userConfig = new JsonManager(userConfigPath);
|
||||
constructor(userConfigPath: string, applicationInfoPath: string, memoryManagerPath: string, clientPort: number) {
|
||||
this.applicationInfo = new JsonManager(applicationInfoPath);
|
||||
this.userConfig = new JsonManager(userConfigPath);
|
||||
this.memoryManager = new MemoryManager(memoryManagerPath);
|
||||
this.clientPort = clientPort;
|
||||
this.destinationPath = destinationPath;
|
||||
this.isBusy = false;
|
||||
this.lastProcessedUserIndex = 0;
|
||||
}
|
||||
|
||||
async start(): Promise<void> {
|
||||
setInterval(async () => {
|
||||
if (!this.isBusy) {
|
||||
this.log('Start successfully. Processing backup tasks.');
|
||||
await this.processBackupTasks();
|
||||
this.isBusy = true;
|
||||
this.log('Start successfully. Backup files to users.');
|
||||
await this.initialize();
|
||||
}
|
||||
}, 10000); // Retry every 10 seconds if there's an error
|
||||
}, 10000); // 10-second interval for testing
|
||||
}
|
||||
|
||||
private async processBackupTasks(): Promise<void> {
|
||||
private async initialize(): Promise<void> {
|
||||
this.isBusy = true;
|
||||
|
||||
try {
|
||||
const encryptionKeyData = await this.userConfig.readValue('encryption_key');
|
||||
if (!encryptionKeyData || !encryptionKeyData.key || !encryptionKeyData.iv) {
|
||||
this.log('Encryption key data is missing in user configuration.', 'error');
|
||||
return;
|
||||
}
|
||||
|
||||
this.fileEncryptor = new FileEncryptor(encryptionKeyData.key, encryptionKeyData.iv);
|
||||
|
||||
const userInfo = await this.userConfig.readValue('user_info');
|
||||
if (!userInfo || !userInfo.name) {
|
||||
throw new Error('User information or name is missing.');
|
||||
this.log('User information is missing in user configuration.', 'error');
|
||||
return;
|
||||
}
|
||||
const userName = userInfo.name;
|
||||
|
||||
const encryptionData = await this.userConfig.readValue('encryption_key');
|
||||
if (!encryptionData || !encryptionData.key || !encryptionData.iv) {
|
||||
throw new Error('Encryption key or IV is missing.');
|
||||
const backupDirectoryData = await this.applicationInfo.readValue('backupDirectory');
|
||||
if (!backupDirectoryData || !backupDirectoryData.id || !backupDirectoryData.path) {
|
||||
this.log('Backup directory information is missing in application info.', 'error');
|
||||
return;
|
||||
}
|
||||
|
||||
this.encryptionKey = Buffer.from(encryptionData.key, 'base64');
|
||||
this.iv = Buffer.from(encryptionData.iv, 'base64');
|
||||
|
||||
const activeUsersIp = await this.applicationInfo.readValue('users_ip');
|
||||
if (!activeUsersIp || !activeUsersIp.length) {
|
||||
this.isBusy = false;
|
||||
throw new Error('No active users found.');
|
||||
this.log('No active users found.', 'error');
|
||||
return;
|
||||
}
|
||||
|
||||
// Process each user, starting from the last processed index
|
||||
let backupSuccessful = true;
|
||||
for (let i = this.lastProcessedUserIndex; i < activeUsersIp.length; i++) {
|
||||
const ip = activeUsersIp[i];
|
||||
const success = await this.processBackupForIp(ip, userName);
|
||||
if (!success) {
|
||||
backupSuccessful = false;
|
||||
this.lastProcessedUserIndex = i; // Remember where it stopped
|
||||
const backupDirectoryId = backupDirectoryData.id;
|
||||
const backupDirectoryPath = backupDirectoryData.path;
|
||||
|
||||
const directoryData = await this.memoryManager.retrieveMetaInformation(backupDirectoryId);
|
||||
if (!directoryData || !directoryData.structure) {
|
||||
this.log('Backup directory structure is missing in memory.', 'error');
|
||||
return;
|
||||
}
|
||||
|
||||
await this.sendFilesToUsers(directoryData.structure, userName, activeUsersIp, backupDirectoryPath);
|
||||
} catch (error: any) {
|
||||
this.log(`Error in initialize process: ${error.message}`, 'error');
|
||||
} finally {
|
||||
this.log('Backup process completed.');
|
||||
this.isBusy = false;
|
||||
}
|
||||
}
|
||||
|
||||
private encryptFile(filePath: string): string {
|
||||
if (!this.fileEncryptor) {
|
||||
return filePath;
|
||||
}
|
||||
|
||||
if (!fs.existsSync(filePath)) {
|
||||
this.log(`File not found: ${filePath}`, 'error');
|
||||
return '';
|
||||
}
|
||||
|
||||
return this.fileEncryptor.encryptFileToBase64(filePath);
|
||||
}
|
||||
|
||||
private async sendFilesToUsers(fileStructure: { [key: string]: string }, userName: string, usersIp: string[], backupDirectoryPath: string): Promise<void> {
|
||||
let unsentFiles = Object.keys(fileStructure);
|
||||
|
||||
for (const fileName of unsentFiles) {
|
||||
const filePath = fileStructure[fileName];
|
||||
const encryptedFileContent = this.encryptFile(filePath);
|
||||
|
||||
if (!encryptedFileContent) {
|
||||
this.log(`Failed to encrypt file: ${fileName}`, 'error');
|
||||
continue;
|
||||
}
|
||||
|
||||
const relativeFilePath = path.relative(backupDirectoryPath, filePath);
|
||||
const metaInfo = { userName, relativeFilePath };
|
||||
|
||||
for (const ip of usersIp) {
|
||||
const tcpCommunicator = new TcpCommunicator(ip, this.clientPort);
|
||||
|
||||
try {
|
||||
await tcpCommunicator.connect();
|
||||
this.log(`Connected to ${ip}`);
|
||||
|
||||
const sendSuccess = await tcpCommunicator.sendMessage(operationCodes.BACKUP_FILE, metaInfo, Buffer.from(encryptedFileContent, 'base64'));
|
||||
if (!sendSuccess) {
|
||||
throw new Error('Failed to send file content.');
|
||||
}
|
||||
|
||||
const responseReceived = await this.waitForResponse(tcpCommunicator);
|
||||
if (!responseReceived) {
|
||||
throw new Error('Timeout waiting for the message response.');
|
||||
}
|
||||
|
||||
this.log(`Successfully sent file: ${fileName} to ${ip}`);
|
||||
unsentFiles = unsentFiles.filter(f => f !== fileName);
|
||||
break;
|
||||
} catch (error) {
|
||||
this.log(`Failed to send file: ${fileName} to ${ip}. Error: ${error}`, 'error');
|
||||
} finally {
|
||||
await tcpCommunicator.disconnect();
|
||||
this.log(`Disconnected from ${ip}`);
|
||||
}
|
||||
}
|
||||
|
||||
if (backupSuccessful) {
|
||||
this.log('Backup retrieval completed successfully for all users.');
|
||||
this.lastProcessedUserIndex = 0; // Reset index to start from the beginning next cycle
|
||||
}
|
||||
} catch (error: any) {
|
||||
this.log(error.message, 'error');
|
||||
}
|
||||
|
||||
this.isBusy = false;
|
||||
}
|
||||
|
||||
private async processBackupForIp(ip: string, userName: string): Promise<boolean> {
|
||||
this.tcpCommunicator = new TcpCommunicator(ip, this.clientPort);
|
||||
if (!await this.tcpCommunicator.connect()) {
|
||||
this.log(`Failed to connect to ${ip}`, 'error');
|
||||
return false;
|
||||
}
|
||||
|
||||
const backupExists = await this.checkIfBackupExists(userName);
|
||||
if (!backupExists) {
|
||||
this.log(`No backup found for user ${userName} on IP ${ip}`);
|
||||
await this.tcpCommunicator.disconnect();
|
||||
return true; // Skip user if no backup found, do not mark as error
|
||||
}
|
||||
|
||||
const backupStructure = await this.requestBackupStructure(userName);
|
||||
if (!backupStructure || Object.keys(backupStructure).length === 0) {
|
||||
this.log(`No files found in backup structure for user ${userName} on IP ${ip}`);
|
||||
await this.tcpCommunicator.disconnect();
|
||||
return true; // Skip user if no files found, do not mark as error
|
||||
}
|
||||
|
||||
for (const relativeFilePath of Object.keys(backupStructure)) {
|
||||
const fileRequestSuccess = await this.requestBackupFile(userName, relativeFilePath);
|
||||
if (!fileRequestSuccess) {
|
||||
this.log(`Failed to retrieve file ${relativeFilePath} from backup for user ${userName} on IP ${ip}`, 'error');
|
||||
await this.tcpCommunicator.disconnect();
|
||||
return false; // Stop if any file fails to be retrieved
|
||||
}
|
||||
}
|
||||
|
||||
await this.tcpCommunicator.disconnect();
|
||||
return true;
|
||||
}
|
||||
|
||||
private async checkIfBackupExists(userName: string): Promise<boolean> {
|
||||
if (!this.tcpCommunicator) return false;
|
||||
|
||||
const metaInfo = { name: userName };
|
||||
if (!await this.tcpCommunicator.sendMessage(operationCodes.IS_BACKUP_CREATED, metaInfo)) return false;
|
||||
|
||||
const response = await this.waitForResponse();
|
||||
return response?.metaInfo?.backupExists === true;
|
||||
}
|
||||
|
||||
private async requestBackupStructure(userName: string): Promise<any> {
|
||||
if (!this.tcpCommunicator) return false;
|
||||
|
||||
const metaInfo = { name: userName };
|
||||
if (!await this.tcpCommunicator.sendMessage(operationCodes.GET_BACKUP_STRUCTURE, metaInfo)) return null;
|
||||
|
||||
const response = await this.waitForResponse();
|
||||
return response?.metaInfo?.structure || null;
|
||||
}
|
||||
|
||||
private async requestBackupFile(userName: string, relativeFilePath: string): Promise<boolean> {
|
||||
if (!this.tcpCommunicator) return false;
|
||||
|
||||
const metaInfo = { name: userName, relativeFilePath };
|
||||
if (!await this.tcpCommunicator.sendMessage(operationCodes.REQ_FILE_FROM_BACKUP, metaInfo)) return false;
|
||||
|
||||
const response = await this.waitForResponse();
|
||||
if (response?.operationCode === operationCodes.OK && response.metaInfo && response.fileContent) {
|
||||
return this.saveFile(response.metaInfo.relativeFilePath, response.fileContent.toString('base64'));
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
private saveFile(relativeFilePath: string, fileContent: string): boolean {
|
||||
if (!this.encryptionKey || !this.iv) {
|
||||
this.log('Encryption key or IV is not set.', 'error');
|
||||
return false;
|
||||
}
|
||||
|
||||
let encryptedBuffer: Buffer;
|
||||
try {
|
||||
encryptedBuffer = Buffer.from(fileContent, 'base64');
|
||||
} catch (error) {
|
||||
this.log(`Error decoding base64 file content: ${error}`, 'error');
|
||||
return false;
|
||||
}
|
||||
|
||||
let decryptedContent: Buffer;
|
||||
try {
|
||||
const decipher = crypto.createDecipheriv('aes-256-cbc', this.encryptionKey, this.iv);
|
||||
decryptedContent = Buffer.concat([decipher.update(encryptedBuffer), decipher.final()]);
|
||||
} catch (error) {
|
||||
this.log(`Error decrypting file: ${error}`, 'error');
|
||||
return false;
|
||||
}
|
||||
|
||||
const fullFilePath = path.join(this.destinationPath, relativeFilePath);
|
||||
try {
|
||||
const dirPath = path.dirname(fullFilePath);
|
||||
if (!fs.existsSync(dirPath)) {
|
||||
fs.mkdirSync(dirPath, { recursive: true });
|
||||
}
|
||||
fs.writeFileSync(fullFilePath, decryptedContent);
|
||||
this.log(`File saved successfully: ${fullFilePath}`);
|
||||
return true;
|
||||
} catch (error: any) {
|
||||
this.log(`Error saving file ${relativeFilePath}: ${error.message}`, 'error');
|
||||
return false;
|
||||
if (unsentFiles.length > 0) {
|
||||
parentPort?.postMessage({ success: false, message: 'Backup could not be completed for all files', unsentFiles });
|
||||
} else {
|
||||
parentPort?.postMessage({ success: true, message: 'Backup completed successfully' });
|
||||
}
|
||||
}
|
||||
|
||||
private async waitForResponse(): Promise<ParsedMessage | null> {
|
||||
private async waitForResponse(tcpCommunicator: TcpCommunicator): Promise<ParsedMessage | null> {
|
||||
return new Promise((resolve) => {
|
||||
const idResponseCheck = setInterval(async () => {
|
||||
if (!this.tcpCommunicator) return null;
|
||||
if (this.tcpCommunicator.hasResponseArrived()) {
|
||||
const idResponseCheck = setInterval(() => {
|
||||
if (!tcpCommunicator) return null;
|
||||
if (tcpCommunicator.hasResponseArrived()) {
|
||||
clearInterval(idResponseCheck);
|
||||
resolve(this.tcpCommunicator.getLastResult());
|
||||
resolve(tcpCommunicator.getLastResult());
|
||||
}
|
||||
}, 100); // Check every 100 milliseconds
|
||||
}, 100);
|
||||
});
|
||||
}
|
||||
|
||||
// Unified logging function
|
||||
private log(message: string, level: 'log' | 'error' = 'log'): void {
|
||||
const prefix = '[BackupRetrievalWorker]';
|
||||
private log(message: string, level: 'log' | 'error' | 'warn' = 'log'): void {
|
||||
const prefix = '[BackupManager]';
|
||||
if (level === 'error') {
|
||||
console.error(`${prefix} ${message}`);
|
||||
} else if (level === 'warn') {
|
||||
console.warn(`${prefix} ${message}`);
|
||||
} else {
|
||||
console.log(`${prefix} ${message}`);
|
||||
}
|
||||
|
||||
@@ -32,6 +32,7 @@ export class DepartmentSharer {
|
||||
async start(): Promise<void> {
|
||||
setInterval(async () => {
|
||||
if (!this.isBusy) {
|
||||
this.isBusy = true;
|
||||
this.log('Start successfully. Sharing files with the department.');
|
||||
await this.shareFilesWithDepartment();
|
||||
}
|
||||
@@ -40,68 +41,69 @@ export class DepartmentSharer {
|
||||
|
||||
// Share files with users in the same department
|
||||
private async shareFilesWithDepartment(): Promise<void> {
|
||||
this.isBusy = true;
|
||||
|
||||
// Get the current user's department information
|
||||
const userInfo = await this.userConfig.readValue('user_info');
|
||||
if (!userInfo || !userInfo.departmentId || !userInfo.name) {
|
||||
this.log('User information or department ID is missing in the configuration.', 'error');
|
||||
return;
|
||||
}
|
||||
|
||||
const departmentId = userInfo.departmentId;
|
||||
const userName = userInfo.name;
|
||||
|
||||
// Get the list of active users from applicationInfo
|
||||
const activeUsersId = await this.applicationInfo.readValue('active_users_info');
|
||||
if (!activeUsersId) {
|
||||
this.log('No active users found.', 'error');
|
||||
return;
|
||||
}
|
||||
|
||||
const activeUsers = await this.memoryManager.retrieveMetaInformation(activeUsersId);
|
||||
if (!activeUsers || activeUsers.length === 0) {
|
||||
this.log('No active users found.', 'error');
|
||||
return;
|
||||
}
|
||||
|
||||
// Filter users who belong to the same department
|
||||
const departmentUsers = activeUsers.filter((user: any) => user.user_info.departmentId === departmentId);
|
||||
if (departmentUsers.length === 0) {
|
||||
this.log('No users found in the same department.', 'log');
|
||||
return;
|
||||
}
|
||||
|
||||
// Get department directory info
|
||||
const departmentData = await this.applicationInfo.readValue('departmentDirectory');
|
||||
if (!departmentData || !departmentData.path || !departmentData.id) {
|
||||
this.log('No department directory found.', 'error');
|
||||
return;
|
||||
}
|
||||
|
||||
this.departmentDirectory = departmentData.path;
|
||||
|
||||
// Read files from the MemoryManager related to this department
|
||||
const departmentFiles = await this.memoryManager.retrieveMetaInformation(departmentData.id);
|
||||
if (!departmentFiles || !departmentFiles.structure) {
|
||||
this.log('No files found for this department in the memory manager.', 'error');
|
||||
return;
|
||||
}
|
||||
|
||||
// Iterate over all department users and perform the operations
|
||||
for (const user of departmentUsers) {
|
||||
const userIp = user.ip;
|
||||
this.tcpCommunicator = new TcpCommunicator(userIp, this.clientPort);
|
||||
|
||||
if(await this.tcpCommunicator.connect()) continue;
|
||||
|
||||
// First clear the department directory
|
||||
const clearSuccess = await this.clearDepartmentDirectory();
|
||||
if (clearSuccess) {
|
||||
await this.sendFilesToUser(departmentFiles.structure, userName);
|
||||
try {
|
||||
// Get the current user's department information
|
||||
const userInfo = await this.userConfig.readValue('user_info');
|
||||
if (!userInfo || !userInfo.departmentId || !userInfo.name) {
|
||||
throw new Error('User information or department ID is missing in the configuration.');
|
||||
}
|
||||
|
||||
await this.tcpCommunicator.disconnect();
|
||||
const departmentId = userInfo.departmentId;
|
||||
const userName = userInfo.name;
|
||||
|
||||
// Get the list of active users from applicationInfo
|
||||
const activeUsersId = await this.applicationInfo.readValue('active_users_info');
|
||||
if (!activeUsersId) {
|
||||
throw new Error('No active users found.');
|
||||
}
|
||||
|
||||
const activeUsers = await this.memoryManager.retrieveMetaInformation(activeUsersId);
|
||||
if (!activeUsers || activeUsers.length === 0) {
|
||||
throw new Error('No active users found.');
|
||||
}
|
||||
|
||||
// Filter users who belong to the same department
|
||||
const departmentUsers = activeUsers.filter((user: any) => user.user_info.departmentId === departmentId);
|
||||
if (departmentUsers.length === 0) {
|
||||
throw new Error('No users found in the same department.');
|
||||
}
|
||||
|
||||
// Get department directory info
|
||||
const departmentData = await this.applicationInfo.readValue('departmentDirectory');
|
||||
if (!departmentData || !departmentData.path || !departmentData.id) {
|
||||
throw new Error('No department directory found.');
|
||||
}
|
||||
|
||||
this.departmentDirectory = departmentData.path;
|
||||
|
||||
// Read files from the MemoryManager related to this department
|
||||
const departmentFiles = await this.memoryManager.retrieveMetaInformation(departmentData.id);
|
||||
if (!departmentFiles || !departmentFiles.structure) {
|
||||
throw new Error('No files found for this department in the memory manager.');
|
||||
}
|
||||
|
||||
// Iterate over all department users and perform the operations
|
||||
for (const user of departmentUsers) {
|
||||
const userIp = user.ip;
|
||||
this.tcpCommunicator = new TcpCommunicator(userIp, this.clientPort);
|
||||
|
||||
if (await this.tcpCommunicator.connect()) continue;
|
||||
|
||||
// First clear the department directory
|
||||
const clearSuccess = await this.clearDepartmentDirectory();
|
||||
if (clearSuccess) {
|
||||
await this.sendFilesToUser(departmentFiles.structure, userName);
|
||||
}
|
||||
|
||||
await this.tcpCommunicator.disconnect();
|
||||
}
|
||||
}
|
||||
catch(error: any) {
|
||||
this.log(error.message, 'error');
|
||||
}
|
||||
finally{
|
||||
this.log('Department sharing completed.');
|
||||
this.isBusy = false;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { workerData } from 'worker_threads';
|
||||
import { UsersInfoFetcher } from '../helpers/users_info_fetcher';
|
||||
import {BackupRetrievalWorker} from "../helpers/backup_manager";
|
||||
import {BackupManager} from "../helpers/backup_manager";
|
||||
import {FileSharer} from "../helpers/file_sharer";
|
||||
import {DepartmentSharer} from "../helpers/department_sharer";
|
||||
|
||||
@@ -10,7 +10,7 @@ const { usersConfigPath, applicationInfoPath, memoryManagerPath, queueManagerPat
|
||||
const usersInfoFetcher = new UsersInfoFetcher(applicationInfoPath, memoryManagerPath, tcpPort);
|
||||
usersInfoFetcher.start();
|
||||
|
||||
const backupManager = new BackupRetrievalWorker(usersConfigPath, applicationInfoPath, memoryManagerPath, tcpPort);
|
||||
const backupManager = new BackupManager(usersConfigPath, applicationInfoPath, memoryManagerPath, tcpPort);
|
||||
backupManager.start();
|
||||
|
||||
const fileSharer = new FileSharer(queueManagerPath, tcpPort);
|
||||
|
||||
Reference in New Issue
Block a user