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" + /> +