import 'dart:async'; import 'dart:convert' show jsonDecode, utf8; import 'dart:io'; import 'dart:typed_data'; import 'package:kolibri/kolibri.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-ядро kolibri, /// которое стримит файл с диска (не держит его целиком в памяти). class FileUploader { final Api api; final MessagesModule messages; FileUploader({required this.api, required this.messages}); Stream 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(); var cancelled = false; StreamSubscription? sub; ctrl.onCancel = () { cancelled = true; sub?.cancel(); }; Future 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(); 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 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 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 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 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 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; } }