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 06172b8cbc..a34c392597 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 @@ -68,6 +68,7 @@ import stirling.software.proprietary.policy.overview.PoliciesOverviewResponse; import stirling.software.proprietary.policy.overview.PolicyOverviewService; import stirling.software.proprietary.policy.progress.PolicyProgressListener; import stirling.software.proprietary.policy.source.EditorSource; +import stirling.software.proprietary.policy.source.Source; import stirling.software.proprietary.policy.source.SourceAccessGuard; import stirling.software.proprietary.policy.source.SourceDocCounter; import stirling.software.proprietary.policy.source.SourceStore; @@ -248,6 +249,7 @@ public class PolicyController { requirePolicyEditingAllowed(); Policy owned = withStoredOutputSecrets(resolveOwnership(policy)); requireAccessibleSources(owned); + requireAccessibleOutput(owned); try { policyValidator.validate(owned); } catch (IllegalArgumentException e) { @@ -290,6 +292,40 @@ public class PolicyController { } } + /** + * A policy's output destination is a {@link Source} used as a write target: it must resolve to + * a source in the caller's team, so a client can neither reference a non-existent location nor + * reach across teams to write to another team's. The editor is virtual and has no writable + * location, so it can't be a destination. The config is then validated on this (request) thread + * so an S3 destination's connection is authorization-checked against the caller - the async + * delivery worker has no principal. A policy with no reference (inline / editor / one-off) has + * nothing to check. + */ + private void requireAccessibleOutput(Policy policy) { + for (String outputId : policy.outputIds()) { + Source destination = + sourceStore + .get(outputId) + .filter(sourceAccessGuard::canAccess) + .orElseThrow( + () -> + new ResponseStatusException( + HttpStatus.BAD_REQUEST, + "Unknown or inaccessible output source: " + + outputId)); + if (EditorSource.TYPE.equals(destination.type())) { + throw new ResponseStatusException( + HttpStatus.BAD_REQUEST, + "The editor can't be used as an output destination"); + } + try { + policyValidator.validateOutput(destination.toOutputSpec()); + } catch (IllegalArgumentException e) { + throw new ResponseStatusException(HttpStatus.BAD_REQUEST, e.getMessage()); + } + } + } + /** * Assign owner + owning team server-side. Create stamps the current user and their team; update * preserves the existing owner and team after verifying the policy belongs to the caller's team @@ -323,6 +359,7 @@ public class PolicyController { policy.sourceIds(), policy.steps(), policy.output(), + policy.outputIds(), teamId); } @@ -358,16 +395,7 @@ public class PolicyController { } private static Policy withOutput(Policy policy, OutputSpec output) { - return new Policy( - policy.id(), - policy.name(), - policy.owner(), - policy.enabled(), - policy.trigger(), - policy.sourceIds(), - policy.steps(), - output, - policy.teamId()); + return policy.withOutput(output); } /** @@ -573,8 +601,9 @@ public class PolicyController { // step dereferences its connection by id on a principal-less worker thread, so this // request thread is the only place that reference can be checked against the caller. policyValidator.validateSteps(definition.steps()); - if (definition.output() != null) { - policyValidator.validateOutput(definition.output()); + // Every destination is checked; an ad-hoc run with no destinations validates nothing. + for (OutputSpec output : definition.outputs()) { + policyValidator.validateOutput(output); } } catch (IllegalArgumentException e) { throw new ResponseStatusException(HttpStatus.BAD_REQUEST, e.getMessage()); diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyEngine.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyEngine.java index abdb69cb07..e6dce0ee7b 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyEngine.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyEngine.java @@ -37,6 +37,7 @@ import stirling.software.proprietary.policy.model.PolicyInputs; import stirling.software.proprietary.policy.model.PolicyRun; import stirling.software.proprietary.policy.model.WaitState; import stirling.software.proprietary.policy.output.OutputDelivery; +import stirling.software.proprietary.policy.output.PolicyOutputResolver; import stirling.software.proprietary.policy.output.PolicyOutputSink; import stirling.software.proprietary.policy.progress.PolicyProgressListener; import stirling.software.proprietary.service.DownstreamEntitlementError; @@ -72,6 +73,7 @@ public class PolicyEngine { private final FileStorage fileStorage; private final JobOwnershipService jobOwnershipService; private final List outputSinks; + private final PolicyOutputResolver outputResolver; private final ResourceMonitor resourceMonitor; private final JobQueue jobQueue; @@ -119,8 +121,14 @@ public class PolicyEngine { // the owner owns those outputs. String triggeringUser = currentActingPrincipal(); String fileOwner = triggeringUser != null ? triggeringUser : policy.owner(); + // Resolve the referenced output destinations live (like sourceIds), so a stored policy + // delivers to each of its saved Source destinations. Unreferenced policies fall back to + // their inline output. + PipelineDefinition definition = + new PipelineDefinition( + policy.name(), policy.steps(), outputResolver.resolve(policy)); return submitForPrincipal( - policy.owner(), fileOwner, policy.id(), policy.toDefinition(), inputs, listener); + policy.owner(), fileOwner, policy.id(), definition, inputs, listener); } private PolicyRunHandle submitForPrincipal( @@ -212,13 +220,21 @@ public class PolicyEngine { run.markRunning(); PolicyExecutionResult result = stepExecutor.execute(run.getDefinition(), inputs, listener); - OutputSpec output = run.getDefinition().output(); - List outputs = - sinkFor(output) - .deliver( - new OutputDelivery(runId, run.getPolicyId()), - result.files(), - output); + // Deliver the run's files to every destination; no destinations means inline + // delivery (results stored/returned to the caller), preserving ad-hoc/AI behaviour. + List destinations = run.getDefinition().outputs(); + if (destinations.isEmpty()) { + destinations = List.of(OutputSpec.inline()); + } + List outputs = new ArrayList<>(); + for (OutputSpec destination : destinations) { + outputs.addAll( + sinkFor(destination) + .deliver( + new OutputDelivery(runId, run.getPolicyId()), + result.files(), + destination)); + } taskManager.setMultipleFileResults(runId, outputs); taskManager.setComplete(runId); run.complete(outputs); diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/migration/CompletedMigration.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/migration/CompletedMigration.java new file mode 100644 index 0000000000..9837cd4f45 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/migration/CompletedMigration.java @@ -0,0 +1,40 @@ +package stirling.software.proprietary.policy.migration; + +import java.io.Serializable; +import java.time.Instant; + +import jakarta.persistence.Column; +import jakarta.persistence.Entity; +import jakarta.persistence.Id; +import jakarta.persistence.Table; + +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.Setter; + +/** + * A one-time policy-subsystem migration that has finished, keyed by a stable migration id. Its + * presence lets a migration skip its (otherwise every-boot) scan once it has run, instead of + * re-scanning and finding nothing to do forever. + */ +@Entity +@Table(name = "policy_completed_migrations") +@NoArgsConstructor +@Getter +@Setter +public class CompletedMigration implements Serializable { + + private static final long serialVersionUID = 1L; + + @Id + @Column(name = "id") + private String id; + + @Column(name = "applied_at") + private Instant appliedAt; + + public CompletedMigration(String id, Instant appliedAt) { + this.id = id; + this.appliedAt = appliedAt; + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/migration/CompletedMigrationRepository.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/migration/CompletedMigrationRepository.java new file mode 100644 index 0000000000..23dec902c5 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/migration/CompletedMigrationRepository.java @@ -0,0 +1,7 @@ +package stirling.software.proprietary.policy.migration; + +import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.stereotype.Repository; + +@Repository +public interface CompletedMigrationRepository extends JpaRepository {} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/migration/CompletedMigrations.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/migration/CompletedMigrations.java new file mode 100644 index 0000000000..f03defa92b --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/migration/CompletedMigrations.java @@ -0,0 +1,18 @@ +package stirling.software.proprietary.policy.migration; + +/** + * Tracks which one-time policy-subsystem migrations have finished, so a migration can skip its + * every-boot scan once done. {@link JpaCompletedMigrations} is the runtime bean; {@link + * InProcessCompletedMigrations} backs tests. + */ +public interface CompletedMigrations { + + /** Whether the migration with this id has already been recorded as complete. */ + boolean isDone(String id); + + /** + * Record the migration as complete. Safe to call concurrently: a race on first boot leaves the + * marker recorded exactly once and never propagates a failure to the caller. + */ + void markDone(String id); +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/migration/InProcessCompletedMigrations.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/migration/InProcessCompletedMigrations.java new file mode 100644 index 0000000000..0744c85965 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/migration/InProcessCompletedMigrations.java @@ -0,0 +1,23 @@ +package stirling.software.proprietary.policy.migration; + +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +/** + * In-memory {@link CompletedMigrations} for tests and any future no-database mode. {@link + * JpaCompletedMigrations} is the runtime bean. + */ +public class InProcessCompletedMigrations implements CompletedMigrations { + + private final Set done = ConcurrentHashMap.newKeySet(); + + @Override + public boolean isDone(String id) { + return done.contains(id); + } + + @Override + public void markDone(String id) { + done.add(id); + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/migration/JpaCompletedMigrations.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/migration/JpaCompletedMigrations.java new file mode 100644 index 0000000000..853e18be51 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/migration/JpaCompletedMigrations.java @@ -0,0 +1,38 @@ +package stirling.software.proprietary.policy.migration; + +import java.time.Instant; + +import org.springframework.dao.DataIntegrityViolationException; +import org.springframework.stereotype.Service; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +/** + * Durable {@link CompletedMigrations} backed by JPA; the runtime bean. {@code markDone} relies on + * the primary-key uniqueness of {@link CompletedMigration#getId()} to stay safe under a concurrent + * first boot: whichever node inserts first wins, and the loser's duplicate insert is swallowed + * rather than propagated, so it never disturbs the migration that called it. + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class JpaCompletedMigrations implements CompletedMigrations { + + private final CompletedMigrationRepository repository; + + @Override + public boolean isDone(String id) { + return repository.existsById(id); + } + + @Override + public void markDone(String id) { + try { + repository.save(new CompletedMigration(id, Instant.now())); + } catch (DataIntegrityViolationException alreadyRecorded) { + // A concurrent boot recorded the same marker first; the row exists, so we are done. + log.debug("Completion marker '{}' was already recorded concurrently", id); + } + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/PipelineDefinition.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/PipelineDefinition.java index 146424b0fb..209756c67f 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/PipelineDefinition.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/model/PipelineDefinition.java @@ -3,13 +3,20 @@ package stirling.software.proprietary.policy.model; import java.util.List; /** - * An ordered chain of tool steps plus an output destination; the unit the engine executes. + * An ordered chain of tool steps plus its output destinations; the unit the engine executes. * - *

{@code output} may be null for callers that handle result files themselves (e.g. the AI - * workflow, which builds its own response payload). + *

{@code outputs} may be empty for callers that handle result files themselves (e.g. the AI + * workflow, which builds its own response payload) - the engine then falls back to inline delivery. + * A run's files are delivered to every destination in the list. */ -public record PipelineDefinition(String name, List steps, OutputSpec output) { +public record PipelineDefinition(String name, List steps, List outputs) { public PipelineDefinition { steps = steps == null ? List.of() : steps; + outputs = outputs == null ? List.of() : List.copyOf(outputs); + } + + /** Convenience for the common single-destination (or inline) case. A null output is empty. */ + public PipelineDefinition(String name, List steps, OutputSpec output) { + this(name, steps, output == null ? List.of() : List.of(output)); } } 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 51dcc4ac0c..ae9fedc46a 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,12 +3,14 @@ package stirling.software.proprietary.policy.model; import java.util.List; /** - * A stored automation: ordered tool steps, input sources, and an output destination. + * A stored automation: ordered tool steps, input sources, 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} connections (resolved live at run time) that decide where files come from; a run - * pulls from every referenced source. + * {@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. */ public record Policy( String id, @@ -19,12 +21,32 @@ public record Policy( List sourceIds, List steps, OutputSpec output, + List outputIds, Long teamId) { public Policy { sourceIds = sourceIds == null ? List.of() : List.copyOf(sourceIds); steps = steps == null ? List.of() : steps; output = output == null ? OutputSpec.inline() : output; + outputIds = outputIds == null ? List.of() : List.copyOf(outputIds); + } + + /** + * Without output references: the inline output is used as-is. Kept for the engine, migrations, + * and tests, and for editor/one-off policies that return results to the caller rather than a + * stored destination. + */ + public Policy( + String id, + String name, + String owner, + boolean enabled, + TriggerConfig trigger, + List sourceIds, + List steps, + OutputSpec output, + Long teamId) { + this(id, name, owner, enabled, trigger, sourceIds, steps, output, List.of(), teamId); } /** @@ -40,7 +62,7 @@ public record Policy( List sourceIds, List steps, OutputSpec output) { - this(id, name, owner, enabled, trigger, sourceIds, steps, output, null); + this(id, name, owner, enabled, trigger, sourceIds, steps, output, List.of(), null); } /** A policy with no configured sources (a generator, or files supplied directly to a run). */ @@ -52,10 +74,25 @@ public record Policy( TriggerConfig trigger, List steps, OutputSpec output) { - this(id, name, owner, enabled, trigger, List.of(), steps, output, null); + this(id, name, owner, enabled, trigger, List.of(), steps, output, List.of(), null); } - /** This policy's pipeline as the engine sees it. */ + /** 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); + } + + /** 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); + } + + /** + * This policy's pipeline as the engine sees it (inline output; destinations resolved + * elsewhere). + */ public PipelineDefinition toDefinition() { return new PipelineDefinition(name, steps, output); } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/PolicyInlineOutputMigration.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/PolicyInlineOutputMigration.java new file mode 100644 index 0000000000..a93b60d000 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/PolicyInlineOutputMigration.java @@ -0,0 +1,144 @@ +package stirling.software.proprietary.policy.output; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import org.springframework.boot.context.event.ApplicationReadyEvent; +import org.springframework.context.event.EventListener; +import org.springframework.core.annotation.Order; +import org.springframework.stereotype.Component; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +import stirling.software.proprietary.policy.migration.CompletedMigrations; +import stirling.software.proprietary.policy.model.OutputSpec; +import stirling.software.proprietary.policy.model.Policy; +import stirling.software.proprietary.policy.source.Source; +import stirling.software.proprietary.policy.source.SourceStore; +import stirling.software.proprietary.policy.store.PolicyStore; + +/** + * One-time, idempotent migration of policies' inline output destinations onto stored {@link Source} + * references: policies written before a destination was a saved location carry their folder/S3 + * destination inline; this points each at a {@link Source} (reusing one at the same location, or + * creating it) so the destination becomes a managed location like any other source. Policies with + * an inline "return to caller" output have no location to store and are left as-is. + * + *

Idempotent by construction: a policy that already carries an {@code outputId} is skipped, so a + * sequential re-run finds nothing to do. Matches are keyed by the write-relevant config within a + * team, so an output to a folder/prefix an input source already covers links to that same source - + * unifying the "output of A is the input of B" case onto one location. A concurrent multi-node boot + * can at worst create a redundant (unreferenced) source row, never corrupt a policy. + */ +@Slf4j +@Component +@RequiredArgsConstructor +public class PolicyInlineOutputMigration { + + /** Completion-marker id: once recorded, later boots skip the scan entirely. */ + private static final String MIGRATION_ID = "policy-inline-output"; + + // Destination types worth persisting as a location; "inline" has nothing to store. + private static final List DESTINATION_TYPES = List.of("folder", "s3"); + // The options that actually address a write destination, per type. Read-only options (e.g. a + // folder's consume mode) are excluded so an output matches an existing input source at the same + // place regardless of how that source reads. + private static final Map> ADDRESS_OPTIONS = + Map.of("folder", List.of("directory"), "s3", List.of("connectionId", "prefix")); + // Field separator for the dedup key: a unit-separator control char that cannot appear in a + // directory/prefix/connection id, so distinct field sets can never collide. + private static final char DELIMITER = '\u001f'; + + private final PolicyStore policyStore; + private final SourceStore sourceStore; + private final CompletedMigrations completedMigrations; + + // Runs after EmbeddedS3CredentialMigration (@Order(1)) so any legacy S3 output has already had + // its embedded credentials extracted into a connection; the Source created here then references + // that connection rather than copying credentials into source_json. Not wrapped in a single + // transaction: each store write is its own (idempotent) commit, so a crash mid-run just re-runs + // next boot, and the marker below is written only once the whole pass succeeds. + @Order(2) + @EventListener(ApplicationReadyEvent.class) + public void migrate() { + if (completedMigrations.isDone(MIGRATION_ID)) { + return; + } + Map byAddress = indexExistingSources(); + int migrated = 0; + for (Policy policy : policyStore.all()) { + if (!policy.outputIds().isEmpty()) { + continue; // already references one or more locations + } + OutputSpec output = policy.output(); + if (output == null || !DESTINATION_TYPES.contains(output.type())) { + continue; // inline / editor / no location to migrate + } + Source destination = destinationFor(policy, output, byAddress); + policyStore.save(policy.withOutputIds(List.of(destination.id()))); + migrated++; + } + if (migrated > 0) { + log.info("Linked {} policy output(s) to stored source locations", migrated); + } + completedMigrations.markDone(MIGRATION_ID); + } + + /** Reuses an existing team source at the same address, else creates a minimal one. */ + private Source destinationFor(Policy policy, OutputSpec spec, Map byAddress) { + String key = addressKey(policy.teamId(), spec.type(), spec.options()); + Source existing = byAddress.get(key); + if (existing != null) { + return existing; + } + Source created = + sourceStore.save( + new Source( + null, + destinationName(spec), + spec.type(), + spec.options(), + true, + policy.owner(), + policy.teamId())); + byAddress.put(key, created); + return created; + } + + private Map indexExistingSources() { + Map byKey = new LinkedHashMap<>(); + for (Source source : sourceStore.all()) { + if (!DESTINATION_TYPES.contains(source.type())) { + continue; + } + byKey.putIfAbsent(addressKey(source.teamId(), source.type(), source.options()), source); + } + return byKey; + } + + private static String addressKey(Long teamId, String type, Map options) { + StringBuilder key = new StringBuilder(); + key.append(teamId == null ? "" : teamId).append(DELIMITER); + key.append(type == null ? "" : type).append(DELIMITER); + for (String option : ADDRESS_OPTIONS.getOrDefault(type, List.of())) { + Object value = options.get(option); + key.append(value == null ? "" : value.toString()).append(DELIMITER); + } + return key.toString(); + } + + /** A readable default name derived from the destination; the user can rename it later. */ + private static String destinationName(OutputSpec spec) { + if ("folder".equals(spec.type())) { + Object directory = spec.options().get("directory"); + return directory == null ? "Folder" : "Folder: " + directory; + } + if ("s3".equals(spec.type())) { + Object prefix = spec.options().get("prefix"); + return prefix == null || prefix.toString().isBlank() ? "S3 bucket" : "S3: " + prefix; + } + return spec.type(); + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/PolicyOutputResolver.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/PolicyOutputResolver.java new file mode 100644 index 0000000000..3e9d804683 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/PolicyOutputResolver.java @@ -0,0 +1,53 @@ +package stirling.software.proprietary.policy.output; + +import java.util.ArrayList; +import java.util.List; + +import org.springframework.stereotype.Service; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +import stirling.software.proprietary.policy.model.OutputSpec; +import stirling.software.proprietary.policy.model.Policy; +import stirling.software.proprietary.policy.source.Source; +import stirling.software.proprietary.policy.source.SourceStore; + +/** + * Resolves a policy's effective output destinations at run time: each {@code outputId} references a + * {@link Source} used as a destination, looked up live (so editing a location updates every policy + * that writes to it), exactly as input {@code sourceIds} are resolved. A run is delivered to every + * resolved destination. A policy with no references keeps its inline output (results returned to + * the caller) - the case for editor and one-off policies. A reference that no longer resolves + * (location deleted out from under a live policy - normally blocked by the source delete guard) is + * skipped; if none resolve, delivery falls back to inline so the run still completes. + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class PolicyOutputResolver { + + private final SourceStore sourceStore; + + public List resolve(Policy policy) { + List outputIds = policy.outputIds(); + if (outputIds.isEmpty()) { + return List.of(policy.output()); + } + List resolved = new ArrayList<>(); + for (String outputId : outputIds) { + sourceStore + .get(outputId) + .map(Source::toOutputSpec) + .ifPresentOrElse( + resolved::add, + () -> + log.warn( + "Policy {} references missing output source {}; skipping" + + " that destination", + policy.id(), + outputId)); + } + return resolved.isEmpty() ? List.of(policy.output()) : resolved; + } +} 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 117d78de40..c927ec199c 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 @@ -4,6 +4,7 @@ import java.util.Comparator; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.stream.Collectors; import org.springframework.stereotype.Service; @@ -75,10 +76,25 @@ public class PolicyOverviewService { triggerSummary(policy.trigger()), sources, steps, - outputSummary(policy.output()), + outputSummary(policy, sourceNames), policy.owner()); } + /** + * A policy that delivers to sources shows those locations' display names, comma-joined (each + * falling back to its id if it's since been deleted or isn't visible); otherwise the inline + * output's type. + */ + private static String outputSummary(Policy policy, Map sourceNames) { + List outputIds = policy.outputIds(); + if (!outputIds.isEmpty()) { + return outputIds.stream() + .map(id -> sourceNames.getOrDefault(id, id)) + .collect(Collectors.joining(", ")); + } + 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(); diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/EmbeddedS3CredentialMigration.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/EmbeddedS3CredentialMigration.java index e324cda525..bbbb9cb8b1 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/EmbeddedS3CredentialMigration.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/EmbeddedS3CredentialMigration.java @@ -6,8 +6,8 @@ import java.util.Map; import org.springframework.boot.context.event.ApplicationReadyEvent; import org.springframework.context.event.EventListener; +import org.springframework.core.annotation.Order; import org.springframework.stereotype.Component; -import org.springframework.transaction.annotation.Transactional; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -18,6 +18,7 @@ import stirling.software.proprietary.integration.model.IntegrationConfig; import stirling.software.proprietary.integration.model.IntegrationType; import stirling.software.proprietary.integration.repository.IntegrationConfigRepository; import stirling.software.proprietary.model.Team; +import stirling.software.proprietary.policy.migration.CompletedMigrations; import stirling.software.proprietary.policy.model.OutputSpec; import stirling.software.proprietary.policy.model.Policy; import stirling.software.proprietary.policy.source.Source; @@ -45,6 +46,9 @@ import tools.jackson.databind.ObjectMapper; @RequiredArgsConstructor public class EmbeddedS3CredentialMigration { + /** Completion-marker id: once recorded, later boots skip the scan entirely. */ + private static final String MIGRATION_ID = "embedded-s3-credentials"; + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); private static final List CONNECTION_OPTIONS = List.of("bucket", "region", "endpoint", "accessKeyId", "secretAccessKey"); @@ -56,10 +60,19 @@ public class EmbeddedS3CredentialMigration { private final PolicyStore policyStore; private final IntegrationConfigRepository connections; private final TeamRepository teamRepository; + private final CompletedMigrations completedMigrations; + // Must run before PolicyInlineOutputMigration: that migration copies a policy's inline output + // options into a Source, so embedded S3 credentials have to be extracted into a connection here + // first, or they would be copied verbatim (plaintext) into the new source row. Each rewrite is + // its own idempotent commit (dedup by credential key), so a crash mid-run just re-runs next + // boot; the completion marker below is written only once the whole pass succeeds. + @Order(1) @EventListener(ApplicationReadyEvent.class) - @Transactional public void migrate() { + if (completedMigrations.isDone(MIGRATION_ID)) { + return; + } Map byCredentialKey = indexExistingConnections(); int migrated = 0; for (Source source : sourceStore.all()) { @@ -89,6 +102,7 @@ public class EmbeddedS3CredentialMigration { if (migrated > 0) { log.info("Extracted embedded S3 credentials from {} row(s) into connections", migrated); } + completedMigrations.markDone(MIGRATION_ID); } private static boolean embedsCredentials(Map options) { @@ -196,15 +210,6 @@ public class EmbeddedS3CredentialMigration { } private static Policy withOutput(Policy policy, OutputSpec output) { - return new Policy( - policy.id(), - policy.name(), - policy.owner(), - policy.enabled(), - policy.trigger(), - policy.sourceIds(), - policy.steps(), - output, - policy.teamId()); + return policy.withOutput(output); } } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/Source.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/Source.java index ca80c1b46f..c064a33d9a 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/Source.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/Source.java @@ -3,14 +3,18 @@ package stirling.software.proprietary.policy.source; import java.util.Map; import stirling.software.proprietary.policy.model.InputSpec; +import stirling.software.proprietary.policy.model.OutputSpec; /** - * A persisted, reusable input connection: the instantiation of a source definition. Policies - * reference sources by {@code id} rather than embedding their config, so one connection is - * configured once and can feed many policies. + * A persisted, reusable storage location: the instantiation of a source definition. Policies + * reference sources by {@code id} rather than embedding their config, so one location is configured + * once and can be used by many policies - as an input (files come from it) and/or as an output (a + * run's files are delivered to it), which is how a folder or bucket can be both the output of one + * pipeline and the input of the next. * - *

{@code type} keys an {@link stirling.software.proprietary.policy.input.InputSource} bean, - * matching {@link InputSpec#type()}; {@code options} is that source's config. {@code owner} and + *

{@code type} keys an {@link stirling.software.proprietary.policy.input.InputSource} bean (and, + * for writable types, a {@link stirling.software.proprietary.policy.output.PolicyOutputSink}), + * matching {@link InputSpec#type()}; {@code options} is that location's config. {@code owner} and * {@code teamId} scope the source to a team, mirroring {@link * stirling.software.proprietary.policy.model.Policy}. */ @@ -27,8 +31,17 @@ public record Source( options = options == null ? Map.of() : options; } - /** The runtime form the policy engine resolves and runs against. */ + /** The runtime form the policy engine resolves and reads inputs from. */ public InputSpec toInputSpec() { return new InputSpec(type, options); } + + /** + * The runtime form the policy engine delivers a run's outputs to, when this source is used as a + * policy's destination. Read-only options (e.g. a folder's consume mode) are simply ignored by + * the output sink. + */ + public OutputSpec toOutputSpec() { + return new OutputSpec(type, options); + } } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceController.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceController.java index 6539d9a34d..73622738f0 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceController.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceController.java @@ -307,10 +307,17 @@ public class SourceController { } } - /** Names of the caller's visible policies that reference the given source. */ + /** + * Names of the caller's visible policies that reference the given source - as an input ({@code + * sourceIds}) or as their output destination ({@code outputId}), so a location in use either + * way is protected from deletion. + */ private List referencingPolicyNames(String sourceId) { return policyAccessGuard.visibleFrom(policyStore).stream() - .filter(policy -> policy.sourceIds().contains(sourceId)) + .filter( + policy -> + policy.sourceIds().contains(sourceId) + || policy.outputIds().contains(sourceId)) .map(Policy::name) .toList(); } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceOverviewService.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceOverviewService.java index 2c02b9f152..0f9df21440 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceOverviewService.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceOverviewService.java @@ -3,8 +3,10 @@ package stirling.software.proprietary.policy.source; import java.util.ArrayList; import java.util.Comparator; import java.util.HashMap; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; +import java.util.Set; import org.springframework.stereotype.Service; @@ -115,11 +117,17 @@ public class SourceOverviewService { return sources instanceof List list && list.contains(EditorSource.ID); } - /** Policies referencing each source id, across the caller's visible policies. */ + /** + * Policies referencing each source id, across the caller's visible policies. A source counts + * whether a policy reads from it ({@code sourceIds}) or writes to it ({@code outputId}); a + * policy that does both counts once. + */ private static Map> referencesBySource(List policies) { Map> bySource = new HashMap<>(); for (Policy policy : policies) { - for (String sourceId : policy.sourceIds()) { + Set referenced = new LinkedHashSet<>(policy.sourceIds()); + referenced.addAll(policy.outputIds()); + for (String sourceId : referenced) { bySource.computeIfAbsent(sourceId, key -> new ArrayList<>()).add(policy); } } 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 685ddb21db..6498e7e28a 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 @@ -36,6 +36,7 @@ public class InProcessPolicyStore implements PolicyStore { policy.sourceIds(), policy.steps(), policy.output(), + policy.outputIds(), policy.teamId()); policies.put(id, stored); // Existing policy keeps its position; a new one appends to the end of its team's queue. 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 41da23c734..1b598b88c1 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 @@ -44,6 +44,7 @@ public class JpaPolicyStore implements PolicyStore { policy.sourceIds(), policy.steps(), policy.output(), + policy.outputIds(), policy.teamId()); PolicyEntity entity = new PolicyEntity(); diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/security/configuration/DatabaseConfig.java b/app/proprietary/src/main/java/stirling/software/proprietary/security/configuration/DatabaseConfig.java index a9c567885f..a5bb305e06 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/security/configuration/DatabaseConfig.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/security/configuration/DatabaseConfig.java @@ -34,6 +34,7 @@ import stirling.software.common.model.exception.UnsupportedProviderException; "stirling.software.proprietary.workflow.repository", "stirling.software.proprietary.policy.store", "stirling.software.proprietary.policy.source", + "stirling.software.proprietary.policy.migration", "stirling.software.proprietary.policy.ledger", "stirling.software.proprietary.accountlink", "stirling.software.proprietary.access.repository", @@ -46,6 +47,7 @@ import stirling.software.common.model.exception.UnsupportedProviderException; "stirling.software.proprietary.workflow.model", "stirling.software.proprietary.policy.store", "stirling.software.proprietary.policy.source", + "stirling.software.proprietary.policy.migration", "stirling.software.proprietary.policy.ledger", "stirling.software.proprietary.accountlink", "stirling.software.proprietary.access.model", 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 cf33de8656..351eecd581 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 @@ -135,7 +135,7 @@ class PolicyControllerTest { private static PipelineDefinition definitionWithStep() { return new PipelineDefinition( - "pipe", List.of(new PipelineStep("/api/v1/misc/compress-pdf", null)), null); + "pipe", List.of(new PipelineStep("/api/v1/misc/compress-pdf", null)), List.of()); } private static Policy policy(String id, Long teamId) { @@ -190,7 +190,7 @@ class PolicyControllerTest { @Test @DisplayName("rejects a pipeline with no steps") void rejectsEmptyPipeline() { - PipelineDefinition empty = new PipelineDefinition("pipe", List.of(), null); + PipelineDefinition empty = new PipelineDefinition("pipe", List.of(), List.of()); assertThatThrownBy(() -> controller.run(empty, new PolicyRunFiles())) .isInstanceOf(ResponseStatusException.class) @@ -241,7 +241,7 @@ class PolicyControllerTest { @Test @DisplayName("rejects a pipeline with no steps") void rejectsEmpty() { - PipelineDefinition empty = new PipelineDefinition("pipe", List.of(), null); + PipelineDefinition empty = new PipelineDefinition("pipe", List.of(), List.of()); assertThatThrownBy(() -> controller.runStream(empty, new PolicyRunFiles())) .isInstanceOf(ResponseStatusException.class); 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 78348fa5c4..e69f0f5476 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 @@ -15,8 +15,10 @@ import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import java.io.IOException; import java.io.InputStream; import java.nio.file.Path; +import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; @@ -37,6 +39,7 @@ import org.springframework.http.ResponseEntity; import org.springframework.web.client.HttpClientErrorException; import stirling.software.common.model.ApplicationProperties; +import stirling.software.common.model.job.ResultFile; import stirling.software.common.service.FileStorage; import stirling.software.common.service.FileStorage.StoredFile; import stirling.software.common.service.InternalApiClient; @@ -55,7 +58,11 @@ import stirling.software.proprietary.policy.model.PolicyInputs; import stirling.software.proprietary.policy.model.PolicyRun; import stirling.software.proprietary.policy.model.PolicyRunStatus; import stirling.software.proprietary.policy.output.InlineOutputSink; +import stirling.software.proprietary.policy.output.OutputDelivery; +import stirling.software.proprietary.policy.output.PolicyOutputResolver; +import stirling.software.proprietary.policy.output.PolicyOutputSink; import stirling.software.proprietary.policy.progress.PolicyProgressListener; +import stirling.software.proprietary.policy.source.InProcessSourceStore; import tools.jackson.databind.json.JsonMapper; @@ -81,6 +88,7 @@ class PolicyEngineTest { @TempDir Path tempDir; + private final RecordingSink recordingSink = new RecordingSink(); private PolicyRunRegistry registry; private PolicyEngine engine; @@ -98,6 +106,7 @@ class PolicyEngineTest { JsonMapper.builder().build()); registry = new PolicyRunRegistry(new ApplicationProperties()); InlineOutputSink sink = new InlineOutputSink(fileStorage); + PolicyOutputResolver outputResolver = new PolicyOutputResolver(new InProcessSourceStore()); engine = new PolicyEngine( executor, @@ -105,7 +114,8 @@ class PolicyEngineTest { registry, fileStorage, jobOwnershipService, - List.of(sink), + List.of(sink, recordingSink), + outputResolver, resourceMonitor, jobQueue); @@ -156,6 +166,36 @@ class PolicyEngineTest { verify(taskManager, atLeastOnce()).addNote(eq(runId), anyString()); } + @Test + void deliversTheRunsFilesToEveryDestination() throws Exception { + when(toolMetadataService.isMultiInput(anyString())).thenReturn(false); + when(toolMetadataService.shouldUnpackZipResponse(anyString())).thenReturn(false); + stubEndpoint(COMPRESS, pdf("compressed", "out.pdf")); + + // Two destinations of a recording sink; each fully reads the (shared) result file, so this + // also proves the result Resources are re-readable across more than one delivery. + PipelineDefinition definition = + new PipelineDefinition( + "multi", + List.of(new PipelineStep(COMPRESS, Map.of())), + List.of( + new OutputSpec("record", Map.of("dest", "a")), + new OutputSpec("record", Map.of("dest", "b")))); + + PolicyRun run = + engine.submit( + definition, + PolicyInputs.of(List.of(pdf("input", "input.pdf"))), + PolicyProgressListener.NOOP) + .completion() + .get(10, TimeUnit.SECONDS); + + assertEquals(PolicyRunStatus.COMPLETED, run.getStatus()); + // One result file per destination, and each destination read the same output content. + assertEquals(2, run.getOutputs().size()); + assertEquals(List.of("a:compressed", "b:compressed"), recordingSink.deliveries()); + } + @Test void submitFailsRunWhenAToolErrors() throws Exception { when(toolMetadataService.isMultiInput(ROTATE)).thenReturn(false); @@ -385,4 +425,50 @@ class PolicyEngineTest { } }; } + + /** + * A test output sink (type "record") that fully reads each delivered file and records + * "{dest}:{content}" per file, so a test can assert the run was delivered to every destination. + */ + private static final class RecordingSink implements PolicyOutputSink { + + private final List deliveries = new ArrayList<>(); + + List deliveries() { + return deliveries; + } + + @Override + public String type() { + return "record"; + } + + @Override + public boolean supports(OutputSpec spec) { + return spec != null && "record".equals(spec.type()); + } + + @Override + public List deliver( + OutputDelivery delivery, List outputs, OutputSpec spec) + throws IOException { + String dest = String.valueOf(spec.options().get("dest")); + List results = new ArrayList<>(); + for (Resource file : outputs) { + byte[] bytes; + try (InputStream is = file.getInputStream()) { + bytes = is.readAllBytes(); + } + deliveries.add(dest + ":" + new String(bytes)); + results.add( + ResultFile.builder() + .fileId("rec-" + dest) + .fileName(dest + "/" + file.getFilename()) + .contentType("application/pdf") + .fileSize(bytes.length) + .build()); + } + return results; + } + } } diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunRegistryTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunRegistryTest.java index 1549ba62ac..0d1361c5d9 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunRegistryTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunRegistryTest.java @@ -95,7 +95,8 @@ class PolicyRunRegistryTest { } private PolicyRun register(String runId) { - PolicyRun run = new PolicyRun(runId, null, new PipelineDefinition(runId, List.of(), null)); + PolicyRun run = + new PolicyRun(runId, null, new PipelineDefinition(runId, List.of(), List.of())); registry.register(run); return run; } 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 new file mode 100644 index 0000000000..082ca8cd42 --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/PolicyInlineOutputMigrationTest.java @@ -0,0 +1,149 @@ +package stirling.software.proprietary.policy.output; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.List; +import java.util.Map; + +import org.junit.jupiter.api.Test; + +import stirling.software.proprietary.policy.migration.InProcessCompletedMigrations; +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.source.InProcessSourceStore; +import stirling.software.proprietary.policy.source.Source; +import stirling.software.proprietary.policy.source.SourceStore; +import stirling.software.proprietary.policy.store.InProcessPolicyStore; +import stirling.software.proprietary.policy.store.PolicyStore; + +/** + * Tests for {@link PolicyInlineOutputMigration}: policies carrying a folder/S3 destination inline + * are rewritten to reference a {@link Source} location by id; inline (return-to-caller) policies + * are left untouched; the pass is idempotent; two policies sharing a destination in one team share + * one source; and an output at a location an input source already covers reuses that source. + */ +class PolicyInlineOutputMigrationTest { + + private final PolicyStore policyStore = new InProcessPolicyStore(); + private final SourceStore sourceStore = new InProcessSourceStore(); + private final PolicyInlineOutputMigration migration = + new PolicyInlineOutputMigration( + policyStore, sourceStore, new InProcessCompletedMigrations()); + + @Test + void migratesAFolderPolicyToAStoredSource() { + Policy saved = policyStore.save(folderPolicy("Archive", "/out")); + + migration.migrate(); + + Policy migrated = policyStore.get(saved.id()).orElseThrow(); + assertEquals(1, migrated.outputIds().size()); + Source destination = sourceStore.get(migrated.outputIds().get(0)).orElseThrow(); + assertEquals("folder", destination.type()); + assertEquals("/out", destination.options().get("directory")); + } + + @Test + void leavesInlinePoliciesUntouched() { + Policy saved = + policyStore.save( + new Policy( + null, + "Editor run", + "owner", + true, + null, + List.of(), + List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), + OutputSpec.inline())); + + migration.migrate(); + + assertTrue(policyStore.get(saved.id()).orElseThrow().outputIds().isEmpty()); + assertTrue(sourceStore.all().isEmpty()); + } + + @Test + void isIdempotent() { + policyStore.save(folderPolicy("Archive", "/out")); + + migration.migrate(); + int afterFirst = sourceStore.all().size(); + migration.migrate(); + + assertEquals(afterFirst, sourceStore.all().size()); + } + + @Test + void skipsTheScanOnceComplete() { + // First pass records the completion marker (even with nothing to migrate). + migration.migrate(); + + // A migratable folder policy created afterwards is left untouched: the marker means the + // migration never scans again, rather than re-scanning and finding it every boot. + Policy later = policyStore.save(folderPolicy("Late", "/late")); + migration.migrate(); + + assertTrue(policyStore.get(later.id()).orElseThrow().outputIds().isEmpty()); + assertTrue(sourceStore.all().isEmpty()); + } + + @Test + void dedupesASharedDestinationWithinATeam() { + policyStore.save(teamFolderPolicy("A", "/shared", 7L)); + policyStore.save(teamFolderPolicy("B", "/shared", 7L)); + + migration.migrate(); + + assertEquals(1, sourceStore.all().size()); + } + + @Test + void reusesAnExistingInputSourceAtTheSameLocation() { + // An input source already reads /shared (with consume mode); a policy that outputs there + // should link to that same source, not mint a duplicate. + Source existing = + sourceStore.save( + new Source( + null, + "Shared", + "folder", + Map.of("directory", "/shared", "mode", "consume"), + true, + "owner", + 7L)); + Policy saved = policyStore.save(teamFolderPolicy("Writer", "/shared", 7L)); + + migration.migrate(); + + assertEquals(1, sourceStore.all().size()); + assertEquals(List.of(existing.id()), policyStore.get(saved.id()).orElseThrow().outputIds()); + } + + private static Policy folderPolicy(String name, String directory) { + return new Policy( + null, + name, + "owner", + true, + null, + List.of(), + List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), + OutputSpec.folder(directory)); + } + + private static Policy teamFolderPolicy(String name, String directory, Long teamId) { + return new Policy( + null, + name, + "owner", + true, + null, + List.of(), + List.of(new PipelineStep("/api/v1/misc/compress-pdf", Map.of())), + OutputSpec.folder(directory), + teamId); + } +} 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 new file mode 100644 index 0000000000..d65c5b24f2 --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/PolicyOutputResolverTest.java @@ -0,0 +1,73 @@ +package stirling.software.proprietary.policy.output; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.util.List; +import java.util.Map; + +import org.junit.jupiter.api.Test; + +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.source.InProcessSourceStore; +import stirling.software.proprietary.policy.source.Source; +import stirling.software.proprietary.policy.source.SourceStore; + +/** + * Tests for {@link PolicyOutputResolver}: a policy's {@code outputIds} resolve live to the stored + * sources used as destinations (one spec each), an unreferenced policy keeps its inline output, and + * a dangling reference falls back to inline delivery rather than failing the run. + */ +class PolicyOutputResolverTest { + + private final SourceStore sourceStore = new InProcessSourceStore(); + private final PolicyOutputResolver resolver = new PolicyOutputResolver(sourceStore); + + @Test + void resolvesEachOutputIdToItsStoredSource() { + Source archive = sourceStore.save(folder("Archive", "/out")); + Source backup = sourceStore.save(folder("Backup", "/backup")); + + List specs = + resolver.resolve(policy().withOutputIds(List.of(archive.id(), backup.id()))); + + assertEquals(2, specs.size()); + assertEquals("/out", specs.get(0).options().get("directory")); + assertEquals("/backup", specs.get(1).options().get("directory")); + } + + @Test + void anUnreferencedPolicyKeepsItsInlineOutput() { + List specs = resolver.resolve(policy()); + + assertEquals(1, specs.size()); + assertEquals("inline", specs.get(0).type()); + } + + @Test + void whenNoReferencesResolveItFallsBackToInline() { + List specs = + resolver.resolve(policy().withOutputIds(List.of("does-not-exist"))); + + assertEquals(1, specs.size()); + assertEquals("inline", specs.get(0).type()); + } + + private static Source folder(String name, String directory) { + return new Source( + null, name, "folder", Map.of("directory", directory), true, "owner", null); + } + + private static Policy policy() { + return new Policy( + "p1", + "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/s3/EmbeddedS3CredentialMigrationTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/s3/EmbeddedS3CredentialMigrationTest.java index ff14d8cf81..a794ee104e 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 @@ -23,6 +23,7 @@ import stirling.software.proprietary.access.model.OwnerScope; import stirling.software.proprietary.integration.model.IntegrationConfig; import stirling.software.proprietary.integration.repository.IntegrationConfigRepository; 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.PipelineStep; import stirling.software.proprietary.policy.model.Policy; @@ -49,7 +50,11 @@ class EmbeddedS3CredentialMigrationTest { void setUp() { migration = new EmbeddedS3CredentialMigration( - sourceStore, policyStore, connections, teamRepository); + sourceStore, + policyStore, + connections, + teamRepository, + new InProcessCompletedMigrations()); AtomicLong ids = new AtomicLong(100); // Lenient: the nothing-to-migrate cases never create a connection. lenient().when(connections.findAll()).thenReturn(List.of()); diff --git a/frontend/editor/public/locales/en-US/translation.toml b/frontend/editor/public/locales/en-US/translation.toml index 48677633c4..913cdf20a4 100644 --- a/frontend/editor/public/locales/en-US/translation.toml +++ b/frontend/editor/public/locales/en-US/translation.toml @@ -7771,20 +7771,16 @@ addTool = "Add tool" cancel = "Cancel" chainEmpty = "Add a tool to start building your pipeline." create = "Create pipeline" -directory = "Output folder" -directoryHelp = "Absolute path on the server. Must be within the configured allowed folders." editingUnsupported = "Displaying these tool params for editing is not supported yet." moveDown = "Move down" moveUp = "Move up" name = "Name" namePlaceholder = "e.g. Redaction sweep" -noSources = "No sources connected yet. The pipeline can still run on files supplied to it directly." noToolSettings = "This tool has no configurable settings." operations_one = "Operation ({{count}})" operations_other = "Operations ({{count}})" -output = "Output" +output = "Destinations" removeStep = "Remove operation" -s3PrefixHelp = "Outputs are uploaded under this key prefix." save = "Save changes" scheduleEvery = "Run every" sources = "Sources" @@ -7819,11 +7815,6 @@ active = "Active" paused = "Paused" total = "Pipelines" -[portal.pipelines.output] -folder = "Write to folder" -inline = "Return files" -s3 = "Write to Amazon S3" - [portal.pipelines.run] allProcessed_one = "Nothing to run: the source's {{count}} document has already been processed." allProcessed_other = "Nothing to run: all {{count}} documents in the sources have already been processed." @@ -8704,7 +8695,7 @@ confirm = "Delete" title = "Delete source?" [portal.sources.empty] -description = "Connect a folder (and, soon, cloud storage) so your policies have somewhere to pull documents from." +description = "Connect a storage location so your policies have somewhere to pull data from." title = "No sources connected yet" [portal.sources.kpi] diff --git a/frontend/editor/src/portal/api/pipelines.ts b/frontend/editor/src/portal/api/pipelines.ts index e75711b912..2a4f25b79e 100644 --- a/frontend/editor/src/portal/api/pipelines.ts +++ b/frontend/editor/src/portal/api/pipelines.ts @@ -29,8 +29,8 @@ export interface OutputSpec { options: Record; } -/** The output destinations the pipeline builder can offer. */ -export type PipelineOutputMode = "inline" | "folder" | "s3"; +/** Source types that can be written to (used as a pipeline's output destination). */ +export type PipelineOutputMode = "folder" | "s3"; /** * The stored policy record: the create/update body (`id` blank on create) and what @@ -45,7 +45,17 @@ export interface Policy { trigger: TriggerConfig | null; sourceIds: string[]; steps: PipelineStep[]; + /** + * Inline output, used only when no destinations are referenced (editor/one-off runs that return + * results to the caller). Portal pipelines set {@link outputIds} instead. + */ output: OutputSpec; + /** + * The saved Sources this policy delivers its output to (each a source used as a write target), + * resolved live at run time; a run is delivered to every one. Empty means the inline {@link + * output} is used. + */ + outputIds: string[]; teamId?: number | null; } diff --git a/frontend/editor/src/portal/components/pipelines/DestinationPicker.tsx b/frontend/editor/src/portal/components/pipelines/DestinationPicker.tsx new file mode 100644 index 0000000000..683c6301b4 --- /dev/null +++ b/frontend/editor/src/portal/components/pipelines/DestinationPicker.tsx @@ -0,0 +1,62 @@ +import { useTranslation } from "react-i18next"; +import AddRoundedIcon from "@mui/icons-material/AddRounded"; +import { Button, Checkbox } 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). + */ +interface DestinationOption { + id: string; + name: string; +} + +interface DestinationPickerProps { + sources: DestinationOption[]; + value: string[]; + onChange: (outputIds: string[]) => void; + /** Leave the builder to create a new source location (navigate-away, like inputs). */ + onCreateNew: () => void; +} + +export function DestinationPicker({ + sources, + value, + onChange, + onCreateNew, +}: 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} + /> + ))} +
+ + + ); +} diff --git a/frontend/editor/src/portal/components/pipelines/outputModes.ts b/frontend/editor/src/portal/components/pipelines/outputModes.ts index a8e698c676..0d76cf122e 100644 --- a/frontend/editor/src/portal/components/pipelines/outputModes.ts +++ b/frontend/editor/src/portal/components/pipelines/outputModes.ts @@ -1,11 +1,11 @@ import type { PipelineOutputMode } from "@portal/api/pipelines"; /** - * The output destinations the pipeline builder offers. An extension point: - * deployments where a destination cannot work shadow this module and filter - * the list (e.g. hosted deployments never write to the server's filesystem, - * so folder outputs are not offered there). + * The source types that can be written to, i.e. offered as a pipeline's output destination. An + * extension point: deployments where a destination cannot work shadow this module and filter the + * list (e.g. hosted deployments never write to the server's filesystem, so folder destinations are + * not offered there and only S3 remains). */ export function availableOutputModes(): PipelineOutputMode[] { - return ["inline", "folder", "s3"]; + return ["folder", "s3"]; } diff --git a/frontend/editor/src/portal/mocks/handlers/pipelines.ts b/frontend/editor/src/portal/mocks/handlers/pipelines.ts index 4c18da8b6e..fdfa22e4b5 100644 --- a/frontend/editor/src/portal/mocks/handlers/pipelines.ts +++ b/frontend/editor/src/portal/mocks/handlers/pipelines.ts @@ -53,6 +53,7 @@ function seedPipelines(): StoredPolicy[] { { operation: "/api/v1/security/sanitize-pdf", parameters: {} }, ], output: { type: "inline", options: {} }, + outputIds: ["src-archive", "src-contracts"], }, { id: "plc-archive", @@ -62,7 +63,8 @@ function seedPipelines(): StoredPolicy[] { trigger: null, sourceIds: ["src-contracts", "src-archive"], steps: [{ operation: "/api/v1/misc/compress-pdf", parameters: {} }], - output: { type: "folder", options: { directory: "/data/archive-out" } }, + output: { type: "inline", options: {} }, + outputIds: ["src-contracts"], }, { id: "plc-onboarding", @@ -76,6 +78,7 @@ function seedPipelines(): StoredPolicy[] { { operation: "/api/v1/misc/flatten", parameters: {} }, ], output: { type: "inline", options: {} }, + outputIds: [], }, ]; } @@ -104,7 +107,10 @@ function toView(policy: StoredPolicy): PipelineView { name: SOURCE_NAMES[id] ?? id, })), steps: policy.steps.map((s) => s.operation), - output: policy.output?.type ?? "inline", + output: + policy.outputIds && policy.outputIds.length > 0 + ? policy.outputIds.map((id) => SOURCE_NAMES[id] ?? id).join(", ") + : (policy.output?.type ?? "inline"), owner: policy.owner ?? "you@acme.com", }; } diff --git a/frontend/editor/src/portal/views/PipelineBuilder.test.tsx b/frontend/editor/src/portal/views/PipelineBuilder.test.tsx index fe88ad7917..9d3f60f54e 100644 --- a/frontend/editor/src/portal/views/PipelineBuilder.test.tsx +++ b/frontend/editor/src/portal/views/PipelineBuilder.test.tsx @@ -8,6 +8,7 @@ import { import { PortalTestProviders } from "@portal/test/TestQueryProvider"; import { MemoryRouter, Route, Routes } from "react-router-dom"; import type { Policy, TriggerOutcome } from "@portal/api/pipelines"; +import type { SourceView } from "@portal/api/sources"; import type { ToolRegistryCatalog } from "@app/contexts/ToolRegistryContext"; import type { ToolRegistryEntry } from "@app/data/toolsTaxonomy"; import { PipelineBuilder } from "@portal/views/PipelineBuilder"; @@ -60,6 +61,22 @@ vi.mock("@portal/api/integrations", () => ({ createIntegration: (...args: unknown[]) => createIntegration(...args), })); +// The destination picker just selects saved sources; stub it to a button that +// picks a fixed source, keeping this suite focused on the builder. +vi.mock("@portal/components/pipelines/DestinationPicker", () => ({ + DestinationPicker: ({ + value, + onChange, + }: { + value: string[]; + onChange: (ids: string[]) => void; + }) => ( + + ), +})); + // One editable tool, Compress, so the picker and step settings have something to render. vi.mock("@app/contexts/ToolRegistryContext", () => { const compress = { @@ -113,6 +130,20 @@ const POLICY: Policy = { sourceIds: [], steps: [], output: { type: "inline", options: {} }, + outputIds: [], +}; + +const SOURCE: SourceView = { + id: "src-in", + name: "Claims intake", + type: "folder", + status: "active", + referenceCount: 0, + referencingPolicies: [], + config: [], + docsTotal: 0, + docs24h: 0, + docs30d: 0, }; function outcome(overrides: Partial): TriggerOutcome { @@ -152,7 +183,7 @@ describe("PipelineBuilder", () => { fetchSources.mockReset(); fetchPipeline.mockResolvedValue(POLICY); fetchTriggers.mockResolvedValue([]); - fetchSources.mockResolvedValue({ kpis: [], sources: [] }); + fetchSources.mockResolvedValue({ kpis: [], sources: [SOURCE] }); savePipeline.mockResolvedValue({}); deletePipeline.mockResolvedValue(undefined); triggerPipeline.mockResolvedValue(outcome({ runIds: ["run-1"] })); @@ -175,6 +206,12 @@ describe("PipelineBuilder", () => { fireEvent.click(screen.getByRole("button", { name: /addTool/ })); fireEvent.click(await screen.findByText("Compress")); + // A pipeline must have at least one input source and one output destination. + fireEvent.click( + await screen.findByRole("checkbox", { name: "Claims intake" }), + ); + fireEvent.click(screen.getByText("pick output")); + fireEvent.click(screen.getByText("portal.pipelines.composer.create")); await waitFor(() => expect(savePipeline).toHaveBeenCalledTimes(1)); @@ -182,6 +219,8 @@ describe("PipelineBuilder", () => { expect.objectContaining({ name: "Nightly compress", trigger: null, + sourceIds: ["src-in"], + outputIds: ["src-1"], steps: [ expect.objectContaining({ operation: "/api/v1/misc/compress-pdf" }), ], @@ -190,76 +229,30 @@ describe("PipelineBuilder", () => { expect(await screen.findByText("pipelines list")).toBeInTheDocument(); }); - it("saves an s3 output referencing an inline-created connection", async () => { - createIntegration.mockResolvedValue({ id: 12, name: "Claims bucket" }); + it("requires at least one source and one destination before saving", async () => { renderBuilder("/processor/pipelines/new"); fireEvent.change(await screen.findByRole("textbox"), { - target: { value: "Bucket to bucket" }, + target: { value: "Needs both" }, }); - fireEvent.click(screen.getByLabelText("portal.pipelines.output.s3")); + const saveButton = () => + screen.getByText("portal.pipelines.composer.create").closest("button"); - // With s3 selected but no connection chosen, saving is blocked. The - // connection picker + prefix are inline (no modal), like the folder output. - expect( - screen.getByText("portal.pipelines.composer.create").closest("button"), - ).toBeDisabled(); + // Name only: blocked (no source, no destination). + expect(saveButton()).toBeDisabled(); - // No connections exist: create one inline from the picker. Target fields by - // label, not position - the picker's Mantine Select also carries an input - // role and would shift index-based queries. + // A source but still no destination: blocked. fireEvent.click( - await screen.findByText("portal.connections.picker.createNew"), - ); - fireEvent.change(screen.getByLabelText(/portal\.integrations\.typedName/), { - target: { value: "Claims bucket" }, - }); - fireEvent.change( - screen.getByLabelText( - /portal\.connections\.types\.s3\.fields\.bucket\.label/, - ), - { target: { value: "claims-processed" } }, - ); - fireEvent.change( - screen.getByLabelText( - /portal\.connections\.types\.s3\.fields\.accessKeyId\.label/, - ), - { target: { value: "AKIAEXAMPLE" } }, - ); - fireEvent.change( - screen.getByLabelText( - /portal\.connections\.types\.s3\.fields\.secretAccessKey\.label/, - ), - { target: { value: "shh-secret" } }, - ); - fireEvent.click(screen.getByText("portal.connections.picker.save")); - await waitFor(() => expect(createIntegration).toHaveBeenCalledTimes(1)); - // The connection modal closes once saved and the connection is selected. - await waitFor(() => - expect( - screen.queryByText("portal.connections.picker.save"), - ).not.toBeInTheDocument(), + await screen.findByRole("checkbox", { name: "Claims intake" }), ); + expect(saveButton()).toBeDisabled(); - fireEvent.change( - screen.getByLabelText( - /portal\.sources\.types\.s3\.fields\.prefix\.label/, - ), - { target: { value: "processed/" } }, - ); + // Both chosen: allowed, and both are sent. + fireEvent.click(screen.getByText("pick output")); fireEvent.click(screen.getByText("portal.pipelines.composer.create")); - await waitFor(() => expect(savePipeline).toHaveBeenCalledTimes(1)); expect(savePipeline).toHaveBeenCalledWith( - expect.objectContaining({ - output: { - type: "s3", - options: { - connectionId: "12", - prefix: "processed/", - }, - }, - }), + expect.objectContaining({ sourceIds: ["src-in"], outputIds: ["src-1"] }), ); }); @@ -408,6 +401,12 @@ describe("PipelineBuilder", () => { ); fireEvent.click(await screen.findByText("Ops alerts")); + // Saving needs at least one input source and one destination. + fireEvent.click( + await screen.findByRole("checkbox", { name: "Claims intake" }), + ); + fireEvent.click(screen.getByText("pick output")); + fireEvent.click(screen.getByText("portal.pipelines.composer.create")); await waitFor(() => expect(savePipeline).toHaveBeenCalledTimes(1)); diff --git a/frontend/editor/src/portal/views/PipelineBuilder.tsx b/frontend/editor/src/portal/views/PipelineBuilder.tsx index fc23ac1516..0955049605 100644 --- a/frontend/editor/src/portal/views/PipelineBuilder.tsx +++ b/frontend/editor/src/portal/views/PipelineBuilder.tsx @@ -14,7 +14,6 @@ import { Button, Checkbox, EmptyState, - FormField, Input, Modal, RadioGroup, @@ -40,17 +39,15 @@ import { fetchTriggers, savePipeline, triggerPipeline, - type OutputSpec, type Policy, type PolicyRunView, - type PipelineOutputMode, type TriggerConfig, type TriggerInfo, type TriggerOutcome, } from "@portal/api/pipelines"; import { clearProcessedHistory } from "@portal/api/policies"; +import { DestinationPicker } from "@portal/components/pipelines/DestinationPicker"; import { availableOutputModes } from "@portal/components/pipelines/outputModes"; -import { S3ConnectionPicker } from "@portal/components/sources/S3ConnectionPicker"; import { type SourceView } from "@portal/api/sources"; import { useSources } from "@portal/queries/sources"; import { EDITOR_SOURCE_TYPE } from "@portal/components/sources/sourceTypes"; @@ -70,21 +67,6 @@ import { } from "@portal/components/pipelines/integrationStep"; import "@portal/views/PipelineBuilder.css"; -type OutputMode = PipelineOutputMode; - -/** New pipelines (and specs of unoffered types) start on the first offered destination. */ -const DEFAULT_OUTPUT_MODE = availableOutputModes()[0]; - -/** The s3 output's options: a stored connection reference plus the per-use prefix. */ -interface S3OutputOptions { - connectionId: string; - prefix: string; -} - -const EMPTY_S3_OUTPUT: S3OutputOptions = { - connectionId: "", - prefix: "", -}; type ScheduleUnit = "MINUTES" | "HOURS" | "DAYS"; const SCHEDULE_UNITS: ScheduleUnit[] = ["MINUTES", "HOURS", "DAYS"]; @@ -123,31 +105,6 @@ function parseTrigger(trigger: TriggerConfig | null): { return { triggerType: trigger.type, count: "1", unit: "HOURS" }; } -function parseOutput(output: OutputSpec | undefined): { - mode: OutputMode; - directory: string; - s3: S3OutputOptions; -} { - if (output?.type === "folder") { - return { - mode: "folder", - directory: String(output.options?.directory ?? ""), - s3: EMPTY_S3_OUTPUT, - }; - } - if (output?.type === "s3") { - return { - mode: "s3", - directory: "", - s3: { - connectionId: String(output.options?.connectionId ?? ""), - prefix: String(output.options?.prefix ?? ""), - }, - }; - } - return { mode: DEFAULT_OUTPUT_MODE, directory: "", s3: EMPTY_S3_OUTPUT }; -} - /** * Full-page pipeline builder (route: /pipelines/new and /pipelines/:id). Pipeline-level settings * (sources, trigger, output) sit above the operation list; the operation list and the selected @@ -193,6 +150,15 @@ export function PipelineBuilder() { ), [sourcesState.data], ); + // A destination is a source used as a write target: only writable types (folder/S3, filtered per + // deployment) can be picked, and the virtual editor is already excluded from availableSources. + const writableSources = useMemo( + () => + availableSources.filter((source) => + (availableOutputModes() as string[]).includes(source.type), + ), + [availableSources], + ); const triggers = useMemo( () => triggersState.data ?? [], [triggersState.data], @@ -207,9 +173,7 @@ export function PipelineBuilder() { const [triggerType, setTriggerType] = useState(MANUAL); const [scheduleCount, setScheduleCount] = useState("1"); const [scheduleUnit, setScheduleUnit] = useState("HOURS"); - const [outputMode, setOutputMode] = useState(DEFAULT_OUTPUT_MODE); - const [outputDirectory, setOutputDirectory] = useState(""); - const [outputS3, setOutputS3] = useState(EMPTY_S3_OUTPUT); + const [outputIds, setOutputIds] = useState([]); const [submitting, setSubmitting] = useState(false); const [error, setError] = useState(null); const [seeded, setSeeded] = useState(false); @@ -234,7 +198,6 @@ export function PipelineBuilder() { if (isEdit && !policyState.data) return; const policy = policyState.data ?? undefined; const trigger = parseTrigger(policy?.trigger ?? null); - const output = parseOutput(policy?.output); setName(policy?.name ?? ""); setEnabled(policy?.enabled ?? true); setSourceIds(policy?.sourceIds ?? []); @@ -244,9 +207,7 @@ export function PipelineBuilder() { setTriggerType(trigger.triggerType); setScheduleCount(trigger.count); setScheduleUnit(trigger.unit); - setOutputMode(output.mode); - setOutputDirectory(output.directory); - setOutputS3(output.s3); + setOutputIds(policy?.outputIds ?? []); setSeeded(true); }, [isEdit, policyState.data, allTools, seeded]); @@ -371,9 +332,7 @@ export function PipelineBuilder() { triggerType, scheduleCount, scheduleUnit, - outputMode, - outputDirectory, - outputS3, + outputIds: [...outputIds].sort(), }); const baseline = useRef(null); useEffect(() => { @@ -383,12 +342,12 @@ export function PipelineBuilder() { const scheduleCountValid = triggerType !== "schedule" || Number(scheduleCount) > 0; - const s3OutputValid = - outputMode !== "s3" || outputS3.connectionId.trim() !== ""; - const outputValid = - (outputMode !== "folder" || outputDirectory.trim() !== "") && s3OutputValid; + // A pipeline must have at least one input source and at least one output destination. + const sourceValid = sourceIds.length > 0; + const outputValid = outputIds.length > 0; const canSave = name.trim() !== "" && + sourceValid && scheduleCountValid && outputValid && !hasUploadSteps && @@ -436,8 +395,8 @@ export function PipelineBuilder() { else navigate(destination); } - // Jump to the Sources page with its create wizard open, for when the source you want to run - // this pipeline over doesn't exist yet. + // Jump to the source builder, for when the source you want to read from or write to doesn't + // exist yet. Inputs and the output destination are both saved sources, so both create one here. function goToSources() { attemptLeave(sourcesPath); } @@ -446,12 +405,6 @@ export function PipelineBuilder() { if (!canSave) return; setSubmitting(true); setError(null); - const output: OutputSpec = - outputMode === "folder" - ? { type: "folder", options: { directory: outputDirectory.trim() } } - : outputMode === "s3" - ? { type: "s3", options: { ...outputS3 } } - : { type: "inline", options: {} }; const policy: Policy = { id: policyState.data?.id ?? undefined, name: name.trim(), @@ -459,7 +412,10 @@ export function PipelineBuilder() { trigger: buildTrigger(), sourceIds, steps: steps.map((step) => serializeToolStep(step, allTools)), - output, + // Destinations are the referenced saved sources; the inline output field is + // preserved as-is (e.g. an editor policy's membership metadata) or defaults to inline. + output: policyState.data?.output ?? { type: "inline", options: {} }, + outputIds, }; try { await savePipeline(policy); @@ -716,10 +672,6 @@ export function PipelineBuilder() {

{t("portal.pipelines.composer.sourcesLoading")}

- ) : availableSources.length === 0 ? ( -

- {t("portal.pipelines.composer.noSources")} -

) : (
{availableSources.map((source) => ( @@ -787,55 +739,12 @@ export function PipelineBuilder() { {t("portal.pipelines.composer.output")} - - name="pipeline-output" - value={outputMode} - onChange={setOutputMode} - options={availableOutputModes().map((mode) => ({ - value: mode, - label: t(`portal.pipelines.output.${mode}`), - }))} + - {outputMode === "folder" && ( - - setOutputDirectory(e.target.value)} - /> - - )} - {outputMode === "s3" && ( - <> - - - setOutputS3((s) => ({ ...s, connectionId })) - } - /> - - - - setOutputS3((s) => ({ ...s, prefix: e.target.value })) - } - /> - - - )}
diff --git a/frontend/editor/src/proprietary/services/policyPipeline.test.ts b/frontend/editor/src/proprietary/services/policyPipeline.test.ts index e6017e99b1..776495c344 100644 --- a/frontend/editor/src/proprietary/services/policyPipeline.test.ts +++ b/frontend/editor/src/proprietary/services/policyPipeline.test.ts @@ -32,7 +32,7 @@ describe("buildPipelineDefinition", () => { expect(unresolved).toEqual([]); expect(definition.name).toBe("Secure Ingestion"); - expect(definition.output).toEqual({ type: "inline", options: {} }); + expect(definition.outputs).toEqual([{ type: "inline", options: {} }]); expect(definition.steps).toEqual([ { operation: "/api/v1/misc/compress-pdf", parameters: {} }, { diff --git a/frontend/editor/src/proprietary/services/policyPipeline.ts b/frontend/editor/src/proprietary/services/policyPipeline.ts index c04adb78b6..ed073b9b1c 100644 --- a/frontend/editor/src/proprietary/services/policyPipeline.ts +++ b/frontend/editor/src/proprietary/services/policyPipeline.ts @@ -32,7 +32,8 @@ export interface BackendOutputSpec { export interface BackendPipelineDefinition { name: string; steps: BackendPipelineStep[]; - output: BackendOutputSpec; + /** Destinations a run's files are delivered to; a single inline entry for one-off/editor runs. */ + outputs: BackendOutputSpec[]; } /** How a stored policy is triggered ("manual" | "folder" | "schedule" | "s3"). */ @@ -191,7 +192,7 @@ export function buildPipelineDefinition( definition: { name: automation.name, steps, - output: { type: "inline", options: {} }, + outputs: [{ type: "inline", options: {} }], }, unresolved, };