import 'dart:async'; import 'dart:convert'; import 'dart:io'; import 'package:logger/logger.dart'; import 'package:path/path.dart' as p; import 'app_paths.dart'; import 'local_node_service.dart'; class WindowsLocalNodeService implements LocalNodeService { WindowsLocalNodeService({required Logger logger}) : _logger = logger; final Logger _logger; Future? _startFuture; LocalNodeConfig? _lastConfig; @override Future isRunning({LocalNodeConfig? config}) async { final resolved = _resolveConfig(config); return _isRpcAvailable(resolved); } @override Future ensureRunning({LocalNodeConfig? config}) async { final effective = _resolveConfig(config); if (await _isRpcAvailable(effective)) { return true; } return _startNode(effective); } @override Future start({LocalNodeConfig? config}) async { final effective = _resolveConfig(config); return _startNode(effective); } @override Future stop() async { final effective = _resolveConfig(null); if (!await _isRpcAvailable(effective)) { return true; } final rpcAccepted = await _stopViaRpc(effective); final stopped = await _waitForStop(effective); if (!stopped) { _logger.w('Failed to stop local node via RPC.'); return false; } if (!rpcAccepted) { _logger.i('Local node stopped despite RPC response/read failure.'); } return true; } @override Future restart({LocalNodeConfig? config}) async { final effective = _resolveConfig(config); final stopped = await stop(); if (!stopped) { return false; } return _startNode(effective); } Future _startNode(LocalNodeConfig config) async { if (_startFuture != null) { return _startFuture!; } _startFuture = _startNodeInternal(config); try { final started = await _startFuture!; if (started) { return true; } final eventuallyAvailable = await _waitForRpc( config, attempts: 8, logOnFailure: false, ); if (eventuallyAvailable) { _logger.i('Local node RPC became available after delayed startup.'); return true; } return false; } finally { _startFuture = null; } } Future _startNodeInternal(LocalNodeConfig config) async { _lastConfig = config; final binary = await _locateBinary(); if (binary == null) { _logger.w( 'Local node binary not found. Expected external\\daemon\\peyad.exe', ); return false; } final daemonLogFile = await AppPaths.localNodeLogFile(); final launcherLogFile = await AppPaths.localNodeLauncherLogFile(); await launcherLogFile.parent.create(recursive: true); final workingDirectory = p.dirname(binary); final args = [ '--non-interactive', '--rpc-bind-ip', config.rpcHost, '--rpc-bind-port', config.rpcPort.toString(), '--log-file', daemonLogFile.path, ...config.extraArgs, ]; try { final process = await Process.start( binary, args, workingDirectory: workingDirectory, mode: ProcessStartMode.detachedWithStdio, ); _logger.i( 'Starting local node: "$binary" ${args.join(' ')} ' '(cwd=$workingDirectory, log=${daemonLogFile.path})', ); unawaited(_pipeLauncherLogs(process, launcherLogFile)); final earlyExitCode = await _waitForEarlyExit(process); if (earlyExitCode != null) { _logger.w( 'Local node exited immediately with code $earlyExitCode. ' 'See ${launcherLogFile.path} and ${daemonLogFile.path}', ); return false; } } catch (error) { _logger.w('Failed to start local node: $error'); return false; } return _waitForRpc(config); } Future _pipeLauncherLogs(Process process, File logFile) async { final sink = logFile.openWrite(mode: FileMode.writeOnlyAppend); sink.writeln('=== ${DateTime.now().toIso8601String()} local node launch ==='); unawaited( process.stdout .transform(utf8.decoder) .transform(const LineSplitter()) .forEach((line) => sink.writeln('[stdout] $line')), ); unawaited( process.stderr .transform(utf8.decoder) .transform(const LineSplitter()) .forEach((line) => sink.writeln('[stderr] $line')), ); unawaited( process.exitCode.then((code) async { sink.writeln('[exit] $code'); await sink.flush(); await sink.close(); }), ); } Future _waitForEarlyExit(Process process) async { const startupGrace = Duration(seconds: 2); final result = await Future.any([ process.exitCode, Future.delayed(startupGrace, () => null), ]); return result is int ? result : null; } Future _waitForRpc( LocalNodeConfig config, { int attempts = 20, bool logOnFailure = true, }) async { for (var i = 0; i < attempts; i++) { if (await _isRpcAvailable(config)) { return true; } await Future.delayed(const Duration(seconds: 1)); } if (logOnFailure) { _logger.w('Local node did not become available after ${attempts}s.'); } return false; } Future _waitForStop(LocalNodeConfig config) async { const attempts = 15; for (var i = 0; i < attempts; i++) { if (!await _isRpcAvailable(config)) { return true; } await Future.delayed(const Duration(seconds: 1)); } return false; } Future _isRpcAvailable(LocalNodeConfig config) async { try { final socket = await Socket.connect( config.rpcHost, config.rpcPort, timeout: const Duration(seconds: 1), ); socket.destroy(); return true; } catch (_) { return false; } } Future _stopViaRpc(LocalNodeConfig config) async { HttpClient? client; try { client = HttpClient(); final uri = Uri.parse('http://${config.rpcHost}:${config.rpcPort}/stop_daemon'); final request = await client.postUrl(uri); request.headers.contentType = ContentType.json; request.write('{}'); final response = await request.close(); final body = await utf8.decoder.bind(response).join(); if (response.statusCode != HttpStatus.ok) { _logger.w( 'Local node stop_daemon returned HTTP ${response.statusCode}: $body', ); return false; } return true; } catch (error) { _logger.w('Failed to call stop_daemon on local node: $error'); return false; } finally { client?.close(force: true); } } LocalNodeConfig _resolveConfig(LocalNodeConfig? config) { final base = config ?? _lastConfig ?? const LocalNodeConfig(); return _applyOverrides(base); } LocalNodeConfig _applyOverrides(LocalNodeConfig config) { var host = config.rpcHost; var port = config.rpcPort; final args = config.extraArgs; for (var i = 0; i < args.length; i++) { final arg = args[i]; if (arg.startsWith('--rpc-bind-ip=')) { host = arg.split('=').last.trim(); continue; } if (arg == '--rpc-bind-ip' && i + 1 < args.length) { host = args[i + 1].trim(); continue; } if (arg.startsWith('--rpc-bind-port=')) { final parsed = int.tryParse(arg.split('=').last.trim()); if (parsed != null) { port = parsed; } continue; } if (arg == '--rpc-bind-port' && i + 1 < args.length) { final parsed = int.tryParse(args[i + 1].trim()); if (parsed != null) { port = parsed; } } } return config.copyWith(rpcHost: host, rpcPort: port); } Future _locateBinary() async { final cwd = Directory.current.path; final executableDir = p.dirname(Platform.resolvedExecutable); final candidates = [ p.join(executableDir, 'external', 'daemon', 'peyad.exe'), p.join(cwd, 'external', 'daemon', 'peyad.exe'), p.join(cwd, '..', 'external', 'daemon', 'peyad.exe'), ]; for (final candidate in candidates) { final file = File(candidate); if (await file.exists()) { return file.path; } } return null; } }