Implement Fetcher#scheduled() for dispatching scheduled events

Like `Fetcher#queue()`, this is gated behind the experimental
`service_binding_extra_handlers` compatibility flag.

Co-authored-by: jjohnson <jjohnson@cloudflare.com>

src/workerd/api/http-test.js

@@ -0,0 +1,34 @@
+// Copyright (c) 2023 Cloudflare, Inc.
+// Licensed under the Apache 2.0 license found in the LICENSE file or at:
+//     https://opensource.org/licenses/Apache-2.0
+
+import assert from "node:assert";
+
+let scheduledLastCtrl;
+
+export default {
+  async scheduled(ctrl, env, ctx) {
+    scheduledLastCtrl = ctrl;
+    if (ctrl.cron === "* * * * 30") ctrl.noRetry();
+  },
+
+  async test(ctrl, env, ctx) {
+    // Call `scheduled()` with no options
+    {
+      const result = await env.SERVICE.scheduled();
+      assert.strictEqual(result.outcome, /* OK */ 1);
+      assert(!result.noRetry);
+      assert(Math.abs(Date.now() - scheduledLastCtrl.scheduledTime) < 3_000);
+      assert.strictEqual(scheduledLastCtrl.cron, "");
+    }
+
+    // Call `scheduled()` with options, and noRetry()
+    {
+      const result = await env.SERVICE.scheduled({ scheduledTime: 1000, cron: "* * * * 30" });
+      assert.strictEqual(result.outcome, /* OK */ 1);
+      assert(result.noRetry);
+      assert.strictEqual(scheduledLastCtrl.scheduledTime, 1000);
+      assert.strictEqual(scheduledLastCtrl.cron, "* * * * 30");
+    }
+  }
+}

src/workerd/api/http-test.wd-test

@@ -0,0 +1,18 @@
+using Workerd = import "/workerd/workerd.capnp";
+
+const unitTests :Workerd.Config = (
+  services = [
+    ( name = "http-test",
+      worker = (
+        modules = [
+          ( name = "worker", esModule = embed "http-test.js" )
+        ],
+        bindings = [
+          ( name = "SERVICE", service = "http-test" )
+        ],
+        compatibilityDate = "2023-08-01",
+        compatibilityFlags = ["nodejs_compat", "service_binding_extra_handlers"],
+      )
+    ),
+  ],
+);

src/workerd/api/http.c++

@@ -2052,6 +2052,32 @@ jsg::Promise<QueueResponse> Fetcher::queue(
   });
 }
 
+jsg::Promise<Fetcher::ScheduledResult> Fetcher::scheduled(
+      jsg::Lock& js, jsg::Optional<ScheduledOptions> options) {
+  auto& ioContext = IoContext::current();
+  auto worker = getClient(ioContext, nullptr, "scheduled"_kjc);
+
+  auto scheduledTime = ioContext.now();
+  auto cron = kj::String();
+  KJ_IF_MAYBE(o, options) {
+    KJ_IF_MAYBE(t, o->scheduledTime) {
+      scheduledTime = *t;
+    }
+    KJ_IF_MAYBE(c, o->cron) {
+      cron = kj::mv(*c);
+    }
+  }
+
+  return ioContext.awaitIo(js, worker->runScheduled(scheduledTime, cron)
+      .attach(kj::mv(worker), kj::mv(cron)),
+      [](jsg::Lock& js, WorkerInterface::ScheduledResult result) {
+    return Fetcher::ScheduledResult{
+      .outcome=static_cast<uint16_t>(result.outcome),
+      .noRetry=!result.retry,
+    };
+  });
+}
+
 kj::Own<WorkerInterface> Fetcher::getClient(
     IoContext& ioContext, kj::Maybe<kj::String> cfStr, kj::ConstString operationName) {
   KJ_SWITCH_ONEOF(channelOrClientFactory) {

src/workerd/api/http.h

@@ -471,12 +471,29 @@ class Fetcher: public jsg::Object {
   jsg::Promise<QueueResponse> queue(
       jsg::Lock& js, kj::String queueName, kj::Array<ServiceBindingQueueMessage> messages);
 
+  struct ScheduledOptions {
+    jsg::Optional<kj::Date> scheduledTime;
+    jsg::Optional<kj::String> cron;
+
+    JSG_STRUCT(scheduledTime, cron);
+  };
+
+  struct ScheduledResult {
+    uint16_t outcome;
+    bool noRetry;
+
+    JSG_STRUCT(outcome, noRetry);
+  };
+
+  jsg::Promise<ScheduledResult> scheduled(jsg::Lock& js, jsg::Optional<ScheduledOptions> options);
+
   JSG_RESOURCE_TYPE(Fetcher, CompatibilityFlags::Reader flags) {
     JSG_METHOD(fetch);
     JSG_METHOD(connect);
 
     if (flags.getServiceBindingExtraHandlers()) {
       JSG_METHOD(queue);
+      JSG_METHOD(scheduled);
     }
 
     JSG_METHOD(get);
@@ -1075,6 +1092,8 @@ kj::String makeRandomBoundaryCharacters();
   api::Request::InitializerDict,      \
   api::Fetcher,                       \
   api::Fetcher::PutOptions,           \
+  api::Fetcher::ScheduledOptions,     \
+  api::Fetcher::ScheduledResult,      \
   api::Fetcher::ServiceBindingQueueMessage
 
 // The list of http.h types that are added to worker.c++'s JSG_DECLARE_ISOLATE_TYPE