feat(notifications): add suppress-after-action notification-uri config option

Author: sentriz

cmd/internal/wrtagflag/wrtagflag.go

@@ -161,15 +161,28 @@ func (r researchLinkParser) String() string {
 type notificationsParser struct{ *notifications.Notifications }
 
 func (n *notificationsParser) Set(value string) error {
-	eventsRaw, uri, ok := strings.Cut(value, " ")
-	if !ok {
-		return errors.New("invalid notification uri format. expected eg \"ev1,ev2 uri\"")
+	parts := strings.Fields(value)
+	if len(parts) < 2 {
+		return errors.New("invalid notification uri format. expected eg \"ev1,ev2 uri [suppress-after-action 30s]\"")
+	}
+	eventsRaw, uri := parts[0], parts[1]
+
+	var suppressAfterAction time.Duration
+	if len(parts) >= 4 && parts[2] == "suppress-after-action" {
+		var err error
+		suppressAfterAction, err = time.ParseDuration(parts[3])
+		if err != nil {
+			return fmt.Errorf("parse suppress duration: %w", err)
+		}
 	}
+
 	var lineErrs []error
 	for ev := range strings.SplitSeq(eventsRaw, ",") {
-		ev, uri = strings.TrimSpace(ev), strings.TrimSpace(uri)
-		err := n.AddURI(ev, uri)
-		lineErrs = append(lineErrs, err)
+		ev = strings.TrimSpace(ev)
+		if err := n.AddDestination(ev, uri, suppressAfterAction); err != nil {
+			lineErrs = append(lineErrs, err)
+			continue
+		}
 	}
 	return errors.Join(lineErrs...)
 }
@@ -178,10 +191,14 @@ func (n notificationsParser) String() string {
 		return ""
 	}
 	var parts []string
-	n.Notifications.IterMappings(func(e string, uri string) {
-		url, _ := url.Parse(uri)
-		parts = append(parts, fmt.Sprintf("%s: %s://%s/...", e, url.Scheme, url.Host))
-	})
+	for ev, dest := range n.Notifications.Destinations() {
+		uri, _ := url.Parse(dest.URI)
+		part := fmt.Sprintf("%s: %s://%s/...", ev, uri.Scheme, uri.Host)
+		if dest.SuppressAfterAction > 0 {
+			part += fmt.Sprintf(" (suppress after action %s)", dest.SuppressAfterAction)
+		}
+		parts = append(parts, part)
+	}
 	return strings.Join(parts, ", ")
 }
 

cmd/wrtag/main.go

@@ -23,6 +23,7 @@ import (
 	"go.senan.xyz/wrtag/cmd/internal/wrtagflag"
 	"go.senan.xyz/wrtag/cmd/internal/wrtaglog"
 	"go.senan.xyz/wrtag/fileutil"
+	"go.senan.xyz/wrtag/notifications"
 	"go.senan.xyz/wrtag/researchlink"
 )
 
@@ -49,7 +50,7 @@ func main() {
 	wrtagflag.DefaultClient()
 	var (
 		cfg                 = wrtagflag.Config()
-		notifications       = wrtagflag.Notifications()
+		notifs              = wrtagflag.Notifications()
 		researchLinkQuerier = wrtagflag.ResearchLinks()
 	)
 	wrtagflag.Parse()
@@ -74,6 +75,8 @@ func main() {
 		)
 		flag.Parse(args)
 
+		ctx := notifications.RecordAction(context.Background())
+
 		var importCondition wrtag.ImportCondition
 		if *yes {
 			importCondition = wrtag.Always
@@ -91,7 +94,7 @@ func main() {
 			return
 		}
 
-		ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+		ctx, cancel := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM)
 		defer cancel()
 
 		op, err := wrtagflag.OperationByName(command, *dryRun)
@@ -115,6 +118,8 @@ func main() {
 		)
 		flag.Parse(args)
 
+		ctx := notifications.RecordAction(context.Background())
+
 		// walk the whole root dir by default, or some user provided dirs if provided
 		var dirs []string
 		if args := flag.Args(); len(args) > 0 {
@@ -132,7 +137,7 @@ func main() {
 			}
 		}
 
-		ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+		ctx, cancel := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM)
 		defer cancel()
 
 		start := time.Now()
@@ -148,10 +153,10 @@ func main() {
 		switch {
 		case stats.errors.Load() > 0:
 			slog.Error("sync finished", "took", took, "", &stats)
-			notifications.Sendf(ctx, notifSyncError, "sync finished in %v %v", took, &stats)
+			notifs.Sendf(ctx, notifSyncError, "sync finished in %v %v", took, &stats)
 		default:
 			slog.Info("sync finished", "took", took, "", &stats)
-			notifications.Sendf(ctx, notifSyncComplete, "sync finished in %v %v", took, &stats)
+			notifs.Sendf(ctx, notifSyncComplete, "sync finished in %v %v", took, &stats)
 		}
 
 	default:

cmd/wrtagweb/main.go

@@ -28,6 +28,7 @@ import (
 	"go.senan.xyz/wrtag"
 	wrtagflag "go.senan.xyz/wrtag/cmd/internal/wrtagflag"
 	"go.senan.xyz/wrtag/cmd/internal/wrtaglog"
+	"go.senan.xyz/wrtag/notifications"
 	"go.senan.xyz/wrtag/researchlink"
 
 	_ "github.com/ncruces/go-sqlite3/driver"
@@ -58,7 +59,7 @@ func main() {
 	wrtagflag.DefaultClient()
 	var (
 		cfg                 = wrtagflag.Config()
-		notifications       = wrtagflag.Notifications()
+		notifs              = wrtagflag.Notifications()
 		researchLinkQuerier = wrtagflag.ResearchLinks()
 		apiKey              = flag.String("web-api-key", "", "API key for web interface")
 		listenAddr          = flag.String("web-listen-addr", ":7373", "Listen address for web interface (optional)")
@@ -185,11 +186,20 @@ func main() {
 			return err
 		}
 
-		switch job.Status {
-		case StatusComplete:
-			go notifications.Send(context.WithoutCancel(ctx), notifComplete, jobNotificationMessage(*publicURL, job))
-		case StatusNeedsInput:
-			go notifications.Send(context.WithoutCancel(ctx), notifNeedsInput, jobNotificationMessage(*publicURL, job))
+		{
+			ctx := context.WithoutCancel(ctx)
+
+			// If we have no action time in the ctx, use the job's updated at
+			if job.UpdatedTime.Valid && notifications.ActionTime(ctx).IsZero() {
+				ctx = notifications.RecordActionTime(ctx, job.UpdatedTime.Time)
+			}
+
+			switch job.Status {
+			case StatusComplete:
+				go notifs.Send(ctx, notifComplete, jobNotificationMessage(*publicURL, job))
+			case StatusNeedsInput:
+				go notifs.Send(ctx, notifNeedsInput, jobNotificationMessage(*publicURL, job))
+			}
 		}
 
 		return nil
@@ -266,7 +276,9 @@ func main() {
 		w.WriteHeader(http.StatusOK)
 		rc.Flush()
 
-		for id := range sse.receive(r.Context(), 0) {
+		ctx := r.Context()
+
+		for id := range sse.receive(ctx, 0) {
 			fmt.Fprintf(w, "data: %d\n\n", id)
 			rc.Flush()
 		}
@@ -276,7 +288,10 @@ func main() {
 		search := r.URL.Query().Get("search")
 		filter := JobStatus(r.URL.Query().Get("filter"))
 		page, _ := strconv.Atoi(r.URL.Query().Get("page"))
-		jl, err := listJobs(r.Context(), filter, search, page)
+
+		ctx := r.Context()
+
+		jl, err := listJobs(ctx, filter, search, page)
 		if err != nil {
 			respErrf(w, http.StatusInternalServerError, "error listing jobs: %v", err)
 			return
@@ -301,8 +316,10 @@ func main() {
 		}
 		path = filepath.Clean(path)
 
+		ctx := r.Context()
+
 		var job Job
-		if err := sqlb.ScanRow(r.Context(), db, &job, "insert into jobs (source_path, operation, time) values (?, ?, ?) returning *", path, operationStr, time.Now()); err != nil {
+		if err := sqlb.ScanRow(ctx, db, &job, "insert into jobs (source_path, operation, time) values (?, ?, ?) returning *", path, operationStr, time.Now()); err != nil {
 			http.Error(w, fmt.Sprintf("error saving job: %v", err), http.StatusInternalServerError)
 			return
 		}
@@ -314,8 +331,11 @@ func main() {
 
 	mux.HandleFunc("GET /jobs/{id}", func(w http.ResponseWriter, r *http.Request) {
 		id, _ := strconv.Atoi(r.PathValue("id"))
+
+		ctx := r.Context()
+
 		var job Job
-		if err := sqlb.ScanRow(r.Context(), db, &job, "select * from jobs where id=?", id); err != nil {
+		if err := sqlb.ScanRow(ctx, db, &job, "select * from jobs where id=?", id); err != nil {
 			respErrf(w, http.StatusInternalServerError, "error getting job")
 			return
 		}
@@ -332,8 +352,10 @@ func main() {
 			useMBID = filepath.Base(useMBID) // accept release URL
 		}
 
+		ctx := r.Context()
+
 		var job Job
-		if err := sqlb.ScanRow(r.Context(), db, &job, "update jobs set confirm=?, use_mbid=?, status=?, updated_time=? where id=? and status<>? returning *", confirm, useMBID, StatusEnqueued, time.Now(), id, StatusInProgress); err != nil {
+		if err := sqlb.ScanRow(ctx, db, &job, "update jobs set confirm=?, use_mbid=?, status=?, updated_time=? where id=? and status<>? returning *", confirm, useMBID, StatusEnqueued, time.Now(), id, StatusInProgress); err != nil {
 			respErrf(w, http.StatusInternalServerError, "error getting job")
 			return
 		}
@@ -345,7 +367,10 @@ func main() {
 
 	mux.HandleFunc("DELETE /jobs/{id}", func(w http.ResponseWriter, r *http.Request) {
 		id, _ := strconv.Atoi(r.PathValue("id"))
-		if err := sqlb.Exec(r.Context(), db, "delete from jobs where id=? and status<>?", id, StatusInProgress); err != nil {
+
+		ctx := r.Context()
+
+		if err := sqlb.Exec(ctx, db, "delete from jobs where id=? and status<>?", id, StatusInProgress); err != nil {
 			respErrf(w, http.StatusInternalServerError, "error getting job")
 			return
 		}
@@ -384,7 +409,9 @@ func main() {
 	})
 
 	mux.HandleFunc("/{$}", func(w http.ResponseWriter, r *http.Request) {
-		jl, err := listJobs(r.Context(), "", "", 0)
+		ctx := r.Context()
+
+		jl, err := listJobs(ctx, "", "", 0)
 		if err != nil {
 			respErrf(w, http.StatusInternalServerError, "error listing jobs: %v", err)
 			return
@@ -417,7 +444,9 @@ func main() {
 		}
 		path = filepath.Clean(path)
 
-		if err := sqlb.Exec(r.Context(), db, "insert into jobs (source_path, operation, time) values (?, ?, ?)", path, operationStr, time.Now()); err != nil {
+		ctx := r.Context()
+
+		if err := sqlb.Exec(ctx, db, "insert into jobs (source_path, operation, time) values (?, ?, ?)", path, operationStr, time.Now()); err != nil {
 			http.Error(w, fmt.Sprintf("error saving job: %v", err), http.StatusInternalServerError)
 			return
 		}

config.example

@@ -36,8 +36,9 @@
 # add notification uri that are triggered by various events
 # see all available services here https://containrrr.dev/shoutrrr/v0.8/services/generic/
 # possible events are complete, needs-input, sync-complete, sync-error
+# optional "suppress-after-action" prevents notifications for specified duration after manual actions
 
-#notification-uri complete,needs-input,sync-error smtp://username:password@host:port/?from=from@example.com&to=recipient@example.com
+#notification-uri complete,needs-input,sync-error smtp://username:password@host:port/?from=from@example.com&to=recipient@example.com suppress-after-action 30s
 #notification-uri complete,sync-complete          generic+https://my.subsonic.com/rest/startScan.view?c=wrtag&v=1.16&u=user&p=password
 
 # set a list of files to keep when moving or copying

notifications/notifications.go

@@ -4,8 +4,11 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"iter"
 	"log/slog"
+	"maps"
 	"net/url"
+	"time"
 
 	"github.com/containrrr/shoutrrr"
 	shoutrrrtypes "github.com/containrrr/shoutrrr/pkg/types"
@@ -15,36 +18,66 @@ var (
 	ErrInvalidURI = errors.New("invalid URI")
 )
 
+type Destination struct {
+	URI                 string
+	SuppressAfterAction time.Duration // suppress this long after manual actions
+}
+
 type Notifications struct {
-	mappings map[string][]string
+	mappings map[string][]Destination
 }
 
-func (n *Notifications) AddURI(event string, uri string) error {
+func (n *Notifications) AddDestination(event string, uri string, suppressAfterAction time.Duration) error {
 	if n.mappings == nil {
-		n.mappings = map[string][]string{}
+		n.mappings = map[string][]Destination{}
 	}
 	if _, err := url.Parse(uri); err != nil {
 		return fmt.Errorf("parse uri: %w", err)
 	}
-	n.mappings[event] = append(n.mappings[event], uri)
+	n.mappings[event] = append(n.mappings[event], Destination{URI: uri, SuppressAfterAction: suppressAfterAction})
 	return nil
 }
 
-func (n *Notifications) IterMappings(f func(string, string)) {
-	for event, uris := range n.mappings {
-		for _, uri := range uris {
-			f(event, uri)
+func (n *Notifications) Destinations() iter.Seq2[string, Destination] {
+	mappings := maps.Clone(n.mappings)
+
+	return func(yield func(string, Destination) bool) {
+		for event, destinations := range mappings {
+			for _, destination := range destinations {
+				if !yield(event, destination) {
+					return
+				}
+			}
 		}
 	}
 }
+
 func (n *Notifications) Sendf(ctx context.Context, event string, f string, a ...any) {
 	n.Send(ctx, event, fmt.Sprintf(f, a...))
 }
 
 // Send a simple string for now, maybe later message could instead be a type which
 // implements a notifications.Bodyer or something so that notifiers can send rich notifications.
 func (n *Notifications) Send(ctx context.Context, event string, message string) {
-	uris := n.mappings[event]
+	destinations := n.mappings[event]
+	if len(destinations) == 0 {
+		return
+	}
+
+	var timeSinceAction time.Duration
+	if actionTime, ok := ctx.Value(actionKey{}).(time.Time); ok {
+		timeSinceAction = time.Since(actionTime)
+	}
+
+	var uris []string
+	for _, dest := range destinations {
+		if timeSinceAction == 0 || timeSinceAction >= dest.SuppressAfterAction {
+			uris = append(uris, dest.URI)
+		} else {
+			slog.DebugContext(ctx, "suppressing notification due to recent manual action",
+				"event", event, "suppress_after_action", dest.SuppressAfterAction, "since_action", timeSinceAction)
+		}
+	}
 	if len(uris) == 0 {
 		return
 	}
@@ -63,3 +96,22 @@ func (n *Notifications) Send(ctx context.Context, event string, message string)
 		return
 	}
 }
+
+type actionKey struct{}
+
+// RecordAction records the current time of a user action and returns a context which may
+// be used to suppres notifications later.
+func RecordAction(ctx context.Context) context.Context {
+	return RecordActionTime(ctx, time.Now())
+}
+
+// RecordActionTime records a specific time as a user action and returns a context which may
+// be used to suppress notifications later.
+func RecordActionTime(ctx context.Context, actionTime time.Time) context.Context {
+	return context.WithValue(ctx, actionKey{}, actionTime)
+}
+
+func ActionTime(ctx context.Context) time.Time {
+	t, _ := ctx.Value(actionKey{}).(time.Time)
+	return t
+}