Add a config for instance wide oauth parameters

airbyte-api/src/main/openapi/config.yaml

@@ -1364,48 +1364,6 @@ paths:
                 $ref: "#/components/schemas/WebBackendConnectionRead"
         "422":
           $ref: "#/components/responses/InvalidInputResponse"
-  /v1/web_backend/sources/recreate:
-    post:
-      tags:
-        - web_backend
-      summary: Recreate a source
-      operationId: webBackendRecreateSource
-      requestBody:
-        content:
-          application/json:
-            schema:
-              $ref: "#/components/schemas/SourceRecreate"
-        required: true
-      responses:
-        "200":
-          description: Successful operation
-          content:
-            application/json:
-              schema:
-                $ref: "#/components/schemas/SourceRead"
-        "422":
-          $ref: "#/components/responses/InvalidInputResponse"
-  /v1/web_backend/destinations/recreate:
-    post:
-      tags:
-        - web_backend
-      summary: Recreate a destination
-      operationId: webBackendRecreateDestination
-      requestBody:
-        content:
-          application/json:
-            schema:
-              $ref: "#/components/schemas/DestinationRecreate"
-        required: true
-      responses:
-        "200":
-          description: Successful operation
-          content:
-            application/json:
-              schema:
-                $ref: "#/components/schemas/DestinationRead"
-        "422":
-          $ref: "#/components/responses/InvalidInputResponse"
   /v1/jobs/list:
     post:
       tags:

airbyte-config/models/src/main/java/io/airbyte/config/ConfigSchema.java

@@ -66,6 +66,14 @@ public enum ConfigSchema implements AirbyteConfig {
       StandardSyncOperation.class,
       standardSyncOperation -> standardSyncOperation.getOperationId().toString(),
       "operationId"),
+
+  SOURCE_OAUTH_PARAM("SourceOAuthParameter.yaml", SourceOAuthParameter.class,
+      sourceOAuthParameter -> sourceOAuthParameter.getOauthParameterId().toString(),
+      "oauthParameterId"),
+  DESTINATION_OAUTH_PARAM("DestinationOAuthParameter.yaml", DestinationOAuthParameter.class,
+      destinationOAuthParameter -> destinationOAuthParameter.getOauthParameterId().toString(),
+      "oauthParameterId"),
+
   STANDARD_SYNC_SUMMARY("StandardSyncSummary.yaml", StandardSyncSummary.class),
 
   // worker

airbyte-config/models/src/main/resources/types/DestinationOAuthParameter.yaml

@@ -0,0 +1,25 @@
+---
+"$schema": http://json-schema.org/draft-07/schema#
+"$id": https://github.com/airbytehq/airbyte/blob/master/airbyte-config/models/src/main/resources/types/DestinationOAuthParameter.yaml
+title: DestinationOAuthParameter
+description: OAuth parameters used when connecting to destination
+type: object
+required:
+  - oauthParameterId
+  - destinationDefinitionId
+  - configuration
+additionalProperties: false
+properties:
+  oauthParameterId:
+    type: string
+    format: uuid
+  destinationDefinitionId:
+    type: string
+    format: uuid
+  workspaceId:
+    type: string
+    format: uuid
+  configuration:
+    description: Integration specific blob. Must be a valid JSON string.
+    type: object
+    existingJavaType: com.fasterxml.jackson.databind.JsonNode

airbyte-config/models/src/main/resources/types/SourceOAuthParameter.yaml

@@ -0,0 +1,25 @@
+---
+"$schema": http://json-schema.org/draft-07/schema#
+"$id": https://github.com/airbytehq/airbyte/blob/master/airbyte-config/models/src/main/resources/types/SourceOAuthParameter.yaml
+title: SourceOAuthParameter
+description: OAuth parameters used when connecting to source
+type: object
+required:
+  - oauthParameterId
+  - sourceDefinitionId
+  - configuration
+additionalProperties: false
+properties:
+  oauthParameterId:
+    type: string
+    format: uuid
+  sourceDefinitionId:
+    type: string
+    format: uuid
+  workspaceId:
+    type: string
+    format: uuid
+  configuration:
+    description: Integration specific blob. Must be a valid JSON string.
+    type: object
+    existingJavaType: com.fasterxml.jackson.databind.JsonNode

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

@@ -29,7 +29,9 @@
 import io.airbyte.config.AirbyteConfig;
 import io.airbyte.config.ConfigSchema;
 import io.airbyte.config.DestinationConnection;
+import io.airbyte.config.DestinationOAuthParameter;
 import io.airbyte.config.SourceConnection;
+import io.airbyte.config.SourceOAuthParameter;
 import io.airbyte.config.StandardDestinationDefinition;
 import io.airbyte.config.StandardSourceDefinition;
 import io.airbyte.config.StandardSync;
@@ -211,6 +213,33 @@ public List<StandardSyncOperation> listStandardSyncOperations() throws IOExcepti
     return persistence.listConfigs(ConfigSchema.STANDARD_SYNC_OPERATION, StandardSyncOperation.class);
   }
 
+  public SourceOAuthParameter getSourceOAuthParams(final UUID SourceOAuthParameterId)
+      throws JsonValidationException, IOException, ConfigNotFoundException {
+    return persistence.getConfig(ConfigSchema.SOURCE_OAUTH_PARAM, SourceOAuthParameterId.toString(), SourceOAuthParameter.class);
+  }
+
+  public void writeSourceOAuthParam(final SourceOAuthParameter SourceOAuthParameter) throws JsonValidationException, IOException {
+    persistence.writeConfig(ConfigSchema.SOURCE_OAUTH_PARAM, SourceOAuthParameter.getOauthParameterId().toString(), SourceOAuthParameter);
+  }
+
+  public List<SourceOAuthParameter> listSourceOAuthParam() throws JsonValidationException, IOException {
+    return persistence.listConfigs(ConfigSchema.SOURCE_OAUTH_PARAM, SourceOAuthParameter.class);
+  }
+
+  public DestinationOAuthParameter getDestinationOAuthParams(final UUID destinationOAuthParameterId)
+      throws JsonValidationException, IOException, ConfigNotFoundException {
+    return persistence.getConfig(ConfigSchema.DESTINATION_OAUTH_PARAM, destinationOAuthParameterId.toString(), DestinationOAuthParameter.class);
+  }
+
+  public void writeDestinationOAuthParam(final DestinationOAuthParameter destinationOAuthParameter) throws JsonValidationException, IOException {
+    persistence.writeConfig(ConfigSchema.DESTINATION_OAUTH_PARAM, destinationOAuthParameter.getOauthParameterId().toString(),
+        destinationOAuthParameter);
+  }
+
+  public List<DestinationOAuthParameter> listDestinationOAuthParam() throws JsonValidationException, IOException {
+    return persistence.listConfigs(ConfigSchema.DESTINATION_OAUTH_PARAM, DestinationOAuthParameter.class);
+  }
+
   public <T> void replaceAllConfigs(final Map<AirbyteConfig, Stream<T>> configs, final boolean dryRun) throws IOException {
     persistence.replaceAllConfigs(configs, dryRun);
   }

airbyte-scheduler/persistence/src/main/java/io/airbyte/scheduler/persistence/job_factory/DefaultSyncJobFactory.java

@@ -44,20 +44,29 @@ public class DefaultSyncJobFactory implements SyncJobFactory {
 
   private final DefaultJobCreator jobCreator;
   private final ConfigRepository configRepository;
+  private final OAuthConfigSupplier oAuthConfigSupplier;
 
   public DefaultSyncJobFactory(final DefaultJobCreator jobCreator,
                                final ConfigRepository configRepository) {
 
     this.jobCreator = jobCreator;
     this.configRepository = configRepository;
+    this.oAuthConfigSupplier = new OAuthConfigSupplier(configRepository);
   }
 
   public Long create(final UUID connectionId) {
     try {
       final StandardSync standardSync = configRepository.getStandardSync(connectionId);
       final SourceConnection sourceConnection = configRepository.getSourceConnection(standardSync.getSourceId());
+      sourceConnection.withConfiguration(oAuthConfigSupplier.injectSourceOAuthParameters(
+          sourceConnection.getSourceDefinitionId(),
+          sourceConnection.getWorkspaceId(),
+          sourceConnection.getConfiguration()));
       final DestinationConnection destinationConnection = configRepository.getDestinationConnection(standardSync.getDestinationId());
-
+      destinationConnection.withConfiguration(oAuthConfigSupplier.injectDestinationOAuthParameters(
+          destinationConnection.getDestinationDefinitionId(),
+          destinationConnection.getWorkspaceId(),
+          destinationConnection.getConfiguration()));
       final StandardSourceDefinition sourceDefinition = configRepository.getStandardSourceDefinition(sourceConnection.getSourceDefinitionId());
       final StandardDestinationDefinition destinationDefinition =
           configRepository.getStandardDestinationDefinition(destinationConnection.getDestinationDefinitionId());

airbyte-scheduler/persistence/src/main/java/io/airbyte/scheduler/persistence/job_factory/OAuthConfigSupplier.java

@@ -0,0 +1,77 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Airbyte
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.airbyte.scheduler.persistence.job_factory;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.config.DestinationOAuthParameter;
+import io.airbyte.config.SourceOAuthParameter;
+import io.airbyte.config.persistence.ConfigRepository;
+import io.airbyte.validation.json.JsonValidationException;
+import java.io.IOException;
+import java.util.Comparator;
+import java.util.UUID;
+
+public class OAuthConfigSupplier {
+
+  final private ConfigRepository configRepository;
+
+  public OAuthConfigSupplier(ConfigRepository configRepository) {
+    this.configRepository = configRepository;
+  }
+
+  public JsonNode injectSourceOAuthParameters(UUID sourceDefinitionId, UUID workspaceId, JsonNode sourceConnectorConfig)
+      throws JsonValidationException, IOException {
+    configRepository.listSourceOAuthParam().stream()
+        .filter(p -> sourceDefinitionId.equals(p.getSourceDefinitionId()))
+        .filter(p -> p.getWorkspaceId() == null || workspaceId.equals(p.getWorkspaceId()))
+        // we prefer params specific to a workspace before global ones (ie workspace is null)
+        .min(Comparator.comparing(SourceOAuthParameter::getWorkspaceId, Comparator.nullsLast(Comparator.naturalOrder()))
+            .thenComparing(SourceOAuthParameter::getOauthParameterId))
+        .ifPresent(sourceOAuthParameter -> injectJsonNode((ObjectNode) sourceConnectorConfig, (ObjectNode) sourceOAuthParameter.getConfiguration()));
+    return sourceConnectorConfig;
+  }
+
+  public JsonNode injectDestinationOAuthParameters(UUID destinationDefinitionId, UUID workspaceId, JsonNode destinationConnectorConfig)
+      throws JsonValidationException, IOException {
+    configRepository.listDestinationOAuthParam().stream()
+        .filter(p -> destinationDefinitionId.equals(p.getDestinationDefinitionId()))
+        .filter(p -> p.getWorkspaceId() == null || workspaceId.equals(p.getWorkspaceId()))
+        // we prefer params specific to a workspace before global ones (ie workspace is null)
+        .min(Comparator.comparing(DestinationOAuthParameter::getWorkspaceId, Comparator.nullsLast(Comparator.naturalOrder()))
+            .thenComparing(DestinationOAuthParameter::getOauthParameterId))
+        .ifPresent(destinationOAuthParameter -> injectJsonNode((ObjectNode) destinationConnectorConfig,
+            (ObjectNode) destinationOAuthParameter.getConfiguration()));
+    return destinationConnectorConfig;
+  }
+
+  private static void injectJsonNode(ObjectNode config, ObjectNode fromConfig) {
+    for (String key : Jsons.keys(fromConfig)) {
+      config.set(key, fromConfig.get(key));
+    }
+  }
+
+}