@ -26,7 +26,6 @@ import { SnodeAPIStore } from '../apis/snode_api/storeMessage';
import { getConversationController } from '../conversations' ;
import { MessageEncrypter } from '../crypto' ;
import { addMessagePadding } from '../crypto/BufferPadding' ;
import { DisappearingMessages } from '../disappearing_messages' ;
import { ContentMessage } from '../messages/outgoing' ;
import { ConfigurationMessage } from '../messages/outgoing/controlMessage/ConfigurationMessage' ;
import { SharedConfigMessage } from '../messages/outgoing/controlMessage/SharedConfigMessage' ;
@ -36,6 +35,7 @@ import { OpenGroupVisibleMessage } from '../messages/outgoing/visibleMessage/Ope
import { ed25519Str } from '../onions/onionPath' ;
import { PubKey } from '../types' ;
import { RawMessage } from '../types/RawMessage' ;
import { UserUtils } from '../utils' ;
import { fromUInt8ArrayToBase64 } from '../utils/String' ;
import { EmptySwarmError } from '../utils/errors' ;
@ -76,7 +76,7 @@ function getMinRetryTimeout() {
return 1000 ;
}
function is SyncMessage( message : ContentMessage ) {
function is Content SyncMessage( message : ContentMessage ) {
if (
message instanceof ConfigurationMessage ||
message instanceof ClosedGroupNewMessage ||
@ -95,16 +95,20 @@ function isSyncMessage(message: ContentMessage) {
* @param message The message to send .
* @param attempts The amount of times to attempt sending . Minimum value is 1 .
* /
async function send (
message : RawMessage ,
attempts : number = 3 ,
retryMinTimeout? : number , // in ms
isASyncMessage? : boolean
) : Promise < { wrappedEnvelope : Uint8Array ; effectiveTimestamp : number } > {
async function send ( {
message ,
retryMinTimeout = 100 ,
attempts = 3 ,
isSyncMessage ,
} : {
message : RawMessage ;
attempts? : number ;
retryMinTimeout? : number ; // in ms
isSyncMessage : boolean ;
} ) : Promise < { wrappedEnvelope : Uint8Array ; effectiveTimestamp : number } > {
return pRetry (
async ( ) = > {
const recipient = PubKey . cast ( message . device ) ;
const { ttl } = message ;
// we can only have a single message in this send function for now
const [ encryptedAndWrapped ] = await encryptMessagesAndWrap ( [
@ -112,9 +116,9 @@ async function send(
destination : message.device ,
plainTextBuffer : message.plainTextBuffer ,
namespace : message . namespace ,
ttl ,
ttl : message.ttl ,
identifier : message.identifier ,
isSyncMessage : Boolean ( is A SyncMessage) ,
isSyncMessage : Boolean ( is SyncMessage) ,
} ,
] ) ;
@ -127,13 +131,30 @@ async function send(
found . set ( { sent_at : encryptedAndWrapped.networkTimestamp } ) ;
await found . commit ( ) ;
}
let foundMessage = encryptedAndWrapped . identifier
? await Data . getMessageById ( encryptedAndWrapped . identifier )
: null ;
const isSyncedDeleteAfterReadMessage =
found &&
UserUtils . isUsFromCache ( recipient . key ) &&
found . getExpirationType ( ) === 'deleteAfterRead' &&
found . getExpireTimerSeconds ( ) > 0 &&
encryptedAndWrapped . isSyncMessage ;
let overridenTtl = encryptedAndWrapped . ttl ;
if ( isSyncedDeleteAfterReadMessage && found . getExpireTimerSeconds ( ) > 0 ) {
const asMs = found . getExpireTimerSeconds ( ) * 1000 ;
window . log . debug ( ` overriding ttl for synced DaR message to ${ asMs } ` ) ;
overridenTtl = asMs ;
}
const batchResult = await MessageSender . sendMessagesDataToSnode (
[
{
pubkey : recipient.key ,
data64 : encryptedAndWrapped.data64 ,
ttl ,
ttl : overridenTtl ,
timestamp : encryptedAndWrapped.networkTimestamp ,
namespace : encryptedAndWrapped . namespace ,
} ,
@ -158,44 +179,19 @@ async function send(
) {
// TODO: the expiration is due to be returned by the storage server on "store" soon, we will then be able to use it instead of doing the storedAt + ttl logic below
// if we have a hash and a storedAt, mark it as seen so we don't reprocess it on the next retrieve
await Data . saveSeenMessageHashes ( [ { expiresAt : storedAt + ttl , hash : storedHash } ] ) ;
await Data . saveSeenMessageHashes ( [
{ expiresAt : storedAt + encryptedAndWrapped . ttl , hash : storedHash } ,
] ) ;
// If message also has a sync message, save that hash. Otherwise save the hash from the regular message send i.e. only closed groups in this case.
if (
encryptedAndWrapped . identifier &&
( encryptedAndWrapped . isSyncMessage || isDestinationClosedGroup )
) {
const foundMessage = await Data . getMessageById ( encryptedAndWrapped . identifier ) ;
// get a fresh copy of the message from the DB
foundMessage = await Data . getMessageById ( encryptedAndWrapped . identifier ) ;
if ( foundMessage ) {
await foundMessage . updateMessageHash ( storedHash ) ;
const convo = foundMessage . getConversation ( ) ;
const expireTimer = foundMessage . getExpireTimer ( ) ;
const expirationType = foundMessage . getExpirationType ( ) ;
if (
convo &&
expirationType &&
expireTimer > 0 &&
// a message has started to disappear
foundMessage . getExpirationStartTimestamp ( )
) {
const expirationMode = DisappearingMessages . changeToDisappearingConversationMode (
convo ,
expirationType ,
expireTimer
) ;
const canBeDeleteAfterRead = ! convo . isMe ( ) && convo . isPrivate ( ) ;
// TODO legacy messages support will be removed in a future release
if (
canBeDeleteAfterRead &&
( expirationMode === 'legacy' || expirationMode === 'deleteAfterRead' )
) {
await DisappearingMessages . updateMessageExpiriesOnSwarm ( [ foundMessage ] ) ;
}
}
await foundMessage . commit ( ) ;
}
}
@ -399,7 +395,7 @@ async function sendMessagesToSnode(
namespace : m . namespace ,
ttl : m.message.ttl ( ) ,
identifier : m.message.identifier ,
isSyncMessage : MessageSender.is SyncMessage( m . message ) ,
isSyncMessage : MessageSender.is Content SyncMessage( m . message ) ,
} ) )
) ;
@ -581,5 +577,5 @@ export const MessageSender = {
getMinRetryTimeout ,
sendToOpenGroupV2 ,
send ,
is SyncMessage,
is Content SyncMessage,
} ;
Audric Ackermann
1 year ago