///|
/// Selects which Gateway shards this process owns.
pub(all) enum ShardConfig {
  /// Use Discord's recommended shard count and run every shard locally.
  Auto
  /// Run one shard from a larger logical shard set. This is the default and
  /// preserves the pre-manager single-shard executor behavior.
  Single(id~ : Int, count~ : Int)
  /// Run shard ids `0.. Unit raise BotError {
  if count <= 0 {
    raise InvalidShardConfig(message="shard count must be positive")
  }
  if ids.is_empty() {
    raise InvalidShardConfig(message="at least one shard id must be selected")
  }
  let seen : Map[Int, Bool] = Map([])
  for id in ids {
    if id < 0 || id >= count {
      raise InvalidShardConfig(message="shard id \{id} is outside 0..<\{count}")
    }
    if seen.get(id) == Some(true) {
      raise InvalidShardConfig(message="duplicate shard id \{id}")
    }
    seen[id] = true
  }
}

///|
async fn Bot::gateway_bot_info(
  self : Bot,
  client : @dhttp.Client,
) -> @model.GatewayBotInfo raise @dhttp.DiscordHttpError {
  match self.gateway_bot_info_.val {
    Some(info) => info
    None => client.get_gateway_bot()
  }
}

///|
fn[X] Bot::start_managed_shard(
  self : Bot,
  group : @async.TaskGroup[X],
  token~ : String,
  intents~ : @model.Intents,
  gateway_url~ : String,
  shard_id~ : Int,
  shard_count~ : Int,
  identify_queue~ : &@queue.IdentifyQueue,
  event_filter~ : (@model.EventKind) -> Bool,
  resume? : @gateway.Session,
) -> @gateway.Shard {
  match self.connector_.val {
    Some(connector) =>
      @gateway.Shard::start(
        group,
        token~,
        intents~,
        capabilities=self.capabilities_,
        gateway_url~,
        shard_id~,
        shard_count~,
        identify_queue~,
        event_filter~,
        connector~,
        compress=self.compress_,
        telemetry=event => self.emit_telemetry(event),
        resume?,
      )
    None =>
      @gateway.Shard::start(
        group,
        token~,
        intents~,
        capabilities=self.capabilities_,
        gateway_url~,
        shard_id~,
        shard_count~,
        identify_queue~,
        event_filter~,
        compress=self.compress_,
        telemetry=event => self.emit_telemetry(event),
        resume?,
      )
  }
}

///|
async fn Bot::resolve_shards(
  self : Bot,
  client : @dhttp.Client,
) -> ResolvedShardConfig {
  match self.shards_ {
    Single(id~, count~) => {
      let ids = [id]
      validate_shard_ids(ids, count)
      let gateway_url = match self.gateway_url_ {
        Some(url) => url
        None => client.get_gateway()
      }
      {
        ids,
        count,
        gateway_url,
        max_concurrency: 1,
        resume: self.sessions_for_layout(ids, count),
      }
    }
    Auto | Fixed(_) | Range(..) => {
      let info = self.gateway_bot_info(client)
      let (ids, count) = match self.shards_ {
        Auto => {
          let ids = []
          for id in 0.. {
          let ids = []
          for id in 0.. (ids.copy(), count)
        Single(..) => abort("handled above")
      }
      validate_shard_ids(ids, count)
      if info.session_start_limit.max_concurrency <= 0 {
        raise BotError::InvalidShardConfig(
          message="gateway max_concurrency must be positive",
        )
      }
      let resume = self.sessions_for_layout(ids, count)
      // A shard with a saved session sends RESUME, which does not consume the
      // identify allowance; only a rejected RESUME falls back to IDENTIFY.
      let required = ids.filter(id => !resume.contains(id)).length()
      if required > info.session_start_limit.remaining {
        raise BotError::SessionStartLimitExceeded(
          required~,
          remaining=info.session_start_limit.remaining,
          reset_after_ms=info.session_start_limit.reset_after,
        )
      }
      {
        ids,
        count,
        gateway_url: self.gateway_url_.unwrap_or(info.url),
        max_concurrency: info.session_start_limit.max_concurrency,
        resume,
      }
    }
  }
}

///|
/// The saved sessions usable for the selected shards. A RESUME keeps the
/// guild assignment of the READY it was recorded under, so a snapshot from
/// another shard layout is dropped with a warning and its shard identifies.
fn Bot::sessions_for_layout(
  self : Bot,
  ids : Array[Int],
  count : Int,
) -> Map[Int, BotSession] {
  let usable : Map[Int, BotSession] = Map([])
  for id in ids {
    guard self.resume_.get(id) is Some(saved) else { continue }
    let recorded = saved.ready.shard.unwrap_or([0, 1])
    if recorded is [recorded_id, recorded_count] &&
      recorded_id == id &&
      recorded_count == count {
      usable[id] = saved
    } else {
      self.app_.warn(
        "ignoring saved session for shard \{id}: its READY was recorded for shard [\{recorded.map(part => part.to_string()).join(", ")}] but the bot resolved shard [\{id}, \{count}], so the shard identifies instead",
      )
    }
  }
  usable
}

///|
async fn close_shards(shards : Array[@gateway.Shard]) -> Unit noraise {
  for shard in shards {
    shard.close()
  }
}