feat: add endpoint to get connector builder project by definiton ID (#13784)

## What
<!--
* Describe what the change is solving.
* It helps to add screenshots if it affects the frontend.
-->
* Close https://github.com/airbytehq/airbyte-internal-issues/issues/9377
* Add an endpoint which allows us to get a (optional) connector builder project ID for a given source definition ID. This allows us to enable the fork buttons
  * To show the fork button, we need to know if the given source def ID has a project associated with it. We then use this to be able to open the project associated with it. See https://github.com/airbytehq/airbyte-platform-internal/pull/13787 

## How
<!--
* Describe how code changes achieve the solution.
-->
* Add a new DB method to look up a connector builder project by actor definition ID. Return an optional, since we will be looking up arbitrary actor definitions (custom or catalog).
* Update the server to host the new endpoint

## Can this PR be safely reverted and rolled back?
<!--
* If you know that your be safely rolled back, check YES.*
* If that is not the case (e.g. a database migration), check NO.
* If unsure, leave it blank.*
-->
- [x] YES 💚
- [ ] NO ❌

Author: Ella Rohm-Ensing

airbyte-api/server-api/src/main/openapi/config.yaml

@@ -1029,6 +1029,25 @@ paths:
             application/json:
               schema:
                 $ref: "#/components/schemas/ConnectorBuilderProjectRead"
+  /v1/connector_builder_projects/get_for_definition_id:
+    post:
+      tags:
+        - connector_builder_project
+      summary: Get a connector builder project by source definition ID
+      operationId: getConnectorBuilderProjectIdForDefinitionId
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/BuilderProjectForDefinitionRequestBody"
+        required: true
+      responses:
+        "200":
+          description: Successful operation
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/BuilderProjectForDefinitionResponse"
   /v1/connector_builder_projects/update_testing_values:
     post:
       tags:
@@ -7742,6 +7761,25 @@ components:
           $ref: "#/components/schemas/WorkspaceId"
         builderProject:
           $ref: "#/components/schemas/ConnectorBuilderProjectDetails"
+    BuilderProjectForDefinitionRequestBody:
+      type: object
+      required:
+        - actorDefinitionId
+        - workspaceId
+      properties:
+        actorDefinitionId:
+          $ref: "#/components/schemas/ActorDefinitionId"
+        workspaceId:
+          $ref: "#/components/schemas/WorkspaceId"
+    BuilderProjectForDefinitionResponse:
+      type: object
+      required:
+        - builderProjectId
+      properties:
+        builderProjectId:
+          type: string
+          format: uuid
+          nullable: true
     ConnectorBuilderProjectRead:
       type: object
       required:

airbyte-commons-server/src/main/java/io/airbyte/commons/server/handlers/ConnectorBuilderProjectsHandler.java

@@ -8,6 +8,8 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import io.airbyte.api.model.generated.BuilderProjectForDefinitionRequestBody;
+import io.airbyte.api.model.generated.BuilderProjectForDefinitionResponse;
 import io.airbyte.api.model.generated.ConnectorBuilderHttpRequest;
 import io.airbyte.api.model.generated.ConnectorBuilderHttpResponse;
 import io.airbyte.api.model.generated.ConnectorBuilderProjectDetailsRead;
@@ -371,6 +373,13 @@ public ConnectorBuilderProjectStreamRead readConnectorBuilderProjectStream(final
     }
   }
 
+  public BuilderProjectForDefinitionResponse getConnectorBuilderProjectForDefinitionId(final BuilderProjectForDefinitionRequestBody requestBody)
+      throws IOException {
+    final Optional<UUID> builderProjectId =
+        connectorBuilderService.getConnectorBuilderProjectIdForActorDefinitionId(requestBody.getActorDefinitionId());
+    return new BuilderProjectForDefinitionResponse().builderProjectId(builderProjectId.orElse(null));
+  }
+
   private ConnectorBuilderProjectStreamRead convertStreamRead(final StreamRead streamRead) {
     return new ConnectorBuilderProjectStreamRead()
         .logs(streamRead.getLogs())

airbyte-commons-server/src/test/java/io/airbyte/commons/server/handlers/ConnectorBuilderProjectsHandlerTest.java

@@ -24,6 +24,8 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.airbyte.api.model.generated.BuilderProjectForDefinitionRequestBody;
+import io.airbyte.api.model.generated.BuilderProjectForDefinitionResponse;
 import io.airbyte.api.model.generated.ConnectorBuilderHttpRequest;
 import io.airbyte.api.model.generated.ConnectorBuilderHttpRequest.HttpMethodEnum;
 import io.airbyte.api.model.generated.ConnectorBuilderHttpResponse;
@@ -85,6 +87,7 @@
 import java.time.OffsetDateTime;
 import java.util.Collections;
 import java.util.List;
+import java.util.Optional;
 import java.util.UUID;
 import java.util.function.Supplier;
 import java.util.stream.Stream;
@@ -755,6 +758,31 @@ void testGetBaseImageForDeclarativeManifest() {
     assertEquals(A_BASE_IMAGE, responseBody.getBaseImage());
   }
 
+  @Test
+  void testGetConnectorBuilderProjectIdBySourceDefinitionId() throws IOException {
+    final UUID actorDefinitionId = UUID.randomUUID();
+    final UUID projectId = UUID.randomUUID();
+    final UUID workspaceId = UUID.randomUUID();
+    when(connectorBuilderService.getConnectorBuilderProjectIdForActorDefinitionId(actorDefinitionId)).thenReturn(Optional.of(projectId));
+
+    final BuilderProjectForDefinitionResponse response = connectorBuilderProjectsHandler.getConnectorBuilderProjectForDefinitionId(
+        new BuilderProjectForDefinitionRequestBody().actorDefinitionId(actorDefinitionId).workspaceId(workspaceId));
+
+    assertEquals(projectId, response.getBuilderProjectId());
+  }
+
+  @Test
+  void testGetConnectorBuilderProjectIdBySourceDefinitionIdWhenNotFound() throws IOException {
+    final UUID actorDefinitionId = UUID.randomUUID();
+    final UUID workspaceId = UUID.randomUUID();
+    when(connectorBuilderService.getConnectorBuilderProjectIdForActorDefinitionId(actorDefinitionId)).thenReturn(Optional.empty());
+
+    final BuilderProjectForDefinitionResponse response = connectorBuilderProjectsHandler.getConnectorBuilderProjectForDefinitionId(
+        new BuilderProjectForDefinitionRequestBody().actorDefinitionId(actorDefinitionId).workspaceId(workspaceId));
+
+    assertNull(response.getBuilderProjectId());
+  }
+
   private static ConnectorBuilderPublishRequestBody anyConnectorBuilderProjectRequest() {
     return new ConnectorBuilderPublishRequestBody().initialDeclarativeManifest(anyInitialManifest());
   }

airbyte-config/config-persistence/src/test/java/io/airbyte/config/persistence/ConnectorBuilderProjectPersistenceTest.java

@@ -49,6 +49,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Optional;
 import java.util.UUID;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -308,6 +309,20 @@ void testDeleteManifestDraftForActorDefinitionId() throws IOException, ConfigNot
     assertNull(connectorBuilderService.getConnectorBuilderProject(project1.getBuilderProjectId(), true).getManifestDraft());
   }
 
+  @Test
+  void testGetConnectorBuilderProjectIdByActorDefinitionId() throws IOException {
+    createBaseObjects();
+    final StandardSourceDefinition sourceDefinition = linkSourceDefinition(project1.getBuilderProjectId());
+    assertEquals(Optional.of(project1.getBuilderProjectId()),
+        connectorBuilderService.getConnectorBuilderProjectIdForActorDefinitionId(sourceDefinition.getSourceDefinitionId()));
+  }
+
+  @Test
+  void testGetConnectorBuilderProjectIdByActorDefinitionIdWhenNoMatch() throws IOException {
+    createBaseObjects();
+    assertEquals(Optional.empty(), connectorBuilderService.getConnectorBuilderProjectIdForActorDefinitionId(UUID.randomUUID()));
+  }
+
   private DeclarativeManifest anyDeclarativeManifest() {
     try {
       return new DeclarativeManifest()

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

@@ -12,6 +12,7 @@
 import io.airbyte.data.exceptions.ConfigNotFoundException;
 import io.airbyte.protocol.models.ConnectorSpecification;
 import java.io.IOException;
+import java.util.Optional;
 import java.util.UUID;
 import java.util.stream.Stream;
 
@@ -22,6 +23,8 @@ public interface ConnectorBuilderService {
 
   ConnectorBuilderProject getConnectorBuilderProject(UUID builderProjectId, boolean fetchManifestDraft) throws IOException, ConfigNotFoundException;
 
+  Optional<UUID> getConnectorBuilderProjectIdForActorDefinitionId(UUID actorDefinitionId) throws IOException;
+
   ConnectorBuilderProjectVersionedManifest getVersionedConnectorBuilderProject(UUID builderProjectId, Long version)
       throws ConfigNotFoundException, IOException;
 

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

@@ -95,6 +95,16 @@ public ConnectorBuilderProject getConnectorBuilderProject(final UUID builderProj
     return projectOptional.orElseThrow(() -> new ConfigNotFoundException(ConfigSchema.CONNECTOR_BUILDER_PROJECT, builderProjectId));
   }
 
+  @Override
+  public Optional<UUID> getConnectorBuilderProjectIdForActorDefinitionId(final UUID actorDefinitionId) throws IOException {
+    return database.query(ctx -> ctx
+        .select(CONNECTOR_BUILDER_PROJECT.ID)
+        .from(CONNECTOR_BUILDER_PROJECT)
+        .where(CONNECTOR_BUILDER_PROJECT.ACTOR_DEFINITION_ID.eq(actorDefinitionId)
+            .andNot(CONNECTOR_BUILDER_PROJECT.TOMBSTONE))
+        .fetchOptional(CONNECTOR_BUILDER_PROJECT.ID));
+  }
+
   /**
    * Return a versioned manifest associated with a builder project.
    *

airbyte-server/src/main/java/io/airbyte/server/apis/ConnectorBuilderProjectApiController.java

@@ -12,6 +12,8 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import io.airbyte.api.generated.ConnectorBuilderProjectApi;
+import io.airbyte.api.model.generated.BuilderProjectForDefinitionRequestBody;
+import io.airbyte.api.model.generated.BuilderProjectForDefinitionResponse;
 import io.airbyte.api.model.generated.ConnectorBuilderProjectIdWithWorkspaceId;
 import io.airbyte.api.model.generated.ConnectorBuilderProjectRead;
 import io.airbyte.api.model.generated.ConnectorBuilderProjectReadList;
@@ -40,6 +42,7 @@
 @Controller("/api/v1/connector_builder_projects")
 @Context
 @Secured(SecurityRule.IS_AUTHENTICATED)
+@SuppressWarnings("PMD.AvoidDuplicateLiterals")
 public class ConnectorBuilderProjectApiController implements ConnectorBuilderProjectApi {
 
   private final ConnectorBuilderProjectsHandler connectorBuilderProjectsHandler;
@@ -137,4 +140,15 @@ public JsonNode updateConnectorBuilderProjectTestingValues(@Body final Connector
         .execute(() -> connectorBuilderProjectsHandler.updateConnectorBuilderProjectTestingValues(connectorBuilderProjectTestingValuesUpdate));
   }
 
+  @Override
+  @Post(uri = "/get_by_definition_id")
+  @Status(HttpStatus.OK)
+  @Secured({WORKSPACE_READER, ORGANIZATION_READER})
+  @ExecuteOn(AirbyteTaskExecutors.IO)
+  @SuppressWarnings("LineLength")
+  public BuilderProjectForDefinitionResponse getConnectorBuilderProjectIdForDefinitionId(@Body final BuilderProjectForDefinitionRequestBody builderProjectForDefinitionRequestBody) {
+    return ApiHelper
+        .execute(() -> connectorBuilderProjectsHandler.getConnectorBuilderProjectForDefinitionId(builderProjectForDefinitionRequestBody));
+  }
+
 }