feat(notify): bundled notifications, task watchers, daily summary

- Per-user routing (own mail / own Discord webhook) for task alarms (sent at once), shares, completed shared tasks and watched tasks; settings as JSON on the user (GET /auth/me, PATCH /auth/settings). - Everything but alarms is queued and sent as one message per user after NOTIFY_DIGEST_MINUTES (default 5); mails now multipart with an HTML version, Discord as one embed. Three failed attempts drop it. - Own changes never notify the actor, except on watched tasks when asked for; "paused" stops everything including alarms. - Task watchers: POST/DELETE /tasks/{id}/watch, isWatching on tasks. - Daily summary at a chosen time: overdue, today, tomorrow, plus the count of the next 7 days; nothing due, nothing sent. - Migration 7; jobs do not run in the demo. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-30 09:49:13 +01:00
1 parent 362dd3b4c6
commit a69bcbaf10
@@ -27,6 +27,7 @@ var migrations = []migration{
 	{4, "Sitzungen", (*Store).migrateSessions},
 	{5, "Medien", (*Store).migrateMedia},
 	{6, "Link-Vorschauen", (*Store).migrateLinkPreviews},
+	{7, "Benachrichtigungen", (*Store).migrateNotifications},
 }

 func (s *Store) migrate() error {
@@ -0,0 +1,253 @@
+package storage
+
+import (
+	"encoding/json"
+	"strings"
+	"time"
+
+	"github.com/sandrodach/mindsift/internal/domain"
+)
+
+// Notifications: per-user settings (JSON on the user), a queue that is sent bundled every few minutes,
+// task watchers and the bookkeeping for the daily summary.
+
+func (s *Store) migrateNotifications() error {
+	if err := s.addColumnIfMissing("users", "notification_settings", "TEXT NOT NULL DEFAULT ''"); err != nil {
+		return err
+	}
+	if err := s.addColumnIfMissing("users", "last_daily_summary", "TEXT NOT NULL DEFAULT ''"); err != nil {
+		return err
+	}
+	return s.execInTx(
+		stmt(`CREATE TABLE IF NOT EXISTS notification_queue (
+			id INTEGER PRIMARY KEY AUTOINCREMENT,
+			user_id TEXT NOT NULL,
+			kind TEXT NOT NULL,
+			payload TEXT NOT NULL,
+			created_at TEXT NOT NULL,
+			attempts INTEGER NOT NULL DEFAULT 0
+		)`),
+		stmt(`CREATE INDEX IF NOT EXISTS idx_notification_queue_user ON notification_queue(user_id, created_at)`),
+		stmt(`CREATE TABLE IF NOT EXISTS task_watchers (
+			task_id TEXT NOT NULL,
+			user_id TEXT NOT NULL,
+			created_at TEXT NOT NULL,
+			PRIMARY KEY (task_id, user_id)
+		)`),
+	)
+}
+
+// NotificationSettingsOf returns the user's settings, defaults filled in for users who never saved any.
+func (s *Store) NotificationSettingsOf(userID string) domain.NotificationSettings {
+	var raw, channels string
+	_ = s.db.QueryRow(`SELECT notification_settings, alarm_channels FROM users WHERE id = ?`, userID).Scan(&raw, &channels)
+	out := domain.DefaultNotificationSettings(parseAlarmChannels(channels))
+	if raw != "" {
+		_ = json.Unmarshal([]byte(raw), &out)
+	}
+	if !validClock(out.DailySummary.Time) {
+		out.DailySummary.Time = "07:00"
+	}
+	return out
+}
+
+func validClock(v string) bool {
+	t, err := time.Parse("15:04", v)
+	return err == nil && t.Format("15:04") == v
+}
+
+func (s *Store) SetNotificationSettings(userID string, n domain.NotificationSettings) error {
+	if !validClock(n.DailySummary.Time) {
+		n.DailySummary.Time = "07:00"
+	}
+	raw, _ := json.Marshal(n)
+	_, err := s.db.Exec(`UPDATE users SET notification_settings = ? WHERE id = ?`, string(raw), userID)
+	return err
+}
+
+// QueuedNotification is one event waiting to be sent bundled.
+type QueuedNotification struct {
+	ID        int64
+	UserID    string
+	Kind      string
+	Payload   map[string]string
+	CreatedAt string
+	Attempts  int
+}
+
+func (s *Store) QueueNotification(userID, kind string, payload map[string]string) error {
+	raw, _ := json.Marshal(payload)
+	_, err := s.db.Exec(`INSERT INTO notification_queue (user_id, kind, payload, created_at) VALUES (?, ?, ?, ?)`,
+		userID, kind, string(raw), time.Now().UTC().Format(time.RFC3339))
+	return err
+}
+
+// DueNotificationUsers lists users whose oldest waiting event is older than the bundling window.
+func (s *Store) DueNotificationUsers(olderThan time.Time) ([]string, error) {
+	rows, err := s.db.Query(`SELECT user_id FROM notification_queue GROUP BY user_id HAVING MIN(created_at) <= ?`,
+		olderThan.UTC().Format(time.RFC3339))
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []string
+	for rows.Next() {
+		var id string
+		if rows.Scan(&id) == nil {
+			out = append(out, id)
+		}
+	}
+	return out, rows.Err()
+}
+
+func (s *Store) QueuedNotifications(userID string) ([]QueuedNotification, error) {
+	rows, err := s.db.Query(`SELECT id, user_id, kind, payload, created_at, attempts FROM notification_queue WHERE user_id = ? ORDER BY id`, userID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []QueuedNotification
+	for rows.Next() {
+		var q QueuedNotification
+		var raw string
+		if rows.Scan(&q.ID, &q.UserID, &q.Kind, &raw, &q.CreatedAt, &q.Attempts) == nil {
+			_ = json.Unmarshal([]byte(raw), &q.Payload)
+			out = append(out, q)
+		}
+	}
+	return out, rows.Err()
+}
+
+func (s *Store) DeleteQueuedNotifications(ids []int64) error {
+	for _, id := range ids {
+		if _, err := s.db.Exec(`DELETE FROM notification_queue WHERE id = ?`, id); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (s *Store) BumpQueuedNotifications(ids []int64) error {
+	for _, id := range ids {
+		if _, err := s.db.Exec(`UPDATE notification_queue SET attempts = attempts + 1 WHERE id = ?`, id); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// --- Task watchers ---
+
+func (s *Store) WatchTask(taskID, userID string, on bool) error {
+	if on {
+		_, err := s.db.Exec(`INSERT OR IGNORE INTO task_watchers (task_id, user_id, created_at) VALUES (?, ?, ?)`, taskID, userID, nowUTC())
+		return err
+	}
+	_, err := s.db.Exec(`DELETE FROM task_watchers WHERE task_id = ? AND user_id = ?`, taskID, userID)
+	return err
+}
+
+func (s *Store) TaskWatchers(taskID string) []string {
+	rows, err := s.db.Query(`SELECT user_id FROM task_watchers WHERE task_id = ?`, taskID)
+	if err != nil {
+		return nil
+	}
+	defer rows.Close()
+	var out []string
+	for rows.Next() {
+		var id string
+		if rows.Scan(&id) == nil {
+			out = append(out, id)
+		}
+	}
+	return out
+}
+
+// WatchedTaskIDs is the set of tasks the user watches (to mark them in task lists).
+func (s *Store) WatchedTaskIDs(userID string) map[string]bool {
+	out := map[string]bool{}
+	rows, err := s.db.Query(`SELECT task_id FROM task_watchers WHERE user_id = ?`, userID)
+	if err != nil {
+		return out
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var id string
+		if rows.Scan(&id) == nil {
+			out[id] = true
+		}
+	}
+	return out
+}
+
+func (s *Store) ForgetTaskWatchers(taskID string) {
+	_, _ = s.db.Exec(`DELETE FROM task_watchers WHERE task_id = ?`, taskID)
+}
+
+// --- Daily summary ---
+
+// SummaryTask is an open task with a due date, for the daily summary.
+type SummaryTask struct {
+	Title, DueDate, ListTitle string
+}
+
+// ActiveUserIDs lists all active accounts (the summary job checks their settings).
+func (s *Store) ActiveUserIDs() []string {
+	rows, err := s.db.Query(`SELECT id FROM users WHERE is_active = 1`)
+	if err != nil {
+		return nil
+	}
+	defer rows.Close()
+	var out []string
+	for rows.Next() {
+		var id string
+		if rows.Scan(&id) == nil {
+			out = append(out, id)
+		}
+	}
+	return out
+}
+
+// OpenTasksDueBefore lists the user's open tasks (own ones and those of shared task lists) due before
+// the given day (YYYY-MM-DD, exclusive), oldest first.
+func (s *Store) OpenTasksDueBefore(userID, beforeDay string) ([]SummaryTask, error) {
+	rows, err := s.db.Query(`SELECT t.content, t.due_date, COALESCE(l.title, '')
+		FROM thoughts t LEFT JOIN task_lists l ON l.id = t.task_list_id
+		WHERE (t.status = 'kept' OR (t.status = 'converted' AND t.converted_type = 'task'))
+		  AND t.is_completed = 0 AND t.due_date != '' AND substr(t.due_date, 1, 10) < ?
+		  AND (t.user_id = ? OR (t.task_list_id != '' AND t.task_list_id IN (SELECT task_list_id FROM task_list_shares WHERE target_type = 'user' AND target_id = ?)))
+		ORDER BY t.due_date`, beforeDay, userID, userID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []SummaryTask
+	for rows.Next() {
+		var t SummaryTask
+		if rows.Scan(&t.Title, &t.DueDate, &t.ListTitle) == nil {
+			t.Title = strings.TrimSpace(strings.SplitN(t.Title, "\n", 2)[0])
+			out = append(out, t)
+		}
+	}
+	return out, rows.Err()
+}
+
+// ClaimDailySummary marks today's summary as sent; false when it was already sent today.
+func (s *Store) ClaimDailySummary(userID, day string) bool {
+	res, err := s.db.Exec(`UPDATE users SET last_daily_summary = ? WHERE id = ? AND last_daily_summary != ?`, day, userID, day)
+	if err != nil {
+		return false
+	}
+	n, _ := res.RowsAffected()
+	return n == 1
+}
+
+// SharedFolderName is the display name of a shared workspace folder.
+func (s *Store) SharedFolderName(folderID string) string {
+	var name string
+	_ = s.db.QueryRow(`SELECT name FROM workspace_shared_folders WHERE id = ?`, folderID).Scan(&name)
+	if name == "" {
+		name = "Ordner"
+	}
+	return name
+}