diff --git a/internal/listener/message_gateway.go b/internal/listener/message_gateway.go new file mode 100644 index 00000000..575c85a0 --- /dev/null +++ b/internal/listener/message_gateway.go @@ -0,0 +1,39 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package listener + +import ( + "context" + + "github.com/Rain-kl/Wavelet/pkg/message_gateway" +) + +// EventMessageGatewayInbound is the domain event name for authorized inbound messages. +const EventMessageGatewayInbound = "message_gateway.inbound" + +// MessageGatewayInbound is emitted when a bound user sends a private message. +type MessageGatewayInbound struct { + Msg message_gateway.InboundMessage +} + +// MessageGatewayInboundHandler handles inbound messaging events. +type MessageGatewayInboundHandler func(ctx context.Context, event MessageGatewayInbound) + +var messageGatewayInboundHandlers []MessageGatewayInboundHandler + +// OnMessageGatewayInbound registers a handler. Call from bootstrap only. +func OnMessageGatewayInbound(handler MessageGatewayInboundHandler) { + messageGatewayInboundHandlers = append(messageGatewayInboundHandlers, handler) +} + +// EmitMessageGatewayInbound dispatches a bound inbound message. +func EmitMessageGatewayInbound(ctx context.Context, msg message_gateway.InboundMessage) { + if msg.BindingUserID == nil { + return + } + event := MessageGatewayInbound{Msg: msg} + for _, handler := range messageGatewayInboundHandlers { + handler(ctx, event) + } +} diff --git a/internal/listener/message_gateway_test.go b/internal/listener/message_gateway_test.go new file mode 100644 index 00000000..edba8d97 --- /dev/null +++ b/internal/listener/message_gateway_test.go @@ -0,0 +1,25 @@ +// Copyright 2026 Arctel.net +// SPDX-License-Identifier: Apache-2.0 + +package listener + +import ( + "context" + "testing" + + "github.com/Rain-kl/Wavelet/pkg/message_gateway" +) + +func TestEmitMessageGatewayInbound_SkipsUnbound(t *testing.T) { + called := 0 + OnMessageGatewayInbound(func(ctx context.Context, ev MessageGatewayInbound) { called++ }) + EmitMessageGatewayInbound(context.Background(), message_gateway.InboundMessage{Text: "x"}) + if called != 0 { + t.Fatal("unbound must not emit") + } + uid := uint64(9) + EmitMessageGatewayInbound(context.Background(), message_gateway.InboundMessage{BindingUserID: &uid, Text: "x"}) + if called != 1 { + t.Fatalf("called=%d", called) + } +} diff --git a/internal/platform/bootstrap/bootstrap.go b/internal/platform/bootstrap/bootstrap.go index d005cb26..cc86c696 100644 --- a/internal/platform/bootstrap/bootstrap.go +++ b/internal/platform/bootstrap/bootstrap.go @@ -15,6 +15,7 @@ import ( admin_push "github.com/Rain-kl/Wavelet/internal/apps/admin/push" "github.com/Rain-kl/Wavelet/internal/apps/admin/push/custom_events" "github.com/Rain-kl/Wavelet/internal/apps/risk_control" + "github.com/Rain-kl/Wavelet/internal/listener" "github.com/Rain-kl/Wavelet/internal/infra/config" taskhandlers "github.com/Rain-kl/Wavelet/internal/infra/task/handlers" "github.com/Rain-kl/Wavelet/internal/model" @@ -38,10 +39,11 @@ type CacheRegistry struct { } var ( - registerTasksOnce sync.Once - registerPushDomainEventsOnce sync.Once - registerTaskListenersOnce sync.Once - initRuntimeOnce sync.Once + registerTasksOnce sync.Once + registerPushDomainEventsOnce sync.Once + registerTaskListenersOnce sync.Once + registerMessageGatewayListenersOnce sync.Once + initRuntimeOnce sync.Once cacheRegistries = make(map[string]CacheRegistry) cacheRegistriesMu sync.RWMutex @@ -99,6 +101,25 @@ func RegisterPushDomainEvents() { }) } +// RegisterMessageGatewayListeners registers the default log-only inbound handler. +func RegisterMessageGatewayListeners() { + registerMessageGatewayListenersOnce.Do(func() { + listener.OnMessageGatewayInbound(func(ctx context.Context, event listener.MessageGatewayInbound) { + userID := uint64(0) + if event.Msg.BindingUserID != nil { + userID = *event.Msg.BindingUserID + } + logger.InfoF(ctx, "[%s] channel=%d type=%s user=%d platform_user=%s", + listener.EventMessageGatewayInbound, + event.Msg.ChannelID, + event.Msg.ChannelType, + userID, + event.Msg.PlatformUserID, + ) + }) + }) +} + // RegisterTaskListeners wires operational listeners to task framework hooks. func RegisterTaskListeners() { registerTaskListenersOnce.Do(func() { @@ -110,12 +131,14 @@ func RegisterTaskListeners() { func RegisterAPI() { RegisterTasks() RegisterPushDomainEvents() + RegisterMessageGatewayListeners() } // RegisterWorker wires integrations required by the task worker process. func RegisterWorker() { RegisterTasks() RegisterTaskListeners() + RegisterMessageGatewayListeners() } // RegisterScheduler wires integrations required by the task scheduler process. @@ -128,6 +151,7 @@ func RegisterAll() { RegisterTasks() RegisterPushDomainEvents() RegisterTaskListeners() + RegisterMessageGatewayListeners() } // Init runs shared runtime bootstrap exactly once per process.