Streaming, Webhooks & Long-Running Nodes

import { IExecuteFunctions, INodeExecutionData } from 'n8n-workflow';
import { Readable, Transform } from 'stream';
import * as csv from 'csv-parser';

export class StreamingCSVProcessor {
  async execute(this: IExecuteFunctions): Promise<INodeExecutionData[][]> {
    const items = this.getInputData();
    const returnData: INodeExecutionData[] = [];

    // Get node parameters
    const operation = this.getNodeParameter('operation', 0) as string;
    const batchSize = this.getNodeParameter('batchSize', 0, 1000) as number;

    for (let i = 0; i < items.length; i++) {
      const fileStream = items[i].binary?.data?.stream;

      if (!fileStream) {
        throw new Error('No file stream found');
      }

      // Create processing pipeline
      const processedItems = await this.processStream(
        fileStream,
        operation,
        batchSize
      );

      returnData.push(...processedItems);
    }

    return [returnData];
  }

  private async processStream(
    stream: Readable,
    operation: string,
    batchSize: number
  ): Promise<INodeExecutionData[]> {
    return new Promise((resolve, reject) => {
      const results: INodeExecutionData[] = [];
      let batch: any[] = [];

      // Create transform stream for processing
      const transformer = new Transform({
        objectMode: true,
        transform: async (chunk, encoding, callback) => {
          try {
            // Apply transformations
            const processed = await this.transformData(chunk, operation);
            batch.push(processed);

            // Process in batches to control memory
            if (batch.length >= batchSize) {
              await this.processBatch(batch);
              batch = [];
            }

            callback();
          } catch (error) {
            callback(error as Error);
          }
        },
        flush: async (callback) => {
          // Process remaining items
          if (batch.length > 0) {
            await this.processBatch(batch);
          }
          callback();
        }
      });

      // Build the pipeline
      stream
        .pipe(csv())
        .pipe(transformer)
        .on('data', (data) => {
          results.push({ json: data });
        })
        .on('end', () => resolve(results))
        .on('error', reject);
    });
  }

  private async transformData(data: any, operation: string): Promise<any> {
    switch (operation) {
      case 'uppercase':
        return Object.fromEntries(
          Object.entries(data).map(([k, v]) => [k, String(v).toUpperCase()])
        );
      case 'clean':
        return Object.fromEntries(
          Object.entries(data).map(([k, v]) => [k.trim(), String(v).trim()])
        );
      default:
        return data;
    }
  }

  private async processBatch(batch: any[]): Promise<void> {
    // Simulate async processing (e.g., database insert)
    await new Promise(resolve => setTimeout(resolve, 10));
    console.log(`Processed batch of ${batch.length} items`);
  }
}

export class MemoryEfficientNode {
  async execute(this: IExecuteFunctions): Promise<INodeExecutionData[][]> {
    const maxMemoryMB = this.getNodeParameter('maxMemoryMB', 0, 100) as number;
    const maxMemoryBytes = maxMemoryMB * 1024 * 1024;

    // Monitor memory usage
    const memoryMonitor = setInterval(() => {
      const usage = process.memoryUsage();
      if (usage.heapUsed > maxMemoryBytes) {
        this.logger.warn('Memory limit approaching', {
          used: Math.round(usage.heapUsed / 1024 / 1024),
          limit: maxMemoryMB
        });

        // Trigger garbage collection if available
        if (global.gc) {
          global.gc();
        }
      }
    }, 1000);

    try {
      // Process with streaming
      return await this.processWithStreaming();
    } finally {
      clearInterval(memoryMonitor);
    }
  }
}

import {
  IWebhookFunctions,
  IWebhookResponseData,
  INodeType,
  INodeTypeDescription,
} from 'n8n-workflow';

export class CustomWebhookTrigger implements INodeType {
  description: INodeTypeDescription = {
    displayName: 'Custom Webhook Trigger',
    name: 'customWebhookTrigger',
    group: ['trigger'],
    version: 1,
    description: 'Receives data via webhook with authentication',
    defaults: {
      name: 'Custom Webhook',
    },
    inputs: [],
    outputs: ['main'],
    webhooks: [
      {
        name: 'default',
        httpMethod: 'POST',
        responseMode: 'onReceived',
        path: 'webhook',
      },
    ],
    properties: [
      {
        displayName: 'Authentication',
        name: 'authentication',
        type: 'options',
        options: [
          { name: 'None', value: 'none' },
          { name: 'Bearer Token', value: 'bearer' },
          { name: 'HMAC Signature', value: 'hmac' },
        ],
        default: 'none',
      },
      {
        displayName: 'Secret',
        name: 'secret',
        type: 'string',
        displayOptions: {
          show: {
            authentication: ['bearer', 'hmac'],
          },
        },
        default: '',
        description: 'Secret for authentication',
      },
    ],
  };

  async webhook(this: IWebhookFunctions): Promise<IWebhookResponseData> {
    const req = this.getRequestObject();
    const resp = this.getResponseObject();
    const authentication = this.getNodeParameter('authentication') as string;

    // Validate authentication
    if (!await this.validateAuth(req, authentication)) {
      resp.status(401).json({ error: 'Unauthorized' });
      return {
        noWebhookResponse: true,
      };
    }

    // Parse and validate payload
    const body = this.getBodyData();
    const headers = this.getHeaderData();
    const query = this.getQueryData();

    // Process webhook data
    const processedData = await this.processWebhookData(body, headers, query);

    // Store data for workflow execution
    const returnData = {
      json: processedData,
      headers,
      query,
    };

    // Send immediate response
    resp.status(200).json({
      received: true,
      timestamp: new Date().toISOString()
    });

    return {
      workflowData: [[returnData]],
    };
  }

  private async validateAuth(
    req: any,
    authType: string
  ): Promise<boolean> {
    const secret = this.getNodeParameter('secret', '') as string;

    switch (authType) {
      case 'bearer':
        const token = req.headers.authorization?.replace('Bearer ', '');
        return token === secret;

      case 'hmac':
        const signature = req.headers['x-signature'];
        const payload = JSON.stringify(req.body);
        const expectedSig = this.createHmacSignature(payload, secret);
        return signature === expectedSig;

      default:
        return true;
    }
  }

  private createHmacSignature(payload: string, secret: string): string {
    const crypto = require('crypto');
    return crypto
      .createHmac('sha256', secret)
      .update(payload)
      .digest('hex');
  }

  private async processWebhookData(
    body: any,
    headers: any,
    query: any
  ): Promise<any> {
    // Custom processing logic
    return {
      ...body,
      receivedAt: new Date().toISOString(),
      source: headers['user-agent'] || 'unknown',
      queryParams: query,
    };
  }
}

export class RateLimitedWebhook {
  private rateLimiter = new Map<string, number[]>();

  async webhook(this: IWebhookFunctions): Promise<IWebhookResponseData> {
    const clientId = this.getClientIdentifier();
    const maxRequests = this.getNodeParameter('maxRequests', 0, 100) as number;
    const windowMs = this.getNodeParameter('windowMs', 0, 60000) as number;

    // Check rate limit
    if (!this.checkRateLimit(clientId, maxRequests, windowMs)) {
      const resp = this.getResponseObject();
      resp.status(429).json({
        error: 'Too many requests',
        retryAfter: windowMs / 1000
      });
      return { noWebhookResponse: true };
    }

    // Process webhook
    return this.processWebhook();
  }

  private checkRateLimit(
    clientId: string,
    maxRequests: number,
    windowMs: number
  ): boolean {
    const now = Date.now();
    const requests = this.rateLimiter.get(clientId) || [];

    // Remove old requests outside window
    const validRequests = requests.filter(time => now - time < windowMs);

    if (validRequests.length >= maxRequests) {
      return false;
    }

    validRequests.push(now);
    this.rateLimiter.set(clientId, validRequests);

    // Clean up old clients
    if (this.rateLimiter.size > 1000) {
      this.cleanupRateLimiter(windowMs);
    }

    return true;
  }
}

export class AsyncProcessingNode {
  async execute(this: IExecuteFunctions): Promise<INodeExecutionData[][]> {
    const operation = this.getNodeParameter('operation', 0) as string;
    const timeout = this.getNodeParameter('timeout', 0, 300000) as number;

    switch (operation) {
      case 'startJob':
        return await this.startAsyncJob();
      case 'checkStatus':
        return await this.checkJobStatus();
      case 'getResults':
        return await this.getJobResults();
      default:
        throw new Error(`Unknown operation: ${operation}`);
    }
  }

  private async startAsyncJob(): Promise<INodeExecutionData[][]> {
    const jobData = this.getNodeParameter('jobData', 0) as any;

    // Start async job (e.g., video processing)
    const jobId = await this.initiateJob(jobData);

    // Store job ID for later retrieval
    await this.storeJobId(jobId);

    return [[{
      json: {
        jobId,
        status: 'processing',
        startedAt: new Date().toISOString(),
        estimatedCompletion: this.estimateCompletion(jobData)
      }
    }]];
  }

  private async checkJobStatus(): Promise<INodeExecutionData[][]> {
    const jobId = this.getNodeParameter('jobId', 0) as string;
    const pollInterval = 5000; // 5 seconds
    const maxAttempts = 60; // 5 minutes total

    for (let i = 0; i < maxAttempts; i++) {
      const status = await this.getJobStatus(jobId);

      if (status.state === 'completed') {
        return [[{
          json: {
            jobId,
            status: 'completed',
            result: status.result,
            completedAt: new Date().toISOString()
          }
        }]];
      }

      if (status.state === 'failed') {
        throw new Error(`Job ${jobId} failed: ${status.error}`);
      }

      // Wait before next poll
      await new Promise(resolve => setTimeout(resolve, pollInterval));

      // Update progress if available
      if (status.progress) {
        this.sendMessageToUI(`Processing: ${status.progress}%`);
      }
    }

    // Timeout reached
    return [[{
      json: {
        jobId,
        status: 'timeout',
        message: 'Job did not complete within timeout period'
      }
    }]];
  }
}

export class QueueProcessor {
  private queue: any[] = [];
  private processing = false;

  async execute(this: IExecuteFunctions): Promise<INodeExecutionData[][]> {
    const items = this.getInputData();
    const batchSize = this.getNodeParameter('batchSize', 0, 10) as number;
    const concurrency = this.getNodeParameter('concurrency', 0, 3) as number;

    // Add items to queue
    this.queue.push(...items);

    // Start processing if not already running
    if (!this.processing) {
      await this.processQueue(batchSize, concurrency);
    }

    return [this.queue];
  }

  private async processQueue(
    batchSize: number,
    concurrency: number
  ): Promise<void> {
    this.processing = true;

    while (this.queue.length > 0) {
      const batch = this.queue.splice(0, batchSize);

      // Process batch with concurrency control
      const promises = [];
      for (let i = 0; i < batch.length; i += concurrency) {
        const chunk = batch.slice(i, i + concurrency);
        promises.push(this.processConcurrent(chunk));
      }

      await Promise.all(promises);
    }

    this.processing = false;
  }

  private async processConcurrent(items: any[]): Promise<void> {
    const results = await Promise.allSettled(
      items.map(item => this.processItem(item))
    );

    // Handle results and errors
    results.forEach((result, index) => {
      if (result.status === 'rejected') {
        this.logger.error(`Failed to process item ${index}:`, result.reason);
        // Add to retry queue if needed
        this.addToRetryQueue(items[index]);
      }
    });
  }
}

export class VideoProcessingNode {
  async execute(this: IExecuteFunctions): Promise<INodeExecutionData[][]> {
    const videoUrl = this.getNodeParameter('videoUrl', 0) as string;
    const operations = this.getNodeParameter('operations', 0) as string[];

    // Start FFmpeg process
    const ffmpeg = spawn('ffmpeg', [
      '-i', videoUrl,
      ...this.buildFFmpegArgs(operations),
      'pipe:1' // Output to stdout
    ]);

    // Stream processing
    return new Promise((resolve, reject) => {
      const chunks: Buffer[] = [];

      ffmpeg.stdout.on('data', (chunk) => {
        chunks.push(chunk);
        // Report progress
        this.reportProgress(chunks.length);
      });

      ffmpeg.stderr.on('data', (data) => {
        // Parse FFmpeg progress
        const progress = this.parseFFmpegProgress(data.toString());
        if (progress) {
          this.sendMessageToUI(`Processing: ${progress}%`);
        }
      });

      ffmpeg.on('close', (code) => {
        if (code === 0) {
          const buffer = Buffer.concat(chunks);
          resolve([[{
            binary: {
              data: {
                data: buffer.toString('base64'),
                mimeType: 'video/mp4',
                fileName: 'processed.mp4'
              }
            }
          }]]);
        } else {
          reject(new Error(`FFmpeg failed with code ${code}`));
        }
      });
    });
  }
}

export class RealtimeSyncNode {
  private connections = new Map<string, WebSocket>();

  async execute(this: IExecuteFunctions): Promise<INodeExecutionData[][]> {
    const endpoint = this.getNodeParameter('endpoint', 0) as string;
    const events = this.getNodeParameter('events', 0) as string[];

    return new Promise((resolve, reject) => {
      const ws = new WebSocket(endpoint);
      const results: INodeExecutionData[] = [];

      ws.on('open', () => {
        // Subscribe to events
        ws.send(JSON.stringify({
          action: 'subscribe',
          events
        }));
      });

      ws.on('message', (data) => {
        const event = JSON.parse(data.toString());

        // Process based on event type
        switch (event.type) {
          case 'data':
            results.push({ json: event.payload });
            // Stream to next node immediately
            this.sendDataToNextNode([{ json: event.payload }]);
            break;

          case 'error':
            this.logger.error('Stream error:', event.error);
            break;

          case 'complete':
            ws.close();
            resolve([results]);
            break;
        }
      });

      ws.on('error', reject);

      // Cleanup on workflow stop
      this.on('stop', () => {
        ws.close();
      });
    });
  }
}

export class RobustAsyncNode {
  async execute(this: IExecuteFunctions): Promise<INodeExecutionData[][]> {
    const retryConfig = {
      maxAttempts: 3,
      backoffMs: 1000,
      maxBackoffMs: 30000
    };

    try {
      return await this.executeWithRetry(retryConfig);
    } catch (error) {
      // Graceful degradation
      return await this.fallbackExecution(error);
    }
  }

  private async executeWithRetry(config: any): Promise<INodeExecutionData[][]> {
    let lastError: Error | undefined;

    for (let attempt = 1; attempt <= config.maxAttempts; attempt++) {
      try {
        return await this.performOperation();
      } catch (error) {
        lastError = error as Error;

        if (attempt < config.maxAttempts) {
          const delay = Math.min(
            config.backoffMs * Math.pow(2, attempt - 1),
            config.maxBackoffMs
          );

          this.logger.warn(`Attempt ${attempt} failed, retrying in ${delay}ms`);
          await new Promise(resolve => setTimeout(resolve, delay));
        }
      }
    }

    throw lastError;
  }
}