seekia/internal/messaging/myMessageQueue/myMessageQueue.go

// myMessageQueue provides functions to add messages we want to send to the message queue
// All messages are added to the queue before being sent
// This is useful for 2 reasons.
//  1. If the account credit server is down/unreachable, we can automatically retry funding the message later
//  2. We can queue messages to send to users whose chat keys we do not yet have (and may not exist on the network at all)

package myMessageQueue

// For users whose keys are missing, we will wait for their chat keys to download in the background
// If we can't download their chat keys after trying for a certain time, we will discard the message

//TODO: If we disable our profile or delete our identity, clear our chat message queue
//TODO: A way to view the message queue through the GUI

import "seekia/internal/encoding"
import "seekia/internal/helpers"
import "seekia/internal/identity"
import "seekia/internal/messaging/myChatMessages"
import "seekia/internal/messaging/peerChatKeys"
import "seekia/internal/messaging/sendMessages"
import "seekia/internal/myDatastores/myMapList"
import "seekia/internal/myIdentity"
import "seekia/internal/network/myAccountCredit"
import "seekia/internal/parameters/getParameters"
import "seekia/internal/unixTime"

import "errors"
import "time"
import "sync"

// This will be locked whenever we are updating the message queue map list
var updatingMessageQueueMapListMutex sync.Mutex

var myMessageQueueMapListDatastore *myMapList.MyMapList

// This function must be called whenever an app user signs in
func InitializeMyMessageQueueDatastore()error{

	updatingMessageQueueMapListMutex.Lock()
	defer updatingMessageQueueMapListMutex.Unlock()

	newMyMessageQueueMapListDatastore, err := myMapList.CreateNewMapList("MyMessageQueue")
	if (err != nil) { return err }

	myMessageQueueMapListDatastore = newMyMessageQueueMapListDatastore

	return nil
}


//Outputs:
//	-bool: Parameters exist
//	-bool: Sufficient credits exist
//	-error
func AddMessageToMyMessageQueue(myIdentityHash [16]byte, recipientIdentityHash [16]byte, messageNetworkType byte, messageDuration int, messageCommunication string)(bool, bool, error){

	// We keep track of our identity to make sure if we don't clear the queue properly between changing user/importing new identity, messages will not be sent by the wrong identity

	myIdentityExists, myIdentityType, err := myIdentity.CheckIfIdentityHashIsMine(myIdentityHash)
	if (err != nil) { return false, false, err }
	if (myIdentityExists == false){
		return false, false, errors.New("AddMessageToMyMessageQueue called with an unknown identity.")
	}

	myIdentityHashString, _, err := identity.EncodeIdentityHashBytesToString(myIdentityHash)
	if (err != nil) {
		myIdentityHashHex := encoding.EncodeBytesToHexString(myIdentityHash[:])
		return false, false, errors.New("CheckIfIdentityHashIsMine returning invalid identity hash: " + myIdentityHashHex)
	}

	recipientIdentityHashString, recipientIdentityType, err := identity.EncodeIdentityHashBytesToString(recipientIdentityHash)
	if (err != nil) {
		recipientIdentityHashHex := encoding.EncodeBytesToHexString(recipientIdentityHash[:])
		return false, false, errors.New("AddMessageToMyMessageQueue called with invalid recipientIdentityHash: " + recipientIdentityHashHex)
	}

	if (myIdentityType != recipientIdentityType){
		return false, false, errors.New("AddMessageToMyMessageQueue called with mismatched my/recipient identityTypes.")
	}

	isValid := helpers.VerifyNetworkType(messageNetworkType)
	if (isValid == false){
		messageNetworkTypeString := helpers.ConvertByteToString(messageNetworkType)
		return false, false, errors.New("AddMessageToMyMessageQueue called with invalid messageNetworkType: " + messageNetworkTypeString)
	}

	messageNetworkTypeString := helpers.ConvertByteToString(messageNetworkType)

	messageDurationString := helpers.ConvertIntToString(messageDuration)

	messageIdentifierBytes, err := helpers.GetNewRandomBytes(20)
	if (err != nil) { return false, false, err }

	messageIdentifier := [20]byte(messageIdentifierBytes)
	messageIdentifierString := encoding.EncodeBytesToHexString(messageIdentifierBytes)

	// TimeAdded will be used as the messageSentTime when we send the message
	currentTime := time.Now().Unix()
	currentTimeString := helpers.ConvertInt64ToString(currentTime)

	newMessageMap := map[string]string{
		"MessageIdentifier": messageIdentifierString,
		"MyIdentityHash": myIdentityHashString,
		"RecipientIdentityHash": recipientIdentityHashString,
		"NetworkType": messageNetworkTypeString,
		"MessageCommunication": messageCommunication,
		"MessageDuration": messageDurationString,
		"TimeAdded": currentTimeString,
	}

	estimatedMessageSize, err := sendMessages.GetEstimatedMessageSize(recipientIdentityHash, messageCommunication)
	if (err != nil) { return false, false, err }

	updatingMessageQueueMapListMutex.Lock()
	defer updatingMessageQueueMapListMutex.Unlock()

	parametersExist, sufficientCreditsExist, err := myAccountCredit.FreezeCreditForMessage(messageIdentifier, messageNetworkType, messageDuration, estimatedMessageSize)
	if (err != nil) { return false, false, err }
	if (parametersExist == false){
		return false, false, nil
	}
	if (sufficientCreditsExist == false){
		return true, false, nil
	}

	err = myMessageQueueMapListDatastore.AddMapListItem(newMessageMap)
	if (err != nil) { return false, false, err }

	// Now we add queued message map to myChatMessages so the user will see a queued message in their GUI

	err = myChatMessages.AddMyNewMessageToMyChatMessages("Queued", [26]byte{}, messageIdentifier, myIdentityHash, recipientIdentityHash, messageNetworkType, currentTime, messageCommunication)
	if (err != nil) { return false, false, err }

	return true, true, nil
}

func PruneMessageQueue()error{

	//TODO: Delete messages that have already been sent

	// Use sendMessages.CheckIfMessageIsSent()

	return nil
}

// This function should be run automatically on a continual period
// It will attempt to send a random selection of messages from the queue
// It will contact 1 host per message
func AttemptToSendMessagesInQueue(identityType string, networkType byte, maximumMessagesToSend int)error{

	if (identityType != "Mate" && identityType != "Moderator"){
		return errors.New("AttemptToSendMessagesInQueue called with invalid identityType: " + identityType)
	}

	isValid := helpers.VerifyNetworkType(networkType)
	if (isValid == false){
		networkTypeString := helpers.ConvertByteToString(networkType)
		return errors.New("AttemptToSendMessagesInQueue called with invalid networkType: " + networkTypeString)
	}

	myIdentityExists, myIdentityHash, err := myIdentity.GetMyIdentityHash(identityType)
	if (err != nil) { return err }
	if (myIdentityExists == false){
		return nil
	}

	parametersExist, err := getParameters.CheckIfChatParametersExist(networkType)
	if (err != nil) { return err }
	if (parametersExist == false){
		// We need these parameters to send messages. We wait for them to download.
		return nil
	}

	updatingMessageQueueMapListMutex.Lock()
	defer updatingMessageQueueMapListMutex.Unlock()

	currentMessageQueueMapList, err := myMessageQueueMapListDatastore.GetMapList()
	if (err != nil) { return err }

	// We randomize the map list order, so we are sending a random subset of the messages

	helpers.RandomizeListOrder(currentMessageQueueMapList)

	// We use this to keep track of how many message sends we have succesfully started
	sentMessagesCount := 0

	messageMapsToDeleteList := make([]map[string]string, 0)

	for _, messageQueueMessageMap := range currentMessageQueueMapList{

		messageMyIdentityHashString, exists := messageQueueMessageMap["MyIdentityHash"]
		if (exists == false) {
			return errors.New("MessageQueue mapList malformed: Item missing MyIdentityHash")
		}

		messageMyIdentityHash, _, err := identity.ReadIdentityHashString(messageMyIdentityHashString)
		if (err != nil){
			return errors.New("MessageQueue mapList malformed: Item contains invalid MyIdentityHash: " + messageMyIdentityHashString)
		}

		if (messageMyIdentityHash != myIdentityHash){
			// Must be identity for different identityType
			continue
		}

		messageNetworkTypeString, exists := messageQueueMessageMap["NetworkType"]
		if (exists == false){
			return errors.New("MessageQueue mapList malformed: Item missing NetworkType.")
		}

		messageNetworkType, err := helpers.ConvertNetworkTypeStringToByte(messageNetworkTypeString)
		if (err != nil){
			return errors.New("MessageQueue mapList malformed: Item contains invalid NetworkType: " + messageNetworkTypeString)
		}
		if (messageNetworkType != networkType){
			// Message belongs to a different networkType.
			continue
		}

		messageIdentifierString, exists := messageQueueMessageMap["MessageIdentifier"]
		if (exists == false) {
			return errors.New("MessageQueue mapList malformed: Item missing MessageIdentifier")
		}

		messageIdentifierBytes, err := encoding.DecodeHexStringToBytes(messageIdentifierString)
		if (err != nil){
			return errors.New("MessageQueue mapList malformed: Item contains invalid messageIdentifier: Not Hex: " + messageIdentifierString)
		}

		if (len(messageIdentifierBytes) != 20){
			return errors.New("MessageQueue mapList malformed: Item contains invalid messageIdentifier: Invalid length: " + messageIdentifierString)
		}

		messageIdentifier := [20]byte(messageIdentifierBytes)

		messageRecipientString, exists := messageQueueMessageMap["RecipientIdentityHash"]
		if (exists == false) {
			return errors.New("MessageQueue mapList malformed: item missing RecipientIdentityHash")
		}

		messageRecipient, _, err := identity.ReadIdentityHashString(messageRecipientString)
		if (err != nil){
			return errors.New("MessageQueue mapList malformed: Item contains invalid RecipientIdentityHash: " + messageRecipientString)
		}

		messageTimeAdded, exists := messageQueueMessageMap["TimeAdded"]
		if (exists == false) {
			return errors.New("MessageQueue mapList malformed: Item missing TimeAdded")
		}

		messageTimeAddedInt64, err := helpers.ConvertBroadcastTimeStringToInt64(messageTimeAdded)
		if (err != nil) {
			return errors.New("MessageQueue mapList malformed: Item contains invalid TimeAdded: " + messageTimeAdded)
		}

		messageCommunication, exists := messageQueueMessageMap["MessageCommunication"]
		if (exists == false) {
			return errors.New("MessageQueue mapList malformed: item missing MessageCommunication")
		}

		// Now we check if recipient keys exist

		peerIsDisabled, peerChatKeysExist, _, _, err := peerChatKeys.GetPeerNewestActiveChatKeys(messageRecipient, networkType)
		if (err != nil) { return err }
		if (peerIsDisabled == true || peerChatKeysExist == false){

			// We see if message entry is more than 1 week old
			// If it is, we will delete it

			currentTime := time.Now().Unix()

			elapsedTime := currentTime - messageTimeAddedInt64

			maximumTime := unixTime.GetWeekUnix()

			if (elapsedTime > maximumTime){

				creditsFound, err := myAccountCredit.ReleaseFrozenCreditForMessage(messageIdentifier)
				if (err != nil){ return err }
				if (creditsFound == false){
					// This should not happen
					// If the user manually released credits, the queued message that the credits were held for should have been deleted from queue
					return errors.New("MessageQueue message frozen credits not found.")
				}

				// We update the existing message status inside myChatMessages to "Failed"
				err = myChatMessages.AddMyNewMessageToMyChatMessages("Failed", [26]byte{}, messageIdentifier, messageMyIdentityHash, messageRecipient, messageNetworkType, messageTimeAddedInt64, messageCommunication)
				if (err != nil) { return err }

				messageMapsToDeleteList = append(messageMapsToDeleteList, messageQueueMessageMap)
			}

			// Peer chat keys were not found.
			// We will keep trying to download them. We skip this message for now.

			continue
		}

		// We initiate message send

		messageDurationString, exists := messageQueueMessageMap["MessageDuration"]
		if (exists == false) {
			return errors.New("MessageQueue map list malformed: item missing MessageDuration")
		}

		messageDurationToFund, err := helpers.ConvertStringToInt(messageDurationString)
		if (err != nil) {
			return errors.New("MessageQueue mapList malformed: Item contains invalid MessageDuration: " + messageDurationString)
		}

		messageIsPendingOrSent, successfullyStartedSend, err := sendMessages.StartSendingMessage(messageIdentifier, myIdentityHash, messageRecipient, messageNetworkType, messageTimeAddedInt64, messageCommunication, messageDurationToFund)
		if (err != nil) { return err }
		if (messageIsPendingOrSent == true){
			// We are already trying to send the message, or the message has already been sent.
			continue
		}
		if (successfullyStartedSend == false){

			// We don't have the required data 
			// We are probably missing parameters
			// We continue looking for messages to send

			continue
		}

		// We successfully started sending the message.
		// It will be deleted from the message queue automatically if the send is successful

		sentMessagesCount += 1

		if (sentMessagesCount >= maximumMessagesToSend){
			// We have sent the number of messages we wanted to
			break
		}
	}

	for _, messageMapToDelete := range messageMapsToDeleteList{

		err := myMessageQueueMapListDatastore.DeleteMapListItems(messageMapToDelete)
		if (err != nil) { return err }
	}

	return nil
}