mail based rss feed aggregator
shuffling some stuff around in the backend and fetcher; mainly readability tbh
+105
-121
+43
-64
src/eater/backend.gleam
+43
-64
src/eater/backend.gleam
···
188
188
use _ <- result.try(
189
189
database.delete_subscription(backend.database, user, feed)
190
190
|> result.map_error(fn(error) {
191
-
log(woof.Error, "Failed to add user to database", [
191
+
log(woof.Error, "Failed to delete subscription in database", [
192
192
woof.str("user-email", user.email),
193
193
woof.str("user-id", user.id |> uuid.to_string()),
194
+
woof.str("feed-url", feed.link |> uri.to_string()),
195
+
woof.str("feed-id", feed.id |> uuid.to_string()),
194
196
woof.str("details", string.inspect(error)),
195
197
])
196
198
}),
···
499
501
user: user.User,
500
502
feed: rss.Location,
501
503
) -> actor.Next(State, Message) {
502
-
// - notify sender
503
-
sender.remove_subscription(state.names.sender_manager, user, feed)
504
-
505
-
use subscribers <- try_twice(
506
-
fn() { database.subscription_count(feed, state.database) },
507
-
otherwise: log_and_stop(
504
+
// we failed to get the subscriber count
505
+
let subscriber_count_failed = fn(error) {
506
+
woof.log(
508
507
state.logger,
508
+
woof.Error,
509
509
"Failed get subscriber count from database",
510
510
[
511
-
woof.str("feed", feed.link |> uri.to_string()),
511
+
woof.str("feed-url", feed.link |> uri.to_string()),
512
+
woof.str("feed-id", feed.id |> uuid.to_string()),
513
+
woof.str("details", string.inspect(error)),
512
514
],
513
-
),
514
-
)
515
+
)
516
+
actor.stop_abnormal("Failed get subscriber count from database")
517
+
}
515
518
516
-
case subscribers {
517
-
// - if noone is subscribed
518
-
0 -> {
519
-
// - stop fetcher
520
-
fetcher.stop_for(state.names.fetcher_manager, feed)
521
-
// - remove feed from database
519
+
// noone is subscribed anymore, so we remove this feed from the database
520
+
// and stop the fetcher for it
521
+
let noone_subscribed = fn() {
522
+
// - stop fetcher
523
+
fetcher.stop_for(state.names.fetcher_manager, feed)
522
524
523
-
case database.delete_feed(state.database, feed) {
524
-
Ok(_) -> actor.continue(state)
525
-
Error(error) -> {
526
-
woof.log(
527
-
state.logger,
528
-
woof.Error,
529
-
"Failed to delete feed from database",
530
-
[
531
-
woof.str("feed-url", feed.link |> uri.to_string()),
532
-
woof.str("feed-id", feed.id |> uuid.to_string()),
533
-
woof.str("details", string.inspect(error)),
534
-
],
535
-
)
536
-
actor.stop_abnormal("Failed to delete feed from database")
537
-
}
525
+
// - remove feed from database
526
+
case database.delete_feed(state.database, feed) {
527
+
Ok(_) -> actor.continue(state)
528
+
Error(error) -> {
529
+
woof.log(
530
+
state.logger,
531
+
woof.Error,
532
+
"Failed to delete feed from database",
533
+
[
534
+
woof.str("feed-url", feed.link |> uri.to_string()),
535
+
woof.str("feed-id", feed.id |> uuid.to_string()),
536
+
woof.str("details", string.inspect(error)),
537
+
],
538
+
)
539
+
actor.stop_abnormal("Failed to delete feed from database")
538
540
}
539
541
}
540
-
_ -> actor.continue(state)
541
542
}
542
-
}
543
543
544
-
// helpers ----------------------------------------------------------------------
544
+
// - notify sender, through manager
545
+
sender.remove_subscription(state.names.sender_manager, user, feed)
545
546
546
-
/// returns a function that logs an error
547
-
/// to the supplied logger with the supplied message
548
-
/// and then returns `actor.stop_abnormal` with that same message
549
-
///
550
-
fn log_and_stop(
551
-
logger logger: woof.Logger,
552
-
message message: String,
553
-
fields fields: List(#(String, _)),
554
-
) -> fn(a) -> actor.Next(b, c) {
555
-
fn(error) {
556
-
woof.log(logger, woof.Error, message, [
557
-
woof.str("details", string.inspect(error)),
558
-
..fields
559
-
])
560
-
actor.stop_abnormal(message)
561
-
}
562
-
}
563
-
564
-
fn try_twice(
565
-
try try: fn() -> Result(a, b),
566
-
otherwise otherwise: fn(b) -> c,
567
-
continue continue: fn(a) -> c,
568
-
) -> c {
569
-
case try() {
570
-
Ok(value) -> continue(value)
571
-
Error(_) ->
572
-
case try() {
573
-
Ok(value) -> continue(value)
574
-
Error(error) -> otherwise(error)
575
-
}
547
+
// check subscription count for this feed
548
+
case database.subscription_count(feed, state.database) {
549
+
// remove it when noone is subscribed anymore
550
+
Ok(0) -> noone_subscribed()
551
+
// or continue otherwise
552
+
Ok(_) -> actor.continue(state)
553
+
// and stop stop abnormally when checking returned an error
554
+
Error(error) -> subscriber_count_failed(error)
576
555
}
577
556
}
+62
-57
src/eater/fetcher.gleam
+62
-57
src/eater/fetcher.gleam
···
400
400
401
401
// if we've failed to fetch the feed too many times
402
402
// stop fetching for this feed, until it gets started again
403
-
case state.feed |> rss.is_fetching_paused() {
404
-
True -> failed_too_often(state)
405
-
False -> {
406
-
let reset_timer_and_continue = fn(state: State) {
407
-
// cancel old timer in case we received another CheckFeeds message that wasnt sent by the timer
408
-
// from the Ui for example
409
-
process.cancel_timer(state.fetch_timer)
403
+
use <- bool.lazy_guard(
404
+
when: rss.is_fetching_paused(state.feed),
405
+
return: fn() { failed_too_often(state) },
406
+
)
410
407
411
-
// update the `next_check` timestamp in the `rss.Location`
412
-
let feed = rss.next_check_in(state.feed, duration_between_checks())
408
+
// only actually check the feed if there are people subscribed to it
409
+
case pubsub.subscriber_count_feed(state.feed, state.registry) {
410
+
[] -> {
411
+
log(woof.Warning, "Skipping", [
412
+
woof.str("feed-url", state.feed.link |> uri.to_string()),
413
+
woof.str("feed-id", state.feed.id |> uuid.to_string()),
414
+
woof.str("reason", "No sender is subscribed"),
415
+
])
416
+
reset_timer_and_continue(state)
417
+
}
413
418
414
-
// and save it to the database
415
-
case database.update_feed_next_check(state.database, feed) {
416
-
Error(_) ->
417
-
actor.stop_abnormal(
418
-
"Failed to persist new next_check for "
419
-
<> state.feed.id |> uuid.to_string()
420
-
<> " "
421
-
<> state.feed.link |> uri.to_string(),
422
-
)
423
-
// we managed to delete it
424
-
// lets notify the other
425
-
Ok(_) -> {
426
-
// check again after interval has elapsed
427
-
let fetch_timer =
428
-
process.send_after(
429
-
state.self,
430
-
duration_between_checks() |> duration.to_milliseconds(),
431
-
CheckFeed,
432
-
)
419
+
[_, ..] ->
420
+
case handle_feed_fetching(state) {
421
+
Ok(state) -> reset_timer_and_continue(state)
422
+
Error(stop) -> stop
423
+
}
424
+
}
425
+
}
426
+
}
427
+
}
428
+
429
+
/// resets the fetch timer, updates the next_feed_check in the database
430
+
/// and `actor.continue(state)`'s
431
+
///
432
+
/// if the database write fails it will stop abnormally
433
+
///
434
+
fn reset_timer_and_continue(state: State) {
435
+
// cancel old timer in case we received another CheckFeeds message that wasnt sent by the timer
436
+
// from the Ui for example
437
+
process.cancel_timer(state.fetch_timer)
433
438
434
-
log(woof.Info, "Next check is at", [
435
-
woof.str(
436
-
"timestamp",
437
-
feed.next_check
438
-
|> timestamp.to_rfc3339(calendar.local_offset()),
439
-
),
440
-
])
439
+
// update the `next_check` timestamp in the `rss.Location`
440
+
let feed = rss.next_check_in(state.feed, duration_between_checks())
441
441
442
-
actor.continue(State(..state, fetch_timer:, feed:))
443
-
}
444
-
}
445
-
}
442
+
// and save it to the database
443
+
case database.update_feed_next_check(state.database, feed) {
444
+
Error(_) ->
445
+
actor.stop_abnormal(
446
+
"Failed to persist new next_check for "
447
+
<> state.feed.id |> uuid.to_string()
448
+
<> " "
449
+
<> state.feed.link |> uri.to_string(),
450
+
)
451
+
// we managed to delete it
452
+
// lets notify the other
453
+
Ok(_) -> {
454
+
// check again after interval has elapsed
455
+
let fetch_timer =
456
+
process.send_after(
457
+
state.self,
458
+
duration_between_checks() |> duration.to_milliseconds(),
459
+
CheckFeed,
460
+
)
446
461
447
-
// only actually check the feed if there are people subscribed to it
448
-
case pubsub.subscriber_count_feed(state.feed, state.registry) {
449
-
[] -> {
450
-
log(woof.Warning, "Skipping", [
451
-
woof.str("feed-url", state.feed.link |> uri.to_string()),
452
-
woof.str("feed-id", state.feed.id |> uuid.to_string()),
453
-
woof.str("reason", "No sender is subscribed"),
454
-
])
455
-
reset_timer_and_continue(state)
456
-
}
462
+
log(woof.Info, "Next check is at", [
463
+
woof.str(
464
+
"timestamp",
465
+
feed.next_check
466
+
|> timestamp.to_rfc3339(calendar.local_offset()),
467
+
),
468
+
])
457
469
458
-
[_, ..] ->
459
-
case handle_feed_fetching(state) {
460
-
Ok(state) -> reset_timer_and_continue(state)
461
-
Error(stop) -> stop
462
-
}
463
-
}
464
-
}
465
-
}
470
+
actor.continue(State(..state, fetch_timer:, feed:))
466
471
}
467
472
}
468
473
}