refactor(cdc): make connector identifiers semantically specific

cdc-service/src/main/java/com/xtrmetl/cdc/config/XtrmetlProperties.java

@@ -57,33 +57,84 @@ public void setSources(java.util.List<Source> sources) {
         }
 
         private static Source defaultPostgresSource() {
-            Source source = new Source();
-            source.setId("pg-main");
-            source.setType("postgres-debezium");
-            source.setEnabled(true);
-            return source;
+            Source sourceConfiguration = new Source();
+            sourceConfiguration.setSourceId("pg-main");
+            sourceConfiguration.setSourceType("postgres-debezium");
+            sourceConfiguration.setEnabled(true);
+            return sourceConfiguration;
         }
     }
 
+    /**
+     * Spring configuration adapter for one CDC source declaration.
+     *
+     * <p>The authoritative Java names are {@code sourceId}/{@code sourceType}. Legacy
+     * {@code id}/{@code type} bean accessors remain only so existing YAML continues to bind
+     * without a breaking configuration migration.</p>
+     */
     public static class Source {
-        private String id = "pg-main";
-        private String type = "postgres-debezium";
+        private String sourceId = "pg-main";
+        private String sourceType = "postgres-debezium";
         private boolean enabled = true;
 
+        public String getSourceId() {
+            return sourceId;
+        }
+
+        public void setSourceId(String sourceId) {
+            this.sourceId = sourceId;
+        }
+
+        public String getSourceType() {
+            return sourceType;
+        }
+
+        public void setSourceType(String sourceType) {
+            this.sourceType = sourceType;
+        }
+
+        /**
+         * Legacy Spring/YAML compatibility accessor for the historical {@code id} key.
+         *
+         * @return the configured CDC source identifier
+         * @deprecated internal callers must use {@link #getSourceId()}
+         */
+        @Deprecated(forRemoval = false)
         public String getId() {
-            return id;
+            return sourceId;
         }
 
-        public void setId(String id) {
-            this.id = id;
+        /**
+         * Legacy Spring/YAML compatibility mutator for the historical {@code id} key.
+         *
+         * @param legacySourceId configured CDC source identifier
+         * @deprecated internal callers must use {@link #setSourceId(String)}
+         */
+        @Deprecated(forRemoval = false)
+        public void setId(String legacySourceId) {
+            this.sourceId = legacySourceId;
         }
 
+        /**
+         * Legacy Spring/YAML compatibility accessor for the historical {@code type} key.
+         *
+         * @return the configured CDC source connector type
+         * @deprecated internal callers must use {@link #getSourceType()}
+         */
+        @Deprecated(forRemoval = false)
         public String getType() {
-            return type;
+            return sourceType;
         }
 
-        public void setType(String type) {
-            this.type = type;
+        /**
+         * Legacy Spring/YAML compatibility mutator for the historical {@code type} key.
+         *
+         * @param legacySourceType configured CDC source connector type
+         * @deprecated internal callers must use {@link #setSourceType(String)}
+         */
+        @Deprecated(forRemoval = false)
+        public void setType(String legacySourceType) {
+            this.sourceType = legacySourceType;
         }
 
         public boolean isEnabled() {

cdc-service/src/main/java/com/xtrmetl/cdc/controller/CdcController.java

@@ -24,22 +24,22 @@
 public class CdcController {
 
     private final CdcService cdcService;
-    private final XtrmetlProperties properties;
+    private final XtrmetlProperties xtrmetlProperties;
     private final CdcSourceRegistry sourceRegistry;
     private final CdcTargetRegistry targetRegistry;
     private final CdcSourceFactory sourceFactory;
     private final ReplicationSlotProbe replicationSlotProbe;
 
     public CdcController(
             CdcService cdcService,
-            XtrmetlProperties properties,
+            XtrmetlProperties xtrmetlProperties,
             CdcSourceRegistry sourceRegistry,
             CdcTargetRegistry targetRegistry,
             CdcSourceFactory sourceFactory,
             ReplicationSlotProbe replicationSlotProbe
     ) {
         this.cdcService = cdcService;
-        this.properties = properties;
+        this.xtrmetlProperties = xtrmetlProperties;
         this.sourceRegistry = sourceRegistry;
         this.targetRegistry = targetRegistry;
         this.sourceFactory = sourceFactory;
@@ -48,64 +48,70 @@ public CdcController(
 
     @GetMapping("/status")
     @Observed(name = "cdc.status", contextualName = "cdc-status")
-    public ResponseEntity<Map<String, Object>> status() {
-        Map<String, Object> body = new LinkedHashMap<>(cdcService.getStatus());
-        body.put("replicaEnabled", properties.getReplica().isEnabled());
-        body.put("replicaDdlEnabled", properties.getReplica().isDdlEnabled());
-        body.put("replicaTopicPattern", properties.getReplica().getTopicPattern());
-        body.put("replicaTables", properties.getReplica().getTables());
-        body.put("replicationSlot", replicationSlotProbe.probeConfiguredSlot());
-        body.put("configuredSources", sourceFactory.describeConfigured(
-                properties.getCdc().getSources().stream()
-                        .map(s -> new CdcSourceFactory.SourceSpec(s.getId(), s.getType(), s.isEnabled()))
+    public ResponseEntity<Map<String, Object>> cdcStatus() {
+        Map<String, Object> statusBody = new LinkedHashMap<>(cdcService.getStatus());
+        statusBody.put("replicaEnabled", xtrmetlProperties.getReplica().isEnabled());
+        statusBody.put("replicaDdlEnabled", xtrmetlProperties.getReplica().isDdlEnabled());
+        statusBody.put("replicaTopicPattern", xtrmetlProperties.getReplica().getTopicPattern());
+        statusBody.put("replicaTables", xtrmetlProperties.getReplica().getTables());
+        statusBody.put("replicationSlot", replicationSlotProbe.probeConfiguredSlot());
+        statusBody.put("configuredSources", sourceFactory.describeConfigured(
+                xtrmetlProperties.getCdc().getSources().stream()
+                        .map(sourceConfiguration -> new CdcSourceFactory.SourceSpec(
+                                sourceConfiguration.getSourceId(),
+                                sourceConfiguration.getSourceType(),
+                                sourceConfiguration.isEnabled()
+                        ))
                         .collect(Collectors.toList())
         ));
-        body.put("registeredSources", sourceRegistry.all().stream()
-                .map(this::sourceEntry)
+        statusBody.put("registeredSources", sourceRegistry.all().stream()
+                .map(this::sourceRegistryEntry)
                 .collect(Collectors.toList()));
-        body.put("registeredTargets", targetRegistry.all().stream()
-                .map(target -> {
-                    Map<String, Object> entry = new LinkedHashMap<>();
-                    entry.put("id", target.id());
-                    entry.put("displayName", target.displayName());
-                    entry.put("scaffoldOnly", target.scaffoldOnly());
-                    return entry;
+        statusBody.put("registeredTargets", targetRegistry.all().stream()
+                .map(targetConnector -> {
+                    Map<String, Object> targetEntry = new LinkedHashMap<>();
+                    targetEntry.put("id", targetConnector.targetId());
+                    targetEntry.put("displayName", targetConnector.displayName());
+                    targetEntry.put("scaffoldOnly", targetConnector.scaffoldOnly());
+                    return targetEntry;
                 })
                 .collect(Collectors.toList()));
-        return ResponseEntity.ok(body);
+        return ResponseEntity.ok(statusBody);
     }
 
     @GetMapping("/sources")
     @Observed(name = "cdc.sources", contextualName = "cdc-sources")
-    public ResponseEntity<List<Map<String, Object>>> sources() {
+    public ResponseEntity<List<Map<String, Object>>> cdcSources() {
         return ResponseEntity.ok(sourceRegistry.all().stream()
-                .map(this::sourceEntry)
+                .map(this::sourceRegistryEntry)
                 .collect(Collectors.toList()));
     }
 
     @GetMapping("/targets")
     @Observed(name = "cdc.targets", contextualName = "cdc-targets")
-    public ResponseEntity<List<Map<String, Object>>> targets() {
-        List<Map<String, Object>> body = targetRegistry.all().stream()
-                .map(target -> {
-                    Map<String, Object> entry = new LinkedHashMap<>();
-                    entry.put("id", target.id());
-                    entry.put("displayName", target.displayName());
-                    entry.put("scaffoldOnly", target.scaffoldOnly());
-                    return entry;
+    public ResponseEntity<List<Map<String, Object>>> cdcTargets() {
+        List<Map<String, Object>> targetEntries = targetRegistry.all().stream()
+                .map(targetConnector -> {
+                    Map<String, Object> targetEntry = new LinkedHashMap<>();
+                    targetEntry.put("id", targetConnector.targetId());
+                    targetEntry.put("displayName", targetConnector.displayName());
+                    targetEntry.put("scaffoldOnly", targetConnector.scaffoldOnly());
+                    return targetEntry;
                 })
                 .collect(Collectors.toList());
-        return ResponseEntity.ok(body);
+        return ResponseEntity.ok(targetEntries);
     }
 
-    private Map<String, Object> sourceEntry(com.xtrmetl.cdc.spi.CdcSourceConnector source) {
-        Map<String, Object> entry = new LinkedHashMap<>();
-        entry.put("id", source.id());
-        entry.put("displayName", source.displayName());
-        entry.put("engine", source.capabilities().engine());
-        entry.put("databases", source.capabilities().databases());
-        entry.put("scaffoldOnly", source.capabilities().scaffoldOnly());
-        return entry;
+    private Map<String, Object> sourceRegistryEntry(
+            com.xtrmetl.cdc.spi.CdcSourceConnector sourceConnector
+    ) {
+        Map<String, Object> sourceEntry = new LinkedHashMap<>();
+        sourceEntry.put("id", sourceConnector.sourceId());
+        sourceEntry.put("displayName", sourceConnector.displayName());
+        sourceEntry.put("engine", sourceConnector.capabilities().engine());
+        sourceEntry.put("databases", sourceConnector.capabilities().databases());
+        sourceEntry.put("scaffoldOnly", sourceConnector.capabilities().scaffoldOnly());
+        return sourceEntry;
     }
 
     @PostMapping("/start")
@@ -121,8 +127,10 @@ public ResponseEntity<String> stopCdc() {
         try {
             cdcService.stop();
             return ResponseEntity.ok("CDC process stopped");
-        } catch (IOException e) {
-            return ResponseEntity.internalServerError().body("Error stopping CDC process: " + e.getMessage());
+        } catch (IOException stopFailure) {
+            return ResponseEntity.internalServerError().body(
+                    "Error stopping CDC process: " + stopFailure.getMessage()
+            );
         }
     }
 }

cdc-service/src/main/java/com/xtrmetl/cdc/service/CdcService.java

@@ -269,7 +269,7 @@ private void maybeMapCanonical(String topic, String key, String value) {
         }
         try {
             boolean ok = changeRecordMapper
-                    .map(PostgresDebeziumCdcSource.ID, topic, key, value)
+                    .map(PostgresDebeziumCdcSource.SOURCE_ID, topic, key, value)
                     .isPresent();
             if (ok) {
                 canonicalMapSuccess.incrementAndGet();

cdc-service/src/main/java/com/xtrmetl/cdc/spi/AbstractScaffoldCdcSource.java

@@ -9,21 +9,35 @@
  */
 public abstract class AbstractScaffoldCdcSource implements CdcSourceConnector {
 
-    private final String id;
+    private final String sourceId;
     private final String displayName;
-    private final String engine;
-    private final Set<String> databases;
+    private final String sourceEngine;
+    private final Set<String> supportedDatabases;
 
-    protected AbstractScaffoldCdcSource(String id, String displayName, String engine, Set<String> databases) {
-        this.id = Objects.requireNonNull(id, "id");
+    protected AbstractScaffoldCdcSource(
+            String sourceId,
+            String displayName,
+            String sourceEngine,
+            Set<String> supportedDatabases
+    ) {
+        this.sourceId = Objects.requireNonNull(sourceId, "sourceId");
         this.displayName = Objects.requireNonNull(displayName, "displayName");
-        this.engine = Objects.requireNonNull(engine, "engine");
-        this.databases = Set.copyOf(databases);
+        this.sourceEngine = Objects.requireNonNull(sourceEngine, "sourceEngine");
+        this.supportedDatabases = Set.copyOf(supportedDatabases);
     }
 
     @Override
+    public final String sourceId() {
+        return sourceId;
+    }
+
+    /**
+     * @deprecated compatibility alias; organization-owned callers use {@link #sourceId()}
+     */
+    @Override
+    @Deprecated(forRemoval = false)
     public final String id() {
-        return id;
+        return sourceId();
     }
 
     @Override
@@ -33,16 +47,16 @@ public final String displayName() {
 
     @Override
     public final SourceCapabilities capabilities() {
-        return new SourceCapabilities(engine, databases, true);
+        return new SourceCapabilities(sourceEngine, supportedDatabases, true);
     }
 
     @Override
-    public void validate(Map<String, String> config) {
-        Objects.requireNonNull(config, "config");
+    public void validate(Map<String, String> sourceConfig) {
+        Objects.requireNonNull(sourceConfig, "sourceConfig");
     }
 
     @Override
-    public final void start(Map<String, String> config) {
+    public final void start(Map<String, String> sourceConfig) {
         throw new UnsupportedOperationException(
                 displayName + " is a scaffold source only (no connector dependency wired). "
                         + "See docs/cdc/any-to-any-cdc.md"

cdc-service/src/main/java/com/xtrmetl/cdc/spi/CdcSourceConnector.java

@@ -8,6 +8,26 @@
  */
 public interface CdcSourceConnector extends AutoCloseable {
 
+    /**
+     * Returns the bounded-context-specific CDC source identifier.
+     *
+     * <p>New organization-owned callers must use this semantic accessor. The generic
+     * {@link #id()} method remains only as an SPI compatibility boundary for existing
+     * external connector implementations and callers.</p>
+     *
+     * @return exact CDC source identifier
+     */
+    default String sourceId() {
+        return id();
+    }
+
+    /**
+     * Legacy compatibility accessor for the historical generic connector identifier.
+     *
+     * @return exact CDC source identifier
+     * @deprecated organization-owned callers must use {@link #sourceId()}
+     */
+    @Deprecated(forRemoval = false)
     String id();
 
     String displayName();
@@ -17,12 +37,12 @@ public interface CdcSourceConnector extends AutoCloseable {
     /**
      * Validate source configuration (host, slot, credentials, include lists).
      */
-    void validate(Map<String, String> config);
+    void validate(Map<String, String> sourceConfig);
 
     /**
      * Begin capturing changes. Implementations publish through the service pipeline.
      */
-    void start(Map<String, String> config);
+    void start(Map<String, String> sourceConfig);
 
     /**
      * Stop capture gracefully.