Antonio Mika
·
2026-10-10
1package feeds
2
3import (
4 "crypto/tls"
5 "errors"
6 "fmt"
7 html "html/template"
8 "io"
9 "log/slog"
10 "math"
11 "net/http"
12 "net/url"
13 "os"
14 "strings"
15 "text/template"
16 "time"
17
18 "github.com/adhocore/gronx"
19 "github.com/emersion/go-sasl"
20 "github.com/emersion/go-smtp"
21 "github.com/mmcdole/gofeed"
22 "github.com/picosh/pico/pkg/db"
23 "github.com/picosh/pico/pkg/shared"
24)
25
26var ErrNoRecentArticles = errors.New("no recent articles")
27
28type UserAgentTransport struct {
29 http.RoundTripper
30}
31
32func (c *UserAgentTransport) RoundTrip(r *http.Request) (*http.Response, error) {
33 userAgent := "linux:feeds:v2 (by /u/pico-sh)"
34 r.Header.Set("User-Agent", userAgent)
35 r.Header.Set("Accept", "*/*")
36 return c.RoundTripper.RoundTrip(r)
37}
38
39var httpClient = http.Client{
40 Transport: &UserAgentTransport{
41 &http.Transport{
42 TLSClientConfig: &tls.Config{},
43 },
44 },
45}
46
47type FeedItemTmpl struct {
48 GUID string
49 Title string
50 Link string
51 PublishedAt *time.Time
52 Content html.HTML
53 Description html.HTML
54}
55
56type Feed struct {
57 Title string
58 Link string
59 Description string
60 Items []*FeedItemTmpl
61 FeedItems []*gofeed.Item
62}
63
64type DigestFeed struct {
65 Feeds []*Feed
66 Options DigestOptions
67 KeepAliveURL string
68 UnsubURL string
69 DaysLeft string
70 ShowBanner bool
71 SizeWarning bool
72 IsPicoPlus bool
73}
74
75type DigestOptions struct {
76 InlineContent bool
77}
78
79func itemToTemplate(item *gofeed.Item) *FeedItemTmpl {
80 return &FeedItemTmpl{
81 Title: item.Title,
82 Link: item.Link,
83 PublishedAt: item.PublishedParsed,
84 Description: html.HTML(item.Description),
85 Content: html.HTML(item.Content),
86 }
87}
88
89func DigestIntervalToCron(interval string) string {
90 switch interval {
91 case "10min":
92 return "*/10 * * * *"
93 case "1hour":
94 return "0 * * * *"
95 case "6hour":
96 return "0 */6 * * *"
97 case "12hour":
98 return "0 */12 * * *"
99 case "1day", "":
100 return "0 13 * * *"
101 case "7day":
102 return "0 13 * * 0"
103 case "30day":
104 return "0 13 1 * *"
105 default:
106 return "0 13 * * *"
107 }
108}
109
110func getFeedItemID(logger *slog.Logger, item *gofeed.Item) string {
111 guid := strings.ToValidUTF8(item.GUID, "")
112 if item.GUID == "" {
113 logger.Info("no <guid> found for feed item, using <link> instead for its unique id")
114 return strings.ToValidUTF8(item.Link, "")
115 }
116 return guid
117}
118
119// see if this feed item should be emailed to user.
120func isValidItem(logger *slog.Logger, item *gofeed.Item, feedItems []*db.FeedItem) bool {
121 for _, feedItem := range feedItems {
122 if getFeedItemID(logger, item) == feedItem.GUID {
123 return false
124 }
125 }
126
127 return true
128}
129
130type Fetcher struct {
131 cfg *Config
132 db DB
133 auth sasl.Client
134 gron *gronx.Gronx
135 host string
136}
137
138func NewFetcher(dbpool DB, cfg *Config) *Fetcher {
139 host := os.Getenv("PICO_SMTP_HOST")
140 smtPass := os.Getenv("PICO_SMTP_PASS")
141 emailLogin := os.Getenv("PICO_SMTP_USER")
142
143 auth := sasl.NewPlainClient("", emailLogin, smtPass)
144 gron := gronx.New()
145 return &Fetcher{
146 db: dbpool,
147 cfg: cfg,
148 auth: auth,
149 gron: gron,
150 host: host,
151 }
152}
153
154func DateToMin(now time.Time) time.Time {
155 return time.Date(
156 now.Year(), now.Month(), now.Day(),
157 now.Hour(), now.Minute(),
158 0, 0, // zero out second and nano-second for cron
159 now.Location(),
160 )
161}
162
163func (f *Fetcher) Validate(logger *slog.Logger, post *db.Post, parsed *shared.ListParsedText, now time.Time) error {
164 expiresAt := post.ExpiresAt
165 if expiresAt != nil {
166 if post.ExpiresAt.Before(now) {
167 return fmt.Errorf("(%s) post has expired, skipping", post.ExpiresAt.Format(time.RFC3339))
168 }
169 }
170
171 cron := parsed.Cron
172 // support for posts with deprecated `digest_interval` property
173 if parsed.DigestInterval != "" {
174 cron = DigestIntervalToCron(parsed.DigestInterval)
175 }
176 logger.Info("found cron", "cron", cron)
177
178 if !f.gron.IsValid(cron) {
179 return fmt.Errorf("(%s) is invalid `cron`, skipping", cron)
180 }
181
182 dt := DateToMin(now)
183 isDue, err := f.gron.IsDue(cron, dt)
184 if err != nil {
185 return fmt.Errorf("cron error, skipping; err: %w", err)
186 }
187 if !isDue {
188 nextTime, _ := gronx.NextTickAfter(cron, dt, true)
189 return fmt.Errorf(
190 "cron not time to digest, skipping; cur run: %s, next run: %s",
191 f.gron.C.GetRef(),
192 nextTime,
193 )
194 }
195 return nil
196}
197
198func (f *Fetcher) RunPost(logger *slog.Logger, user *db.User, isPicoPlus bool, post *db.Post, skipValidation bool, now time.Time) error {
199 logger = logger.With("filename", post.Filename)
200 logger.Info("running feed post")
201
202 parsed := shared.ListParseText(post.Text)
203
204 if parsed.Email == "" {
205 logger.Error("post does not have an email associated, removing post")
206 err := f.db.RemovePosts([]string{post.ID})
207 if err != nil {
208 return err
209 }
210 }
211
212 if post.Data.LastDigest != nil {
213 logger.Info("last digest", "at", post.Data.LastDigest.Format(time.RFC3339))
214 } else {
215 logger.Info("first time running post")
216 }
217 err := f.Validate(logger, post, parsed, now)
218 if err != nil {
219 logger.Info("validation failed", "err", err)
220 if skipValidation {
221 logger.Info("overriding validation error, continuing")
222 } else {
223 return nil
224 }
225 }
226 logger.Info("validation success")
227
228 urls := []string{}
229 for _, item := range parsed.Items {
230 rawUrl := ""
231 if item.IsText {
232 rawUrl = string(item.Value)
233 } else if item.IsURL {
234 rawUrl = string(item.URL)
235 }
236
237 if rawUrl == "" {
238 continue
239 }
240
241 _, err := url.Parse(rawUrl)
242 if err != nil {
243 logger.Info("invalid url", "url", rawUrl)
244 continue
245 }
246
247 logger.Info("found rss feed url", "url", rawUrl)
248 urls = append(urls, rawUrl)
249 }
250
251 if post.ExpiresAt == nil {
252 expiresAt := now.AddDate(0, 12, 0)
253 post.ExpiresAt = &expiresAt
254 }
255 _, err = f.db.UpdatePost(post)
256 if err != nil {
257 return err
258 }
259
260 subject := fmt.Sprintf("%s feed digest", post.Filename)
261 unsubURL := getUnsubURL(post)
262
263 msgBody, err := f.FetchAll(logger, urls, parsed.InlineContent, user.Name, isPicoPlus, post)
264 if err != nil {
265 errForUser := err
266
267 // we don't want to increment in this case
268 if errors.Is(errForUser, ErrNoRecentArticles) {
269 return nil
270 }
271
272 post.Data.Attempts += 1
273 logger.Error("could not fetch urls", "err", err, "attempts", post.Data.Attempts)
274
275 maxAttempts := 100
276 errBody := fmt.Sprintf(`There was an error attempting to fetch your feeds (%d) times. After (%d) attempts we remove the file from our system. Please check all the URLs and re-upload.
277Also, we have centralized logs in our pico.sh TUI that will display realtime feed errors so you can debug.
278
279
280%s
281
282
283%s`, post.Data.Attempts, maxAttempts, errForUser.Error(), post.Text)
284 err = f.SendEmail(
285 logger,
286 user.Name,
287 parsed.Email,
288 subject,
289 unsubURL,
290 &MsgBody{Html: strings.ReplaceAll(errBody, "\n", "<br />"), Text: errBody},
291 )
292 if err != nil {
293 return err
294 }
295
296 if post.Data.Attempts >= maxAttempts {
297 err = f.db.RemovePosts([]string{post.ID})
298 if err != nil {
299 return err
300 }
301 } else {
302 _, err = f.db.UpdatePost(post)
303 if err != nil {
304 return err
305 }
306 }
307 return errForUser
308 } else {
309 post.Data.Attempts = 0
310 _, err := f.db.UpdatePost(post)
311 if err != nil {
312 return err
313 }
314 }
315
316 if msgBody != nil {
317 err = f.SendEmail(logger, user.Name, parsed.Email, subject, unsubURL, msgBody)
318 if err != nil {
319 return err
320 }
321 }
322
323 post.Data.LastDigest = &now
324 _, err = f.db.UpdatePost(post)
325 if err != nil {
326 return err
327 }
328
329 return nil
330}
331
332func (f *Fetcher) RunUser(user *db.User, now time.Time) error {
333 logger := shared.LoggerWithUser(f.cfg.Logger, user)
334 logger.Info("run user")
335 posts, err := f.db.FindPostsByUser(&db.Pager{Num: 100}, user.ID, "feeds")
336 if err != nil {
337 return err
338 }
339 isPicoPlus := db.HasPlusOrSpace(f.db, user.ID, "feeds")
340 if !isPicoPlus {
341 return fmt.Errorf("must have pico+ or feeds access to receive rss digests")
342 }
343
344 if len(posts.Data) > 0 {
345 logger.Info("found feed posts", "len", len(posts.Data))
346 }
347
348 for _, post := range posts.Data {
349 err = f.RunPost(logger, user, isPicoPlus, post, false, now)
350 if err != nil {
351 logger.Error("run post failed", "err", err)
352 }
353 }
354
355 return nil
356}
357
358func (f *Fetcher) ParseURL(fp *gofeed.Parser, url string) (*gofeed.Feed, error) {
359 req, err := http.NewRequest("GET", url, nil)
360 if err != nil {
361 return nil, err
362 }
363
364 resp, err := httpClient.Do(req)
365 if err != nil {
366 return nil, err
367 }
368
369 defer func() {
370 _ = resp.Body.Close()
371 }()
372 body, err := io.ReadAll(resp.Body)
373 if err != nil {
374 return nil, err
375 }
376
377 if resp.StatusCode < 200 || resp.StatusCode > 300 {
378 return nil, fmt.Errorf("fetching feed resulted in an error: %s %s", resp.Status, body)
379 }
380
381 feed, err := fp.ParseString(string(body))
382 if err != nil {
383 return nil, err
384 }
385
386 return feed, nil
387}
388
389func (f *Fetcher) Fetch(logger *slog.Logger, fp *gofeed.Parser, url string, username string, feedItems []*db.FeedItem) (*Feed, error) {
390 logger.Info("fetching feed", "url", url)
391
392 feed, err := f.ParseURL(fp, url)
393 if err != nil {
394 return nil, err
395 }
396
397 feedTmpl := &Feed{
398 Title: feed.Title,
399 Description: feed.Description,
400 Link: feed.Link,
401 }
402
403 items := []*FeedItemTmpl{}
404 gofeedItems := []*gofeed.Item{}
405 // we only want to return feed items published since the last digest time we fetched
406 for _, item := range feed.Items {
407 if item == nil {
408 continue
409 }
410
411 if !isValidItem(logger, item, feedItems) {
412 logger.Info("feed item already served", "guid", item.GUID)
413 continue
414 }
415
416 gofeedItems = append(gofeedItems, item)
417 items = append(items, itemToTemplate(item))
418 }
419
420 if len(items) == 0 {
421 return nil, fmt.Errorf(
422 "%s %w, skipping",
423 url,
424 ErrNoRecentArticles,
425 )
426 }
427
428 feedTmpl.FeedItems = gofeedItems
429 feedTmpl.Items = items
430 return feedTmpl, nil
431}
432
433func (f *Fetcher) PrintText(feedTmpl *DigestFeed) (string, error) {
434 ts, err := template.ParseFiles(
435 f.cfg.StaticPath("html/digest_text.page.tmpl"),
436 )
437
438 if err != nil {
439 return "", err
440 }
441
442 w := new(strings.Builder)
443 err = ts.Execute(w, feedTmpl)
444 if err != nil {
445 return "", err
446 }
447
448 return w.String(), nil
449}
450
451func (f *Fetcher) PrintHtml(feedTmpl *DigestFeed) (string, error) {
452 ts, err := html.ParseFiles(
453 f.cfg.StaticPath("html/digest.page.tmpl"),
454 )
455
456 if err != nil {
457 return "", err
458 }
459
460 w := new(strings.Builder)
461 err = ts.Execute(w, feedTmpl)
462 if err != nil {
463 return "", err
464 }
465
466 return w.String(), nil
467}
468
469type MsgBody struct {
470 Html string
471 Text string
472}
473
474func getUnsubURL(post *db.Post) string {
475 return fmt.Sprintf("https://feeds.pico.sh/unsub/%s", post.ID)
476}
477
478func (f *Fetcher) FetchAll(logger *slog.Logger, urls []string, inlineContent bool, username string, isPicoPlus bool, post *db.Post) (*MsgBody, error) {
479 logger.Info("fetching feeds", "inlineContent", inlineContent)
480 fp := gofeed.NewParser()
481 daysLeft := ""
482 showBanner := false
483 if post.ExpiresAt != nil {
484 diff := time.Until(*post.ExpiresAt)
485 daysLeftInt := int(math.Ceil(diff.Hours() / 24))
486 daysLeft = fmt.Sprintf("%d", daysLeftInt)
487 if daysLeftInt <= 30 {
488 showBanner = true
489 }
490 }
491 feeds := &DigestFeed{
492 KeepAliveURL: fmt.Sprintf("https://feeds.pico.sh/keep-alive/%s", post.ID),
493 UnsubURL: getUnsubURL(post),
494 DaysLeft: daysLeft,
495 ShowBanner: showBanner,
496 Options: DigestOptions{InlineContent: inlineContent},
497 IsPicoPlus: isPicoPlus,
498 }
499 feedItems, err := f.db.FindFeedItemsByPostID(post.ID)
500 if err != nil {
501 return nil, err
502 }
503
504 if len(urls) == 0 {
505 return nil, fmt.Errorf("feed file does not contain any urls")
506 }
507
508 var allErrors error
509 for _, url := range urls {
510 feedTmpl, err := f.Fetch(logger, fp, url, username, feedItems)
511 if err != nil {
512 if errors.Is(err, ErrNoRecentArticles) {
513 logger.Info("no recent articles", "err", err)
514 } else {
515 allErrors = errors.Join(allErrors, fmt.Errorf("%s: %w", url, err))
516 logger.Error("fetch error", "err", err)
517 }
518 continue
519 }
520 feeds.Feeds = append(feeds.Feeds, feedTmpl)
521 }
522
523 if len(feeds.Feeds) == 0 {
524 if allErrors != nil {
525 return nil, allErrors
526 }
527 return nil, fmt.Errorf("%w, skipping email", ErrNoRecentArticles)
528 }
529
530 fdi := []*db.FeedItem{}
531 for _, feed := range feeds.Feeds {
532 for _, item := range feed.FeedItems {
533 uid := getFeedItemID(logger, item)
534 fdi = append(fdi, &db.FeedItem{
535 PostID: post.ID,
536 GUID: uid,
537 Data: db.FeedItemData{
538 Title: strings.ToValidUTF8(item.Title, ""),
539 Description: strings.ToValidUTF8(item.Description, ""),
540 Content: strings.ToValidUTF8(item.Content, ""),
541 Link: item.Link,
542 PublishedAt: item.PublishedParsed,
543 },
544 })
545 }
546 }
547 err = f.db.InsertFeedItems(post.ID, fdi)
548 if err != nil {
549 return nil, err
550 }
551
552 text, err := f.PrintText(feeds)
553 if err != nil {
554 return nil, err
555 }
556
557 html, err := f.PrintHtml(feeds)
558 if err != nil {
559 return nil, err
560 }
561
562 // cap body size to prevent abuse
563 if len(html)+len(text) > 5*shared.MB {
564 feeds.Options.InlineContent = false
565 feeds.SizeWarning = true
566 html, err = f.PrintHtml(feeds)
567 if err != nil {
568 return nil, err
569 }
570 }
571
572 if allErrors != nil {
573 text = fmt.Sprintf("> %s\n\n%s", allErrors, text)
574 html = fmt.Sprintf("<blockquote>%s</blockquote><br /><br/>%s", allErrors, html)
575 }
576
577 return &MsgBody{
578 Text: text,
579 Html: html,
580 }, nil
581}
582
583func (f *Fetcher) SendEmail(logger *slog.Logger, username, email, subject, unsubURL string, msg *MsgBody) error {
584 if email == "" {
585 return fmt.Errorf("(%s) does not have an email associated with their feed post", username)
586 }
587 smtpAddr := f.host
588 fromEmail := "hello@pico.sh"
589 to := []string{email}
590 headers := map[string]string{
591 "From": fromEmail,
592 "Subject": subject,
593 "To": email,
594 "MIME-Version": "1.0",
595 "Content-Type": `multipart/alternative; boundary="boundary123"`,
596 "List-Unsubscribe": "<" + unsubURL + ">",
597 }
598 var content strings.Builder
599 for k, v := range headers {
600 fmt.Fprintf(&content, "%s: %s\r\n", k, v)
601 }
602 content.WriteString("\r\n")
603 content.WriteString("\r\n--boundary123\r\n")
604 content.WriteString("Content-Type: text/plain; charset=\"utf-8\"\r\n")
605 content.WriteString("\r\n" + msg.Text + "\r\n")
606 content.WriteString("--boundary123\r\n")
607 content.WriteString("Content-Type: text/html; charset=\"utf-8\"\r\n")
608 content.WriteString("\r\n" + msg.Html + "\r\n")
609 content.WriteString("--boundary123--")
610
611 reader := strings.NewReader(content.String())
612 logger.Info("sending email digest")
613 err := smtp.SendMail(
614 smtpAddr,
615 f.auth,
616 fromEmail,
617 to,
618 reader,
619 )
620 return err
621}
622
623func (f *Fetcher) Run(now time.Time) error {
624 users, err := f.db.FindUsersWithPost("feeds")
625 if err != nil {
626 return err
627 }
628
629 for _, user := range users {
630 err := f.RunUser(user, now)
631 if err != nil {
632 f.cfg.Logger.Error("run user failed", "err", err)
633 continue
634 }
635 }
636
637 return nil
638}
639
640func (f *Fetcher) Loop() {
641 logger := f.cfg.Logger
642 for {
643 logger.Info("running digest emailer")
644
645 err := f.Run(time.Now().UTC())
646 if err != nil {
647 logger.Error("run failed", "err", err)
648 }
649
650 logger.Info("digest emailer finished, waiting 1min ...")
651 time.Sleep(1 * time.Minute)
652 }
653}