diff --git a/gleam.toml b/gleam.toml index 03f0496..60ea062 100644 --- a/gleam.toml +++ b/gleam.toml @@ -13,12 +13,16 @@ version = "1.0.0" # https://gleam.run/writing-gleam/gleam-toml/. [dependencies] -gleam_stdlib = ">= 1.0.0 and < 2.0.0" collie = ">= 3.0.0 and < 4.0.0" gleam_erlang = ">= 1.3.0 and < 2.0.0" -gleam_otp = ">= 1.3.0 and < 2.0.0" -grom = ">= 7.0.0-beta.1.1 and < 7.0.0" gleam_http = ">= 4.4.0 and < 5.0.0" +gleam_json = ">= 3.1.0 and < 4.0.0" +gleam_otp = ">= 1.3.0 and < 2.0.0" +gleam_stdlib = ">= 1.0.0 and < 2.0.0" +grom = { git = "https://tangled.org/folospior.dev/grom", ref = "one-module" } +group_registry = ">= 1.0.0 and < 2.0.0" +operating_system = ">= 1.0.1 and < 2.0.0" +gleam_time = ">= 1.11.0 and < 2.0.0" [dev_dependencies] gleeunit = ">= 1.0.0 and < 2.0.0" diff --git a/manifest.toml b/manifest.toml index fed5c28..0622024 100644 --- a/manifest.toml +++ b/manifest.toml @@ -13,15 +13,18 @@ packages = [ { name = "gleam_community_colour", version = "2.0.4", build_tools = ["gleam"], requirements = ["gleam_json", "gleam_stdlib"], otp_app = "gleam_community_colour", source = "hex", outer_checksum = "6DB4665555D7D2B27F0EA32EF47E8BEBC4303821765F9C73D483F38EE24894F0" }, { name = "gleam_crypto", version = "1.6.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_crypto", source = "hex", outer_checksum = "2DE9E4EF53CF6FEE049D4F765731F7178F7A11AEFAE00EEE63BF7536B354AD3F" }, { name = "gleam_erlang", version = "1.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_erlang", source = "hex", outer_checksum = "1124AD3AA21143E5AF0FC5CF3D9529F6DB8CA03E43A55711B60B6B7B3874375C" }, - { name = "gleam_http", version = "4.4.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_http", source = "hex", outer_checksum = "F065CE9941A7756ABBAEAE7F9F32A96CAD29DA3311A381DB2D926EA8E8C156DC" }, + { name = "gleam_http", version = "4.5.0", build_tools = ["gleam"], requirements = ["gleam_stdlib", "splitter"], otp_app = "gleam_http", source = "hex", outer_checksum = "38D7E5698759D44F21436F365259F7015104F6FAFA23181CD0B2BB0E5B67BE2B" }, { name = "gleam_json", version = "3.1.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_json", source = "hex", outer_checksum = "44FDAA8847BE8FC48CA7A1C089706BD54BADCC4C45B237A992EDDF9F2CDB2836" }, { name = "gleam_otp", version = "1.3.0", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_stdlib"], otp_app = "gleam_otp", source = "hex", outer_checksum = "DE4CA6850842F0266EE95317A25DD6A0A0F20CDFAB7C0ADC2E63251D7C3C72EC" }, { name = "gleam_stdlib", version = "1.0.5", build_tools = ["gleam"], requirements = [], otp_app = "gleam_stdlib", source = "hex", outer_checksum = "CEE5B6C076A85B45F60C585F4316C63EC8B7127C119D5738C3958A9C4D50404E" }, { name = "gleam_time", version = "1.11.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_time", source = "hex", outer_checksum = "3AB789C3ED4EB9DDA7283F46C3CF83A5487C9CFB790DC3FF8F8ACB5BF7B7A6C1" }, { name = "gleeunit", version = "1.11.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleeunit", source = "hex", outer_checksum = "EC31ABA74256AEA531EDF8169931D775BBB384FED0A8A1BDC4DD9354E3E21826" }, - { name = "grom", version = "7.0.0-beta.2.0", build_tools = ["gleam"], requirements = ["datebook", "gleam_community_colour", "gleam_http", "gleam_json", "gleam_stdlib", "gleam_time", "multipart_form", "status_code"], otp_app = "grom", source = "hex", outer_checksum = "0E700A6CC4EC74CFEF82A88923D20EFC7B184CF84095D1F58FF7BE80091107C8" }, + { name = "grom", version = "7.0.0-beta.2.0", build_tools = ["gleam"], requirements = ["datebook", "gleam_community_colour", "gleam_http", "gleam_json", "gleam_stdlib", "gleam_time", "multipart_form", "status_code"], source = "git", repo = "https://tangled.org/folospior.dev/grom", commit = "78caacaaaad6b2da052a0cf2375c80eb3dc4a143" }, + { name = "group_registry", version = "1.0.0", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_otp", "gleam_stdlib"], otp_app = "group_registry", source = "hex", outer_checksum = "BC798A53D6F2406DB94E27CB45C57052CB56B32ACF7CC16EA20F6BAEC7E36B90" }, { name = "logging", version = "1.5.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "logging", source = "hex", outer_checksum = "BC5F18CE5DD9686100229FE5409BDC3DD5C46D5A7DF2F804AD2D8F0DD6C5060E" }, { name = "multipart_form", version = "1.1.0", build_tools = ["gleam"], requirements = ["gleam_http", "gleam_stdlib"], otp_app = "multipart_form", source = "hex", outer_checksum = "082C77A0C3BB1128FCD55491665E9B72BC943E849B67D02B08CFA6808AD8E47C" }, + { name = "operating_system", version = "1.0.1", build_tools = ["gleam"], requirements = [], otp_app = "operating_system", source = "hex", outer_checksum = "682D4D19496E607A6E6229AA51A38A1400ED5067C2C54D56843DBCB0C61FFA6D" }, + { name = "splitter", version = "1.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "splitter", source = "hex", outer_checksum = "8F1485306E0D59168A39E12ED8844F9F616EC59C3C63AFCFDAE04DBD68690CBC" }, { name = "status_code", version = "1.1.0", build_tools = ["gleam"], requirements = [], otp_app = "status_code", source = "hex", outer_checksum = "F6EC765345A634602AD030EB1B4D34A425873CBC223CEC595EBF062073CD70BC" }, { name = "websocks", version = "4.0.1", build_tools = ["gleam"], requirements = ["gleam_crypto", "gleam_erlang", "gleam_stdlib"], otp_app = "websocks", source = "hex", outer_checksum = "89B0C31A032CBE28D4C5FB5CC25A8A690669FEBA501C63F99A4E8F5748750B2A" }, ] @@ -30,7 +33,11 @@ packages = [ collie = { version = ">= 3.0.0 and < 4.0.0" } gleam_erlang = { version = ">= 1.3.0 and < 2.0.0" } gleam_http = { version = ">= 4.4.0 and < 5.0.0" } +gleam_json = { version = ">= 3.1.0 and < 4.0.0" } gleam_otp = { version = ">= 1.3.0 and < 2.0.0" } gleam_stdlib = { version = ">= 1.0.0 and < 2.0.0" } +gleam_time = { version = ">= 1.11.0 and < 2.0.0" } gleeunit = { version = ">= 1.0.0 and < 2.0.0" } -grom = { version = ">= 7.0.0-beta.1.1 and < 7.0.0" } +grom = { git = "https://tangled.org/folospior.dev/grom", ref = "one-module" } +group_registry = { version = ">= 1.0.0 and < 2.0.0" } +operating_system = { version = ">= 1.0.1 and < 2.0.0" } diff --git a/src/agat.gleam b/src/agat.gleam index 7d2f869..de5217e 100644 --- a/src/agat.gleam +++ b/src/agat.gleam @@ -1,21 +1,28 @@ import collie import gleam/dict.{type Dict} +import gleam/dynamic/decode.{type Decoder} import gleam/erlang/process.{type Subject} +import gleam/float import gleam/http/request.{type Request} import gleam/int +import gleam/json.{type Json} import gleam/list +import gleam/option.{type Option, None, Some} import gleam/otp/actor import gleam/otp/static_supervisor import gleam/otp/supervision +import gleam/result import gleam/string import grom +import group_registry pub opaque type Builder(user_state) { Builder( shards: ShardSettings, - connection_info: grom.GatewayConnectionInfo, - compress: Bool, - name: process.Name(CoordinatorMessage), + connection_info: GatewayConnectionInfo, + identification: Identification, + coordinator_name: process.Name(CoordinatorMessage), + registry_name: process.Name(group_registry.Message(ShardMessage)), user_state: user_state, on_close: fn(user_state) -> Nil, on_event: fn(user_state, Event) -> Next(user_state), @@ -27,9 +34,36 @@ type ShardSettings { SpecificShards(ids: List(Int), total_shard_count: Int) } +fn get_total_shard_count(of shards: ShardSettings) { + case shards { + ShardRange(from:, to:) -> to - from + SpecificShards(total_shard_count:, ..) -> total_shard_count + } +} + +// a group that all shards are part of +const shards_group = "agat-shards" + +// a group that only shards of a specific ID are part of +// +// NOTE: be sure to remember that there can be multiple shards with the same ID +// (redundancy!!) +fn get_shard_group(shard_id id: Int) -> String { + "agat-shard-" <> int.to_string(id) +} + +/// Defaults to the recommended amount of shards. +/// +/// The `named` parameter refers to the name of the process to send messages to. +/// +/// The `registry_name` parameter refers to an internal registry of shards. pub fn new( named name: process.Name(CoordinatorMessage), - connection_info connection_info: grom.GatewayConnectionInfo, + registry_name registry_name: process.Name( + group_registry.Message(ShardMessage), + ), + connection_info connection_info: GatewayConnectionInfo, + identification identification: Identification, state user_state: user_state, ) -> Builder(user_state) { Builder( @@ -39,8 +73,9 @@ pub fn new( to: connection_info.recommended_shard_amount, ), connection_info:, - compress: False, - name:, + identification:, + coordinator_name: name, + registry_name:, user_state:, on_close: fn(_user_state) { Nil }, on_event: fn(user_state, _event) { Continue(user_state) }, @@ -55,6 +90,10 @@ pub fn with_shards( Builder(..builder, shards: ShardRange(from:, to:)) } +pub type UpdatePresenceMessage { + UpdatePresenceMessage +} + /// Allows you to specify the specific shards you want to start up on this instance. /// /// The shard count should include any shards you're running on other instances of your bot. @@ -66,8 +105,15 @@ pub fn with_shard_ids( Builder(..builder, shards: SpecificShards(shard_ids, total_shard_count)) } -pub fn with_compression(builder: Builder(_)) -> Builder(_) { - Builder(..builder, compress: True) +pub fn identify( + token token: String, + intents intents: List(grom.GatewayIntent), +) -> Identification { + Identification(token:, intents:) +} + +pub opaque type Identification { + Identification(token: String, intents: List(grom.GatewayIntent)) } pub fn on_close( @@ -80,10 +126,17 @@ pub fn on_close( pub opaque type CoordinatorMessage { MessageFromDiscord(Event) MessageFromUser - MessageFromShard(id: Int, state: ShardState) } -pub type Event +pub type Event { + ErrorEvent(Error) +} + +pub type Error { + CouldNotDecodeEvent(error: json.DecodeError) + EventNotValidUtf8(event_payload: BitArray) + CouldNotSendEvent(event_payload: String, reason: collie.SocketReason) +} pub opaque type Next(user_state) { Continue(user_state: user_state) @@ -107,63 +160,130 @@ pub fn supervised(builder: Builder(user_state)) { supervision.supervisor(fn() { start_supervisor(builder) }) } -type CoordinatorState(user_state) { - CoordinatorState(user_state: user_state, shards: Dict(Int, RegisteredShard)) +type ResumingInfo { + ResumingInfo( + session_id: String, + gateway_request: Request(String), + sequence: Option(Int), + ) } -type RegisteredShard { - RegisteredShard( - subject: Subject(ShardMessage), - /// Used for resuming if the shard dies - state: ShardState, +type CoordinatorState(user_state) { + CoordinatorState( + registry_name: process.Name(group_registry.Message(ShardMessage)), + user_state: user_state, + resuming_info: Dict(Int, ResumingInfo), + on_close: fn(user_state) -> Nil, + on_event: fn(user_state, Event) -> Next(user_state), ) } -type ShardMessage { - ShardMessageFromDiscord - ShardMessageFromUser - ShardStart +pub opaque type ShardMessage { + ShardIdentify + ShardScheduledHeartbeat + ShardRequestedHeartbeat + ShardShutdown(reason: collie.CloseReason, type_: ShardShutdownType) } fn supervised_coordinator(builder: Builder(user_state)) { supervision.worker(fn() { start_coordinator(builder) }) + |> supervision.restart(supervision.Transient) } fn start_coordinator( builder: Builder(user_state), ) -> Result(actor.Started(Subject(CoordinatorMessage)), actor.StartError) { - actor.new(CoordinatorState(user_state: builder.user_state, shards: dict.new())) - |> actor.on_message(on_event_listener_message) - |> actor.named(builder.name) + actor.new(CoordinatorState( + registry_name: builder.registry_name, + user_state: builder.user_state, + resuming_info: dict.new(), + on_close: builder.on_close, + on_event: builder.on_event, + )) + |> actor.on_message(on_coordinator_message) + |> actor.named(builder.coordinator_name) |> actor.start } -fn on_event_listener_message( +fn on_coordinator_message( state: CoordinatorState(user_state), message: CoordinatorMessage, ) { - todo + case message { + MessageFromDiscord(event) -> { + let next = state.on_event(state.user_state, event) + case next { + Continue(user_state:) -> + actor.continue(CoordinatorState(..state, user_state:)) + Stop -> { + shutdown_all_shards( + registry: state.registry_name, + reason: collie.CloseReason(code: collie.NormalClosure, reason: ""), + shutdown_type: RegularShardShutdown, + ) + actor.stop() + } + StopAbnormal(reason:) -> { + shutdown_all_shards( + registry: state.registry_name, + reason: collie.NoCloseReason, + shutdown_type: AbnormalShardShutdown(reason:), + ) + actor.stop_abnormal(reason) + } + } + } + MessageFromUser -> todo + } } -fn start_supervisor(builder: Builder(user_state)) { - let supervisor = static_supervisor.new(static_supervisor.OneForOne) +fn shutdown_all_shards( + registry registry_name: process.Name(group_registry.Message(ShardMessage)), + shutdown_type type_: ShardShutdownType, + reason reason: collie.CloseReason, +) -> Nil { + group_registry.get_registry(registry_name) + |> group_registry.members(shards_group) + |> list.each(fn(subject) { + subject + |> process.send(ShardShutdown(reason:, type_:)) + }) +} - let supervisor = case builder.shards { - SpecificShards(ids:, ..) -> - list.index_fold(ids, supervisor, fn(supervisor, id, idx) { - add_shard_to_supervisor(builder, supervisor, id, idx) - }) +fn start_supervisor( + builder: Builder(user_state), +) -> Result(actor.Started(static_supervisor.Supervisor), actor.StartError) { + static_supervisor.new(strategy: static_supervisor.OneForOne) + // if a coordinator dies, all the shards + |> static_supervisor.add(supervised_coordinator(builder)) + // if the registry dies... all the shards have to re-register + // the shards only register on start up + // so they gotta start again + // LET EM DIE + |> static_supervisor.add(group_registry.supervised(builder.registry_name)) + // if a shard dies, restart it + |> add_shards_to_supervisor(builder) + |> static_supervisor.start +} +fn add_shards_to_supervisor( + supervisor: static_supervisor.Builder, + builder: Builder(_), +) -> static_supervisor.Builder { + case builder.shards { ShardRange(from:, to:) -> int.range(from:, to:, with: supervisor, run: fn(supervisor, idx) { add_shard_to_supervisor(builder, supervisor, idx, idx) }) + SpecificShards(ids:, ..) -> + list.index_fold( + over: ids, + from: supervisor, + with: fn(supervisor, id, idx) { + add_shard_to_supervisor(builder, supervisor, id, idx) + }, + ) } - - supervisor - |> static_supervisor.add(supervised_coordinator(builder)) - // todo: add all the shards - |> static_supervisor.start } fn add_shard_to_supervisor( @@ -172,94 +292,177 @@ fn add_shard_to_supervisor( id id: Int, index idx: Int, ) -> static_supervisor.Builder { - let shard_builder = - ShardBuilder( - id:, - connection_info: builder.connection_info, - compress: builder.compress, - coordinator_name: builder.name, - ) + let shard_builder = builder_to_shard_builder(builder, id, idx) supervisor - |> static_supervisor.add(supervised_shard( - shard_builder, - place_in_shard_list: idx, - )) + |> static_supervisor.add(supervised_shard(shard_builder)) +} + +fn builder_to_shard_builder( + builder: Builder(_), + id: Int, + idx: Int, +) -> ShardBuilder { + ShardBuilder( + id:, + connection_info: builder.connection_info, + identification: builder.identification, + coordinator_name: builder.coordinator_name, + registry_name: builder.registry_name, + num_shards: builder.shards |> get_total_shard_count, + shard_list_index: idx, + ) } type ShardState { - ShardWaitingForInitialization + ShardWaitingForInitialization( + id: Int, + num_shards: Int, + connection_info: GatewayConnectionInfo, + identification: Identification, + coordinator_name: process.Name(CoordinatorMessage), + self_subject: process.Subject(collie.WebsocketMessage(ShardMessage)), + ) + ShardWelcomed( + id: Int, + num_shards: Int, + connection_info: GatewayConnectionInfo, + identification: Identification, + coordinator_name: process.Name(CoordinatorMessage), + self_subject: process.Subject(collie.WebsocketMessage(ShardMessage)), + resuming_info: Option(ResumingInfo), + heartbeat_interval_ms: Int, + heartbeats: Int, + heartbeat_acks: Int, + ) } type ShardBuilder { ShardBuilder( id: Int, num_shards: Int, - connection_info: grom.GatewayConnectionInfo, - compress: Bool, + connection_info: GatewayConnectionInfo, + identification: Identification, coordinator_name: process.Name(CoordinatorMessage), + registry_name: process.Name(group_registry.Message(ShardMessage)), /// Place in the list of shards to be started up shard_list_index: Int, ) } -fn build_connection_request( - url connection_url: String, -) -> Result(Request(String), Nil) { +/// Returns `Error(Nil)` on invalid connection URL. +pub fn build_connection_info( + based_on info: grom.GatewayConnectionInfo, +) -> Result(GatewayConnectionInfo, Nil) { let connection_url = - connection_url + info.connection_url |> string.replace(each: "wss://", with: "https://") - request.to(connection_url <> "?v=10&encoding=json") + use request <- result.try(request.to(connection_url <> "?v=10&encoding=json")) + + Ok(GatewayConnectionInfo( + connection_request: request, + recommended_shard_amount: info.recommended_shard_amount, + connection_start_limits: info.connection_start_limits, + )) +} + +pub opaque type GatewayConnectionInfo { + GatewayConnectionInfo( + connection_request: Request(String), + recommended_shard_amount: Int, + connection_start_limits: grom.GatewayConnectionStartLimits, + ) } fn supervised_shard( builder: ShardBuilder, - place_in_shard_list shard_list_idx: Int, -) -> supervision.ChildSpecification(Subject(ShardMessage)) { - supervision.worker(fn() { - builder - |> start_shard(place_in_shard_list: shard_list_idx) - }) +) -> supervision.ChildSpecification( + Subject(collie.WebsocketMessage(ShardMessage)), +) { + supervision.worker(fn() { builder |> start_shard }) + // if we use collie.stop(), we do NOT want to restart + // only if we use collie.stop_abnormal() + |> supervision.restart(supervision.Transient) +} + +fn shard_builder_to_initial_state( + builder: ShardBuilder, + self_subject: process.Subject(collie.WebsocketMessage(ShardMessage)), +) { + ShardWaitingForInitialization( + id: builder.id, + num_shards: builder.num_shards, + identification: builder.identification, + connection_info: builder.connection_info, + coordinator_name: builder.coordinator_name, + self_subject:, + ) } fn start_shard( builder: ShardBuilder, - connection_request request: Request(String), - place_in_shard_list shard_list_idx: Int, ) -> Result( actor.Started(Subject(collie.WebsocketMessage(ShardMessage))), actor.StartError, ) { - collie.new_with_initialiser(request, fn(self) { - let time_to_wait_milliseconds = - { - shard_list_idx - / builder.connection_info.connection_start_limits.max_identify_per_5_seconds - } - * 5000 - - let delay_subject = process.new_subject() - delay_shard_start(delay_subject, time_to_wait: time_to_wait_milliseconds) - - let selector = process.new_selector() |> process.select(delay_subject) - - collie.initialised(ShardWaitingForInitialization) - |> collie.returning(self) - |> collie.selecting(selector) - |> Ok - }) + collie.new_with_initialiser( + builder.connection_info.connection_request, + fn(self) { + let time_to_wait_milliseconds = + { + builder.shard_list_index + / builder.connection_info.connection_start_limits.max_identify_per_5_seconds + } + // we have to wait (batch * 5 seconds) + * 5000 + + let delay_subject = process.new_subject() + delay_shard_identify( + delay_subject, + time_to_wait: time_to_wait_milliseconds, + ) + + let registry = + builder.registry_name + |> group_registry.get_registry + + let all_shards_subject = + registry + |> group_registry.join(shards_group, process.self()) + + let shard_group_subject = + registry + |> group_registry.join( + get_shard_group(shard_id: builder.id), + process.self(), + ) + + let selector = + process.new_selector() + |> process.select(all_shards_subject) + |> process.select(shard_group_subject) + |> process.select(delay_subject) + + let state = shard_builder_to_initial_state(builder, self) + + collie.initialised(state) + |> collie.returning(self) + |> collie.selecting(selector) + |> Ok + }, + ) |> collie.on_message(on_shard_message) |> collie.on_close(on_shard_close) |> collie.start } -fn delay_shard_start( +fn delay_shard_identify( subject: Subject(ShardMessage), time_to_wait time_to_wait: Int, ) { subject - |> process.send_after(time_to_wait, ShardStart) + |> process.send_after(time_to_wait, ShardIdentify) } fn on_shard_message( @@ -267,9 +470,289 @@ fn on_shard_message( state: ShardState, message: collie.Message(ShardMessage), ) { + case message { + collie.User(ShardShutdown(reason, type_)) -> + handle_shard_shutdown(connection, reason, type_) + collie.User(ShardScheduledHeartbeat) -> + handle_scheduled_heartbeat(connection, state) + collie.User(ShardRequestedHeartbeat) -> + handle_requested_heartbeat(connection, state) + collie.User(ShardIdentify) -> handle_shard_identify(connection, state) + collie.Text(text) -> handle_text_message(state, text) + collie.Binary(binary) -> handle_binary_message(state, binary) + } +} + +type IdentifyMessage { + IdentifyMessage( + token: String, + intents: List(grom.GatewayIntent), + os: String, + shard_id: Int, + num_shards: Int, + ) +} + +fn identify_message_to_json(identify: IdentifyMessage) -> Json { + [] + |> json.object +} + +fn handle_shard_identify( + connection: collie.Connection, + state: ShardState, +) -> collie.Next(ShardState, a) { todo } +fn handle_binary_message( + state: ShardState, + binary: BitArray, +) -> collie.Next(ShardState, a) { + process.named_subject(state.coordinator_name) + |> process.send( + MessageFromDiscord(ErrorEvent(EventNotValidUtf8(event_payload: binary))), + ) + + collie.continue(state) +} + +fn handle_requested_heartbeat( + connection: collie.Connection, + state: ShardState, +) -> collie.Next(ShardState, a) { + use <- handle_heartbeat_inequality(connection, state) + let sequence = case state { + ShardWaitingForInitialization(..) -> None + ShardWelcomed(resuming_info:, ..) -> + case resuming_info { + Some(info) -> info.sequence + None -> None + } + } + + let body = + heartbeat_to_json(sequence) + |> json.to_string + + let result = collie.send_text_frame(connection, body) + + case result { + Ok(_) -> { + let state = case state { + ShardWaitingForInitialization(..) -> state + ShardWelcomed(..) -> + ShardWelcomed(..state, heartbeats: state.heartbeats + 1) + } + + collie.continue(state) + } + + Error(error) -> { + // allows the user to choose whether they want to stop in the on_event handler + // does push the user response to the bottom of the mailbox, but this shouldn't be a problem + // as long as no blocking code is running in `on_event` + process.named_subject(state.coordinator_name) + |> process.send( + MessageFromDiscord( + ErrorEvent(CouldNotSendEvent(event_payload: body, reason: error)), + ), + ) + + collie.continue(state) + } + } +} + +fn heartbeat_to_json(sequence sequence: Option(Int)) -> Json { + [#("op", json.int(1)), #("d", json.nullable(sequence, json.int))] + |> json.object +} + +fn handle_scheduled_heartbeat( + connection: collie.Connection, + state: ShardState, +) -> collie.Next(ShardState, a) { + case state { + // this shouldn't be possible, but if it does happen, let's just reconnect + ShardWaitingForInitialization(..) -> + handle_shard_shutdown( + connection, + collie.CloseReason(code: collie.GoingAway, reason: ""), + AbnormalShardShutdown("Heartbeating before welcomed. Reconnecting."), + ) + ShardWelcomed(..) -> { + use <- handle_heartbeat_inequality(connection, state) + let sequence = case state.resuming_info { + Some(info) -> info.sequence + None -> None + } + + let body = + heartbeat_to_json(sequence) + |> json.to_string + + let result = collie.send_text_frame(connection, body) + + // schedule the next heartbeat + state.self_subject + |> process.send_after( + state.heartbeat_interval_ms, + ShardScheduledHeartbeat |> collie.to_user_message, + ) + + case result { + Ok(_) -> + collie.continue( + ShardWelcomed(..state, heartbeats: state.heartbeats + 1), + ) + Error(error) -> { + // allows the user to choose whether they want to stop in the on_event handler + // does push the user response to the bottom of the mailbox, but this shouldn't be a problem + // as long as no blocking code is running in `on_event` + process.named_subject(state.coordinator_name) + |> process.send( + MessageFromDiscord( + ErrorEvent(CouldNotSendEvent(event_payload: body, reason: error)), + ), + ) + + collie.continue(state) + } + } + } + } +} + +fn handle_heartbeat_inequality( + connection: collie.Connection, + state: ShardState, + next: fn() -> collie.Next(ShardState, _), +) -> collie.Next(ShardState, _) { + case state { + ShardWaitingForInitialization(..) -> next() + ShardWelcomed(..) -> + case state.heartbeats == state.heartbeat_acks { + True -> next() + False -> + handle_shard_shutdown( + connection, + collie.NoCloseReason, + AbnormalShardShutdown( + "Heartbeat inequality - possible zombied connection. Reconnecting.", + ), + ) + } + } +} + +fn handle_text_message( + state: ShardState, + text: String, +) -> collie.Next(ShardState, _) { + case json.parse(text, using: discord_message_decoder()) { + Ok(message) -> handle_discord_message(state, message) + Error(error) -> { + state.coordinator_name + |> process.named_subject + |> process.send( + MessageFromDiscord(ErrorEvent(CouldNotDecodeEvent(error))), + ) + + collie.continue(state) + } + } +} + +fn handle_discord_message( + state: ShardState, + message: DiscordMessage, +) -> collie.Next(ShardState, _) { + case message { + HelloDiscordMessage(hello) -> handle_hello_message(state, hello) + } +} + +fn handle_hello_message( + state: ShardState, + hello: HelloMessage, +) -> collie.Next(ShardState, _) { + state.self_subject + |> process.send_after( + hello.heartbeat_interval_ms + |> int.to_float + // jittering to prevent mass client startup + // on discord servers + |> float.multiply(float.random()) + |> float.round, + ShardScheduledHeartbeat |> collie.to_user_message, + ) + + let resuming_info = case state { + ShardWaitingForInitialization(..) -> None + ShardWelcomed(resuming_info:, ..) -> resuming_info + } + + collie.continue(ShardWelcomed( + id: state.id, + num_shards: state.num_shards, + connection_info: state.connection_info, + identification: state.identification, + coordinator_name: state.coordinator_name, + self_subject: state.self_subject, + resuming_info:, + heartbeat_interval_ms: hello.heartbeat_interval_ms, + heartbeats: 0, + heartbeat_acks: 0, + )) +} + +type DiscordMessage { + HelloDiscordMessage(HelloMessage) +} + +type HelloMessage { + HelloMessage(heartbeat_interval_ms: Int) +} + +fn discord_message_decoder() -> Decoder(DiscordMessage) { + use opcode <- decode.field("op", decode.int) + case opcode { + 10 -> { + use message <- decode.field("d", hello_message_decoder()) + decode.success(HelloDiscordMessage(message)) + } + _ -> decode.failure(HelloDiscordMessage(HelloMessage(0)), "DiscordMessage") + } +} + +fn hello_message_decoder() -> Decoder(HelloMessage) { + use heartbeat_interval_ms <- decode.field("heartbeat_interval", decode.int) + decode.success(HelloMessage(heartbeat_interval_ms:)) +} + +type ShardShutdownType { + /// Will not restart the shard. + RegularShardShutdown + /// Will restart the shard. + AbnormalShardShutdown(reason: String) +} + +fn handle_shard_shutdown( + connection: collie.Connection, + reason: collie.CloseReason, + type_: ShardShutdownType, +) { + // i can discard the result here, because if graceful shutdown fails, + // `collie.stop()` will just close the connection anyway + let _ = collie.send_close_frame(connection, reason) + + case type_ { + RegularShardShutdown -> collie.stop() + AbnormalShardShutdown(reason:) -> collie.stop_abnormal(reason) + } +} + fn on_shard_close(state: ShardState, reason: collie.CloseReason) -> Nil { todo }