feat(delivery): openframe-machine-delivery — engine, dispatcher and the first two specs (PR 1 of the plan)

openframe-data-mongo-common/src/main/java/com/openframe/data/document/delivery/DeliveryFailure.java

@@ -0,0 +1,7 @@
+package com.openframe.data.document.delivery;
+
+public enum DeliveryFailure {
+    EXHAUSTED,
+    OFFLINE,
+    TIMEOUT
+}

openframe-data-mongo-common/src/main/java/com/openframe/data/document/delivery/DeliveryStatus.java

@@ -0,0 +1,8 @@
+package com.openframe.data.document.delivery;
+
+public enum DeliveryStatus {
+    PENDING,
+    ACKED,
+    DONE,
+    FAILED
+}

openframe-data-mongo-common/src/main/java/com/openframe/data/document/delivery/DeliveryType.java

@@ -0,0 +1,9 @@
+package com.openframe.data.document.delivery;
+
+public enum DeliveryType {
+    SCRIPT_SCHEDULE,
+    TOOL_INSTALLATION,
+    TOOL_UPDATE,
+    CLIENT_UPDATE,
+    CLIENT_UNINSTALL
+}

openframe-data-mongo-common/src/main/java/com/openframe/data/document/delivery/MachineDelivery.java

@@ -0,0 +1,48 @@
+package com.openframe.data.document.delivery;
+
+import com.openframe.data.document.rmm.schedule.ScheduleOfflineBehavior;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.mongodb.core.index.CompoundIndex;
+import org.springframework.data.mongodb.core.index.Indexed;
+import org.springframework.data.mongodb.core.mapping.Document;
+
+import java.time.Instant;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@Document(collection = "machine_delivery")
+@CompoundIndex(name = "machine_delivery_sweep", def = "{'type': 1, 'status': 1, 'lastAttemptAt': 1}")
+public class MachineDelivery {
+
+    @Id
+    private String id;
+
+    private DeliveryType type;
+    private String targetId;
+    private String machineId;
+    private String tenantId;
+
+    private DeliveryStatus status;
+    private int attempts;
+    private String payloadJson;
+
+    private Instant dispatchedAt;
+    private Instant lastAttemptAt;
+    private Instant ackedAt;
+    private Instant finishedAt;
+
+    private DeliveryFailure failure;
+    private String error;
+
+    private ScheduleOfflineBehavior offlineBehavior;
+    private Long reconnectWindowSeconds;
+
+    @Indexed(name = "machine_delivery_ttl", expireAfterSeconds = 0)
+    private Instant expiresAt;
+}

openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/delivery/MachineDeliveryRepository.java

@@ -0,0 +1,18 @@
+package com.openframe.data.repository.delivery;
+
+import com.openframe.data.document.delivery.DeliveryStatus;
+import com.openframe.data.document.delivery.DeliveryType;
+import com.openframe.data.document.delivery.MachineDelivery;
+import org.springframework.data.mongodb.repository.MongoRepository;
+import org.springframework.stereotype.Repository;
+
+import java.time.Instant;
+import java.util.List;
+
+@Repository
+public interface MachineDeliveryRepository extends MongoRepository<MachineDelivery, String> {
+
+    List<MachineDelivery> findByTypeAndStatusAndLastAttemptAtBefore(DeliveryType type, DeliveryStatus status, Instant before);
+
+    List<MachineDelivery> findByTypeAndStatusAndAckedAtBefore(DeliveryType type, DeliveryStatus status, Instant before);
+}

openframe-data-nats/pom.xml

@@ -28,6 +28,10 @@
             <groupId>com.openframe.oss</groupId>
             <artifactId>openframe-data-mongo-sync</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.openframe.oss</groupId>
+            <artifactId>openframe-machine-delivery</artifactId>
+        </dependency>
 
         <dependency>
             <groupId>org.springframework.boot</groupId>

openframe-data-nats/src/main/java/com/openframe/data/nats/delivery/ClientUninstallDeliverySpec.java

@@ -0,0 +1,89 @@
+package com.openframe.data.nats.delivery;
+
+import com.openframe.data.document.delivery.DeliveryFailure;
+import com.openframe.data.document.delivery.DeliveryType;
+import com.openframe.data.document.delivery.MachineDelivery;
+import com.openframe.data.document.device.DeviceStatus;
+import com.openframe.data.document.device.Machine;
+import com.openframe.data.nats.model.ClientUninstallMessage;
+import com.openframe.data.nats.publisher.ClientUninstallNatsPublisher;
+import com.openframe.data.repository.device.MachineRepository;
+import com.openframe.delivery.DeliveryRequest;
+import com.openframe.delivery.DeliverySeed;
+import com.openframe.delivery.DeliverySpec;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Component;
+
+@Component
+@RequiredArgsConstructor
+@ConditionalOnProperty("spring.cloud.stream.enabled")
+public class ClientUninstallDeliverySpec implements DeliverySpec<ClientUninstallDeliverySpec.Seed, ClientUninstallMessage> {
+
+    private final ClientUninstallNatsPublisher publisher;
+    private final MachineRepository machineRepository;
+
+    @Getter
+    @AllArgsConstructor
+    public static class Seed implements DeliverySeed {
+        private final String machineId;
+
+        @Override
+        public DeliveryType type() {
+            return DeliveryType.CLIENT_UNINSTALL;
+        }
+    }
+
+    @Override
+    public DeliveryType getType() {
+        return DeliveryType.CLIENT_UNINSTALL;
+    }
+
+    @Override
+    public Class<Seed> getSeedClass() {
+        return Seed.class;
+    }
+
+    @Override
+    public Class<ClientUninstallMessage> getPayloadClass() {
+        return ClientUninstallMessage.class;
+    }
+
+    // targetId is the machine itself: the agent confirms over HTTP /api/agents/uninstall with X-Machine-Id
+    @Override
+    public DeliveryRequest<ClientUninstallMessage> request(Seed seed) {
+        ClientUninstallMessage message = publisher.buildMessage();
+        String machineId = seed.getMachineId();
+        return DeliveryRequest.<ClientUninstallMessage>builder()
+                .type(DeliveryType.CLIENT_UNINSTALL)
+                .targetId(machineId)
+                .machineId(machineId)
+                .payload(message)
+                .build();
+    }
+
+    @Override
+    public void publish(String machineId, ClientUninstallMessage payload) {
+        publisher.publish(machineId, payload);
+    }
+
+    // the agent never ran the uninstall: keep the machine visible instead of parking it in PENDING_DELETION forever
+    @Override
+    public void onFailed(MachineDelivery delivery, DeliveryFailure failure) {
+        String machineId = delivery.getMachineId();
+        machineRepository.findByMachineId(machineId)
+                .filter(ClientUninstallDeliverySpec::isPendingDeletion)
+                .ifPresent(this::restoreOffline);
+    }
+
+    private void restoreOffline(Machine machine) {
+        machine.setStatus(DeviceStatus.OFFLINE);
+        machineRepository.save(machine);
+    }
+
+    private static boolean isPendingDeletion(Machine machine) {
+        return machine.getStatus() == DeviceStatus.PENDING_DELETION;
+    }
+}

openframe-data-nats/src/main/java/com/openframe/data/nats/delivery/ToolInstallationDeliverySpec.java

@@ -0,0 +1,80 @@
+package com.openframe.data.nats.delivery;
+
+import com.openframe.data.document.delivery.DeliveryFailure;
+import com.openframe.data.document.delivery.DeliveryType;
+import com.openframe.data.document.delivery.MachineDelivery;
+import com.openframe.data.document.tool.IntegratedTool;
+import com.openframe.data.document.toolagent.IntegratedToolAgent;
+import com.openframe.data.nats.model.ToolInstallationMessage;
+import com.openframe.data.nats.publisher.ToolInstallationNatsPublisher;
+import com.openframe.delivery.DeliveryRequest;
+import com.openframe.delivery.DeliverySeed;
+import com.openframe.delivery.DeliverySpec;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Component;
+
+@Component
+@RequiredArgsConstructor
+@ConditionalOnProperty("spring.cloud.stream.enabled")
+public class ToolInstallationDeliverySpec implements DeliverySpec<ToolInstallationDeliverySpec.Seed, ToolInstallationMessage> {
+
+    private final ToolInstallationNatsPublisher publisher;
+
+    @Getter
+    @AllArgsConstructor
+    public static class Seed implements DeliverySeed {
+        private final String machineId;
+        private final IntegratedToolAgent toolAgent;
+        private final IntegratedTool tool;
+        private final boolean reinstall;
+
+        @Override
+        public DeliveryType type() {
+            return DeliveryType.TOOL_INSTALLATION;
+        }
+    }
+
+    @Override
+    public DeliveryType getType() {
+        return DeliveryType.TOOL_INSTALLATION;
+    }
+
+    @Override
+    public Class<Seed> getSeedClass() {
+        return Seed.class;
+    }
+
+    @Override
+    public Class<ToolInstallationMessage> getPayloadClass() {
+        return ToolInstallationMessage.class;
+    }
+
+    // targetId is the tool agent key: the agent reports it back as agentType in installed-agent
+    @Override
+    public DeliveryRequest<ToolInstallationMessage> request(Seed seed) {
+        IntegratedToolAgent toolAgent = seed.getToolAgent();
+        IntegratedTool tool = seed.getTool();
+        boolean reinstall = seed.isReinstall();
+        ToolInstallationMessage message = publisher.buildMessage(toolAgent, tool, reinstall);
+        String targetId = toolAgent.getKey();
+        return DeliveryRequest.<ToolInstallationMessage>builder()
+                .type(DeliveryType.TOOL_INSTALLATION)
+                .targetId(targetId)
+                .machineId(seed.getMachineId())
+                .payload(message)
+                .build();
+    }
+
+    @Override
+    public void publish(String machineId, ToolInstallationMessage payload) {
+        publisher.publish(machineId, payload);
+    }
+
+    @Override
+    public void onFailed(MachineDelivery delivery, DeliveryFailure failure) {
+        // nothing beyond FAILED + metric: an install is safe to re-run by hand
+    }
+}

openframe-data-nats/src/main/java/com/openframe/data/nats/publisher/ClientUninstallNatsPublisher.java

@@ -21,14 +21,17 @@ public class ClientUninstallNatsPublisher {
     private final NatsMessagePublisher natsMessagePublisher;
 
     public void publish(String machineId) {
-        String topicName = format(TOPIC_NAME_TEMPLATE, machineId);
         ClientUninstallMessage message = buildMessage();
-        natsMessagePublisher.publishPersistent(topicName, message);
+        publish(machineId, message);
+    }
 
+    public void publish(String machineId, ClientUninstallMessage message) {
+        String topicName = format(TOPIC_NAME_TEMPLATE, machineId);
+        natsMessagePublisher.publishPersistent(topicName, message);
         log.info("Published client uninstall command for machine {}", machineId);
     }
 
-    private ClientUninstallMessage buildMessage() {
+    public ClientUninstallMessage buildMessage() {
         ClientUninstallMessage message = new ClientUninstallMessage();
         message.setIssuedAt(Instant.now().toString());
         return message;

openframe-data-nats/src/main/java/com/openframe/data/nats/publisher/ToolInstallationNatsPublisher.java

@@ -35,8 +35,12 @@ public void publish(String machineId, IntegratedToolAgent toolAgent, IntegratedT
     }
 
     public void publish(String machineId, IntegratedToolAgent toolAgent, IntegratedTool tool, boolean reinstall) {
-        String topicName = buildTopicName(machineId);
         ToolInstallationMessage message = buildMessage(toolAgent, tool, reinstall);
+        publish(machineId, message);
+    }
+
+    public void publish(String machineId, ToolInstallationMessage message) {
+        String topicName = buildTopicName(machineId);
         natsMessagePublisher.publishPersistent(topicName, message);
     }
 
@@ -48,7 +52,7 @@ private ToolInstallationMessage buildMessage(IntegratedToolAgent toolAgent, Inte
         return buildMessage(toolAgent, tool, false);
     }
 
-    private ToolInstallationMessage buildMessage(IntegratedToolAgent toolAgent, IntegratedTool tool, boolean reinstall) {
+    public ToolInstallationMessage buildMessage(IntegratedToolAgent toolAgent, IntegratedTool tool, boolean reinstall) {
         ToolInstallationMessage message = new ToolInstallationMessage();
         message.setToolAgentId(toolAgent.getKey());
         // TODO: need refactoring