From adc355caa9c264dcbad19ac6f5bee47e2b0f60b9 Mon Sep 17 00:00:00 2001 From: zhx Date: Thu, 16 Jul 2026 16:14:49 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=9B=BD=E5=AE=B6=E7=BB=B4?= =?UTF-8?q?=E5=BA=A6=E5=B1=95=E7=A4=BA=E4=B8=8E=E8=B7=A8=E5=BA=94=E7=94=A8?= =?UTF-8?q?=E6=88=BF=E9=97=B4=E4=BA=8B=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../admin/internal/modules/databi/service.go | 167 +++++++++++++++++- .../internal/modules/databi/service_test.go | 144 +++++++++++++++ .../internal/room/outbox/outbox.go | 5 +- .../internal/room/service/admin_robot_room.go | 2 + .../room/service/admin_robot_room_test.go | 18 +- .../service/outbox_worker_internal_test.go | 63 +++++++ .../internal/room/service/pipeline.go | 2 +- .../room/service/robot_display_sampler.go | 11 +- .../internal/room/service/vip_test.go | 3 + .../storage/mysql/command_log_test.go | 49 +++++ .../storage/mysql/outbox_retention_test.go | 40 +++++ .../internal/storage/mysql/query.go | 23 ++- .../internal/storage/mysql/query_test.go | 96 ++++++++++ .../internal/storage/mysql/repository.go | 9 +- .../mysql/social_requirements_query.go | 18 +- 15 files changed, 622 insertions(+), 28 deletions(-) diff --git a/server/admin/internal/modules/databi/service.go b/server/admin/internal/modules/databi/service.go index 4b33545f..fe91fb4d 100644 --- a/server/admin/internal/modules/databi/service.go +++ b/server/admin/internal/modules/databi/service.go @@ -20,8 +20,9 @@ import ( ) const ( - appKindHyapp = "hyapp" - appKindLegacy = "legacy" + appKindHyapp = "hyapp" + appKindLegacy = "legacy" + unknownCountryName = "未知国家" overviewConcurrency = 4 ) @@ -341,14 +342,18 @@ type RequirementsResult struct { // Overview 服务端并发聚合各 App 的统计总览：hyapp App 走 statistics-service， // legacy App 走 dashboard 外部源；国家行按区域目录卷积出大区行，并按用户数据范围裁剪。 func (s *Service) Overview(ctx context.Context, access repository.MoneyAccess, query OverviewQuery) (*OverviewResult, error) { - apps, err := s.listAllowedApps(ctx, query.AppCodes, access) + registeredApps, err := s.listApps(ctx) if err != nil { return nil, err } + apps := filterAllowedApps(registeredApps, query.AppCodes, access) catalog, err := s.regionCatalog(ctx, apps) if err != nil { return nil, err } + // 国家主数据属于 App 业务域；即使 country_id 是物理全局自增，也必须按当前可查询 App 建目录， + // 否则跨 App 污染会被另一个 App 的主数据包装成合法国家，掩盖上游事实错误。 + countryDirectories := s.countryDirectories(ctx, apps) results := make([]AppOverview, len(apps)) var wg sync.WaitGroup @@ -359,7 +364,7 @@ func (s *Service) Overview(ctx context.Context, access repository.MoneyAccess, q defer wg.Done() semaphore <- struct{}{} defer func() { <-semaphore }() - results[index] = s.appOverview(ctx, app, catalog[app.AppCode], access, query) + results[index] = s.appOverview(ctx, app, catalog[app.AppCode], countryDirectories[app.AppCode], access, query) }(index, app) } wg.Wait() @@ -600,7 +605,7 @@ func supportsAppTrackingFunnel(appCode string) bool { return ok } -func (s *Service) appOverview(ctx context.Context, app AppInfo, regions []RegionInfo, access repository.MoneyAccess, query OverviewQuery) AppOverview { +func (s *Service) appOverview(ctx context.Context, app AppInfo, regions []RegionInfo, countries map[int64]countryDirectoryEntry, access repository.MoneyAccess, query OverviewQuery) AppOverview { out := AppOverview{ AppCode: app.AppCode, AppName: app.AppName, @@ -652,8 +657,10 @@ func (s *Service) appOverview(ctx context.Context, app AppInfo, regions []Region return out } - countryRows := anyRowSlice(overview["country_breakdown"]) - dailyCountryRows := anyRowSlice(overview["daily_country_breakdown"]) + // statisticsOverview 会短时缓存并共享底层 map；补展示字段时必须复制行，避免并发请求互相改写缓存。 + preserveSourceNames := app.Kind == appKindLegacy + countryRows := enrichCountryRows(anyRowSlice(overview["country_breakdown"]), countries, preserveSourceNames) + dailyCountryRows := enrichCountryRows(anyRowSlice(overview["daily_country_breakdown"]), countries, preserveSourceNames) resolve := regionResolver(regions) restricted := !allowAll && query.RegionID == 0 && len(requestRegionIDs) == 0 @@ -1199,6 +1206,12 @@ func (s *Service) listAllowedApps(ctx context.Context, requested []string, acces if err != nil { return nil, err } + return filterAllowedApps(apps, requested, access), nil +} + +// filterAllowedApps 只裁剪可查询的 App，不改变 appregistry 给出的稳定顺序。 +// Overview 会保留裁剪前的注册 App 集合，供全局 country_id 展示目录使用。 +func filterAllowedApps(apps []AppInfo, requested []string, access repository.MoneyAccess) []AppInfo { requestedSet := map[string]struct{}{} for _, appCode := range requested { appCode = appctx.Normalize(appCode) @@ -1221,7 +1234,145 @@ func (s *Service) listAllowedApps(ctx context.Context, requested []string, acces } out = append(out, app) } - return out, nil + return out +} + +type countryDirectoryEntry struct { + AppCode string + ID int64 + Code string + Name string + DisplayName string + Flag string + Enabled bool +} + +// countryDirectories 按 app_code + country_id 汇总当前可查询 HyApp 的国家主数据。 +// country_id 虽是全局自增主键，但国家启停和归属仍受 app_code 隔离；目录不能跨 App 复用。 +func (s *Service) countryDirectories(ctx context.Context, apps []AppInfo) map[string]map[int64]countryDirectoryEntry { + byApp := make(map[string]map[int64]countryDirectoryEntry) + if s.user == nil { + return byApp + } + + // appregistry 当前按名称排序；改按 app_code 查询让下游调用顺序和日志保持稳定。 + hyappApps := make([]AppInfo, 0, len(apps)) + for _, app := range apps { + if app.Kind == appKindHyapp { + hyappApps = append(hyappApps, app) + } + } + sort.Slice(hyappApps, func(i, j int) bool { + return hyappApps[i].AppCode < hyappApps[j].AppCode + }) + + for _, app := range hyappApps { + // user-service 依赖 RequestMeta.app_code 选择 App 数据域；每次调用都必须显式覆盖请求上下文。 + appContext := appctx.WithContext(ctx, app.AppCode) + countries, err := s.user.ListCountries(appContext, userclient.ListCountriesRequest{ + Caller: "admin-server", + // nil 表示同时读取 enabled 与 disabled。已停用国家仍可能存在于历史统计事实中，不能从展示目录丢失。 + Enabled: nil, + }) + if err != nil { + // 国家目录只用于展示；单 App 故障时该 App 降级为未知，不能拖垮整个 Social BI Overview 接口。 + slog.Warn("databi_list_hyapp_countries_failed", "app_code", app.AppCode, "error", err) + continue + } + byID := make(map[int64]countryDirectoryEntry, len(countries)) + mergeCountryDirectory(byID, app.AppCode, countries) + byApp[app.AppCode] = byID + } + return byApp +} + +// mergeCountryDirectory 将一次 App 查询投影为 BI 展示字段；ID=0 是统计未知值，不伪造国家主数据。 +// 同一 App 返回重复 ID 时保留第一项，避免异常目录响应在同一次请求内反复覆盖展示语义。 +func mergeCountryDirectory(byID map[int64]countryDirectoryEntry, appCode string, countries []*userclient.Country) { + for _, country := range countries { + if country == nil || country.CountryID <= 0 { + continue + } + if _, exists := byID[country.CountryID]; exists { + continue + } + byID[country.CountryID] = countryDirectoryEntry{ + AppCode: appctx.Normalize(appCode), + ID: country.CountryID, + Code: strings.TrimSpace(country.CountryCode), + Name: strings.TrimSpace(country.CountryName), + DisplayName: strings.TrimSpace(country.CountryDisplayName), + Flag: strings.TrimSpace(country.Flag), + Enabled: country.Enabled, + } + } +} + +// enrichCountryRows 复制统计行后补展示字段，不能原地修改 statisticsOverview 的共享缓存。 +func enrichCountryRows(rows []map[string]any, countries map[int64]countryDirectoryEntry, preserveSourceNames bool) []map[string]any { + out := make([]map[string]any, 0, len(rows)) + for _, source := range rows { + row := make(map[string]any, len(source)+4) + for key, value := range source { + row[key] = value + } + enrichCountryRow(row, countries, preserveSourceNames) + out = append(out, row) + } + return out +} + +func enrichCountryRow(row map[string]any, countries map[int64]countryDirectoryEntry, preserveSourceNames bool) { + // 外接 legacy 源可能使用 country_name、country 或 country_code 标识国家；任一字段存在时都以源数据为准， + // 避免它携带的本地数字 ID 碰巧等于某个 HyApp 全局 ID 后被错误覆盖。 + if preserveSourceNames { + sourceDisplayName := rowString(row, "country_display_name") + sourceName := rowString(row, "country_name", "country") + sourceCode := rowString(row, "country_code") + if firstNonEmptyString(sourceDisplayName, sourceName, sourceCode) != "" { + row["country_name"] = firstNonEmptyString(sourceDisplayName, sourceName, sourceCode) + setCountryFieldIfEmpty(row, "country_code", "") + // 字段必须存在供前端统一读取，但不把 legacy 的英文名伪装成原始中文 display_name。 + setCountryFieldIfEmpty(row, "country_display_name", "") + setCountryFieldIfEmpty(row, "flag", "") + return + } + // legacy 的数字 ID 由外部数据源定义，不能碰巧命中 HyApp ID 后借用其名称； + // 源没有返回名称/国家码时明确保持未知，等待该源补齐自己的展示契约。 + row["country_name"] = unknownCountryName + setCountryFieldIfEmpty(row, "country_code", "") + setCountryFieldIfEmpty(row, "country_display_name", unknownCountryName) + setCountryFieldIfEmpty(row, "flag", "") + return + } + countryID, hasCountryID := rowInt64(row, "country_id") + country, found := countries[countryID] + if !hasCountryID || countryID <= 0 || !found { + // 0 和目录未命中都属于未知维度；保留原 country_id 便于继续追查事实来源，不伪造 ID 或国家码。 + setCountryFieldIfEmpty(row, "country_code", "") + row["country_name"] = unknownCountryName + setCountryFieldIfEmpty(row, "country_display_name", unknownCountryName) + setCountryFieldIfEmpty(row, "flag", "") + return + } + + row["country_code"] = country.Code + // HyApp 数字统计行以 country_id 对应的主数据为准；country_display_name 保持原始字段， + // country_name 则按中文展示名、英文名、国家码依次 fallback，供现有前端直接渲染。 + row["country_display_name"] = country.DisplayName + row["flag"] = country.Flag + row["country_name"] = firstNonEmptyString( + country.DisplayName, + country.Name, + country.Code, + unknownCountryName, + ) +} + +func setCountryFieldIfEmpty(row map[string]any, key string, fallback string) { + if rowString(row, key) == "" { + row[key] = fallback + } } // regionCatalog 汇总每个 App 的区域目录：hyapp App 共用 user-service 的区域主数据， diff --git a/server/admin/internal/modules/databi/service_test.go b/server/admin/internal/modules/databi/service_test.go index f28e36ca..8f8c7f45 100644 --- a/server/admin/internal/modules/databi/service_test.go +++ b/server/admin/internal/modules/databi/service_test.go @@ -2,9 +2,13 @@ package databi import ( "context" + "errors" + "reflect" "testing" "time" + "hyapp-admin-server/internal/appctx" + "hyapp-admin-server/internal/integration/userclient" "hyapp-admin-server/internal/model" "hyapp-admin-server/internal/repository" ) @@ -102,3 +106,143 @@ func TestRegionDisplayToleratesRoundedIDs(t *testing.T) { t.Fatalf("expected rounded id to match catalog") } } + +func TestMergeAndEnrichCountryDirectory(t *testing.T) { + laluDirectory := map[int64]countryDirectoryEntry{} + mergeCountryDirectory(laluDirectory, "lalu", []*userclient.Country{ + {CountryID: 158, CountryCode: "PK", CountryName: "Pakistan", CountryDisplayName: "巴基斯坦", Flag: "🇵🇰", Enabled: true}, + // disabled 国家仍可能出现在历史统计中，必须进入目录。 + {CountryID: 157, CountryCode: "BS", CountryName: "Philippines", CountryDisplayName: "菲律宾", Enabled: false}, + // country_id=0 是未知维度，不允许目录把它伪装成真实国家。 + {CountryID: 0, CountryCode: "UNKNOWN", CountryName: "Unknown", Enabled: true}, + }) + famiDirectory := map[int64]countryDirectoryEntry{} + mergeCountryDirectory(famiDirectory, "fami", []*userclient.Country{ + {CountryID: 1163, CountryCode: "BD", CountryName: "Bangladesh", CountryDisplayName: "孟加拉国", Enabled: true}, + }) + huwaaDirectory := map[int64]countryDirectoryEntry{} + mergeCountryDirectory(huwaaDirectory, "huwaa", []*userclient.Country{ + {CountryID: 953, CountryCode: "IN", CountryName: "India", CountryDisplayName: "印度", Enabled: true}, + }) + + if len(laluDirectory) != 2 { + t.Fatalf("expected two lalu country ids, got %#v", laluDirectory) + } + if country := laluDirectory[158]; country.AppCode != "lalu" || country.Name != "Pakistan" { + t.Fatalf("expected lalu country retained, got %+v", country) + } + if country := laluDirectory[157]; country.Enabled { + t.Fatalf("expected disabled country retained, got %+v", country) + } + + sourceRows := []map[string]any{ + {"country_id": int64(158)}, + {"country_id": int64(1163)}, + {"country_id": int64(953)}, + {"country_id": int64(0)}, + {"country_id": int64(9999)}, + } + rows := enrichCountryRows(sourceRows, laluDirectory, false) + if got := rowString(rows[0], "country_name"); got != "巴基斯坦" { + t.Fatalf("expected display name to win for country 158, got %q", got) + } + for _, index := range []int{1, 2, 3, 4} { + if got := rowString(rows[index], "country_name"); got != unknownCountryName { + t.Fatalf("expected row %d to remain unknown in lalu scope, got %q", index, got) + } + if got := rowString(rows[index], "country_display_name"); got != unknownCountryName { + t.Fatalf("expected row %d unknown display name, got %q", index, got) + } + } + if got := rowString(enrichCountryRows([]map[string]any{{"country_id": int64(1163)}}, famiDirectory, false)[0], "country_name"); got != "孟加拉国" { + t.Fatalf("expected fami country resolved only in fami scope, got %q", got) + } + if got := rowString(enrichCountryRows([]map[string]any{{"country_id": int64(953)}}, huwaaDirectory, false)[0], "country_name"); got != "印度" { + t.Fatalf("expected huwaa country resolved only in huwaa scope, got %q", got) + } + if _, mutated := sourceRows[0]["country_name"]; mutated { + t.Fatalf("expected enrichment to copy shared statistics rows") + } + + legacyRows := enrichCountryRows([]map[string]any{ + { + "country_id": int64(158), + "country_code": "LEGACY-PK", + "country_name": "Legacy Pakistan", + }, + {"country_id": int64(158), "country": "Legacy country field"}, + {"country_id": int64(158), "country_code": "LEGACY-CODE"}, + {"country_id": int64(158)}, + }, laluDirectory, true) + if got := rowString(legacyRows[0], "country_name"); got != "Legacy Pakistan" { + t.Fatalf("expected legacy source name preserved, got %q", got) + } + if got := rowString(legacyRows[0], "country_code"); got != "LEGACY-PK" { + t.Fatalf("expected legacy source code preserved, got %q", got) + } + if got := rowString(legacyRows[1], "country_name"); got != "Legacy country field" { + t.Fatalf("expected legacy country field preserved, got %q", got) + } + if got := rowString(legacyRows[2], "country_name"); got != "LEGACY-CODE" { + t.Fatalf("expected legacy code used instead of colliding HyApp id, got %q", got) + } + if got := rowString(legacyRows[3], "country_name"); got != unknownCountryName { + t.Fatalf("legacy numeric id must not borrow HyApp country name, got %q", got) + } +} + +type countryDirectoryClientStub struct { + userclient.Client + countries map[string][]*userclient.Country + errors map[string]error + calls []string + filtered bool +} + +func (c *countryDirectoryClientStub) ListCountries(ctx context.Context, req userclient.ListCountriesRequest) ([]*userclient.Country, error) { + appCode := appctx.FromContext(ctx) + c.calls = append(c.calls, appCode) + if req.Enabled != nil { + c.filtered = true + } + if err := c.errors[appCode]; err != nil { + return nil, err + } + return c.countries[appCode], nil +} + +func TestCountryDirectoriesQueryEachAllowedHyappAndContinueAfterFailure(t *testing.T) { + client := &countryDirectoryClientStub{ + countries: map[string][]*userclient.Country{ + "fami": {{CountryID: 1163, CountryCode: "BD", CountryName: "Bangladesh", CountryDisplayName: "孟加拉国", Enabled: true}}, + "lalu": {{CountryID: 158, CountryCode: "PK", CountryName: "Pakistan", CountryDisplayName: "巴基斯坦", Enabled: true}}, + }, + errors: map[string]error{"huwaa": errors.New("temporary user-service failure")}, + } + service := NewService(nil, nil, nil, client) + directories := service.countryDirectories(context.Background(), []AppInfo{ + {AppCode: "lalu", Kind: appKindHyapp}, + {AppCode: "aslan", Kind: appKindLegacy}, + {AppCode: "huwaa", Kind: appKindHyapp}, + {AppCode: "fami", Kind: appKindHyapp}, + }) + + if want := []string{"fami", "huwaa", "lalu"}; !reflect.DeepEqual(client.calls, want) { + t.Fatalf("expected each HyApp queried with its own context, got %v want %v", client.calls, want) + } + if client.filtered { + t.Fatalf("expected enabled filter omitted so disabled history can resolve") + } + if _, ok := directories["fami"][1163]; !ok { + t.Fatalf("expected successful app before failure retained") + } + if _, ok := directories["lalu"][158]; !ok { + t.Fatalf("expected apps after failure still queried") + } + if _, ok := directories["huwaa"]; ok { + t.Fatalf("failed app must not receive another app's directory: %#v", directories) + } + if len(directories) != 2 { + t.Fatalf("unexpected directories after one app failure: %#v", directories) + } +} diff --git a/services/room-service/internal/room/outbox/outbox.go b/services/room-service/internal/room/outbox/outbox.go index e36feb1e..c5d351f5 100644 --- a/services/room-service/internal/room/outbox/outbox.go +++ b/services/room-service/internal/room/outbox/outbox.go @@ -6,7 +6,6 @@ import ( "google.golang.org/protobuf/proto" roomeventsv1 "hyapp.local/api/proto/events/room/v1" - "hyapp/pkg/appcode" "hyapp/pkg/idgen" ) @@ -62,14 +61,14 @@ func Build(roomID string, eventType string, roomVersion int64, occurredAt time.T // EventID 在 room-service 生成，后续同步广播和异步补偿都用同一个 ID 去重。 eventID := idgen.New("evt") + // Build 没有请求上下文，不能在多 App 环境猜租户。SaveMutation/SaveOutbox 会在持久化前用 + // 当前请求的 app_code 同时补齐 Record 和 Envelope；直发 IM/机器人展示也在发布边界做同样处理。 return Record{ - AppCode: appcode.Default, EventID: eventID, EventType: eventType, RoomID: roomID, Status: StatusPending, Envelope: &roomeventsv1.EventEnvelope{ - AppCode: appcode.Default, EventId: eventID, RoomId: roomID, EventType: eventType, diff --git a/services/room-service/internal/room/service/admin_robot_room.go b/services/room-service/internal/room/service/admin_robot_room.go index 5158f688..981651b4 100644 --- a/services/room-service/internal/room/service/admin_robot_room.go +++ b/services/room-service/internal/room/service/admin_robot_room.go @@ -235,6 +235,8 @@ func (s *Service) bootstrapRobotRoom(ctx context.Context, config RobotRoomConfig if _, err := s.JoinRoom(ctx, &roomv1.JoinRoomRequest{ Meta: robotRoomCommandMeta(config, userID, fmt.Sprintf("join:%d", userID)), Role: "audience", + // 首批机器人与运行期替换机器人必须使用同一事实标记；否则 RoomUserJoined 会被统计成缺国家的真人 DAU。 + ActorIsRobot: true, }); err != nil { return err } diff --git a/services/room-service/internal/room/service/admin_robot_room_test.go b/services/room-service/internal/room/service/admin_robot_room_test.go index d380d078..6eed9d48 100644 --- a/services/room-service/internal/room/service/admin_robot_room_test.go +++ b/services/room-service/internal/room/service/admin_robot_room_test.go @@ -139,16 +139,21 @@ func TestAdminCreateRobotRoomMarksRobotJoinOutbox(t *testing.T) { _, err := svc.AdminCreateRobotRoom(ctx, &roomv1.AdminCreateRobotRoomRequest{ Meta: roomservice.NewRequestMeta("", 9001), OwnerRobotUserId: 2001, - CandidateRobotUserIds: []int64{2001}, - MinRobotCount: 1, - MaxRobotCount: 1, + CandidateRobotUserIds: []int64{2001, 2002}, + MinRobotCount: 2, + MaxRobotCount: 2, RoomName: "Robot Join Outbox", RoomAvatar: testRoomCoverURL, VisibleRegionId: 686, OwnerCountryCode: "ae", GiftRule: &roomv1.AdminRobotRoomGiftRule{ GiftIds: []string{"84"}, + LuckyGiftIds: []string{"28"}, NormalGiftIntervalMs: 10000, + LuckyComboMin: 100, + LuckyComboMax: 10000, + LuckyPauseMinMs: 5000, + LuckyPauseMaxMs: 20000, }, AdminId: 9001, }) @@ -161,6 +166,7 @@ func TestAdminCreateRobotRoomMarksRobotJoinOutbox(t *testing.T) { t.Fatalf("list pending outbox failed: %v", err) } joinEvents := 0 + bootstrapRobotJoined := false for _, record := range records { if record.EventType != "RoomUserJoined" { continue @@ -170,13 +176,17 @@ func TestAdminCreateRobotRoomMarksRobotJoinOutbox(t *testing.T) { if err := proto.Unmarshal(record.Envelope.GetBody(), &joined); err != nil { t.Fatalf("decode room join event failed: %v", err) } - if joined.GetUserId() == 2001 && !joined.GetIsRobot() { + if !joined.GetIsRobot() { t.Fatalf("robot join outbox must mark is_robot=true: %+v", &joined) } + bootstrapRobotJoined = bootstrapRobotJoined || joined.GetUserId() == 2002 } if joinEvents == 0 { t.Fatalf("robot room creation should write RoomUserJoined outbox") } + if !bootstrapRobotJoined { + t.Fatalf("bootstrap robot should write RoomUserJoined outbox") + } } func occupiedSeatUserCount(snapshot *roomv1.RoomSnapshot) int { diff --git a/services/room-service/internal/room/service/outbox_worker_internal_test.go b/services/room-service/internal/room/service/outbox_worker_internal_test.go index 98cb701d..0942f77b 100644 --- a/services/room-service/internal/room/service/outbox_worker_internal_test.go +++ b/services/room-service/internal/room/service/outbox_worker_internal_test.go @@ -14,6 +14,24 @@ type countingOutboxPublisher struct { events []*roomeventsv1.EventEnvelope } +type scopedOutboxPublication struct { + contextApp string + envelopeApp string +} + +type scopedOutboxPublisher struct { + published chan scopedOutboxPublication +} + +func newScopedOutboxPublisher() *scopedOutboxPublisher { + return &scopedOutboxPublisher{published: make(chan scopedOutboxPublication, 2)} +} + +func (p *scopedOutboxPublisher) PublishOutboxEvent(ctx context.Context, envelope *roomeventsv1.EventEnvelope) error { + p.published <- scopedOutboxPublication{contextApp: appcode.FromContext(ctx), envelopeApp: envelope.GetAppCode()} + return nil +} + type crossAppOutboxRepository struct { Repository claims []string @@ -46,6 +64,51 @@ func TestOutboxSingleRecordLeaseCoversPublishAndStateWrite(t *testing.T) { } } +func TestDirectAndRobotPublishersScopeBuiltRecordsFromContext(t *testing.T) { + ctx := appcode.WithContext(context.Background(), "fami") + directPublisher := newScopedOutboxPublisher() + robotPublisher := newScopedOutboxPublisher() + svc := &Service{roomDisplayPublisher: directPublisher, robotDisplayPublisher: robotPublisher} + build := func(eventType string) outbox.Record { + record, err := outbox.Build("fami-room-publish", eventType, 3, time.UnixMilli(3000), &roomeventsv1.RoomUserJoined{UserId: 1001}) + if err != nil { + t.Fatalf("build %s outbox: %v", eventType, err) + } + return record + } + + svc.publishDirectIMRecordsBestEffort(ctx, []outbox.Record{build("RoomUserJoined")}) + svc.publishRobotDisplayRecordsBestEffort(ctx, []outbox.Record{build("RoomRobotUserJoined")}) + for name, publisher := range map[string]*scopedOutboxPublisher{"direct": directPublisher, "robot": robotPublisher} { + select { + case publication := <-publisher.published: + if publication.contextApp != "fami" || publication.envelopeApp != "fami" { + t.Fatalf("%s publisher lost scoped app: context=%q envelope=%q", name, publication.contextApp, publication.envelopeApp) + } + case <-time.After(time.Second): + t.Fatalf("%s publisher did not receive built record", name) + } + } +} + +func TestRobotDisplaySamplerUsesCommandAppForContextFreeRecords(t *testing.T) { + svc := &Service{robotDisplaySampler: newRobotDisplaySampler()} + build := func() outbox.Record { + record, err := outbox.Build("shared-room-id", "RoomRobotLuckyGiftDrawn", 3, time.UnixMilli(3000), &roomeventsv1.RoomRobotLuckyGiftDrawn{}) + if err != nil { + t.Fatalf("build robot display outbox: %v", err) + } + return record + } + // 同房间标识、事件类型和窗口在不同 App 下必须各保留一条；Build 尚未补 App，采样器只能从命令上下文取租户。 + if got := svc.sampleRobotDisplayRecords(appcode.WithContext(context.Background(), "fami"), []outbox.Record{build()}); len(got) != 1 { + t.Fatalf("fami robot display unexpectedly sampled: %+v", got) + } + if got := svc.sampleRobotDisplayRecords(appcode.WithContext(context.Background(), "huwaa"), []outbox.Record{build()}); len(got) != 1 { + t.Fatalf("huwaa robot display shared fami sample window: %+v", got) + } +} + func TestClaimPendingOutboxFindsNonDefaultApp(t *testing.T) { repository := &crossAppOutboxRepository{} svc := &Service{nodeID: "room-cross-app", repository: repository} diff --git a/services/room-service/internal/room/service/pipeline.go b/services/room-service/internal/room/service/pipeline.go index 92a50496..668c52d2 100644 --- a/services/room-service/internal/room/service/pipeline.go +++ b/services/room-service/internal/room/service/pipeline.go @@ -107,7 +107,7 @@ func (s *Service) mutateRoom(ctx context.Context, cmd command.Command, replayabl return nil, err } // 机器人展示只是 App 视觉补偿，提交前按短窗口降采样；命令日志、房间状态和主 outbox 不受影响。 - robotDisplayRecords := s.sampleRobotDisplayRecords(result.robotDisplayRecords) + robotDisplayRecords := s.sampleRobotDisplayRecords(ctx, result.robotDisplayRecords) durableOutboxRecords, directIMRecords := splitRoomDirectIMRecords(outboxRecords, result.suppressDirectIMEventTypes) // mutate 可额外返回只服务客户端展示的直发 IM，典型场景是重复 JoinRoom 的入房飘窗； // 这类事件不能进入 durable outbox，否则会污染活跃统计和补偿消费。 diff --git a/services/room-service/internal/room/service/robot_display_sampler.go b/services/room-service/internal/room/service/robot_display_sampler.go index 3bf30608..eb3d378a 100644 --- a/services/room-service/internal/room/service/robot_display_sampler.go +++ b/services/room-service/internal/room/service/robot_display_sampler.go @@ -1,6 +1,7 @@ package service import ( + "context" "sync" "time" @@ -28,14 +29,14 @@ func newRobotDisplaySampler() *robotDisplaySampler { return &robotDisplaySampler{lastWindow: make(map[robotDisplaySampleKey]int64)} } -func (s *Service) sampleRobotDisplayRecords(records []outbox.Record) []outbox.Record { +func (s *Service) sampleRobotDisplayRecords(ctx context.Context, records []outbox.Record) []outbox.Record { if len(records) == 0 || s == nil || s.robotDisplaySampler == nil { return records } - return s.robotDisplaySampler.filter(records, robotDisplaySampleWindow) + return s.robotDisplaySampler.filter(records, robotDisplaySampleWindow, appcode.FromContext(ctx)) } -func (s *robotDisplaySampler) filter(records []outbox.Record, window time.Duration) []outbox.Record { +func (s *robotDisplaySampler) filter(records []outbox.Record, window time.Duration, scopedApp string) []outbox.Record { if s == nil || window <= 0 || len(records) == 0 { return records } @@ -48,7 +49,9 @@ func (s *robotDisplaySampler) filter(records []outbox.Record, window time.Durati var newestWindow int64 for _, record := range records { key := robotDisplaySampleKey{ - appCode: appcode.Normalize(record.AppCode), + // Build 阶段故意不猜 App；降采样发生在发布边界补值之前，因此必须使用当前命令上下文兜底， + // 否则不同 App 的同房间标识会共用默认 Lalu 窗口并误丢展示事件。 + appCode: appcode.Normalize(firstNonEmpty(record.AppCode, scopedApp)), roomID: record.RoomID, eventType: record.EventType, } diff --git a/services/room-service/internal/room/service/vip_test.go b/services/room-service/internal/room/service/vip_test.go index b8ddcabf..39643004 100644 --- a/services/room-service/internal/room/service/vip_test.go +++ b/services/room-service/internal/room/service/vip_test.go @@ -48,6 +48,9 @@ func TestJoinRoomCarriesEffectiveVIPAndQueriesWalletOnce(t *testing.T) { if len(records) != 1 { t.Fatalf("joined direct IM record mismatch: %+v", records) } + if records[0].GetAppCode() != vipTestAppCode { + t.Fatalf("joined direct IM must inherit request app: got=%q want=%q", records[0].GetAppCode(), vipTestAppCode) + } var joined roomeventsv1.RoomUserJoined if err := proto.Unmarshal(records[0].GetBody(), &joined); err != nil { t.Fatalf("decode joined event failed: %v", err) diff --git a/services/room-service/internal/storage/mysql/command_log_test.go b/services/room-service/internal/storage/mysql/command_log_test.go index 9c25935e..c6a43c82 100644 --- a/services/room-service/internal/storage/mysql/command_log_test.go +++ b/services/room-service/internal/storage/mysql/command_log_test.go @@ -6,12 +6,61 @@ import ( "errors" "sync" "testing" + "time" + "google.golang.org/protobuf/proto" + roomeventsv1 "hyapp.local/api/proto/events/room/v1" "hyapp/internal/testutil/mysqlschema" "hyapp/pkg/appcode" + "hyapp/services/room-service/internal/room/outbox" roomservice "hyapp/services/room-service/internal/room/service" ) +func TestSaveMutationScopesBuiltOutboxToContextApp(t *testing.T) { + ctx := appcode.WithContext(context.Background(), "fami") + repo := newCommandLogTestRepository(t, ctx) + record, err := outbox.Build("fami-room-outbox", "RoomCreated", 1, time.UnixMilli(1000), &roomeventsv1.RoomCreated{}) + if err != nil { + t.Fatalf("build room outbox: %v", err) + } + if record.AppCode != "" || record.Envelope.GetAppCode() != "" { + t.Fatalf("context-free Build must not guess an app: record=%q envelope=%q", record.AppCode, record.Envelope.GetAppCode()) + } + + if err := repo.SaveMutation(ctx, roomservice.MutationCommit{ + Command: roomservice.CommandRecord{ + RoomID: record.RoomID, + RoomVersion: 1, + CommandID: "cmd-fami-room-outbox", + CommandType: "create_room", + Payload: []byte(`{"command_id":"cmd-fami-room-outbox"}`), + Replayable: true, + OwnerNodeID: "node-fami", + LeaseToken: "lease-fami", + CreatedAtMS: record.CreatedAtMS, + }, + OutboxRecords: []outbox.Record{record}, + }); err != nil { + t.Fatalf("save fami mutation: %v", err) + } + + var storedApp string + var envelopeBytes []byte + if err := repo.db.QueryRowContext(ctx, + `SELECT app_code, envelope FROM room_outbox WHERE app_code = ? AND event_id = ?`, + "fami", record.EventID, + ).Scan(&storedApp, &envelopeBytes); err != nil { + t.Fatalf("read persisted fami outbox: %v", err) + } + var envelope roomeventsv1.EventEnvelope + if err := proto.Unmarshal(envelopeBytes, &envelope); err != nil { + t.Fatalf("decode persisted fami envelope: %v", err) + } + if storedApp != "fami" || envelope.GetAppCode() != "fami" { + t.Fatalf("persistence boundary lost scoped app: row=%q envelope=%q", storedApp, envelope.GetAppCode()) + } +} + func TestSaveMutationRejectsSameCommandInsteadOfContinuingSideEffects(t *testing.T) { ctx := appcode.WithContext(context.Background(), appcode.Default) repo := newCommandLogTestRepository(t, ctx) diff --git a/services/room-service/internal/storage/mysql/outbox_retention_test.go b/services/room-service/internal/storage/mysql/outbox_retention_test.go index 124d2d51..e1673b60 100644 --- a/services/room-service/internal/storage/mysql/outbox_retention_test.go +++ b/services/room-service/internal/storage/mysql/outbox_retention_test.go @@ -3,13 +3,53 @@ package mysql import ( "context" "testing" + "time" + "google.golang.org/protobuf/proto" roomeventsv1 "hyapp.local/api/proto/events/room/v1" "hyapp/internal/testutil/mysqlschema" "hyapp/pkg/appcode" "hyapp/services/room-service/internal/room/outbox" ) +func TestSaveOutboxScopesBuiltRecordToContextApp(t *testing.T) { + schema := mysqlschema.New(t, mysqlschema.Config{ + EnvVar: "ROOM_SERVICE_MYSQL_TEST_DSN", + InitDBPath: mysqlschema.InitDBPath(t, mysqlschema.CallerFile(t, 1), "..", "..", "..", "deploy", "mysql", "initdb", "001_room_service.sql"), + DatabasePrefix: "hyapp_room_outbox_scope_test", + }) + ctx := appcode.WithContext(context.Background(), "huwaa") + repo, err := Open(ctx, schema.DSN, 4, 4) + if err != nil { + t.Fatalf("open repository: %v", err) + } + defer repo.Close() + + record, err := outbox.Build("huwaa-room-outbox", "RoomUserJoined", 9, time.UnixMilli(2000), &roomeventsv1.RoomUserJoined{UserId: 99}) + if err != nil { + t.Fatalf("build room outbox: %v", err) + } + if err := repo.SaveOutbox(ctx, []outbox.Record{record}); err != nil { + t.Fatalf("save scoped outbox: %v", err) + } + + var storedApp string + var envelopeBytes []byte + if err := schema.DB.QueryRowContext(ctx, + `SELECT app_code, envelope FROM room_outbox WHERE app_code = ? AND event_id = ?`, + "huwaa", record.EventID, + ).Scan(&storedApp, &envelopeBytes); err != nil { + t.Fatalf("read scoped outbox: %v", err) + } + var envelope roomeventsv1.EventEnvelope + if err := proto.Unmarshal(envelopeBytes, &envelope); err != nil { + t.Fatalf("decode scoped outbox envelope: %v", err) + } + if storedApp != "huwaa" || envelope.GetAppCode() != "huwaa" { + t.Fatalf("SaveOutbox lost scoped app: row=%q envelope=%q", storedApp, envelope.GetAppCode()) + } +} + func retentionTestRecord(eventID string, createdAtMS int64) outbox.Record { return outbox.Record{ AppCode: appcode.Default, diff --git a/services/statistics-service/internal/storage/mysql/query.go b/services/statistics-service/internal/storage/mysql/query.go index cad634f1..a16305b6 100644 --- a/services/statistics-service/internal/storage/mysql/query.go +++ b/services/statistics-service/internal/storage/mysql/query.go @@ -1582,9 +1582,19 @@ func applyCountryBreakdownDerivedMetrics(item *CountryBreakdown) { // overlayCanonicalCountryActive preserves monetary and business facts from the // legacy aggregate while replacing its incomplete activity counter. A -// canonical-only country is added so country and region rollups still reconcile -// with the top-level canonical user-day total. +// canonical-only valid countries are added to preserve resolved detail. Unknown-country +// DAU intentionally remains only in the top-level total, so country rows need not sum to it. func overlayCanonicalCountryActive(items []CountryBreakdown, cube canonicalOverviewActiveCube) []CountryBreakdown { + // 旧聚合表可能仍保存 country_id=0 的金额或历史活跃事实；总览总计会继续包含这些真实事实， + // 但国家明细不能把“未知维度”渲染成一个国家。先过滤旧行，避免 canonical cube 不再产出 0 后仍残留空壳行。 + validItems := items[:0] + for _, item := range items { + if item.CountryID > 0 { + validItems = append(validItems, item) + } + } + items = validItems + indexByCountry := make(map[int64]int, len(items)) for index := range items { indexByCountry[items[index].CountryID] = index @@ -1629,6 +1639,15 @@ func overlayCanonicalCountryActive(items []CountryBreakdown, cube canonicalOverv // country overlay. It prevents stale legacy DAU from surviving on rows with no // canonical activity and fills client-only detail rows before final ordering. func overlayCanonicalDailyCountryActive(items []DailyCountryStat, cube canonicalOverviewActiveCube) []DailyCountryStat { + // 与汇总国家行保持同一边界：未知国家事实只进入 App/日期总计，不作为 day×country 明细返回。 + validItems := items[:0] + for _, item := range items { + if item.CountryID > 0 { + validItems = append(validItems, item) + } + } + items = validItems + indexByDayCountry := make(map[string]int, len(items)) for index := range items { key := dailyCountryKey(items[index].StatDay, items[index].CountryID) diff --git a/services/statistics-service/internal/storage/mysql/query_test.go b/services/statistics-service/internal/storage/mysql/query_test.go index 0ce9bd09..90e3191a 100644 --- a/services/statistics-service/internal/storage/mysql/query_test.go +++ b/services/statistics-service/internal/storage/mysql/query_test.go @@ -57,6 +57,63 @@ func TestCanonicalizeSocialUserDaysMergesIdentityWithoutLosingDeviceCount(t *tes } } +func TestCanonicalizeSocialUserDaysDirectTupleWithoutCountryDoesNotEraseKnownDimension(t *testing.T) { + rows := []socialRequirementUserDay{ + {StatDay: "2026-07-15", Subject: "d:device-A", DeviceID: "device-A", CountryID: 158, RegionID: 19, UserRole: "user", Active: 1}, + {StatDay: "2026-07-15", Subject: "u:123", UserID: 123, DirectUser: true, DeviceID: "device-A", CountryID: 0, RegionID: 999, UserRole: "host", Active: 1}, + } + resolved := map[socialIdentityDeviceDayKey]int64{ + {StatDay: "2026-07-15", DeviceID: "device-A"}: 123, + } + + got := canonicalizeSocialUserDays(rows, resolved, 0, 0, nil) + if len(got) != 1 { + t.Fatalf("expected one canonical user-day, got %+v", got) + } + if got[0].CountryID != 158 || got[0].RegionID != 19 || !got[0].DirectUser || got[0].UserRole != "host" { + t.Fatalf("direct tuple without country must preserve known device dimension: %+v", got[0]) + } +} + +func TestCanonicalOverviewActiveCubeKeepsUnknownCountryOnlyInTotals(t *testing.T) { + day := "2026-07-15" + cube := buildCanonicalOverviewActiveCube([]socialRequirementUserDay{ + {StatDay: day, Subject: "u:1", UserID: 1, CountryID: 158, RegionID: 19, Active: 1}, + {StatDay: day, Subject: "d:anonymous", DeviceID: "anonymous", CountryID: 0, RegionID: 0, Active: 1}, + {StatDay: day, Subject: "d:inactive", DeviceID: "inactive", CountryID: 0, RegionID: 0, Active: 0}, + }) + + if cube.Total != 2 || cube.ByDay[day] != 2 { + t.Fatalf("unknown-country DAU must remain in total counters: %+v", cube) + } + if cube.ByCountry[158].ActiveUsers != 1 || cube.ByDayCountry[dailyCountryKey(day, 158)].ActiveUsers != 1 { + t.Fatalf("known-country DAU mismatch: %+v", cube) + } + if _, ok := cube.ByCountry[0]; ok { + t.Fatalf("country breakdown must not expose country_id=0: %+v", cube.ByCountry) + } + if _, ok := cube.ByDayCountry[dailyCountryKey(day, 0)]; ok { + t.Fatalf("daily country breakdown must not expose country_id=0: %+v", cube.ByDayCountry) + } + + // 旧 stat_app_day_country 里的 0 维度可能先于 canonical Social 逻辑存在；overlay 也必须删掉旧行， + // 否则页面仍会看到 active_users=0 的“国家 0”空壳，金额字段还可能让它继续参与排序。 + countryRows := overlayCanonicalCountryActive([]CountryBreakdown{ + {CountryID: 0, RechargeUSDMinor: 100}, + {CountryID: 158, RechargeUSDMinor: 200}, + }, cube) + if len(countryRows) != 1 || countryRows[0].CountryID != 158 || countryRows[0].ActiveUsers != 1 { + t.Fatalf("country overlay must remove unknown legacy row: %+v", countryRows) + } + dailyRows := overlayCanonicalDailyCountryActive([]DailyCountryStat{ + {StatDay: day, CountryBreakdown: CountryBreakdown{CountryID: 0, RechargeUSDMinor: 100}}, + {StatDay: day, CountryBreakdown: CountryBreakdown{CountryID: 158, RechargeUSDMinor: 200}}, + }, cube) + if len(dailyRows) != 1 || dailyRows[0].CountryID != 158 || dailyRows[0].ActiveUsers != 1 { + t.Fatalf("daily country overlay must remove unknown legacy row: %+v", dailyRows) + } +} + func TestSocialRetentionUserDayProjectionStaysNarrow(t *testing.T) { for _, required := range []string{ "stat_day", "subject_key", "user_id", "device_id", "country_id", "region_id", "user_role", @@ -1079,6 +1136,45 @@ func TestConsumeUserRegisteredCountsOnlyNewRegistration(t *testing.T) { } } +func TestConsumeUserActiveFallsBackToStatisticsUserDimension(t *testing.T) { + ctx := context.Background() + _, file, _, _ := runtime.Caller(0) + statsSchema := mysqlschema.New(t, mysqlschema.Config{ + InitDBPath: mysqlschema.InitDBPath(t, file, "..", "..", "..", "deploy", "mysql", "initdb", "001_statistics_service.sql"), + DatabasePrefix: "hy_stats_active_dimension_fallback_test", + }) + repository, err := Open(ctx, statsSchema.DSN) + if err != nil { + t.Fatalf("open repository: %v", err) + } + t.Cleanup(func() { _ = repository.Close() }) + + userID := int64(327300172394536960) + if err := repository.ConsumeUserRegionChanged(ctx, UserRegionChangedEvent{ + AppCode: "lalu", EventID: "region:known-before-room", UserID: userID, + CountryID: 158, RegionID: 19, OccurredAtMS: time.Date(2026, 7, 15, 1, 0, 0, 0, time.UTC).UnixMilli(), + }); err != nil { + t.Fatalf("consume user dimension: %v", err) + } + if err := repository.ConsumeUserActive(ctx, UserActiveEvent{ + AppCode: "lalu", EventID: "room:legacy-without-dimension", EventType: "RoomUserJoined", UserID: userID, + CountryID: 0, RegionID: 0, OccurredAtMS: time.Date(2026, 7, 15, 2, 0, 0, 0, time.UTC).UnixMilli(), + }); err != nil { + t.Fatalf("consume room active: %v", err) + } + + for _, table := range []string{"stat_user_day_activity", "stat_social_user_day"} { + var countryID, regionID int64 + query := "SELECT country_id, region_id FROM " + table + " WHERE app_code = 'lalu' AND stat_tz = 'UTC' AND stat_day = '2026-07-15' AND user_id = ?" + if err := repository.db.QueryRowContext(ctx, query, userID).Scan(&countryID, ®ionID); err != nil { + t.Fatalf("query %s dimension: %v", table, err) + } + if countryID != 158 || regionID != 19 { + t.Fatalf("%s must use statistics-owned user dimension, got country=%d region=%d", table, countryID, regionID) + } + } +} + func TestConsumeReportMetricsAggregatesWalletSalaryGrantAndMic(t *testing.T) { ctx := context.Background() _, file, _, _ := runtime.Caller(0) diff --git a/services/statistics-service/internal/storage/mysql/repository.go b/services/statistics-service/internal/storage/mysql/repository.go index 1a047478..9a87008e 100644 --- a/services/statistics-service/internal/storage/mysql/repository.go +++ b/services/statistics-service/internal/storage/mysql/repository.go @@ -1537,7 +1537,14 @@ func (r *Repository) ConsumeUserRoomDailyStatsUpdated(ctx context.Context, event func (r *Repository) ConsumeUserActive(ctx context.Context, event UserActiveEvent) error { return r.withEvent(ctx, SourceRoom, event.EventID, event.EventType, func(tx *sql.Tx, nowMS int64) error { - countryID, regionID := normalizeDimension(event.CountryID, event.RegionID) + // RoomUserJoined 的新版事实应自带完整国家/区域，但历史 outbox 和迁移期 producer 可能缺字段。 + // 只回退 statistics-service 自己消费 UserRegistered/UserRegionChanged 建立的维度投影，既修正已知用户， + // 又不跨服务回查 user owner 明细；维度投影尚未到达时仍保留 0，Social 宽表可由后续有效事实补齐， + // 旧 activity 聚合则保持原始事实并由查询层排除未知国家，不能在这里无账搬迁历史计数。 + countryID, regionID, err := r.dimensionForUser(ctx, tx, event.AppCode, event.UserID, event.CountryID, event.RegionID) + if err != nil { + return err + } for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) { if err := applyActive(ctx, tx, event.AppCode, scope.tz, scope.day, countryID, regionID, event.UserID, event.OccurredAtMS, nowMS); err != nil { return err diff --git a/services/statistics-service/internal/storage/mysql/social_requirements_query.go b/services/statistics-service/internal/storage/mysql/social_requirements_query.go index f7ba4451..4a18406a 100644 --- a/services/statistics-service/internal/storage/mysql/social_requirements_query.go +++ b/services/statistics-service/internal/storage/mysql/social_requirements_query.go @@ -549,6 +549,11 @@ func buildCanonicalOverviewActiveCube(canonicalRows []socialRequirementUserDay) } cube.Total++ cube.ByDay[row.StatDay]++ + // 匿名设备在无法安全归并到账号时仍属于真实 DAU，因此保留在总计和按日总计；country_id=0 + // 只是“尚无国家维度”，不是一个国家。国家/按日国家明细只能承载有效国家，不能把 0 暴露成伪国家。 + if row.CountryID <= 0 { + continue + } country := cube.ByCountry[row.CountryID] country.ActiveUsers++ @@ -656,12 +661,15 @@ func mergeSocialRequirementUserDay(current, next socialRequirementUserDay) socia if current.DeviceID == "" { current.DeviceID = next.DeviceID } - // Country, region and role are one dimension tuple. A direct same-day u: row - // is authoritative over the d: row even when the anonymous tuple is nonzero; - // copying fields independently would create a country/region pair that never - // existed. No cross-day dimension is borrowed, so history is query-window stable. + // Country and region are one dimension tuple. A direct same-day u: row is + // authoritative only when it carries a real country; a legacy row with just a + // region cannot erase the anonymous row's known country because region alone + // has no valid country ownership semantics. No cross-day dimension is borrowed. if next.DirectUser && !current.DirectUser { - current.CountryID, current.RegionID, current.UserRole = next.CountryID, next.RegionID, next.UserRole + if next.CountryID > 0 { + current.CountryID, current.RegionID = next.CountryID, next.RegionID + } + current.UserRole = next.UserRole current.DirectUser = true } else if next.DirectUser == current.DirectUser { if current.CountryID == 0 && current.RegionID == 0 && (next.CountryID > 0 || next.RegionID > 0) {