/*
* Copyright 2017 The Data-Portability Project Authors.
*
* 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 org.dataportabilityproject.worker;

import com.google.gson.Gson;
import com.google.inject.Inject;
import java.io.IOException;
import java.util.UUID;
import org.dataportabilityproject.PortabilityCopier;
import org.dataportabilityproject.ServiceProviderRegistry;
import org.dataportabilityproject.cloud.interfaces.CloudFactory;
import org.dataportabilityproject.job.Crypter;
import org.dataportabilityproject.job.CrypterFactory;
import org.dataportabilityproject.shared.PortableDataType;
import org.dataportabilityproject.spi.cloud.storage.JobStore;
import org.dataportabilityproject.spi.cloud.types.JobAuthorization;
import org.dataportabilityproject.spi.cloud.types.LegacyPortabilityJob;
import org.dataportabilityproject.types.transfer.auth.AuthData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

final class WorkerImpl {
  private static final Logger logger = LoggerFactory.getLogger(WorkerImpl.class);
  private static final Gson GSON = new Gson();

  private final CloudFactory cloudFactory;
  private final JobPollingService jobPollingService;
  private final JobStore store;
  private final ServiceProviderRegistry registry;
  private final WorkerJobMetadata workerJobMetadata;

  @Inject
  WorkerImpl(
      CloudFactory cloudFactory,
      JobPollingService jobPollingService,
      ServiceProviderRegistry registry,
      WorkerJobMetadata workerJobMetadata) {
    this.cloudFactory = cloudFactory;
    this.jobPollingService = jobPollingService;
    this.store = cloudFactory.getJobStore();
    this.registry = registry;
    this.workerJobMetadata = workerJobMetadata;
  }

  void processJob() {
    // Start the polling service to poll for an unassigned job and when it's ready.
    pollForJob();

    // Start the processing
    UUID jobId = workerJobMetadata.getJobId();
    logger.debug("Begin processing jobId: {}", jobId);
    LegacyPortabilityJob job = store.find(jobId, JobAuthorization.State.CREDS_ENCRYPTED);

    // Only load the two providers that are doing actually work.
    // TODO(willard): Only load two needed services here, after converting service name to class
    // name in storage.

    processJob(jobId, job);
    logger.info("Successfully processed jobId: {}", workerJobMetadata.getJobId());
  }

  private void pollForJob() {
    jobPollingService.startAsync();
    jobPollingService.awaitTerminated();
  }

  private void processJob(UUID jobId, LegacyPortabilityJob job) {
    PortableDataType dataType = PortableDataType.valueOf(job.dataType());
    try {
      Crypter decrypter = CrypterFactory.create(workerJobMetadata.getKeyPair().getPrivate());
      String serializedExportAuthData = decrypter.decrypt(job.encryptedExportAuthData());
      AuthData exportAuthData = deSerialize(serializedExportAuthData);
      String serializedImportAuthData = decrypter.decrypt(job.encryptedImportAuthData());
      AuthData importAuthData = deSerialize(serializedImportAuthData);
      PortabilityCopier
          .copyDataType(registry, dataType, job.exportService(), exportAuthData,
              job.importService(), importAuthData, jobId);
    } catch (IOException e) {
      logger.error("Error processing jobId: {}" + jobId, e);
    } finally {
      cloudFactory.clearJobData(jobId);
    }
  }


  // TODO: Switch to using Jackson in the new transfer types
  private static AuthData deSerialize(String serialized) {
    return GSON.fromJson(serialized, AuthData.class);
  }
}

@@ -641,16 +641,61 @@
 orker;%0A%0A
+import com.google.common.base.Preconditions;%0A
 import c
@@ -1316,22 +1316,16 @@
 d.types.
-Legacy
 Portabil
@@ -2499,22 +2499,16 @@
 d);%0A    
-Legacy
 Portabil
@@ -2530,23 +2530,105 @@
 ore.find
+Job
 (jobId
-,
+);%0A    Preconditions.checkState(%0A        job.getJobAuthorization().getState() ==
  JobAuth
@@ -3119,14 +3119,8 @@
 Id, 
-Legacy
 Port
@@ -3198,17 +3198,28 @@
 eOf(job.
-d
+getTransferD
 ataType(
@@ -3322,24 +3322,93 @@
 Private());%0A
+      JobAuthorization jobAuthorization = job.getJobAuthorization();%0A
       String
@@ -3426,32 +3426,42 @@
 ExportAuthData =
+%0A         
  decrypter.decry
@@ -3458,34 +3458,50 @@
 pter.decrypt(job
-.e
+Authorization.getE
 ncryptedExportAu
@@ -3621,16 +3621,26 @@
 thData =
+%0A         
  decrypt
@@ -3653,18 +3653,34 @@
 rypt(job
-.e
+Authorization.getE
 ncrypted
@@ -3841,17 +3841,20 @@
 pe, job.
-e
+getE
 xportSer
@@ -3895,17 +3895,20 @@
     job.
-i
+getI
 mportSer

Migrate WorkerImpl to use PortabilityJob