From 3bee6d212e28e7ff9b11622e1bc7c4e02c3c251e Mon Sep 17 00:00:00 2001 From: James Brunton Date: Fri, 31 Jul 2026 09:53:01 +0100 Subject: [PATCH] Change pipelines to have 1 input and 1 output (#7121) # Description of Changes Change pipelines so that sources and triggers are grouped into a list of inputs, so you can have a different trigger for each source in the list. This is necessary because triggers are not universally supported by all source types. If you wanted to have a pipeline pull from both a folder and an S3 bucket, the current system allows you to choose "Folder Watch" as the trigger, which will either do nothing or crash when it's paired with the S3 bucket. I've got reservations about actually allowing different triggers for every source because it allows for user workflows that I don't believe exist, like "I want this folder to be polled every minute and this other one to be polled every hour, but they should run the same tools and should output to the same place". Because of this (with agreement from Connor, Anthony and Matt) I've changed this PR to artificially limit pipelines to having 1 input & output at this stage. The backend is still shaped to support multiple inputs & outputs so it should be trivial to re-add support for them in the future if we decide we want to, but the UI can be much simpler and easier to understand with just 1 input and output. image --- .../policy/controller/PolicyController.java | 3 +- .../policy/engine/PolicyRunner.java | 42 ++- .../policy/engine/PolicyValidator.java | 52 ++- .../policy/model/PipelineInput.java | 16 + .../proprietary/policy/model/Policy.java | 61 +-- .../policy/model/PolicyBinding.java | 31 ++ .../overview/PolicyOverviewService.java | 13 +- .../DefaultClassificationPolicySeeder.java | 1 - .../policy/store/InProcessPolicyStore.java | 13 +- .../policy/store/JpaPolicyStore.java | 54 ++- .../policy/store/PolicyEntity.java | 11 +- .../policy/store/PolicyRepository.java | 8 +- .../proprietary/policy/store/PolicyStore.java | 8 +- .../policy/trigger/FolderWatchTrigger.java | 77 ++-- .../policy/trigger/PolicyTrigger.java | 13 +- .../policy/trigger/ScheduleTrigger.java | 44 ++- .../policy/trigger/WebhookTrigger.java | 64 ++-- .../policy/config/FolderAccessGuardTest.java | 10 +- .../policy/config/PolicyAccessGuardTest.java | 2 +- .../controller/PolicyControllerTest.java | 10 +- .../policy/engine/PolicyEngineTest.java | 4 +- .../policy/engine/PolicyRunnerTest.java | 5 +- .../policy/engine/PolicyValidatorTest.java | 59 ++- .../PolicyInlineOutputMigrationTest.java | 3 - .../output/PolicyOutputResolverTest.java | 1 - .../overview/PolicyOverviewServiceTest.java | 13 +- .../s3/EmbeddedS3CredentialMigrationTest.java | 4 +- .../s3/PolicyS3ConnectionUsageCheckTest.java | 1 - ...DefaultClassificationPolicySeederTest.java | 1 - .../policy/source/SourceControllerTest.java | 4 +- .../source/SourceOverviewServiceTest.java | 8 +- .../store/InProcessPolicyStoreTest.java | 20 +- .../policy/store/JpaPolicyStoreTest.java | 54 ++- .../trigger/FolderWatchTriggerTest.java | 99 +++-- .../policy/trigger/ScheduleTriggerTest.java | 87 +++-- .../policy/trigger/WebhookTriggerTest.java | 39 +- .../public/locales/en-US/translation.toml | 8 +- frontend/editor/src/portal/api/pipelines.ts | 18 +- .../pipelines/DestinationPicker.tsx | 47 ++- .../src/portal/mocks/handlers/pipelines.ts | 42 ++- .../src/portal/views/PipelineBuilder.css | 23 ++ .../src/portal/views/PipelineBuilder.test.tsx | 131 +++++-- .../src/portal/views/PipelineBuilder.tsx | 350 +++++++++++------- 43 files changed, 1006 insertions(+), 548 deletions(-) create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/model/PipelineInput.java create mode 100644 app/proprietary/src/main/java/stirling/software/proprietary/policy/model/PolicyBinding.java 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 a34c392597..b6156bc2b8 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 @@ -355,8 +355,7 @@ public class PolicyController { policy.name(), owner, policy.enabled(), - policy.trigger(), - policy.sourceIds(), + policy.inputs(), policy.steps(), policy.output(), policy.outputIds(), 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 index 866fe0910d..837e3bd0ae 100644 --- 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 @@ -15,6 +15,7 @@ import stirling.software.proprietary.policy.input.ResolvedInput; import stirling.software.proprietary.policy.ledger.ProcessedLedger; import stirling.software.proprietary.policy.model.InputSpec; import stirling.software.proprietary.policy.model.PipelineDefinition; +import stirling.software.proprietary.policy.model.PipelineInput; import stirling.software.proprietary.policy.model.Policy; import stirling.software.proprietary.policy.model.PolicyInputs; import stirling.software.proprietary.policy.model.PolicyRun; @@ -42,30 +43,46 @@ public class PolicyRunner { private final SourceDocCounter docCounter; private final ProcessedLedger processedLedger; - /** Full-listing sweep: resolve every source, then reconcile the ledger. */ + /** Full-listing sweep over every input: resolve each source, then reconcile the ledger. */ public SweepOutcome run(Policy policy) { return run(policy, SweepKind.FULL); } - /** - * Trigger entry point. Pulls every referenced source; each yielded unit becomes its own run so - * one failure does not affect the others. No sources means one run with no input (generator - * pipeline). Missing or disabled sources are skipped so one broken reference does not stop the - * rest. Returns the ids of the runs it started plus what the sweep skipped, so a manual trigger - * can report which runs to follow or why nothing ran. - */ + /** Sweep every input of the policy at the given listing depth. */ public SweepOutcome run(Policy policy, SweepKind sweep) { + return run(policy, policy.inputs(), sweep); + } + + /** + * Fire one input binding: a background trigger pulling its own source without touching the + * policy's other inputs. Never reconciles the ledger (it sees a single source, so pruning would + * wrongly forget the rest); a full-policy sweep handles that. + */ + public SweepOutcome runInput(Policy policy, PipelineInput input, SweepKind sweep) { + return run(policy, List.of(input), sweep); + } + + /** + * Core sweep: pulls each of the given inputs' sources; each yielded unit becomes its own run so + * one failure does not affect the others. No inputs means one run with no input (generator + * pipeline). Missing or disabled sources are skipped so one broken reference does not stop the + * rest. Presence cleanup only runs when the sweep covered every input of the policy - a + * single-binding fire cannot reconcile the whole policy's ledger. Returns the ids of the runs + * it started plus what the sweep skipped, so a manual trigger can report which runs to follow + * or why nothing ran. + */ + public SweepOutcome run(Policy policy, List inputs, SweepKind sweep) { long sweepStart = System.currentTimeMillis(); PolicySweep context = new PolicySweep(policy.id(), sweep, processedLedger); List runIds = new ArrayList<>(); - List sourceIds = policy.sourceIds(); - if (sourceIds.isEmpty()) { + if (inputs.isEmpty()) { // Generator pipeline: one run with no input. Still fall through to the cleanup // below so rows recorded for its folder outputs are pruned like anything else, // instead of accumulating until the policy is deleted. runIds.add(startRun(policy, PolicyInputs.of(List.of()), unused -> {})); } - for (String sourceId : sourceIds) { + for (PipelineInput input : inputs) { + String sourceId = input.sourceId(); Source source = sourceStore.get(sourceId).orElse(null); if (source == null) { // No veto: a deleted source's rows should age out via the cleanup below. @@ -84,7 +101,8 @@ public class PolicyRunner { } runIds.addAll(pullAndRun(policy, sourceId, source.toInputSpec(), context)); } - if (context.cleanupAllowed()) { + boolean fullPolicy = inputs.size() == policy.inputs().size(); + if (fullPolicy && context.cleanupAllowed()) { processedLedger.markSeen(policy.id(), context.presentIdentities()); int removed = processedLedger.deleteUnseen(policy.id(), sweepStart); if (removed > 0) { 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 index c2d1357889..b30a760fe3 100644 --- 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 @@ -9,6 +9,7 @@ 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.PipelineInput; import stirling.software.proprietary.policy.model.PipelineStep; import stirling.software.proprietary.policy.model.Policy; import stirling.software.proprietary.policy.model.TriggerConfig; @@ -18,10 +19,11 @@ import stirling.software.proprietary.policy.source.SourceStore; import stirling.software.proprietary.policy.trigger.PolicyTrigger; /** - * Validates a policy at save time by delegating each facet (trigger, sources, steps, output) to the - * bean that handles its type, so a misconfiguration fails fast rather than at run time. A null - * trigger is a manual-only policy and skips trigger validation. Each referenced {@code sourceId} - * must resolve to a persisted {@link Source} whose config its {@link InputSource} bean accepts. + * Validates a policy at save time by delegating each facet (inputs, their triggers, output) to the + * bean that handles its type, so a misconfiguration fails fast rather than at run time. Each + * input's {@code sourceId} must resolve to a persisted {@link Source} whose config its {@link + * InputSource} bean accepts; its optional trigger must be a known type compatible with that source. + * A null trigger is a manual-only input and skips trigger validation. */ @Service @RequiredArgsConstructor @@ -34,21 +36,31 @@ public class PolicyValidator { private final SourceStore sourceStore; /** - * @throws IllegalArgumentException if any facet's type is unknown, a referenced source does not - * exist, or any config is invalid + * @throws IllegalArgumentException if the policy has more than one input or output, any facet's + * type is unknown, a referenced source does not exist, a trigger is incompatible with its + * input's source, or any config is invalid */ public void validate(Policy policy) { - if (policy.trigger() != null) { - triggerFor(policy.trigger()).validate(policy); + // Deliberate product cap, not a model limit: the lists stay lists so multiple + // inputs/outputs can be supported later, but today a policy carries at most one of + // each (zero of either remains fine - run on demand / inline output). + if (policy.inputs().size() > 1) { + throw new IllegalArgumentException("a policy supports at most one input"); } - for (String sourceId : policy.sourceIds()) { + if (policy.outputIds().size() > 1) { + throw new IllegalArgumentException("a policy supports at most one output"); + } + for (PipelineInput input : policy.inputs()) { Source source = sourceStore - .get(sourceId) + .get(input.sourceId()) .orElseThrow( () -> new IllegalArgumentException( - "unknown source: " + sourceId)); + "unknown source: " + input.sourceId())); + if (input.trigger() != null) { + validateTrigger(policy, input, source); + } InputSpec spec = source.toInputSpec(); inputSourceFor(spec).validate(spec); } @@ -73,6 +85,24 @@ public class PolicyValidator { } } + /** + * Check an input's trigger is a known type whose source constraints its source satisfies (e.g. + * folder-watch only on a folder source), then let the trigger validate its own options. + */ + private void validateTrigger(Policy policy, PipelineInput input, Source source) { + PolicyTrigger trigger = triggerFor(input.trigger()); + if (!trigger.supportedSourceTypes().isEmpty() + && !trigger.supportedSourceTypes().contains(source.type())) { + throw new IllegalArgumentException( + "trigger '" + + trigger.type() + + "' is not compatible with source type '" + + source.type() + + "'"); + } + trigger.validate(policy, input); + } + /** * Validate an output spec against its sink. Must be called on a request thread (caller's * principal present) so an S3 output's connection is authorization-checked against the caller - diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/PipelineInput.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/PipelineInput.java new file mode 100644 index 0000000000..15242a9916 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/PipelineInput.java @@ -0,0 +1,16 @@ +package stirling.software.proprietary.policy.model; + +/** + * One input of a policy: a reference to a persisted {@code Source} paired with the {@link + * TriggerConfig} that decides when this source is pulled. The trigger lives on the + * binding, not on the source (so one connection can feed many policies on different schedules) and + * not on the policy (so a folder input can be watched while an S3 input on the same policy polls). + * A {@code null} trigger means this input is pulled only when the policy is run on demand. + */ +public record PipelineInput(String sourceId, TriggerConfig trigger) { + + /** An input with no automatic trigger: pulled only on a manual run. */ + public static PipelineInput manual(String sourceId) { + return new PipelineInput(sourceId, null); + } +} 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 ae9fedc46a..a3d43b712d 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,29 +3,31 @@ package stirling.software.proprietary.policy.model; import java.util.List; /** - * A stored automation: ordered tool steps, input sources, and output destinations. + * A stored automation: ordered tool steps, input bindings, and output destinations. * - *

Always runnable on demand. An optional {@link TriggerConfig} fires it automatically; a {@code - * null} trigger means manual-only. Trigger decides when; {@code sourceIds} reference the persisted - * {@code Source} locations (resolved live at run time) files come from; a run pulls from every - * referenced source. {@code outputIds} reference the {@code Source} locations (resolved live) a - * run's files are delivered to - a run is delivered to every one; when empty the inline {@link - * #output} is used (results returned to the caller), the case for editor and one-off policies. + *

Always runnable on demand. Each {@link PipelineInput} references a persisted {@code Source} + * connection (resolved live at run time) and carries its own optional {@link TriggerConfig}: the + * trigger decides when that source is pulled, so one input can be watched while another polls, and + * a {@code null} trigger makes that input manual-only. An input with no trigger, or a policy with + * no triggered inputs, still runs when the policy is run on demand; a manual run pulls every input. + * + *

{@code outputIds} reference the {@code Source} locations (resolved live) a run's files are + * delivered to - a run is delivered to every one; when empty the inline {@link #output} is used + * (results returned to the caller), the case for editor and one-off policies. */ public record Policy( String id, String name, String owner, boolean enabled, - TriggerConfig trigger, - List sourceIds, + List inputs, List steps, OutputSpec output, List outputIds, Long teamId) { public Policy { - sourceIds = sourceIds == null ? List.of() : List.copyOf(sourceIds); + inputs = inputs == null ? List.of() : List.copyOf(inputs); steps = steps == null ? List.of() : steps; output = output == null ? OutputSpec.inline() : output; outputIds = outputIds == null ? List.of() : List.copyOf(outputIds); @@ -41,12 +43,11 @@ public record Policy( String name, String owner, boolean enabled, - TriggerConfig trigger, - List sourceIds, + List inputs, List steps, OutputSpec output, Long teamId) { - this(id, name, owner, enabled, trigger, sourceIds, steps, output, List.of(), teamId); + this(id, name, owner, enabled, inputs, steps, output, List.of(), teamId); } /** @@ -58,35 +59,35 @@ public record Policy( String name, String owner, boolean enabled, - TriggerConfig trigger, - List sourceIds, + List inputs, List steps, OutputSpec output) { - this(id, name, owner, enabled, trigger, sourceIds, steps, output, List.of(), null); + this(id, name, owner, enabled, inputs, steps, output, List.of(), null); } - /** 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, List.of(), null); + /** The source ids this policy pulls from, in input order; a derived view for reads. */ + public List sourceIds() { + return inputs.stream().map(PipelineInput::sourceId).toList(); + } + + /** The distinct trigger types configured across this policy's inputs (manual inputs aside). */ + public List triggerTypes() { + return inputs.stream() + .map(PipelineInput::trigger) + .filter(trigger -> trigger != null) + .map(TriggerConfig::type) + .distinct() + .toList(); } /** A copy with the inline output replaced (e.g. resolved for the engine, or migrated). */ public Policy withOutput(OutputSpec resolved) { - return new Policy( - id, name, owner, enabled, trigger, sourceIds, steps, resolved, outputIds, teamId); + return new Policy(id, name, owner, enabled, inputs, steps, resolved, outputIds, teamId); } /** A copy referencing the given saved output destinations. */ public Policy withOutputIds(List newOutputIds) { - return new Policy( - id, name, owner, enabled, trigger, sourceIds, steps, output, newOutputIds, teamId); + return new Policy(id, name, owner, enabled, inputs, steps, output, newOutputIds, teamId); } /** diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/PolicyBinding.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/PolicyBinding.java new file mode 100644 index 0000000000..18e93ce1bc --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/PolicyBinding.java @@ -0,0 +1,31 @@ +package stirling.software.proprietary.policy.model; + +import java.util.List; + +/** + * A policy paired with one of its {@link PipelineInput}s: the unit a background trigger fires. A + * policy with two triggered inputs yields two bindings, so each fires independently on its own + * trigger and pulls only its own source. + */ +public record PolicyBinding(Policy policy, PipelineInput input) { + + /** + * The bindings across these policies whose input carries a trigger of the given type. Shared by + * the {@code PolicyStore} implementations so every backend derives a trigger's bindings the + * same way. Callers pass the policies a background trigger should consider (i.e. the enabled + * ones). + */ + public static List matching(List policies, String triggerType) { + return policies.stream() + .flatMap( + policy -> + policy.inputs().stream() + .filter( + input -> + input.trigger() != null + && triggerType.equals( + input.trigger().type())) + .map(input -> new PolicyBinding(policy, input))) + .toList(); + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/overview/PolicyOverviewService.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/overview/PolicyOverviewService.java index c927ec199c..272917ec7f 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/overview/PolicyOverviewService.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/overview/PolicyOverviewService.java @@ -14,7 +14,6 @@ import stirling.software.proprietary.policy.config.PolicyAccessGuard; 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.source.Source; import stirling.software.proprietary.policy.source.SourceAccessGuard; import stirling.software.proprietary.policy.source.SourceStore; @@ -73,7 +72,7 @@ public class PolicyOverviewService { policy.name(), policy.enabled(), policy.enabled() ? "active" : "paused", - triggerSummary(policy.trigger()), + triggerSummary(policy), sources, steps, outputSummary(policy, sourceNames), @@ -95,9 +94,13 @@ public class PolicyOverviewService { return outputSummary(policy.output()); } - /** A null trigger is a manual-only policy; otherwise the trigger's type keys the summary. */ - private static String triggerSummary(TriggerConfig trigger) { - return trigger == null ? "manual" : trigger.type(); + /** + * Summarise a policy's triggers for the overview row: "manual" when no input is triggered, + * otherwise the distinct trigger types across its inputs (e.g. "folder-watch, schedule"). + */ + private static String triggerSummary(Policy policy) { + List types = policy.triggerTypes(); + return types.isEmpty() ? "manual" : String.join(", ", types); } private static String outputSummary(OutputSpec output) { diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/seed/DefaultClassificationPolicySeeder.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/seed/DefaultClassificationPolicySeeder.java index 4150dc2f31..63ebae833a 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/seed/DefaultClassificationPolicySeeder.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/seed/DefaultClassificationPolicySeeder.java @@ -87,7 +87,6 @@ public class DefaultClassificationPolicySeeder { POLICY_NAME, "system", true, - null, List.of(), List.of(new PipelineStep(CLASSIFY_ENDPOINT, Map.of())), new OutputSpec("inline", options), 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 6498e7e28a..de6be8ef5d 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 @@ -9,6 +9,7 @@ import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import stirling.software.proprietary.policy.model.Policy; +import stirling.software.proprietary.policy.model.PolicyBinding; /** * In-memory {@link PolicyStore} for tests and any future no-database mode. {@link JpaPolicyStore} @@ -32,8 +33,7 @@ public class InProcessPolicyStore implements PolicyStore { policy.name(), policy.owner(), policy.enabled(), - policy.trigger(), - policy.sourceIds(), + policy.inputs(), policy.steps(), policy.output(), policy.outputIds(), @@ -77,12 +77,9 @@ public class InProcessPolicyStore implements PolicyStore { } @Override - 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(); + public List findBindingsByTriggerType(String triggerType) { + List enabled = policies.values().stream().filter(Policy::enabled).toList(); + return PolicyBinding.matching(enabled, triggerType); } @Override 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 1b598b88c1..4f91f9e930 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 @@ -12,8 +12,12 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import stirling.software.proprietary.policy.model.Policy; +import stirling.software.proprietary.policy.model.PolicyBinding; +import tools.jackson.databind.JsonNode; import tools.jackson.databind.ObjectMapper; +import tools.jackson.databind.node.ArrayNode; +import tools.jackson.databind.node.ObjectNode; /** * Durable {@link PolicyStore} backed by JPA; the runtime store. Policies are persisted as JSON via @@ -40,8 +44,7 @@ public class JpaPolicyStore implements PolicyStore { policy.name(), policy.owner(), policy.enabled(), - policy.trigger(), - policy.sourceIds(), + policy.inputs(), policy.steps(), policy.output(), policy.outputIds(), @@ -52,7 +55,6 @@ public class JpaPolicyStore implements PolicyStore { entity.setName(stored.name()); entity.setOwner(stored.owner()); entity.setEnabled(stored.enabled()); - entity.setTriggerType(stored.trigger() == null ? null : stored.trigger().type()); entity.setTeamId(stored.teamId()); // Preserve an existing policy's run-order position; append a new one to the end of its // team's queue (max + 1), so setting up a policy adds it last by default. @@ -119,11 +121,13 @@ public class JpaPolicyStore implements PolicyStore { } @Override - public List findByTriggerType(String triggerType) { - return repository.findByTriggerTypeAndEnabledTrue(triggerType).stream() - .map(this::toPolicy) - .flatMap(Optional::stream) - .toList(); + public List findBindingsByTriggerType(String triggerType) { + List enabled = + repository.findByEnabledTrue().stream() + .map(this::toPolicy) + .flatMap(Optional::stream) + .toList(); + return PolicyBinding.matching(enabled, triggerType); } @Override @@ -139,7 +143,8 @@ public class JpaPolicyStore implements PolicyStore { // One unreadable row must never abort a bulk read or crash startup. private Optional toPolicy(PolicyEntity entity) { try { - return Optional.of(objectMapper.readValue(entity.getPolicyJson(), Policy.class)); + JsonNode node = upgradeLegacyShape(objectMapper.readTree(entity.getPolicyJson())); + return Optional.of(objectMapper.treeToValue(node, Policy.class)); } catch (Exception e) { log.error( "Skipping unreadable policy id={} name={}: stored JSON could not be parsed" @@ -150,4 +155,35 @@ public class JpaPolicyStore implements PolicyStore { return Optional.empty(); } } + + /** + * Migrate a policy JSON blob written before triggers moved onto inputs. The old shape carried a + * single policy-level {@code trigger} and a {@code sourceIds} list; pair each source with that + * trigger so an upgraded policy keeps firing. A trigger incompatible with a source + * (folder-watch on an S3 source) is simply inert at run time, matching the old behaviour where + * such a source was never watched. New-shape blobs (already carrying {@code inputs}) are + * returned untouched. + */ + private JsonNode upgradeLegacyShape(JsonNode root) { + if (!(root instanceof ObjectNode obj) || obj.has("inputs")) { + return root; + } + JsonNode trigger = obj.get("trigger"); + JsonNode sourceIds = obj.get("sourceIds"); + ArrayNode inputs = objectMapper.createArrayNode(); + if (sourceIds != null && sourceIds.isArray()) { + for (JsonNode sourceId : sourceIds) { + ObjectNode input = objectMapper.createObjectNode(); + input.set("sourceId", sourceId); + if (trigger != null && !trigger.isNull()) { + input.set("trigger", trigger); + } + inputs.add(input); + } + } + obj.set("inputs", inputs); + obj.remove("trigger"); + obj.remove("sourceIds"); + return obj; + } } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/PolicyEntity.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/PolicyEntity.java index c871267bc4..fa639b5c19 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/PolicyEntity.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/PolicyEntity.java @@ -17,10 +17,10 @@ import stirling.software.proprietary.integration.crypto.LegacyDecryptStringConve /** * JPA row for a {@link stirling.software.proprietary.policy.model.Policy}. The whole policy lives * as JSON in {@code policyJson} (authoritative on read); the scalar columns are denormalized copies - * for querying, notably {@code triggerType} + {@code enabled} so background triggers can fetch - * their policies, and {@code teamId} so the caller's team can be loaded without scanning every - * team's rows. {@code owner} and {@code teamId} are plain values, not foreign keys, to stay - * decoupled from the security entities. + * for querying, notably {@code enabled} so background triggers can scan the active policies, and + * {@code teamId} so the caller's team can be loaded without scanning every team's rows. {@code + * owner} and {@code teamId} are plain values, not foreign keys, to stay decoupled from the security + * entities. */ @Entity @Table(name = "policies") @@ -44,9 +44,6 @@ public class PolicyEntity implements Serializable { @Column(name = "enabled") private boolean enabled; - @Column(name = "trigger_type") - private String triggerType; - @Column(name = "team_id") private Long teamId; diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/PolicyRepository.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/PolicyRepository.java index 82ae4355dd..1c8465c82a 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/PolicyRepository.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/PolicyRepository.java @@ -13,8 +13,12 @@ import jakarta.persistence.LockModeType; @Repository public interface PolicyRepository extends JpaRepository { - /** Enabled policies of a given trigger type, for background triggers to activate. */ - List findByTriggerTypeAndEnabledTrue(String triggerType); + /** + * Enabled policies, for background triggers to scan for inputs of their trigger type. Which + * inputs (and their trigger types) a policy carries lives in the JSON blob, so the type filter + * is applied after parsing rather than in SQL. + */ + List findByEnabledTrue(); /** * Policies belonging to a team, in run order (ascending {@code sortOrder}; a null order sorts diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/PolicyStore.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/PolicyStore.java index ac9e210439..fa2a7e9b87 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/PolicyStore.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/store/PolicyStore.java @@ -4,6 +4,7 @@ import java.util.List; import java.util.Optional; import stirling.software.proprietary.policy.model.Policy; +import stirling.software.proprietary.policy.model.PolicyBinding; /** Stores {@link Policy} definitions. */ public interface PolicyStore { @@ -18,8 +19,11 @@ public interface PolicyStore { /** Policies owned by the given team, loaded scoped rather than fetched globally. */ List findByTeam(Long teamId); - /** Enabled policies with the given trigger type, for background triggers. */ - List findByTriggerType(String triggerType); + /** + * Enabled inputs with the given trigger type, as {@code (policy, input)} bindings, so a + * background trigger fires each input independently and pulls only its own source. + */ + List findBindingsByTriggerType(String triggerType); /** * Set the team's run order from {@code orderedIds} (position → sortOrder). Only policies that 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 index a4470a418b..5b859a4786 100644 --- 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 @@ -31,7 +31,9 @@ import stirling.software.proprietary.policy.engine.PolicyRunner; import stirling.software.proprietary.policy.engine.SweepKind; import stirling.software.proprietary.policy.input.InputSource; import stirling.software.proprietary.policy.model.InputSpec; +import stirling.software.proprietary.policy.model.PipelineInput; import stirling.software.proprietary.policy.model.Policy; +import stirling.software.proprietary.policy.model.PolicyBinding; import stirling.software.proprietary.policy.source.Source; import stirling.software.proprietary.policy.source.SourceStore; import stirling.software.proprietary.policy.store.PolicyStore; @@ -85,10 +87,10 @@ public class FolderWatchTrigger implements PolicyTrigger { } @Override - public void validate(Policy policy) { - if (watchDirsOf(policy).isEmpty()) { + public void validate(Policy policy, PipelineInput input) { + if (watchDirsOf(input).isEmpty()) { throw new IllegalArgumentException( - "folder-watch trigger requires at least one watchable (folder) input source"); + "folder-watch trigger requires a watchable (folder) input source"); } } @@ -185,24 +187,30 @@ public class FolderWatchTrigger implements PolicyTrigger { return changed; } - /** Run every folder-watch policy that draws from one of the changed directories. */ + /** Fire every folder-watch input that draws from one of the changed directories. */ void runForChangedDirs(Set changedDirs) { if (changedDirs.isEmpty()) { return; } - for (Policy policy : policyStore.findByTriggerType(TYPE)) { + for (PolicyBinding binding : policyStore.findBindingsByTriggerType(TYPE)) { List dirs; try { - dirs = watchDirsOf(policy); + dirs = watchDirsOf(binding.input()); } catch (RuntimeException e) { log.warn( - "Folder-watch policy {} is misconfigured: {}", policy.id(), e.getMessage()); + "Folder-watch input {}/{} is misconfigured: {}", + binding.policy().id(), + binding.input().sourceId(), + e.getMessage()); continue; } if (dirs.stream().anyMatch(changedDirs::contains)) { - log.debug("Folder-watch policy {} ({}) saw activity", policy.id(), policy.name()); + log.debug( + "Folder-watch input {}/{} saw activity", + binding.policy().id(), + binding.input().sourceId()); // Light: the periodic reconcile does the full sweep. - policyRunner.run(policy, SweepKind.LIGHT); + policyRunner.runInput(binding.policy(), binding.input(), SweepKind.LIGHT); } } } @@ -216,15 +224,16 @@ public class FolderWatchTrigger implements PolicyTrigger { } } - /** Reconcile safety net: run every folder-watch policy regardless of watch events. */ + /** Reconcile safety net: run every folder-watch input regardless of watch events. */ void runAll() { - for (Policy policy : policyStore.findByTriggerType(TYPE)) { + for (PolicyBinding binding : policyStore.findBindingsByTriggerType(TYPE)) { try { - policyRunner.run(policy); + policyRunner.runInput(binding.policy(), binding.input(), SweepKind.FULL); } catch (RuntimeException e) { log.warn( - "Folder-watch reconcile run failed for policy {}: {}", - policy.id(), + "Folder-watch reconcile run failed for input {}/{}: {}", + binding.policy().id(), + binding.input().sourceId(), e.getMessage()); } } @@ -269,41 +278,43 @@ public class FolderWatchTrigger implements PolicyTrigger { return Set.copyOf(keysByDir.keySet()); } - /** Every existing directory any current folder-watch policy wants watched. */ + /** Every existing directory any current folder-watch input wants watched. */ private Set desiredDirs() { Set dirs = new HashSet<>(); - for (Policy policy : policyStore.findByTriggerType(TYPE)) { + for (PolicyBinding binding : policyStore.findBindingsByTriggerType(TYPE)) { try { - for (Path dir : watchDirsOf(policy)) { + for (Path dir : watchDirsOf(binding.input())) { if (Files.isDirectory(dir)) { dirs.add(dir); } } } catch (RuntimeException e) { log.warn( - "Folder-watch policy {} is misconfigured: {}", policy.id(), e.getMessage()); + "Folder-watch input {}/{} is misconfigured: {}", + binding.policy().id(), + binding.input().sourceId(), + e.getMessage()); } } return dirs; } // Absolute + normalised so registration keys and event-time matching compare regardless of how - // the path was configured. - private List watchDirsOf(Policy policy) { + // the path was configured. Empty for a non-folder or missing source (that input is never + // watched), so a folder-watch trigger paired with an S3 input is simply inert. + private List watchDirsOf(PipelineInput input) { List dirs = new ArrayList<>(); - for (String sourceId : policy.sourceIds()) { - Source source = sourceStore.get(sourceId).orElse(null); - if (source == null) { - continue; - } - InputSpec spec = source.toInputSpec(); - InputSource inputSource = sourceFor(spec); - if (inputSource == null) { - continue; - } - for (Path dir : inputSource.watchTargets(spec)) { - dirs.add(dir.toAbsolutePath().normalize()); - } + Source source = sourceStore.get(input.sourceId()).orElse(null); + if (source == null) { + return dirs; + } + InputSpec spec = source.toInputSpec(); + InputSource inputSource = sourceFor(spec); + if (inputSource == null) { + return dirs; + } + for (Path dir : inputSource.watchTargets(spec)) { + dirs.add(dir.toAbsolutePath().normalize()); } return dirs; } 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 ade5357162..e1e0cc7873 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 @@ -2,11 +2,13 @@ package stirling.software.proprietary.policy.trigger; import java.util.Set; +import stirling.software.proprietary.policy.model.PipelineInput; import stirling.software.proprietary.policy.model.Policy; /** - * Decides when a policy runs. On firing it hands the policy to {@code PolicyRunner}; it - * never resolves sources itself. New trigger kinds are just new beans of this type. + * Decides when a policy input runs. On firing it hands the binding to {@code + * PolicyRunner}, which pulls only that input's source; it never resolves sources itself. New + * trigger kinds are just new beans of this type. */ public interface PolicyTrigger { @@ -32,10 +34,11 @@ public interface PolicyTrigger { } /** - * Validate at save time so misconfiguration fails fast, not at fire time. Receives the whole - * {@link Policy} so triggers that depend on the policy's sources (folder-watch) can check that. + * Validate one input's use of this trigger at save time so misconfiguration fails fast, not at + * fire time. Receives the owning {@link Policy} and the specific {@link PipelineInput} so a + * trigger that depends on the input's source (folder-watch) can check it. */ - default void validate(Policy policy) {} + default void validate(Policy policy, PipelineInput input) {} default void start() {} 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 index 2018ffd126..f4ffb30146 100644 --- 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 @@ -17,14 +17,18 @@ import lombok.extern.slf4j.Slf4j; import stirling.software.common.model.ApplicationProperties; import stirling.software.proprietary.policy.engine.PolicyRunner; +import stirling.software.proprietary.policy.engine.SweepKind; +import stirling.software.proprietary.policy.model.PipelineInput; import stirling.software.proprietary.policy.model.Policy; +import stirling.software.proprietary.policy.model.PolicyBinding; 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}: a fixed-interval sweep runs each due "schedule" policy. + * Fires policy inputs on a {@link Schedule}: a fixed-interval sweep pulls each due "schedule" + * input, independently of the policy's other inputs. * *

Last-fire times are in memory, so this assumes a single node and resets on restart. */ @@ -40,17 +44,20 @@ public class ScheduleTrigger implements PolicyTrigger { private final ObjectMapper objectMapper; private final ApplicationProperties applicationProperties; - private final Map lastFiredByPolicy = new ConcurrentHashMap<>(); + private final Map lastFiredByBinding = new ConcurrentHashMap<>(); private volatile ScheduledExecutorService scheduler; + /** Identifies a schedule binding: one input (by source) of one policy. */ + private record BindingKey(String policyId, String sourceId) {} + @Override public String type() { return TYPE; } @Override - public void validate(Policy policy) { - ScheduleConfig.from(objectMapper, policy.trigger().options()); + public void validate(Policy policy, PipelineInput input) { + ScheduleConfig.from(objectMapper, input.trigger().options()); } @Override @@ -83,19 +90,26 @@ public class ScheduleTrigger implements PolicyTrigger { } } - /** Fire every scheduled policy that is due as of {@code now}. Package-visible for testing. */ + /** Fire every scheduled input that is due as of {@code now}. Package-visible for testing. */ void sweep(Instant now) { - for (Policy policy : policyStore.findByTriggerType(TYPE)) { + for (PolicyBinding binding : policyStore.findBindingsByTriggerType(TYPE)) { + Policy policy = binding.policy(); + PipelineInput input = binding.input(); ScheduleConfig config; try { - config = ScheduleConfig.from(objectMapper, policy.trigger().options()); + config = ScheduleConfig.from(objectMapper, input.trigger().options()); } catch (IllegalArgumentException e) { - log.warn("Scheduled policy {} is misconfigured: {}", policy.id(), e.getMessage()); + log.warn( + "Scheduled input {}/{} is misconfigured: {}", + policy.id(), + input.sourceId(), + e.getMessage()); continue; } - // Baseline a newly-seen policy to now so it does not fire immediately. - Instant last = lastFiredByPolicy.computeIfAbsent(policy.id(), id -> now); + // Baseline a newly-seen binding to now so it does not fire immediately. + BindingKey key = new BindingKey(policy.id(), input.sourceId()); + Instant last = lastFiredByBinding.computeIfAbsent(key, id -> now); ZonedDateTime next = config.schedule().nextAfter(last.atZone(config.zone())); if (next.toInstant().isAfter(now)) { continue; @@ -105,9 +119,13 @@ public class ScheduleTrigger implements PolicyTrigger { next = later; later = config.schedule().nextAfter(later); } - lastFiredByPolicy.put(policy.id(), next.toInstant()); - log.info("Scheduled policy {} ({}) is due", policy.id(), policy.name()); - policyRunner.run(policy); + lastFiredByBinding.put(key, next.toInstant()); + log.info( + "Scheduled input {}/{} ({}) is due", + policy.id(), + input.sourceId(), + policy.name()); + policyRunner.runInput(policy, input, SweepKind.FULL); } } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/WebhookTrigger.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/WebhookTrigger.java index 816f510bba..d5bb412a7c 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/WebhookTrigger.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/WebhookTrigger.java @@ -13,7 +13,9 @@ import lombok.extern.slf4j.Slf4j; import stirling.software.common.model.ApplicationProperties; import stirling.software.proprietary.policy.engine.PolicyRunner; import stirling.software.proprietary.policy.engine.SweepKind; +import stirling.software.proprietary.policy.model.PipelineInput; import stirling.software.proprietary.policy.model.Policy; +import stirling.software.proprietary.policy.model.PolicyBinding; import stirling.software.proprietary.policy.source.Source; import stirling.software.proprietary.policy.source.SourceStore; import stirling.software.proprietary.policy.store.PolicyStore; @@ -50,15 +52,14 @@ public class WebhookTrigger implements PolicyTrigger { } @Override - public void validate(Policy policy) { - boolean hasWebhookSource = - policy.sourceIds().stream() - .map(sourceStore::get) - .flatMap(java.util.Optional::stream) - .anyMatch(source -> WEBHOOK_SOURCE_TYPE.equals(source.type())); - if (!hasWebhookSource) { - throw new IllegalArgumentException( - "webhook trigger requires at least one webhook input source"); + public void validate(Policy policy, PipelineInput input) { + boolean isWebhookSource = + sourceStore + .get(input.sourceId()) + .filter(source -> WEBHOOK_SOURCE_TYPE.equals(source.type())) + .isPresent(); + if (!isWebhookSource) { + throw new IllegalArgumentException("webhook trigger requires a webhook input source"); } } @@ -83,29 +84,38 @@ public class WebhookTrigger implements PolicyTrigger { } } + /** Fire every webhook input fed by this webhook, pulling only that input's source. */ public void fireForWebhook(String webhookId) { - for (Policy policy : policyStore.findByTriggerType(TYPE)) { - if (!referencesWebhook(policy, webhookId)) { + for (PolicyBinding binding : policyStore.findBindingsByTriggerType(TYPE)) { + if (!referencesWebhook(binding.input(), webhookId)) { continue; } try { - log.debug("Webhook policy {} ({}) saw a delivery", policy.id(), policy.name()); - policyRunner.run(policy, SweepKind.LIGHT); + log.debug( + "Webhook input {}/{} saw a delivery", + binding.policy().id(), + binding.input().sourceId()); + policyRunner.runInput(binding.policy(), binding.input(), SweepKind.LIGHT); } catch (RuntimeException e) { - log.warn("Webhook run failed for policy {}: {}", policy.id(), e.getMessage()); + log.warn( + "Webhook run failed for input {}/{}: {}", + binding.policy().id(), + binding.input().sourceId(), + e.getMessage()); } } } private void safeReconcile() { try { - for (Policy policy : policyStore.findByTriggerType(TYPE)) { + for (PolicyBinding binding : policyStore.findBindingsByTriggerType(TYPE)) { try { - policyRunner.run(policy); + policyRunner.runInput(binding.policy(), binding.input(), SweepKind.FULL); } catch (RuntimeException e) { log.warn( - "Webhook reconcile run failed for policy {}: {}", - policy.id(), + "Webhook reconcile run failed for input {}/{}: {}", + binding.policy().id(), + binding.input().sourceId(), e.getMessage()); } } @@ -114,17 +124,13 @@ public class WebhookTrigger implements PolicyTrigger { } } - private boolean referencesWebhook(Policy policy, String webhookId) { - for (String sourceId : policy.sourceIds()) { - Source source = sourceStore.get(sourceId).orElse(null); - if (source == null || !WEBHOOK_SOURCE_TYPE.equals(source.type())) { - continue; - } - Object configured = source.options().get(WebhookConfig.WEBHOOK_ID_OPTION); - if (configured != null && configured.toString().equals(webhookId)) { - return true; - } + /** Whether this input draws from the webhook source the delivery arrived on. */ + private boolean referencesWebhook(PipelineInput input, String webhookId) { + Source source = sourceStore.get(input.sourceId()).orElse(null); + if (source == null || !WEBHOOK_SOURCE_TYPE.equals(source.type())) { + return false; } - return false; + Object configured = source.options().get(WebhookConfig.WEBHOOK_ID_OPTION); + return configured != null && configured.toString().equals(webhookId); } } 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 index 351dcd1d2f..431d74d8b8 100644 --- 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 @@ -17,6 +17,7 @@ import stirling.software.common.configuration.RuntimePathConfig; 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.PipelineInput; import stirling.software.proprietary.policy.model.Policy; import stirling.software.proprietary.policy.source.InProcessSourceStore; import stirling.software.proprietary.policy.source.Source; @@ -189,6 +190,13 @@ class FolderAccessGuardTest { null)) .id()) .toList(); - return new Policy("p1", "p", "owner", true, null, sourceIds, List.of(), output); + return new Policy( + "p1", + "p", + "owner", + true, + sourceIds.stream().map(PipelineInput::manual).toList(), + List.of(), + output); } } diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/config/PolicyAccessGuardTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/config/PolicyAccessGuardTest.java index 35c4d2631e..c3f6552dc5 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/config/PolicyAccessGuardTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/config/PolicyAccessGuardTest.java @@ -90,6 +90,6 @@ class PolicyAccessGuardTest { private static Policy inTeam(Long teamId) { return new Policy( - null, "p", "owner", true, null, List.of(), List.of(), OutputSpec.inline(), teamId); + null, "p", "owner", true, List.of(), List.of(), OutputSpec.inline(), teamId); } } diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/controller/PolicyControllerTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/controller/PolicyControllerTest.java index 351eecd581..c29c96a8ed 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/controller/PolicyControllerTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/controller/PolicyControllerTest.java @@ -139,7 +139,7 @@ class PolicyControllerTest { } private static Policy policy(String id, Long teamId) { - return new Policy(id, "name", "owner", true, null, List.of(), List.of(), null, teamId); + return new Policy(id, "name", "owner", true, List.of(), List.of(), null, teamId); } private static Policy s3OutputPolicy(String id, String secret) { @@ -150,7 +150,7 @@ class PolicyControllerTest { "bucket", "outbox", "accessKeyId", "AKIAEXAMPLE", "secretAccessKey", secret)); - return new Policy(id, "name", "owner", true, null, List.of(), List.of(), output, 1L); + return new Policy(id, "name", "owner", true, List.of(), List.of(), output, 1L); } private static PolicyRunHandle handle(String runId) { @@ -401,8 +401,7 @@ class PolicyControllerTest { void updatePreservesOwnership() { applicationProperties.getSecurity().setEnableLogin(false); Policy existing = - new Policy( - "p2", "name", "origOwner", true, null, List.of(), List.of(), null, 3L); + new Policy("p2", "name", "origOwner", true, List.of(), List.of(), null, 3L); when(policyStore.get("p2")).thenReturn(Optional.of(existing)); when(policyAccessGuard.canAccess(existing)).thenReturn(true); when(policyStore.save(any())).thenAnswer(i -> i.getArgument(0)); @@ -410,8 +409,7 @@ class PolicyControllerTest { ResponseEntity response = controller.savePolicy( new Policy( - "p2", "name", "forged", true, null, List.of(), List.of(), null, - 77L)); + "p2", "name", "forged", true, List.of(), List.of(), null, 77L)); assertThat(response.getBody().owner()).isEqualTo("origOwner"); assertThat(response.getBody().teamId()).isEqualTo(3L); 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 e69f0f5476..560c80d7cd 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 @@ -262,7 +262,7 @@ class PolicyEngineTest { "rotate", "owner", true, - null, + List.of(), List.of(new PipelineStep(ROTATE, Map.of())), OutputSpec.inline()); @@ -306,7 +306,7 @@ class PolicyEngineTest { "rotate", "alice", true, - null, + List.of(), List.of(new PipelineStep(ROTATE, Map.of())), OutputSpec.inline()); 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 index e21f371305..51959a2525 100644 --- 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 @@ -36,6 +36,7 @@ import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger; import stirling.software.proprietary.policy.ledger.ProcessedLedger; import stirling.software.proprietary.policy.model.InputSpec; import stirling.software.proprietary.policy.model.OutputSpec; +import stirling.software.proprietary.policy.model.PipelineInput; import stirling.software.proprietary.policy.model.PipelineStep; import stirling.software.proprietary.policy.model.Policy; import stirling.software.proprietary.policy.model.PolicyInputs; @@ -306,7 +307,6 @@ class PolicyRunnerTest { "p", "owner", true, - null, List.of(), List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), OutputSpec.inline(), @@ -338,8 +338,7 @@ class PolicyRunnerTest { "p", "owner", true, - null, - sourceIds, + sourceIds.stream().map(PipelineInput::manual).toList(), 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 index 2e9613a024..6440cefa4a 100644 --- 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 @@ -20,6 +20,7 @@ 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.PipelineInput; import stirling.software.proprietary.policy.model.Policy; import stirling.software.proprietary.policy.model.TriggerConfig; import stirling.software.proprietary.policy.output.PolicyOutputSink; @@ -60,25 +61,27 @@ class PolicyValidatorTest { validator.validate(policy); - verify(trigger).validate(policy); + verify(trigger).validate(policy, policy.inputs().get(0)); verify(inputSource).validate(InputSpec.folder("/in")); verify(outputSink).validate(policy.output()); } @Test - void skipsTriggerValidationForAManualOnlyPolicy() { + void skipsTriggerValidationForAManualOnlyInput() { when(inputSource.supports(any())).thenReturn(true); when(outputSink.supports(any())).thenReturn(true); validator.validate(manualOnly()); - verify(trigger, never()).validate(any()); + verify(trigger, never()).validate(any(), any()); } @Test void surfacesAnInvalidConfigFromAHandler() { when(trigger.type()).thenReturn("schedule"); - doThrow(new IllegalArgumentException("invalid schedule")).when(trigger).validate(any()); + doThrow(new IllegalArgumentException("invalid schedule")) + .when(trigger) + .validate(any(), any()); IllegalArgumentException ex = assertThrows( @@ -120,14 +123,55 @@ class PolicyValidatorTest { assertTrue(ex.getMessage().contains("unknown trigger type")); } + // The one-input/one-output caps are a product decision, not a model limit: the lists stay so + // multiple can be supported later, but saving more than one of either is rejected today. + + @Test + void rejectsMoreThanOneInput() { + Policy twoInputs = + new Policy( + "p1", + "p", + "owner", + true, + List.of( + PipelineInput.manual(folderSourceId()), + PipelineInput.manual(folderSourceId())), + List.of(), + OutputSpec.inline()); + + IllegalArgumentException ex = + assertThrows(IllegalArgumentException.class, () -> validator.validate(twoInputs)); + assertTrue(ex.getMessage().contains("at most one input")); + } + + @Test + void rejectsMoreThanOneOutput() { + Policy twoOutputs = manualOnly().withOutputIds(List.of("out-a", "out-b")); + + IllegalArgumentException ex = + assertThrows(IllegalArgumentException.class, () -> validator.validate(twoOutputs)); + assertTrue(ex.getMessage().contains("at most one output")); + } + + @Test + void allowsZeroInputsAndZeroOutputs() { + when(outputSink.supports(any())).thenReturn(true); + Policy bare = + new Policy("p1", "p", "owner", true, List.of(), List.of(), OutputSpec.inline()); + + validator.validate(bare); + } + private Policy policy(String triggerType) { return new Policy( "p1", "p", "owner", true, - new TriggerConfig(triggerType, Map.of()), - List.of(folderSourceId()), + List.of( + new PipelineInput( + folderSourceId(), new TriggerConfig(triggerType, Map.of()))), List.of(), OutputSpec.inline()); } @@ -138,8 +182,7 @@ class PolicyValidatorTest { "p", "owner", true, - null, - List.of(folderSourceId()), + List.of(PipelineInput.manual(folderSourceId())), List.of(), OutputSpec.inline()); } diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/PolicyInlineOutputMigrationTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/PolicyInlineOutputMigrationTest.java index 082ca8cd42..8cd91c56e9 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/PolicyInlineOutputMigrationTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/PolicyInlineOutputMigrationTest.java @@ -54,7 +54,6 @@ class PolicyInlineOutputMigrationTest { "Editor run", "owner", true, - null, List.of(), List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), OutputSpec.inline())); @@ -128,7 +127,6 @@ class PolicyInlineOutputMigrationTest { name, "owner", true, - null, List.of(), List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), OutputSpec.folder(directory)); @@ -140,7 +138,6 @@ class PolicyInlineOutputMigrationTest { name, "owner", true, - null, List.of(), List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), OutputSpec.folder(directory), diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/PolicyOutputResolverTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/PolicyOutputResolverTest.java index d65c5b24f2..acaa68dae1 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/PolicyOutputResolverTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/PolicyOutputResolverTest.java @@ -65,7 +65,6 @@ class PolicyOutputResolverTest { "Pipeline", "owner", true, - null, List.of(), 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/overview/PolicyOverviewServiceTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/overview/PolicyOverviewServiceTest.java index 221cbc4482..8610926c2a 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/overview/PolicyOverviewServiceTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/overview/PolicyOverviewServiceTest.java @@ -16,6 +16,7 @@ import stirling.software.common.service.UserServiceInterface; import stirling.software.proprietary.policy.config.PolicyAccessGuard; import stirling.software.proprietary.policy.config.PolicyManagementAuthority; import stirling.software.proprietary.policy.model.OutputSpec; +import stirling.software.proprietary.policy.model.PipelineInput; import stirling.software.proprietary.policy.model.PipelineStep; import stirling.software.proprietary.policy.model.Policy; import stirling.software.proprietary.policy.model.TriggerConfig; @@ -57,8 +58,9 @@ class PolicyOverviewServiceTest { "Redaction", "owner", true, - new TriggerConfig("schedule", Map.of()), - List.of(claims.id()), + List.of( + new PipelineInput( + claims.id(), new TriggerConfig("schedule", Map.of()))), List.of(new PipelineStep("/api/v1/security/auto-redact", Map.of())), OutputSpec.inline())); policyStore.save( @@ -67,7 +69,6 @@ class PolicyOverviewServiceTest { "Archive (paused)", "owner", false, - null, List.of(), List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), OutputSpec.inline())); @@ -102,8 +103,7 @@ class PolicyOverviewServiceTest { "Orphan", "owner", true, - null, - List.of("src-missing"), + List.of(PipelineInput.manual("src-missing")), List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), OutputSpec.inline())); @@ -170,8 +170,7 @@ class PolicyOverviewServiceTest { name, "owner", true, - null, - List.of(sourceIds), + List.of(sourceIds).stream().map(PipelineInput::manual).toList(), List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), OutputSpec.inline(), teamId)); diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/s3/EmbeddedS3CredentialMigrationTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/s3/EmbeddedS3CredentialMigrationTest.java index a794ee104e..41f80461cd 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/s3/EmbeddedS3CredentialMigrationTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/s3/EmbeddedS3CredentialMigrationTest.java @@ -25,6 +25,7 @@ import stirling.software.proprietary.integration.repository.IntegrationConfigRep import stirling.software.proprietary.model.Team; import stirling.software.proprietary.policy.migration.InProcessCompletedMigrations; import stirling.software.proprietary.policy.model.OutputSpec; +import stirling.software.proprietary.policy.model.PipelineInput; import stirling.software.proprietary.policy.model.PipelineStep; import stirling.software.proprietary.policy.model.Policy; import stirling.software.proprietary.policy.source.InProcessSourceStore; @@ -97,8 +98,7 @@ class EmbeddedS3CredentialMigrationTest { "Rotate", "alice", true, - null, - List.of(source.id()), + List.of(PipelineInput.manual(source.id())), List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), new OutputSpec( "s3", diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/s3/PolicyS3ConnectionUsageCheckTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/s3/PolicyS3ConnectionUsageCheckTest.java index 847a553ba6..4af725428d 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/s3/PolicyS3ConnectionUsageCheckTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/s3/PolicyS3ConnectionUsageCheckTest.java @@ -39,7 +39,6 @@ class PolicyS3ConnectionUsageCheckTest { "Rotate", "alice", true, - null, List.of(), List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), new OutputSpec("s3", Map.of("connectionId", "5")), diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/seed/DefaultClassificationPolicySeederTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/seed/DefaultClassificationPolicySeederTest.java index 2f270afddb..49159e7d6e 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/seed/DefaultClassificationPolicySeederTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/seed/DefaultClassificationPolicySeederTest.java @@ -41,7 +41,6 @@ class DefaultClassificationPolicySeederTest { "Classification Policy", "system", true, - null, List.of(), List.of(), new OutputSpec("inline", Map.of("categoryId", "classification")), diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceControllerTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceControllerTest.java index 545ea8e50f..20f783f75c 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceControllerTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceControllerTest.java @@ -30,6 +30,7 @@ import stirling.software.proprietary.policy.config.PolicyManagementAuthority; import stirling.software.proprietary.policy.input.InputSource; import stirling.software.proprietary.policy.input.WebhookInputSource; import stirling.software.proprietary.policy.model.OutputSpec; +import stirling.software.proprietary.policy.model.PipelineInput; import stirling.software.proprietary.policy.model.PipelineStep; import stirling.software.proprietary.policy.model.Policy; import stirling.software.proprietary.policy.store.InProcessPolicyStore; @@ -274,8 +275,7 @@ class SourceControllerTest { name, "owner", true, - null, - List.of(sourceId), + List.of(PipelineInput.manual(sourceId)), 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/source/SourceOverviewServiceTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceOverviewServiceTest.java index 3f63e18407..a8c295acc8 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceOverviewServiceTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceOverviewServiceTest.java @@ -16,6 +16,7 @@ import stirling.software.common.service.UserServiceInterface; import stirling.software.proprietary.policy.config.PolicyAccessGuard; import stirling.software.proprietary.policy.config.PolicyManagementAuthority; import stirling.software.proprietary.policy.model.OutputSpec; +import stirling.software.proprietary.policy.model.PipelineInput; import stirling.software.proprietary.policy.model.PipelineStep; import stirling.software.proprietary.policy.model.Policy; import stirling.software.proprietary.policy.store.InProcessPolicyStore; @@ -216,8 +217,7 @@ class SourceOverviewServiceTest { name, "owner", true, - null, - List.of(sourceIds), + List.of(sourceIds).stream().map(PipelineInput::manual).toList(), List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), OutputSpec.inline())); } @@ -232,7 +232,6 @@ class SourceOverviewServiceTest { name, "owner", true, - null, List.of(), List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), new OutputSpec("inline", Map.of("sources", List.of("editor"))))); @@ -245,8 +244,7 @@ class SourceOverviewServiceTest { name, "owner", true, - null, - List.of(sourceIds), + List.of(sourceIds).stream().map(PipelineInput::manual).toList(), List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), OutputSpec.inline(), teamId)); 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 60f14be20d..f74a4cfb5b 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 @@ -12,11 +12,13 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import stirling.software.proprietary.policy.model.OutputSpec; +import stirling.software.proprietary.policy.model.PipelineInput; import stirling.software.proprietary.policy.model.PipelineStep; import stirling.software.proprietary.policy.model.Policy; +import stirling.software.proprietary.policy.model.PolicyBinding; import stirling.software.proprietary.policy.model.TriggerConfig; -/** Tests for {@link InProcessPolicyStore}: id assignment, upsert, trigger-type lookup, delete. */ +/** Tests for {@link InProcessPolicyStore}: id assignment, upsert, binding lookup, delete. */ class InProcessPolicyStoreTest { private PolicyStore store; @@ -45,7 +47,7 @@ class InProcessPolicyStoreTest { "after", "owner", true, - null, + List.of(), List.of(), OutputSpec.inline())); @@ -54,16 +56,16 @@ class InProcessPolicyStoreTest { } @Test - void findByTriggerTypeReturnsOnlyEnabledMatches() { + void findBindingsByTriggerTypeReturnsOnlyEnabledMatches() { 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 scheduled = store.findByTriggerType("schedule"); + List scheduled = store.findBindingsByTriggerType("schedule"); assertEquals(1, scheduled.size()); - assertEquals("nightly", scheduled.get(0).name()); + assertEquals("nightly", scheduled.get(0).policy().name()); } @Test @@ -76,14 +78,16 @@ class InProcessPolicyStoreTest { } private static Policy policy(String id, String name, String triggerType, boolean enabled) { - TriggerConfig trigger = - triggerType == null ? null : new TriggerConfig(triggerType, Map.of()); + PipelineInput input = + triggerType == null + ? PipelineInput.manual("src") + : new PipelineInput("src", new TriggerConfig(triggerType, Map.of())); return new Policy( id, name, "owner", enabled, - trigger, + List.of(input), 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 e9667f2e8d..2a1d2b4f11 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 @@ -19,8 +19,10 @@ import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; import stirling.software.proprietary.policy.model.OutputSpec; +import stirling.software.proprietary.policy.model.PipelineInput; import stirling.software.proprietary.policy.model.PipelineStep; import stirling.software.proprietary.policy.model.Policy; +import stirling.software.proprietary.policy.model.PolicyBinding; import stirling.software.proprietary.policy.model.TriggerConfig; import tools.jackson.databind.ObjectMapper; @@ -53,8 +55,9 @@ class JpaPolicyStoreTest { "compress incoming", "alice", true, - new TriggerConfig("schedule", Map.of()), - List.of("src-in"), + List.of( + new PipelineInput( + "src-in", new TriggerConfig("schedule", Map.of()))), List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), OutputSpec.inline())); @@ -63,7 +66,6 @@ class JpaPolicyStoreTest { verify(repository).save(captor.capture()); PolicyEntity entity = captor.getValue(); assertEquals(saved.id(), entity.getId()); - 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)); @@ -77,7 +79,7 @@ class JpaPolicyStoreTest { "rotate", "alice", true, - null, // manual-only: no automatic trigger + List.of(), // no inputs: run on demand only List.of( new PipelineStep( "/api/v1/general/rotate-pdf", Map.of("angle", 90))), @@ -87,6 +89,30 @@ class JpaPolicyStoreTest { assertEquals(policy, store.get("p1").orElseThrow()); } + @Test + void getUpgradesLegacyTriggerAndSourceIdsToPerInputTriggers() { + // A blob written before triggers moved onto inputs: one policy-level trigger + sourceIds. + String legacyJson = + "{\"id\":\"p1\",\"name\":\"legacy\",\"owner\":\"alice\",\"enabled\":true," + + "\"trigger\":{\"type\":\"schedule\",\"options\":{}}," + + "\"sourceIds\":[\"s1\",\"s2\"],\"steps\":[]," + + "\"output\":{\"type\":\"inline\",\"options\":{}}}"; + PolicyEntity entity = new PolicyEntity(); + entity.setId("p1"); + entity.setName("legacy"); + entity.setEnabled(true); + entity.setPolicyJson(legacyJson); + when(repository.findById("p1")).thenReturn(Optional.of(entity)); + + Policy upgraded = store.get("p1").orElseThrow(); + + assertEquals( + List.of( + new PipelineInput("s1", new TriggerConfig("schedule", Map.of())), + new PipelineInput("s2", new TriggerConfig("schedule", Map.of()))), + upgraded.inputs()); + } + @Test void saveDenormalizesTeamIdForScopedQueries() { store.save( @@ -95,7 +121,6 @@ class JpaPolicyStoreTest { "scoped", "alice", true, - null, List.of(), List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), OutputSpec.inline(), @@ -114,7 +139,6 @@ class JpaPolicyStoreTest { "ours", "alice", true, - null, List.of(), List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), OutputSpec.inline(), @@ -128,23 +152,28 @@ class JpaPolicyStoreTest { } @Test - void findByTriggerTypeUsesTheEnabledQuery() { + void findBindingsByTriggerTypeScansEnabledPoliciesForMatchingInputs() { Policy policy = new Policy( "p1", "watch", "alice", true, - new TriggerConfig("schedule", Map.of()), + List.of( + new PipelineInput( + "src-in", new TriggerConfig("schedule", Map.of())), + PipelineInput.manual("src-manual")), List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), OutputSpec.inline()); - when(repository.findByTriggerTypeAndEnabledTrue("schedule")) - .thenReturn(List.of(entityFor(policy))); + when(repository.findByEnabledTrue()).thenReturn(List.of(entityFor(policy))); - List scheduled = store.findByTriggerType("schedule"); + List scheduled = store.findBindingsByTriggerType("schedule"); + // Only the scheduled input yields a binding; the manual input on the same policy does not. assertEquals(1, scheduled.size()); - assertEquals("p1", scheduled.get(0).id()); + assertEquals("p1", scheduled.get(0).policy().id()); + assertEquals("src-in", scheduled.get(0).input().sourceId()); + assertEquals("schedule", scheduled.get(0).input().trigger().type()); } @Test @@ -163,7 +192,6 @@ class JpaPolicyStoreTest { entity.setName(policy.name()); entity.setOwner(policy.owner()); entity.setEnabled(policy.enabled()); - entity.setTriggerType(policy.trigger() == null ? null : policy.trigger().type()); entity.setTeamId(policy.teamId()); 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 index 2c3adf691e..c955e9ed8d 100644 --- 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 @@ -14,6 +14,7 @@ import java.nio.file.FileSystems; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.WatchService; +import java.util.Arrays; import java.util.List; import java.util.Map; import java.util.Set; @@ -31,8 +32,10 @@ import stirling.software.proprietary.policy.engine.SweepKind; 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.PipelineInput; import stirling.software.proprietary.policy.model.PipelineStep; import stirling.software.proprietary.policy.model.Policy; +import stirling.software.proprietary.policy.model.PolicyBinding; import stirling.software.proprietary.policy.model.TriggerConfig; import stirling.software.proprietary.policy.source.InProcessSourceStore; import stirling.software.proprietary.policy.source.Source; @@ -41,10 +44,10 @@ 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}. + * runForChangedDirs}/{@code runAll}, plus its per-input 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 { @@ -83,39 +86,43 @@ class FolderWatchTriggerTest { } @Test - void validateRejectsPolicyWithNoWatchableSource() { + void validateRejectsInputWithNoWatchableSource() { + PolicyBinding binding = + bindings(folderWatch("p1", List.of(new InputSpec("folder", Map.of())))).get(0); assertThrows( IllegalArgumentException.class, - () -> trigger.validate(folderWatch("p1", List.of()))); + () -> trigger.validate(binding.policy(), binding.input())); } @Test - void validateAcceptsPolicyWithAFolderSource() { - trigger.validate(folderWatch("p1", List.of(InputSpec.folder("/in")))); + void validateAcceptsAFolderSource() { + PolicyBinding binding = + bindings(folderWatch("p1", List.of(InputSpec.folder("/in")))).get(0); + trigger.validate(binding.policy(), binding.input()); } @Test - void runsOnlyPoliciesDrawingFromTheChangedDirectory() { + void runsOnlyInputsDrawingFromTheChangedDirectory() { 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)); + when(policyStore.findBindingsByTriggerType("folder-watch")).thenReturn(bindings(a, b)); trigger.runForChangedDirs(Set.of(normalized("/in/a"))); - verify(policyRunner).run(a, SweepKind.LIGHT); - verify(policyRunner, never()).run(eq(b), any()); + verify(policyRunner).runInput(a, a.inputs().get(0), SweepKind.LIGHT); + verify(policyRunner, never()).runInput(eq(b), any(), any()); } @Test - void skipsAMisconfiguredPolicyButStillRunsTheOthers() { + void skipsAMisconfiguredInputButStillRunsTheOthers() { 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)); + when(policyStore.findBindingsByTriggerType("folder-watch")).thenReturn(bindings(bad, good)); trigger.runForChangedDirs(Set.of(normalized("/in/a"))); - verify(policyRunner).run(good, SweepKind.LIGHT); - verify(policyRunner, never()).run(eq(bad), any()); + verify(policyRunner).runInput(good, good.inputs().get(0), SweepKind.LIGHT); + verify(policyRunner, never()).runInput(eq(bad), any(), any()); } @Test @@ -126,15 +133,15 @@ class FolderWatchTriggerTest { } @Test - void reconcileRunsEveryFolderWatchPolicyAsASafetyNet() { + void reconcileRunsEveryFolderWatchInputAsASafetyNet() { 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)); + when(policyStore.findBindingsByTriggerType("folder-watch")).thenReturn(bindings(a, b)); trigger.runAll(); - verify(policyRunner).run(a); - verify(policyRunner).run(b); + verify(policyRunner).runInput(a, a.inputs().get(0), SweepKind.FULL); + verify(policyRunner).runInput(b, b.inputs().get(0), SweepKind.FULL); } @Test @@ -151,15 +158,16 @@ class FolderWatchTriggerTest { try { trigger.watchService = service; - when(policyStore.findByTriggerType("folder-watch")).thenReturn(List.of(a, b, m)); + when(policyStore.findBindingsByTriggerType("folder-watch")) + .thenReturn(bindings(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)); + // b's input is removed: its registration is cancelled, a remains. + when(policyStore.findBindingsByTriggerType("folder-watch")).thenReturn(bindings(a)); trigger.syncRegistrations(); assertEquals(Set.of(normalized(dirA.toString())), trigger.watchedDirs()); } finally { @@ -175,15 +183,15 @@ class FolderWatchTriggerTest { WatchService service = FileSystems.getDefault().newWatchService(); try { trigger.watchService = service; - when(policyStore.findByTriggerType("folder-watch")).thenReturn(List.of(p)); + when(policyStore.findBindingsByTriggerType("folder-watch")).thenReturn(bindings(p)); - // The mutation hook registers the new policy's directory without waiting for a + // The mutation hook registers the new input's directory without waiting for a // reconcile. trigger.onPoliciesChanged(); assertEquals(Set.of(normalized(dir.toString())), trigger.watchedDirs()); - // Once the policy is gone, the same hook cancels its registration. - when(policyStore.findByTriggerType("folder-watch")).thenReturn(List.of()); + // Once the input is gone, the same hook cancels its registration. + when(policyStore.findBindingsByTriggerType("folder-watch")).thenReturn(List.of()); trigger.onPoliciesChanged(); assertEquals(Set.of(), trigger.watchedDirs()); } finally { @@ -195,31 +203,40 @@ class FolderWatchTriggerTest { return Path.of(dir).toAbsolutePath().normalize(); } + /** Every (policy, input) binding across the given policies, as the store would return them. */ + private static List bindings(Policy... policies) { + return Arrays.stream(policies) + .flatMap( + policy -> policy.inputs().stream().map(in -> new PolicyBinding(policy, in))) + .toList(); + } + /** Persists each spec as a source and returns a folder-watch policy referencing them by id. */ private Policy folderWatch(String id, List sources) { - List sourceIds = + List inputs = sources.stream() .map( spec -> - sourceStore - .save( - new Source( - null, - "src", - spec.type(), - spec.options(), - true, - "owner", - null)) - .id()) + new PipelineInput( + sourceStore + .save( + new Source( + null, + "src", + spec.type(), + spec.options(), + true, + "owner", + null)) + .id(), + new TriggerConfig("folder-watch", Map.of()))) .toList(); return new Policy( id, "watcher", "owner", true, - new TriggerConfig("folder-watch", Map.of()), - sourceIds, + inputs, 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/ScheduleTriggerTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/ScheduleTriggerTest.java index 3f81392285..b7b7dccc9a 100644 --- 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 @@ -25,8 +25,10 @@ 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.PipelineInput; import stirling.software.proprietary.policy.model.PipelineStep; import stirling.software.proprietary.policy.model.Policy; +import stirling.software.proprietary.policy.model.PolicyBinding; import stirling.software.proprietary.policy.model.Schedule; import stirling.software.proprietary.policy.model.TriggerConfig; import stirling.software.proprietary.policy.store.PolicyStore; @@ -35,9 +37,9 @@ 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. + * sweep(Instant)}. The trigger only decides when an input is due; pulling the source and starting + * runs is the {@link PolicyRunner}'s job, so these assert it delegates to the runner per binding. + * Schedules default to UTC, so explicit UTC instants make these deterministic. */ @ExtendWith(MockitoExtension.class) class ScheduleTriggerTest { @@ -59,102 +61,105 @@ class ScheduleTriggerTest { @Test void firesOncePerScheduleWhenItComesDue() { - Policy policy = scheduled("p1", new Schedule.Every(1, Schedule.Unit.MINUTES)); - when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy)); + PolicyBinding binding = scheduled("p1", new Schedule.Every(1, Schedule.Unit.MINUTES)); + when(policyStore.findBindingsByTriggerType("schedule")).thenReturn(List.of(binding)); Instant t0 = Instant.parse("2026-06-05T10:00:30Z"); trigger.sweep(t0); // first sight: baseline, must not fire immediately - verify(policyRunner, never()).run(any()); + verify(policyRunner, never()).runInput(any(), any(), any()); trigger.sweep(t0.plusSeconds(120)); // the one-minute mark has passed - verify(policyRunner, times(1)).run(eq(policy)); + verify(policyRunner, times(1)).runInput(eq(binding.policy()), eq(binding.input()), any()); } @Test void anIntervalMatchingTheSweepPeriodFiresEverySweepDespiteJitter() { - Policy policy = scheduled("p1", new Schedule.Every(1, Schedule.Unit.MINUTES)); - when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy)); + PolicyBinding binding = scheduled("p1", new Schedule.Every(1, Schedule.Unit.MINUTES)); + when(policyStore.findBindingsByTriggerType("schedule")).thenReturn(List.of(binding)); Instant t0 = Instant.parse("2026-06-05T10:00:00Z"); trigger.sweep(t0); // baseline // The sweep that fires runs a few ms late (scheduler jitter)... trigger.sweep(t0.plusSeconds(60).plusMillis(5)); - verify(policyRunner, times(1)).run(eq(policy)); + verify(policyRunner, times(1)).runInput(eq(binding.policy()), eq(binding.input()), any()); // ...and the next sweep lands exactly on the 60s grid. Anchoring lastFired to the due // time (not the jittered observation) means this must still fire, not alias to skip. trigger.sweep(t0.plusSeconds(120)); - verify(policyRunner, times(2)).run(eq(policy)); + verify(policyRunner, times(2)).runInput(eq(binding.policy()), eq(binding.input()), any()); } @Test void aGapFiresOnceNotOncePerMissedInterval() { - Policy policy = scheduled("p1", new Schedule.Every(1, Schedule.Unit.MINUTES)); - when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy)); + PolicyBinding binding = scheduled("p1", new Schedule.Every(1, Schedule.Unit.MINUTES)); + when(policyStore.findBindingsByTriggerType("schedule")).thenReturn(List.of(binding)); Instant t0 = Instant.parse("2026-06-05T10:00:00Z"); trigger.sweep(t0); // baseline // Ten minutes of downtime: nine missed due points collapse into one firing. trigger.sweep(t0.plusSeconds(600)); - verify(policyRunner, times(1)).run(eq(policy)); + verify(policyRunner, times(1)).runInput(eq(binding.policy()), eq(binding.input()), any()); // Not due again until a full interval after the latest due point. trigger.sweep(t0.plusSeconds(630)); - verify(policyRunner, times(1)).run(eq(policy)); + verify(policyRunner, times(1)).runInput(eq(binding.policy()), eq(binding.input()), any()); trigger.sweep(t0.plusSeconds(660)); - verify(policyRunner, times(2)).run(eq(policy)); + verify(policyRunner, times(2)).runInput(eq(binding.policy()), eq(binding.input()), any()); } @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)); + PolicyBinding binding = + scheduled("p1", new Schedule.Daily(LocalTime.of(3, 0))); // 03:00 UTC daily + when(policyStore.findBindingsByTriggerType("schedule")).thenReturn(List.of(binding)); 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()); + verify(policyRunner, never()).runInput(any(), any(), any()); } @Test void firesWeeklyOnAChosenDay() { // 2026-06-05 is a Friday; the next Monday 09:00 is the soonest firing. - Policy policy = + PolicyBinding binding = scheduled("p1", new Schedule.Weekly(Set.of(DayOfWeek.MONDAY), LocalTime.of(9, 0))); - when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy)); + when(policyStore.findBindingsByTriggerType("schedule")).thenReturn(List.of(binding)); 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)); + verify(policyRunner, times(1)).runInput(eq(binding.policy()), eq(binding.input()), any()); } @Test - void skipsPoliciesWithAnInvalidSchedule() { - Policy policy = scheduledWithRawOptions("p1", Map.of()); // no schedule - when(policyStore.findByTriggerType("schedule")).thenReturn(List.of(policy)); + void skipsInputsWithAnInvalidSchedule() { + PolicyBinding binding = scheduledWithRawOptions("p1", Map.of()); // no schedule + when(policyStore.findBindingsByTriggerType("schedule")).thenReturn(List.of(binding)); trigger.sweep(Instant.parse("2026-06-05T10:00:00Z")); - verify(policyRunner, never()).run(any()); + verify(policyRunner, never()).runInput(any(), any(), any()); } @Test void validateRejectsMissingSchedule() { + PolicyBinding binding = scheduledWithRawOptions("p1", Map.of()); assertThrows( IllegalArgumentException.class, - () -> trigger.validate(scheduledWithRawOptions("p1", Map.of()))); + () -> trigger.validate(binding.policy(), binding.input())); } @Test void validateRejectsAnInvalidSchedule() { Map options = Map.of("schedule", Map.of("type", "every", "count", -5, "unit", "MINUTES")); + PolicyBinding binding = scheduledWithRawOptions("p1", options); assertThrows( IllegalArgumentException.class, - () -> trigger.validate(scheduledWithRawOptions("p1", options))); + () -> trigger.validate(binding.policy(), binding.input())); } @Test @@ -162,21 +167,25 @@ class ScheduleTriggerTest { Map options = new LinkedHashMap<>(); options.put("schedule", new Schedule.Daily(LocalTime.of(2, 0))); options.put("zone", "Europe/London"); - trigger.validate(scheduledWithRawOptions("p1", options)); + PolicyBinding binding = scheduledWithRawOptions("p1", options); + trigger.validate(binding.policy(), binding.input()); } - private static Policy scheduled(String id, Schedule schedule) { + private static PolicyBinding 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()); + private static PolicyBinding scheduledWithRawOptions(String id, Map options) { + PipelineInput input = new PipelineInput("s1", new TriggerConfig("schedule", options)); + Policy policy = + new Policy( + id, + "nightly", + "owner", + true, + List.of(input), + List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), + OutputSpec.inline()); + return new PolicyBinding(policy, input); } } diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/WebhookTriggerTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/WebhookTriggerTest.java index 63b721dd6d..a5b06bde01 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/WebhookTriggerTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/WebhookTriggerTest.java @@ -2,10 +2,12 @@ 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.verify; import static org.mockito.Mockito.when; +import java.util.Arrays; import java.util.List; import java.util.Map; @@ -19,8 +21,10 @@ import stirling.software.common.model.ApplicationProperties; import stirling.software.proprietary.policy.engine.PolicyRunner; import stirling.software.proprietary.policy.engine.SweepKind; import stirling.software.proprietary.policy.model.OutputSpec; +import stirling.software.proprietary.policy.model.PipelineInput; import stirling.software.proprietary.policy.model.PipelineStep; import stirling.software.proprietary.policy.model.Policy; +import stirling.software.proprietary.policy.model.PolicyBinding; import stirling.software.proprietary.policy.model.TriggerConfig; import stirling.software.proprietary.policy.source.InProcessSourceStore; import stirling.software.proprietary.policy.source.Source; @@ -46,37 +50,45 @@ class WebhookTriggerTest { } @Test - void firesOnlyPoliciesReferencingTheDeliveredWebhook() { + void firesOnlyInputsReferencingTheDeliveredWebhook() { Policy matching = webhookPolicy("a", "whkA"); Policy other = webhookPolicy("b", "whkB"); - when(policyStore.findByTriggerType(TYPE)).thenReturn(List.of(matching, other)); + when(policyStore.findBindingsByTriggerType(TYPE)).thenReturn(bindings(matching, other)); trigger.fireForWebhook("whkA"); - verify(policyRunner).run(matching, SweepKind.LIGHT); - verify(policyRunner, never()).run(other, SweepKind.LIGHT); + verify(policyRunner).runInput(matching, matching.inputs().get(0), SweepKind.LIGHT); + verify(policyRunner, never()).runInput(eq(other), any(), any()); } @Test void ignoresADeliveryForAnUnknownWebhookId() { Policy policy = webhookPolicy("a", "whkA"); - when(policyStore.findByTriggerType(TYPE)).thenReturn(List.of(policy)); + when(policyStore.findBindingsByTriggerType(TYPE)).thenReturn(bindings(policy)); trigger.fireForWebhook("whkZ"); - verify(policyRunner, never()).run(any(), any(SweepKind.class)); + verify(policyRunner, never()).runInput(any(), any(), any()); } @Test void validateRequiresAWebhookSource() { + // A non-webhook source (here: an id that resolves to nothing) is rejected. + Policy notWebhook = policy("p", PipelineInput.manual("missing-source")); assertThrows( IllegalArgumentException.class, - () -> trigger.validate(policy("p", webhookTriggerConfig(), List.of()))); - trigger.validate(webhookPolicy("p", "whkA")); + () -> trigger.validate(notWebhook, notWebhook.inputs().get(0))); + + Policy hooked = webhookPolicy("p", "whkA"); + trigger.validate(hooked, hooked.inputs().get(0)); } - private static TriggerConfig webhookTriggerConfig() { - return new TriggerConfig(TYPE, Map.of()); + /** Every (policy, input) binding across the given policies, as the store would return them. */ + private static List bindings(Policy... policies) { + return Arrays.stream(policies) + .flatMap( + policy -> policy.inputs().stream().map(in -> new PolicyBinding(policy, in))) + .toList(); } private Policy webhookPolicy(String id, String webhookId) { @@ -98,17 +110,16 @@ class WebhookTriggerTest { "owner", null)) .id(); - return policy(id, webhookTriggerConfig(), List.of(sourceId)); + return policy(id, new PipelineInput(sourceId, new TriggerConfig(TYPE, Map.of()))); } - private static Policy policy(String id, TriggerConfig trigger, List sourceIds) { + private static Policy policy(String id, PipelineInput input) { return new Policy( id, "hook", "owner", true, - trigger, - sourceIds, + List.of(input), List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), OutputSpec.inline()); } diff --git a/frontend/editor/public/locales/en-US/translation.toml b/frontend/editor/public/locales/en-US/translation.toml index abbe1446ff..d02e60b84c 100644 --- a/frontend/editor/public/locales/en-US/translation.toml +++ b/frontend/editor/public/locales/en-US/translation.toml @@ -7745,11 +7745,17 @@ newPipeline = "New pipeline" addStep = "Add tool" back = "Back to pipelines" chooseAccount = "Choose an account" +chooseDestination = "Choose a destination" chooseOperation = "Choose what this step does" +chooseSource = "Choose a source" discard = "Discard changes" enabled = "Enabled" +inputs = "Input" +inputSource = "Input source" +inputTrigger = "Trigger" keepEditing = "Keep editing" needsUpload = "Needs an uploaded file" +noSources = "No sources connected yet. Create one to use it as an input." noToolMatches = "No tools match your search." pipelineSettings = "Pipeline settings" searchTools = "Search tools" @@ -7777,7 +7783,7 @@ namePlaceholder = "e.g. Redaction sweep" noToolSettings = "This tool has no configurable settings." operations_one = "Operation ({{count}})" operations_other = "Operations ({{count}})" -output = "Destinations" +output = "Destination" removeStep = "Remove operation" save = "Save changes" scheduleEvery = "Run every" diff --git a/frontend/editor/src/portal/api/pipelines.ts b/frontend/editor/src/portal/api/pipelines.ts index 2a4f25b79e..d6088d06ff 100644 --- a/frontend/editor/src/portal/api/pipelines.ts +++ b/frontend/editor/src/portal/api/pipelines.ts @@ -17,12 +17,22 @@ export interface PipelineStep { fileParameters?: Record; } -/** When a policy fires automatically. `type` keys a trigger bean (e.g. "schedule"). */ +/** When a policy input fires automatically. `type` keys a trigger bean (e.g. "schedule"). */ export interface TriggerConfig { type: string; options: Record; } +/** + * One input of a pipeline: a persisted source paired with the trigger that decides when that + * source is pulled. A `null` trigger means the input is pulled only on a manual run. Mirrors the + * backend `PipelineInput`. + */ +export interface PipelineInput { + sourceId: string; + trigger: TriggerConfig | null; +} + /** Where a run's outputs are delivered. `type` keys an output sink (e.g. "inline"). */ export interface OutputSpec { type: string; @@ -35,15 +45,15 @@ export type PipelineOutputMode = "folder" | "s3"; /** * The stored policy record: the create/update body (`id` blank on create) and what * the backend returns from GET/POST. Mirrors Policy.java exactly; `owner`/`teamId` - * are stamped server-side. A `null` trigger means manual-only. + * are stamped server-side. Each input pairs a source with its own trigger; an input + * with a `null` trigger (or a policy with no triggered inputs) runs only on demand. */ export interface Policy { id?: string; name: string; owner?: string | null; enabled: boolean; - trigger: TriggerConfig | null; - sourceIds: string[]; + inputs: PipelineInput[]; steps: PipelineStep[]; /** * Inline output, used only when no destinations are referenced (editor/one-off runs that return diff --git a/frontend/editor/src/portal/components/pipelines/DestinationPicker.tsx b/frontend/editor/src/portal/components/pipelines/DestinationPicker.tsx index 683c6301b4..de1187c3da 100644 --- a/frontend/editor/src/portal/components/pipelines/DestinationPicker.tsx +++ b/frontend/editor/src/portal/components/pipelines/DestinationPicker.tsx @@ -1,14 +1,15 @@ import { useTranslation } from "react-i18next"; import AddRoundedIcon from "@mui/icons-material/AddRounded"; -import { Button, Checkbox } from "@app/ui"; +import { Button, Select } from "@app/ui"; /** - * Picks the saved sources a pipeline delivers its output to. A destination is just - * a source used as a write target, and a pipeline may write to several, so this is - * a checklist over the same locations the builder loaded (filtered to writable - * types by the caller) - mirroring the input-sources checklist. Creating a new one - * is delegated to {@code onCreateNew} (the builder navigates to the source builder, - * prompting about unsaved edits first). + * Picks the saved source a pipeline delivers its output to. A destination is just a + * source used as a write target. The value stays a list ({@code outputIds}) because + * the model supports several, but the product caps a pipeline at one destination + * today, so this renders a single dropdown over the same locations the builder + * loaded (filtered to writable types by the caller). Creating a new one is delegated + * to {@code onCreateNew} (the builder navigates to the source builder, prompting + * about unsaved edits first). */ interface DestinationOption { id: string; @@ -31,23 +32,21 @@ export function DestinationPicker({ }: DestinationPickerProps) { const { t } = useTranslation(); - function toggle(id: string, checked: boolean) { - onChange( - checked ? [...value, id] : value.filter((existing) => existing !== id), - ); - } - return ( - <> -

- {sources.map((source) => ( - toggle(source.id, e.target.checked)} - label={source.name} - /> - ))} +
+
+ setScheduleCount(e.target.value)} - className="portal-pipelines__schedule-count" - /> - changeInputSource(value ?? "")} + options={sourceOptions} + /> +
+
+ + updateInput({ scheduleCount: e.target.value }) + } + className="portal-pipelines__schedule-count" + /> +