Commit `29548d6557`

29548d6557f45c2b5a36a3570713ce06f8e020c5

parent: c9cf3e5403

Verified · cmc

cmc <hello@cleberg.net> · 2026-09-20 09:23 UTC

store: the push queue, swept like the mail queue

One row per device per notice, so a retry to one device does not
resend to another. EnqueuePush writes nothing when the account has
push off or no devices. [retention] push caps the table.

Ref #89

cmd/gitbayd/main.go +2 −2

		@@ -515,9 +515,9 @@ func sweep(ctx context.Context, st *store.Store, cfg config.Config) {
515	515	tick = d
516	516	}
517	517	}
518		audit, events, deliveries, mail := cfg.Retention.Durations()
	518	audit, events, deliveries, mail, push := cfg.Retention.Durations()
519	519	r := store.Retention{Audit: audit, Events: events,
520		WebhookDeliveries: deliveries, Mail: mail}
	520	WebhookDeliveries: deliveries, Mail: mail, Push: push}
521	521	t := time.NewTicker(tick)
522	522	defer t.Stop()
523	523	for {

internal/config/config.go +5 −3

		@@ -125,10 +125,12 @@ type Retention struct {
125	125	WebhookDeliveries string `toml:"webhook_deliveries"`
126	126	// Mail is the outbound queue: rows already sent or given up on.
127	127	Mail string `toml:"mail"`
	128	// Push is the outbound device queue: rows already sent or given up on.
	129	Push string `toml:"push"`
128	130	}
129	131
130		// Durations parses the four, mapping each to zero when unset or bad.
131		func (r Retention) Durations() (audit, events, deliveries, mail time.Duration) {
	132	// Durations parses the five, mapping each to zero when unset or bad.
	133	func (r Retention) Durations() (audit, events, deliveries, mail, push time.Duration) {
132	134	parse := func(s string) time.Duration {
133	135	d, err := time.ParseDuration(s)
134	136	if err != nil \|\| d < 0 {
		@@ -136,7 +138,7 @@ func (r Retention) Durations() (audit, events, deliveries, mail time.Duration) {
136	138	}
137	139	return d
138	140	}
139		return parse(r.Audit), parse(r.Events), parse(r.WebhookDeliveries), parse(r.Mail)
	141	return parse(r.Audit), parse(r.Events), parse(r.WebhookDeliveries), parse(r.Mail), parse(r.Push)
140	142	}
141	143
142	144	// LFS stores large-file objects content-addressed under Root (default

internal/store/push.go +78

		@@ -3,6 +3,7 @@ package store
3	3	import (
4	4	"database/sql"
5	5	"errors"
	6	"time"
6	7	)
7	8
8	9	// PushDevice is one Apple device an account has registered. Token is the
		@@ -90,3 +91,80 @@ func (s *Store) SetPushEnabled(userID int64, on bool) error {
90	91	_, err := s.DB.Exec("UPDATE users SET notify_push = ? WHERE id = ?", v, userID)
91	92	return err
92	93	}
	94
	95	// QueuedPush is one pending push, joined to the token it is bound for so
	96	// the drainer needs one query rather than two.
	97	type QueuedPush struct {
	98	ID int64
	99	DeviceID int64
	100	Token string
	101	Title string
	102	Body string
	103	Path string
	104	Attempts int
	105	}
	106
	107	// EnqueuePush writes one row per registered device, and nothing when the
	108	// account has push off or no devices — the same shape as
	109	// ActivityMailAddress returning "" when notify_mail is off. Mute, watch
	110	// and actor-exclusion are already settled by NotifyRecipients before a
	111	// caller reaches here.
	112	func (s *Store) EnqueuePush(userID int64, title, body, path string) error {
	113	on, err := s.PushEnabled(userID)
	114	if err != nil \|\| !on {
	115	return err
	116	}
	117	_, err = s.DB.Exec(`
	118	INSERT INTO push_queue (device_id, title, body, path)
	119	SELECT id, ?, ?, ? FROM push_devices WHERE user_id = ?`,
	120	title, body, path, userID)
	121	return err
	122	}
	123
	124	func (s *Store) DuePush(limit int) ([]QueuedPush, error) {
	125	rows, err := s.DB.Query(`
	126	SELECT q.id, q.device_id, d.token, q.title, q.body, q.path, q.attempts
	127	FROM push_queue q JOIN push_devices d ON d.id = q.device_id
	128	WHERE q.sent_at IS NULL AND q.failed_at IS NULL
	129	AND (q.next_attempt_at IS NULL OR q.next_attempt_at <= ?)
	130	ORDER BY q.id LIMIT ?`, fmtTime(time.Now()), limit)
	131	if err != nil {
	132	return nil, err
	133	}
	134	defer rows.Close()
	135	var out []QueuedPush
	136	for rows.Next() {
	137	var p QueuedPush
	138	if err := rows.Scan(&p.ID, &p.DeviceID, &p.Token, &p.Title, &p.Body, &p.Path, &p.Attempts); err != nil {
	139	return nil, err
	140	}
	141	out = append(out, p)
	142	}
	143	return out, rows.Err()
	144	}
	145
	146	func (s *Store) MarkPushSent(id int64) error {
	147	_, err := s.DB.Exec(
	148	"UPDATE push_queue SET sent_at = strftime('%Y-%m-%dT%H:%M:%fZ','now'), attempts = attempts + 1 WHERE id = ?", id)
	149	return err
	150	}
	151
	152	func (s Store) MarkPushFailed(id int64, errMsg string, nextAt time.Time) error {
	153	if nextAt == nil {
	154	_, err := s.DB.Exec(
	155	"UPDATE push_queue SET failed_at = strftime('%Y-%m-%dT%H:%M:%fZ','now'), attempts = attempts + 1, last_error = ? WHERE id = ?",
	156	errMsg, id)
	157	return err
	158	}
	159	_, err := s.DB.Exec(
	160	"UPDATE push_queue SET attempts = attempts + 1, last_error = ?, next_attempt_at = ? WHERE id = ?",
	161	errMsg, fmtTime(*nextAt), id)
	162	return err
	163	}
	164
	165	// DeletePushDeviceByToken drops a device Apple has told us is gone. The
	166	// queue rows cascade, so nothing is left retrying at a dead token.
	167	func (s *Store) DeletePushDeviceByToken(token string) error {
	168	_, err := s.DB.Exec("DELETE FROM push_devices WHERE token = ?", token)
	169	return err
	170	}

internal/store/push_test.go +95 −1

		@@ -1,6 +1,9 @@
1	1	package store
2	2
3		import "testing"
	3	import (
	4	"testing"
	5	"time"
	6	)
4	7
5	8	func pushFixture(t testing.T) Store {
6	9	t.Helper()
		@@ -92,3 +95,94 @@ func TestPushEnabledDefaultsOn(t *testing.T) {
92	95	t.Fatal("SetPushEnabled(false) did not stick")
93	96	}
94	97	}
	98
	99	func TestEnqueuePush(t *testing.T) {
	100	s := pushFixture(t)
	101	uid, err := s.CreateUser("alice", false)
	102	if err != nil {
	103	t.Fatal(err)
	104	}
	105	s.AddPushDevice(uid, "tok-a", "iphone")
	106	s.AddPushDevice(uid, "tok-b", "ipad")
	107
	108	// One row per device, so a retry to the phone does not resend to the
	109	// iPad.
	110	if err := s.EnqueuePush(uid, "krz/gitbay", "cmc opened issue #12", "krz/gitbay/issues/12"); err != nil {
	111	t.Fatalf("EnqueuePush: %v", err)
	112	}
	113	due, err := s.DuePush(20)
	114	if err != nil {
	115	t.Fatalf("DuePush: %v", err)
	116	}
	117	if len(due) != 2 {
	118	t.Fatalf("want a row per device, got %d", len(due))
	119	}
	120	if due[0].Token == "" \|\| due[0].Body != "cmc opened issue #12" {
	121	t.Fatalf("got %+v", due[0])
	122	}
	123
	124	// Sent rows stop being due.
	125	if err := s.MarkPushSent(due[0].ID); err != nil {
	126	t.Fatalf("MarkPushSent: %v", err)
	127	}
	128	if due, _ := s.DuePush(20); len(due) != 1 {
	129	t.Fatalf("sent row still due")
	130	}
	131
	132	// A failure with a next attempt in the future is not due yet.
	133	next := time.Now().Add(time.Hour)
	134	if err := s.MarkPushFailed(due[1].ID, "503", &next); err != nil {
	135	t.Fatalf("MarkPushFailed: %v", err)
	136	}
	137	if due, _ := s.DuePush(20); len(due) != 0 {
	138	t.Fatalf("backed-off row is due too early")
	139	}
	140	}
	141
	142	func TestEnqueuePushRespectsSettingAndDevices(t *testing.T) {
	143	s := pushFixture(t)
	144	uid, err := s.CreateUser("alice", false)
	145	if err != nil {
	146	t.Fatal(err)
	147	}
	148
	149	// No devices: nothing queued, no error.
	150	if err := s.EnqueuePush(uid, "t", "b", "p"); err != nil {
	151	t.Fatalf("EnqueuePush with no devices: %v", err)
	152	}
	153	if due, _ := s.DuePush(20); len(due) != 0 {
	154	t.Fatalf("queued for an account with no devices")
	155	}
	156
	157	// Setting off: nothing queued.
	158	s.AddPushDevice(uid, "tok-a", "iphone")
	159	s.SetPushEnabled(uid, false)
	160	if err := s.EnqueuePush(uid, "t", "b", "p"); err != nil {
	161	t.Fatalf("EnqueuePush with push off: %v", err)
	162	}
	163	if due, _ := s.DuePush(20); len(due) != 0 {
	164	t.Fatalf("queued with notify_push off")
	165	}
	166	}
	167
	168	func TestDeletePushDeviceByTokenTakesItsQueue(t *testing.T) {
	169	s := pushFixture(t)
	170	uid, err := s.CreateUser("alice", false)
	171	if err != nil {
	172	t.Fatal(err)
	173	}
	174	s.AddPushDevice(uid, "tok-a", "iphone")
	175	s.EnqueuePush(uid, "t", "b", "p")
	176
	177	if err := s.DeletePushDeviceByToken("tok-a"); err != nil {
	178	t.Fatalf("DeletePushDeviceByToken: %v", err)
	179	}
	180	if d, _ := s.PushDevices(uid); len(d) != 0 {
	181	t.Fatalf("device survived")
	182	}
	183	// push_queue.device_id is ON DELETE CASCADE, so the queued rows go
	184	// with it rather than being retried at a dead token forever.
	185	if due, _ := s.DuePush(20); len(due) != 0 {
	186	t.Fatalf("queued rows outlived their device")
	187	}
	188	}

internal/store/retention.go +2

		@@ -32,6 +32,7 @@ type Retention struct {
32	32	Events time.Duration
33	33	WebhookDeliveries time.Duration
34	34	Mail time.Duration
	35	Push time.Duration
35	36	}
36	37
37	38	// Sweep deletes expired sessions and tokens, then the rows older than
		@@ -79,6 +80,7 @@ func (s *Store) Sweep(r Retention, now time.Time) (Swept, error) {
79	80	SELECT 1 FROM webhook_deliveries d
80	81	WHERE d.event_id = events.id AND d.delivered_at IS NULL AND d.failed_at IS NULL)`, r.Events},
81	82	{"notifications", "created_at < ? AND (sent_at IS NOT NULL OR failed_at IS NOT NULL)", r.Mail},
	83	{"push_queue", "created_at < ? AND (sent_at IS NOT NULL OR failed_at IS NOT NULL)", r.Push},
82	84	}
83	85	for _, a := range aged {
84	86	if a.keep <= 0 {

Commit 29548d6557

Commit `29548d6557`