feat(F18): plugin-to-core connector bridge

syncflow-api/src/main/java/com/syncflow/api/plugin/PluginManager.java

@@ -1,5 +1,7 @@
 package com.syncflow.api.plugin;
 
+import com.syncflow.api.plugin.adapter.PluginConnectorAdapter;
+import com.syncflow.api.plugin.registry.DelegatingConnectorRegistry;
 import com.syncflow.plugin.descriptor.PluginDescriptor;
 import com.syncflow.plugin.lifecycle.PluginLifecycle;
 import com.syncflow.plugin.spi.PluginConnector;
@@ -19,6 +21,19 @@ public class PluginManager {
 
     private final Map<String, PluginEntry> plugins = new ConcurrentHashMap<>();
     private final Map<String, URLClassLoader> classLoaders = new ConcurrentHashMap<>();
+    private final Map<String, PluginConnectorAdapter> adapters = new ConcurrentHashMap<>();
+    private final DelegatingConnectorRegistry registry;
+
+    public PluginManager(DelegatingConnectorRegistry registry) {
+        this.registry = registry;
+    }
+
+    /**
+     * Test-only constructor — registry operations are no-ops when registry is null.
+     */
+    PluginManager() {
+        this.registry = null;
+    }
 
     public PluginInstallResult install(File jarFile) {
         try (var jar = new JarFile(jarFile)) {
@@ -64,6 +79,15 @@ public boolean enable(String pluginId) {
         var entry = plugins.get(pluginId);
         if (entry == null)
             return false;
+        // Already enabled? Idempotent.
+        if (entry.lifecycle() == PluginLifecycle.ENABLED) {
+            return true;
+        }
+        var adapter = adapters.computeIfAbsent(pluginId,
+                id -> new PluginConnectorAdapter(entry.connector(), entry.descriptor()));
+        if (registry != null) {
+            registry.register(adapter);
+        }
         plugins.put(pluginId, new PluginEntry(entry.descriptor(), entry.connector(), PluginLifecycle.ENABLED));
         return true;
     }
@@ -72,12 +96,19 @@ public boolean disable(String pluginId) {
         var entry = plugins.get(pluginId);
         if (entry == null)
             return false;
+        if (registry != null) {
+            registry.unregisterPlugin(pluginId);
+        }
         plugins.put(pluginId, new PluginEntry(entry.descriptor(), entry.connector(), PluginLifecycle.DISABLED));
         return true;
     }
 
     public boolean uninstall(String pluginId) {
         var removed = plugins.remove(pluginId);
+        if (registry != null) {
+            registry.unregisterPlugin(pluginId);
+        }
+        adapters.remove(pluginId);
         var cl = classLoaders.remove(pluginId);
         if (cl != null) {
             try {

syncflow-api/src/main/java/com/syncflow/api/plugin/adapter/ConnectorTypeResolver.java

@@ -0,0 +1,35 @@
+package com.syncflow.api.plugin.adapter;
+
+import com.syncflow.core.model.ConnectorType;
+
+import java.util.Locale;
+
+/**
+ * Maps a free-form {@code PluginDescriptor.connectorType} string to the
+ * closed {@link ConnectorType} enum. Recognized database names map to the
+ * matching enum value; everything else collapses to
+ * {@link ConnectorType#GENERIC_PLUGIN}.
+ */
+public final class ConnectorTypeResolver {
+
+    private ConnectorTypeResolver() {
+    }
+
+    public static ConnectorType resolve(String pluginType) {
+        if (pluginType == null || pluginType.isBlank()) {
+            return ConnectorType.GENERIC_PLUGIN;
+        }
+        return switch (pluginType.toLowerCase(Locale.ROOT)) {
+            case "postgresql", "postgres" -> ConnectorType.POSTGRESQL;
+            case "mysql", "mariadb" -> ConnectorType.MYSQL;
+            case "mongodb", "mongo" -> ConnectorType.MONGODB;
+            case "kafka" -> ConnectorType.KAFKA;
+            case "sqlserver", "mssql" -> ConnectorType.SQLSERVER;
+            case "oracle" -> ConnectorType.ORACLE;
+            case "elasticsearch", "elastic", "es" -> ConnectorType.ELASTICSEARCH;
+            case "redis" -> ConnectorType.REDIS;
+            case "jdbc", "generic_jdbc" -> ConnectorType.GENERIC_JDBC;
+            default -> ConnectorType.GENERIC_PLUGIN;
+        };
+    }
+}

syncflow-api/src/main/java/com/syncflow/api/plugin/adapter/PluginCapabilitiesAdapter.java

@@ -0,0 +1,31 @@
+package com.syncflow.api.plugin.adapter;
+
+import com.syncflow.core.spi.ConnectorCapabilities;
+
+/**
+ * Maps the 6-boolean plugin capabilities to the 5-boolean core capabilities.
+ *
+ * <p>
+ * Plugin SPI: metadata, snapshot, cdc, destination, transactions, streaming.
+ * <br>
+ * Core SPI: cdc, snapshot, schemaDiscovery, transactions, offsetTracking.
+ */
+public final class PluginCapabilitiesAdapter {
+
+    private PluginCapabilitiesAdapter() {
+    }
+
+    public static ConnectorCapabilities map(
+            com.syncflow.plugin.capabilities.ConnectorCapabilities plugin) {
+        if (plugin == null) {
+            return ConnectorCapabilities.none();
+        }
+        return new ConnectorCapabilities(
+                plugin.supportsCdc(),
+                plugin.supportsSnapshot(),
+                plugin.supportsMetadata(), // → schemaDiscovery
+                plugin.supportsTransactions(),
+                // No direct mapping: streaming/cdc imply offset tracking.
+                plugin.supportsCdc() || plugin.supportsStreaming());
+    }
+}

syncflow-api/src/main/java/com/syncflow/api/plugin/adapter/PluginCdcEventAdapter.java

@@ -0,0 +1,58 @@
+package com.syncflow.api.plugin.adapter;
+
+import com.syncflow.core.cdc.CDCOperation;
+import com.syncflow.core.cdc.CDCEvent;
+import com.syncflow.core.cdc.EventHeader;
+import com.syncflow.core.cdc.EventMetadata;
+import com.syncflow.core.cdc.EventPayload;
+import com.syncflow.core.cdc.EventSource;
+import com.syncflow.core.cdc.OffsetInformation;
+import com.syncflow.plugin.descriptor.PluginDescriptor;
+import com.syncflow.plugin.spi.CdcProvider;
+
+import java.time.Instant;
+import java.util.Locale;
+import java.util.Map;
+
+/**
+ * Converts the flat {@link CdcProvider.CdcEvent} emitted by a plugin into
+ * the deeply nested core {@link CDCEvent}. Synthesizes fields the plugin
+ * SPI does not provide (pipelineId, connectionId, eventNumber, version,
+ * capturedAt, captureLatencyMs, transaction) with sensible defaults.
+ */
+public final class PluginCdcEventAdapter {
+
+    private PluginCdcEventAdapter() {
+    }
+
+    public static CDCEvent toCore(CdcProvider.CdcEvent event, PluginDescriptor descriptor) {
+        var eventId = event.eventId() != null
+                ? event.eventId()
+                : java.util.UUID.randomUUID().toString();
+        var now = Instant.now();
+        var pipelineId = "plugin:" + descriptor.pluginId();
+        var offsetMap = event.offset() == null ? Map.<String, String>of() : event.offset();
+
+        return new CDCEvent(
+                new EventHeader(eventId, pipelineId, pipelineId, 0L, 1, Map.of()),
+                new EventSource(descriptor.pluginId(), event.schema(), event.table(),
+                        descriptor.connectorType()),
+                parseOperation(event.operation()),
+                new EventPayload(event.before(), event.after(), Map.of()),
+                new EventMetadata(0L, now, 0L),
+                null,
+                new OffsetInformation(descriptor.connectorType(), offsetMap, null, now));
+    }
+
+    static CDCOperation parseOperation(String op) {
+        if (op == null || op.isBlank()) {
+            return CDCOperation.READ;
+        }
+        return switch (op.trim().toUpperCase(Locale.ROOT)) {
+            case "INSERT", "I", "CREATE" -> CDCOperation.INSERT;
+            case "UPDATE", "U" -> CDCOperation.UPDATE;
+            case "DELETE", "D" -> CDCOperation.DELETE;
+            default -> CDCOperation.READ;
+        };
+    }
+}