Align webhook source with S3 connections

app/proprietary/src/main/java/stirling/software/proprietary/policy/input/WebhookInputSource.java

@@ -28,14 +28,22 @@ import stirling.software.proprietary.policy.webhook.WebhookSpool;
 
 /**
  * A push source: documents arrive by signed HTTP POST to {@code /api/v1/webhooks/{webhookId}} and
- * are staged by {@link WebhookSpool}. This source then reads that per-webhook spool exactly as
- * {@link FolderInputSource} reads a watched directory - each spooled file is a unit of work claimed
- * through the {@link ResolveContext} ledger and tracked in place - so the whole claim/consume model
- * is reused rather than reinvented. Options (see {@link WebhookConfig}): "mode" is "consume"
- * (default: a spooled file is removed once every policy that claimed it has settled successfully)
- * or "snapshot" (stateless, every run re-reads whatever is currently spooled). "webhookId" and
- * "signingSecret" are minted server-side on create, never supplied by the client. An empty or
- * not-yet-created spool means nothing has been delivered, which reads as "verifiably no files".
+ * are staged for the referencing policies. Where they are staged depends on the source's config
+ * (see {@link WebhookConfig}):
+ *
+ * <ul>
+ *   <li>no {@code connectionId}: a node-local {@link WebhookSpool} directory, read exactly as
+ *       {@link FolderInputSource} reads a watched directory (self-hosted);
+ *   <li>a {@code connectionId}: the referenced S3 {@code IntegrationConfig} connection, under a
+ *       reserved per-webhook prefix - read, claimed, and consumed by delegating to {@link
+ *       S3InputSource}, so the durable, multi-node model hosted deployments need is reused, not
+ *       reinvented.
+ * </ul>
+ *
+ * <p>Either way the {@link ResolveContext} ledger does the claiming, "mode" is "consume" (default:
+ * a staged document is removed once every policy that claimed it has settled successfully) or
+ * "snapshot" (stateless), and "webhookId"/"signingSecret" are minted server-side on create. Empty
+ * staging reads as "verifiably no files".
  */
 @Slf4j
 @Service
@@ -47,6 +55,10 @@ public class WebhookInputSource implements InputSource {
 
     private final WebhookSpool spool;
     private final FileReadinessChecker readinessChecker;
+    // Deliveries staged to an S3 connection are read/claimed/consumed exactly like an S3 source, so
+    // that whole path is reused rather than reimplemented; a connection-less webhook uses the
+    // spool.
+    private final S3InputSource s3InputSource;
 
     @Override
     public String type() {
@@ -60,7 +72,28 @@ public class WebhookInputSource implements InputSource {
 
     @Override
     public void validate(InputSpec spec) {
-        WebhookConfig.from(spec.options());
+        WebhookConfig config = WebhookConfig.from(spec.options());
+        if (config.usesConnection()) {
+            // Resolves the connection with the caller present, so save fails if their team can't
+            // use it (the ownership check the delivery path then trusts).
+            s3InputSource.validate(stagingSpec(config));
+        }
+    }
+
+    /**
+     * The S3 view of a webhook's durable staging: its connection plus the reserved per-webhook
+     * prefix, as an {@code s3} input spec {@link S3InputSource} can resolve directly.
+     */
+    private static InputSpec stagingSpec(WebhookConfig config) {
+        return new InputSpec(
+                "s3",
+                Map.of(
+                        WebhookConfig.CONNECTION_ID_OPTION,
+                        config.connectionId(),
+                        "prefix",
+                        config.stagingPrefix(),
+                        "mode",
+                        config.mode()));
     }
 
     /**
@@ -91,6 +124,12 @@ public class WebhookInputSource implements InputSource {
     @Override
     public List<ResolvedInput> resolve(InputSpec spec, ResolveContext ctx) throws IOException {
         WebhookConfig config = WebhookConfig.from(spec.options());
+        if (config.usesConnection()) {
+            // Durable staging: the delivery objects live in the connection's bucket and are read,
+            // claimed, and consumed by the S3 source (no principal here - the save-time check the
+            // resolver trusts already ran).
+            return s3InputSource.resolve(stagingSpec(config), ctx);
+        }
         Path dir = spool.dirFor(config.webhookId());
         if (!Files.isDirectory(dir)) {
             // No deliveries yet (or none since the last consume): a verifiably empty source, so the

app/proprietary/src/main/java/stirling/software/proprietary/policy/webhook/WebhookConfig.java

@@ -5,21 +5,30 @@ import java.util.Map;
 /**
  * Config for a webhook input source, parsed from a spec's options map. A webhook is a push source:
  * external systems POST documents to a signed URL keyed by {@code webhookId}, and each delivery is
- * verified against {@code signingSecret} before it is spooled for the referencing policies. Both
+ * verified against {@code signingSecret} before it is staged for the referencing policies. Both
  * {@code webhookId} (the public URL token) and {@code signingSecret} (the HMAC key) are generated
- * server-side on create - a client never supplies them - so the receiver's identity and the
- * sender's proof of authenticity are always Stirling's own. {@code mode} is "consume" (default: a
- * spooled document is deleted once every policy that claimed it has settled successfully) or
- * "snapshot" (stateless, every run re-reads whatever is currently spooled).
+ * server-side on create - a client never supplies them.
+ *
+ * <p>Where deliveries are staged depends on {@code connectionId}: absent, they go to a node-local
+ * spool directory (self-hosted); present, they go to the S3 {@code IntegrationConfig} connection it
+ * references, under a reserved per-webhook prefix, so the staging is durable and reachable from any
+ * node (the model hosted/SaaS needs). Either way {@code mode} is "consume" (default: a staged
+ * document is removed once every policy that claimed it has settled successfully) or "snapshot"
+ * (stateless, every run re-reads whatever is currently staged).
  */
-public record WebhookConfig(String webhookId, String signingSecret, boolean snapshot) {
+public record WebhookConfig(
+        String webhookId, String signingSecret, Long connectionId, boolean snapshot) {
 
     public static final String WEBHOOK_ID_OPTION = "webhookId";
     public static final String SIGNING_SECRET_OPTION = "signingSecret";
+    public static final String CONNECTION_ID_OPTION = "connectionId";
     private static final String MODE_OPTION = "mode";
     private static final String MODE_CONSUME = "consume";
     private static final String MODE_SNAPSHOT = "snapshot";
 
+    /** Reserved key namespace webhook deliveries are staged under in a connection's bucket. */
+    private static final String STAGING_ROOT = "stirling-webhook";
+
     public static WebhookConfig from(Map<String, Object> options) {
         String webhookId = trimmed(options.get(WEBHOOK_ID_OPTION));
         if (webhookId == null) {
@@ -37,7 +46,43 @@ public record WebhookConfig(String webhookId, String signingSecret, boolean snap
             throw new IllegalArgumentException(
                     "webhook config 'mode' must be 'consume' or 'snapshot'");
         }
-        return new WebhookConfig(webhookId, signingSecret, MODE_SNAPSHOT.equals(mode));
+        return new WebhookConfig(
+                webhookId, signingSecret, connectionId(options), MODE_SNAPSHOT.equals(mode));
+    }
+
+    /** Whether deliveries are staged to a durable S3 connection rather than the local spool. */
+    public boolean usesConnection() {
+        return connectionId != null;
+    }
+
+    /**
+     * The reserved key prefix this webhook's deliveries are staged under inside its connection's
+     * bucket. Server-derived from the (unguessable) id, so two webhooks sharing a connection never
+     * collide and an S3 source pointed at the same bucket won't see another webhook's inbox.
+     */
+    public String stagingPrefix() {
+        return STAGING_ROOT + "/" + webhookId;
+    }
+
+    /** The mode string as the S3 source expects it. */
+    public String mode() {
+        return snapshot ? MODE_SNAPSHOT : MODE_CONSUME;
+    }
+
+    private static Long connectionId(Map<String, Object> options) {
+        Object reference = options.get(CONNECTION_ID_OPTION);
+        if (reference == null || (reference instanceof String s && s.isBlank())) {
+            return null;
+        }
+        if (reference instanceof Number number) {
+            return number.longValue();
+        }
+        try {
+            return Long.valueOf(reference.toString().trim());
+        } catch (NumberFormatException e) {
+            throw new IllegalArgumentException(
+                    "webhook 'connectionId' is not a valid connection reference: " + reference);
+        }
     }
 
     private static String trimmed(Object value) {
@@ -51,6 +96,12 @@ public record WebhookConfig(String webhookId, String signingSecret, boolean snap
     /** Never prints the signing secret, so an accidental log line cannot leak it. */
     @Override
     public String toString() {
-        return "WebhookConfig[webhookId=" + webhookId + ", snapshot=" + snapshot + "]";
+        return "WebhookConfig[webhookId="
+                + webhookId
+                + ", connectionId="
+                + connectionId
+                + ", snapshot="
+                + snapshot
+                + "]";
     }
 }

app/proprietary/src/main/java/stirling/software/proprietary/policy/webhook/WebhookReceiverController.java

@@ -3,6 +3,7 @@ package stirling.software.proprietary.policy.webhook;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.Map;
 
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty;
 import org.springframework.http.HttpStatus;
@@ -25,10 +26,18 @@ import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 
 import stirling.software.common.model.ApplicationProperties;
+import stirling.software.proprietary.policy.s3.S3Config;
+import stirling.software.proprietary.policy.s3.S3ConnectionPool;
+import stirling.software.proprietary.policy.s3.S3ConnectionResolver;
 import stirling.software.proprietary.policy.source.Source;
 import stirling.software.proprietary.policy.source.SourceStore;
 import stirling.software.proprietary.policy.trigger.WebhookTrigger;
 
+import software.amazon.awssdk.core.exception.SdkException;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+
 /**
  * Public receiver for webhook input sources. External systems POST a document to {@code
  * /api/v1/webhooks/{webhookId}}; the request is authenticated not by a login session but by an
@@ -62,6 +71,8 @@ public class WebhookReceiverController {
     private final WebhookSpool spool;
     private final WebhookTrigger webhookTrigger;
     private final ApplicationProperties applicationProperties;
+    private final S3ConnectionResolver connectionResolver;
+    private final S3ConnectionPool connectionPool;
 
     @PostMapping("/{webhookId}")
     @Operation(
@@ -97,16 +108,10 @@ public class WebhookReceiverController {
             throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "Empty request body");
         }
 
-        String storedName;
-        try {
-            storedName =
-                    WebhookSpool.displayName(
-                            spool.store(webhookId, filename, body).getFileName().toString());
-        } catch (IOException e) {
-            log.error("Could not spool webhook delivery for {}: {}", webhookId, e.getMessage());
-            throw new ResponseStatusException(
-                    HttpStatus.INTERNAL_SERVER_ERROR, "Could not store delivery");
-        }
+        String storedName =
+                config.usesConnection()
+                        ? stageToConnection(config, filename, body)
+                        : stageToSpool(webhookId, filename, body);
 
         // Fire the referencing policies now; the trigger's reconcile is the safety net.
         webhookTrigger.fireForWebhook(webhookId);
@@ -134,6 +139,58 @@ public class WebhookReceiverController {
         return null;
     }
 
+    /** Stage a delivery to the node-local spool; returns its display (original) name. */
+    private String stageToSpool(String webhookId, String filename, byte[] body) {
+        try {
+            return WebhookSpool.displayName(
+                    spool.store(webhookId, filename, body).getFileName().toString());
+        } catch (IOException e) {
+            log.error("Could not spool webhook delivery for {}: {}", webhookId, e.getMessage());
+            throw new ResponseStatusException(
+                    HttpStatus.INTERNAL_SERVER_ERROR, "Could not store delivery");
+        }
+    }
+
+    /**
+     * Stage a delivery to the webhook's S3 connection under its reserved prefix. Resolved with no
+     * principal - the referencing source's right to use the connection was checked when it was
+     * saved - so a public delivery never needs the caller to hold connection rights.
+     */
+    private String stageToConnection(WebhookConfig config, String filename, byte[] body) {
+        S3Config s3 =
+                connectionResolver.resolve(
+                        Map.of(
+                                WebhookConfig.CONNECTION_ID_OPTION,
+                                config.connectionId(),
+                                "prefix",
+                                config.stagingPrefix()));
+        S3Client client = connectionPool.clientFor(s3);
+        String key = keyPrefix(s3.prefix()) + WebhookSpool.objectKeySuffix(filename);
+        try {
+            client.putObject(
+                    PutObjectRequest.builder().bucket(s3.bucket()).key(key).build(),
+                    RequestBody.fromBytes(body));
+        } catch (SdkException e) {
+            log.error(
+                    "Could not stage webhook delivery for {} to s3://{}/{}: {}",
+                    config.webhookId(),
+                    s3.bucket(),
+                    key,
+                    e.getMessage());
+            throw new ResponseStatusException(
+                    HttpStatus.INTERNAL_SERVER_ERROR, "Could not store delivery");
+        }
+        return WebhookSpool.objectDisplayName(filename);
+    }
+
+    /** The configured prefix as a key-path prefix ("inbox" and "inbox/" mean the same). */
+    private static String keyPrefix(String prefix) {
+        if (prefix == null || prefix.isEmpty() || prefix.endsWith("/")) {
+            return prefix == null ? "" : prefix;
+        }
+        return prefix + "/";
+    }
+
     /**
      * Read the body into memory, capped at {@code policies.webhookMaxBytes}. Reading one byte past
      * the limit is enough to reject an over-sized (or unbounded, chunked) delivery with 413 before

app/proprietary/src/main/java/stirling/software/proprietary/policy/webhook/WebhookSpool.java

@@ -90,6 +90,23 @@ public class WebhookSpool {
         return spoolFileName;
     }
 
+    /**
+     * The object-key suffix for an S3-staged delivery: a unique segment then the sanitised name
+     * ({@code <unique>/<name>}). A subfolder rather than a name prefix so the object's basename -
+     * and thus the resolved input's filename - stays the clean original name, matching the local
+     * spool.
+     */
+    public static String objectKeySuffix(String filename) {
+        return UUID.randomUUID().toString().replace("-", "") + "/" + sanitize(filename);
+    }
+
+    /**
+     * The sanitised display name of a staged delivery (the basename of {@link #objectKeySuffix}).
+     */
+    public static String objectDisplayName(String filename) {
+        return sanitize(filename);
+    }
+
     /** Reduce a client-supplied filename to a safe bare basename; fall back to a default. */
     private static String sanitize(String filename) {
         if (filename == null) {

app/proprietary/src/test/java/stirling/software/proprietary/policy/input/WebhookInputSourceTest.java

@@ -6,7 +6,10 @@ import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 import java.io.IOException;
 import java.nio.file.Files;
@@ -21,6 +24,7 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.io.TempDir;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 
@@ -42,6 +46,7 @@ class WebhookInputSourceTest {
     private static final String WEBHOOK_ID = "testwebhookid1234";
 
     @Mock private FileReadinessChecker readinessChecker;
+    @Mock private S3InputSource s3InputSource;
 
     @TempDir Path tempDir;
 
@@ -53,7 +58,7 @@ class WebhookInputSourceTest {
     @BeforeEach
     void setUp() {
         spool = new WebhookSpool(tempDir.resolve("spool"));
-        source = new WebhookInputSource(spool, readinessChecker);
+        source = new WebhookInputSource(spool, readinessChecker, s3InputSource);
         ledger = new InProcessProcessedLedger();
         ctx = new RecordingContext();
         lenient().when(readinessChecker.isReady(any())).thenReturn(true);
@@ -146,6 +151,53 @@ class WebhookInputSourceTest {
         assertEquals("keepme", prepared.get("signingSecret"));
     }
 
+    @Test
+    void aConnectionBackedWebhookDelegatesToTheS3SourceUnderItsReservedPrefix() throws IOException {
+        when(s3InputSource.resolve(any(), any())).thenReturn(List.of());
+        InputSpec spec =
+                new InputSpec(
+                        "webhook",
+                        Map.of(
+                                "webhookId",
+                                WEBHOOK_ID,
+                                "signingSecret",
+                                "secret",
+                                "mode",
+                                "consume",
+                                "connectionId",
+                                7));
+
+        source.resolve(spec, ctx);
+
+        ArgumentCaptor<InputSpec> delegated = ArgumentCaptor.forClass(InputSpec.class);
+        verify(s3InputSource).resolve(delegated.capture(), eq(ctx));
+        InputSpec s3 = delegated.getValue();
+        assertEquals("s3", s3.type());
+        assertEquals(7L, ((Number) s3.options().get("connectionId")).longValue());
+        assertEquals("stirling-webhook/" + WEBHOOK_ID, s3.options().get("prefix"));
+        assertEquals("consume", s3.options().get("mode"));
+    }
+
+    @Test
+    void aConnectionBackedWebhookValidatesTheConnectionThroughTheS3Source() {
+        InputSpec spec =
+                new InputSpec(
+                        "webhook",
+                        Map.of(
+                                "webhookId",
+                                WEBHOOK_ID,
+                                "signingSecret",
+                                "secret",
+                                "connectionId",
+                                7));
+
+        source.validate(spec);
+
+        // Save-time validation defers to the S3 source, which resolves + ownership-checks the
+        // connection (the check the delivery path later trusts).
+        verify(s3InputSource).validate(any());
+    }
+
     /** Policy-scoped context backed by the in-process ledger, recording presence reports. */
     private class RecordingContext implements ResolveContext {
 

app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceControllerTest.java

@@ -28,6 +28,7 @@ import stirling.software.common.util.FileReadinessChecker;
 import stirling.software.proprietary.policy.config.PolicyAccessGuard;
 import stirling.software.proprietary.policy.config.PolicyManagementAuthority;
 import stirling.software.proprietary.policy.input.InputSource;
+import stirling.software.proprietary.policy.input.S3InputSource;
 import stirling.software.proprietary.policy.input.WebhookInputSource;
 import stirling.software.proprietary.policy.model.OutputSpec;
 import stirling.software.proprietary.policy.model.PipelineStep;
@@ -88,7 +89,10 @@ class SourceControllerTest {
                         properties,
                         List.of(folderInput));
         WebhookInputSource webhookInput =
-                new WebhookInputSource(new WebhookSpool(tempDir), mock(FileReadinessChecker.class));
+                new WebhookInputSource(
+                        new WebhookSpool(tempDir),
+                        mock(FileReadinessChecker.class),
+                        mock(S3InputSource.class));
         webhookController =
                 new SourceController(
                         sourceStore,

app/proprietary/src/test/java/stirling/software/proprietary/policy/webhook/WebhookReceiverControllerTest.java

@@ -23,6 +23,8 @@ import org.springframework.mock.web.MockHttpServletRequest;
 import org.springframework.web.server.ResponseStatusException;
 
 import stirling.software.common.model.ApplicationProperties;
+import stirling.software.proprietary.policy.s3.S3ConnectionPool;
+import stirling.software.proprietary.policy.s3.S3ConnectionResolver;
 import stirling.software.proprietary.policy.source.InProcessSourceStore;
 import stirling.software.proprietary.policy.source.Source;
 import stirling.software.proprietary.policy.source.SourceStore;
@@ -55,7 +57,15 @@ class WebhookReceiverControllerTest {
         spool = new WebhookSpool(tempDir.resolve("spool"));
         trigger = mock(WebhookTrigger.class);
         properties = new ApplicationProperties();
-        controller = new WebhookReceiverController(sourceStore, spool, trigger, properties);
+        // The local-disk tests never touch a connection; mocks stand in for the S3 collaborators.
+        controller =
+                new WebhookReceiverController(
+                        sourceStore,
+                        spool,
+                        trigger,
+                        properties,
+                        mock(S3ConnectionResolver.class),
+                        mock(S3ConnectionPool.class));
     }
 
     private static Source webhookSource(boolean enabled) {

app/proprietary/src/test/java/stirling/software/proprietary/policy/webhook/WebhookS3ConnectionMinioTest.java

@@ -0,0 +1,273 @@
+package stirling.software.proprietary.policy.webhook;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Supplier;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.springframework.mock.web.MockHttpServletRequest;
+import org.testcontainers.containers.MinIOContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import stirling.software.common.model.ApplicationProperties;
+import stirling.software.common.util.FileReadinessChecker;
+import stirling.software.proprietary.access.service.OwnershipService;
+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.policy.input.ResolveContext;
+import stirling.software.proprietary.policy.input.ResolvedInput;
+import stirling.software.proprietary.policy.input.S3InputSource;
+import stirling.software.proprietary.policy.input.WebhookInputSource;
+import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger;
+import stirling.software.proprietary.policy.model.InputSpec;
+import stirling.software.proprietary.policy.s3.S3ConnectionPool;
+import stirling.software.proprietary.policy.s3.S3ConnectionResolver;
+import stirling.software.proprietary.policy.source.InProcessSourceStore;
+import stirling.software.proprietary.policy.source.Source;
+import stirling.software.proprietary.policy.trigger.WebhookTrigger;
+import stirling.software.proprietary.security.service.UserService;
+
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.S3Configuration;
+import software.amazon.awssdk.services.s3.model.CreateBucketRequest;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
+import software.amazon.awssdk.services.s3.model.S3Object;
+import tools.jackson.databind.ObjectMapper;
+
+/**
+ * End-to-end test of a connection-backed webhook against a real S3 API (MinIO): a signed delivery
+ * is staged by the receiver into the S3 connection's bucket under the reserved per-webhook prefix,
+ * then read and consumed by the source (which delegates to {@link S3InputSource}). Proves the
+ * durable, multi-node staging model hosted deployments use, through the production S3 client
+ * factory.
+ */
+@Testcontainers(disabledWithoutDocker = true)
+class WebhookS3ConnectionMinioTest {
+
+    private static final String POLICY = "p1";
+    private static final String ACCESS_KEY = "minioadmin";
+    private static final String SECRET_KEY = "minioadmin";
+    private static final String WEBHOOK_ID = "miniowebhookid12";
+    private static final String SECRET = "topsecret";
+    private static final long CONNECTION_ID = 7L;
+
+    @Container
+    static MinIOContainer minio =
+            new MinIOContainer("minio/minio:latest")
+                    .withUserName(ACCESS_KEY)
+                    .withPassword(SECRET_KEY);
+
+    private static S3Client adminClient;
+    private static int bucketCounter;
+
+    @TempDir Path tempDir;
+
+    private String bucket;
+    private String stagingPrefix;
+    private WebhookReceiverController receiver;
+    private WebhookInputSource inputSource;
+    private InProcessProcessedLedger ledger;
+    private RecordingContext ctx;
+
+    @BeforeEach
+    void setUp() {
+        if (adminClient == null) {
+            adminClient =
+                    S3Client.builder()
+                            .endpointOverride(URI.create(minio.getS3URL()))
+                            .httpClient(UrlConnectionHttpClient.create())
+                            .region(Region.US_EAST_1)
+                            .credentialsProvider(
+                                    StaticCredentialsProvider.create(
+                                            AwsBasicCredentials.create(ACCESS_KEY, SECRET_KEY)))
+                            .serviceConfiguration(
+                                    S3Configuration.builder().pathStyleAccessEnabled(true).build())
+                            .build();
+        }
+        bucket = "webhook-inbox-" + ++bucketCounter;
+        adminClient.createBucket(CreateBucketRequest.builder().bucket(bucket).build());
+        stagingPrefix = "stirling-webhook/" + WEBHOOK_ID + "/";
+
+        // The MinIO endpoint resolves to loopback, so the operator opt-in must be on.
+        ApplicationProperties properties = new ApplicationProperties();
+        properties.getPolicies().setAllowPrivateS3Endpoints(true);
+
+        // A resolver over a stored S3 connection pointing at MinIO. No SecurityContext in the test,
+        // so the resolver's ownership check is skipped (the delivery path's trust-the-save model).
+        S3ConnectionResolver resolver = resolverFor(minioConnection());
+        S3ConnectionPool pool = new S3ConnectionPool(properties);
+        S3InputSource s3 = new S3InputSource(pool, resolver);
+
+        inputSource =
+                new WebhookInputSource(
+                        new WebhookSpool(tempDir.resolve("spool")),
+                        mock(FileReadinessChecker.class),
+                        s3);
+
+        InProcessSourceStore sourceStore = new InProcessSourceStore();
+        sourceStore.save(
+                new Source(
+                        "s1",
+                        "Partner uploads",
+                        "webhook",
+                        Map.of(
+                                "webhookId", WEBHOOK_ID,
+                                "signingSecret", SECRET,
+                                "mode", "consume",
+                                "connectionId", CONNECTION_ID),
+                        true,
+                        "owner",
+                        null));
+        receiver =
+                new WebhookReceiverController(
+                        sourceStore,
+                        new WebhookSpool(tempDir.resolve("spool")),
+                        mock(WebhookTrigger.class),
+                        properties,
+                        resolver,
+                        pool);
+        ledger = new InProcessProcessedLedger();
+        ctx = new RecordingContext();
+    }
+
+    @Test
+    void aDeliveryIsStagedToTheConnectionThenReadAndConsumed() throws IOException {
+        byte[] body = "a pdf".getBytes(StandardCharsets.UTF_8);
+        String signature = WebhookSignatures.sign(SECRET, body);
+
+        // Deliver: the receiver stages the object into the connection's bucket under the prefix.
+        var response = receiver.receive(WEBHOOK_ID, signature, "invoice.pdf", request(body));
+        assertThat(response.getStatusCode().value()).isEqualTo(202);
+
+        List<S3Object> staged = listUnder(stagingPrefix);
+        assertThat(staged).hasSize(1);
+        // Staged under a unique subfolder so the object basename stays the clean original name.
+        assertThat(staged.get(0).key()).endsWith("/invoice.pdf");
+
+        // Read: the webhook source resolves the delivery by delegating to the S3 source.
+        List<ResolvedInput> work = inputSource.resolve(webhookSpec(), ctx);
+        assertThat(work).hasSize(1);
+        assertThat(work.get(0).inputs().primary().get(0).getFilename()).isEqualTo("invoice.pdf");
+        assertThat(read(work.get(0))).isEqualTo("a pdf");
+        // In flight: a second sweep claims nothing.
+        assertThat(inputSource.resolve(webhookSpec(), ctx)).isEmpty();
+
+        // Consume: a successful run removes the staged object from the bucket.
+        work.get(0).onComplete().accept(true);
+        assertThat(listUnder(stagingPrefix)).isEmpty();
+        assertThat(inputSource.resolve(webhookSpec(), ctx)).isEmpty();
+    }
+
+    @Test
+    void aRejectedDeliveryStagesNothing() {
+        byte[] body = "a pdf".getBytes(StandardCharsets.UTF_8);
+
+        try {
+            receiver.receive(WEBHOOK_ID, "sha256=deadbeef", "x.pdf", request(body));
+        } catch (RuntimeException expected) {
+            // 401 invalid signature
+        }
+
+        assertThat(listUnder(stagingPrefix)).isEmpty();
+    }
+
+    private IntegrationConfig minioConnection() {
+        IntegrationConfig connection = new IntegrationConfig();
+        connection.setId(CONNECTION_ID);
+        connection.setIntegrationType(IntegrationType.S3);
+        connection.setName("minio");
+        connection.setEnabled(true);
+        connection.setConfig(
+                new ObjectMapper()
+                        .writeValueAsString(
+                                Map.of(
+                                        "bucket", bucket,
+                                        "region", "us-east-1",
+                                        "endpoint", minio.getS3URL(),
+                                        "accessKeyId", ACCESS_KEY,
+                                        "secretAccessKey", SECRET_KEY)));
+        return connection;
+    }
+
+    private static S3ConnectionResolver resolverFor(IntegrationConfig connection) {
+        IntegrationConfigRepository connections = mock(IntegrationConfigRepository.class);
+        when(connections.findById(connection.getId())).thenReturn(Optional.of(connection));
+        return new S3ConnectionResolver(
+                connections, mock(OwnershipService.class), mock(UserService.class));
+    }
+
+    private static InputSpec webhookSpec() {
+        return new InputSpec(
+                "webhook",
+                Map.of(
+                        "webhookId", WEBHOOK_ID,
+                        "signingSecret", SECRET,
+                        "mode", "consume",
+                        "connectionId", CONNECTION_ID));
+    }
+
+    private static MockHttpServletRequest request(byte[] body) {
+        MockHttpServletRequest req =
+                new MockHttpServletRequest("POST", "/api/v1/webhooks/" + WEBHOOK_ID);
+        req.setContent(body);
+        return req;
+    }
+
+    private List<S3Object> listUnder(String prefix) {
+        return adminClient
+                .listObjectsV2(ListObjectsV2Request.builder().bucket(bucket).prefix(prefix).build())
+                .contents();
+    }
+
+    private static String read(ResolvedInput unit) throws IOException {
+        try (InputStream stream = unit.inputs().primary().get(0).getInputStream()) {
+            return new String(stream.readAllBytes(), StandardCharsets.UTF_8);
+        }
+    }
+
+    private class RecordingContext implements ResolveContext {
+
+        private final List<String> present = new ArrayList<>();
+
+        @Override
+        public boolean claim(String identity, String gate, Supplier<String> contentHash) {
+            return ledger.claim(POLICY, identity, gate, contentHash);
+        }
+
+        @Override
+        public void settle(
+                String identity, String finalGate, String finalContentHash, boolean success) {
+            ledger.settle(POLICY, identity, finalGate, finalContentHash, success);
+        }
+
+        @Override
+        public boolean allSettledDone(String identity) {
+            return ledger.allSettledDone(identity);
+        }
+
+        @Override
+        public void reportPresent(Collection<String> identities) {
+            present.addAll(identities);
+        }
+    }
+}

frontend/editor/public/locales/en-US/translation.toml

@@ -8147,6 +8147,10 @@ label = "Webhook"
 deliveryUrl = "Delivery URL"
 secretNote = "The signing secret is shown once, when the webhook is created. Recreate the source to roll it."
 
+[portal.sources.types.webhook.fields.connection]
+helperText = "Choose an S3 connection to stage deliveries durably (required in hosted). Leave blank on a self-hosted server to stage them on local disk."
+label = "Storage connection"
+
 [portal.sources.types.webhook.fields.mode]
 helperText = "Consume removes each delivered document once every policy has processed it."
 label = "Read mode"

frontend/editor/src/portal-saas/components/sources/creatableSourceTypes.test.ts

@@ -3,10 +3,17 @@ import { describe, expect, it } from "vitest";
 import { creatableSourceTypes } from "@portal/components/sources/creatableSourceTypes";
 
 describe("creatableSourceTypes (SaaS)", () => {
-  it("never offers server-local-disk sources (folder, webhook) in hosted deployments", () => {
-    const offered = creatableSourceTypes().map((t) => t.type);
-    expect(offered).not.toContain("folder");
-    expect(offered).not.toContain("webhook");
+  it("never offers folder sources: hosted deployments do not read the server filesystem", () => {
+    expect(creatableSourceTypes().map((t) => t.type)).not.toContain("folder");
+  });
+
+  it("offers webhook with a required S3 connection (durable staging in hosted)", () => {
+    const webhook = creatableSourceTypes().find((t) => t.type === "webhook");
+    expect(webhook).toBeDefined();
+    const connection = webhook?.fields.find(
+      (f) => f.control === "s3Connection",
+    );
+    expect(connection?.required).toBe(true);
   });
 
   it("still offers the cloud source types", () => {

frontend/editor/src/portal-saas/components/sources/creatableSourceTypes.ts

@@ -1,19 +1,31 @@
 import {
   CREATABLE_SOURCE_TYPES,
+  WEBHOOK_SOURCE_TYPE,
   type CreatableSourceType,
 } from "@portal/components/sources/sourceTypes";
 
 /**
- * Hosted deployments don't rely on the server's local filesystem, so the two
- * server-local-disk source types are not offered in the connect wizard: folder
- * (denied outright by the backend's FolderAccessGuard) and webhook (whose
- * delivery spool is node-local, so it can't be relied on across a multi-node
- * fleet). Cloud sources such as S3 remain.
+ * Hosted deployments don't rely on the server's local filesystem. Folder sources
+ * are dropped outright (denied by the backend's FolderAccessGuard). Webhook stays
+ * - it works in hosted by staging deliveries to a durable S3 connection - but its
+ * connection becomes required, since the node-local fallback can't be relied on
+ * across a multi-node fleet. Cloud sources such as S3 are unchanged.
  */
-const SERVER_LOCAL_TYPES = new Set(["folder", "webhook"]);
-
 export function creatableSourceTypes(): CreatableSourceType[] {
-  return CREATABLE_SOURCE_TYPES.filter(
-    (type) => !SERVER_LOCAL_TYPES.has(type.type),
+  return CREATABLE_SOURCE_TYPES.filter((type) => type.type !== "folder").map(
+    (type) =>
+      type.type === WEBHOOK_SOURCE_TYPE ? withRequiredConnection(type) : type,
   );
 }
+
+/** The type with its S3-connection field marked required (durable staging is mandatory). */
+function withRequiredConnection(
+  type: CreatableSourceType,
+): CreatableSourceType {
+  return {
+    ...type,
+    fields: type.fields.map((field) =>
+      field.control === "s3Connection" ? { ...field, required: true } : field,
+    ),
+  };
+}

frontend/editor/src/portal/components/sources/sourceTypes.ts

@@ -22,6 +22,9 @@ export interface SourceTypeMeta {
  */
 export const EDITOR_SOURCE_TYPE = "editor";
 
+/** The webhook source type. Its delivery URL + signing secret are minted server-side on create. */
+export const WEBHOOK_SOURCE_TYPE = "webhook";
+
 const SOURCE_TYPE_META: Record<string, SourceTypeMeta> = {
   folder: {
     labelKey: "portal.sources.types.folder.label",
@@ -38,6 +41,11 @@ const SOURCE_TYPE_META: Record<string, SourceTypeMeta> = {
     icon: "☁",
     accent: "brand",
   },
+  webhook: {
+    labelKey: "portal.sources.types.webhook.label",
+    icon: "↯",
+    accent: "warning",
+  },
 };
 
 const UNKNOWN_TYPE_META: SourceTypeMeta = {
@@ -180,6 +188,42 @@ export const CREATABLE_SOURCE_TYPES: CreatableSourceType[] = [
       },
     ],
   },
+  {
+    // The delivery URL + signing secret are minted server-side on create (revealed once). An S3
+    // connection makes staging durable + multi-node (required in hosted); left blank, deliveries
+    // stage to the server's local disk (self-hosted only).
+    type: WEBHOOK_SOURCE_TYPE,
+    labelKey: "portal.sources.types.webhook.label",
+    descriptionKey: "portal.sources.types.webhook.description",
+    fields: [
+      {
+        key: "connectionId",
+        labelKey: "portal.sources.types.webhook.fields.connection.label",
+        control: "s3Connection",
+        helperTextKey:
+          "portal.sources.types.webhook.fields.connection.helperText",
+      },
+      {
+        key: "mode",
+        labelKey: "portal.sources.types.webhook.fields.mode.label",
+        control: "select",
+        defaultValue: "consume",
+        helperTextKey: "portal.sources.types.webhook.fields.mode.helperText",
+        options: [
+          {
+            value: "consume",
+            labelKey:
+              "portal.sources.types.webhook.fields.mode.options.consume",
+          },
+          {
+            value: "snapshot",
+            labelKey:
+              "portal.sources.types.webhook.fields.mode.options.snapshot",
+          },
+        ],
+      },
+    ],
+  },
 ];
 
 /** Default option values for a type's create form. */

frontend/editor/src/portal/views/SourceBuilder.css

@@ -85,3 +85,27 @@
   justify-content: flex-end;
   gap: 0.5rem;
 }
+
+/* Webhook delivery URL / signing secret: read-only value + copy button. */
+.portal-source-builder__copy-row {
+  display: flex;
+  gap: 0.5rem;
+  align-items: center;
+}
+
+.portal-source-builder__copy-row > :first-child {
+  flex: 1 1 auto;
+  min-width: 0;
+}
+
+.portal-source-builder__reveal {
+  display: flex;
+  flex-direction: column;
+  gap: 1rem;
+}
+
+.portal-source-builder__muted {
+  margin: 0;
+  font-size: 0.8125rem;
+  color: var(--color-text-4);
+}

frontend/editor/src/portal/views/SourceBuilder.test.tsx

@@ -102,6 +102,40 @@ describe("SourceBuilder", () => {
     ).toBeDisabled();
   });
 
+  it("reveals the delivery URL and signing secret once after creating a webhook", async () => {
+    createSource.mockResolvedValue({
+      id: "wh-1",
+      options: { webhookId: "whk_abc123", signingSecret: "whsec_topsecret" },
+    });
+    renderBuilder("/processor/sources/new");
+
+    fireEvent.change(screen.getByLabelText(/portal\.sources\.wizard\.name/), {
+      target: { value: "Partner uploads" },
+    });
+    // Webhook's connection is optional (self-hosted local-disk), so a name is enough to create.
+    fireEvent.click(screen.getByText("portal.sources.types.webhook.label"));
+    fireEvent.click(screen.getByText("portal.sources.builder.create"));
+
+    await waitFor(() => expect(createSource).toHaveBeenCalledTimes(1));
+    expect(createSource).toHaveBeenCalledWith(
+      expect.objectContaining({ type: "webhook", name: "Partner uploads" }),
+    );
+
+    // The reveal shows the secret + URL and stays put until Done is clicked.
+    expect(
+      await screen.findByDisplayValue("whsec_topsecret"),
+    ).toBeInTheDocument();
+    expect(
+      screen.getByDisplayValue(/\/api\/v1\/webhooks\/whk_abc123$/),
+    ).toBeInTheDocument();
+    expect(screen.queryByText("sources list")).not.toBeInTheDocument();
+
+    fireEvent.click(
+      screen.getByText("portal.sources.types.webhook.reveal.done"),
+    );
+    expect(await screen.findByText("sources list")).toBeInTheDocument();
+  });
+
   it("blocks create until required fields are filled", async () => {
     renderBuilder("/processor/sources/new");
     // Name given but directory (required) still blank -> Create disabled.

frontend/editor/src/portal/views/SourceBuilder.tsx

@@ -27,11 +27,17 @@ import {
   CREATABLE_SOURCE_TYPES,
   defaultOptions,
   sourceTypeMeta,
+  WEBHOOK_SOURCE_TYPE,
   type CreatableSourceType,
 } from "@portal/components/sources/sourceTypes";
 import { S3ConnectionPicker } from "@portal/components/sources/S3ConnectionPicker";
 import "@portal/views/SourceBuilder.css";
 
+/** Absolute delivery URL a sender POSTs to for a webhook's routing id. */
+function webhookUrl(webhookId: string): string {
+  return `${window.location.origin}/api/v1/webhooks/${webhookId}`;
+}
+
 const OFFERED_TYPES = creatableSourceTypes();
 
 /** A source's stored type resolved to its create-form metadata (edit falls back to any type). */
@@ -84,6 +90,11 @@ export function SourceBuilder() {
   const [error, setError] = useState<string | null>(null);
   const [pendingDelete, setPendingDelete] = useState(false);
   const [deleting, setDeleting] = useState(false);
+  // Set after a webhook is created: its one-time delivery id + signing secret, shown before leaving.
+  const [reveal, setReveal] = useState<{
+    webhookId: string;
+    secret: string;
+  } | null>(null);
 
   // Seed once: immediately for a new source, or after the record loads for edit.
   useEffect(() => {
@@ -112,18 +123,42 @@ export function SourceBuilder() {
   );
   const canSave = name.trim() !== "" && requiredComplete && !submitting;
 
+  // An existing webhook's delivery URL, shown read-only in the edit view (the secret is not - it
+  // is revealed only once, at creation).
+  const editingWebhookId =
+    isEdit && sourceState.data?.type === WEBHOOK_SOURCE_TYPE
+      ? String(sourceState.data.options?.webhookId ?? "")
+      : "";
+  const revealUrl = reveal ? webhookUrl(reveal.webhookId) : "";
+  const revealSecret = reveal ? reveal.secret : "";
+
+  function dismissReveal() {
+    setReveal(null);
+    navigate(listPath);
+  }
+
   async function save() {
     if (!canSave) return;
     setSubmitting(true);
     setError(null);
     try {
-      await createSource({
+      const saved = await createSource({
         id: isEdit ? id : undefined,
         name: name.trim(),
         type: type.type,
         options,
         enabled,
       });
+      // A new webhook returns its server-minted routing id + signing secret once; reveal them
+      // (with the delivery URL) before leaving so the operator can copy the secret.
+      if (!isEdit && type.type === WEBHOOK_SOURCE_TYPE) {
+        const webhookId = String(saved.options?.webhookId ?? "");
+        const secret = String(saved.options?.signingSecret ?? "");
+        if (webhookId && secret) {
+          setReveal({ webhookId, secret });
+          return;
+        }
+      }
       navigate(listPath);
     } catch (e) {
       setError(errorMessage(e));
@@ -131,6 +166,10 @@ export function SourceBuilder() {
     }
   }
 
+  function copy(text: string) {
+    void navigator.clipboard?.writeText(text);
+  }
+
   async function confirmDelete() {
     if (!id || deleting) return;
     setDeleting(true);
@@ -290,6 +329,28 @@ export function SourceBuilder() {
           </FormField>
         ))}
 
+        {editingWebhookId && (
+          <FormField
+            label={t("portal.sources.types.webhook.detail.deliveryUrl")}
+            helperText={t("portal.sources.types.webhook.detail.secretNote")}
+          >
+            <div className="portal-source-builder__copy-row">
+              <Input
+                value={webhookUrl(editingWebhookId)}
+                readOnly
+                onFocus={(e) => e.currentTarget.select()}
+              />
+              <Button
+                variant="tertiary"
+                size="sm"
+                onClick={() => copy(webhookUrl(editingWebhookId))}
+              >
+                {t("portal.sources.types.webhook.reveal.copy")}
+              </Button>
+            </div>
+          </FormField>
+        )}
+
         {error && <Banner tone="danger" description={error} />}
       </div>
 
@@ -321,6 +382,69 @@ export function SourceBuilder() {
       >
         <p>{t("portal.sources.delete.body", { name })}</p>
       </Modal>
+
+      <Modal
+        open={reveal !== null}
+        onClose={dismissReveal}
+        width="md"
+        title={t("portal.sources.types.webhook.reveal.title")}
+        footer={
+          <div className="portal-source-builder__delete-actions">
+            <Button size="sm" onClick={dismissReveal}>
+              {t("portal.sources.types.webhook.reveal.done")}
+            </Button>
+          </div>
+        }
+      >
+        {reveal && (
+          <div className="portal-source-builder__reveal">
+            <Banner
+              tone="warning"
+              description={t(
+                "portal.sources.types.webhook.reveal.secretWarning",
+              )}
+            />
+            <FormField label={t("portal.sources.types.webhook.reveal.url")}>
+              <div className="portal-source-builder__copy-row">
+                <Input
+                  value={revealUrl}
+                  readOnly
+                  onFocus={(e) => e.currentTarget.select()}
+                />
+                <Button
+                  variant="tertiary"
+                  size="sm"
+                  onClick={() => copy(revealUrl)}
+                >
+                  {t("portal.sources.types.webhook.reveal.copy")}
+                </Button>
+              </div>
+            </FormField>
+            <FormField
+              label={t("portal.sources.types.webhook.reveal.secret")}
+              helperText={t("portal.sources.types.webhook.reveal.secretHelp")}
+            >
+              <div className="portal-source-builder__copy-row">
+                <Input
+                  value={revealSecret}
+                  readOnly
+                  onFocus={(e) => e.currentTarget.select()}
+                />
+                <Button
+                  variant="tertiary"
+                  size="sm"
+                  onClick={() => copy(revealSecret)}
+                >
+                  {t("portal.sources.types.webhook.reveal.copy")}
+                </Button>
+              </div>
+            </FormField>
+            <p className="portal-source-builder__muted">
+              {t("portal.sources.types.webhook.reveal.usage")}
+            </p>
+          </div>
+        )}
+      </Modal>
     </div>
   );
 }