ubiquity-core 0.1.1

// Cloudflare Worker for command execution with Durable Objects

export default {
  async fetch(request, env) {
    const url = new URL(request.url);
    const path = url.pathname;

    // Route requests to appropriate handlers
    if (path.startsWith('/api/commands/execute')) {
      return handleExecute(request, env);
    } else if (path.match(/^\/api\/commands\/[^\/]+\/cancel$/)) {
      return handleCancel(request, env);
    } else if (path.match(/^\/api\/commands\/[^\/]+\/status$/)) {
      return handleStatus(request, env);
    } else if (path.startsWith('/ws/')) {
      return handleWebSocket(request, env);
    }

    return new Response('Not Found', { status: 404 });
  }
};

async function handleExecute(request, env) {
  try {
    const { request: commandRequest, namespace_id } = await request.json();
    
    // Create a new Durable Object for this command
    const id = env.COMMAND_EXECUTOR.newUniqueId();
    const executor = env.COMMAND_EXECUTOR.get(id);
    
    // Initialize the executor with the command
    const response = await executor.fetch(new Request('http://internal/execute', {
      method: 'POST',
      body: JSON.stringify(commandRequest)
    }));
    
    if (!response.ok) {
      return new Response('Failed to initialize command', { status: 500 });
    }
    
    // Return the WebSocket URL for streaming
    const wsUrl = `${new URL(request.url).origin}/ws/${id.toString()}`;
    
    return new Response(JSON.stringify({
      durable_object_id: id.toString(),
      websocket_url: wsUrl
    }), {
      headers: { 'Content-Type': 'application/json' }
    });
  } catch (error) {
    return new Response(JSON.stringify({ error: error.message }), {
      status: 500,
      headers: { 'Content-Type': 'application/json' }
    });
  }
}

async function handleCancel(request, env) {
  const url = new URL(request.url);
  const commandId = url.pathname.split('/')[3];
  
  try {
    const id = env.COMMAND_EXECUTOR.idFromString(commandId);
    const executor = env.COMMAND_EXECUTOR.get(id);
    
    return await executor.fetch(new Request('http://internal/cancel', {
      method: 'POST'
    }));
  } catch (error) {
    return new Response(JSON.stringify({ error: error.message }), {
      status: 500,
      headers: { 'Content-Type': 'application/json' }
    });
  }
}

async function handleStatus(request, env) {
  const url = new URL(request.url);
  const commandId = url.pathname.split('/')[3];
  
  try {
    const id = env.COMMAND_EXECUTOR.idFromString(commandId);
    const executor = env.COMMAND_EXECUTOR.get(id);
    
    return await executor.fetch(new Request('http://internal/status'));
  } catch (error) {
    return new Response(JSON.stringify({ error: error.message }), {
      status: 404,
      headers: { 'Content-Type': 'application/json' }
    });
  }
}

async function handleWebSocket(request, env) {
  const url = new URL(request.url);
  const durableObjectId = url.pathname.split('/')[2];
  
  try {
    const id = env.COMMAND_EXECUTOR.idFromString(durableObjectId);
    const executor = env.COMMAND_EXECUTOR.get(id);
    
    return await executor.fetch(new Request('http://internal/websocket', {
      headers: request.headers
    }));
  } catch (error) {
    return new Response('WebSocket connection failed', { status: 500 });
  }
}

// Durable Object implementation
export class CommandExecutor {
  constructor(state, env) {
    this.state = state;
    this.env = env;
    this.websockets = [];
    this.commandResult = null;
    this.eventHistory = [];
    this.command = null;
    this.cancelled = false;
    this.startTime = null;
  }

  async fetch(request) {
    const url = new URL(request.url);
    const path = url.pathname;

    switch (path) {
      case '/execute':
        return this.handleExecute(request);
      case '/websocket':
        return this.handleWebSocket(request);
      case '/cancel':
        return this.handleCancel(request);
      case '/status':
        return this.handleStatus(request);
      default:
        return new Response('Not Found', { status: 404 });
    }
  }

  async handleExecute(request) {
    try {
      this.command = await request.json();
      this.startTime = Date.now();
      
      // Start execution in the background
      this.executeCommand();
      
      return new Response('OK', { status: 200 });
    } catch (error) {
      return new Response(JSON.stringify({ error: error.message }), {
        status: 400,
        headers: { 'Content-Type': 'application/json' }
      });
    }
  }

  async handleWebSocket(request) {
    const upgradeHeader = request.headers.get('Upgrade');
    if (!upgradeHeader || upgradeHeader !== 'websocket') {
      return new Response('Expected Upgrade: websocket', { status: 426 });
    }

    const [client, server] = Object.values(new WebSocketPair());
    
    await this.handleWebSocketConnection(server);
    
    return new Response(null, {
      status: 101,
      webSocket: client,
    });
  }

  async handleWebSocketConnection(websocket) {
    websocket.accept();
    this.websockets.push(websocket);
    
    // Send any historical events
    for (const event of this.eventHistory) {
      websocket.send(JSON.stringify({
        type: 'Event',
        event: event
      }));
    }
    
    // Handle incoming messages
    websocket.addEventListener('message', async (event) => {
      try {
        const message = JSON.parse(event.data);
        
        switch (message.type) {
          case 'Subscribe':
            // Already subscribed by connecting
            break;
          case 'Cancel':
            this.cancelled = true;
            break;
        }
      } catch (error) {
        websocket.send(JSON.stringify({
          type: 'Error',
          message: error.message
        }));
      }
    });
    
    // Clean up on close
    websocket.addEventListener('close', () => {
      this.websockets = this.websockets.filter(ws => ws !== websocket);
    });
  }

  async handleCancel(request) {
    this.cancelled = true;
    
    const event = {
      type: 'Cancelled',
      command_id: this.command.id,
      duration_ms: Date.now() - this.startTime,
      timestamp: new Date().toISOString()
    };
    
    this.broadcastEvent(event);
    
    return new Response('OK', { status: 200 });
  }

  async handleStatus(request) {
    if (!this.commandResult) {
      return new Response('Not Found', { status: 404 });
    }
    
    return new Response(JSON.stringify(this.commandResult), {
      headers: { 'Content-Type': 'application/json' }
    });
  }

  async executeCommand() {
    const commandId = this.command.id;
    
    // Send start event
    this.broadcastEvent({
      type: 'Started',
      command_id: commandId,
      command: this.command.command,
      args: this.command.args,
      timestamp: new Date().toISOString()
    });
    
    try {
      // Simulate command execution
      const result = await this.simulateCommand(this.command);
      
      // Send output events
      for (const line of result.stdout) {
        if (this.cancelled) break;
        
        this.broadcastEvent({
          type: 'Stdout',
          command_id: commandId,
          data: line,
          timestamp: new Date().toISOString()
        });
        
        await this.delay(10);
      }
      
      for (const line of result.stderr) {
        if (this.cancelled) break;
        
        this.broadcastEvent({
          type: 'Stderr',
          command_id: commandId,
          data: line,
          timestamp: new Date().toISOString()
        });
        
        await this.delay(10);
      }
      
      // Send progress events
      for (const [percentage, message] of result.progress || []) {
        if (this.cancelled) break;
        
        this.broadcastEvent({
          type: 'Progress',
          command_id: commandId,
          percentage: percentage,
          message: message,
          timestamp: new Date().toISOString()
        });
        
        await this.delay(50);
      }
      
      const duration = Date.now() - this.startTime;
      
      if (!this.cancelled) {
        // Send completion event
        this.broadcastEvent({
          type: 'Completed',
          command_id: commandId,
          exit_code: result.exitCode,
          duration_ms: duration,
          timestamp: new Date().toISOString()
        });
        
        // Store result
        this.commandResult = {
          id: commandId,
          exit_code: result.exitCode,
          stdout: result.stdout.join('\n'),
          stderr: result.stderr.join('\n'),
          duration_ms: duration,
          cancelled: false
        };
      } else {
        this.commandResult = {
          id: commandId,
          exit_code: null,
          stdout: '',
          stderr: '',
          duration_ms: duration,
          cancelled: true
        };
      }
    } catch (error) {
      const duration = Date.now() - this.startTime;
      
      this.broadcastEvent({
        type: 'Failed',
        command_id: commandId,
        error: error.message,
        duration_ms: duration,
        timestamp: new Date().toISOString()
      });
      
      this.commandResult = {
        id: commandId,
        exit_code: -1,
        stdout: '',
        stderr: error.message,
        duration_ms: duration,
        cancelled: false
      };
    }
  }

  async simulateCommand(command) {
    // This is a simplified simulation
    // In production, this would execute in a secure sandbox
    
    const commands = {
      echo: () => ({
        stdout: command.args,
        stderr: [],
        exitCode: 0
      }),
      
      ls: () => ({
        stdout: ['file1.txt', 'file2.js', 'directory/', 'README.md'],
        stderr: [],
        exitCode: 0,
        progress: [[50, 'Listing files...']]
      }),
      
      cat: () => {
        if (command.stdin) {
          return {
            stdout: command.stdin.split('\n'),
            stderr: [],
            exitCode: 0
          };
        } else if (command.args.length > 0) {
          return {
            stdout: [`Contents of ${command.args[0]}`],
            stderr: [],
            exitCode: 0
          };
        } else {
          return {
            stdout: [],
            stderr: ['cat: missing file operand'],
            exitCode: 1
          };
        }
      },
      
      sleep: () => {
        const seconds = parseInt(command.args[0]) || 1;
        const progress = [];
        for (let i = 0; i < 10; i++) {
          progress.push([(i + 1) * 10, `Sleeping... ${(i + 1) * 10}%`]);
        }
        return {
          stdout: [],
          stderr: [],
          exitCode: 0,
          progress,
          duration: seconds * 1000
        };
      }
    };
    
    const executor = commands[command.command];
    if (executor) {
      const result = executor();
      
      // Simulate execution time
      if (result.duration) {
        await this.delay(result.duration);
      }
      
      return result;
    } else {
      return {
        stdout: [],
        stderr: [`${command.command}: command not found`],
        exitCode: 127
      };
    }
  }

  broadcastEvent(event) {
    this.eventHistory.push(event);
    
    const message = JSON.stringify({
      type: 'Event',
      event: event
    });
    
    // Send to all connected WebSockets
    this.websockets = this.websockets.filter(ws => {
      try {
        ws.send(message);
        return true;
      } catch (error) {
        // Remove closed connections
        return false;
      }
    });
  }

  delay(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }
}