@ -39,8 +39,7 @@ public enum MessageSendJob: JobExecutor {
// / a l r e a d y h a v e a t t a c h m e n t s i n a v a l i d s t a t e
// / a l r e a d y h a v e a t t a c h m e n t s i n a v a l i d s t a t e
if
if
details . message is VisibleMessage ,
details . message is VisibleMessage ,
( details . message as ? VisibleMessage ) ? . reaction = = nil &&
( details . message as ? VisibleMessage ) ? . reaction = = nil
details . isSyncMessage = = false
{
{
guard
guard
let jobId : Int64 = job . id ,
let jobId : Int64 = job . id ,
@ -51,122 +50,111 @@ public enum MessageSendJob: JobExecutor {
return
return
}
}
// I f t h e o r i g i n a l i n t e r a c t i o n n o l o n g e r e x i s t s t h e n d o n ' t b o t h e r s e n d i n g t h e m e s s a g e ( i e . t h e
// R e t r i e v e t h e c u r r e n t a t t a c h m e n t s t a t e
// m e s s a g e w a s d e l e t e d b e f o r e i t e v e n g o t s e n t )
typealias AttachmentState = ( error : Error ? , pendingUploadAttachmentIds : [ String ] , preparedFileIds : [ String ] )
guard Storage . shared . read ( { db in try Interaction . exists ( db , id : interactionId ) } ) = = true else {
SNLog ( " [MessageSendJob] Failing due to missing interaction " )
failure ( job , StorageError . objectNotFound , true )
return
}
// C h e c k i f t h e r e a r e a n y a t t a c h m e n t s a s s o c i a t e d t o t h i s m e s s a g e , a n d i f s o
// u p l o a d t h e m n o w
//
// N o t e : N o r m a l a t t a c h m e n t s s h o u l d b e s e n t i n a n o n - d u r a b l e w a y b u t a n y
// a t t a c h m e n t s f o r L i n k P r e v i e w s a n d Q u o t e s w i l l b e p r o c e s s e d t h r o u g h t h i s m e c h a n i s m
let attachmentState : ( shouldFail : Bool , shouldDefer : Bool , fileIds : [ String ] ) ? = Storage . shared . write { db in
let allAttachmentStateInfo : [ Attachment . StateInfo ] = try Attachment
. stateInfo ( interactionId : interactionId )
. fetchAll ( db )
let maybeFileIds : [ String ? ] = allAttachmentStateInfo
. sorted { lhs , rhs in lhs . albumIndex < rhs . albumIndex }
. map { Attachment . fileId ( for : $0 . downloadUrl ) }
let fileIds : [ String ] = maybeFileIds . compactMap { $0 }
// I f t h e r e w e r e f a i l e d a t t a c h m e n t s t h e n t h i s j o b s h o u l d f a i l ( c a n ' t s e n d a
// m e s s a g e w h i c h h a s a s s o c i a t e d a t t a c h m e n t s i f t h e a t t a c h m e n t s f a i l t o u p l o a d )
guard ! allAttachmentStateInfo . contains ( where : { $0 . state = = . failedDownload } ) else {
return ( true , false , fileIds )
}
// C r e a t e j o b s f o r a n y p e n d i n g ( o r f a i l e d ) a t t a c h m e n t j o b s a n d i n s e r t t h e m i n t o t h e
let attachmentState : AttachmentState = Storage . shared
// q u e u e b e f o r e t h e c u r r e n t j o b ( t h i s w i l l m e a n t h e c u r r e n t j o b w i l l r e - r u n
. read { db in
// a f t e r t h e s e i n s e r t e d j o b s c o m p l e t e )
// I f t h e o r i g i n a l i n t e r a c t i o n n o l o n g e r e x i s t s t h e n d o n ' t b o t h e r s e n d i n g t h e m e s s a g e ( i e . t h e
//
// m e s s a g e w a s d e l e t e d b e f o r e i t e v e n g o t s e n t )
// N o t e : I f t h e r e a r e a n y ' d o w n l o a d e d ' a t t a c h m e n t s t h e n t h e y a l s o n e e d t o b e
guard try Interaction . exists ( db , id : interactionId ) else {
// u p l o a d e d ( a s a ' d o w n l o a d e d ' a t t a c h m e n t w i l l b e o n t h e c u r r e n t u s e r s d e v i c e
SNLog ( " [MessageSendJob] Failing due to missing interaction " )
// b u t n o t o n t h e m e s s a g e r e c i p i e n t s d e v i c e - b o t h L i n k P r e v i e w a n d Q u o t e c a n
return ( StorageError . objectNotFound , [ ] , [ ] )
// h a v e t h i s c a s e )
try allAttachmentStateInfo
. filter { attachment -> Bool in
// N o n - m e d i a q u o t e s w o n ' t h a v e t h u m b n a i l s s o s o d o n ' t t r y t o u p l o a d t h e m
guard attachment . downloadUrl != Attachment . nonMediaQuoteFileId else { return false }
switch attachment . state {
case . uploading , . pendingDownload , . downloading , . failedUpload , . downloaded :
return true
default : return false
}
}
. filter { stateInfo in
// D o n ' t a d d a n e w j o b i f t h e r e i s o n e a l r e a d y i n t h e q u e u e
! JobRunner . hasPendingOrRunningJob (
with : . attachmentUpload ,
details : AttachmentUploadJob . Details (
messageSendJobId : jobId ,
attachmentId : stateInfo . attachmentId
)
)
}
. compactMap { stateInfo -> ( jobId : Int64 , job : Job ) ? in
JobRunner
. insert (
db ,
job : Job (
variant : . attachmentUpload ,
behaviour : . runOnce ,
threadId : job . threadId ,
interactionId : interactionId ,
details : AttachmentUploadJob . Details (
messageSendJobId : jobId ,
attachmentId : stateInfo . attachmentId
)
) ,
before : job
)
}
}
. forEach { otherJobId , _ in
// C r e a t e t h e d e p e n d e n c y b e t w e e n t h e j o b s
// G e t t h e c u r r e n t s t a t e o f t h e a t t a c h m e n t s
try JobDependencies (
let allAttachmentStateInfo : [ Attachment . StateInfo ] = try Attachment
jobId : jobId ,
. stateInfo ( interactionId : interactionId )
dependantId : otherJobId
. fetchAll ( db )
)
let maybeFileIds : [ String ? ] = allAttachmentStateInfo
. insert ( db )
. sorted { lhs , rhs in lhs . albumIndex < rhs . albumIndex }
. map { Attachment . fileId ( for : $0 . downloadUrl ) }
let fileIds : [ String ] = maybeFileIds . compactMap { $0 }
// I f t h e r e w e r e f a i l e d a t t a c h m e n t s t h e n t h i s j o b s h o u l d f a i l ( c a n ' t s e n d a
// m e s s a g e w h i c h h a s a s s o c i a t e d a t t a c h m e n t s i f t h e a t t a c h m e n t s f a i l t o u p l o a d )
guard ! allAttachmentStateInfo . contains ( where : { $0 . state = = . failedDownload } ) else {
SNLog ( " [MessageSendJob] Failing due to failed attachment upload " )
return ( AttachmentError . notUploaded , [ ] , fileIds )
}
}
// I f t h e r e w e r e p e n d i n g o r u p l o a d i n g a t t a c h m e n t s t h e n s t o p h e r e ( w e w a n t t o
// / F i n d a l l a t t a c h m e n t I d s f o r a t t a c h m e n t s w h i c h n e e d t o b e u p l o a d e d
// u p l o a d t h e m f i r s t a n d t h e n r e - r u n t h i s s e n d j o b - t h e ' J o b R u n n e r . i n s e r t '
// /
// m e t h o d w i l l t a k e c a r e o f t h i s )
// / * * N o t e : * * I f t h e r e a r e a n y ' d o w n l o a d e d ' a t t a c h m e n t s t h e n t h e y a l s o n e e d t o b e u p l o a d e d ( a s a
let isMissingFileIds : Bool = ( maybeFileIds . count != fileIds . count )
// / ' d o w n l o a d e d ' a t t a c h m e n t w i l l b e o n t h e c u r r e n t u s e r s d e v i c e b u t n o t o n t h e m e s s a g e r e c i p i e n t s
let hasPendingUploads : Bool = allAttachmentStateInfo . contains ( where : { $0 . state != . uploaded } )
// / d e v i c e - b o t h ` L i n k P r e v i e w ` a n d ` Q u o t e ` c a n h a v e t h i s c a s e )
let pendingUploadAttachmentIds : [ String ] = allAttachmentStateInfo
. filter { attachment -> Bool in
// N o n - m e d i a q u o t e s w o n ' t h a v e t h u m b n a i l s s o s o d o n ' t t r y t o u p l o a d t h e m
guard attachment . downloadUrl != Attachment . nonMediaQuoteFileId else { return false }
switch attachment . state {
case . uploading , . pendingDownload , . downloading , . failedUpload , . downloaded :
return true
default : return false
}
}
. map { $0 . attachmentId }
return ( nil , pendingUploadAttachmentIds , fileIds )
}
. defaulting ( to : ( MessageSenderError . invalidMessage , [ ] , [ ] ) )
return (
// / I f w e g o t a n e r r o r w h e n t r y i n g t o r e t r i e v e t h e a t t a c h m e n t s t a t e t h e n t h i s j o b i s a c t u a l l y i n v a l i d s o i t
( isMissingFileIds && ! hasPendingUploads ) ,
// / s h o u l d p e r m a n e n t l y f a i l
hasPendingUploads ,
guard attachmentState . error = = nil else {
fileIds
return failure ( job , ( attachmentState . error ? ? MessageSenderError . invalidMessage ) , true )
)
}
}
// D o n ' t s e n d m e s s a g e s w i t h f a i l e d a t t a c h m e n t u p l o a d s
// / I f w e h a v e a n y p e n d i n g ( o r f a i l e d ) a t t a c h m e n t u p l o a d s t h e n w e s h o u l d c r e a t e j o b s f o r t h e m a n d i n s e r t t h e m i n t o t h e
//
// / q u e u e b e f o r e t h e c u r r e n t j o b a n d d e f e r i t ( t h i s w i l l m e a n t h e c u r r e n t j o b w i l l r e - r u n a f t e r t h e s e i n s e r t e d j o b s c o m p l e t e )
// N o t e : I f w e h a v e g o t t e n t o t h i s p o i n t t h e n a n y d e p e n d a n t a t t a c h m e n t u p l o a d
guard attachmentState . pendingUploadAttachmentIds . isEmpty else {
// j o b s w i l l h a v e p e r m a n e n t l y f a i l e d s o t h i s m e s s a g e s e n d s h o u l d a l s o d o s o
Storage . shared . write { db in
guard attachmentState ? . shouldFail = = false else {
try attachmentState . pendingUploadAttachmentIds
SNLog ( " [MessageSendJob] Failing due to failed attachment upload " )
. filter { attachmentId in
failure ( job , AttachmentError . notUploaded , true )
// D o n ' t a d d a n e w j o b i f t h e r e i s o n e a l r e a d y i n t h e q u e u e
return
! JobRunner . hasPendingOrRunningJob (
}
with : . attachmentUpload ,
details : AttachmentUploadJob . Details (
messageSendJobId : jobId ,
attachmentId : attachmentId
)
)
}
. compactMap { attachmentId -> ( jobId : Int64 , job : Job ) ? in
JobRunner
. insert (
db ,
job : Job (
variant : . attachmentUpload ,
behaviour : . runOnce ,
threadId : job . threadId ,
interactionId : interactionId ,
details : AttachmentUploadJob . Details (
messageSendJobId : jobId ,
attachmentId : attachmentId
)
) ,
before : job
)
}
. forEach { otherJobId , _ in
// C r e a t e t h e d e p e n d e n c y b e t w e e n t h e j o b s
try JobDependencies (
jobId : jobId ,
dependantId : otherJobId
)
. insert ( db )
}
}
// D e f e r t h e j o b i f w e f o u n d i n c o m p l e t e u p l o a d s
SNLog ( " [MessageSendJob] Deferring due to pending attachment uploads " )
guard attachmentState ? . shouldDefer = = false else {
return deferred ( job )
SNLog ( " [MessageSendJob] Deferring pending attachment uploads " )
deferred ( job )
return
}
}
// S t o r e t h e f i l e I d s s o t h e y c a n b e s e n t w i t h t h e o p e n g r o u p m e s s a g e c o n t e n t
// S t o r e t h e f i l e I d s s o t h e y c a n b e s e n t w i t h t h e o p e n g r o u p m e s s a g e c o n t e n t
messageFileIds = ( attachmentState ? . fileIds ? ? [ ] )
messageFileIds = attachmentState . preparedFileIds
}
}
// S t o r e t h e s e n t T i m e s t a m p f r o m t h e m e s s a g e i n c a s e i t f a i l s d u e t o a c l o c k O u t O f S y n c e r r o r
// S t o r e t h e s e n t T i m e s t a m p f r o m t h e m e s s a g e i n c a s e i t f a i l s d u e t o a c l o c k O u t O f S y n c e r r o r
Morgan Pretty
2 years ago