zclaw_openfang/desktop/src/lib/pipeline-client.ts

/**
 * Pipeline Client (Tauri)
 *
 * Client for discovering, running, and monitoring Pipelines.
 * Pipelines are DSL-based workflows that orchestrate Skills and Hands.
 */

import { invoke } from '@tauri-apps/api/core';
import { listen, type UnlistenFn } from '@tauri-apps/api/event';

// Re-export UnlistenFn for external use
export type { UnlistenFn };

// === Tauri Runtime Detection ===

function isTauriRuntime(): boolean {
  return typeof window !== 'undefined' && '__TAURI_INTERNALS__' in window;
}

const DEV_SERVER_URL = 'http://localhost:50051';

async function devServerFetch<T>(endpoint: string, options?: RequestInit): Promise<T> {
  const response = await fetch(`${DEV_SERVER_URL}${endpoint}`, {
    ...options,
    headers: {
      'Content-Type': 'application/json',
      ...options?.headers,
    },
  });
  
  if (!response.ok) {
    throw new Error(`Dev server error: ${response.status}`);
  }
  
  return response.json();
}

// === Types ===

export interface PipelineInputInfo {
  name: string;
  inputType: string;
  required: boolean;
  label: string;
  placeholder?: string;
  default?: unknown;
  options: string[];
}

export interface PipelineInfo {
  id: string;
  displayName: string;
  description: string;
  category: string;
  industry: string;
  tags: string[];
  icon: string;
  version: string;
  author: string;
  inputs: PipelineInputInfo[];
}

export interface RunPipelineRequest {
  pipelineId: string;
  inputs: Record<string, unknown>;
}

export interface RunPipelineResponse {
  runId: string;
  pipelineId: string;
  status: string;
}

export interface PipelineRunResponse {
  runId: string;
  pipelineId: string;
  status: 'pending' | 'running' | 'completed' | 'failed' | 'cancelled';
  currentStep?: string;
  percentage: number;
  message: string;
  outputs?: unknown;
  error?: string;
  startedAt: string;
  endedAt?: string;
}

export interface PipelineCompleteEvent {
  runId: string;
  pipelineId: string;
  status: string;
  outputs?: unknown;
  error?: string;
}

// === Pipeline Client ===

export class PipelineClient {
  /**
   * List all available pipelines
   */
  static async listPipelines(options?: {
    category?: string;
    industry?: string;
  }): Promise<PipelineInfo[]> {
    if (!isTauriRuntime()) {
      return devServerFetch<PipelineInfo[]>('/api/pipelines');
    }
    
    try {
      const pipelines = await invoke<PipelineInfo[]>('pipeline_list', {
        category: options?.category || null,
        industry: options?.industry || null,
      });
      return pipelines;
    } catch (error) {
      console.error('Failed to list pipelines:', error);
      throw new Error(`Failed to list pipelines: ${error}`);
    }
  }

  /**
   * Get a specific pipeline by ID
   */
  static async getPipeline(pipelineId: string): Promise<PipelineInfo> {
    if (!isTauriRuntime()) {
      return devServerFetch<PipelineInfo>(`/api/pipelines/${pipelineId}`);
    }
    
    try {
      const pipeline = await invoke<PipelineInfo>('pipeline_get', {
        pipelineId,
      });
      return pipeline;
    } catch (error) {
      console.error(`Failed to get pipeline ${pipelineId}:`, error);
      throw new Error(`Failed to get pipeline: ${error}`);
    }
  }

  /**
   * Run a pipeline with the given inputs
   */
  static async runPipeline(request: RunPipelineRequest): Promise<RunPipelineResponse> {
    if (!isTauriRuntime()) {
      return devServerFetch<RunPipelineResponse>('/api/pipelines/run', {
        method: 'POST',
        body: JSON.stringify({ request }),
      });
    }
    
    try {
      const response = await invoke<RunPipelineResponse>('pipeline_run', {
        request,
      });
      return response;
    } catch (error) {
      console.error('Failed to run pipeline:', error);
      throw new Error(`Failed to run pipeline: ${error}`);
    }
  }

  /**
   * Get the progress of a running pipeline
   */
  static async getProgress(runId: string): Promise<PipelineRunResponse> {
    if (!isTauriRuntime()) {
      return devServerFetch<PipelineRunResponse>(`/api/pipelines/${runId}/progress`);
    }
    
    try {
      const progress = await invoke<PipelineRunResponse>('pipeline_progress', {
        runId,
      });
      return progress;
    } catch (error) {
      console.error(`Failed to get progress for run ${runId}:`, error);
      throw new Error(`Failed to get progress: ${error}`);
    }
  }

  /**
   * Get the result of a completed pipeline run
   */
  static async getResult(runId: string): Promise<PipelineRunResponse> {
    if (!isTauriRuntime()) {
      return devServerFetch<PipelineRunResponse>(`/api/pipelines/${runId}/result`);
    }
    
    try {
      const result = await invoke<PipelineRunResponse>('pipeline_result', {
        runId,
      });
      return result;
    } catch (error) {
      console.error(`Failed to get result for run ${runId}:`, error);
      throw new Error(`Failed to get result: ${error}`);
    }
  }

  /**
   * Cancel a running pipeline
   */
  static async cancel(runId: string): Promise<void> {
    if (!isTauriRuntime()) {
      await devServerFetch<void>(`/api/pipelines/${runId}/cancel`, { method: 'POST' });
      return;
    }
    
    try {
      await invoke('pipeline_cancel', { runId });
    } catch (error) {
      console.error(`Failed to cancel run ${runId}:`, error);
      throw new Error(`Failed to cancel run: ${error}`);
    }
  }

  /**
   * List all runs
   */
  static async listRuns(): Promise<PipelineRunResponse[]> {
    if (!isTauriRuntime()) {
      return devServerFetch<PipelineRunResponse[]>('/api/pipelines/runs');
    }
    
    try {
      const runs = await invoke<PipelineRunResponse[]>('pipeline_runs');
      return runs;
    } catch (error) {
      console.error('Failed to list runs:', error);
      throw new Error(`Failed to list runs: ${error}`);
    }
  }

  /**
   * Refresh pipeline discovery (rescan filesystem)
   */
  static async refresh(): Promise<PipelineInfo[]> {
    if (!isTauriRuntime()) {
      return devServerFetch<PipelineInfo[]>('/api/pipelines/refresh', { method: 'POST' });
    }
    
    try {
      const pipelines = await invoke<PipelineInfo[]>('pipeline_refresh');
      return pipelines;
    } catch (error) {
      console.error('Failed to refresh pipelines:', error);
      throw new Error(`Failed to refresh pipelines: ${error}`);
    }
  }

  /**
   * Subscribe to pipeline completion events
   */
  static async onComplete(
    callback: (event: PipelineCompleteEvent) => void
  ): Promise<UnlistenFn> {
    return listen<PipelineCompleteEvent>('pipeline-complete', (event) => {
      callback(event.payload);
    });
  }

  /**
   * Run a pipeline and wait for completion
   * Returns the final result
   */
  static async runAndWait(
    request: RunPipelineRequest,
    onProgress?: (progress: PipelineRunResponse) => void,
    pollIntervalMs: number = 1000
  ): Promise<PipelineRunResponse> {
    // Start the pipeline
    const { runId } = await this.runPipeline(request);

    // Poll for progress until completion
    let result = await this.getProgress(runId);

    while (result.status === 'running' || result.status === 'pending') {
      if (onProgress) {
        onProgress(result);
      }

      await new Promise((resolve) => setTimeout(resolve, pollIntervalMs));
      result = await this.getProgress(runId);
    }

    return result;
  }
}

// === Utility Functions ===

/**
 * Format pipeline input type for display
 */
export function formatInputType(type: string): string {
  const typeMap: Record<string, string> = {
    string: '文本',
    number: '数字',
    boolean: '布尔值',
    select: '单选',
    'multi-select': '多选',
    file: '文件',
    text: '多行文本',
  };
  return typeMap[type] || type;
}

/**
 * Get default value for input type
 */
export function getDefaultForType(type: string): unknown {
  switch (type) {
    case 'string':
    case 'text':
      return '';
    case 'number':
      return 0;
    case 'boolean':
      return false;
    case 'select':
      return null;
    case 'multi-select':
      return [];
    case 'file':
      return null;
    default:
      return null;
  }
}

/**
 * Validate pipeline inputs against schema
 */
export function validateInputs(
  inputs: PipelineInputInfo[],
  values: Record<string, unknown>
): { valid: boolean; errors: string[] } {
  const errors: string[] = [];

  for (const input of inputs) {
    const value = values[input.name];

    // Check required
    if (input.required && (value === undefined || value === null || value === '')) {
      errors.push(`${input.label || input.name} 是必填项`);
      continue;
    }

    // Skip validation if not provided and not required
    if (value === undefined || value === null) {
      continue;
    }

    // Type-specific validation
    switch (input.inputType) {
      case 'number':
        if (typeof value !== 'number') {
          errors.push(`${input.label || input.name} 必须是数字`);
        }
        break;
      case 'boolean':
        if (typeof value !== 'boolean') {
          errors.push(`${input.label || input.name} 必须是布尔值`);
        }
        break;
      case 'select':
        if (input.options.length > 0 && !input.options.includes(String(value))) {
          errors.push(`${input.label || input.name} 必须是有效选项`);
        }
        break;
      case 'multi-select':
        if (!Array.isArray(value)) {
          errors.push(`${input.label || input.name} 必须是数组`);
        } else if (input.options.length > 0) {
          const invalid = value.filter((v) => !input.options.includes(String(v)));
          if (invalid.length > 0) {
            errors.push(`${input.label || input.name} 包含无效选项`);
          }
        }
        break;
    }
  }

  return {
    valid: errors.length === 0,
    errors,
  };
}

// === React Hook ===

import { useState, useEffect, useCallback } from 'react';

export interface UsePipelineOptions {
  category?: string;
  industry?: string;
  autoRefresh?: boolean;
  refreshInterval?: number;
}

export function usePipelines(options: UsePipelineOptions = {}) {
  const [pipelines, setPipelines] = useState<PipelineInfo[]>([]);
  const [loading, setLoading] = useState(true);
  const [error, setError] = useState<string | null>(null);

  const loadPipelines = useCallback(async () => {
    setLoading(true);
    setError(null);
    try {
      const result = await PipelineClient.listPipelines({
        category: options.category,
        industry: options.industry,
      });
      setPipelines(result);
    } catch (err) {
      setError(err instanceof Error ? err.message : String(err));
    } finally {
      setLoading(false);
    }
  }, [options.category, options.industry]);

  const refresh = useCallback(async () => {
    setLoading(true);
    setError(null);
    try {
      const result = await PipelineClient.refresh();
      // Filter by category and industry if specified
      let filtered = result;
      if (options.category) {
        filtered = filtered.filter((p) => p.category === options.category);
      }
      if (options.industry) {
        filtered = filtered.filter((p) => p.industry === options.industry);
      }
      setPipelines(filtered);
    } catch (err) {
      setError(err instanceof Error ? err.message : String(err));
    } finally {
      setLoading(false);
    }
  }, [options.category, options.industry]);

  useEffect(() => {
    loadPipelines();
  }, [loadPipelines]);

  useEffect(() => {
    if (options.autoRefresh && options.refreshInterval) {
      const interval = setInterval(loadPipelines, options.refreshInterval);
      return () => clearInterval(interval);
    }
  }, [options.autoRefresh, options.refreshInterval, loadPipelines]);

  return {
    pipelines,
    loading,
    error,
    refresh,
    reload: loadPipelines,
  };
}

export interface UsePipelineRunOptions {
  onComplete?: (result: PipelineRunResponse) => void;
  onProgress?: (progress: PipelineRunResponse) => void;
}

export function usePipelineRun(options: UsePipelineRunOptions = {}) {
  const [running, setRunning] = useState(false);
  const [progress, setProgress] = useState<PipelineRunResponse | null>(null);
  const [error, setError] = useState<string | null>(null);

  const run = useCallback(
    async (pipelineId: string, inputs: Record<string, unknown>) => {
      setRunning(true);
      setError(null);
      setProgress(null);

      try {
        const result = await PipelineClient.runAndWait(
          { pipelineId, inputs },
          (p) => {
            setProgress(p);
            options.onProgress?.(p);
          }
        );

        setProgress(result);
        options.onComplete?.(result);
        return result;
      } catch (err) {
        const errorMsg = err instanceof Error ? err.message : String(err);
        setError(errorMsg);
        throw err;
      } finally {
        setRunning(false);
      }
    },
    [options]
  );

  const cancel = useCallback(async () => {
    if (progress?.runId) {
      await PipelineClient.cancel(progress.runId);
      setRunning(false);
    }
  }, [progress?.runId]);

  return {
    run,
    cancel,
    running,
    progress,
    error,
  };
}