Add send_after to JS runtime

drift_js/CHANGELOG

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+### Added
+- Adds the `send_after` function to `runtime`, which allows sending messages
+  with a delay, or scheduling them on the task queue with a delay of zero.
+
 ## [1.0.1] - 2025-17-20
 
 ### Fixed

drift_js/src/drift/js/channel.gleam

@@ -2,9 +2,8 @@
 //// Provided for convenience, not required to use `drift_js`.
 
 import gleam/javascript/promise.{type Promise}
-import gleam/result
 
-/// An unbounded single-consumer channel with synchronous sending
+/// An unbounded single-consumer channel with non-blocking sending
 /// and asynchronous receiving.
 /// Intended to be used similarly to a `gleam/erlang` `Subject`.
 pub type Channel(a)

drift_js/src/drift/js/internal/event_loop.gleam

@@ -44,6 +44,10 @@ pub fn error_if_stopped(
 @external(javascript, "../../../drift_event_loop.mjs", "send")
 pub fn send(loop: EventLoop(i), input: i) -> Nil
 
+/// Uses `setTimeout` to call `send` after a delay.
+@external(javascript, "../../../drift_event_loop.mjs", "send_after")
+pub fn send_after(loop: EventLoop(i), delay: Int, input: i) -> Nil
+
 /// Sets the time to the next time `receive` should return `Tick`.
 /// Only one timeout can be set at a time.
 /// Returns an error if a timeout is already set.

drift_js/src/drift/js/runtime.gleam

@@ -35,10 +35,20 @@ pub type TerminalResult(a, e) {
 }
 
 /// Sends an input to be handled by the runtime.
+/// This will resolve a promise under the hood, and thus completing the receive
+/// will be scheduled as a microtask.
 pub fn send(runtime: Runtime(i), input: i) -> Nil {
   event_loop.send(runtime.loop, input)
 }
 
+/// Sends an input to be handled by the runtime after a delay (in milliseconds).
+/// Since triggering the receive from `send` will be scheduled as a microtask,
+/// using `send_after` with a delay of 0 can be used to handle an input as
+/// a task instead.
+pub fn send_after(runtime: Runtime(i), delay: Int, input: i) -> Nil {
+  event_loop.send_after(runtime.loop, delay, input)
+}
+
 /// Similar to `process.call_forever` on Gleam on Erlang.
 pub fn call_forever(
   runtime: Runtime(i),

drift_js/src/drift_event_loop.mjs

@@ -38,6 +38,10 @@ export function send(loop, message) {
     return loop.send(new HandleInput(message));
 }
 
+export function send_after(loop, delay, message) {
+    setTimeout(() => send(loop, message), delay);
+}
+
 export function set_timeout(loop, after) {
     return loop.setTimeout(after);
 }

drift_js/test/drift/js/internal/event_loop_test.gleam

@@ -2,6 +2,7 @@ import drift/js/internal/event_loop.{
   type Event, type EventLoop, AlreadyReceiving, AlreadyTicking, HandleInput,
   Stopped, Tick,
 }
+import drift/js/runtime
 import gleam/javascript/promise.{type Promise, await}
 
 pub fn receive_when_empty_test() {
@@ -84,6 +85,24 @@ pub fn send_cancels_timeout_test() {
   promise.resolve(Nil)
 }
 
+pub fn send_after_test() {
+  let loop = event_loop.start()
+  let start = runtime.now()
+
+  event_loop.send_after(loop, 10, 42)
+
+  let assert Ok(result) = event_loop.receive(loop)
+  use result <- await(timeout(result, 20))
+  assert result == Ok(HandleInput(42))
+
+  // Assert reasonably accurate delay
+  let elapsed = runtime.now() - start
+  assert elapsed >= 10
+  assert elapsed <= 12
+
+  promise.resolve(Nil)
+}
+
 pub fn queued_receive_cancels_timeout_test() {
   let loop = event_loop.start()
 

drift_js/test/drift/shceduling_test.gleam

@@ -0,0 +1,66 @@
+import drift
+import drift/js/channel.{type Channel}
+import drift/js/runtime
+import gleam/javascript/promise.{type Promise}
+import gleam/list
+
+type State {
+  State(value: Int, remaining_sends: Int)
+}
+
+// Checks that a single runtime that sends messages to itself repeatedly
+// still allows other runtimes to run.
+pub fn multiple_runtime_scheduling_test() -> Promise(Nil) {
+  let results = channel.new()
+  let #(termination1, runtime1) =
+    runtime.start(State(1, 5), fn(_) { results }, handle_input, handle_output)
+  let #(termination2, runtime2) =
+    runtime.start(State(2, 5), fn(_) { results }, handle_input, handle_output)
+
+  // Kick off the event handling
+  runtime.send(runtime1, Nil)
+  runtime.send(runtime2, Nil)
+
+  // Wait for termination
+  use _ <- promise.await(termination1)
+  use _ <- promise.await(termination2)
+
+  assert channel_to_list(results, []) == [1, 2, 1, 2, 1, 2, 1, 2, 1, 2]
+
+  promise.resolve(Nil)
+}
+
+fn handle_output(
+  context: drift.EffectContext(channel.Channel(Int)),
+  value: Int,
+  send: fn(Nil) -> Nil,
+) -> Result(drift.EffectContext(channel.Channel(Int)), Nil) {
+  // Send who was scheduled to results
+  let results = drift.read_effect_context(context)
+  channel.send(results, value)
+
+  // Trigger another input to be handled
+  send(Nil)
+  Ok(context)
+}
+
+fn handle_input(
+  ctx: drift.Context(a, Int),
+  state: State,
+  _,
+) -> drift.Step(State, a, Int, c) {
+  case state.remaining_sends {
+    0 -> drift.stop(ctx, state)
+    remaining_sends ->
+      ctx
+      |> drift.output(state.value)
+      |> drift.continue(State(..state, remaining_sends: remaining_sends - 1))
+  }
+}
+
+fn channel_to_list(channel: Channel(a), values: List(a)) -> List(a) {
+  case channel.try_receive(channel) {
+    Ok(value) -> channel_to_list(channel, [value, ..values])
+    Error(Nil) -> list.reverse(values)
+  }
+}

drift_js/test/drift_js_test.gleam

@@ -1,9 +1,10 @@
 import drift.{type Action, type Context, type Effect, type Step}
 import drift/js/runtime.{
-  type Runtime, type TerminalResult, CallTimedOut, RuntimeStopped,
+  type Runtime, type TerminalResult, CallTimedOut, RuntimeStopped, Terminated,
 }
 import exemplify
 import gleam/javascript/promise.{type Promise, await}
+import gleam/list
 import gleeunit
 
 pub fn main() -> Nil {
@@ -112,6 +113,32 @@ pub fn call_timeout_test() {
   promise.resolve(Nil)
 }
 
+pub fn send_after_0_is_delayed_test() {
+  use <- timeout(100)
+
+  let #(result, rt) =
+    start_without_io([], fn(ctx, state, input) {
+      let state = [input, ..state]
+      case list.length(state) {
+        4 -> drift.stop(ctx, state)
+        _ -> drift.continue(ctx, state)
+      }
+    })
+
+  runtime.send_after(rt, 0, "after1")
+  runtime.send(rt, "immediate1")
+  runtime.send_after(rt, 0, "after2")
+  runtime.send(rt, "immediate2")
+  use result <- promise.await(result)
+
+  // Reverse list for clearer assertion
+  let assert Terminated(result) = result
+  let result = list.reverse(result)
+  assert result == ["immediate1", "immediate2", "after1", "after2"]
+
+  promise.resolve(Nil)
+}
+
 fn noop(ctx: Context(i, o), state: s, _: i) -> Step(s, i, o, e) {
   ctx |> drift.continue(state)
 }
@@ -133,7 +160,7 @@ fn stop(ctx: Context(Bool, o), state: s, stop: Bool) -> Step(s, Bool, o, e) {
   }
 }
 
-pub fn start_with_action_executor(
+fn start_with_action_executor(
   state: s,
   next: fn(Context(i, Action(a)), s, i) -> Step(s, i, Action(a), e),
 ) -> #(Promise(TerminalResult(s, e)), Runtime(i)) {