generic-converter.ts

/**
 * Generic Qdrant Converter
 * Main conversion engine that orchestrates field analysis, transformation, and compression
 */

import {
  ConversionConfig,
  ConversionResult,
  ValidationResult,
  ConversionMetrics,
  InferQdrantPayload,
  CompressibleField,
} from '../types/generic-converter.js';
import { QdrantPayloadBase } from '../types/qdrant-payload.js';
import { QdrantStorageMetadata } from '../types/response-filter.js';
import { performance } from 'perf_hooks';
import {
  isQueryResultData,
  isExtendedClusterData,
  isDataprocJob,
} from '../types/dataproc-responses.js';
import { FieldAnalyzer } from './field-analyzer.js';
import { TransformationEngine } from './transformation-engine.js';
import { CompressionService } from './compression.js';
import { logger } from '../utils/logger.js';

/**
 * Main generic converter service
 */
export class GenericQdrantConverter {
  private fieldAnalyzer: FieldAnalyzer;
  private transformationEngine: TransformationEngine;
  private compressionService: CompressionService;
  private metrics: ConversionMetrics;

  constructor(compressionService: CompressionService) {
    this.fieldAnalyzer = new FieldAnalyzer();
    this.transformationEngine = new TransformationEngine();
    this.compressionService = compressionService;
    this.metrics = this.initializeMetrics();
  }

  /**
   * Convert any source type to Qdrant-compatible payload with automatic type inference
   */
  async convert<TSource extends Record<string, any>>(
    source: TSource,
    metadata: QdrantStorageMetadata,
    config?: ConversionConfig<TSource>
  ): Promise<ConversionResult<InferQdrantPayload<TSource>>> {
    const startTime = performance.now();

    try {
      logger.info('Starting generic conversion', {
        sourceType: this.detectSourceType(source),
        fieldsCount: Object.keys(source).length,
        hasConfig: !!config,
      });

      // 1. Validate source data
      const validation = await this.validateSource(source);
      if (!validation.isValid) {
        throw new Error(`Source validation failed: ${validation.errors.join(', ')}`);
      }

      // 2. Analyze source fields
      const fieldAnalysis = await this.fieldAnalyzer.analyzeFields(source);

      // 3. Apply default transformations
      const transformedSource = this.transformationEngine.applyDefaultTransformations(source);

      // 4. Apply custom field mappings
      const mappedFields = this.transformationEngine.applyFieldMappings(
        transformedSource,
        config?.fieldMappings
      );

      // 5. Apply custom transformations
      const customTransformed = this.transformationEngine.applyTransformations(
        mappedFields as TSource,
        config?.transformations
      );

      // 6. Handle field-level compression
      const compressedFields = await this.handleFieldCompression(
        customTransformed,
        fieldAnalysis,
        config?.compressionRules
      );

      // 7. Merge compressed fields back into payload
      const payloadWithCompression = this.transformationEngine.mergeCompressedFields(
        customTransformed,
        compressedFields
      );

      // 8. Inject metadata
      const finalPayload = this.transformationEngine.injectMetadata(
        payloadWithCompression,
        metadata,
        config?.metadata
      );

      // 9. Calculate metrics
      const processingTime = performance.now() - startTime;
      const conversionMetadata = this.calculateConversionMetadata(
        source,
        compressedFields,
        processingTime
      );

      // 10. Update global metrics
      this.updateMetrics(conversionMetadata);

      logger.info('Generic conversion completed', {
        processingTime: `${processingTime.toFixed(2)}ms`,
        fieldsProcessed: conversionMetadata.fieldsProcessed,
        fieldsCompressed: conversionMetadata.fieldsCompressed,
        compressionRatio: conversionMetadata.compressionRatio,
      });

      return {
        payload: finalPayload as unknown as InferQdrantPayload<TSource>,
        metadata: conversionMetadata,
      };
    } catch (error) {
      logger.error('Generic conversion failed', {
        error: error instanceof Error ? error.message : String(error),
        sourceType: this.detectSourceType(source),
      });
      throw error;
    }
  }

  /**
   * Convert with explicit type specification for better type safety
   */
  async convertTyped<TSource extends Record<string, any>, TTarget extends QdrantPayloadBase>(
    source: TSource,
    metadata: QdrantStorageMetadata,
    targetType: new () => TTarget,
    config?: ConversionConfig<TSource>
  ): Promise<ConversionResult<TTarget>> {
    const result = await this.convert(source, metadata, config);
    return result as unknown as ConversionResult<TTarget>;
  }

  /**
   * Validate source data for conversion compatibility
   */
  async validateSource<T extends Record<string, any>>(source: T): Promise<ValidationResult> {
    const errors: string[] = [];
    const warnings: string[] = [];
    const suggestions: string[] = [];

    // Basic validation
    if (!source || typeof source !== 'object') {
      errors.push('Source must be a non-null object');
      return { isValid: false, errors, warnings, suggestions };
    }

    if (Object.keys(source).length === 0) {
      warnings.push('Source object is empty');
    }

    // Use field analyzer for detailed validation
    const compatibility = this.fieldAnalyzer.validateQdrantCompatibility(source);
    errors.push(...compatibility.issues);
    suggestions.push(...compatibility.suggestions);

    // Check for required metadata fields
    const requiredFields = ['toolName', 'timestamp', 'projectId', 'region'];
    for (const field of requiredFields) {
      if (!(field in source)) {
        warnings.push(`Recommended field '${field}' not found in source`);
      }
    }

    return {
      isValid: errors.length === 0,
      errors,
      warnings,
      suggestions,
    };
  }

  /**
   * Handle field-level compression based on analysis and configuration
   */
  private async handleFieldCompression<T extends Record<string, any>>(
    source: T,
    fieldAnalysis: any,
    compressionRules?: any
  ): Promise<Record<string, CompressibleField<any>>> {
    const compressedFields: Record<string, CompressibleField<any>> = {};

    // Determine fields to compress
    let fieldsToCompress: (keyof T)[] = [];

    if (compressionRules?.fields) {
      fieldsToCompress = compressionRules.fields;
    } else {
      // Use automatic detection
      const recommendations = this.fieldAnalyzer.generateCompressionRecommendations(fieldAnalysis);
      fieldsToCompress = [...recommendations.recommended, ...recommendations.optional];
    }

    // Compress each field
    for (const fieldName of fieldsToCompress) {
      if (fieldName in source) {
        const fieldValue = source[fieldName];
        const compressed = await this.compressionService.compressIfNeeded(fieldValue);

        compressedFields[String(fieldName)] = {
          data: compressed.data,
          isCompressed: compressed.isCompressed,
          compressionType: compressed.compressionType,
          originalSize: compressed.originalSize,
          compressedSize: compressed.compressedSize,
        };
      }
    }

    return compressedFields;
  }

  /**
   * Detect the source data type for automatic configuration
   */
  private detectSourceType(source: unknown): string {
    if (isQueryResultData(source)) {
      return 'QueryResultData';
    }
    if (isExtendedClusterData(source)) {
      return 'ExtendedClusterData';
    }
    if (isDataprocJob(source)) {
      return 'DataprocJob';
    }
    return 'Unknown';
  }

  /**
   * Calculate conversion metadata for result
   */
  private calculateConversionMetadata(
    source: Record<string, any>,
    compressedFields: Record<string, CompressibleField<any>>,
    processingTime: number
  ) {
    const fieldsProcessed = Object.keys(source).length;
    const fieldsCompressed = Object.values(compressedFields).filter((f) => f.isCompressed).length;

    const totalOriginalSize = Object.values(compressedFields).reduce(
      (sum, field) => sum + (field.originalSize || 0),
      0
    );

    const totalCompressedSize = Object.values(compressedFields).reduce(
      (sum, field) => sum + (field.compressedSize || field.originalSize || 0),
      0
    );

    const compressionRatio = totalOriginalSize > 0 ? totalCompressedSize / totalOriginalSize : 1;

    return {
      fieldsProcessed,
      fieldsCompressed,
      totalOriginalSize,
      totalCompressedSize,
      compressionRatio,
      processingTime,
    };
  }

  /**
   * Update global conversion metrics
   */
  private updateMetrics(conversionMetadata: any): void {
    this.metrics.totalConversions++;

    // Update averages
    const totalTime =
      this.metrics.averageProcessingTime * (this.metrics.totalConversions - 1) +
      conversionMetadata.processingTime;
    this.metrics.averageProcessingTime = totalTime / this.metrics.totalConversions;

    const totalRatio =
      this.metrics.averageCompressionRatio * (this.metrics.totalConversions - 1) +
      conversionMetadata.compressionRatio;
    this.metrics.averageCompressionRatio = totalRatio / this.metrics.totalConversions;
  }

  /**
   * Get current conversion metrics
   */
  getMetrics(): ConversionMetrics {
    return { ...this.metrics };
  }

  /**
   * Reset conversion metrics
   */
  resetMetrics(): void {
    this.metrics = this.initializeMetrics();
  }

  /**
   * Initialize metrics object
   */
  private initializeMetrics(): ConversionMetrics {
    return {
      totalConversions: 0,
      averageProcessingTime: 0,
      averageCompressionRatio: 1,
      fieldCompressionStats: {},
    };
  }

  /**
   * Create conversion configuration for known types
   */
  async createConfigForType<T extends Record<string, any>>(
    source: T,
    type: 'query' | 'cluster' | 'job' | 'auto'
  ): Promise<ConversionConfig<T>> {
    const detectedType = type === 'auto' ? this.detectSourceType(source) : type;

    // Generate automatic field mappings
    const fieldMappings = this.transformationEngine.generateAutoFieldMappings(
      source,
      detectedType as any
    );

    // Generate compression recommendations
    const fieldAnalysis = await this.fieldAnalyzer.analyzeFields(source);
    const compressionRecommendations =
      this.fieldAnalyzer.generateCompressionRecommendations(fieldAnalysis);

    return {
      fieldMappings,
      compressionRules: {
        fields: compressionRecommendations.recommended,
        sizeThreshold: 10240, // 10KB
        compressionType: 'gzip',
      },
      transformations: this.getDefaultTransformationsForType(detectedType),
      metadata: {
        autoTimestamp: true,
        autoUUID: false,
      },
    };
  }

  /**
   * Get default transformations for specific types
   */
  private getDefaultTransformationsForType(type: string): any {
    switch (type) {
      case 'QueryResultData':
        return {
          totalRows: (value: any) => value || 0,
          timestamp: () => new Date().toISOString(),
        };
      case 'ExtendedClusterData':
        return {
          createTime: (value: any) => value || new Date().toISOString(),
          status: (value: any) => value || 'UNKNOWN',
        };
      case 'DataprocJob':
        return {
          submissionTime: (value: any) => value || new Date().toISOString(),
          status: (value: any) => value || 'UNKNOWN',
        };
      default:
        return {};
    }
  }
}

/**
 * Factory function to create a configured generic converter
 */
export function createGenericConverter(
  compressionService: CompressionService
): GenericQdrantConverter {
  return new GenericQdrantConverter(compressionService);
}

/**
 * Utility function for quick conversions with automatic configuration
 */
export async function quickConvert<TSource extends Record<string, any>>(
  source: TSource,
  metadata: QdrantStorageMetadata,
  compressionService: CompressionService
): Promise<ConversionResult<InferQdrantPayload<TSource>>> {
  const converter = createGenericConverter(compressionService);
  const config = await converter.createConfigForType(source, 'auto');
  return converter.convert(source, metadata, config);
}