off_topic

import gleam/dynamic.{type Dynamic}
import gleam/dynamic/decode.{type Decoder}
import gleam/int
import gleam/json.{type Json}
import gleam/list
import gleam/option.{type Option, None, Some}
import gleam/result
import lustre.{type App}
import lustre/effect.{type Effect}
import lustre/element.{type Element}
import lustre/event
import lustre/server_component
import off_topic/component

// -- TYPES -------------------------------------------------------------------

/// An active listener that dispatches messages to your Lustre application.
///
/// Build subscriptions with the functions in this module and return them from
/// your application's `subscriptions` callback. Use [batch](#batch) to combine
/// multiple subscriptions, and [none](#none) to return an empty one.
pub opaque type Subscription(message) {
  RequestAnimationFrame(callback: fn() -> message)
  Simple(start: fn(fn(message) -> Nil) -> fn() -> Nil, config: Config)
  Element(start: fn(fn(message) -> Nil, Dynamic) -> fn() -> Nil, config: Config)
  Remote(
    name: String,
    params: List(Json),
    decoder: Decoder(message),
    config: Config,
  )
  Batch(subscriptions: List(Subscription(message)))
}

type Config {
  Config(
    dependencies: List(Dependency),
    throttle: Int,
    delay: Int,
    include: List(String),
  )
}

fn config(dependencies) {
  Config(dependencies:, throttle: 0, delay: 0, include: [])
}

fn update_config(subscription, updater) {
  case subscription {
    RequestAnimationFrame(..) -> subscription
    Simple(config:, ..) -> Simple(..subscription, config: updater(config))
    Element(config:, ..) -> Element(..subscription, config: updater(config))
    Remote(config:, ..) -> Remote(..subscription, config: updater(config))
    Batch(subscriptions:) ->
      Batch(list.map(subscriptions, update_config(_, updater)))
  }
}

/// An opaque value off_topic uses to detect when a subscription's inputs have
/// changed and it needs to be restarted.
///
/// Create one with [dep](#dep).
pub type Dependency

/// The subscription runtime's wrapper around your application model.
///
/// Use with [init](#init), [update](#update), and [view](#view) when wiring
/// up off_topic manually instead of through [application](#application).
pub opaque type Model(model, message) {
  Model(
    tick: Int,
    app: model,
    mounted: Bool,
    state: Running(message),
    connected_clients: Int,
  )
}

/// The subscription runtime's wrapper around your application message type.
///
/// Use with [init](#init), [update](#update), and [view](#view) when wiring
/// up off_topic manually instead of through [application](#application).
pub opaque type Message(message) {
  ClientRuntimeMounted
  GotAppMessage(message: message)
  SubscriptionStarted(tick: Int, path: List(Int), cleanup: fn() -> Nil)
  RemoteSubscriptionSentMessage(tick: Int, path: List(Int), message: Dynamic)
  SubscriptionSentDeferredMessage(
    tick: Int,
    path: List(Int),
    message: message,
    ts: Int,
  )
  DelayTimeoutElapsed(
    tick: Int,
    path: List(Int),
    delay_tick: Int,
    message: message,
  )
  AnimationFrameTriggered(tick: Int, path: List(Int))
  ComponentConnected(forward: Option(message))
  ComponentDisconnected(forward: Option(message))
}

type Running(message) {
  Running(
    tick: Int,
    config: Config,
    cleanup: fn() -> Nil,
    last_fired: Int,
    delay_tick: Int,
  )
  Animating(tick: Int, callback: fn() -> message)
  Listening(
    tick: Int,
    config: Config,
    name: String,
    params: List(Json),
    decoder: Decoder(message),
  )
  Group(group: List(Running(message)))
}

// -- MAKING SUBSCRIPTIONS ----------------------------------------------------

/// A subscription that never fires.
///
/// See also: [batch](#batch)
pub fn none() -> Subscription(message) {
  Batch([])
}

/// Build a subscription from an arbitrary setup callback.
///
/// The callback receives a `dispatch` function and must return a cleanup
/// function that will be called when the subscription is torn down. off_topic
/// restarts the subscription whenever any value in `dependencies` changes.
///
/// For the exact rules on when and how subscriptions and their cleanup
/// functions fire, see [the guide](#TODO).
///
/// See also: [element](#element), [resource](#resource)
pub fn from(
  watching dependencies: List(Dependency),
  run callback: fn(fn(message) -> Nil) -> fn() -> Nil,
) -> Subscription(message) {
  Simple(start: callback, config: config(dependencies))
}

/// Build a subscription with access to the Lustre root element. <small>JS</small>
///
/// Like [from](#from), but the setup callback also receives the DOM element
/// that Lustre is mounted on, allowing you to attach listeners scoped to that
/// element rather than the window or document.
///
/// See also: [from](#from), [resource](#resource)
pub fn element(
  watching dependencies: List(Dependency),
  run callback: fn(fn(message) -> Nil, Dynamic) -> fn() -> Nil,
) -> Subscription(message) {
  Element(start: callback, config: config(dependencies))
}

pub fn animation(run callback: fn() -> message) -> Subscription(message) {
  RequestAnimationFrame(callback)
}

/// Combine a list of subscriptions into one.
///
/// See also: [none](#none)
pub fn batch(list: List(Subscription(message))) -> Subscription(message) {
  Batch(list)
}

/// Wrap a value as a subscription dependency.
///
/// off_topic restarts a subscription whenever any of its dependency values
/// change. When making your own subscriptions, keep in mind that you should
/// typically **not** pass the message callback as a dependency.
///
/// See also: [from](#from), [watching](#watching)
@external(erlang, "gleam_stdlib", "identity")
@external(javascript, "../../../gleam_stdlib/gleam/function.mjs", "identity")
pub fn dep(value: any) -> Dependency

/// Transform the messages a subscription dispatches.
pub fn map(
  subscription: Subscription(a),
  with tagger: fn(a) -> b,
) -> Subscription(b) {
  case subscription {
    RequestAnimationFrame(callback:) ->
      RequestAnimationFrame(callback: fn() { tagger(callback()) })

    Simple(start:, ..) ->
      Simple(..subscription, start: fn(dispatch) {
        start(fn(message) { dispatch(tagger(message)) })
      })

    Element(start:, ..) ->
      Element(..subscription, start: fn(dispatch, root) {
        start(fn(message) { dispatch(tagger(message)) }, root)
      })

    Remote(decoder:, ..) ->
      Remote(..subscription, decoder: decode.map(decoder, tagger))

    Batch(subscriptions) -> Batch(list.map(subscriptions, map(_, tagger)))
  }
}

/// Run a side effect whenever dependencies change.
///
/// The callback fires once at subscription setup and again whenever any
/// dependency changes.
///
/// See also: [from](#from), [resource](#resource)
pub fn watch(
  watching dependencies: List(Dependency),
  run callback: fn() -> Nil,
) -> Subscription(message) {
  use _ <- from(watching: dependencies)
  callback()
  fn() { Nil }
}

/// Build a subscription that runs on the server component client.
///
/// See also: [including](#including)
pub fn remote(
  name name: String,
  with params: List(Json),
  run decoder: Decoder(message),
) -> Subscription(message) {
  Remote(name:, params:, decoder:, config: config([]))
}

/// Dispatch a named browser command through the off_topic client runtime. <small>SC</small>
///
/// The server-component counterpart to `effect.from`: dispatches an
/// `off-topic:command` event that the `ot-client-runtime` web component
/// routes to the registered `window.OffTopic.commands[name]` handler on the
/// client.
///
/// For the built-in commands (`set_local_storage`, `scroll_to`, etc.) both
/// paths are handled automatically — use those directly. Reserve `command`
/// for custom handlers you register yourself.
///
/// See also: [remote](#remote)
pub fn command(name: String, params: List(Json)) -> Effect(message) {
  event.emit(
    "off-topic:command",
    json.object([
      #("name", json.string(name)),
      #("params", json.preprocessed_array(params)),
    ]),
  )
}

/// Specify which fields are forwarded from a remote subscription's events.
///
/// Remote subscriptions have to explicitely list all fields they want to
/// forward to the server. Has no effect on non-`Remote` subscriptions.
///
/// See also: [remote](#remote)
pub fn including(
  subscription: Subscription(message),
  paths: List(String),
) -> Subscription(message) {
  use config <- update_config(subscription)
  Config(..config, include: list.append(paths, config.include))
}

/// Limit how often a subscription dispatches messages.
///
/// When a subscription fires more frequently than `duration`, intermediate
/// messages are dropped. The first message always passes through immediately.
///
/// See also: [delay](#delay)
pub fn throttle(
  subscription: Subscription(message),
  wait ms: Int,
) -> Subscription(message) {
  use config <- update_config(subscription)
  Config(..config, throttle: int.max(0, ms))
}

/// Hold a subscription's message until the subscription has been quiet for `duration`.
///
/// When combined with [throttle](#throttle), the throttle pass-through fires
/// immediately and a delayed message is also queued for the final event in
/// each burst.
///
/// See also: [throttle](#throttle)
pub fn delay(
  subscription: Subscription(message),
  wait ms: Int,
) -> Subscription(message) {
  use config <- update_config(subscription)
  Config(..config, delay: int.max(0, ms))
}

/// Add extra dependencies to a subscription.
///
/// off_topic restarts the subscription whenever any dependency changes. Useful
/// when a subscription's behaviour depends on model values not captured in its
/// own parameters.
///
/// See also: [dep](#dep), [from](#from)
pub fn watching(
  subscription: Subscription(message),
  watching dependencies: List(Dependency),
) -> Subscription(message) {
  use config <- update_config(subscription)
  Config(..config, dependencies: list.append(dependencies, config.dependencies))
}

// -- FUNDAMENTAL EFFECT ------------------------------------------------------

type Timer

/// Dispatch a message after a delay.
pub fn after(ms: Int, send message: message) -> Effect(message) {
  use dispatch <- effect.from
  let _timer = do_send_after(ms, message, dispatch)
  Nil
}

@external(erlang, "off_topic_ffi", "send_after")
@external(javascript, "../../off_topic_ffi.mjs", "send_after")
fn do_send_after(
  interval: Int,
  message: message,
  dispatch: fn(message) -> Nil,
) -> Timer

@external(erlang, "off_topic_ffi", "now_ms")
@external(javascript, "../../off_topic_ffi.mjs", "now_ms")
fn now_ms() -> Int

// -- STARTING APPLICATIONS ---------------------------------------------------

/// Create a Lustre application with subscription support.
///
/// A drop-in replacement for `lustre.application` that adds a `subscriptions`
/// callback. The returned `App` is passed directly to `lustre.start` or
/// `lustre.start_server_component`.
///
/// See also: [component](#component), [init](#init), [update](#update), [view](#view)
pub fn application(
  init app_init: fn(flags) -> #(model, Effect(message)),
  update app_update: fn(model, message) -> #(model, Effect(message)),
  subscriptions app_subscriptions: fn(model) -> Subscription(message),
  view app_view: fn(model) -> Element(message),
) -> App(flags, Model(model, message), Message(message)) {
  let #(init, update, view) =
    wrap_runtime(app_init, app_update, app_subscriptions, app_view)
  lustre.application(init:, update:, view:)
}

/// Create a Lustre [component](https://hexdocs.pm/lustre/lustre/component.html)
/// that runs the off_topic subscription runtime. This is the component
/// equivalent of [application](#application).
///
/// If you're using component options, swap your import to `off_topic/component`,
/// offering the same API.
///
/// See also: [application](#application)
pub fn component(
  init app_init: fn(Nil) -> #(model, Effect(message)),
  update app_update: fn(model, message) -> #(model, Effect(message)),
  subscriptions app_subscriptions: fn(model) -> Subscription(message),
  view app_view: fn(model) -> Element(message),
  options options: List(component.Option(message)),
) -> App(Nil, Model(model, message), Message(message)) {
  let #(init, update, view) =
    wrap_runtime(app_init, app_update, app_subscriptions, app_view)

  let #(user_on_connect, user_on_disconnect) =
    component.extract_lifecycle(options)

  // User options lifted to Message(message) via ToApp, followed by our runtime
  // lifecycle options. Lustre's configure is last-wins, so appending last means
  // our on_connect / on_disconnect always take effect.
  let options =
    list.map(options, component.map(_, GotAppMessage))
    |> list.append([
      component.on_connect(ComponentConnected(user_on_connect)),
      component.on_disconnect(ComponentDisconnected(user_on_disconnect)),
    ])
    |> component.to_lustre_options

  lustre.component(init:, update:, view:, options:)
}

fn wrap_runtime(app_init, app_update, app_subscriptions, app_view) {
  let init = init(_, app_init)
  let view = view(_, app_view)
  let update = fn(model, msg) {
    update(model, msg, app_subscriptions, app_update)
  }
  #(init, update, view)
}

// -- APPLICATION BUILDING BLOCKS ---------------------------------------------

/// Run the application init function and start subscriptions.
///
/// The low-level counterpart to [application](#application), for use when you
/// need to compose off_topic into an existing `lustre.application` manually.
///
/// See also: [update](#update), [view](#view), [application](#application)
pub fn init(
  flags flags: flags,
  init init: fn(flags) -> #(model, Effect(message)),
) -> #(Model(model, message), Effect(Message(message))) {
  let #(model, app_effect) = init(flags)

  // Figure out if we will receive connect/disconnect messages.
  //
  // We will always receive a ClientRuntimeMounted mesasge:
  // - Server components don't fire before_paint, but <ot-client-runtime> will emit an event
  // - For applications and components, before_paint always fires synchronously,
  //   but AFTER a potential connected message, if we do receive those.
  //
  // Waiting until the Mounted message fired, we can then check if we just received a
  // connected message prior and if we did, we are in component mode!
  let effect =
    effect.batch([
      effect.map(app_effect, GotAppMessage),
      effect.before_paint(fn(dispatch, _root) { dispatch(ClientRuntimeMounted) }),
    ])

  let model =
    Model(
      tick: 0,
      app: model,
      mounted: False,
      state: Group([]),
      connected_clients: 0,
    )

  #(model, effect)
}

/// Handle a message in the subscription runtime.
///
/// The low-level counterpart to [application](#application), for use when you
/// need to compose off_topic into an existing `lustre.application` manually.
///
/// See also: [init](#init), [view](#view), [application](#application)
pub fn update(
  model: Model(model, message),
  message: Message(message),
  subscriptions: fn(model) -> Subscription(message),
  update: fn(model, message) -> #(model, Effect(message)),
) -> #(Model(model, message), Effect(Message(message))) {
  case message {
    GotAppMessage(message:) ->
      to_app(model, model.state, message, subscriptions, update)

    // client components receive mounted exactly once on startup,
    // server components can repeatedly receive mounted whenever a client joins (and have to ignore connected instead)
    ClientRuntimeMounted -> {
      case model.mounted && model.connected_clients > 0 {
        // Server component — a new client joined while subscriptions are live.
        // Re-emit remote subscriptions so the new client receives them.
        True -> {
          #(model, effect.batch(re_emit(model.state, [], [])))
        }

        // All other cases:
        //
        // (False, 0) Plain application — no component lifecycle, subscriptions run forever.
        //
        // (False, _) Custom element or server component — clients connected before mount,
        // so subscriptions weren't started yet. Start them now.
        //
        // (True, 0) Server component re-mounted after all clients disconnected.
        // Subscriptions were torn down on last disconnect; restart them.
        False -> {
          let connected_clients = int.max(1, model.connected_clients)
          step_lifecycle(model, True, connected_clients, subscriptions)
        }
      }
    }

    ComponentConnected(msg) -> {
      // Client component going from 0→1 while already mounted — restart subscriptions.
      // Apply forward first so the initial subscription set reflects the post-connect model.
      //
      // Otherwise subscriptions are running, will start on mount, or this is a server component.
      let connected_clients = model.connected_clients + 1
      // When we hit this in a browser while not mounted , we know for sure that
      // we are in a custom element, so we can mount early as a slight optimisation.
      // The mounted dispatch afterwards will re_emit, which we ignore.
      let mounted = model.mounted || lustre.is_browser()
      forward(msg, model, mounted, connected_clients, subscriptions, update)
    }

    ComponentDisconnected(msg) -> {
      let mounted = model.mounted
      let connected_clients = int.max(0, model.connected_clients - 1)
      forward(msg, model, mounted, connected_clients, subscriptions, update)
    }

    SubscriptionSentDeferredMessage(tick:, path:, message:, ts:) -> {
      case deferred(ts, tick, path, model.state, message) {
        Ok(#(state, Immediate(message:))) ->
          to_app(model, state, message, subscriptions, update)
        Ok(#(state, Delayed(message:, delay:, delay_tick:))) -> {
          let delay_message =
            DelayTimeoutElapsed(tick:, path:, delay_tick:, message:)
          #(Model(..model, state:), after(delay, delay_message))
        }
        Error(Nil) -> #(model, effect.none())
      }
    }

    SubscriptionStarted(tick:, path:, cleanup:) ->
      case started(tick, path, model.state, cleanup) {
        Ok(state) -> #(Model(..model, state:), effect.none())
        Error(_) -> #(model, effect.from(fn(_) { cleanup() }))
      }

    RemoteSubscriptionSentMessage(tick:, path:, message:) -> {
      use state <- emit_message(model, path, subscriptions, update)
      case state {
        Listening(decoder:, ..) if state.tick == tick ->
          decode.run(message, decoder) |> result.replace_error(Nil)
        _ -> Error(Nil)
      }
    }

    DelayTimeoutElapsed(tick:, path:, delay_tick:, message:) -> {
      use state <- emit_message(model, path, subscriptions, update)
      case state {
        Running(..) if state.tick == tick && state.delay_tick == delay_tick ->
          Ok(message)
        _ -> Error(Nil)
      }
    }

    AnimationFrameTriggered(tick:, path:) -> {
      let effect = case get(model.state, path) {
        Ok(Animating(tick: state_tick, callback:)) if state_tick == tick ->
          effect.batch([
            effect.from(fn(dispatch) { dispatch(GotAppMessage(callback())) }),
            effect.after_paint(fn(dispatch, _root) { dispatch(message) }),
          ])
        _ -> effect.none()
      }

      #(model, effect)
    }
  }
}

/// Render the application view through the subscription runtime.
///
/// The low-level counterpart to [application](#application), for use when you
/// need to compose off_topic into an existing `lustre.application` manually.
/// On the server target, this also injects the `ot-client-runtime` web
/// component into the rendered output.
///
/// See also: [init](#init), [update](#update), [application](#application)
pub fn view(
  model: Model(model, message),
  view: fn(model) -> Element(message),
) -> Element(Message(message)) {
  let runtime = case lustre.is_browser() {
    True -> element.none()
    False -> {
      let detail_decoder = {
        use tick <- decode.field("tick", decode.int)
        use path <- decode.field("path", decode.list(decode.int))
        use message <- decode.field("message", decode.dynamic)
        decode.success(RemoteSubscriptionSentMessage(tick:, path:, message:))
      }

      let on_dispatch =
        event.on("off-topic:dispatch", decode.at(["detail"], detail_decoder))
        |> server_component.include(["detail"])

      let on_mount =
        event.on("off-topic:mount", decode.success(ClientRuntimeMounted))

      element.element("ot-client-runtime", [on_dispatch, on_mount], [])
    }
  }

  let app = {
    // we increment the `tick` whenever we update the subscriptions, which
    // happens whenever the user model got updated.
    // This means we can skip "internal" updates in our runtime, which never
    // influence the view, allowing us to not re-render throttled events etc.
    element.memo([element.ref([model.tick])], fn() { view(model.app) })
  }

  element.fragment([runtime, element.map(app, GotAppMessage)])
}

// -- SUB DISPATCH ------------------------------------------------------------

fn to_app(model: Model(_, _), state, message, subscriptions, update) {
  let clients = model.connected_clients
  let #(app, app_effect) = update(model.app, message)
  let #(model, sync_effect) =
    step(model.tick, app, model.mounted, state, clients, subscriptions)
  #(model, effect(sync_effect, app_effect))
}

fn effect(sync_effect, app_effect) -> Effect(Message(message)) {
  effect.batch([sync_effect, effect.map(app_effect, GotAppMessage)])
}

fn step_lifecycle(model: Model(_, _), mounted, clients, subscriptions) {
  step(model.tick, model.app, mounted, model.state, clients, subscriptions)
}

fn forward(msg, model: Model(_, _), mounted, clients, subscriptions, update) {
  case msg {
    Some(msg) -> {
      let #(app, app_effect) = update(model.app, msg)
      let #(model, sync_effect) =
        step(model.tick, app, mounted, model.state, clients, subscriptions)
      #(model, effect(sync_effect, app_effect))
    }
    None -> step_lifecycle(model, mounted, clients, subscriptions)
  }
}

fn emit_message(model: Model(_, _), path, subscriptions, update, f) {
  let state = model.state
  case result.try(get(state, path), f) {
    Ok(msg) -> to_app(model, state, msg, subscriptions, update)
    Error(_) -> #(model, effect.none())
  }
}

fn get(state, path) {
  case path, state {
    [], _ -> Ok(state)
    [index, ..rest], Group(children) ->
      result.try(at(children, index), get(_, rest))
    _, _ -> Error(Nil)
  }
}

fn at(list, index) {
  case list {
    [] -> Error(Nil)
    [first, ..] if index <= 0 -> Ok(first)
    [_, ..rest] -> at(rest, index - 1)
  }
}

fn step(tick, app, mounted, state, connected_clients, subscriptions) {
  // let model = Model(..model, mounted:, connected_clients:)
  // step(model, model.app, subscriptions)

  // if we are not mounted or have no connected clients, we treat everything
  // as if there are no subscriptions -
  // this will automatically re-start or stop the current state accordingly
  // as clients connect and disconnect.
  let subscriptions = case mounted && connected_clients > 0 {
    True -> subscriptions(app)
    False -> none()
  }

  let tick = next_tick(tick)
  let #(state, effects) = diff(tick, [], state, subscriptions, [])
  let model = Model(tick:, state:, app:, mounted:, connected_clients:)
  #(model, effect.batch(effects))
}

fn next_tick(tick) {
  case tick < 0xffffffff {
    True -> tick + 1
    False -> 0
  }
}

fn started(tick, path, state, cleanup) {
  case path, state {
    [], Running(..) if state.tick == tick -> Ok(Running(..state, cleanup:))
    [index, ..rest], Group(children) -> {
      use #(prefix, child, suffix) <- result.try(split(children, index, []))
      use child <- result.try(started(tick, rest, child, cleanup))
      Ok(Group(unsplit(prefix, child, suffix)))
    }
    _, _ -> Error(Nil)
  }
}

fn split(list, index, prefix) {
  case list {
    [] -> Error(Nil)
    [first, ..rest] if index > 0 -> split(rest, index - 1, [first, ..prefix])
    [first, ..rest] -> Ok(#(prefix, first, rest))
  }
}

fn unsplit(prefix, child, rest) {
  list.fold(from: [child, ..rest], over: prefix, with: list.prepend)
}

type Dispatch(message) {
  Immediate(message: message)
  Delayed(message: message, delay: Int, delay_tick: Int)
}

fn deferred(now, tick, path, state, message) {
  case path, state {
    [], Running(config:, last_fired:, delay_tick:, ..) if state.tick == tick -> {
      let Config(throttle:, delay:, ..) = config

      let delay_tick = next_tick(delay_tick)
      case throttle > 0 && now - last_fired >= throttle, delay > 0 {
        True, _ -> {
          let state = Running(..state, last_fired: now, delay_tick:)
          Ok(#(state, Immediate(message)))
        }
        False, True -> {
          let state = Running(..state, delay_tick:)
          Ok(#(state, Delayed(message, delay, delay_tick)))
        }
        False, False -> Error(Nil)
      }
    }

    [index, ..rest], Group(children) -> {
      use #(prefix, child, suffix) <- result.try(split(children, index, []))
      use #(child, result) <- result.try({
        deferred(now, tick, rest, child, message)
      })
      Ok(#(Group(unsplit(prefix, child, suffix)), result))
    }
    _, _ -> Error(Nil)
  }
}

// -- DIFF --------------------------------------------------------------------

fn diff(tick, path, left, right, effects) {
  case left, right {
    Running(config: c1, ..), Simple(config: c2, ..)
    | Running(config: c1, ..), Element(config: c2, ..)
      if c1.dependencies == c2.dependencies
    -> #(Running(..left, config: c2), effects)

    Listening(name: n1, params: p1, config: c1, ..),
      Remote(name: n2, params: p2, config: c2, ..)
      if n1 == n2 && p1 == p2 && c1.dependencies == c2.dependencies
    -> #(left, effects)

    Animating(tick:, ..), RequestAnimationFrame(callback:) -> {
      #(Animating(tick:, callback:), effects)
    }

    Group(olds), Batch(news) ->
      diff_batch(tick, 0, path, olds, news, [], effects)

    _, _ -> start(tick, path, right, cleanup(left, path, effects))
  }
}

fn diff_batch(tick, index, path, left, right, out, effects) {
  case left, right {
    [old, ..left], [new, ..right] -> {
      let #(child, effects) = diff(tick, [index, ..path], old, new, effects)
      diff_batch(tick, index + 1, path, left, right, [child, ..out], effects)
    }

    _, _ -> diff_rest(tick, index, path, left, right, out, effects)
  }
}

fn diff_rest(tick, index, path, left, right, out, effects) {
  let effects = fold_children(left, index, path, effects, cleanup)
  let #(children, effects) =
    start_children(tick, index, path, right, out, effects)
  #(Group(children), effects)
}

fn cleanup(state, path, effects) {
  case state {
    Running(cleanup:, ..) -> {
      [effect.from(fn(_) { cleanup() }), ..effects]
    }
    Animating(..) -> effects
    Listening(tick:, name:, ..) -> {
      let details =
        json.object([
          #("tick", json.int(tick)),
          #("name", json.string(name)),
          #("path", json.array(list.reverse(path), json.int)),
        ])
      [event.emit("off-topic:unsubscribe", details), ..effects]
    }
    Group(children) -> fold_children(children, 0, path, effects, cleanup)
  }
}

fn re_emit(state, path, effects) {
  case state {
    Running(..) | Animating(..) -> effects
    Listening(tick:, name:, params:, config:, ..) -> {
      [subscribe_effect(tick, path, name, params, config), ..effects]
    }
    Group(children) -> fold_children(children, 0, path, effects, re_emit)
  }
}

fn fold_children(children, index, path, effects, f) {
  case children {
    [] -> effects
    [child, ..rest] -> {
      let effects = f(child, [index, ..path], effects)
      fold_children(rest, index + 1, path, effects, f)
    }
  }
}

fn start(tick, path, subscription, effects) {
  case subscription {
    RequestAnimationFrame(callback:) -> {
      let effect =
        effect.after_paint(fn(dispatch, _root) {
          dispatch(AnimationFrameTriggered(tick, list.reverse(path)))
        })
      #(Animating(tick:, callback:), [effect, ..effects])
    }
    Simple(start:, config:) -> {
      let effect =
        effect.from(fn(dispatch) {
          let path = list.reverse(path)
          let cleanup = start(wrap_dispatch(dispatch, config, tick, path))
          dispatch(SubscriptionStarted(tick:, path:, cleanup:))
        })
      #(running(tick, config), [effect, ..effects])
    }
    Element(start:, config:) -> {
      let effect =
        effect.before_paint(fn(dispatch, root) {
          let path = list.reverse(path)
          let cleanup = start(wrap_dispatch(dispatch, config, tick, path), root)
          dispatch(SubscriptionStarted(tick:, path:, cleanup:))
        })
      #(running(tick, config), [effect, ..effects])
    }
    Remote(name:, params:, decoder:, config:) -> {
      let effect = subscribe_effect(tick, path, name, params, config)
      let state = Listening(tick:, name:, params:, decoder:, config:)
      #(state, [effect, ..effects])
    }
    Batch(subscriptions:) ->
      diff_batch(tick, 0, path, [], subscriptions, [], effects)
  }
}

fn start_children(tick, index, path, right, out, effects) {
  case right {
    [] -> #(list.reverse(out), effects)
    [new, ..right] -> {
      let #(child, effects) = start(tick, [index, ..path], new, effects)
      start_children(tick, index + 1, path, right, [child, ..out], effects)
    }
  }
}

fn running(tick: Int, config: Config) -> Running(message) {
  Running(tick:, config:, cleanup: fn() { Nil }, last_fired: 0, delay_tick: 0)
}

fn wrap_dispatch(dispatch, config: Config, tick, path) {
  case config.throttle == 0 && config.delay == 0 {
    True -> fn(message) { dispatch(GotAppMessage(message)) }
    False -> fn(message) {
      let ts = now_ms()
      dispatch(SubscriptionSentDeferredMessage(tick:, path:, message:, ts:))
    }
  }
}

fn subscribe_effect(tick, path, name, params, config: Config) {
  let details =
    json.object([
      #("tick", json.int(tick)),
      #("name", json.string(name)),
      #("path", json.array(list.reverse(path), json.int)),
      #("params", json.preprocessed_array(params)),
      #("include", json.array(config.include, json.string)),
      #("throttle", json.int(config.throttle)),
      #("delay", json.int(config.delay)),
    ])
  event.emit("off-topic:subscribe", details)
}