Mercury-IM/data/src/main/java/org/mercury_im/messenger/data/repository/RosterRepository.java

package org.mercury_im.messenger.data.repository;
import org.mercury_im.messenger.data.model.AccountModel;
import org.mercury_im.messenger.data.model.ContactModel;
import org.mercury_im.messenger.data.model.PeerModel;
import org.mercury_im.messenger.util.ThreadUtils;

import javax.inject.Inject;
import javax.inject.Named;

import io.reactivex.Completable;
import io.reactivex.Observable;
import io.reactivex.Scheduler;
import io.reactivex.Single;
import io.requery.Persistable;
import io.requery.reactivex.ReactiveEntityStore;
import io.requery.reactivex.ReactiveResult;

public class RosterRepository extends RequeryRepository {

    @Inject
    public RosterRepository(ReactiveEntityStore<Persistable> data,
                            @Named(value = ThreadUtils.SCHEDULER_IO) Scheduler subscriberScheduler,
                            @Named(value = ThreadUtils.SCHEDULER_UI) Scheduler observerScheduler) {
        super(data, subscriberScheduler, observerScheduler);
    }

    /*
    ContactModel related methods
     */

    public Observable<ReactiveResult<ContactModel>> getAllContactsOfAccount(AccountModel accountModel) {
        return getAllContactsOfAccount(accountModel.getId());
    }

    public Observable<ReactiveResult<ContactModel>> getAllContactsOfAccount(long accountId) {
        return data().select(ContactModel.class).join(PeerModel.class).on(ContactModel.ENTITY_ID.eq(PeerModel.ID))
                .where(PeerModel.ACCOUNT_ID.eq(accountId))
                .get().observableResult()
                .subscribeOn(subscriberScheduler()).observeOn(observerScheduler());
    }

    public Single<ContactModel> upsertContact(ContactModel contact) {
        return data().upsert(contact).subscribeOn(subscriberScheduler()).observeOn(observerScheduler());
    }

    public Completable deleteContact(ContactModel contact) {
        return data().delete(contact).subscribeOn(subscriberScheduler()).observeOn(observerScheduler());
    }

    public Completable deleteContact(long accountId, String address) {
        return data().delete(ContactModel.class).from(ContactModel.class)
                .join(PeerModel.class).on(ContactModel.ENTITY_ID.eq(PeerModel.ID))
                .where(PeerModel.ACCOUNT_ID.eq(accountId).and(PeerModel.ADDRESS.eq(address)))
                .get().single().ignoreElement()
                .subscribeOn(subscriberScheduler()).observeOn(observerScheduler());
    }

    public Single<Integer> deleteAllContactsOfAccount(AccountModel account) {
        return deleteAllContactsOfAccount(account.getId());
    }

    public Single<Integer> deleteAllContactsOfAccount(long accountId) {
        return data().delete(ContactModel.class).from(ContactModel.class)
                .join(PeerModel.class).on(ContactModel.ENTITY_ID.eq(PeerModel.ID))
                .where(PeerModel.ACCOUNT_ID.eq(accountId))
                .get().single()
                .subscribeOn(subscriberScheduler()).observeOn(observerScheduler());
    }


    /*
    PeerModel related methods
     */

    public Observable<ReactiveResult<PeerModel>> getAllEntitiesOfAccount(AccountModel account) {
        return getAllEntitiesOfAccount(account.getId());
    }

    public Observable<ReactiveResult<PeerModel>> getAllEntitiesOfAccount(long accountId) {
        return data().select(PeerModel.class).where(PeerModel.ACCOUNT_ID.eq(accountId))
                .get().observableResult()
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    public Observable<ReactiveResult<PeerModel>> getEntityById(long entityId) {
        return data().select(PeerModel.class).where(PeerModel.ID.eq(entityId))
                .get().observableResult()
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    public Observable<ReactiveResult<PeerModel>> getEntityByJid(AccountModel account, String address) {
        return getEntityByJid(account.getId(), address);
    }

    public Observable<ReactiveResult<PeerModel>> getEntityByJid(long accountId, String address) {
        return data().select(PeerModel.class)
                .where(PeerModel.ACCOUNT_ID.eq(accountId).and(PeerModel.ADDRESS.eq(address)))
                .get().observableResult()
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    public Single<PeerModel> getOrCreateEntity(long accountId, String address) {
        return Single.fromCallable(() -> {
            AccountModel account = data().select(AccountModel.class).where(AccountModel.ID.eq(accountId))
                    .get().first();
            return getOrCreateEntity(account, address).blockingGet();
        })
                .observeOn(observerScheduler())
                .subscribeOn(subscriberScheduler());
    }

    public Single<PeerModel> getOrCreateEntity(AccountModel account, String address) {
        return Single.fromCallable(() -> {
            PeerModel entity = data().select(PeerModel.class)
                    .where(PeerModel.ACCOUNT_ID.eq(account.getId()).and(PeerModel.ADDRESS.eq(address)))
                    .get().firstOrNull();
            if (entity == null) {
                entity = new PeerModel();
                entity.setAccount(account);
                entity.setAddress(address);
                entity = data().insert(entity).blockingGet();
            }
            return entity;
        })
                .observeOn(observerScheduler())
                .subscribeOn(subscriberScheduler());

    }

    public Single<PeerModel> upsertEntity(PeerModel entity) {
        return data().upsert(entity)
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    public Completable deleteEntity(PeerModel entity) {
        return data().delete(entity)
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    /*
    RosterVersion related methods
     */

    public Single<Integer> updateRosterVersion(long accountId, String rosterVer) {
        return data().update(AccountModel.class).set(AccountModel.ROSTER_VERSION, rosterVer)
                .where(AccountModel.ID.eq(accountId))
                .get().single()
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    public Observable<String> getRosterVersion(AccountModel account) {
        return getRosterVersion(account.getId());
    }

    public Observable<String> getRosterVersion(long accountId) {
        return data().select(AccountModel.class).where(AccountModel.ID.eq(accountId))
                .get().observableResult()
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler())
                .map(accountModels -> {
                    AccountModel accountModel = accountModels.firstOrNull();
                    if (accountModel == null || accountModel.getRosterVersion() == null) {
                        return "";
                    }
                    return accountModel.getRosterVersion();
                });
    }

    public Single<AccountModel> updateRosterVersion(AccountModel account, String rosterVersion) {
        account.setRosterVersion(rosterVersion);
        return data().upsert(account)
                .subscribeOn(subscriberScheduler()).observeOn(observerScheduler());
    }

    public Observable<ReactiveResult<ContactModel>> getContact(AccountModel account, String address) {
        return getContact(account.getId(), address);
    }

    public Observable<ReactiveResult<ContactModel>> getContact(long accountId, String address) {
        return data().select(ContactModel.class).from(ContactModel.class)
                .join(PeerModel.class).on(ContactModel.ENTITY_ID.eq(PeerModel.ID))
                .where(PeerModel.ACCOUNT_ID.eq(accountId).and(PeerModel.ADDRESS.eq(address)))
                .get().observableResult()
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    public Observable<ReactiveResult<ContactModel>> getAllContacts() {
        return data().select(ContactModel.class)
                .get().observableResult()
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }
}