add the ability to use a secret persistence

airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/ConfigRepository.java

@@ -25,6 +25,7 @@
 package io.airbyte.config.persistence;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import com.google.api.client.util.Preconditions;
 import io.airbyte.commons.lang.MoreBooleans;
 import io.airbyte.config.AirbyteConfig;
 import io.airbyte.config.ConfigSchema;
@@ -37,6 +38,9 @@
 import io.airbyte.config.StandardSync;
 import io.airbyte.config.StandardSyncOperation;
 import io.airbyte.config.StandardWorkspace;
+import io.airbyte.config.persistence.split_secrets.SecretPersistence;
+import io.airbyte.config.persistence.split_secrets.SecretsHelpers;
+import io.airbyte.config.persistence.split_secrets.SplitSecretConfig;
 import io.airbyte.protocol.models.ConnectorSpecification;
 import io.airbyte.validation.json.JsonSchemaValidator;
 import io.airbyte.validation.json.JsonValidationException;
@@ -52,9 +56,11 @@
 public class ConfigRepository {
 
   private final ConfigPersistence persistence;
+  private final Optional<SecretPersistence> secretPersistence;
 
-  public ConfigRepository(final ConfigPersistence persistence) {
+  public ConfigRepository(final ConfigPersistence persistence, final Optional<SecretPersistence> secretPersistence) {
     this.persistence = persistence;
+    this.secretPersistence = secretPersistence;
   }
 
   public StandardWorkspace getStandardWorkspace(final UUID workspaceId, final boolean includeTombstone)
@@ -167,19 +173,75 @@ public void writeStandardDestinationDefinition(final StandardDestinationDefiniti
         destinationDefinition);
   }
 
-  public SourceConnection getSourceConnection(final UUID sourceId) throws JsonValidationException, IOException, ConfigNotFoundException {
+  public SourceConnection getSourceConnection(final UUID sourceId) throws JsonValidationException, ConfigNotFoundException, IOException {
     return persistence.getConfig(ConfigSchema.SOURCE_CONNECTION, sourceId.toString(), SourceConnection.class);
   }
 
+  public SourceConnection getSourceConnectionWithSecrets(final UUID sourceId) throws JsonValidationException, IOException, ConfigNotFoundException {
+    final var source = getSourceConnection(sourceId);
+
+    if (secretPersistence.isPresent()) {
+      final var partialConfig = source.getConfiguration();
+      final var fullConfig = SecretsHelpers.combineConfig(partialConfig, secretPersistence.get());
+
+      source.setConfiguration(fullConfig);
+    }
+
+    return source;
+  }
+
+  private Optional<SourceConnection> getOptionalSourceConnection(final UUID sourceId) throws JsonValidationException, IOException {
+    try {
+      return Optional.of(getSourceConnection(sourceId));
+    } catch (ConfigNotFoundException e) {
+      return Optional.empty();
+    }
+  }
+
   public void writeSourceConnection(final SourceConnection source, final ConnectorSpecification connectorSpecification)
       throws JsonValidationException, IOException {
     // actual validation is only for sanity checking
     final JsonSchemaValidator validator = new JsonSchemaValidator();
     validator.ensure(connectorSpecification.getConnectionSpecification(), source.getConfiguration());
 
+    if (secretPersistence.isPresent()) {
+      final var previousSourceConnection = getOptionalSourceConnection(source.getSourceId());
+      final var splitConfig = getSplitSourceConfig(previousSourceConnection, source, connectorSpecification);
+
+      // todo: should figure out how to do this more transactionally with getSplitSourceConfig /
+      // getSplitDestinationConfig
+      splitConfig.getCoordinateToPayload().forEach(secretPersistence.get()::write);
+
+      // todo: can we do this better without editing the input object? SourceConnection doesn't have a
+      // deepCopy but we could reserialize
+      source.setConfiguration(splitConfig.getPartialConfig());
+    }
+
     persistence.writeConfig(ConfigSchema.SOURCE_CONNECTION, source.getSourceId().toString(), source);
   }
 
+  private SplitSecretConfig getSplitSourceConfig(final Optional<SourceConnection> previousSource,
+                                                 final SourceConnection source,
+                                                 final ConnectorSpecification connectorSpecification) {
+    Preconditions.checkArgument(secretPersistence.isPresent());
+
+    if (previousSource.isPresent()) {
+      return SecretsHelpers.splitAndUpdateConfig(
+          source.getWorkspaceId(),
+          previousSource.get().getConfiguration(),
+          source.getConfiguration(),
+          connectorSpecification,
+          secretPersistence.get()::read);
+    } else {
+      return SecretsHelpers.splitConfig(
+          source.getWorkspaceId(),
+          source.getConfiguration(),
+          connectorSpecification);
+    }
+  }
+
+  // todo: what should list source connection return? full or partial?
+  // probably partial if it isn't used for migrations
   public List<SourceConnection> listSourceConnection() throws JsonValidationException, IOException {
     return persistence.listConfigs(ConfigSchema.SOURCE_CONNECTION, SourceConnection.class);
   }
@@ -189,15 +251,71 @@ public DestinationConnection getDestinationConnection(final UUID destinationId)
     return persistence.getConfig(ConfigSchema.DESTINATION_CONNECTION, destinationId.toString(), DestinationConnection.class);
   }
 
-  public void writeDestinationConnection(final DestinationConnection destinationConnection, final ConnectorSpecification connectorSpecification)
+  public DestinationConnection getDestinationConnectionWithSecrets(final UUID destinationId)
+      throws JsonValidationException, IOException, ConfigNotFoundException {
+    final var destination = getDestinationConnection(destinationId);
+
+    if (secretPersistence.isPresent()) {
+      final var partialConfig = destination.getConfiguration();
+      final var fullConfig = SecretsHelpers.combineConfig(partialConfig, secretPersistence.get());
+
+      destination.setConfiguration(fullConfig);
+    }
+
+    return destination;
+
+  }
+
+  private Optional<DestinationConnection> getOptionalDestinationConnection(final UUID destinationId) throws JsonValidationException, IOException {
+    try {
+      return Optional.of(getDestinationConnection(destinationId));
+    } catch (ConfigNotFoundException e) {
+      return Optional.empty();
+    }
+  }
+
+  public void writeDestinationConnection(final DestinationConnection destination, final ConnectorSpecification connectorSpecification)
       throws JsonValidationException, IOException {
     // actual validation is only for sanity checking
     final JsonSchemaValidator validator = new JsonSchemaValidator();
-    validator.ensure(connectorSpecification.getConnectionSpecification(), destinationConnection.getConfiguration());
+    validator.ensure(connectorSpecification.getConnectionSpecification(), destination.getConfiguration());
+
+    if (secretPersistence.isPresent()) {
+      final var previousDestination = getOptionalDestinationConnection(destination.getDestinationId());
+      final var splitConfig = getSplitDestinationConfig(previousDestination, destination, connectorSpecification);
 
-    persistence.writeConfig(ConfigSchema.DESTINATION_CONNECTION, destinationConnection.getDestinationId().toString(), destinationConnection);
+      splitConfig.getCoordinateToPayload().forEach(secretPersistence.get()::write);
+
+      // todo: can we do this better without editing the input object? DestinationCOnnection doesn't have
+      // a deepCopy but we could reserialize
+      destination.setConfiguration(splitConfig.getPartialConfig());
+    }
+
+    persistence.writeConfig(ConfigSchema.DESTINATION_CONNECTION, destination.getDestinationId().toString(), destination);
+  }
+
+  private SplitSecretConfig getSplitDestinationConfig(final Optional<DestinationConnection> previousDestination,
+                                                      final DestinationConnection destination,
+                                                      final ConnectorSpecification connectorSpecification) {
+    Preconditions.checkArgument(secretPersistence.isPresent());
+
+    if (previousDestination.isPresent()) {
+      return SecretsHelpers.splitAndUpdateConfig(
+          destination.getWorkspaceId(),
+          previousDestination.get().getConfiguration(),
+          destination.getConfiguration(),
+          connectorSpecification,
+          secretPersistence.get()::read);
+    } else {
+      return SecretsHelpers.splitConfig(
+          destination.getWorkspaceId(),
+          destination.getConfiguration(),
+          connectorSpecification);
+    }
   }
 
+  // todo: what should list source connection return? full or partial?
+  // probably partial if it isn't used for migrations
   public List<DestinationConnection> listDestinationConnection() throws JsonValidationException, IOException {
     return persistence.listConfigs(ConfigSchema.DESTINATION_CONNECTION, DestinationConnection.class);
   }
@@ -276,14 +394,17 @@ public List<DestinationOAuthParameter> listDestinationOAuthParam() throws JsonVa
     return persistence.listConfigs(ConfigSchema.DESTINATION_OAUTH_PARAM, DestinationOAuthParameter.class);
   }
 
+  // todo: support here
   public void replaceAllConfigs(final Map<AirbyteConfig, Stream<?>> configs, final boolean dryRun) throws IOException {
     persistence.replaceAllConfigs(configs, dryRun);
   }
 
+  // todo: support here
   public Map<String, Stream<JsonNode>> dumpConfigs() throws IOException {
     return persistence.dumpConfigs();
   }
 
+  // todo: support here
   public void loadData(ConfigPersistence seedPersistence) throws IOException {
     persistence.loadData(seedPersistence);
   }