fix(server): make notify config update transactional

server/internal/core/event.go

@@ -294,7 +294,7 @@ type eventBroker struct {
 	subscribe   chan eventSubscription
 	unsubscribe chan chan Event
 	send        chan Event
-	notifier    inotify.Notifier
+	notifier    *inotify.Notifier
 
 	cache *RingCache
 
@@ -813,5 +813,20 @@ var (
 )
 
 func (broker *eventBroker) InitService(config *configs.NotifyConfig) error {
+	if broker == nil || broker.notifier == nil {
+		return ErrEventBrokerUnavailable
+	}
 	return broker.notifier.InitService(config)
 }
+
+// PrepareNotifyConfig builds and validates the replacement notification
+// service for config without changing the running one. The returned commit
+// swaps it in atomically and cannot fail; callers persist the configuration
+// first and commit only after the write succeeds, so the configuration on
+// disk and the running service change together or not at all.
+func (broker *eventBroker) PrepareNotifyConfig(config *configs.NotifyConfig) (commit func(), err error) {
+	if broker == nil || broker.notifier == nil {
+		return nil, ErrEventBrokerUnavailable
+	}
+	return broker.notifier.Prepare(config)
+}

server/internal/core/notify_config_test.go

@@ -0,0 +1,35 @@
+package core
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/chainreactors/malice-network/server/internal/configs"
+	inotify "github.com/chainreactors/malice-network/server/internal/notify"
+)
+
+func TestPrepareNotifyConfigTransaction(t *testing.T) {
+	broker := &eventBroker{notifier: inotify.NewNotifier()}
+
+	if _, err := broker.PrepareNotifyConfig(&configs.NotifyConfig{
+		Enable:     true,
+		ServerChan: &configs.ServerChanConfig{Enable: true, URL: "not-a-url"},
+	}); err == nil {
+		t.Fatal("expected a validation error")
+	}
+
+	commit, err := broker.PrepareNotifyConfig(&configs.NotifyConfig{Enable: false})
+	if err != nil {
+		t.Fatalf("PrepareNotifyConfig failed: %v", err)
+	}
+	commit()
+
+	for _, unavailable := range []*eventBroker{nil, {}} {
+		if _, err := unavailable.PrepareNotifyConfig(&configs.NotifyConfig{}); !errors.Is(err, ErrEventBrokerUnavailable) {
+			t.Fatalf("unavailable broker prepare: %v", err)
+		}
+		if err := unavailable.InitService(&configs.NotifyConfig{}); !errors.Is(err, ErrEventBrokerUnavailable) {
+			t.Fatalf("unavailable broker init: %v", err)
+		}
+	}
+}

server/internal/notify/notify.go

@@ -2,8 +2,11 @@ package notify
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net/url"
+	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/chainreactors/logs"
@@ -15,39 +18,67 @@ import (
 	"github.com/nikoksr/notify/service/telegram"
 )
 
+// Notifier publishes complete, immutable service sets: a replacement is built
+// and validated in memory first, then atomically swapped in, so a failed
+// rebuild never touches the running services and a send never observes a
+// half-initialized set. A Notifier must not be copied after first use.
 type Notifier struct {
-	notify *notify.Notify
-	enable bool
+	active atomic.Pointer[notify.Notify]
 }
 
-func NewNotifier() Notifier {
-	return Notifier{notify: notify.New(), enable: false}
+func NewNotifier() *Notifier {
+	return &Notifier{}
 }
 
-func (n *Notifier) InitService(config *configs.NotifyConfig) error {
+// buildServices validates cfg and constructs a complete replacement service
+// set without touching the currently active one.
+func buildServices(config *configs.NotifyConfig) (*notify.Notify, error) {
+	next := notify.New()
 	if config == nil || !config.Enable {
-		return nil
+		next.Disabled = true
+		return next, nil
+	}
+	// Validate every enabled channel before constructing any provider, so a
+	// misconfiguration fails the whole update instead of partially building it.
+	if config.Telegram != nil && config.Telegram.Enable &&
+		(strings.TrimSpace(config.Telegram.APIKey) == "" || config.Telegram.ChatID == 0) {
+		return nil, errors.New("server.notify.telegram requires api_key and a nonzero chat_id")
+	}
+	if config.DingTalk != nil && config.DingTalk.Enable && strings.TrimSpace(config.DingTalk.Token) == "" {
+		return nil, errors.New("server.notify.dingtalk requires token")
+	}
+	if config.Lark != nil && config.Lark.Enable {
+		if err := validateWebhook("server.notify.lark.webhook_url", config.Lark.WebHookUrl); err != nil {
+			return nil, err
+		}
+	}
+	if config.ServerChan != nil && config.ServerChan.Enable {
+		if err := validateWebhook("server.notify.serverchan.url", config.ServerChan.URL); err != nil {
+			return nil, err
+		}
+	}
+	if config.PushPlus != nil && config.PushPlus.Enable && strings.TrimSpace(config.PushPlus.Token) == "" {
+		return nil, errors.New("server.notify.pushplus requires token")
 	}
-	n.enable = true
 	if config.Telegram != nil && config.Telegram.Enable {
 		tg, err := telegram.New(config.Telegram.APIKey)
 		if err != nil {
-			return err
+			return nil, err
 		}
 		tg.SetParseMode(telegram.ModeMarkdown)
 		tg.AddReceivers(config.Telegram.ChatID)
-		n.notify.UseServices(tg)
+		next.UseServices(tg)
 	}
 	if config.DingTalk != nil && config.DingTalk.Enable {
 		dt := dingding.New(&dingding.Config{
 			Token:  config.DingTalk.Token,
 			Secret: config.DingTalk.Secret,
 		})
-		n.notify.UseServices(dt)
+		next.UseServices(dt)
 	}
 	if config.Lark != nil && config.Lark.Enable {
 		larkSvc := NewLarkWebhookService(config.Lark.WebHookUrl, config.Lark.Secret)
-		n.notify.UseServices(larkSvc)
+		next.UseServices(larkSvc)
 	}
 	if config.ServerChan != nil && config.ServerChan.Enable {
 		sc := http.New()
@@ -62,7 +93,7 @@ func (n *Notifier) InitService(config *configs.NotifyConfig) error {
 				return data.Encode()
 			},
 		})
-		n.notify.UseServices(sc)
+		next.UseServices(sc)
 	}
 	if config.PushPlus != nil && config.PushPlus.Enable {
 		pp := http.New()
@@ -81,17 +112,56 @@ func (n *Notifier) InitService(config *configs.NotifyConfig) error {
 				}
 			},
 		})
-		n.notify.UseServices(pp)
+		next.UseServices(pp)
+	}
+	return next, nil
+}
+
+func validateWebhook(path, raw string) error {
+	parsed, err := url.Parse(raw)
+	if err != nil || parsed.Hostname() == "" || (parsed.Scheme != "http" && parsed.Scheme != "https") {
+		return fmt.Errorf("%s: expected an absolute http(s) url", path)
 	}
 	return nil
 }
 
+// Prepare builds and validates the replacement service set for config without
+// changing the active one. The returned commit activates the replacement
+// atomically and cannot fail; callers that persist the configuration first
+// invoke commit only after the write succeeds, keeping the persisted
+// configuration and the running service in lockstep.
+func (n *Notifier) Prepare(config *configs.NotifyConfig) (commit func(), err error) {
+	if n == nil {
+		return nil, errors.New("notifier is unavailable")
+	}
+	next, err := buildServices(config)
+	if err != nil {
+		return nil, err
+	}
+	return func() { n.active.Store(next) }, nil
+}
+
+// InitService atomically replaces the active service set. It is the boot-time
+// equivalent of Prepare+commit for callers that have no persistence step.
+func (n *Notifier) InitService(config *configs.NotifyConfig) error {
+	commit, err := n.Prepare(config)
+	if err != nil {
+		return err
+	}
+	commit()
+	return nil
+}
+
 func (n *Notifier) Send(eventType, op, message string) {
-	if !n.enable {
+	if n == nil {
+		return
+	}
+	active := n.active.Load()
+	if active == nil || active.Disabled {
 		return
 	}
 	title := fmt.Sprintf("[%s] %s", eventType, op)
-	err := n.notify.Send(context.Background(), title, message)
+	err := active.Send(context.Background(), title, message)
 	if err != nil {
 		logs.Log.Errorf("Failed to send notification: %s", err)
 	}

server/internal/notify/notify_test.go

@@ -6,6 +6,7 @@ import (
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"sync/atomic"
 	"testing"
 
 	"github.com/chainreactors/malice-network/server/internal/configs"
@@ -104,7 +105,8 @@ func TestNotifierInitServiceUsesConfiguredChannels(t *testing.T) {
 	if err := n.InitService(cfg); err != nil {
 		t.Fatalf("InitService failed: %v", err)
 	}
-	if !n.enable {
+	active := n.active.Load()
+	if active == nil || active.Disabled {
 		t.Fatal("expected notifier to be enabled after loading config")
 	}
 }
@@ -115,7 +117,95 @@ func TestNotifierInitServiceSkipsDisabledConfig(t *testing.T) {
 	if err := n.InitService(&configs.NotifyConfig{Enable: false}); err != nil {
 		t.Fatalf("InitService failed: %v", err)
 	}
-	if n.enable {
-		t.Fatal("expected notifier to remain disabled")
+	active := n.active.Load()
+	if active == nil || !active.Disabled {
+		t.Fatal("expected notifier to be disabled")
+	}
+}
+
+func TestBuildServicesValidatesEnabledChannels(t *testing.T) {
+	cases := []struct {
+		name string
+		cfg  *configs.NotifyConfig
+	}{
+		{"telegram missing credentials", &configs.NotifyConfig{
+			Enable:   true,
+			Telegram: &configs.TelegramConfig{Enable: true},
+		}},
+		{"dingtalk missing token", &configs.NotifyConfig{
+			Enable:   true,
+			DingTalk: &configs.DingTalkConfig{Enable: true},
+		}},
+		{"lark invalid webhook", &configs.NotifyConfig{
+			Enable: true,
+			Lark:   &configs.LarkConfig{Enable: true, WebHookUrl: "not-a-url"},
+		}},
+		{"serverchan invalid url", &configs.NotifyConfig{
+			Enable:     true,
+			ServerChan: &configs.ServerChanConfig{Enable: true, URL: "not-a-url"},
+		}},
+		{"pushplus missing token", &configs.NotifyConfig{
+			Enable:   true,
+			PushPlus: &configs.PushPlusConfig{Enable: true},
+		}},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if _, err := buildServices(tc.cfg); err == nil {
+				t.Fatal("expected a validation error")
+			}
+		})
+	}
+}
+
+func TestNotifierPrepareKeepsActiveServiceOnFailure(t *testing.T) {
+	n := NewNotifier()
+	good := &configs.NotifyConfig{
+		Enable:     true,
+		ServerChan: &configs.ServerChanConfig{Enable: true, URL: "https://serverchan.example/send"},
+	}
+	if err := n.InitService(good); err != nil {
+		t.Fatalf("InitService failed: %v", err)
+	}
+	active := n.active.Load()
+	if _, err := n.Prepare(&configs.NotifyConfig{
+		Enable:     true,
+		ServerChan: &configs.ServerChanConfig{Enable: true, URL: "not-a-url"},
+	}); err == nil {
+		t.Fatal("expected a validation error")
+	}
+	if n.active.Load() != active {
+		t.Fatal("failed prepare replaced the active service")
+	}
+}
+
+func TestNotifierInitServiceReplacesPreviousServices(t *testing.T) {
+	var deliveries int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		atomic.AddInt32(&deliveries, 1)
+		w.Header().Set("Content-Type", "application/json")
+		w.Write([]byte(`{"StatusCode":0,"StatusMessage":"success"}`))
+	}))
+	defer srv.Close()
+
+	n := NewNotifier()
+	if err := n.InitService(&configs.NotifyConfig{
+		Enable: true,
+		Lark:   &configs.LarkConfig{Enable: true, WebHookUrl: srv.URL},
+	}); err != nil {
+		t.Fatalf("InitService failed: %v", err)
+	}
+	n.Send("test", "op", "message")
+	if atomic.LoadInt32(&deliveries) != 1 {
+		t.Fatal("expected the configured channel to deliver")
+	}
+
+	// A disabled replacement must swap the old service set out, not leave it active.
+	if err := n.InitService(&configs.NotifyConfig{Enable: false}); err != nil {
+		t.Fatalf("InitService failed: %v", err)
+	}
+	n.Send("test", "op", "message")
+	if atomic.LoadInt32(&deliveries) != 1 {
+		t.Fatal("disabled config left the previous service active")
 	}
 }

server/rpc/rpc-config.go

@@ -2,11 +2,14 @@ package rpc
 
 import (
 	"context"
+
 	"github.com/chainreactors/IoM-go/proto/client/clientpb"
 	"github.com/chainreactors/IoM-go/types"
 	"github.com/chainreactors/malice-network/helper/utils/configutil"
 	"github.com/chainreactors/malice-network/server/internal/configs"
 	"github.com/chainreactors/malice-network/server/internal/core"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 func (rpc *Server) GetGithubConfig(ctx context.Context, req *clientpb.Empty) (*clientpb.GithubActionBuildConfig, error) {
@@ -98,6 +101,9 @@ func (rpc *Server) GetNotifyConfig(ctx context.Context, req *clientpb.Empty) (*c
 }
 
 func (rpc *Server) UpdateNotifyConfig(ctx context.Context, req *clientpb.Notify) (*clientpb.Empty, error) {
+	if req == nil {
+		return nil, status.Error(codes.InvalidArgument, "notify config is required")
+	}
 	notifyConfig := &configs.NotifyConfig{
 		Enable: req.TelegramEnable || req.DingtalkEnable || req.LarkEnable || req.ServerchanEnable || req.PushplusEnable,
 		Telegram: &configs.TelegramConfig{
@@ -126,14 +132,17 @@ func (rpc *Server) UpdateNotifyConfig(ctx context.Context, req *clientpb.Notify)
 			Channel: req.PushplusChannel,
 		},
 	}
-	err := configs.UpdateNotifyConfig(notifyConfig)
+	// Build and validate the replacement service before persisting anything.
+	// Persisting is the last fallible step, so the configuration on disk and
+	// the running service change together or not at all.
+	commit, err := core.EventBroker.PrepareNotifyConfig(notifyConfig)
 	if err != nil {
 		return nil, err
 	}
-	err = core.EventBroker.InitService(notifyConfig)
-	if err != nil {
+	if err = configs.UpdateNotifyConfig(notifyConfig); err != nil {
 		return nil, err
 	}
+	commit()
 	return &clientpb.Empty{}, nil
 }