Send notification with change details when schema is propagated (#9008)

airbytehq · Sep 28, 2023 · eb0539e · eb0539e
1 parent 7535228
commit eb0539e
Show file tree

Hide file tree

Showing 4 changed files with 27 additions and 5 deletions.
diff --git a/airbyte-notification/src/main/java/io/airbyte/notification/CustomerioNotificationClient.java b/airbyte-notification/src/main/java/io/airbyte/notification/CustomerioNotificationClient.java
@@ -10,7 +10,6 @@
 import io.airbyte.config.ActorDefinitionBreakingChange;
 import io.airbyte.config.ActorType;
 import io.airbyte.config.EnvConfigs;
-import io.airbyte.config.SlackNotificationConfiguration;
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;
@@ -180,11 +179,16 @@ public boolean notifyFailure(final String message) {
   @Override
   public boolean notifySchemaChange(final UUID connectionId,
                                     final boolean isBreaking,
-                                    final SlackNotificationConfiguration config,
                                     final String url) {
     throw new NotImplementedException();
   }
 
+  @Override
+  public boolean notifySchemaPropagated(final UUID connectionId, final List<String> changes, final String url, boolean isBreaking)
+      throws IOException, InterruptedException {
+    throw new NotImplementedException();
+  }
+
   @Override
   public String getNotificationClientType() {
     return CUSTOMERIO_TYPE;

diff --git a/airbyte-notification/src/main/java/io/airbyte/notification/NotificationClient.java b/airbyte-notification/src/main/java/io/airbyte/notification/NotificationClient.java
@@ -7,7 +7,6 @@
 import io.airbyte.commons.resources.MoreResources;
 import io.airbyte.config.ActorDefinitionBreakingChange;
 import io.airbyte.config.ActorType;
-import io.airbyte.config.SlackNotificationConfiguration;
 import java.io.IOException;
 import java.util.List;
 import java.util.UUID;
@@ -73,10 +72,12 @@ public abstract boolean notifyBreakingChangeSyncsDisabled(List<String> receiverE
 
   public abstract boolean notifySchemaChange(final UUID connectionId,
                                              final boolean isBreaking,
-                                             final SlackNotificationConfiguration config,
                                              final String url)
       throws IOException, InterruptedException;
 
+  public abstract boolean notifySchemaPropagated(final UUID connectionId, final List<String> changes, final String url, boolean isBreaking)
+      throws IOException, InterruptedException;
+
   public abstract String getNotificationClientType();
 
   String renderTemplate(final String templateFile, final String... data) throws IOException {

diff --git a/airbyte-notification/src/main/java/io/airbyte/notification/SlackNotificationClient.java b/airbyte-notification/src/main/java/io/airbyte/notification/SlackNotificationClient.java
@@ -147,7 +147,7 @@ public boolean notifyBreakingChangeSyncsDisabled(final List<String> receiverEmai
   }
 
   @Override
-  public boolean notifySchemaChange(final UUID connectionId, final boolean isBreaking, final SlackNotificationConfiguration config, final String url)
+  public boolean notifySchemaChange(final UUID connectionId, final boolean isBreaking, final String url)
       throws IOException, InterruptedException {
     final String message = renderTemplate(
         isBreaking ? "slack/breaking_schema_change_slack_notification_template.txt"
@@ -160,6 +160,19 @@ public boolean notifySchemaChange(final UUID connectionId, final boolean isBreak
     return false;
   }
 
+  @Override
+  public boolean notifySchemaPropagated(final UUID connectionId, final List<String> changes, final String url, boolean isBreaking)
+      throws IOException, InterruptedException {
+    final String summary = String.join("\n", changes);
+    final String message = isBreaking ? renderTemplate("slack/breaking_schema_change_slack_notification_template.txt", connectionId.toString(), url)
+        : renderTemplate("slack/schema_propagation_slack_notification.txt", connectionId.toString(), summary, url);
+    final String webhookUrl = config.getWebhook();
+    if (!Strings.isEmpty(webhookUrl)) {
+      return notify(message);
+    }
+    return false;
+  }
+
   private boolean notify(final String message) throws IOException, InterruptedException {
     final HttpClient httpClient = HttpClient.newBuilder()
         .version(HttpClient.Version.HTTP_2)

diff --git a/...-notification/src/main/resources/slack/schema_propagation_slack_notification_template.txt b/...-notification/src/main/resources/slack/schema_propagation_slack_notification_template.txt
@@ -0,0 +1,4 @@
+Your source schema has changed for connection ID: %s and the following changes were automatically propagated:
+%s
+
+Visit your connection page here: %s