mail based rss feed aggregator
add `backend.gleam`, turn `sender.starter` and `fetcher.starter` into `*.manager` implemented most of the features in the readme, do still need to test em
+831
-299
+20
-19
README.md
+20
-19
README.md
···
39
39
- [ ] ewe logs
40
40
41
41
- [ ] backend manager
42
-
- [ ] `backend.new_subscription(backend_name, user, feed)`
43
-
- [ ] save new subscription to database
44
-
- [ ] notify responsible sender
45
-
- [ ] start sender if not running
46
-
- [ ] start fetcher if not running
47
-
- [ ] `backend.remove_subscription(backend_name, user, feed)`
48
-
- [ ] delete from database
49
-
- [ ] notify sender
50
-
- [ ] if noone is subscribed anymore
51
-
- [ ] remove feed from database
52
-
- [ ] stop fetcher
42
+
- [ ] TESTING THIS
43
+
- [x] `backend.new_subscription(backend_name, user, feed)`
44
+
- [x] save new subscription to database
45
+
- [x] notify responsible sender
46
+
- [x] start sender if not running
47
+
- [x] start fetcher if not running
48
+
- [x] `backend.remove_subscription(backend_name, user, feed)`
49
+
- [x] delete from database
50
+
- [x] notify sender
51
+
- [x] if noone is subscribed anymore
52
+
- [x] remove feed from database
53
+
- [x] stop fetcher
53
54
- [ ] `backend.status(backend_name)`
54
55
- [ ] fetcher data
55
56
- [ ] time till next fetch
···
57
58
- [ ] sender data
58
59
- [ ] status ok / issues
59
60
- [ ] maybe mails sent? (we do have this data)
60
-
- [ ] `backend.restart_feed(backend_name, feed)`
61
+
- [x] `backend.restart_feed(backend_name, feed)`
61
62
- restart the fetcher for this feed
62
-
- [ ] `backend.refetch(backend_name, feed)`
63
+
- [x] `backend.refetch(backend_name, feed)`
63
64
- trigger early fetching of a given feed
64
-
- [ ] `backend.restart_user(backend_name, user)`
65
+
- [x] `backend.restart_user(backend_name, user)`
65
66
- restart sender for this user
66
-
- [ ] `backend.find_feed(backend_name, uri) -> Result(rss.Location, Nil)`
67
-
- [ ] check database
68
-
- [ ] found -> found `rss.Location`
69
-
- [ ] not found -> `rss.new_location`
70
-
- [ ] db_error -> Nil
67
+
- [x] `backend.find_feed(backend_name, uri) -> Result(rss.Location, Nil)`
68
+
- [x] check database
69
+
- [x] found -> found `rss.Location`
70
+
- [x] not found -> `rss.new_location`
71
+
- [x] db_error -> Nil
71
72
72
73
73
74
+14
-14
src/eater.gleam
+14
-14
src/eater.gleam
···
16
16
// This software is provided "AS IS", WITHOUT WARRANTY OF ANY KIND. [cite: 5]
17
17
// See the Licence for the specific language governing permissions and limitations. [cite: 6]
18
18
19
+
import eater/backend
19
20
import eater/configuration
20
21
import eater/database
21
22
import eater/fetcher
···
50
51
let assert Ok(_) =
51
52
ensure_default_admin_exists(database, smtp_environment, configuration)
52
53
54
+
let backend = process.new_name("backend")
53
55
let registry = process.new_name("registry")
56
+
let sender_factory = process.new_name("sender_factory")
57
+
let sender_manager = process.new_name("sender_starter")
54
58
let fetcher_factory = process.new_name("fetcher_factory")
55
-
let fetcher_starter = process.new_name("fetcher_starter")
56
-
let sender_factory = process.new_name("sender_factory")
57
-
let sender_starter = process.new_name("sender_starter")
59
+
let fetcher_manager = process.new_name("fetcher_starter")
58
60
59
61
let assert Ok(_supervisor) =
60
62
supervisor.new(supervisor.RestForOne)
61
-
|> supervisor.add(group_registry.supervised(registry))
62
-
|> supervisor.add(fetcher.factory(
63
-
name: fetcher_factory,
64
-
starter: fetcher_starter,
65
-
registry:,
66
-
database:,
67
-
))
68
-
|> supervisor.add(sender.factory(
69
-
name: sender_factory,
70
-
starter: sender_starter,
71
-
registry:,
63
+
|> supervisor.add(backend.supervised(
64
+
names: backend.Names(
65
+
backend:,
66
+
registry:,
67
+
sender_factory:,
68
+
sender_manager:,
69
+
fetcher_factory:,
70
+
fetcher_manager:,
71
+
),
72
72
database:,
73
73
smtp_environment:,
74
74
))
+409
src/eater/backend.gleam
+409
src/eater/backend.gleam
···
1
+
// eater
2
+
// Copyright (C) 2026 Olivia Streun and contributors. [cite: 4]
3
+
//
4
+
// This software is licensed under the European Union Public Licence (EUPL) v1.2.
5
+
// You may not use this work except in compliance with the Licence.
6
+
// You may obtain a copy of the Licence at: https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
7
+
//
8
+
// AI TRAINING NOTICE: Rights for TDM and AI training are EXPRESSLY RESERVED
9
+
// under Art 4(3) Dir 2019/790. AI training constitutes a Derivative Work.
10
+
// See LICENSE file in the repository root for full details.
11
+
//
12
+
//
13
+
// This software is provided "AS IS", WITHOUT WARRANTY OF ANY KIND. [cite: 5]
14
+
// See the Licence for the specific language governing permissions and limitations. [cite: 6]
15
+
16
+
import eater/database
17
+
import eater/feed/rss
18
+
import eater/fetcher
19
+
import eater/pubsub
20
+
import eater/sender
21
+
import eater/smtp
22
+
import eater/user
23
+
import gleam/erlang/process
24
+
import gleam/otp/actor
25
+
import gleam/otp/static_supervisor as supervisor
26
+
import gleam/otp/supervision
27
+
import gleam/result
28
+
import gleam/string
29
+
import gleam/uri
30
+
import sqlight
31
+
import woof
32
+
33
+
// public stuff -----------------------------------------------------------------
34
+
35
+
pub type Names {
36
+
Names(
37
+
backend: process.Name(Message),
38
+
registry: pubsub.Registry,
39
+
sender_factory: sender.FactoryName,
40
+
sender_manager: sender.ManagerName,
41
+
fetcher_factory: fetcher.FactoryName,
42
+
fetcher_manager: fetcher.ManagerName,
43
+
)
44
+
}
45
+
46
+
pub fn supervised(
47
+
names names: Names,
48
+
database database: sqlight.Connection,
49
+
smtp_environment smtp_environment: smtp.SmtpEnvironment,
50
+
) -> supervision.ChildSpecification(supervisor.Supervisor) {
51
+
let Names(
52
+
backend: _,
53
+
registry:,
54
+
sender_factory:,
55
+
sender_manager:,
56
+
fetcher_factory:,
57
+
fetcher_manager:,
58
+
) = names
59
+
60
+
supervisor.new(supervisor.RestForOne)
61
+
|> supervisor.add(pubsub.supervised(registry))
62
+
|> supervisor.add(fetcher.factory(
63
+
name: fetcher_factory,
64
+
starter: fetcher_manager,
65
+
registry:,
66
+
database:,
67
+
))
68
+
|> supervisor.add(sender.factory(
69
+
name: sender_factory,
70
+
starter: sender_manager,
71
+
registry:,
72
+
database:,
73
+
smtp_environment:,
74
+
))
75
+
|> supervisor.add(backend_manager(names, database, smtp_environment))
76
+
|> supervisor.supervised()
77
+
}
78
+
79
+
// main api ---------------------------------------------------------------------
80
+
81
+
/// add a new subscription for a user
82
+
///
83
+
pub fn new_subscription(
84
+
backend backend: process.Name(Message),
85
+
user user: user.User,
86
+
feed feed: rss.Location,
87
+
) -> Nil {
88
+
let backend = process.named_subject(backend)
89
+
process.send(backend, NewSubscription(user, feed))
90
+
}
91
+
92
+
/// remove a subscription from a user
93
+
///
94
+
pub fn remove_subscription(
95
+
backend backend: process.Name(Message),
96
+
user user: user.User,
97
+
feed feed: rss.Location,
98
+
) -> Nil {
99
+
let backend = process.named_subject(backend)
100
+
process.send(backend, RemoveSubscription(user, feed))
101
+
}
102
+
103
+
/// to be fleshed out as it becomes clearer what this entails
104
+
pub type Status
105
+
106
+
/// get the status of the backend
107
+
///
108
+
/// waits for the response from the backend
109
+
///
110
+
pub fn status(backend backend: process.Name(Message)) -> Result(Status, Nil) {
111
+
let backend = process.named_subject(backend)
112
+
let send_to = process.new_subject()
113
+
114
+
process.send(backend, Status(send_to:))
115
+
116
+
process.receive(send_to, within: 1000)
117
+
}
118
+
119
+
/// restart the fetcher for this feed
120
+
///
121
+
pub fn restart_feed(
122
+
backend backend: process.Name(Message),
123
+
feed feed: rss.Location,
124
+
) -> Nil {
125
+
let backend = process.named_subject(backend)
126
+
process.send(backend, RestartFeed(feed))
127
+
}
128
+
129
+
/// trigger an early refetch for this feed
130
+
///
131
+
pub fn refetch(
132
+
backend backend: process.Name(Message),
133
+
feed feed: rss.Location,
134
+
) -> Nil {
135
+
let backend = process.named_subject(backend)
136
+
process.send(backend, Refetch(feed))
137
+
}
138
+
139
+
/// restart the sender for this user
140
+
///
141
+
pub fn restart_user(
142
+
backend backend: process.Name(Message),
143
+
user user: user.User,
144
+
) -> Nil {
145
+
let backend = process.named_subject(backend)
146
+
process.send(backend, RestartUser(user))
147
+
}
148
+
149
+
/// find a feed using a uri
150
+
///
151
+
pub fn find_feed(
152
+
backend backend: process.Name(Message),
153
+
uri uri: uri.Uri,
154
+
) -> Result(rss.Location, Nil) {
155
+
let backend = process.named_subject(backend)
156
+
157
+
let send_to = process.new_subject()
158
+
159
+
process.send(backend, FindFeed(uri, send_to:))
160
+
161
+
process.receive(send_to, within: 1000)
162
+
|> result.flatten()
163
+
}
164
+
165
+
pub opaque type Message {
166
+
167
+
// external messages ----------------------------------------------------------
168
+
/// add a subscription for a user
169
+
///
170
+
/// - save to database
171
+
/// - notify sender
172
+
/// - start sender if not running
173
+
/// - start fetcher if not running
174
+
///
175
+
NewSubscription(user.User, rss.Location)
176
+
177
+
/// remove a subscription from a user
178
+
///
179
+
/// - delete from database
180
+
/// - notify sender
181
+
/// - if noone is subscribed
182
+
/// - stop fetcher
183
+
/// - remove feed from database
184
+
///
185
+
RemoveSubscription(user.User, rss.Location)
186
+
187
+
/// get the status of the backend
188
+
///
189
+
/// - fetcher data
190
+
/// - time till next fetch
191
+
/// - status ok / issue
192
+
/// - sender data
193
+
/// - status ok / issue
194
+
/// - total mails sent (maybe)
195
+
///
196
+
Status(send_to: process.Subject(Status))
197
+
198
+
/// restart the fetcher for this feed
199
+
///
200
+
RestartFeed(rss.Location)
201
+
202
+
/// trigger an early refetch for this feed
203
+
///
204
+
Refetch(rss.Location)
205
+
206
+
/// restart the sender for this user
207
+
///
208
+
RestartUser(user.User)
209
+
210
+
/// find a feed using a uri
211
+
///
212
+
/// - check the database
213
+
/// - found -> send the associated `rss.Location` to `send_to`
214
+
/// - not found -> `rss.new_location` and send to `send_to`
215
+
///
216
+
FindFeed(uri.Uri, send_to: process.Subject(Result(rss.Location, Nil)))
217
+
// internal messages ----------------------------------------------------------
218
+
}
219
+
220
+
type State {
221
+
State(
222
+
self: process.Subject(Message),
223
+
names: Names,
224
+
database: sqlight.Connection,
225
+
smtp_environment: smtp.SmtpEnvironment,
226
+
logger: woof.Logger,
227
+
)
228
+
}
229
+
230
+
// actor ------------------------------------------------------------------------
231
+
232
+
fn backend_manager(
233
+
names: Names,
234
+
database: sqlight.Connection,
235
+
smtp_environment: smtp.SmtpEnvironment,
236
+
) -> supervision.ChildSpecification(Nil) {
237
+
supervision.worker(fn() {
238
+
actor.new_with_initialiser(200, fn(self) {
239
+
actor.initialised(State(
240
+
self:,
241
+
names:,
242
+
database:,
243
+
smtp_environment:,
244
+
logger: woof.new("BACKEND-MANAGER"),
245
+
))
246
+
|> Ok
247
+
})
248
+
|> actor.named(names.backend)
249
+
|> actor.on_message(on_message)
250
+
|> actor.start()
251
+
})
252
+
}
253
+
254
+
fn on_message(state: State, message: Message) -> actor.Next(State, Message) {
255
+
case message {
256
+
// external messages --------------------------------------------------------
257
+
NewSubscription(user, feed) -> handle_new_subscription(state, user, feed)
258
+
RemoveSubscription(user, feed) ->
259
+
handle_remove_subscription(state, user, feed)
260
+
Status(send_to: _) -> todo as "get status of system"
261
+
RestartFeed(feed) -> {
262
+
fetcher.restart_for(state.names.fetcher_manager, feed)
263
+
actor.continue(state)
264
+
}
265
+
Refetch(feed) -> {
266
+
fetcher.refetch_for(state.names.fetcher_manager, feed)
267
+
actor.continue(state)
268
+
}
269
+
RestartUser(user) -> {
270
+
sender.restart_for(state.names.sender_manager, user)
271
+
actor.continue(state)
272
+
}
273
+
FindFeed(uri, send_to:) -> {
274
+
case database.feed_by_link(uri, state.database) {
275
+
Error(_) -> Error(Nil)
276
+
Ok([]) -> Ok(rss.new_location(uri))
277
+
Ok([feed, ..]) -> Ok(feed)
278
+
}
279
+
|> process.send(send_to, _)
280
+
actor.continue(state)
281
+
}
282
+
// internal messages --------------------------------------------------------
283
+
}
284
+
}
285
+
286
+
/// add a subscription for a user
287
+
///
288
+
/// - save to database
289
+
/// - notify sender
290
+
/// - start sender if not running
291
+
/// - start fetcher if not running
292
+
///
293
+
fn handle_new_subscription(
294
+
state: State,
295
+
user: user.User,
296
+
feed: rss.Location,
297
+
) -> actor.Next(State, Message) {
298
+
// - save to database
299
+
use _ <- try_twice(
300
+
fn() { database.add_feed(feed, state.database) },
301
+
otherwise: log_and_stop(state.logger, "Failed to add feed to database", [
302
+
woof.field("feed", feed.link |> uri.to_string()),
303
+
]),
304
+
)
305
+
306
+
// - notify sender
307
+
pubsub.publish_subscribed_to_feed(user, feed, state.names.registry)
308
+
309
+
// - start sender if not running
310
+
sender.start_for(state.names.sender_manager, user)
311
+
// - start fetcher if not running
312
+
fetcher.start_for(state.names.fetcher_manager, feed)
313
+
314
+
actor.continue(state)
315
+
}
316
+
317
+
/// remove a subscription from a user
318
+
///
319
+
/// - delete from database
320
+
/// - notify sender
321
+
/// - if noone is subscribed
322
+
/// - stop fetcher
323
+
/// - remove feed from database
324
+
///
325
+
fn handle_remove_subscription(
326
+
state: State,
327
+
user: user.User,
328
+
feed: rss.Location,
329
+
) -> actor.Next(State, Message) {
330
+
// - delete from database
331
+
use _ <- try_twice(
332
+
fn() { database.delete_subscription(user, feed, state.database) },
333
+
otherwise: log_and_stop(state.logger, "Failed delete subscription", [
334
+
woof.field("feed", feed.link |> uri.to_string()),
335
+
woof.field("user", user.email),
336
+
]),
337
+
)
338
+
339
+
// - notify sender
340
+
pubsub.publish_unsubscribed_from_feed(user, feed, state.names.registry)
341
+
342
+
use subscribers <- try_twice(
343
+
fn() { database.subscription_count(feed, state.database) },
344
+
otherwise: log_and_stop(
345
+
state.logger,
346
+
"Failed get subscriber count from database",
347
+
[
348
+
woof.field("feed", feed.link |> uri.to_string()),
349
+
],
350
+
),
351
+
)
352
+
353
+
case subscribers {
354
+
// - if noone is subscribed
355
+
0 -> {
356
+
// - stop fetcher
357
+
fetcher.stop_for(state.names.fetcher_manager, feed)
358
+
// - remove feed from database
359
+
use _ <- try_twice(
360
+
fn() { database.delete_feed(feed, state.database) },
361
+
otherwise: log_and_stop(
362
+
state.logger,
363
+
"Failed to delete feed from database",
364
+
[
365
+
woof.field("feed", feed.link |> uri.to_string()),
366
+
],
367
+
),
368
+
)
369
+
370
+
actor.continue(state)
371
+
}
372
+
_ -> actor.continue(state)
373
+
}
374
+
}
375
+
376
+
// helpers ----------------------------------------------------------------------
377
+
378
+
/// returns a function that logs an error
379
+
/// to the supplied logger with the supplied message
380
+
/// and then returns `actor.stop_abnormal` with that same message
381
+
///
382
+
fn log_and_stop(
383
+
logger logger: woof.Logger,
384
+
message message: String,
385
+
fields fields: List(#(String, String)),
386
+
) -> fn(a) -> actor.Next(b, c) {
387
+
fn(error) {
388
+
woof.log(logger, woof.Error, message, [
389
+
woof.field("details", string.inspect(error)),
390
+
..fields
391
+
])
392
+
actor.stop_abnormal(message)
393
+
}
394
+
}
395
+
396
+
fn try_twice(
397
+
try try: fn() -> Result(a, b),
398
+
otherwise otherwise: fn(b) -> c,
399
+
continue continue: fn(a) -> c,
400
+
) -> c {
401
+
case try() {
402
+
Ok(value) -> continue(value)
403
+
Error(_) ->
404
+
case try() {
405
+
Ok(value) -> continue(value)
406
+
Error(error) -> otherwise(error)
407
+
}
408
+
}
409
+
}
+1
-1
src/eater/database.gleam
+1
-1
src/eater/database.gleam
+160
-90
src/eater/fetcher.gleam
+160
-90
src/eater/fetcher.gleam
···
19
19
import eater/database
20
20
import eater/feed/rss
21
21
import eater/pubsub
22
+
import gleam/dict
22
23
import gleam/erlang/process
23
24
import gleam/http
24
25
import gleam/http/request
···
35
36
import parsed_it/xml
36
37
import sqlight
37
38
import woof
39
+
import youid/uuid
38
40
39
41
const interval_in_ms: Int = 3_600_000
40
42
···
52
54
/// reexport of the underlying `process.Name` for convenience
53
55
///
54
56
pub type FactoryName =
55
-
process.Name(factory.Message(Start, Nil))
57
+
process.Name(factory.Message(Start, process.Subject(Message)))
56
58
57
59
/// the factory for the fetcher actors
58
60
///
59
61
pub fn factory(
60
62
name factory: FactoryName,
61
-
starter starter_name: StarterName,
63
+
starter starter_name: ManagerName,
62
64
registry registry: pubsub.Registry,
63
65
database database: sqlight.Connection,
64
66
) -> supervision.ChildSpecification(_) {
···
72
74
|> static_supervisor.supervised()
73
75
}
74
76
75
-
type StarterState {
76
-
StarterState(
77
-
self: process.Subject(StarterMessage),
77
+
type ManagerState {
78
+
ManagerState(
79
+
self: process.Subject(ManagerMessage),
78
80
factory: FactoryName,
79
81
registry: pubsub.Registry,
80
82
database: sqlight.Connection,
83
+
fetchers: dict.Dict(uuid.Uuid, #(process.Pid, process.Subject(Message))),
81
84
)
82
85
}
83
86
84
-
pub opaque type StarterMessage {
87
+
pub opaque type ManagerMessage {
85
88
// initial startup ------------------------------------------------------------
86
89
/// start the process
87
90
StartAll
88
-
/// fetch feeds from db (attempt 1)
89
-
DatabaseReturnedFeeds(Result(List(rss.Location), sqlight.Error))
90
-
/// fetch feeds from db (attempt 2)
91
-
FetchAgain
91
+
/// fetch feeds from db
92
+
DatabaseReturnedFeeds(List(rss.Location))
92
93
93
-
/// start fetcher (attempt 1)
94
-
Started(Result(actor.Started(Nil), actor.StartError), rss.Location)
95
-
/// start fetcher (attempt 2)
96
-
StartAgain(rss.Location)
94
+
/// start fetcher
95
+
Started(
96
+
Result(actor.Started(process.Subject(Message)), actor.StartError),
97
+
rss.Location,
98
+
)
97
99
98
100
// runtime additions ----------------------------------------------------------
99
101
/// start a sender for a given user at runtime
100
102
StartFor(rss.Location)
103
+
/// stop the fetcher for a given feed at runtime
104
+
StopFor(rss.Location)
105
+
/// restart the fetcher for a given feed at runtime
106
+
RestartFor(rss.Location)
107
+
/// trigger an early refetching of this feed
108
+
RecheckFor(rss.Location)
101
109
}
102
110
103
-
pub type StarterName =
104
-
process.Name(StarterMessage)
111
+
pub type ManagerName =
112
+
process.Name(ManagerMessage)
105
113
106
-
pub fn start_for(starter: StarterName, feed: rss.Location) {
114
+
pub fn start_for(starter: ManagerName, feed: rss.Location) {
107
115
process.named_subject(starter)
108
116
|> process.send(StartFor(feed))
109
117
}
110
118
119
+
pub fn stop_for(manager: ManagerName, feed: rss.Location) {
120
+
process.named_subject(manager)
121
+
|> process.send(StopFor(feed))
122
+
}
123
+
124
+
pub fn restart_for(manager: ManagerName, feed: rss.Location) {
125
+
process.named_subject(manager)
126
+
|> process.send(RestartFor(feed))
127
+
}
128
+
129
+
pub fn refetch_for(manager: ManagerName, feed: rss.Location) {
130
+
process.named_subject(manager)
131
+
|> process.send(RecheckFor(feed))
132
+
}
133
+
111
134
fn starter(
112
-
name: StarterName,
135
+
name: ManagerName,
113
136
factory: FactoryName,
114
137
registry: pubsub.Registry,
115
138
database: sqlight.Connection,
···
123
146
actor.new_with_initialiser(1000, fn(self) {
124
147
process.send_after(self, 100, StartAll)
125
148
126
-
actor.initialised(StarterState(self:, factory:, registry:, database:))
149
+
actor.initialised(ManagerState(
150
+
self:,
151
+
factory:,
152
+
registry:,
153
+
database:,
154
+
fetchers: dict.new(),
155
+
))
127
156
|> Ok
128
157
})
129
158
|> actor.on_message(fn(state, message) {
130
159
case message {
131
160
StartAll -> {
132
-
database.all_feeds(database)
133
-
|> DatabaseReturnedFeeds
134
-
|> actor.send(state.self, _)
161
+
case database.all_feeds(database) {
162
+
Ok(feeds) -> {
163
+
DatabaseReturnedFeeds(feeds) |> actor.send(state.self, _)
164
+
actor.continue(state)
165
+
}
166
+
// have the supervision handle retrying
167
+
Error(_) -> actor.stop_abnormal("Failed to get feeds from database")
168
+
}
135
169
136
170
actor.continue(state)
137
171
}
138
-
DatabaseReturnedFeeds(Ok(feeds)) -> {
172
+
DatabaseReturnedFeeds(feeds) -> {
139
173
feeds
140
174
|> list.map(fn(feed) {
141
-
start_new(factory, Start(database:, registry:, feed:))
175
+
start_new(
176
+
factory,
177
+
Start(database:, registry:, feed:, manager: name),
178
+
)
142
179
|> Started(feed)
143
180
|> actor.send(state.self, _)
144
181
})
145
182
146
183
actor.continue(state)
147
184
}
148
-
DatabaseReturnedFeeds(Error(db_error)) -> {
149
-
log(woof.Warning, "Failed to get feeds from database once", [
150
-
woof.field("details", string.inspect(db_error)),
151
-
])
152
185
153
-
actor.send(state.self, FetchAgain)
154
-
actor.continue(state)
155
-
}
156
-
FetchAgain -> {
157
-
case database.all_feeds(database) {
158
-
Ok(feeds) -> {
159
-
DatabaseReturnedFeeds(Ok(feeds))
160
-
|> actor.send(state.self, _)
186
+
// all starts go through this
187
+
Started(Ok(started), feed) -> {
188
+
// if there is an existing sender for this
189
+
// tell it to shut down
190
+
case dict.get(state.fetchers, feed.id) {
191
+
Ok(#(pid, _)) ->
192
+
case process.is_alive(pid) {
193
+
True -> process.send_exit(pid)
194
+
False -> Nil
195
+
}
196
+
Error(_) -> Nil
197
+
}
161
198
162
-
actor.continue(state)
163
-
}
164
-
Error(error) -> {
165
-
log(woof.Warning, "Failed to get feeds from database twice", [
166
-
woof.field("details", string.inspect(error)),
167
-
])
199
+
let state =
200
+
ManagerState(
201
+
..state,
202
+
fetchers: dict.insert(state.fetchers, feed.id, #(
203
+
started.pid,
204
+
started.data,
205
+
)),
206
+
)
168
207
169
-
actor.stop_abnormal("Failed to get feeds from database twice")
170
-
}
171
-
}
208
+
actor.continue(state)
172
209
}
173
210
174
-
// all starts go through this
175
-
// TODO: track started subjects and add StopFor
176
-
Started(Ok(_), _user) -> actor.continue(state)
177
-
178
211
Started(Error(start_error), feed) -> {
179
-
log(woof.Warning, "Failed to start fetcher once", [
212
+
log(woof.Error, "Failed to start fetcher", [
180
213
woof.field("feed", feed.link |> uri.to_string()),
181
214
woof.field("details", string.inspect(start_error)),
182
215
])
183
216
184
-
StartAgain(feed)
185
-
|> actor.send(state.self, _)
217
+
actor.stop_abnormal(
218
+
"Fetcher failed to start, i dont know how this would happen",
219
+
)
220
+
}
221
+
// runtime additions ----------------------------------------------------
222
+
StartFor(feed) -> {
223
+
let start_new = fn() {
224
+
let _ =
225
+
start_new(
226
+
factory,
227
+
Start(database:, registry:, feed:, manager: name),
228
+
)
229
+
Nil
230
+
}
231
+
232
+
case dict.get(state.fetchers, feed.id) {
233
+
Ok(#(pid, _)) -> {
234
+
case process.is_alive(pid) {
235
+
True -> Nil
236
+
False -> start_new()
237
+
}
238
+
}
239
+
Error(_) -> start_new()
240
+
}
241
+
186
242
actor.continue(state)
187
243
}
188
-
StartAgain(feed) -> {
189
-
case start_new(factory, Start(database:, registry:, feed:)) {
190
-
Ok(started) -> {
191
-
actor.send(state.self, Started(Ok(started), feed))
192
-
actor.continue(state)
244
+
StopFor(feed) -> {
245
+
case dict.get(state.fetchers, feed.id) {
246
+
Ok(#(pid, _)) -> {
247
+
case process.is_alive(pid) {
248
+
True -> process.send_exit(pid)
249
+
False -> Nil
250
+
}
193
251
}
194
-
Error(error) -> {
195
-
log(woof.Warning, "Failed to start fetcher twice", [
196
-
woof.field("feed", feed.link |> uri.to_string()),
197
-
woof.field("details", string.inspect(error)),
198
-
])
252
+
Error(_) -> Nil
253
+
}
254
+
255
+
actor.continue(state)
256
+
}
257
+
RestartFor(user) -> {
258
+
process.send(state.self, StopFor(user))
259
+
process.send(state.self, StartFor(user))
199
260
200
-
actor.stop_abnormal(
201
-
"Failed to start fetcher for feed "
202
-
<> feed.link |> uri.to_string()
203
-
<> " twice",
204
-
)
261
+
actor.continue(state)
262
+
}
263
+
RecheckFor(feed) -> {
264
+
case dict.get(state.fetchers, feed.id) {
265
+
Ok(#(pid, subject)) -> {
266
+
case process.is_alive(pid) {
267
+
True -> {
268
+
process.send(subject, CheckFeeds)
269
+
}
270
+
False -> Nil
271
+
}
205
272
}
273
+
Error(_) -> Nil
206
274
}
207
-
}
208
-
// runtime additions ----------------------------------------------------
209
-
StartFor(feed) -> {
210
-
start_new(factory, Start(database:, registry:, feed:))
211
-
|> Started(feed)
212
-
|> actor.send(state.self, _)
213
275
214
276
actor.continue(state)
215
277
}
···
240
302
feed: rss.Location,
241
303
registry: pubsub.Registry,
242
304
database: sqlight.Connection,
305
+
manager: ManagerName,
243
306
)
244
307
}
245
308
···
264
327
/// starts a new fetcher child
265
328
///
266
329
fn start(args: Start) -> Result(actor.Started(_), actor.StartError) {
267
-
let Start(feed:, registry:, database:) = args
330
+
let Start(feed:, registry:, database:, manager:) = args
268
331
269
332
log(woof.Info, "Starting", [woof.field("feed", feed.link |> uri.to_string())])
270
333
271
-
actor.new_with_initialiser(100, fn(subject) {
272
-
// TODO: persist the next timestamp to check at to the database
273
-
// so the actor restarting doesnt preeptively recheck the feed
274
-
let fetch_timer = process.send_after(subject, 150, CheckFeeds)
334
+
let started =
335
+
actor.new_with_initialiser(100, fn(subject) {
336
+
// TODO: persist the next timestamp to check at to the database
337
+
// so the actor restarting doesnt preeptively recheck the feed
338
+
let fetch_timer = process.send_after(subject, 150, CheckFeeds)
275
339
276
-
actor.initialised(State(
277
-
self: subject,
278
-
fetch_timer:,
279
-
feed:,
280
-
registry:,
281
-
database:,
282
-
))
283
-
|> actor.returning(Nil)
284
-
|> Ok
285
-
})
286
-
|> actor.on_message(on_message)
287
-
|> actor.start()
340
+
actor.initialised(State(
341
+
self: subject,
342
+
fetch_timer:,
343
+
feed:,
344
+
registry:,
345
+
database:,
346
+
))
347
+
|> actor.returning(subject)
348
+
|> Ok
349
+
})
350
+
|> actor.on_message(on_message)
351
+
|> actor.start()
352
+
353
+
// let the manager know we started
354
+
let manager = process.named_subject(manager)
355
+
process.send(manager, Started(started, feed))
356
+
357
+
started
288
358
}
289
359
290
360
/// handle messages
+196
-146
src/eater/sender.gleam
+196
-146
src/eater/sender.gleam
···
23
23
import eater/user
24
24
import gcourier/smtp as gsmtp
25
25
import gleam/bool
26
+
import gleam/dict
26
27
import gleam/erlang/process
27
28
import gleam/list
28
29
import gleam/otp/actor
···
34
35
import gleam/uri
35
36
import sqlight
36
37
import woof
38
+
import youid/uuid
37
39
38
40
/// log with module related structured data
39
41
///
···
55
57
56
58
pub fn factory(
57
59
name factory: FactoryName,
58
-
starter starter_name: StarterName,
60
+
starter starter_name: ManagerName,
59
61
registry registry: pubsub.Registry,
60
62
database database: sqlight.Connection,
61
63
smtp_environment smtp_environment: smtp.SmtpEnvironment,
···
66
68
|> factory.named(factory)
67
69
|> factory.supervised(),
68
70
)
69
-
|> static_supervisor.add(starter(
71
+
|> static_supervisor.add(manager(
70
72
starter_name,
71
73
factory,
72
74
registry,
···
78
80
79
81
// starting ---------------------------------------------------------------------
80
82
81
-
pub type StarterName =
82
-
process.Name(StarterMessage)
83
+
pub type ManagerName =
84
+
process.Name(ManagerMessage)
83
85
84
-
type StarterState {
85
-
StarterState(
86
-
self: process.Subject(StarterMessage),
86
+
type ManagerState {
87
+
ManagerState(
88
+
self: process.Subject(ManagerMessage),
87
89
factory: FactoryName,
88
90
registry: pubsub.Registry,
89
91
database: sqlight.Connection,
90
92
smtp_environment: smtp.SmtpEnvironment,
93
+
senders: dict.Dict(uuid.Uuid, process.Pid),
91
94
)
92
95
}
93
96
94
-
pub opaque type StarterMessage {
97
+
pub opaque type ManagerMessage {
95
98
// initial startup ------------------------------------------------------------
96
99
/// start the process
97
100
StartAll
98
-
/// fetch users from db (attempt 1)
99
-
DatabaseReturnedUsers(Result(List(user.User), sqlight.Error))
100
-
/// fetch users from db (attempt 2)
101
-
FetchAgain
102
-
103
-
/// start sender (attempt 1)
101
+
/// fetch users from db
102
+
DatabaseReturnedUsers(List(user.User))
103
+
/// start sender
104
104
Started(Result(actor.Started(Nil), actor.StartError), user.User)
105
-
/// start sender (attempt 2)
106
-
StartAgain(user.User)
107
105
108
106
// runtime additions ------------------------------------------------------------
109
107
/// start a sender for a given user at runtime
110
108
StartFor(user.User)
109
+
/// stop the sender for a given user at runtime
110
+
StopFor(user.User)
111
+
/// restart the sender for a given user at runtime
112
+
RestartFor(user.User)
111
113
}
112
114
113
-
pub fn start_for(starter: StarterName, user: user.User) {
114
-
process.named_subject(starter)
115
+
pub fn start_for(manager: ManagerName, user: user.User) {
116
+
process.named_subject(manager)
115
117
|> process.send(StartFor(user))
116
118
}
117
119
120
+
pub fn stop_for(manager: ManagerName, user: user.User) {
121
+
process.named_subject(manager)
122
+
|> process.send(StopFor(user))
123
+
}
124
+
125
+
pub fn restart_for(manager: ManagerName, user: user.User) {
126
+
process.named_subject(manager)
127
+
|> process.send(RestartFor(user))
128
+
}
129
+
118
130
/// handles starting all senders on startup and then shuts down
119
131
///
120
-
fn starter(
121
-
name: StarterName,
132
+
fn manager(
133
+
name: ManagerName,
122
134
factory: FactoryName,
123
135
registry: pubsub.Registry,
124
136
database: sqlight.Connection,
···
133
145
actor.new_with_initialiser(1000, fn(self) {
134
146
process.send_after(self, 100, StartAll)
135
147
136
-
actor.initialised(StarterState(
148
+
actor.initialised(ManagerState(
137
149
self:,
138
150
factory:,
139
151
registry:,
140
152
database:,
141
153
smtp_environment:,
154
+
senders: dict.new(),
142
155
))
143
156
|> Ok
144
157
})
···
146
159
case message {
147
160
// initial start --------------------------------------------------------
148
161
StartAll -> {
149
-
database.all_users(database)
150
-
|> DatabaseReturnedUsers
151
-
|> actor.send(state.self, _)
162
+
case database.all_users(database) {
163
+
Ok(users) -> {
164
+
DatabaseReturnedUsers(users)
165
+
|> actor.send(state.self, _)
166
+
167
+
actor.continue(state)
168
+
}
169
+
Error(_) -> actor.stop_abnormal("Failed to get users from database")
170
+
}
152
171
153
172
actor.continue(state)
154
173
}
155
-
DatabaseReturnedUsers(Ok(users)) -> {
174
+
DatabaseReturnedUsers(users) -> {
156
175
users
157
176
|> list.map(fn(user) {
158
177
start_new(
159
178
factory,
160
-
Start(database:, registry:, user:, smtp_environment:),
179
+
Start(
180
+
database:,
181
+
registry:,
182
+
user:,
183
+
smtp_environment:,
184
+
manager: name,
185
+
),
161
186
)
162
187
|> Started(user)
163
188
|> actor.send(state.self, _)
···
165
190
166
191
actor.continue(state)
167
192
}
168
-
DatabaseReturnedUsers(Error(db_error)) -> {
169
-
log(woof.Warning, "Failed to get users from database once", [
170
-
woof.field("details", string.inspect(db_error)),
171
-
])
193
+
// all starts go through this
194
+
Started(Ok(started), user) -> {
195
+
// if there is an existing sender for this
196
+
// tell it to shut down
197
+
case dict.get(state.senders, user.id) {
198
+
Ok(pid) ->
199
+
case process.is_alive(pid) {
200
+
True -> process.send_exit(pid)
201
+
False -> Nil
202
+
}
203
+
Error(_) -> Nil
204
+
}
172
205
173
-
actor.send(state.self, FetchAgain)
206
+
let state =
207
+
ManagerState(
208
+
..state,
209
+
senders: dict.insert(state.senders, user.id, started.pid),
210
+
)
211
+
174
212
actor.continue(state)
175
213
}
176
-
FetchAgain -> {
177
-
case database.all_users(database) {
178
-
Ok(users) -> {
179
-
DatabaseReturnedUsers(Ok(users))
180
-
|> actor.send(state.self, _)
181
-
182
-
actor.continue(state)
183
-
}
184
-
Error(error) -> {
185
-
log(woof.Warning, "Failed to get users from database twice", [
186
-
woof.field("details", string.inspect(error)),
187
-
])
188
-
189
-
actor.stop_abnormal("Failed to get users from database twice")
190
-
}
191
-
}
192
-
}
193
-
// all starts go through this
194
-
// TODO: track started subjects and add StopFor
195
-
Started(Ok(_), _user) -> actor.continue(state)
196
214
197
215
Started(Error(start_error), user) -> {
198
-
log(woof.Warning, "Failed to start sender once", [
216
+
log(woof.Error, "Failed to start sender", [
199
217
woof.field("user", user.email),
200
218
woof.field("details", string.inspect(start_error)),
201
219
])
202
220
203
-
StartAgain(user)
204
-
|> actor.send(state.self, _)
205
-
actor.continue(state)
221
+
actor.stop_abnormal(
222
+
"Sender failed to start, i dont know how this would happen",
223
+
)
206
224
}
207
-
StartAgain(user) -> {
208
-
case
209
-
start_new(
210
-
factory,
211
-
Start(database:, registry:, user:, smtp_environment:),
212
-
)
213
-
{
214
-
Ok(started) -> {
215
-
actor.send(state.self, Started(Ok(started), user))
216
-
actor.continue(state)
217
-
}
218
-
Error(error) -> {
219
-
log(woof.Warning, "Failed to start sender twice", [
220
-
woof.field("user", user.email),
221
-
woof.field("details", string.inspect(error)),
222
-
])
223
225
224
-
actor.stop_abnormal(
225
-
"Failed to start sender for user " <> user.email <> " twice",
226
+
// runtime additions ----------------------------------------------------
227
+
StartFor(user) -> {
228
+
let start_new = fn() {
229
+
let _ =
230
+
start_new(
231
+
factory,
232
+
Start(
233
+
database:,
234
+
registry:,
235
+
user:,
236
+
smtp_environment:,
237
+
manager: name,
238
+
),
226
239
)
240
+
Nil
241
+
}
242
+
243
+
case dict.get(state.senders, user.id) {
244
+
Ok(pid) -> {
245
+
case process.is_alive(pid) {
246
+
True -> Nil
247
+
False -> start_new()
248
+
}
227
249
}
250
+
Error(_) -> start_new()
228
251
}
252
+
253
+
actor.continue(state)
229
254
}
255
+
StopFor(user) -> {
256
+
case dict.get(state.senders, user.id) {
257
+
Ok(pid) -> {
258
+
case process.is_alive(pid) {
259
+
True -> process.send_exit(pid)
260
+
False -> Nil
261
+
}
262
+
}
263
+
Error(_) -> Nil
264
+
}
230
265
231
-
// runtime additions ----------------------------------------------------
232
-
StartFor(user) -> {
233
-
start_new(
234
-
factory,
235
-
Start(database:, registry:, user:, smtp_environment:),
236
-
)
237
-
|> Started(user)
238
-
|> actor.send(state.self, _)
266
+
actor.continue(state)
267
+
}
268
+
RestartFor(user) -> {
269
+
process.send(state.self, StopFor(user))
270
+
process.send(state.self, StartFor(user))
239
271
240
272
actor.continue(state)
241
273
}
···
252
284
registry: pubsub.Registry,
253
285
user: user.User,
254
286
smtp_environment: smtp.SmtpEnvironment,
287
+
manager: ManagerName,
255
288
)
256
289
}
257
290
258
291
/// start a new sender in the sender factory
259
292
///
260
-
/// make sure the user you are starting this sender for, is actually subscribed to some feeds
293
+
/// the sender will register itself with the manager
261
294
///
262
-
pub fn start_new(
263
-
factory_name: process.Name(_),
264
-
with: Start,
265
-
) -> Result(actor.Started(Nil), actor.StartError) {
295
+
fn start_new(factory factory_name: FactoryName, with with: Start) {
266
296
log(woof.Info, "Starting", [woof.field("user", with.user.email)])
297
+
267
298
let factory = factory.get_by_name(factory_name)
299
+
268
300
factory.start_child(factory, with)
269
301
}
270
302
···
286
318
// actor ------------------------------------------------------------------------
287
319
288
320
fn sender(args: Start) -> Result(actor.Started(Nil), actor.StartError) {
289
-
let initial_selector = fn(state: State) -> State {
290
-
log(woof.Info, "Initializing selector", [
291
-
woof.field("user", state.user.email),
292
-
])
293
-
294
-
let self = process.self()
295
-
296
-
let selector =
297
-
state.selector
298
-
|> pubsub.select_user(
299
-
user: state.user,
300
-
in: state.registry,
301
-
self:,
302
-
handler: PubSubMessage,
303
-
)
321
+
let Start(database:, registry:, user:, smtp_environment:, manager:) = args
304
322
305
-
let selector =
306
-
list.fold(state.feeds, selector, fn(selector, feed) {
307
-
log(woof.Info, "User is subscribed to feed", [
308
-
woof.field("user", state.user.email),
309
-
woof.field("feed", feed.link |> uri.to_string()),
310
-
])
311
-
312
-
pubsub.select_feed(
313
-
selector:,
314
-
feed:,
315
-
in: state.registry,
316
-
self:,
317
-
handler: PubSubMessage,
323
+
let actor_started =
324
+
actor.new_with_initialiser(300, fn(subject) {
325
+
let state =
326
+
State(
327
+
database:,
328
+
registry:,
329
+
user:,
330
+
feeds: [],
331
+
smtp_environment:,
332
+
self: subject,
333
+
sending_failed_n_times: 0,
334
+
selector: process.new_selector() |> process.select(subject),
318
335
)
319
-
})
320
336
321
-
State(..state, selector:)
322
-
}
337
+
actor.initialised(state)
338
+
|> actor.selecting(state.selector)
339
+
|> Ok
340
+
})
341
+
|> actor.on_message(on_message)
342
+
|> actor.start()
323
343
324
-
let Start(database:, registry:, user:, smtp_environment:) = args
344
+
// let the manager know we started
345
+
let manager = process.named_subject(manager)
346
+
process.send(manager, Started(actor_started, user))
325
347
326
-
actor.new_with_initialiser(100, fn(subject) {
327
-
use feeds <- result.try(
328
-
database.feeds_for_user(for: user, in: database)
329
-
|> result.map_error(fn(error) {
330
-
"Failed to get feeds from database with: " <> string.inspect(error)
331
-
}),
332
-
)
333
-
334
-
let state =
335
-
State(
336
-
database:,
337
-
registry:,
338
-
user:,
339
-
feeds:,
340
-
smtp_environment:,
341
-
self: subject,
342
-
sending_failed_n_times: 0,
343
-
selector: process.new_selector() |> process.select(subject),
344
-
)
345
-
|> initial_selector
346
-
347
-
actor.initialised(state)
348
-
|> actor.selecting(state.selector)
349
-
|> Ok
350
-
})
351
-
|> actor.on_message(on_message)
352
-
|> actor.start()
348
+
actor_started
353
349
}
354
350
355
351
// message ----------------------------------------------------------------------
356
352
357
353
type Message {
354
+
/// sent once at the start to fetch all feeds
355
+
///
356
+
GetFeeds
358
357
PubSubMessage(pubsub.Message)
359
358
Retry(message: pubsub.Message, attempt: Int)
360
359
}
···
374
373
)
375
374
376
375
case message {
376
+
GetFeeds -> {
377
+
case database.feeds_for_user(for: state.user, in: state.database) {
378
+
Ok(feeds) -> {
379
+
let state =
380
+
State(..state, feeds:)
381
+
|> initial_selector()
382
+
383
+
actor.continue(state)
384
+
|> actor.with_selector(state.selector)
385
+
}
386
+
Error(_) -> actor.stop_abnormal("Failed to get feeds from the database")
387
+
}
388
+
}
377
389
PubSubMessage(pubsub.FeedUpdate(update:)) -> {
378
390
let handled = handle_feed_update(state, update)
379
391
case handled {
···
462
474
}
463
475
}
464
476
477
+
/// initializes the selector from the supplied state
478
+
///
479
+
/// > !! should only be called once on startup !!
480
+
///
481
+
fn initial_selector(state: State) {
482
+
log(woof.Info, "Initializing selector", [
483
+
woof.field("user", state.user.email),
484
+
])
485
+
486
+
let self = process.self()
487
+
488
+
let selector =
489
+
state.selector
490
+
|> pubsub.select_user(
491
+
user: state.user,
492
+
in: state.registry,
493
+
self:,
494
+
handler: PubSubMessage,
495
+
)
496
+
497
+
let selector =
498
+
list.fold(state.feeds, selector, fn(selector, feed) {
499
+
log(woof.Info, "User is subscribed to feed", [
500
+
woof.field("user", state.user.email),
501
+
woof.field("feed", feed.link |> uri.to_string()),
502
+
])
503
+
504
+
pubsub.select_feed(
505
+
selector:,
506
+
feed:,
507
+
in: state.registry,
508
+
self:,
509
+
handler: PubSubMessage,
510
+
)
511
+
})
512
+
513
+
State(..state, selector:)
514
+
}
515
+
465
516
/// add a given feed
466
517
///
467
518
fn add_feed(state: State, feed: rss.Location) -> State {
···
551
602
smtp.feed_update_to_email(update, state.user)
552
603
|> smtp.send_message(state.smtp_environment)
553
604
}
554
-
// subscriptions changed --------------------------------------------------------
+2
-2
src/eater/sql.gleam
+2
-2
src/eater/sql.gleam
+1
-1
src/eater/sql/feeds.sql
+1
-1
src/eater/sql/feeds.sql
-2
src/eater/sql/subscriptions.sql
-2
src/eater/sql/subscriptions.sql
+1
-1
src/eater/sql/users.sql
+1
-1
src/eater/sql/users.sql
+15
-13
src/eater/ui/main_ui.gleam
+15
-13
src/eater/ui/main_ui.gleam
···
972
972
973
973
case pubsub.subscriber_count_user(user, registry) {
974
974
[] -> {
975
-
use dispatch <- effect.from
976
-
sender.start_new(
977
-
sender_factory,
978
-
sender.Start(database:, registry:, user:, smtp_environment:),
979
-
)
980
-
|> ServerStartedNewSender(user)
981
-
|> dispatch
975
+
// use dispatch <- effect.from
976
+
// sender.start_new(
977
+
// sender_factory,
978
+
// sender.Start(database:, registry:, user:, smtp_environment:),
979
+
// )
980
+
// |> ServerStartedNewSender(user)
981
+
// |> dispatch
982
+
todo
982
983
}
983
984
[_, ..] -> effect.none()
984
985
}
···
996
997
[] -> {
997
998
use dispatch <- effect.from
998
999
999
-
fetcher.start_new(
1000
-
fetcher_factory,
1001
-
fetcher.Start(feed:, registry:, database:),
1002
-
)
1003
-
|> ServerStartedNewFetcher(feed)
1004
-
|> dispatch
1000
+
todo as "this is nolonger needed with the new backend"
1001
+
// fetcher.start_new(
1002
+
// fetcher_factory,
1003
+
// fetcher.Start(feed:, registry:, database:),
1004
+
// )
1005
+
// |> ServerStartedNewFetcher(feed)
1006
+
// |> dispatch
1005
1007
}
1006
1008
[_, ..] -> effect.none()
1007
1009
}
+12
-10
test/eater_test.gleam
+12
-10
test/eater_test.gleam
···
6
6
import gleam/erlang/process
7
7
import gleam/int
8
8
import gleam/option
9
+
import gleam/otp/actor
10
+
import gleam/otp/factory_supervisor
9
11
import gleam/otp/static_supervisor
10
12
import gleam/string
11
13
import gleam/uri
···
124
126
|> static_supervisor.start()
125
127
as "failed to start supervisor"
126
128
127
-
let assert Ok(_) =
128
-
sender.start_new(
129
-
factory,
130
-
sender.Start(
131
-
database:,
132
-
registry:,
133
-
user:,
134
-
smtp_environment: test_smtp_env(),
135
-
),
136
-
)
129
+
// let assert Ok(_) =
130
+
// sender.start_new(
131
+
// factory,
132
+
// sender.Start(
133
+
// database:,
134
+
// registry:,
135
+
// user:,
136
+
// smtp_environment: test_smtp_env(),
137
+
// ),
138
+
// )
137
139
138
140
pubsub.publish_feed_update(feed_update("0"), feed, registry)
139
141