@ -2,6 +2,7 @@
import Foundation
import GRDB
import SessionSnodeKit
// / A b s t r a c t b a s e c l a s s f o r ` V i s i b l e M e s s a g e ` a n d ` C o n t r o l M e s s a g e ` .
public class Message : Codable {
@ -76,6 +77,258 @@ public class Message: Codable {
}
}
// MARK: - M e s s a g e P a r s i n g / P r o c e s s i n g
public typealias ProcessedMessage = (
threadId : String ? ,
proto : SNProtoContent ,
messageInfo : MessageReceiveJob . Details . MessageInfo
)
public extension Message {
static let nonThreadMessageId : String = " NON_THREAD_MESSAGE "
enum Variant : String , Codable {
case readReceipt
case typingIndicator
case closedGroupControlMessage
case dataExtractionNotification
case expirationTimerUpdate
case configurationMessage
case unsendRequest
case messageRequestResponse
case visibleMessage
init ? ( from type : Message ) {
switch type {
case is ReadReceipt : self = . readReceipt
case is TypingIndicator : self = . typingIndicator
case is ClosedGroupControlMessage : self = . closedGroupControlMessage
case is DataExtractionNotification : self = . dataExtractionNotification
case is ExpirationTimerUpdate : self = . expirationTimerUpdate
case is ConfigurationMessage : self = . configurationMessage
case is UnsendRequest : self = . unsendRequest
case is MessageRequestResponse : self = . messageRequestResponse
case is VisibleMessage : self = . visibleMessage
default : return nil
}
}
var messageType : Message . Type {
switch self {
case . readReceipt : return ReadReceipt . self
case . typingIndicator : return TypingIndicator . self
case . closedGroupControlMessage : return ClosedGroupControlMessage . self
case . dataExtractionNotification : return DataExtractionNotification . self
case . expirationTimerUpdate : return ExpirationTimerUpdate . self
case . configurationMessage : return ConfigurationMessage . self
case . unsendRequest : return UnsendRequest . self
case . messageRequestResponse : return MessageRequestResponse . self
case . visibleMessage : return VisibleMessage . self
}
}
func decode < CodingKeys : CodingKey > ( from container : KeyedDecodingContainer < CodingKeys > , forKey key : CodingKeys ) throws -> Message {
switch self {
case . readReceipt : return try container . decode ( ReadReceipt . self , forKey : key )
case . typingIndicator : return try container . decode ( TypingIndicator . self , forKey : key )
case . closedGroupControlMessage :
return try container . decode ( ClosedGroupControlMessage . self , forKey : key )
case . dataExtractionNotification :
return try container . decode ( DataExtractionNotification . self , forKey : key )
case . expirationTimerUpdate : return try container . decode ( ExpirationTimerUpdate . self , forKey : key )
case . configurationMessage : return try container . decode ( ConfigurationMessage . self , forKey : key )
case . unsendRequest : return try container . decode ( UnsendRequest . self , forKey : key )
case . messageRequestResponse : return try container . decode ( MessageRequestResponse . self , forKey : key )
case . visibleMessage : return try container . decode ( VisibleMessage . self , forKey : key )
}
}
}
static func createMessageFrom ( _ proto : SNProtoContent , sender : String ) -> Message ? {
// N o t e : T h i s a r r a y i s o r d e r e d i n t e n t i o n a l l y t o e n s u r e t h e c o r r e c t t y p e s a r e p r o c e s s e d
// a n d a r e n ' t p a r s e d a s t h e w r o n g t y p e
let prioritisedVariants : [ Variant ] = [
. readReceipt ,
. typingIndicator ,
. closedGroupControlMessage ,
. dataExtractionNotification ,
. expirationTimerUpdate ,
. configurationMessage ,
. unsendRequest ,
. messageRequestResponse ,
. visibleMessage
]
return prioritisedVariants
. reduce ( nil ) { prev , variant in
guard prev = = nil else { return prev }
return variant . messageType . fromProto ( proto , sender : sender )
}
}
static func processRawReceivedMessage (
_ db : Database ,
rawMessage : SnodeReceivedMessage
) throws -> ProcessedMessage ? {
guard let envelope = SNProtoEnvelope . from ( rawMessage ) else {
throw MessageReceiverError . invalidMessage
}
do {
let processedMessage : ProcessedMessage ? = try processRawReceivedMessage (
db ,
envelope : envelope ,
serverExpirationTimestamp : ( TimeInterval ( rawMessage . info . expirationDateMs ) / 1000 ) ,
serverHash : rawMessage . info . hash ,
handleClosedGroupKeyUpdateMessages : true
)
// R e t r i e v e t h e n u m b e r o f e n t r i e s w e h a v e f o r t h e h a s h o f t h i s m e s s a g e
let numExistingHashes : Int = ( try ? SnodeReceivedMessageInfo
. filter ( SnodeReceivedMessageInfo . Columns . hash = = rawMessage . info . hash )
. fetchCount ( db ) )
. defaulting ( to : 0 )
// T r y t o i n s e r t t h e r a w m e s s a g e i n f o i n t o t h e d a t a b a s e ( u s e d f o r b o t h r e q u e s t p a g i n g a n d
// d e - d u p i n g p u r p o s e s )
_ = try rawMessage . info . inserted ( db )
// I f t h e a b o v e i n s e r t i o n w o r k e d t h e n w e h a d n ' t p r o c e s s e d t h i s m e s s a g e f o r t h i s s p e c i f i c
// s e r v i c e n o d e , b u t m a y h a v e d o n e s o f o r a n o t h e r n o d e - i f t h e h a s h a l r e a d y e x i s t e d i n
// t h e d a t a b a s e b e f o r e w e i n s e r t e d i t f o r t h i s n o d e t h e n w e c a n i g n o r e t h i s m e s s a g e a s a
// d u p l i c a t e
guard numExistingHashes = = 0 else { throw MessageReceiverError . duplicateMessage }
return processedMessage
}
catch {
// I f w e g e t ' s e l f S e n d ' o r ' d u p l i c a t e C o n t r o l M e s s a g e ' e r r o r s t h e n w e s t i l l w a n t t o i n s e r t
// t h e S n o d e R e c e i v e d M e s s a g e I n f o t o p r e v e n t r e t r i e v i n g a n d a t t e m p t i n g t o p r o c e s s t h e s a m e
// m e s s a g e a g a i n ( a s w e l l a s e n s u r e t h e n e x t p o l l d o e s n ' t r e t r i e v e t h e s a m e m e s s a g e )
switch error {
case MessageReceiverError . selfSend , MessageReceiverError . duplicateControlMessage :
_ = try ? rawMessage . info . inserted ( db )
break
default : break
}
throw error
}
}
static func processRawReceivedMessage (
_ db : Database ,
serializedData : Data ,
serverHash : String ?
) throws -> ProcessedMessage ? {
guard let envelope = try ? SNProtoEnvelope . parseData ( serializedData ) else {
throw MessageReceiverError . invalidMessage
}
return try processRawReceivedMessage (
db ,
envelope : envelope ,
serverExpirationTimestamp : ( Date ( ) . timeIntervalSince1970 + ControlMessageProcessRecord . defaultExpirationSeconds ) ,
serverHash : serverHash ,
handleClosedGroupKeyUpdateMessages : true
)
}
// / T h i s m e t h o d b e h a v e s s l i g h t l y d i f f e r e n t l y f r o m t h e o t h e r ` p r o c e s s R a w R e c e i v e d M e s s a g e ` m e t h o d s a s i t d o e s n ' t
// / i n s e r t t h e " m e s s a g e i n f o " f o r d e d u p i n g ( w e w a n t t h e p o l l e r t o r e - p r o c e s s t h e m e s s a g e ) a n d a l s o a v o i d s h a n d l i n g a n y
// / c l o s e d g r o u p k e y u p d a t e m e s s a g e s ( t h e ` N o t i f i c a t i o n S e r v i c e E x t e n s i o n ` d o e s t h i s i t s e l f )
static func processRawReceivedMessageAsNotification (
_ db : Database ,
envelope : SNProtoEnvelope
) throws -> ProcessedMessage ? {
let processedMessage : ProcessedMessage ? = try processRawReceivedMessage (
db ,
envelope : envelope ,
serverExpirationTimestamp : ( Date ( ) . timeIntervalSince1970 + ControlMessageProcessRecord . defaultExpirationSeconds ) ,
serverHash : nil ,
handleClosedGroupKeyUpdateMessages : false
)
return processedMessage
}
private static func processRawReceivedMessage (
_ db : Database ,
envelope : SNProtoEnvelope ,
serverExpirationTimestamp : TimeInterval ,
serverHash : String ? ,
// TODO: T h e s e
openGroupId : String ? = nil ,
openGroupMessageServerId : UInt64 ? = nil ,
handleClosedGroupKeyUpdateMessages : Bool
) throws -> ProcessedMessage ? {
let ( message , proto , threadId ) = try MessageReceiver . parse (
db ,
envelope : envelope ,
serverExpirationTimestamp : serverExpirationTimestamp ,
openGroupId : openGroupId ,
openGroupMessageServerId : openGroupMessageServerId
)
message . serverHash = serverHash
// I g n o r e i n v a l i d m e s s a g e s a n d h a s h e s f o r m e s s a g e s w e h a v e p r e v i o u s l y h a n d l e d
guard let variant : Message . Variant = Message . Variant ( from : message ) else {
throw MessageReceiverError . invalidMessage
}
// / * * N o t e : * * W e w a n t t o i m m e d i a t e l y h a n d l e a n y ` C l o s e d G r o u p C o n t r o l M e s s a g e ` w i t h t h e k i n d ` e n c r y p t i o n K e y P a i r ` a s
// / w e n e e d t h e k e y P a i r i n s t o r a g e i n o r d e r t o b e a b l e t o p a r s e a n d m e s s a g e s w h i c h w e r e s i g n e d w i t h t h e n e w k e y ( a l s o n o n e e d t o a d d
// / t h e s e a s j o b s a s t h e y w i l l b e f u l l y h a n d l e d i n h e r e )
if handleClosedGroupKeyUpdateMessages {
switch message {
case let closedGroupControlMessage as ClosedGroupControlMessage :
switch closedGroupControlMessage . kind {
case . encryptionKeyPair :
try MessageReceiver . handleClosedGroupControlMessage ( db , closedGroupControlMessage )
return nil
default : break
}
default : break
}
}
// 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
do {
try ControlMessageProcessRecord (
threadId : threadId ,
message : message ,
serverExpirationTimestamp : serverExpirationTimestamp
) ? . insert ( db )
}
catch {
// W e w a n t t o c u s t o m h a n d l e t h i s
if case DatabaseError . SQLITE_CONSTRAINT_UNIQUE = error {
throw MessageReceiverError . duplicateControlMessage
}
throw error
}
return (
threadId ,
proto ,
try MessageReceiveJob . Details . MessageInfo (
message : message ,
variant : variant ,
proto : proto
)
)
}
}
// MARK: - M u t a t i o n
internal extension Message {
Morgan Pretty
3 years ago