Qlyra/lib/backend/modules/file_uploader.dart

import 'dart:async';
import 'dart:convert' show jsonDecode, utf8;
import 'dart:io';
import 'dart:typed_data';

import 'package:kusoft/kusoft.dart' as kb;

import '../api.dart';
import '../../core/protocol/opcode_map.dart';
import '../../core/utils/logger.dart';
import 'messages.dart';

sealed class UploadEvent {
  const UploadEvent();
}

class UploadProgress extends UploadEvent {
  final int sent;
  final int total;
  const UploadProgress({required this.sent, required this.total});
}

class UploadDone extends UploadEvent {
  final int fileId;
  final String? token;
  final String? url;
  final String filename;
  final int size;

  /// Server-assigned message id. Without it the optimistic message keeps its
  /// local temp id and download URLs cannot be resolved until a restart.
  final String? messageId;

  const UploadDone({
    required this.fileId,
    required this.filename,
    required this.size,
    this.token,
    this.url,
    this.messageId,
  });
}

class UploadError extends UploadEvent {
  final String message;
  const UploadError(this.message);
}

/// Оркестратор медиа-загрузок: control-plane (URL, отправка сообщения) идёт
/// обычными опкодами, data-plane (заливка на CDN) — через Rust-ядро kusoft,
/// которое стримит файл с диска (не держит его целиком в памяти).
class FileUploader {
  final Api api;
  final MessagesModule messages;

  FileUploader({required this.api, required this.messages});

  Stream<UploadEvent> upload({
    required int chatId,
    required File file,
    required String filename,
    required int totalSize,
    int? scheduledTime,
    Duration autoForceAfter = const Duration(seconds: 1),
    Duration overallTimeout = const Duration(minutes: 5),
    Duration progressThrottle = const Duration(milliseconds: 16),
  }) {
    final ctrl = StreamController<UploadEvent>();
    var cancelled = false;
    StreamSubscription<kb.UploadEvent>? sub;

    ctrl.onCancel = () {
      cancelled = true;
      sub?.cancel();
    };

    Future<void> run() async {
      try {
        final info = await messages.requestUploadUrl();
        if (cancelled) return;
        if (info == null) {
          ctrl.add(const UploadError('no_upload_url'));
          return;
        }

        final session = api.session;
        if (session == null) {
          ctrl.add(const UploadError('no_session'));
          return;
        }

        unawaited(() async {
          try {
            await api.sendRequest(Opcode.msgTyping, {
              'chatId': chatId,
              'type': 'FILE',
            });
          } catch (_) {}
        }());

        var status = 0;
        String? error;
        final done = Completer<void>();
        sub = session
            .uploadFilePath(
              url: info.url,
              path: file.path,
              filename: filename,
              connection: 'close',
            )
            .listen(
              (e) {
                switch (e) {
                  case kb.UploadEvent_Progress(:final sent, :final total):
                    ctrl.add(
                      UploadProgress(sent: sent.toInt(), total: total.toInt()),
                    );
                  case kb.UploadEvent_Done(status: final s):
                    status = s;
                  case kb.UploadEvent_Error(:final message):
                    error = message;
                }
              },
              onError: (Object err) {
                error = err.toString();
                if (!done.isCompleted) done.complete();
              },
              onDone: () {
                if (!done.isCompleted) done.complete();
              },
              cancelOnError: true,
            );
        await done.future;
        if (cancelled) return;
        if (error != null) {
          ctrl.add(UploadError(error!));
          return;
        }
        if (status != 200 && status != 0) {
          ctrl.add(UploadError('http_$status'));
          return;
        }

        final messageId = await messages.sendFileMessage(
          chatId,
          info.fileId,
          token: info.token,
          scheduledTime: scheduledTime,
        );
        if (cancelled) return;
        if (messageId == null) {
          ctrl.add(const UploadError('send_failed'));
          return;
        }

        ctrl.add(
          UploadDone(
            fileId: info.fileId,
            token: info.token,
            url: info.url,
            filename: filename,
            size: totalSize,
            messageId: messageId,
          ),
        );
      } catch (e) {
        if (!cancelled) ctrl.add(UploadError(e.toString()));
      } finally {
        await ctrl.close();
      }
    }

    unawaited(run());
    return ctrl.stream;
  }

  Future<bool> uploadMediaFile(
    Uri uri,
    File file, {
    void Function(int sent, int total)? onProgress,
    Duration overallTimeout = const Duration(minutes: 5),
    Duration progressThrottle = const Duration(milliseconds: 16),
  }) async {
    final session = api.session;
    if (session == null) return false;
    try {
      final result = await _consume(
        session.uploadFilePath(
          url: uri.toString(),
          path: file.path,
          filename: _syntheticFilename(),
          contentType: 'application/octet-stream',
          connection: 'close',
        ),
        onProgress: onProgress,
      );
      if (result.error != null) {
        logger.w('uploadMediaFile: ${result.error}');
        return false;
      }
      final respBody = utf8.decode(result.body, allowMalformed: true);
      final hasError =
          respBody.contains('error_msg') || respBody.contains('error_code');
      if (result.status != 200 || hasError) {
        logger.w(
          'uploadMediaFile rejected: status=${result.status} '
          'body=${respBody.length > 500 ? respBody.substring(0, 500) : respBody}',
        );
      }
      return result.status == 200 && !hasError;
    } catch (e) {
      logger.w('uploadMediaFile: $e');
      return false;
    }
  }

  Future<String?> uploadImage(
    Uri uri,
    Uint8List bytes, {
    String filename = 'avatar.jpg',
  }) async {
    final session = api.session;
    if (session == null) return null;
    try {
      final result = await _consume(
        session.uploadPhoto(
          url: uri.toString(),
          data: bytes,
          filename: filename,
        ),
      );
      if (result.error != null || result.status != 200) {
        logger.w('uploadImage: status=${result.status} error=${result.error}');
        return null;
      }
      return _parsePhotoToken(utf8.decode(result.body, allowMalformed: true));
    } catch (e) {
      logger.w('uploadImage: $e');
      return null;
    }
  }

  Future<String?> uploadPhoto(
    Uri uri,
    File file, {
    String filename = 'photo.jpg',
    void Function(int sent, int total)? onProgress,
    Duration progressThrottle = const Duration(milliseconds: 16),
  }) async {
    final session = api.session;
    if (session == null) return null;
    try {
      final result = await _consume(
        session.uploadPhotoPath(
          url: uri.toString(),
          path: file.path,
          filename: filename,
        ),
        onProgress: onProgress,
      );
      if (result.error != null || result.status != 200) {
        logger.w('uploadPhoto: status=${result.status} error=${result.error}');
        return null;
      }
      return _parsePhotoToken(utf8.decode(result.body, allowMalformed: true));
    } catch (e) {
      logger.w('uploadPhoto: $e');
      return null;
    }
  }

  Future<bool> uploadVideoFile(
    Uri uri,
    File file, {
    void Function(int sent, int total)? onProgress,
    int chunkSize = 2 * 1024 * 1024,
    int concurrency = 4,
    Duration overallTimeout = const Duration(minutes: 30),
  }) async {
    final session = api.session;
    if (session == null) return false;
    try {
      final result = await _consume(
        session.uploadVideoPath(
          url: uri.toString(),
          path: file.path,
          chunkSize: chunkSize,
          concurrency: concurrency,
        ),
        onProgress: onProgress,
      );
      if (result.error != null || result.status != 200) {
        logger.w(
          'uploadVideoFile: status=${result.status} error=${result.error}',
        );
      }
      return result.error == null && result.status == 200;
    } catch (e) {
      logger.w('uploadVideoFile: $e');
      return false;
    }
  }

  /// Загрузка видео для истории. В отличие от чата (чанковый `uploadVideoPath`
  /// с GET-handshake), story-эндпоинт `su.oneme.ru/uploadVideo` ждёт **один POST
  /// на весь файл** (как у оригинального клиента) и возвращает медиа-токен в теле
  /// ответа — `[{"token":"..."}]`. Именно этот токен идёт в `media.token`
  /// STORIES_SEND, а не токен из ответа VIDEO_UPLOAD.
  Future<({bool ok, String? token})> uploadVideoWithToken(
    Uri uri,
    File file, {
    void Function(int sent, int total)? onProgress,
  }) async {
    final session = api.session;
    if (session == null) return (ok: false, token: null);
    try {
      final result = await _consume(
        session.uploadFilePath(
          url: uri.toString(),
          path: file.path,
          filename: _syntheticFilename(),
          contentType: 'application/octet-stream',
          connection: 'close',
        ),
        onProgress: onProgress,
      );
      if (result.error != null || result.status != 200) {
        logger.w(
          'uploadVideoWithToken: status=${result.status} error=${result.error}',
        );
        return (ok: false, token: null);
      }
      final token = _parseVideoToken(
        utf8.decode(result.body, allowMalformed: true),
      );
      return (ok: true, token: token);
    } catch (e) {
      logger.w('uploadVideoWithToken: $e');
      return (ok: false, token: null);
    }
  }

  String? _parseVideoToken(String body) {
    try {
      final json = jsonDecode(body);
      // Сервер отвечает массивом: [{"token":"..."}]
      if (json is List) {
        for (final v in json) {
          if (v is Map) {
            final token = v['token'];
            if (token is String && token.isNotEmpty) return token;
          }
        }
      }
      if (json is Map) {
        for (final key in const ['videos', 'video', 'photos']) {
          final node = json[key];
          if (node is Map) {
            for (final v in node.values) {
              if (v is Map) {
                final token = v['token'];
                if (token is String && token.isNotEmpty) return token;
              }
            }
          }
        }
        for (final key in const ['token', 'videoToken', 'photoToken']) {
          final t = json[key];
          if (t is String && t.isNotEmpty) return t;
        }
      }
    } catch (e) {
      logger.w('parseVideoToken: $e');
    }
    return null;
  }

  /// Прогоняет стрим ядра до конца, форвардит прогресс, отдаёт итог.
  Future<({int status, Uint8List body, String? error})> _consume(
    Stream<kb.UploadEvent> stream, {
    void Function(int sent, int total)? onProgress,
  }) async {
    var status = 0;
    var body = Uint8List(0);
    String? error;
    await for (final event in stream) {
      switch (event) {
        case kb.UploadEvent_Progress(:final sent, :final total):
          onProgress?.call(sent.toInt(), total.toInt());
        case kb.UploadEvent_Done(status: final s, body: final b):
          status = s;
          body = b;
        case kb.UploadEvent_Error(:final message):
          error = message;
      }
    }
    return (status: status, body: body, error: error);
  }

  String _syntheticFilename() =>
      (DateTime.now().microsecondsSinceEpoch & 0x7FFFFFFF).toString();

  String? _parsePhotoToken(String body) {
    try {
      final json = jsonDecode(body);
      if (json is Map) {
        final photos = json['photos'];
        if (photos is Map) {
          for (final v in photos.values) {
            if (v is Map) {
              final token = v['token'];
              if (token is String && token.isNotEmpty) return token;
            }
          }
        }
        final pt = json['photoToken'];
        if (pt is String && pt.isNotEmpty) return pt;
      }
    } catch (e) {
      logger.w('parsePhotoToken: $e');
    }
    return null;
  }
}