accounts/notify.go (4830 bytes)
1 package main 2 3 import ( 4 "database/sql" 5 "errors" 6 "fmt" 7 "net/url" 8 "strings" 9 "time" 10 ) 11 12 // notify sends one email to each subscriber who has new posts. 13 // 14 // A post is new for a user if the service first saw it in the index at or 15 // after the time the user signed up. On the first run the posts table is 16 // empty, and notify records every post with first_seen 0, so that the 17 // posts that were on the site before the service started go to nobody. 18 // 19 // notify writes a row in deliveries for each new (user, post) before it 20 // sends the email, and sends only the posts whose rows it wrote. It also 21 // writes the row when the user does not want the post (email off, or all 22 // tags of the post muted), so a later change of settings does not send 23 // old posts. A second run, or a second process at the same time, never 24 // sends a post two times to one user. If the send fails, notify removes the rows, and the next 25 // run tries again. If the process stops between the write and the send, 26 // the user does not get that email. This design accepts a lost email to 27 // prevent a duplicate email. 28 func (a *app) notify() error { 29 posts, err := loadPosts(a.cfg.SiteDir) 30 if err != nil { 31 return err 32 } 33 firstSeen, err := recordPosts(a.db, posts, a.now().Unix()) 34 if err != nil { 35 return err 36 } 37 38 rows, err := a.db.Query(`SELECT id, email, created_at, email_on, unsub_token FROM users`) 39 if err != nil { 40 return err 41 } 42 var users []user 43 for rows.Next() { 44 var u user 45 if err := rows.Scan(&u.ID, &u.Email, &u.CreatedAt, &u.EmailOn, &u.UnsubToken); err != nil { 46 rows.Close() 47 return err 48 } 49 users = append(users, u) 50 } 51 rows.Close() 52 if err := rows.Err(); err != nil { 53 return err 54 } 55 56 var errs []error 57 for _, u := range users { 58 if err := a.notifyUser(u, posts, firstSeen); err != nil { 59 errs = append(errs, fmt.Errorf("user %d: %w", u.ID, err)) 60 } 61 } 62 return errors.Join(errs...) 63 } 64 65 // recordPosts adds the posts that the posts table does not have, and 66 // returns first_seen for each post in posts. 67 func recordPosts(db *sql.DB, posts []Post, now int64) (map[string]int64, error) { 68 tx, err := db.Begin() 69 if err != nil { 70 return nil, err 71 } 72 defer tx.Rollback() 73 var known int 74 if err := tx.QueryRow(`SELECT count(*) FROM posts`).Scan(&known); err != nil { 75 return nil, err 76 } 77 seen := now 78 if known == 0 { 79 seen = 0 80 } 81 firstSeen := map[string]int64{} 82 for _, p := range posts { 83 if _, err := tx.Exec(`INSERT OR IGNORE INTO posts (slug, first_seen) VALUES (?, ?)`, p.Slug, seen); err != nil { 84 return nil, err 85 } 86 var t int64 87 if err := tx.QueryRow(`SELECT first_seen FROM posts WHERE slug = ?`, p.Slug).Scan(&t); err != nil { 88 return nil, err 89 } 90 firstSeen[p.Slug] = t 91 } 92 return firstSeen, tx.Commit() 93 } 94 95 func (a *app) notifyUser(u user, posts []Post, firstSeen map[string]int64) error { 96 muted, err := mutedTags(a.db, u.ID) 97 if err != nil { 98 return err 99 } 100 var claimed, send []Post 101 for _, p := range posts { 102 if firstSeen[p.Slug] < u.CreatedAt { 103 continue 104 } 105 res, err := a.db.Exec(`INSERT OR IGNORE INTO deliveries (user_id, slug) VALUES (?, ?)`, u.ID, p.Slug) 106 if err != nil { 107 return err 108 } 109 if n, _ := res.RowsAffected(); n == 1 { 110 claimed = append(claimed, p) 111 if u.EmailOn && wanted(p, muted) { 112 send = append(send, p) 113 } 114 } 115 } 116 if len(send) == 0 { 117 return nil 118 } 119 if err := a.mail.Send(a.digest(u, send)); err != nil { 120 for _, p := range claimed { 121 // The next run sends the post again only if this row is gone. 122 if _, derr := a.db.Exec(`DELETE FROM deliveries WHERE user_id = ? AND slug = ?`, u.ID, p.Slug); derr != nil { 123 err = errors.Join(err, derr) 124 } 125 } 126 return err 127 } 128 return nil 129 } 130 131 func (a *app) unsubscribeURL(u user) string { 132 return a.cfg.BaseURL + "/account/unsubscribe?t=" + url.QueryEscape(u.UnsubToken) 133 } 134 135 func (a *app) digest(u user, posts []Post) message { 136 subject := fmt.Sprintf("%d new posts on Recently Written", len(posts)) 137 if len(posts) == 1 { 138 subject = "New post: " + posts[0].Title 139 } 140 var b strings.Builder 141 b.WriteString("New on Recently Written:\n\n") 142 for _, p := range posts { 143 fmt.Fprintf(&b, "%s\n%s/%s.html\n\n", p.Title, a.cfg.BaseURL, p.Slug) 144 } 145 fmt.Fprintf(&b, "-- \nChange your settings: %s/account/\nStop these emails: %s\n", 146 a.cfg.BaseURL, a.unsubscribeURL(u)) 147 b.WriteString("\nAlso from the author: SubRead, read along with an audiobook: https://subread.space/\n") 148 return message{To: u.Email, Subject: subject, Body: b.String(), Unsubscribe: a.unsubscribeURL(u)} 149 } 150 151 // notifyLoop runs notify now and then one time in each interval, until 152 // stop closes. 153 func (a *app) notifyLoop(interval time.Duration, stop <-chan struct{}) { 154 t := time.NewTicker(interval) 155 defer t.Stop() 156 for { 157 if err := a.notify(); err != nil { 158 a.log.Printf("notify: %v", err) 159 } 160 select { 161 case <-stop: 162 return 163 case <-t.C: 164 } 165 } 166 }