feat(amazon-photos): harden Photos importer and add shared import helper (#1512)

## Summary

Refactors the Amazon Photos importer for reliability and testability.
Introduces typed API error handling, structured failure classification,
platform-level retry integration, and multipart upload with robust
completion polling. Client lifecycle and shared import utilities are
extracted into a new `AmazonImportHelper`.

> **Note:** This PR has been reviewed internally already.

This is a behavior-preserving refactor for existing flows, plus
hardening for transient failures and large uploads.

## What's changed

### Error handling

- New `AmazonPhotosApiException` carrying HTTP status + service
`errorCode`. The client parses the error body and throws the typed
exception, logging only `errorCode`/`message` (never the raw response
body).
- `UploadErrorCodes` centralizes the client-facing codes we classify on:
  - Duplicates (`DuplicatesConflictError`) are skipped idempotently.
- Storage-quota conditions (`InsufficientStorage`,
`NoActiveSubscriptionFound`) map to a terminal
`DestinationMemoryFullException`.

### Retry integration

- Wires the platform `RetryingIdempotentImportExecutor` (opt-in via the
`enableRetrying` setting) so transient failures are retried per the host
`RetryStrategyLibrary`, consistent with other providers.

### Multipart upload

- Initiate → upload parts → complete, followed by completion polling
that:
  - fails fast on a definitive 4xx,
- retries transient 5xx/429/malformed responses until the budget is
exhausted, and
  - guards against a missing status field.
- `RetryUtils` backoff is exponential-with-jitter; explicit
connect/read/write timeouts on the OkHttp client.

### Shared `AmazonImportHelper`

- Owns per-job client creation keyed by `jobId` (bounded access-order
LRU), resolving endpoints once per job. Client build + endpoint
resolution happen outside the lock so one job's latency can't stall
others.
- Provides MD5/streamed download (untrusted `dataId` is sanitized for
the temp-file prefix) and the shared error classification used by the
importer.
- Endpoint resolution now requires `uploadServiceUrl` (dropped the
`contentUrl` fallback).

## Testing

Unit tests added/updated across:

- **Helper** — per-job client isolation, error classification.
- **Client** — multipart poll success / fail-fast / transient-retry /
budget-exhausted, error extraction, configured timeouts, lazy endpoint
resolution.
- **Importer** — retry-executor selection, quota → terminal failure,
duplicate skip, upload happy paths.

Completed transfers and validated.

## Notes

- `okhttp`/`mockwebserver` now use the shared `${okHttpVersion}` instead
of a hardcoded version.

Author: vipulkala95

extensions/data-transfer/portability-data-transfer-amazon/build.gradle

@@ -30,7 +30,7 @@ dependencies {
     implementation "com.fasterxml.jackson.core:jackson-annotations:${jacksonVersion}"
     implementation "com.fasterxml.jackson.core:jackson-databind:${jacksonVersion}"
     implementation "com.google.guava:guava:${guavaVersion}"
-    implementation "com.squareup.okhttp3:okhttp:4.9.0"
+    implementation "com.squareup.okhttp3:okhttp:${okHttpVersion}"
 
-    testImplementation "com.squareup.okhttp3:mockwebserver:4.9.0"
+    testImplementation "com.squareup.okhttp3:mockwebserver:${okHttpVersion}"
 }

extensions/data-transfer/portability-data-transfer-amazon/src/main/java/org/datatransferproject/transfer/amazon/AmazonTransferExtension.java

@@ -22,6 +22,8 @@
 import org.datatransferproject.spi.cloud.storage.AppCredentialStore;
 import org.datatransferproject.spi.cloud.storage.TemporaryPerJobDataStore;
 import org.datatransferproject.spi.transfer.extension.TransferExtension;
+import org.datatransferproject.spi.transfer.idempotentexecutor.IdempotentImportExecutor;
+import org.datatransferproject.spi.transfer.idempotentexecutor.IdempotentImportExecutorExtension;
 import org.datatransferproject.spi.transfer.provider.Exporter;
 import org.datatransferproject.spi.transfer.provider.Importer;
 import org.datatransferproject.transfer.amazon.photos.AmazonPhotosImporter;
@@ -34,7 +36,7 @@ public class AmazonTransferExtension implements TransferExtension {
 
   private static final String SERVICE_ID = "Amazon";
 
-  private AmazonPhotosImporter importer;
+  private AmazonPhotosImporter photosImporter;
   private volatile boolean initialized = false;
 
   @Override
@@ -51,8 +53,10 @@ public String getServiceId() {
   @Override
   public Importer<?, ?> getImporter(DataVertical transferDataType) {
     Preconditions.checkArgument(initialized, "Extension not initialized");
-    Preconditions.checkArgument(transferDataType == DataVertical.PHOTOS);
-    return importer;
+    if (transferDataType == DataVertical.PHOTOS) {
+      return photosImporter;
+    }
+    throw new IllegalArgumentException("Unsupported data type: " + transferDataType);
   }
 
   @Override
@@ -70,9 +74,15 @@ public synchronized void initialize(ExtensionContext context) {
       return;
     }
 
-    importer = new AmazonPhotosImporter(
+    IdempotentImportExecutor retryingIdempotentExecutor =
+        context.getService(IdempotentImportExecutorExtension.class)
+            .getRetryingIdempotentImportExecutor(context);
+    boolean enableRetrying = context.getSetting("enableRetrying", false);
+
+    photosImporter = new AmazonPhotosImporter(
         monitor, appCredentials.getKey(), appCredentials.getSecret(),
-        context.getService(TemporaryPerJobDataStore.class));
+        context.getService(TemporaryPerJobDataStore.class),
+        retryingIdempotentExecutor, enableRetrying);
 
     initialized = true;
   }

extensions/data-transfer/portability-data-transfer-amazon/src/main/java/org/datatransferproject/transfer/amazon/photos/AmazonImportHelper.java

@@ -0,0 +1,183 @@
+/*
+ * Copyright 2026 The Data Transfer 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.datatransferproject.transfer.amazon.photos;
+
+import org.datatransferproject.spi.cloud.connection.ConnectionProvider;
+import org.datatransferproject.spi.cloud.storage.TemporaryPerJobDataStore;
+import org.datatransferproject.api.launcher.Monitor;
+import org.datatransferproject.spi.transfer.idempotentexecutor.IdempotentImportExecutor;
+import org.datatransferproject.types.common.DownloadableItem;
+import org.datatransferproject.types.transfer.auth.TokensAndUrlAuthData;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.security.DigestInputStream;
+import java.security.MessageDigest;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+
+/**
+ * Shared helper for Amazon Photos/Videos importers.
+ * Holds common collaborators and provides reusable import operations.
+ */
+class AmazonImportHelper {
+
+  static final String IMPORTED_SUFFIX = " - Imported from ";
+
+  private final TemporaryPerJobDataStore dataStore;
+  private final ConnectionProvider connectionProvider;
+  private final String clientId;
+  private final String clientSecret;
+  private final Monitor monitor;
+  // Test seam: when set it is always returned in place of a real client.
+  private final AmazonPhotosInterface injectedClient;
+  // Safeguard to avoid sharing a client across jobs: keying by jobId keeps each job's credentials
+  // separate. Bounded, access-order LRU so a long-lived (multi-job) process can't accumulate
+  // clients/tokens without limit; an evicted job just rebuilds its client on its next chunk.
+  // Guarded by getOrCreateClient (synchronized).
+  private static final int MAX_CACHED_CLIENTS = 1000;
+  private final Map<UUID, AmazonPhotosInterface> clientsByJob =
+      new LinkedHashMap<UUID, AmazonPhotosInterface>(16, 0.75f, true) {
+        @Override
+        protected boolean removeEldestEntry(Map.Entry<UUID, AmazonPhotosInterface> eldest) {
+          return size() > MAX_CACHED_CLIENTS;
+        }
+      };
+
+  /** Classification/download-only helper (no client provisioning); used where no client is built. */
+  AmazonImportHelper(TemporaryPerJobDataStore dataStore) {
+    this(dataStore, null, null, null, null);
+  }
+
+  /** Production helper that builds a real Amazon client per job from the given app credentials. */
+  AmazonImportHelper(TemporaryPerJobDataStore dataStore, String clientId, String clientSecret,
+                     Monitor monitor) {
+    this(dataStore, clientId, clientSecret, monitor, null);
+  }
+
+  /** Test helper: {@code injectedClient} is always returned by {@link #getOrCreateClient}. */
+  AmazonImportHelper(TemporaryPerJobDataStore dataStore, AmazonPhotosInterface injectedClient) {
+    this(dataStore, null, null, null, injectedClient);
+  }
+
+  private AmazonImportHelper(TemporaryPerJobDataStore dataStore, String clientId,
+                             String clientSecret, Monitor monitor,
+                             AmazonPhotosInterface injectedClient) {
+    this.dataStore = dataStore;
+    this.connectionProvider = new ConnectionProvider(dataStore);
+    this.clientId = clientId;
+    this.clientSecret = clientSecret;
+    this.monitor = monitor;
+    this.injectedClient = injectedClient;
+  }
+
+  /**
+   * Returns the client for this job, creating and caching it (and resolving endpoints) on first
+   * use. Injected clients are returned as-is; per-job caching keeps each job's credentials
+   * isolated and avoids re-resolving endpoints on every chunk of the same job.
+   */
+  AmazonPhotosInterface getOrCreateClient(UUID jobId, TokensAndUrlAuthData authData)
+      throws IOException {
+    if (injectedClient != null) {
+      return injectedClient;
+    }
+    synchronized (this) {
+      AmazonPhotosInterface existing = clientsByJob.get(jobId);
+      if (existing != null) {
+        return existing;
+      }
+    }
+    // Build and resolve endpoints outside the lock: resolveEndpoints() is a network call, so
+    // holding the shared lock across it would let one job's latency stall getOrCreateClient for
+    // every other concurrent job. Two concurrent first-chunks of the same new job may each build
+    // once; putIfAbsent dedupes storage and the loser is discarded.
+    AmazonPhotosInterface created = createClient(authData);
+    created.resolveEndpoints();
+    synchronized (this) {
+      AmazonPhotosInterface race = clientsByJob.putIfAbsent(jobId, created);
+      return race != null ? race : created;
+    }
+  }
+
+  // Visible-for-testing seam: endpoint resolution otherwise requires a live network call, so
+  // tests override this to inject a fake client.
+  AmazonPhotosInterface createClient(TokensAndUrlAuthData authData) {
+    return new AmazonPhotosClient(
+        AmazonPhotosClient.createDefaultHttpClient(),
+        authData.getAccessToken(), authData.getRefreshToken(), clientId, clientSecret, monitor);
+  }
+
+  /** Creates a new MD5 MessageDigest instance. */
+  MessageDigest newMd5Digest() {
+    return Md5Utils.newDigest();
+  }
+
+  /** Converts a byte array to its lowercase hex string representation. */
+  String toHexString(byte[] bytes) {
+    return Md5Utils.toHexString(bytes);
+  }
+
+  /** Whether the API error indicates the item already exists (duplicate). */
+  boolean isDuplicate(AmazonPhotosApiException e) {
+    return e.isErrorCode(UploadErrorCodes.DUPLICATES_CONFLICT_ERROR);
+  }
+
+  /** Whether the API error indicates the destination storage quota is exceeded. */
+  boolean isStorageQuotaExceeded(AmazonPhotosApiException e) {
+    return e.isErrorCode(UploadErrorCodes.INSUFFICIENT_STORAGE)
+        || e.isErrorCode(UploadErrorCodes.NO_ACTIVE_SUBSCRIPTION_FOUND);
+  }
+
+  /**
+   * Downloads a content item to a temp file, computing MD5 in a single pass.
+   *
+   * <p>The provider-supplied {@code dataId} is untrusted filesystem input, so path separators are
+   * stripped before using it as the temp-file prefix — otherwise {@code Files.createTempFile}
+   * rejects it with an IllegalArgumentException. Only the local prefix is affected; the content
+   * sent to Amazon is unchanged.
+   */
+  File downloadToTempFile(UUID jobId, DownloadableItem item, String dataId,
+                          MessageDigest md5) throws IOException {
+    String prefix = dataId.replaceAll("[/\\\\]", "_");
+    try (InputStream raw = connectionProvider.getInputStreamForItem(jobId, item).getStream();
+         DigestInputStream dis = new DigestInputStream(raw, md5)) {
+      return dataStore.getTempFileFromInputStream(dis, prefix, ".tmp");
+    }
+  }
+
+  /**
+   * Resolves the target album ID from the executor cache.
+   * If albumId is provided but not cached (album creation failed), throws to prevent
+   * silent data loss — photos/videos should not be uploaded without their album.
+   */
+  String resolveTargetAlbumId(String albumId, IdempotentImportExecutor executor)
+      throws Exception {
+    if (albumId == null) {
+      return null;
+    }
+    // This will throw if album creation failed, which marks the item as failed too,
+    // preventing silent loss of album organization.
+    return executor.getCachedValue(albumId);
+  }
+
+  /** Removes temp data from the job store. */
+  void cleanupTempData(UUID jobId, String fetchableUrl) throws IOException {
+    dataStore.removeData(jobId, fetchableUrl);
+  }
+}

extensions/data-transfer/portability-data-transfer-amazon/src/main/java/org/datatransferproject/transfer/amazon/photos/AmazonPhotosApiException.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2026 The Data Transfer 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.datatransferproject.transfer.amazon.photos;
+
+import java.io.IOException;
+
+/**
+ * Typed exception for non-success Amazon Photos / Upload Service API responses.
+ *
+ * <p>Carries the HTTP status and the service {@code errorCode} (see {@link UploadErrorCodes})
+ * so callers can classify failures on a structured field rather than by substring-matching a
+ * raw response body. Extends {@link IOException} so it flows through the existing DTP
+ * import error handling (e.g. {@code executeAndSwallowIOExceptions}).
+ */
+public class AmazonPhotosApiException extends IOException {
+
+  private final int httpStatus;
+  private final String errorCode;
+
+  /**
+   * @param httpStatus the HTTP status code of the response
+   * @param errorCode the service errorCode from the response body, or {@code null} if absent
+   * @param message a human-readable message (must not embed sensitive raw response detail)
+   */
+  public AmazonPhotosApiException(int httpStatus, String errorCode, String message) {
+    super(message);
+    this.httpStatus = httpStatus;
+    this.errorCode = errorCode;
+  }
+
+  public int getHttpStatus() {
+    return httpStatus;
+  }
+
+  /** The service errorCode, or {@code null} if the response carried none. */
+  public String getErrorCode() {
+    return errorCode;
+  }
+
+  /** Whether this error's {@code errorCode} equals the given code. */
+  public boolean isErrorCode(String code) {
+    return code != null && code.equals(errorCode);
+  }
+}