Instagram and Facebook Pages oAuth backend (#7412)

airbyte-oauth/src/main/java/io/airbyte/oauth/BaseOAuthFlow.java

@@ -169,11 +169,11 @@ public Map<String, Object> completeDestinationOAuth(final UUID workspaceId,
         redirectUrl, oAuthParamConfig);
   }
 
-  private Map<String, Object> completeOAuthFlow(final String clientId,
-                                                final String clientSecret,
-                                                final String authCode,
-                                                final String redirectUrl,
-                                                JsonNode oAuthParamConfig)
+  protected Map<String, Object> completeOAuthFlow(final String clientId,
+                                                  final String clientSecret,
+                                                  final String authCode,
+                                                  final String redirectUrl,
+                                                  JsonNode oAuthParamConfig)
       throws IOException {
     var accessTokenUrl = getAccessTokenUrl();
     final HttpRequest request = HttpRequest.newBuilder()

airbyte-oauth/src/main/java/io/airbyte/oauth/OAuthImplementationFactory.java

@@ -7,10 +7,12 @@
 import com.google.common.collect.ImmutableMap;
 import io.airbyte.config.persistence.ConfigRepository;
 import io.airbyte.oauth.flows.AsanaOAuthFlow;
-import io.airbyte.oauth.flows.FacebookMarketingOAuthFlow;
 import io.airbyte.oauth.flows.GithubOAuthFlow;
 import io.airbyte.oauth.flows.SalesforceOAuthFlow;
 import io.airbyte.oauth.flows.TrelloOAuthFlow;
+import io.airbyte.oauth.flows.facebook.FacebookMarketingOAuthFlow;
+import io.airbyte.oauth.flows.facebook.FacebookPagesOAuthFlow;
+import io.airbyte.oauth.flows.facebook.InstagramOAuthFlow;
 import io.airbyte.oauth.flows.google.GoogleAdsOAuthFlow;
 import io.airbyte.oauth.flows.google.GoogleAnalyticsOAuthFlow;
 import io.airbyte.oauth.flows.google.GoogleSearchConsoleOAuthFlow;
@@ -26,11 +28,13 @@ public OAuthImplementationFactory(final ConfigRepository configRepository) {
     OAUTH_FLOW_MAPPING = ImmutableMap.<String, OAuthFlowImplementation>builder()
         .put("airbyte/source-asana", new AsanaOAuthFlow(configRepository))
         .put("airbyte/source-facebook-marketing", new FacebookMarketingOAuthFlow(configRepository))
+        .put("airbyte/source-facebook-pages", new FacebookPagesOAuthFlow(configRepository))
         .put("airbyte/source-github", new GithubOAuthFlow(configRepository))
         .put("airbyte/source-google-ads", new GoogleAdsOAuthFlow(configRepository))
         .put("airbyte/source-google-analytics-v4", new GoogleAnalyticsOAuthFlow(configRepository))
         .put("airbyte/source-google-search-console", new GoogleSearchConsoleOAuthFlow(configRepository))
         .put("airbyte/source-google-sheets", new GoogleSheetsOAuthFlow(configRepository))
+        .put("airbyte/source-instagram", new InstagramOAuthFlow(configRepository))
         .put("airbyte/source-salesforce", new SalesforceOAuthFlow(configRepository))
         .put("airbyte/source-trello", new TrelloOAuthFlow(configRepository))
         .build();

airbyte-oauth/src/main/java/io/airbyte/oauth/flows/facebook/FacebookMarketingOAuthFlow.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.oauth.flows.facebook;
+
+import com.google.common.annotations.VisibleForTesting;
+import io.airbyte.config.persistence.ConfigRepository;
+import java.net.http.HttpClient;
+import java.util.function.Supplier;
+
+public class FacebookMarketingOAuthFlow extends FacebookOAuthFlow {
+
+  private static final String SCOPES = "ads_management,ads_read,read_insights";
+
+  public FacebookMarketingOAuthFlow(final ConfigRepository configRepository) {
+    super(configRepository);
+  }
+
+  @VisibleForTesting
+  FacebookMarketingOAuthFlow(final ConfigRepository configRepository, final HttpClient httpClient, final Supplier<String> stateSupplier) {
+    super(configRepository, httpClient, stateSupplier);
+  }
+
+  @Override
+  protected String getScopes() {
+    return SCOPES;
+  }
+
+}

airbyte-oauth/src/main/java/io/airbyte/oauth/flows/facebook/FacebookOAuthFlow.java

@@ -0,0 +1,123 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.oauth.flows.facebook;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.config.persistence.ConfigRepository;
+import io.airbyte.oauth.BaseOAuthFlow;
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Supplier;
+import org.apache.http.client.utils.URIBuilder;
+
+/**
+ * Following docs from
+ * https://developers.facebook.com/docs/facebook-login/manually-build-a-login-flow
+ */
+public abstract class FacebookOAuthFlow extends BaseOAuthFlow {
+
+  private static final String ACCESS_TOKEN_URL = "https://graph.facebook.com/v12.0/oauth/access_token";
+  private static final String AUTH_CODE_TOKEN_URL = "https://www.facebook.com/v12.0/dialog/oauth";
+
+  public FacebookOAuthFlow(final ConfigRepository configRepository) {
+    super(configRepository);
+  }
+
+  @VisibleForTesting
+  FacebookOAuthFlow(final ConfigRepository configRepository, final HttpClient httpClient, final Supplier<String> stateSupplier) {
+    super(configRepository, httpClient, stateSupplier);
+  }
+
+  protected abstract String getScopes();
+
+  @Override
+  protected String formatConsentUrl(final UUID definitionId, final String clientId, final String redirectUrl) throws IOException {
+    try {
+      return new URIBuilder(AUTH_CODE_TOKEN_URL)
+          .addParameter("client_id", clientId)
+          .addParameter("redirect_uri", redirectUrl)
+          .addParameter("state", getState())
+          .addParameter("scope", getScopes())
+          .build().toString();
+    } catch (final URISyntaxException e) {
+      throw new IOException("Failed to format Consent URL for OAuth flow", e);
+    }
+  }
+
+  @Override
+  protected String getAccessTokenUrl() {
+    return ACCESS_TOKEN_URL;
+  }
+
+  @Override
+  protected Map<String, Object> extractRefreshToken(final JsonNode data, String accessTokenUrl) throws IOException {
+    // Facebook does not have refresh token but calls it "long lived access token" instead:
+    // see https://developers.facebook.com/docs/facebook-login/access-tokens/refreshing
+    Preconditions.checkArgument(data.has("access_token"), "Missing 'access_token' in query params from %s", ACCESS_TOKEN_URL);
+    return Map.of("access_token", data.get("access_token").asText());
+  }
+
+  @Override
+  protected Map<String, Object> completeOAuthFlow(final String clientId,
+                                                  final String clientSecret,
+                                                  final String authCode,
+                                                  final String redirectUrl,
+                                                  JsonNode oAuthParamConfig)
+      throws IOException {
+    // Access tokens generated via web login are short-lived tokens
+    // they arre valid for 1 hour and need to be exchanged for long-lived access token
+    // https://developers.facebook.com/docs/facebook-login/access-tokens (Short-Term Tokens and
+    // https://developers.facebook.com/docs/instagram-basic-display-api/overview#short-lived-access-tokens
+    // Long-Term Tokens section)
+
+    final Map<String, Object> data = super.completeOAuthFlow(clientId, clientSecret, authCode, redirectUrl, oAuthParamConfig);
+    Preconditions.checkArgument(data.containsKey("access_token"));
+    final String shortLivedAccessToken = (String) data.get("access_token");
+    final String longLivedAccessToken = getLongLivedAccessToken(clientId, clientSecret, shortLivedAccessToken);
+    return Map.of("access_token", longLivedAccessToken);
+  }
+
+  protected URI createLongLivedTokenURI(final String clientId, final String clientSecret, final String shortLivedAccessToken)
+      throws URISyntaxException {
+    // Exchange Short-lived Access token for Long-lived one
+    // https://developers.facebook.com/docs/facebook-login/access-tokens/refreshing
+    // It's valid for 60 days and resreshed once per day if using in requests.
+    // If no requests are made, the token will expire after about 60 days and
+    // the person will have to go through the login flow again to get a new
+    // token.
+    return new URIBuilder(ACCESS_TOKEN_URL)
+        .addParameter("client_secret", clientSecret)
+        .addParameter("client_id", clientId)
+        .addParameter("grant_type", "fb_exchange_token")
+        .addParameter("fb_exchange_token", shortLivedAccessToken)
+        .build();
+  }
+
+  protected String getLongLivedAccessToken(final String clientId, final String clientSecret, final String shortLivedAccessToken) throws IOException {
+    try {
+      final URI uri = createLongLivedTokenURI(clientId, clientSecret, shortLivedAccessToken);
+      final HttpRequest request = HttpRequest.newBuilder()
+          .GET()
+          .uri(uri)
+          .build();
+      final HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
+      final JsonNode responseJson = Jsons.deserialize(response.body());
+      Preconditions.checkArgument(responseJson.hasNonNull("access_token"), "%s response should have access_token", responseJson);
+      return responseJson.get("access_token").asText();
+    } catch (final InterruptedException | URISyntaxException e) {
+      throw new IOException("Failed to complete OAuth flow", e);
+    }
+  }
+
+}

airbyte-oauth/src/main/java/io/airbyte/oauth/flows/facebook/FacebookPagesOAuthFlow.java

@@ -0,0 +1,22 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.oauth.flows.facebook;
+
+import io.airbyte.config.persistence.ConfigRepository;
+
+public class FacebookPagesOAuthFlow extends FacebookOAuthFlow {
+
+  private static final String SCOPES = "pages_manage_ads,pages_manage_metadata,pages_read_engagement,pages_read_user_content";
+
+  public FacebookPagesOAuthFlow(final ConfigRepository configRepository) {
+    super(configRepository);
+  }
+
+  @Override
+  protected String getScopes() {
+    return SCOPES;
+  }
+
+}

airbyte-oauth/src/main/java/io/airbyte/oauth/flows/facebook/InstagramOAuthFlow.java

@@ -0,0 +1,23 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.oauth.flows.facebook;
+
+import io.airbyte.config.persistence.ConfigRepository;
+
+// Instagram Graph API require Facebook API User token
+public class InstagramOAuthFlow extends FacebookMarketingOAuthFlow {
+
+  private static final String SCOPES = "ads_management,instagram_basic,instagram_manage_insights,read_insights";
+
+  public InstagramOAuthFlow(final ConfigRepository configRepository) {
+    super(configRepository);
+  }
+
+  @Override
+  protected String getScopes() {
+    return SCOPES;
+  }
+
+}

airbyte-oauth/src/test-integration/java/io.airbyte.oauth.flows/FacebookMarketingOAuthFlowIntegrationTest.java → airbyte-oauth/src/test-integration/java/io.airbyte.oauth.flows/FacebookOAuthFlowIntegrationTest.java

@@ -2,7 +2,7 @@
  * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
  */
 
-package io.airbyte.oauth.flows;
+package io.airbyte.oauth.flows.facebook;
 
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.when;
@@ -14,6 +14,7 @@
 import io.airbyte.config.persistence.ConfigNotFoundException;
 import io.airbyte.config.persistence.ConfigRepository;
 import io.airbyte.oauth.OAuthFlowImplementation;
+import io.airbyte.oauth.flows.OAuthFlowIntegrationTest;
 import io.airbyte.validation.json.JsonValidationException;
 import java.io.IOException;
 import java.nio.file.Files;
@@ -24,9 +25,10 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
-public class FacebookMarketingOAuthFlowIntegrationTest extends OAuthFlowIntegrationTest {
+public class FacebookOAuthFlowIntegrationTest extends OAuthFlowIntegrationTest {
 
   protected static final Path CREDENTIALS_PATH = Path.of("secrets/facebook_marketing.json");
+  protected static final String REDIRECT_URL = "http://localhost:9000/auth_flow";
 
   @Override
   protected Path get_credentials_path() {
@@ -43,9 +45,13 @@ public void setup() throws IOException {
     super.setup();
   }
 
+  @Override
+  protected int getServerListeningPort() {
+    return 9000;
+  }
+
   @Test
-  public void testFullGoogleOAuthFlow() throws InterruptedException, ConfigNotFoundException, IOException, JsonValidationException {
-    int limit = 20;
+  public void testFullFacebookOAuthFlow() throws InterruptedException, ConfigNotFoundException, IOException, JsonValidationException {
     final UUID workspaceId = UUID.randomUUID();
     final UUID definitionId = UUID.randomUUID();
     final String fullConfigAsString = new String(Files.readAllBytes(CREDENTIALS_PATH));
@@ -60,12 +66,7 @@ public void testFullGoogleOAuthFlow() throws InterruptedException, ConfigNotFoun
             .build()))));
     final String url = flow.getSourceConsentUrl(workspaceId, definitionId, REDIRECT_URL);
     LOGGER.info("Waiting for user consent at: {}", url);
-    // TODO: To automate, start a selenium job to navigate to the Consent URL and click on allowing
-    // access...
-    while (!serverHandler.isSucceeded() && limit > 0) {
-      Thread.sleep(1000);
-      limit -= 1;
-    }
+    waitForResponse(20);
     assertTrue(serverHandler.isSucceeded(), "Failed to get User consent on time");
     final Map<String, Object> params = flow.completeSourceOAuth(workspaceId, definitionId,
         Map.of("code", serverHandler.getParamValue()), REDIRECT_URL);

airbyte-oauth/src/test-integration/java/io/airbyte/oauth/flows/OAuthFlowIntegrationTest.java

@@ -71,7 +71,16 @@ void tearDown() {
     server.stop(1);
   }
 
-  static class ServerHandler implements HttpHandler {
+  protected void waitForResponse(int limit) throws InterruptedException {
+    // TODO: To automate, start a selenium job to navigate to the Consent URL and click on allowing
+    // access...
+    while (!serverHandler.isSucceeded() && limit > 0) {
+      Thread.sleep(1000);
+      limit -= 1;
+    }
+  }
+
+  public static class ServerHandler implements HttpHandler {
 
     final private String expectedParam;
     private String paramValue;