diff --git a/.gitignore b/.gitignore index 8ea68f1..a4fc4eb 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ /.gradle/ /.idea/ /build/ +/temp/run.sh diff --git a/CLAUDE.md b/CLAUDE.md index fb39dbc..c5fee68 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -22,6 +22,37 @@ suffix). The Gradle `validateExtensionJars` task enforces this and fails the build if the directory is missing, a JAR is absent, versions are mixed, or multiple versions of either JAR are present. +### Dependency scoping — do not revert + +The `jar` task produces a **minimal fat jar**: only libraries actually used +by extension code are shaded in. Anything the agent provides at runtime +(`extension-api`, Spring, Jackson, Jakarta, SLF4J) is declared `compileOnly` +and stays out of `build/libs/opa-extensions-.jar`. Shading +agent-provided types would cause classloader conflicts in the agent — see +`extension-api/README-extension-api.md` → "do not shade it into your JAR". + +Scoping rules — follow them when adding a new dependency: + +- **Shaded (`implementation`)** — any new third-party library that the agent + does NOT already provide. Today: `commons-codec`, `commons-lang3`. Add a + trailing comment pointing at the extension that uses the library. +- **Provided by the agent (`compileOnly`)** — anything from Spring, Jackson, + Jakarta, SLF4J, Apache HttpClient, or further `com.workato.*` packages. + Compile against it, do not shade. +- **Test-only (`testImplementation`)** — anything only the test harness + needs. `jackson-dataformat-yaml` (used by `AgentExtensionHarness` to parse + YAML fixtures) and `extension-test-*.jar` live here. + +The `configurations { testImplementation.extendsFrom compileOnly }` block at +the top of `dependencies { }` forwards every `compileOnly` entry onto the +test classpath — Gradle does NOT do this automatically, and without this +line `AgentExtensionHarness` loses Spring at test runtime. + +Verify after any dependency change: `./gradlew clean jar && unzip -l +build/libs/opa-extensions-*.jar | grep -c org/springframework` must print +`0`. If a Spring/Jackson/SLF4J class is in the jar you shaded something you +shouldn't have. + ## Authoritative documentation — read before coding The Gradle build extracts the README from each JAR into `extension-api/` on @@ -130,6 +161,41 @@ Current connectors (deliverable → test project): - Real HTTP in a test is fine when it's a tiny in-process server (see `NtlmExtensionTest` → `com.sun.net.httpserver.HttpServer` on `127.0.0.1:0`). +### Mandatory `*DeployedTest` for every extension + +Every extension under `src/main/java/com/mycompany/onprem//` **must** +have a companion `ExtensionDeployedTest.java` under +`src/test/java/com/mycompany/onprem//` that runs the extension inside +the real agent via `DeployedAgentLauncher` (see `README-extension-test.md` → +"Running against the real agent"). This is the only test type that exercises +classloader isolation, the real Jetty binding, and — in gateway mode — the +mTLS tunnel to Workato. `AgentExtensionHarness` cannot cover any of these. + +Conventions, copied verbatim from `SecurityExtensionDeployedTest` / +`NtlmExtensionDeployedTest` / `TimerExtensionDeployedTest`: + +- `@Disabled("Manual-only — needs deployed+activated agent at AGENT_DIST; run one method per JVM.")` + at the class level. These tests mutate process-global state inside the + agent (singleton Spring context, Netty loops, …), so run **one `@Test` + method per JVM invocation** from the IDE — never the whole class at once. +- A `private static final Path AGENT_DIST = Paths.get("...")` at the top, + pointing at the local deployed+activated agent dist. The path is + per-machine — leave it as a constant; the `@Disabled` keeps it out of CI. +- At minimum two methods: + 1. **Local-HTTP smoke test** — one or more happy-path calls via + `openLocalAccess()` + `agent.httpPost(Controller::method, body)` / + `agent.httpGet(Controller::method)`. Asserts controller actually + answers through the real agent dispatcher. + 2. **`connectsToWorkatoGateway`** — no `openLocalAccess()`, call + `agent.awaitGatewayConnected(Duration.ofSeconds(60))`, assert the + returned line contains `"Connected to gateway"`. Proves the dist's + activation certs, trust store, and the extension's bean wiring do + not block mTLS startup. +- If the extension emits outbound HTTP (webhook-style — see + `TimerExtension` + `AgentWebhookApiV1`), start a tiny loopback + `com.sun.net.httpserver.HttpServer` sink in the test to capture the + delivered payload and headers, and assert on them. + ### Writing Ruby tests - Test framework: RSpec + VCR + WebMock (all declared in the shared diff --git a/README.md b/README.md index 3940af9..36b6848 100644 --- a/README.md +++ b/README.md @@ -1,10 +1,26 @@ +OPA extension SDK allows customers, external developers/partners to connect to legacy on-prem systems using Java within +the Workato Agent infrastructure. -OPA extension SDK allows customers, external developers/partners to connect to legacy on-prem systems using Java within the Workato Agent infrastructure. +## Available extensions +Each extension is a standalone Java `@RestController` plus a companion Workato +custom connector. Per-extension documentation (what it does, what Extension +API features it demonstrates, HTTP endpoints, YAML keys it reads) lives next +to its Java class — follow the links. + +| Extension | What it demonstrates | HTTP surface | +|-----------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-------------------------------------------------------------------------| +| **[security](src/main/java/com/mycompany/onprem/security/README.md)** | The minimum-viable extension — `@Inject AgentExtensionApiV1`, YAML read via `getProfileConfiguration()`, one `@PostMapping` endpoint. | `POST /computeDigest` | +| **[ntlm](src/main/java/com/mycompany/onprem/ntlm/README.md)** | `@Value("${…}")` property injection with defaults, lazy-init of a shared `HttpClient` (DCL pattern), `@RequestHeader`-driven inputs, forwarding upstream status via `HttpServletResponse`. | `POST /request` | +| **[timer](src/main/java/com/mycompany/onprem/timer/README.md)** | **`AgentWebhookApiV1.send(...)`** with per-profile retry policy, `webhook_subscribe` / `_unsubscribe` HTTP contract, Jackson records with `@JsonProperty`, background `ScheduledExecutorService` with `DisposableBean.destroy()` lifecycle. | `GET /webhook/health`, `POST /webhook/{subscribe,unsubscribe,fire_now}` | + +Add a new extension by cloning any of the three subdirectories under +`src/main/java/com/mycompany/onprem/` and following the six-step **Development +workflow** below. ## Configuration -Extension mechanism is enabled by adding the `extensions` section to the config file, `conf/config.yml`: +Extension mechanism is enabled by adding the `extensions` section to the config file, `conf/config.yml`: ```yml server: @@ -29,47 +45,22 @@ Endpoint request handling logic is defined by Spring REST controller class, prov controller class is registered inside an endpoint-specific Spring `WebApplicationContext` and allows retrieving configuration properties from the `config.yml` file. -## Endpoint controller sample - -Full code can be -found [here](https://github.com/workato/opa-extensions/blob/master/src/main/java/com/mycompany/onprem/SecurityExtension.java). +## Endpoint controller shape -```java -package com.mycompany.onprem; - -// import rest of the packages here: -@Controller -public class SecurityExtension { - - @Inject - private Environment env; - - @RequestMapping(method = RequestMethod.POST) - public @ResponseBody Map computeDigest(@RequestBody Map body) throws Exception { - Charset encoding = Charset.forName("UTF-8"); - String payload = (String) body.get("payload"); - MessageDigest digest = MessageDigest.getInstance("SHA-256"); - digest.update(env.getProperty("secret").getBytes(encoding)); - byte[] result = digest.digest(payload.getBytes(encoding)); - return Collections.singletonMap("signature", Hex.encodeHexString(result)); - } -} -``` +Every extension is a Spring `@RestController` the agent mounts at +`/ext/{profile}/...`. The authoritative API surface — allowed packages, +injection points, lifecycle, and a worked example — is in +[`extension-api/README-extension-api.md`](extension-api/README-extension-api.md), +which the build regenerates from the Extension API JAR on every compile. -Given the configuration: - -```yml - extensions: - security: - controllerClass: com.mycompany.onprem.security.SecurityExtension - secret: HA63A3043AMMMM -``` +For concrete examples of each pattern in use, open the per-extension READMEs +linked in the **Available extensions** table above. ## Development workflow End-to-end for a new extension. The six steps below mirror the layered test -pyramid that already covers `SecurityExtension` and `NtlmExtension`; each has -its own detailed section further down with code, Gradle commands, and +pyramid that already covers every extension in **Available extensions** above; +each has its own detailed section further down with code, Gradle commands, and troubleshooting. ``` @@ -103,11 +94,14 @@ troubleshooting. ### Step 1 — Write the Java extension class -Put the controller under `src/main/java/com/mycompany/onprem//`. It is a -regular Spring `@RestController` the agent will mount at -`/ext//...`. Follow `extension-api/README-extension-api.md` for -the exhaustive contract (allowed packages, injection points, lifecycle). See -`SecurityExtension.java` / `NtlmExtension.java` for working examples. +Put the controller under `src/main/java/com/mycompany/onprem//` along +with a `README.md` next to it (what it does + which Extension API features it +demonstrates — mirror the shape of the existing per-extension READMEs linked +in **Available extensions**). It is a regular Spring `@RestController` the +agent will mount at `/ext//...`. Follow +`extension-api/README-extension-api.md` for the exhaustive contract (allowed +packages, injection points, lifecycle). Any existing extension listed in +**Available extensions** is a working example. Make the controller methods throw only unchecked exceptions — this keeps their method references usable with the typed `DeployedAgentLauncher` @@ -319,7 +313,34 @@ Steps to build an extension: (see the preceding section). 3. Run `./gradlew jar` — this also triggers `validateExtensionJars` and `extractExtensionReadmes`. -4. The output is in `build/libs`. +4. The output is `build/libs/opa-extensions-.jar`. + +The `jar` task produces a **minimal fat jar** — it bundles only the Apache +libraries that extension code actually imports (today: `commons-codec` for +`Hex`, `commons-lang3` for `StringUtils`), plus our own classes under +`com/mycompany/onprem/`. Everything else the agent provides at runtime — +`extension-api`, Spring, Jackson, Jakarta, SLF4J, Apache HttpClient — is +declared `compileOnly` in `build.gradle` and is **not** shaded. Packaging +what the agent already loads would cause classloader conflicts (see +`extension-api/README-extension-api.md` → "Add the API as a provided-scope +dependency") and bloat the deliverable; today the jar weighs under 1 MB. + +When you add a new third-party dependency, decide its scope with one +question: *does the agent itself provide this at runtime?* + +- **Yes** (any Spring / Jackson / Jakarta / SLF4J / Apache HttpClient + module, further `com.workato.*` types) → `compileOnly`. Nothing extra + for tests — the `configurations { testImplementation.extendsFrom + compileOnly }` block near the top of `dependencies { }` already forwards + every `compileOnly` entry onto the test classpath, so + `AgentExtensionHarness` keeps working. +- **No** (anything else — any new Apache Commons module, any client library + for the system this extension talks to) → `implementation`. It gets + shaded into the fat jar. Add a one-line comment next to the entry + pointing at the extension that uses it. + +The build strips third-party JAR signatures (`META-INF/*.SF` / `*.DSA` / +`*.RSA`) so the merged archive loads without `SignatureException`. Useful standalone Gradle tasks: @@ -455,24 +476,27 @@ cloud account. ``` src/main/ruby/ -├── security/connector.rb ← connector for SecurityExtension (fixed filename) -└── ntlm/connector.rb ← connector for NtlmExtension +├── security/connector.rb ← one connector.rb per extension (filename is fixed by the SDK) +├── ntlm/connector.rb +└── timer/connector.rb src/test/ruby/ ├── Gemfile ← shared — pulls in workato-connector-sdk + rspec + vcr + webmock -├── Gemfile.lock ← shared — one `bundle install` serves both suites +├── Gemfile.lock ← shared — one `bundle install` serves every suite ├── spec_helper.rb ← shared — VCR + RSpec config -├── security/ ← test harness for the security connector +├── security/ ← test harness for one connector │ ├── .rspec │ ├── settings.yaml ← connection fields the specs load as settings │ └── spec/ │ ├── spec_helper.rb ← one-liner: `require_relative '../../spec_helper'` -│ └── connector_spec.rb ← CONNECTOR_PATH points at ../../../../main/ruby/security/connector.rb -└── ntlm/ ← same layout, points at ../../../../main/ruby/ntlm/connector.rb +│ └── connector_spec.rb ← CONNECTOR_PATH points at ../../../../main/ruby//connector.rb +├── ntlm/ ← same layout +└── timer/ ← same layout src/test/resources/tape_library/ -├── security/ ← VCR cassettes for the security connector (one YAML per HTTP interaction set) -└── ntlm/ ← VCR cassettes for the ntlm connector +├── security/ ← VCR cassettes per connector (one YAML per HTTP interaction set) +├── ntlm/ +└── timer/ ``` Bundler walks up from any connector subdir to the shared `Gemfile`, so each @@ -483,15 +507,14 @@ under `src/test/resources/tape_library/`. ### Running -| Command | What runs | -|--------------------------------------------------------------------------|--------------------------------------------------------------| -| `./gradlew test` | Java JUnit **and** both Ruby RSpec suites. | -| `./gradlew testRubyConnectors` | Both Ruby suites (no Java). | -| `./gradlew :securityRspec` | Whole security suite. | -| `./gradlew :ntlmRspec` | Whole ntlm suite. | -| `./gradlew :securityRspec -Pspec='posts the payload'` | One example by description substring (forwarded as `-e`). | -| `./gradlew :ntlmRspec -Pspec=spec/connector_spec.rb:12` | One example by file:line (path containing `/` or `:` wins). | -| `cd src/test/ruby/ && bundle exec rspec [spec_path[:line]] [-e …]` | Fastest local iteration — skips Gradle, full rspec CLI. | +| Command | What runs | +|--------------------------------------------------------------------------|----------------------------------------------------------------------| +| `./gradlew test` | Java JUnit **and** every Ruby RSpec suite. | +| `./gradlew testRubyConnectors` | Every Ruby suite (no Java). | +| `./gradlew :Rspec` | One Ruby suite — e.g. `:securityRspec`, `:ntlmRspec`, `:timerRspec`. | +| `./gradlew :Rspec -Pspec='posts the payload'` | One example by description substring (forwarded as `-e`). | +| `./gradlew :Rspec -Pspec=spec/connector_spec.rb:12` | One example by file:line (path containing `/` or `:` wins). | +| `cd src/test/ruby/ && bundle exec rspec [spec_path[:line]] [-e …]` | Fastest local iteration — skips Gradle, full rspec CLI. | Prerequisites: Ruby 3.2.x (pinned in `.tool-versions` for asdf; works with any version manager or system Ruby that satisfies it) and `bundler`. The Gradle @@ -517,12 +540,12 @@ delegates to our existing `:Rspec` task, which already knows the right `workingDir`, bundle, and env: 1. **Run → Edit Configurations → + → Gradle.** -2. **Tasks:** `:ntlmRspec` (or `:securityRspec`). +2. **Tasks:** `:Rspec` — one of `:securityRspec`, `:ntlmRspec`, `:timerRspec`. 3. **Arguments** — pick one: - - *whole file:* `-Pspec=spec/connector_spec.rb` - - *single example by `:line`:* `-Pspec=spec/connector_spec.rb:12` - - *single example by description:* `-Pspec='posts the payload'` - - *whole suite:* leave empty. + - *whole file:* `-Pspec=spec/connector_spec.rb` + - *single example by `:line`:* `-Pspec=spec/connector_spec.rb:12` + - *single example by description:* `-Pspec='posts the payload'` + - *whole suite:* leave empty. 4. Save the config (one per spec file / view you want quick access to). Convention: each `connector_spec.rb` has a header comment with the exact @@ -923,39 +946,13 @@ below cover the surrounding Workato tooling. - [Custom connectors API](https://docs.workato.com/workato-api/custom_connectors.html) — if you prefer to publish via HTTP instead of `workato push`. -## Using the extension in Recipe - -In order to use the extension in a recipe, we need a custom adapter in Workato. Sample adapter for the extension -can be -found [here](https://github.com/workato/connector_sdk/blob/master/custom_connectors/basic_auth/onprem_security.rb). - -```ruby -{ - title: 'On-prem security', - secure_tunnel: true, - - connection: { - fields: [{ name: 'profile', hint: 'On-prem security connection profile' }], - authorization: { type: 'none'} - }, - - test: ->(connection) { - post("http://localhost/ext/#{connection['profile']}/computeDigest", { payload: 'test' }).headers('X-Workato-Connector': 'enforce') - }, +## Using an extension in a recipe - actions: { - sha256_digest: { +Each extension ships with a companion Workato custom connector under +`src/main/ruby//connector.rb`. The connector is what the recipe talks +to; the extension is what the connector talks to. For the per-connector +contract (actions, triggers, connection fields), open the matching +per-extension README linked in the **Available extensions** table above. - title: 'Create SHA-256 digest', - description: 'Create SHA-256 digest', - - input_fields: ->(_) { [{ name: 'payload' }] }, - output_fields: ->(_) { [{name: 'signature'}] }, - - execute: ->(connection, input) { - post("http://localhost/ext/#{connection['profile']}/computeDigest", input).headers('X-Workato-Connector': 'enforce') - } - } - } -} -``` +To publish and wire up a connector in a Workato workspace, see +**Deploying the connector to Workato** earlier in this document. diff --git a/build.gradle b/build.gradle index 6675036..88a9a01 100644 --- a/build.gradle +++ b/build.gradle @@ -96,39 +96,57 @@ task extractExtensionReadmes { tasks.withType(AbstractCompile).configureEach { it.dependsOn extractExtensionReadmes } processResources.dependsOn extractExtensionReadmes +// Make everything in `compileOnly` also visible to the test classpath. Gradle +// does NOT propagate compileOnly to tests by default, so without this the +// AgentExtensionHarness tests would lose Spring/Jackson/Jakarta/SLF4J at test +// runtime and fail with ClassNotFoundException. +configurations { + testImplementation.extendsFrom compileOnly +} + dependencies { - implementation fileTree(dir: extensionApiDir, include: 'extension-api-*.jar') + // ---- Provided by the agent at runtime — compileOnly, NOT shaded ---- + // + // extension-api is the API we compile against; the agent exposes it at + // runtime. Shading it would cause classloader conflicts (see + // extension-api/README-extension-api.md). + compileOnly fileTree(dir: extensionApiDir, include: 'extension-api-*.jar') testImplementation fileTree(dir: extensionApiDir, include: 'extension-test-*.jar') - implementation "org.slf4j:slf4j-api:${slf4jVersion}" - implementation "org.slf4j:jcl-over-slf4j:${slf4jVersion}" - implementation "org.slf4j:jul-to-slf4j:${slf4jVersion}" - implementation "org.slf4j:log4j-over-slf4j:${slf4jVersion}" + // Spring, Jackson, Jakarta, SLF4J — the agent is itself a Spring app and + // provides all four stacks at runtime. We compile against them (every + // extension uses @RestController / @Value / @JsonProperty / @Inject / + // Logger), but packaging them would duplicate code the agent already has. + compileOnly "org.springframework:spring-context:${springVersion}" + compileOnly "org.springframework:spring-webmvc:${springVersion}" - implementation "jakarta.servlet:jakarta.servlet-api:6.0.0" - implementation "jakarta.websocket:jakarta.websocket-api:2.1.0" - implementation "jakarta.validation:jakarta.validation-api:3.0.2" - implementation "jakarta.inject:jakarta.inject-api:2.0.1" + compileOnly "com.fasterxml.jackson.core:jackson-core:${jacksonVersion}" + compileOnly "com.fasterxml.jackson.core:jackson-databind:${jacksonVersion}" - implementation "org.springframework:spring-context:${springVersion}" - implementation "org.springframework:spring-jdbc:${springVersion}" - implementation "org.springframework:spring-jms:${springVersion}" - implementation "org.springframework:spring-webmvc:${springVersion}" + compileOnly "jakarta.servlet:jakarta.servlet-api:6.0.0" + compileOnly "jakarta.inject:jakarta.inject-api:2.0.1" - implementation "com.fasterxml.jackson.core:jackson-core:${jacksonVersion}" - implementation "com.fasterxml.jackson.core:jackson-databind:${jacksonVersion}" - implementation "com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:${jacksonVersion}" + compileOnly "org.slf4j:slf4j-api:${slf4jVersion}" - implementation "commons-io:commons-io:2.11.0" - implementation "commons-codec:commons-codec:1.15" - implementation "net.sf.jopt-simple:jopt-simple:5.0.3" - implementation "org.apache.commons:commons-lang3:3.12.0" - implementation "org.apache.commons:commons-collections4:4.4" - implementation "org.apache.httpcomponents:httpclient:4.5.14" + // Apache HttpClient 4 — the agent ships it as part of its own runtime, so + // NtlmExtension can import org.apache.http.* without us shading it in. + compileOnly "org.apache.httpcomponents:httpclient:4.5.14" + + // ---- Actually used by extension code — shaded into the fat jar ---- + // + // Keep the comment next to each entry pointing at the one or two source + // files that reference the library, so a future reader can confirm the + // dependency is still needed before upgrading or dropping it. + implementation "commons-codec:commons-codec:1.15" // SecurityExtension — Hex + implementation "org.apache.commons:commons-lang3:3.12.0" // NtlmExtension, TimerExtension — StringUtils testImplementation "org.springframework:spring-test:${springVersion}" testImplementation "org.junit.jupiter:junit-jupiter:5.10.0" testRuntimeOnly "org.junit.platform:junit-platform-launcher:1.10.0" + // AgentExtensionHarness parses the YAML fixtures under + // src/test/resources/extensions/ via jackson-dataformat-yaml. Not referenced + // by any extension production code, so it stays strictly on the test side. + testImplementation "com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:${jacksonVersion}" } test { @@ -140,6 +158,26 @@ test { } } +// --------------------------------------------------------------------------- +// Fat jar — bundle every runtime dependency into build/libs/opa-extensions-*.jar +// so the deliverable is a single file that drops into the agent's lib_ext/. +// extension-api is declared compileOnly above, so it is NOT on runtimeClasspath +// and therefore is NOT shaded here — exactly as README-extension-api.md requires +// ("The API is compile-only / provided — do not shade it into your JAR. The +// agent provides it at runtime; shading would cause classloader conflicts"). +// --------------------------------------------------------------------------- +jar { + duplicatesStrategy = DuplicatesStrategy.EXCLUDE + from { + configurations.runtimeClasspath.collect { it.isDirectory() ? it : zipTree(it) } + } + // Strip signatures from third-party JARs — otherwise the JVM rejects the + // merged archive on load with "SignatureException: invalid signature file digest". + exclude 'META-INF/*.SF' + exclude 'META-INF/*.DSA' + exclude 'META-INF/*.RSA' +} + // --------------------------------------------------------------------------- // Ruby custom connectors — Workato SDK (workato-connector-sdk gem). // Each connector lives in its own self-contained SDK project under @@ -207,10 +245,18 @@ task ntlmRspec(type: Exec) { commandLine rspecCommand(project.findProperty('spec')) } +task timerRspec(type: Exec) { + group = 'verification' + description = rspecDescription('timer') + dependsOn rubyBundleInstall + workingDir file('src/test/ruby/timer') + commandLine rspecCommand(project.findProperty('spec')) +} + task testRubyConnectors { group = 'verification' description = 'Runs RSpec for every Ruby custom connector using VCR cassettes.' - dependsOn securityRspec, ntlmRspec + dependsOn securityRspec, ntlmRspec, timerRspec } // Hook into both `test` and `check` so `./gradlew test` also runs the Ruby suite. diff --git a/extension-api/README-extension-api.md b/extension-api/README-extension-api.md index 10a1331..7549d9a 100644 --- a/extension-api/README-extension-api.md +++ b/extension-api/README-extension-api.md @@ -40,7 +40,7 @@ agent provides it at runtime; shading would cause classloader conflicts. package com.example.my_extension; import com.workato.agent.extension.api.AgentExtensionApiV1; -import com.workato.agent.extension.api.AgentConfigurationV1; +import com.workato.agent.extension.api.AgentExtensionContextV1.AgentConfiguration; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.web.bind.annotation.*; @@ -59,7 +59,7 @@ public class MyController { @PostMapping public Map handle(@RequestBody Map payload) { - AgentConfigurationV1 cfg = agent.getContext().getProfileConfiguration(); + AgentConfiguration cfg = agent.getContext().getProfileConfiguration(); // ... business logic ... return Map.of("profile", agent.getContext().getProfileName(), "received", payload); @@ -116,24 +116,227 @@ but you should autowire by type, not by name. public interface AgentExtensionContextV1 { String getProfileName(); - AgentConfigurationV1 getProfileConfiguration(); + AgentConfiguration getProfileConfiguration(); } ``` -| Method | What you get | -|-----------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| `getProfileName()` | The YAML key your extension is declared under (`my-extension` in the example above). Stable for the lifetime of the agent. | -| `getProfileConfiguration()` | Your extension's YAML section as a map, including the `controller` / `controllerClass` key alongside your custom properties. Snapshot — does not change across agent YAML reloads. | +| Method | What you get | +|-----------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| `getProfileName()` | The YAML key your extension is declared under (`my-extension` in the example above). Stable for the lifetime of the agent. | +| `getProfileConfiguration()` | Your extension's YAML section as a map, including the `controller` / `controllerClass` key alongside your custom properties. Snapshot — does not change across agent YAML reloads. | > **Each extension sees only its own profile.** There is no API for discovering > other extensions' configuration. (Note that this does not provide full > isolation on its own: an extension controller still shares a Spring parent > context with agent internals and can reach other beans by autowiring.) -### `AgentConfigurationV1` — map view of a YAML section +### `AgentWebhookApiV1` — outbound HTTP transport + +Obtained via `agent.getWebhookApi()`. A thin wrapper over `POST` with automatic retry and +exponential backoff — the only outbound-HTTP primitive extensions need when they want to push an +event to a recipe webhook URL or wake a pull-mode trigger. All nested types +(`WebhookResponse`, `RetryPolicy`, `WebhookDeliveryException`, `WebhookResponseException`) live +on the interface itself, so a single import brings in the whole vocabulary: + +```java +import com.workato.agent.extension.api.AgentWebhookApiV1; +import com.workato.agent.extension.api.AgentWebhookApiV1.WebhookResponse; +import com.workato.agent.extension.api.AgentWebhookApiV1.RetryPolicy; +``` + +**`send` ladder** — pick the shortest overload that carries exactly what you need: + +| Overload | When to use | +|------------------------------------|--------------------------------------------------------------------| +| `send(url, body)` | Simplest — POST JSON, profile's effective retry policy, no headers | +| `send(url, body, headers)` | Add correlation IDs, auth tokens, custom trace headers | +| `send(url, body, policy)` | Override retry policy per call (tighter timeout, more attempts) | +| `send(url, body, policy, headers)` | Both policy and headers | + +**`wakeup` ladder** — same four shapes, minus the body parameter. Sends an empty POST +(`Content-Length: 0`, not `{}`) for pull-mode triggers that fetch records through a separate +endpoint: + +```java +agent.getWebhookApi(). + +wakeup("https://recipe/trigger"); +agent. + +getWebhookApi(). + +wakeup("https://recipe/trigger", + Map.of("X-Request-ID", "req-1")); +``` + +**Response & deserialization.** Every call returns a `WebhookResponse`: + +```java +WebhookResponse resp = agent.getWebhookApi().send(url, payload); +if(resp. + +isSuccessful()){ +Ack ack = resp.as(Ack.class); // JSON → POJO/record via a shared Jackson ObjectMapper +} +``` + +`WebhookResponse.as(Class)` throws `WebhookResponseException` on `null` body, invalid JSON, +or type mismatch. For `List` / `Map` generics, call `.body()` and parse the +string yourself (a `TypeReference` overload is on the V1.1 shortlist). + +**Delivery semantics.** + +- **At-least-once** — the agent may deliver the same request more than once if the recipe + answered with a retryable status (`5xx` / `408` / `429` by default) or a network error. +- **Synchronous** — the calling thread blocks for the whole retry chain. +- **Only `application/json`** — the agent sets `Content-Type: application/json` for `send` and + omits it for `wakeup`; any such header you pass in `headers` is ignored. +- **Exhausted → `WebhookDeliveryException`.** Non-retryable non-2xx is **returned** as a + `WebhookResponse` for your code to decide — only exhaustion or a non-retryable network error + raises an exception. + +**Configuration via YAML.** The retry defaults for each profile come from an optional +`webhook:` section: + +```yaml +extensions: + my-extension: + controller: com.example.MyController + webhook: + retries: 10 # default 5 + initial-delay-ms: 500 # default 1000 + multiplier: 2.0 # default 2.0 + max-delay-ms: 60000 # default 30000 + jitter-factor: 0.2 # default 0.0, max 1.0 + timeout-ms: 10000 # default 30000 + response-cap-bytes: 524288 # default 1048576 (1 MB) +``` + +Everything is optional — a profile without a `webhook:` section still gets sensible defaults +(5 attempts, 1s → 2s → 4s → 8s backoff, no jitter, 30s timeout, 1 MB response cap). Invalid +values (`retries: -1`, `multiplier: 0.5`) log a WARN and fall back to the default **for that +specific key** — a typo must not keep the extension from loading. + +#### subscribe/unsubscribe pattern + +The webhook API does not include lifecycle management for subscriptions. Build it on top of +your own `@RestController` plus the webhook transport. The contract below is the one the +Workato SDK's `webhook_subscribe` / `webhook_unsubscribe` lambdas use for a webhook-proxy +connector — any extension that wants to serve as such a connector follows it verbatim. + +> ⚠ **The subscribe/unsubscribe pattern below is single-node only — treat it as legacy.** +> OPA agents commonly run as a group (several instances behind a load balancer). The +> `/subscribe` call that Workato routes to one agent registers the subscription in that +> agent's in-memory map; the other agents in the group know nothing about it. The next event +> picked up by a different agent is dropped on the floor; the SDK's `unsubscribe` is also +> routed to a single arbitrary node and leaves orphan entries on the others. Making this work +> under a group requires either a shared registry (database, Redis, …) or a group-aware +> subscribe fan-out — both out of scope for this API and currently unaddressed by the SDK. +> +> **The outbound API itself — {@link AgentWebhookApiV1#send send} and +> {@link AgentWebhookApiV1#wakeup wakeup} — has no such limitation.** It POSTs to a URL +> Workato controls; the agent is just an HTTP client. Any agent in the group that holds the +> URL can deliver, and nothing in the webhook transport cares about group membership. If your +> extension stores subscription state out-of-band (e.g., pulled from an external system on +> each event), the webhook API alone is enough and the pattern below is not needed. + +**The contract (from the SDK side):** + +| Direction | Endpoint | Body | +|-----------------------------------|-------------------------------------------|------------------------------------------------------------------------------| +| Workato → extension (subscribe) | `POST /webhook/subscribe` | `{"workato_webhook_url": "...", "recipe_id": "...", "connection_id": "..."}` | +| Workato → extension (unsubscribe) | `POST /webhook/unsubscribe` | `{"recipe_id": "...", "connection_id": "..."}` | +| extension → Workato (event) | `POST {workato_webhook_url}` with payload | `X-Workato-Event-Id` header carries a dedup id | + +> **The webhook URL is per-run; identity is whatever the extension chooses.** Workato mints a +> fresh `workato_webhook_url` every time a recipe (re)starts — so the URL is the *address* you +> POST to, not the *key* you look up by. The SDK's unsubscribe body is +> `{"recipe_id": …, "connection_id": …}` and no URL is passed. **The `recipe_id` field, however, +> is not necessarily the Workato recipe id — the SDK fills it in as** +> `webhook_subscribe_output['id'] || webhook_subscribe_output['recipe_id']`: if the extension +> returned `{"id": ""}` from `/subscribe`, that same string arrives in `recipe_id` on +> `/unsubscribe`. That lets the extension pick its own identity — the recommended +> approach — instead of relying on `connection_id` or the real recipe id. +> +> Possible keys for the registry, from least to most flexible: +> +> | Key | When to use | +> |--------------------------------------------|-----------------------------------------------------------------------------| +> | `(connection_id, recipe_id)` | Multiple recipes may share one connection — the usual case | +> | **Extension-generated UUID** (recommended) | Full control over identity; survives SDK quirks and extension-side rekeying | +> +> A `connection_id`-only key is not safe: nothing prevents several recipes from sharing one +> connection, so the second subscribe would overwrite the first. ```java -public interface AgentConfigurationV1 extends Map { + +@RestController +@RequestMapping("/webhook") +public class WebhookSubscriptionController { + + @Autowired + private AgentExtensionApiV1 agent; + + /** Keyed by an extension-generated UUID returned to the SDK as the subscribe response id. */ + private final Map active = new ConcurrentHashMap<>(); + + /** SDK bodies — field names must match the Workato SDK JSON. */ + public record SubscribeRequest( + @JsonProperty("workato_webhook_url") String workatoWebhookUrl, + @JsonProperty("recipe_id") String recipeId, + @JsonProperty("connection_id") String connectionId) { + } + + public record UnsubscribeRequest( + // Carries whatever the extension returned under `id` at subscribe time + // (SDK fallback: `webhook_subscribe_output['id'] || webhook_subscribe_output['recipe_id']`). + @JsonProperty("recipe_id") String subscriptionId, + @JsonProperty("connection_id") String connectionId) { + } + + /** Per-subscription state the extension owns. */ + private record Subscription(String webhookUrl, String workatoRecipeId, String connectionId, + /* extension-defined routing state: topic, filter, auth, … */ Object state) { + } + + @PostMapping("/subscribe") + public Map subscribe(@RequestBody SubscribeRequest req) { + String subscriptionId = UUID.randomUUID().toString(); + active.put(subscriptionId, + new Subscription(req.workatoWebhookUrl(), req.recipeId(), req.connectionId(), /* state */ null)); + // The SDK echoes this `id` back as `recipe_id` on unsubscribe — see the SDK fallback above. + return Map.of("id", subscriptionId); + } + + @PostMapping("/unsubscribe") + public void unsubscribe(@RequestBody UnsubscribeRequest req) { + active.remove(req.subscriptionId()); + } + + // Called from the extension's own event source (polling loop, external bus listener, …). + // The dedup header lets the Workato trigger drop duplicates on its side. + private void dispatch(String subscriptionId, Object payload) { + Subscription sub = active.get(subscriptionId); + if (sub == null) + return; + WebhookResponse r = agent.getWebhookApi().send(sub.webhookUrl(), payload, + Map.of("X-Workato-Event-Id", UUID.randomUUID().toString())); + log.debug("Delivered to {} (sub {}, recipe {}), status {}", + sub.webhookUrl(), subscriptionId, sub.workatoRecipeId(), r.statusCode()); + } +} +``` + +### `AgentExtensionContextV1.AgentConfiguration` — map view of a YAML section + +Nested on `AgentExtensionContextV1`, so one `AgentExtensionContextV1` import is enough — the +configuration type comes with it. + +```java +public interface AgentExtensionContextV1 { + // ... + interface AgentConfiguration extends Map { + } } ``` @@ -160,6 +363,7 @@ extensions: ``` ```java + @Value("${api.endpoint}") private String endpoint; @@ -183,13 +387,17 @@ this API. Record historically autowired as: ```java -@Autowired DefaultExtensionEndpoint workatoEndpoint; + +@Autowired +DefaultExtensionEndpoint workatoEndpoint; ``` Replace with: ```java -@Autowired AgentExtensionApiV1 agent; + +@Autowired +AgentExtensionApiV1 agent; ``` Then migrate each call site: @@ -199,10 +407,10 @@ Then migrate each call site: | `workatoEndpoint.name()` | `agent.getContext().getProfileName()` | | `workatoEndpoint.configuration()` | `agent.getContext().getProfileConfiguration()` | -The returned `AgentConfigurationV1` is a `Map`, so if your old +The returned `AgentConfiguration` is a `Map`, so if your old code called `.get("key")` on the legacy `Configuration`, the new call is identical. Typed accessors from the legacy `Configuration` (e.g., coerce to -`int`/`boolean`) are not currently on `AgentConfigurationV1` — do the coercion +`int`/`boolean`) are not currently on `AgentConfiguration` — do the coercion yourself, or inject the leaf with `@Value`. ### `ExtensionEndpoint` interface — not supported @@ -232,14 +440,16 @@ before the legacy types are removed. ## Versioning -The API is named `AgentExtensionApiV1` / `AgentExtensionContextV1` / `AgentConfigurationV1`. +The API is named `AgentExtensionApiV1` / `AgentExtensionContextV1` (with its nested +`AgentConfiguration`) / `AgentWebhookApiV1` (with its nested `WebhookResponse`, `RetryPolicy`, +`WebhookDeliveryException`, `WebhookResponseException`). - `AgentExtensionApiV1` and `AgentExtensionContextV1` are **consumer-only** — customer code calls their methods but does not implement them (the implementation is supplied by the agent). Methods may be added to these interfaces without breaking customer compile or runtime, as long as the customer does not provide their own implementation. -- `AgentConfigurationV1` extends `Map`; additional typed +- `AgentConfiguration` extends `Map`; additional typed accessors may be added here in future iterations, source- and binary-compatible with existing calls. - Incompatible changes will introduce a parallel `V2` family and keep `V1` diff --git a/extension-api/README-extension-test.md b/extension-api/README-extension-test.md index c833216..f074649 100644 --- a/extension-api/README-extension-test.md +++ b/extension-api/README-extension-test.md @@ -310,12 +310,113 @@ implementation returns: `put`/`remove`/`clear`/`putAll` throw configuration, the test fails here rather than silently drifting from production behaviour. +## Testing webhook calls — `FakeAgentWebhookApiV1` + +If your extension calls `agent.getWebhookApi().send(...)` or `.wakeup(...)`, the harness ships a +`FakeAgentWebhookApiV1` that captures every attempt and lets you stub responses. Access it with +`harness.webhookApi()` — no cast needed. + +> **Nested-type imports.** The examples below use `CallKind` and `SentCall` (both nested on +> `FakeAgentWebhookApiV1`) and `RetryPolicy` / `WebhookResponse` / `WebhookDeliveryException` +> (nested on `AgentWebhookApiV1`). Import them as e.g. +> `import com.workato.agent.extension.test.FakeAgentWebhookApiV1.CallKind;` or qualify them +> inline — IDE auto-import handles both. + +```java +@RegisterExtension +static AgentExtensionHarness agent = AgentExtensionHarness.extension("my-profile") + .controller(MyController.class); + +@BeforeEach +void resetWebhookState() { + // Class-level harness → same fake across tests; reset log + stubs before each case. + agent.webhookApi().reset(); +} + +@Test +void controllerDeliversExpectedPayload() throws Exception { + agent.webhookApi() + .whenSent("https://recipe/webhook") + .thenReturn(200, "{\"ack\":true}"); + + agent.mockMvc().perform(agent.request(MyController::fire) + .contentType(APPLICATION_JSON).content("{\"x\":1}")); + + var call = agent.webhookApi().lastCall(); + assertEquals(Map.of("x", 1), call.body()); + assertEquals(CallKind.SEND, call.kind()); +} +``` + +### Stubbing responses + +| Matcher | Matches… | +|---------------------------------|------------------------------------------| +| `whenCalled(url)` | Any `send` or `wakeup` to the exact URL | +| `whenCalledMatching(predicate)` | Any kind, URL filtered by your predicate | +| `whenSent(url)` | Only `send` to the exact URL | +| `whenWokenUp(url)` | Only `wakeup` to the exact URL | + +Each matcher returns a builder with: + +| Builder method | Behaviour | +|----------------------------------|----------------------------------------------------------------| +| `.thenReturn(status, body)` | Same response on every attempt | +| `.thenReturn(WebhookResponse)` | Same `WebhookResponse` instance on every attempt | +| `.thenFail(IOException cause)` | Fail every attempt with the given exception | +| `.thenFailAll()` | Shorthand for a generic IO failure | +| `.thenReturnSequence(r1, r2, …)` | Nth attempt → `rn`; reuses the last entry if attempts > length | + +Multiple matching stubs resolve **last-wins** — a broad `@BeforeEach` default plus a narrower +per-test override is the typical pattern. + +### Observing captured calls + +```java +List all = agent.webhookApi().allCalls(); +List toUrl = agent.webhookApi().callsFor("https://recipe/webhook"); +List wakeups = agent.webhookApi().callsOfKind(CallKind.WAKEUP); +SentCall last = agent.webhookApi().lastCall(); // throws if empty +``` + +`SentCall` records one HTTP attempt, so a `send` that retried 3 times produces 3 `SentCall`s +with the same `url`/`body`/`headers`/`policy` and increasing `attemptNumber`. Exactly one of +`response()` / `exception()` is non-null per attempt. + +### Retry fidelity + +The fake honours the full `RetryPolicy` — attempt count, retryable-status predicate, retryable- +exception predicate, exhaustion → `WebhookDeliveryException`. The one deliberate deviation from +production is that **all delays are zero** (unit tests must not sleep). The retry count and +order are preserved exactly as production produces them. + +Tune the policy per-test by passing a `RetryPolicy` to the overload: + +```java +agent.webhookApi().setDefaultRetryPolicy(RetryPolicy.DEFAULTS.withAttempts(3)); +// or override per-call via send(url, body, policy) +``` + +### Response deserialization + +`WebhookResponse.as(Class)` on both fake and production uses a shared Jackson +`ObjectMapper`: + +```java +Ack ack = agent.webhookApi().lastCall().response().as(Ack.class); +``` + +Throws `WebhookResponseException` on `null` body, invalid JSON, or type mismatch. For generic +types (`List`, `Map`), call `.body()` and parse yourself — a `TypeReference` +overload is on the V1.1 shortlist. + ## What the harness exposes | Method | What it returns | |-----------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------| | `mockMvc()` | Spring's `MockMvc` for issuing requests | | `agentExtensionApi()` | The `AgentExtensionApiV1` the controller sees — handy for direct assertions | +| `webhookApi()` | The `FakeAgentWebhookApiV1` the controller sees through `agent.getWebhookApi()` — stub responses, assert against the captured call log | | `webApplicationContext()` | The underlying `AnnotationConfigWebApplicationContext` for advanced wiring | | `profileName()` | The configured profile name | | `urlFor(path)` | The full request path including the `/ext/{profile}` prefix | @@ -695,21 +796,21 @@ Builder methods on `DeployedAgentLauncher`: Accessors on `RunningAgent` (`AutoCloseable`): -| Method | Returns / effect | -|------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| `port()` | Bound TCP port opened by `openLocalAccess(...)`. Throws `IllegalStateException` when local access was not opened. | -| `profileName()` | The default profile name. | -| `profile()` | The final merged body of the default profile. | -| `extensionProfile(name)` | The final merged body of any registered profile. | -| `extensionProfiles()` | Snapshot of all registered profile bodies. | -| `httpGet(Controller::method [, pathVars])` | Issues GET at the URL derived from the controller method's `@GetMapping` + the profile mount point (connector mode only). Arity overloads cover controller signatures from zero through ten declared parameters. | -| `httpPost(Controller::method, body [, pathVars])` | Issues POST with a JSON-serialised body (connector mode only). Same ten-arity ladder as `httpGet` — enough for `@RequestBody` plus `@PathVariable` / `@RequestHeader` / `@RequestParam` / … combos. | -| `urlFor(Controller::method [, pathVars])` | The full URL (`http://127.0.0.1:{port}/ext/{profile}{route}`) a request to the given controller method would target. Same ten-arity ladder as `httpGet`. | -| `requestFor(Controller::method [, pathVars])` | An `HttpRequest.Builder` with URL + HTTP method pre-filled from the annotation; caller adds `.header(...)` / `.POST(publisher)` / `.timeout(...)` and finishes with `.build()`. Pairs with `send(request)`. | -| `send(HttpRequest)` | Sends the request through the same `HttpClient` the convenience methods use (5-second connect timeout, string body handler). | -| `awaitGatewayConnected(Duration)` | Blocks until `"Connected to gateway"` appears in the agent's log, or times out. Returns the full matching line. | -| `classLoader()` | The agent's isolated `URLClassLoader`. | -| `stop()` / `close()` | Shuts the agent down, restores stdout, releases the classloader. | +| Method | Returns / effect | +|---------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| `port()` | Bound TCP port opened by `openLocalAccess(...)`. Throws `IllegalStateException` when local access was not opened. | +| `profileName()` | The default profile name. | +| `profile()` | The final merged body of the default profile. | +| `extensionProfile(name)` | The final merged body of any registered profile. | +| `extensionProfiles()` | Snapshot of all registered profile bodies. | +| `httpGet(Controller::method [, pathVars])` | Issues GET at the URL derived from the controller method's `@GetMapping` + the profile mount point (connector mode only). Arity overloads cover controller signatures from zero through ten declared parameters. | +| `httpPost(Controller::method, body [, pathVars])` | Issues POST with a JSON-serialised body (connector mode only). Same ten-arity ladder as `httpGet` — enough for `@RequestBody` plus `@PathVariable` / `@RequestHeader` / `@RequestParam` / … combos. | +| `urlFor(Controller::method [, pathVars])` | The full URL (`http://127.0.0.1:{port}/ext/{profile}{route}`) a request to the given controller method would target. Same ten-arity ladder as `httpGet`. | +| `requestFor(Controller::method [, pathVars])` | An `HttpRequest.Builder` with URL + HTTP method pre-filled from the annotation; caller adds `.header(...)` / `.POST(publisher)` / `.timeout(...)` and finishes with `.build()`. Pairs with `send(request)`. | +| `send(HttpRequest)` | Sends the request through the same `HttpClient` the convenience methods use (5-second connect timeout, string body handler). | +| `awaitGatewayConnected(Duration)` | Blocks until `"Connected to gateway"` appears in the agent's log, or times out. Returns the full matching line. | +| `classLoader()` | The agent's isolated `URLClassLoader`. | +| `stop()` / `close()` | Shuts the agent down, restores stdout, releases the classloader. | Static helpers: diff --git a/extension-api/extension-api-34.0.jar b/extension-api/extension-api-34.0.jar index 509fcd1..2e0de59 100644 Binary files a/extension-api/extension-api-34.0.jar and b/extension-api/extension-api-34.0.jar differ diff --git a/extension-api/extension-test-34.0.jar b/extension-api/extension-test-34.0.jar index cf8e2aa..906396f 100644 Binary files a/extension-api/extension-test-34.0.jar and b/extension-api/extension-test-34.0.jar differ diff --git a/src/main/java/com/mycompany/onprem/ntlm/NtlmExtension.java b/src/main/java/com/mycompany/onprem/ntlm/NtlmExtension.java index 5fc5797..3cd7f22 100644 --- a/src/main/java/com/mycompany/onprem/ntlm/NtlmExtension.java +++ b/src/main/java/com/mycompany/onprem/ntlm/NtlmExtension.java @@ -2,6 +2,7 @@ import com.workato.agent.extension.api.AgentExtensionApiV1; import jakarta.servlet.http.HttpServletResponse; +import org.apache.commons.lang3.StringUtils; import org.apache.http.HttpEntity; import org.apache.http.HttpStatus; import org.apache.http.auth.AuthScope; @@ -60,7 +61,7 @@ public class NtlmExtension { private volatile CloseableHttpClient httpClient; private static URI parseHttpUri(String raw) { - if (raw == null || raw.isBlank()) { + if (StringUtils.isBlank(raw)) { return null; } URI uri; @@ -108,7 +109,7 @@ private List allowedHostPatterns() { if (cached != null) { return cached; } - if (allowedHostsRaw == null || allowedHostsRaw.isBlank()) { + if (StringUtils.isBlank(allowedHostsRaw)) { cached = Collections.emptyList(); } else { diff --git a/src/main/java/com/mycompany/onprem/ntlm/README.md b/src/main/java/com/mycompany/onprem/ntlm/README.md new file mode 100644 index 0000000..df28e8d --- /dev/null +++ b/src/main/java/com/mycompany/onprem/ntlm/README.md @@ -0,0 +1,70 @@ +# NtlmExtension + +Proxies an HTTPS/HTTP GET through NTLM authentication. The Workato recipe +hands the extension a target URL and credentials; the extension performs the +NTLM-authenticated request from inside the on-prem network and streams the +response body back verbatim. + +Exists because Workato cloud cannot speak NTLM directly to a customer's +intranet host — the handshake only works from a process that lives inside the +customer's network boundary. This is the archetype of why on-prem extensions +exist in the first place. + +## Extension API features demonstrated + +| Feature | How | +|------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| `@Autowired AgentExtensionApiV1` | Spring form of the autowire — alternative to `@Inject` seen in **SecurityExtension**. | +| `@Value("${connect-timeout-ms:5000}")` with defaults | Property injection with a fallback; honours the agent's dot-literal lookup semantics (dots in keys are literal, not nesting markers). | +| Multiple injected scalars + a list-shaped key | `connect-timeout-ms`, `read-timeout-ms`, `ntlm.allowed-hosts` — the last one is parsed into a comma-separated allow-list. | +| Lazy init of a shared resource (DCL pattern) | `volatile CloseableHttpClient httpClient;` + double-check locking in `client()`. Mirrors the pattern **TimerExtension** uses for its `ScheduledExecutorService`. | +| Request headers via `@RequestHeader` | NTLM creds and the target URL arrive as agent-side request headers (`X-Workato-NTLM-auth` / `X-Workato-NTLM-url`), not JSON body — reduces chances of accidental logging. | +| Writing to the HTTP response status directly | Uses `HttpServletResponse.setStatus(...)` to forward upstream status codes (400 for bad URL, 403 for host not in allow-list, upstream's 2xx/4xx/5xx otherwise). | + +Not demonstrated here (see other extensions): + +- `AgentExtensionContextV1.getProfileConfiguration()` map read → **SecurityExtension**. +- Outbound HTTP via `AgentWebhookApiV1.send(...)` → **TimerExtension**. +- Background scheduler + `DisposableBean.destroy()` lifecycle → **TimerExtension**. + +## HTTP endpoint + +``` +POST /ext/{profile}/request +X-Workato-NTLM-auth : username:password[@domain[/workstation]] +X-Workato-NTLM-url : +``` + +Response: + +``` +200 OK + upstream body — upstream answered 2xx +400 Bad Request — target URL malformed / non-http scheme / no host +403 Forbidden — target host not in ntlm.allowed-hosts + — forwarded as-is, empty body +``` + +## Configuration + +```yaml +extensions: + ntlm: + controller: com.mycompany.onprem.ntlm.NtlmExtension + connect-timeout-ms: 5000 # default 5000 + read-timeout-ms: 30000 # default 30000 + ntlm.allowed-hosts: "intranet.corp.local, *.corp.local" +``` + +- `connect-timeout-ms` / `read-timeout-ms` — straight to Apache HttpClient's + `RequestConfig`. Optional; defaults in the `@Value` annotations. +- `ntlm.allowed-hosts` — comma-separated. Each entry is matched + case-insensitively against the target URL's host. `*.suffix` matches any + single-level-or-deeper subdomain. **Empty value disables the check** + entirely — fine for a lab, risky in production. Configure explicitly. + +## Workato connector + +Companion Ruby connector: [`src/main/ruby/ntlm/connector.rb`](../../../../../main/ruby/ntlm/connector.rb). +Credentials live in the Workato connection (not in each action input), get +assembled into the `X-Workato-NTLM-auth` header via a shared connector +`method`, and the target URL is the action's one input field. diff --git a/src/main/java/com/mycompany/onprem/security/README.md b/src/main/java/com/mycompany/onprem/security/README.md new file mode 100644 index 0000000..fe93638 --- /dev/null +++ b/src/main/java/com/mycompany/onprem/security/README.md @@ -0,0 +1,65 @@ +# SecurityExtension + +Simplest extension in the repo. One synchronous endpoint that takes a UTF-8 +string from a Workato recipe, prefixes it with a secret the agent reads from +`config.yml`, and returns the SHA-256 digest. + +The point is to show the **smallest possible shape** of an OPA extension — a +plain Spring `@RestController` that pulls one key out of its YAML profile and +answers one `POST`. Everything else in the project is built on this skeleton. + +## Extension API features demonstrated + +| Feature | How | +|---------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| Autowiring `AgentExtensionApiV1` | `@Inject AgentExtensionApiV1 agent;` — jakarta.inject form, equivalent to Spring `@Autowired`. | +| `AgentExtensionContextV1.getProfileName()` | Used as a prefix in every log line so multi-profile agents are debuggable. | +| `AgentExtensionContextV1.getProfileConfiguration()` | Reads the `signing-secret` key out of the profile's YAML map at request time. | +| YAML → config read via the context (not `@Value`) | A deliberate choice — shows the `Map` view, including the subtle gotcha that `secret` as a key name is reserved by the agent's `SecretsPreprocessor` and must be avoided (see the comment in the class). | +| `@RestController` + `@PostMapping` + `@RequestBody Map` | Vanilla Spring MVC — no custom body deserialisation. | +| Failing fast on missing config | `IllegalStateException` carrying the profile name — surfaces to the caller as HTTP 500 but with a message a human can act on. | + +Not demonstrated here (see other extensions): + +- `@Value("${…}")` property injection → **NtlmExtension**. +- Lazy init of a shared resource via double-check locking → **NtlmExtension**. +- Outbound HTTP via `AgentWebhookApiV1.send(...)` → **TimerExtension**. +- Background scheduler + `DisposableBean.destroy()` lifecycle → **TimerExtension**. + +## HTTP endpoint + +``` +POST /ext/{profile}/computeDigest +Content-Type: application/json + +{"payload": ""} +``` + +Response: + +``` +200 OK +Content-Type: application/json + +{"signature": "<64-char lowercase hex of SHA-256(secret || payload)>"} +``` + +## Configuration + +```yaml +extensions: + security: + controller: com.mycompany.onprem.security.SecurityExtension + signing-secret: +``` + +- `signing-secret` (required, non-empty) — prepended to the payload before + hashing. **Do not** rename this key to just `secret` — the agent's + `SecretsPreprocessor` interprets any map containing `secret:` as a + secret-provider spec and will fail to load the profile in local-HTTP mode. + +## Workato connector + +Companion Ruby connector: [`src/main/ruby/security/connector.rb`](../../../../../main/ruby/security/connector.rb). +It exposes one action (`sha256_digest`) that forwards `{payload}` to this +endpoint and surfaces the returned signature. diff --git a/src/main/java/com/mycompany/onprem/security/SecurityExtension.java b/src/main/java/com/mycompany/onprem/security/SecurityExtension.java index fcc9d9e..b5f0526 100644 --- a/src/main/java/com/mycompany/onprem/security/SecurityExtension.java +++ b/src/main/java/com/mycompany/onprem/security/SecurityExtension.java @@ -4,8 +4,8 @@ package com.mycompany.onprem.security; -import com.workato.agent.extension.api.AgentConfigurationV1; import com.workato.agent.extension.api.AgentExtensionApiV1; +import com.workato.agent.extension.api.AgentExtensionContextV1.AgentConfiguration; import jakarta.inject.Inject; import org.apache.commons.codec.binary.Hex; import org.slf4j.Logger; @@ -31,7 +31,7 @@ public class SecurityExtension { @PostMapping("/computeDigest") public Map computeDigest(@RequestBody Map body) { String profile = agent.getContext().getProfileName(); - AgentConfigurationV1 cfg = agent.getContext().getProfileConfiguration(); + AgentConfiguration cfg = agent.getContext().getProfileConfiguration(); // Avoid the config key name `secret` — the agent's SecretsPreprocessor // treats any map containing `secret:` as a secret-provider spec and tries diff --git a/src/main/java/com/mycompany/onprem/timer/README.md b/src/main/java/com/mycompany/onprem/timer/README.md new file mode 100644 index 0000000..20e8b89 --- /dev/null +++ b/src/main/java/com/mycompany/onprem/timer/README.md @@ -0,0 +1,137 @@ +# TimerExtension + +Periodic webhook emitter. A Workato recipe subscribes with +`{message, period_seconds}`; the extension schedules a task that pushes +`{message, generated_at, tick, subscription_id, period_seconds}` back to the +recipe every N seconds through `AgentWebhookApiV1.send(...)`. + +Exists purely to showcase the webhook half of the Extension API — the +`AgentWebhookApiV1` interface that was added in 34.0, together with the SDK's +`webhook_subscribe` / `webhook_notification` / `webhook_unsubscribe` trigger +contract. + +> ⚠ **The subscribe/unsubscribe part of this extension is demonstrative and +> considered legacy** — treat it as a reference for how the SDK contract +> looks on the agent side, not as a template to copy into production. +> +> Subscription state lives in a single `ConcurrentHashMap` inside **one** +> agent process. OPA agents commonly run as a group (several instances behind +> a load balancer). Workato routes the `webhook_subscribe` call to one +> arbitrary member of the group — only that one agent knows the subscription +> exists. If a scheduled tick, an inbound event, or the matching +> `webhook_unsubscribe` later lands on a different member, it is dropped on +> the floor. That directly contradicts the whole point of running agents as +> a group (fault-tolerance + horizontal scale), so this pattern is not +> acceptable for production without a shared registry (database, Redis, …) +> or a group-aware subscribe fan-out — both out of scope for this API and +> currently unaddressed by the SDK. See +> `extension-api/README-extension-api.md` → "subscribe/unsubscribe pattern" +> for the authoritative statement of this limitation. +> +> The **outbound** half — `AgentWebhookApiV1.send(...)` and its retry / +> dedup machinery — does **not** have this limitation. Any agent in the +> group that holds the webhook URL can deliver, and the API itself does not +> care about group membership. If your extension sources subscription state +> from an external system (polled per tick, pushed by that system, …), the +> outbound half alone is enough and the subscribe/unsubscribe lambdas in +> this demo are not needed. + +## Extension API features demonstrated + +| Feature | How | +|-----------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| **`AgentWebhookApiV1.send(url, body, headers)`** | The whole point of the demo. One call per tick — delivery, retries, and backoff are the API's job. | +| Automatic retries + exponential backoff | Configured per-profile under `extensions.timer.webhook:` in YAML (see **Configuration** below). We do not implement retry ourselves. | +| Dedup header pattern | Every outbound call carries a fresh `X-Workato-Event-Id: ` — Workato uses it to drop duplicates when at-least-once delivery doubles up a tick. | +| Jackson records with `@JsonProperty("snake_case")` | All JSON boundaries — inbound request bodies, outbound response bodies, and the outbound webhook payload (`TimerEvent`) — go through typed records. Component names compile to camelCase so the annotations are mandatory. | +| `webhook_subscribe` / `/unsubscribe` HTTP contract | The three endpoints under `/webhook/*` implement the SDK contract documented in `README-extension-api.md` verbatim. Subscription identity is an extension-generated UUID — the recommended path per that document. | +| `@Value("${min-period-seconds:5}")` | Soft floor on the period, overridable from YAML — tests set it to `1` so the scheduler can tick inside a bounded wait. | +| `DisposableBean.destroy()` lifecycle | Cancels in-flight `ScheduledFuture`s, shuts the executor down, and `awaitTermination`s. Fires when the extension's Spring context closes (profile unload or `AgentExtensionHarness.stop()` in tests). | +| Lazy init of a shared `ScheduledExecutorService` | Double-check-locked, daemon-threaded, named `timer-ext-`. Same pattern as **NtlmExtension**'s HttpClient. | +| `scheduleWithFixedDelay` over `scheduleAtFixedRate` | `AgentWebhookApiV1.send` is **synchronous and blocks for the whole retry chain** (seconds, not millis). Fixed-rate would queue up back-to-back ticks on the single-threaded executor during a slow one; fixed-delay keeps a full `period_seconds` gap between the end of one tick and the start of the next. | +| Swallow + log in the scheduler thread | A `WebhookDeliveryException` on a scheduled tick is caught, logged `WARN`, and the thread stays alive to run the next tick. Contrast with `/fire_now` which propagates the exception to the HTTP response — manual triggers want visible failures. | + +Not demonstrated here (see other extensions): + +- `AgentExtensionContextV1.getProfileConfiguration()` map read → **SecurityExtension**. +- `@RequestHeader`-driven controllers → **NtlmExtension**. +- Writing `HttpServletResponse` status directly → **NtlmExtension**. + +## HTTP endpoints + +All under `/ext/{profile}/webhook/*`. + +| Method | Path | Request | Response | +|--------|----------------|----------------------------------------------------------------------------|------------------------------------------------------------------------------------------------| +| `GET` | `/health` | — | `200 OK` | +| `POST` | `/subscribe` | `{workato_webhook_url, recipe_id, connection_id, message, period_seconds}` | `200 {"id":""}` — or `400 {"error":"period_seconds must be >= "}` | +| `POST` | `/unsubscribe` | `{recipe_id (= our UUID), connection_id}` | `204 No Content` | +| `POST` | `/fire_now` | `{subscription_id}` | `200 {"delivered": , "status_code": }` or `404 {"error":"unknown subscription_id"}` | + +Outbound tick payload — the body `AgentWebhookApiV1.send(...)` POSTs to +`workato_webhook_url`: + +```json +{ + "message": "", + "generated_at": "2026-10-19T14:03:21.337Z", + "tick": 1, + "subscription_id": "", + "period_seconds": 10 +} +``` + +## Configuration + +Only `controller:` is required. Everything else is optional — the extension +runs out of the box with the YAML snippet below: + +```yaml +extensions: + timer: + controller: com.mycompany.onprem.timer.TimerExtension +``` + +Full configuration surface, every key shown with its default: + +```yaml +extensions: + timer: + controller: com.mycompany.onprem.timer.TimerExtension + + # Optional — soft floor on period_seconds a recipe may request. + # Default: 5. Tests lower it to 1 so the scheduler can tick inside a + # bounded wait. + min-period-seconds: 5 + + # Optional — per-profile override of the retry policy + # AgentWebhookApiV1.send(...) uses. Omitting the whole `webhook:` block + # leaves every key at its default (5 attempts, 1s → 2s → 4s → 8s backoff, + # no jitter, 30 s per-attempt timeout, 1 MB response cap). Individual + # keys inside the block are also optional — a typo on one key logs a + # WARN and falls back to the default for *that key only*, the profile + # still loads. + webhook: + retries: 5 # default 5 + initial-delay-ms: 1000 # default 1000 + multiplier: 2.0 # default 2.0 + max-delay-ms: 30000 # default 30000 + jitter-factor: 0.0 # default 0.0, max 1.0 + timeout-ms: 30000 # default 30000 + response-cap-bytes: 1048576 # default 1 MB +``` + +- `min-period-seconds` — rejects `/subscribe` calls with a smaller + `period_seconds`. Keeps a demo from DoS-ing the recipe. +- `webhook:` — read by the agent's `AgentWebhookApiV1` implementation, not + by our code. See `extension-api/README-extension-api.md` section + "Configuration via YAML" for the full policy semantics. + +## Workato connector + +Companion Ruby connector: [`src/main/ruby/timer/connector.rb`](../../../../../main/ruby/timer/connector.rb). +Exposes the trigger `tick` (full `webhook_subscribe` / `_notification` / +`_unsubscribe` lambdas) and an `fire_now` action for one-shot delivery. The +connector caches `connection_id` inside the subscribe-output hash so +`webhook_unsubscribe` — which the SDK calls with only that hash, no +connection — can rebuild the unsubscribe body without a side-channel lookup. diff --git a/src/main/java/com/mycompany/onprem/timer/TimerExtension.java b/src/main/java/com/mycompany/onprem/timer/TimerExtension.java new file mode 100644 index 0000000..7d29843 --- /dev/null +++ b/src/main/java/com/mycompany/onprem/timer/TimerExtension.java @@ -0,0 +1,215 @@ +package com.mycompany.onprem.timer; + +import com.fasterxml.jackson.annotation.JsonProperty; +import com.workato.agent.extension.api.AgentExtensionApiV1; +import com.workato.agent.extension.api.AgentWebhookApiV1.WebhookResponse; +import org.apache.commons.lang3.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.http.HttpStatus; +import org.springframework.http.ResponseEntity; +import org.springframework.web.bind.annotation.*; + +import java.time.Instant; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicLong; + +@RestController +@RequestMapping("/webhook") +public class TimerExtension implements InitializingBean, DisposableBean { + + private static final Logger log = LoggerFactory.getLogger(TimerExtension.class); + + private final ConcurrentHashMap active = new ConcurrentHashMap<>(); + + @Autowired + private AgentExtensionApiV1 agent; + + @Value("${min-period-seconds:5}") + private int minPeriodSeconds; + + private String profile; + private ScheduledExecutorService scheduler; + + @Override + public void afterPropertiesSet() { + this.profile = agent.getContext().getProfileName(); + this.scheduler = Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r, "timer-ext"); + t.setDaemon(true); + return t; + }); + } + + @GetMapping("/health") + public void health() { + } + + @PostMapping("/subscribe") + public SubscribeResponse subscribe(@RequestBody SubscribeRequest req) { + if (StringUtils.isBlank(req.message())) { + throw new BadRequestException("message must not be blank"); + } + if (StringUtils.isBlank(req.workatoWebhookUrl())) { + throw new BadRequestException("workato_webhook_url must not be blank"); + } + if (req.periodSeconds() == null || req.periodSeconds() < minPeriodSeconds) { + throw new BadRequestException("period_seconds must be >= " + minPeriodSeconds); + } + String id = UUID.randomUUID().toString(); + Subscription sub = new Subscription(id, req.workatoWebhookUrl(), req.message(), req.periodSeconds()); + // Schedule and set `sub.future` BEFORE publishing to `active`, so any + // concurrent /unsubscribe that finds the subscription is guaranteed to + // see a non-null future to cancel. + sub.future = scheduler.scheduleWithFixedDelay( + () -> deliver(sub), + req.periodSeconds(), req.periodSeconds(), TimeUnit.SECONDS); + active.put(id, sub); + log.info("[{}] subscribed id={}, period={}s, url={}", + profile, id, req.periodSeconds(), req.workatoWebhookUrl()); + return new SubscribeResponse(id); + } + + @PostMapping("/unsubscribe") + @ResponseStatus(HttpStatus.NO_CONTENT) + public void unsubscribe(@RequestBody UnsubscribeRequest req) { + Subscription sub = active.remove(req.subscriptionId()); + if (sub != null) { + sub.future.cancel(false); + } + log.info("[{}] unsubscribed id={}", profile, req.subscriptionId()); + } + + @PostMapping("/fire_now") + public FireNowResponse fireNow(@RequestBody FireNowRequest req) { + Subscription sub = active.get(req.subscriptionId()); + if (sub == null) { + throw new NotFoundException("unknown subscription_id: " + req.subscriptionId()); + } + WebhookResponse resp = emit(sub); + return new FireNowResponse(resp.isSuccessful(), resp.statusCode()); + } + + private void deliver(Subscription sub) { + try { + emit(sub); + } catch (RuntimeException e) { + // Scheduler-thread invariant: swallow + log so the next tick still runs. + // Manual `/fire_now` deliberately does NOT swallow — callers want visible failures. + log.warn("[{}] scheduled delivery failed for sub={}: {}", profile, sub.id, e.toString()); + } + } + + private WebhookResponse emit(Subscription sub) { + TimerEvent event = new TimerEvent( + sub.message, + Instant.now().toString(), + sub.tick.incrementAndGet(), + sub.id, + sub.periodSeconds); + return agent.getWebhookApi().send(sub.webhookUrl, event, + Map.of("X-Workato-Event-Id", UUID.randomUUID().toString())); + } + + @Override + public void destroy() { + cancelAllSubscriptions(); + if (scheduler != null) { + scheduler.shutdownNow(); + try { + scheduler.awaitTermination(2, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + } + + /** + * Cancels every active subscription without touching the scheduler lifecycle. + * Used by tests to reset state between cases on a shared static harness. + */ + void cancelAllSubscriptions() { + active.values().forEach(s -> s.future.cancel(false)); + active.clear(); + } + + @ExceptionHandler(BadRequestException.class) + public ResponseEntity handleBadRequest(BadRequestException e) { + return ResponseEntity.badRequest().body(new ErrorResponse(e.getMessage())); + } + + @ExceptionHandler(NotFoundException.class) + public ResponseEntity handleNotFound(NotFoundException e) { + return ResponseEntity.status(HttpStatus.NOT_FOUND).body(new ErrorResponse(e.getMessage())); + } + + public record SubscribeRequest( + @JsonProperty("workato_webhook_url") String workatoWebhookUrl, + @JsonProperty("recipe_id") String recipeId, + @JsonProperty("connection_id") String connectionId, + @JsonProperty("message") String message, + @JsonProperty("period_seconds") Integer periodSeconds) { + } + + public record UnsubscribeRequest( + @JsonProperty("recipe_id") String subscriptionId, + @JsonProperty("connection_id") String connectionId) { + } + + public record FireNowRequest( + @JsonProperty("subscription_id") String subscriptionId) { + } + + public record SubscribeResponse(@JsonProperty("id") String id) { + } + + public record FireNowResponse( + @JsonProperty("delivered") boolean delivered, + @JsonProperty("status_code") int statusCode) { + } + + public record ErrorResponse(@JsonProperty("error") String error) { + } + + public record TimerEvent( + @JsonProperty("message") String message, + @JsonProperty("generated_at") String generatedAt, + @JsonProperty("tick") long tick, + @JsonProperty("subscription_id") String subscriptionId, + @JsonProperty("period_seconds") int periodSeconds) { + } + + private static final class Subscription { + final String id; + final String webhookUrl; + final String message; + final int periodSeconds; + final AtomicLong tick = new AtomicLong(0); + volatile ScheduledFuture future; + + Subscription(String id, String webhookUrl, String message, int periodSeconds) { + this.id = id; + this.webhookUrl = webhookUrl; + this.message = message; + this.periodSeconds = periodSeconds; + } + } + + private static final class BadRequestException extends RuntimeException { + BadRequestException(String message) { + super(message); + } + } + + private static final class NotFoundException extends RuntimeException { + NotFoundException(String message) { + super(message); + } + } +} diff --git a/src/main/ruby/timer/connector.rb b/src/main/ruby/timer/connector.rb new file mode 100644 index 0000000..f610b7c --- /dev/null +++ b/src/main/ruby/timer/connector.rb @@ -0,0 +1,178 @@ +# Workato custom connector for the TimerExtension on-prem agent extension. +# +# Companion for: src/main/java/com/mycompany/onprem/timer/TimerExtension.java +# Agent endpoints (all under /ext/{profile}/webhook/): +# GET health — liveness check (used by the connection test lambda) +# POST subscribe — register a periodic timer, returns {"id": ""} +# POST unsubscribe — cancel a timer by its subscription id +# POST fire_now — synchronously emit one webhook event for a subscription +# +# Demonstrates AgentWebhookApiV1 + the Workato SDK webhook_subscribe / +# webhook_notification / webhook_unsubscribe trigger pattern. Single-node / +# in-memory on the agent side (see PLAN.md). + +{ + title: 'On-prem timer (webhook demo)', + secure_tunnel: true, + + connection: { + fields: [ + { + name: 'profile', + label: 'Extension profile', + hint: 'Name of the on-prem TimerExtension profile declared under ' \ + 'extensions: in the agent config.yml.', + optional: false + }, + { + name: 'connection_id', + label: 'Connection ID', + hint: 'A stable identifier for this connection. Threaded through to ' \ + 'the extension on subscribe/unsubscribe so the agent can key ' \ + 'subscriptions per connection.', + optional: false + } + ], + + authorization: { + type: 'custom_auth', + acquire: lambda do |_connection| + {} + end, + apply: lambda do |_connection| + headers('X-Workato-Connector': 'enforce') + end + }, + + base_uri: lambda do |connection| + "http://localhost/ext/#{connection['profile']}/" + end + }, + + test: lambda do |_connection| + get('webhook/health'). + after_error_response(/.*/) do |_code, body, _headers, message| + error("Connection test failed: #{message}. Response: #{body}") + end + end, + + object_definitions: { + timer_event: { + fields: lambda do |_connection, _config_fields, _object_definitions| + [ + { name: 'message', label: 'Message', type: 'string' }, + { name: 'generated_at', label: 'Generated at', type: 'date_time' }, + { name: 'tick', label: 'Tick', type: 'integer' }, + { name: 'subscription_id', label: 'Subscription ID', type: 'string' }, + { name: 'period_seconds', label: 'Period (seconds)', type: 'integer' } + ] + end + } + }, + + actions: { + fire_now: { + title: 'Fire now', + subtitle: 'Emit one webhook event for a subscription immediately', + description: "Synchronously emit one timer event " \ + "to a subscribed webhook URL", + + input_fields: lambda do |_object_definitions| + [ + { name: 'subscription_id', label: 'Subscription ID', + hint: 'The id returned by webhook_subscribe (or logged by the extension).', + optional: false } + ] + end, + + output_fields: lambda do |_object_definitions| + [ + { name: 'delivered', label: 'Delivered', type: 'boolean' }, + { name: 'status_code', label: 'Status code', type: 'integer' } + ] + end, + + execute: lambda do |_connection, input| + post('webhook/fire_now').payload(subscription_id: input['subscription_id']). + after_error_response(/.*/) do |_code, body, _headers, message| + error("fire_now failed: #{message}. Response: #{body}") + end + end, + + sample_output: lambda do |_connection, _input| + { 'delivered' => true, 'status_code' => 200 } + end + } + }, + + triggers: { + tick: { + title: 'New timer tick', + subtitle: 'Fires every N seconds with a user-defined message', + description: "New tick from the " \ + "on-prem timer", + help: 'The extension pushes one event every period_seconds ' \ + 'to the Workato-provided webhook URL. Minimum period is enforced ' \ + 'by the agent (default 5 seconds in production).', + + input_fields: lambda do |_object_definitions| + [ + { name: 'message', label: 'Message', + hint: 'The string the extension echoes back on every tick.', + optional: false }, + { name: 'period_seconds', label: 'Period (seconds)', + type: 'integer', default: 10, + hint: 'Minimum 5 in production — the extension rejects smaller values.', + optional: false } + ] + end, + + webhook_subscribe: lambda do |webhook_url, connection, input, recipe_id| + response = post('webhook/subscribe').payload( + workato_webhook_url: webhook_url, + recipe_id: recipe_id, + connection_id: connection['connection_id'], + message: input['message'], + period_seconds: input['period_seconds'] + ).after_error_response(/.*/) do |_code, body, _headers, message| + error("Failed to subscribe webhook: #{message}. Response: #{body}") + end + # Cache connection_id in the subscribe output so webhook_unsubscribe + # (which the SDK calls with just the subscribe output, no connection) + # can tear the subscription down without a side-channel lookup. + response.merge('connection_id' => connection['connection_id']) + end, + + webhook_notification: lambda do |_input, payload, _ext_input_schema, _ext_output_schema, _headers, _params, _settings, _subscribe_output| + payload + end, + + webhook_unsubscribe: lambda do |webhook_subscribe_output| + post('webhook/unsubscribe').payload( + recipe_id: webhook_subscribe_output['id'], + connection_id: webhook_subscribe_output['connection_id'] + ).after_error_response(/.*/) do |_code, body, _headers, message| + error("Failed to unsubscribe webhook: #{message}. Response: #{body}") + end + end, + + dedup: lambda do |payload| + "#{payload['subscription_id']}-#{payload['tick']}" + end, + + output_fields: lambda do |object_definitions| + object_definitions['timer_event'] + end, + + sample_output: lambda do |_connection, _input| + { + 'message' => 'hello', + 'generated_at' => '2026-10-16T14:03:21Z', + 'tick' => 1, + 'subscription_id' => '0000-aaaa', + 'period_seconds' => 10 + } + end + } + } +} diff --git a/src/test/java/com/mycompany/onprem/timer/TimerExtensionDeployedTest.java b/src/test/java/com/mycompany/onprem/timer/TimerExtensionDeployedTest.java new file mode 100644 index 0000000..c322838 --- /dev/null +++ b/src/test/java/com/mycompany/onprem/timer/TimerExtensionDeployedTest.java @@ -0,0 +1,100 @@ +package com.mycompany.onprem.timer; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.mycompany.onprem.timer.TimerExtension.FireNowRequest; +import com.mycompany.onprem.timer.TimerExtension.FireNowResponse; +import com.mycompany.onprem.timer.TimerExtension.SubscribeRequest; +import com.mycompany.onprem.timer.TimerExtension.SubscribeResponse; +import com.sun.net.httpserver.HttpServer; +import com.workato.agent.extension.test.DeployedAgentLauncher; +import com.workato.agent.extension.test.DeployedAgentLauncher.RunningAgent; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; + +import java.net.InetSocketAddress; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.time.Duration; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Runs {@link TimerExtension} inside the real on-prem agent via + * {@link DeployedAgentLauncher}. Manual-only — see + * {@code extension-api/README-extension-test.md}, section + * "Running against the real agent": each start() mutates process-global state + * in the agent, so only one @Test method may run per JVM invocation. + * Run one method at a time from the IDE. + */ +@Disabled("Manual-only — needs deployed+activated agent at AGENT_DIST; run one method per JVM.") +class TimerExtensionDeployedTest { + + private static final Path AGENT_DIST = Paths.get("/Users/krocodl/Apps/OPA-Preview-US-34"); + private static final ObjectMapper MAPPER = new ObjectMapper(); + + @Test + void firesEventToLocalSinkOverLocalHttp() throws Exception { + AtomicReference capturedBody = new AtomicReference<>(); + AtomicReference capturedEventId = new AtomicReference<>(); + + HttpServer sink = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + sink.createContext("/", exchange -> { + capturedBody.set(new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8)); + capturedEventId.set(exchange.getRequestHeaders().getFirst("X-Workato-Event-Id")); + exchange.sendResponseHeaders(200, 0); + exchange.close(); + }); + sink.start(); + String sinkUrl = "http://127.0.0.1:" + sink.getAddress().getPort() + "/"; + + try (RunningAgent agent = DeployedAgentLauncher.create("timer", AGENT_DIST) + .controllerClass(TimerExtension.class) + // 60s period keeps the scheduler from ticking during this test; we drive + // delivery through fire_now instead to make the assertion deterministic. + .property("min-period-seconds", "60") + .openLocalAccess() + .start()) { + + HttpResponse subRes = agent.httpPost( + TimerExtension::subscribe, + new SubscribeRequest(sinkUrl, "recipe-1", "conn-1", "hello", 60)); + assertEquals(200, subRes.statusCode(), "subscribe body=" + subRes.body()); + String id = MAPPER.readValue(subRes.body(), SubscribeResponse.class).id(); + assertNotNull(id, "no subscription id in subscribe response"); + + HttpResponse fireRes = agent.httpPost( + TimerExtension::fireNow, new FireNowRequest(id)); + assertEquals(200, fireRes.statusCode(), "fire_now body=" + fireRes.body()); + FireNowResponse fr = MAPPER.readValue(fireRes.body(), FireNowResponse.class); + assertTrue(fr.delivered(), "extension reported delivery failure: " + fireRes.body()); + assertEquals(200, fr.statusCode()); + + assertNotNull(capturedBody.get(), "sink did not receive the webhook event"); + assertTrue(capturedBody.get().contains("\"message\":\"hello\""), capturedBody.get()); + assertTrue(capturedBody.get().contains("\"tick\":1"), capturedBody.get()); + assertTrue(capturedBody.get().contains("\"subscription_id\":\"" + id + "\""), capturedBody.get()); + assertNotNull(capturedEventId.get(), "X-Workato-Event-Id header missing on delivery"); + } finally { + sink.stop(0); + } + } + + @Test + void connectsToWorkatoGateway() throws Exception { + try (RunningAgent agent = DeployedAgentLauncher.create("timer", AGENT_DIST) + .controllerClass(TimerExtension.class) + .property("min-period-seconds", "10") + // No .openLocalAccess(...) → full gateway mode. + .start()) { + + String line = agent.awaitGatewayConnected(Duration.ofSeconds(60)); + assertNotNull(line, "awaitGatewayConnected returned null"); + assertTrue(line.contains("Connected to gateway"), + "unexpected gateway marker line: " + line); + } + } + +} diff --git a/src/test/java/com/mycompany/onprem/timer/TimerExtensionTest.java b/src/test/java/com/mycompany/onprem/timer/TimerExtensionTest.java new file mode 100644 index 0000000..f5d5196 --- /dev/null +++ b/src/test/java/com/mycompany/onprem/timer/TimerExtensionTest.java @@ -0,0 +1,247 @@ +package com.mycompany.onprem.timer; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.mycompany.onprem.timer.TimerExtension.ErrorResponse; +import com.mycompany.onprem.timer.TimerExtension.FireNowResponse; +import com.mycompany.onprem.timer.TimerExtension.SubscribeResponse; +import com.mycompany.onprem.timer.TimerExtension.TimerEvent; +import com.workato.agent.extension.api.AgentWebhookApiV1.RetryPolicy; +import com.workato.agent.extension.test.AgentExtensionHarness; +import com.workato.agent.extension.test.FakeAgentWebhookApiV1.CallKind; +import com.workato.agent.extension.test.FakeAgentWebhookApiV1.SentCall; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.springframework.http.MediaType; +import org.springframework.test.web.servlet.MvcResult; + +import java.time.Duration; +import java.time.Instant; +import java.util.Map; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.*; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; + +class TimerExtensionTest { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + @RegisterExtension + static AgentExtensionHarness agent = AgentExtensionHarness.extension("timer") + .yaml("extensions/timer.yml"); + + private static String subscribe(AgentExtensionHarness h, + String url, int periodSeconds, String message) throws Exception { + String json = MAPPER.writeValueAsString(Map.of( + "workato_webhook_url", url, + "recipe_id", "recipe-xyz", + "connection_id", "conn-1", + "message", message, + "period_seconds", periodSeconds)); + MvcResult result = postJson(h, "/webhook/subscribe", json); + assertEquals(200, result.getResponse().getStatus(), + "subscribe body=" + result.getResponse().getContentAsString()); + return MAPPER.readValue(result.getResponse().getContentAsString(), SubscribeResponse.class).id(); + } + + private static MvcResult postJson(AgentExtensionHarness h, String path, String json) throws Exception { + return h.mockMvc().perform(post(h.urlFor(path)) + .contentType(MediaType.APPLICATION_JSON) + .content(json)) + .andReturn(); + } + + private static void awaitAtLeast(AgentExtensionHarness h, int n, Duration deadline) throws Exception { + Instant end = Instant.now().plus(deadline); + while (Instant.now().isBefore(end)) { + if (h.webhookApi().allCalls().size() >= n) { + return; + } + Thread.sleep(50); + } + throw new AssertionError("expected at least " + n + " webhook calls within " + deadline + + ", got " + h.webhookApi().allCalls().size()); + } + + @BeforeEach + void resetWebhookState() { + agent.webhookApi().reset(); + } + + @AfterEach + void cancelSubscriptions() { + // Clear every subscription registered during the test — this stops the + // scheduler ticking into the next test's fake-webhook assertions. Reaches + // into the bean because we deliberately don't expose the state over HTTP. + agent.webApplicationContext().getBean(TimerExtension.class).cancelAllSubscriptions(); + } + + @Test + void subscribeReturnsIdUuid() throws Exception { + agent.webhookApi().whenSent("https://recipe/wh").thenReturn(200, "{}"); + String id = subscribe(agent, "https://recipe/wh", 5, "hi"); + UUID parsed = assertDoesNotThrow(() -> UUID.fromString(id), "not a UUID: " + id); + assertEquals(4, parsed.version(), "expected a UUIDv4"); + } + + @Test + void fireNowPostsExpectedPayload() throws Exception { + agent.webhookApi().whenSent("https://recipe/wh").thenReturn(200, "{}"); + String id = subscribe(agent, "https://recipe/wh", 5, "hi"); + + MvcResult fire = postJson(agent, "/webhook/fire_now", + MAPPER.writeValueAsString(Map.of("subscription_id", id))); + assertEquals(200, fire.getResponse().getStatus()); + + FireNowResponse fr = MAPPER.readValue(fire.getResponse().getContentAsString(), FireNowResponse.class); + assertTrue(fr.delivered()); + assertEquals(200, fr.statusCode()); + + SentCall call = agent.webhookApi().lastCall(); + assertEquals(CallKind.SEND, call.kind()); + assertEquals("https://recipe/wh", call.url()); + assertNotNull(call.headers().get("X-Workato-Event-Id")); + + TimerEvent event = MAPPER.convertValue(call.body(), TimerEvent.class); + assertEquals("hi", event.message()); + assertEquals(1L, event.tick()); + assertEquals(id, event.subscriptionId()); + assertEquals(5, event.periodSeconds()); + assertTrue(event.generatedAt().matches("\\d{4}-\\d{2}-\\d{2}T.*Z"), + "not ISO-8601: " + event.generatedAt()); + } + + @Test + void fireNowUnknownIdReturns404() throws Exception { + MvcResult result = postJson(agent, "/webhook/fire_now", + MAPPER.writeValueAsString(Map.of("subscription_id", "does-not-exist"))); + assertEquals(404, result.getResponse().getStatus()); + assertTrue(agent.webhookApi().allCalls().isEmpty()); + } + + @Test + void subscribeWithBlankMessageReturns400() throws Exception { + String json = MAPPER.writeValueAsString(Map.of( + "workato_webhook_url", "https://recipe/wh", + "recipe_id", "r", "connection_id", "c", + "message", " ", "period_seconds", 5)); + MvcResult result = postJson(agent, "/webhook/subscribe", json); + assertEquals(400, result.getResponse().getStatus()); + ErrorResponse err = MAPPER.readValue(result.getResponse().getContentAsString(), ErrorResponse.class); + assertTrue(err.error().contains("message must not be blank"), + "unexpected error: " + err.error()); + } + + @Test + void subscribeWithBlankUrlReturns400() throws Exception { + String json = MAPPER.writeValueAsString(Map.of( + "workato_webhook_url", "", + "recipe_id", "r", "connection_id", "c", + "message", "hi", "period_seconds", 5)); + MvcResult result = postJson(agent, "/webhook/subscribe", json); + assertEquals(400, result.getResponse().getStatus()); + ErrorResponse err = MAPPER.readValue(result.getResponse().getContentAsString(), ErrorResponse.class); + assertTrue(err.error().contains("workato_webhook_url"), + "unexpected error: " + err.error()); + } + + @Test + void subscribeBelowMinReturns400() throws Exception { + AgentExtensionHarness strict = AgentExtensionHarness.extension("timer-strict") + .controller(TimerExtension.class) + .property("min-period-seconds", "10") + .start(); + try { + String json = MAPPER.writeValueAsString(Map.of( + "workato_webhook_url", "https://recipe/wh", + "recipe_id", "r", "connection_id", "c", + "message", "hi", "period_seconds", 1)); + MvcResult result = postJson(strict, "/webhook/subscribe", json); + assertEquals(400, result.getResponse().getStatus()); + ErrorResponse err = MAPPER.readValue(result.getResponse().getContentAsString(), ErrorResponse.class); + assertTrue(err.error().contains("must be >="), "unexpected error: " + err.error()); + } finally { + strict.stop(); + } + } + + @Test + void unsubscribeCancelsScheduledTask() throws Exception { + agent.webhookApi().whenSent("https://recipe/wh").thenReturn(200, "{}"); + String id = subscribe(agent, "https://recipe/wh", 1, "tick"); + + awaitAtLeast(agent, 1, Duration.ofSeconds(3)); + postJson(agent, "/webhook/unsubscribe", + MAPPER.writeValueAsString(Map.of("recipe_id", id, "connection_id", "conn-1"))); + + int n = agent.webhookApi().allCalls().size(); + // One period + a cushion — long enough for a stray tick to show up if cancel didn't take. + Thread.sleep(1500); + assertEquals(n, agent.webhookApi().allCalls().size(), + "scheduler kept firing after unsubscribe"); + } + + @Test + void contextShutdownStopsScheduler() throws Exception { + AgentExtensionHarness alt = AgentExtensionHarness.extension("timer-alt") + .controller(TimerExtension.class) + .property("min-period-seconds", "1") + .start(); + try { + var fake = alt.webhookApi(); + fake.whenSent("https://recipe/wh").thenReturn(200, "{}"); + subscribe(alt, "https://recipe/wh", 1, "tick"); + awaitAtLeast(alt, 1, Duration.ofSeconds(3)); + int n = fake.allCalls().size(); + alt.stop(); + Thread.sleep(1500); + assertEquals(n, fake.allCalls().size(), + "scheduler kept firing after context shutdown"); + } finally { + // Idempotent — stop() above is the happy path, this covers early failures. + try { + alt.stop(); + } catch (Exception ignored) { + } + } + } + + @Test + void multipleSubscriptionsFireIndependently() throws Exception { + agent.webhookApi().whenSent("https://a/wh").thenReturn(200, "{}"); + agent.webhookApi().whenSent("https://b/wh").thenReturn(200, "{}"); + String idA = subscribe(agent, "https://a/wh", 5, "A"); + String idB = subscribe(agent, "https://b/wh", 5, "B"); + + postJson(agent, "/webhook/fire_now", MAPPER.writeValueAsString(Map.of("subscription_id", idA))); + postJson(agent, "/webhook/fire_now", MAPPER.writeValueAsString(Map.of("subscription_id", idB))); + + assertEquals(1, agent.webhookApi().callsFor("https://a/wh").size()); + assertEquals(1, agent.webhookApi().callsFor("https://b/wh").size()); + + TimerEvent ea = MAPPER.convertValue( + agent.webhookApi().callsFor("https://a/wh").get(0).body(), TimerEvent.class); + TimerEvent eb = MAPPER.convertValue( + agent.webhookApi().callsFor("https://b/wh").get(0).body(), TimerEvent.class); + assertEquals(idA, ea.subscriptionId()); + assertEquals(idB, eb.subscriptionId()); + assertEquals("A", ea.message()); + assertEquals("B", eb.message()); + } + + @Test + void swallowsDeliveryFailuresInScheduledTick() throws Exception { + // Cap retries at 1 so each tick = exactly one captured attempt — then + // waiting for ≥2 attempts proves the scheduler thread survived the + // first tick's failure and kept firing. With the default (5 attempts) + // the first tick alone would satisfy ≥2 and the test would pass even + // if the scheduler died. + agent.webhookApi().setDefaultRetryPolicy(RetryPolicy.DEFAULTS.withAttempts(1)); + agent.webhookApi().whenSent("https://recipe/wh").thenFailAll(); + subscribe(agent, "https://recipe/wh", 1, "tick"); + + awaitAtLeast(agent, 2, Duration.ofSeconds(5)); + } +} diff --git a/src/test/resources/extensions/timer.yml b/src/test/resources/extensions/timer.yml new file mode 100644 index 0000000..7b72dbd --- /dev/null +++ b/src/test/resources/extensions/timer.yml @@ -0,0 +1,4 @@ +extensions: + timer: + controller: com.mycompany.onprem.timer.TimerExtension + min-period-seconds: 1 diff --git a/src/test/resources/tape_library/timer/fire_now.yml b/src/test/resources/tape_library/timer/fire_now.yml new file mode 100644 index 0000000..8efa270 --- /dev/null +++ b/src/test/resources/tape_library/timer/fire_now.yml @@ -0,0 +1,22 @@ +--- +http_interactions: + - request: + method: post + uri: http://localhost/ext/timer/webhook/fire_now + body: + encoding: UTF-8 + string: '' + headers: { } + response: + status: + code: 200 + message: OK + headers: + Content-Type: + - application/json + body: + encoding: UTF-8 + string: '{"delivered":true,"status_code":200}' + http_version: '1.1' + recorded_at: Thu, 16 Oct 2026 10:00:00 GMT +recorded_with: VCR 6.3.1 diff --git a/src/test/resources/tape_library/timer/subscribe.yml b/src/test/resources/tape_library/timer/subscribe.yml new file mode 100644 index 0000000..3d7b170 --- /dev/null +++ b/src/test/resources/tape_library/timer/subscribe.yml @@ -0,0 +1,22 @@ +--- +http_interactions: + - request: + method: post + uri: http://localhost/ext/timer/webhook/subscribe + body: + encoding: UTF-8 + string: '' + headers: { } + response: + status: + code: 200 + message: OK + headers: + Content-Type: + - application/json + body: + encoding: UTF-8 + string: '{"id":"00000000-0000-4000-8000-00000000aaaa"}' + http_version: '1.1' + recorded_at: Thu, 16 Oct 2026 10:00:00 GMT +recorded_with: VCR 6.3.1 diff --git a/src/test/resources/tape_library/timer/test.yml b/src/test/resources/tape_library/timer/test.yml new file mode 100644 index 0000000..53cfd2d --- /dev/null +++ b/src/test/resources/tape_library/timer/test.yml @@ -0,0 +1,20 @@ +--- +http_interactions: + - request: + method: get + uri: http://localhost/ext/timer/webhook/health + body: + encoding: UTF-8 + string: '' + headers: { } + response: + status: + code: 200 + message: OK + headers: { } + body: + encoding: UTF-8 + string: '' + http_version: '1.1' + recorded_at: Thu, 16 Oct 2026 10:00:00 GMT +recorded_with: VCR 6.3.1 diff --git a/src/test/resources/tape_library/timer/unsubscribe.yml b/src/test/resources/tape_library/timer/unsubscribe.yml new file mode 100644 index 0000000..5b2af74 --- /dev/null +++ b/src/test/resources/tape_library/timer/unsubscribe.yml @@ -0,0 +1,20 @@ +--- +http_interactions: + - request: + method: post + uri: http://localhost/ext/timer/webhook/unsubscribe + body: + encoding: UTF-8 + string: '' + headers: { } + response: + status: + code: 204 + message: No Content + headers: { } + body: + encoding: UTF-8 + string: '' + http_version: '1.1' + recorded_at: Thu, 16 Oct 2026 10:00:00 GMT +recorded_with: VCR 6.3.1 diff --git a/src/test/ruby/timer/.rspec b/src/test/ruby/timer/.rspec new file mode 100644 index 0000000..4e33a32 --- /dev/null +++ b/src/test/ruby/timer/.rspec @@ -0,0 +1,3 @@ +--require spec_helper +--color +--format documentation diff --git a/src/test/ruby/timer/settings.yaml b/src/test/ruby/timer/settings.yaml new file mode 100644 index 0000000..130f2f4 --- /dev/null +++ b/src/test/ruby/timer/settings.yaml @@ -0,0 +1,2 @@ +profile: timer +connection_id: test-conn diff --git a/src/test/ruby/timer/spec/connector_spec.rb b/src/test/ruby/timer/spec/connector_spec.rb new file mode 100644 index 0000000..16a6cdf --- /dev/null +++ b/src/test/ruby/timer/spec/connector_spec.rb @@ -0,0 +1,90 @@ +# frozen_string_literal: true +# +# How to run this spec +# -------------------- +# From Gradle (works everywhere, uses the shared src/test/ruby/Gemfile): +# ./gradlew :timerRspec # whole file +# ./gradlew :timerRspec -Pspec='fire_now' # one example by description +# ./gradlew :timerRspec -Pspec=spec/connector_spec.rb:12 # one example by file:line +# +# From IDEA with the JetBrains Ruby plugin (Ultimate / RubyMine): +# right-click the gutter triangle on describe/it. Make sure the RSpec +# run-config template points Working directory at src/test/ruby/timer +# (VCR resolves cassette paths from CWD). +# +# From IDEA Community (no Ruby plugin): Run → Edit Configurations → + → Gradle, +# Tasks: :timerRspec, Arguments: one of the -Pspec=... forms above. + +CONNECTOR_PATH = File.expand_path('../../../../main/ruby/timer/connector.rb', __dir__).freeze + +RSpec.describe 'onprem timer connector' do + let(:settings) { Workato::Connector::Sdk::Settings.from_default_file } + let(:connector) { Workato::Connector::Sdk::Connector.from_file(CONNECTOR_PATH, settings) } + + describe 'test' do + it 'pings the health endpoint without raising' do + VCR.use_cassette('test') do + expect { connector.test(settings) }.not_to raise_error + end + end + end + + describe 'actions.fire_now' do + it 'posts the subscription id and returns the delivered/status pair' do + VCR.use_cassette('fire_now') do + output = connector.actions.fire_now.execute(settings, 'subscription_id' => 'sub-abc') + expect(output).to be_a(Hash) + expect(output['delivered']).to be(true) + expect(output['status_code']).to eq(200) + end + end + end + + describe 'triggers.tick' do + it 'webhook_subscribe posts the subscription body and returns {id, connection_id}' do + VCR.use_cassette('subscribe') do + output = connector.triggers.tick.webhook_subscribe( + 'https://recipe/wh', + settings, + { 'message' => 'hi', 'period_seconds' => 10 }, + 'recipe-xyz' + ) + expect(output['id']).to match(/\A[0-9a-f-]{36}\z/) + # connection_id is cached into the subscribe output so webhook_unsubscribe + # can access it later (SDK does not re-supply connection on unsubscribe). + expect(output['connection_id']).to eq('test-conn') + end + end + + it 'webhook_notification returns the payload hash as the event' do + payload = { + 'tick' => 1, + 'message' => 'hi', + 'subscription_id' => 'sub-abc', + 'period_seconds' => 10, + 'generated_at' => '2026-10-16T14:03:21Z' + } + event = connector.triggers.tick.webhook_notification( + {}, payload, [], [], {}, {}, nil, {} + ) + expect(event).to eq(payload) + end + + it 'webhook_unsubscribe posts recipe_id (= subscription id) and connection_id' do + VCR.use_cassette('unsubscribe') do + expect { + connector.triggers.tick.webhook_unsubscribe( + { 'id' => 'sub-abc', 'connection_id' => 'test-conn' } + ) + }.not_to raise_error + end + end + + it 'dedup is deterministic from subscription_id + tick (no wall clock)' do + key1 = connector.triggers.tick.dedup({ 'subscription_id' => 'sub-abc', 'tick' => 7 }) + key2 = connector.triggers.tick.dedup({ 'subscription_id' => 'sub-abc', 'tick' => 7 }) + expect(key1).to eq('sub-abc-7') + expect(key1).to eq(key2) + end + end +end diff --git a/src/test/ruby/timer/spec/spec_helper.rb b/src/test/ruby/timer/spec/spec_helper.rb new file mode 100644 index 0000000..928a3e7 --- /dev/null +++ b/src/test/ruby/timer/spec/spec_helper.rb @@ -0,0 +1,3 @@ +# frozen_string_literal: true + +require_relative '../../spec_helper'