fileTransferService.js

Go to the documentation of this file.

/**
 * File Transfer Service via WebSocket
 * 
 * Handles file uploads and downloads through agent WebSocket connections
 * This allows file transfer to/from agents behind NAT/firewalls
 * 
 * Flow:
 * Upload: Dashboard → Backend → WebSocket → Agent → Filesystem
 * Download: Filesystem → Agent → WebSocket → Backend → Dashboard
 */
 
const crypto = require('crypto');
const fs = require('fs').promises;
const path = require('path');
const EventEmitter = require('events');
const wsManager = require('./websocketManager');
 
/**
 *
 */
class FileTransferService extends EventEmitter {
  /**
   *
   */
  constructor() {
    super();
    // Map: transferId → { agent_uuid, type, filePath, progress, status, ...}
    this.activeTransfers = new Map();
    
    // Temporary upload directory
    this.uploadDir = path.join(__dirname, '../uploads');
  }
 
  /**
   *
   */
  async initialize() {
    // Ensure upload directory exists
    try {
      await fs.mkdir(this.uploadDir, { recursive: true });
    } catch (err) {
      console.error('[FileTransfer] Failed to create upload directory:', err);
    }
  }
 
  /**
   * Start file upload to agent
   * @param {string} agent_uuid - Agent UUID
   * @param {string} localPath - Local file path on backend
   * @param {string} remotePath - Destination path on agent
   * @param {string} user_email - User initiating transfer
   * @returns {Promise<string>} transferId
   */
  async startUpload(agent_uuid, localPath, remotePath, user_email = 'system') {
    const agentWs = wsManager.agentConnections.get(agent_uuid);
    if (!agentWs || agentWs.readyState !== 1) {
      throw new Error('Agent not connected');
    }
 
    const transferId = crypto.randomUUID();
    const stats = await fs.stat(localPath);
    const fileSize = stats.size;
 
    // Store transfer info
    this.activeTransfers.set(transferId, {
      transferId,
      agent_uuid,
      type: 'upload',
      localPath,
      remotePath,
      fileSize,
      transferred: 0,
      status: 'starting',
      user_email,
      startedAt: Date.now()
    });
 
    // Send start command to agent
    agentWs.send(JSON.stringify({
      type: 'file_upload_start',
      transferId,
      remotePath,
      fileSize
    }));
 
    console.log(`[FileTransfer] Started upload ${transferId} to agent ${agent_uuid}: ${remotePath}`);
 
    // Start streaming file in chunks
    this.streamFileToAgent(transferId, agent_uuid, localPath);
 
    // Audit log
    this.logActivity(agent_uuid, 'file_upload_started', {
      transfer_id: transferId,
      remote_path: remotePath,
      file_size: fileSize,
      user: user_email
    });
 
    return transferId;
  }
 
  /**
   * Stream file to agent in chunks
   * @param {string} transferId - Transfer ID
   * @param {string} agent_uuid - Agent UUID
   * @param {string} localPath - Local file path
   */
  async streamFileToAgent(transferId, agent_uuid, localPath) {
    const CHUNK_SIZE = 64 * 1024; // 64KB chunks
    const transfer = this.activeTransfers.get(transferId);
    const agentWs = wsManager.agentConnections.get(agent_uuid);
 
    if (!transfer || !agentWs) {
      console.error(`[FileTransfer] Transfer ${transferId} cancelled or agent disconnected`);
      return;
    }
 
    try {
      const fileHandle = await fs.open(localPath, 'r');
      const buffer = Buffer.allocUnsafe(CHUNK_SIZE);
      let offset = 0;
 
      transfer.status = 'transferring';
 
      while (true) {
        const { bytesRead } = await fileHandle.read(buffer, 0, CHUNK_SIZE, offset);
        
        if (bytesRead === 0) break; // EOF
 
        // Send chunk to agent
        agentWs.send(JSON.stringify({
          type: 'file_upload_chunk',
          transferId,
          data: buffer.slice(0, bytesRead).toString('base64'),
          offset,
          size: bytesRead
        }));
 
        offset += bytesRead;
        transfer.transferred = offset;
        transfer.progress = (offset / transfer.fileSize) * 100;
 
        this.emit('progress', { transferId, progress: transfer.progress });
      }
 
      await fileHandle.close();
 
      // Send completion
      agentWs.send(JSON.stringify({
        type: 'file_upload_complete',
        transferId
      }));
 
      transfer.status = 'completed';
      transfer.completedAt = Date.now();
 
      console.log(`[FileTransfer] Upload ${transferId} completed`);
 
      // Audit log
      this.logActivity(agent_uuid, 'file_upload_completed', {
        transfer_id: transferId,
        remote_path: transfer.remotePath,
        file_size: transfer.fileSize,
        duration_ms: transfer.completedAt - transfer.startedAt
      });
 
    } catch (err) {
      console.error(`[FileTransfer] Upload ${transferId} failed:`, err);
      transfer.status = 'failed';
      transfer.error = err.message;
 
      // Notify agent
      if (agentWs && agentWs.readyState === 1) {
        agentWs.send(JSON.stringify({
          type: 'file_upload_cancel',
          transferId,
          error: err.message
        }));
      }
    }
  }
 
  /**
   * Start file download from agent
   * @param {string} agent_uuid - Agent UUID
   * @param {string} remotePath - File path on agent
   * @param {string} user_email - User initiating transfer
   * @returns {Promise<string>} transferId
   */
  async startDownload(agent_uuid, remotePath, user_email = 'system') {
    const agentWs = wsManager.agentConnections.get(agent_uuid);
    if (!agentWs || agentWs.readyState !== 1) {
      throw new Error('Agent not connected');
    }
 
    const transferId = crypto.randomUUID();
    const localPath = path.join(this.uploadDir, `${transferId}_${path.basename(remotePath)}`);
 
    // Store transfer info
    this.activeTransfers.set(transferId, {
      transferId,
      agent_uuid,
      type: 'download',
      localPath,
      remotePath,
      fileSize: 0,
      transferred: 0,
      status: 'starting',
      user_email,
      startedAt: Date.now()
    });
 
    // Send download request to agent
    agentWs.send(JSON.stringify({
      type: 'file_download_start',
      transferId,
      remotePath
    }));
 
    console.log(`[FileTransfer] Started download ${transferId} from agent ${agent_uuid}: ${remotePath}`);
 
    // Audit log
    this.logActivity(agent_uuid, 'file_download_started', {
      transfer_id: transferId,
      remote_path: remotePath,
      user: user_email
    });
 
    return transferId;
  }
 
  /**
   * Handle incoming chunk from agent (download)
   * @param {object} data - Message data
   */
  async handleDownloadChunk(data) {
    const { transferId, data: base64Data, offset, size, fileSize } = data;
    const transfer = this.activeTransfers.get(transferId);
 
    if (!transfer || transfer.type !== 'download') {
      console.error(`[FileTransfer] Invalid download chunk for ${transferId}`);
      return;
    }
 
    try {
      // Initialize file size on first chunk
      if (fileSize && transfer.fileSize === 0) {
        transfer.fileSize = fileSize;
      }
 
      // Write chunk to file
      const buffer = Buffer.from(base64Data, 'base64');
      await fs.appendFile(transfer.localPath, buffer);
 
      transfer.transferred += size;
      transfer.progress = transfer.fileSize > 0 ? (transfer.transferred / transfer.fileSize) * 100 : 0;
      transfer.status = 'transferring';
 
      this.emit('progress', { transferId, progress: transfer.progress });
 
    } catch (err) {
      console.error(`[FileTransfer] Failed to write chunk for ${transferId}:`, err);
      transfer.status = 'failed';
      transfer.error = err.message;
    }
  }
 
  /**
   * Handle download completion from agent
   * @param {object} data - Message data
   */
  async handleDownloadComplete(data) {
    const { transferId } = data;
    const transfer = this.activeTransfers.get(transferId);
 
    if (!transfer || transfer.type !== 'download') {
      console.error(`[FileTransfer] Invalid download completion for ${transferId}`);
      return;
    }
 
    transfer.status = 'completed';
    transfer.completedAt = Date.now();
 
    console.log(`[FileTransfer] Download ${transferId} completed: ${transfer.localPath}`);
 
    // Audit log
    this.logActivity(transfer.agent_uuid, 'file_download_completed', {
      transfer_id: transferId,
      remote_path: transfer.remotePath,
      file_size: transfer.fileSize,
      duration_ms: transfer.completedAt - transfer.startedAt
    });
  }
 
  /**
   * Get transfer status
   * @param {string} transferId - Transfer ID
   * @returns {object | null}
   */
  getTransfer(transferId) {
    return this.activeTransfers.get(transferId) || null;
  }
 
  /**
   * Get all active transfers for agent
   * @param {string} agent_uuid - Agent UUID
   * @returns {Array}
   */
  getAgentTransfers(agent_uuid) {
    const transfers = [];
    for (const transfer of this.activeTransfers.values()) {
      if (transfer.agent_uuid === agent_uuid) {
        transfers.push(transfer);
      }
    }
    return transfers;
  }
 
  /**
   * Cancel transfer
   * @param {string} transferId - Transfer ID
   */
  async cancelTransfer(transferId) {
    const transfer = this.activeTransfers.get(transferId);
    if (!transfer) return;
 
    const agentWs = wsManager.agentConnections.get(transfer.agent_uuid);
    if (agentWs && agentWs.readyState === 1) {
      agentWs.send(JSON.stringify({
        type: transfer.type === 'upload' ? 'file_upload_cancel' : 'file_download_cancel',
        transferId
      }));
    }
 
    transfer.status = 'cancelled';
    console.log(`[FileTransfer] Cancelled transfer ${transferId}`);
  }
 
  /**
   * Clean up completed/failed transfers older than 1 hour
   */
  async cleanup() {
    const oneHourAgo = Date.now() - (60 * 60 * 1000);
    
    for (const [transferId, transfer] of this.activeTransfers.entries()) {
      if (['completed', 'failed', 'cancelled'].includes(transfer.status) &&
          transfer.completedAt && transfer.completedAt < oneHourAgo) {
        
        // Delete download file if exists
        if (transfer.type === 'download' && transfer.localPath) {
          try {
            await fs.unlink(transfer.localPath);
          } catch (err) {
            // Ignore
          }
        }
        
        this.activeTransfers.delete(transferId);
      }
    }
  }
 
  /**
   * Log activity to database for audit trail
   * @param {string} agent_uuid - Agent UUID
   * @param {string} action - Action type
   * @param {object} details - Additional details
   */
  async logActivity(agent_uuid, action, details = {}) {
    try {
      const db = require('../db');
      
      // Get tenant_id from agent
      const agentResult = await db.query(
        'SELECT tenant_id FROM agents WHERE agent_uuid = $1',
        [agent_uuid]
      );
      
      if (agentResult.rows.length === 0) {
        console.error('[FileTransfer] Cannot log activity: agent not found');
        return;
      }
      
      const tenant_id = agentResult.rows[0].tenant_id;
      
      await db.query(
        `INSERT INTO activity_logs (agent_uuid, tenant_id, action, details, user_email, created_at)
         VALUES ($1, $2, $3, $4, $5, NOW())`,
        [agent_uuid, tenant_id, action, JSON.stringify(details), details.user || 'system']
      );
      
      console.log(`[FileTransfer] Logged activity: ${action} for agent ${agent_uuid}`);
    } catch (err) {
      console.error('[FileTransfer] Failed to log activity:', err.message);
    }
  }
}
 
// Create singleton
const fileTransferService = new FileTransferService();
fileTransferService.initialize();
 
// Cleanup every hour
setInterval(() => fileTransferService.cleanup(), 60 * 60 * 1000);
 
module.exports = fileTransferService;