package com.contentful.java.cda;

import io.reactivex.Flowable;
import io.reactivex.functions.Function;
import retrofit2.Response;

import static com.contentful.java.cda.Util.checkNotNull;

/**
 * Represents a query to the Sync API.
 */
public class SyncQuery {
  final CDAClient client;

  final String syncToken;

  final SynchronizedSpace space;

  final boolean initial;

  final SyncType type;

  SyncQuery(Builder builder) {
    this.client = checkNotNull(builder.client, "Client must not be null.");
    this.syncToken = builder.syncToken;
    this.space = builder.space;
    this.initial = space == null && syncToken == null;
    this.type = builder.type;
  }

  /**
   * Returns an {@link Flowable} to which one can subscribe in order to fulfill this sync query.
   *
   * @return {@link Flowable} instance.
   */
  public Flowable<SynchronizedSpace> observe() {
    final String token;
    if (space != null) {
      String nextSyncUrl = space.nextSyncUrl();
      if (nextSyncUrl == null) {
        throw new IllegalArgumentException("Provided space for synchronization is corrupt.");
      } else {
        token = Util.queryParam(space.nextSyncUrl(), "sync_token");
      }
    } else {
      token = syncToken;
    }
    return client.cacheAll(true)
        .flatMap(new Function<Cache, Flowable<Response<SynchronizedSpace>>>() {
          @Override public Flowable<Response<SynchronizedSpace>> apply(Cache cache) {
            return client.service.sync(client.spaceId,
                initial ? initial : null, token,
                type != null ? type.getName() : null,
                type != null ? type.getContentType() : null);
          }
        }).map(new Function<Response<SynchronizedSpace>, SynchronizedSpace>() {
          @Override public SynchronizedSpace apply(Response<SynchronizedSpace> synchronizedSpace) {
            return ResourceFactory.sync(synchronizedSpace, space, client);
          }
        });
  }

  /**
   * Invokes the request to sync (blocking).
   *
   * @return {@link SynchronizedSpace} instance.
   */
  public SynchronizedSpace fetch() {
    return observe().blockingFirst();
  }

  /**
   * Invokes the request to sync (asynchronously) with the provided {@code callback}.
   *
   * @param callback callback.
   * @param <C>      callback type.
   * @return the given callback instance.
   */
  @SuppressWarnings("unchecked")
  public <C extends CDACallback<SynchronizedSpace>> C fetch(C callback) {
    return (C) Callbacks.subscribeAsync(observe(), callback, client);
  }

  static Builder builder() {
    return new Builder();
  }

  static class Builder {
    CDAClient client;

    String syncToken;

    SynchronizedSpace space;

    SyncType type;

    Builder setClient(CDAClient client) {
      this.client = client;
      return this;
    }

    Builder setSyncToken(String syncToken) {
      this.syncToken = syncToken;
      return this;
    }

    Builder setSpace(SynchronizedSpace space) {
      this.space = space;
      return this;
    }

    Builder setType(SyncType type) {
      this.type = type;
      return this;
    }

    SyncQuery build() {
      return new SyncQuery(this);
    }
  }
}

@@ -595,42 +595,27 @@
 l = 
-space == null && syncToken == null
+builder.isInitial()
 ;%0A  
@@ -1519,32 +1519,43 @@
 %0A               
+ initial &&
  type != null ? 
@@ -1592,16 +1592,27 @@
         
+ initial &&
  type !=
@@ -3046,24 +3046,51 @@
 ype type) %7B%0A
+      if (isInitial()) %7B%0A  
       this.t
@@ -3101,16 +3101,24 @@
 = type;%0A
+      %7D%0A
       re
@@ -3131,24 +3131,106 @@
 his;%0A    %7D%0A%0A
+    boolean isInitial() %7B%0A      return space == null && syncToken == null;%0A    %7D%0A%0A
     SyncQuer

set type only on initial sync