eldridge.cam
Trading card city builder game?
remove notifications code
-219
-219
server/src/db/notification_listener.gleam
-219
server/src/db/notification_listener.gleam
···
1
-
import gleam/dynamic/decode
2
-
import gleam/erlang/process.{type Name}
3
-
import gleam/erlang/reference
4
-
import gleam/io
5
-
import gleam/json
6
-
import gleam/option.{type Option}
7
-
import gleam/otp/actor
8
-
import gleam/result
9
-
import gleam/string
10
-
import palabres
11
-
import pog
12
-
13
-
pub type State(st) {
14
-
State(
15
-
state: st,
16
-
reference: reference.Reference,
17
-
notifications: pog.NotificationsConnection,
18
-
)
19
-
}
20
-
21
-
pub type Message(msg) {
22
-
Notification(pog.Notification)
23
-
Unlisten
24
-
Msg(msg)
25
-
}
26
-
27
-
pub type Never
28
-
29
-
pub opaque type Next(state) {
30
-
Continue(state)
31
-
Stop
32
-
StopAbnormal(String)
33
-
}
34
-
35
-
pub fn continue(state: state) {
36
-
Continue(state)
37
-
}
38
-
39
-
pub fn stop() {
40
-
Stop
41
-
}
42
-
43
-
pub fn stop_abnormal(reason: String) {
44
-
StopAbnormal(reason)
45
-
}
46
-
47
-
pub opaque type Builder(state, event, msg) {
48
-
Builder(
49
-
initial_state: state,
50
-
channel: Option(#(pog.NotificationsConnection, String)),
51
-
handler: Option(fn(state, msg) -> Next(state)),
52
-
event_handler: Option(
53
-
#(decode.Decoder(event), fn(state, event) -> Next(state)),
54
-
),
55
-
name: Option(Name(Message(msg))),
56
-
)
57
-
}
58
-
59
-
pub fn new(state: state) -> Builder(state, Never, Never) {
60
-
Builder(
61
-
initial_state: state,
62
-
channel: option.None,
63
-
handler: option.None,
64
-
event_handler: option.None,
65
-
name: option.None,
66
-
)
67
-
}
68
-
69
-
pub fn named(
70
-
builder: Builder(state, event, msg),
71
-
name: Name(Message(msg)),
72
-
) -> Builder(state, event, msg) {
73
-
Builder(..builder, name: option.Some(name))
74
-
}
75
-
76
-
pub fn listen_to(
77
-
builder: Builder(state, event, msg),
78
-
notifications: pog.NotificationsConnection,
79
-
channel: String,
80
-
) -> Builder(state, event, msg) {
81
-
Builder(..builder, channel: option.Some(#(notifications, channel)))
82
-
}
83
-
84
-
pub fn on_message(
85
-
builder: Builder(state, event, _msg),
86
-
handler: fn(state, msg) -> Next(state),
87
-
) -> Builder(state, event, msg) {
88
-
Builder(..builder, handler: option.Some(handler))
89
-
}
90
-
91
-
pub fn on_notification(
92
-
builder: Builder(state, _event, msg),
93
-
decoder: decode.Decoder(event),
94
-
handler: fn(state, event) -> Next(state),
95
-
) -> Builder(state, event, msg) {
96
-
Builder(..builder, event_handler: option.Some(#(decoder, handler)))
97
-
}
98
-
99
-
fn with(state: t, option: Option(v), apply: fn(t, v) -> t) -> t {
100
-
case option {
101
-
option.None -> state
102
-
option.Some(v) -> apply(state, v)
103
-
}
104
-
}
105
-
106
-
fn shutdown(state: State(st)) {
107
-
pog.unlisten(state.notifications, state.reference)
108
-
}
109
-
110
-
pub fn unlisten(listener: process.Subject(Message(msg))) {
111
-
process.send(listener, Unlisten)
112
-
}
113
-
114
-
pub fn start(
115
-
builder: Builder(state, event, msg),
116
-
) -> Result(actor.Started(process.Subject(Message(msg))), actor.StartError) {
117
-
use #(notifications, channel) <- result.try(
118
-
option.to_result(builder.channel, "missing channel")
119
-
|> result.map_error(actor.InitFailed),
120
-
)
121
-
use #(event_decoder, event_handler) <- result.try(
122
-
option.to_result(builder.event_handler, "missing event handler")
123
-
|> result.map_error(actor.InitFailed),
124
-
)
125
-
126
-
actor.new_with_initialiser(100, fn(subject) -> Result(
127
-
actor.Initialised(State(state), Message(msg), process.Subject(Message(msg))),
128
-
String,
129
-
) {
130
-
palabres.info("starting database listener")
131
-
|> palabres.string("channel", channel)
132
-
|> palabres.log()
133
-
134
-
use reference <- result.try(
135
-
pog.listen(notifications, channel)
136
-
|> result.map_error(fn(_) { "failed to start listener" }),
137
-
)
138
-
139
-
let selector: process.Selector(Message(msg)) =
140
-
process.new_selector()
141
-
|> process.select(subject)
142
-
|> process.merge_selector(
143
-
pog.notification_selector()
144
-
|> process.map_selector(Notification),
145
-
)
146
-
|> process.select_other(fn(dyn) {
147
-
io.println(string.inspect(dyn))
148
-
Unlisten
149
-
})
150
-
151
-
Ok(
152
-
actor.initialised(State(
153
-
state: builder.initial_state,
154
-
reference:,
155
-
notifications:,
156
-
))
157
-
|> actor.returning(subject)
158
-
|> actor.selecting(selector),
159
-
)
160
-
})
161
-
|> with(builder.name, actor.named)
162
-
|> actor.on_message(fn(state, message) {
163
-
use delegate_message <- handle_generic(state, message, event_decoder)
164
-
let sub_next = case delegate_message {
165
-
Event(event) -> event_handler(state.state, event)
166
-
Message(msg) -> {
167
-
let assert option.Some(handler) = builder.handler
168
-
handler(state.state, msg)
169
-
}
170
-
}
171
-
case sub_next {
172
-
Stop -> {
173
-
shutdown(state)
174
-
actor.stop()
175
-
}
176
-
StopAbnormal(reason) -> {
177
-
shutdown(state)
178
-
actor.stop_abnormal(reason)
179
-
}
180
-
Continue(substate) -> actor.continue(State(..state, state: substate))
181
-
}
182
-
})
183
-
|> actor.start()
184
-
}
185
-
186
-
fn handle_generic(
187
-
state: State(state),
188
-
message: Message(msg),
189
-
decoder: decode.Decoder(event),
190
-
delegate_fn: fn(DelegateMessage(event, msg)) ->
191
-
actor.Next(State(state), Message(msg)),
192
-
) {
193
-
case message {
194
-
Notification(pog.Notify(_, _, channel, payload)) -> {
195
-
palabres.debug("database notification received")
196
-
|> palabres.string("channel", channel)
197
-
|> palabres.string("payload", payload)
198
-
|> palabres.log()
199
-
200
-
case json.parse(payload, using: decoder) {
201
-
Ok(event) -> delegate_fn(Event(event))
202
-
Error(_) -> {
203
-
shutdown(state)
204
-
actor.stop_abnormal("unexpected event")
205
-
}
206
-
}
207
-
}
208
-
Unlisten -> {
209
-
shutdown(state)
210
-
actor.stop()
211
-
}
212
-
Msg(msg) -> delegate_fn(Message(msg))
213
-
}
214
-
}
215
-
216
-
type DelegateMessage(event, message) {
217
-
Event(event)
218
-
Message(message)
219
-
}