/*
 * Copyright 2019 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.google.android.gnd.persistence.sync;

import static com.google.android.gnd.util.ImmutableListCollector.toImmutableList;
import static java8.util.stream.StreamSupport.stream;

import android.content.Context;
import android.util.Log;
import androidx.annotation.NonNull;
import androidx.work.Data;
import androidx.work.Worker;
import androidx.work.WorkerParameters;
import com.google.android.gnd.model.Mutation;
import com.google.android.gnd.model.User;
import com.google.android.gnd.persistence.local.LocalDataStore;
import com.google.android.gnd.persistence.remote.RemoteDataStore;
import com.google.common.collect.ImmutableList;
import io.reactivex.Completable;
import io.reactivex.Observable;
import java.util.Map;
import java.util.Set;
import java8.util.stream.Collectors;

/**
 * A worker that syncs local changes to the remote data store. Each instance handles mutations for a
 * specific map feature, whose id is provided in the {@link Data} object built by {@link
 * #createInputData} and provided to the worker request while being enqueued.
 */
public class LocalMutationSyncWorker extends Worker {

  private static final String TAG = LocalMutationSyncWorker.class.getSimpleName();
  public static final String FEATURE_ID_PARAM_KEY = "featureId";

  private final LocalDataStore localDataStore;
  private final RemoteDataStore remoteDataStore;
  private final String featureId;

  public LocalMutationSyncWorker(
      @NonNull Context context,
      @NonNull WorkerParameters params,
      LocalDataStore localDataStore,
      RemoteDataStore remoteDataStore) {
    super(context, params);
    this.localDataStore = localDataStore;
    this.remoteDataStore = remoteDataStore;
    this.featureId = params.getInputData().getString(FEATURE_ID_PARAM_KEY);
  }

  /** Returns a new work {@link Data} object containing the specified feature id. */
  public static Data createInputData(String featureId) {
    return new Data.Builder().putString(FEATURE_ID_PARAM_KEY, featureId).build();
  }

  @NonNull
  @Override
  public Result doWork() {
    Log.d(TAG, "Connected. Syncing changes to feature " + featureId);
    ImmutableList<Mutation> mutations = localDataStore.getPendingMutations(featureId).blockingGet();
    try {
      Log.v(TAG, "Mutations: " + mutations);
      processMutations(mutations).blockingAwait();
      return Result.success();
    } catch (Throwable t) {
      Log.d(TAG, "Remote updates for feature " + featureId + " failed", t);
      localDataStore.updateMutations(incrementRetryCounts(mutations, t)).blockingAwait();
      return Result.retry();
    }
  }

  /**
   * Groups mutations by user id, loads each user, applies mutations, and removes processed
   * mutations.
   */
  private Completable processMutations(ImmutableList<Mutation> pendingMutations) {
    Map<String, ImmutableList<Mutation>> mutationsByUserId = groupByUserId(pendingMutations);
    Set<String> userIds = mutationsByUserId.keySet();
    return Observable.fromIterable(userIds)
        .flatMapCompletable(userId -> processMutations(mutationsByUserId.get(userId), userId));
  }

  /** Loads each user with specified id, applies mutations, and removes processed mutations. */
  private Completable processMutations(ImmutableList<Mutation> mutations, String userId) {
    return localDataStore
        .loadUser(userId)
        .flatMapCompletable(user -> processMutations(mutations, user))
        .doOnError(__ -> Log.d(TAG, "User account removed before mutation processed"))
        .onErrorComplete();
  }

  /** Applies mutations to remote data store. Once successful, removes them from the local db. */
  private Completable processMutations(ImmutableList<Mutation> mutations, User user) {
    return remoteDataStore
        .applyMutations(mutations, user)
        .andThen(localDataStore.removePendingMutations(mutations));
  }

  private Map<String, ImmutableList<Mutation>> groupByUserId(
      ImmutableList<Mutation> pendingMutations) {
    return stream(pendingMutations)
        .collect(Collectors.groupingBy(Mutation::getUserId, toImmutableList()));
  }

  private ImmutableList<Mutation> incrementRetryCounts(
      ImmutableList<Mutation> mutations, Throwable error) {
    return stream(mutations).map(m -> incrementRetryCount(m, error)).collect(toImmutableList());
  }

  private Mutation incrementRetryCount(Mutation mutation, Throwable error) {
    return mutation
        .toBuilder()
        .setRetryCount(mutation.getRetryCount() + 1)
        .setLastError(error.toString())
        .build();
  }
}

@@ -812,33 +812,8 @@
 xt;%0A
-import android.util.Log;%0A
 impo
@@ -844,16 +844,16 @@
 onNull;%0A
+
 import a
@@ -1350,16 +1350,42 @@
 lectors;
+%0Aimport timber.log.Timber;
 %0A%0A/**%0A *
@@ -1743,90 +1743,8 @@
 ing 
-TAG = LocalMutationSyncWorker.class.getSimpleName();%0A  public static final String 
 FEAT
@@ -2569,27 +2569,25 @@
 ) %7B%0A    
-Log.d(TAG, 
+Timber.d(
 %22Connect
@@ -2609,35 +2609,36 @@
 nges to feature 
-%22 +
+%25s%22,
  featureId);%0A   
@@ -2755,19 +2755,17 @@
     
-Log.v(TAG, 
+Timber.v(
 %22Mut
@@ -2772,19 +2772,20 @@
 ations: 
-%22 +
+%25s%22,
  mutatio
@@ -2905,25 +2905,26 @@
 %7B%0A      
-Log.d(TAG
+Timber.d(t
 , %22Remot
@@ -2949,25 +2949,10 @@
 ure 
-%22 + featureId + %22
+%25s
  fai
@@ -2957,17 +2957,25 @@
 ailed%22, 
-t
+featureId
 );%0A     
@@ -3934,19 +3934,17 @@
  -%3E 
-Log.d(TAG, 
+Timber.d(
 %22Use

Replace log with Timber