tspm/ts/classes.ipcclient.ts

import * as plugins from './plugins.js';
import * as paths from './paths.js';
import { spawn } from 'child_process';
import type {
  IpcMethodMap,
  RequestForMethod,
  ResponseForMethod,
} from './ipc.types.js';

/**
 * IPC client for communicating with the TSPM daemon
 */
export class TspmIpcClient {
  private ipcClient: plugins.smartipc.IpcClient | null = null;
  private socketPath: string;
  private daemonPidFile: string;
  private isConnected: boolean = false;

  constructor() {
    this.socketPath = plugins.path.join(paths.tspmDir, 'tspm.sock');
    this.daemonPidFile = plugins.path.join(paths.tspmDir, 'daemon.pid');
  }

  /**
   * Connect to the daemon, starting it if necessary
   */
  public async connect(): Promise<void> {
    // Check if already connected
    if (this.isConnected && this.ipcClient) {
      return;
    }

    // Check if daemon is running
    const daemonRunning = await this.isDaemonRunning();

    if (!daemonRunning) {
      console.log('Daemon not running, starting it...');
      await this.startDaemon();
      // Wait a bit for daemon to initialize
      await new Promise((resolve) => setTimeout(resolve, 1000));
    }

    // Create IPC client
    this.ipcClient = plugins.smartipc.SmartIpc.createClient({
      id: 'tspm-cli',
      socketPath: this.socketPath,
      clientId: `cli-${process.pid}`,
      connectRetry: {
        enabled: true,
        initialDelay: 100,
        maxDelay: 2000,
        maxAttempts: 30,
        totalTimeout: 15000,
      },
      registerTimeoutMs: 8000,
      heartbeat: true,
      heartbeatInterval: 5000,
      heartbeatTimeout: 20000,
      heartbeatInitialGracePeriodMs: 10000,
      heartbeatThrowOnTimeout: false  // Don't throw, emit events instead
    });

    // Connect to the daemon
    try {
      await this.ipcClient.connect({ waitForReady: true });
      this.isConnected = true;
      
      // Handle heartbeat timeouts gracefully
      this.ipcClient.on('heartbeatTimeout', () => {
        console.warn('Heartbeat timeout detected, connection may be degraded');
        this.isConnected = false;
      });
      
      console.log('Connected to TSPM daemon');
    } catch (error) {
      console.error('Failed to connect to daemon:', error);
      throw new Error(
        'Could not connect to TSPM daemon. Please try running "tspm daemon start" manually.',
      );
    }
  }

  /**
   * Disconnect from the daemon
   */
  public async disconnect(): Promise<void> {
    if (this.ipcClient) {
      await this.ipcClient.disconnect();
      this.ipcClient = null;
      this.isConnected = false;
    }
  }

  /**
   * Send a request to the daemon
   */
  public async request<M extends keyof IpcMethodMap>(
    method: M,
    params: RequestForMethod<M>,
  ): Promise<ResponseForMethod<M>> {
    if (!this.isConnected || !this.ipcClient) {
      await this.connect();
    }

    try {
      const response = await this.ipcClient!.request<
        RequestForMethod<M>,
        ResponseForMethod<M>
      >(method, params);

      return response;
    } catch (error) {
      // Handle connection errors by trying to reconnect once
      if (
        error.message?.includes('ECONNREFUSED') ||
        error.message?.includes('ENOENT')
      ) {
        console.log('Connection lost, attempting to reconnect...');
        this.isConnected = false;
        await this.connect();

        // Retry the request
        return await this.ipcClient!.request<
          RequestForMethod<M>,
          ResponseForMethod<M>
        >(method, params);
      }

      throw error;
    }
  }
  
  /**
   * Subscribe to log updates for a specific process
   */
  public async subscribe(processId: string, handler: (log: any) => void): Promise<void> {
    if (!this.ipcClient || !this.isConnected) {
      throw new Error('Not connected to daemon');
    }
    
    const topic = `logs.${processId}`;
    await this.ipcClient.subscribe(`topic:${topic}`, handler);
  }
  
  /**
   * Unsubscribe from log updates for a specific process
   */
  public async unsubscribe(processId: string): Promise<void> {
    if (!this.ipcClient || !this.isConnected) {
      throw new Error('Not connected to daemon');
    }
    
    const topic = `logs.${processId}`;
    await this.ipcClient.unsubscribe(`topic:${topic}`);
  }

  /**
   * Check if the daemon is running
   */
  private async isDaemonRunning(): Promise<boolean> {
    try {
      const fs = await import('fs');

      // Check if PID file exists
      try {
        const pidContent = await fs.promises.readFile(
          this.daemonPidFile,
          'utf-8',
        );
        const pid = parseInt(pidContent.trim(), 10);

        // Check if process is running
        try {
          process.kill(pid, 0);

          // Also check if socket exists and is accessible
          try {
            await fs.promises.access(this.socketPath);
            return true;
          } catch {
            // Socket doesn't exist, daemon might be starting
            return false;
          }
        } catch {
          // Process doesn't exist, clean up stale PID file
          await fs.promises.unlink(this.daemonPidFile).catch(() => {});
          return false;
        }
      } catch {
        // PID file doesn't exist
        return false;
      }
    } catch {
      return false;
    }
  }

  /**
   * Start the daemon process
   */
  private async startDaemon(): Promise<void> {
    const daemonScript = plugins.path.join(
      paths.packageDir,
      'dist_ts',
      'daemon.js',
    );

    // Spawn the daemon as a detached process
    const daemonProcess = spawn(process.execPath, [daemonScript], {
      detached: true,
      stdio: ['ignore', 'ignore', 'ignore'],
      env: {
        ...process.env,
        TSPM_DAEMON_MODE: 'true',
      },
    });

    // Unref the process so the parent can exit
    daemonProcess.unref();

    console.log(`Started daemon process with PID: ${daemonProcess.pid}`);

    // Wait for daemon to be ready using SmartIPC's helper
    try {
      await plugins.smartipc.SmartIpc.waitForServer({
        socketPath: this.socketPath,
        timeoutMs: 15000,
      });
    } catch (error) {
      throw new Error(`Daemon failed to start: ${error.message}`);
    }
  }

  /**
   * Stop the daemon
   */
  public async stopDaemon(graceful: boolean = true): Promise<void> {
    if (!(await this.isDaemonRunning())) {
      console.log('Daemon is not running');
      return;
    }

    try {
      await this.connect();
      await this.request('daemon:shutdown', {
        graceful,
        timeout: 10000,
      });

      console.log('Daemon shutdown initiated');

      // Wait for daemon to actually stop
      const maxWaitTime = 15000; // 15 seconds
      const startTime = Date.now();

      while (Date.now() - startTime < maxWaitTime) {
        if (!(await this.isDaemonRunning())) {
          console.log('Daemon stopped successfully');
          return;
        }
        await new Promise((resolve) => setTimeout(resolve, 500));
      }

      console.warn(
        'Daemon did not stop within timeout, it may still be running',
      );
    } catch (error) {
      console.error('Error stopping daemon:', error);

      // Try to kill the process directly if graceful shutdown failed
      try {
        const fs = await import('fs');
        const pidContent = await fs.promises.readFile(
          this.daemonPidFile,
          'utf-8',
        );
        const pid = parseInt(pidContent.trim(), 10);
        process.kill(pid, 'SIGKILL');
        console.log('Force killed daemon process');
      } catch {
        console.error('Could not force kill daemon');
      }
    }
  }

  /**
   * Get daemon status
   */
  public async getDaemonStatus(): Promise<ResponseForMethod<'daemon:status'> | null> {
    try {
      if (!(await this.isDaemonRunning())) {
        return null;
      }

      await this.connect();
      return await this.request('daemon:status', {});
    } catch (error) {
      console.error('Error getting daemon status:', error);
      return null;
    }
  }
}

// Singleton instance
export const tspmIpcClient = new TspmIpcClient();