// packages/core/src/state/workflow-state-manager.ts
import mongoose from 'mongoose'

// 执行上下文Schema
const ExecutionContextSchema = new mongoose.Schema({
  executionId: { type: String, required: true, unique: true },
  workflowId: { type: String, required: true },
  userId: { type: String, required: true },
  clientId: { type: String }, // 用于断线重连
  status: { 
    type: String, 
    enum: ['pending', 'running', 'completed', 'failed', 'cancelled', 'skipped'],
    default: 'pending'
  },
  progress: {
    currentNode: String,
    completedNodes: [String],
    totalNodes: Number,
    percentage: { type: Number, default: 0 }
  },
  result: mongoose.Schema.Types.Mixed,
  error: {
    message: String,
    stack: String,
    timestamp: Date
  },
  metadata: mongoose.Schema.Types.Mixed,
  createdAt: { type: Date, default: Date.now },
  updatedAt: { type: Date, default: Date.now },
  completedAt: Date
})

// 节点执行结果Schema
const NodeResultSchema = new mongoose.Schema({
  executionId: { type: String, required: true },
  nodeId: { type: String, required: true },
  nodeName: String,
  nodeType: String,
  status: { type: String, enum: ['completed', 'failed', 'skipped'] },
  input: mongoose.Schema.Types.Mixed,
  output: mongoose.Schema.Types.Mixed,
  error: {
    message: String,
    stack: String
  },
  executionTime: Number, // 毫秒
  startedAt: Date,
  completedAt: Date
})

export class WorkflowStateManager {
  private ExecutionContext: mongoose.Model<any>
  private NodeResult: mongoose.Model<any>

  constructor(mongoConfig: any) {
    // 连接MongoDB
    mongoose.connect(mongoConfig.uri, mongoConfig.options)
    
    this.ExecutionContext = mongoose.model('ExecutionContext', ExecutionContextSchema)
    this.NodeResult = mongoose.model('NodeResult', NodeResultSchema)
  }

  /**
   * 保存执行上下文
   */
  async saveExecutionContext(context: WorkflowExecutionContext) {
    const doc = new this.ExecutionContext(context)
    return await doc.save()
  }

  /**
   * 更新执行状态
   */
  async updateExecutionStatus(executionId: string, status: string) {
    return await this.ExecutionContext.updateOne(
      { executionId },
      { 
        status, 
        updatedAt: new Date(),
        ...(status === 'completed' && { completedAt: new Date() })
      }
    )
  }

  /**
   * 更新执行进度
   */
  async updateExecutionProgress(executionId: string, progress: any) {
    return await this.ExecutionContext.updateOne(
      { executionId },
      { 
        progress, 
        updatedAt: new Date() 
      }
    )
  }

  /**
   * 保存执行结果
   */
  async saveExecutionResult(executionId: string, result: any) {
    return await this.ExecutionContext.updateOne(
      { executionId },
      { 
        result, 
        updatedAt: new Date() 
      }
    )
  }

  /**
   * 保存执行错误
   */
  async saveExecutionError(executionId: string, error: Error) {
    return await this.ExecutionContext.updateOne(
      { executionId },
      { 
        error: {
          message: error.message,
          stack: error.stack,
          timestamp: new Date()
        },
        updatedAt: new Date() 
      }
    )
  }

  /**
   * 保存节点执行结果
   */
  async saveNodeResult(executionId: string, nodeResult: any) {
    const doc = new this.NodeResult({
      executionId,
      ...nodeResult
    })
    return await doc.save()
  }

  /**
   * 获取执行上下文
   */
  async getExecutionContext(executionId: string) {
    return await this.ExecutionContext.findOne({ executionId }).lean()
  }

  /**
   * 获取执行进度
   */
  async getExecutionProgress(executionId: string) {
    const context = await this.ExecutionContext.findOne({ executionId }, 'progress').lean()
    return context?.progress
  }

  /**
   * 获取执行结果
   */
  async getExecutionResult(executionId: string) {
    const context = await this.ExecutionContext.findOne({ executionId }, 'result').lean()
    return context?.result
  }

  /**
   * 获取客户端的活跃执行
   */
  async getActiveExecutionsByClient(clientId: string) {
    return await this.ExecutionContext.find({
      clientId,
      status: { $in: ['pending', 'running'] }
    }).lean()
  }

  /**
   * 获取节点执行历史
   */
  async getNodeResults(executionId: string) {
    return await this.NodeResult.find({ executionId }).sort({ startedAt: 1 }).lean()
  }

  /**
   * 清理过期数据
   */
  async cleanupExpiredData(daysOld: number = 30) {
    const cutoffDate = new Date()
    cutoffDate.setDate(cutoffDate.getDate() - daysOld)

    // 清理执行上下文
    await this.ExecutionContext.deleteMany({
      status: { $in: ['completed', 'failed', 'cancelled'] },
      updatedAt: { $lt: cutoffDate }
    })

    // 清理节点结果
    await this.NodeResult.deleteMany({
      completedAt: { $lt: cutoffDate }
    })
  }

  /**
   * 获取执行统计
   */
  async getExecutionStats() {
    const stats = await this.ExecutionContext.aggregate([
      {
        $group: {
          _id: '$status',
          count: { $sum: 1 }
        }
      }
    ])

    return stats.reduce((acc, stat) => {
      acc[stat._id] = stat.count
      return acc
    }, {})
  }

  async close() {
    await mongoose.connection.close()
  }
}

// packages/api/src/server.ts
import Fastify from 'fastify'
import { DistributedWorkflowEngine } from '@genn/workflow-core'
import { WorkflowController } from './controllers/workflow-controller'

const fastify = Fastify({ 
  logger: true,
  bodyLimit: 10485760 // 10MB
})

// 初始化工作流引擎
const workflowEngine = new DistributedWorkflowEngine(
  // Redis配置
  {
    host: process.env.REDIS_HOST || 'localhost',
    port: parseInt(process.env.REDIS_PORT || '6379'),
    password: process.env.REDIS_PASSWORD,
    db: parseInt(process.env.REDIS_DB || '0')
  },
  // MongoDB配置
  {
    uri: process.env.MONGODB_URI || 'mongodb://localhost:27017/workflow',
    options: {}
  }
)

// 注册控制器
const workflowController = new WorkflowController(workflowEngine)

// 路由定义
fastify.register(async function (fastify) {
  // 提交工作流执行
  fastify.post('/api/workflow/execute', workflowController.executeWorkflow.bind(workflowController))
  
  // 获取执行状态
  fastify.get('/api/workflow/status/:executionId', workflowController.getExecutionStatus.bind(workflowController))
  
  // 断线重连
  fastify.get('/api/workflow/reconnect/:clientId', workflowController.reconnectClient.bind(workflowController))
  
  // 取消执行
  fastify.post('/api/workflow/cancel/:executionId', workflowController.cancelExecution.bind(workflowController))
  
  // 系统统计
  fastify.get('/api/workflow/stats', workflowController.getSystemStats.bind(workflowController))
  
  // 健康检查
  fastify.get('/health', async () => ({ status: 'ok', timestamp: new Date().toISOString() }))
})

// WebSocket支持（用于实时进度推送）
fastify.register(require('@fastify/websocket'))

fastify.register(async function (fastify) {
  fastify.get('/ws/:clientId', { websocket: true }, (connection, req) => {
    const clientId = (req.params as any).clientId
    
    connection.socket.on('message', async (message) => {
      // 处理客户端消息
      try {
        const data = JSON.parse(message.toString())
        
        if (data.type === 'subscribe') {
          // 订阅执行状态更新
          workflowController.subscribeToUpdates(clientId, connection.socket)
        }
      } catch (error) {
        connection.socket.send(JSON.stringify({
          type: 'error',
          message: 'Invalid message format'
        }))
      }
    })
    
    connection.socket.on('close', () => {
      // 清理订阅
      workflowController.unsubscribeFromUpdates(clientId)
    })
  })
})

// 启动服务器
const start = async () => {
  try {
    const port = parseInt(process.env.PORT || '3000')
    await fastify.listen({ port, host: '0.0.0.0' })
    console.log(`🚀 Workflow Engine API running on port ${port}`)
  } catch (err) {
    fastify.log.error(err)
    process.exit(1)
  }
}

// 优雅关闭
process.on('SIGTERM', async () => {
  console.log('Received SIGTERM, shutting down gracefully...')
  await workflowEngine.close()
  await fastify.close()
  process.exit(0)
})

start()

# deploy/docker/Dockerfile.api
FROM node:18-alpine

WORKDIR /app

# 复制package.json和安装依赖
COPY package*.json ./
COPY packages/api/package*.json ./packages/api/
RUN npm ci --only=production

# 复制源代码
COPY packages/api ./packages/api
COPY packages/shared ./packages/shared

# 构建
RUN npm run build

EXPOSE 3000

CMD ["npm", "run", "start:api"]

# deploy/docker/Dockerfile.worker
FROM node:18-alpine

WORKDIR /app

# 复制依赖和构建worker
COPY package*.json ./
COPY packages/worker/package*.json ./packages/worker/
COPY packages/core/package*.json ./packages/core/
RUN npm ci --only=production

COPY packages/worker ./packages/worker
COPY packages/core ./packages/core
COPY packages/shared ./packages/shared

RUN npm run build

CMD ["npm", "run", "start:worker"]

# docker-compose.yml
version: '3.8'

services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    command: redis-server --appendonly yes
    
  mongodb:
    image: mongo:7
    ports:
      - "27017:27017"
    volumes:
      - mongo_data:/data/db
    environment:
      MONGO_INITDB_ROOT_USERNAME: admin
      MONGO_INITDB_ROOT_PASSWORD: password
      
  workflow-api:
    build:
      context: .
      dockerfile: deploy/docker/Dockerfile.api
    ports:
      - "3000:3000"
    environment:
      - NODE_ENV=production
      - REDIS_HOST=redis
      - REDIS_PORT=6379
      - MONGODB_URI=mongodb://admin:password@mongodb:27017/workflow?authSource=admin
    depends_on:
      - redis
      - mongodb
    restart: unless-stopped
    
  workflow-worker-1:
    build:
      context: .
      dockerfile: deploy/docker/Dockerfile.worker
    environment:
      - NODE_ENV=production
      - REDIS_HOST=redis
      - REDIS_PORT=6379
      - MONGODB_URI=mongodb://admin:password@mongodb:27017/workflow?authSource=admin
      - WORKER_ID=worker-1
    depends_on:
      - redis
      - mongodb
    restart: unless-stopped
    
  workflow-worker-2:
    build:
      context: .
      dockerfile: deploy/docker/Dockerfile.worker
    environment:
      - NODE_ENV=production
      - REDIS_HOST=redis
      - REDIS_PORT=6379
      - MONGODB_URI=mongodb://admin:password@mongodb:27017/workflow?authSource=admin
      - WORKER_ID=worker-2
    depends_on:
      - redis
      - mongodb
    restart: unless-stopped

  # Nginx负载均衡
  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./deploy/nginx/nginx.conf:/etc/nginx/nginx.conf:ro
    depends_on:
      - workflow-api
    restart: unless-stopped

volumes:
  redis_data:
  mongo_data:

# deploy/k8s/workflow-engine.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: workflow-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: workflow-api
  template:
    metadata:
      labels:
        app: workflow-api
    spec:
      containers:
      - name: workflow-api
        image: genn/workflow-engine-api:latest
        ports:
        - containerPort: 3000
        env:
        - name: REDIS_HOST
          value: "redis-service"
        - name: MONGODB_URI
          valueFrom:
            secretKeyRef:
              name: mongodb-secret
              key: uri
        resources:
          requests:
            memory: "256Mi"
            cpu: "250m"
          limits:
            memory: "512Mi"
            cpu: "500m"
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: workflow-worker
spec:
  replicas: 5
  selector:
    matchLabels:
      app: workflow-worker
  template:
    metadata:
      labels:
        app: workflow-worker
    spec:
      containers:
      - name: workflow-worker
        image: genn/workflow-engine-worker:latest
        env:
        - name: REDIS_HOST
          value: "redis-service"
        - name: MONGODB_URI
          valueFrom:
            secretKeyRef:
              name: mongodb-secret
              key: uri
        resources:
          requests:
            memory: "512Mi"
            cpu: "500m"
          limits:
            memory: "1Gi"
            cpu: "1000m"
---
apiVersion: v1
kind: Service
metadata:
  name: workflow-api-service
spec:
  selector:
    app: workflow-api
  ports:
  - port: 80
    targetPort: 3000
  type: LoadBalancer

// 在主系统中的集成
class WorkflowEngineClient {
  private baseUrl: string
  private apiKey: string

  constructor(baseUrl: string, apiKey: string) {
    this.baseUrl = baseUrl
    this.apiKey = apiKey
  }

  async executeWorkflow(workflowDefinition: any, options: any) {
    const response = await fetch(`${this.baseUrl}/api/workflow/execute`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${this.apiKey}`
      },
      body: JSON.stringify({
        workflowDefinition,
        ...options
      })
    })

    return response.json()
  }

  async getExecutionStatus(executionId: string) {
    const response = await fetch(`${this.baseUrl}/api/workflow/status/${executionId}`, {
      headers: {
        'Authorization': `Bearer ${this.apiKey}`
      }
    })

    return response.json()
  }

  // WebSocket连接用于实时更新
  connectToUpdates(clientId: string, onUpdate: Function) {
    const ws = new WebSocket(`${this.baseUrl.replace('http', 'ws')}/ws/${clientId}`)
    
    ws.onmessage = (event) => {
      const data = JSON.parse(event.data)
      onUpdate(data)
    }

    return ws
  }
}