From eb0539e0ff56df188b456b8347c91c8741908554 Mon Sep 17 00:00:00 2001
From: Malik Diarra <malik@airbyte.io>
Date: Wed, 27 Sep 2023 17:22:40 -0700
Subject: [PATCH] Send notification with change details when schema is
 propagated (#9008)

---
 .../CustomerioNotificationClient.java             |  8 ++++++--
 .../airbyte/notification/NotificationClient.java  |  5 +++--
 .../notification/SlackNotificationClient.java     | 15 ++++++++++++++-
 ...ma_propagation_slack_notification_template.txt |  4 ++++
 4 files changed, 27 insertions(+), 5 deletions(-)
 create mode 100644 airbyte-notification/src/main/resources/slack/schema_propagation_slack_notification_template.txt

diff --git a/airbyte-notification/src/main/java/io/airbyte/notification/CustomerioNotificationClient.java b/airbyte-notification/src/main/java/io/airbyte/notification/CustomerioNotificationClient.java
index 9101e090748..8c7639a95f4 100644
--- 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
index 9948a606b07..afb7daa1e73 100644
--- 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
index 3d8a1ef7e3d..bf99f23ab71 100644
--- 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/airbyte-notification/src/main/resources/slack/schema_propagation_slack_notification_template.txt b/airbyte-notification/src/main/resources/slack/schema_propagation_slack_notification_template.txt
new file mode 100644
index 00000000000..0b7ebd31694
--- /dev/null
+++ b/airbyte-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
\ No newline at end of file