implement meta delivered/received status for split-parts messages

Signed-off-by: xsbfh <z6MkpzZDoFM7tPZPXqvaVCtMFxKJQAxghVRJZxsf8s8SqBvm@dev.smashchats.com>

Author: xsbfh

TODO

@@ -20,7 +20,15 @@
 - [x] clearing another user from the neighborhood
 
 - [x] embedded media type
-- [x] message parts
+- [x] message parts & limit message queue size to keep websocket alive
+- [ ] emit sent and delivered status when all parts have been ack'd as such
+
+    - [ ] Parts feedback to Client to allow them to restore queue ??? **OR RATHER**, think through the state restore & queue recovery –> half-a, half-b, full
+    - [ ] add tests for concurrent messages re-assembling
+    - [ ] add tests for out of order parts
+    - [ ] test persistence and resilience (save/load state between restarts)
+
+- [ ] **Investigate INITCHATS: new Date() or new Date(0)–why?? FIX DOCS.**
 
 - [ ] livestream embedded media (snap) (before sharing decryption key at last, then reconstitute) – PARTS IN PARTS??? or binary only for content???
 
@@ -62,6 +70,10 @@
 - [ ] investigate performance and memory usage (eg, dangling sessions, dlqs, parts, no redundant compute)
 - [ ] Use UTF8 vs Base64 for raw content encoding??
 
+WARNINGS/DISCLAIMERS
+
+[Warning] AES-CBC and AES-CTR do not provide authentication by default, and implementing it manually can result in minor, but serious mistakes. We recommended using authenticated encryption like AES-GCM to protect against chosen-ciphertext attacks. (2key-ratchet.js, line 6011)
+
 ### TESTS REFACTORING (keeping or increasing coverage):
 
 - [ ] generate unit .test files with mocks/spies for each module corresponding to the library API reference (**testing only publicly exported**) --> should have 100% coverage for publicly exposed APIs (public and internal)

src/api/SmashMessaging.ts

@@ -7,6 +7,7 @@ import { PeerRegistry } from '@src/core/messaging/peer/PeerRegistry.js';
 import { SmashPeer } from '@src/core/messaging/peer/SmashPeer.js';
 import { MessageMiddleware } from '@src/core/messaging/protocol/MessageMiddleware.js';
 import { MessageReassembler } from '@src/core/messaging/protocol/MessageReassembler.js';
+import { MessageSplitter } from '@src/core/messaging/protocol/MessageSplitter.js';
 import { SessionManager } from '@src/core/messaging/session/SessionManager.js';
 import { EndpointManager } from '@src/infrastructure/network/endpoints/EndpointManager.js';
 import { SMESocketManager } from '@src/infrastructure/network/sme/SMESocketManager.js';
@@ -28,6 +29,7 @@ import type {
     IDIDResolver,
     IIMPeerIdentity,
     IMProtoMessage,
+    IMReceivedACKMessage,
     reverseDNS,
     sha256,
 } from '@src/shared/types/index.js';
@@ -36,6 +38,9 @@ import {
     IMProfile,
     IMReadACKMessage,
     MessageStatus,
+    MessageStatusDelivered,
+    MessageStatusRead,
+    MessageStatusReceived,
 } from '@src/shared/types/messages/index.js';
 import { LogLevel, Logger } from '@src/shared/utils/Logger.js';
 import { EventEmitter } from 'events';
@@ -50,6 +55,7 @@ export class SmashMessaging extends EventEmitter {
     protected readonly peers: PeerRegistry;
     private readonly sessionManager: SessionManager;
     private readonly smeSocketManager: SMESocketManager;
+    private readonly messageSplitter: MessageSplitter;
     private readonly messageReassembler: MessageReassembler;
     private meta: Partial<IMProfile> = {};
 
@@ -66,9 +72,11 @@ export class SmashMessaging extends EventEmitter {
     ) {
         super();
         this.logger = new Logger(logId, logLevel);
+        const messageStatusHandler = this.messagesStatusHandler.bind(this);
+
         this.smeSocketManager = new SMESocketManager(
             this.logger,
-            this.messagesStatusHandler.bind(this),
+            messageStatusHandler,
         );
 
         this.peers = new PeerRegistry(
@@ -96,6 +104,10 @@ export class SmashMessaging extends EventEmitter {
         );
 
         this.messageReassembler = new MessageReassembler(this.logger);
+        this.messageSplitter = new MessageSplitter(
+            this.logger,
+            messageStatusHandler,
+        );
 
         this.registerEventHandlers();
         this.logger.info(
@@ -194,6 +206,7 @@ export class SmashMessaging extends EventEmitter {
             lastMessageTime,
             this.sessionManager,
             this.smeSocketManager,
+            this.messageSplitter,
             this.peers,
         );
     }
@@ -252,6 +265,12 @@ export class SmashMessaging extends EventEmitter {
                         `Successfully reassembled message ${reassembledMessage.sha256} from parts`,
                     );
                     await this.notifyNewMessage(sender, reassembledMessage);
+                    (await this.peers.getOrCreate(sender)).send(
+                        await encapsulateMessage({
+                            type: IM_ACK_RECEIVED,
+                            data: [reassembledMessage.sha256],
+                        } as IMReceivedACKMessage),
+                    );
                 } else {
                     this.logger.debug(
                         `Waiting for more parts for message ${(message.data as IMPartData).originalSha256}`,
@@ -284,11 +303,11 @@ export class SmashMessaging extends EventEmitter {
         };
 
         this.on(IM_ACK_RECEIVED, (from, message) => {
-            handleAck('received', from, message.data);
+            handleAck(MessageStatusReceived, from, message.data);
         });
 
         this.on(IM_ACK_READ, (from, message) => {
-            handleAck('read', from, message.data);
+            handleAck(MessageStatusRead, from, message.data);
         });
     }
 
@@ -300,6 +319,9 @@ export class SmashMessaging extends EventEmitter {
             `messagesStatusHandler ACK:${status} : ${messageIds.join(', ')}`,
         );
         this.emit('status', status, messageIds);
+        if (status === MessageStatusDelivered) {
+            this.messageSplitter.markAsDelivered(messageIds);
+        }
     }
 
     /**

src/core/messaging/peer/SmashPeer.ts

@@ -54,6 +54,7 @@ export class SmashPeer {
         private readonly lastMessageTime: number,
         private readonly sessionManager: SessionManager,
         private readonly smeSocketManager: SMESocketManager,
+        private readonly messageSplitter: MessageSplitter,
         private readonly peerRegistry: PeerRegistry,
         config: PeerConfig = {},
     ) {
@@ -164,7 +165,7 @@ export class SmashPeer {
         const messageSize = MessageSplitter.getMessageSize(encapsulatedMessage);
         const queue =
             messageSize > MAX_MESSAGE_SIZE
-                ? await MessageSplitter.split(this.logger, encapsulatedMessage)
+                ? await this.messageSplitter.split(encapsulatedMessage)
                 : [encapsulatedMessage];
         return this.mutex.acquire(`flushQueue-${this.id}`, async () => {
             for (let i = 0; i < queue.length; i++) {

src/core/messaging/protocol/MessageSplitter.ts

@@ -1,20 +1,36 @@
 import { encapsulateMessage } from '@src/api/tools/encapsulateMessage.js';
 import { MAX_MESSAGE_SIZE } from '@src/shared/constants/protocol.js';
 import { IM_PART } from '@src/shared/lexicon/improto.lexicon.js';
+import {
+    MessageStatusDelivered,
+    onMessagesStatusFn,
+} from '@src/shared/types/index.js';
 import type {
     EncapsulatedIMProtoMessage,
     IMProtoMessage,
 } from '@src/shared/types/message.types.js';
 import type { IMPartData } from '@src/shared/types/messages/IMPartMessage.js';
+import { sha256 } from '@src/shared/types/string.types.js';
 import { Logger } from '@src/shared/utils/Logger.js';
 
+interface PartsCounter {
+    originalSha256: sha256;
+    counter: number;
+}
+
 export class MessageSplitter {
+    private readonly sending: Map<sha256, PartsCounter> = new Map();
+
+    constructor(
+        private readonly logger: Logger,
+        private readonly onMessagesStatusCallback: onMessagesStatusFn,
+    ) {}
+
     /**
      * Split a message into parts.
      * Each part will contain a portion of the original message data.
      */
-    public static async split(
-        logger: Logger,
+    public async split(
         message: EncapsulatedIMProtoMessage,
     ): Promise<EncapsulatedIMProtoMessage[]> {
         if (!message?.sha256) {
@@ -24,7 +40,7 @@ export class MessageSplitter {
         const messageJson = JSON.stringify(message);
         const totalParts = Math.ceil(messageJson.length / MAX_MESSAGE_SIZE);
 
-        logger.info(
+        this.logger.info(
             `Splitting message (${messageJson.length} bytes) into ${totalParts} parts`,
         );
 
@@ -43,7 +59,7 @@ export class MessageSplitter {
                 content: partContent, // No need for base64 encoding
             };
 
-            logger.debug(
+            this.logger.debug(
                 `Created part ${i + 1}/${totalParts} (${partContent.length} bytes)`,
             );
 
@@ -55,6 +71,11 @@ export class MessageSplitter {
                 }),
             );
         }
+        const counter = {
+            counter: parts.length,
+            originalSha256: message.sha256,
+        } as PartsCounter;
+        parts.forEach((part) => this.sending.set(part.sha256, counter));
         return parts;
     }
 
@@ -64,4 +85,32 @@ export class MessageSplitter {
     public static getMessageSize(message: IMProtoMessage): number {
         return JSON.stringify(message).length;
     }
+
+    /**
+     * ACK potential parts as being delivered
+     * @param parts - Array of message parts to be marked as delivered.
+     */
+    public markAsDelivered(parts: sha256[]) {
+        if (this.sending.size === 0) return;
+        this.logger.debug(
+            `Trying to mark ${parts.length} potential parts as delivered`,
+        );
+        parts.forEach((part) => this.markPartAsDelivered(part));
+    }
+
+    private markPartAsDelivered(part: sha256) {
+        const counter = this.sending.get(part);
+        if (counter) {
+            this.logger.debug(`Marking part ${part} as delivered`);
+            if (--counter.counter === 0) {
+                this.sending.delete(part);
+                this.logger.debug(
+                    `All parts have been delivered for ${counter.originalSha256}`,
+                );
+                this.onMessagesStatusCallback(MessageStatusDelivered, [
+                    counter.originalSha256,
+                ]);
+            }
+        }
+    }
 }

src/infrastructure/network/sme/SMESocketWriteOnly.ts

@@ -1,4 +1,5 @@
 import type { onMessagesStatusFn } from '@src/shared/types/callbacks.types.js';
+import { MessageStatusDelivered } from '@src/shared/types/index.js';
 import type { sha256 } from '@src/shared/types/string.types.js';
 import type { Logger } from '@src/shared/utils/Logger.js';
 import { Timeout } from '@src/shared/utils/Timeout.js';
@@ -148,7 +149,10 @@ export class SMESocketWriteOnly {
                 this.logger.debug(
                     `> ${buffer.byteLength} bytes (${messageIds.length} messages) sent to ${this.url} [${this.socket?.id}]`,
                 );
-                this.onMessagesStatusCallback('delivered', messageIds);
+                this.onMessagesStatusCallback(
+                    MessageStatusDelivered,
+                    messageIds,
+                );
                 resolve();
             });
         });

src/shared/types/messages/IMAckMessage.ts

@@ -20,12 +20,18 @@ export interface IMReceivedACKMessage extends IMACKMessage {
     type: typeof IM_ACK_RECEIVED;
 }
 
+export const MessageStatusSending = 'sending' as const;
+export const MessageStatusDelivered = 'delivered' as const;
+export const MessageStatusReceived = 'received' as const;
+export const MessageStatusRead = 'read' as const;
+export const MessageStatusError = 'error' as const;
+
 export type MessageStatus =
-    | 'sending'
-    | 'delivered'
-    | 'received'
-    | 'read'
-    | 'error';
+    | typeof MessageStatusSending
+    | typeof MessageStatusDelivered
+    | typeof MessageStatusReceived
+    | typeof MessageStatusRead
+    | typeof MessageStatusError;
 // TODO: double check status lifecycle implementation accross lib
 // in particular, error states and (re)sending states
 

tests/how-to/edgecases.spec.ts

@@ -13,6 +13,7 @@ import {
     IMTextMessage,
     IM_CHAT_TEXT,
     Logger,
+    MessageStatusDelivered,
     SESSION_TTL_MS,
     SmashMessaging,
     sha256,
@@ -118,7 +119,7 @@ describe('SmashMessaging: Edge cases', () => {
                 logger.info('>> Verify Alice received the ACK');
                 await delay(TEST_CONFIG.MESSAGE_DELIVERY);
                 expect(alice.onStatus).toHaveBeenCalledWith(
-                    'delivered',
+                    MessageStatusDelivered,
                     expect.arrayContaining([sent.sha256]),
                 );
 

tests/how-to/index.spec.ts

@@ -9,6 +9,7 @@ import {
     IM_CHAT_TEXT,
     IM_PROFILE,
     Logger,
+    MessageStatusDelivered,
     SmashMessaging,
 } from 'smash-node-lib';
 
@@ -139,7 +140,7 @@ describe('[SmashMessaging] Between peers registered to a SME', () => {
                 await delay(TEST_CONFIG.DEFAULT_SETUP_DELAY);
 
                 expect(alice.onStatus).toHaveBeenCalledWith(
-                    'delivered',
+                    MessageStatusDelivered,
                     expect.arrayContaining([TEST_MESSAGE.sha256]),
                 );
             },

tests/how-to/parts.spec.ts

@@ -83,11 +83,16 @@ describe('Message Splitting', () => {
             const onBobMessage = jest.fn();
             bob.messaging.on(IM_CHAT_TEXT, onBobMessage);
 
+            const onAliceStatus = jest.fn();
+            alice.messaging.on('status', onAliceStatus);
+
             const bobReceivedMessage = waitFor(bob.messaging, IM_CHAT_TEXT);
 
+            // Alice sends an oversized message to Bob
             const sent = await alice.messaging.send(bob.did, message);
             await bobReceivedMessage;
 
+            // Bob receives the original message and it is correctly handled
             expect(onBobMessage).toHaveBeenCalledWith(
                 alice.did.id,
                 expect.objectContaining<IMProtoMessage>({
@@ -98,6 +103,20 @@ describe('Message Splitting', () => {
                     after: sent.after,
                 }),
             );
+
+            await delay(TEST_CONFIG.MESSAGE_DELIVERY);
+
+            // Alice receives a delivered ACK for the original message
+            expect(onAliceStatus).toHaveBeenCalledWith(
+                'delivered',
+                expect.arrayContaining([sent.sha256]),
+            );
+
+            // Alice receives a received ACK for the original message
+            expect(onAliceStatus).toHaveBeenCalledWith(
+                'received',
+                expect.arrayContaining([sent.sha256]),
+            );
         });
 
         test('Sending a large media message', async () => {