@ -42,10 +42,16 @@ public final class MessageSender {
// MARK: - C o n v e n i e n c e
// MARK: - C o n v e n i e n c e
public static func sendImmediate ( _ db : Database , message : Message , to destination : Message . Destination , interactionId : Int64 ? ) throws -> Promise < Void > {
public static func sendImmediate (
_ db : Database ,
message : Message ,
to destination : Message . Destination ,
interactionId : Int64 ? ,
isSyncMessage : Bool
) throws -> Promise < Void > {
switch destination {
switch destination {
case . contact , . closedGroup :
case . contact , . closedGroup :
return try sendToSnodeDestination ( db , message : message , to : destination , interactionId : interactionId )
return try sendToSnodeDestination ( db , message : message , to : destination , interactionId : interactionId , isSyncMessage : isSyncMessage )
case . openGroup :
case . openGroup :
return sendToOpenGroupDestination ( db , message : message , to : destination , interactionId : interactionId )
return sendToOpenGroupDestination ( db , message : message , to : destination , interactionId : interactionId )
@ -65,7 +71,7 @@ public final class MessageSender {
isSyncMessage : Bool = false
isSyncMessage : Bool = false
) throws -> Promise < Void > {
) throws -> Promise < Void > {
let ( promise , seal ) = Promise < Void > . pending ( )
let ( promise , seal ) = Promise < Void > . pending ( )
let : String = getUserHexEncodedPublicKey ( db )
let c urrentU serPublicKey: String = getUserHexEncodedPublicKey ( db )
let messageSendTimestamp : Int64 = SnodeAPI . currentOffsetTimestampMs ( )
let messageSendTimestamp : Int64 = SnodeAPI . currentOffsetTimestampMs ( )
// S e t t h e t i m e s t a m p , s e n d e r a n d r e c i p i e n t
// S e t t h e t i m e s t a m p , s e n d e r a n d r e c i p i e n t
@ -73,7 +79,7 @@ public final class MessageSender {
message . sentTimestamp ? ? // V i s i b l e m e s s a g e s w i l l a l r e a d y h a v e t h e i r s e n t t i m e s t a m p s e t
message . sentTimestamp ? ? // V i s i b l e m e s s a g e s w i l l a l r e a d y h a v e t h e i r s e n t t i m e s t a m p s e t
UInt64 ( messageSendTimestamp )
UInt64 ( messageSendTimestamp )
)
)
message . sender =
message . sender = c urrentU serPublicKey
message . recipient = {
message . recipient = {
switch destination {
switch destination {
case . contact ( let publicKey ) : return publicKey
case . contact ( let publicKey ) : return publicKey
@ -84,7 +90,7 @@ public final class MessageSender {
// S e t t h e f a i l u r e h a n d l e r ( n e e d i t h e r e a l r e a d y f o r p r e c o n d i t i o n f a i l u r e h a n d l i n g )
// S e t t h e f a i l u r e h a n d l e r ( n e e d i t h e r e a l r e a d y f o r p r e c o n d i t i o n f a i l u r e h a n d l i n g )
func handleFailure ( _ db : Database , with error : MessageSenderError ) {
func handleFailure ( _ db : Database , with error : MessageSenderError ) {
MessageSender . handleFailedMessageSend ( db , message : message , with : error , interactionId : interactionId
MessageSender . handleFailedMessageSend ( db , message : message , with : error , interactionId : interactionId , isSyncMessage : isSyncMessage )
seal . reject ( error )
seal . reject ( error )
}
}
@ -94,21 +100,8 @@ public final class MessageSender {
return promise
return promise
}
}
// S t o p h e r e i f t h i s i s a s e l f - s e n d , u n l e s s w e s h o u l d s y n c t h e m e s s a g e
let isSelfSend : Bool = ( message . recipient = = userPublicKey )
guard
! isSelfSend ||
isSyncMessage ||
Message . shouldSync ( message : message )
else {
try MessageSender . handleSuccessfulMessageSend ( db , message : message , to : destination , interactionId : interactionId )
seal . fulfill ( ( ) )
return promise
}
// A t t a c h t h e u s e r ' s p r o f i l e i f n e e d e d
// A t t a c h t h e u s e r ' s p r o f i l e i f n e e d e d
if var messageWithProfile : MessageWithProfile = message as ? MessageWithProfile {
if ! isSyncMessage , var messageWithProfile : MessageWithProfile = message as ? MessageWithProfile {
let profile : Profile = Profile . fetchOrCreateCurrentUser ( db )
let profile : Profile = Profile . fetchOrCreateCurrentUser ( db )
if let profileKey : Data = profile . profileEncryptionKey ? . keyData , let profilePictureUrl : String = profile . profilePictureUrl {
if let profileKey : Data = profile . profileEncryptionKey ? . keyData , let profilePictureUrl : String = profile . profilePictureUrl {
@ -123,6 +116,9 @@ public final class MessageSender {
}
}
}
}
// P e r f o r m a n y p r e - s e n d a c t i o n s
handleMessageWillSend ( db , message : message , interactionId : interactionId , isSyncMessage : isSyncMessage )
// C o n v e r t i t t o p r o t o b u f
// C o n v e r t i t t o p r o t o b u f
guard let proto = message . toProto ( db ) else {
guard let proto = message . toProto ( db ) else {
handleFailure ( db , with : . protoConversionFailed )
handleFailure ( db , with : . protoConversionFailed )
@ -233,6 +229,9 @@ public final class MessageSender {
)
)
let shouldNotify : Bool = {
let shouldNotify : Bool = {
// D o n ' t s e n d a n o t i f i c a t i o n w h e n s e n d i n g m e s s a g e s i n ' N o t e t o S e l f '
guard message . recipient != currentUserPublicKey else { return false }
switch message {
switch message {
case is VisibleMessage , is UnsendRequest : return ! isSyncMessage
case is VisibleMessage , is UnsendRequest : return ! isSyncMessage
case let callMessage as CallMessage :
case let callMessage as CallMessage :
@ -402,6 +401,9 @@ public final class MessageSender {
return promise
return promise
}
}
// P e r f o r m a n y p r e - s e n d a c t i o n s
handleMessageWillSend ( db , message : message , interactionId : interactionId )
// C o n v e r t i t t o p r o t o b u f
// C o n v e r t i t t o p r o t o b u f
guard let proto = message . toProto ( db ) else {
guard let proto = message . toProto ( db ) else {
handleFailure ( db , with : . protoConversionFailed )
handleFailure ( db , with : . protoConversionFailed )
@ -465,7 +467,7 @@ public final class MessageSender {
dependencies : SMKDependencies = SMKDependencies ( )
dependencies : SMKDependencies = SMKDependencies ( )
) -> Promise < Void > {
) -> Promise < Void > {
let ( promise , seal ) = Promise < Void > . pending ( )
let ( promise , seal ) = Promise < Void > . pending ( )
let : String = getUserHexEncodedPublicKey ( db , dependencies : dependencies )
let c urrentU serPublicKey: String = getUserHexEncodedPublicKey ( db , dependencies : dependencies )
guard case . openGroupInbox ( let server , let openGroupPublicKey , let recipientBlindedPublicKey ) = destination else {
guard case . openGroupInbox ( let server , let openGroupPublicKey , let recipientBlindedPublicKey ) = destination else {
preconditionFailure ( )
preconditionFailure ( )
@ -476,7 +478,7 @@ public final class MessageSender {
message . sentTimestamp = UInt64 ( SnodeAPI . currentOffsetTimestampMs ( ) )
message . sentTimestamp = UInt64 ( SnodeAPI . currentOffsetTimestampMs ( ) )
}
}
message . sender =
message . sender = c urrentU serPublicKey
message . recipient = recipientBlindedPublicKey
message . recipient = recipientBlindedPublicKey
// S e t t h e f a i l u r e h a n d l e r ( n e e d i t h e r e a l r e a d y f o r p r e c o n d i t i o n f a i l u r e h a n d l i n g )
// S e t t h e f a i l u r e h a n d l e r ( n e e d i t h e r e a l r e a d y f o r p r e c o n d i t i o n f a i l u r e h a n d l i n g )
@ -501,6 +503,9 @@ public final class MessageSender {
}
}
}
}
// P e r f o r m a n y p r e - s e n d a c t i o n s
handleMessageWillSend ( db , message : message , interactionId : interactionId )
// C o n v e r t i t t o p r o t o b u f
// C o n v e r t i t t o p r o t o b u f
guard let proto = message . toProto ( db ) else {
guard let proto = message . toProto ( db ) else {
handleFailure ( db , with : . protoConversionFailed )
handleFailure ( db , with : . protoConversionFailed )
@ -569,6 +574,32 @@ public final class MessageSender {
// MARK: S u c c e s s & F a i l u r e H a n d l i n g
// MARK: S u c c e s s & F a i l u r e H a n d l i n g
public static func handleMessageWillSend (
_ db : Database ,
message : Message ,
interactionId : Int64 ? ,
isSyncMessage : Bool = false
) {
// I f t h e m e s s a g e w a s a r e a c t i o n t h e n w e d o n ' t w a n t t o d o a n y t h i n g t o t h e o r i g i n a l
// i n t e r a c t i o n ( w h i c h t h e ' i n t e r a c t i o n I d ' i s p o i n t i n g t o
guard ( message as ? VisibleMessage ) ? . reaction = = nil else { return }
// M a r k m e s s a g e s a s " s e n d i n g " / " s y n c i n g " i f n e e d e d ( t h i s i s f o r r e t r i e s )
_ = try ? RecipientState
. filter ( RecipientState . Columns . interactionId = = interactionId )
. filter ( isSyncMessage ?
RecipientState . Columns . state = = RecipientState . State . failedToSync :
RecipientState . Columns . state = = RecipientState . State . failed
)
. updateAll (
db ,
RecipientState . Columns . state . set ( to : isSyncMessage ?
RecipientState . State . syncing :
RecipientState . State . sending
)
)
}
private static func handleSuccessfulMessageSend (
private static func handleSuccessfulMessageSend (
_ db : Database ,
_ db : Database ,
message : Message ,
message : Message ,
@ -578,7 +609,7 @@ public final class MessageSender {
isSyncMessage : Bool = false
isSyncMessage : Bool = false
) throws {
) throws {
// I f t h e m e s s a g e w a s a r e a c t i o n t h e n w e w a n t t o u p d a t e t h e r e a c t i o n i n s t e a d o f t h e o r i g i n a l
// I f t h e m e s s a g e w a s a r e a c t i o n t h e n w e w a n t t o u p d a t e t h e r e a c t i o n i n s t e a d o f t h e o r i g i n a l
// i n t e r a c it o n ( w h i c h t h e ' i n t e r a c t i o n I d ' i s p o i n t i n g t o
// i n t e r a c ti o n ( w h i c h t h e ' i n t e r a c t i o n I d ' i s p o i n t i n g t o
if let visibleMessage : VisibleMessage = message as ? VisibleMessage , let reaction : VisibleMessage . VMReaction = visibleMessage . reaction {
if let visibleMessage : VisibleMessage = message as ? VisibleMessage , let reaction : VisibleMessage . VMReaction = visibleMessage . reaction {
try Reaction
try Reaction
. filter ( Reaction . Columns . interactionId = = interactionId )
. filter ( Reaction . Columns . interactionId = = interactionId )
@ -624,18 +655,20 @@ public final class MessageSender {
}
}
}
}
let threadId : String = {
switch destination {
case . contact ( let publicKey ) : return publicKey
case . closedGroup ( let groupPublicKey ) : return groupPublicKey
case . openGroup ( let roomToken , let server , _ , _ , _ ) :
return OpenGroup . idFor ( roomToken : roomToken , server : server )
case . openGroupInbox ( _ , _ , let blindedPublicKey ) : return blindedPublicKey
}
} ( )
// P r e v e n t C o n t r o l M e s s a g e s f r o m b e i n g h a n d l e d m u l t i p l e t i m e s i f n o t s u p p o r t e d
// P r e v e n t C o n t r o l M e s s a g e s f r o m b e i n g h a n d l e d m u l t i p l e t i m e s i f n o t s u p p o r t e d
try ? ControlMessageProcessRecord (
try ? ControlMessageProcessRecord (
threadId : {
threadId : threadId ,
switch destination {
case . contact ( let publicKey ) : return publicKey
case . closedGroup ( let groupPublicKey ) : return groupPublicKey
case . openGroup ( let roomToken , let server , _ , _ , _ ) :
return OpenGroup . idFor ( roomToken : roomToken , server : server )
case . openGroupInbox ( _ , _ , let blindedPublicKey ) : return blindedPublicKey
}
} ( ) ,
message : message ,
message : message ,
serverExpirationTimestamp : (
serverExpirationTimestamp : (
( TimeInterval ( SnodeAPI . currentOffsetTimestampMs ( ) ) / 1000 ) +
( TimeInterval ( SnodeAPI . currentOffsetTimestampMs ( ) ) / 1000 ) +
@ -643,36 +676,27 @@ public final class MessageSender {
)
)
) ? . insert ( db )
) ? . insert ( db )
// S y n c t h e m e s s a g e i f :
// S y n c t h e m e s s a g e i f n e e d e d
// • i t ' s a v i s i b l e m e s s a g e o r a n e x p i r a t i o n t i m e r u p d a t e
scheduleSyncMessageIfNeeded (
// • t h e d e s t i n a t i o n w a s a c o n t a c t
db ,
// • w e d i d n ' t s y n c i t a l r e a d y
message : message ,
// • i t w a s n ' t s e t t o ' N o t e t o S e l f '
destination : destination ,
let userPublicKey = getUserHexEncodedPublicKey ( db )
threadId : threadId ,
if case . contact ( let publicKey ) = destination , ! isSyncMessage , publicKey != userPublicKey {
interactionId : interactionId ,
if let message = message as ? VisibleMessage { message . syncTarget = publicKey }
isAlreadySyncMessage : isSyncMessage
if let message = message as ? ExpirationTimerUpdate { message . syncTarget = publicKey }
)
// FIXME: M a k e t h i s a j o b
try sendToSnodeDestination (
db ,
message : message ,
to : . contact ( publicKey : userPublicKey ) ,
interactionId : interactionId ,
isSyncMessage : true
) . retainUntilComplete ( )
}
}
}
public static func handleFailedMessageSend (
public static func handleFailedMessageSend (
_ db : Database ,
_ db : Database ,
message : Message ,
message : Message ,
with error : MessageSenderError ,
with error : MessageSenderError ,
interactionId : Int64 ?
interactionId : Int64 ? ,
isSyncMessage : Bool = false
) {
) {
// TODO: R e v e r t t h e l o c a l d a t a b a s e c h a n g e
// TODO: R e v e r t t h e l o c a l d a t a b a s e c h a n g e
// I f t h e m e s s a g e w a s a r e a c t i o n t h e n w e d o n ' t w a n t t o d o a n y t h i n g t o t h e o r i g i n a l
// I f t h e m e s s a g e w a s a r e a c t i o n t h e n w e d o n ' t w a n t t o d o a n y t h i n g t o t h e o r i g i n a l
// i n t e r a c it o n ( w h i c h t h e ' i n t e r a c t i o n I d ' i s p o i n t i n g t o
// i n t e r a c ti o n ( w h i c h t h e ' i n t e r a c t i o n I d ' i s p o i n t i n g t o
guard ( message as ? VisibleMessage ) ? . reaction = = nil else { return }
guard ( message as ? VisibleMessage ) ? . reaction = = nil else { return }
// C h e c k i f w e n e e d t o m a r k a n y " s e n d i n g " r e c i p i e n t s a s " f a i l e d "
// C h e c k i f w e n e e d t o m a r k a n y " s e n d i n g " r e c i p i e n t s a s " f a i l e d "
@ -683,7 +707,12 @@ public final class MessageSender {
let rowIds : [ Int64 ] = ( try ? RecipientState
let rowIds : [ Int64 ] = ( try ? RecipientState
. select ( Column . rowID )
. select ( Column . rowID )
. filter ( RecipientState . Columns . interactionId = = interactionId )
. filter ( RecipientState . Columns . interactionId = = interactionId )
. filter ( RecipientState . Columns . state = = RecipientState . State . sending )
. filter ( ! isSyncMessage ?
RecipientState . Columns . state = = RecipientState . State . sending : (
RecipientState . Columns . state = = RecipientState . State . syncing ||
RecipientState . Columns . state = = RecipientState . State . sent
)
)
. asRequest ( of : Int64 . self )
. asRequest ( of : Int64 . self )
. fetchAll ( db ) )
. fetchAll ( db ) )
. defaulting ( to : [ ] )
. defaulting ( to : [ ] )
@ -698,7 +727,9 @@ public final class MessageSender {
. filter ( rowIds . contains ( Column . rowID ) )
. filter ( rowIds . contains ( Column . rowID ) )
. updateAll (
. updateAll (
db ,
db ,
RecipientState . Columns . state . set ( to : RecipientState . State . failed ) ,
RecipientState . Columns . state . set (
to : ( isSyncMessage ? RecipientState . State . failedToSync : RecipientState . State . failed )
) ,
RecipientState . Columns . mostRecentFailureText . set ( to : error . localizedDescription )
RecipientState . Columns . mostRecentFailureText . set ( to : error . localizedDescription )
)
)
}
}
@ -720,6 +751,43 @@ public final class MessageSender {
return nil
return nil
}
}
public static func scheduleSyncMessageIfNeeded (
_ db : Database ,
message : Message ,
destination : Message . Destination ,
threadId : String ? ,
interactionId : Int64 ? ,
isAlreadySyncMessage : Bool
) {
// S y n c t h e m e s s a g e i f i t ' s n o t a s y n c m e s s a g e , w a s n ' t a l r e a d y s e n t t o t h e c u r r e n t u s e r a n d
// i t ' s a m e s s a g e t y p e w h i c h s h o u l d b e s y n c e d
let currentUserPublicKey = getUserHexEncodedPublicKey ( db )
if
case . contact ( let publicKey ) = destination ,
! isAlreadySyncMessage ,
publicKey != currentUserPublicKey ,
Message . shouldSync ( message : message )
{
if let message = message as ? VisibleMessage { message . syncTarget = publicKey }
if let message = message as ? ExpirationTimerUpdate { message . syncTarget = publicKey }
JobRunner . add (
db ,
job : Job (
variant : . messageSend ,
threadId : threadId ,
interactionId : interactionId ,
details : MessageSendJob . Details (
destination : . contact ( publicKey : currentUserPublicKey ) ,
message : message ,
isSyncMessage : true
)
)
)
}
}
} }
// MARK: - O b j e c t i v e - C S u p p o r t // MARK: - O b j e c t i v e - C S u p p o r t
Morgan Pretty
2 years ago