Mercury-IM/domain/src/main/java/org/mercury_im/messenger/core/store/message/MercuryMessageStore.java

package org.mercury_im.messenger.core.store.message;

import org.jivesoftware.smack.chat2.IncomingChatMessageListener;
import org.jivesoftware.smack.chat2.OutgoingChatMessageListener;
import org.jivesoftware.smack.packet.MessageBuilder;
import org.jivesoftware.smackx.delay.packet.DelayInformation;
import org.jivesoftware.smackx.eme.element.ExplicitMessageEncryptionElement;
import org.jivesoftware.smackx.ox.OpenPgpContact;
import org.jivesoftware.smackx.ox.element.SigncryptElement;
import org.jivesoftware.smackx.ox_im.OxMessageListener;
import org.jxmpp.jid.EntityBareJid;
import org.mercury_im.messenger.core.SchedulersFacade;
import org.mercury_im.messenger.core.data.repository.DirectChatRepository;
import org.mercury_im.messenger.core.data.repository.MessageRepository;
import org.mercury_im.messenger.core.data.repository.PeerRepository;
import org.mercury_im.messenger.entity.Account;
import org.mercury_im.messenger.entity.message.Message;
import org.mercury_im.messenger.entity.message.MessageDirection;
import org.pgpainless.decryption_verification.OpenPgpMetadata;

import java.util.Date;
import java.util.logging.Level;
import java.util.logging.Logger;

import javax.inject.Inject;
import javax.inject.Singleton;

import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;

@Singleton
public class MercuryMessageStore implements IncomingChatMessageListener, OxMessageListener {

    private static final Logger LOGGER = Logger.getLogger(MercuryMessageStore.class.getName());

    private final MessageRepository messageRepository;
    private final DirectChatRepository directChatRepository;
    private final PeerRepository peerRepository;
    private final SchedulersFacade schedulers;

    private Account account;

    private CompositeDisposable disposable = new CompositeDisposable();

    @Inject
    public MercuryMessageStore(Account account,
                               PeerRepository peerRepository,
                               DirectChatRepository directChatRepository,
                               MessageRepository messageRepository,
                               SchedulersFacade schedulers) {
        this.account = account;
        this.peerRepository = peerRepository;
        this.directChatRepository = directChatRepository;
        this.messageRepository = messageRepository;
        this.schedulers = schedulers;
    }

    @Override
    public void newIncomingMessage(EntityBareJid from,
                                   org.jivesoftware.smack.packet.Message smackMessage,
                                   org.jivesoftware.smack.chat2.Chat smackChat) {
        if (ExplicitMessageEncryptionElement.from(smackMessage) != null) {
            return;
        }
        Message message = new Message();
        message.setDirection(MessageDirection.incoming);
        DelayInformation delayInformation = DelayInformation.from(smackMessage);
        message.setTimestamp(delayInformation != null ? delayInformation.getStamp() : new Date());
        message.setSender(from);
        message.setRecipient(smackMessage.getTo().asEntityJidOrThrow());
        message.setXml(smackMessage.toXML().toString());
        if (smackMessage.getBody() != null) {
            message.setBody(smackMessage.getBody());
        }
        disposable.add(writeMessageToStore(from, message));
    }

    private Disposable writeMessageToStore(EntityBareJid peer, Message message) {
        return peerRepository.getOrCreatePeer(account, peer)
                .flatMap(directChatRepository::getOrCreateChatWithPeer)
                .map(chat -> {
                    message.setChatId(chat.getId());
                    return chat;
                })
                .flatMap(chat -> messageRepository.insertMessage(message))
                .subscribeOn(schedulers.getIoScheduler())
                .subscribe(m -> LOGGER.log(Level.INFO, "Message written"), e -> LOGGER.log(Level.SEVERE, "Error: ", e));
    }

    @Override
    public void newIncomingOxMessage(OpenPgpContact contact,
                                     org.jivesoftware.smack.packet.Message smackMessage,
                                     SigncryptElement decryptedPayload,
                                     OpenPgpMetadata metadata) {
        Message message = new Message();
        message.setDirection(MessageDirection.incoming);
        DelayInformation delayInformation = DelayInformation.from(smackMessage);
        message.setTimestamp(delayInformation != null ? delayInformation.getStamp() : new Date());
        message.setSender(contact.getJid().asEntityJidOrThrow());
        message.setRecipient(smackMessage.getTo().asEntityJidOrThrow());
        message.setXml(smackMessage.toXML().toString());
        org.jivesoftware.smack.packet.Message.Body body = decryptedPayload.getExtension(org.jivesoftware.smack.packet.Message.Body.ELEMENT,
                org.jivesoftware.smack.packet.Message.Body.NAMESPACE);
        if (body != null) {
            message.setBody(body.getMessage());
        }
        disposable.add(writeMessageToStore(contact.getJid().asEntityBareJidOrThrow(), message));
    }
}