[#2051] Support instagram as a source

airyhq · Jun 30, 2021 · 9d93f13 · 9d93f13
1 parent eda9c63
commit 9d93f13
Show file tree

Hide file tree

Showing 13 changed files with 398 additions and 8 deletions.
diff --git a/...es/facebook/connector/src/main/java/co/airy/core/sources/facebook/ChannelsController.java b/...es/facebook/connector/src/main/java/co/airy/core/sources/facebook/ChannelsController.java
@@ -2,10 +2,12 @@
 
 import co.airy.avro.communication.Channel;
 import co.airy.avro.communication.ChannelConnectionState;
+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.PageWithConnectInfo;
-import co.airy.core.sources.facebook.payload.ConnectRequestPayload;
+import co.airy.core.sources.facebook.payload.ConnectInstagramRequestPayload;
+import co.airy.core.sources.facebook.payload.ConnectPageRequestPayload;
 import co.airy.core.sources.facebook.payload.DisconnectChannelRequestPayload;
 import co.airy.core.sources.facebook.payload.ExploreRequestPayload;
 import co.airy.core.sources.facebook.payload.ExploreResponsePayload;
@@ -78,7 +80,7 @@ ResponseEntity<?> explore(@RequestBody @Valid ExploreRequestPayload requestPaylo
     }
 
     @PostMapping("/channels.facebook.connect")
-    ResponseEntity<?> connect(@RequestBody @Valid ConnectRequestPayload requestPayload) {
+    ResponseEntity<?> connectFacebook(@RequestBody @Valid ConnectPageRequestPayload requestPayload) {
         final String token = requestPayload.getPageToken();
         final String pageId = requestPayload.getPageId();
 
@@ -115,7 +117,53 @@ ResponseEntity<?> connect(@RequestBody @Valid ConnectRequestPayload requestPaylo
         }
     }
 
-    @PostMapping("/channels.facebook.disconnect")
+    @PostMapping("/channels.instagram.connect")
+    ResponseEntity<?> connectInstagram(@RequestBody @Valid ConnectInstagramRequestPayload requestPayload) {
+        final String token = requestPayload.getPageToken();
+        final String pageId = requestPayload.getPageId();
+        final String accountId = requestPayload.getAccountId();
+
+        final String channelId = UUIDv5.fromNamespaceAndName("instagram", accountId).toString();
+
+        try {
+            final String longLivingUserToken = api.exchangeToLongLivingUserAccessToken(token);
+            final PageWithConnectInfo fbPageWithConnectInfo = api.getPageForUser(pageId, longLivingUserToken);
+
+            api.connectPageToApp(fbPageWithConnectInfo.getAccessToken());
+
+            final MetadataMap metadataMap = MetadataMap.from(List.of(
+                    newChannelMetadata(channelId, MetadataKeys.ChannelKeys.NAME, Optional.ofNullable(requestPayload.getName()).orElse(String.format("%s Instagram account", fbPageWithConnectInfo.getNameWithLocationDescriptor())))
+            ));
+
+            Optional.ofNullable(requestPayload.getImageUrl())
+                    .ifPresent((imageUrl) -> {
+                        final Metadata metadata = newChannelMetadata(channelId, MetadataKeys.ChannelKeys.IMAGE_URL, imageUrl);
+                        metadataMap.put(metadata.getKey(), metadata);
+                    });
+
+            final ChannelContainer container = ChannelContainer.builder()
+                    .channel(
+                            Channel.newBuilder()
+                                    .setId(channelId)
+                                    .setConnectionState(ChannelConnectionState.CONNECTED)
+                                    .setSource("instagram")
+                                    .setSourceChannelId(accountId)
+                                    .setToken(longLivingUserToken)
+                                    .build()
+                    )
+                    .metadataMap(metadataMap).build();
+
+            stores.storeChannelContainer(container);
+
+            return ResponseEntity.ok(fromChannelContainer(container));
+        } catch (ApiException e) {
+            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body(new RequestErrorResponsePayload(e.getMessage()));
+        } catch (Exception e) {
+            return ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE).build();
+        }
+    }
+
+    @PostMapping(path = {"/channels.facebook.disconnect", "/channels.instagram.disconnect"})
     ResponseEntity<?> disconnect(@RequestBody @Valid DisconnectChannelRequestPayload requestPayload) {
         final String channelId = requestPayload.getChannelId().toString();
 

diff --git a/backend/sources/facebook/connector/src/main/java/co/airy/core/sources/facebook/Stores.java b/backend/sources/facebook/connector/src/main/java/co/airy/core/sources/facebook/Stores.java
@@ -30,6 +30,7 @@
 import org.springframework.stereotype.Service;
 
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
@@ -63,14 +64,16 @@ public void onApplicationEvent(ApplicationStartedEvent applicationStartedEvent)
 
         channelStream.toTable(Materialized.as(channelsStore));
 
+        final List<String> sources = List.of("facebook", "instagram");
+
         // Channels table
         KTable<String, Channel> channelsTable = channelStream
-                .filter((sourceChannelId, channel) -> "facebook".equalsIgnoreCase(channel.getSource())
+                .filter((sourceChannelId, channel) -> sources.contains(channel.getSource())
                         && channel.getConnectionState().equals(ChannelConnectionState.CONNECTED)).toTable();
 
         // Facebook messaging stream by conversation-id
         final KStream<String, Message> messageStream = builder.<String, Message>stream(new ApplicationCommunicationMessages().name())
-                .filter((messageId, message) -> "facebook".equalsIgnoreCase(message.getSource()))
+                .filter((messageId, message) -> sources.contains(message.getSource()))
                 .selectKey((messageId, message) -> message.getConversationId());
 
         // Metadata table

diff --git a/...r/src/main/java/co/airy/core/sources/facebook/payload/ConnectInstagramRequestPayload.java b/...r/src/main/java/co/airy/core/sources/facebook/payload/ConnectInstagramRequestPayload.java
@@ -0,0 +1,21 @@
+package co.airy.core.sources.facebook.payload;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import javax.validation.constraints.NotNull;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class ConnectInstagramRequestPayload {
+    @NotNull
+    private String pageId;
+    @NotNull
+    private String accountId;
+    @NotNull
+    private String pageToken;
+    private String name;
+    private String imageUrl;
+}
diff --git a/...cebook/payload/ConnectRequestPayload.java → ...ok/payload/ConnectPageRequestPayload.java b/...cebook/payload/ConnectRequestPayload.java → ...ok/payload/ConnectPageRequestPayload.java
@@ -9,7 +9,7 @@
 @Data
 @NoArgsConstructor
 @AllArgsConstructor
-public class ConnectRequestPayload {
+public class ConnectPageRequestPayload {
     @NotNull
     private String pageId;
     @NotNull

diff --git a/...rces/facebook/events-router/src/main/java/co/airy/core/sources/facebook/EventsRouter.java b/...rces/facebook/events-router/src/main/java/co/airy/core/sources/facebook/EventsRouter.java
@@ -50,11 +50,13 @@ public class EventsRouter implements DisposableBean, ApplicationListener<Applica
     public void startStream() {
         final StreamsBuilder builder = new StreamsBuilder();
 
+        final List<String> sources = List.of("facebook", "instagram");
+
         // Channels table
         KTable<String, Channel> channelsTable = builder.<String, Channel>stream(new ApplicationCommunicationChannels().name())
                 .groupBy((k, v) -> v.getSourceChannelId())
                 .reduce((aggValue, newValue) -> newValue)
-                .filter((sourceChannelId, channel) -> "facebook".equalsIgnoreCase(channel.getSource())
+                .filter((sourceChannelId, channel) -> sources.contains(channel.getSource())
                         && channel.getConnectionState().equals(ChannelConnectionState.CONNECTED));
 
         builder.<String, String>stream(new SourceFacebookEvents().name())

diff --git a/docs/docs/api/endpoints/channels.md b/docs/docs/api/endpoints/channels.md
@@ -146,6 +146,12 @@ import ConnectFacebook from './connect-facebook.mdx'
 
 <ConnectFacebook />
 
+### Instagram
+
+import ConnectInstagram from './connect-instagram.mdx'
+
+<ConnectInstagram />
+
 ### Google
 
 import ConnectGoogle from './connect-google.mdx'
@@ -186,6 +192,16 @@ POST /channels.facebook.disconnect
 
 <ChannelDisconnect />
 
+### Instagram
+
+Disconnects an instagram account from Airy Core.
+
+```
+POST /channels.instagram.disconnect
+```
+
+<ChannelDisconnect />
+
 ### Google
 
 ```

diff --git a/docs/docs/api/endpoints/connect-instagram.mdx b/docs/docs/api/endpoints/connect-instagram.mdx
@@ -0,0 +1,38 @@
+Connects an Instagram account to Airy Core.
+
+```
+POST /channels.instagram.connect
+```
+
+- `page_id` is ID of the Facebook page connected to the Instagram account
+- `page_token` is the Access Token of the Facebook page
+- `account_id` is the ID of the Instagram account
+- `name` is the custom name for the connected page
+- `image_url` (optional) is the custom image URL
+
+**Sample request**
+
+```json5
+{
+  "page_id": "fb-page-id-1",
+  "account_id": "ig-account-id",
+  "page_token": "authentication token",
+  "name": "My custom name for this account",
+  "image_url": "https://example.org/custom-image.jpg" // optional
+}
+```
+
+**Sample response**
+
+```json5
+{
+  "id": "channel-uuid-1",
+  "source": "instagram",
+  "source_channel_id": "ig-account-id",
+  "metadata": {
+    "name": "My custom name for this account",
+    // optional
+    "image_url": "https://example.org/custom-image.jpg"
+  }
+}
+```