import { Injectable, inject } from '@angular/core'; import { take } from 'rxjs'; import { Store } from '@ngrx/store'; import { recordDebugNetworkFileChunk } from '../../../../infrastructure/realtime/logging/debug-network-metrics'; import { RealtimeSessionFacade } from '../../../../core/realtime'; import { AppI18nService } from '../../../../core/i18n'; import { selectCurrentUserId } from '../../../../store/users/users.selectors'; import { AttachmentStorageService } from '../../infrastructure/services/attachment-storage.service'; import { MAX_AUTO_SAVE_SIZE_BYTES } from '../../domain/constants/attachment.constants'; import { isImageAttachment, resolvePublishAttachmentIsImage } from '../../domain/logic/attachment-image.rules'; import { base64DecodedByteLength, decodeBase64ToUint8Array } from '../../domain/logic/attachment-blob.rules'; import { isSharingFromThisDevice, canHostAttachment } from '../../domain/logic/attachment-sharing.rules'; import { selectFileRequestPeer } from '../../domain/logic/attachment-request.rules'; import { canReceiveAttachment, needsAttachmentDisplayHydration, shouldCopyLargeUploaderFileToAppData, shouldPersistDownloadedAttachment, shouldStreamAttachmentReceiveToDisk } from '../../domain/logic/attachment.logic'; import { normalizeAttachmentMeta } from '../../domain/logic/attachment-normalize.rules'; import { resolveAttachmentMime } from '../../domain/logic/attachment-mime.rules'; import { shouldServeAttachmentFromDiskPath } from '../../domain/logic/attachment-serve.rules'; import type { Attachment, AttachmentMeta } from '../../domain/models/attachment.model'; import { ATTACHMENT_TRANSFER_EWMA_CURRENT_WEIGHT, ATTACHMENT_TRANSFER_EWMA_PREVIOUS_WEIGHT, DEFAULT_ATTACHMENT_MIME_TYPE, ATTACHMENT_DOWNLOAD_FAILED_KEY, ATTACHMENT_FILE_TOO_LARGE_KEY, ATTACHMENT_CHUNKS_OUT_OF_ORDER_KEY, ATTACHMENT_PREPARE_DOWNLOAD_FAILED_KEY, ATTACHMENT_WRITE_DOWNLOAD_FAILED_KEY, FILE_NOT_FOUND_REQUEST_ERROR_KEY, NO_CONNECTED_PEERS_REQUEST_ERROR_KEY, UPLOADER_LOCAL_FILE_MISSING_ERROR_KEY } from '../../domain/constants/attachment-transfer.constants'; import { type FileAnnounceEvent, type FileAnnouncePayload, type FileCancelEvent, type FileCancelPayload, type FileChunkPayload, type FileChunkAckPayload, type FileChunkAckEvent, type FileNotFoundEvent, type FileNotFoundPayload, type FileRequestEvent, type FileRequestPayload, type LocalFileWithPath } from '../../domain/models/attachment-transfer.model'; import { AttachmentPersistenceService } from './attachment-persistence.service'; import { AttachmentRuntimeStore } from './attachment-runtime.store'; import { AttachmentTransferTransportService } from './attachment-transfer-transport.service'; import { AttachmentChunkAckService } from './attachment-chunk-ack.service'; interface DiskReceiveAssembly { path: string; receivedCount: number; receivedIndexes: Set; total: number; } interface ValidFileChunkPayload { data: string; fileId: string; fromPeerId?: string; index: number; messageId: string; total: number; } function isValidFileChunkPayload(payload: FileChunkPayload): payload is FileChunkPayload & ValidFileChunkPayload { const { messageId, fileId, index, total, data } = payload; return !!messageId && !!fileId && typeof data === 'string' && typeof index === 'number' && typeof total === 'number' && Number.isInteger(index) && Number.isInteger(total) && total > 0 && index >= 0 && index < total; } @Injectable({ providedIn: 'root' }) export class AttachmentTransferService { private readonly ngrxStore = inject(Store); private readonly webrtc = inject(RealtimeSessionFacade); private readonly appI18n = inject(AppI18nService); private readonly runtimeStore = inject(AttachmentRuntimeStore); private readonly attachmentStorage = inject(AttachmentStorageService); private readonly persistence = inject(AttachmentPersistenceService); private readonly transport = inject(AttachmentTransferTransportService); private readonly chunkAcks = inject(AttachmentChunkAckService); private readonly diskReceiveAssemblies = new Map(); private readonly diskReceiveLocks = new Map>(); private readonly activeOutboundTransfers = new Set(); getAttachmentMetasForMessages(messageIds: string[]): Record { const result: Record = {}; for (const messageId of messageIds) { const attachments = this.runtimeStore.getAttachmentsForMessage(messageId); if (attachments.length > 0) { result[messageId] = attachments.map((attachment) => ({ id: attachment.id, messageId: attachment.messageId, filename: attachment.filename, size: attachment.size, mime: attachment.mime, isImage: attachment.isImage, uploaderPeerId: attachment.uploaderPeerId, filePath: undefined, savedPath: undefined })); } } return result; } async registerSyncedAttachments( attachmentMap: Record, messageRoomIds?: Record ): Promise { await this.persistence.whenReady(); if (messageRoomIds) { for (const [messageId, roomId] of Object.entries(messageRoomIds)) { this.runtimeStore.rememberMessageRoom(messageId, roomId); } } const newAttachments: Attachment[] = []; for (const [messageId, metas] of Object.entries(attachmentMap)) { const existing = [...this.runtimeStore.getAttachmentsForMessage(messageId)]; for (const meta of metas) { const alreadyKnown = existing.find((entry) => entry.id === meta.id); if (!alreadyKnown) { const attachment: Attachment = { ...normalizeAttachmentMeta(meta), available: false, receivedBytes: 0 }; existing.push(attachment); newAttachments.push(attachment); } } this.runtimeStore.setAttachmentsForMessage(messageId, existing); } if (newAttachments.length > 0) { this.runtimeStore.touch(); for (const attachment of newAttachments) { void this.persistence.persistAttachmentMeta(attachment); } } } async requestFromAnyPeer(messageId: string, attachment: Attachment): Promise { const requestKey = this.buildRequestKey(messageId, attachment.id); const clearedRequestError = this.clearAttachmentRequestError(attachment); // Mark the request pending synchronously so concurrent triggers (file-announce, // message sync, peer connect) cannot double-request the same file - a duplicate // request makes the sender stream the file twice and corrupts byte accounting. this.runtimeStore.setPendingRequestPeers(requestKey, new Set()); if (needsAttachmentDisplayHydration(attachment)) { const hydratedLocally = await this.persistence.tryRestoreAttachmentFromLocal(attachment); if (hydratedLocally) { this.runtimeStore.deletePendingRequest(requestKey); this.runtimeStore.touch(); return; } } if (!attachment.available) { const restoredLocally = await this.persistence.tryRestoreAttachmentFromLocal(attachment); if (restoredLocally) { this.runtimeStore.deletePendingRequest(requestKey); this.runtimeStore.touch(); return; } } const connectedPeers = this.webrtc.getConnectedPeers(); const currentUserId = await this.resolveCurrentUserId(); // Only the device that actually still holds the original bytes should report a // missing local upload. A second device of the same user that merely synced the // metadata is not the sharing device, so it falls back to the regular peer-request // flow (and the "no connected peers" error when offline) like any other recipient. const sharingFromThisDevice = isSharingFromThisDevice(attachment, currentUserId); if (connectedPeers.length === 0) { this.runtimeStore.deletePendingRequest(requestKey); attachment.requestError = sharingFromThisDevice ? this.appI18n.instant(UPLOADER_LOCAL_FILE_MISSING_ERROR_KEY) : this.appI18n.instant(NO_CONNECTED_PEERS_REQUEST_ERROR_KEY); this.runtimeStore.touch(); console.warn('[Attachments] No connected peers to request file from'); return; } if (!canReceiveAttachment(attachment, this.receiveCapabilities())) { this.runtimeStore.deletePendingRequest(requestKey); attachment.requestError = this.appI18n.instant(ATTACHMENT_FILE_TOO_LARGE_KEY); this.runtimeStore.touch(); return; } if (clearedRequestError) this.runtimeStore.touch(); this.sendFileRequestToNextPeer(messageId, attachment.id, attachment.uploaderPeerId); } handleFileNotFound(payload: FileNotFoundPayload): void { const { messageId, fileId } = payload; if (!messageId || !fileId) return; const attachments = this.runtimeStore.getAttachmentsForMessage(messageId); const attachment = attachments.find((entry) => entry.id === fileId); const didSendRequest = this.sendFileRequestToNextPeer(messageId, fileId, attachment?.uploaderPeerId); if (!didSendRequest && attachment) { attachment.requestError = this.appI18n.instant(FILE_NOT_FOUND_REQUEST_ERROR_KEY); this.runtimeStore.touch(); } } requestImageFromAnyPeer(messageId: string, attachment: Attachment): Promise { if (needsAttachmentDisplayHydration(attachment)) { return this.requestFromAnyPeer(messageId, attachment); } if ((attachment.receivedBytes ?? 0) > 0 || this.hasPendingRequest(messageId, attachment.id)) { this.cancelRequest(messageId, attachment); } return this.requestFromAnyPeer(messageId, attachment); } requestFile(messageId: string, attachment: Attachment): Promise { return this.requestFromAnyPeer(messageId, attachment); } hasPendingRequest(messageId: string, fileId: string): boolean { return this.runtimeStore.hasPendingRequest(this.buildRequestKey(messageId, fileId)); } async publishAttachments( messageId: string, files: File[], uploaderPeerId?: string ): Promise { const attachments: Attachment[] = []; for (const file of files) { const fileId = crypto.randomUUID?.() ?? `${Date.now()}-${Math.random()}`; const attachment: Attachment = { id: fileId, messageId, filename: file.name, size: file.size, mime: resolveAttachmentMime(file.name, file.type || DEFAULT_ATTACHMENT_MIME_TYPE), isImage: resolvePublishAttachmentIsImage(file), uploaderPeerId, filePath: (file as LocalFileWithPath).path, available: false }; attachments.push(attachment); this.runtimeStore.setOriginalFile(`${messageId}:${fileId}`, file); const fileUrl = attachment.filePath && this.isPlayableMedia(attachment) ? await this.attachmentStorage.getFileUrl(attachment.filePath) : null; if (fileUrl) { attachment.objectUrl = fileUrl; attachment.available = true; } else { try { attachment.objectUrl = URL.createObjectURL(file); attachment.available = true; } catch { /* non-critical */ } } await this.persistPublishedAttachment(attachment, file); this.releaseInMemoryUploadCopyIfPersisted(`${messageId}:${fileId}`, attachment); const fileAnnounceEvent: FileAnnounceEvent = { type: 'file-announce', messageId, file: { id: fileId, filename: attachment.filename, size: attachment.size, mime: attachment.mime, isImage: attachment.isImage, uploaderPeerId } }; this.webrtc.broadcastMessage(fileAnnounceEvent); } const existingList = this.runtimeStore.getAttachmentsForMessage(messageId); this.runtimeStore.setAttachmentsForMessage(messageId, [...existingList, ...attachments]); this.runtimeStore.touch(); for (const attachment of attachments) { void this.persistence.persistAttachmentMeta(attachment); } } handleFileAnnounce(payload: FileAnnouncePayload): boolean { const { messageId, file } = payload; if (!messageId || !file) { return false; } if (payload.fromPeerId) { this.runtimeStore.addAnnouncedHost(this.buildRequestKey(messageId, file.id), payload.fromPeerId); } const list = [...this.runtimeStore.getAttachmentsForMessage(messageId)]; const alreadyKnown = list.find((entry) => entry.id === file.id); if (alreadyKnown) { alreadyKnown.filename = file.filename; alreadyKnown.size = file.size; alreadyKnown.mime = resolveAttachmentMime(file.filename, file.mime); alreadyKnown.isImage = isImageAttachment({ filename: file.filename, isImage: !!file.isImage, mime: alreadyKnown.mime }); alreadyKnown.uploaderPeerId = file.uploaderPeerId ?? alreadyKnown.uploaderPeerId; this.runtimeStore.touch(); void this.persistence.persistAttachmentMeta(alreadyKnown); return false; } const normalizedMeta = normalizeAttachmentMeta({ id: file.id, messageId, filename: file.filename, size: file.size, mime: file.mime, isImage: !!file.isImage, uploaderPeerId: file.uploaderPeerId }); const runtimeAttachment: Attachment = { ...normalizedMeta, available: false, receivedBytes: 0 }; list.push(runtimeAttachment); this.runtimeStore.setAttachmentsForMessage(messageId, list); this.runtimeStore.touch(); void this.persistence.persistAttachmentMeta(runtimeAttachment); return true; } handleFileChunk(payload: FileChunkPayload): void { if (!isValidFileChunkPayload(payload)) { return; } const { messageId, fileId, fromPeerId, index, total, data } = payload; const list = this.runtimeStore.getAttachmentsForMessage(messageId); const attachment = list.find((entry) => entry.id === fileId); if (!attachment) return; if (attachment.available) { // Transfer already completed (or restored locally) - trailing chunks from // a redundant stream must not restart accounting or rewrite the file. return; } if ((attachment.receivedBytes ?? 0) > attachment.size) { return; } if (!canReceiveAttachment(attachment, this.receiveCapabilities())) { attachment.requestError = this.appI18n.instant(ATTACHMENT_FILE_TOO_LARGE_KEY); this.runtimeStore.touch(); return; } if (this.shouldReceiveToDisk(attachment)) { void this.receiveDiskChunk(attachment, { data, fileId, fromPeerId, index, messageId, total }); return; } if (attachment.size > MAX_AUTO_SAVE_SIZE_BYTES) { attachment.requestError = this.appI18n.instant(ATTACHMENT_FILE_TOO_LARGE_KEY); this.runtimeStore.touch(); return; } const decodedBytes = this.transport.decodeBase64(data); const assemblyKey = `${messageId}:${fileId}`; const requestKey = this.buildRequestKey(messageId, fileId); this.runtimeStore.deletePendingRequest(requestKey); this.clearAttachmentRequestError(attachment); const chunkBuffer = this.getOrCreateChunkBuffer(assemblyKey, total); if (index >= chunkBuffer.length || chunkBuffer[index]) { // Duplicate delivery (e.g. a redundant concurrent stream) - the chunk is // already buffered, so it must not count toward transfer progress. return; } chunkBuffer[index] = decodedBytes.buffer as ArrayBuffer; this.runtimeStore.setChunkCount(assemblyKey, (this.runtimeStore.getChunkCount(assemblyKey) ?? 0) + 1); this.updateTransferProgress(attachment, decodedBytes.byteLength, fromPeerId); this.runtimeStore.touch(); void this.finalizeTransferIfComplete(attachment, assemblyKey, total); this.emitChunkAck({ fileId, fromPeerId, index, messageId }); } handleFileChunkAck(payload: FileChunkAckPayload): void { const { messageId, fileId, index } = payload; if (!messageId || !fileId || typeof index !== 'number' || !Number.isInteger(index) || index < 0) { return; } this.chunkAcks.resolveAck(messageId, fileId, index); } async handleFileRequest(payload: FileRequestPayload): Promise { const { messageId, fileId, fromPeerId } = payload; if (!messageId || !fileId || !fromPeerId) return; const transferKey = this.buildTransferKey(messageId, fileId, fromPeerId); if (this.activeOutboundTransfers.has(transferKey)) { // A stream for this exact request is already in flight - sending the file // twice in parallel duplicates chunks and corrupts the receiver's assembly. return; } // A fresh request supersedes any earlier cancellation from this peer. this.runtimeStore.deleteCancelledTransfer(transferKey); this.activeOutboundTransfers.add(transferKey); try { await this.streamRequestedFile(messageId, fileId, fromPeerId); } finally { this.activeOutboundTransfers.delete(transferKey); } } cancelRequest(messageId: string, attachment: Attachment): void { try { const requestKey = this.buildRequestKey(messageId, attachment.id); const assemblyKey = `${messageId}:${attachment.id}`; const pendingPeers = this.runtimeStore.getPendingRequestPeers(requestKey); this.runtimeStore.deleteChunkBuffer(assemblyKey); this.runtimeStore.deleteChunkCount(assemblyKey); this.runtimeStore.deletePendingRequest(requestKey); void this.deleteDiskReceiveAssembly(assemblyKey); this.chunkAcks.cancelPendingForFile(messageId, attachment.id); attachment.receivedBytes = 0; attachment.speedBps = 0; attachment.startedAtMs = undefined; attachment.lastUpdateMs = undefined; attachment.requestError = undefined; if (attachment.objectUrl) { try { URL.revokeObjectURL(attachment.objectUrl); } catch { /* ignore */ } attachment.objectUrl = undefined; } attachment.available = false; this.runtimeStore.touch(); const fileCancelEvent: FileCancelEvent = { type: 'file-cancel', messageId, fileId: attachment.id }; const peersToNotify = new Set(); if (pendingPeers) { for (const peerId of pendingPeers) { peersToNotify.add(peerId); } } if (attachment.uploaderPeerId) { peersToNotify.add(attachment.uploaderPeerId); } for (const peerId of peersToNotify) { this.webrtc.sendToPeer(peerId, fileCancelEvent); } } catch { /* best-effort */ } } handleFileCancel(payload: FileCancelPayload): void { const { messageId, fileId, fromPeerId } = payload; if (!messageId || !fileId || !fromPeerId) return; this.runtimeStore.addCancelledTransfer( this.buildTransferKey(messageId, fileId, fromPeerId) ); } async fulfillRequestWithFile( messageId: string, fileId: string, targetPeerId: string, file: File ): Promise { this.runtimeStore.setOriginalFile(`${messageId}:${fileId}`, file); const transferKey = this.buildTransferKey(messageId, fileId, targetPeerId); if (this.activeOutboundTransfers.has(transferKey)) { return; } this.runtimeStore.deleteCancelledTransfer(transferKey); this.activeOutboundTransfers.add(transferKey); try { await this.transport.streamFileToPeer( targetPeerId, messageId, fileId, file, () => this.isTransferCancelled(targetPeerId, messageId, fileId) ); } finally { this.activeOutboundTransfers.delete(transferKey); } } private async streamRequestedFile( messageId: string, fileId: string, fromPeerId: string ): Promise { const exactKey = `${messageId}:${fileId}`; const list = this.runtimeStore.getAttachmentsForMessage(messageId); const attachment = list.find((entry) => entry.id === fileId); const diskPath = attachment ? await this.attachmentStorage.resolveExistingPath(attachment) : null; if (diskPath && shouldServeAttachmentFromDiskPath(await this.attachmentStorage.getFileSize(diskPath))) { await this.transport.streamFileFromDiskToPeer( fromPeerId, messageId, fileId, diskPath, () => this.isTransferCancelled(fromPeerId, messageId, fileId) ); return; } const originalFile = this.runtimeStore.getOriginalFile(exactKey) ?? this.runtimeStore.findOriginalFileByFileId(fileId); if (originalFile) { await this.transport.streamFileToPeer( fromPeerId, messageId, fileId, originalFile, () => this.isTransferCancelled(fromPeerId, messageId, fileId) ); return; } if (attachment?.isImage) { const roomName = await this.persistence.resolveCurrentRoomName(); const legacyDiskPath = await this.attachmentStorage.resolveLegacyImagePath( attachment.filename, roomName ); if (legacyDiskPath && shouldServeAttachmentFromDiskPath(await this.attachmentStorage.getFileSize(legacyDiskPath))) { await this.transport.streamFileFromDiskToPeer( fromPeerId, messageId, fileId, legacyDiskPath, () => this.isTransferCancelled(fromPeerId, messageId, fileId) ); return; } } if (attachment?.available && attachment.objectUrl) { try { const response = await fetch(attachment.objectUrl); const blob = await response.blob(); const file = new File([blob], attachment.filename, { type: attachment.mime }); await this.transport.streamFileToPeer( fromPeerId, messageId, fileId, file, () => this.isTransferCancelled(fromPeerId, messageId, fileId) ); return; } catch { /* fall through */ } } const fileNotFoundEvent: FileNotFoundEvent = { type: 'file-not-found', messageId, fileId }; this.webrtc.sendToPeer(fromPeerId, fileNotFoundEvent); } private resolveCurrentUserId(): Promise { return new Promise((resolve) => { this.ngrxStore .select(selectCurrentUserId) .pipe(take(1)) .subscribe((userId) => resolve(userId)); }); } private buildTransferKey(messageId: string, fileId: string, peerId: string): string { return `${messageId}:${fileId}:${peerId}`; } private buildRequestKey(messageId: string, fileId: string): string { return `${messageId}:${fileId}`; } private clearAttachmentRequestError(attachment: Attachment): boolean { if (!attachment.requestError) return false; attachment.requestError = undefined; return true; } private isTransferCancelled(targetPeerId: string, messageId: string, fileId: string): boolean { return this.runtimeStore.hasCancelledTransfer( this.buildTransferKey(messageId, fileId, targetPeerId) ); } private sendFileRequestToNextPeer( messageId: string, fileId: string, preferredPeerId?: string ): boolean { const connectedPeers = this.webrtc.getConnectedPeers(); const requestKey = this.buildRequestKey(messageId, fileId); const triedPeers = this.runtimeStore.getPendingRequestPeers(requestKey) ?? new Set(); const announcedHosts = this.runtimeStore.getAnnouncedHosts(requestKey); const targetPeerId = selectFileRequestPeer({ connectedPeers, triedPeers, announcedHosts, uploaderPeerId: preferredPeerId }); if (!targetPeerId) { this.runtimeStore.deletePendingRequest(requestKey); return false; } triedPeers.add(targetPeerId); this.runtimeStore.setPendingRequestPeers(requestKey, triedPeers); const fileRequestEvent: FileRequestEvent = { type: 'file-request', messageId, fileId }; this.webrtc.sendToPeer(targetPeerId, fileRequestEvent); return true; } private getOrCreateChunkBuffer(assemblyKey: string, total: number): (ArrayBuffer | undefined)[] { const existingChunkBuffer = this.runtimeStore.getChunkBuffer(assemblyKey); if (existingChunkBuffer) { return existingChunkBuffer; } // Dense initialization - sparse arrays from `new Array(total)` skip holes in // `every`/`some`, which lets incomplete transfers pass completion checks. const createdChunkBuffer = Array.from({ length: total }); this.runtimeStore.setChunkBuffer(assemblyKey, createdChunkBuffer); this.runtimeStore.setChunkCount(assemblyKey, 0); return createdChunkBuffer; } private updateTransferProgress( attachment: Attachment, chunkByteLength: number, fromPeerId?: string ): void { const now = Date.now(); const previousReceived = attachment.receivedBytes ?? 0; attachment.receivedBytes = previousReceived + chunkByteLength; if (fromPeerId) { recordDebugNetworkFileChunk(fromPeerId, chunkByteLength, now); } if (!attachment.startedAtMs) attachment.startedAtMs = now; if (!attachment.lastUpdateMs) attachment.lastUpdateMs = now; const elapsedMs = Math.max(1, now - attachment.lastUpdateMs); const instantaneousBps = (chunkByteLength / elapsedMs) * 1000; const previousSpeed = attachment.speedBps ?? instantaneousBps; attachment.speedBps = ATTACHMENT_TRANSFER_EWMA_PREVIOUS_WEIGHT * previousSpeed + ATTACHMENT_TRANSFER_EWMA_CURRENT_WEIGHT * instantaneousBps; attachment.lastUpdateMs = now; } private async finalizeTransferIfComplete( attachment: Attachment, assemblyKey: string, total: number ): Promise { const receivedChunkCount = this.runtimeStore.getChunkCount(assemblyKey) ?? 0; const completeBuffer = this.runtimeStore.getChunkBuffer(assemblyKey); if (!completeBuffer || receivedChunkCount < total) { return; } const bufferedChunks = completeBuffer.filter( (part): part is ArrayBuffer => part instanceof ArrayBuffer ); // Every chunk index must be present - byte counters are never a substitute // for chunk completeness, otherwise partial files finalize as corrupt blobs. if (bufferedChunks.length !== total || completeBuffer.length !== total) { return; } const blob = new Blob(bufferedChunks, { type: attachment.mime }); if (shouldPersistDownloadedAttachment(attachment)) { await this.persistence.saveFileToDisk(attachment, blob); } attachment.objectUrl = URL.createObjectURL(blob); attachment.available = true; // Release assembly state only after the attachment is marked available so a // trailing duplicate chunk cannot restart accounting in a fresh buffer. this.runtimeStore.deleteChunkBuffer(assemblyKey); this.runtimeStore.deleteChunkCount(assemblyKey); this.runtimeStore.touch(); void this.persistence.persistAttachmentMeta(attachment); void this.announceLocalHost(attachment); if (this.isPlayableMedia(attachment)) { await this.hydratePlayableMediaAfterDiskReceive(attachment); } } /** * Persist an outgoing attachment so it survives restart/logout-login: small * files are auto-saved, oversized uploader media is copied or streamed to the * active store, and the inline object URL is upgraded to the saved file when * the store provides one. Oversized media on capped stores (browser) stays in * memory / peer-served and degrades gracefully. */ private async persistPublishedAttachment(attachment: Attachment, file: File): Promise { if (attachment.size <= MAX_AUTO_SAVE_SIZE_BYTES) { void this.persistence.saveFileToDisk(attachment, file); return; } if (!this.attachmentStorage.canPersistSize(attachment.size)) { return; } if (shouldCopyLargeUploaderFileToAppData( attachment, attachment.filePath, this.attachmentStorage.canCopyFiles() ) && attachment.filePath) { await this.applySavedPathObjectUrl( attachment, await this.persistence.persistUploadCopyFromSourcePath(attachment, attachment.filePath) ); return; } if (this.isPlayableMedia(attachment) && this.attachmentStorage.canWriteFiles()) { await this.applySavedPathObjectUrl(attachment, await this.persistence.saveFileToDisk(attachment, file)); } } async reannounceHostedAttachments(currentUserId: string | null | undefined): Promise { if (!currentUserId) { return; } for (const [, attachments] of this.runtimeStore.getAttachmentEntries()) { for (const attachment of attachments) { if (!canHostAttachment(attachment)) { continue; } const canServe = await this.attachmentStorage.resolveExistingPath(attachment); if (!canServe) { continue; } await this.announceLocalHost(attachment, currentUserId); } } } private releaseInMemoryUploadCopyIfPersisted(exactKey: string, attachment: Attachment): void { if (!attachment.savedPath?.trim() || attachment.size <= MAX_AUTO_SAVE_SIZE_BYTES) { return; } this.runtimeStore.deleteOriginalFile(exactKey); if (!attachment.objectUrl?.startsWith('blob:')) { return; } try { URL.revokeObjectURL(attachment.objectUrl); } catch { /* ignore */ } if (!this.isPlayableMedia(attachment)) { attachment.objectUrl = undefined; attachment.available = true; } } private async announceLocalHost(attachment: Attachment, hostPeerId?: string | null): Promise { if (!canHostAttachment(attachment)) { return; } const announcingPeerId = hostPeerId ?? await this.resolveCurrentUserId(); if (!announcingPeerId) { return; } this.runtimeStore.addAnnouncedHost( this.buildRequestKey(attachment.messageId, attachment.id), announcingPeerId ); const fileAnnounceEvent: FileAnnounceEvent = { type: 'file-announce', messageId: attachment.messageId, file: { id: attachment.id, filename: attachment.filename, size: attachment.size, mime: attachment.mime, isImage: attachment.isImage, uploaderPeerId: attachment.uploaderPeerId } }; this.webrtc.broadcastMessage(fileAnnounceEvent); } private async applySavedPathObjectUrl(attachment: Attachment, savedPath: string | null): Promise { if (!savedPath) { return; } const fileUrl = await this.attachmentStorage.getFileUrl(savedPath); if (fileUrl) { attachment.objectUrl = fileUrl; attachment.available = true; } } private isPlayableMedia(attachment: Pick): boolean { return attachment.mime.startsWith('video/') || attachment.mime.startsWith('audio/'); } private shouldReceiveToDisk(attachment: Attachment): boolean { return shouldStreamAttachmentReceiveToDisk(attachment, this.receiveCapabilities()); } private receiveCapabilities() { return { canStreamToDisk: this.attachmentStorage.canStreamToDisk(), canPersistSize: (bytes: number) => this.attachmentStorage.canPersistSize(bytes) }; } private receiveDiskChunk(attachment: Attachment, payload: ValidFileChunkPayload): void { const assemblyKey = `${payload.messageId}:${payload.fileId}`; const previous = this.diskReceiveLocks.get(assemblyKey) ?? Promise.resolve(); const next = previous .catch(() => undefined) .then(async () => { await this.handleDiskFileChunk(attachment, assemblyKey, payload); this.emitChunkAck(payload); }) .catch((error: unknown) => this.handleDiskReceiveFailure(attachment, assemblyKey, error)); this.diskReceiveLocks.set(assemblyKey, next); void next.finally(() => { if (this.diskReceiveLocks.get(assemblyKey) === next) { this.diskReceiveLocks.delete(assemblyKey); } }); } private emitChunkAck(payload: Pick): void { if (!payload.fromPeerId) { return; } const ack: FileChunkAckEvent = { type: 'file-chunk-ack', messageId: payload.messageId, fileId: payload.fileId, index: payload.index }; this.webrtc.sendToPeer(payload.fromPeerId, ack); } private async handleDiskFileChunk( attachment: Attachment, assemblyKey: string, payload: ValidFileChunkPayload ): Promise { const chunkByteLength = base64DecodedByteLength(payload.data); const chunkBytes = decodeBase64ToUint8Array(payload.data); const requestKey = this.buildRequestKey(payload.messageId, payload.fileId); this.runtimeStore.deletePendingRequest(requestKey); this.clearAttachmentRequestError(attachment); const assembly = await this.getOrCreateDiskReceiveAssembly(attachment, assemblyKey, payload.total); if (!assembly) { throw new Error(this.appI18n.instant(ATTACHMENT_PREPARE_DOWNLOAD_FAILED_KEY)); } if (assembly.receivedIndexes.has(payload.index)) { return; } if (payload.index !== assembly.receivedCount) { throw new Error(this.appI18n.instant(ATTACHMENT_CHUNKS_OUT_OF_ORDER_KEY)); } const didAppend = await this.attachmentStorage.appendBytes(assembly.path, chunkBytes); if (!didAppend) { throw new Error(this.appI18n.instant(ATTACHMENT_WRITE_DOWNLOAD_FAILED_KEY)); } assembly.receivedIndexes.add(payload.index); assembly.receivedCount += 1; this.updateTransferProgress(attachment, chunkByteLength, payload.fromPeerId); this.runtimeStore.touch(); if (assembly.receivedCount < assembly.total) { return; } attachment.savedPath = assembly.path; attachment.available = true; attachment.objectUrl = undefined; this.diskReceiveAssemblies.delete(assemblyKey); this.runtimeStore.touch(); void this.persistence.persistAttachmentMeta(attachment); void this.announceLocalHost(attachment); void this.hydratePlayableMediaAfterDiskReceive(attachment); } private async hydratePlayableMediaAfterDiskReceive(attachment: Attachment): Promise { if (!this.isPlayableMedia(attachment) || !attachment.savedPath) { return; } const nativeUrl = await this.attachmentStorage.getFileUrl(attachment.savedPath); if (nativeUrl) { attachment.objectUrl = nativeUrl; this.runtimeStore.touch(); return; } await this.persistence.ensureInlineDisplayObjectUrl(attachment); } private async getOrCreateDiskReceiveAssembly( attachment: Attachment, assemblyKey: string, total: number ): Promise { const existing = this.diskReceiveAssemblies.get(assemblyKey); if (existing) { return existing; } const storageContainer = await this.persistence.resolveStorageContainerName(attachment); const path = await this.attachmentStorage.createWritableFile(attachment, storageContainer); if (!path) { return null; } const assembly: DiskReceiveAssembly = { path, receivedCount: 0, receivedIndexes: new Set(), total }; this.diskReceiveAssemblies.set(assemblyKey, assembly); return assembly; } private async handleDiskReceiveFailure( attachment: Attachment, assemblyKey: string, error: unknown ): Promise { await this.deleteDiskReceiveAssembly(assemblyKey); attachment.available = false; attachment.objectUrl = undefined; attachment.receivedBytes = 0; attachment.speedBps = 0; attachment.startedAtMs = undefined; attachment.lastUpdateMs = undefined; attachment.requestError = error instanceof Error && error.message ? error.message : this.appI18n.instant(ATTACHMENT_DOWNLOAD_FAILED_KEY); this.runtimeStore.touch(); } private async deleteDiskReceiveAssembly(assemblyKey: string): Promise { const assembly = this.diskReceiveAssemblies.get(assemblyKey); this.diskReceiveAssemblies.delete(assemblyKey); if (assembly?.path) { await this.attachmentStorage.deleteFile(assembly.path); } } }