status updates: switch to per-contact-method behavior

engine/message/db.go

@@ -190,15 +190,14 @@ func NewDB(ctx context.Context, db *sql.DB, a *alertlog.Store, pausable lifecycl
 		`),
 		cleanupStatusUpdateOptOut: p.P(`
 			delete from outgoing_messages msg
-			using users usr
+			using user_contact_methods cm
 			where
 				msg.message_type = 'alert_status_update' and
 				(
 					msg.last_status = 'pending' or
 					(msg.last_status = 'failed' and msg.next_retry_at notnull)
 				) and
-				usr.alert_status_log_contact_method_id isnull and
-				usr.id = msg.user_id
+				not cm.enable_status_updates and cm.id = msg.contact_method_id
 		`),
 		setSending: p.P(`
 			update outgoing_messages

engine/statusupdatemanager/db.go

@@ -19,8 +19,7 @@ type DB struct {
 	deleteSub      *sql.Stmt
 	cmWantsUpdates *sql.Stmt
 
-	cmUnsub  *sql.Stmt
-	usrUnsub *sql.Stmt
+	cmUnsub *sql.Stmt
 }
 
 // Name returns the name of the module.
@@ -30,7 +29,7 @@ func (db *DB) Name() string { return "Engine.StatusUpdateManager" }
 func NewDB(ctx context.Context, db *sql.DB) (*DB, error) {
 	lock, err := processinglock.NewLock(ctx, db, processinglock.Config{
 		Type:    processinglock.TypeStatusUpdate,
-		Version: 3,
+		Version: 4,
 	})
 	if err != nil {
 		return nil, err
@@ -41,32 +40,24 @@ func NewDB(ctx context.Context, db *sql.DB) (*DB, error) {
 		lock: lock,
 
 		cmUnsub: p.P(`
-			delete from alert_status_subscriptions where id in (
-				select stat.id from alert_status_subscriptions stat
-				where stat.contact_method_id notnull and exists (
-					select true from user_contact_methods cm where cm.id = stat.contact_method_id and cm.disabled
-				)
-				limit 100
-				for update skip locked
+			with _update as (
+				UPDATE user_contact_methods
+				SET enable_status_updates = TRUE
+				WHERE TYPE = 'SLACK_DM'
+					AND NOT enable_status_updates
 			)
-		`),
-
-		usrUnsub: p.P(`
-			delete from alert_status_subscriptions where id in (
-				select stat.id from alert_status_subscriptions stat
-				where stat.contact_method_id notnull and not exists (
-					select true from users u where u.alert_status_log_contact_method_id = stat.contact_method_id
+			delete from alert_status_subscriptions sub
+			using user_contact_methods cm
+			where
+				sub.contact_method_id = cm.id and (
+					cm.disabled or not cm.enable_status_updates
 				)
-				limit 100
-				for update skip locked
-			)
 		`),
 
 		cmWantsUpdates: p.P(`
-			select u.id
-			from user_contact_methods cm
-			join users u on u.id = cm.user_id and u.alert_status_log_contact_method_id = $1
-			where cm.id = $1 and not cm.disabled
+			select user_id, type
+			from user_contact_methods
+			where id = $1 and not disabled and enable_status_updates
 		`),
 
 		needsUpdate: p.P(`

engine/statusupdatemanager/update.go

@@ -11,6 +11,7 @@ import (
 	"github.com/target/goalert/alert/alertlog"
 	"github.com/target/goalert/engine/processinglock"
 	"github.com/target/goalert/permission"
+	"github.com/target/goalert/user/contactmethod"
 	"github.com/target/goalert/util/log"
 	"github.com/target/goalert/util/sqlutil"
 )
@@ -30,12 +31,6 @@ func (db *DB) UpdateAll(ctx context.Context) error {
 		log.Log(ctx, fmt.Errorf("delete status subscriptions for disabled contact methods: %w", err))
 	}
 
-	_, err = db.lock.Exec(ctx, db.usrUnsub)
-	if err != nil && !errors.Is(err, processinglock.ErrNoLock) {
-		// okay to proceed
-		log.Log(ctx, fmt.Errorf("delete status subscriptions for disabled users: %w", err))
-	}
-
 	// process up to 100
 	for i := 0; i < 100; i++ {
 		err = db.update(ctx)
@@ -73,9 +68,10 @@ func (db *DB) update(ctx context.Context) error {
 
 	isSubscribed := chanID.Valid
 	var userID sql.NullString
+	var cmType contactmethod.Type
 	if cmID.Valid {
 		isSubscribed = true
-		err = tx.StmtContext(ctx, db.cmWantsUpdates).QueryRowContext(ctx, cmID).Scan(&userID)
+		err = tx.StmtContext(ctx, db.cmWantsUpdates).QueryRowContext(ctx, cmID).Scan(&userID, &cmType)
 		if errors.Is(err, sql.ErrNoRows) {
 			isSubscribed = false
 			err = nil
@@ -110,7 +106,7 @@ func (db *DB) update(ctx context.Context) error {
 		}
 
 		// Only insert message if the user is not the same as the log event user and we have a recent log entry.
-		if logID > 0 && (!userID.Valid || userID.String != logUserID.String) {
+		if cmType.StatusUpdatesAlways() || (logID > 0 && (!userID.Valid || userID.String != logUserID.String)) {
 			_, err = tx.StmtContext(ctx, db.insertMessage).ExecContext(ctx, uuid.New(), chanID, cmID, userID, alertID, logID)
 			if err != nil {
 				return fmt.Errorf("insert status update message for id=%d: %w", id, err)

graphql2/graphqlapp/contactmethod.go

@@ -35,6 +35,22 @@ func (a *ContactMethod) Value(ctx context.Context, obj *contactmethod.ContactMet
 	return webhook.MaskURLPass(u), nil
 }
 
+func (a *ContactMethod) StatusUpdates(ctx context.Context, obj *contactmethod.ContactMethod) (graphql2.StatusUpdateState, error) {
+	if obj.Type.StatusUpdatesAlways() {
+		return graphql2.StatusUpdateStateEnabledForced, nil
+	}
+
+	if obj.Type.StatusUpdatesNever() {
+		return graphql2.StatusUpdateStateDisabledForced, nil
+	}
+
+	if obj.StatusUpdates {
+		return graphql2.StatusUpdateStateEnabled, nil
+	}
+
+	return graphql2.StatusUpdateStateDisabled, nil
+}
+
 func (a *ContactMethod) FormattedValue(ctx context.Context, obj *contactmethod.ContactMethod) (string, error) {
 	return a.FormatDestFunc(ctx, notification.ScannableDestType{CM: obj.Type}.DestType(), obj.Value), nil
 }
@@ -145,6 +161,9 @@ func (m *Mutation) UpdateUserContactMethod(ctx context.Context, input graphql2.U
 		if input.Value != nil {
 			cm.Value = *input.Value
 		}
+		if input.EnableStatusUpdates != nil {
+			cm.StatusUpdates = *input.EnableStatusUpdates
+		}
 
 		return m.CMStore.UpdateTx(ctx, tx, cm)
 	})