Merge branch 'release/0.37.0'

.github/workflows/release-drafter.yml

@@ -3,7 +3,7 @@ name: Release Drafter
 on:
   push:
     branches:
-      - develop
+      - release/*
 
 jobs:
   update_release_draft:

VERSION

@@ -1 +1 @@
-0.36.1
+0.37.0

backend/model/metadata/src/main/java/co/airy/model/metadata/MetadataKeys.java

@@ -50,6 +50,7 @@ public static class MessageKeys {
         public static class Source {
             public static final String ID = "source.id";
             public static final String DELIVERY_STATE = "source.delivery_state";
+            public static final String ERROR = "source.error";
         }
 
         public static class Reaction {

backend/sources/facebook/connector/BUILD

@@ -35,6 +35,7 @@ springboot(
             ":app",
             "//backend:base_test",
             "//lib/java/kafka/test:kafka-test",
+            "//lib/java/spring/test:spring-test",
         ] + app_deps,
     )
     for file in glob(["src/test/java/**/*Test.java"])

backend/sources/facebook/connector/src/main/java/co/airy/core/sources/facebook/ChannelsController.java

@@ -5,6 +5,7 @@
 import co.airy.avro.communication.Metadata;
 import co.airy.core.sources.facebook.api.Api;
 import co.airy.core.sources.facebook.api.ApiException;
+import co.airy.core.sources.facebook.api.model.FaceBookMetadataKeys;
 import co.airy.core.sources.facebook.api.model.PageWithConnectInfo;
 import co.airy.core.sources.facebook.payload.ConnectInstagramRequestPayload;
 import co.airy.core.sources.facebook.payload.ConnectPageRequestPayload;
@@ -104,7 +105,9 @@ ResponseEntity<?> connectFacebook(@RequestBody @Valid ConnectPageRequestPayload
                     )
                     .metadataMap(MetadataMap.from(List.of(
                             newChannelMetadata(channelId, MetadataKeys.ChannelKeys.NAME, Optional.ofNullable(payload.getName()).orElse(pageWithConnectInfo.getNameWithLocationDescriptor())),
-                            newChannelMetadata(channelId, MetadataKeys.ChannelKeys.IMAGE_URL, Optional.ofNullable(payload.getImageUrl()).orElse(pageWithConnectInfo.getPicture().getData().getUrl()))
+                            newChannelMetadata(channelId, MetadataKeys.ChannelKeys.IMAGE_URL, Optional.ofNullable(payload.getImageUrl()).orElse(pageWithConnectInfo.getPicture().getData().getUrl())),
+                            newChannelMetadata(channelId, FaceBookMetadataKeys.ChannelKeys.PAGE_ID, payload.getPageId()),
+                            newChannelMetadata(channelId, FaceBookMetadataKeys.ChannelKeys.PAGE_TOKEN, payload.getPageToken())
                     ))).build();
 
             stores.storeChannelContainer(container);
@@ -132,7 +135,10 @@ ResponseEntity<?> connectInstagram(@RequestBody @Valid ConnectInstagramRequestPa
             api.connectPageToApp(pageWithConnectInfo.getAccessToken());
 
             final MetadataMap metadataMap = MetadataMap.from(List.of(
-                    newChannelMetadata(channelId, MetadataKeys.ChannelKeys.NAME, Optional.ofNullable(payload.getName()).orElse(String.format("%s Instagram account", pageWithConnectInfo.getNameWithLocationDescriptor())))
+                    newChannelMetadata(channelId, MetadataKeys.ChannelKeys.NAME, Optional.ofNullable(payload.getName()).orElse(String.format("%s Instagram account", pageWithConnectInfo.getNameWithLocationDescriptor()))),
+                    newChannelMetadata(channelId, FaceBookMetadataKeys.ChannelKeys.PAGE_ID, payload.getPageId()),
+                    newChannelMetadata(channelId, FaceBookMetadataKeys.ChannelKeys.PAGE_TOKEN, payload.getPageToken()),
+                    newChannelMetadata(channelId, FaceBookMetadataKeys.ChannelKeys.ACCOUNT_ID, payload.getAccountId())
             ));
 
             Optional.ofNullable(payload.getImageUrl())

backend/sources/facebook/connector/src/main/java/co/airy/core/sources/facebook/Connector.java

@@ -78,13 +78,23 @@ public List<KeyValue<String, SpecificRecordBase>> sendMessage(SendMessageRequest
 
             return List.of(KeyValue.pair(message.getId(), message), KeyValue.pair(getId(metadata).toString(), metadata));
         } catch (ApiException e) {
-            log.error(String.format("Failed to send a message to Facebook \n SendMessageRequest: %s \n Error Message: %s \n", sendMessageRequest, e.getMessage()), e);
+            log.error(String.format("Failed to send a message to Facebook \n SendMessageRequest: %s \n Api Exception: %s \n", sendMessageRequest, e.getMessage()), e);
+            final ArrayList<KeyValue<String, SpecificRecordBase>> results = new ArrayList<>();
+            final Metadata error = newMessageMetadata(message.getId(), MetadataKeys.MessageKeys.ERROR, e.getMessage());
+            results.add(KeyValue.pair(getId(error).toString(), error));
+
+            if (e.getErrorPayload() != null) {
+                final Metadata errorPayload = newMessageMetadata(message.getId(), MetadataKeys.MessageKeys.Source.ERROR, e.getErrorPayload());
+                results.add(KeyValue.pair(getId(errorPayload).toString(), errorPayload));
+            }
+            updateDeliveryState(message, DeliveryState.FAILED);
+            return results;
         } catch (Exception e) {
             log.error(String.format("Failed to send a message to Facebook \n SendMessageRequest: %s", sendMessageRequest), e);
+            final Metadata metadata = newMessageMetadata(message.getId(), MetadataKeys.MessageKeys.ERROR, e.getMessage());
+            updateDeliveryState(message, DeliveryState.FAILED);
+            return List.of(KeyValue.pair(message.getId(), message), KeyValue.pair(getId(metadata).toString(), metadata));
         }
-
-        updateDeliveryState(message, DeliveryState.FAILED);
-        return List.of(KeyValue.pair(message.getId(), message));
     }
 
     private boolean isMessageStale(Message message) {

backend/sources/facebook/connector/src/main/java/co/airy/core/sources/facebook/api/Api.java

@@ -8,7 +8,12 @@
 import co.airy.core.sources.facebook.api.model.SendMessagePayload;
 import co.airy.core.sources.facebook.api.model.SendMessageResponse;
 import co.airy.core.sources.facebook.api.model.UserProfile;
+import co.airy.log.AiryLoggerFactory;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.PropertyNamingStrategy;
+import org.slf4j.Logger;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.context.event.ApplicationReadyEvent;
 import org.springframework.boot.web.client.RestTemplateBuilder;
@@ -39,6 +44,7 @@
  */
 @Service
 public class Api implements ApplicationListener<ApplicationReadyEvent> {
+    private static final Logger log = AiryLoggerFactory.getLogger(Api.class);
     private final RestTemplateBuilder restTemplateBuilder;
     private final ObjectMapper objectMapper;
     private RestTemplate restTemplate;
@@ -51,16 +57,15 @@ public class Api implements ApplicationListener<ApplicationReadyEvent> {
     private final HttpHeaders httpHeaders = new HttpHeaders();
     private final String appId;
     private final String apiSecret;
-    private static final String errorMessageTemplate =
-            "Exception while sending a message to Facebook: \n" +
-                    "Http Status Code: %s \n" +
-                    "Error Message: %s \n";
 
-    public Api(ObjectMapper objectMapper, RestTemplateBuilder restTemplateBuilder,
+    public Api(RestTemplateBuilder restTemplateBuilder,
                @Value("${facebook.app-id}") String appId,
                @Value("${facebook.app-secret}") String apiSecret) {
         httpHeaders.setContentType(MediaType.APPLICATION_JSON);
-        this.objectMapper = objectMapper;
+        this.objectMapper = new ObjectMapper()
+                .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
+                .configure(DeserializationFeature.FAIL_ON_MISSING_EXTERNAL_TYPE_ID_PROPERTY, false)
+                .setPropertyNamingStrategy(PropertyNamingStrategy.SNAKE_CASE);
         this.restTemplateBuilder = restTemplateBuilder;
         this.appId = appId;
         this.apiSecret = apiSecret;
@@ -111,24 +116,21 @@ public UserProfile getInstagramProfile(String sourceConversationId, String token
     public UserProfile getProfileFromContact(String sourceConversationId, String token) {
         String reqUrl = String.format(baseUrl + "/%s?fields=first_name,last_name,profile_pic&access_token=%s",
                 sourceConversationId, token);
-        ResponseEntity<UserProfile> responseEntity = restTemplate.getForEntity(reqUrl, UserProfile.class);
-        if (responseEntity.getStatusCode() != HttpStatus.OK) {
-            throw new ApiException("Call unsuccessful, received HTTP status " + responseEntity.getStatusCodeValue());
-        }
-        return responseEntity.getBody();
+        ResponseEntity<UserProfile> response = restTemplate.getForEntity(reqUrl, UserProfile.class);
+        return response.getBody();
     }
 
     // See https://developers.facebook.com/docs/graph-api/reference/v9.0/conversation#edges
     public UserProfile getProfileFromParticipants(String sourceConversationId, String token) {
         String reqUrl = String.format(baseUrl + "/me/conversations?user_id=%s&fields=participants&access_token=%s",
                 sourceConversationId, token);
 
-        ResponseEntity<Participants> responseEntity = restTemplate.getForEntity(reqUrl, Participants.class);
-        if (responseEntity.getBody() == null || responseEntity.getStatusCode() != HttpStatus.OK) {
-            throw new ApiException("Call unsuccessful");
+        ResponseEntity<Participants> response = restTemplate.getForEntity(reqUrl, Participants.class);
+        if (response.getBody() == null) {
+            throw new ApiException(String.format("Response body was null, status code %s", response.getStatusCode()));
         }
 
-        return fromParticipants(responseEntity.getBody(), sourceConversationId);
+        return fromParticipants(response.getBody(), sourceConversationId);
     }
 
     public PageWithConnectInfo getPageForUser(final String pageId, final String accessToken) throws Exception {
@@ -180,7 +182,18 @@ public boolean hasError(ClientHttpResponse response) throws IOException {
 
                     @Override
                     public void handleError(ClientHttpResponse response) throws IOException {
-                        throw new ApiException(String.format(errorMessageTemplate, response.getRawStatusCode(), new String(response.getBody().readAllBytes())));
+                        final String errorPayload = new String(response.getBody().readAllBytes());
+                        final int statusCode = response.getRawStatusCode();
+
+                        final JsonNode jsonNode = objectMapper.readTree(errorPayload);
+                        final String errorMessage = Optional.of(jsonNode.get("error"))
+                                .map((node) -> node.get("message"))
+                                .map(JsonNode::textValue)
+                                .orElseGet(() -> {
+                                    log.warn("Could not parse error message from response: {}", errorPayload);
+                                    return String.format("Api replied with status code %s and payload %s", statusCode, errorPayload);
+                                });
+                        throw new ApiException(errorMessage, errorPayload);
                     }
                 })
                 .additionalMessageConverters(new MappingJackson2HttpMessageConverter(objectMapper))

backend/sources/facebook/connector/src/main/java/co/airy/core/sources/facebook/api/ApiException.java

@@ -1,7 +1,15 @@
 package co.airy.core.sources.facebook.api;
 
+import lombok.Getter;
+
 public class ApiException extends RuntimeException {
-    public ApiException(String msg) {
-        super(msg);
+    @Getter
+    private String errorPayload;
+    public ApiException(String message) {
+        super(message);
+    }
+    public ApiException(String message, String errorPayload) {
+        super(message);
+        this.errorPayload = errorPayload;
     }
 }

backend/sources/facebook/connector/src/main/java/co/airy/core/sources/facebook/api/model/FaceBookMetadataKeys.java

@@ -0,0 +1,10 @@
+package co.airy.core.sources.facebook.api.model;
+
+
+public class FaceBookMetadataKeys {
+    public static class ChannelKeys {
+        public static final String PAGE_ID = "page_id";
+        public static final String PAGE_TOKEN = "page_token";
+        public static final String ACCOUNT_ID = "account_id";
+    }
+}