From 3675db5907b24c918abf821b79604d71a8865331 Mon Sep 17 00:00:00 2001 From: James Brunton Date: Wed, 10 Jun 2026 11:55:29 +0100 Subject: [PATCH] Add new triggers, sources and sinks to policies (#6543) # Description of Changes Add new triggers: - Schedule (fires every X amount of time) - Folder watch (fires whenever the OS tells us a folder has a new file in it; on Mac this is technically a 2s schedule but that's just how Java implements it) Add new sources: - Folder (reads from this directory) Add new sinks: - Inline (stores in FileStorage) - Folder (stores in specified directory) Still want to do S3 buckets and web hooks and stuff, but they can come in a future PR. I'm hoping this should make it sufficient to be able to integrate with processing folders frontend etc. I've also changed it so that policies can have multiple sources and triggers at once, which seems like it might be useful. --- .../common/model/ApplicationProperties.java | 40 +++ .../src/main/resources/settings.yml.template | 12 + .../policy/config/FolderAccessGuard.java | 106 ++++++ .../policy/controller/PolicyController.java | 56 +++- .../policy/engine/PolicyRunRegistry.java | 12 +- .../policy/engine/PolicyRunner.java | 116 +++++++ .../policy/engine/PolicyValidator.java | 74 +++++ .../policy/input/FolderInputSource.java | 186 +++++++++++ .../proprietary/policy/input/InputSource.java | 49 +++ .../policy/input/ResolvedInput.java | 23 ++ .../proprietary/policy/model/InputSpec.java | 26 ++ .../proprietary/policy/model/OutputSpec.java | 5 + .../proprietary/policy/model/Policy.java | 33 +- .../proprietary/policy/model/Schedule.java | 141 ++++++++ .../policy/model/TriggerConfig.java | 20 +- .../policy/output/FolderOutputSink.java | 130 ++++++++ .../policy/output/PolicyOutputSink.java | 6 + .../policy/store/InProcessPolicyStore.java | 2 + .../policy/store/JpaPolicyStore.java | 3 +- .../policy/trigger/FolderWatchTrigger.java | 302 ++++++++++++++++++ .../policy/trigger/ManualTrigger.java | 41 --- .../policy/trigger/PolicyTrigger.java | 32 +- .../policy/trigger/PolicyTriggerManager.java | 56 ++++ .../policy/trigger/ScheduleTrigger.java | 158 +++++++++ .../policy/config/FolderAccessGuardTest.java | 99 ++++++ .../policy/engine/PolicyEngineTest.java | 5 +- .../policy/engine/PolicyRunRegistryTest.java | 3 +- .../policy/engine/PolicyRunnerTest.java | 159 +++++++++ .../policy/engine/PolicyValidatorTest.java | 114 +++++++ .../policy/input/FolderInputSourceTest.java | 137 ++++++++ .../policy/output/FolderOutputSinkTest.java | 105 ++++++ .../store/InProcessPolicyStoreTest.java | 23 +- .../policy/store/JpaPolicyStoreTest.java | 20 +- .../trigger/FolderWatchTriggerTest.java | 178 +++++++++++ .../trigger/PolicyTriggerManagerTest.java | 51 +++ .../policy/trigger/ScheduleTriggerTest.java | 147 +++++++++ 36 files changed, 2556 insertions(+), 114 deletions(-) create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/config/FolderAccessGuard.java create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunner.java create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyValidator.java create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/input/FolderInputSource.java create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/input/InputSource.java create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/input/ResolvedInput.java create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/model/InputSpec.java create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/model/Schedule.java create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/output/FolderOutputSink.java create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/FolderWatchTrigger.java delete mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/ManualTrigger.java create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/PolicyTriggerManager.java create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/ScheduleTrigger.java create mode 100644 app/proprietary/src/test/java/stirling/software/proprietary/policy/config/FolderAccessGuardTest.java create mode 100644 app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunnerTest.java create mode 100644 app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyValidatorTest.java create mode 100644 app/proprietary/src/test/java/stirling/software/proprietary/policy/input/FolderInputSourceTest.java create mode 100644 app/proprietary/src/test/java/stirling/software/proprietary/policy/output/FolderOutputSinkTest.java create mode 100644 app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/FolderWatchTriggerTest.java create mode 100644 app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/PolicyTriggerManagerTest.java create mode 100644 app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/ScheduleTriggerTest.java diff --git a/app/common/src/main/java/stirling/software/common/model/ApplicationProperties.java b/app/common/src/main/java/stirling/software/common/model/ApplicationProperties.java index 00e26ec64..067be54eb 100644 --- a/app/common/src/main/java/stirling/software/common/model/ApplicationProperties.java +++ b/app/common/src/main/java/stirling/software/common/model/ApplicationProperties.java @@ -80,6 +80,7 @@ public class ApplicationProperties { private Mcp mcp = new Mcp(); private InternalApi internalApi = new InternalApi(); private Cluster cluster = new Cluster(); + private Policies policies = new Policies(); @Bean public PropertySource dynamicYamlPropertySource(ConfigurableEnvironment environment) @@ -203,6 +204,45 @@ public class ApplicationProperties { } } + @Data + public static class Policies { + /** + * Absolute directories that policy folder input sources and output sinks may read from or + * write to. Empty (the default) disables folder access entirely, so a policy can never be + * pointed at an arbitrary server path. Stirling's own config directory is always + * off-limits, and folder access is always disabled in SaaS mode regardless of this list. + */ + private List allowedFolderRoots = new java.util.ArrayList<>(); + + /** How often (seconds) the schedule trigger checks for policies whose schedule is due. */ + private long scheduleSweepSeconds = 60; + + /** + * How often (seconds) the folder-watch trigger reconciles its watch registrations and + * re-runs every folder-watch policy as a safety net for filesystem events that were missed + * (NFS, bind mounts, inotify-queue overflow). + */ + private long watchReconcileSeconds = 300; + + /** + * How long (milliseconds) the folder-watch trigger keeps draining filesystem events after + * the first, so a burst from a single file copy coalesces into one run instead of many. + */ + private long watchQuietPeriodMs = 500; + + /** + * SSE emitter timeout (milliseconds) for streamed runs; generous for long multi-step runs. + */ + private long streamTimeoutMs = 1800000; + + /** + * How long (minutes) a finished run's in-memory state is retained before eviction, + * mirroring the job-result expiry so rich run state does not outlive the process. Active + * and paused runs are kept regardless of age. + */ + private int runExpiryMinutes = 30; + } + @Data public static class PdfEditor { private Cache cache = new Cache(); diff --git a/app/core/src/main/resources/settings.yml.template b/app/core/src/main/resources/settings.yml.template index d4ecb35f7..413f39706 100644 --- a/app/core/src/main/resources/settings.yml.template +++ b/app/core/src/main/resources/settings.yml.template @@ -364,6 +364,18 @@ aiEngine: url: http://localhost:5001 # URL of the Python AI engine timeoutSeconds: 120 # Timeout in seconds for AI engine requests +policies: + # Folder automations can read from and write to the directories you allow here, so treat this as a + # security boundary. Leave allowedFolderRoots empty (default) to disable folder sources/outputs + # entirely; list absolute directories to permit folder access only within them. Stirling's own + # config directory is always off-limits, and folder access is always disabled in SaaS mode. + allowedFolderRoots: [] # e.g. ["/data/inbox", "/data/outbox"] + scheduleSweepSeconds: 60 # How often (seconds) scheduled policies are checked for being due + watchReconcileSeconds: 300 # How often (seconds) folder-watch re-syncs watches and re-runs as a safety net for missed events + watchQuietPeriodMs: 500 # How long (ms) folder-watch coalesces a burst of file events into a single run + streamTimeoutMs: 1800000 # SSE timeout (ms) for live run-progress streams + runExpiryMinutes: 30 # How long (minutes) a finished run's in-memory state is kept before eviction + # Model Context Protocol (MCP) server. Exposes Stirling's PDF tools (grouped by namespace) # plus the AI agents to MCP clients (Inspector, Claude Desktop, custom). OAuth-protected. # Disabled by default - enable explicitly per deployment after configuring mcp.auth. diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/config/FolderAccessGuard.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/config/FolderAccessGuard.java new file mode 100644 index 000000000..3abf919f1 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/config/FolderAccessGuard.java @@ -0,0 +1,106 @@ +package stirling.software.proprietary.policy.config; + +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import org.springframework.core.env.Environment; +import org.springframework.stereotype.Component; + +import stirling.software.common.configuration.InstallationPathConfig; +import stirling.software.common.model.ApplicationProperties; +import stirling.software.proprietary.policy.model.Policy; + +/** + * The single authority on which filesystem locations a policy may read from or write to. Folder + * sources and sinks take a configured directory, so without this a user who can save a policy could + * point one at Stirling's own config/secrets directory and exfiltrate (or overwrite) it. Every + * folder source and sink runs its directory through {@link #requirePermitted(Path)} at save time + * and again at run time. + * + *

Enforced fail-closed, in order: + * + *

+ * + *

Paths are compared after normalisation, so {@code ..} segments cannot walk out of an allowed + * root. (Symlink escape is not defended here; an operator who configures an allowed root containing + * a symlink to a sensitive location is trusted.) + */ +@Component +public class FolderAccessGuard { + + public static final String FOLDER_TYPE = "folder"; + + private final boolean saasActive; + private final List allowedRoots; + private final List protectedRoots; + + public FolderAccessGuard(ApplicationProperties applicationProperties, Environment environment) { + this.saasActive = Arrays.asList(environment.getActiveProfiles()).contains("saas"); + this.allowedRoots = + normalizeAll(applicationProperties.getPolicies().getAllowedFolderRoots()); + this.protectedRoots = List.of(normalize(Path.of(InstallationPathConfig.getConfigPath()))); + } + + /** + * Check that {@code dir} is a permitted folder location, returning its normalised absolute + * form. + * + * @throws IllegalArgumentException if folder access is disabled (SaaS or no roots configured), + * the path is inside a protected directory, or it falls outside every allowed root + */ + public Path requirePermitted(Path dir) { + if (saasActive) { + throw new IllegalArgumentException( + "folder sources and outputs are not available in SaaS mode"); + } + Path normalized = normalize(dir); + for (Path protectedRoot : protectedRoots) { + if (normalized.startsWith(protectedRoot)) { + throw new IllegalArgumentException( + "folder may not point inside a protected Stirling directory"); + } + } + if (allowedRoots.isEmpty()) { + throw new IllegalArgumentException( + "folder access is disabled; set policies.allowedFolderRoots to permit it"); + } + boolean within = allowedRoots.stream().anyMatch(normalized::startsWith); + if (!within) { + throw new IllegalArgumentException( + "folder '" + normalized + "' is outside the allowed folder roots"); + } + return normalized; + } + + /** Whether this policy reads from or writes to a folder, and so is subject to these rules. */ + public boolean usesFolderAccess(Policy policy) { + boolean readsFolder = + policy.sources().stream().anyMatch(spec -> FOLDER_TYPE.equals(spec.type())); + boolean writesFolder = + policy.output() != null && FOLDER_TYPE.equals(policy.output().type()); + return readsFolder || writesFolder; + } + + private static List normalizeAll(List roots) { + List result = new ArrayList<>(); + for (String root : roots) { + if (root != null && !root.isBlank()) { + result.add(normalize(Path.of(root))); + } + } + return result; + } + + private static Path normalize(Path path) { + return path.toAbsolutePath().normalize(); + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/controller/PolicyController.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/controller/PolicyController.java index 0a94eef1f..df65fbd99 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/controller/PolicyController.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/controller/PolicyController.java @@ -6,7 +6,6 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; -import org.springframework.beans.factory.annotation.Value; import org.springframework.core.io.FileSystemResource; import org.springframework.core.io.Resource; import org.springframework.http.HttpStatus; @@ -34,11 +33,16 @@ import io.swagger.v3.oas.annotations.tags.Tag; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import stirling.software.common.model.ApplicationProperties; import stirling.software.common.model.job.JobResponse; +import stirling.software.common.service.UserServiceInterface; import stirling.software.common.util.TempFile; import stirling.software.common.util.TempFileManager; +import stirling.software.proprietary.policy.config.FolderAccessGuard; import stirling.software.proprietary.policy.engine.PolicyRunHandle; import stirling.software.proprietary.policy.engine.PolicyRunRegistry; +import stirling.software.proprietary.policy.engine.PolicyRunner; +import stirling.software.proprietary.policy.engine.PolicyValidator; import stirling.software.proprietary.policy.model.PipelineDefinition; import stirling.software.proprietary.policy.model.Policy; import stirling.software.proprietary.policy.model.PolicyInputs; @@ -47,7 +51,6 @@ import stirling.software.proprietary.policy.model.PolicyRunStatus; import stirling.software.proprietary.policy.model.PolicyRunView; import stirling.software.proprietary.policy.progress.PolicyProgressListener; import stirling.software.proprietary.policy.store.PolicyStore; -import stirling.software.proprietary.policy.trigger.ManualTrigger; import stirling.software.proprietary.security.config.PremiumEndpoint; import tools.jackson.core.JacksonException; @@ -71,16 +74,16 @@ import tools.jackson.databind.ObjectMapper; @Tag(name = "Policies", description = "Run tool pipelines on the backend") public class PolicyController { - private final ManualTrigger manualTrigger; + private final PolicyRunner policyRunner; private final PolicyRunRegistry runRegistry; private final PolicyStore policyStore; + private final PolicyValidator policyValidator; + private final FolderAccessGuard folderAccessGuard; + private final UserServiceInterface userService; + private final ApplicationProperties applicationProperties; private final ObjectMapper objectMapper; private final TempFileManager tempFileManager; - /** SSE emitter timeout, generous enough for long multi-step runs on large files. */ - @Value("${stirling.policies.streamTimeoutMs:1800000}") - private long streamTimeoutMs; - @PostMapping(value = "/run", consumes = MediaType.MULTIPART_FORM_DATA_VALUE) @Operation( summary = "Run a tool pipeline", @@ -95,7 +98,8 @@ public class PolicyController { throws IOException { PipelineDefinition definition = parseDefinition(json); PolicyInputs inputs = collectInputs(request); - String runId = manualTrigger.fire(definition, inputs, PolicyProgressListener.NOOP).runId(); + String runId = + policyRunner.runAdHoc(definition, inputs, PolicyProgressListener.NOOP).runId(); return ResponseEntity.accepted().body(new JobResponse<>(true, runId, null)); } @@ -112,10 +116,11 @@ public class PolicyController { PipelineDefinition definition = parseDefinition(json); PolicyInputs inputs = collectInputs(request); - SseEmitter emitter = new SseEmitter(streamTimeoutMs); + SseEmitter emitter = + new SseEmitter(applicationProperties.getPolicies().getStreamTimeoutMs()); emitter.onError(e -> log.warn("Policy run SSE emitter error", e)); - PolicyRunHandle handle = manualTrigger.fire(definition, inputs, streamListener(emitter)); + PolicyRunHandle handle = policyRunner.runAdHoc(definition, inputs, streamListener(emitter)); // Close the stream with a terminal event once the run finishes. whenComplete runs on the // engine's worker thread after the run is done, so this never races the step events. handle.completion() @@ -155,7 +160,34 @@ public class PolicyController { "Stores a policy (trigger config + steps + output + metadata). A blank id is" + " assigned; returns the stored policy with its id.") public ResponseEntity savePolicy(@RequestBody String json) { - return ResponseEntity.ok(policyStore.save(parsePolicy(json))); + Policy policy = parsePolicy(json); + requireAuthorizedForFolderAccess(policy); + try { + policyValidator.validate(policy); + } catch (IllegalArgumentException e) { + throw new ResponseStatusException(HttpStatus.BAD_REQUEST, e.getMessage()); + } + return ResponseEntity.ok(policyStore.save(policy)); + } + + /** + * A policy that reads from or writes to a server folder grants whoever saves it access to that + * path, so restrict it to administrators on multi-user deployments. Single-user deployments + * (login disabled, e.g. desktop) trust the local operator. The {@link FolderAccessGuard} still + * enforces SaaS-off and the path allowlist during validation regardless of who saves. + */ + private void requireAuthorizedForFolderAccess(Policy policy) { + if (!folderAccessGuard.usesFolderAccess(policy)) { + return; + } + if (!applicationProperties.getSecurity().isEnableLogin()) { + return; + } + if (!userService.isCurrentUserAdmin()) { + throw new ResponseStatusException( + HttpStatus.FORBIDDEN, + "Folder sources and outputs may only be configured by an administrator"); + } } @GetMapping @@ -199,7 +231,7 @@ public class PolicyController { new ResponseStatusException( HttpStatus.NOT_FOUND, "No policy: " + policyId)); PolicyInputs inputs = collectInputs(request); - String runId = manualTrigger.run(policy, inputs, PolicyProgressListener.NOOP).runId(); + String runId = policyRunner.runWith(policy, inputs, PolicyProgressListener.NOOP).runId(); return ResponseEntity.accepted().body(new JobResponse<>(true, runId, null)); } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunRegistry.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunRegistry.java index 0b613726d..acf5a3168 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunRegistry.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunRegistry.java @@ -9,13 +9,13 @@ import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import jakarta.annotation.PreDestroy; import lombok.extern.slf4j.Slf4j; +import stirling.software.common.model.ApplicationProperties; import stirling.software.proprietary.policy.model.PolicyRun; /** @@ -24,9 +24,9 @@ import stirling.software.proprietary.policy.model.PolicyRun; * TaskManager}. * *

Finished runs are evicted on a fixed interval once they age past {@code - * stirling.policies.runExpiryMinutes}, mirroring the job-result expiry in {@code TaskManager} so a - * run's rich in-memory state does not outlive the process. Only terminal runs are evicted; active - * and paused ({@code WAITING_FOR_INPUT}) runs are retained regardless of age. Result files are not + * policies.runExpiryMinutes}, mirroring the job-result expiry in {@code TaskManager} so a run's + * rich in-memory state does not outlive the process. Only terminal runs are evicted; active and + * paused ({@code WAITING_FOR_INPUT}) runs are retained regardless of age. Result files are not * touched here: a run shares its runId with a {@code TaskManager} job, which owns file-lifecycle * cleanup, so eviction only frees this map's entry. */ @@ -41,8 +41,8 @@ public class PolicyRunRegistry { Executors.newSingleThreadScheduledExecutor( Thread.ofVirtual().name("policy-run-cleanup-", 0).factory()); - public PolicyRunRegistry( - @Value("${stirling.policies.runExpiryMinutes:30}") int runExpiryMinutes) { + public PolicyRunRegistry(ApplicationProperties applicationProperties) { + int runExpiryMinutes = applicationProperties.getPolicies().getRunExpiryMinutes(); this.runExpiry = Duration.ofMinutes(runExpiryMinutes); cleanupExecutor.scheduleAtFixedRate(this::evictExpiredRuns, 10, 10, TimeUnit.MINUTES); log.debug( diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunner.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunner.java new file mode 100644 index 000000000..8685bf020 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunner.java @@ -0,0 +1,116 @@ +package stirling.software.proprietary.policy.engine; + +import java.io.IOException; +import java.util.List; +import java.util.function.Consumer; + +import org.springframework.stereotype.Service; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +import stirling.software.proprietary.policy.input.InputSource; +import stirling.software.proprietary.policy.input.ResolvedInput; +import stirling.software.proprietary.policy.model.InputSpec; +import stirling.software.proprietary.policy.model.PipelineDefinition; +import stirling.software.proprietary.policy.model.Policy; +import stirling.software.proprietary.policy.model.PolicyInputs; +import stirling.software.proprietary.policy.model.PolicyRun; +import stirling.software.proprietary.policy.model.PolicyRunStatus; +import stirling.software.proprietary.policy.progress.PolicyProgressListener; + +/** + * Runs policies, and is the one place that knows how to turn a policy's configured {@link InputSpec + * sources} into actual runs. Triggers (schedule, and future webhook/folder-watch) decide + * when to run and call {@link #run(Policy)}; they never touch sources themselves. The + * controller uses the supplied-input and ad-hoc entry points for on-demand work. + * + *

This is the seam that keeps triggers and sources independent: a trigger depends on the runner, + * the runner depends on the {@link InputSource} beans, and a source depends on neither - it just + * yields {@link ResolvedInput units of work}, each carrying its own completion hook. + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class PolicyRunner { + + private final PolicyEngine policyEngine; + private final List inputSources; + + /** + * Run a policy by pulling from every source it configures: each source yields zero or more + * units of work, and each unit becomes its own run so one failure does not affect the others. A + * policy with no sources runs once with no input files (a generator pipeline). Used by + * automatic triggers. + */ + public void run(Policy policy) { + List sources = policy.sources(); + if (sources.isEmpty()) { + startRun(policy, PolicyInputs.of(List.of()), unused -> {}); + return; + } + for (InputSpec spec : sources) { + pullAndRun(policy, spec); + } + } + + /** + * Run a stored policy on files supplied directly by the caller (e.g. a manual run with + * uploads), bypassing its configured sources. Returns the run handle so callers can stream + * progress. + */ + public PolicyRunHandle runWith( + Policy policy, PolicyInputs inputs, PolicyProgressListener listener) { + return policyEngine.runPolicy(policy, inputs, listener); + } + + /** Run an ad-hoc pipeline with no stored policy (AI/Automate one-offs). */ + public PolicyRunHandle runAdHoc( + PipelineDefinition definition, PolicyInputs inputs, PolicyProgressListener listener) { + return policyEngine.submit(definition, inputs, listener); + } + + private void pullAndRun(Policy policy, InputSpec spec) { + InputSource source = sourceFor(spec); + if (source == null) { + log.warn( + "No input source for type '{}' (policy {}); skipping", + spec.type(), + policy.id()); + return; + } + List work; + try { + work = source.resolve(spec); + } catch (IOException | RuntimeException e) { + log.warn( + "Failed to resolve source '{}' for policy {}: {}", + spec.type(), + policy.id(), + e.getMessage()); + return; + } + for (ResolvedInput unit : work) { + startRun(policy, unit.inputs(), unit.onComplete()); + } + } + + private void startRun(Policy policy, PolicyInputs inputs, Consumer onComplete) { + log.info("Running policy {} ({})", policy.id(), policy.name()); + PolicyRunHandle handle = + policyEngine.runPolicy(policy, inputs, PolicyProgressListener.NOOP); + handle.completion() + .whenComplete((run, throwable) -> onComplete.accept(succeeded(run, throwable))); + } + + private static boolean succeeded(PolicyRun run, Throwable throwable) { + return throwable == null && run != null && run.getStatus() == PolicyRunStatus.COMPLETED; + } + + private InputSource sourceFor(InputSpec spec) { + return inputSources.stream() + .filter(source -> source.supports(spec)) + .findFirst() + .orElse(null); + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyValidator.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyValidator.java new file mode 100644 index 000000000..2aa4d98be --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyValidator.java @@ -0,0 +1,74 @@ +package stirling.software.proprietary.policy.engine; + +import java.util.List; + +import org.springframework.stereotype.Service; + +import lombok.RequiredArgsConstructor; + +import stirling.software.proprietary.policy.input.InputSource; +import stirling.software.proprietary.policy.model.InputSpec; +import stirling.software.proprietary.policy.model.OutputSpec; +import stirling.software.proprietary.policy.model.Policy; +import stirling.software.proprietary.policy.model.TriggerConfig; +import stirling.software.proprietary.policy.output.PolicyOutputSink; +import stirling.software.proprietary.policy.trigger.PolicyTrigger; + +/** + * Validates a policy's trigger, sources, and output configuration by delegating each facet to the + * bean that handles its type. Called when a policy is saved so a misconfigured schedule, missing + * folder directory, or unknown type fails fast instead of silently misbehaving at run time. + * + *

The trigger is optional (a {@code null} trigger is a manual-only policy and needs no + * validation); every configured source is validated. + */ +@Service +@RequiredArgsConstructor +public class PolicyValidator { + + private final List triggers; + private final List inputSources; + private final List outputSinks; + + /** + * @throws IllegalArgumentException if any facet's type is unknown or its configuration is + * invalid + */ + public void validate(Policy policy) { + if (policy.trigger() != null) { + triggerFor(policy.trigger()).validate(policy); + } + for (InputSpec source : policy.sources()) { + inputSourceFor(source).validate(source); + } + outputSinkFor(policy.output()).validate(policy.output()); + } + + private PolicyTrigger triggerFor(TriggerConfig config) { + return triggers.stream() + .filter(trigger -> trigger.type().equals(config.type())) + .findFirst() + .orElseThrow( + () -> + new IllegalArgumentException( + "unknown trigger type: " + config.type())); + } + + private InputSource inputSourceFor(InputSpec spec) { + return inputSources.stream() + .filter(source -> source.supports(spec)) + .findFirst() + .orElseThrow( + () -> + new IllegalArgumentException( + "unknown input source type: " + spec.type())); + } + + private PolicyOutputSink outputSinkFor(OutputSpec spec) { + return outputSinks.stream() + .filter(sink -> sink.supports(spec)) + .findFirst() + .orElseThrow( + () -> new IllegalArgumentException("unknown output type: " + spec.type())); + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/FolderInputSource.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/FolderInputSource.java new file mode 100644 index 000000000..d67820275 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/FolderInputSource.java @@ -0,0 +1,186 @@ +package stirling.software.proprietary.policy.input; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardCopyOption; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.stream.Stream; + +import org.springframework.core.io.FileSystemResource; +import org.springframework.core.io.Resource; +import org.springframework.stereotype.Service; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +import stirling.software.common.util.FileReadinessChecker; +import stirling.software.proprietary.policy.config.FolderAccessGuard; +import stirling.software.proprietary.policy.model.InputSpec; +import stirling.software.proprietary.policy.model.PolicyInputs; + +/** + * Reads input files from a directory. Each ready file becomes its own unit of work (one run per + * file) so a failure on one file does not affect the others. + * + *

Two modes via the {@code mode} option: + * + *

+ * + * Readiness is checked first (via {@link FileReadinessChecker}) so files mid-write are skipped. + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class FolderInputSource implements InputSource { + + private static final String TYPE = FolderAccessGuard.FOLDER_TYPE; + // Bookkeeping lives under one hidden namespace dir so the watched folder stays tidy. + private static final String WORK_SUBDIR = ".stirling"; + private static final String PROCESSING_SUBDIR = "processing"; + private static final String DONE_SUBDIR = "done"; + private static final String ERROR_SUBDIR = "error"; + + private final FileReadinessChecker readinessChecker; + private final FolderAccessGuard accessGuard; + + @Override + public String type() { + return TYPE; + } + + @Override + public boolean supports(InputSpec spec) { + return spec != null && TYPE.equals(spec.type()); + } + + @Override + public void validate(InputSpec spec) { + accessGuard.requirePermitted(FolderConfig.from(spec.options()).directory()); + } + + @Override + public List watchTargets(InputSpec spec) { + return List.of(FolderConfig.from(spec.options()).directory()); + } + + @Override + public List resolve(InputSpec spec) throws IOException { + FolderConfig config = FolderConfig.from(spec.options()); + Path inputDir = accessGuard.requirePermitted(config.directory()); + if (!Files.isDirectory(inputDir)) { + log.debug("Folder input dir does not exist: {}", inputDir); + return List.of(); + } + + List ready = new ArrayList<>(); + try (Stream entries = Files.list(inputDir)) { + entries.filter(Files::isRegularFile) + .filter(readinessChecker::isReady) + .forEach(ready::add); + } + + List work = new ArrayList<>(); + for (Path file : ready) { + if (config.snapshot()) { + work.add(ResolvedInput.of(PolicyInputs.of(List.of(fileResource(file))))); + } else { + Path claimed = claim(inputDir, file); + if (claimed == null) { + continue; // another sweep/process grabbed it + } + work.add( + new ResolvedInput( + PolicyInputs.of(List.of(fileResource(claimed))), + success -> route(inputDir, claimed, success))); + } + } + return work; + } + + private Path claim(Path inputDir, Path file) { + try { + Path processingDir = workDir(inputDir, PROCESSING_SUBDIR); + Files.createDirectories(processingDir); + Path claimed = uniqueTarget(processingDir, file.getFileName().toString()); + Files.move(file, claimed, StandardCopyOption.ATOMIC_MOVE); + return claimed; + } catch (IOException e) { + log.debug("Could not claim {}: {}", file, e.getMessage()); + return null; + } + } + + private void route(Path inputDir, Path claimed, boolean success) { + String subdir = success ? DONE_SUBDIR : ERROR_SUBDIR; + try { + Path destDir = workDir(inputDir, subdir); + Files.createDirectories(destDir); + Files.move( + claimed, + uniqueTarget(destDir, claimed.getFileName().toString()), + StandardCopyOption.ATOMIC_MOVE); + } catch (IOException e) { + log.warn( + "Could not move processed input {} to {}: {}", claimed, subdir, e.getMessage()); + } + } + + /** A bookkeeping subdirectory under the watched folder's {@code .stirling} namespace. */ + private static Path workDir(Path inputDir, String subdir) { + return inputDir.resolve(WORK_SUBDIR).resolve(subdir); + } + + private static Resource fileResource(Path path) { + String name = path.getFileName().toString(); + return new FileSystemResource(path.toFile()) { + @Override + public String getFilename() { + return name; + } + }; + } + + private static Path uniqueTarget(Path dir, String filename) { + Path candidate = dir.resolve(filename); + if (!Files.exists(candidate)) { + return candidate; + } + int dot = filename.lastIndexOf('.'); + String base = dot < 0 ? filename : filename.substring(0, dot); + String ext = dot < 0 ? "" : filename.substring(dot); + for (int n = 1; ; n++) { + Path next = dir.resolve(base + " (" + n + ")" + ext); + if (!Files.exists(next)) { + return next; + } + } + } + + /** The typed, validated form of a folder source's options: the directory and dedup mode. */ + record FolderConfig(Path directory, boolean snapshot) { + + private static final String DIRECTORY_OPTION = "directory"; + private static final String MODE_OPTION = "mode"; + private static final String MODE_SNAPSHOT = "snapshot"; + + static FolderConfig from(Map options) { + Object directory = options.get(DIRECTORY_OPTION); + if (directory == null || directory.toString().isBlank()) { + throw new IllegalArgumentException("folder input requires a 'directory' option"); + } + Object mode = options.get(MODE_OPTION); + boolean snapshot = mode != null && MODE_SNAPSHOT.equals(mode.toString()); + return new FolderConfig(Path.of(directory.toString()), snapshot); + } + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/InputSource.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/InputSource.java new file mode 100644 index 000000000..a6f116d02 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/InputSource.java @@ -0,0 +1,49 @@ +package stirling.software.proprietary.policy.input; + +import java.io.IOException; +import java.nio.file.Path; +import java.util.List; + +import stirling.software.proprietary.policy.model.InputSpec; + +/** + * Resolves one of a policy's {@link InputSpec sources} into the files to run on - answering + * where a run's files come from, independent of when it runs. The counterpart of + * {@code PolicyOutputSink}: implementations are beans selected by {@link #supports(InputSpec)}, so + * a new source kind (folder, S3) is just a new bean. + * + *

Driven by the {@code PolicyRunner}, which a trigger calls when a policy is due; a source is + * passive and knows nothing about what triggered the run. A manual run may instead supply files + * directly and bypass sources entirely. + */ +public interface InputSource { + + /** Stable identifier for this source, matching {@code InputSpec.type()} (e.g. "folder"). */ + String type(); + + /** Whether this source can handle the given spec. */ + boolean supports(InputSpec spec); + + /** + * Check that an input spec is usable, throwing {@link IllegalArgumentException} if not. Called + * when a policy is saved so misconfiguration fails fast rather than at run time. + */ + default void validate(InputSpec spec) {} + + /** + * Resolve the spec into zero or more units of work, each carrying the files for one run and a + * completion hook. Returning an empty list means there is nothing to run right now. + */ + List resolve(InputSpec spec) throws IOException; + + /** + * The local filesystem directories this source draws from, if any, for triggers that want to + * react to changes there (the folder-watch trigger) rather than poll. Advisory only: it merely + * tells a trigger where to watch; resolving the spec into files is still this source's + * job via {@link #resolve}. Non-filesystem sources (S3, ...) return an empty list and so are + * simply not watchable. Default: nothing to watch. + */ + default List watchTargets(InputSpec spec) { + return List.of(); + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/ResolvedInput.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/ResolvedInput.java new file mode 100644 index 000000000..285413436 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/ResolvedInput.java @@ -0,0 +1,23 @@ +package stirling.software.proprietary.policy.input; + +import java.util.function.Consumer; + +import stirling.software.proprietary.policy.model.PolicyInputs; + +/** + * One unit of work produced by an {@link InputSource}: the files to run plus a completion callback + * invoked with the run's success once it finishes (e.g. a folder source routes the input to {@code + * .stirling/done} or {@code .stirling/error}). A source may return several of these (e.g. one per + * file). + */ +public record ResolvedInput(PolicyInputs inputs, Consumer onComplete) { + + public ResolvedInput { + onComplete = onComplete == null ? success -> {} : onComplete; + } + + /** A unit of work with no completion side effect. */ + public static ResolvedInput of(PolicyInputs inputs) { + return new ResolvedInput(inputs, success -> {}); + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/InputSpec.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/InputSpec.java new file mode 100644 index 000000000..d6816c60f --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/InputSpec.java @@ -0,0 +1,26 @@ +package stirling.software.proprietary.policy.model; + +import java.util.Map; + +/** + * One source a policy's input files come from. {@code type} selects an {@code InputSource} + * ("folder", and in future "s3", ...); {@code options} carries source-specific configuration (a + * directory, a bucket, a dedup mode, ...). + * + *

A policy holds a list of these; a run pulls from every one. An empty list means the policy + * runs with no input files (a generator pipeline, or files supplied directly to a manual run). + * + *

Data-driven and parallel to {@link OutputSpec}/{@link TriggerConfig}: a new source kind is a + * new {@code type} handled by a new {@code InputSource} bean. + */ +public record InputSpec(String type, Map options) { + + public InputSpec { + options = options == null ? Map.of() : options; + } + + /** Read input files from a directory on disk. */ + public static InputSpec folder(String directory) { + return new InputSpec("folder", Map.of("directory", directory)); + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/OutputSpec.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/OutputSpec.java index 4759810bc..3caab8afc 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/OutputSpec.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/OutputSpec.java @@ -18,4 +18,9 @@ public record OutputSpec(String type, Map options) { public static OutputSpec inline() { return new OutputSpec("inline", Map.of()); } + + /** Write outputs to a directory on disk. */ + public static OutputSpec folder(String directory) { + return new OutputSpec("folder", Map.of("directory", directory)); + } } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/Policy.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/Policy.java index 3a2acaf0c..14fd35b63 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/Policy.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/Policy.java @@ -3,18 +3,16 @@ package stirling.software.proprietary.policy.model; import java.util.List; /** - * A stored, owned automation: how it is triggered, the ordered tool steps to run, and where its - * output goes, plus identity and metadata. + * A stored automation: an ordered chain of tool steps, the sources its input files come from, and + * an output destination for the results. * - *

This is the central object of the feature. Everything that runs a chain of tools is a use of a - * Policy: a watched folder is a Policy with a folder {@link TriggerConfig} and a folder {@link - * OutputSpec}; a scheduled job is a Policy with a schedule trigger; manual/Automate/AI runs execute - * a Policy (or an ad-hoc {@link PipelineDefinition}) on demand. The engine itself only ever - * executes the {@link PipelineDefinition} this exposes via {@link #toDefinition()} - it is - * trigger-agnostic. + *

Every policy can always be run on demand (manually). It may additionally carry one automatic + * {@link TriggerConfig} - usually a schedule - that fires it without a person asking; a {@code + * null} trigger means manual-only. A trigger decides when a run happens and a {@link + * InputSpec source} decides where its files come from; the two are independent, and a run + * pulls from every configured source. * - *

{@code enabled} gates automatic triggering (a disabled policy is not picked up by its - * trigger); it does not block an explicit manual run. + *

This is the feature's central configuration object - what a user defines and the engine runs. */ public record Policy( String id, @@ -22,15 +20,28 @@ public record Policy( String owner, boolean enabled, TriggerConfig trigger, + List sources, List steps, OutputSpec output) { public Policy { - trigger = trigger == null ? TriggerConfig.manual() : trigger; + sources = sources == null ? List.of() : List.copyOf(sources); steps = steps == null ? List.of() : steps; output = output == null ? OutputSpec.inline() : output; } + /** A policy with no configured sources (a generator, or files supplied directly to a run). */ + public Policy( + String id, + String name, + String owner, + boolean enabled, + TriggerConfig trigger, + List steps, + OutputSpec output) { + this(id, name, owner, enabled, trigger, List.of(), steps, output); + } + /** The engine-level, trigger-agnostic view of this policy's pipeline. */ public PipelineDefinition toDefinition() { return new PipelineDefinition(name, steps, output); diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/Schedule.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/Schedule.java new file mode 100644 index 000000000..148534636 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/Schedule.java @@ -0,0 +1,141 @@ +package stirling.software.proprietary.policy.model; + +import java.time.DayOfWeek; +import java.time.LocalTime; +import java.time.ZonedDateTime; +import java.util.EnumSet; +import java.util.Set; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonSubTypes; +import com.fasterxml.jackson.annotation.JsonTypeInfo; + +/** + * When a scheduled policy should fire, expressed as intent ("every day at 02:00") rather than a + * cron string. A frontend builds one of these from a friendly picker; the schedule trigger asks it + * for the next firing after a given moment. + * + *

Sealed so every cadence is an explicit, self-validating shape and adding a new one forces a + * new subtype rather than another string convention to learn. The {@code type} discriminator stays + * on the wire so the frontend can switch on it without knowing the Java hierarchy. + * + *

Wall-clock kinds ({@link Daily}, {@link Weekly}, {@link Monthly}) are evaluated in the zone of + * the {@code after} argument; {@link Every} is a fixed offset and ignores wall-clock time. + */ +@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type") +@JsonSubTypes({ + @JsonSubTypes.Type(value = Schedule.Every.class, name = "every"), + @JsonSubTypes.Type(value = Schedule.Daily.class, name = "daily"), + @JsonSubTypes.Type(value = Schedule.Weekly.class, name = "weekly"), + @JsonSubTypes.Type(value = Schedule.Monthly.class, name = "monthly"), +}) +@JsonIgnoreProperties(ignoreUnknown = true) +public sealed interface Schedule { + + /** The next firing strictly after {@code after}, evaluated in {@code after}'s zone. */ + ZonedDateTime nextAfter(ZonedDateTime after); + + /** The granularities a fixed-interval schedule can repeat on. */ + enum Unit { + MINUTES, + HOURS, + DAYS + } + + /** + * A fixed cadence repeating from when the policy was last seen: "every 15 minutes", "every 6 + * hours". Time of day is irrelevant. + */ + record Every(long count, Unit unit) implements Schedule { + public Every { + if (count <= 0) { + throw new IllegalArgumentException("'every' schedule needs a positive count"); + } + if (unit == null) { + throw new IllegalArgumentException("'every' schedule needs a unit"); + } + } + + @Override + public ZonedDateTime nextAfter(ZonedDateTime after) { + return switch (unit) { + case MINUTES -> after.plusMinutes(count); + case HOURS -> after.plusHours(count); + case DAYS -> after.plusDays(count); + }; + } + } + + /** Once a day at a wall-clock time: "every day at 02:00". */ + record Daily(LocalTime at) implements Schedule { + public Daily { + requireTime(at); + } + + @Override + public ZonedDateTime nextAfter(ZonedDateTime after) { + ZonedDateTime today = after.with(at); + return today.isAfter(after) ? today : today.plusDays(1); + } + } + + /** On chosen weekdays at a wall-clock time: "every Monday and Thursday at 09:00". */ + record Weekly(Set days, LocalTime at) implements Schedule { + public Weekly { + if (days == null || days.isEmpty()) { + throw new IllegalArgumentException("'weekly' schedule needs at least one day"); + } + requireTime(at); + days = EnumSet.copyOf(days); + } + + @Override + public ZonedDateTime nextAfter(ZonedDateTime after) { + // The soonest of the next 7 days that lands on a chosen weekday, at the configured + // time. + for (int i = 0; i <= 7; i++) { + ZonedDateTime candidate = after.plusDays(i).with(at); + if (candidate.isAfter(after) && days.contains(candidate.getDayOfWeek())) { + return candidate; + } + } + throw new IllegalStateException("unreachable: a chosen weekday recurs within 8 days"); + } + } + + /** + * On a day of the month at a wall-clock time: "the 1st at 00:00". Months too short for the + * chosen day (e.g. the 31st in February) are skipped, not clamped. + */ + record Monthly(int dayOfMonth, LocalTime at) implements Schedule { + public Monthly { + if (dayOfMonth < 1 || dayOfMonth > 31) { + throw new IllegalArgumentException("'monthly' day-of-month must be 1-31"); + } + requireTime(at); + } + + @Override + public ZonedDateTime nextAfter(ZonedDateTime after) { + ZonedDateTime firstOfMonth = after.withDayOfMonth(1).with(at); + // Scan forward a few years' worth of months to skip ones without the chosen day. + for (int i = 0; i < 48; i++) { + ZonedDateTime month = firstOfMonth.plusMonths(i); + if (month.toLocalDate().lengthOfMonth() >= dayOfMonth) { + ZonedDateTime fire = month.withDayOfMonth(dayOfMonth); + if (fire.isAfter(after)) { + return fire; + } + } + } + throw new IllegalStateException( + "unreachable: a month with the chosen day recurs yearly"); + } + } + + private static void requireTime(LocalTime at) { + if (at == null) { + throw new IllegalArgumentException("schedule needs a time of day ('at')"); + } + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/TriggerConfig.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/TriggerConfig.java index 848c48e4f..e18347dc3 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/TriggerConfig.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/TriggerConfig.java @@ -3,23 +3,21 @@ package stirling.software.proprietary.policy.model; import java.util.Map; /** - * How a {@link Policy} is automatically triggered. {@code type} selects a trigger kind ("manual", - * "folder", "schedule", "s3"); {@code options} carries type-specific configuration (a folder path, - * a cron expression, a bucket, ...). + * A {@link Policy}'s optional automatic trigger - what fires it without a person asking. {@code + * type} selects a trigger kind ("schedule", and in future "webhook", "folder-watch", ...); {@code + * options} carries type-specific configuration (a {@link Schedule}, a webhook secret, ...). + * + *

Manual running is not a trigger kind: every policy can always be run on demand, so a + * policy with no automatic trigger simply has a {@code null} {@code TriggerConfig}. A trigger + * answers only "when"; where a run's files come from is a separate concern owned by the policy's + * {@link InputSpec sources}. * *

Data-driven and parallel to {@link OutputSpec}: new trigger kinds are new {@code type} values - * handled by a new trigger bean, with no change to the model. {@code "manual"} means there is no - * automatic trigger - the policy is only ever run on demand. + * handled by a new trigger bean, with no change to the model. */ public record TriggerConfig(String type, Map options) { public TriggerConfig { - type = type == null || type.isBlank() ? "manual" : type; options = options == null ? Map.of() : options; } - - /** No automatic trigger; the policy is run on demand only. */ - public static TriggerConfig manual() { - return new TriggerConfig("manual", Map.of()); - } } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/FolderOutputSink.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/FolderOutputSink.java new file mode 100644 index 000000000..6ecec3c93 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/FolderOutputSink.java @@ -0,0 +1,130 @@ +package stirling.software.proprietary.policy.output; + +import java.io.IOException; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; + +import org.apache.commons.io.FilenameUtils; +import org.springframework.core.io.Resource; +import org.springframework.http.MediaType; +import org.springframework.http.MediaTypeFactory; +import org.springframework.stereotype.Service; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +import stirling.software.common.model.job.ResultFile; +import stirling.software.proprietary.policy.config.FolderAccessGuard; +import stirling.software.proprietary.policy.model.OutputSpec; + +/** + * Writes a run's output files to a directory on disk. The destination is the {@code directory} + * option of the {@link OutputSpec}. + * + *

Files are streamed to disk (so large outputs are not buffered) and given unique names to avoid + * clobbering existing files. The returned {@link ResultFile}s describe what was written (path + + * size); they carry a synthetic id because the deliverable is the file on disk, not a {@code + * FileStorage} entry, so folder outputs are not downloadable via {@code /files/{id}}. + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class FolderOutputSink implements PolicyOutputSink { + + static final String TYPE = FolderAccessGuard.FOLDER_TYPE; + static final String DIRECTORY_OPTION = "directory"; + + private final FolderAccessGuard accessGuard; + + @Override + public String type() { + return TYPE; + } + + @Override + public boolean supports(OutputSpec spec) { + return spec != null && TYPE.equals(spec.type()); + } + + @Override + public void validate(OutputSpec spec) { + accessGuard.requirePermitted(directoryOf(spec)); + } + + @Override + public List deliver(String runId, List outputs, OutputSpec spec) + throws IOException { + Path targetDir = accessGuard.requirePermitted(directoryOf(spec)); + Files.createDirectories(targetDir); + + List results = new ArrayList<>(); + for (int i = 0; i < outputs.size(); i++) { + Resource resource = outputs.get(i); + String name = safeName(resource.getFilename(), i); + Path target = uniqueTarget(targetDir, name); + try (InputStream is = resource.getInputStream()) { + Files.copy(is, target); + } + long size = Files.size(target); + String contentType = + MediaTypeFactory.getMediaType(name) + .orElse(MediaType.APPLICATION_OCTET_STREAM) + .toString(); + results.add( + ResultFile.builder() + .fileId(UUID.randomUUID().toString()) + .fileName(target.toString()) + .contentType(contentType) + .fileSize(size) + .build()); + log.debug("Wrote policy run {} output to {}", runId, target); + } + return results; + } + + private static Path directoryOf(OutputSpec spec) { + Object directory = spec.options().get(DIRECTORY_OPTION); + if (directory == null || directory.toString().isBlank()) { + throw new IllegalArgumentException( + "folder output requires a '" + DIRECTORY_OPTION + "' option"); + } + return Path.of(directory.toString()); + } + + /** + * The resource's filename reduced to a bare, traversal-free name: any directory component or + * "../" is stripped so a crafted output name cannot escape {@code targetDir}. Falls back to a + * synthetic name when the filename is absent or reduces to nothing usable. + */ + private static String safeName(String filename, int index) { + if (filename == null || filename.isBlank()) { + return "output-" + index; + } + String name = FilenameUtils.getName(filename); + if (name.isBlank() || ".".equals(name) || "..".equals(name)) { + return "output-" + index; + } + return name; + } + + /** Resolve a non-colliding path in {@code dir}, appending " (n)" before the extension. */ + private static Path uniqueTarget(Path dir, String filename) { + Path candidate = dir.resolve(filename); + if (!Files.exists(candidate)) { + return candidate; + } + String base = FilenameUtils.getBaseName(filename); + String ext = FilenameUtils.getExtension(filename); + String suffix = ext.isEmpty() ? "" : "." + ext; + for (int n = 1; ; n++) { + Path next = dir.resolve(base + " (" + n + ")" + suffix); + if (!Files.exists(next)) { + return next; + } + } + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/PolicyOutputSink.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/PolicyOutputSink.java index f6bf478e7..c98b55ad6 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/PolicyOutputSink.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/PolicyOutputSink.java @@ -23,6 +23,12 @@ public interface PolicyOutputSink { /** Whether this sink can handle the given output spec. */ boolean supports(OutputSpec spec); + /** + * Check that an output spec is usable, throwing {@link IllegalArgumentException} if not. Called + * when a policy is saved so misconfiguration fails fast rather than at run time. + */ + default void validate(OutputSpec spec) {} + /** * Persist/deliver the output files and return their descriptors. * diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/InProcessPolicyStore.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/InProcessPolicyStore.java index df8fc9339..2ec94d1eb 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/InProcessPolicyStore.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/InProcessPolicyStore.java @@ -30,6 +30,7 @@ public class InProcessPolicyStore implements PolicyStore { policy.owner(), policy.enabled(), policy.trigger(), + policy.sources(), policy.steps(), policy.output()); policies.put(id, stored); @@ -50,6 +51,7 @@ public class InProcessPolicyStore implements PolicyStore { public List findByTriggerType(String triggerType) { return policies.values().stream() .filter(Policy::enabled) + .filter(policy -> policy.trigger() != null) .filter(policy -> triggerType.equals(policy.trigger().type())) .toList(); } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/JpaPolicyStore.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/JpaPolicyStore.java index 977e8c9c1..e4e05ea4c 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/JpaPolicyStore.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/JpaPolicyStore.java @@ -37,6 +37,7 @@ public class JpaPolicyStore implements PolicyStore { policy.owner(), policy.enabled(), policy.trigger(), + policy.sources(), policy.steps(), policy.output()); @@ -45,7 +46,7 @@ public class JpaPolicyStore implements PolicyStore { entity.setName(stored.name()); entity.setOwner(stored.owner()); entity.setEnabled(stored.enabled()); - entity.setTriggerType(stored.trigger().type()); + entity.setTriggerType(stored.trigger() == null ? null : stored.trigger().type()); entity.setPolicyJson(objectMapper.writeValueAsString(stored)); repository.save(entity); return stored; diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/FolderWatchTrigger.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/FolderWatchTrigger.java new file mode 100644 index 000000000..dbff801b3 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/FolderWatchTrigger.java @@ -0,0 +1,302 @@ +package stirling.software.proprietary.policy.trigger; + +import static java.nio.file.StandardWatchEventKinds.ENTRY_CREATE; +import static java.nio.file.StandardWatchEventKinds.ENTRY_MODIFY; + +import java.io.IOException; +import java.nio.file.ClosedWatchServiceException; +import java.nio.file.FileSystems; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.WatchKey; +import java.nio.file.WatchService; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +import org.springframework.stereotype.Service; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +import stirling.software.common.model.ApplicationProperties; +import stirling.software.proprietary.policy.engine.PolicyRunner; +import stirling.software.proprietary.policy.input.InputSource; +import stirling.software.proprietary.policy.model.InputSpec; +import stirling.software.proprietary.policy.model.Policy; +import stirling.software.proprietary.policy.store.PolicyStore; + +/** + * Fires policies the moment a file lands in one of their folder sources, instead of polling on a + * timer. The trigger only reads that location; turning it into files is still the source's job. + * + *

The watcher is a latency optimisation, not a source of truth, so this pairs an event watch + * with a low-frequency reconcile sweep ({@code watchReconcileSeconds}). The reconcile both + * (a) re-syncs which directories are watched as policies are created/edited/deleted and folders + * appear on disk, and (b) runs every folder-watch policy once, catching files that pre-dated the + * watch, events lost to inotify-queue overflow, and changes on filesystems that do not deliver + * events at all (NFS, many container bind mounts). Both paths just call {@link PolicyRunner#run}; + * the {@link InputSource} does the claiming, so a redundant run finds nothing to claim and is + * harmless. + * + *

Like {@link ScheduleTrigger}, watch state is per-node and in memory; this assumes a single + * node and rebuilds its registrations on restart from the {@link PolicyStore}. + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class FolderWatchTrigger implements PolicyTrigger { + + private static final String TYPE = "folder-watch"; + + private final PolicyStore policyStore; + private final PolicyRunner policyRunner; + private final List inputSources; + private final ApplicationProperties applicationProperties; + + private final Map keysByDir = new ConcurrentHashMap<>(); + private final Map dirByKey = new ConcurrentHashMap<>(); + + private volatile boolean running; + + // Package-visible (not private) so tests can drive syncRegistrations() against a real service. + volatile WatchService watchService; + + private volatile ScheduledExecutorService reconciler; + + @Override + public String type() { + return TYPE; + } + + @Override + public void validate(Policy policy) { + if (watchDirsOf(policy).isEmpty()) { + throw new IllegalArgumentException( + "folder-watch trigger requires at least one watchable (folder) input source"); + } + } + + @Override + public synchronized void start() { + if (watchService != null) { + return; + } + try { + watchService = FileSystems.getDefault().newWatchService(); + } catch (IOException e) { + log.error("Could not start folder-watch trigger: {}", e.getMessage(), e); + return; + } + running = true; + Thread.ofVirtual().name("policy-folder-watch").start(this::watchLoop); + long reconcileSeconds = applicationProperties.getPolicies().getWatchReconcileSeconds(); + reconciler = + Executors.newSingleThreadScheduledExecutor( + Thread.ofVirtual().name("policy-folder-reconcile-", 0).factory()); + // First reconcile runs immediately so pre-existing files are picked up at startup. + reconciler.scheduleAtFixedRate(this::safeReconcile, 0, reconcileSeconds, TimeUnit.SECONDS); + log.info("Folder-watch trigger started (reconcile every {}s)", reconcileSeconds); + } + + @Override + public synchronized void stop() { + running = false; + if (reconciler != null) { + reconciler.shutdownNow(); + reconciler = null; + } + if (watchService != null) { + try { + watchService.close(); // wakes the watch loop with ClosedWatchServiceException + } catch (IOException e) { + log.debug("Error closing folder watch service: {}", e.getMessage()); + } + watchService = null; + } + keysByDir.clear(); + dirByKey.clear(); + } + + private void watchLoop() { + // Capture the service once: stop() may null the field, and a local avoids racing that to an + // NPE (close() still wakes take()/poll() on this same instance). + WatchService watcher = watchService; + if (watcher == null) { + return; + } + while (running) { + WatchKey first; + try { + first = watcher.take(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } catch (ClosedWatchServiceException e) { + return; + } + runForChangedDirs(drainBurst(watcher, first)); + } + } + + /** + * Collect the directories touched by {@code first} and any further events that arrive within + * the quiet period, so a burst of file-system events becomes a single set of affected + * directories. The event kinds are irrelevant: any event on a watched dir just means "go look". + */ + private Set drainBurst(WatchService watcher, WatchKey first) { + long quietPeriodMs = applicationProperties.getPolicies().getWatchQuietPeriodMs(); + Set changed = new HashSet<>(); + WatchKey key = first; + while (key != null) { + key.pollEvents(); + Path dir = dirByKey.get(key); + if (dir != null) { + changed.add(dir); + } + key.reset(); + try { + key = watcher.poll(quietPeriodMs, TimeUnit.MILLISECONDS); + } catch (ClosedWatchServiceException | InterruptedException e) { + break; + } + } + return changed; + } + + /** Run every folder-watch policy that draws from one of the changed directories. */ + void runForChangedDirs(Set changedDirs) { + if (changedDirs.isEmpty()) { + return; + } + for (Policy policy : policyStore.findByTriggerType(TYPE)) { + List dirs; + try { + dirs = watchDirsOf(policy); + } catch (RuntimeException e) { + log.warn( + "Folder-watch policy {} is misconfigured: {}", policy.id(), e.getMessage()); + continue; + } + if (dirs.stream().anyMatch(changedDirs::contains)) { + log.debug("Folder-watch policy {} ({}) saw activity", policy.id(), policy.name()); + policyRunner.run(policy); + } + } + } + + private void safeReconcile() { + try { + syncRegistrations(); + runAll(); + } catch (RuntimeException e) { + log.error("Folder-watch reconcile failed: {}", e.getMessage(), e); + } + } + + /** The reconcile safety net: run every folder-watch policy regardless of watch events. */ + void runAll() { + for (Policy policy : policyStore.findByTriggerType(TYPE)) { + try { + policyRunner.run(policy); + } catch (RuntimeException e) { + log.warn( + "Folder-watch reconcile run failed for policy {}: {}", + policy.id(), + e.getMessage()); + } + } + } + + /** + * Bring the set of watched directories in line with the current folder-watch policies: register + * newly required directories that exist on disk, and cancel ones no longer wanted. + */ + synchronized void syncRegistrations() { + if (watchService == null) { + return; + } + Set desired = desiredDirs(); + + keysByDir + .entrySet() + .removeIf( + entry -> { + if (desired.contains(entry.getKey())) { + return false; + } + entry.getValue().cancel(); + dirByKey.remove(entry.getValue()); + return true; + }); + + for (Path dir : desired) { + if (keysByDir.containsKey(dir)) { + continue; + } + try { + WatchKey key = dir.register(watchService, ENTRY_CREATE, ENTRY_MODIFY); + keysByDir.put(dir, key); + dirByKey.put(key, dir); + log.info("Watching {} for folder-watch policies", dir); + } catch (IOException | RuntimeException e) { + log.warn("Could not watch {}: {}", dir, e.getMessage()); + } + } + } + + /** The directories currently registered with the watch service. Visible for tests. */ + Set watchedDirs() { + return Set.copyOf(keysByDir.keySet()); + } + + /** Every existing directory any current folder-watch policy wants watched. */ + private Set desiredDirs() { + Set dirs = new HashSet<>(); + for (Policy policy : policyStore.findByTriggerType(TYPE)) { + try { + for (Path dir : watchDirsOf(policy)) { + if (Files.isDirectory(dir)) { + dirs.add(dir); + } + } + } catch (RuntimeException e) { + log.warn( + "Folder-watch policy {} is misconfigured: {}", policy.id(), e.getMessage()); + } + } + return dirs; + } + + /** + * The normalised, absolute directories this policy's sources expose to watch. Normalisation + * makes registration keys and event-time matching comparable regardless of how the path was + * configured. + */ + private List watchDirsOf(Policy policy) { + List dirs = new ArrayList<>(); + for (InputSpec spec : policy.sources()) { + InputSource source = sourceFor(spec); + if (source == null) { + continue; + } + for (Path dir : source.watchTargets(spec)) { + dirs.add(dir.toAbsolutePath().normalize()); + } + } + return dirs; + } + + private InputSource sourceFor(InputSpec spec) { + return inputSources.stream() + .filter(source -> source.supports(spec)) + .findFirst() + .orElse(null); + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/ManualTrigger.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/ManualTrigger.java deleted file mode 100644 index 2ae24671a..000000000 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/ManualTrigger.java +++ /dev/null @@ -1,41 +0,0 @@ -package stirling.software.proprietary.policy.trigger; - -import org.springframework.stereotype.Service; - -import lombok.RequiredArgsConstructor; - -import stirling.software.proprietary.policy.engine.PolicyEngine; -import stirling.software.proprietary.policy.engine.PolicyRunHandle; -import stirling.software.proprietary.policy.model.PipelineDefinition; -import stirling.software.proprietary.policy.model.Policy; -import stirling.software.proprietary.policy.model.PolicyInputs; -import stirling.software.proprietary.policy.progress.PolicyProgressListener; - -/** - * Runs policies on demand, in response to a request (the {@code PolicyController} endpoints, an AI, - * or another automation). It is the request-driven trigger: no background lifecycle, it just - * forwards to the engine. Any policy can be run manually regardless of its configured trigger type. - */ -@Service -@RequiredArgsConstructor -public class ManualTrigger implements PolicyTrigger { - - private final PolicyEngine policyEngine; - - @Override - public String type() { - return "manual"; - } - - /** Run a stored policy immediately and return its run handle. */ - public PolicyRunHandle run( - Policy policy, PolicyInputs inputs, PolicyProgressListener listener) { - return policyEngine.runPolicy(policy, inputs, listener); - } - - /** Run an ad-hoc pipeline (no stored policy), e.g. for AI or Automate one-offs. */ - public PolicyRunHandle fire( - PipelineDefinition definition, PolicyInputs inputs, PolicyProgressListener listener) { - return policyEngine.submit(definition, inputs, listener); - } -} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/PolicyTrigger.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/PolicyTrigger.java index c63e5a1da..1e3718d7f 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/PolicyTrigger.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/PolicyTrigger.java @@ -1,22 +1,36 @@ package stirling.software.proprietary.policy.trigger; +import stirling.software.proprietary.policy.model.Policy; + /** - * Activates policies of one trigger type. A trigger owns a {@link #type()} (matching {@code - * TriggerConfig.type()}); when its condition fires it runs the relevant {@code Policy} through the - * {@code PolicyEngine}. + * An automatic trigger: the thing that decides when a policy runs without a person asking. + * A trigger owns a {@link #type()} (matching {@code TriggerConfig.type()}); when its condition + * fires it hands the policy to the {@code PolicyRunner}, which pulls the policy's sources and + * starts the runs. A trigger never resolves sources itself. * - *

Background triggers (folder watcher, schedule) are driven by configuration: on {@link - * #start()} they begin watching/scheduling for the policies returned by {@code - * PolicyStore.findByTriggerType(type())}, and stop on {@link #stop()}. Request-driven triggers - * (manual) have no background lifecycle and run a policy directly in response to a call. New - * trigger kinds are new beans of this type; the engine and the {@code Policy} model do not change. + *

Triggers are background, configuration-driven beans (schedule, and in future webhook or + * folder-watch): on {@link #start()} they begin watching/scheduling for the policies returned by + * {@code PolicyStore.findByTriggerType(type())}, and stop on {@link #stop()}. New trigger kinds are + * new beans of this type; the runner and the {@code Policy} model do not change. + * + *

Manual running is not a trigger - every policy can always be run on demand via the {@code + * PolicyRunner} regardless of whether it has a trigger. */ public interface PolicyTrigger { /** Stable identifier for this trigger kind, matching {@code TriggerConfig.type()}. */ String type(); - /** Begin activating policies of this type (e.g. start a folder watcher). No-op for manual. */ + /** + * Check that this trigger is usable for the given policy, throwing {@link + * IllegalArgumentException} if not. Called when a policy is saved so misconfiguration fails + * fast rather than at fire time. Receives the whole {@link Policy} (not just its {@code + * TriggerConfig}) so a trigger whose firing depends on the policy's sources (folder-watch) can + * assert that relationship; most triggers only inspect {@code policy.trigger()}. + */ + default void validate(Policy policy) {} + + /** Begin activating policies of this type (e.g. start the schedule sweep). */ default void start() {} /** Stop activating and release any resources. */ diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/PolicyTriggerManager.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/PolicyTriggerManager.java new file mode 100644 index 000000000..1e87dcd4d --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/PolicyTriggerManager.java @@ -0,0 +1,56 @@ +package stirling.software.proprietary.policy.trigger; + +import java.util.List; + +import org.springframework.context.SmartLifecycle; +import org.springframework.stereotype.Service; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +/** + * Starts and stops every {@link PolicyTrigger} with the application lifecycle. Background triggers + * (schedule, and future folder/S3) begin watching on {@link #start()} and release resources on + * {@link #stop()}; request-driven triggers (manual) are no-ops. + * + *

This is the single activation point for triggers - a new background trigger only has to be a + * {@link PolicyTrigger} bean. + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class PolicyTriggerManager implements SmartLifecycle { + + private final List triggers; + + private volatile boolean running; + + @Override + public void start() { + for (PolicyTrigger trigger : triggers) { + try { + trigger.start(); + } catch (RuntimeException e) { + log.error("Failed to start trigger '{}': {}", trigger.type(), e.getMessage(), e); + } + } + running = true; + } + + @Override + public void stop() { + for (PolicyTrigger trigger : triggers) { + try { + trigger.stop(); + } catch (RuntimeException e) { + log.error("Failed to stop trigger '{}': {}", trigger.type(), e.getMessage(), e); + } + } + running = false; + } + + @Override + public boolean isRunning() { + return running; + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/ScheduleTrigger.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/ScheduleTrigger.java new file mode 100644 index 000000000..f138710d7 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/ScheduleTrigger.java @@ -0,0 +1,158 @@ +package stirling.software.proprietary.policy.trigger; + +import java.time.Instant; +import java.time.ZoneId; +import java.time.ZoneOffset; +import java.time.ZonedDateTime; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +import org.springframework.stereotype.Service; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +import stirling.software.common.model.ApplicationProperties; +import stirling.software.proprietary.policy.engine.PolicyRunner; +import stirling.software.proprietary.policy.model.Policy; +import stirling.software.proprietary.policy.model.Schedule; +import stirling.software.proprietary.policy.store.PolicyStore; + +import tools.jackson.databind.ObjectMapper; + +/** + * Fires policies on a {@link Schedule}. On {@link #start()} it sweeps on a fixed interval; each + * sweep finds the enabled "schedule" policies and runs any whose next firing has come due since it + * last fired. + * + *

The trigger only decides when: once a policy is due it hands it to the {@link + * PolicyRunner}, which pulls from the policy's configured sources and starts the runs. The trigger + * knows nothing about folders, buckets, or how many runs a sweep produces. + * + *

Caveat: last-fire times are tracked in memory, so this assumes a single node and resets + * on restart; cluster-wide coordination (leader election) is a follow-up. + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class ScheduleTrigger implements PolicyTrigger { + + private static final String TYPE = "schedule"; + + private final PolicyStore policyStore; + private final PolicyRunner policyRunner; + private final ObjectMapper objectMapper; + private final ApplicationProperties applicationProperties; + + private final Map lastFiredByPolicy = new ConcurrentHashMap<>(); + private volatile ScheduledExecutorService scheduler; + + @Override + public String type() { + return TYPE; + } + + @Override + public void validate(Policy policy) { + ScheduleConfig.from(objectMapper, policy.trigger().options()); + } + + @Override + public synchronized void start() { + if (scheduler != null) { + return; + } + long sweepSeconds = applicationProperties.getPolicies().getScheduleSweepSeconds(); + scheduler = + Executors.newSingleThreadScheduledExecutor( + Thread.ofVirtual().name("policy-schedule-", 0).factory()); + scheduler.scheduleAtFixedRate( + this::safeSweep, sweepSeconds, sweepSeconds, TimeUnit.SECONDS); + log.info("Schedule trigger started (sweep every {}s)", sweepSeconds); + } + + @Override + public synchronized void stop() { + if (scheduler != null) { + scheduler.shutdownNow(); + scheduler = null; + } + } + + private void safeSweep() { + try { + sweep(Instant.now()); + } catch (RuntimeException e) { + log.error("Schedule sweep failed: {}", e.getMessage(), e); + } + } + + /** Fire every scheduled policy that is due as of {@code now}. Package-visible for testing. */ + void sweep(Instant now) { + for (Policy policy : policyStore.findByTriggerType(TYPE)) { + ScheduleConfig config; + try { + config = ScheduleConfig.from(objectMapper, policy.trigger().options()); + } catch (IllegalArgumentException e) { + log.warn("Scheduled policy {} is misconfigured: {}", policy.id(), e.getMessage()); + continue; + } + + // First time we see a policy, baseline its last-fire to now so it does not fire + // immediately; subsequent sweeps fire it once its next firing has passed. + Instant last = lastFiredByPolicy.computeIfAbsent(policy.id(), id -> now); + ZonedDateTime next = config.schedule().nextAfter(last.atZone(config.zone())); + if (!next.toInstant().isAfter(now)) { + lastFiredByPolicy.put(policy.id(), now); + log.info("Scheduled policy {} ({}) is due", policy.id(), policy.name()); + policyRunner.run(policy); + } + } + } + + /** + * The typed, validated form of a schedule trigger's options: the {@link Schedule} and the zone + * its wall-clock kinds are evaluated in (default UTC). Construction fails for a missing/invalid + * schedule or zone. + */ + record ScheduleConfig(Schedule schedule, ZoneId zone) { + + private static final String SCHEDULE_OPTION = "schedule"; + private static final String ZONE_OPTION = "zone"; + + static ScheduleConfig from(ObjectMapper mapper, Map options) { + Object scheduleNode = options.get(SCHEDULE_OPTION); + if (scheduleNode == null) { + throw new IllegalArgumentException("schedule trigger requires a 'schedule'"); + } + Schedule schedule; + try { + schedule = mapper.convertValue(scheduleNode, Schedule.class); + } catch (RuntimeException e) { + throw new IllegalArgumentException("invalid schedule: " + rootMessage(e), e); + } + + ZoneId zone = ZoneOffset.UTC; + Object zoneNode = options.get(ZONE_OPTION); + if (zoneNode != null && !zoneNode.toString().isBlank()) { + try { + zone = ZoneId.of(zoneNode.toString()); + } catch (RuntimeException e) { + throw new IllegalArgumentException("invalid zone '" + zoneNode + "'"); + } + } + return new ScheduleConfig(schedule, zone); + } + + private static String rootMessage(Throwable t) { + Throwable cause = t; + while (cause.getCause() != null) { + cause = cause.getCause(); + } + return cause.getMessage(); + } + } +} diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/config/FolderAccessGuardTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/config/FolderAccessGuardTest.java new file mode 100644 index 000000000..ea33319a8 --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/config/FolderAccessGuardTest.java @@ -0,0 +1,99 @@ +package stirling.software.proprietary.policy.config; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import java.util.List; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.springframework.core.env.StandardEnvironment; + +import stirling.software.common.configuration.InstallationPathConfig; +import stirling.software.common.model.ApplicationProperties; +import stirling.software.proprietary.policy.model.InputSpec; +import stirling.software.proprietary.policy.model.OutputSpec; +import stirling.software.proprietary.policy.model.Policy; + +/** + * Tests for {@link FolderAccessGuard}: folder access is fail-closed, confined to the configured + * allowed roots, never reaches Stirling's own config directory, and is off entirely under SaaS. + */ +class FolderAccessGuardTest { + + @TempDir Path tempDir; + + private FolderAccessGuard guard(List allowedRoots, String... activeProfiles) { + ApplicationProperties properties = new ApplicationProperties(); + properties.getPolicies().setAllowedFolderRoots(allowedRoots); + StandardEnvironment environment = new StandardEnvironment(); + environment.setActiveProfiles(activeProfiles); + return new FolderAccessGuard(properties, environment); + } + + @Test + void permitsAndNormalisesADirectoryWithinAnAllowedRoot() { + FolderAccessGuard guard = guard(List.of(tempDir.toString())); + Path within = tempDir.resolve("inbox"); + + assertEquals(within.toAbsolutePath().normalize(), guard.requirePermitted(within)); + } + + @Test + void rejectsADirectoryOutsideEveryAllowedRoot() { + FolderAccessGuard guard = guard(List.of(tempDir.toString())); + assertThrows( + IllegalArgumentException.class, + () -> guard.requirePermitted(tempDir.resolveSibling("elsewhere"))); + } + + @Test + void rejectsTraversalThatWalksOutOfAnAllowedRoot() { + FolderAccessGuard guard = guard(List.of(tempDir.toString())); + assertThrows( + IllegalArgumentException.class, + () -> guard.requirePermitted(tempDir.resolve("..").resolve("escaped"))); + } + + @Test + void rejectsEverythingWhenNoRootsAreConfigured() { + FolderAccessGuard guard = guard(List.of()); + assertThrows(IllegalArgumentException.class, () -> guard.requirePermitted(tempDir)); + } + + @Test + void rejectsTheStirlingConfigDirectoryEvenWhenItWouldBeInsideAnAllowedRoot() { + Path configDir = + Path.of(InstallationPathConfig.getConfigPath()).toAbsolutePath().normalize(); + // Allow the config dir's parent, so only the protected-path rule can reject it. + FolderAccessGuard guard = guard(List.of(configDir.getParent().toString())); + + assertThrows( + IllegalArgumentException.class, + () -> guard.requirePermitted(configDir.resolve("settings.yml"))); + } + + @Test + void refusesAllFolderAccessUnderTheSaasProfile() { + FolderAccessGuard guard = guard(List.of(tempDir.toString()), "saas"); + assertThrows(IllegalArgumentException.class, () -> guard.requirePermitted(tempDir)); + } + + @Test + void usesFolderAccessDetectsFolderSourcesAndOutputs() { + FolderAccessGuard guard = guard(List.of(tempDir.toString())); + + assertTrue( + guard.usesFolderAccess( + policy(List.of(InputSpec.folder("/in")), OutputSpec.inline()))); + assertTrue(guard.usesFolderAccess(policy(List.of(), OutputSpec.folder("/out")))); + assertFalse(guard.usesFolderAccess(policy(List.of(), OutputSpec.inline()))); + } + + private static Policy policy(List sources, OutputSpec output) { + return new Policy("p1", "p", "owner", true, null, sources, List.of(), output); + } +} diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyEngineTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyEngineTest.java index bf6442b42..e47916dbe 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyEngineTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyEngineTest.java @@ -50,7 +50,6 @@ import stirling.software.proprietary.policy.model.Policy; import stirling.software.proprietary.policy.model.PolicyInputs; import stirling.software.proprietary.policy.model.PolicyRun; import stirling.software.proprietary.policy.model.PolicyRunStatus; -import stirling.software.proprietary.policy.model.TriggerConfig; import stirling.software.proprietary.policy.output.InlineOutputSink; import stirling.software.proprietary.policy.progress.PolicyProgressListener; @@ -93,7 +92,7 @@ class PolicyEngineTest { toolMetadataService, tempFileManager, JsonMapper.builder().build()); - registry = new PolicyRunRegistry(30); + registry = new PolicyRunRegistry(new ApplicationProperties()); InlineOutputSink sink = new InlineOutputSink(fileStorage); engine = new PolicyEngine( @@ -190,7 +189,7 @@ class PolicyEngineTest { "rotate", "owner", true, - TriggerConfig.manual(), + null, List.of(new PipelineStep(ROTATE, Map.of())), OutputSpec.inline()); diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunRegistryTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunRegistryTest.java index 823d53edc..51645e71b 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunRegistryTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunRegistryTest.java @@ -12,6 +12,7 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import stirling.software.common.model.ApplicationProperties; import stirling.software.proprietary.policy.model.PipelineDefinition; import stirling.software.proprietary.policy.model.PolicyRun; import stirling.software.proprietary.policy.model.WaitState; @@ -25,7 +26,7 @@ class PolicyRunRegistryTest { @BeforeEach void setUp() { - registry = new PolicyRunRegistry(30); + registry = new PolicyRunRegistry(new ApplicationProperties()); } @AfterEach diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunnerTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunnerTest.java new file mode 100644 index 000000000..2cefb1b27 --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunnerTest.java @@ -0,0 +1,159 @@ +package stirling.software.proprietary.policy.engine; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import stirling.software.proprietary.policy.input.InputSource; +import stirling.software.proprietary.policy.input.ResolvedInput; +import stirling.software.proprietary.policy.model.InputSpec; +import stirling.software.proprietary.policy.model.OutputSpec; +import stirling.software.proprietary.policy.model.PipelineStep; +import stirling.software.proprietary.policy.model.Policy; +import stirling.software.proprietary.policy.model.PolicyInputs; +import stirling.software.proprietary.policy.model.PolicyRun; +import stirling.software.proprietary.policy.model.PolicyRunStatus; +import stirling.software.proprietary.policy.progress.PolicyProgressListener; + +/** + * Tests for {@link PolicyRunner}: the one place that turns a policy's sources into runs. Verifies + * it pulls every source, runs one job per unit of work, feeds each unit's completion hook the run + * outcome, and that a generator (no sources) still runs once. + */ +@ExtendWith(MockitoExtension.class) +class PolicyRunnerTest { + + @Mock private PolicyEngine policyEngine; + @Mock private InputSource folderSource; + + private PolicyRunner runner; + + @BeforeEach + void setUp() { + runner = new PolicyRunner(policyEngine, List.of(folderSource)); + } + + @Test + void runsOnceWithNoFilesWhenThePolicyHasNoSources() { + Policy policy = policy(List.of()); + when(policyEngine.runPolicy(eq(policy), any(), any())) + .thenReturn(new PolicyRunHandle("r", new CompletableFuture<>())); + + runner.run(policy); + + ArgumentCaptor inputs = ArgumentCaptor.forClass(PolicyInputs.class); + verify(policyEngine).runPolicy(eq(policy), inputs.capture(), any()); + assertTrue(inputs.getValue().primary().isEmpty()); + } + + @Test + void pullsEverySourceAndRunsOnePerUnitOfWork() throws Exception { + InputSpec spec = InputSpec.folder("/in"); + Policy policy = policy(List.of(spec)); + when(folderSource.supports(spec)).thenReturn(true); + when(folderSource.resolve(spec)) + .thenReturn( + List.of( + ResolvedInput.of(PolicyInputs.of(List.of())), + ResolvedInput.of(PolicyInputs.of(List.of())))); + when(policyEngine.runPolicy(any(), any(), any())) + .thenReturn(new PolicyRunHandle("r", new CompletableFuture<>())); + + runner.run(policy); + + verify(policyEngine, times(2)).runPolicy(eq(policy), any(), any()); + } + + @Test + void feedsEachUnitsCompletionHookTheRunOutcome() throws Exception { + InputSpec spec = InputSpec.folder("/in"); + Policy policy = policy(List.of(spec)); + AtomicBoolean outcome = new AtomicBoolean(false); + ResolvedInput unit = new ResolvedInput(PolicyInputs.of(List.of()), outcome::set); + when(folderSource.supports(spec)).thenReturn(true); + when(folderSource.resolve(spec)).thenReturn(List.of(unit)); + CompletableFuture completion = new CompletableFuture<>(); + when(policyEngine.runPolicy(any(), any(), any())) + .thenReturn(new PolicyRunHandle("r", completion)); + + runner.run(policy); + + PolicyRun run = mock(PolicyRun.class); + when(run.getStatus()).thenReturn(PolicyRunStatus.COMPLETED); + completion.complete(run); + + assertTrue(outcome.get()); + } + + @Test + void reportsFailureToTheCompletionHookWhenTheRunDoesNotComplete() throws Exception { + InputSpec spec = InputSpec.folder("/in"); + Policy policy = policy(List.of(spec)); + AtomicBoolean outcome = new AtomicBoolean(true); + ResolvedInput unit = new ResolvedInput(PolicyInputs.of(List.of()), outcome::set); + when(folderSource.supports(spec)).thenReturn(true); + when(folderSource.resolve(spec)).thenReturn(List.of(unit)); + CompletableFuture completion = new CompletableFuture<>(); + when(policyEngine.runPolicy(any(), any(), any())) + .thenReturn(new PolicyRunHandle("r", completion)); + + runner.run(policy); + completion.completeExceptionally(new RuntimeException("boom")); + + assertFalse(outcome.get()); + } + + @Test + void skipsSourcesWithNoMatchingBean() { + InputSpec spec = new InputSpec("s3", Map.of()); + Policy policy = policy(List.of(spec)); + when(folderSource.supports(spec)).thenReturn(false); + + runner.run(policy); + + verifyNoInteractions(policyEngine); + } + + @Test + void runWithSuppliedInputsBypassesSources() { + Policy policy = policy(List.of(InputSpec.folder("/in"))); + PolicyInputs inputs = PolicyInputs.of(List.of()); + PolicyRunHandle handle = new PolicyRunHandle("r", new CompletableFuture<>()); + when(policyEngine.runPolicy(policy, inputs, PolicyProgressListener.NOOP)) + .thenReturn(handle); + + assertSame(handle, runner.runWith(policy, inputs, PolicyProgressListener.NOOP)); + verifyNoInteractions(folderSource); + } + + private static Policy policy(List sources) { + return new Policy( + "p1", + "p", + "owner", + true, + null, + sources, + List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), + OutputSpec.inline()); + } +} diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyValidatorTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyValidatorTest.java new file mode 100644 index 000000000..edaa58651 --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyValidatorTest.java @@ -0,0 +1,114 @@ +package stirling.software.proprietary.policy.engine; + +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.util.List; +import java.util.Map; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import stirling.software.proprietary.policy.input.InputSource; +import stirling.software.proprietary.policy.model.InputSpec; +import stirling.software.proprietary.policy.model.OutputSpec; +import stirling.software.proprietary.policy.model.Policy; +import stirling.software.proprietary.policy.model.TriggerConfig; +import stirling.software.proprietary.policy.output.PolicyOutputSink; +import stirling.software.proprietary.policy.trigger.PolicyTrigger; + +/** Tests for {@link PolicyValidator}: routes each facet to its handler and surfaces failures. */ +@ExtendWith(MockitoExtension.class) +class PolicyValidatorTest { + + @Mock private PolicyTrigger trigger; + @Mock private InputSource inputSource; + @Mock private PolicyOutputSink outputSink; + + private PolicyValidator validator; + + @BeforeEach + void setUp() { + validator = + new PolicyValidator(List.of(trigger), List.of(inputSource), List.of(outputSink)); + } + + @Test + void delegatesEachFacetToItsHandler() { + when(trigger.type()).thenReturn("schedule"); + when(inputSource.supports(any())).thenReturn(true); + when(outputSink.supports(any())).thenReturn(true); + Policy policy = policy("schedule"); + + validator.validate(policy); + + verify(trigger).validate(policy); + verify(inputSource).validate(policy.sources().get(0)); + verify(outputSink).validate(policy.output()); + } + + @Test + void skipsTriggerValidationForAManualOnlyPolicy() { + when(inputSource.supports(any())).thenReturn(true); + when(outputSink.supports(any())).thenReturn(true); + + validator.validate(manualOnly()); + + verify(trigger, never()).validate(any()); + } + + @Test + void surfacesAnInvalidConfigFromAHandler() { + when(trigger.type()).thenReturn("schedule"); + doThrow(new IllegalArgumentException("invalid schedule")).when(trigger).validate(any()); + + IllegalArgumentException ex = + assertThrows( + IllegalArgumentException.class, + () -> validator.validate(policy("schedule"))); + assertTrue(ex.getMessage().contains("schedule")); + } + + @Test + void rejectsAnUnknownTriggerType() { + when(trigger.type()).thenReturn("schedule"); + + IllegalArgumentException ex = + assertThrows( + IllegalArgumentException.class, + () -> validator.validate(policy("mystery"))); + assertTrue(ex.getMessage().contains("unknown trigger type")); + } + + private static Policy policy(String triggerType) { + return new Policy( + "p1", + "p", + "owner", + true, + new TriggerConfig(triggerType, Map.of()), + List.of(InputSpec.folder("/in")), + List.of(), + OutputSpec.inline()); + } + + private static Policy manualOnly() { + return new Policy( + "p1", + "p", + "owner", + true, + null, + List.of(InputSpec.folder("/in")), + List.of(), + OutputSpec.inline()); + } +} diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/input/FolderInputSourceTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/input/FolderInputSourceTest.java new file mode 100644 index 000000000..3e9d34ccf --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/input/FolderInputSourceTest.java @@ -0,0 +1,137 @@ +package stirling.software.proprietary.policy.input; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.lenient; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.Map; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.io.TempDir; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.core.env.StandardEnvironment; + +import stirling.software.common.model.ApplicationProperties; +import stirling.software.common.util.FileReadinessChecker; +import stirling.software.proprietary.policy.config.FolderAccessGuard; +import stirling.software.proprietary.policy.model.InputSpec; + +/** Tests for {@link FolderInputSource}: consume (claim + route) and snapshot (read-only) modes. */ +@ExtendWith(MockitoExtension.class) +class FolderInputSourceTest { + + @Mock private FileReadinessChecker readinessChecker; + + @TempDir Path tempDir; + + private FolderInputSource source; + + @BeforeEach + void setUp() { + ApplicationProperties properties = new ApplicationProperties(); + properties.getPolicies().setAllowedFolderRoots(List.of(tempDir.toString())); + FolderAccessGuard guard = new FolderAccessGuard(properties, new StandardEnvironment()); + source = new FolderInputSource(readinessChecker, guard); + // Lenient: the missing-dir / nonexistent-dir cases return before any readiness check. + lenient().when(readinessChecker.isReady(any())).thenReturn(true); + } + + @Test + void consumeClaimsFilesAndRoutesToDoneOnSuccess() throws IOException { + Path inputDir = Files.createDirectories(tempDir.resolve("in")); + Files.writeString(inputDir.resolve("doc.pdf"), "data"); + + List work = source.resolve(InputSpec.folder(inputDir.toString())); + + assertEquals(1, work.size()); + assertEquals(1, work.get(0).inputs().primary().size()); + // Claimed out of the input dir. + assertFalse(Files.exists(inputDir.resolve("doc.pdf"))); + assertTrue( + Files.exists( + inputDir.resolve(".stirling").resolve("processing").resolve("doc.pdf"))); + + work.get(0).onComplete().accept(true); + assertTrue(Files.exists(inputDir.resolve(".stirling").resolve("done").resolve("doc.pdf"))); + assertFalse( + Files.exists( + inputDir.resolve(".stirling").resolve("processing").resolve("doc.pdf"))); + } + + @Test + void consumeRoutesToErrorOnFailure() throws IOException { + Path inputDir = Files.createDirectories(tempDir.resolve("in")); + Files.writeString(inputDir.resolve("doc.pdf"), "data"); + + List work = source.resolve(InputSpec.folder(inputDir.toString())); + work.get(0).onComplete().accept(false); + + assertTrue(Files.exists(inputDir.resolve(".stirling").resolve("error").resolve("doc.pdf"))); + } + + @Test + void snapshotReadsWithoutClaiming() throws IOException { + Path inputDir = Files.createDirectories(tempDir.resolve("in")); + Files.writeString(inputDir.resolve("doc.pdf"), "data"); + + List work = + source.resolve( + new InputSpec( + "folder", + Map.of("directory", inputDir.toString(), "mode", "snapshot"))); + + assertEquals(1, work.size()); + // Not moved, and completing the run is a no-op. + assertTrue(Files.exists(inputDir.resolve("doc.pdf"))); + work.get(0).onComplete().accept(true); + assertTrue(Files.exists(inputDir.resolve("doc.pdf"))); + } + + @Test + void missingDirectoryOptionFails() { + assertThrows( + IllegalArgumentException.class, + () -> source.resolve(new InputSpec("folder", Map.of()))); + } + + @Test + void nonexistentDirectoryYieldsNoWork() throws IOException { + List work = + source.resolve(InputSpec.folder(tempDir.resolve("nope").toString())); + assertTrue(work.isEmpty()); + } + + @Test + void validateRejectsMissingDirectory() { + assertThrows( + IllegalArgumentException.class, + () -> source.validate(new InputSpec("folder", Map.of()))); + } + + @Test + void rejectsADirectoryOutsideTheAllowedRoots() { + Path outside = tempDir.resolveSibling("not-allowed"); + assertThrows( + IllegalArgumentException.class, + () -> source.resolve(InputSpec.folder(outside.toString()))); + assertThrows( + IllegalArgumentException.class, + () -> source.validate(InputSpec.folder(outside.toString()))); + } + + @Test + void watchTargetsIsTheConfiguredDirectory() { + Path inputDir = tempDir.resolve("in"); + assertEquals(List.of(inputDir), source.watchTargets(InputSpec.folder(inputDir.toString()))); + } +} diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/FolderOutputSinkTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/FolderOutputSinkTest.java new file mode 100644 index 000000000..0d714162d --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/FolderOutputSinkTest.java @@ -0,0 +1,105 @@ +package stirling.software.proprietary.policy.output; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.Map; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.springframework.core.env.StandardEnvironment; +import org.springframework.core.io.ByteArrayResource; +import org.springframework.core.io.Resource; + +import stirling.software.common.model.ApplicationProperties; +import stirling.software.common.model.job.ResultFile; +import stirling.software.proprietary.policy.config.FolderAccessGuard; +import stirling.software.proprietary.policy.model.OutputSpec; + +/** Tests for {@link FolderOutputSink}: outputs are written to the configured directory on disk. */ +class FolderOutputSinkTest { + + @TempDir Path tempDir; + + private FolderOutputSink sink; + + @BeforeEach + void setUp() { + ApplicationProperties properties = new ApplicationProperties(); + properties.getPolicies().setAllowedFolderRoots(List.of(tempDir.toString())); + sink = new FolderOutputSink(new FolderAccessGuard(properties, new StandardEnvironment())); + } + + @Test + void writesEachOutputToTheDirectory() throws IOException { + Path out = tempDir.resolve("out"); + List outputs = List.of(named("a.pdf", "aaa"), named("b.pdf", "bb")); + + List results = + sink.deliver("run-1", outputs, OutputSpec.folder(out.toString())); + + assertEquals(2, results.size()); + assertTrue(Files.exists(out.resolve("a.pdf"))); + assertEquals("aaa", Files.readString(out.resolve("a.pdf"))); + assertEquals("bb", Files.readString(out.resolve("b.pdf"))); + } + + @Test + void collidingNamesGetAUniqueSuffix() throws IOException { + Path out = tempDir.resolve("out"); + List outputs = List.of(named("a.pdf", "first"), named("a.pdf", "second")); + + sink.deliver("run-1", outputs, OutputSpec.folder(out.toString())); + + assertTrue(Files.exists(out.resolve("a.pdf"))); + assertTrue(Files.exists(out.resolve("a (1).pdf"))); + } + + @Test + void missingDirectoryOptionIsRejected() { + OutputSpec noDir = new OutputSpec("folder", Map.of()); + assertThrows(IllegalArgumentException.class, () -> sink.validate(noDir)); + assertThrows( + IllegalArgumentException.class, + () -> sink.deliver("run-1", List.of(named("a.pdf", "x")), noDir)); + } + + @Test + void aDirectoryOutsideTheAllowedRootsIsRejected() { + OutputSpec outside = OutputSpec.folder(tempDir.resolveSibling("not-allowed").toString()); + assertThrows(IllegalArgumentException.class, () -> sink.validate(outside)); + assertThrows( + IllegalArgumentException.class, + () -> sink.deliver("run-1", List.of(named("a.pdf", "x")), outside)); + } + + @Test + void filenamesWithPathTraversalAreConfinedToTheDirectory() throws IOException { + Path out = tempDir.resolve("out"); + List outputs = + List.of(named("../escape.pdf", "x"), named("nested/deep.pdf", "y")); + + sink.deliver("run-1", outputs, OutputSpec.folder(out.toString())); + + // Each name is reduced to its bare form inside the target dir; nothing escapes. + assertTrue(Files.exists(out.resolve("escape.pdf"))); + assertTrue(Files.exists(out.resolve("deep.pdf"))); + assertFalse(Files.exists(tempDir.resolve("escape.pdf"))); + } + + private static ByteArrayResource named(String filename, String content) { + return new ByteArrayResource(content.getBytes()) { + @Override + public String getFilename() { + return filename; + } + }; + } +} diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/store/InProcessPolicyStoreTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/store/InProcessPolicyStoreTest.java index 2e9701918..60f14be20 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/store/InProcessPolicyStoreTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/store/InProcessPolicyStoreTest.java @@ -28,7 +28,7 @@ class InProcessPolicyStoreTest { @Test void savedPolicyGetsAnIdAndIsRetrievable() { - Policy saved = store.save(policy(null, "compress", "manual", true)); + Policy saved = store.save(policy(null, "compress", null, true)); assertNotNull(saved.id()); assertFalse(saved.id().isBlank()); @@ -37,7 +37,7 @@ class InProcessPolicyStoreTest { @Test void savingWithAnExistingIdUpdatesInPlace() { - Policy created = store.save(policy(null, "before", "manual", true)); + Policy created = store.save(policy(null, "before", null, true)); store.save( new Policy( @@ -45,7 +45,7 @@ class InProcessPolicyStoreTest { "after", "owner", true, - TriggerConfig.manual(), + null, List.of(), OutputSpec.inline())); @@ -55,19 +55,20 @@ class InProcessPolicyStoreTest { @Test void findByTriggerTypeReturnsOnlyEnabledMatches() { - store.save(policy(null, "watch", "folder", true)); - store.save(policy(null, "watch-disabled", "folder", false)); store.save(policy(null, "nightly", "schedule", true)); + store.save(policy(null, "nightly-disabled", "schedule", false)); + store.save(policy(null, "hooked", "webhook", true)); + store.save(policy(null, "on-demand", null, true)); // manual-only: no trigger - List folder = store.findByTriggerType("folder"); + List scheduled = store.findByTriggerType("schedule"); - assertEquals(1, folder.size()); - assertEquals("watch", folder.get(0).name()); + assertEquals(1, scheduled.size()); + assertEquals("nightly", scheduled.get(0).name()); } @Test void deleteRemovesThePolicy() { - Policy saved = store.save(policy(null, "p", "manual", true)); + Policy saved = store.save(policy(null, "p", null, true)); assertTrue(store.delete(saved.id())); assertTrue(store.get(saved.id()).isEmpty()); @@ -75,12 +76,14 @@ class InProcessPolicyStoreTest { } private static Policy policy(String id, String name, String triggerType, boolean enabled) { + TriggerConfig trigger = + triggerType == null ? null : new TriggerConfig(triggerType, Map.of()); return new Policy( id, name, "owner", enabled, - new TriggerConfig(triggerType, Map.of()), + trigger, List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), OutputSpec.inline()); } diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/store/JpaPolicyStoreTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/store/JpaPolicyStoreTest.java index 2a8212e2d..2bdbc08a7 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/store/JpaPolicyStoreTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/store/JpaPolicyStoreTest.java @@ -18,6 +18,7 @@ import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; +import stirling.software.proprietary.policy.model.InputSpec; import stirling.software.proprietary.policy.model.OutputSpec; import stirling.software.proprietary.policy.model.PipelineStep; import stirling.software.proprietary.policy.model.Policy; @@ -53,7 +54,8 @@ class JpaPolicyStoreTest { "compress incoming", "alice", true, - new TriggerConfig("folder", Map.of("path", "/in")), + new TriggerConfig("schedule", Map.of()), + List.of(InputSpec.folder("/in")), List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), OutputSpec.inline())); @@ -62,7 +64,7 @@ class JpaPolicyStoreTest { verify(repository).save(captor.capture()); PolicyEntity entity = captor.getValue(); assertEquals(saved.id(), entity.getId()); - assertEquals("folder", entity.getTriggerType()); + assertEquals("schedule", entity.getTriggerType()); assertTrue(entity.isEnabled()); // The stored JSON round-trips back to an equal policy. assertEquals(saved, objectMapper.readValue(entity.getPolicyJson(), Policy.class)); @@ -76,7 +78,7 @@ class JpaPolicyStoreTest { "rotate", "alice", true, - TriggerConfig.manual(), + null, // manual-only: no automatic trigger List.of( new PipelineStep( "/api/v1/general/rotate-pdf", Map.of("angle", 90))), @@ -94,16 +96,16 @@ class JpaPolicyStoreTest { "watch", "alice", true, - new TriggerConfig("folder", Map.of()), + new TriggerConfig("schedule", Map.of()), List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), OutputSpec.inline()); - when(repository.findByTriggerTypeAndEnabledTrue("folder")) + when(repository.findByTriggerTypeAndEnabledTrue("schedule")) .thenReturn(List.of(entityFor(policy))); - List folder = store.findByTriggerType("folder"); + List scheduled = store.findByTriggerType("schedule"); - assertEquals(1, folder.size()); - assertEquals("p1", folder.get(0).id()); + assertEquals(1, scheduled.size()); + assertEquals("p1", scheduled.get(0).id()); } @Test @@ -122,7 +124,7 @@ class JpaPolicyStoreTest { entity.setName(policy.name()); entity.setOwner(policy.owner()); entity.setEnabled(policy.enabled()); - entity.setTriggerType(policy.trigger().type()); + entity.setTriggerType(policy.trigger() == null ? null : policy.trigger().type()); entity.setPolicyJson(objectMapper.writeValueAsString(policy)); return entity; } diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/FolderWatchTriggerTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/FolderWatchTriggerTest.java new file mode 100644 index 000000000..9b0b9f1d2 --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/FolderWatchTriggerTest.java @@ -0,0 +1,178 @@ +package stirling.software.proprietary.policy.trigger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +import java.nio.file.FileSystems; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.WatchService; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.io.TempDir; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import stirling.software.common.model.ApplicationProperties; +import stirling.software.proprietary.policy.engine.PolicyRunner; +import stirling.software.proprietary.policy.input.InputSource; +import stirling.software.proprietary.policy.model.InputSpec; +import stirling.software.proprietary.policy.model.OutputSpec; +import stirling.software.proprietary.policy.model.PipelineStep; +import stirling.software.proprietary.policy.model.Policy; +import stirling.software.proprietary.policy.model.TriggerConfig; +import stirling.software.proprietary.policy.store.PolicyStore; + +/** + * Tests for {@link FolderWatchTrigger}'s dispatch logic via the package-visible {@code + * runForChangedDirs}/{@code runAll}, plus its cross-facet validation. The OS watch loop and + * scheduled reconcile are thin glue around these and are not exercised here (a real {@code + * WatchService} is timing-dependent), mirroring how {@link ScheduleTriggerTest} drives {@code + * sweep} directly. The folder source is stubbed to mirror {@code FolderInputSource.watchTargets}. + */ +@ExtendWith(MockitoExtension.class) +class FolderWatchTriggerTest { + + @Mock private PolicyStore policyStore; + @Mock private PolicyRunner policyRunner; + @Mock private InputSource folderSource; + + @TempDir Path tempDir; + + private FolderWatchTrigger trigger; + + @BeforeEach + void setUp() { + trigger = + new FolderWatchTrigger( + policyStore, + policyRunner, + List.of(folderSource), + new ApplicationProperties()); + lenient().when(folderSource.supports(any())).thenReturn(true); + lenient() + .when(folderSource.watchTargets(any())) + .thenAnswer( + invocation -> { + InputSpec spec = invocation.getArgument(0); + Object dir = spec.options().get("directory"); + if (dir == null) { + throw new IllegalArgumentException( + "folder input requires a 'directory' option"); + } + return List.of(Path.of(dir.toString())); + }); + } + + @Test + void validateRejectsPolicyWithNoWatchableSource() { + assertThrows( + IllegalArgumentException.class, + () -> trigger.validate(folderWatch("p1", List.of()))); + } + + @Test + void validateAcceptsPolicyWithAFolderSource() { + trigger.validate(folderWatch("p1", List.of(InputSpec.folder("/in")))); + } + + @Test + void runsOnlyPoliciesDrawingFromTheChangedDirectory() { + Policy a = folderWatch("a", List.of(InputSpec.folder("/in/a"))); + Policy b = folderWatch("b", List.of(InputSpec.folder("/in/b"))); + when(policyStore.findByTriggerType("folder-watch")).thenReturn(List.of(a, b)); + + trigger.runForChangedDirs(Set.of(normalized("/in/a"))); + + verify(policyRunner).run(a); + verify(policyRunner, never()).run(b); + } + + @Test + void skipsAMisconfiguredPolicyButStillRunsTheOthers() { + Policy bad = folderWatch("bad", List.of(new InputSpec("folder", Map.of()))); + Policy good = folderWatch("good", List.of(InputSpec.folder("/in/a"))); + when(policyStore.findByTriggerType("folder-watch")).thenReturn(List.of(bad, good)); + + trigger.runForChangedDirs(Set.of(normalized("/in/a"))); + + verify(policyRunner).run(good); + verify(policyRunner, never()).run(bad); + } + + @Test + void anEmptyChangeSetDoesNothing() { + trigger.runForChangedDirs(Set.of()); + + verifyNoInteractions(policyStore, policyRunner); + } + + @Test + void reconcileRunsEveryFolderWatchPolicyAsASafetyNet() { + Policy a = folderWatch("a", List.of(InputSpec.folder("/in/a"))); + Policy b = folderWatch("b", List.of(InputSpec.folder("/in/b"))); + when(policyStore.findByTriggerType("folder-watch")).thenReturn(List.of(a, b)); + + trigger.runAll(); + + verify(policyRunner).run(a); + verify(policyRunner).run(b); + } + + @Test + void syncRegistrationsWatchesExistingDirsAndCancelsRemovedOnes() throws Exception { + Path dirA = Files.createDirectories(tempDir.resolve("a")); + Path dirB = Files.createDirectories(tempDir.resolve("b")); + Path missing = tempDir.resolve("missing"); // never created on disk + + Policy a = folderWatch("a", List.of(InputSpec.folder(dirA.toString()))); + Policy b = folderWatch("b", List.of(InputSpec.folder(dirB.toString()))); + Policy m = folderWatch("m", List.of(InputSpec.folder(missing.toString()))); + + WatchService service = FileSystems.getDefault().newWatchService(); + try { + trigger.watchService = service; + + when(policyStore.findByTriggerType("folder-watch")).thenReturn(List.of(a, b, m)); + trigger.syncRegistrations(); + // Existing dirs are watched; the non-existent one is skipped. + assertEquals( + Set.of(normalized(dirA.toString()), normalized(dirB.toString())), + trigger.watchedDirs()); + + // b's policy is removed: its registration is cancelled, a remains. + when(policyStore.findByTriggerType("folder-watch")).thenReturn(List.of(a)); + trigger.syncRegistrations(); + assertEquals(Set.of(normalized(dirA.toString())), trigger.watchedDirs()); + } finally { + service.close(); + } + } + + private static Path normalized(String dir) { + return Path.of(dir).toAbsolutePath().normalize(); + } + + private static Policy folderWatch(String id, List sources) { + return new Policy( + id, + "watcher", + "owner", + true, + new TriggerConfig("folder-watch", Map.of()), + sources, + List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), + OutputSpec.inline()); + } +} diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/PolicyTriggerManagerTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/PolicyTriggerManagerTest.java new file mode 100644 index 000000000..41504424a --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/PolicyTriggerManagerTest.java @@ -0,0 +1,51 @@ +package stirling.software.proprietary.policy.trigger; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.verify; + +import java.util.List; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +/** + * Tests for {@link PolicyTriggerManager}: starts/stops every trigger, tolerating individual + * failures. + */ +@ExtendWith(MockitoExtension.class) +class PolicyTriggerManagerTest { + + @Mock private PolicyTrigger triggerA; + @Mock private PolicyTrigger triggerB; + + @Test + void startsAndStopsAllTriggers() { + PolicyTriggerManager manager = new PolicyTriggerManager(List.of(triggerA, triggerB)); + assertFalse(manager.isRunning()); + + manager.start(); + verify(triggerA).start(); + verify(triggerB).start(); + assertTrue(manager.isRunning()); + + manager.stop(); + verify(triggerA).stop(); + verify(triggerB).stop(); + assertFalse(manager.isRunning()); + } + + @Test + void oneTriggerFailingToStartDoesNotBlockTheOthers() { + doThrow(new RuntimeException("boom")).when(triggerA).start(); + PolicyTriggerManager manager = new PolicyTriggerManager(List.of(triggerA, triggerB)); + + manager.start(); + + verify(triggerB).start(); + assertTrue(manager.isRunning()); + } +} diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/ScheduleTriggerTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/ScheduleTriggerTest.java new file mode 100644 index 000000000..889c50ee4 --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/ScheduleTriggerTest.java @@ -0,0 +1,147 @@ +package stirling.software.proprietary.policy.trigger; + +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.time.DayOfWeek; +import java.time.Instant; +import java.time.LocalTime; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import stirling.software.common.model.ApplicationProperties; +import stirling.software.proprietary.policy.engine.PolicyRunner; +import stirling.software.proprietary.policy.model.OutputSpec; +import stirling.software.proprietary.policy.model.PipelineStep; +import stirling.software.proprietary.policy.model.Policy; +import stirling.software.proprietary.policy.model.Schedule; +import stirling.software.proprietary.policy.model.TriggerConfig; +import stirling.software.proprietary.policy.store.PolicyStore; + +import tools.jackson.databind.json.JsonMapper; + +/** + * Tests for {@link ScheduleTrigger}'s due-firing logic via the package-visible {@code + * sweep(Instant)}. The trigger only decides when a policy is due; pulling sources and starting runs + * is the {@link PolicyRunner}'s job, so these assert it delegates to the runner. Schedules default + * to UTC, so explicit UTC instants make these deterministic. + */ +@ExtendWith(MockitoExtension.class) +class ScheduleTriggerTest { + + @Mock private PolicyStore policyStore; + @Mock private PolicyRunner policyRunner; + + private ScheduleTrigger trigger; + + @BeforeEach + void setUp() { + trigger = + new ScheduleTrigger( + policyStore, + policyRunner, + JsonMapper.builder().build(), + new ApplicationProperties()); + } + + @Test + void firesOncePerScheduleWhenItComesDue() { + Policy policy = scheduled("p1", new Schedule.Every(1, Schedule.Unit.MINUTES)); + when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy)); + + Instant t0 = Instant.parse("2026-06-05T10:00:30Z"); + trigger.sweep(t0); // first sight: baseline, must not fire immediately + verify(policyRunner, never()).run(any()); + + trigger.sweep(t0.plusSeconds(120)); // the one-minute mark has passed + verify(policyRunner, times(1)).run(eq(policy)); + } + + @Test + void doesNotFireBeforeTheNextScheduledTime() { + Policy policy = scheduled("p1", new Schedule.Daily(LocalTime.of(3, 0))); // 03:00 UTC daily + when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy)); + + Instant t0 = Instant.parse("2026-06-05T10:00:00Z"); + trigger.sweep(t0); + trigger.sweep(t0.plusSeconds(60)); // next 03:00 is far away + + verify(policyRunner, never()).run(any()); + } + + @Test + void firesWeeklyOnAChosenDay() { + // 2026-06-05 is a Friday; the next Monday 09:00 is the soonest firing. + Policy policy = + scheduled("p1", new Schedule.Weekly(Set.of(DayOfWeek.MONDAY), LocalTime.of(9, 0))); + when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy)); + + Instant friday = Instant.parse("2026-06-05T10:00:00Z"); + trigger.sweep(friday); // baseline + trigger.sweep(Instant.parse("2026-06-08T09:00:00Z")); // Monday 09:00 + + verify(policyRunner, times(1)).run(eq(policy)); + } + + @Test + void skipsPoliciesWithAnInvalidSchedule() { + Policy policy = scheduledWithRawOptions("p1", Map.of()); // no schedule + when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy)); + + trigger.sweep(Instant.parse("2026-06-05T10:00:00Z")); + + verify(policyRunner, never()).run(any()); + } + + @Test + void validateRejectsMissingSchedule() { + assertThrows( + IllegalArgumentException.class, + () -> trigger.validate(scheduledWithRawOptions("p1", Map.of()))); + } + + @Test + void validateRejectsAnInvalidSchedule() { + Map options = + Map.of("schedule", Map.of("type", "every", "count", -5, "unit", "MINUTES")); + assertThrows( + IllegalArgumentException.class, + () -> trigger.validate(scheduledWithRawOptions("p1", options))); + } + + @Test + void validateAcceptsAValidScheduleAndZone() { + Map options = new LinkedHashMap<>(); + options.put("schedule", new Schedule.Daily(LocalTime.of(2, 0))); + options.put("zone", "Europe/London"); + trigger.validate(scheduledWithRawOptions("p1", options)); + } + + private static Policy scheduled(String id, Schedule schedule) { + return scheduledWithRawOptions(id, Map.of("schedule", schedule)); + } + + private static Policy scheduledWithRawOptions(String id, Map options) { + return new Policy( + id, + "nightly", + "owner", + true, + new TriggerConfig("schedule", options), + List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), + OutputSpec.inline()); + } +}