feat: add function to retrieve connections per destinationId (#14850)

airbyte-data/src/main/java/io/airbyte/data/services/ConnectionService.java

@@ -44,6 +44,8 @@ Map<UUID, List<StandardSync>> listWorkspaceStandardSyncsPaginated(List<UUID> wor
 
   List<StandardSync> listConnectionsBySource(UUID sourceId, boolean includeDeleted) throws IOException;
 
+  List<StandardSync> listConnectionsByDestination(UUID destinationId, boolean includeDeleted) throws IOException;
+
   List<StandardSync> listConnectionsByActorDefinitionIdAndType(UUID actorDefinitionId, String actorTypeValue, boolean includeDeleted)
       throws IOException;
 

airbyte-data/src/main/java/io/airbyte/data/services/impls/jooq/ConnectionServiceJooqImpl.java

@@ -331,6 +331,34 @@ public List<StandardSync> listConnectionsBySource(final UUID sourceId, final boo
     return getStandardSyncsFromResult(connectionAndOperationIdsResult, getNotificationConfigurationByConnectionIds(connectionIds));
   }
 
+  /**
+   * List connections that use a source.
+   *
+   * @param destinationId destination id
+   * @param includeDeleted include deleted
+   * @return connections that use the provided destination
+   * @throws IOException if there is an issue while interacting with db.
+   */
+  @Override
+  public List<StandardSync> listConnectionsByDestination(final UUID destinationId, final boolean includeDeleted)
+      throws IOException {
+    final Result<Record> connectionAndOperationIdsResult = database.query(ctx -> ctx
+        .select(
+            CONNECTION.asterisk(),
+            groupConcat(CONNECTION_OPERATION.OPERATION_ID).separator(OPERATION_IDS_AGG_DELIMITER).as(OPERATION_IDS_AGG_FIELD),
+            SCHEMA_MANAGEMENT.AUTO_PROPAGATION_STATUS, SCHEMA_MANAGEMENT.BACKFILL_PREFERENCE)
+        .from(CONNECTION)
+        .leftJoin(CONNECTION_OPERATION).on(CONNECTION_OPERATION.CONNECTION_ID.eq(CONNECTION.ID))
+        .leftJoin(SCHEMA_MANAGEMENT).on(SCHEMA_MANAGEMENT.CONNECTION_ID.eq(CONNECTION.ID))
+        .where(CONNECTION.DESTINATION_ID.eq(destinationId)
+            .and(includeDeleted ? noCondition() : CONNECTION.STATUS.notEqual(StatusType.deprecated)))
+        .groupBy(CONNECTION.ID, SCHEMA_MANAGEMENT.AUTO_PROPAGATION_STATUS, SCHEMA_MANAGEMENT.BACKFILL_PREFERENCE)).fetch();
+
+    final List<UUID> connectionIds = connectionAndOperationIdsResult.map(record -> record.get(CONNECTION.ID));
+
+    return getStandardSyncsFromResult(connectionAndOperationIdsResult, getNotificationConfigurationByConnectionIds(connectionIds));
+  }
+
   /**
    * List connections that use a particular actor definition.
    *