Stirling-Tools/Stirling-PDF
1
0
Fork 0
mirror of https://github.com/Stirling-Tools/Stirling-PDF.git synced 2026-09-02 21:03:34 +03:00
Code Issues Packages Projects Releases Wiki Activity

Add persistent outputs to Processor (#7071)

Browse Source 
# Description of Changes
<img width="1270" height="487" alt="image"
src="https://github.com/user-attachments/assets/64894e2b-aab9-42ab-96c2-2c11ba427b52"
/>

Change policies to point towards a source for its output instead of a
dynamically defined output location for the pipeline. This allows for
easy reuse of outputs in different pipelines and makes it impossible to
break complex pipelines by accidentally updating the source but not the
output and vice versa. Also makes outputs a list to match the inputs, so
it's possible for a pipeline to output to multiple locations.

We should consider whether we want to continue calling these Sources
since they're now being used as both inputs and outputs, but that
decision is beyond the scope of this PR.

Also updates the existing S3 DB migration script and adds a new one to
migrate to the new schema. Neither of these scripts are possible with
SQL since it involves parsing and restructuring JSON. I've updated them
so that they only ever run once on startup and mark themselves as
completed.
This commit is contained in:
James Brunton
2026-07-29 11:22:47 +00:00
committed by GitHub
parent 66b80a80c0
commit 999b5e5995
34 changed files with 1017 additions and 260 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/proprietary/src/main/java/stirling/software/proprietary/policy/controller/PolicyController.java
+41 -12
View File
@@ -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());
app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyEngine.java
+24 -8
View File
@@ -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<PolicyOutputSink> 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<ResultFile> 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<OutputSpec> destinations = run.getDefinition().outputs();
if (destinations.isEmpty()) {
destinations = List.of(OutputSpec.inline());
}
List<ResultFile> 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);
app/proprietary/src/main/java/stirling/software/proprietary/policy/migration/CompletedMigration.java
+40
View File
@@ -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;
}
}
app/proprietary/src/main/java/stirling/software/proprietary/policy/migration/CompletedMigrationRepository.java
+7
View File
@@ -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<CompletedMigration, String> {}
app/proprietary/src/main/java/stirling/software/proprietary/policy/migration/CompletedMigrations.java
+18
View File
@@ -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);
}
app/proprietary/src/main/java/stirling/software/proprietary/policy/migration/InProcessCompletedMigrations.java
+23
View File
@@ -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<String> done = ConcurrentHashMap.newKeySet();
@Override
public boolean isDone(String id) {
return done.contains(id);
}
@Override
public void markDone(String id) {
done.add(id);
}
}
app/proprietary/src/main/java/stirling/software/proprietary/policy/migration/JpaCompletedMigrations.java
+38
View File
@@ -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);
}
}
}
app/proprietary/src/main/java/stirling/software/proprietary/policy/model/PipelineDefinition.java
+11 -4
View File
@@ -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.
*
* <p>{@code output} may be null for callers that handle result files themselves (e.g. the AI
* workflow, which builds its own response payload).
* <p>{@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<PipelineStep> steps, OutputSpec output) {
public record PipelineDefinition(String name, List<PipelineStep> steps, List<OutputSpec> 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<PipelineStep> steps, OutputSpec output) {
this(name, steps, output == null ? List.of() : List.of(output));
}
}
app/proprietary/src/main/java/stirling/software/proprietary/policy/model/Policy.java
+43 -6
View File
@@ -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.
*
* <p>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<String> sourceIds,
List<PipelineStep> steps,
OutputSpec output,
List<String> 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<String> sourceIds,
List<PipelineStep> 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<String> sourceIds,
List<PipelineStep> 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<PipelineStep> 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<String> 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);
}
app/proprietary/src/main/java/stirling/software/proprietary/policy/output/PolicyInlineOutputMigration.java
+144
View File
@@ -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.
*
* <p>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<String> 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<String, List<String>> 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<String, Source> 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<String, Source> 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<String, Source> indexExistingSources() {
Map<String, Source> 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<String, Object> 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();
}
}
app/proprietary/src/main/java/stirling/software/proprietary/policy/output/PolicyOutputResolver.java
+53
View File
@@ -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<OutputSpec> resolve(Policy policy) {
List<String> outputIds = policy.outputIds();
if (outputIds.isEmpty()) {
return List.of(policy.output());
}
List<OutputSpec> 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;
}
}
app/proprietary/src/main/java/stirling/software/proprietary/policy/overview/PolicyOverviewService.java
+17 -1
View File
@@ -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<String, String> sourceNames) {
List<String> 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();
app/proprietary/src/main/java/stirling/software/proprietary/policy/s3/EmbeddedS3CredentialMigration.java
+17 -12
View File
@@ -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<String> 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<String, IntegrationConfig> 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<String, Object> 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);
}
}
app/proprietary/src/main/java/stirling/software/proprietary/policy/source/Source.java
+19 -6
View File
@@ -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.
*
* <p>{@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
* <p>{@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);
}
}
app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceController.java
+9 -2
View File
@@ -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<String> 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();
}
app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceOverviewService.java
+10 -2
View File
@@ -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<String, List<Policy>> referencesBySource(List<Policy> policies) {
Map<String, List<Policy>> bySource = new HashMap<>();
for (Policy policy : policies) {
for (String sourceId : policy.sourceIds()) {
Set<String> referenced = new LinkedHashSet<>(policy.sourceIds());
referenced.addAll(policy.outputIds());
for (String sourceId : referenced) {
bySource.computeIfAbsent(sourceId, key -> new ArrayList<>()).add(policy);
}
}
app/proprietary/src/main/java/stirling/software/proprietary/policy/store/InProcessPolicyStore.java
+1
View File
@@ -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.
app/proprietary/src/main/java/stirling/software/proprietary/policy/store/JpaPolicyStore.java
+1
View File
@@ -44,6 +44,7 @@ public class JpaPolicyStore implements PolicyStore {
policy.sourceIds(),
policy.steps(),
policy.output(),
policy.outputIds(),
policy.teamId());
PolicyEntity entity = new PolicyEntity();
app/proprietary/src/main/java/stirling/software/proprietary/security/configuration/DatabaseConfig.java
+2
View File
@@ -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",
app/proprietary/src/test/java/stirling/software/proprietary/policy/controller/PolicyControllerTest.java
+3 -3
View File
@@ -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);
app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyEngineTest.java
+87 -1
View File
@@ -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<String> deliveries = new ArrayList<>();
List<String> 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<ResultFile> deliver(
OutputDelivery delivery, List<Resource> outputs, OutputSpec spec)
throws IOException {
String dest = String.valueOf(spec.options().get("dest"));
List<ResultFile> 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;
}
}
}
app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunRegistryTest.java
+2 -1
View File
@@ -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;
}
app/proprietary/src/test/java/stirling/software/proprietary/policy/output/PolicyInlineOutputMigrationTest.java
+149
View File
@@ -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);
}
}
app/proprietary/src/test/java/stirling/software/proprietary/policy/output/PolicyOutputResolverTest.java
+73
View File
@@ -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<OutputSpec> 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<OutputSpec> specs = resolver.resolve(policy());
assertEquals(1, specs.size());
assertEquals("inline", specs.get(0).type());
}
@Test
void whenNoReferencesResolveItFallsBackToInline() {
List<OutputSpec> 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());
}
}
app/proprietary/src/test/java/stirling/software/proprietary/policy/s3/EmbeddedS3CredentialMigrationTest.java
+6 -1
View File
@@ -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());
frontend/editor/public/locales/en-US/translation.toml
+2 -11
View File
@@ -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]
frontend/editor/src/portal/api/pipelines.ts
+12 -2
View File
@@ -29,8 +29,8 @@ export interface OutputSpec {
options: Record<string, unknown>;
}
/** 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;
}
frontend/editor/src/portal/components/pipelines/DestinationPicker.tsx
+62
View File
@@ -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 (
<>
<div className="portal-pipelines__source-list">
{sources.map((source) => (
<Checkbox
key={source.id}
checked={value.includes(source.id)}
onChange={(e) => toggle(source.id, e.target.checked)}
label={source.name}
/>
))}
</div>
<Button
variant="tertiary"
size="sm"
onClick={onCreateNew}
leftSection={<AddRoundedIcon style={{ fontSize: "1.125rem" }} />}
>
{t("portal.sources.actions.connectSource")}
</Button>
</>
);
}
frontend/editor/src/portal/components/pipelines/outputModes.ts
+5 -5
View File
@@ -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"];
}
frontend/editor/src/portal/mocks/handlers/pipelines.ts
+8 -2
View File
@@ -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",
};
}
frontend/editor/src/portal/views/PipelineBuilder.test.tsx
+58 -59
View File
@@ -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;
}) => (
<button type="button" onClick={() => onChange(["src-1"])}>
{value.length > 0 ? `output:${value.join(",")}` : "pick output"}
</button>
),
}));
// 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>): 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));
frontend/editor/src/portal/views/PipelineBuilder.tsx
+28 -119
View File
@@ -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<SourceView[]>(
() =>
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<string>(MANUAL);
const [scheduleCount, setScheduleCount] = useState("1");
const [scheduleUnit, setScheduleUnit] = useState<ScheduleUnit>("HOURS");
const [outputMode, setOutputMode] = useState<OutputMode>(DEFAULT_OUTPUT_MODE);
const [outputDirectory, setOutputDirectory] = useState("");
const [outputS3, setOutputS3] = useState<S3OutputOptions>(EMPTY_S3_OUTPUT);
const [outputIds, setOutputIds] = useState<string[]>([]);
const [submitting, setSubmitting] = useState(false);
const [error, setError] = useState<string | null>(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<string | null>(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() {
<p className="portal-pipelines__muted">
{t("portal.pipelines.composer.sourcesLoading")}
</p>
) : availableSources.length === 0 ? (
<p className="portal-pipelines__muted">
{t("portal.pipelines.composer.noSources")}
</p>
) : (
<div className="portal-pipelines__source-list">
{availableSources.map((source) => (
@@ -787,55 +739,12 @@ export function PipelineBuilder() {
<span className="portal-pipelines__detail-heading">
{t("portal.pipelines.composer.output")}
</span>
<RadioGroup<OutputMode>
name="pipeline-output"
value={outputMode}
onChange={setOutputMode}
options={availableOutputModes().map((mode) => ({
value: mode,
label: t(`portal.pipelines.output.${mode}`),
}))}
<DestinationPicker
sources={writableSources}
value={outputIds}
onChange={setOutputIds}
onCreateNew={goToSources}
/>
{outputMode === "folder" && (
<FormField
label={t("portal.pipelines.composer.directory")}
helperText={t("portal.pipelines.composer.directoryHelp")}
required
>
<Input
value={outputDirectory}
placeholder="/data/processed"
onChange={(e) => setOutputDirectory(e.target.value)}
/>
</FormField>
)}
{outputMode === "s3" && (
<>
<FormField
label={t("portal.sources.types.s3.fields.connection.label")}
required
>
<S3ConnectionPicker
value={outputS3.connectionId}
onChange={(connectionId) =>
setOutputS3((s) => ({ ...s, connectionId }))
}
/>
</FormField>
<FormField
label={t("portal.sources.types.s3.fields.prefix.label")}
helperText={t("portal.pipelines.composer.s3PrefixHelp")}
>
<Input
value={outputS3.prefix}
placeholder="processed/"
onChange={(e) =>
setOutputS3((s) => ({ ...s, prefix: e.target.value }))
}
/>
</FormField>
</>
)}
</div>
</div>
</section>
frontend/editor/src/proprietary/services/policyPipeline.test.ts
+1 -1
View File
@@ -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: {} },
{
frontend/editor/src/proprietary/services/policyPipeline.ts
+3 -2
View File
@@ -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,
};