New Feature Notification per Connection

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

@@ -235,7 +235,7 @@ paths:
         content:
           application/json:
             schema:
-              $ref: "#/components/schemas/Notification"
+              $ref: "#/components/schemas/NotificationLegacy"
         required: true
       responses:
         "200":
@@ -248,6 +248,50 @@ paths:
           $ref: "#/components/responses/NotFoundResponse"
         "422":
           $ref: "#/components/responses/InvalidInputResponse"
+  /v1/notifications/create:
+    post:
+      tags:
+        - notifications
+      summary: >
+        Create a new Notification
+      operationId: createNotification
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/NotificationCreate"
+        required: true
+      responses:
+        "200":
+          description: Successful operation
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/NotificationRead"
+        "404":
+          $ref: "#/components/responses/NotFoundResponse"
+  /v1/notifications/add_connection:
+    post:
+      tags:
+        - notifications
+      summary: >
+        Create a link between a connection and a notification
+      operationId: createNotificationConnection
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/NotificationConnectionCreate"
+        required: true
+      responses:
+        "200":
+          description: Successful operation
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/NotificationRead"
+        "404":
+          $ref: "#/components/responses/NotFoundResponse"
   /v1/source_definitions/create:
     post:
       tags:
@@ -1940,10 +1984,61 @@ components:
         notifications:
           type: array
           items:
-            $ref: "#/components/schemas/Notification"
+            $ref: "#/components/schemas/NotificationLegacy"
         displaySetupWizard:
           type: boolean
-    Notification:
+    NotificationCreate:
+      type: object
+      required:
+        - workspaceId
+        - name
+        - webhookUrl
+        - notificationType
+      properties:
+        workspaceId:
+          type: string
+          format: uuid
+        name:
+          type: string
+        webhookUrl:
+          type: string
+        defaultNotification:
+          type: boolean
+          default: false
+        notificationType:
+          $ref: "#/components/schemas/NotificationType"
+        tombstone:
+          type: boolean
+          default: false
+    NotificationConnectionCreate:
+      type: object
+      required:
+        - workspaceId
+        - notificationId
+        - connectionId
+        - sendOnSuccess
+        - sendOnFailure
+        - tombstone
+      properties:
+        workspaceId:
+          type: string
+          format: uuid
+        notificationId:
+          type: string
+          format: uuid
+        connectionId:
+          type: string
+          format: uuid
+        sendOnSuccess:
+          type: boolean
+          default: false
+        sendOnFailure:
+          type: boolean
+          default: true
+        tombstone:
+          type: boolean
+          default: false
+    NotificationLegacy:
       type: object
       required:
         - notificationType
@@ -2036,7 +2131,7 @@ components:
         notifications:
           type: array
           items:
-            $ref: "#/components/schemas/Notification"
+            $ref: "#/components/schemas/NotificationLegacy"
         firstCompletedSync:
           type: boolean
         feedbackDone:
@@ -2078,7 +2173,7 @@ components:
         notifications:
           type: array
           items:
-            $ref: "#/components/schemas/Notification"
+            $ref: "#/components/schemas/NotificationLegacy"
     WorkspaceGiveFeedback:
       type: object
       required:

airbyte-bootloader/src/test/java/io/airbyte/bootloader/BootloaderAppTest.java

@@ -72,15 +72,15 @@ void testBootloaderAppBlankDb() throws Exception {
         container.getPassword(),
         container.getJdbcUrl()).getInitialized();
     val jobsMigrator = new JobsDatabaseMigrator(jobDatabase, this.getClass().getName());
-    assertEquals("0.35.40.001", jobsMigrator.getLatestMigration().getVersion().getVersion());
+    assertEquals("0.35.5.001", jobsMigrator.getLatestMigration().getVersion().getVersion());
 
     val configDatabase = new ConfigsDatabaseInstance(
         mockedConfigs.getConfigDatabaseUser(),
         mockedConfigs.getConfigDatabasePassword(),
         mockedConfigs.getConfigDatabaseUrl())
             .getAndInitialize();
     val configsMigrator = new ConfigsDatabaseMigrator(configDatabase, this.getClass().getName());
-    assertEquals("0.35.56.001", configsMigrator.getLatestMigration().getVersion().getVersion());
+    assertEquals("0.35.50.001", configsMigrator.getLatestMigration().getVersion().getVersion());
 
     val jobsPersistence = new DefaultJobPersistence(jobDatabase);
     assertEquals(version, jobsPersistence.getVersion().get());

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

@@ -17,6 +17,18 @@ public enum ConfigSchema implements AirbyteConfig {
       standardWorkspace -> standardWorkspace.getWorkspaceId().toString(),
       "workspaceId"),
 
+  // notification
+  NOTIFICATION("Notification.yaml",
+      Notification.class,
+      Notification -> Notification.getNotificationId().toString(),
+      "notificationId"),
+
+  // notification connection
+  NOTIFICATION_CONNECTION("NotificationConnection.yaml",
+      NotificationConnection.class,
+      NotificationConnection -> NotificationConnection.getNotificationConnectionId().toString(),
+      "notificationConnectionId"),
+
   // source
   STANDARD_SOURCE_DEFINITION("StandardSourceDefinition.yaml",
       StandardSourceDefinition.class,

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

@@ -2,21 +2,32 @@
 "$schema": http://json-schema.org/draft-07/schema#
 "$id": https://github.com/airbytehq/airbyte/blob/master/airbyte-config/models/src/main/resources/types/Notification.yaml
 title: Notification
-description: Notification Settings
+description: notification configuration
 type: object
 required:
+  - workspaceId
+  - name
+  - webhookUrl
   - notificationType
-additionalProperties: true
+additionalProperties: false
 properties:
-  # Instead of this type field, we would prefer a json schema "oneOf" but unfortunately,
-  # the jsonschema2pojo does not seem to support it yet: https://github.com/joelittlejohn/jsonschema2pojo/issues/392
+  workspaceId:
+    type: string
+    format: uuid
+  notificationId:
+    type: string
+    format: uuid
+  name:
+    type: string
+  webhookUrl:
+    type: string
   notificationType:
     "$ref": NotificationType.yaml
-  sendOnSuccess:
+  defaultNotification:
     type: boolean
-    default: false
-  sendOnFailure:
+    description: if set true the webhook will be triggered to all connections
+  tombstone:
+    description:
+      if not set or false, the configuration is active. if true, then this
+      configuration is permanently off.
     type: boolean
-    default: true
-  slackConfiguration:
-    "$ref": SlackNotificationConfiguration.yaml

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

@@ -0,0 +1,35 @@
+---
+"$schema": http://json-schema.org/draft-07/schema#
+"$id": https://github.com/airbytehq/airbyte/blob/master/airbyte-config/models/src/main/resources/types/NotificationConnection.yaml
+title: NotificationConnection
+description: notification connection link
+type: object
+required:
+  - workspaceId
+  - notificationId
+  - sendOnSuccess
+  - sendOnFailure
+  - tombstone
+additionalProperties: false
+properties:
+  notificationConnectionId:
+    type: string
+    format: uuid
+  workspaceId:
+    type: string
+    format: uuid
+  notificationId:
+    type: string
+    format: uuid
+  connectionId:
+    type: string
+    format: uuid
+  sendOnSuccess:
+    type: boolean
+  sendOnFailure:
+    type: boolean
+  tombstone:
+    description:
+      if not set or false, the configuration is active. if true, then this
+      configuration is permanently off.
+    type: boolean

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

@@ -0,0 +1,22 @@
+---
+"$schema": http://json-schema.org/draft-07/schema#
+"$id": https://github.com/airbytehq/airbyte/blob/master/airbyte-config/models/src/main/resources/types/NotificationLegacy.yaml
+title: NotificationLegacy
+description: Notification Settings
+type: object
+required:
+  - notificationType
+additionalProperties: true
+properties:
+  # Instead of this type field, we would prefer a json schema "oneOf" but unfortunately,
+  # the jsonschema2pojo does not seem to support it yet: https://github.com/joelittlejohn/jsonschema2pojo/issues/392
+  notificationType:
+    "$ref": NotificationType.yaml
+  sendOnSuccess:
+    type: boolean
+    default: false
+  sendOnFailure:
+    type: boolean
+    default: true
+  slackConfiguration:
+    "$ref": SlackNotificationConfiguration.yaml

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

@@ -42,7 +42,7 @@ properties:
   notifications:
     type: array
     items:
-      "$ref": Notification.yaml
+      "$ref": NotificationLegacy.yaml
   firstCompletedSync:
     type: boolean
   feedbackDone:

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

@@ -24,6 +24,8 @@
 import io.airbyte.config.ConfigSchema;
 import io.airbyte.config.DestinationConnection;
 import io.airbyte.config.DestinationOAuthParameter;
+import io.airbyte.config.Notification;
+import io.airbyte.config.NotificationConnection;
 import io.airbyte.config.SourceConnection;
 import io.airbyte.config.SourceOAuthParameter;
 import io.airbyte.config.StandardDestinationDefinition;
@@ -137,6 +139,28 @@ public void setFeedback(final UUID workflowId) throws JsonValidationException, C
     persistence.writeConfig(ConfigSchema.STANDARD_WORKSPACE, workspace.getWorkspaceId().toString(), workspace);
   }
 
+  public List<Notification> listNotifications(final boolean includeTombstone) throws JsonValidationException, IOException {
+
+    final List<Notification> notifications = new ArrayList<>();
+
+    for (final Notification notification : persistence.listConfigs(ConfigSchema.NOTIFICATION, Notification.class)) {
+      if (!MoreBooleans.isTruthy(notification.getTombstone()) || includeTombstone) {
+        notifications.add(notification);
+      }
+    }
+
+    return notifications;
+  }
+
+  public void writeNotification(final Notification notification) throws JsonValidationException, IOException {
+    persistence.writeConfig(ConfigSchema.NOTIFICATION, notification.getNotificationId().toString(), notification);
+  }
+
+  public void writeNotificationConnection(final NotificationConnection notificationConnection) throws JsonValidationException, IOException {
+    persistence.writeConfig(ConfigSchema.NOTIFICATION_CONNECTION, notificationConnection.getNotificationConnectionId().toString(),
+        notificationConnection);
+  }
+
   public StandardSourceDefinition getStandardSourceDefinition(final UUID sourceDefinitionId)
       throws JsonValidationException, IOException, ConfigNotFoundException {
 

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

@@ -11,6 +11,8 @@
 import static io.airbyte.db.instance.configs.jooq.Tables.ACTOR_OAUTH_PARAMETER;
 import static io.airbyte.db.instance.configs.jooq.Tables.CONNECTION;
 import static io.airbyte.db.instance.configs.jooq.Tables.CONNECTION_OPERATION;
+import static io.airbyte.db.instance.configs.jooq.Tables.NOTIFICATION_CONFIG;
+import static io.airbyte.db.instance.configs.jooq.Tables.NOTIFICATION_CONNECTION;
 import static io.airbyte.db.instance.configs.jooq.Tables.OPERATION;
 import static io.airbyte.db.instance.configs.jooq.Tables.STATE;
 import static io.airbyte.db.instance.configs.jooq.Tables.WORKSPACE;
@@ -33,6 +35,8 @@
 import io.airbyte.config.ConfigWithMetadata;
 import io.airbyte.config.DestinationConnection;
 import io.airbyte.config.DestinationOAuthParameter;
+import io.airbyte.config.Notification;
+import io.airbyte.config.NotificationConnection;
 import io.airbyte.config.OperatorDbt;
 import io.airbyte.config.OperatorNormalization;
 import io.airbyte.config.SourceConnection;
@@ -762,6 +766,48 @@ private void writeStandardWorkspace(final List<StandardWorkspace> configs) throw
     });
   }
 
+  private void writeNotification(final List<Notification> configs) throws IOException {
+    database.transaction(ctx -> {
+      writeNotification(configs, ctx);
+      return null;
+    });
+  }
+
+  private void writeNotificationConnection(final List<NotificationConnection> configs) throws IOException {
+    database.transaction(ctx -> {
+      writeNotificationConnection(configs, ctx);
+      return null;
+    });
+  }
+
+  private void writeNotification(final List<Notification> configs, final DSLContext ctx) {
+    configs.forEach((Notification) -> {
+      ctx.insertInto(NOTIFICATION_CONFIG)
+          .set(NOTIFICATION_CONFIG.ID, Notification.getNotificationId())
+          .set(NOTIFICATION_CONFIG.WORKSPACE_ID, Notification.getWorkspaceId())
+          .set(NOTIFICATION_CONFIG.NAME, Notification.getName())
+          .set(NOTIFICATION_CONFIG.WEBHOOK_URL, Notification.getWebhookUrl())
+          .set(NOTIFICATION_CONFIG.DEFAULT_NOTIFICATION, Notification.getDefaultNotification())
+          .set(NOTIFICATION_CONFIG.NOTIFICATION_TYPE, Notification.getNotificationType().value())
+          .set(NOTIFICATION_CONFIG.TOMBSTONE, Notification.getTombstone())
+          .execute();
+    });
+  }
+
+  private void writeNotificationConnection(final List<NotificationConnection> configs, final DSLContext ctx) {
+    configs.forEach((NotificationConnection) -> {
+      ctx.insertInto(NOTIFICATION_CONNECTION)
+          .set(NOTIFICATION_CONNECTION.NOTIFICATION_CONNECTION_ID, NotificationConnection.getNotificationConnectionId())
+          .set(NOTIFICATION_CONNECTION.WORKSPACE_ID, NotificationConnection.getWorkspaceId())
+          .set(NOTIFICATION_CONNECTION.NOTIFICATION_ID, NotificationConnection.getNotificationId())
+          .set(NOTIFICATION_CONNECTION.CONNECTION_ID, NotificationConnection.getConnectionId())
+          .set(NOTIFICATION_CONNECTION.SEND_ON_SUCCESS, NotificationConnection.getSendOnSuccess())
+          .set(NOTIFICATION_CONNECTION.SEND_ON_FAILURE, NotificationConnection.getSendOnFailure())
+          .set(NOTIFICATION_CONNECTION.TOMBSTONE, NotificationConnection.getTombstone())
+          .execute();
+    });
+  }
+
   private void writeStandardWorkspace(final List<StandardWorkspace> configs, final DSLContext ctx) {
     final OffsetDateTime timestamp = OffsetDateTime.now();
     configs.forEach((standardWorkspace) -> {