sevenhill/Qlyra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(chat): отложенные сообщения и отправка видео

Browse Source
This commit is contained in:
klockky
2026-06-13 08:02:12 +00:00
parent d6acfa52fa
commit c8a64e2871
6 changed files with 1262 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/backend/modules/file_uploader.dart
+131
View File
@@ -52,6 +52,7 @@ class FileUploader {
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),
@@ -125,6 +126,7 @@ class FileUploader {
chatId,
info.fileId,
token: info.token,
scheduledTime: scheduledTime,
);
if (cancelled) return;
if (!ok) {
@@ -308,6 +310,135 @@ class FileUploader {
}
}
/// Загружает видео на CDN-URL (vu.okcdn.ru/upload.do), полученный из
/// [MessagesModule.requestVideoUploadUrl], по протоколу OK с докачкой:
/// сначала GET-хендшейк (возвращает уже загруженный оффсет), затем
/// параллельная отправка чанков по [chunkSize] байт через `Content-Range`
/// ([concurrency] одновременных соединений, режим `X-Uploading-Mode:
/// parallel`). Токен уже известен, поэтому возвращается только признак
/// успеха.
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 total = await file.length();
if (total <= 0) return false;
final fileName =
(DateTime.now().microsecondsSinceEpoch & 0x7FFFFFFF).toString();
final handshake = await _okCdnRequest(
uri,
method: 'GET',
fileName: fileName,
timeout: const Duration(seconds: 30),
);
if (handshake == null || handshake.$1 != 200) return false;
var startOffset = 0;
final resumed = int.tryParse(handshake.$2.trim());
if (resumed != null && resumed > 0 && resumed <= total) {
startOffset = resumed;
}
final ranges = <(int, int)>[];
for (var o = startOffset; o < total; o += chunkSize) {
ranges.add((o, o + chunkSize < total ? o + chunkSize : total));
}
if (ranges.isEmpty) return true;
var nextIndex = 0;
var sent = startOffset;
var failed = false;
Future<void> worker() async {
while (!failed) {
final i = nextIndex++;
if (i >= ranges.length) return;
final (start, end) = ranges[i];
final bytes = await _readRange(file, start, end);
final resp = await _okCdnRequest(
uri,
method: 'POST',
fileName: fileName,
body: bytes,
contentRange: 'bytes $start-${end - 1}/$total',
timeout: overallTimeout,
);
if (resp == null || (resp.$1 != 200 && resp.$1 != 201)) {
logger.w('uploadVideoFile: chunk status=${resp?.$1}');
failed = true;
return;
}
sent += end - start;
onProgress?.call(sent, total);
}
}
final workerCount = concurrency < ranges.length
? concurrency
: ranges.length;
await Future.wait(List.generate(workerCount, (_) => worker()));
return !failed;
}
Future<Uint8List> _readRange(File file, int start, int end) async {
final builder = BytesBuilder(copy: false);
await for (final chunk in file.openRead(start, end)) {
builder.add(chunk);
}
return builder.takeBytes();
}
Future<(int, String)?> _okCdnRequest(
Uri uri, {
required String method,
required String fileName,
Uint8List? body,
String? contentRange,
required Duration timeout,
}) async {
Socket? socket;
try {
socket = await _openSocket(uri);
final path = '${uri.path}${uri.hasQuery ? "?${uri.query}" : ""}';
final headers = StringBuffer()
..write('$method $path HTTP/1.1\r\n')
..write('Host: ${uri.host}\r\n')
..write('Content-Type: application/x-binary; charset=x-user-defined\r\n')
..write('Content-Disposition: attachment; fileName="$fileName"\r\n');
if (contentRange != null) {
headers.write('Content-Range: $contentRange\r\n');
}
headers
..write('Content-Length: ${body?.length ?? 0}\r\n')
..write('X-Uploading-Mode: parallel\r\n')
..write('Connection: close\r\n')
..write('\r\n');
socket.add(utf8.encode(headers.toString()));
if (body != null && body.isNotEmpty) socket.add(body);
await socket.flush();
final response = await _readFullResponse(socket, timeout: timeout);
try {
socket.destroy();
} catch (_) {}
return response;
} catch (e) {
logger.w('_okCdnRequest($method): $e');
try {
socket?.destroy();
} catch (_) {}
return null;
}
}
void _writeImageHeaders(Socket socket, Uri uri, int total, {required String boundary}) {
final path = '${uri.path}${uri.hasQuery ? "?${uri.query}" : ""}';
final headers = StringBuffer()
lib/backend/modules/messages.dart
+237 -19
View File
@@ -174,6 +174,18 @@ class FileUploadInfo {
});
}
class VideoUploadInfo {
final String url;
final int videoId;
final String token;
VideoUploadInfo({
required this.url,
required this.videoId,
required this.token,
});
}
class CachedMessage {
final String id;
final int accountId;
@@ -250,6 +262,18 @@ class CachedMessage {
);
}
int? get delayedTimeToFire {
final attrs = payload?['delayedAttributes'];
if (attrs is Map) {
final t = attrs['timeToFire'];
if (t is int) return t;
if (t is String) return int.tryParse(t);
}
return null;
}
bool get isDelayed => delayedTimeToFire != null;
static List<CachedMessage> _decodeRows(List<Map<String, dynamic>> rows) =>
rows.map(CachedMessage.fromDbRow).toList();
@@ -435,15 +459,23 @@ class MessagesModule {
int chatId,
String text, {
bool notify = true,
int? scheduledTime,
}) async {
final message = <String, dynamic>{
'text': text,
'cid': DateTime.now().millisecondsSinceEpoch * -1,
'elements': [],
'attaches': [],
};
if (scheduledTime != null) {
message['delayedAttributes'] = {
'timeToFire': scheduledTime,
'notifySender': true,
};
}
final payload = {
'chatId': chatId,
'message': {
'text': text,
'cid': DateTime.now().millisecondsSinceEpoch * -1,
'elements': [],
'attaches': [],
},
'message': message,
'notify': notify,
};
@@ -467,10 +499,107 @@ class MessagesModule {
return '';
}
/// Загружает отложенные (запланированные) сообщения чата.
///
/// В отличие от обычной истории, отложенные сообщения не сохраняются
/// в локальную БД — они живут только до момента отправки.
Future<List<CachedMessage>> fetchDelayedMessages(
int accountId,
int chatId,
) async {
final payload = {
'chatId': chatId,
'forward': 0,
'backwardTime': 0,
'getChat': false,
'from': 1,
'itemType': 'DELAYED',
'getMessages': true,
'forwardTime': 0,
'interactive': true,
'backward': 150,
};
final response = await _api.sendRequest(Opcode.chatHistory, payload);
if (!response.isOk) return [];
final data = response.payload;
if (data is! Map) return [];
final messagesData = data['messages'];
if (messagesData is! List) return [];
final results = <CachedMessage>[];
for (final m in messagesData) {
if (m is! Map) continue;
final msg = _parseMessage(m.cast<dynamic, dynamic>(), accountId, chatId);
if (msg != null) results.add(msg);
}
results.sort(
(a, b) => (a.delayedTimeToFire ?? a.time).compareTo(
b.delayedTimeToFire ?? b.time,
),
);
return results;
}
/// Редактирует текст (подпись) обычного сообщения.
///
/// Поле `attachments` не передаётся — сервер сохраняет существующие
/// вложения.
Future<bool> editMessage(
int chatId,
String messageId, {
required String text,
}) async {
final id = int.tryParse(messageId);
if (id == null) return false;
final payload = {
'messageId': id,
'chatId': chatId,
'elements': <dynamic>[],
'text': text,
};
final response = await _api.sendRequest(Opcode.msgEdit, payload);
return response.isOk;
}
/// Редактирует отложенное сообщение: меняет текст и/или время отправки.
///
/// Вложения сервер сохраняет сам — в payload они не передаются.
Future<bool> editScheduledMessage(
int chatId,
String messageId, {
required String text,
required int timeToFire,
}) async {
final id = int.tryParse(messageId);
if (id == null) return false;
final payload = {
'messageId': id,
'chatId': chatId,
'elements': <dynamic>[],
'text': text,
'delayedAttributes': {
'timeToFire': timeToFire,
'notifySender': true,
},
};
final response = await _api.sendRequest(Opcode.msgEdit, payload);
return response.isOk;
}
Future<bool> deleteMessages(
int chatId,
List<String> messageIds, {
bool forEveryone = false,
String itemType = 'REGULAR',
}) async {
final ids = messageIds
.map((id) => int.tryParse(id))
@@ -482,7 +611,7 @@ class MessagesModule {
'messageIds': ids,
'chatId': chatId,
'forMe': !forEveryone,
'itemType': 'REGULAR',
'itemType': itemType,
};
final response = await _api.sendRequest(Opcode.msgDelete, payload);
@@ -547,24 +676,32 @@ class MessagesModule {
int fileId, {
String? token,
bool notify = true,
int? scheduledTime,
int maxAttempts = 20,
Duration retryDelay = const Duration(seconds: 1),
Duration initialDelay = const Duration(seconds: 3),
}) async {
final message = <String, dynamic>{
'isLive': false,
'detectShare': false,
'elements': <dynamic>[],
'cid': DateTime.now().millisecondsSinceEpoch,
'attaches': [
if (token != null)
{'_type': 'FILE', 'token': token}
else
{'_type': 'FILE', 'fileId': fileId},
],
};
if (scheduledTime != null) {
message['delayedAttributes'] = {
'timeToFire': scheduledTime,
'notifySender': true,
};
}
final payload = {
'chatId': chatId,
'message': {
'isLive': false,
'detectShare': false,
'elements': <dynamic>[],
'cid': DateTime.now().millisecondsSinceEpoch,
'attaches': [
if (token != null)
{'_type': 'FILE', 'token': token}
else
{'_type': 'FILE', 'fileId': fileId},
],
},
'message': message,
'notify': notify,
};
@@ -597,6 +734,7 @@ class MessagesModule {
List<String> photoTokens, {
String? caption,
bool notify = true,
int? scheduledTime,
int maxAttempts = 20,
Duration retryDelay = const Duration(seconds: 1),
}) async {
@@ -608,6 +746,86 @@ class MessagesModule {
],
};
if (caption != null && caption.isNotEmpty) message['text'] = caption;
if (scheduledTime != null) {
message['delayedAttributes'] = {
'timeToFire': scheduledTime,
'notifySender': true,
};
}
final payload = {'chatId': chatId, 'message': message, 'notify': notify};
for (var attempt = 0; attempt < maxAttempts; attempt++) {
try {
final response = await _api.sendRequest(Opcode.msgSend, payload);
if (!response.isOk) return null;
final data = response.payload;
if (data is Map) {
final msg = data['message'];
if (msg is Map) return Map<String, dynamic>.from(msg);
}
return null;
} on PacketError catch (e) {
if (e.errorKey != 'attachment.not.ready') rethrow;
if (attempt == maxAttempts - 1) return null;
await Future.delayed(retryDelay);
}
}
return null;
}
/// Запрашивает URL для загрузки видео (опкод 82).
Future<VideoUploadInfo?> requestVideoUploadUrl() async {
final response = await _api.sendRequest(Opcode.videoUpload, {
'uploaderType': 0,
'type': 0,
'count': 1,
});
if (!response.isOk) return null;
final data = response.payload;
if (data is! Map) return null;
final infoList = data['info'] as List?;
if (infoList == null || infoList.isEmpty) return null;
final info = infoList.first;
if (info is! Map) return null;
return VideoUploadInfo(
url: info['url'] as String? ?? '',
videoId: info['videoId'] as int? ?? 0,
token: info['token'] as String? ?? '',
);
}
/// Отправляет сообщение с видео по [token], полученному из
/// [requestVideoUploadUrl]. Сервер может ответить `attachment.not.ready`,
/// пока обрабатывает загруженное видео — в этом случае запрос повторяется.
Future<Map<String, dynamic>?> sendVideoMessage(
int chatId,
String token, {
String? caption,
bool notify = true,
int? scheduledTime,
int maxAttempts = 30,
Duration retryDelay = const Duration(seconds: 1),
}) async {
final message = <String, dynamic>{
'isLive': false,
'detectShare': false,
'elements': <dynamic>[],
'cid': DateTime.now().millisecondsSinceEpoch * -1,
'attaches': [
{'videoType': 0, '_type': 'VIDEO', 'token': token},
],
};
if (caption != null && caption.isNotEmpty) message['text'] = caption;
if (scheduledTime != null) {
message['delayedAttributes'] = {
'timeToFire': scheduledTime,
'notifySender': true,
};
}
final payload = {'chatId': chatId, 'message': message, 'notify': notify};
for (var attempt = 0; attempt < maxAttempts; attempt++) {