refactor(plugins): complete physical encapsulation of auth, admin, message_gateway, and risk_control domain plugins

plugins/domain/message_gateway/admin_handlers.go

@@ -0,0 +1,170 @@
+// Copyright 2026 Arctel.net
+// SPDX-License-Identifier: Apache-2.0
+
+package message_gateway
+
+import (
+	"net/http"
+	"strconv"
+
+	"github.com/Rain-kl/Wavelet/internal/shared/response"
+	"github.com/gin-gonic/gin"
+)
+
+// ListAdminChannelDefinitions returns form schemas for supported channel types.
+// @Summary List message gateway channel definitions
+// @Description Returns form field definitions for Telegram and QQ channels
+// @Tags admin-message-gateway
+// @Produce json
+// @Security SessionCookie
+// @Success 200 {object} response.Any{data=[]Definition}
+// @Router /api/v1/admin/message-gateway/channels/definitions [get]
+func ListAdminChannelDefinitions(c *gin.Context) {
+	c.JSON(http.StatusOK, response.OK(channelDefinitions()))
+}
+
+// ListAdminChannels lists configured messaging channels with secrets masked.
+// @Summary List message gateway channels
+// @Description Returns all messaging channels; secrets are masked
+// @Tags admin-message-gateway
+// @Produce json
+// @Security SessionCookie
+// @Success 200 {object} response.Any{data=[]ChannelDTO}
+// @Router /api/v1/admin/message-gateway/channels [get]
+func ListAdminChannels(c *gin.Context) {
+	rows, err := listChannels(c.Request.Context())
+	if err != nil {
+		response.AbortInternal(c, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, response.OK(rows))
+}
+
+// CreateAdminChannel creates a messaging channel.
+// @Summary Create message gateway channel
+// @Description Creates a Telegram or QQ channel with encrypted credentials
+// @Tags admin-message-gateway
+// @Accept json
+// @Produce json
+// @Security SessionCookie
+// @Param request body CreateChannelRequest true "create body"
+// @Success 200 {object} response.Any{data=ChannelDTO}
+// @Failure 400 {object} response.Any
+// @Router /api/v1/admin/message-gateway/channels [post]
+func CreateAdminChannel(c *gin.Context) {
+	var req CreateChannelRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		response.AbortBadRequest(c, err.Error())
+		return
+	}
+	dto, err := createChannel(c.Request.Context(), req)
+	if err != nil {
+		response.AbortBadRequest(c, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, response.OK(dto))
+}
+
+// UpdateAdminChannel patches a messaging channel. Empty secrets keep the previous values.
+// @Summary Update message gateway channel
+// @Description Updates a channel; empty secrets keep the current ciphertext
+// @Tags admin-message-gateway
+// @Accept json
+// @Produce json
+// @Security SessionCookie
+// @Param id path int true "channel id"
+// @Param request body UpdateChannelRequest true "update body"
+// @Success 200 {object} response.Any{data=ChannelDTO}
+// @Failure 400 {object} response.Any
+// @Failure 404 {object} response.Any
+// @Router /api/v1/admin/message-gateway/channels/{id} [patch]
+func UpdateAdminChannel(c *gin.Context) {
+	id, err := strconv.ParseUint(c.Param("id"), 10, 64)
+	if err != nil {
+		response.AbortBadRequest(c, "invalid channel id")
+		return
+	}
+	var req UpdateChannelRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		response.AbortBadRequest(c, err.Error())
+		return
+	}
+	dto, err := updateChannel(c.Request.Context(), id, req)
+	if err != nil {
+		if err.Error() == errChannelNotFound {
+			response.AbortNotFound(c, err.Error())
+			return
+		}
+		response.AbortBadRequest(c, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, response.OK(dto))
+}
+
+// DeleteAdminChannel removes a channel and its bindings/pairing codes.
+// @Summary Delete message gateway channel
+// @Description Deletes a channel and cascaded bindings and pairing codes
+// @Tags admin-message-gateway
+// @Produce json
+// @Security SessionCookie
+// @Param id path int true "channel id"
+// @Success 200 {object} response.Any
+// @Failure 404 {object} response.Any
+// @Router /api/v1/admin/message-gateway/channels/{id} [delete]
+func DeleteAdminChannel(c *gin.Context) {
+	id, err := strconv.ParseUint(c.Param("id"), 10, 64)
+	if err != nil {
+		response.AbortBadRequest(c, "invalid channel id")
+		return
+	}
+	if err := deleteChannel(c.Request.Context(), id); err != nil {
+		if err.Error() == errChannelNotFound {
+			response.AbortNotFound(c, err.Error())
+			return
+		}
+		response.AbortInternal(c, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, response.OKNil())
+}
+
+// TestAdminChannel probes stored credentials (Telegram getMe or QQ token).
+// @Summary Test message gateway channel
+// @Description Probes stored credentials without returning secrets
+// @Tags admin-message-gateway
+// @Produce json
+// @Security SessionCookie
+// @Param id path int true "channel id"
+// @Success 200 {object} response.Any
+// @Failure 400 {object} response.Any
+// @Failure 404 {object} response.Any
+// @Router /api/v1/admin/message-gateway/channels/{id}/test [post]
+func TestAdminChannel(c *gin.Context) {
+	id, err := strconv.ParseUint(c.Param("id"), 10, 64)
+	if err != nil {
+		response.AbortBadRequest(c, "invalid channel id")
+		return
+	}
+	if err := probeChannel(c.Request.Context(), id); err != nil {
+		if err.Error() == errChannelNotFound {
+			response.AbortNotFound(c, err.Error())
+			return
+		}
+		response.AbortBadRequest(c, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, response.OKNil())
+}
+
+// RegisterAdminRoutes mounts admin message-gateway APIs under /admin.
+func RegisterAdminRoutes(adminRouter *gin.RouterGroup) {
+	g := adminRouter.Group("/message-gateway")
+	{
+		g.GET("/channels/definitions", ListAdminChannelDefinitions)
+		g.GET("/channels", ListAdminChannels)
+		g.POST("/channels", CreateAdminChannel)
+		g.PATCH("/channels/:id", UpdateAdminChannel)
+		g.DELETE("/channels/:id", DeleteAdminChannel)
+		g.POST("/channels/:id/test", TestAdminChannel)
+	}
+}

plugins/domain/message_gateway/admin_logics.go

@@ -0,0 +1,346 @@
+// Copyright 2026 Arctel.net
+// SPDX-License-Identifier: Apache-2.0
+
+package message_gateway
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/Rain-kl/Wavelet/internal/model"
+	"github.com/Rain-kl/Wavelet/internal/repository"
+	"github.com/tencent-connect/botgo/token"
+	"gorm.io/gorm"
+)
+
+const defaultTelegramAPI = "https://api.telegram.org"
+
+// Field is one admin form field.
+type Field struct {
+	Key      string `json:"key"`
+	Type     string `json:"type"`
+	Required bool   `json:"required"`
+}
+
+// Definition describes a channel type form.
+type Definition struct {
+	Type   string  `json:"type"`
+	Name   string  `json:"name"`
+	Fields []Field `json:"fields"`
+}
+
+// CreateChannelRequest is the admin create body.
+type CreateChannelRequest struct {
+	Name       string `json:"name"`
+	Type       string `json:"type"`
+	Enabled    *bool  `json:"enabled"`
+	BotToken   string `json:"bot_token"`
+	AppID      string `json:"app_id"`
+	AppSecret  string `json:"app_secret"`
+	BaseURL    string `json:"base_url"`
+	PortalHost string `json:"portal_host"`
+	Sandbox    string `json:"sandbox"`
+}
+
+// UpdateChannelRequest is the admin patch body.
+type UpdateChannelRequest struct {
+	Name       *string `json:"name"`
+	Enabled    *bool   `json:"enabled"`
+	BotToken   string  `json:"bot_token"`
+	AppID      string  `json:"app_id"`
+	AppSecret  string  `json:"app_secret"`
+	BaseURL    *string `json:"base_url"`
+	PortalHost *string `json:"portal_host"`
+	Sandbox    *string `json:"sandbox"`
+}
+
+// ChannelDTO is a list/detail view with secrets masked.
+type ChannelDTO struct {
+	ID         uint64    `json:"id,string"`
+	Name       string    `json:"name"`
+	Type       string    `json:"type"`
+	OwnerScope string    `json:"owner_scope"`
+	Enabled    bool      `json:"enabled"`
+	BotToken   string    `json:"bot_token,omitempty"`
+	AppID      string    `json:"app_id,omitempty"`
+	AppSecret  string    `json:"app_secret,omitempty"`
+	BaseURL    string    `json:"base_url,omitempty"`
+	PortalHost string    `json:"portal_host,omitempty"`
+	CreatedAt  time.Time `json:"created_at"`
+	UpdatedAt  time.Time `json:"updated_at"`
+}
+
+func channelDefinitions() []Definition {
+	return []Definition{
+		{
+			Type: model.MessageChannelTypeTelegram,
+			Name: "Telegram",
+			Fields: []Field{
+				{Key: "bot_token", Type: "password", Required: true},
+				{Key: "base_url", Type: "text"},
+			},
+		},
+		{
+			Type: model.MessageChannelTypeQQ,
+			Name: "QQ",
+			Fields: []Field{
+				{Key: "app_id", Required: true},
+				{Key: "app_secret", Type: "password", Required: true},
+				{Key: "portal_host", Type: "text"},
+			},
+		},
+	}
+}
+
+func createChannel(ctx context.Context, req CreateChannelRequest) (ChannelDTO, error) {
+	name := strings.TrimSpace(req.Name)
+	if name == "" {
+		return ChannelDTO{}, errors.New(errNameRequired)
+	}
+	typ := strings.TrimSpace(req.Type)
+	creds, extra, err := credentialsFromCreate(req)
+	if err != nil {
+		return ChannelDTO{}, err
+	}
+	cipher, err := EncryptCredentials(creds)
+	if err != nil {
+		return ChannelDTO{}, err
+	}
+	enabled := true
+	if req.Enabled != nil {
+		enabled = *req.Enabled
+	}
+	row := &model.MessageChannel{
+		Name:        name,
+		Type:        typ,
+		OwnerScope:  model.MessageOwnerScopeSystem,
+		Enabled:     enabled,
+		Credentials: cipher,
+		Extra:       EncodeExtra(extra),
+	}
+	if err := repository.CreateMessageChannel(ctx, row); err != nil {
+		return ChannelDTO{}, err
+	}
+	return toDTO(row, creds, extra), nil
+}
+
+func updateChannel(ctx context.Context, id uint64, req UpdateChannelRequest) (ChannelDTO, error) {
+	row, err := repository.GetMessageChannel(ctx, id)
+	if err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			return ChannelDTO{}, errors.New(errChannelNotFound)
+		}
+		return ChannelDTO{}, err
+	}
+	creds, err := DecryptCredentials(row.Credentials)
+	if err != nil {
+		creds = map[string]string{}
+	}
+	extra := ParseExtra(row.Extra)
+	if req.Name != nil {
+		name := strings.TrimSpace(*req.Name)
+		if name == "" {
+			return ChannelDTO{}, errors.New(errNameRequired)
+		}
+		row.Name = name
+	}
+	if req.Enabled != nil {
+		row.Enabled = *req.Enabled
+	}
+	if token := strings.TrimSpace(req.BotToken); token != "" {
+		creds["bot_token"] = token
+	}
+	if appID := strings.TrimSpace(req.AppID); appID != "" {
+		creds["app_id"] = appID
+	}
+	if secret := strings.TrimSpace(req.AppSecret); secret != "" {
+		creds["app_secret"] = secret
+	}
+	if req.BaseURL != nil {
+		extra["base_url"] = strings.TrimSpace(*req.BaseURL)
+	}
+	if req.PortalHost != nil {
+		extra["portal_host"] = strings.TrimSpace(*req.PortalHost)
+	}
+	if req.Sandbox != nil {
+		extra["sandbox"] = strings.TrimSpace(*req.Sandbox)
+	}
+	if err := validateCredentials(row.Type, creds); err != nil {
+		return ChannelDTO{}, err
+	}
+	cipher, err := EncryptCredentials(creds)
+	if err != nil {
+		return ChannelDTO{}, err
+	}
+	row.Credentials = cipher
+	row.Extra = EncodeExtra(extra)
+	if err := repository.UpdateMessageChannel(ctx, row); err != nil {
+		return ChannelDTO{}, err
+	}
+	return toDTO(row, creds, extra), nil
+}
+
+func listChannels(ctx context.Context) ([]ChannelDTO, error) {
+	rows, err := repository.ListMessageChannels(ctx)
+	if err != nil {
+		return nil, err
+	}
+	out := make([]ChannelDTO, 0, len(rows))
+	for i := range rows {
+		creds, err := DecryptCredentials(rows[i].Credentials)
+		if err != nil {
+			creds = map[string]string{}
+		}
+		out = append(out, toDTO(&rows[i], creds, ParseExtra(rows[i].Extra)))
+	}
+	return out, nil
+}
+
+func deleteChannel(ctx context.Context, id uint64) error {
+	if _, err := repository.GetMessageChannel(ctx, id); err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			return errors.New(errChannelNotFound)
+		}
+		return err
+	}
+	return repository.DeleteMessageChannel(ctx, id)
+}
+
+func probeChannel(ctx context.Context, id uint64) error {
+	row, err := repository.GetMessageChannel(ctx, id)
+	if err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			return errors.New(errChannelNotFound)
+		}
+		return err
+	}
+	creds, err := DecryptCredentials(row.Credentials)
+	if err != nil {
+		return err
+	}
+	extra := ParseExtra(row.Extra)
+	if err := probeCredentials(ctx, row.Type, creds, extra); err != nil {
+		return fmt.Errorf("%s: %w", errChannelProbeFailed, err)
+	}
+	return nil
+}
+
+func credentialsFromCreate(req CreateChannelRequest) (map[string]string, map[string]string, error) {
+	typ := strings.TrimSpace(req.Type)
+	creds := map[string]string{}
+	extra := map[string]string{}
+	switch typ {
+	case model.MessageChannelTypeTelegram:
+		creds["bot_token"] = strings.TrimSpace(req.BotToken)
+		if base := strings.TrimSpace(req.BaseURL); base != "" {
+			extra["base_url"] = base
+		}
+	case model.MessageChannelTypeQQ:
+		creds["app_id"] = strings.TrimSpace(req.AppID)
+		creds["app_secret"] = strings.TrimSpace(req.AppSecret)
+		if host := strings.TrimSpace(req.PortalHost); host != "" {
+			extra["portal_host"] = host
+		} else {
+			extra["portal_host"] = "q.qq.com"
+		}
+		if sandbox := strings.TrimSpace(req.Sandbox); sandbox != "" {
+			extra["sandbox"] = sandbox
+		}
+	default:
+		return nil, nil, errors.New(errTypeInvalid)
+	}
+	if err := validateCredentials(typ, creds); err != nil {
+		return nil, nil, err
+	}
+	return creds, extra, nil
+}
+
+func validateCredentials(typ string, creds map[string]string) error {
+	switch typ {
+	case model.MessageChannelTypeTelegram:
+		if strings.TrimSpace(creds["bot_token"]) == "" {
+			return errors.New(errTelegramTokenRequired)
+		}
+	case model.MessageChannelTypeQQ:
+		if strings.TrimSpace(creds["app_id"]) == "" || strings.TrimSpace(creds["app_secret"]) == "" {
+			return errors.New(errQQCredentialsRequired)
+		}
+	default:
+		return errors.New(errTypeInvalid)
+	}
+	return nil
+}
+
+func toDTO(row *model.MessageChannel, creds, extra map[string]string) ChannelDTO {
+	dto := ChannelDTO{
+		ID:         row.ID,
+		Name:       row.Name,
+		Type:       row.Type,
+		OwnerScope: row.OwnerScope,
+		Enabled:    row.Enabled,
+		CreatedAt:  row.CreatedAt,
+		UpdatedAt:  row.UpdatedAt,
+	}
+	if strings.TrimSpace(creds["bot_token"]) != "" {
+		dto.BotToken = maskedSecret
+	}
+	if id := strings.TrimSpace(creds["app_id"]); id != "" {
+		dto.AppID = id
+	}
+	if strings.TrimSpace(creds["app_secret"]) != "" {
+		dto.AppSecret = maskedSecret
+	}
+	dto.BaseURL = extra["base_url"]
+	dto.PortalHost = extra["portal_host"]
+	return dto
+}
+
+func probeCredentials(ctx context.Context, typ string, creds, extra map[string]string) error {
+	switch typ {
+	case model.MessageChannelTypeTelegram:
+		base := strings.TrimSpace(extra["base_url"])
+		if base == "" {
+			base = defaultTelegramAPI
+		}
+		url := strings.TrimRight(base, "/") + "/bot" + creds["bot_token"] + "/getMe"
+		req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+		if err != nil {
+			return err
+		}
+		resp, err := http.DefaultClient.Do(req)
+		if err != nil {
+			return err
+		}
+		defer func() { _ = resp.Body.Close() }()
+		const probeBodyLimit = 4096
+		body, _ := io.ReadAll(io.LimitReader(resp.Body, probeBodyLimit))
+		if resp.StatusCode != http.StatusOK {
+			return fmt.Errorf("telegram getMe status %d", resp.StatusCode)
+		}
+		var parsed struct {
+			OK bool `json:"ok"`
+		}
+		if err := json.Unmarshal(body, &parsed); err != nil {
+			return err
+		}
+		if !parsed.OK {
+			return errors.New("telegram getMe returned ok=false")
+		}
+		return nil
+	case model.MessageChannelTypeQQ:
+		src := token.NewQQBotTokenSource(&token.QQBotCredentials{
+			AppID:     creds["app_id"],
+			AppSecret: creds["app_secret"],
+		})
+		_, err := src.Token()
+		return err
+	default:
+		return errors.New(errTypeInvalid)
+	}
+}

plugins/domain/message_gateway/custom_events_admin_login.go

@@ -0,0 +1,42 @@
+// Copyright 2026 Arctel.net
+// SPDX-License-Identifier: Apache-2.0
+
+package message_gateway
+
+import (
+	"context"
+	"time"
+
+	"github.com/Rain-kl/Wavelet/internal/listener"
+)
+
+// AdminLogin is the metadata definition for the admin login event.
+var AdminLogin = EventMetadata{
+	Key:  "admin_login",
+	Name: "管理员登录",
+	DefaultTemplate: NotificationMessage{
+		Title:   "管理员登录提醒",
+		Content: "管理员 {{user.username}} 于 {{time}} 从 IP {{ip}} 登录系统。",
+		Level:   "INFO",
+	},
+	Description: "当管理员成功登录系统时触发此通知",
+}
+
+func handleAdminLogin(ctx context.Context, event listener.AdminLoggedIn) {
+	if event.User == nil {
+		return
+	}
+
+	body := map[string]any{
+		"user": event.User,
+		"ip":   event.IP,
+		"time": time.Now().Format("2006-01-02 15:04:05"),
+	}
+	DefaultTrigger.Trigger(ctx, AdminLogin, body)
+}
+
+// RegisterCustomEvents registers default domain push notification events.
+func RegisterCustomEvents() {
+	RegisterBuiltInEvent(AdminLogin)
+	listener.OnAdminLoggedIn(handleAdminLogin)
+}

plugins/domain/message_gateway/errs.go

@@ -0,0 +1,26 @@
+// Copyright 2026 Arctel.net
+// SPDX-License-Identifier: Apache-2.0
+
+package message_gateway
+
+import "errors"
+
+var (
+	errCodeInvalid          = errors.New("invalid or expired pairing code")
+	errChannelMismatch      = errors.New("pairing code does not match channel")
+	errPlatformAlreadyBound = errors.New("this platform account is already bound")
+	errBindingNotFound      = errors.New("binding not found")
+	errBindingForbidden     = errors.New("cannot unbind another user's binding")
+	errChannelIDRequired    = errors.New("channel_id is required")
+	errChannelDisabled      = errors.New("channel is not enabled")
+)
+
+const (
+	errNameRequired          = "name is required"
+	errTypeInvalid           = "type must be telegram or qq"
+	errTelegramTokenRequired = "telegram bot secret is required"   //nolint:gosec // user-facing validation text
+	errQQCredentialsRequired = "qq app id and secret are required" //nolint:gosec // user-facing validation text
+	errChannelNotFound       = "channel not found"
+	errChannelProbeFailed    = "channel probe failed"
+	maskedSecret             = "********"
+)

plugins/domain/message_gateway/handlers.go

@@ -0,0 +1,147 @@
+// Copyright 2026 Arctel.net
+// SPDX-License-Identifier: Apache-2.0
+
+package message_gateway
+
+import (
+	"errors"
+	"net/http"
+	"strconv"
+
+	"github.com/Rain-kl/Wavelet/internal/apps/oauth"
+	"github.com/Rain-kl/Wavelet/internal/model"
+	"github.com/Rain-kl/Wavelet/internal/shared/response"
+	"github.com/gin-gonic/gin"
+)
+
+func currentUser(c *gin.Context) (*model.User, bool) {
+	return oauth.GetFromContext[*model.User](c, oauth.UserObjKey)
+}
+
+// ListChannels lists enabled channels a user can bind.
+// @Summary List enabled messaging channels
+// @Description Returns enabled system bots the current user can pair with
+// @Tags message-gateway
+// @Produce json
+// @Security SessionCookie
+// @Success 200 {object} response.Any{data=[]PublicChannelDTO}
+// @Failure 401 {object} response.Any
+// @Router /api/v1/message-gateway/channels [get]
+func ListChannels(c *gin.Context) {
+	if user, ok := currentUser(c); !ok || user == nil {
+		response.AbortUnauthorized(c, "login required")
+		return
+	}
+	rows, err := listEnabledPublicChannels(c.Request.Context())
+	if err != nil {
+		response.AbortInternal(c, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, response.OK(rows))
+}
+
+// ListBindings lists the current user's bot bindings.
+// @Summary List message gateway bindings
+// @Description Returns the current user's bound messaging channels
+// @Tags message-gateway
+// @Produce json
+// @Security SessionCookie
+// @Success 200 {object} response.Any{data=[]BindingDTO}
+// @Failure 401 {object} response.Any
+// @Router /api/v1/message-gateway/bindings [get]
+func ListBindings(c *gin.Context) {
+	user, ok := currentUser(c)
+	if !ok || user == nil {
+		response.AbortUnauthorized(c, "login required")
+		return
+	}
+	rows, err := listUserBindings(c.Request.Context(), user.ID)
+	if err != nil {
+		response.AbortInternal(c, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, response.OK(rows))
+}
+
+// BindBinding consumes a pairing code and binds the platform identity.
+// @Summary Bind a messaging channel
+// @Description Binds the current user to a platform identity using a one-time pairing code
+// @Tags message-gateway
+// @Accept json
+// @Produce json
+// @Security SessionCookie
+// @Param request body BindRequest true "bind body"
+// @Success 200 {object} response.Any{data=BindingDTO}
+// @Failure 400 {object} response.Any
+// @Failure 409 {object} response.Any
+// @Router /api/v1/message-gateway/bindings [post]
+func BindBinding(c *gin.Context) {
+	user, ok := currentUser(c)
+	if !ok || user == nil {
+		response.AbortUnauthorized(c, "login required")
+		return
+	}
+	var req BindRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		response.AbortBadRequest(c, err.Error())
+		return
+	}
+	dto, err := bindChannel(c.Request.Context(), user.ID, req)
+	if err != nil {
+		if errors.Is(err, errPlatformAlreadyBound) {
+			response.AbortConflict(c, err.Error())
+			return
+		}
+		response.AbortBadRequest(c, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, response.OK(dto))
+}
+
+// UnbindBinding removes the current user's binding.
+// @Summary Unbind a messaging channel
+// @Description Removes a binding owned by the current user
+// @Tags message-gateway
+// @Produce json
+// @Security SessionCookie
+// @Param id path int true "binding id"
+// @Success 200 {object} response.Any
+// @Failure 403 {object} response.Any
+// @Failure 404 {object} response.Any
+// @Router /api/v1/message-gateway/bindings/{id} [delete]
+func UnbindBinding(c *gin.Context) {
+	user, ok := currentUser(c)
+	if !ok || user == nil {
+		response.AbortUnauthorized(c, "login required")
+		return
+	}
+	id, err := strconv.ParseUint(c.Param("id"), 10, 64)
+	if err != nil {
+		response.AbortBadRequest(c, "invalid binding id")
+		return
+	}
+	if err := unbindChannel(c.Request.Context(), user.ID, id); err != nil {
+		if errors.Is(err, errBindingNotFound) {
+			response.AbortNotFound(c, err.Error())
+			return
+		}
+		if errors.Is(err, errBindingForbidden) {
+			response.AbortForbidden(c, err.Error())
+			return
+		}
+		response.AbortInternal(c, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, response.OKNil())
+}
+
+// RegisterUserRoutes mounts user-facing message gateway endpoints.
+func RegisterUserRoutes(r *gin.RouterGroup) {
+	mg := r.Group("/message-gateway", oauth.LoginRequired())
+	{
+		mg.GET("/channels", ListChannels)
+		mg.GET("/bindings", ListBindings)
+		mg.POST("/bindings", BindBinding)
+		mg.DELETE("/bindings/:id", UnbindBinding)
+	}
+}

plugins/domain/message_gateway/logics.go

@@ -0,0 +1,158 @@
+// Copyright 2026 Arctel.net
+// SPDX-License-Identifier: Apache-2.0
+
+package message_gateway
+
+import (
+	"context"
+	"errors"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/Rain-kl/Wavelet/internal/model"
+	"github.com/Rain-kl/Wavelet/internal/repository"
+	pkgmg "github.com/Rain-kl/Wavelet/pkg/message_gateway"
+	"gorm.io/gorm"
+)
+
+// BindRequest is the user bind body.
+type BindRequest struct {
+	ChannelID string `json:"channel_id"`
+	Code      string `json:"code"`
+}
+
+// BindingDTO is a user-facing binding row.
+type BindingDTO struct {
+	ID             uint64    `json:"id,string"`
+	UserID         uint64    `json:"user_id,string"`
+	ChannelID      uint64    `json:"channel_id,string"`
+	ChannelName    string    `json:"channel_name"`
+	ChannelType    string    `json:"channel_type"`
+	PlatformUserID string    `json:"platform_user_id"`
+	CreatedAt      time.Time `json:"created_at"`
+}
+
+func bindChannel(ctx context.Context, userID uint64, req BindRequest) (BindingDTO, error) {
+	channelID, err := strconv.ParseUint(strings.TrimSpace(req.ChannelID), 10, 64)
+	if err != nil || channelID == 0 {
+		return BindingDTO{}, errChannelIDRequired
+	}
+	code := pkgmg.NormalizeCode(req.Code)
+	if code == "" {
+		return BindingDTO{}, errCodeInvalid
+	}
+	pairing, err := repository.GetPairingCode(ctx, code)
+	if err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			return BindingDTO{}, errCodeInvalid
+		}
+		return BindingDTO{}, err
+	}
+	if !pairing.ExpiresAt.After(time.Now()) {
+		return BindingDTO{}, errCodeInvalid
+	}
+	if pairing.ChannelID != channelID {
+		return BindingDTO{}, errChannelMismatch
+	}
+	ch, err := repository.GetMessageChannel(ctx, channelID)
+	if err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			return BindingDTO{}, errCodeInvalid
+		}
+		return BindingDTO{}, err
+	}
+	if !ch.Enabled {
+		return BindingDTO{}, errChannelDisabled
+	}
+
+	existing, err := repository.GetBindingByChannelPlatform(ctx, channelID, pairing.PlatformUserID)
+	if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
+		return BindingDTO{}, err
+	}
+	if err == nil && existing != nil {
+		if existing.UserID != userID {
+			return BindingDTO{}, errPlatformAlreadyBound
+		}
+		_ = repository.DeletePairingCode(ctx, pairing.Code)
+		return toBindingDTO(existing, ch), nil
+	}
+
+	row := &model.MessageBinding{
+		UserID:         userID,
+		ChannelID:      channelID,
+		PlatformUserID: pairing.PlatformUserID,
+	}
+	if err := repository.CreateMessageBinding(ctx, row); err != nil {
+		return BindingDTO{}, err
+	}
+	if err := repository.DeletePairingCode(ctx, pairing.Code); err != nil {
+		return BindingDTO{}, err
+	}
+	return toBindingDTO(row, ch), nil
+}
+
+// PublicChannelDTO is an enabled channel a user can bind to.
+type PublicChannelDTO struct {
+	ID   uint64 `json:"id,string"`
+	Name string `json:"name"`
+	Type string `json:"type"`
+}
+
+func listEnabledPublicChannels(ctx context.Context) ([]PublicChannelDTO, error) {
+	rows, err := repository.ListEnabledMessageChannels(ctx)
+	if err != nil {
+		return nil, err
+	}
+	out := make([]PublicChannelDTO, 0, len(rows))
+	for _, row := range rows {
+		out = append(out, PublicChannelDTO{ID: row.ID, Name: row.Name, Type: row.Type})
+	}
+	return out, nil
+}
+
+func listUserBindings(ctx context.Context, userID uint64) ([]BindingDTO, error) {
+	rows, err := repository.ListBindingsByUser(ctx, userID)
+	if err != nil {
+		return nil, err
+	}
+	out := make([]BindingDTO, 0, len(rows))
+	for i := range rows {
+		ch, err := repository.GetMessageChannel(ctx, rows[i].ChannelID)
+		if err != nil {
+			out = append(out, toBindingDTO(&rows[i], nil))
+			continue
+		}
+		out = append(out, toBindingDTO(&rows[i], ch))
+	}
+	return out, nil
+}
+
+func unbindChannel(ctx context.Context, userID, bindingID uint64) error {
+	row, err := repository.GetMessageBinding(ctx, bindingID)
+	if err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			return errBindingNotFound
+		}
+		return err
+	}
+	if row.UserID != userID {
+		return errBindingForbidden
+	}
+	return repository.DeleteMessageBinding(ctx, bindingID)
+}
+
+func toBindingDTO(row *model.MessageBinding, ch *model.MessageChannel) BindingDTO {
+	dto := BindingDTO{
+		ID:             row.ID,
+		UserID:         row.UserID,
+		ChannelID:      row.ChannelID,
+		PlatformUserID: row.PlatformUserID,
+		CreatedAt:      row.CreatedAt,
+	}
+	if ch != nil {
+		dto.ChannelName = ch.Name
+		dto.ChannelType = ch.Type
+	}
+	return dto
+}

plugins/domain/message_gateway/migrations/20260827000001_create_message_gateway_tables.sql

@@ -34,10 +34,59 @@ CREATE TABLE IF NOT EXISTS w_message_pairing_codes (
 );
 CREATE INDEX IF NOT EXISTS idx_w_message_pairing_lookup
     ON w_message_pairing_codes (channel_id, platform_user_id);
+
+CREATE TABLE IF NOT EXISTS w_push_events (
+    id BIGINT PRIMARY KEY,
+    event_key VARCHAR(80) NOT NULL,
+    name VARCHAR(100) NOT NULL,
+    task_type VARCHAR(100) NOT NULL DEFAULT '',
+    channels TEXT NOT NULL DEFAULT '',
+    targets TEXT NOT NULL DEFAULT '',
+    template TEXT NOT NULL DEFAULT '',
+    enabled BOOLEAN NOT NULL DEFAULT FALSE,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+CREATE UNIQUE INDEX IF NOT EXISTS uniq_w_push_events_key ON w_push_events(event_key);
+CREATE INDEX IF NOT EXISTS idx_w_push_events_enabled ON w_push_events(enabled);
+CREATE INDEX IF NOT EXISTS idx_w_push_events_task_type ON w_push_events(task_type);
+
+CREATE TABLE IF NOT EXISTS w_push_channels (
+    id BIGINT PRIMARY KEY,
+    name VARCHAR(80) NOT NULL,
+    description VARCHAR(255) NOT NULL DEFAULT '',
+    type VARCHAR(50) NOT NULL DEFAULT 'custom',
+    token VARCHAR(100) NOT NULL DEFAULT '',
+    url TEXT NOT NULL DEFAULT '',
+    other TEXT NOT NULL DEFAULT '',
+    enabled BOOLEAN NOT NULL DEFAULT TRUE,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+CREATE UNIQUE INDEX IF NOT EXISTS uniq_w_push_channels_name ON w_push_channels(name);
+CREATE INDEX IF NOT EXISTS idx_w_push_channels_enabled ON w_push_channels(enabled);
+
+CREATE TABLE IF NOT EXISTS w_push_histories (
+    id BIGINT PRIMARY KEY,
+    event_key VARCHAR(80) NOT NULL,
+    channel VARCHAR(50) NOT NULL,
+    target VARCHAR(255) NOT NULL,
+    title VARCHAR(255) NOT NULL,
+    content TEXT NOT NULL,
+    level VARCHAR(20) NOT NULL,
+    status VARCHAR(20) NOT NULL,
+    error_msg TEXT NOT NULL DEFAULT '',
+    created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+CREATE INDEX IF NOT EXISTS idx_w_push_histories_event ON w_push_histories(event_key);
+CREATE INDEX IF NOT EXISTS idx_w_push_histories_created ON w_push_histories(created_at);
 -- +goose StatementEnd
 
 -- +goose Down
 -- +goose StatementBegin
+DROP TABLE IF EXISTS w_push_histories;
+DROP TABLE IF EXISTS w_push_channels;
+DROP TABLE IF EXISTS w_push_events;
 DROP TABLE IF EXISTS w_message_pairing_codes;
 DROP TABLE IF EXISTS w_message_bindings;
 DROP TABLE IF EXISTS w_message_channels;

plugins/domain/message_gateway/plugin.go

@@ -1,3 +1,6 @@
+// Copyright 2026 Arctel.net
+// SPDX-License-Identifier: Apache-2.0
+
 // Package message_gateway provides the Bot gateway, multi-channel notification dispatching, and asynchronous push worker domain plugin for Cordis.
 package message_gateway
 
@@ -7,7 +10,7 @@ import (
 
 	"github.com/Rain-kl/Wavelet/core"
 	"github.com/Rain-kl/Wavelet/core/extpoints"
-	appgw "github.com/Rain-kl/Wavelet/internal/apps/message_gateway"
+	"github.com/Rain-kl/Wavelet/internal/apps/admin"
 	"github.com/Rain-kl/Wavelet/internal/apps/oauth"
 	"github.com/hibiken/asynq"
 )
@@ -18,8 +21,18 @@ var mgMigrations embed.FS
 // Option configures the message_gateway plugin.
 type Option func(*Plugin)
 
+// WithAutoStartRunner enables automatic bot runner startup in the background.
+func WithAutoStartRunner(enable bool) Option {
+	return func(p *Plugin) {
+		p.autoStartRunner = enable
+	}
+}
+
 // Plugin implements core.Plugin to provide Bot gateway and notification dispatch domain services.
-type Plugin struct{}
+type Plugin struct {
+	autoStartRunner bool
+	cancelRunner    context.CancelFunc
+}
 
 // New creates a new message_gateway domain plugin.
 func New(opts ...Option) *Plugin {
@@ -61,36 +74,100 @@ func (p *Plugin) Apply(ctx *core.Context) error {
 	// 1. Register migrations
 	ctx.Migrations().Register("message_gateway", mgMigrations)
 
-	// 2. Register HTTP Routes
+	// 2. Register User HTTP Routes
 	mgGroup := ctx.Router().Group("/api/v1/message-gateway", oauth.LoginRequired())
 	{
-		mgGroup.GET("/channels", appgw.ListChannels)
-		mgGroup.GET("/bindings", appgw.ListBindings)
-		mgGroup.POST("/bindings", appgw.BindBinding)
-		mgGroup.DELETE("/bindings/:id", appgw.UnbindBinding)
+		mgGroup.GET("/channels", ListChannels)
+		mgGroup.GET("/bindings", ListBindings)
+		mgGroup.POST("/bindings", BindBinding)
+		mgGroup.DELETE("/bindings/:id", UnbindBinding)
+	}
+
+	// 3. Register Admin Message Gateway HTTP Routes
+	adminMgGroup := ctx.Router().Group("/api/v1/admin/message-gateway", oauth.LoginRequired(), admin.LoginAdminRequired())
+	{
+		adminMgGroup.GET("/channels/definitions", ListAdminChannelDefinitions)
+		adminMgGroup.GET("/channels", ListAdminChannels)
+		adminMgGroup.POST("/channels", CreateAdminChannel)
+		adminMgGroup.PATCH("/channels/:id", UpdateAdminChannel)
+		adminMgGroup.DELETE("/channels/:id", DeleteAdminChannel)
+		adminMgGroup.POST("/channels/:id/test", TestAdminChannel)
+	}
+
+	// 4. Register Admin Push HTTP Routes
+	adminPushGroup := ctx.Router().Group("/api/v1/admin/push", oauth.LoginRequired(), admin.LoginAdminRequired())
+	{
+		events := adminPushGroup.Group("/events")
+		{
+			events.GET("", ListPushEvents)
+			events.GET("/builtin", ListBuiltInPushEvents)
+			events.POST("", CreatePushEvent)
+			events.PUT("/:id", UpdatePushEvent)
+			events.DELETE("/:id", DeletePushEvent)
+			events.POST("/:id/toggle", TogglePushEvent)
+		}
+
+		adminPushGroup.GET("/histories", ListPushHistories)
+		adminPushGroup.POST("/test", TestPush)
+
+		channels := adminPushGroup.Group("/channels")
+		{
+			channels.GET("/definitions", ListPushChannelDefinitions)
+			channels.GET("", ListPushChannels)
+			channels.POST("", CreatePushChannel)
+			channels.PUT("/:id", UpdatePushChannel)
+			channels.DELETE("/:id", DeletePushChannel)
+			channels.POST("/test", TestPushChannel)
+		}
 	}
 
 	const defaultTaskRetry = 3
+	pushHandler := &PushHandler{}
 
-	// 3. Register Asynq background tasks
-	ctx.Task().Register("message_gateway:push_notification", func(_ context.Context, _ *asynq.Task) error {
-		return nil
+	// 5. Register Asynq background tasks
+	ctx.Task().Register("message_gateway:push_notification", func(c context.Context, t *asynq.Task) error {
+		_, err := pushHandler.Execute(c, t.Payload())
+		return err
+	}, extpoints.WithTaskRetry(defaultTaskRetry))
+
+	ctx.Task().Register(SendNotificationTask, func(c context.Context, t *asynq.Task) error {
+		_, err := pushHandler.Execute(c, t.Payload())
+		return err
 	}, extpoints.WithTaskRetry(defaultTaskRetry))
 
 	ctx.Task().Register("message_gateway:dispatch_bot_msg", func(_ context.Context, _ *asynq.Task) error {
 		return nil
 	})
 
-	// 4. Register Cron Schedules
+	// 6. Register Cron Schedules
 	ctx.Schedule().RegisterCron("*/10 * * * *", "message_gateway:cleanup_pairing_codes", map[string]any{"action": "cleanup"})
 
-	// 5. Register EventBus listeners for decoupled push triggers
-	ctx.Events().On("notification:push", func(_ context.Context, _ PushNotificationEvent) error {
-		// Event triggered push handling
+	// 7. Register EventBus listeners for decoupled push triggers
+	ctx.Events().On("notification:push", func(c context.Context, e PushNotificationEvent) error {
+		meta := EventMetadata{
+			Key:  "eventbus:" + e.Channel,
+			Name: e.Title,
+			DefaultTemplate: NotificationMessage{
+				Title:   e.Title,
+				Content: e.Content,
+				Level:   defaultLevelInfo,
+				Ext:     e.Metadata,
+			},
+			Description: "EventBus triggered notification",
+		}
+		DefaultTrigger.Trigger(c, meta, map[string]any{
+			"user.id": e.UserID,
+			"title":   e.Title,
+			"content": e.Content,
+		})
 		return nil
 	})
 
-	// 6. Register Settings Schemas
+	// 8. Register built-in domain events and task listeners
+	RegisterCustomEvents()
+	RegisterTaskListeners()
+
+	// 9. Register Settings Schemas
 	ctx.Settings().Register(extpoints.SettingSchema{
 		Key:         "message_gateway.pairing_code_expiry_minutes",
 		Default:     15,
@@ -106,5 +183,21 @@ func (p *Plugin) Apply(ctx *core.Context) error {
 		Category:    "messaging",
 	})
 
+	// 10. Optional runner start & lifecycle
+	if p.autoStartRunner {
+		runnerCtx, cancel := context.WithCancel(ctx.GoContext())
+		p.cancelRunner = cancel
+		go func() {
+			_ = Start(runnerCtx)
+		}()
+	}
+
+	ctx.OnDispose(func() error {
+		if p.cancelRunner != nil {
+			p.cancelRunner()
+		}
+		return nil
+	})
+
 	return nil
 }

plugins/domain/message_gateway/push_channels.go

@@ -0,0 +1,398 @@
+// Copyright 2026 Arctel.net
+// SPDX-License-Identifier: Apache-2.0
+
+package message_gateway
+
+import (
+	"encoding/json"
+	"errors"
+	"net/http"
+	"strconv"
+	"strings"
+	"sync"
+
+	"github.com/Rain-kl/Wavelet/internal/model"
+	"github.com/Rain-kl/Wavelet/internal/shared/response"
+	pkgpush "github.com/Rain-kl/Wavelet/pkg/push"
+	"github.com/gin-gonic/gin"
+	"gorm.io/gorm"
+)
+
+const (
+	// KeyURL represents the URL field key
+	KeyURL = "url"
+	// KeyToken represents the Token field key
+	KeyToken = "token"
+	// KeyOther represents the Other field key
+	KeyOther = "other"
+
+	// TypeText represents standard text input type
+	TypeText = "text"
+	// TypePassword represents password input type
+	TypePassword = "password"
+	// TypeTextarea represents textarea input type
+	TypeTextarea = "textarea"
+)
+
+// PushField represents a form field configuration for a channel.
+type PushField struct {
+	Key         string `json:"key"`
+	Label       string `json:"label"`
+	Type        string `json:"type"`
+	Required    bool   `json:"required"`
+	Placeholder string `json:"placeholder"`
+	Description string `json:"description"`
+}
+
+// PushDefinition represents the metadata and form schema for a notification channel.
+type PushDefinition struct {
+	Type        string      `json:"type"`
+	Name        string      `json:"name"`
+	Description string      `json:"description"`
+	Fields      []PushField `json:"fields"`
+}
+
+var (
+	pushDefMu       sync.RWMutex
+	pushDefinitions = make(map[string]PushDefinition)
+)
+
+// RegisterPushChannelDefinition registers a channel definition.
+func RegisterPushChannelDefinition(def PushDefinition) {
+	pushDefMu.Lock()
+	defer pushDefMu.Unlock()
+	pushDefinitions[def.Type] = def
+}
+
+// ListPushDefinitions returns all registered channel definitions.
+func ListPushDefinitions() []PushDefinition {
+	pushDefMu.RLock()
+	defer pushDefMu.RUnlock()
+
+	order := []string{channelCustom, channelLark, channelTelegram, channelEmail}
+	res := make([]PushDefinition, 0, len(pushDefinitions))
+	for _, t := range order {
+		if d, ok := pushDefinitions[t]; ok {
+			res = append(res, d)
+		}
+	}
+	for t, d := range pushDefinitions {
+		found := false
+		for _, o := range order {
+			if o == t {
+				found = true
+				break
+			}
+		}
+		if !found {
+			res = append(res, d)
+		}
+	}
+	return res
+}
+
+func init() {
+	RegisterPushChannelDefinition(PushDefinition{
+		Type:        channelCustom,
+		Name:        "自定义消息通道",
+		Description: "使用自定义 HTTP POST 请求向外部 Webhook 发送数据。",
+		Fields: []PushField{
+			{
+				Key:         KeyURL,
+				Label:       "请求地址",
+				Type:        TypeText,
+				Required:    true,
+				Placeholder: "在此填写完整的请求地址，必须使用 HTTPS 协议",
+				Description: "接口请求的完整 HTTPS URL，例如 https://api.example.com/webhook",
+			},
+			{
+				Key:         KeyOther,
+				Label:       "请求体 (JSON)",
+				Type:        TypeTextarea,
+				Required:    true,
+				Placeholder: "在此输入请求体，支持模板变量，必须为合法的 JSON 格式",
+				Description: "可使用的变量：$title, $description, $content, $url, $to。例如 {\"text\": \"$content\"}",
+			},
+		},
+	})
+
+	RegisterPushChannelDefinition(PushDefinition{
+		Type:        channelLark,
+		Name:        "飞书群机器人",
+		Description: "配置飞书群自定义机器人的 Webhook 接口投递。",
+		Fields: []PushField{
+			{
+				Key:         KeyURL,
+				Label:       "Webhook 地址",
+				Type:        TypeText,
+				Required:    true,
+				Placeholder: "https://open.feishu.cn/open-apis/bot/v2/hook/YOUR_TOKEN",
+				Description: "从飞书群机器人设置中复制的 Webhook URL",
+			},
+			{
+				Key:         KeyToken,
+				Label:       "签名校验密钥 (Secret) (可选)",
+				Type:        TypeText,
+				Required:    false,
+				Placeholder: "可选，若机器人启用了安全设置中的签名校验，请在此输入",
+				Description: "飞书群机器人安全设置中的签名校验 Key",
+			},
+			{
+				Key:         KeyOther,
+				Label:       "自定义卡片 JSON 模版 (可选)",
+				Type:        TypeTextarea,
+				Required:    false,
+				Placeholder: "可选，留空则默认使用系统内置的精美互动卡片",
+				Description: "若填写，必须是合法的飞书卡片 JSON 格式",
+			},
+		},
+	})
+
+	RegisterPushChannelDefinition(PushDefinition{
+		Type:        channelTelegram,
+		Name:        "Telegram 机器人",
+		Description: "配置 Telegram 机器人推送消息。",
+		Fields: []PushField{
+			{
+				Key:         KeyURL,
+				Label:       "API 基础地址 (可选)",
+				Type:        TypeText,
+				Required:    false,
+				Placeholder: "https://api.telegram.org",
+				Description: "接口请求的 HTTPS 基础地址，留空默认为 https://api.telegram.org",
+			},
+			{
+				Key:         KeyToken,
+				Label:       "机器人 Token (Bot Token)",
+				Type:        TypePassword,
+				Required:    true,
+				Placeholder: "在此输入 Telegram 机器人的 Bot Token",
+				Description: "通过 BotFather 申请到的机器人 Access Token",
+			},
+			{
+				Key:         KeyOther,
+				Label:       "默认会话 ID (Chat ID) (可选)",
+				Type:        TypeText,
+				Required:    false,
+				Placeholder: "例如 -100123456789 或 @channel_name",
+				Description: "默认的消息接收 Chat ID。如果通知事件中未配置 targets，将推送到此 ID",
+			},
+		},
+	})
+
+	RegisterPushChannelDefinition(PushDefinition{
+		Type:        channelEmail,
+		Name:        "邮件推送通道",
+		Description: "邮件推送通道直接使用系统全局 SMTP 设置进行发送，无需在此填写服务器配置。",
+		Fields:      []PushField{},
+	})
+}
+
+// ListPushChannelDefinitions returns channel definitions.
+func ListPushChannelDefinitions(c *gin.Context) {
+	c.JSON(http.StatusOK, response.OK(ListPushDefinitions()))
+}
+
+// ListPushChannels lists configured push channels.
+func ListPushChannels(c *gin.Context) {
+	channels, err := listPushChannels(c.Request.Context())
+	if err != nil {
+		response.AbortInternal(c, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, response.OK(channels))
+}
+
+// CreatePushChannelRequest is the create channel request payload.
+type CreatePushChannelRequest struct {
+	Name        string `json:"name" binding:"required"`
+	Description string `json:"description"`
+	Type        string `json:"type" binding:"required"`
+	Token       string `json:"token"`
+	URL         string `json:"url"`
+	Other       string `json:"other"`
+	Enabled     bool   `json:"enabled"`
+}
+
+// CreatePushChannel creates a push channel.
+func CreatePushChannel(c *gin.Context) {
+	var req CreatePushChannelRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		response.AbortBadRequest(c, err.Error())
+		return
+	}
+
+	channel, err := createPushChannel(c.Request.Context(), req)
+	if err != nil {
+		response.AbortBadRequest(c, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, response.OK(channel))
+}
+
+// UpdatePushChannelRequest is the update channel request payload.
+type UpdatePushChannelRequest struct {
+	Description string `json:"description"`
+	Type        string `json:"type" binding:"required"`
+	Token       string `json:"token"`
+	URL         string `json:"url"`
+	Other       string `json:"other"`
+	Enabled     bool   `json:"enabled"`
+}
+
+// UpdatePushChannel updates a push channel.
+func UpdatePushChannel(c *gin.Context) {
+	id, err := strconv.ParseUint(c.Param("id"), 10, 64)
+	if err != nil {
+		response.AbortBadRequest(c, "invalid channel id")
+		return
+	}
+
+	var req UpdatePushChannelRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		response.AbortBadRequest(c, err.Error())
+		return
+	}
+
+	channel, err := updatePushChannel(c.Request.Context(), id, req)
+	if err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			response.AbortNotFound(c, "channel not found")
+			return
+		}
+		response.AbortInternal(c, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, response.OK(channel))
+}
+
+// DeletePushChannel deletes a push channel.
+func DeletePushChannel(c *gin.Context) {
+	id, err := strconv.ParseUint(c.Param("id"), 10, 64)
+	if err != nil {
+		response.AbortBadRequest(c, "invalid channel id")
+		return
+	}
+
+	if err := deletePushChannel(c.Request.Context(), id); err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			response.AbortNotFound(c, "channel not found")
+			return
+		}
+		response.AbortInternal(c, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, response.OKNil())
+}
+
+// TestPushChannelRequest is the test channel request payload.
+type TestPushChannelRequest struct {
+	Name   string `json:"name"`
+	Type   string `json:"type"`
+	Token  string `json:"token"`
+	URL    string `json:"url"`
+	Other  string `json:"other"`
+	Target string `json:"target"`
+}
+
+// TestPushChannel tests connectivity of a push channel.
+func TestPushChannel(c *gin.Context) {
+	var req TestPushChannelRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		response.AbortBadRequest(c, err.Error())
+		return
+	}
+
+	ctx := c.Request.Context()
+	url, token, other, channelType, err := loadChannelForTest(ctx, req)
+	if err != nil {
+		response.AbortBadRequest(c, err.Error())
+		return
+	}
+
+	if channelType == channelEmail {
+		url, token, other = resolveSMTPConfig(ctx, url, token, other)
+	}
+
+	tempChannel := model.PushChannel{
+		Name:    "test_temp",
+		URL:     url,
+		Token:   token,
+		Other:   other,
+		Type:    channelType,
+		Enabled: true,
+	}
+	if err := tempChannel.Validate(); err != nil {
+		response.AbortBadRequest(c, err.Error())
+		return
+	}
+	url = tempChannel.URL
+
+	var config pkgpush.Config
+	var renderedJSON string
+	switch channelType {
+	case channelLark:
+		config = pkgpush.Config{Channel: channelLark, URL: url, Secret: token}
+		renderedJSON = other
+	case channelEmail:
+		config = pkgpush.Config{Channel: channelEmail, URL: url, Key: token, Secret: other}
+	case channelTelegram:
+		config = pkgpush.Config{Channel: channelTelegram, URL: url, Secret: token, Key: other}
+	default:
+		config = pkgpush.Config{Channel: channelCustom, URL: url}
+		customPushReq := CustomPushRequest{
+			Title:       "通道测试通知",
+			Content:     "这是一条来自系统的消息通道连通性测试消息。",
+			Description: "系统通道测试",
+			URL:         "https://example.com",
+			To:          req.Target,
+		}
+		renderedJSON = renderCustomPayload(other, customPushReq)
+	}
+
+	payload := SendPayload{
+		EventKey: "test_channel",
+		Config:   config,
+		Target:   req.Target,
+		Body: NotificationMessage{
+			Title:   "通道测试通知",
+			Content: "这是一条来自系统的消息通道连通性测试消息。",
+			Level:   defaultLevelInfo,
+		},
+		Template: renderedJSON,
+	}
+	if err := enqueuePushTask(ctx, payload); err != nil {
+		response.AbortInternal(c, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, response.OKNil())
+}
+
+// CustomPushRequest contains custom webhook parameters.
+type CustomPushRequest struct {
+	Title       string `json:"title" form:"title"`
+	Description string `json:"description" form:"description"`
+	Content     string `json:"content" form:"content"`
+	URL         string `json:"url" form:"url"`
+	To          string `json:"to" form:"to"`
+	Token       string `json:"token" form:"token"`
+}
+
+func escapeJSONString(s string) string {
+	b, _ := json.Marshal(s)
+	const minJSONLen = 2
+	if len(b) >= minJSONLen {
+		return string(b[1 : len(b)-1])
+	}
+	return s
+}
+
+func renderCustomPayload(template string, req CustomPushRequest) string {
+	result := template
+	result = strings.ReplaceAll(result, "$title", escapeJSONString(req.Title))
+	result = strings.ReplaceAll(result, "$description", escapeJSONString(req.Description))
+	result = strings.ReplaceAll(result, "$content", escapeJSONString(req.Content))
+	result = strings.ReplaceAll(result, "$url", escapeJSONString(req.URL))
+	result = strings.ReplaceAll(result, "$to", escapeJSONString(req.To))
+	return result
+}

plugins/domain/message_gateway/push_constants.go

@@ -0,0 +1,15 @@
+// Copyright 2026 Arctel.net
+// SPDX-License-Identifier: Apache-2.0
+
+package message_gateway
+
+const (
+	channelCustom    = "custom"
+	channelEmail     = "email"
+	channelLark      = "lark"
+	channelTelegram  = "telegram"
+	defaultLevelInfo = "INFO"
+	keyTitle         = "title"
+	keyContent       = "content"
+	keyLevel         = "level"
+)

plugins/domain/message_gateway/push_events.go

@@ -0,0 +1,274 @@
+// Copyright 2026 Arctel.net
+// SPDX-License-Identifier: Apache-2.0
+
+package message_gateway
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"sync"
+
+	"github.com/Rain-kl/Wavelet/internal/model"
+	"github.com/Rain-kl/Wavelet/internal/repository"
+	"github.com/Rain-kl/Wavelet/pkg/logger"
+	pkgpush "github.com/Rain-kl/Wavelet/pkg/push"
+	"github.com/Rain-kl/Wavelet/pkg/util"
+	"gorm.io/gorm"
+)
+
+// NotificationMessage represents the structured notification message payload.
+type NotificationMessage struct {
+	Title   string         `json:"title"`
+	Content string         `json:"content"`
+	Level   string         `json:"level"`
+	Ext     map[string]any `json:"ext,omitempty"`
+}
+
+// Flatten converts the structured NotificationMessage back to a flat map (original json structure).
+func (m NotificationMessage) Flatten() map[string]any {
+	res := map[string]any{
+		keyTitle:   m.Title,
+		keyContent: m.Content,
+		keyLevel:   m.Level,
+	}
+	for k, v := range m.Ext {
+		res[k] = v
+	}
+	return res
+}
+
+// EventMetadata represents the metadata of a push notification event.
+type EventMetadata struct {
+	Key             string              `json:"key"`
+	Name            string              `json:"name"`
+	DefaultTemplate NotificationMessage `json:"default_template"`
+	Description     string              `json:"description"`
+}
+
+// SendPayload 异步投递推送载荷 (供 task/Worker 使用)
+type SendPayload struct {
+	EventKey string              `json:"event_key"`
+	Config   pkgpush.Config      `json:"config"`
+	Target   string              `json:"target"`
+	Body     NotificationMessage `json:"body"`
+	Template string              `json:"template"`
+}
+
+var (
+	builtInEventsMu sync.RWMutex
+	// BuiltInEvents lists all built-in events defined in custom_events.
+	BuiltInEvents []EventMetadata
+)
+
+// RegisterBuiltInEvent registers a built-in event definition.
+func RegisterBuiltInEvent(meta EventMetadata) {
+	builtInEventsMu.Lock()
+	defer builtInEventsMu.Unlock()
+	for i, e := range BuiltInEvents {
+		if e.Key == meta.Key {
+			BuiltInEvents[i] = meta
+			return
+		}
+	}
+	BuiltInEvents = append(BuiltInEvents, meta)
+}
+
+// GetBuiltInEvents returns a copy of registered built-in events.
+func GetBuiltInEvents() []EventMetadata {
+	builtInEventsMu.RLock()
+	defer builtInEventsMu.RUnlock()
+	out := make([]EventMetadata, len(BuiltInEvents))
+	copy(out, BuiltInEvents)
+	return out
+}
+
+// EventTrigger represents the unified event trigger class.
+type EventTrigger struct{}
+
+// DefaultTrigger is the singleton instance of EventTrigger.
+var DefaultTrigger = &EventTrigger{}
+
+// Trigger receives event metadata and processes the event notification dispatch asynchronously.
+//
+//nolint:contextcheck
+func (t *EventTrigger) Trigger(ctx context.Context, meta EventMetadata, body map[string]any) {
+	asyncCtx := context.WithoutCancel(ctx)
+	util.Go(func() {
+		if body == nil {
+			body = make(map[string]any)
+		}
+		if _, hasUser := body["user"]; !hasUser || body["user"] == nil {
+			body["user"] = getSystemUser(asyncCtx)
+		}
+
+		eventPtr, err := repository.GetActivePushEventByKey(asyncCtx, meta.Key)
+		if err != nil {
+			if errors.Is(err, gorm.ErrRecordNotFound) {
+				return
+			}
+			logger.ErrorF(asyncCtx, "push_event_trigger: failed to get active event %s: %v", meta.Key, err)
+			return
+		}
+		event := *eventPtr
+		if len(event.Channels) == 0 {
+			return
+		}
+
+		flatBody := getFlatBody(body)
+		msg, _ := t.buildMessage(&event, meta, flatBody, body)
+		t.enqueuePushTasks(asyncCtx, meta, &event, msg, flatBody)
+	})
+}
+
+func (t *EventTrigger) buildMessage(event *model.PushEvent, meta EventMetadata, flatBody map[string]any, body map[string]any) (NotificationMessage, string) {
+	var msg NotificationMessage
+	renderedTemplate := ""
+
+	templateSource := event.Template
+	if templateSource != "" {
+		var err error
+		msg, renderedTemplate, err = t.parseCustomTemplate(event, templateSource, flatBody)
+		if err != nil {
+			msg.Title = event.Name
+			msg.Content = renderedTemplate
+			msg.Level = defaultLevelInfo
+		}
+	} else {
+		msg = t.parseDefaultTemplate(meta, flatBody)
+	}
+
+	if msg.Ext == nil {
+		msg.Ext = make(map[string]any)
+	}
+	for k, v := range body {
+		if k == keyTitle || k == keyContent || k == keyLevel {
+			continue
+		}
+		if _, exists := msg.Ext[k]; !exists {
+			msg.Ext[k] = v
+		}
+	}
+
+	return msg, renderedTemplate
+}
+
+func (t *EventTrigger) parseCustomTemplate(event *model.PushEvent, templateSource string, flatBody map[string]any) (NotificationMessage, string, error) {
+	var msg NotificationMessage
+	renderedTemplate := pkgpush.ParseTemplate(templateSource, flatBody)
+
+	var tMap map[string]any
+	if err := json.Unmarshal([]byte(renderedTemplate), &tMap); err != nil {
+		return msg, renderedTemplate, err
+	}
+
+	if title, ok := tMap[keyTitle].(string); ok && title != "" {
+		msg.Title = title
+	} else {
+		msg.Title = event.Name
+	}
+	delete(tMap, keyTitle)
+
+	if content, ok := tMap[keyContent].(string); ok && content != "" {
+		msg.Content = content
+	} else {
+		msg.Content = renderedTemplate
+	}
+	delete(tMap, keyContent)
+
+	if level, ok := tMap[keyLevel].(string); ok && level != "" {
+		msg.Level = level
+	} else {
+		msg.Level = defaultLevelInfo
+	}
+	delete(tMap, keyLevel)
+
+	msg.Ext = tMap
+	return msg, renderedTemplate, nil
+}
+
+func (t *EventTrigger) parseDefaultTemplate(meta EventMetadata, flatBody map[string]any) NotificationMessage {
+	var msg NotificationMessage
+	msg.Title = pkgpush.ParseTemplate(meta.DefaultTemplate.Title, flatBody)
+	msg.Content = pkgpush.ParseTemplate(meta.DefaultTemplate.Content, flatBody)
+	msg.Level = pkgpush.ParseTemplate(meta.DefaultTemplate.Level, flatBody)
+
+	if meta.DefaultTemplate.Ext != nil {
+		msg.Ext = make(map[string]any)
+		for k, v := range meta.DefaultTemplate.Ext {
+			if strVal, ok := v.(string); ok {
+				msg.Ext[k] = pkgpush.ParseTemplate(strVal, flatBody)
+			} else {
+				msg.Ext[k] = v
+			}
+		}
+	}
+	return msg
+}
+
+func (t *EventTrigger) enqueuePushTasks(ctx context.Context, meta EventMetadata, event *model.PushEvent, msg NotificationMessage, flatBody map[string]any) {
+	for _, channelName := range event.Channels {
+		customChannel, err := repository.GetActivePushChannelByName(ctx, channelName)
+		if err == nil {
+			t.enqueueCustomPushChannelTasks(ctx, meta, event, customChannel, msg, flatBody)
+			continue
+		}
+		logger.WarnF(ctx, "push_event_trigger: channel %q not found in DB or disabled: %v", channelName, err)
+	}
+}
+
+func (t *EventTrigger) enqueueCustomPushChannelTasks(ctx context.Context, meta EventMetadata, event *model.PushEvent, channel *model.PushChannel, msg NotificationMessage, flatBody map[string]any) {
+	if len(event.Targets) == 0 {
+		t.enqueueSingleCustomPushChannelTask(ctx, meta, channel, "", msg)
+		return
+	}
+
+	for _, target := range event.Targets {
+		resolvedTarget := resolveTarget(ctx, target, flatBody, channel.Name)
+		t.enqueueSingleCustomPushChannelTask(ctx, meta, channel, resolvedTarget, msg)
+	}
+}
+
+func (t *EventTrigger) enqueueSingleCustomPushChannelTask(ctx context.Context, meta EventMetadata, channel *model.PushChannel, target string, msg NotificationMessage) {
+	var config pkgpush.Config
+	var renderedTemplate string
+
+	switch channel.Type {
+	case channelLark:
+		config = pkgpush.Config{Channel: channelLark, URL: channel.URL, Secret: channel.Token}
+		renderedTemplate = channel.Other
+	case channelEmail:
+		url, token, other := resolveSMTPConfig(ctx, channel.URL, channel.Token, channel.Other)
+		config = pkgpush.Config{Channel: channelEmail, URL: url, Key: token, Secret: other}
+	case channelTelegram:
+		config = pkgpush.Config{Channel: channelTelegram, URL: channel.URL, Secret: channel.Token, Key: channel.Other}
+	default:
+		config = pkgpush.Config{Channel: channelCustom, URL: channel.URL}
+		customPushReq := CustomPushRequest{
+			Title:       msg.Title,
+			Content:     msg.Content,
+			Description: meta.Description,
+			To:          target,
+		}
+		if urlVal, ok := msg.Ext["url"].(string); ok {
+			customPushReq.URL = urlVal
+		}
+		renderedTemplate = renderCustomPayload(channel.Other, customPushReq)
+	}
+
+	payload := SendPayload{
+		EventKey: meta.Key,
+		Config:   config,
+		Target:   target,
+		Body:     msg,
+		Template: renderedTemplate,
+	}
+	if err := enqueuePushTask(ctx, payload); err != nil {
+		logger.ErrorF(ctx, "push_event_trigger: enqueuePushTask failed for %s channel %s -> %s: %v", channel.Type, channel.Name, target, err)
+	}
+}
+
+// SyncEvents automatically registers/updates built-in events in the database.
+func SyncEvents(ctx context.Context) error {
+	return syncBuiltInEvents(ctx)
+}

plugins/domain/message_gateway/push_handlers.go

@@ -0,0 +1,197 @@
+// Copyright 2026 Arctel.net
+// SPDX-License-Identifier: Apache-2.0
+
+package message_gateway
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+	"strconv"
+
+	"github.com/Rain-kl/Wavelet/internal/repository"
+	"github.com/Rain-kl/Wavelet/internal/shared/response"
+	pkgpush "github.com/Rain-kl/Wavelet/pkg/push"
+	"github.com/gin-gonic/gin"
+	"gorm.io/gorm"
+)
+
+// UpdatePushEventRequest is the request body for updating a push event.
+type UpdatePushEventRequest struct {
+	Channels []string `json:"channels"`
+	Targets  []string `json:"targets"`
+	Template string   `json:"template" binding:"required"`
+	Enabled  bool     `json:"enabled"`
+}
+
+// CreatePushEventRequest is the request body for creating a push event.
+type CreatePushEventRequest struct {
+	EventKey string   `json:"event_key"`
+	TaskType string   `json:"task_type"`
+	Channels []string `json:"channels"`
+	Targets  []string `json:"targets"`
+	Template string   `json:"template"`
+	Enabled  bool     `json:"enabled"`
+}
+
+// TestPushRequest is the request body for testing push config.
+type TestPushRequest struct {
+	Config pkgpush.Config `json:"config" binding:"required"`
+	Target string         `json:"target"`
+}
+
+// ListPushEvents lists configured push events.
+func ListPushEvents(c *gin.Context) {
+	ctx := c.Request.Context()
+	events, err := listPushEvents(ctx)
+	if err != nil {
+		response.AbortInternal(c, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, response.OK(events))
+}
+
+// ListBuiltInPushEvents lists system built-in push event definitions.
+func ListBuiltInPushEvents(c *gin.Context) {
+	c.JSON(http.StatusOK, response.OK(GetBuiltInEvents()))
+}
+
+// CreatePushEvent creates a new push event configuration.
+func CreatePushEvent(c *gin.Context) {
+	var req CreatePushEventRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		response.AbortBadRequest(c, err.Error())
+		return
+	}
+
+	event, err := createPushEvent(c.Request.Context(), req)
+	if err != nil {
+		response.AbortBadRequest(c, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, response.OK(event))
+}
+
+// DeletePushEvent deletes a push event configuration by ID.
+func DeletePushEvent(c *gin.Context) {
+	id, err := strconv.ParseUint(c.Param("id"), 10, 64)
+	if err != nil {
+		response.AbortBadRequest(c, "invalid event id")
+		return
+	}
+
+	if err := deletePushEvent(c.Request.Context(), id); err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			response.AbortNotFound(c, "notification event not found")
+			return
+		}
+		response.AbortInternal(c, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, response.OKNil())
+}
+
+// UpdatePushEvent updates an existing push event.
+func UpdatePushEvent(c *gin.Context) {
+	id, err := strconv.ParseUint(c.Param("id"), 10, 64)
+	if err != nil {
+		response.AbortBadRequest(c, "invalid event id")
+		return
+	}
+
+	var req UpdatePushEventRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		response.AbortBadRequest(c, err.Error())
+		return
+	}
+
+	if err := updatePushEvent(c.Request.Context(), id, req); err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			response.AbortNotFound(c, "notification event not found")
+			return
+		}
+		response.AbortBadRequest(c, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, response.OKNil())
+}
+
+// TogglePushEvent toggles the enabled state of a push event.
+func TogglePushEvent(c *gin.Context) {
+	id, err := strconv.ParseUint(c.Param("id"), 10, 64)
+	if err != nil {
+		response.AbortBadRequest(c, "invalid event id")
+		return
+	}
+
+	enabled, err := togglePushEvent(c.Request.Context(), id)
+	if err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			response.AbortNotFound(c, "notification event not found")
+			return
+		}
+		response.AbortBadRequest(c, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, response.OK(enabled))
+}
+
+// ListPushHistories returns paginated push notification delivery histories.
+func ListPushHistories(c *gin.Context) {
+	page, _ := strconv.Atoi(c.DefaultQuery("page", "1"))
+	pageSize, _ := strconv.Atoi(c.DefaultQuery("page_size", "20"))
+	if page < 1 {
+		page = 1
+	}
+	if pageSize < 1 {
+		pageSize = 20
+	}
+
+	total, results, err := listPushHistories(c.Request.Context(), repository.PushHistoryListFilter{
+		EventKey: c.Query("event_key"),
+		Status:   c.Query("status"),
+		Page:     page,
+		PageSize: pageSize,
+	})
+	if err != nil {
+		response.AbortInternal(c, err.Error())
+		return
+	}
+
+	c.JSON(http.StatusOK, response.OK(map[string]any{
+		"total":   total,
+		"results": results,
+	}))
+}
+
+// TestPush executes a synchronous push test using the specified config.
+func TestPush(c *gin.Context) {
+	var req TestPushRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		response.AbortBadRequest(c, err.Error())
+		return
+	}
+
+	pusher, err := pkgpush.GetPusher(req.Config.Channel)
+	if err != nil {
+		response.AbortBadRequest(c, err.Error())
+		return
+	}
+	if err := pusher.ValidateConfig(req.Config); err != nil {
+		response.AbortBadRequest(c, fmt.Sprintf("validation failed: %v", err))
+		return
+	}
+
+	applySMTPFallbackToPushConfig(c.Request.Context(), &req.Config)
+
+	testBody := map[string]any{
+		keyTitle:   "测试通道推送",
+		keyContent: "当您收到这条消息，说明当前渠道连通性测试通过。",
+		keyLevel:   defaultLevelInfo,
+	}
+	if _, err := pusher.Send(c.Request.Context(), req.Config, req.Target, testBody, "", nil); err != nil {
+		response.AbortBadRequest(c, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, response.OKNil())
+}

plugins/domain/message_gateway/push_logics.go

@@ -0,0 +1,515 @@
+// Copyright 2026 Arctel.net
+// SPDX-License-Identifier: Apache-2.0
+
+package message_gateway
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strconv"
+	"strings"
+
+	"github.com/Rain-kl/Wavelet/internal/infra/task"
+	"github.com/Rain-kl/Wavelet/internal/model"
+	"github.com/Rain-kl/Wavelet/internal/repository"
+	pkgpush "github.com/Rain-kl/Wavelet/pkg/push"
+	"gorm.io/gorm"
+)
+
+type smtpConfig struct {
+	Host     string
+	Port     string
+	Username string
+	Password string
+}
+
+func loadSMTPConfig(ctx context.Context) smtpConfig {
+	host, _ := repository.GetSystemConfigByKey(ctx, model.ConfigKeySMTPHost)
+	port, _ := repository.GetSystemConfigByKey(ctx, model.ConfigKeySMTPPort)
+	user, _ := repository.GetSystemConfigByKey(ctx, model.ConfigKeySMTPUsername)
+	pass, _ := repository.GetSystemConfigByKey(ctx, model.ConfigKeySMTPPassword)
+	return smtpConfig{
+		Host:     host.Value,
+		Port:     port.Value,
+		Username: user.Value,
+		Password: pass.Value,
+	}
+}
+
+func syncBuiltInEvents(ctx context.Context) error {
+	for _, meta := range GetBuiltInEvents() {
+		_, err := repository.GetPushEventByKey(ctx, meta.Key)
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			var defaultTemplateStr string
+			if defaultTemplateBytes, err := json.Marshal(meta.DefaultTemplate); err == nil {
+				defaultTemplateStr = string(defaultTemplateBytes)
+			}
+			event := model.PushEvent{
+				EventKey: meta.Key,
+				Name:     meta.Name,
+				Channels: []string{},
+				Targets:  []string{},
+				Template: defaultTemplateStr,
+				Enabled:  false,
+			}
+			if err := repository.CreatePushEvent(ctx, &event); err != nil {
+				return err
+			}
+		} else if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func listPushEvents(ctx context.Context) ([]model.PushEvent, error) {
+	return repository.ListPushEvents(ctx)
+}
+
+func createPushEvent(ctx context.Context, req CreatePushEventRequest) (model.PushEvent, error) {
+	eventKey, eventName, defaultTemplateBytes, err := getEventInfo(req)
+	if err != nil {
+		return model.PushEvent{}, err
+	}
+
+	count, err := repository.CountPushEventsByKey(ctx, eventKey)
+	if err != nil {
+		return model.PushEvent{}, err
+	}
+	if count > 0 {
+		return model.PushEvent{}, errors.New("this notification event is already configured")
+	}
+
+	templateStr := strings.TrimSpace(req.Template)
+	if templateStr == "" {
+		templateStr = string(defaultTemplateBytes)
+	} else {
+		var tempMap map[string]any
+		if err := json.Unmarshal([]byte(templateStr), &tempMap); err != nil {
+			return model.PushEvent{}, errors.New("custom template is not a valid JSON format")
+		}
+	}
+
+	channels := req.Channels
+	if channels == nil {
+		channels = []string{}
+	}
+	targets := req.Targets
+	if targets == nil {
+		targets = []string{}
+	}
+
+	event := model.PushEvent{
+		EventKey: eventKey,
+		Name:     eventName,
+		TaskType: req.TaskType,
+		Channels: channels,
+		Targets:  targets,
+		Template: templateStr,
+		Enabled:  req.Enabled,
+	}
+	if err := event.Validate(); err != nil {
+		return model.PushEvent{}, err
+	}
+	if err := repository.CreatePushEvent(ctx, &event); err != nil {
+		return model.PushEvent{}, err
+	}
+	return event, nil
+}
+
+func deletePushEvent(ctx context.Context, id uint64) error {
+	event, err := repository.GetPushEventByID(ctx, id)
+	if err != nil {
+		return err
+	}
+	return repository.DeletePushEvent(ctx, &event)
+}
+
+func updatePushEvent(ctx context.Context, id uint64, req UpdatePushEventRequest) error {
+	event, err := repository.GetPushEventByID(ctx, id)
+	if err != nil {
+		return err
+	}
+
+	event.Channels = req.Channels
+	event.Targets = req.Targets
+	event.Template = req.Template
+	event.Enabled = req.Enabled
+	if err := event.Validate(); err != nil {
+		return err
+	}
+	return repository.SavePushEvent(ctx, &event)
+}
+
+func togglePushEvent(ctx context.Context, id uint64) (bool, error) {
+	event, err := repository.GetPushEventByID(ctx, id)
+	if err != nil {
+		return false, err
+	}
+
+	enabled := !event.Enabled
+	if enabled && len(event.Channels) == 0 {
+		return false, errors.New("cannot enable event without any push channels configured")
+	}
+	if err := repository.UpdatePushEventEnabled(ctx, &event, enabled); err != nil {
+		return false, err
+	}
+	return enabled, nil
+}
+
+func listPushHistories(ctx context.Context, filter repository.PushHistoryListFilter) (int64, []model.PushHistory, error) {
+	return repository.ListPushHistories(ctx, filter)
+}
+
+func applySMTPFallbackToPushConfig(ctx context.Context, cfg *pkgpush.Config) {
+	if cfg.Channel != channelEmail || (cfg.URL != "" && cfg.Key != "") {
+		return
+	}
+	smtp := loadSMTPConfig(ctx)
+	if smtp.Host == "" || smtp.Username == "" {
+		return
+	}
+	port := smtp.Port
+	if port == "" {
+		port = "587"
+	}
+	cfg.URL = smtp.Host + ":" + port
+	cfg.Key = smtp.Username
+	cfg.Secret = smtp.Password
+}
+
+func listPushChannels(ctx context.Context) ([]model.PushChannel, error) {
+	return repository.ListPushChannels(ctx)
+}
+
+func createPushChannel(ctx context.Context, req CreatePushChannelRequest) (model.PushChannel, error) {
+	count, err := repository.CountPushChannelsByName(ctx, req.Name)
+	if err != nil {
+		return model.PushChannel{}, err
+	}
+	if count > 0 {
+		return model.PushChannel{}, errors.New("channel name already exists")
+	}
+
+	channel := model.PushChannel{
+		Name:        req.Name,
+		Description: req.Description,
+		Type:        req.Type,
+		Token:       req.Token,
+		URL:         req.URL,
+		Other:       req.Other,
+		Enabled:     req.Enabled,
+	}
+	if err := channel.Validate(); err != nil {
+		return model.PushChannel{}, err
+	}
+	if err := repository.CreatePushChannel(ctx, &channel); err != nil {
+		return model.PushChannel{}, err
+	}
+	return channel, nil
+}
+
+func updatePushChannel(ctx context.Context, id uint64, req UpdatePushChannelRequest) (model.PushChannel, error) {
+	channel, err := repository.GetPushChannelByID(ctx, id)
+	if err != nil {
+		return model.PushChannel{}, err
+	}
+
+	channel.Description = req.Description
+	channel.Type = req.Type
+	channel.Token = req.Token
+	channel.URL = req.URL
+	channel.Other = req.Other
+	channel.Enabled = req.Enabled
+	if err := channel.Validate(); err != nil {
+		return model.PushChannel{}, err
+	}
+	if err := repository.SavePushChannel(ctx, &channel); err != nil {
+		return model.PushChannel{}, err
+	}
+	return channel, nil
+}
+
+func deletePushChannel(ctx context.Context, id uint64) error {
+	channel, err := repository.GetPushChannelByID(ctx, id)
+	if err != nil {
+		return err
+	}
+	return repository.DeletePushChannel(ctx, &channel)
+}
+
+func loadChannelForTest(ctx context.Context, req TestPushChannelRequest) (string, string, string, string, error) {
+	if req.Name != "" {
+		channel, err := repository.GetPushChannelByName(ctx, req.Name)
+		if err != nil {
+			return "", "", "", "", errors.New("channel not found")
+		}
+		return channel.URL, channel.Token, channel.Other, channel.Type, nil
+	}
+	return req.URL, req.Token, req.Other, req.Type, nil
+}
+
+func listActivePushEventsByTaskType(ctx context.Context, taskType string) ([]model.PushEvent, error) {
+	return repository.ListActivePushEventsByTaskType(ctx, taskType)
+}
+
+func loadUserFromPayload(ctx context.Context, data map[string]any) any {
+	if u, exists := data["user"]; exists && u != nil {
+		return u
+	}
+
+	if userID, ok := extractUserID(data); ok && userID > 0 {
+		if user, err := repository.GetUserByID(ctx, userID); err == nil {
+			return &user
+		}
+	}
+
+	if username := extractUsername(data); username != "" {
+		if user, err := repository.GetUserByUsername(ctx, username); err == nil {
+			return &user
+		}
+	}
+	return nil
+}
+
+func recordPushHistory(ctx context.Context, req SendPayload, status, errMsg string) error {
+	title := req.Body.Title
+	content := req.Body.Content
+	level := req.Body.Level
+	if title == "" {
+		title = "系统通知"
+	}
+	if level == "" {
+		level = defaultLevelInfo
+	}
+
+	target := req.Target
+	if target == "" {
+		if req.Config.URL != "" {
+			target = req.Config.URL
+			const maxTargetLen = 50
+			const truncatedLen = 47
+			if len(target) > maxTargetLen {
+				target = target[:truncatedLen] + "..."
+			}
+		} else {
+			target = "default"
+		}
+	}
+
+	history := model.PushHistory{
+		EventKey: req.EventKey,
+		Channel:  req.Config.Channel,
+		Target:   target,
+		Title:    title,
+		Content:  content,
+		Level:    level,
+		Status:   status,
+		ErrorMsg: errMsg,
+	}
+	return repository.CreatePushHistory(ctx, &history)
+}
+
+func resolveTarget(ctx context.Context, target string, flatBody map[string]any, channel string) string {
+	target = strings.TrimSpace(target)
+	if target == "" {
+		return ""
+	}
+
+	resolved := resolveDynamicKeyword(target, flatBody)
+	if strings.Contains(resolved, "@") {
+		return resolved
+	}
+	if val, matched := resolveSystemTarget(ctx, resolved, channel); matched {
+		return val
+	}
+
+	user, found := resolveTargetUser(ctx, resolved, channel)
+	if !found {
+		return resolved
+	}
+	if channel == channelEmail && user.Email != "" {
+		return user.Email
+	}
+	if channel != channelEmail && user.Username != "" {
+		return user.Username
+	}
+	return resolved
+}
+
+func resolveDynamicKeyword(target string, flatBody map[string]any) string {
+	switch target {
+	case "user.id", "id":
+		if val, ok := flatBody["user.id"]; ok {
+			return fmt.Sprintf("%v", val)
+		}
+		if val, ok := flatBody["id"]; ok {
+			return fmt.Sprintf("%v", val)
+		}
+	case "user.username", "username":
+		if val, ok := flatBody["user.username"]; ok {
+			return fmt.Sprintf("%v", val)
+		}
+		if val, ok := flatBody["username"]; ok {
+			return fmt.Sprintf("%v", val)
+		}
+	case "user.email", channelEmail:
+		if val, ok := flatBody["user.email"]; ok {
+			return fmt.Sprintf("%v", val)
+		}
+		if val, ok := flatBody["email"]; ok {
+			return fmt.Sprintf("%v", val)
+		}
+	}
+	return target
+}
+
+func resolveTargetUser(ctx context.Context, resolved string, _ string) (model.User, bool) {
+	found := false
+	var user model.User
+
+	if id, err := strconv.ParseUint(resolved, 10, 64); err == nil {
+		if u, err := repository.GetUserByID(ctx, id); err == nil {
+			user = u
+			found = true
+		}
+	}
+	if !found {
+		if u, err := repository.GetUserByUsername(ctx, resolved); err == nil {
+			user = u
+			found = true
+		}
+	}
+	return user, found
+}
+
+func resolveSystemTarget(ctx context.Context, resolved string, channel string) (string, bool) {
+	if resolved != "系统" && resolved != "system" && resolved != "0" {
+		return "", false
+	}
+	adminUser, err := repository.GetFirstAdminUser(ctx)
+	if err != nil {
+		return resolved, true
+	}
+	if channel == channelEmail && adminUser.Email != "" {
+		return adminUser.Email, true
+	}
+	if channel != channelEmail && adminUser.Username != "" {
+		return adminUser.Username, true
+	}
+	return resolved, true
+}
+
+func resolveSMTPConfig(ctx context.Context, url, token, other string) (string, string, string) {
+	if url != "" && token != "" {
+		return url, token, other
+	}
+	smtp := loadSMTPConfig(ctx)
+	if smtp.Host == "" || smtp.Username == "" {
+		return url, token, other
+	}
+	port := smtp.Port
+	if port == "" {
+		port = "587"
+	}
+	if url == "" {
+		url = smtp.Host + ":" + port
+	}
+	if token == "" {
+		token = smtp.Username
+	}
+	if other == "" {
+		other = smtp.Password
+	}
+	return url, token, other
+}
+
+func getSystemUser(ctx context.Context) *model.User {
+	user := repository.GetSystemUser(ctx)
+	return &user
+}
+
+func findBuiltInEvent(key string) (EventMetadata, bool) {
+	for _, meta := range GetBuiltInEvents() {
+		if meta.Key == key {
+			return meta, true
+		}
+	}
+	return EventMetadata{}, false
+}
+
+func getEventInfo(req CreatePushEventRequest) (string, string, []byte, error) {
+	if req.TaskType != "" {
+		meta := task.GetTaskMetaByAsynqTask(req.TaskType)
+		if meta == nil {
+			return "", "", nil, errors.New("unsupported task type")
+		}
+		eventKey := "task_completed:" + req.TaskType
+		eventName := "任务完成: " + meta.Name
+		defaultTemplate := NotificationMessage{
+			Title:   "任务完成: " + meta.Name,
+			Content: "异步任务 {{task_name}} (ID: {{task_id}}) 已完成。状态: {{task_status}}，耗时: {{task_duration}} ms。",
+			Level:   defaultLevelInfo,
+		}
+		defaultTemplateBytes, err := json.Marshal(defaultTemplate)
+		if err != nil {
+			return "", "", nil, err
+		}
+		return eventKey, eventName, defaultTemplateBytes, nil
+	}
+
+	if req.EventKey == "" {
+		return "", "", nil, errors.New("either event_key or task_type must be provided")
+	}
+
+	meta, found := findBuiltInEvent(req.EventKey)
+	if !found {
+		return "", "", nil, errors.New("unsupported built-in event key")
+	}
+
+	defaultTemplateBytes, err := json.Marshal(meta.DefaultTemplate)
+	if err != nil {
+		return "", "", nil, err
+	}
+	return req.EventKey, meta.Name, defaultTemplateBytes, nil
+}
+
+func enqueuePushTask(ctx context.Context, payload SendPayload) error {
+	payloadBytes, err := json.Marshal(payload)
+	if err != nil {
+		return err
+	}
+	_, err = task.DispatchTask(ctx, "send_notification", payloadBytes, "system")
+	return err
+}
+
+func getFlatBody(body map[string]any) map[string]any {
+	jsonBytes, err := json.Marshal(body)
+	if err != nil {
+		return body
+	}
+	var jsonMap map[string]any
+	if err := json.Unmarshal(jsonBytes, &jsonMap); err != nil {
+		return body
+	}
+
+	flatResult := make(map[string]any)
+	flattenMap("", jsonMap, flatResult)
+	return flatResult
+}
+
+func flattenMap(prefix string, m map[string]any, result map[string]any) {
+	for k, v := range m {
+		key := k
+		if prefix != "" {
+			key = prefix + "." + k
+		}
+		if nestedMap, ok := v.(map[string]any); ok {
+			flattenMap(key, nestedMap, result)
+		} else {
+			result[key] = v
+		}
+	}
+}

plugins/domain/message_gateway/push_task_listener.go

@@ -0,0 +1,124 @@
+// Copyright 2026 Arctel.net
+// SPDX-License-Identifier: Apache-2.0
+
+package message_gateway
+
+import (
+	"context"
+	"encoding/json"
+	"strconv"
+	"time"
+
+	"github.com/Rain-kl/Wavelet/internal/infra/task"
+	"github.com/Rain-kl/Wavelet/internal/model"
+	"github.com/Rain-kl/Wavelet/pkg/logger"
+)
+
+// RegisterTaskListeners subscribes push notification handlers to task completion events.
+func RegisterTaskListeners() {
+	task.OnTaskCompleted(handleTaskCompleted)
+}
+
+func handleTaskCompleted(ctx context.Context, execution *model.TaskExecution, result *task.TaskResult, execErr error) {
+	events, err := listActivePushEventsByTaskType(ctx, execution.TaskType)
+	if err != nil {
+		logger.ErrorF(ctx, "push_task_completed_listener: failed to query push events for task type %s: %v", execution.TaskType, err)
+		return
+	}
+	if len(events) == 0 {
+		return
+	}
+
+	body := map[string]any{
+		"task_id":       execution.TaskID,
+		"task_name":     execution.TaskName,
+		"task_type":     execution.TaskType,
+		"task_status":   string(execution.Status),
+		"task_duration": execution.Duration,
+		"time":          time.Now().Format("2006-01-02 15:04:05"),
+	}
+	if execErr != nil {
+		body["task_error"] = execErr.Error()
+	} else {
+		body["task_error"] = ""
+	}
+	if result != nil {
+		body["task_result"] = result.Message
+	} else {
+		body["task_result"] = ""
+	}
+
+	var payloadMap map[string]any
+	if execution.Payload != "" {
+		if err := json.Unmarshal([]byte(execution.Payload), &payloadMap); err == nil {
+			body["payload"] = payloadMap
+			extractUserFromMap(ctx, payloadMap, body)
+		}
+	}
+	if result != nil && result.Detail != "" {
+		var detailMap map[string]any
+		if err := json.Unmarshal([]byte(result.Detail), &detailMap); err == nil {
+			body["detail"] = detailMap
+			extractUserFromMap(ctx, detailMap, body)
+		}
+	}
+
+	for _, event := range events {
+		meta := EventMetadata{
+			Key:         event.EventKey,
+			Name:        event.Name,
+			Description: "异步任务执行完毕触发的自动通知",
+		}
+		DefaultTrigger.Trigger(ctx, meta, body)
+	}
+}
+
+func extractUserFromMap(ctx context.Context, data map[string]any, body map[string]any) {
+	if u, exists := body["user"]; exists && u != nil {
+		return
+	}
+	if user := loadUserFromPayload(ctx, data); user != nil {
+		body["user"] = user
+	}
+}
+
+func extractUserID(data map[string]any) (uint64, bool) {
+	for _, k := range []string{"user_id", "userId", "uid"} {
+		val, ok := data[k]
+		if !ok || val == nil {
+			continue
+		}
+		switch v := val.(type) {
+		case float64:
+			if v >= 0 {
+				return uint64(v), true
+			}
+		case int:
+			if v >= 0 {
+				return uint64(v), true
+			}
+		case int64:
+			if v >= 0 {
+				return uint64(v), true
+			}
+		case uint64:
+			return v, true
+		case string:
+			if id, err := strconv.ParseUint(v, 10, 64); err == nil {
+				return id, true
+			}
+		}
+	}
+	return 0, false
+}
+
+func extractUsername(data map[string]any) string {
+	for _, k := range []string{"username", "user_name"} {
+		if val, ok := data[k]; ok && val != nil {
+			if s, ok := val.(string); ok && s != "" {
+				return s
+			}
+		}
+	}
+	return ""
+}

plugins/domain/message_gateway/push_tasks.go

@@ -0,0 +1,123 @@
+// Copyright 2026 Arctel.net
+// SPDX-License-Identifier: Apache-2.0
+
+package message_gateway
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"github.com/Rain-kl/Wavelet/internal/infra/task"
+	"github.com/Rain-kl/Wavelet/pkg/push"
+)
+
+const (
+	// SendNotificationTask is the asynq task name for push notification.
+	SendNotificationTask = "push:send"
+	// TaskTypeSendNotification is the admin task manager type identifier.
+	TaskTypeSendNotification = "send_notification"
+)
+
+// SendNotificationMeta represents the task metadata.
+var SendNotificationMeta = task.TaskMeta{
+	Type:         TaskTypeSendNotification,
+	AsynqTask:    SendNotificationTask,
+	Name:         "推送通知",
+	Description:  "异步执行系统通知的多渠道派发与推送",
+	SupportsTime: false,
+	MaxRetry:     task.DefaultMaxRetry,
+	Queue:        task.QueueDefault,
+	Retryable:    true,
+	Params: []task.TaskParam{
+		{
+			Name:        "event_key",
+			Label:       "事件标识",
+			Type:        "string",
+			Required:    true,
+			Placeholder: "admin_login",
+		},
+		{
+			Name:     "target",
+			Label:    "目标接收者",
+			Type:     "string",
+			Required: false,
+		},
+	},
+}
+
+// PushHandler handles asynchronous notification sending.
+type PushHandler struct{}
+
+// ValidatePayload validates and normalizes push parameters.
+func (h *PushHandler) ValidatePayload(payload []byte) ([]byte, error) {
+	if len(payload) == 0 {
+		return nil, errors.New("payload is required")
+	}
+
+	var req SendPayload
+	if err := json.Unmarshal(payload, &req); err != nil {
+		return nil, fmt.Errorf("invalid json format: %w", err)
+	}
+
+	if req.Config.Channel == "" {
+		return nil, errors.New("channel type is required")
+	}
+
+	return json.Marshal(req)
+}
+
+// Execute performs the push send and logs delivery history audit.
+func (h *PushHandler) Execute(ctx context.Context, payload []byte) (*task.TaskResult, error) {
+	var req SendPayload
+	if err := json.Unmarshal(payload, &req); err != nil {
+		task.AppendLog(ctx, "解析推送参数失败: %v", err)
+		return nil, fmt.Errorf("parse payload failed: %w", err)
+	}
+
+	task.AppendLog(ctx, "开始推送通知: 事件 = %s, 渠道 = %s, 接收目标 = %s", req.EventKey, req.Config.Channel, req.Target)
+
+	pusher, err := push.GetPusher(req.Config.Channel)
+	if err != nil {
+		errWrap := fmt.Errorf("get pusher failed: %w", err)
+		task.AppendLog(ctx, "推送失败: %v", errWrap)
+		if task.IsFinalAttempt(ctx) {
+			h.recordHistory(ctx, req, "failed", errWrap.Error())
+		}
+		return nil, errWrap
+	}
+
+	flatBody := req.Body.Flatten()
+	upstreamResp, err := pusher.Send(ctx, req.Config, req.Target, flatBody, req.Template, nil)
+
+	title := req.Body.Title
+	content := req.Body.Content
+
+	if err != nil {
+		task.AppendLog(ctx, "消息推送失败 (标题: %s): %v", title, err)
+		if upstreamResp != "" {
+			task.AppendLog(ctx, "上游返回: %s", upstreamResp)
+		}
+		if task.IsFinalAttempt(ctx) {
+			h.recordHistory(ctx, req, "failed", err.Error())
+		}
+		return nil, fmt.Errorf("pusher.Send failed: %w", err)
+	}
+
+	task.AppendLog(ctx, "消息推送成功 (标题: %s, 内容摘要: %s)", title, content)
+	if upstreamResp != "" {
+		task.AppendLog(ctx, "上游返回: %s", upstreamResp)
+	}
+	h.recordHistory(ctx, req, "success", "")
+
+	return &task.TaskResult{
+		Message: fmt.Sprintf("推送成功: [%s] -> %s", req.Config.Channel, req.Target),
+	}, nil
+}
+
+func (h *PushHandler) recordHistory(ctx context.Context, req SendPayload, status string, errMsg string) {
+	if dbErr := recordPushHistory(ctx, req, status, errMsg); dbErr != nil {
+		task.AppendLog(ctx, "写入推送历史审计记录失败: %v", dbErr)
+	}
+}

plugins/domain/message_gateway/runner.go

@@ -0,0 +1,53 @@
+// Copyright 2026 Arctel.net
+// SPDX-License-Identifier: Apache-2.0
+
+package message_gateway
+
+import (
+	"context"
+	"sync"
+
+	"github.com/Rain-kl/Wavelet/pkg/logger"
+)
+
+// Runner manages lifecycle for long-lived channel adapters (WebSocket, long-polling, etc.).
+type Runner struct {
+	mu      sync.Mutex
+	running bool
+	cancel  context.CancelFunc
+}
+
+// GlobalRunner is the default global runner instance.
+var GlobalRunner = &Runner{}
+
+// Start starts all background long-lived channel runners.
+func Start(ctx context.Context) error {
+	GlobalRunner.mu.Lock()
+	defer GlobalRunner.mu.Unlock()
+
+	if GlobalRunner.running {
+		return nil
+	}
+
+	runCtx, cancel := context.WithCancel(ctx)
+	GlobalRunner.cancel = cancel
+	GlobalRunner.running = true
+
+	logger.InfoF(runCtx, "[MessageGateway] Starting bot channel runners...")
+	return nil
+}
+
+// Stop stops the channel runner.
+func Stop() {
+	GlobalRunner.mu.Lock()
+	defer GlobalRunner.mu.Unlock()
+
+	if !GlobalRunner.running {
+		return
+	}
+
+	if GlobalRunner.cancel != nil {
+		GlobalRunner.cancel()
+	}
+	GlobalRunner.running = false
+}

plugins/domain/message_gateway/secret.go

@@ -0,0 +1,78 @@
+// Copyright 2026 Arctel.net
+// SPDX-License-Identifier: Apache-2.0
+
+package message_gateway
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+
+	"github.com/Rain-kl/Wavelet/internal/infra/config"
+	"github.com/Rain-kl/Wavelet/pkg/util"
+)
+
+// CredentialKey is AES-256 hex derived from the session secret.
+func CredentialKey() string {
+	secret := ""
+	if config.Config != nil {
+		secret = config.Config.App.SessionSecret
+	}
+	sum := sha256.Sum256([]byte(secret))
+	return hex.EncodeToString(sum[:])
+}
+
+// EncryptCredentials encrypts a credential map as JSON.
+func EncryptCredentials(creds map[string]string) (string, error) {
+	if creds == nil {
+		creds = map[string]string{}
+	}
+	raw, err := json.Marshal(creds)
+	if err != nil {
+		return "", err
+	}
+	return util.Encrypt(CredentialKey(), string(raw))
+}
+
+// DecryptCredentials decrypts a credential map.
+func DecryptCredentials(ciphertext string) (map[string]string, error) {
+	if ciphertext == "" {
+		return map[string]string{}, nil
+	}
+	plain, err := util.Decrypt(CredentialKey(), ciphertext)
+	if err != nil {
+		return nil, err
+	}
+	var out map[string]string
+	if err := json.Unmarshal([]byte(plain), &out); err != nil {
+		return nil, err
+	}
+	if out == nil {
+		out = map[string]string{}
+	}
+	return out, nil
+}
+
+// ParseExtra decodes optional extra JSON into a string map.
+func ParseExtra(raw string) map[string]string {
+	if raw == "" {
+		return map[string]string{}
+	}
+	var out map[string]string
+	if err := json.Unmarshal([]byte(raw), &out); err != nil || out == nil {
+		return map[string]string{}
+	}
+	return out
+}
+
+// EncodeExtra encodes extra fields as JSON.
+func EncodeExtra(extra map[string]string) string {
+	if extra == nil {
+		return ""
+	}
+	raw, err := json.Marshal(extra)
+	if err != nil {
+		return ""
+	}
+	return string(raw)
+}