debezium/dbz#2090 Add SSH config file watcher with host reconciliation

debezium-platform-conductor/pom.xml

@@ -466,6 +466,11 @@
             <artifactId>quarkus-rest-client-jackson</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-scheduler</artifactId>
+        </dependency>
+
         <!-- Flyway -->
         <dependency>
             <groupId>io.quarkus</groupId>

debezium-platform-conductor/src/main/java/io/debezium/platform/domain/HostStatusService.java

@@ -0,0 +1,120 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.platform.domain;
+
+import static jakarta.transaction.Transactional.TxType.SUPPORTS;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Optional;
+
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.persistence.EntityManager;
+import jakarta.transaction.Transactional;
+
+import org.jboss.logging.Logger;
+
+import com.blazebit.persistence.CriteriaBuilderFactory;
+import com.blazebit.persistence.view.EntityViewManager;
+import com.blazebit.persistence.view.EntityViewSetting;
+
+import io.debezium.platform.data.model.HostStatusEntity;
+import io.debezium.platform.data.model.ProvisioningStatus;
+import io.debezium.platform.domain.views.HostStatus;
+import io.debezium.platform.domain.views.refs.HostStatusReference;
+
+/**
+ * Service for managing {@link HostStatusEntity} lifecycle via the
+ * Blaze-Persistence entity view layer.
+ *
+ * <p>Extends {@link AbstractService} to inherit standard CRUD operations
+ * ({@code list}, {@code create}, {@code update}, {@code delete}, {@code findById})
+ * and adds host-specific queries needed by the SSH config watcher's
+ * reconciliation logic.
+ */
+@ApplicationScoped
+public class HostStatusService extends AbstractService<HostStatusEntity, HostStatus, HostStatusReference> {
+
+    private static final int DEFAULT_AGENT_PORT = 8090;
+
+    private static final Logger logger = Logger.getLogger(HostStatusService.class);
+
+    public HostStatusService(EntityManager em, CriteriaBuilderFactory cbf, EntityViewManager evm) {
+        super(HostStatusEntity.class, HostStatus.class, HostStatusReference.class, em, cbf, evm);
+    }
+
+    /**
+     * Returns all host views whose provisioning status is not
+     * {@link ProvisioningStatus#REMOVED}.
+     */
+    @Transactional(SUPPORTS)
+    public List<HostStatus> findAllActiveHosts() {
+        return evm.applySetting(
+                EntityViewSetting.create(HostStatus.class),
+                cb().where("provisioningStatus").notEq(ProvisioningStatus.REMOVED)).getResultList();
+    }
+
+    /**
+     * Finds a host by its SSH config alias.
+     *
+     * @param alias the {@code Host} alias from {@code ~/.ssh/config}
+     * @return the matching view, or empty if no host with that alias exists
+     */
+    @Transactional(SUPPORTS)
+    public Optional<HostStatus> findBySshAlias(String alias) {
+        List<HostStatus> results = evm.applySetting(
+                EntityViewSetting.create(HostStatus.class),
+                cb().where("sshAlias").eq(alias)).getResultList();
+        return results.stream().findFirst();
+    }
+
+    /**
+     * Creates a new host with {@link ProvisioningStatus#PENDING}.
+     *
+     * @param alias    the SSH config alias (natural key)
+     * @param hostname the resolved hostname or IP address
+     */
+    public void createPendingHost(String alias, String hostname) {
+        HostStatus view = createEmpty();
+        view.setSshAlias(alias);
+        view.setHostname(hostname);
+        view.setProvisioningStatus(ProvisioningStatus.PENDING);
+        view.setAgentPort(DEFAULT_AGENT_PORT);
+        view.setLastCheckedAt(Instant.now());
+        create(view);
+        logger.infov("Created pending host: {0}", alias);
+    }
+
+    /**
+     * Marks the host with the given alias as {@link ProvisioningStatus#REMOVED}.
+     *
+     * @param alias the SSH config alias
+     */
+    public void markHostRemoved(String alias) {
+        findBySshAlias(alias).ifPresent(host -> {
+            host.setProvisioningStatus(ProvisioningStatus.REMOVED);
+            update(host);
+            logger.infov("Marked host as REMOVED: {0}", alias);
+        });
+    }
+
+    /**
+     * Updates the cached hostname and resets status to
+     * {@link ProvisioningStatus#PENDING} so the host gets re-provisioned.
+     *
+     * @param alias       the SSH config alias
+     * @param newHostname the new hostname or IP address
+     */
+    public void updateHostDetails(String alias, String newHostname) {
+        findBySshAlias(alias).ifPresent(host -> {
+            host.setHostname(newHostname);
+            host.setProvisioningStatus(ProvisioningStatus.PENDING);
+            host.setLastCheckedAt(Instant.now());
+            update(host);
+            logger.infov("Updated host details and reset to PENDING: {0}", alias);
+        });
+    }
+}

debezium-platform-conductor/src/main/java/io/debezium/platform/domain/views/HostStatus.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.platform.domain.views;
+
+import java.time.Instant;
+
+import com.blazebit.persistence.view.CreatableEntityView;
+import com.blazebit.persistence.view.EntityView;
+import com.blazebit.persistence.view.UpdatableEntityView;
+
+import io.debezium.platform.data.model.HostStatusEntity;
+import io.debezium.platform.data.model.ProvisioningStatus;
+import io.debezium.platform.domain.views.refs.HostStatusReference;
+
+@EntityView(HostStatusEntity.class)
+@CreatableEntityView
+@UpdatableEntityView
+public interface HostStatus extends HostStatusReference {
+
+    String getHostname();
+
+    void setHostname(String hostname);
+
+    void setSshAlias(String sshAlias);
+
+    ProvisioningStatus getProvisioningStatus();
+
+    void setProvisioningStatus(ProvisioningStatus status);
+
+    int getAgentPort();
+
+    void setAgentPort(int port);
+
+    String getAgentToken();
+
+    void setAgentToken(String token);
+
+    Instant getLastCheckedAt();
+
+    void setLastCheckedAt(Instant lastCheckedAt);
+}

debezium-platform-conductor/src/main/java/io/debezium/platform/domain/views/refs/HostStatusReference.java

@@ -0,0 +1,16 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.platform.domain.views.refs;
+
+import com.blazebit.persistence.view.EntityView;
+
+import io.debezium.platform.data.model.HostStatusEntity;
+import io.debezium.platform.domain.views.base.IdView;
+
+@EntityView(HostStatusEntity.class)
+public interface HostStatusReference extends IdView {
+    String getSshAlias();
+}

debezium-platform-conductor/src/main/java/io/debezium/platform/environment/host/discovery/HostReconciliationPlanBuilder.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.platform.environment.host.discovery;
+
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import io.debezium.platform.domain.views.HostStatus;
+
+/**
+ * Pure-function builder that compares SSH config file state against database
+ * state and produces a {@link ReconciliationPlan}.
+ *
+ * <p>This class is intentionally <strong>not</strong> a CDI bean — it has no
+ * injected dependencies, no mutable state, and no side effects. Every method
+ * is a deterministic transformation of its inputs.
+ */
+class HostReconciliationPlanBuilder {
+
+    /**
+     * Compares SSH config file state against database state and produces
+     * a reconciliation plan.
+     *
+     * @param fileHosts hosts parsed from the SSH config file
+     * @param dbHosts   active host views from the database (status ≠ REMOVED)
+     * @return the reconciliation plan describing what needs to change
+     */
+    ReconciliationPlan buildPlan(List<SshHostEntry> fileHosts,
+                                 List<HostStatus> dbHosts) {
+        Map<String, HostStatus> dbByAlias = dbHosts.stream()
+                .collect(Collectors.toMap(HostStatus::getSshAlias, Function.identity()));
+        Map<String, SshHostEntry> fileByAlias = fileHosts.stream()
+                .collect(Collectors.toMap(SshHostEntry::alias, Function.identity()));
+
+        List<SshHostEntry> toAdd = fileHosts.stream()
+                .filter(fileHost -> !dbByAlias.containsKey(fileHost.alias()))
+                .toList();
+
+        List<SshHostEntry> toUpdate = fileHosts.stream()
+                .filter(fileHost -> dbByAlias.containsKey(fileHost.alias()))
+                .filter(fileHost -> isModified(fileHost, dbByAlias.get(fileHost.alias())))
+                .toList();
+
+        List<String> toRemove = dbHosts.stream()
+                .map(HostStatus::getSshAlias)
+                .filter(alias -> !fileByAlias.containsKey(alias))
+                .toList();
+
+        return new ReconciliationPlan(toAdd, toRemove, toUpdate);
+    }
+
+    /**
+     * Per ssh_config(5): if {@code HostName} is absent, the alias is used as hostname.
+     */
+    String resolveHostname(SshHostEntry entry) {
+        return entry.hostname() != null ? entry.hostname() : entry.alias();
+    }
+
+    /**
+     * Hostname is the critical field — if the IP/address changed, re-provisioning
+     * is needed. Per ssh_config(5), absent HostName defaults to the alias.
+     */
+    private boolean isModified(SshHostEntry fileHost, HostStatus dbHost) {
+        String fileHostname = resolveHostname(fileHost);
+        String dbHostname = dbHost.getHostname();
+        return !fileHostname.equals(dbHostname);
+    }
+}

debezium-platform-conductor/src/main/java/io/debezium/platform/environment/host/discovery/ReconciliationPlan.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright Debezium Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.debezium.platform.environment.host.discovery;
+
+import java.util.List;
+
+/**
+ * Immutable result of comparing SSH config file state against database state.
+ *
+ * <p>Produced by
+ * {@link HostReconciliationPlanBuilder#buildPlan(List, List)}
+ * — a pure function with zero side effects and zero database access.
+ *
+ * @param toAdd    hosts present in the file but not in the database
+ * @param toRemove aliases of hosts present in the database but not in the file
+ * @param toUpdate hosts present in both but with a changed hostname
+ */
+record ReconciliationPlan(
+        List<SshHostEntry> toAdd,
+        List<String> toRemove,
+        List<SshHostEntry> toUpdate) {
+}