diff --git a/openframe-api-lib/src/main/java/com/openframe/api/dto/rmm/execution/ScriptExecutionResponse.java b/openframe-api-lib/src/main/java/com/openframe/api/dto/rmm/execution/ScriptExecutionResponse.java index a91916975..1b8666a21 100644 --- a/openframe-api-lib/src/main/java/com/openframe/api/dto/rmm/execution/ScriptExecutionResponse.java +++ b/openframe-api-lib/src/main/java/com/openframe/api/dto/rmm/execution/ScriptExecutionResponse.java @@ -24,6 +24,8 @@ public class ScriptExecutionResponse { private String id; private String executionId; private String scriptId; + /** Schedule this execution came from; null for ad-hoc runs. */ + private String scheduleId; private String machineId; private PrivilegeLevel privilegeLevel; private String initiatedBy; diff --git a/openframe-api-lib/src/main/java/com/openframe/api/dto/rmm/schedule/CreateScriptScheduleInput.java b/openframe-api-lib/src/main/java/com/openframe/api/dto/rmm/schedule/CreateScriptScheduleInput.java index f7375ca53..907ad223b 100644 --- a/openframe-api-lib/src/main/java/com/openframe/api/dto/rmm/schedule/CreateScriptScheduleInput.java +++ b/openframe-api-lib/src/main/java/com/openframe/api/dto/rmm/schedule/CreateScriptScheduleInput.java @@ -1,9 +1,11 @@ package com.openframe.api.dto.rmm.schedule; import com.openframe.data.document.rmm.ScriptPlatform; +import jakarta.validation.constraints.Min; import jakarta.validation.constraints.NotBlank; import lombok.Data; +import java.time.Instant; import java.util.List; /** @@ -22,4 +24,14 @@ public class CreateScriptScheduleInput { private List supportedPlatforms; private List scriptIds; + + /** + * First scheduled run as an absolute UTC instant (the dashboard converts the + * chosen Date + Time to UTC). Optional — a schedule with no startAt is never + * picked up by the runner until one is set. + */ + private Instant startAt; + + @Min(value = 1800, message = "repeat must be at least 1800 seconds (30 minutes)") + private Long repeat; } diff --git a/openframe-api-lib/src/main/java/com/openframe/api/dto/rmm/schedule/ScriptScheduleResponse.java b/openframe-api-lib/src/main/java/com/openframe/api/dto/rmm/schedule/ScriptScheduleResponse.java index c7c52fb16..7e1ab4f66 100644 --- a/openframe-api-lib/src/main/java/com/openframe/api/dto/rmm/schedule/ScriptScheduleResponse.java +++ b/openframe-api-lib/src/main/java/com/openframe/api/dto/rmm/schedule/ScriptScheduleResponse.java @@ -22,6 +22,11 @@ public class ScriptScheduleResponse { private List scriptIds; + private Instant startAt; + private Long repeat; + private Instant nextRunAt; + private Instant lastRunAt; + private String createdBy; private String status; diff --git a/openframe-api-lib/src/main/java/com/openframe/api/dto/rmm/schedule/UpdateScriptScheduleInput.java b/openframe-api-lib/src/main/java/com/openframe/api/dto/rmm/schedule/UpdateScriptScheduleInput.java index f0320c7d8..fe677aa86 100644 --- a/openframe-api-lib/src/main/java/com/openframe/api/dto/rmm/schedule/UpdateScriptScheduleInput.java +++ b/openframe-api-lib/src/main/java/com/openframe/api/dto/rmm/schedule/UpdateScriptScheduleInput.java @@ -1,9 +1,11 @@ package com.openframe.api.dto.rmm.schedule; import com.openframe.data.document.rmm.ScriptPlatform; +import jakarta.validation.constraints.Min; import jakarta.validation.constraints.NotBlank; import lombok.Data; +import java.time.Instant; import java.util.List; /** @@ -25,4 +27,16 @@ public class UpdateScriptScheduleInput { private List supportedPlatforms; private List scriptIds; + + /** + * First scheduled run as an absolute UTC instant. PUT semantics: null clears + * the timing (the schedule stops being picked up). Changing this value + * reschedules the next run (see {@code ScriptScheduleService.update}). + */ + private Instant startAt; + + /** Recurrence interval in seconds; null clears recurrence (one-shot). Must be a whole number + * of 30-minute slots (1800, 3600, 5400, …) — the runner ticks on that grid. */ + @Min(value = 1800, message = "repeat must be at least 1800 seconds (30 minutes)") + private Long repeat; } diff --git a/openframe-api-lib/src/main/java/com/openframe/api/mapper/ScriptExecutionMapper.java b/openframe-api-lib/src/main/java/com/openframe/api/mapper/ScriptExecutionMapper.java index c9f49bb92..a2898c6cb 100644 --- a/openframe-api-lib/src/main/java/com/openframe/api/mapper/ScriptExecutionMapper.java +++ b/openframe-api-lib/src/main/java/com/openframe/api/mapper/ScriptExecutionMapper.java @@ -17,6 +17,7 @@ public ScriptExecutionResponse toResponse(ScriptExecution entity) { .id(entity.getId()) .executionId(entity.getExecutionId()) .scriptId(entity.getScriptId()) + .scheduleId(entity.getScheduleId()) .machineId(entity.getMachineId()) .privilegeLevel(entity.getPrivilegeLevel()) .initiatedBy(entity.getInitiatedBy()) diff --git a/openframe-api-lib/src/main/java/com/openframe/api/mapper/ScriptScheduleMapper.java b/openframe-api-lib/src/main/java/com/openframe/api/mapper/ScriptScheduleMapper.java index 3876ab970..8f0623bda 100644 --- a/openframe-api-lib/src/main/java/com/openframe/api/mapper/ScriptScheduleMapper.java +++ b/openframe-api-lib/src/main/java/com/openframe/api/mapper/ScriptScheduleMapper.java @@ -25,6 +25,8 @@ public ScriptSchedule toEntity(String tenantId, CreateScriptScheduleInput input) .description(input.getDescription()) .supportedPlatforms(input.getSupportedPlatforms()) .scriptIds(input.getScriptIds()) + .startAt(input.getStartAt()) + .repeat(input.getRepeat()) .build(); } @@ -33,6 +35,8 @@ public void updateEntity(ScriptSchedule existing, UpdateScriptScheduleInput inpu existing.setDescription(input.getDescription()); existing.setSupportedPlatforms(input.getSupportedPlatforms()); existing.setScriptIds(input.getScriptIds()); + existing.setStartAt(input.getStartAt()); + existing.setRepeat(input.getRepeat()); } public ScriptScheduleResponse toResponse(ScriptSchedule entity) { @@ -42,6 +46,10 @@ public ScriptScheduleResponse toResponse(ScriptSchedule entity) { .description(entity.getDescription()) .supportedPlatforms(mapPlatformsToResponse(entity.getSupportedPlatforms())) .scriptIds(entity.getScriptIds()) + .startAt(entity.getStartAt()) + .repeat(entity.getRepeat()) + .nextRunAt(entity.getNextRunAt()) + .lastRunAt(entity.getLastRunAt()) .createdBy(entity.getCreatedBy()) .status(entity.getStatus() != null ? entity.getStatus().name() : ScriptStatus.ACTIVE.name()) .statusChangedAt(entity.getStatusChangedAt()) diff --git a/openframe-api-lib/src/main/java/com/openframe/api/service/rmm/ScriptDispatchService.java b/openframe-api-lib/src/main/java/com/openframe/api/service/rmm/ScriptDispatchService.java index a3dd1e832..2ef152872 100644 --- a/openframe-api-lib/src/main/java/com/openframe/api/service/rmm/ScriptDispatchService.java +++ b/openframe-api-lib/src/main/java/com/openframe/api/service/rmm/ScriptDispatchService.java @@ -1,26 +1,40 @@ package com.openframe.api.service.rmm; import com.openframe.api.dto.rmm.DispatchResponse; +import com.openframe.api.dto.rmm.schedule.ScriptScheduleResponse; import com.openframe.api.dto.rmm.script.BatchRunScriptInput; import com.openframe.api.dto.rmm.script.RunScriptInput; import com.openframe.api.dto.rmm.script.ScriptEnvVarInput; import com.openframe.api.dto.rmm.script.ScriptResponse; import com.openframe.api.exception.DeviceNotFoundException; import com.openframe.api.service.DeviceService; +import com.openframe.core.exception.BadRequestException; +import com.openframe.data.document.rmm.ExecutionStatus; +import com.openframe.data.document.rmm.PrivilegeLevel; +import com.openframe.data.document.rmm.ScheduleScriptExecution; import com.openframe.data.document.rmm.ScriptEnvVar; import com.openframe.data.document.rmm.ScriptShell; +import com.openframe.data.document.rmm.ScriptStatus; +import com.openframe.data.nats.rmm.model.ScriptScheduleExecutionItem; import com.openframe.data.nats.rmm.model.ScriptMessage; +import com.openframe.data.nats.rmm.model.ScriptScheduleExecutionMessage; import com.openframe.data.nats.rmm.publisher.ScriptNatsPublisher; +import com.openframe.data.nats.rmm.publisher.ScriptScheduleExecutionNatsPublisher; +import com.openframe.data.repository.rmm.ScheduleScriptExecutionRepository; +import com.openframe.data.service.TenantIdProvider; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.stereotype.Service; +import java.time.Instant; import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.function.Function; +import java.util.stream.Collectors; /** * Dispatches a saved script from the dashboard to the target agent over core NATS. @@ -39,8 +53,13 @@ public class ScriptDispatchService { private final ScriptService scriptService; private final ScriptNatsPublisher scriptNatsPublisher; + private final ScriptScheduleExecutionNatsPublisher scriptScheduleExecutionNatsPublisher; private final DeviceService deviceService; private final ScriptExecutionService scriptExecutionService; + private final ScriptScheduleService scriptScheduleService; + private final ScriptScheduleDeviceService scriptScheduleDeviceService; + private final ScheduleScriptExecutionRepository scheduleScriptExecutionRepository; + private final TenantIdProvider tenantIdProvider; public DispatchResponse runScript(RunScriptInput input, String initiatedBy) { deviceService.findByMachineId(input.getMachineId()) @@ -58,6 +77,7 @@ public DispatchResponse runScript(RunScriptInput input, String initiatedBy) { ScriptMessage message = ScriptMessage.builder() .executionId(executionId) + .scriptId(script.getId()) .machineId(input.getMachineId()) .code(script.getScriptBody()) .shell(ScriptShell.valueOf(script.getShell())) @@ -81,44 +101,170 @@ public DispatchResponse batchRunScript(BatchRunScriptInput input, String initiat // Verify every target up front — reject the whole batch if any is unknown, // so we never half-dispatch. - machineIds.forEach(machineId -> deviceService.findByMachineId(machineId) - .orElseThrow(() -> new DeviceNotFoundException("Machine not found: " + machineId))); + verifyMachines(machineIds); // Resolve the saved script once; every machine shares it. ScriptResponse script = scriptService.get(input.getScriptId()); String executionId = UUID.randomUUID().toString(); - Integer timeoutSeconds = effectiveTimeout(input.getTimeoutSeconds(), script.getDefaultTimeoutSeconds()); + return dispatchBatch(executionId, script, machineIds, input.getPrivilegeLevel(), + input.getArgs(), input.getTimeoutSeconds(), input.getEnvVars(), initiatedBy); + } + + /** + * Ad-hoc run of a saved schedule: fan every ACTIVE script the schedule references out + * to every assigned device over core NATS as ONE + * {@link ScriptScheduleExecutionMessage} per machine (the whole script list is + * batched into a single wire payload — subject + * {@code machine.{machineId}.script-schedule-execution}). The whole run shares a + * single {@code executionId} across all scripts and machines; each leaf + * {@code ScriptExecution} row is disambiguated by {@code scriptId} and the header + * {@link ScheduleScriptExecution} record ties them all together. + * + *

Only ACTIVE scripts are dispatched; scripts the schedule has outlived + * (deleted/archived) are skipped, not fatal. Throws {@link BadRequestException} only + * when there is genuinely nothing to run — no scripts, no assigned devices, or none + * of the referenced scripts are runnable — so we never mint a hollow executionId that + * correlates to no execution rows. + */ + public DispatchResponse runSchedule(String scheduleId, String initiatedBy) { + // Tenant-scoped lookup; throws if the schedule is missing or soft-deleted. + ScriptScheduleResponse schedule = scriptScheduleService.get(scheduleId); + + List scriptIds = schedule.getScriptIds(); + if (scriptIds == null || scriptIds.isEmpty()) { + throw new BadRequestException("Schedule has no scripts to run: " + scheduleId); + } + + List machineIds = scriptScheduleDeviceService.getMachineIds(scheduleId); + if (machineIds.isEmpty()) { + throw new BadRequestException("Schedule has no assigned devices: " + scheduleId); + } + + // Verify every target up front — reject the whole run if any is unknown, + // so we never half-dispatch across the schedule's scripts. + verifyMachines(machineIds); + + // Resolve every referenced script in ONE query (no N+1). Only ACTIVE scripts are + // dispatched; a schedule can outlive some of its scripts (deleted/archived), and + // those are skipped rather than failing the run. + Map scriptsById = scriptService.getScriptsByIds(scriptIds).stream() + .filter(script -> ScriptStatus.ACTIVE.name().equals(script.getStatus())) + .collect(Collectors.toMap(ScriptResponse::getId, Function.identity(), (a, b) -> a)); + + // Preserve run order; dedup (a shared executionId can't carry the same + // scriptId twice on one machine — the leaf unique key would collide). + List runnableScripts = scriptIds.stream().distinct() + .map(scriptsById::get).filter(java.util.Objects::nonNull).toList(); + List skippedIds = scriptIds.stream().distinct().filter(id -> !scriptsById.containsKey(id)).toList(); + if (!skippedIds.isEmpty()) { + log.warn("Schedule run scheduleId={} skipping non-runnable scripts (missing/deleted/archived): {}", + scheduleId, skippedIds); + } + if (runnableScripts.isEmpty()) { + throw new BadRequestException("Schedule has no runnable scripts: " + scheduleId); + } + + String executionId = UUID.randomUUID().toString(); + Instant now = Instant.now(); + List runnableScriptIds = runnableScripts.stream().map(ScriptResponse::getId).toList(); + + // 1. Header: one ScheduleScriptExecution row per fire — snapshot of what was + // attempted. Persisted BEFORE the leaves + NATS publish so the fact of the fire + // is durably recorded even if downstream persistence/publish fails midway. + scheduleScriptExecutionRepository.save(ScheduleScriptExecution.builder() + .tenantId(tenantIdProvider.getTenantId()) + .executionId(executionId) + .scheduleId(scheduleId) + .initiatedBy(initiatedBy) + .scriptIds(runnableScriptIds) + .machineIds(machineIds) + .status(ExecutionStatus.RUNNING) + .dispatchedAt(now) + .build()); + + // 2. Leaves: N × M ScriptExecution rows (persist per-script batch), so the watchdog + // and per-(script, machine) history keep working exactly as before. + for (ScriptResponse script : runnableScripts) { + scriptExecutionService.createBatch(executionId, script.getId(), scheduleId, machineIds, + script.getPrivilegeLevel(), + effectiveTimeout(null, script.getDefaultTimeoutSeconds()), + initiatedBy); + } + + // 3. Build the batched agent payload once — shared across every target machine. + List scheduledScripts = runnableScripts.stream() + .map(script -> ScriptScheduleExecutionItem.builder() + .scriptId(script.getId()) + .code(script.getScriptBody()) + .shell(ScriptShell.valueOf(script.getShell())) + .privilegeLevel(script.getPrivilegeLevel()) + .args(script.getDefaultArgs()) + .timeoutSeconds(script.getDefaultTimeoutSeconds()) + .envVars(mergeEnvVars(script.getEnvVars(), null)) + .build()) + .toList(); + + // 4. Fan out: ONE message per machine (vs. the old N-per-machine). subject: + // machine.{machineId}.script-schedule-execution. + machineIds.forEach(machineId -> scriptScheduleExecutionNatsPublisher.publish(machineId, + ScriptScheduleExecutionMessage.builder() + .executionId(executionId) + .scheduleId(scheduleId) + .machineId(machineId) + .initiatedBy(initiatedBy) + .scripts(scheduledScripts) + .build())); + + // "Run now" is an extra, out-of-band execution: record it, but leave the cadence + // untouched — the schedule still fires at whatever slot it was already heading for. + scriptScheduleService.recordManualRun(scheduleId, Instant.now()); + + log.info("Dispatched schedule run scheduleId={} executionId={} scripts={} machines={}", + scheduleId, executionId, runnableScripts.size(), machineIds.size()); + return DispatchResponse.builder().executionId(executionId).build(); + } + + private DispatchResponse dispatchBatch(String executionId, ScriptResponse script, List machineIds, + PrivilegeLevel privilegeLevel, List argsOverride, + Integer timeoutOverride, List envVarsOverride, + String initiatedBy) { + Integer timeoutSeconds = effectiveTimeout(timeoutOverride, script.getDefaultTimeoutSeconds()); // Persist the effective timeout per row so the watchdog can derive a // per-execution stuck-threshold from it. - scriptExecutionService.createBatch(executionId, script.getId(), - machineIds, input.getPrivilegeLevel(), timeoutSeconds, initiatedBy); + scriptExecutionService.createBatch(executionId, script.getId(), null, machineIds, privilegeLevel, timeoutSeconds, initiatedBy); ScriptShell shell = ScriptShell.valueOf(script.getShell()); - List args = input.getArgs() != null ? input.getArgs() : script.getDefaultArgs(); - List envVars = mergeEnvVars(script.getEnvVars(), input.getEnvVars()); + List args = argsOverride != null ? argsOverride : script.getDefaultArgs(); + List envVars = mergeEnvVars(script.getEnvVars(), envVarsOverride); - // Fan out the same script (one executionId) to every machine. + // Fan out the same script (shared executionId) to every machine. machineIds.forEach(machineId -> scriptNatsPublisher.publishScript(machineId, ScriptMessage.builder() .executionId(executionId) + .scriptId(script.getId()) .machineId(machineId) .code(script.getScriptBody()) .shell(shell) - .privilegeLevel(input.getPrivilegeLevel()) + .privilegeLevel(privilegeLevel) .args(args) .timeoutSeconds(timeoutSeconds) .envVars(envVars) .build())); log.info("Dispatched batch script executionId={} scriptId={} machines={} shell={} privilegeLevel={}", - executionId, input.getScriptId(), machineIds.size(), script.getShell(), input.getPrivilegeLevel()); + executionId, script.getId(), machineIds.size(), script.getShell(), privilegeLevel); return DispatchResponse.builder() .executionId(executionId) .build(); } + private void verifyMachines(List machineIds) { + machineIds.forEach(machineId -> deviceService.findByMachineId(machineId) + .orElseThrow(() -> new DeviceNotFoundException("Machine not found: " + machineId))); + } + private static Integer effectiveTimeout(Integer override, Integer scriptDefault) { return override != null ? override : scriptDefault; } diff --git a/openframe-api-lib/src/main/java/com/openframe/api/service/rmm/ScriptExecutionService.java b/openframe-api-lib/src/main/java/com/openframe/api/service/rmm/ScriptExecutionService.java index a8c0a5c6f..0382ce03e 100644 --- a/openframe-api-lib/src/main/java/com/openframe/api/service/rmm/ScriptExecutionService.java +++ b/openframe-api-lib/src/main/java/com/openframe/api/service/rmm/ScriptExecutionService.java @@ -60,41 +60,44 @@ public Optional findById(String id) { * later rename of the source {@code Script} is reflected in History without * duplicating the name onto every row. */ - public ScriptExecution create(String executionId, - String scriptId, - String machineId, - PrivilegeLevel privilegeLevel, - Integer timeoutSeconds, - String initiatedBy) { + public ScriptExecutionResponse create(String executionId, + String scriptId, + String machineId, + PrivilegeLevel privilegeLevel, + Integer timeoutSeconds, + String initiatedBy) { Instant now = Instant.now(); - ScriptExecution scriptExecution = buildRunningRow(executionId, scriptId, machineId, privilegeLevel, timeoutSeconds, initiatedBy, now); + // Single ad-hoc run (runScript) never originates from a schedule → scheduleId null. + ScriptExecution scriptExecution = buildRunningRow(executionId, scriptId, null, machineId, privilegeLevel, timeoutSeconds, initiatedBy, now); ScriptExecution saved = scriptExecutionRepository.save(scriptExecution); log.info("Persisted execution row: executionId={} scriptId={} machineId={} initiatedBy={} status=RUNNING", executionId, scriptId, machineId, initiatedBy); - return saved; + return scriptExecutionMapper.toResponse(saved); } /** * Bulk-persist one {@link ExecutionStatus#RUNNING} row per target machine * under a shared {@code executionId} — backs batch dispatch. Unique - * constraint is {@code (tenantId, executionId, machineId)}, so the same - * {@code executionId} repeats across rows while each {@code machineId} - * stays distinct. + * constraint is {@code (tenantId, executionId, machineId, scriptId)}: the same + * {@code executionId} repeats across rows (a schedule run shares it across all + * its scripts too), with {@code machineId}/{@code scriptId} keeping each row + * distinct. */ - public List createBatch(String executionId, - String scriptId, - List machineIds, - PrivilegeLevel privilegeLevel, - Integer timeoutSeconds, - String initiatedBy) { + public List createBatch(String executionId, + String scriptId, + String scheduleId, + List machineIds, + PrivilegeLevel privilegeLevel, + Integer timeoutSeconds, + String initiatedBy) { Instant now = Instant.now(); List rows = machineIds.stream() - .map(machineId -> buildRunningRow(executionId, scriptId, machineId, privilegeLevel, timeoutSeconds, initiatedBy, now)) + .map(machineId -> buildRunningRow(executionId, scriptId, scheduleId, machineId, privilegeLevel, timeoutSeconds, initiatedBy, now)) .toList(); List saved = scriptExecutionRepository.saveAll(rows); - log.info("Persisted batch execution rows: executionId={} scriptId={} machineCount={} initiatedBy={} status=RUNNING", - executionId, scriptId, machineIds.size(), initiatedBy); - return saved; + log.info("Persisted batch execution rows: executionId={} scriptId={} scheduleId={} machineCount={} initiatedBy={} status=RUNNING", + executionId, scriptId, scheduleId, machineIds.size(), initiatedBy); + return saved.stream().map(scriptExecutionMapper::toResponse).toList(); } /** @@ -111,6 +114,7 @@ public List getBatchResults(String executionId, List ma private ScriptExecution buildRunningRow(String executionId, String scriptId, + String scheduleId, String machineId, PrivilegeLevel privilegeLevel, Integer timeoutSeconds, @@ -120,6 +124,7 @@ private ScriptExecution buildRunningRow(String executionId, .tenantId(tenantIdProvider.getTenantId()) .executionId(executionId) .scriptId(scriptId) + .scheduleId(scheduleId) .machineId(machineId) .privilegeLevel(privilegeLevel) .timeoutSeconds(timeoutSeconds) diff --git a/openframe-api-lib/src/main/java/com/openframe/api/service/rmm/ScriptScheduleDeviceService.java b/openframe-api-lib/src/main/java/com/openframe/api/service/rmm/ScriptScheduleDeviceService.java index c01731b48..53596d803 100644 --- a/openframe-api-lib/src/main/java/com/openframe/api/service/rmm/ScriptScheduleDeviceService.java +++ b/openframe-api-lib/src/main/java/com/openframe/api/service/rmm/ScriptScheduleDeviceService.java @@ -46,7 +46,7 @@ public class ScriptScheduleDeviceService { */ public void setDevices(String scheduleId, List machineIds, String actorUserId) { String tenantId = tenantIdProvider.getTenantId(); - requireVisibleSchedule(tenantId, scheduleId); + requireVisibleSchedule(tenantId, scheduleId); // existence check — throws NotFound if missing / DELETED List distinct = machineIds == null ? List.of() : new LinkedHashSet<>(machineIds).stream().toList(); @@ -61,6 +61,7 @@ public void setDevices(String scheduleId, List machineIds, String actorU doc.setMachineIds(distinct); assignedRepository.save(doc); + log.info("Set {} device(s) on script schedule id={} tenantId={}", distinct.size(), scheduleId, tenantId); } @@ -99,8 +100,8 @@ public Map> getMachineIdsByScheduleIds(Collection s return result; } - private void requireVisibleSchedule(String tenantId, String scheduleId) { - scheduleRepository.findByTenantIdAndId(tenantId, scheduleId) + private ScriptSchedule requireVisibleSchedule(String tenantId, String scheduleId) { + return scheduleRepository.findByTenantIdAndId(tenantId, scheduleId) .filter(schedule -> schedule.getStatus() != ScriptStatus.DELETED) .orElseThrow(() -> new NotFoundException("Script schedule not found: " + scheduleId)); } diff --git a/openframe-api-lib/src/main/java/com/openframe/api/service/rmm/ScriptScheduleService.java b/openframe-api-lib/src/main/java/com/openframe/api/service/rmm/ScriptScheduleService.java index fca5106df..a07c5d6be 100644 --- a/openframe-api-lib/src/main/java/com/openframe/api/service/rmm/ScriptScheduleService.java +++ b/openframe-api-lib/src/main/java/com/openframe/api/service/rmm/ScriptScheduleService.java @@ -11,6 +11,7 @@ import com.openframe.api.dto.shared.SortDirection; import com.openframe.api.dto.shared.SortInput; import com.openframe.api.mapper.ScriptScheduleMapper; +import com.openframe.core.exception.BadRequestException; import com.openframe.core.exception.ConflictException; import com.openframe.core.exception.NotFoundException; import com.openframe.data.document.rmm.ScriptSchedule; @@ -26,6 +27,7 @@ import java.time.Instant; import java.util.Collection; import java.util.List; +import java.util.Objects; import java.util.Optional; /** @@ -40,6 +42,13 @@ @RequiredArgsConstructor public class ScriptScheduleService { + /** + * Schedules live on a half-hour grid: every run happens at xx:00 or xx:30 and every + * repeat is a whole number of these slots. The management runner ticks on the same + * grid, so an off-grid instant would simply never coincide with a tick. + */ + private static final long SLOT_SECONDS = 1800L; + private static final List NAME_UNIQUE_STATUSES = List.of(ScriptStatus.ACTIVE, ScriptStatus.ARCHIVED); @@ -59,8 +68,11 @@ public ScriptScheduleResponse create(CreateScriptScheduleInput input, String cre throw new ConflictException("Script schedule with name '" + input.getName() + "' already exists"); } + validateGrid(input.getStartAt(), input.getRepeat()); + ScriptSchedule entity = scheduleMapper.toEntity(tenantId, input); entity.setCreatedBy(createdBy); + entity.setNextRunAt(entity.getStartAt()); ScriptSchedule saved = scheduleRepository.save(entity); log.info("Created script schedule id={} name='{}' tenantId={}", saved.getId(), saved.getName(), tenantId); return scheduleMapper.toResponse(saved); @@ -126,7 +138,7 @@ public CountedGenericQueryResult list(ScriptScheduleFilt return CountedGenericQueryResult.builder() .items(views) - .pageInfo(buildPageInfo(views, hasMore, normalized)) + .pageInfo(buildPageInfo(items, hasMore, normalized, sortField)) .filteredCount((int) filteredCount) .build(); } @@ -178,7 +190,13 @@ public ScriptScheduleResponse update(UpdateScriptScheduleInput input) { throw new ConflictException("Script schedule with name '" + input.getName() + "' already exists"); } + validateGrid(input.getStartAt(), input.getRepeat()); + + Instant priorStartAt = existing.getStartAt(); scheduleMapper.updateEntity(existing, input); + if (!Objects.equals(priorStartAt, existing.getStartAt())) { + existing.setNextRunAt(existing.getStartAt()); + } ScriptSchedule saved = scheduleRepository.save(existing); log.info("Updated script schedule id={} tenantId={}", saved.getId(), tenantId); return scheduleMapper.toResponse(saved); @@ -217,6 +235,27 @@ public ScriptScheduleResponse unarchive(String id) { return transitionTo(id, ScriptStatus.ACTIVE); } + /** + * Record that a schedule was run manually ("Run now"). + * + *

Only {@code lastRunAt} moves. The cadence is deliberately left alone: a manual run + * is an extra, out-of-band execution, not a replacement for the scheduled one, so + * {@code nextRunAt} keeps whatever slot the schedule was already heading for. Shifting it + * would either delay the planned run (re-anchoring to the manual instant) or pull it in + * ahead of the interval — both surprise the author, who never asked to change the schedule. + * + * @throws NotFoundException if the schedule does not exist or is soft-deleted. + */ + public void recordManualRun(String scheduleId, Instant runAt) { + String tenantId = tenantIdProvider.getTenantId(); + ScriptSchedule schedule = loadVisibleOrThrow(tenantId, scheduleId); + + schedule.setLastRunAt(runAt); + scheduleRepository.save(schedule); + log.info("Recorded manual run for schedule id={} tenantId={} lastRunAt={} (nextRunAt left at {})", + scheduleId, tenantId, runAt, schedule.getNextRunAt()); + } + private ScriptScheduleResponse transitionTo(String id, ScriptStatus target) { String tenantId = tenantIdProvider.getTenantId(); ScriptSchedule existing = loadVisibleOrThrow(tenantId, id); @@ -233,6 +272,22 @@ private ScriptScheduleResponse transitionTo(String id, ScriptStatus target) { return scheduleMapper.toResponse(saved); } + private static void validateGrid(Instant startAt, Long repeatSeconds) { + if (startAt != null && !isOnSlot(startAt)) { + throw new BadRequestException( + "startAt must fall on a 30-minute boundary (xx:00 or xx:30), got " + startAt); + } + if (repeatSeconds != null && repeatSeconds % SLOT_SECONDS != 0) { + throw new BadRequestException( + "repeat must be a whole number of 30-minute slots (multiple of " + SLOT_SECONDS + + " seconds), got " + repeatSeconds); + } + } + + private static boolean isOnSlot(Instant instant) { + return instant.getNano() == 0 && Math.floorMod(instant.getEpochSecond(), SLOT_SECONDS) == 0; + } + private ScriptSchedule loadOrThrow(String tenantId, String id) { return scheduleRepository.findByTenantIdAndId(tenantId, id) .orElseThrow(() -> new NotFoundException("Script schedule not found: " + id)); @@ -246,10 +301,17 @@ private ScriptSchedule loadVisibleOrThrow(String tenantId, String id) { return schedule; } - private static PageInfo buildPageInfo(List items, boolean hasMore, - CursorPaginationCriteria criteria) { - String startCursor = items.isEmpty() ? null : CursorCodec.encode(items.getFirst().getId()); - String endCursor = items.isEmpty() ? null : CursorCodec.encode(items.getLast().getId()); + /** + * Cursors are built from the ENTITIES (not the mapped views) and via the repository, + * because the cursor must encode the active sort value alongside the id — the keyset + * predicate on the other side has to match it exactly. + */ + private PageInfo buildPageInfo(List items, boolean hasMore, + CursorPaginationCriteria criteria, String sortField) { + String startCursor = items.isEmpty() ? null + : CursorCodec.encode(scheduleRepository.encodeCursor(items.getFirst(), sortField)); + String endCursor = items.isEmpty() ? null + : CursorCodec.encode(scheduleRepository.encodeCursor(items.getLast(), sortField)); boolean hasNextPage = criteria.isBackward() ? criteria.hasCursor() : hasMore; boolean hasPreviousPage = criteria.isBackward() ? hasMore : criteria.hasCursor(); diff --git a/openframe-api-service-core/src/main/java/com/openframe/api/datafetcher/ScriptDataFetcher.md b/openframe-api-service-core/src/main/java/com/openframe/api/datafetcher/.ScriptDataFetcher.md similarity index 100% rename from openframe-api-service-core/src/main/java/com/openframe/api/datafetcher/ScriptDataFetcher.md rename to openframe-api-service-core/src/main/java/com/openframe/api/datafetcher/.ScriptDataFetcher.md diff --git a/openframe-api-service-core/src/main/java/com/openframe/api/datafetcher/ScriptScheduleDataFetcher.java b/openframe-api-service-core/src/main/java/com/openframe/api/datafetcher/ScriptScheduleDataFetcher.java index 41f38d713..b6590d872 100644 --- a/openframe-api-service-core/src/main/java/com/openframe/api/datafetcher/ScriptScheduleDataFetcher.java +++ b/openframe-api-service-core/src/main/java/com/openframe/api/datafetcher/ScriptScheduleDataFetcher.java @@ -9,6 +9,7 @@ import com.openframe.api.dto.CountedGenericConnection; import com.openframe.api.dto.CountedGenericQueryResult; import com.openframe.api.dto.GenericEdge; +import com.openframe.api.dto.rmm.DispatchResponse; import com.openframe.api.dto.rmm.schedule.CreateScriptScheduleInput; import com.openframe.api.dto.rmm.schedule.ScriptScheduleFilterInput; import com.openframe.api.dto.rmm.schedule.ScriptScheduleFilters; @@ -21,6 +22,7 @@ import com.openframe.api.dto.shared.SortInput; import com.openframe.api.dto.user.UserResponse; import com.openframe.api.mapper.GraphQLScriptScheduleMapper; +import com.openframe.api.service.rmm.ScriptDispatchService; import com.openframe.api.service.rmm.ScriptScheduleDeviceService; import com.openframe.api.service.rmm.ScriptScheduleFilterService; import com.openframe.api.service.rmm.ScriptScheduleService; @@ -33,7 +35,10 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.dataloader.DataLoader; +import org.springframework.security.core.Authentication; import org.springframework.security.core.annotation.AuthenticationPrincipal; +import org.springframework.security.core.context.SecurityContextHolder; +import org.springframework.security.oauth2.jwt.Jwt; import org.springframework.validation.annotation.Validated; import java.util.List; @@ -61,6 +66,7 @@ public class ScriptScheduleDataFetcher { private final ScriptScheduleFilterService scheduleFilterService; private final ScriptService scriptService; private final ScriptScheduleDeviceService scheduleDeviceService; + private final ScriptDispatchService scriptDispatchService; private final GraphQLScriptScheduleMapper scheduleMapper; @DgsQuery @@ -143,6 +149,14 @@ public ScriptScheduleResponse setScriptScheduleDevices(@InputArgument @NotBlank return scheduleService.get(rawScheduleId); } + /** + * Ad-hoc "run now" of a schedule. + */ + @DgsMutation + public DispatchResponse runScheduleJobNow(@InputArgument @NotBlank String scheduleId) { + return scriptDispatchService.runSchedule(decodeId(scheduleId), getCurrentUserId()); + } + /** Returns the Relay global id ("ScriptSchedule:<rawId>") for the {@code id} field. */ @DgsData(parentType = "ScriptSchedule", field = "id") public String scriptScheduleNodeId(DgsDataFetchingEnvironment dfe) { @@ -213,4 +227,9 @@ private static void encodeNodeOptions(List options, String n } options.forEach(o -> o.setValue(RELAY.toGlobalId(nodeType, o.getValue()))); } + + private String getCurrentUserId() { + Authentication auth = SecurityContextHolder.getContext().getAuthentication(); + return AuthPrincipal.fromJwt((Jwt) auth.getPrincipal()).getId(); + } } diff --git a/openframe-api-service-core/src/main/resources/schema/script-execution.graphqls b/openframe-api-service-core/src/main/resources/schema/script-execution.graphqls index 479549e90..8c466c808 100644 --- a/openframe-api-service-core/src/main/resources/schema/script-execution.graphqls +++ b/openframe-api-service-core/src/main/resources/schema/script-execution.graphqls @@ -47,6 +47,9 @@ type ScriptExecution implements Node { scriptId: ID! # Resolved at read time from scriptId scriptName: String + # Schedule this execution originated from, stamped at dispatch. + # Null for ad-hoc runs (runScript / batchRunScript). Backs "execution history per schedule job". + scheduleId: ID machine: Machine privilegeLevel: PrivilegeLevel! initiator: User diff --git a/openframe-api-service-core/src/main/resources/schema/script-schedule.graphqls b/openframe-api-service-core/src/main/resources/schema/script-schedule.graphqls index 4c1b2253e..c327fce40 100644 --- a/openframe-api-service-core/src/main/resources/schema/script-schedule.graphqls +++ b/openframe-api-service-core/src/main/resources/schema/script-schedule.graphqls @@ -11,7 +11,8 @@ extend type Query { """Cursor-paginated list of schedules within the current tenant (Relay Connection Spec). Default order is newest-first by _id. Optional filter / search / sort. - Sortable fields: _id (default), name, createdAt, updatedAt. Search is a + Sortable fields: _id (default), name, createdAt, updatedAt, repeat, deviceCount. + Ties are broken by _id so the order is stable. Search is a case-insensitive substring match on name.""" scriptSchedules( filter: ScriptScheduleFilterInput, @@ -45,6 +46,11 @@ extend type Mutation { """Replace the full set of devices assigned to a schedule (PUT semantics — backs "Edit Devices"). machineIds are Machine global ids. Returns the updated schedule. Throws if the schedule is not found or soft-deleted.""" setScriptScheduleDevices(scheduleId: ID!, machineIds: [ID!]!): ScriptSchedule! + + """Run a schedule now: fan every script the schedule references out to all of its assigned + devices over core NATS under a single run executionId (shared across all scripts and machines). + Returns that one DispatchResponse.""" + runScheduleJobNow(scheduleId: ID!): DispatchResponse! } type ScriptSchedule implements Node { @@ -52,6 +58,14 @@ type ScriptSchedule implements Node { name: String! description: String supportedPlatforms: [ScriptPlatform!] + """First scheduled run as an absolute UTC instant (the dashboard's Date + Time). Null if the schedule has no timing set.""" + startAt: Instant + """Recurrence interval in seconds. Null for a one-shot schedule that fires once at startAt.""" + repeat: Long + """Next instant the runner will fire this schedule. Null when not scheduled (no timing, archived, or a one-shot that already fired).""" + nextRunAt: Instant + """Instant of the most recent run performed by the runner. Null until the first fire.""" + lastRunAt: Instant """The scripts this schedule runs, in run order (resolved from the stored script ids).""" scripts: [Script!]! """Machines assigned to this schedule (resolved from the assignment collection via the machine loader).""" @@ -91,6 +105,10 @@ input CreateScriptScheduleInput { supportedPlatforms: [ScriptPlatform!] "Ids of existing Scripts to run, in run order." scriptIds: [ID!] + "First scheduled run as an absolute UTC instant. Must fall on a 30-minute boundary (xx:00 or xx:30). Optional; a schedule with no startAt is never run until one is set." + startAt: Instant + "Recurrence interval in seconds. Must be a whole number of 30-minute slots (1800, 3600, 5400, …); the runner ticks on that grid. Null means run once at startAt." + repeat: Long } """ @@ -103,6 +121,10 @@ input UpdateScriptScheduleInput { description: String supportedPlatforms: [ScriptPlatform!] scriptIds: [ID!] + "First scheduled run (absolute UTC instant). Must fall on a 30-minute boundary (xx:00 or xx:30). PUT semantics: null clears the timing; changing it reschedules the next run." + startAt: Instant + "Recurrence interval in seconds. Must be a whole number of 30-minute slots (1800, 3600, 5400, …). Null clears recurrence (one-shot)." + repeat: Long } """ diff --git a/openframe-api-service-core/src/test/java/com/openframe/api/service/rmm/ScriptDispatchServiceTest.java b/openframe-api-service-core/src/test/java/com/openframe/api/service/rmm/ScriptDispatchServiceTest.java index 147cce6b6..867153df0 100644 --- a/openframe-api-service-core/src/test/java/com/openframe/api/service/rmm/ScriptDispatchServiceTest.java +++ b/openframe-api-service-core/src/test/java/com/openframe/api/service/rmm/ScriptDispatchServiceTest.java @@ -247,6 +247,7 @@ void batchRunScript_fansOutWithSharedExecutionId() { inOrder.verify(scriptExecutionService).createBatch( eq(response.getExecutionId()), eq(SCRIPT_ID), + eq((String) null), // ad-hoc batch — no schedule origin eq(machines), eq(PrivilegeLevel.ADMIN), eq(60), @@ -291,7 +292,7 @@ void batchRunScript_dedupsMachineIds() { scriptDispatchService.batchRunScript(batchInput(List.of("machine-1", "machine-1")), USER_ID); verify(scriptExecutionService).createBatch( - any(), eq(SCRIPT_ID), eq(List.of("machine-1")), eq(PrivilegeLevel.ADMIN), eq(60), eq(USER_ID)); + any(), eq(SCRIPT_ID), eq((String) null), eq(List.of("machine-1")), eq(PrivilegeLevel.ADMIN), eq(60), eq(USER_ID)); verify(scriptNatsPublisher, times(1)).publishScript(eq("machine-1"), any(ScriptMessage.class)); } diff --git a/openframe-api-service-core/src/test/java/com/openframe/api/service/rmm/ScriptExecutionServiceTest.java b/openframe-api-service-core/src/test/java/com/openframe/api/service/rmm/ScriptExecutionServiceTest.java index 05dea2880..f01c38cd3 100644 --- a/openframe-api-service-core/src/test/java/com/openframe/api/service/rmm/ScriptExecutionServiceTest.java +++ b/openframe-api-service-core/src/test/java/com/openframe/api/service/rmm/ScriptExecutionServiceTest.java @@ -28,6 +28,7 @@ import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.lenient; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -52,6 +53,9 @@ class ScriptExecutionServiceTest { void setUp() { when(tenantIdProvider.getTenantId()).thenReturn(TENANT_ID); service = new ScriptExecutionService(scriptExecutionRepository, tenantIdProvider, new ScriptExecutionMapper()); + // create()/createBatch() now map the SAVED entity to a DTO, so save must echo its + // argument back — otherwise the mock's null return NPEs in the mapper. + lenient().when(scriptExecutionRepository.save(any(ScriptExecution.class))).thenAnswer(inv -> inv.getArgument(0)); } @Test @@ -60,7 +64,7 @@ void create_persistsRunningRow() { when(scriptExecutionRepository.save(any(ScriptExecution.class))).thenAnswer(inv -> inv.getArgument(0)); Instant before = Instant.now().minus(Duration.ofSeconds(1)); - ScriptExecution result = service.create(EXECUTION_ID, SCRIPT_ID, MACHINE_ID, PrivilegeLevel.ADMIN, TIMEOUT_SECONDS, INITIATED_BY); + ScriptExecutionResponse result = service.create(EXECUTION_ID, SCRIPT_ID, MACHINE_ID, PrivilegeLevel.ADMIN, TIMEOUT_SECONDS, INITIATED_BY); ArgumentCaptor captor = ArgumentCaptor.forClass(ScriptExecution.class); verify(scriptExecutionRepository).save(captor.capture()); @@ -84,8 +88,13 @@ void create_persistsRunningRow() { assertThat(saved.getStderr()).isNull(); assertThat(saved.getError()).isNull(); - // Service returns the persisted row (id assigned by Mongo on save). - assertThat(result).isSameAs(saved); + // Service returns a DTO (never the entity), mapped from the persisted row. + assertThat(result.getExecutionId()).isEqualTo(EXECUTION_ID); + assertThat(result.getScriptId()).isEqualTo(SCRIPT_ID); + assertThat(result.getScheduleId()).isNull(); // ad-hoc run — no schedule origin + assertThat(result.getMachineId()).isEqualTo(MACHINE_ID); + assertThat(result.getStatus()).isEqualTo(ExecutionStatus.RUNNING); + assertThat(result.getInitiatedBy()).isEqualTo(INITIATED_BY); } @Test @@ -127,7 +136,7 @@ void createBatch_persistsOneRowPerMachine() { List machines = List.of("m-1", "m-2", "m-3"); Instant before = Instant.now().minus(Duration.ofSeconds(1)); - service.createBatch(EXECUTION_ID, SCRIPT_ID, machines, PrivilegeLevel.ADMIN, TIMEOUT_SECONDS, INITIATED_BY); + service.createBatch(EXECUTION_ID, SCRIPT_ID, "sched-1", machines, PrivilegeLevel.ADMIN, TIMEOUT_SECONDS, INITIATED_BY); @SuppressWarnings("unchecked") ArgumentCaptor> captor = ArgumentCaptor.forClass(List.class); @@ -140,6 +149,7 @@ void createBatch_persistsOneRowPerMachine() { assertThat(r.getTenantId()).isEqualTo(TENANT_ID); assertThat(r.getExecutionId()).isEqualTo(EXECUTION_ID); assertThat(r.getScriptId()).isEqualTo(SCRIPT_ID); + assertThat(r.getScheduleId()).isEqualTo("sched-1"); assertThat(r.getPrivilegeLevel()).isEqualTo(PrivilegeLevel.ADMIN); assertThat(r.getTimeoutSeconds()).isEqualTo(TIMEOUT_SECONDS); assertThat(r.getInitiatedBy()).isEqualTo(INITIATED_BY); diff --git a/openframe-api-service-core/src/test/java/com/openframe/api/service/rmm/ScriptScheduleDeviceServiceTest.java b/openframe-api-service-core/src/test/java/com/openframe/api/service/rmm/ScriptScheduleDeviceServiceTest.java index 9c416d2e2..14e0a96f9 100644 --- a/openframe-api-service-core/src/test/java/com/openframe/api/service/rmm/ScriptScheduleDeviceServiceTest.java +++ b/openframe-api-service-core/src/test/java/com/openframe/api/service/rmm/ScriptScheduleDeviceServiceTest.java @@ -50,8 +50,13 @@ void setUp() { } private void scheduleExists(ScriptStatus status) { + scheduleExistsReturning(status); + } + + private ScriptSchedule scheduleExistsReturning(ScriptStatus status) { ScriptSchedule schedule = ScriptSchedule.builder().id(SCHEDULE_ID).status(status).build(); when(scheduleRepository.findByTenantIdAndId(TENANT_ID, SCHEDULE_ID)).thenReturn(Optional.of(schedule)); + return schedule; } @Test @@ -144,6 +149,19 @@ void getMachineIdsByScheduleIds_mapsAndUnions() { assertThat(result.get("sch-2")).containsExactly("m-2", "m-3"); } + @Test + @DisplayName("setDevices: does NOT write back to the schedule document — device count is computed at read time from script_schedules_machines_assigned") + void setDevices_doesNotWriteScheduleDoc() { + scheduleExistsReturning(ScriptStatus.ACTIVE); + when(assignedRepository.findByTenantIdAndScriptScheduleIdsContaining(TENANT_ID, SCHEDULE_ID)) + .thenReturn(Optional.empty()); + when(assignedRepository.save(any())).thenAnswer(inv -> inv.getArgument(0)); + + service.setDevices(SCHEDULE_ID, List.of("m-1", "m-2", "m-1"), "user-1"); + + verify(scheduleRepository, never()).save(any()); + } + @Test @DisplayName("getMachineIdsByScheduleIds: empty input short-circuits without a repository call") void getMachineIdsByScheduleIds_empty_noLookup() { diff --git a/openframe-api-service-core/src/test/java/com/openframe/api/service/rmm/ScriptScheduleServiceTest.java b/openframe-api-service-core/src/test/java/com/openframe/api/service/rmm/ScriptScheduleServiceTest.java index 6a533a46b..6edb5cc0d 100644 --- a/openframe-api-service-core/src/test/java/com/openframe/api/service/rmm/ScriptScheduleServiceTest.java +++ b/openframe-api-service-core/src/test/java/com/openframe/api/service/rmm/ScriptScheduleServiceTest.java @@ -5,9 +5,12 @@ import com.openframe.api.dto.rmm.schedule.ScriptScheduleFilterInput; import com.openframe.api.dto.rmm.schedule.ScriptScheduleResponse; import com.openframe.api.dto.rmm.schedule.UpdateScriptScheduleInput; +import com.openframe.api.dto.shared.CursorCodec; import com.openframe.api.dto.shared.CursorPaginationCriteria; +import com.openframe.api.dto.shared.SortDirection; import com.openframe.api.dto.shared.SortInput; import com.openframe.api.mapper.ScriptScheduleMapper; +import com.openframe.core.exception.BadRequestException; import com.openframe.core.exception.ConflictException; import com.openframe.core.exception.NotFoundException; import com.openframe.data.document.rmm.ScriptPlatform; @@ -22,6 +25,8 @@ import org.mockito.ArgumentCaptor; import org.springframework.data.domain.Sort; +import java.time.Duration; +import java.time.Instant; import java.util.List; import java.util.Optional; @@ -194,6 +199,70 @@ void list_filterForwardedToRepository() { assertThat(forwarded.getCreatedByIds()).containsExactly("user-7"); } + @Test + @DisplayName("list: sort by repeat ASC is validated against the allowlist and forwarded verbatim to the repository") + void list_sortByRepeatAscending_forwarded() { + when(scheduleRepository.isSortableField("repeat")).thenReturn(true); + when(scheduleRepository.countForTenant(any(), any(), any())).thenReturn(0L); + when(scheduleRepository.findPageForTenant(any(), any(), any(), any(), any(), any(), eq(false), anyInt())) + .thenReturn(List.of()); + + scheduleService.list(null, null, + SortInput.builder().field("repeat").direction(SortDirection.ASC).build(), + CursorPaginationCriteria.builder().limit(20).build()); + + verify(scheduleRepository).findPageForTenant( + eq(TENANT_ID), any(), any(), eq("repeat"), eq(Sort.Direction.ASC), eq(null), eq(false), eq(21)); + // Allowlisted → the default sort field is never consulted. + verify(scheduleRepository, never()).getDefaultSortField(); + } + + @Test + @DisplayName("list: sort by repeat defaults to DESC when no direction is given") + void list_sortByRepeatDefaultsToDescending() { + when(scheduleRepository.isSortableField("repeat")).thenReturn(true); + when(scheduleRepository.countForTenant(any(), any(), any())).thenReturn(0L); + when(scheduleRepository.findPageForTenant(any(), any(), any(), any(), any(), any(), eq(false), anyInt())) + .thenReturn(List.of()); + + scheduleService.list(null, null, + SortInput.builder().field("repeat").build(), + CursorPaginationCriteria.builder().limit(20).build()); + + verify(scheduleRepository).findPageForTenant( + any(), any(), any(), eq("repeat"), eq(Sort.Direction.DESC), any(), eq(false), anyInt()); + } + + @Test + @DisplayName("list: page cursors are built for the ACTIVE sort field — repeat rows yield the compound cursor, not a bare id") + void list_sortByRepeat_buildsCompoundCursors() { + ScriptSchedule first = active(); + first.setRepeat(604800L); + ScriptSchedule last = active(); + last.setId("65f4a8000000000000000002"); + last.setRepeat(1800L); + + when(scheduleRepository.isSortableField("repeat")).thenReturn(true); + when(scheduleRepository.countForTenant(any(), any(), any())).thenReturn(2L); + when(scheduleRepository.findPageForTenant(any(), any(), any(), any(), any(), any(), eq(false), anyInt())) + .thenReturn(List.of(first, last)); + when(scheduleRepository.encodeCursor(first, "repeat")).thenReturn("604800|" + first.getId()); + when(scheduleRepository.encodeCursor(last, "repeat")).thenReturn("1800|" + last.getId()); + + CountedGenericQueryResult result = scheduleService.list(null, null, + SortInput.builder().field("repeat").build(), + CursorPaginationCriteria.builder().limit(20).build()); + + // Cursors must come from the repository (which owns the keyset format) and be + // built from the ENTITIES under the active sort field. + verify(scheduleRepository).encodeCursor(first, "repeat"); + verify(scheduleRepository).encodeCursor(last, "repeat"); + assertThat(CursorCodec.decode(result.getPageInfo().getStartCursor())) + .isEqualTo("604800|" + first.getId()); + assertThat(CursorCodec.decode(result.getPageInfo().getEndCursor())) + .isEqualTo("1800|" + last.getId()); + } + @Test @DisplayName("list: an invalid sort field falls back to the repository default (no exception)") void list_invalidSortField_fallsBackToDefault() { @@ -296,6 +365,103 @@ void archive_setsArchived() { verify(scheduleRepository).save(active); } + @Test + @DisplayName("recordManualRun: only lastRunAt moves — nextRunAt keeps the slot the schedule was already heading for") + void recordManualRun_doesNotTouchNextRun() { + ScriptSchedule active = active(); + active.setRepeat(7200L); // 2 h + Instant plannedNext = Instant.parse("2026-07-17T10:00:00Z"); + active.setNextRunAt(plannedNext); + when(scheduleRepository.findByTenantIdAndId(TENANT_ID, SCHEDULE_ID)).thenReturn(Optional.of(active)); + + // "Run now" at an arbitrary 09:17 — an extra run, not a replacement for the planned one. + Instant runAt = Instant.parse("2026-07-17T09:17:00Z"); + scheduleService.recordManualRun(SCHEDULE_ID, runAt); + + assertThat(active.getLastRunAt()).isEqualTo(runAt); + assertThat(active.getNextRunAt()).isEqualTo(plannedNext); // untouched + verify(scheduleRepository).save(active); + } + + @Test + @DisplayName("recordManualRun: a one-shot schedule keeps its null nextRunAt — a manual run never revives it") + void recordManualRun_oneShotStaysUnscheduled() { + ScriptSchedule active = active(); + active.setRepeat(null); + active.setNextRunAt(null); + when(scheduleRepository.findByTenantIdAndId(TENANT_ID, SCHEDULE_ID)).thenReturn(Optional.of(active)); + + scheduleService.recordManualRun(SCHEDULE_ID, Instant.parse("2026-07-17T09:17:00Z")); + + assertThat(active.getNextRunAt()).isNull(); + assertThat(active.getLastRunAt()).isEqualTo(Instant.parse("2026-07-17T09:17:00Z")); + } + + @Test + @DisplayName("recordManualRun: soft-deleted schedule throws, nothing saved") + void recordManualRun_deletedThrows() { + when(scheduleRepository.findByTenantIdAndId(TENANT_ID, SCHEDULE_ID)).thenReturn(Optional.of(deleted())); + + assertThatThrownBy(() -> scheduleService.recordManualRun(SCHEDULE_ID, Instant.now())) + .isInstanceOf(NotFoundException.class); + verify(scheduleRepository, never()).save(any()); + } + + // ---- half-hour grid (xx:00 / xx:30) ---- + + @Test + @DisplayName("create: startAt off the 30-minute grid is rejected — the runner only ticks at xx:00/xx:30") + void create_startAtOffGrid_rejected() { + createInput.setStartAt(Instant.parse("2026-09-15T02:07:00Z")); + when(scheduleRepository.existsByTenantIdAndNameAndStatusIn(any(), any(), any())).thenReturn(false); + + assertThatThrownBy(() -> scheduleService.create(createInput, "user-1")) + .isInstanceOf(BadRequestException.class) + .hasMessageContaining("30-minute boundary"); + verify(scheduleRepository, never()).save(any()); + } + + @Test + @DisplayName("create: both xx:00 and xx:30 are accepted") + void create_startAtOnGrid_accepted() { + when(scheduleRepository.existsByTenantIdAndNameAndStatusIn(any(), any(), any())).thenReturn(false); + when(scheduleRepository.save(any())).thenAnswer(inv -> inv.getArgument(0)); + + for (String iso : List.of("2026-09-15T02:00:00Z", "2026-09-15T02:30:00Z")) { + createInput.setStartAt(Instant.parse(iso)); + ScriptScheduleResponse result = scheduleService.create(createInput, "user-1"); + assertThat(result.getStartAt()).isEqualTo(Instant.parse(iso)); + } + } + + @Test + @DisplayName("create: repeat that is not a whole number of 30-minute slots is rejected") + void create_repeatNotSlotMultiple_rejected() { + createInput.setStartAt(Instant.parse("2026-09-15T02:00:00Z")); + createInput.setRepeat(2700L); // 45 min — above the floor, but off the grid + when(scheduleRepository.existsByTenantIdAndNameAndStatusIn(any(), any(), any())).thenReturn(false); + + assertThatThrownBy(() -> scheduleService.create(createInput, "user-1")) + .isInstanceOf(BadRequestException.class) + .hasMessageContaining("30-minute slots"); + verify(scheduleRepository, never()).save(any()); + } + + @Test + @DisplayName("create: 30m / 1h / 1h30 / 2h repeats are accepted") + void create_repeatSlotMultiples_accepted() { + when(scheduleRepository.existsByTenantIdAndNameAndStatusIn(any(), any(), any())).thenReturn(false); + when(scheduleRepository.save(any())).thenAnswer(inv -> inv.getArgument(0)); + createInput.setStartAt(Instant.parse("2026-09-15T02:00:00Z")); + + for (long repeat : List.of(1800L, 3600L, 5400L, 7200L)) { + createInput.setRepeat(repeat); + assertThat(scheduleService.create(createInput, "user-1").getRepeat()).isEqualTo(repeat); + } + } + + + private static ScriptSchedule deleted() { ScriptSchedule s = new ScriptSchedule(); s.setId(SCHEDULE_ID); diff --git a/openframe-client-core/src/main/java/com/openframe/client/service/RmmResultService.java b/openframe-client-core/src/main/java/com/openframe/client/service/RmmResultService.java index 33b6cc17d..d11921828 100644 --- a/openframe-client-core/src/main/java/com/openframe/client/service/RmmResultService.java +++ b/openframe-client-core/src/main/java/com/openframe/client/service/RmmResultService.java @@ -69,6 +69,11 @@ private RmmResultEvent getRmmResultEvent(String machineId, RmmResultMessage mess data.setTimedOut(message.getTimedOut()); data.setError(message.getError()); data.setEventTimestamp(now); + // Script-only ids: a command result has no saved script or schedule behind it. + if (message instanceof ScriptResultMessage script) { + data.setScriptId(script.getScriptId()); + data.setScheduleId(script.getScheduleId()); + } return data; } diff --git a/openframe-client-core/src/test/java/com/openframe/client/service/RmmResultServiceTest.java b/openframe-client-core/src/test/java/com/openframe/client/service/RmmResultServiceTest.java index 5fd4f7b76..8abdd1795 100644 --- a/openframe-client-core/src/test/java/com/openframe/client/service/RmmResultServiceTest.java +++ b/openframe-client-core/src/test/java/com/openframe/client/service/RmmResultServiceTest.java @@ -99,6 +99,42 @@ void processResult_scriptResult_publishesWithScriptExecutedHeader() { .containsEntry(KafkaHeader.MESSAGE_TYPE_HEADER, MessageType.SCRIPT_EXECUTED.name()); } + @Test + @DisplayName("processResult(ScriptResultMessage): the agent-echoed scriptId/scheduleId reach payload.after — scriptId is what pins a shared-executionId schedule result to its row") + void processResult_scriptResult_carriesScriptAndScheduleIds() { + ScriptResultMessage message = ScriptResultMessage.builder() + .executionId("exec-4") + .scriptId("script-b") + .scheduleId("sched-1") + .build(); + + rmmResultService.processResult(MACHINE_ID, message); + + ArgumentCaptor envelope = ArgumentCaptor.forClass(CommonDebeziumMessage.class); + verify(eventLogsPublisher).publish(eq(MACHINE_ID), envelope.capture(), any()); + JsonNode after = envelope.getValue().getPayload().getAfter(); + assertThat(after.get("scriptId").asText()).isEqualTo("script-b"); + assertThat(after.get("scheduleId").asText()).isEqualTo("sched-1"); + } + + @Test + @DisplayName("processResult(CommandResultMessage): no scriptId/scheduleId on the event — an ad-hoc command has neither a saved script nor a schedule behind it") + void processResult_commandResult_hasNoScriptOrScheduleIds() { + CommandResultMessage message = CommandResultMessage.builder() + .executionId("exec-5") + .machineId(MACHINE_ID) + .build(); + + rmmResultService.processResult(MACHINE_ID, message); + + ArgumentCaptor envelope = ArgumentCaptor.forClass(CommonDebeziumMessage.class); + verify(eventLogsPublisher).publish(eq(MACHINE_ID), envelope.capture(), any()); + JsonNode after = envelope.getValue().getPayload().getAfter(); + // NON_NULL on the event: absent rather than null. + assertThat(after.has("scriptId")).isFalse(); + assertThat(after.has("scheduleId")).isFalse(); + } + @Test @DisplayName("processResult: a sparse payload (only executionId) still publishes; absent fields are omitted from payload.after") void processResult_sparsePayload() { diff --git a/openframe-data-kafka/src/main/java/com/openframe/kafka/model/RmmResultEvent.java b/openframe-data-kafka/src/main/java/com/openframe/kafka/model/RmmResultEvent.java index e3186e9bc..71c4ee3fd 100644 --- a/openframe-data-kafka/src/main/java/com/openframe/kafka/model/RmmResultEvent.java +++ b/openframe-data-kafka/src/main/java/com/openframe/kafka/model/RmmResultEvent.java @@ -20,6 +20,8 @@ public class RmmResultEvent extends DebeziumMessage implements K private String tenantId; private String machineId; private String executionId; + private String scriptId; + private String scheduleId; private String stdout; private String stderr; private Integer exitCode; diff --git a/openframe-data-mongo-common/src/main/java/com/openframe/data/document/rmm/ScheduleScriptExecution.java b/openframe-data-mongo-common/src/main/java/com/openframe/data/document/rmm/ScheduleScriptExecution.java new file mode 100644 index 000000000..640f000e5 --- /dev/null +++ b/openframe-data-mongo-common/src/main/java/com/openframe/data/document/rmm/ScheduleScriptExecution.java @@ -0,0 +1,81 @@ +package com.openframe.data.document.rmm; + +import com.openframe.data.document.TenantScoped; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; +import org.springframework.data.annotation.CreatedDate; +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; +import java.util.List; + +/** + * Header record for a single {@link ScriptSchedule} fire — one document per run, + * regardless of how many scripts × machines the run fans out to. Ties together + * the per-(script, machine) leaf {@link ScriptExecution} rows produced by the + * same fire via a shared {@link #executionId}. + * + *

Lifecycle: + *

    + *
  1. Persisted with {@link ExecutionStatus#RUNNING} at dispatch time (before + * NATS publish), alongside a snapshot of the scripts / machines it + * targeted.
    2. + *
  2. Transitions to {@code SUCCESS} or {@code FAILED} once every leaf + * {@code ScriptExecution} sharing its {@code executionId} is terminal — + * the aggregator is TODO (waits for stream-service work to be in scope).
    3. + *
+ * + *

{@code scriptIds} and {@code machineIds} are snapshots at dispatch time: + * they must keep displaying what was actually attempted even if the schedule + * is later edited or the target set changes. + */ +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +@Document(collection = "schedule_script_execution") +@CompoundIndex( + name = "tenant_executionId_unique", + def = "{'tenantId': 1, 'executionId': 1}", + unique = true +) +@CompoundIndex( + name = "tenant_schedule_dispatchedAt", + def = "{'tenantId': 1, 'scheduleId': 1, 'dispatchedAt': -1}" +) +public class ScheduleScriptExecution implements TenantScoped { + + @Id + private String id; + + private String tenantId; + + /** Correlation id shared with every leaf {@link ScriptExecution} row this fire produced. */ + @Indexed + private String executionId; + + @Indexed + private String scheduleId; + + private String initiatedBy; + + /** Snapshot at fire time — the scripts that were runnable when the schedule fired. */ + private List scriptIds; + + /** Snapshot at fire time — the machines the schedule was fanned out to. */ + private List machineIds; + + @Indexed + private ExecutionStatus status; + + private Instant dispatchedAt; + private Instant finishedAt; + + @CreatedDate + private Instant createdAt; +} diff --git a/openframe-data-mongo-common/src/main/java/com/openframe/data/document/rmm/ScriptExecution.java b/openframe-data-mongo-common/src/main/java/com/openframe/data/document/rmm/ScriptExecution.java index c8e3cb36d..48116536d 100644 --- a/openframe-data-mongo-common/src/main/java/com/openframe/data/document/rmm/ScriptExecution.java +++ b/openframe-data-mongo-common/src/main/java/com/openframe/data/document/rmm/ScriptExecution.java @@ -22,8 +22,8 @@ @AllArgsConstructor @Document(collection = "script_executions") @CompoundIndex( - name = "tenant_executionId_machineId_unique", - def = "{'tenantId': 1, 'executionId': 1, 'machineId': 1}", + name = "tenant_executionId_machineId_scriptId_unique", + def = "{'tenantId': 1, 'executionId': 1, 'machineId': 1, 'scriptId': 1}", unique = true ) @CompoundIndex( @@ -63,6 +63,8 @@ public class ScriptExecution implements TenantScoped { @Indexed private String scriptId; + private String scheduleId; + @Indexed private String machineId; diff --git a/openframe-data-mongo-common/src/main/java/com/openframe/data/document/rmm/ScriptSchedule.java b/openframe-data-mongo-common/src/main/java/com/openframe/data/document/rmm/ScriptSchedule.java index b3d7cdebd..25b9f3a31 100644 --- a/openframe-data-mongo-common/src/main/java/com/openframe/data/document/rmm/ScriptSchedule.java +++ b/openframe-data-mongo-common/src/main/java/com/openframe/data/document/rmm/ScriptSchedule.java @@ -25,12 +25,13 @@ @NoArgsConstructor @AllArgsConstructor @Document(collection = "script_schedules") -// Non-unique tenant-scoped lookup index on (tenantId, name), matching the -// Script convention. If name must be unique per tenant, add a PARTIAL unique -// index in MongoIndexConfig (see Script) — pending the uniqueness decision. @CompoundIndex( def = "{'tenantId': 1, 'name': 1}" ) +@CompoundIndex( + name = "status_nextRunAt", + def = "{'status': 1, 'nextRunAt': 1}" +) public class ScriptSchedule implements TenantScoped { @Id @@ -45,6 +46,14 @@ public class ScriptSchedule implements TenantScoped { private List supportedPlatforms; private List scriptIds; + private Instant startAt; + + private Long repeat; + + private Instant nextRunAt; + + private Instant lastRunAt; + private String createdBy; @CreatedDate diff --git a/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomScriptScheduleRepository.java b/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomScriptScheduleRepository.java index 202396472..035637057 100644 --- a/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomScriptScheduleRepository.java +++ b/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomScriptScheduleRepository.java @@ -35,4 +35,9 @@ List findPageForTenant(String tenantId, boolean isSortableField(String field); String getDefaultSortField(); + + /** + * Build the raw (pre-base64) pagination cursor for a row under the active sort. + */ + String encodeCursor(ScriptSchedule schedule, String sortField); } diff --git a/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomScriptScheduleRepositoryImpl.java b/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomScriptScheduleRepositoryImpl.java index f9a236367..012c9efc5 100644 --- a/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomScriptScheduleRepositoryImpl.java +++ b/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/CustomScriptScheduleRepositoryImpl.java @@ -16,10 +16,13 @@ import org.springframework.data.mongodb.core.query.Query; import org.springframework.stereotype.Repository; +import java.time.Instant; import java.util.ArrayList; +import java.util.Date; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.regex.Pattern; @@ -28,10 +31,12 @@ /** * MongoTemplate-backed implementation of {@link CustomScriptScheduleRepository}. * - *

Cursor pagination is implemented on {@code _id} (descending by default), - * with the comparison flipped when paginating backward. All queries are - * tenant-scoped. Mirrors {@code CustomScriptRepositoryImpl}; an invalid cursor - * is logged and treated as "no cursor" (first page). + *

Cursor pagination is a keyset over the active sort: plain {@code _id} when sorting by + * {@code _id} (the default, descending), otherwise a compound {@code (sortField, _id)} + * cursor — required because sort values repeat (many schedules share a {@code repeat} + * interval) and an {@code _id}-only cursor would skip and duplicate rows across a tie + * boundary. The comparison is flipped when paginating backward. All queries are + * tenant-scoped. An invalid cursor is logged and treated as "no cursor" (first page). */ @Slf4j @Repository @@ -47,10 +52,16 @@ public class CustomScriptScheduleRepositoryImpl implements CustomScriptScheduleR private static final String FIELD_UPDATED_AT = "updatedAt"; private static final String FIELD_CREATED_BY = "createdBy"; private static final String FIELD_COUNT = "count"; + private static final String FIELD_REPEAT = "repeat"; + private static final String FIELD_DEVICE_COUNT = "deviceCount"; + private static final String FIELD_SCRIPT_SCHEDULE_IDS = "scriptScheduleIds"; + private static final String FIELD_MACHINE_IDS = "machineIds"; + private static final String ASSIGNMENTS_COLLECTION = "script_schedules_machines_assigned"; + private static final String LOOKUP_ALIAS = "assignments"; + private static final String CURSOR_SEPARATOR = "|"; - /** Sort-field allowlist. Anything not in here falls back to {@link #getDefaultSortField()}. */ private static final Set SORTABLE_FIELDS = - Set.of(FIELD_ID, FIELD_NAME, FIELD_CREATED_AT, FIELD_UPDATED_AT); + Set.of(FIELD_ID, FIELD_NAME, FIELD_CREATED_AT, FIELD_UPDATED_AT, FIELD_REPEAT, FIELD_DEVICE_COUNT); private final MongoTemplate mongoTemplate; @@ -63,17 +74,104 @@ public List findPageForTenant(String tenantId, String cursor, boolean backward, int limit) { + Sort.Direction effectiveDir = backward ? flip(sortDirection) : sortDirection; + + if (FIELD_DEVICE_COUNT.equals(sortField)) { + return findPageAggregatedByDeviceCount(tenantId, filter, search, effectiveDir, cursor, limit); + } + Criteria criteria = buildBaseCriteria(tenantId, filter, search); - applyCursor(criteria, cursor, backward, sortDirection); + applyCursor(criteria, cursor, effectiveDir, sortField); - Sort.Direction effectiveDir = backward ? flip(sortDirection) : sortDirection; Query query = new Query(criteria) - .with(Sort.by(effectiveDir, sortField)) + .with(sortWithIdTiebreaker(effectiveDir, sortField)) .limit(limit); return mongoTemplate.find(query, ScriptSchedule.class); } + private List findPageAggregatedByDeviceCount(String tenantId, + ScriptScheduleQueryFilter filter, + String search, + Sort.Direction effectiveDir, + String cursor, + int limit) { + List ops = new ArrayList<>(); + ops.add(Aggregation.match(buildBaseCriteria(tenantId, filter, search))); + ops.add(deviceCountLookupStage()); + ops.add(deviceCountAddFieldsStage()); + appendDeviceCountCursor(ops, cursor, effectiveDir); + ops.add(Aggregation.sort(Sort.by( + new Sort.Order(effectiveDir, FIELD_DEVICE_COUNT), + new Sort.Order(effectiveDir, FIELD_ID)))); + ops.add(Aggregation.limit(limit)); + + AggregationResults results = + mongoTemplate.aggregate(Aggregation.newAggregation(ops), "script_schedules", Document.class); + return results.getMappedResults().stream().map(this::toSchedule).toList(); + } + + private static AggregationOperation deviceCountLookupStage() { + Document lookup = new Document("$lookup", new Document() + .append("from", ASSIGNMENTS_COLLECTION) + .append("let", new Document("schedId", "$_id").append("tenId", "$" + FIELD_TENANT_ID)) + .append("pipeline", List.of(new Document("$match", new Document("$expr", new Document("$and", List.of( + new Document("$eq", List.of("$" + FIELD_TENANT_ID, "$$tenId")), + new Document("$in", List.of("$$schedId", "$" + FIELD_SCRIPT_SCHEDULE_IDS)))))))) + .append("as", LOOKUP_ALIAS)); + return ctx -> lookup; + } + + private static AggregationOperation deviceCountAddFieldsStage() { + Document addFields = new Document("$addFields", new Document(FIELD_DEVICE_COUNT, + new Document("$sum", new Document("$map", new Document() + .append("input", "$" + LOOKUP_ALIAS) + .append("as", "a") + .append("in", new Document("$size", new Document("$ifNull", + List.of("$$a." + FIELD_MACHINE_IDS, List.of())))))))); + return ctx -> addFields; + } + + private static void appendDeviceCountCursor(List ops, String cursor, Sort.Direction effectiveDir) { + if (isBlank(cursor)) { + return; + } + int separator = cursor.lastIndexOf(CURSOR_SEPARATOR); + if (separator < 0) { + log.warn("Invalid deviceCount cursor (no separator): '{}' — falling back to first page", cursor); + return; + } + ObjectId cursorId = parseObjectId(cursor.substring(separator + 1)); + if (cursorId == null) { + return; + } + int count; + try { + count = Integer.parseInt(cursor.substring(0, separator)); + } catch (NumberFormatException ex) { + log.warn("Unparseable deviceCount in cursor '{}' — falling back to first page", cursor); + return; + } + + String cmp = effectiveDir == Sort.Direction.ASC ? "$gt" : "$lt"; + Document orExpr = new Document("$or", List.of( + new Document(FIELD_DEVICE_COUNT, new Document(cmp, count)), + new Document(FIELD_DEVICE_COUNT, count).append(FIELD_ID, new Document(cmp, cursorId)))); + ops.add(ctx -> new Document("$match", orExpr)); + } + + private ScriptSchedule toSchedule(Document doc) { + return mongoTemplate.getConverter().read(ScriptSchedule.class, doc); + } + + private int computeDeviceCount(ScriptSchedule schedule) { + Query q = new Query(Criteria.where(FIELD_TENANT_ID).is(schedule.getTenantId()) + .and(FIELD_SCRIPT_SCHEDULE_IDS).is(schedule.getId())); + return mongoTemplate.find(q, com.openframe.data.document.rmm.ScriptScheduleMachineAssigned.class).stream() + .mapToInt(a -> a.getMachineIds() == null ? 0 : a.getMachineIds().size()) + .sum(); + } + @Override public long countForTenant(String tenantId, ScriptScheduleQueryFilter filter, String search) { Query query = new Query(buildBaseCriteria(tenantId, filter, search)); @@ -180,30 +278,172 @@ private static void applySearch(Criteria criteria, String search) { criteria.and(FIELD_NAME).regex(Pattern.quote(search.trim()), "i"); } - private static void applyCursor(Criteria criteria, String cursor, boolean backward, Sort.Direction sortDirection) { + /** + * Keyset predicate matching the active sort. {@code effectiveDir} already folds in + * backward paging (forward+DESC and backward+ASC both walk "downward"), so the + * comparison operator follows it directly. + */ + private static void applyCursor(Criteria criteria, String cursor, Sort.Direction effectiveDir, String sortField) { if (isBlank(cursor)) { return; } - - ObjectId cursorId; - try { - cursorId = new ObjectId(cursor); - } catch (IllegalArgumentException ex) { - log.warn("Invalid ObjectId cursor for schedule pagination: '{}' — falling back to first page", cursor); + if (FIELD_ID.equals(sortField)) { + applyIdCursor(criteria, cursor, effectiveDir); return; } + applyCompoundCursor(criteria, cursor, effectiveDir, sortField); + } - // forward+DESC and backward+ASC both want {@code _id < cursor}; - // forward+ASC and backward+DESC want {@code _id > cursor}. - boolean useLessThan = (sortDirection == Sort.Direction.DESC) ^ backward; - if (useLessThan) { + private static void applyIdCursor(Criteria criteria, String cursor, Sort.Direction effectiveDir) { + ObjectId cursorId = parseObjectId(cursor); + if (cursorId == null) { + return; + } + if (effectiveDir == Sort.Direction.DESC) { criteria.and(FIELD_ID).lt(cursorId); } else { criteria.and(FIELD_ID).gt(cursorId); } } + /** + * Compound keyset over {@code (sortField, _id)} — the only correct way to page a sort + * whose values repeat (a handful of distinct {@code repeat} intervals across the whole + * list). Sorting on the value alone with an {@code _id}-only cursor silently skips and + * repeats rows across a tie boundary. + * + *

Nulls need explicit handling: MongoDB's range operators never match {@code null} + * (a number-bracket {@code $lt} skips it), while BSON ordering puts null before + * numbers. So ASC must additionally sweep in every non-null row once the null group is + * exhausted, and DESC must append the null tail that sorts after every value. + */ + private static void applyCompoundCursor(Criteria criteria, String cursor, + Sort.Direction effectiveDir, String sortField) { + int separator = cursor.lastIndexOf(CURSOR_SEPARATOR); + if (separator < 0) { + log.warn("Invalid compound cursor (no separator) for schedule pagination: '{}' — falling back to first page", cursor); + return; + } + ObjectId cursorId = parseObjectId(cursor.substring(separator + 1)); + if (cursorId == null) { + return; + } + + Object value; + try { + value = parseSortValue(cursor.substring(0, separator), sortField); + } catch (RuntimeException ex) { + log.warn("Unparseable '{}' value in schedule cursor '{}' — falling back to first page", sortField, cursor); + return; + } + + boolean ascending = effectiveDir == Sort.Direction.ASC; + List or = new ArrayList<>(); + + if (value == null) { + // Cursor sits inside the null group: continue it by _id... + or.add(tieBreak(sortField, null, cursorId, ascending)); + if (ascending) { + // ...and, ascending, everything non-null still lies ahead. + or.add(Criteria.where(sortField).ne(null).getCriteriaObject()); + } + } else { + or.add(ascending + ? Criteria.where(sortField).gt(value).getCriteriaObject() + : Criteria.where(sortField).lt(value).getCriteriaObject()); + or.add(tieBreak(sortField, value, cursorId, ascending)); + if (!ascending) { + // Descending, the null tail sorts after every value — none of it seen yet. + or.add(Criteria.where(sortField).is(null).getCriteriaObject()); + } + } + + // $or as an explicit key: a second keyless criteria would clash with the base + // filter's chained predicate (MongoDB Query rejects it). Mirrors CustomOrganizationRepositoryImpl. + criteria.and("$or").is(or); + } + + /** {@code {sortField: value, _id: {$gt|$lt: cursorId}}} — same sort value, next id. */ + private static Document tieBreak(String sortField, Object value, ObjectId cursorId, boolean ascending) { + Criteria sameValue = Criteria.where(sortField).is(value); + return (ascending ? sameValue.and(FIELD_ID).gt(cursorId) : sameValue.and(FIELD_ID).lt(cursorId)) + .getCriteriaObject(); + } + + private static ObjectId parseObjectId(String raw) { + try { + return new ObjectId(raw); + } catch (IllegalArgumentException ex) { + log.warn("Invalid ObjectId in schedule cursor: '{}' — falling back to first page", raw); + return null; + } + } + + /** Empty means "the row's sort value was null" (see {@link #encodeSortValue}). */ + private static Object parseSortValue(String raw, String sortField) { + if (raw.isEmpty()) { + return null; + } + return switch (sortField) { + case FIELD_REPEAT -> Long.parseLong(raw); + case FIELD_DEVICE_COUNT -> Integer.parseInt(raw); + case FIELD_CREATED_AT, FIELD_UPDATED_AT -> Date.from(Instant.ofEpochMilli(Long.parseLong(raw))); + default -> raw; + }; + } + + @Override + public String encodeCursor(ScriptSchedule schedule, String sortField) { + if (schedule == null) { + return null; + } + if (FIELD_ID.equals(sortField)) { + return schedule.getId(); + } + return encodeSortValue(schedule, sortField) + CURSOR_SEPARATOR + schedule.getId(); + } + + /** + * Sort value as a string, empty for null. Parsed back by {@link #parseSortValue}; the + * separator is located from the right, so a value containing it (a schedule name) is + * still split correctly against the fixed-length ObjectId. + */ + private String encodeSortValue(ScriptSchedule schedule, String sortField) { + Object value = switch (sortField) { + case FIELD_NAME -> schedule.getName(); + case FIELD_CREATED_AT -> schedule.getCreatedAt(); + case FIELD_UPDATED_AT -> schedule.getUpdatedAt(); + case FIELD_REPEAT -> schedule.getRepeat(); + case FIELD_DEVICE_COUNT -> computeDeviceCount(schedule); + default -> null; + }; + if (value == null) { + return ""; + } + if (value instanceof Instant instant) { + return String.valueOf(instant.toEpochMilli()); + } + return String.valueOf(value); + } + private static Sort.Direction flip(Sort.Direction direction) { return direction == Sort.Direction.ASC ? Sort.Direction.DESC : Sort.Direction.ASC; } + + /** + * Sort by the requested field with {@code _id} as a tie-breaker, so rows sharing a + * sort value (very common for {@code repeat} — a handful of distinct intervals across + * the whole list) come back in a stable, repeatable order instead of Mongo's arbitrary + * one. Redundant when already sorting by {@code _id}. + * + *

Note: the cursor predicate is still {@code _id}-only, so deep paging across a tie + * boundary on a non-{@code _id} sort can skip/repeat rows. Pre-existing for + * name/createdAt/updatedAt; a compound keyset cursor is the proper fix. + */ + private static Sort sortWithIdTiebreaker(Sort.Direction direction, String sortField) { + if (FIELD_ID.equals(sortField)) { + return Sort.by(direction, FIELD_ID); + } + return Sort.by(direction, sortField).and(Sort.by(direction, FIELD_ID)); + } } diff --git a/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/ScheduleScriptExecutionRepository.java b/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/ScheduleScriptExecutionRepository.java new file mode 100644 index 000000000..a9286314e --- /dev/null +++ b/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/ScheduleScriptExecutionRepository.java @@ -0,0 +1,17 @@ +package com.openframe.data.repository.rmm; + +import com.openframe.data.document.rmm.ScheduleScriptExecution; +import org.springframework.data.mongodb.repository.MongoRepository; +import org.springframework.stereotype.Repository; + +import java.util.Optional; + +/** + * Repository for {@link ScheduleScriptExecution} header rows (one per schedule fire). + * Per-(script, machine) leaf rows live in {@code ScriptExecutionRepository}. + */ +@Repository +public interface ScheduleScriptExecutionRepository extends MongoRepository { + + Optional findByTenantIdAndExecutionId(String tenantId, String executionId); +} diff --git a/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/ScriptExecutionRepository.java b/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/ScriptExecutionRepository.java index 41d5c8afe..0c3c35b3a 100644 --- a/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/ScriptExecutionRepository.java +++ b/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/ScriptExecutionRepository.java @@ -26,6 +26,8 @@ public interface ScriptExecutionRepository Optional findByMachineIdAndExecutionId(String machineId, String executionId); + Optional findByMachineIdAndExecutionIdAndScriptId(String machineId, String executionId, String scriptId); + Optional findFirstByTenantIdAndExecutionId(String tenantId, String executionId); List findByStatusAndDispatchedAtBefore(ExecutionStatus status, Instant dispatchedAtBefore); diff --git a/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/ScriptScheduleRepository.java b/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/ScriptScheduleRepository.java index 5b2ce01a0..003799a51 100644 --- a/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/ScriptScheduleRepository.java +++ b/openframe-data-mongo-sync/src/main/java/com/openframe/data/repository/rmm/ScriptScheduleRepository.java @@ -5,6 +5,7 @@ import org.springframework.data.mongodb.repository.MongoRepository; import org.springframework.stereotype.Repository; +import java.time.Instant; import java.util.Collection; import java.util.List; import java.util.Optional; @@ -29,4 +30,6 @@ public interface ScriptScheduleRepository boolean existsByTenantIdAndNameAndStatusIn(String tenantId, String name, Collection statuses); boolean existsByTenantIdAndNameAndIdNotAndStatusIn(String tenantId, String name, String excludeId, Collection statuses); + + List findByStatusAndNextRunAtLessThanEqual(ScriptStatus status, Instant cutoff); } diff --git a/openframe-data-mongo-sync/src/test/java/com/openframe/data/integration/repository/rmm/ScriptScheduleRepositoryIT.java b/openframe-data-mongo-sync/src/test/java/com/openframe/data/integration/repository/rmm/ScriptScheduleRepositoryIT.java index accf7e2fc..56ca6a051 100644 --- a/openframe-data-mongo-sync/src/test/java/com/openframe/data/integration/repository/rmm/ScriptScheduleRepositoryIT.java +++ b/openframe-data-mongo-sync/src/test/java/com/openframe/data/integration/repository/rmm/ScriptScheduleRepositoryIT.java @@ -14,10 +14,12 @@ import org.junit.jupiter.api.condition.EnabledIfSystemProperty; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; +import org.bson.types.ObjectId; import org.springframework.data.domain.Sort; import org.springframework.data.mongodb.core.MongoTemplate; import org.springframework.data.mongodb.core.query.Query; +import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -101,6 +103,248 @@ void findPageForTenant_forwardCursor_olderRows() { assertThat(page).extracting(ScriptSchedule::getId).containsExactly(s2.getId(), s1.getId()); } + @Test + @DisplayName("repeat is a sortable field; _id is not (allowlist guards the sort input)") + void isSortableField_includesRepeat() { + assertThat(scheduleRepository.isSortableField("repeat")).isTrue(); + assertThat(scheduleRepository.isSortableField("bogus")).isFalse(); + } + + @Test + @DisplayName("findPageForTenant sorts by repeat ASC — nulls (one-shot schedules) first, then ascending interval") + void findPageForTenant_sortsByRepeatAscending() { + ScriptSchedule weekly = saveWithRepeat(TENANT_A, "weekly", 604800L); + ScriptSchedule halfHour = saveWithRepeat(TENANT_A, "halfHour", 1800L); + ScriptSchedule oneShot = saveWithRepeat(TENANT_A, "oneShot", null); + + List page = scheduleRepository.findPageForTenant( + TENANT_A, null, null, "repeat", Sort.Direction.ASC, null, false, 10); + + assertThat(page).extracting(ScriptSchedule::getId) + .containsExactly(oneShot.getId(), halfHour.getId(), weekly.getId()); + } + + @Test + @DisplayName("findPageForTenant sorts by repeat DESC — largest interval first") + void findPageForTenant_sortsByRepeatDescending() { + ScriptSchedule halfHour = saveWithRepeat(TENANT_A, "halfHour", 1800L); + ScriptSchedule weekly = saveWithRepeat(TENANT_A, "weekly", 604800L); + ScriptSchedule daily = saveWithRepeat(TENANT_A, "daily", 86400L); + + List page = scheduleRepository.findPageForTenant( + TENANT_A, null, null, "repeat", Sort.Direction.DESC, null, false, 10); + + assertThat(page).extracting(ScriptSchedule::getId) + .startsWith(weekly.getId(), daily.getId(), halfHour.getId()); + } + + @Test + @DisplayName("equal repeat values fall back to the _id tie-breaker — order is stable, not arbitrary") + void findPageForTenant_repeatTiesBrokenById() { + ScriptSchedule a = saveWithRepeat(TENANT_A, "a", 604800L); + ScriptSchedule b = saveWithRepeat(TENANT_A, "b", 604800L); + ScriptSchedule c = saveWithRepeat(TENANT_A, "c", 604800L); + + List desc = scheduleRepository.findPageForTenant( + TENANT_A, null, null, "repeat", Sort.Direction.DESC, null, false, 10); + // All three tie on repeat → _id DESC decides, and repeats identically across calls. + assertThat(desc).extracting(ScriptSchedule::getId) + .containsExactly(c.getId(), b.getId(), a.getId()); + + List again = scheduleRepository.findPageForTenant( + TENANT_A, null, null, "repeat", Sort.Direction.DESC, null, false, 10); + assertThat(again).extracting(ScriptSchedule::getId) + .containsExactlyElementsOf(desc.stream().map(ScriptSchedule::getId).toList()); + } + + @Test + @DisplayName("compound cursor pages through a repeat tie group without skipping or repeating rows") + void findPageForTenant_repeatKeysetPagesCleanlyAcrossTies() { + // 5 schedules, only 2 distinct repeat values → every page boundary lands inside a tie. + ScriptSchedule w1 = saveWithRepeat(TENANT_A, "w1", 604800L); + ScriptSchedule w2 = saveWithRepeat(TENANT_A, "w2", 604800L); + ScriptSchedule w3 = saveWithRepeat(TENANT_A, "w3", 604800L); + ScriptSchedule h1 = saveWithRepeat(TENANT_A, "h1", 1800L); + ScriptSchedule h2 = saveWithRepeat(TENANT_A, "h2", 1800L); + + List walked = new ArrayList<>(); + String cursor = null; + for (int page = 0; page < 5; page++) { + List chunk = scheduleRepository.findPageForTenant( + TENANT_A, null, null, "repeat", Sort.Direction.DESC, cursor, false, 2); + if (chunk.isEmpty()) { + break; + } + chunk.forEach(s -> walked.add(s.getId())); + cursor = scheduleRepository.encodeCursor(chunk.getLast(), "repeat"); + } + + // Every row exactly once, in (repeat DESC, _id DESC) order. + assertThat(walked).containsExactly(w3.getId(), w2.getId(), w1.getId(), h2.getId(), h1.getId()); + assertThat(walked).doesNotHaveDuplicates(); + } + + @Test + @DisplayName("compound cursor ASC crosses the null (one-shot) group into the numeric group exactly once") + void findPageForTenant_repeatKeysetCrossesNullBoundaryAscending() { + ScriptSchedule n1 = saveWithRepeat(TENANT_A, "n1", null); + ScriptSchedule n2 = saveWithRepeat(TENANT_A, "n2", null); + ScriptSchedule half = saveWithRepeat(TENANT_A, "half", 1800L); + ScriptSchedule week = saveWithRepeat(TENANT_A, "week", 604800L); + + List walked = new ArrayList<>(); + String cursor = null; + for (int page = 0; page < 5; page++) { + List chunk = scheduleRepository.findPageForTenant( + TENANT_A, null, null, "repeat", Sort.Direction.ASC, cursor, false, 1); + if (chunk.isEmpty()) { + break; + } + chunk.forEach(s -> walked.add(s.getId())); + cursor = scheduleRepository.encodeCursor(chunk.getLast(), "repeat"); + } + + // Nulls first (by _id ASC), then ascending intervals — no row lost at the null→number boundary. + assertThat(walked).containsExactly(n1.getId(), n2.getId(), half.getId(), week.getId()); + } + + @Test + @DisplayName("encodeCursor: plain id for an _id sort, value|id for a compound sort, empty value for null repeat") + void encodeCursor_formats() { + ScriptSchedule weekly = saveWithRepeat(TENANT_A, "weekly", 604800L); + ScriptSchedule oneShot = saveWithRepeat(TENANT_A, "oneShot", null); + + assertThat(scheduleRepository.encodeCursor(weekly, "_id")).isEqualTo(weekly.getId()); + assertThat(scheduleRepository.encodeCursor(weekly, "repeat")).isEqualTo("604800|" + weekly.getId()); + assertThat(scheduleRepository.encodeCursor(oneShot, "repeat")).isEqualTo("|" + oneShot.getId()); + } + + @Test + @DisplayName("a malformed cursor falls back to the first page instead of throwing") + void findPageForTenant_malformedCursorFallsBackToFirstPage() { + saveWithRepeat(TENANT_A, "a", 1800L); + saveWithRepeat(TENANT_A, "b", 1800L); + + List noSeparator = scheduleRepository.findPageForTenant( + TENANT_A, null, null, "repeat", Sort.Direction.DESC, "garbage", false, 10); + List badId = scheduleRepository.findPageForTenant( + TENANT_A, null, null, "repeat", Sort.Direction.DESC, "1800|not-an-objectid", false, 10); + List badValue = scheduleRepository.findPageForTenant( + TENANT_A, null, null, "repeat", Sort.Direction.DESC, "abc|" + new ObjectId(), false, 10); + + assertThat(noSeparator).hasSize(2); + assertThat(badId).hasSize(2); + assertThat(badValue).hasSize(2); + } + + private ScriptSchedule saveWithRepeat(String tenantId, String name, Long repeat) { + ScriptSchedule schedule = ScriptSchedule.builder() + .tenantId(tenantId) + .name(name) + .status(ScriptStatus.ACTIVE) + .createdBy("user-1") + .supportedPlatforms(List.of(ScriptPlatform.WINDOWS)) + .repeat(repeat) + .build(); + return scheduleRepository.save(schedule); + } + + @Test + @DisplayName("deviceCount IS sortable — served via aggregation ($lookup + $addFields on assignments), no denormalised field on the schedule document") + void isSortableField_includesDeviceCount() { + assertThat(scheduleRepository.isSortableField("deviceCount")).isTrue(); + } + + @Test + @DisplayName("sort by deviceCount DESC — schedules with the most assigned machines come first; deviceCount is populated on the returned entities from the aggregation") + void findPageForTenant_sortsByDeviceCountDesc() { + ScriptSchedule few = saveActive(TENANT_A, "few"); + ScriptSchedule many = saveActive(TENANT_A, "many"); + ScriptSchedule mid = saveActive(TENANT_A, "mid"); + assignMachines(TENANT_A, few.getId(), List.of("m1", "m2")); + assignMachines(TENANT_A, mid.getId(), List.of("m1", "m2", "m3", "m4")); + assignMachines(TENANT_A, many.getId(), List.of("m1", "m2", "m3", "m4", "m5", "m6")); + + List page = scheduleRepository.findPageForTenant( + TENANT_A, null, null, "deviceCount", Sort.Direction.DESC, null, false, 10); + + assertThat(page).extracting(ScriptSchedule::getId) + .containsExactly(many.getId(), mid.getId(), few.getId()); + // Cursor encoding reads the count via an independent indexed lookup — verifies the + // schedule doc stays clean (no denormalised deviceCount field) while still supporting sort. + assertThat(scheduleRepository.encodeCursor(page.get(0), "deviceCount")).startsWith("6|"); + assertThat(scheduleRepository.encodeCursor(page.get(1), "deviceCount")).startsWith("4|"); + assertThat(scheduleRepository.encodeCursor(page.get(2), "deviceCount")).startsWith("2|"); + } + + @Test + @DisplayName("sort by deviceCount ASC — schedules with NO assignments (missing assignment doc) count as 0 and come first, not last") + void findPageForTenant_sortsByDeviceCountAsc_missingAssignmentIsZero() { + ScriptSchedule none = saveActive(TENANT_A, "none"); // no assignment doc at all + ScriptSchedule some = saveActive(TENANT_A, "some"); + assignMachines(TENANT_A, some.getId(), List.of("m1", "m2")); + + List page = scheduleRepository.findPageForTenant( + TENANT_A, null, null, "deviceCount", Sort.Direction.ASC, null, false, 10); + + assertThat(page).extracting(ScriptSchedule::getId).containsExactly(none.getId(), some.getId()); + // cursor for the no-assignment schedule encodes 0 (matching the aggregation's $sum semantics) + assertThat(scheduleRepository.encodeCursor(page.get(0), "deviceCount")).startsWith("0|"); + assertThat(scheduleRepository.encodeCursor(page.get(1), "deviceCount")).startsWith("2|"); + } + + @Test + @DisplayName("aggregation cursor pages through a deviceCount tie group without skipping or repeating rows (infinite-scroll semantics)") + void findPageForTenant_deviceCountCursorPagesCleanlyAcrossTies() { + ScriptSchedule a = saveActive(TENANT_A, "a"); + ScriptSchedule b = saveActive(TENANT_A, "b"); + ScriptSchedule c = saveActive(TENANT_A, "c"); + ScriptSchedule d = saveActive(TENANT_A, "d"); + // three schedules tied on 3 machines each; the fourth trails on 1 + assignMachines(TENANT_A, a.getId(), List.of("m1", "m2", "m3")); + assignMachines(TENANT_A, b.getId(), List.of("m1", "m2", "m3")); + assignMachines(TENANT_A, c.getId(), List.of("m1", "m2", "m3")); + assignMachines(TENANT_A, d.getId(), List.of("m1")); + + List walked = new ArrayList<>(); + String cursor = null; + for (int pass = 0; pass < 5; pass++) { + List chunk = scheduleRepository.findPageForTenant( + TENANT_A, null, null, "deviceCount", Sort.Direction.DESC, cursor, false, 2); + if (chunk.isEmpty()) { + break; + } + chunk.forEach(x -> walked.add(x.getId())); + cursor = scheduleRepository.encodeCursor(chunk.getLast(), "deviceCount"); + } + + assertThat(walked).hasSize(4); + assertThat(walked).doesNotHaveDuplicates(); + assertThat(walked.getLast()).isEqualTo(d.getId()); // "d" (count=1) always last on DESC + } + + @Test + @DisplayName("sort by deviceCount is tenant-isolated — a schedule in tenant B does not leak into tenant A's page even if their assignment docs coexist in the collection") + void findPageForTenant_deviceCountAggregation_isTenantIsolated() { + ScriptSchedule aOnly = saveActive(TENANT_A, "a-only"); + ScriptSchedule bOnly = save(TENANT_B, "b-only", ScriptStatus.ACTIVE, "user-1", List.of(ScriptPlatform.LINUX)); + assignMachines(TENANT_A, aOnly.getId(), List.of("m1")); + assignMachines(TENANT_B, bOnly.getId(), List.of("m1", "m2", "m3", "m4")); + + List page = scheduleRepository.findPageForTenant( + TENANT_A, null, null, "deviceCount", Sort.Direction.DESC, null, false, 10); + + assertThat(page).extracting(ScriptSchedule::getId).containsExactly(aOnly.getId()); + } + + private void assignMachines(String tenantId, String scheduleId, List machineIds) { + mongoTemplate.getCollection("script_schedules_machines_assigned").insertOne( + new org.bson.Document() + .append("tenantId", tenantId) + .append("scriptScheduleIds", List.of(scheduleId)) + .append("machineIds", machineIds)); + } + @Test @DisplayName("findPageForTenant excludes soft-deleted schedules by default") void findPageForTenant_excludesDeleted() { diff --git a/openframe-data-nats/src/main/java/com/openframe/data/nats/rmm/model/RmmResultMessage.java b/openframe-data-nats/src/main/java/com/openframe/data/nats/rmm/model/RmmResultMessage.java index 9f2293511..015622bf5 100644 --- a/openframe-data-nats/src/main/java/com/openframe/data/nats/rmm/model/RmmResultMessage.java +++ b/openframe-data-nats/src/main/java/com/openframe/data/nats/rmm/model/RmmResultMessage.java @@ -12,12 +12,13 @@ /** * Wire payload published by the OpenFrame agent over core NATS for the result * of an RMM execution. Sealed: exactly two concrete subtypes — - * {@link CommandResultMessage} and {@link ScriptResultMessage} — share this - * shape verbatim and only differ in their Java type so downstream code (the - * result service, audit, etc.) can distinguish a command result from a - * saved-script result without an in-payload discriminator. The sealed - * declaration also makes any pattern-matching switch on this type - * compile-time exhaustive — a new subtype can't be added without forcing + * {@link CommandResultMessage} and {@link ScriptResultMessage}. This class holds + * the fields common to both; {@link ScriptResultMessage} adds the script-specific + * {@code scriptId}/{@code scheduleId} (meaningless for an ad-hoc command). The + * distinct Java types also let downstream code (the result service, audit, etc.) + * distinguish a command result from a saved-script result without an in-payload + * discriminator. The sealed declaration makes any pattern-matching switch on this + * type compile-time exhaustive — a new subtype can't be added without forcing * every consumer to consider it. * *

Mirrors the agent's execution-result struct. The agent serializes with diff --git a/openframe-data-nats/src/main/java/com/openframe/data/nats/rmm/model/ScriptMessage.java b/openframe-data-nats/src/main/java/com/openframe/data/nats/rmm/model/ScriptMessage.java index 00f87f647..253395f39 100644 --- a/openframe-data-nats/src/main/java/com/openframe/data/nats/rmm/model/ScriptMessage.java +++ b/openframe-data-nats/src/main/java/com/openframe/data/nats/rmm/model/ScriptMessage.java @@ -28,6 +28,10 @@ public class ScriptMessage { private String executionId; + private String scheduleId; + + private String scriptId; + private String machineId; private String code; diff --git a/openframe-data-nats/src/main/java/com/openframe/data/nats/rmm/model/ScriptResultMessage.java b/openframe-data-nats/src/main/java/com/openframe/data/nats/rmm/model/ScriptResultMessage.java index 7309674b7..5b252d8cf 100644 --- a/openframe-data-nats/src/main/java/com/openframe/data/nats/rmm/model/ScriptResultMessage.java +++ b/openframe-data-nats/src/main/java/com/openframe/data/nats/rmm/model/ScriptResultMessage.java @@ -9,13 +9,20 @@ * Wire payload for a result of a saved script execution. * Subject: {@code machine.{machineId}.script-execution.result}. * - *

Same shape as {@link RmmResultMessage}; the distinct Java type is what - * lets the result service route this to {@code MessageType.SCRIPT_EXECUTED} - * downstream without an in-payload discriminator field. + *

Extends {@link RmmResultMessage} with the two script-specific ids the agent + * echoes back from the dispatch payload. The distinct Java type is also what lets + * the result service route this to {@code MessageType.SCRIPT_EXECUTED} downstream + * without an in-payload discriminator field. + * + *

Snake_case mapping ({@code script_id} → {@code scriptId}) comes from the + * {@code @JsonNaming} on the superclass. */ @Data @EqualsAndHashCode(callSuper = true) @SuperBuilder @NoArgsConstructor public final class ScriptResultMessage extends RmmResultMessage { + + private String scriptId; + private String scheduleId; } diff --git a/openframe-data-nats/src/main/java/com/openframe/data/nats/rmm/model/ScriptScheduleExecutionItem.java b/openframe-data-nats/src/main/java/com/openframe/data/nats/rmm/model/ScriptScheduleExecutionItem.java new file mode 100644 index 000000000..dc3009b74 --- /dev/null +++ b/openframe-data-nats/src/main/java/com/openframe/data/nats/rmm/model/ScriptScheduleExecutionItem.java @@ -0,0 +1,39 @@ +package com.openframe.data.nats.rmm.model; + +import com.fasterxml.jackson.annotation.JsonInclude; +import com.openframe.data.document.rmm.PrivilegeLevel; +import com.openframe.data.document.rmm.ScriptEnvVar; +import com.openframe.data.document.rmm.ScriptShell; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.util.List; + +/** + * One saved-script payload inside a {@link ScriptScheduleExecutionMessage}. Mirrors the + * per-script fields the agent needs to execute (code / shell / privilegeLevel / defaults), + * with no override slots — schedules always run scripts with their stored defaults. + */ +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +@JsonInclude(JsonInclude.Include.NON_NULL) +public class ScriptScheduleExecutionItem { + + private String scriptId; + + private String code; + + private ScriptShell shell; + + private PrivilegeLevel privilegeLevel; + + private List args; + + private Integer timeoutSeconds; + + private List envVars; +} diff --git a/openframe-data-nats/src/main/java/com/openframe/data/nats/rmm/model/ScriptScheduleExecutionMessage.java b/openframe-data-nats/src/main/java/com/openframe/data/nats/rmm/model/ScriptScheduleExecutionMessage.java new file mode 100644 index 000000000..d3d10bc67 --- /dev/null +++ b/openframe-data-nats/src/main/java/com/openframe/data/nats/rmm/model/ScriptScheduleExecutionMessage.java @@ -0,0 +1,43 @@ +package com.openframe.data.nats.rmm.model; + +import com.fasterxml.jackson.annotation.JsonInclude; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.util.List; + +/** + * Wire payload sent to the OpenFrame agent over core NATS for a schedule fire — + * one message per target machine, carrying the full set of scripts the + * schedule runs. Replaces the per-script fan-out ({@link ScriptMessage}) for + * schedule-triggered dispatches: for M machines and N scripts, the schedule + * flow emits M messages instead of N×M. + * + * 

+ *   Subject: machine.{machineId}.script-schedule-execution
+ *
+ * + *

{@code executionId} is shared across every message in the run and every + * leaf {@code ScriptExecution} row it produces — the agent reports each + * (scriptId, machineId) result back on the existing result channel and it + * correlates via that shared id. + */ +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +@JsonInclude(JsonInclude.Include.NON_NULL) +public class ScriptScheduleExecutionMessage { + + private String executionId; + + private String scheduleId; + + private String machineId; + + private String initiatedBy; + + private List scripts; +} diff --git a/openframe-data-nats/src/main/java/com/openframe/data/nats/rmm/publisher/ScriptScheduleExecutionNatsPublisher.java b/openframe-data-nats/src/main/java/com/openframe/data/nats/rmm/publisher/ScriptScheduleExecutionNatsPublisher.java new file mode 100644 index 000000000..d6dbee9f7 --- /dev/null +++ b/openframe-data-nats/src/main/java/com/openframe/data/nats/rmm/publisher/ScriptScheduleExecutionNatsPublisher.java @@ -0,0 +1,49 @@ +package com.openframe.data.nats.rmm.publisher; + +import com.openframe.data.nats.publisher.NatsMessagePublisher; +import com.openframe.data.nats.rmm.model.ScriptScheduleExecutionMessage; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.stereotype.Component; + +import static java.lang.String.format; + +/** + * Domain publisher for RMM schedule fires sent to an agent over core NATS. One + * message per target machine carries every script the schedule runs — see + * {@link ScriptScheduleExecutionMessage}. The per-script {@code ScriptNatsPublisher} + * stays alive for ad-hoc {@code runScript} / {@code batchRunScript} dispatches. + */ +@Component +@RequiredArgsConstructor +@ConditionalOnProperty("spring.cloud.stream.enabled") +@Slf4j +public class ScriptScheduleExecutionNatsPublisher { + + private static final String SUBJECT_TEMPLATE = "machine.%s.script-schedule-execution"; + + private final NatsMessagePublisher natsMessagePublisher; + + /** + * Publish a schedule-fire batch to the target agent over core NATS. + * + *

{@code machineId} is expected to be a subject-safe token — that is + * validated at the API boundary ({@code @Pattern} on the GraphQL input). + * + * @throws IllegalArgumentException if {@code message} is null + * @throws com.openframe.core.exception.NatsException if the underlying + * NATS publish fails + */ + public void publish(String machineId, ScriptScheduleExecutionMessage message) { + if (message == null) { + throw new IllegalArgumentException("ScriptScheduleExecutionMessage must not be null"); + } + + String subject = format(SUBJECT_TEMPLATE, machineId); + natsMessagePublisher.publish(subject, message); + int scriptCount = message.getScripts() == null ? 0 : message.getScripts().size(); + log.info("Published schedule-execution batch: machineId={} subject={} executionId={} scheduleId={} scripts={}", + machineId, subject, message.getExecutionId(), message.getScheduleId(), scriptCount); + } +} diff --git a/openframe-data-nats/src/test/java/com/openframe/data/nats/rmm/publisher/ScriptScheduleExecutionNatsPublisherTest.java b/openframe-data-nats/src/test/java/com/openframe/data/nats/rmm/publisher/ScriptScheduleExecutionNatsPublisherTest.java new file mode 100644 index 000000000..f5784696f --- /dev/null +++ b/openframe-data-nats/src/test/java/com/openframe/data/nats/rmm/publisher/ScriptScheduleExecutionNatsPublisherTest.java @@ -0,0 +1,107 @@ +package com.openframe.data.nats.rmm.publisher; + +import com.openframe.core.exception.NatsException; +import com.openframe.data.document.rmm.PrivilegeLevel; +import com.openframe.data.document.rmm.ScriptShell; +import com.openframe.data.nats.publisher.NatsMessagePublisher; +import com.openframe.data.nats.rmm.model.ScriptScheduleExecutionItem; +import com.openframe.data.nats.rmm.model.ScriptScheduleExecutionMessage; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; + +class ScriptScheduleExecutionNatsPublisherTest { + + private NatsMessagePublisher messagePublisher; + private ScriptScheduleExecutionNatsPublisher publisher; + + @BeforeEach + void setUp() { + messagePublisher = mock(NatsMessagePublisher.class); + publisher = new ScriptScheduleExecutionNatsPublisher(messagePublisher); + } + + @Test + @DisplayName("publish: routes to machine..script-schedule-execution and forwards the batched payload verbatim") + void publish_routesToMachineSubjectWithUnchangedPayload() { + ScriptScheduleExecutionMessage message = ScriptScheduleExecutionMessage.builder() + .executionId("exec-1") + .scheduleId("sched-1") + .machineId("machine-42") + .initiatedBy("cron") + .scripts(List.of( + ScriptScheduleExecutionItem.builder() + .scriptId("s1").code("df -h") + .shell(ScriptShell.BASH).privilegeLevel(PrivilegeLevel.ADMIN) + .build(), + ScriptScheduleExecutionItem.builder() + .scriptId("s2").code("Get-Process") + .shell(ScriptShell.POWERSHELL).privilegeLevel(PrivilegeLevel.USER) + .build())) + .build(); + + publisher.publish("machine-42", message); + + ArgumentCaptor subject = ArgumentCaptor.forClass(String.class); + ArgumentCaptor body = + ArgumentCaptor.forClass(ScriptScheduleExecutionMessage.class); + verify(messagePublisher).publish(subject.capture(), body.capture()); + assertThat(subject.getValue()).isEqualTo("machine.machine-42.script-schedule-execution"); + // Same instance reaches the transport — publisher does not repackage the payload. + assertThat(body.getValue()).isSameAs(message); + } + + @Test + @DisplayName("publish: uses core NATS publish, NOT publishPersistent — schedule fires must not be durable / replayable") + void publish_usesCoreNatsNotJetStream() { + ScriptScheduleExecutionMessage message = ScriptScheduleExecutionMessage.builder() + .executionId("exec-1").scheduleId("sched-1").machineId("machine-42") + .scripts(List.of(ScriptScheduleExecutionItem.builder().scriptId("s1").code("ls") + .shell(ScriptShell.BASH).privilegeLevel(PrivilegeLevel.USER).build())) + .build(); + + publisher.publish("machine-42", message); + + verify(messagePublisher).publish(anyString(), any(ScriptScheduleExecutionMessage.class)); + // Locks in the transport choice: a future "let's add durability" change + // can't silently flip this back to JetStream without breaking this test. + verify(messagePublisher, never()).publishPersistent(anyString(), any()); + } + + @Test + @DisplayName("publish: null message is rejected before any broker call") + void publish_rejectsNullMessage() { + assertThatThrownBy(() -> publisher.publish("machine-42", null)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("ScriptScheduleExecutionMessage"); + verifyNoInteractions(messagePublisher); + } + + @Test + @DisplayName("publish: NatsException from the transport propagates so the caller can surface dispatch failure") + void publish_propagatesTransportFailure() { + ScriptScheduleExecutionMessage message = ScriptScheduleExecutionMessage.builder() + .executionId("exec-1").scheduleId("sched-1").machineId("machine-42") + .scripts(List.of(ScriptScheduleExecutionItem.builder().scriptId("s1").code("ls") + .shell(ScriptShell.BASH).privilegeLevel(PrivilegeLevel.USER).build())) + .build(); + doThrow(new NatsException("broker offline")) + .when(messagePublisher).publish(anyString(), any(ScriptScheduleExecutionMessage.class)); + + assertThatThrownBy(() -> publisher.publish("machine-42", message)) + .isInstanceOf(NatsException.class); + } +} diff --git a/openframe-management-service-core/src/main/java/com/openframe/management/scheduler/ScriptScheduleScheduler.java b/openframe-management-service-core/src/main/java/com/openframe/management/scheduler/ScriptScheduleScheduler.java new file mode 100644 index 000000000..295a26be6 --- /dev/null +++ b/openframe-management-service-core/src/main/java/com/openframe/management/scheduler/ScriptScheduleScheduler.java @@ -0,0 +1,53 @@ +package com.openframe.management.scheduler; + +import com.openframe.management.service.ScriptScheduleExecutionService; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import net.javacrumbs.shedlock.spring.annotation.SchedulerLock; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +/** + * Cron wiring for the time-driven RMM schedule runner. Delegates to + * {@link ScriptScheduleExecutionService}; this class holds only the schedule + * and the ShedLock guard. + * + *

Disabled by default — enable per environment via + * {@code openframe.rmm.schedule.runner.enabled=true}. The cron is tuneable via + * {@code openframe.rmm.schedule.runner.cron}. + * + *

The sweep runs on the half-hour grid (xx:00 and xx:30) rather than on a + * fixed delay, because every schedule is constrained to that same grid: {@code startAt} + * must land on a boundary and {@code repeat} must be a whole number of 30-minute slots. + * Ticking on the grid makes a due schedule fire at its slot instead of up to a poll + * interval late, and keeps firing times identical across restarts (a fixed delay drifts + * with whenever the pod happened to start). Pinned to UTC so the boundaries do not move + * with the pod's local zone — offsets like +05:45 would otherwise shift them. + * + *

{@link SchedulerLock} serialises the sweep across management replicas so a + * due schedule is dispatched exactly once per fire even when several pods run. + */ +@Component +@RequiredArgsConstructor +@Slf4j +@ConditionalOnProperty(name = "openframe.rmm.schedule.runner.enabled", havingValue = "true") +public class ScriptScheduleScheduler { + + private final ScriptScheduleExecutionService scheduleExecutionService; + + @Scheduled(cron = "${openframe.rmm.schedule.runner.cron:0 0,30 * * * *}", zone = "UTC") + @SchedulerLock( + name = "scriptScheduleRunner", + lockAtMostFor = "${openframe.rmm.schedule.runner.lock-at-most-for:5m}", + lockAtLeastFor = "${openframe.rmm.schedule.runner.lock-at-least-for:10s}" + ) + public void run() { + log.debug("Running script schedule sweep"); + try { + scheduleExecutionService.runDueSchedules(); + } catch (Exception e) { + log.error("Script schedule sweep failed", e); + } + } +} diff --git a/openframe-management-service-core/src/main/java/com/openframe/management/service/ScriptScheduleExecutionService.java b/openframe-management-service-core/src/main/java/com/openframe/management/service/ScriptScheduleExecutionService.java new file mode 100644 index 000000000..eea8a704c --- /dev/null +++ b/openframe-management-service-core/src/main/java/com/openframe/management/service/ScriptScheduleExecutionService.java @@ -0,0 +1,222 @@ +package com.openframe.management.service; + +import com.openframe.data.document.rmm.ExecutionStatus; +import com.openframe.data.document.rmm.ScheduleScriptExecution; +import com.openframe.data.document.rmm.Script; +import com.openframe.data.document.rmm.ScriptExecution; +import com.openframe.data.document.rmm.ScriptSchedule; +import com.openframe.data.document.rmm.ScriptScheduleMachineAssigned; +import com.openframe.data.document.rmm.ScriptStatus; +import com.openframe.data.nats.rmm.model.ScriptScheduleExecutionItem; +import com.openframe.data.nats.rmm.model.ScriptScheduleExecutionMessage; +import com.openframe.data.nats.rmm.publisher.ScriptScheduleExecutionNatsPublisher; +import com.openframe.data.repository.rmm.ScheduleScriptExecutionRepository; +import com.openframe.data.repository.rmm.ScriptExecutionRepository; +import com.openframe.data.repository.rmm.ScriptRepository; +import com.openframe.data.repository.rmm.ScriptScheduleMachineAssignedRepository; +import com.openframe.data.repository.rmm.ScriptScheduleRepository; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; + +import java.time.Duration; +import java.time.Instant; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.UUID; +import java.util.function.Function; +import java.util.stream.Collectors; + +/** + * Time-driven runner for RMM {@link ScriptSchedule}s. Fires every schedule whose + * {@code nextRunAt} has come due, fanning each of its scripts out to every + * assigned device over core NATS — the same wire contract as the ad-hoc + * {@code runScheduleJobNow} mutation, but originated by the clock rather than a + * user. + * + *

Lives in the management service (like the {@code ScriptExecutionWatchdog}) + * because that is where the scheduled/ShedLock machinery runs. It talks to the + * repositories and {@link ScriptScheduleExecutionNatsPublisher} directly rather + * than reusing the api-lib dispatch service, which is request-scoped + * (GraphQL/tenant-from-principal) and not on the management classpath. + * + *

Tenancy: the sweep query is tenant-agnostic (mirrors the watchdog); each + * due schedule carries its own {@code tenantId}, which is used verbatim for all + * downstream reads (scripts, assignments) and writes (execution rows) so a run + * stays within the owning tenant. + * + *

Semantics

+ *
    + *
  • One {@code executionId} per schedule run, shared across every script + * and machine of that fire; {@code scheduleId} is stamped on every + * {@link ScriptScheduleExecutionMessage} and {@code scriptId} disambiguates + * each leaf row (correlation key: {@code executionId + machineId + scriptId}).
    • + *
  • Missed runs (runner was down / lock held past several intervals): the + * schedule fires once and {@code nextRunAt} is rolled forward to the + * next slot strictly after "now" — no backfill storm.
    • + *
  • A one-shot schedule ({@code repeat == null}) fires once and then has + * {@code nextRunAt} cleared to null.
    • + *
  • A schedule with no scripts or no assigned devices still advances its + * {@code nextRunAt} (nothing is dispatched) so it does not hot-loop.
    • + *
+ */ +@Service +@RequiredArgsConstructor +@Slf4j +public class ScriptScheduleExecutionService { + + private final ScriptScheduleRepository scheduleRepository; + private final ScriptScheduleMachineAssignedRepository assignedRepository; + private final ScriptRepository scriptRepository; + private final ScriptExecutionRepository scriptExecutionRepository; + private final ScheduleScriptExecutionRepository scheduleScriptExecutionRepository; + private final ScriptScheduleExecutionNatsPublisher scriptScheduleExecutionNatsPublisher; + + /** + * Fire every ACTIVE schedule that is due (nextRunAt ≤ now). Each schedule + * is handled independently: a failure on one is logged and its + * {@code nextRunAt} still advanced, so one broken schedule cannot wedge the + * sweep or hot-loop. + */ + public void runDueSchedules() { + Instant now = Instant.now(); + List due = scheduleRepository.findByStatusAndNextRunAtLessThanEqual(ScriptStatus.ACTIVE, now); + if (due.isEmpty()) { + log.debug("No due script schedules"); + return; + } + + log.info("Found {} due script schedule(s) — running", due.size()); + for (ScriptSchedule schedule : due) { + try { + fire(schedule, now); + } catch (Exception e) { + log.error("Script schedule run failed scheduleId={} tenantId={} — advancing nextRunAt anyway", + schedule.getId(), schedule.getTenantId(), e); + advanceNextRun(schedule, now); + scheduleRepository.save(schedule); + } + } + } + + private void fire(ScriptSchedule schedule, Instant now) { + String tenantId = schedule.getTenantId(); + List scriptIds = schedule.getScriptIds(); + List machineIds = resolveMachineIds(tenantId, schedule.getId()); + + if (scriptIds == null || scriptIds.isEmpty() || machineIds.isEmpty()) { + log.info("Schedule scheduleId={} has no scripts or no assigned devices — nothing dispatched", + schedule.getId()); + } else { + // One executionId for the entire run: shared across all scripts and machines. + String executionId = UUID.randomUUID().toString(); + dispatchScripts(schedule, executionId, tenantId, scriptIds, machineIds, now); + } + + schedule.setLastRunAt(now); + advanceNextRun(schedule, now); + scheduleRepository.save(schedule); + } + + private void dispatchScripts(ScriptSchedule schedule, String executionId, String tenantId, + List scriptIds, List machineIds, Instant now) { + // Resolve all referenced scripts once; only ACTIVE scripts are dispatched. + Map byId = scriptRepository.findByTenantIdAndIdIn(tenantId, scriptIds).stream() + .filter(s -> s.getStatus() == ScriptStatus.ACTIVE) + .collect(Collectors.toMap(Script::getId, Function.identity(), (a, b) -> a)); + + // Preserve the schedule's stored run order; dedup and drop missing/inactive. + List