Files
Nukumizu/internal/controller/pipes/ntfy.go
T
NanamiAdminandClaude Code cb2df5076d refactor(config): publish config singletons through atomic pointers
The three configuration singletons (C_globalConfig, C_botUserConfig,
C_botNodeConfig) were plain variables: LoadGlobalConfig and friends assigned
them from the goroutine handling a settings update, while bot pipes, the node
tracker and the HTTP handlers read them from their own goroutines. That is an
unsynchronized read of a concurrently written variable — a data race the race
detector reports, and one that already existed before any hot-reload work
because /api/webhook/add reloads the configuration while the bots run.

Replace them with atomic.Pointer values behind Current(), BotUsers() and
BotNodes(). Each reload builds a fresh value and publishes it atomically, so a
reader either sees the previous configuration or the new one, never a partial
one. Callers read through the accessor on every use instead of caching it.

Two spots that read several fields of one guard now snapshot once per call, so
a reload cannot split a combined check mid-flight:
- qq_napcat.handleNapcatEvent, which evaluates the debug guards per event
- the komari task-echo guard, now behind taskEchoEnabled()

The NapCat HTTP methods keep logging on showNapcatAction alone (without
requiring debugMode), matching their existing behaviour; that inconsistency
with the WebSocket path is preserved, not introduced, and is called out in
actionLogEnabled.

Also adds TestConcurrentReloadAndRead, which drives every accessor from four
reader goroutines while two writers reload the configuration, and sanitises the
member IDs used as test fixtures in config/settings_test.go.

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-28 22:47:37 +08:00

149 lines
3.9 KiB
Go

package pipes
import (
"fmt"
"net/http"
"strings"
"time"
"nukumizu-backend/config"
"nukumizu-backend/internal/controller"
"nukumizu-backend/internal/netproxy"
"nukumizu-backend/internal/node"
"nukumizu-backend/internal/template"
"nukumizu-backend/postLog"
)
// NtfyController handles notifications via ntfy.sh or a self-hosted ntfy server.
type NtfyController struct {
cfg config.NtfyConfig
httpClient *http.Client
}
// NewNtfyController creates a new Ntfy controller.
func NewNtfyController(cfg config.NtfyConfig) *NtfyController {
return &NtfyController{
cfg: cfg,
httpClient: netproxy.HTTPClient(cfg.NetworkUseProxy, 10*time.Second),
}
}
// Name returns the controller name.
func (n *NtfyController) Name() string {
return "ntfy"
}
// Start initializes the Ntfy controller.
func (n *NtfyController) Start() error {
if !n.cfg.Enabled {
postLog.Info("Ntfy controller is disabled")
return nil
}
postLog.Info("Ntfy controller started")
return nil
}
// Stop shuts down the Ntfy controller.
func (n *NtfyController) Stop() {
postLog.Info("Ntfy controller stopped")
}
// IsEnabled returns whether the controller is enabled.
func (n *NtfyController) IsEnabled() bool {
return n.cfg.Enabled
}
// IsMarkdown returns whether the channel renders Markdown, per its markdown
// setting in config.json.
func (n *NtfyController) IsMarkdown() bool {
return n.cfg.Markdown
}
// SendStatusChange sends a status change notification via Ntfy.
func (n *NtfyController) SendStatusChange(change node.StatusChange) error {
if !n.cfg.Enabled {
return nil
}
cfg := config.Current()
params := template.BuildParamsFromStatusChange(change)
message := template.Render(cfg.ControllerMessage.ServerStatusChanged, params, n.cfg.Markdown)
title := fmt.Sprintf("Server %s: %s", change.Name, change.Event)
return n.publish(title, message)
}
// SendServerList sends the server list via Ntfy.
func (n *NtfyController) SendServerList(onlineServers, offlineServers string) error {
if !n.cfg.Enabled {
return nil
}
cfg := config.Current()
params := template.BuildParamsFromServerList()
message := template.Render(cfg.ControllerMessage.ServerList, params, n.cfg.Markdown)
return n.publish("Server List", message)
}
// SendExecuteResult sends a command execution result via Ntfy.
func (n *NtfyController) SendExecuteResult(serverName, serverUUID, command, result string) error {
if !n.cfg.Enabled {
return nil
}
cfg := config.Current()
params := template.BuildParamsFromExecResult(serverName, serverUUID, command, result)
message := template.Render(cfg.ControllerMessage.ServerExecuteResult, params, n.cfg.Markdown)
title := fmt.Sprintf("Command Result: %s on %s", command, serverName)
return n.publish(title, message)
}
// SendAlert sends an alert submitted through the incoming webhook API to the
// configured topic.
func (n *NtfyController) SendAlert(alert controller.Alert) error {
if !n.cfg.Enabled {
return nil
}
return n.publish(alert.Subject, alert.Render(n.cfg.Markdown))
}
func (n *NtfyController) publish(title, message string) error {
serverURL := n.cfg.Server
if serverURL == "" {
serverURL = "https://ntfy.sh"
}
serverURL = strings.TrimRight(serverURL, "/")
publishURL := fmt.Sprintf("%s/%s", serverURL, n.cfg.Topic)
req, err := http.NewRequest("POST", publishURL, strings.NewReader(message))
if err != nil {
return fmt.Errorf("failed to create ntfy request: %w", err)
}
req.Header.Set("Title", title)
if n.cfg.Priority != "" && n.cfg.Priority != "default" {
req.Header.Set("Priority", n.cfg.Priority)
}
if n.cfg.Token != "" {
req.Header.Set("Authorization", "Bearer "+n.cfg.Token)
}
resp, err := n.httpClient.Do(req)
if err != nil {
postLog.Warning("Failed to publish to ntfy: " + err.Error())
return err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
postLog.Warning(fmt.Sprintf("Ntfy publish returned status %d", resp.StatusCode))
}
postLog.Debug("Ntfy notification sent to topic: " + n.cfg.Topic)
return nil
}