Introduce webhook configs into workspace api and persistence

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

@@ -2280,6 +2280,22 @@ components:
           type: boolean
         defaultGeography:
           $ref: "#/components/schemas/Geography"
+        webhookConfigs:
+          type: array
+          items:
+            $ref: "#/components/schemas/WebhookConfigWrite"
+    WebhookConfigWrite:
+      type: object
+      properties:
+        name:
+          type: string
+          description: human readable name for this webhook e.g. for UI display.
+        authToken:
+          type: string
+          description: an auth token, to be passed as the value for an HTTP Authorization header.
+        validationUrl:
+          type: string
+          description: if supplied, the webhook config will be validated by checking that this URL returns a 2xx response.
     Notification:
       type: object
       required:
@@ -2384,6 +2400,23 @@ components:
           type: boolean
         defaultGeography:
           $ref: "#/components/schemas/Geography"
+        webhookConfigs:
+          type: array
+          items:
+            # Note: this omits any sensitive info e.g. auth token
+            $ref: "#/components/schemas/WebhookConfigRead"
+    WebhookConfigRead:
+      type: object
+      description: the readable info for a webhook config; omits sensitive info e.g. auth token
+      required:
+        - id
+      properties:
+        id:
+          type: string
+          format: uuid
+        name:
+          type: string
+          description: human-readable name e.g. for display in UI
     WorkspaceUpdateName:
       type: object
       required:

airbyte-commons/src/main/java/io/airbyte/commons/json/Jsons.java

@@ -16,10 +16,12 @@
 import com.fasterxml.jackson.databind.ObjectWriter;
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
 import com.google.common.base.Charsets;
 import com.google.common.base.Preconditions;
 import io.airbyte.commons.jackson.MoreMappers;
 import io.airbyte.commons.stream.MoreStreams;
+import java.io.File;
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collections;
@@ -39,6 +41,8 @@ public class Jsons {
 
   // Object Mapper is thread-safe
   private static final ObjectMapper OBJECT_MAPPER = MoreMappers.initMapper();
+
+  private static final ObjectMapper YAML_OBJECT_MAPPER = MoreMappers.initYamlMapper(new YAMLFactory());
   private static final ObjectWriter OBJECT_WRITER = OBJECT_MAPPER.writer(new JsonPrettyPrinter());
 
   public static <T> String serialize(final T object) {
@@ -89,6 +93,10 @@ public static <T> JsonNode jsonNode(final T object) {
     return OBJECT_MAPPER.valueToTree(object);
   }
 
+  public static JsonNode jsonNodeFromFile(final File file) throws IOException {
+    return YAML_OBJECT_MAPPER.readTree(file);
+  }
+
   public static JsonNode emptyObject() {
     return jsonNode(Collections.emptyMap());
   }

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

@@ -23,6 +23,9 @@ public enum ConfigSchema implements AirbyteConfig {
       workspaceServiceAccount -> workspaceServiceAccount.getWorkspaceId().toString(),
       "workspaceId"),
 
+  WORKSPACE_WEBHOOK_OPERATION_CONFIGS("WebhookOperationConfigs.yaml",
+      WebhookOperationConfigs.class),
+
   // source
   STANDARD_SOURCE_DEFINITION("StandardSourceDefinition.yaml",
       StandardSourceDefinition.class,

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

@@ -50,3 +50,9 @@ properties:
     type: boolean
   defaultGeography:
     "$ref": Geography.yaml
+  webhookOperationConfigs:
+    description:
+      Configurations for webhooks operations, stored as a JSON object so we can replace sensitive info with
+      coordinates in the secrets manager. Must conform to WebhookOperationConfigs.yaml.
+    type: object
+    existingJavaType: com.fasterxml.jackson.databind.JsonNode

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

@@ -0,0 +1,29 @@
+---
+"$schema": http://json-schema.org/draft-07/schema#
+"$id": https://github.com/airbytehq/airbyte/blob/master/airbyte-config/models/src/main/resources/types/WebhookOperationConfigs.yaml
+title: WebhookOperationConfigs
+description: List of configurations for webhook operations
+additionalProperties: false
+# NOTE: we have an extra layer of object nesting because the generator has some weird behavior with arrays.
+# See https://github.com/OpenAPITools/openapi-generator/issues/7802.
+type: object
+properties:
+  webhookConfigs:
+    type: array
+    items:
+      type: object
+      required:
+        - id
+        - name
+        - authToken
+      properties:
+        id:
+          type: string
+          format: uuid
+        name:
+          type: string
+          description: human readable name for this webhook e.g., for UI display
+        authToken:
+          type: string
+          airbyte_secret: true
+          description: An auth token, to be passed as the value for an HTTP Authorization header. Note - must include prefix such as "Bearer <credential>".

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

@@ -766,6 +766,8 @@ private void writeStandardWorkspace(final List<StandardWorkspace> configs, final
             .set(WORKSPACE.FIRST_SYNC_COMPLETE, standardWorkspace.getFirstCompletedSync())
             .set(WORKSPACE.FEEDBACK_COMPLETE, standardWorkspace.getFeedbackDone())
             .set(WORKSPACE.UPDATED_AT, timestamp)
+            .set(WORKSPACE.WEBHOOK_OPERATION_CONFIGS, standardWorkspace.getWebhookOperationConfigs() == null ? null
+                : JSONB.valueOf(Jsons.serialize(standardWorkspace.getWebhookOperationConfigs())))
             .where(WORKSPACE.ID.eq(standardWorkspace.getWorkspaceId()))
             .execute();
       } else {
@@ -786,6 +788,8 @@ private void writeStandardWorkspace(final List<StandardWorkspace> configs, final
             .set(WORKSPACE.FEEDBACK_COMPLETE, standardWorkspace.getFeedbackDone())
             .set(WORKSPACE.CREATED_AT, timestamp)
             .set(WORKSPACE.UPDATED_AT, timestamp)
+            .set(WORKSPACE.WEBHOOK_OPERATION_CONFIGS, standardWorkspace.getWebhookOperationConfigs() == null ? null
+                : JSONB.valueOf(Jsons.serialize(standardWorkspace.getWebhookOperationConfigs())))
             .execute();
       }
     });

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

@@ -111,7 +111,8 @@ public WorkspaceServiceAccount getWorkspaceServiceAccountWithSecrets(final UUID
   public StandardWorkspace getWorkspaceWithSecrets(final UUID workspaceId, final boolean includeTombstone)
       throws JsonValidationException, ConfigNotFoundException, IOException {
     final StandardWorkspace workspace = configRepository.getStandardWorkspaceNoSecrets(workspaceId, includeTombstone);
-    // TODO: hydrate any secrets once they're introduced.
+    final JsonNode webhookConfigs = secretsHydrator.hydrate(workspace.getWebhookOperationConfigs());
+    workspace.withWebhookOperationConfigs(webhookConfigs);
     return workspace;
   }
 

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

@@ -83,7 +83,7 @@ public void writeSourceConnection(final SourceConnection source, final Connector
         source.getWorkspaceId(),
         previousSourceConnection,
         source.getConfiguration(),
-        connectorSpecification,
+        connectorSpecification.getConnectionSpecification(),
         source.getTombstone() == null || !source.getTombstone());
     final SourceConnection partialSource = Jsons.clone(source).withConfiguration(partialConfig);
 
@@ -107,7 +107,7 @@ public void writeDestinationConnection(final DestinationConnection destination,
         destination.getWorkspaceId(),
         previousDestinationConnection,
         destination.getConfiguration(),
-        connectorSpecification,
+        connectorSpecification.getConnectionSpecification(),
         destination.getTombstone() == null || !destination.getTombstone());
     final DestinationConnection partialDestination = Jsons.clone(destination).withConfiguration(partialConfig);
 
@@ -146,11 +146,11 @@ private JsonNode statefulSplitSecrets(final UUID workspaceId, final JsonNode ful
   private JsonNode statefulUpdateSecrets(final UUID workspaceId,
                                          final Optional<JsonNode> oldConfig,
                                          final JsonNode fullConfig,
-                                         final ConnectorSpecification spec,
+                                         final JsonNode spec,
                                          final boolean validate)
       throws JsonValidationException {
     if (validate) {
-      validator.ensure(spec.getConnectionSpecification(), fullConfig);
+      validator.ensure(spec, fullConfig);
     }
 
     if (longLivedSecretPersistence.isEmpty()) {
@@ -163,13 +163,13 @@ private JsonNode statefulUpdateSecrets(final UUID workspaceId,
           workspaceId,
           oldConfig.get(),
           fullConfig,
-          spec.getConnectionSpecification(),
+          spec,
           longLivedSecretPersistence.get());
     } else {
       splitSecretConfig = SecretsHelpers.splitConfig(
           workspaceId,
           fullConfig,
-          spec.getConnectionSpecification());
+          spec);
     }
     splitSecretConfig.getCoordinateToPayload().forEach(longLivedSecretPersistence.get()::write);
     return splitSecretConfig.getPartialConfig();
@@ -324,8 +324,35 @@ public Optional<WorkspaceServiceAccount> getOptionalWorkspaceServiceAccount(fina
 
   public void writeWorkspace(final StandardWorkspace workspace)
       throws JsonValidationException, IOException {
-    // TODO(msiega): split secrets once they're introduced.
-    configRepository.writeStandardWorkspaceNoSecrets(workspace);
+    // Get the schema for the webhook config so we can split out any secret fields.
+    final JsonNode webhookConfigSchema = Jsons.jsonNodeFromFile(ConfigSchema.WORKSPACE_WEBHOOK_OPERATION_CONFIGS.getConfigSchemaFile());
+    // Check if there's an existing config, so we can re-use the secret coordinates.
+    final var previousWorkspace = getWorkspaceIfExists(workspace.getWorkspaceId(), false);
+    Optional<JsonNode> previousWebhookConfigs = Optional.empty();
+    if (previousWorkspace.isPresent() && previousWorkspace.get().getWebhookOperationConfigs() != null) {
+      previousWebhookConfigs = Optional.of(previousWorkspace.get().getWebhookOperationConfigs());
+    }
+    // Split out the secrets from the webhook config.
+    final JsonNode partialConfig = workspace.getWebhookOperationConfigs() == null ? null
+        : statefulUpdateSecrets(
+            workspace.getWorkspaceId(),
+            previousWebhookConfigs,
+            workspace.getWebhookOperationConfigs(),
+            webhookConfigSchema, true);
+    final StandardWorkspace partialWorkspace = Jsons.clone(workspace);
+    if (partialConfig != null) {
+      partialWorkspace.withWebhookOperationConfigs(partialConfig);
+    }
+    configRepository.writeStandardWorkspaceNoSecrets(partialWorkspace);
+  }
+
+  private Optional<StandardWorkspace> getWorkspaceIfExists(final UUID workspaceId, final boolean includeTombstone) {
+    try {
+      final StandardWorkspace existingWorkspace = configRepository.getStandardWorkspaceNoSecrets(workspaceId, includeTombstone);
+      return existingWorkspace == null ? Optional.empty() : Optional.of(existingWorkspace);
+    } catch (JsonValidationException | IOException | ConfigNotFoundException e) {
+      return Optional.empty();
+    }
   }
 
 }

airbyte-config/config-persistence/src/test/java/io/airbyte/config/persistence/SecretsRepositoryWriterTest.java

@@ -11,6 +11,7 @@
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
@@ -30,9 +31,13 @@
 import io.airbyte.config.AirbyteConfig;
 import io.airbyte.config.ConfigSchema;
 import io.airbyte.config.DestinationConnection;
+import io.airbyte.config.Geography;
 import io.airbyte.config.SourceConnection;
 import io.airbyte.config.StandardDestinationDefinition;
 import io.airbyte.config.StandardSourceDefinition;
+import io.airbyte.config.StandardWorkspace;
+import io.airbyte.config.WebhookConfig;
+import io.airbyte.config.WebhookOperationConfigs;
 import io.airbyte.config.WorkspaceServiceAccount;
 import io.airbyte.config.persistence.split_secrets.MemorySecretPersistence;
 import io.airbyte.config.persistence.split_secrets.RealSecretsHydrator;
@@ -42,6 +47,7 @@
 import io.airbyte.validation.json.JsonSchemaValidator;
 import io.airbyte.validation.json.JsonValidationException;
 import java.io.IOException;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -88,6 +94,11 @@ class SecretsRepositoryWriterTest {
 
   private static final String PASSWORD_PROPERTY_NAME = "password";
   private static final String PASSWORD_FIELD_NAME = "_secret";
+  private static final String TEST_EMAIL = "test-email";
+  private static final String TEST_WORKSPACE_NAME = "test-workspace-name";
+  private static final String TEST_WORKSPACE_SLUG = "test-workspace-slug";
+  private static final String TEST_WEBHOOK_NAME = "test-webhook-name";
+  private static final String TEST_AUTH_TOKEN = "test-auth-token";
 
   private ConfigRepository configRepository;
   private MemorySecretPersistence longLivedSecretPersistence;
@@ -433,4 +444,37 @@ void testWriteDifferentStagingConfiguration() throws JsonValidationException, Co
             Map.of(PASSWORD_FIELD_NAME, hmacSecretNewCoordinate.getFullCoordinate()))));
   }
 
+  @Test
+  void testWriteWorkspaceSplitsSecrets() throws JsonValidationException, IOException {
+    final ConfigRepository configRepository = mock(ConfigRepository.class);
+    final SecretPersistence secretPersistence = mock(SecretPersistence.class);
+    final SecretsRepositoryWriter secretsRepositoryWriter =
+        spy(new SecretsRepositoryWriter(configRepository, jsonSchemaValidator, Optional.of(secretPersistence), Optional.of(secretPersistence)));
+    final var webhookConfigs = new WebhookOperationConfigs().withWebhookConfigs(List.of(
+        new WebhookConfig()
+            .withName(TEST_WEBHOOK_NAME)
+            .withAuthToken(TEST_AUTH_TOKEN)
+            .withId(UUID.randomUUID())));
+    final var workspace = new StandardWorkspace()
+        .withWorkspaceId(UUID.randomUUID())
+        .withCustomerId(UUID.randomUUID())
+        .withEmail(TEST_EMAIL)
+        .withName(TEST_WORKSPACE_NAME)
+        .withSlug(TEST_WORKSPACE_SLUG)
+        .withInitialSetupComplete(false)
+        .withDisplaySetupWizard(true)
+        .withNews(false)
+        .withAnonymousDataCollection(false)
+        .withSecurityUpdates(false)
+        .withTombstone(false)
+        .withNotifications(Collections.emptyList())
+        .withDefaultGeography(Geography.AUTO)
+        // Serialize it to a string, then deserialize it to a JsonNode.
+        .withWebhookOperationConfigs(Jsons.jsonNode(webhookConfigs));
+    secretsRepositoryWriter.writeWorkspace(workspace);
+    final var workspaceArgumentCaptor = ArgumentCaptor.forClass(StandardWorkspace.class);
+    verify(configRepository, times(1)).writeStandardWorkspaceNoSecrets(workspaceArgumentCaptor.capture());
+    assertFalse(Jsons.serialize(workspaceArgumentCaptor.getValue().getWebhookOperationConfigs()).contains(TEST_AUTH_TOKEN));
+  }
+
 }

airbyte-server/src/main/java/io/airbyte/server/converters/WebhookOperationConfigsConverter.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.server.converters;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.airbyte.api.model.generated.WebhookConfigRead;
+import io.airbyte.api.model.generated.WebhookConfigWrite;
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.config.WebhookConfig;
+import io.airbyte.config.WebhookOperationConfigs;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+public class WebhookOperationConfigsConverter {
+
+  public static JsonNode toPersistenceWrite(List<WebhookConfigWrite> apiWebhookConfigs) {
+    if (apiWebhookConfigs == null) {
+      return Jsons.emptyObject();
+    }
+
+    final WebhookOperationConfigs configs = new WebhookOperationConfigs()
+        .withWebhookConfigs(apiWebhookConfigs.stream().map(WebhookOperationConfigsConverter::toPersistenceConfig).collect(Collectors.toList()));
+
+    return new ObjectMapper().convertValue(configs, JsonNode.class);
+  }
+
+  public static List<WebhookConfigRead> toApiReads(List<WebhookConfig> persistenceConfig) {
+    if (persistenceConfig.isEmpty()) {
+      return Collections.emptyList();
+    }
+    return persistenceConfig.stream().map(WebhookOperationConfigsConverter::toApiRead).collect(Collectors.toList());
+  }
+
+  private static WebhookConfig toPersistenceConfig(final WebhookConfigWrite input) {
+    return new WebhookConfig()
+        .withId(UUID.randomUUID())
+        .withName(input.getName())
+        .withAuthToken(input.getAuthToken());
+  }
+
+  private static WebhookConfigRead toApiRead(final WebhookConfig persistenceConfig) {
+    final var read = new WebhookConfigRead();
+    read.setId(persistenceConfig.getId());
+    read.setName(persistenceConfig.getName());
+    return read;
+  }
+
+}