didvan-app/lib/services/ai/realtime_voice_session.dart

import 'dart:async';
import 'dart:collection';
import 'dart:convert';
import 'dart:io';
import 'dart:math' as math;

import 'package:flutter/foundation.dart';
import 'package:permission_handler/permission_handler.dart';
import 'package:web_socket_channel/io.dart';

import 'package:didvan/services/ai/voice_audio_engine.dart';
import 'package:didvan/services/network/request.dart';
import 'package:didvan/services/network/request_helper.dart';

enum VoiceSessionPhase {
  connecting,
  reconnecting,
  listening,
  hearing,
  searching,
  thinking,
  speaking,
  muted,
  error,
  closed,
}

class RealtimeVoiceSession extends ChangeNotifier {
  static const String _webSocketUrl =
      'wss://api.x.ai/v1/realtime?model=grok-voice-latest';
  static const String _voice = 'leo';
  static const int sampleRate = 24000;
  static const int _maximumPreSessionAudioBytes = sampleRate * 2 * 3;

  final VoiceAudioEngine _audio = VoiceAudioEngine();
  final ListQueue<Uint8List> _preSessionAudio = ListQueue<Uint8List>();
  final List<String> _assistantMessages = <String>[];

  IOWebSocketChannel? _socket;
  StreamSubscription<dynamic>? _socketSubscription;
  StreamSubscription<VoiceAudioEvent>? _audioSubscription;
  Timer? _reconnectTimer;
  int _reconnectAttempt = 0;
  int _connectionGeneration = 0;
  int _preSessionAudioBytes = 0;
  int _playbackGeneration = 0;
  bool _started = false;
  bool _closing = false;
  bool _sessionReady = false;
  bool _playbackActive = false;
  bool _responseDone = false;
  bool _userSpeaking = false;
  bool _micMuted = false;
  bool _hardwareAecEnabled = false;
  String? _conversationId;
  String? _activeResponseId;
  String _liveAssistantTranscript = '';
  Future<void> _playbackChain = Future<void>.value();
  DateTime _lastLevelNotification = DateTime.fromMillisecondsSinceEpoch(0);

  VoiceSessionPhase _phase = VoiceSessionPhase.connecting;
  String? _errorMessage;
  double _microphoneLevel = 0;
  double _assistantLevel = 0;

  VoiceSessionPhase get phase => _phase;
  String? get errorMessage => _errorMessage;
  bool get isMicMuted => _micMuted;
  bool get isUserSpeaking => _userSpeaking;
  bool get isAssistantActive => _activeResponseId != null || _playbackActive;
  bool get hardwareAecEnabled => _hardwareAecEnabled;
  bool get supportsAutomaticInterruption => _hardwareAecEnabled;
  bool get isSessionReady => _sessionReady;
  double get microphoneLevel => _microphoneLevel;
  double get assistantLevel => _assistantLevel;
  String get liveAssistantTranscript => _liveAssistantTranscript;
  List<String> get assistantMessages =>
      List<String>.unmodifiable(_assistantMessages);

  String get statusText {
    switch (_phase) {
      case VoiceSessionPhase.connecting:
        return 'در حال اتصال';
      case VoiceSessionPhase.reconnecting:
        return 'اتصال مجدد';
      case VoiceSessionPhase.listening:
        return 'گوش می‌دهم';
      case VoiceSessionPhase.hearing:
        return 'صدایت را می‌شنوم';
      case VoiceSessionPhase.searching:
        return 'در حال جست‌وجوی وب';
      case VoiceSessionPhase.thinking:
        return 'در حال فکر کردن';
      case VoiceSessionPhase.speaking:
        return 'هوشان در حال صحبت است';
      case VoiceSessionPhase.muted:
        return 'میکروفن خاموش است';
      case VoiceSessionPhase.error:
        return _errorMessage ?? 'خطایی رخ داد';
      case VoiceSessionPhase.closed:
        return 'مکالمه پایان یافت';
    }
  }

  String get _instructions {
    final now = DateTime.now();
    final date = '${now.year.toString().padLeft(4, '0')}-'
        '${now.month.toString().padLeft(2, '0')}-'
        '${now.day.toString().padLeft(2, '0')}';
    return 'تو هوشان، دستیار هوشمند دیدوان هستی. فقط فارسی، کوتاه، روشن و '
        'طبیعی صحبت کن. تاریخ امروز $date و منطقه زمانی Asia/Tehran است. '
        'برای خبر، قیمت، آب‌وهوا، سیاست، ورزش، افراد، شرکت‌ها و هر اطلاعات '
        'زمان‌حساس قبل از پاسخ از web_search استفاده کن. اگر صدای کاربر '
        'نامفهوم یا ناقص بود حدس نزن و از او بخواه دوباره بگوید.';
  }

  Future<void> start() async {
    if (_started || _closing) return;
    _started = true;
    _setPhase(VoiceSessionPhase.connecting);

    final permission = await Permission.microphone.request();
    if (!permission.isGranted) {
      _setError('دسترسی میکروفن لازم است');
      return;
    }

    try {
      // Subscribe before start so the first captured syllable cannot be lost.
      _audioSubscription = _audio.events.listen(
        _handleAudioEvent,
        onError: (Object error, StackTrace stackTrace) {
          _setError('خطا در مسیر صوتی: $error');
        },
      );
      final capabilities = await _audio.start(sampleRate: sampleRate);
      _hardwareAecEnabled = capabilities.aecEnabled;
      debugPrint(
        '🎧 native voice audio ready: '
        'aec=${capabilities.aecEnabled} '
        'ns=${capabilities.noiseSuppressionEnabled} '
        'rate=${capabilities.sampleRate}',
      );
      await _connect();
    } catch (error) {
      _setError('راه‌اندازی صدا ناموفق بود: $error');
    }
  }

  void _handleAudioEvent(VoiceAudioEvent event) {
    if (_closing) return;
    switch (event.type) {
      case VoiceAudioEventType.microphone:
        final pcm = event.pcm;
        if (pcm != null) _handleMicrophoneFrame(pcm, event.rms);
        break;
      case VoiceAudioEventType.playbackStarted:
        _playbackActive = true;
        _setPhase(VoiceSessionPhase.speaking);
        break;
      case VoiceAudioEventType.playbackIdle:
        _playbackActive = false;
        _assistantLevel = 0;
        if (_responseDone) {
          _activeResponseId = null;
        }
        if (!_userSpeaking && _activeResponseId == null) {
          _setPhase(_micMuted
              ? VoiceSessionPhase.muted
              : VoiceSessionPhase.listening);
        } else {
          _notify();
        }
        break;
      case VoiceAudioEventType.error:
        _setError(event.message ?? 'خطا در موتور صوتی');
        break;
    }
  }

  void _handleMicrophoneFrame(Uint8List pcm, double rms) {
    _microphoneLevel =
        math.max(_microphoneLevel * 0.58, (rms * 8).clamp(0.0, 1.0));
    final now = DateTime.now();
    if (now.difference(_lastLevelNotification).inMilliseconds >= 50) {
      _lastLevelNotification = now;
      _notify();
    }

    if (_micMuted) return;

    // A device without platform AEC is kept strictly half-duplex. This
    // guarantees that assistant playback can never be submitted as user audio.
    // The UI exposes immediate tap-to-interrupt on those rare devices.
    if (!_hardwareAecEnabled && isAssistantActive) return;

    if (_sessionReady) {
      _sendAudio(pcm);
      return;
    }

    _preSessionAudio.addLast(pcm);
    _preSessionAudioBytes += pcm.length;
    while (_preSessionAudioBytes > _maximumPreSessionAudioBytes &&
        _preSessionAudio.isNotEmpty) {
      _preSessionAudioBytes -= _preSessionAudio.removeFirst().length;
    }
  }

  Future<String?> _fetchEphemeralToken() async {
    final service = RequestService(
      '${RequestHelper.newApiBaseUrl}/ai/voice/token',
    );
    await service.post();
    if (!service.isSuccess) {
      debugPrint('Voice token failed: ${service.errorMessage}');
      return null;
    }
    final token = service.result['token']?.toString();
    return token == null || token.isEmpty ? null : token;
  }

  Future<void> _connect() async {
    if (_closing) return;
    final generation = ++_connectionGeneration;
    _sessionReady = false;
    _setPhase(
      _reconnectAttempt == 0
          ? VoiceSessionPhase.connecting
          : VoiceSessionPhase.reconnecting,
    );

    try {
      final token = await _fetchEphemeralToken();
      if (_closing || generation != _connectionGeneration) return;
      if (token == null) throw StateError('توکن صوتی دریافت نشد');

      final resumeId = _conversationId;
      final url = resumeId == null
          ? _webSocketUrl
          : '$_webSocketUrl&conversation_id='
              '${Uri.encodeQueryComponent(resumeId)}';
      final webSocket = await WebSocket.connect(
        url,
        protocols: <String>['xai-client-secret.$token'],
      );
      webSocket.pingInterval = const Duration(seconds: 25);
      if (_closing || generation != _connectionGeneration) {
        await webSocket.close();
        return;
      }

      final channel = IOWebSocketChannel(webSocket);
      _socket = channel;
      await _socketSubscription?.cancel();
      _socketSubscription = channel.stream.listen(
        _handleServerEvent,
        onError: (Object error, StackTrace stackTrace) {
          _handleDisconnect(generation, error);
        },
        onDone: () => _handleDisconnect(generation, null),
      );
      _sendSessionConfiguration();
    } catch (error) {
      if (_closing || generation != _connectionGeneration) return;
      debugPrint('Voice WebSocket connection failed: $error');
      _scheduleReconnect();
    }
  }

  void _sendSessionConfiguration() {
    _send(<String, Object?>{
      'type': 'session.update',
      'session': <String, Object?>{
        'voice': _voice,
        'instructions': _instructions,
        'turn_detection': <String, Object?>{
          'type': 'server_vad',
          'threshold': 0.5,
          'prefix_padding_ms': 700,
          'silence_duration_ms': 650,
        },
        'resumption': <String, Object?>{'enabled': true},
        'tools': <Object?>[
          <String, Object?>{
            'type': 'web_search',
            'location': <String, Object?>{
              'country': 'IR',
              'city': 'Tehran',
              'timezone': 'Asia/Tehran',
            },
          },
          <String, Object?>{'type': 'x_search'},
        ],
        'audio': <String, Object?>{
          'input': <String, Object?>{
            'format': <String, Object?>{
              'type': 'audio/pcm',
              'rate': sampleRate,
            },
            'transcription': <String, Object?>{
              'model': 'grok-transcribe',
              'language_hint': 'fa-IR',
            },
          },
          'output': <String, Object?>{
            'format': <String, Object?>{
              'type': 'audio/pcm',
              'rate': sampleRate,
            },
          },
        },
      },
    });
  }

  void _handleServerEvent(dynamic raw) {
    if (_closing) return;
    try {
      final decoded = jsonDecode(
        raw is String ? raw : utf8.decode(raw as List<int>),
      );
      final event = Map<String, dynamic>.from(decoded as Map);
      final type = event['type']?.toString() ?? '';

      switch (type) {
        case 'conversation.created':
          final id = event['conversation']?['id']?.toString();
          if (id != null && id.isNotEmpty) _conversationId = id;
          break;

        case 'session.updated':
          _sessionReady = true;
          _reconnectAttempt = 0;
          if (!_micMuted) {
            while (_preSessionAudio.isNotEmpty) {
              _sendAudio(_preSessionAudio.removeFirst());
            }
          } else {
            _preSessionAudio.clear();
          }
          _preSessionAudioBytes = 0;
          _setPhase(
            _micMuted ? VoiceSessionPhase.muted : VoiceSessionPhase.listening,
          );
          break;

        case 'input_audio_buffer.speech_started':
          if (_micMuted) {
            _send(<String, Object?>{'type': 'input_audio_buffer.clear'});
            break;
          }
          _userSpeaking = true;
          if (isAssistantActive) {
            // With server VAD xAI cancels generation automatically. Only stop
            // local playback; sending response.cancel here creates the
            // "no active response found" race seen in the old implementation.
            _stopLocalPlayback();
            _activeResponseId = null;
            _responseDone = true;
          }
          _setPhase(VoiceSessionPhase.hearing);
          break;

        case 'input_audio_buffer.speech_stopped':
          _userSpeaking = false;
          _setPhase(VoiceSessionPhase.thinking);
          break;

        case 'response.created':
          _startResponse(event);
          break;

        case 'response.output_audio.delta':
          _handleAssistantAudio(event);
          break;

        case 'response.output_audio_transcript.delta':
          final delta = event['delta']?.toString() ?? '';
          if (delta.isNotEmpty) {
            _liveAssistantTranscript += delta;
            _notify();
          }
          break;

        case 'response.output_audio_transcript.done':
          _commitAssistantTranscript(event['transcript']?.toString());
          break;

        case 'response.output_item.added':
          final item = jsonEncode(event['item'] ?? '').toLowerCase();
          if (item.contains('web_search') || item.contains('x_search')) {
            _setPhase(VoiceSessionPhase.searching);
          }
          break;

        case 'response.done':
          if (!_matchesActiveResponse(event['response']?['id']?.toString())) {
            break;
          }
          _responseDone = true;
          _commitAssistantTranscript();
          if (!_playbackActive) {
            _activeResponseId = null;
            _setPhase(_micMuted
                ? VoiceSessionPhase.muted
                : VoiceSessionPhase.listening);
          }
          break;

        case 'error':
          _handleServerError(event);
          break;

        default:
          if (kDebugMode &&
              type.isNotEmpty &&
              type != 'ping' &&
              type != 'response.content_part.added' &&
              type != 'response.content_part.done') {
            debugPrint('Voice event: $type');
          }
      }
    } catch (error) {
      debugPrint('Voice event parse failed: $error');
    }
  }

  void _startResponse(Map<String, dynamic> event) {
    if (_liveAssistantTranscript.trim().isNotEmpty) {
      _commitAssistantTranscript();
    }
    _activeResponseId = event['response']?['id']?.toString() ??
        event['response_id']?.toString();
    _responseDone = false;
    _assistantLevel = 0;
    _playbackGeneration++;
    unawaited(_audio.stopPlayback(generation: _playbackGeneration));
    _setPhase(VoiceSessionPhase.thinking);
  }

  void _handleAssistantAudio(Map<String, dynamic> event) {
    final responseId = event['response_id']?.toString();
    if (_activeResponseId == null ||
        (responseId != null && responseId != _activeResponseId)) {
      return;
    }
    final delta = event['delta']?.toString();
    if (delta == null || delta.isEmpty) return;

    final pcm = base64Decode(delta);
    _assistantLevel = math.max(
      _assistantLevel * 0.55,
      (_pcmRms(pcm) * 4.8).clamp(0.0, 1.0),
    );
    _setPhase(VoiceSessionPhase.speaking);

    final generation = _playbackGeneration;
    _playbackChain = _playbackChain
        .then<void>((_) => _audio.enqueuePlayback(
              pcm,
              generation: generation,
            ))
        .catchError((Object error, StackTrace stackTrace) {
      debugPrint('Playback enqueue failed: $error');
    });
  }

  bool _matchesActiveResponse(String? responseId) {
    if (_activeResponseId == null) return false;
    return responseId == null || responseId == _activeResponseId;
  }

  void _handleServerError(Map<String, dynamic> event) {
    final nested = event['error'];
    final error = nested is Map
        ? Map<String, dynamic>.from(nested)
        : const <String, dynamic>{};
    final message = error['message']?.toString() ??
        event['message']?.toString() ??
        'خطای سرویس صوتی';
    final normalized = message.toLowerCase();
    if (normalized.contains('no active response') ||
        normalized.contains('not active')) {
      return;
    }
    _setError(message);
  }

  void _commitAssistantTranscript([String? finalText]) {
    final text = (finalText?.trim().isNotEmpty ?? false)
        ? finalText!.trim()
        : _liveAssistantTranscript.trim();
    if (text.isNotEmpty &&
        (_assistantMessages.isEmpty || _assistantMessages.last != text)) {
      _assistantMessages.add(text);
    }
    _liveAssistantTranscript = '';
    _notify();
  }

  void _sendAudio(Uint8List pcm) {
    _send(<String, Object?>{
      'type': 'input_audio_buffer.append',
      'audio': base64Encode(pcm),
    });
  }

  void _send(Map<String, Object?> message) {
    try {
      _socket?.sink.add(jsonEncode(message));
    } catch (error) {
      debugPrint('Voice WebSocket send failed: $error');
    }
  }

  void _handleDisconnect(int generation, Object? error) {
    if (_closing || generation != _connectionGeneration) return;
    _sessionReady = false;
    _socket = null;
    _socketSubscription = null;
    _stopLocalPlayback();
    _activeResponseId = null;
    _scheduleReconnect();
  }

  void _scheduleReconnect() {
    if (_closing || _reconnectTimer?.isActive == true) return;
    _setPhase(VoiceSessionPhase.reconnecting);
    final seconds = math.min(1 << math.min(_reconnectAttempt, 4), 15);
    _reconnectAttempt++;
    _reconnectTimer = Timer(Duration(seconds: seconds), () {
      _reconnectTimer = null;
      unawaited(_connect());
    });
  }

  void toggleMute() {
    _micMuted = !_micMuted;
    _preSessionAudio.clear();
    _preSessionAudioBytes = 0;
    if (_micMuted) {
      _send(<String, Object?>{'type': 'input_audio_buffer.clear'});
      _setPhase(VoiceSessionPhase.muted);
    } else {
      _setPhase(
        _sessionReady
            ? VoiceSessionPhase.listening
            : VoiceSessionPhase.connecting,
      );
    }
  }

  void interruptAssistant() {
    if (!isAssistantActive) return;
    // Manual stop is the one place where response.cancel is correct.
    _send(<String, Object?>{'type': 'response.cancel'});
    _stopLocalPlayback();
    _activeResponseId = null;
    _responseDone = true;
    _setPhase(
        _micMuted ? VoiceSessionPhase.muted : VoiceSessionPhase.listening);
  }

  void _stopLocalPlayback() {
    _playbackGeneration++;
    _playbackActive = false;
    _assistantLevel = 0;
    unawaited(_audio.stopPlayback(generation: _playbackGeneration));
  }

  double _pcmRms(Uint8List pcm) {
    if (pcm.length < 2) return 0;
    final data = ByteData.sublistView(pcm);
    var sum = 0.0;
    var count = 0;
    for (var index = 0; index + 1 < pcm.length; index += 2) {
      final sample = data.getInt16(index, Endian.little) / 32768.0;
      sum += sample * sample;
      count++;
    }
    return count == 0 ? 0 : math.sqrt(sum / count);
  }

  void _setPhase(VoiceSessionPhase value) {
    if (_phase == value) {
      _notify();
      return;
    }
    _phase = value;
    _notify();
  }

  void _setError(String message) {
    _errorMessage = message;
    _phase = VoiceSessionPhase.error;
    _notify();
  }

  void _notify() {
    if (!_closing) notifyListeners();
  }

  Future<void> close() async {
    if (_closing) return;
    _closing = true;
    _phase = VoiceSessionPhase.closed;
    _connectionGeneration++;
    _reconnectTimer?.cancel();
    _reconnectTimer = null;
    _sessionReady = false;
    _preSessionAudio.clear();
    _preSessionAudioBytes = 0;

    final socketSubscription = _socketSubscription;
    _socketSubscription = null;
    await socketSubscription?.cancel();
    try {
      await _socket?.sink.close();
    } catch (_) {}
    _socket = null;

    final audioSubscription = _audioSubscription;
    _audioSubscription = null;
    await audioSubscription?.cancel();
    try {
      await _audio.stop();
    } catch (_) {}
  }

  @override
  void dispose() {
    unawaited(close());
    super.dispose();
  }
}