krz/gitbay

A CLI-first git forge. cli forge git self-hosted https://gitbay.org

Commit ada2c022c4

ada2c022c4701ad3d06e36ff79a54629ed9ff501

parent: d2d6505218

Verified · cmc

cmc <hello@cleberg.net> · 2026-08-24 15:57 UTC 

email notifications (#3)

Activity mail for issue and MR events: new issue/MR notifies the repo
owner (or org admins); comments, reviews, closes, reopens, and merges
notify participants (author, commenters, reviewers). The actor is
never notified; only verified primary addresses receive mail; no-op
without SMTP. Delivery rides a durable queue (migration 0009) drained
with the webhook retry discipline — bounded backoff, dead-letter —
so a flaky relay delays feedback instead of losing it.

Layout: unified · split

cmd/gitbayd/main.go +4
@@ -21,6 +21,7 @@ import (
21 "gitbay.org/gitbay/internal/config" 21 "gitbay.org/gitbay/internal/config"
22 "gitbay.org/gitbay/internal/control" 22 "gitbay.org/gitbay/internal/control"
23 "gitbay.org/gitbay/internal/mail" 23 "gitbay.org/gitbay/internal/mail"
24 "gitbay.org/gitbay/internal/notify"
24 "gitbay.org/gitbay/internal/gitd" 25 "gitbay.org/gitbay/internal/gitd"
25 "gitbay.org/gitbay/internal/hookd" 26 "gitbay.org/gitbay/internal/hookd"
26 "gitbay.org/gitbay/internal/httpd" 27 "gitbay.org/gitbay/internal/httpd"
@@ -133,6 +134,9 @@ func serveCmd() *cobra.Command {
133 whCtx, whCancel := context.WithCancel(context.Background()) 134 whCtx, whCancel := context.WithCancel(context.Background())
134 defer whCancel() 135 defer whCancel()
135 go webhook.New(st, cfg.Webhooks.AllowLocal, retryBase).Run(whCtx) 136 go webhook.New(st, cfg.Webhooks.AllowLocal, retryBase).Run(whCtx)
137 if cfg.Mail.SMTPHost != "" {
138 go notify.New(st, cfg, retryBase).Run(whCtx)
139 }
136 140
137 errCh := make(chan error, 3) 141 errCh := make(chan error, 3)
138 if cfg.SSH.Mode == "embedded" { 142 if cfg.SSH.Mode == "embedded" {
docs/users.org +8
@@ -166,6 +166,14 @@ Semantics worth knowing:
166 verified commits merge; everything server-created is refused with 166 verified commits merge; everything server-created is refused with
167 instructions to rebase locally. 167 instructions to rebase locally.
168 168
169* Notifications
170
171When the instance has SMTP configured, activity mails you: someone
172opens an issue or MR on your repository, comments where you are a
173participant (author, commenter, reviewer), reviews, closes, or merges.
174You are never mailed about your own actions, and only verified primary
175addresses receive anything. Delivery retries on relay failure.
176
169* Scripting 177* Scripting
170 178
171Every read command takes =--json= and emits one envelope: 179Every read command takes =--json= and emits one envelope:
e2e/notify_test.go added +165
@@ -0,0 +1,165 @@
1package e2e
2
3import (
4 "fmt"
5 "net"
6 "os"
7 "os/exec"
8 "strings"
9 "testing"
10 "time"
11)
12
13// mailTo returns captured messages addressed to one recipient.
14func (f *fakeSMTP) mailTo(recipient string) []string {
15 f.mu.Lock()
16 defer f.mu.Unlock()
17 var out []string
18 for _, m := range f.mail {
19 if strings.Contains(m, "To: "+recipient) {
20 out = append(out, m)
21 }
22 }
23 return out
24}
25
26func (f *fakeSMTP) waitFor(t *testing.T, recipient, substr string) string {
27 t.Helper()
28 deadline := time.Now().Add(15 * time.Second)
29 for time.Now().Before(deadline) {
30 for _, m := range f.mailTo(recipient) {
31 if strings.Contains(m, substr) {
32 return m
33 }
34 }
35 time.Sleep(100 * time.Millisecond)
36 }
37 t.Fatalf("no mail to %s containing %q", recipient, substr)
38 return ""
39}
40
41func TestActivityNotifications(t *testing.T) {
42 smtp := startFakeSMTP(t)
43 inst := startInstanceWith(t, fmt.Sprintf(
44 "[mail]\nsmtp_host = %q\nfrom = \"noreply@gitbay.test\"\n", smtp.addr))
45 // Fast retries for the mailer loop.
46 inst.proc.Process.Kill()
47 inst.proc.Wait()
48 inst.proc = exec.Command(inst.gitbayd, "--config", inst.config, "serve")
49 inst.proc.Env = append(os.Environ(), "GITBAY_WEBHOOK_RETRY_BASE=500ms")
50 inst.proc.Stderr = os.Stderr
51 if err := inst.proc.Start(); err != nil {
52 t.Fatal(err)
53 }
54 t.Cleanup(func() { inst.proc.Process.Kill(); inst.proc.Wait() })
55 deadline := time.Now().Add(10 * time.Second)
56 for {
57 conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", inst.port), 200*time.Millisecond)
58 if err == nil {
59 conn.Close()
60 break
61 }
62 if time.Now().After(deadline) {
63 t.Fatal("daemon did not restart")
64 }
65 time.Sleep(50 * time.Millisecond)
66 }
67
68 aliceKey := inst.newKey(t, "alice")
69 bobKey := inst.newKey(t, "bob")
70 eveKey := inst.newKey(t, "eve")
71 inst.admin(t, "admin", "user", "create", "alice",
72 "--key", aliceKey+".pub", "--email", "alice@example.test", "--verified")
73 inst.admin(t, "admin", "user", "create", "bob",
74 "--key", bobKey+".pub", "--email", "bob@example.test", "--verified")
75 // eve has NO verified email: she must be skipped silently.
76 inst.admin(t, "admin", "user", "create", "eve",
77 "--key", eveKey+".pub", "--email", "eve@example.test")
78
79 if _, errOut, code := inst.ssh(t, aliceKey, "", "repo", "create", "alice/app"); code != 0 {
80 t.Fatalf("repo create: %s", errOut)
81 }
82
83 // Bob opens an issue: the repo owner (alice) is notified with subject,
84 // excerpt, and link; bob (the actor) is not.
85 if _, errOut, code := inst.ssh(t, bobKey, "",
86 "issue", "create", "alice/app", "--title", "'it leaks'", "--body", "'memory climbs forever'"); code != 0 {
87 t.Fatalf("issue create: %s", errOut)
88 }
89 m := smtp.waitFor(t, "alice@example.test", "#1")
90 if !strings.Contains(m, "[alice/app] #1: it leaks") ||
91 !strings.Contains(m, "bob opened issue #1") ||
92 !strings.Contains(m, "memory climbs forever") ||
93 !strings.Contains(m, "/alice/app/issues/1") {
94 t.Fatalf("issue-open mail:\n%s", m)
95 }
96 if n := len(smtp.mailTo("bob@example.test")); n != 0 {
97 t.Fatalf("actor notified about own action (%d mails)", n)
98 }
99
100 // Alice comments: bob (author/participant) is notified; alice is not.
101 if _, _, code := inst.ssh(t, aliceKey, "", "issue", "comment", "alice/app", "1", "--message", "'on it'"); code != 0 {
102 t.Fatal("comment failed")
103 }
104 m = smtp.waitFor(t, "bob@example.test", "commented")
105 if !strings.Contains(m, "alice commented on #1") || !strings.Contains(m, "on it") {
106 t.Fatalf("comment mail:\n%s", m)
107 }
108
109 // Eve comments (unverified email, still allowed to act): both alice
110 // and bob get mail; eve never receives any.
111 if _, _, code := inst.ssh(t, eveKey, "", "issue", "comment", "alice/app", "1", "--message", "'same here'"); code != 0 {
112 t.Fatal("eve comment failed")
113 }
114 smtp.waitFor(t, "alice@example.test", "same here")
115 smtp.waitFor(t, "bob@example.test", "same here")
116
117 // Close notifies participants.
118 if _, _, code := inst.ssh(t, aliceKey, "", "issue", "close", "alice/app", "1"); code != 0 {
119 t.Fatal("close failed")
120 }
121 m = smtp.waitFor(t, "bob@example.test", "closed #1")
122 if !strings.Contains(m, "alice closed #1") {
123 t.Fatalf("close mail:\n%s", m)
124 }
125 if n := len(smtp.mailTo("eve@example.test")); n != 0 {
126 t.Fatalf("unverified recipient got mail (%d)", n)
127 }
128
129 // MR flow: bob opens (alice notified), alice reviews (bob notified),
130 // alice merges (bob notified).
131 if _, _, code := inst.ssh(t, aliceKey, "", "repo", "access", "grant", "alice/app", "bob", "write"); code != 0 {
132 t.Fatal("grant failed")
133 }
134 work := t.TempDir()
135 bobEnv := inst.gitEnv(bobKey)
136 mustGit(t, work, bobEnv, "clone", inst.sshURL("alice/app"), "w")
137 dir := work + "/w"
138 os.WriteFile(dir+"/f.txt", []byte("x\n"), 0o644)
139 mustGit(t, dir, bobEnv, "checkout", "-q", "-b", "main")
140 mustGit(t, dir, bobEnv, "add", ".")
141 mustGit(t, dir, bobEnv, "commit", "-q", "-m", "base")
142 mustGit(t, dir, bobEnv, "push", "-q", "origin", "main")
143 mustGit(t, dir, bobEnv, "checkout", "-q", "-b", "feat")
144 mustGit(t, dir, bobEnv, "commit", "-q", "--allow-empty", "-m", "work")
145 mustGit(t, dir, bobEnv, "push", "-q", "origin", "feat")
146 if _, errOut, code := inst.ssh(t, bobKey, "", "mr", "create", "alice/app",
147 "--source", "feat", "--target", "main", "--title", "'ship it'"); code != 0 {
148 t.Fatalf("mr create: %s", errOut)
149 }
150 m = smtp.waitFor(t, "alice@example.test", "!1")
151 if !strings.Contains(m, "bob opened merge request !1") {
152 t.Fatalf("mr-open mail:\n%s", m)
153 }
154 if _, _, code := inst.ssh(t, aliceKey, "", "mr", "review", "alice/app", "1", "--approve"); code != 0 {
155 t.Fatal("review failed")
156 }
157 smtp.waitFor(t, "bob@example.test", "reviewed !1: approve")
158 if _, errOut, code := inst.ssh(t, aliceKey, "", "mr", "merge", "alice/app", "1"); code != 0 {
159 t.Fatalf("merge: %s", errOut)
160 }
161 m = smtp.waitFor(t, "bob@example.test", "merged !1")
162 if !strings.Contains(m, "alice merged !1 into main") {
163 t.Fatalf("merge mail:\n%s", m)
164 }
165}
internal/control/issue.go +13
@@ -139,6 +139,10 @@ func runIssueCreate(c *Ctx, args []string) int {
139 return c.fail(protocol.ExitFailure, "%v", err) 139 return c.fail(protocol.ExitFailure, "%v", err)
140 } 140 }
141 c.Store.RecordEvent(repo.ID, c.User.ID, "issue.created", fmt.Sprintf(`{"number":%d}`, n)) 141 c.Store.RecordEvent(repo.ID, c.User.ID, "issue.created", fmt.Sprintf(`{"number":%d}`, n))
142 if targets, err := c.Store.RepoNotifyTargets(repo); err == nil {
143 notifyUsers(c, targets, issueSubject(repo, n, title),
144 notifyBody(c, fmt.Sprintf("opened issue #%d", n), b, fmt.Sprintf("%s/issues/%d", repo.Path(), n)))
145 }
142 return c.emit(map[string]any{"number": n}, func(w io.Writer) { 146 return c.emit(map[string]any{"number": n}, func(w io.Writer) {
143 fmt.Fprintf(w, "created %s#%d\n", repo.Path(), n) 147 fmt.Fprintf(w, "created %s#%d\n", repo.Path(), n)
144 }) 148 })
@@ -263,6 +267,10 @@ func runIssueComment(c *Ctx, args []string) int {
263 return c.fail(protocol.ExitFailure, "%v", err) 267 return c.fail(protocol.ExitFailure, "%v", err)
264 } 268 }
265 c.Store.RecordEvent(repo.ID, c.User.ID, "issue.commented", fmt.Sprintf(`{"number":%d}`, issue.Number)) 269 c.Store.RecordEvent(repo.ID, c.User.ID, "issue.commented", fmt.Sprintf(`{"number":%d}`, issue.Number))
270 if parts, err := c.Store.IssueParticipants(issue.ID); err == nil {
271 notifyUsers(c, parts, issueSubject(repo, issue.Number, issue.Title),
272 notifyBody(c, fmt.Sprintf("commented on #%d", issue.Number), body, fmt.Sprintf("%s/issues/%d", repo.Path(), issue.Number)))
273 }
266 return c.emit(map[string]any{"number": issue.Number}, func(w io.Writer) { 274 return c.emit(map[string]any{"number": issue.Number}, func(w io.Writer) {
267 fmt.Fprintf(w, "commented on %s#%d\n", repo.Path(), issue.Number) 275 fmt.Fprintf(w, "commented on %s#%d\n", repo.Path(), issue.Number)
268 }) 276 })
@@ -292,6 +300,11 @@ func setIssueState(c *Ctx, args []string, state string) int {
292 return c.fail(protocol.ExitFailure, "%v", err) 300 return c.fail(protocol.ExitFailure, "%v", err)
293 } 301 }
294 c.Store.RecordEvent(repo.ID, c.User.ID, "issue."+state, fmt.Sprintf(`{"number":%d}`, issue.Number)) 302 c.Store.RecordEvent(repo.ID, c.User.ID, "issue."+state, fmt.Sprintf(`{"number":%d}`, issue.Number))
303 if parts, err := c.Store.IssueParticipants(issue.ID); err == nil {
304 verb := map[string]string{"open": "reopened", "closed": "closed"}[state]
305 notifyUsers(c, parts, issueSubject(repo, issue.Number, issue.Title),
306 notifyBody(c, fmt.Sprintf("%s #%d", verb, issue.Number), "", fmt.Sprintf("%s/issues/%d", repo.Path(), issue.Number)))
307 }
295 return c.emit(map[string]any{"number": issue.Number, "state": state}, func(w io.Writer) { 308 return c.emit(map[string]any{"number": issue.Number, "state": state}, func(w io.Writer) {
296 fmt.Fprintf(w, "%s#%d is now %s\n", repo.Path(), issue.Number, state) 309 fmt.Fprintf(w, "%s#%d is now %s\n", repo.Path(), issue.Number, state)
297 }) 310 })
internal/control/mr.go +16
@@ -233,6 +233,10 @@ func runMRCreate(c *Ctx, args []string) int {
233 return c.fail(protocol.ExitFailure, "recording MR head: %v", err) 233 return c.fail(protocol.ExitFailure, "recording MR head: %v", err)
234 } 234 }
235 c.Store.RecordEvent(repo.ID, c.User.ID, "mr.created", fmt.Sprintf(`{"number":%d}`, n)) 235 c.Store.RecordEvent(repo.ID, c.User.ID, "mr.created", fmt.Sprintf(`{"number":%d}`, n))
236 if targets, err := c.Store.RepoNotifyTargets(repo); err == nil {
237 notifyUsers(c, targets, mrSubject(repo, n, title),
238 notifyBody(c, fmt.Sprintf("opened merge request !%d (%s -> %s)", n, source, target), b, fmt.Sprintf("%s/mrs/%d", repo.Path(), n)))
239 }
236 return c.emit(map[string]any{"number": n, "head_sha": headSHA}, func(w io.Writer) { 240 return c.emit(map[string]any{"number": n, "head_sha": headSHA}, func(w io.Writer) {
237 fmt.Fprintf(w, "created %s!%d (%s -> %s)\n", repo.Path(), n, source, target) 241 fmt.Fprintf(w, "created %s!%d (%s -> %s)\n", repo.Path(), n, source, target)
238 }) 242 })
@@ -441,6 +445,10 @@ func runMRComment(c *Ctx, args []string) int {
441 if err := c.Store.AddMRComment(mr.ID, c.User.ID, body); err != nil { 445 if err := c.Store.AddMRComment(mr.ID, c.User.ID, body); err != nil {
442 return c.fail(protocol.ExitFailure, "%v", err) 446 return c.fail(protocol.ExitFailure, "%v", err)
443 } 447 }
448 if parts, err := c.Store.MRParticipants(mr.ID); err == nil {
449 notifyUsers(c, parts, mrSubject(repo, mr.Number, mr.Title),
450 notifyBody(c, fmt.Sprintf("commented on !%d", mr.Number), body, fmt.Sprintf("%s/mrs/%d", repo.Path(), mr.Number)))
451 }
444 return c.emit(map[string]any{"number": mr.Number}, func(w io.Writer) { 452 return c.emit(map[string]any{"number": mr.Number}, func(w io.Writer) {
445 fmt.Fprintf(w, "commented on %s!%d\n", repo.Path(), mr.Number) 453 fmt.Fprintf(w, "commented on %s!%d\n", repo.Path(), mr.Number)
446 }) 454 })
@@ -474,6 +482,10 @@ func runMRReview(c *Ctx, args []string) int {
474 if err := c.Store.AddMRReview(mr.ID, c.User.ID, verdict, mr.HeadSHA); err != nil { 482 if err := c.Store.AddMRReview(mr.ID, c.User.ID, verdict, mr.HeadSHA); err != nil {
475 return c.fail(protocol.ExitFailure, "%v", err) 483 return c.fail(protocol.ExitFailure, "%v", err)
476 } 484 }
485 if parts, err := c.Store.MRParticipants(mr.ID); err == nil {
486 notifyUsers(c, parts, mrSubject(repo, mr.Number, mr.Title),
487 notifyBody(c, fmt.Sprintf("reviewed !%d: %s", mr.Number, verdict), "", fmt.Sprintf("%s/mrs/%d", repo.Path(), mr.Number)))
488 }
477 return c.emit(map[string]any{"number": mr.Number, "verdict": verdict}, func(w io.Writer) { 489 return c.emit(map[string]any{"number": mr.Number, "verdict": verdict}, func(w io.Writer) {
478 fmt.Fprintf(w, "reviewed %s!%d: %s\n", repo.Path(), mr.Number, verdict) 490 fmt.Fprintf(w, "reviewed %s!%d: %s\n", repo.Path(), mr.Number, verdict)
479 }) 491 })
@@ -735,6 +747,10 @@ func runMRMerge(c *Ctx, args []string) int {
735 return c.fail(protocol.ExitFailure, "%v", err) 747 return c.fail(protocol.ExitFailure, "%v", err)
736 } 748 }
737 c.Store.RecordEvent(repo.ID, c.User.ID, "mr.merged", fmt.Sprintf(`{"number":%d,"sha":%q}`, mr.Number, newSHA)) 749 c.Store.RecordEvent(repo.ID, c.User.ID, "mr.merged", fmt.Sprintf(`{"number":%d,"sha":%q}`, mr.Number, newSHA))
750 if parts, err := c.Store.MRParticipants(mr.ID); err == nil {
751 notifyUsers(c, parts, mrSubject(repo, mr.Number, mr.Title),
752 notifyBody(c, fmt.Sprintf("merged !%d into %s (%s)", mr.Number, mr.TargetRef, strategy), "", fmt.Sprintf("%s/mrs/%d", repo.Path(), mr.Number)))
753 }
738 return c.emit(map[string]any{"number": mr.Number, "strategy": strategy, "sha": newSHA}, func(w io.Writer) { 754 return c.emit(map[string]any{"number": mr.Number, "strategy": strategy, "sha": newSHA}, func(w io.Writer) {
739 fmt.Fprintf(w, "merged %s!%d into %s (%s) at %.10s\n", repo.Path(), mr.Number, mr.TargetRef, strategy, newSHA) 755 fmt.Fprintf(w, "merged %s!%d into %s (%s) at %.10s\n", repo.Path(), mr.Number, mr.TargetRef, strategy, newSHA)
740 }) 756 })
internal/control/notifications.go added +53
@@ -0,0 +1,53 @@
1package control
2
3import (
4 "fmt"
5 "strings"
6
7 "gitbay.org/gitbay/internal/store"
8)
9
10// notifyUsers enqueues activity mail for the given user ids, excluding the
11// acting user and anyone without a verified primary email. A best-effort
12// side channel: failures are ignored, the action itself already succeeded.
13// No-op when the instance has no SMTP.
14func notifyUsers(c *Ctx, userIDs []int64, subject, body string) {
15 if c.Cfg.Mail.SMTPHost == "" {
16 return
17 }
18 seen := map[int64]bool{c.User.ID: true}
19 for _, id := range userIDs {
20 if seen[id] {
21 continue
22 }
23 seen[id] = true
24 email, err := c.Store.PrimaryVerifiedEmail(id)
25 if err != nil || email == "" {
26 continue
27 }
28 c.Store.EnqueueMail(email, subject, body)
29 }
30}
31
32// notifyBody builds the standard notification body: who did what, an
33// excerpt, and the web link.
34func notifyBody(c *Ctx, action, excerpt, path string) string {
35 var b strings.Builder
36 fmt.Fprintf(&b, "%s %s\n", c.User.Username, action)
37 if e := strings.TrimSpace(excerpt); e != "" {
38 if len(e) > 500 {
39 e = e[:500] + "…"
40 }
41 fmt.Fprintf(&b, "\n%s\n", e)
42 }
43 fmt.Fprintf(&b, "\n%s/%s\n", strings.TrimSuffix(c.Cfg.Server.SiteURL, "/"), path)
44 return b.String()
45}
46
47func issueSubject(repo store.Repo, number int64, title string) string {
48 return fmt.Sprintf("[%s] #%d: %s", repo.Path(), number, title)
49}
50
51func mrSubject(repo store.Repo, number int64, title string) string {
52 return fmt.Sprintf("[%s] !%d: %s", repo.Path(), number, title)
53}
internal/notify/notify.go added +57
@@ -0,0 +1,57 @@
1// Package notify drains the notification queue: activity mail with the same
2// bounded-retry discipline as webhook delivery, so a flaky relay delays
3// feedback instead of losing it.
4package notify
5
6import (
7 "context"
8 "log/slog"
9 "time"
10
11 "gitbay.org/gitbay/internal/config"
12 "gitbay.org/gitbay/internal/mail"
13 "gitbay.org/gitbay/internal/store"
14)
15
16type Mailer struct {
17 St *store.Store
18 Cfg config.Config
19 RetryBase time.Duration
20 MaxAttempts int
21}
22
23func New(st *store.Store, cfg config.Config, retryBase time.Duration) *Mailer {
24 return &Mailer{St: st, Cfg: cfg, RetryBase: retryBase, MaxAttempts: 5}
25}
26
27// Run polls for due mail until ctx is done.
28func (m *Mailer) Run(ctx context.Context) {
29 tick := time.NewTicker(2 * time.Second)
30 defer tick.Stop()
31 for {
32 select {
33 case <-ctx.Done():
34 return
35 case <-tick.C:
36 due, err := m.St.DueMail(20)
37 if err != nil {
38 slog.Error("notify: listing due mail", "err", err)
39 continue
40 }
41 for _, q := range due {
42 if err := mail.Send(m.Cfg, q.Recipient, q.Subject, q.Body); err != nil {
43 attempt := q.Attempts + 1
44 if attempt >= m.MaxAttempts {
45 m.St.MarkMailFailed(q.ID, err.Error(), nil)
46 slog.Warn("notification dead-lettered", "recipient", q.Recipient, "err", err)
47 } else {
48 next := time.Now().Add(m.RetryBase << (attempt - 1))
49 m.St.MarkMailFailed(q.ID, err.Error(), &next)
50 }
51 continue
52 }
53 m.St.MarkMailSent(q.ID)
54 }
55 }
56 }
57}
internal/store/migrations/0009_notifications.down.sql added +1
@@ -0,0 +1 @@
1DROP TABLE notifications;
internal/store/migrations/0009_notifications.up.sql added +14
@@ -0,0 +1,14 @@
1CREATE TABLE notifications (
2 id INTEGER PRIMARY KEY,
3 recipient TEXT NOT NULL,
4 subject TEXT NOT NULL,
5 body TEXT NOT NULL,
6 attempts INTEGER NOT NULL DEFAULT 0,
7 next_attempt_at TEXT,
8 sent_at TEXT,
9 failed_at TEXT,
10 last_error TEXT,
11 created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
12);
13CREATE INDEX notifications_due ON notifications(next_attempt_at)
14 WHERE sent_at IS NULL AND failed_at IS NULL;
internal/store/notify.go added +102
@@ -0,0 +1,102 @@
1package store
2
3import "time"
4
5type QueuedMail struct {
6 ID int64
7 Recipient string
8 Subject string
9 Body string
10 Attempts int
11}
12
13func (s *Store) EnqueueMail(recipient, subject, body string) error {
14 _, err := s.DB.Exec(
15 "INSERT INTO notifications (recipient, subject, body) VALUES (?, ?, ?)",
16 recipient, subject, body)
17 return err
18}
19
20func (s *Store) DueMail(limit int) ([]QueuedMail, error) {
21 rows, err := s.DB.Query(`
22 SELECT id, recipient, subject, body, attempts FROM notifications
23 WHERE sent_at IS NULL AND failed_at IS NULL
24 AND (next_attempt_at IS NULL OR next_attempt_at <= ?)
25 ORDER BY id LIMIT ?`, fmtTime(time.Now()), limit)
26 if err != nil {
27 return nil, err
28 }
29 defer rows.Close()
30 var out []QueuedMail
31 for rows.Next() {
32 var m QueuedMail
33 if err := rows.Scan(&m.ID, &m.Recipient, &m.Subject, &m.Body, &m.Attempts); err != nil {
34 return nil, err
35 }
36 out = append(out, m)
37 }
38 return out, rows.Err()
39}
40
41func (s *Store) MarkMailSent(id int64) error {
42 _, err := s.DB.Exec(
43 "UPDATE notifications SET sent_at = strftime('%Y-%m-%dT%H:%M:%fZ','now'), attempts = attempts + 1 WHERE id = ?", id)
44 return err
45}
46
47func (s *Store) MarkMailFailed(id int64, errMsg string, nextAt *time.Time) error {
48 if nextAt == nil {
49 _, err := s.DB.Exec(
50 "UPDATE notifications SET failed_at = strftime('%Y-%m-%dT%H:%M:%fZ','now'), attempts = attempts + 1, last_error = ? WHERE id = ?",
51 errMsg, id)
52 return err
53 }
54 _, err := s.DB.Exec(
55 "UPDATE notifications SET attempts = attempts + 1, last_error = ?, next_attempt_at = ? WHERE id = ?",
56 errMsg, fmtTime(*nextAt), id)
57 return err
58}
59
60// IssueParticipants returns distinct user ids involved in an issue: the
61// author and every commenter.
62func (s *Store) IssueParticipants(issueID int64) ([]int64, error) {
63 return s.idQuery(`
64 SELECT author_id FROM issues WHERE id = ?
65 UNION SELECT author_id FROM issue_comments WHERE issue_id = ?`, issueID, issueID)
66}
67
68// MRParticipants returns distinct user ids involved in an MR: author,
69// commenters, reviewers.
70func (s *Store) MRParticipants(mrID int64) ([]int64, error) {
71 return s.idQuery(`
72 SELECT author_id FROM merge_requests WHERE id = ?
73 UNION SELECT author_id FROM mr_comments WHERE mr_id = ?
74 UNION SELECT reviewer_id FROM mr_reviews WHERE mr_id = ?`, mrID, mrID, mrID)
75}
76
77// RepoNotifyTargets returns who should hear about new activity on a repo:
78// the owning user, or every admin of the owning org.
79func (s *Store) RepoNotifyTargets(repo Repo) ([]int64, error) {
80 if repo.OwnerKind == "user" {
81 return []int64{repo.OwnerID}, nil
82 }
83 return s.idQuery(
84 "SELECT user_id FROM org_members WHERE org_id = ? AND role = 'admin'", repo.OwnerID)
85}
86
87func (s *Store) idQuery(q string, args ...any) ([]int64, error) {
88 rows, err := s.DB.Query(q, args...)
89 if err != nil {
90 return nil, err
91 }
92 defer rows.Close()
93 var out []int64
94 for rows.Next() {
95 var id int64
96 if err := rows.Scan(&id); err != nil {
97 return nil, err
98 }
99 out = append(out, id)
100 }
101 return out, rows.Err()
102}