remove config persistence from seeding logic (#18749)

airbyte-bootloader/src/test/java/io/airbyte/bootloader/BootloaderAppTest.java

@@ -24,10 +24,8 @@
 import io.airbyte.config.Geography;
 import io.airbyte.config.SourceConnection;
 import io.airbyte.config.StandardWorkspace;
-import io.airbyte.config.init.DefinitionProviderToConfigPersistenceAdapter;
 import io.airbyte.config.init.DefinitionsProvider;
 import io.airbyte.config.init.LocalDefinitionsProvider;
-import io.airbyte.config.persistence.ConfigPersistence;
 import io.airbyte.config.persistence.ConfigRepository;
 import io.airbyte.config.persistence.DatabaseConfigPersistence;
 import io.airbyte.config.persistence.SecretsRepositoryReader;
@@ -213,8 +211,7 @@ void testBootloaderAppRunSecretMigration() throws Exception {
       initBootloader.load();
 
       final DefinitionsProvider localDefinitions = new LocalDefinitionsProvider(LocalDefinitionsProvider.DEFAULT_SEED_DEFINITION_RESOURCE_CLASS);
-      final ConfigPersistence localConfigPersistence = new DefinitionProviderToConfigPersistenceAdapter(localDefinitions);
-      configRepository.loadDataNoSecrets(localConfigPersistence);
+      configRepository.seedActorDefinitions(localDefinitions.getSourceDefinitions(), localDefinitions.getDestinationDefinitions());
 
       final String sourceSpecs = """
                                  {

airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/ClassEnforcingConfigPersistence.java

@@ -91,9 +91,4 @@ public Map<String, Stream<JsonNode>> dumpConfigs() throws IOException {
     return decoratedPersistence.dumpConfigs();
   }
 
-  @Override
-  public void loadData(final ConfigPersistence seedPersistence) throws IOException {
-    decoratedPersistence.loadData(seedPersistence);
-  }
-
 }

airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/ConfigPersistence.java

@@ -38,6 +38,4 @@ <T> ConfigWithMetadata<T> getConfigWithMetadata(AirbyteConfig configType, String
 
   Map<String, Stream<JsonNode>> dumpConfigs() throws IOException;
 
-  void loadData(ConfigPersistence seedPersistence) throws IOException;
-
 }

airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/ConfigRepository.java

@@ -94,14 +94,17 @@ public class ConfigRepository {
 
   private final ConfigPersistence persistence;
   private final ExceptionWrappingDatabase database;
+  private final ActorDefinitionMigrator actorDefinitionMigrator;
 
   public ConfigRepository(final ConfigPersistence persistence, final Database database) {
-    this.persistence = persistence;
-    this.database = new ExceptionWrappingDatabase(database);
+    this(persistence, database, new ActorDefinitionMigrator(new ExceptionWrappingDatabase(database)));
   }
 
-  public ConfigPersistence getConfigPersistence() {
-    return persistence;
+  @VisibleForTesting
+  ConfigRepository(final ConfigPersistence persistence, final Database database, final ActorDefinitionMigrator actorDefinitionMigrator) {
+    this.persistence = persistence;
+    this.database = new ExceptionWrappingDatabase(database);
+    this.actorDefinitionMigrator = actorDefinitionMigrator;
   }
 
   /**
@@ -657,7 +660,7 @@ public List<DestinationConnection> listDestinationConnection() throws JsonValida
    * @return destinations
    * @throws IOException - you never know when you IO
    */
-  public List<DestinationConnection> listWorkspaceDestinationConnection(UUID workspaceId) throws IOException {
+  public List<DestinationConnection> listWorkspaceDestinationConnection(final UUID workspaceId) throws IOException {
     final Result<Record> result = database.query(ctx -> ctx.select(asterisk())
         .from(ACTOR)
         .where(ACTOR.ACTOR_TYPE.eq(ActorType.destination))
@@ -673,7 +676,7 @@ public List<DestinationConnection> listWorkspaceDestinationConnection(UUID works
    * @return sources
    * @throws IOException
    */
-  public List<SourceConnection> listSourcesForDefinition(UUID definitionId) throws IOException {
+  public List<SourceConnection> listSourcesForDefinition(final UUID definitionId) throws IOException {
     final Result<Record> result = database.query(ctx -> ctx.select(asterisk())
         .from(ACTOR)
         .where(ACTOR.ACTOR_TYPE.eq(ActorType.source))
@@ -689,7 +692,7 @@ public List<SourceConnection> listSourcesForDefinition(UUID definitionId) throws
    * @return destinations
    * @throws IOException
    */
-  public List<DestinationConnection> listDestinationsForDefinition(UUID definitionId) throws IOException {
+  public List<DestinationConnection> listDestinationsForDefinition(final UUID definitionId) throws IOException {
     final Result<Record> result = database.query(ctx -> ctx.select(asterisk())
         .from(ACTOR)
         .where(ACTOR.ACTOR_TYPE.eq(ActorType.destination))
@@ -915,6 +918,19 @@ private Map<UUID, AirbyteCatalog> findCatalogByHash(final String catalogHash, fi
     return result;
   }
 
+  /**
+   * Updates the database with the most up-to-date source and destination definitions in the connector
+   * catalog.
+   *
+   * @param seedSourceDefs - most up-to-date source definitions
+   * @param seedDestDefs - most up-to-date destination definitions
+   * @throws IOException - throws if exception when interacting with db
+   */
+  public void seedActorDefinitions(final List<StandardSourceDefinition> seedSourceDefs, final List<StandardDestinationDefinition> seedDestDefs)
+      throws IOException {
+    actorDefinitionMigrator.migrate(seedSourceDefs, seedDestDefs);
+  }
+
   // Data-carrier records to hold combined result of query for a Source or Destination and its
   // corresponding Definition. This enables the API layer to
   // process combined information about a Source/Destination/Definition pair without requiring two
@@ -1146,19 +1162,6 @@ public Map<String, Stream<JsonNode>> dumpConfigsNoSecrets() throws IOException {
     return persistence.dumpConfigs();
   }
 
-  /**
-   * MUST NOT ACCEPT SECRETS - Package private so that secrets are not accidentally passed in. Should
-   * only be called from { @link SecretsRepositoryWriter }
-   *
-   * Loads all Data from a ConfigPersistence into the database.
-   *
-   * @param seedPersistenceWithoutSecrets - seed persistence WITHOUT secrets
-   * @throws IOException - you never know when you IO
-   */
-  public void loadDataNoSecrets(final ConfigPersistence seedPersistenceWithoutSecrets) throws IOException {
-    persistence.loadData(seedPersistenceWithoutSecrets);
-  }
-
   /**
    * The following methods are present to allow the JobCreationAndStatusUpdateActivity class to emit
    * metrics without exposing the underlying database connection.

airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/DatabaseConfigPersistence.java

@@ -1638,14 +1638,6 @@ public Map<String, Stream<JsonNode>> dumpConfigs() throws IOException {
     return result;
   }
 
-  @Override
-  public void loadData(final ConfigPersistence seedConfigPersistence) throws IOException {
-    database.transaction(ctx -> {
-      updateConfigsFromSeed(ctx, seedConfigPersistence);
-      return null;
-    });
-  }
-
   @VisibleForTesting
   void updateConfigsFromSeed(final DSLContext ctx, final ConfigPersistence seedConfigPersistence) throws SQLException {
     LOGGER.info("Updating connector definitions from the seed if necessary...");
@@ -1940,8 +1932,8 @@ static class ConnectorInfo {
     ConnectorInfo(final String definitionId, final JsonNode definition) {
       this.definitionId = definitionId;
       this.definition = definition;
-      this.dockerRepository = definition.get("dockerRepository").asText();
-      this.dockerImageTag = definition.get("dockerImageTag").asText();
+      dockerRepository = definition.get("dockerRepository").asText();
+      dockerImageTag = definition.get("dockerImageTag").asText();
     }
 
     @Override

airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/ValidatingConfigPersistence.java

@@ -115,11 +115,6 @@ public Map<String, Stream<JsonNode>> dumpConfigs() throws IOException {
     return decoratedPersistence.dumpConfigs();
   }
 
-  @Override
-  public void loadData(final ConfigPersistence seedPersistence) throws IOException {
-    decoratedPersistence.loadData(seedPersistence);
-  }
-
   private <T> void validateJson(final T config, final AirbyteConfig configType) throws JsonValidationException {
     final JsonNode schema = JsonSchemaValidator.getSchema(configType.getConfigSchemaFile());
     schemaValidator.ensure(schema, Jsons.jsonNode(config));

airbyte-config/init/src/main/java/io/airbyte/config/init/ApplyDefinitionsHelper.java

@@ -6,7 +6,6 @@
 
 import io.airbyte.config.StandardDestinationDefinition;
 import io.airbyte.config.StandardSourceDefinition;
-import io.airbyte.config.persistence.ConfigPersistence;
 import io.airbyte.config.persistence.ConfigRepository;
 import io.airbyte.validation.json.JsonValidationException;
 import java.io.IOException;
@@ -49,9 +48,9 @@ public void apply(final boolean updateAll) throws JsonValidationException, IOExc
     } else {
       // todo (pedroslopez): Logic to apply definitions should be moved outside of the
       // DatabaseConfigPersistence class and behavior standardized
-      final ConfigPersistence dbConfigPersistence = configRepository.getConfigPersistence();
-      final ConfigPersistence providerConfigPersistence = new DefinitionProviderToConfigPersistenceAdapter(definitionsProvider);
-      dbConfigPersistence.loadData(providerConfigPersistence);
+      configRepository.seedActorDefinitions(
+          definitionsProvider.getSourceDefinitions(),
+          definitionsProvider.getDestinationDefinitions());
     }
   }