cyndikator 0.2.2

A cli rss reader
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154

use crate::db::Database;
use crate::fetcher::Fetcher;
use ticker::Ticker;
use tracker::{Trackee, Tracker};

use chrono::Local;
use futures::{future::join_all, select, StreamExt};
use log::{debug, error, info, trace};
use std::time::Duration;
use url::Url;

use cyndikator_dispatch::{Dispatch, Event};
use perform::Invoker;

mod perform;
mod ticker;
mod tracker;

pub struct Daemon {
    db: Database,
    dispatch: Dispatch,
    tick: usize,
}

impl Daemon {
    pub fn new(db: Database, dispatch: Dispatch, tick: usize) -> Daemon {
        Daemon { db, dispatch, tick }
    }

    pub async fn run(mut self) -> eyre::Result<()> {
        let mut tracker = Tracker::default();
        let mut ticker = Ticker::new(Duration::from_secs(self.tick as u64));

        let mut fetches = Vec::new();
        for feed in self.db.tracking()? {
            info!("fetching {} ...", &feed.url);
            let url = match Url::parse(&feed.url) {
                Ok(url) => url,
                Err(_) => continue,
            };

            if feed.last_fetch.is_none() {
                fetches.push(self.fetch(url.clone()));
            }

            info!("tracking {}", &feed.title);

            tracker.track(Trackee {
                url,
                last: feed.last_fetch.unwrap_or_else(Local::now),
                ttl: feed.ttl.unwrap_or(60),
            })
        }

        let (events, errs) = sep(join_all(fetches).await);
        for (url, err) in errs {
            error!("error while tracking feed '{}': {}", url, err);
        }

        self.dispatch(events).await;

        loop {
            select! {
                now = ticker.next() => {
                    trace!("tick");
                    let now = if let Some(now) = now { now } else { continue };
                    debug!("checking expired {}", now);

                    let expired = tracker.expired(&now);

                    let mut fetches = Vec::new();
                    for mut trackee in expired {
                        info!("fetching {}", trackee.url);
                        fetches.push(self.fetch(trackee.url.clone()));

                        trackee.fetched(&now);
                        tracker.track(trackee);
                    }

                    let (events, errs) = sep(join_all(fetches).await);
                    for (url, err) in errs {
                        error!("{} {}", err, url);
                    }


                    self.dispatch(events).await;
                }

            }
        }
    }

    async fn fetch(&self, url: Url) -> (Url, eyre::Result<Vec<Event>>) {
        let mut fetcher = Fetcher::new(&url);
        let events = fetcher.events().await;
        (url, events)
    }

    async fn dispatch(&mut self, feeds: Vec<(Url, Vec<Event>)>) {
        for (url, events) in feeds {
            let last_fetch = self.db.last_fetch(url.as_str()).ok();

            for event in events {
                match (&last_fetch, &event.date) {
                    (Some(lf), Some(pd)) if lf >= pd => {
                        debug!(
                            "skipping {} {}",
                            &event.feed_title.as_deref().unwrap_or("''"),
                            event.title.as_deref().unwrap_or("''")
                        );

                        continue;
                    }

                    _ => (),
                };

                debug!(
                    "dispatching event {} {} {} {}",
                    event.feed_title.as_deref().unwrap_or("''"),
                    event.title.as_deref().unwrap_or("''"),
                    event.feed_url,
                    event.url.as_deref().unwrap_or("''"),
                );

                let actions = self.dispatch.dispatch(&event);

                let mut invoker = Invoker::new(&mut self.db);
                for action in &actions {
                    invoker.invoke(action, &event);
                }
            }

            if let Err(err) = self.db.mark_clean(url.as_str()) {
                error!("failed to mark {} as clean: {}", &url, err);
            }
        }
    }
}

type ResSplit<S, T, E> = (Vec<(S, T)>, Vec<(S, E)>);
fn sep<S, T, E>(v: Vec<(S, Result<T, E>)>) -> ResSplit<S, T, E> {
    let mut ts = Vec::new();
    let mut es = Vec::new();

    for (s, e) in v.into_iter() {
        match e {
            Ok(t) => ts.push((s, t)),
            Err(e) => es.push((s, e)),
        }
    }

    (ts, es)
}