修复 Social 用户身份归并
This commit is contained in:
zhx
parent
8a40bec075
commit
5ef0773ed3
@ -22,6 +22,8 @@ type AppTrackingEvent struct {
|
||||
UserID int64 `json:"user_id"`
|
||||
DeviceID string `json:"device_id"`
|
||||
SessionID string `json:"session_id"`
|
||||
ClientSessionID string `json:"client_session_id"`
|
||||
AuthSessionID string `json:"auth_session_id"`
|
||||
Platform string `json:"platform"`
|
||||
AppVersion string `json:"app_version"`
|
||||
Language string `json:"language"`
|
||||
|
||||
@ -27,6 +27,7 @@ type appEventRequest struct {
|
||||
TargetID string `json:"target_id"`
|
||||
DeviceID string `json:"device_id"`
|
||||
SessionID string `json:"session_id"`
|
||||
ClientSessionID string `json:"client_session_id"`
|
||||
Platform string `json:"platform"`
|
||||
AppVersion string `json:"app_version"`
|
||||
Language string `json:"language"`
|
||||
@ -157,6 +158,8 @@ func appTrackingEventFromRequest(item appEventRequest, defaults appTrackingEvent
|
||||
UserID: defaults.userID,
|
||||
DeviceID: deviceID,
|
||||
SessionID: firstTrimmed(defaults.sessionID, item.SessionID),
|
||||
ClientSessionID: item.ClientSessionID,
|
||||
AuthSessionID: defaults.sessionID,
|
||||
Platform: strings.ToLower(firstTrimmed(item.Platform, defaults.platform)),
|
||||
AppVersion: firstTrimmed(item.AppVersion, defaults.appVersion),
|
||||
Language: firstTrimmed(item.Language, defaults.language),
|
||||
@ -201,6 +204,8 @@ func normalizeAppTrackingEventText(event *client.AppTrackingEvent) bool {
|
||||
{value: &event.TargetID, maxBytes: apptracking.MaxTargetIDBytes},
|
||||
{value: &event.DeviceID, maxBytes: apptracking.MaxDeviceIDBytes},
|
||||
{value: &event.SessionID, maxBytes: apptracking.MaxSessionIDBytes},
|
||||
{value: &event.ClientSessionID, maxBytes: apptracking.MaxSessionIDBytes},
|
||||
{value: &event.AuthSessionID, maxBytes: apptracking.MaxSessionIDBytes},
|
||||
{value: &event.Platform, maxBytes: apptracking.MaxPlatformBytes},
|
||||
{value: &event.AppVersion, maxBytes: apptracking.MaxAppVersionBytes},
|
||||
{value: &event.Language, maxBytes: apptracking.MaxLanguageBytes},
|
||||
|
||||
@ -15,13 +15,17 @@ COPY . .
|
||||
RUN --mount=type=cache,id=hyapp-go-mod,target=/go/pkg/mod \
|
||||
--mount=type=cache,id=hyapp-go-build,target=/root/.cache/go-build \
|
||||
mkdir -p /out && \
|
||||
GOWORK=off CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /out/ ./services/statistics-service/cmd/server
|
||||
GOWORK=off CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /out/server ./services/statistics-service/cmd/server && \
|
||||
GOWORK=off CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /out/backfill-social-identities ./services/statistics-service/cmd/backfill-social-identities
|
||||
|
||||
FROM alpine:3.20
|
||||
ENV TZ=UTC
|
||||
WORKDIR /app
|
||||
RUN apk add --no-cache ca-certificates wget && adduser -D -g '' appuser
|
||||
COPY --from=builder /out/server /app/server
|
||||
# 历史身份回放只随 statistics 镜像交付，不开放 HTTP；生产由单节点
|
||||
# docker exec 显式 dry-run/apply，复用容器内只读配置与最小连接数保护。
|
||||
COPY --from=builder /out/backfill-social-identities /app/backfill-social-identities
|
||||
COPY services/statistics-service/configs/config.docker.yaml /app/config.yaml
|
||||
USER appuser
|
||||
EXPOSE 13110
|
||||
|
||||
@ -0,0 +1,356 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"flag"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"hyapp/pkg/appcode"
|
||||
statisticsconfig "hyapp/services/statistics-service/internal/config"
|
||||
|
||||
_ "github.com/go-sql-driver/mysql"
|
||||
)
|
||||
|
||||
const (
|
||||
legacyDayResolutionSource = "legacy_day_unique"
|
||||
legacyWindowResolutionSource = "legacy_window_unique"
|
||||
legacyAmbiguousResolutionSource = "legacy_ambiguous"
|
||||
)
|
||||
|
||||
type replayWindow struct {
|
||||
start time.Time
|
||||
end time.Time
|
||||
loc *time.Location
|
||||
}
|
||||
|
||||
type deviceResolution struct {
|
||||
userID int64
|
||||
ambiguous bool
|
||||
}
|
||||
|
||||
type dayResolution struct {
|
||||
deviceID string
|
||||
userID int64
|
||||
state string
|
||||
source string
|
||||
}
|
||||
|
||||
type daySummary struct {
|
||||
day string
|
||||
anonymous int
|
||||
resolved int
|
||||
ambiguous int
|
||||
unresolved int
|
||||
}
|
||||
|
||||
func main() {
|
||||
var (
|
||||
appsValue = flag.String("apps", envDefault("APP_CODES", "lalu,huwaa,fami"), "comma-separated app_code list")
|
||||
statTZ = flag.String("timezone", envDefault("STAT_TZ", "Asia/Shanghai"), "UTC or Asia/Shanghai")
|
||||
startDay = flag.String("start-day", "", "inclusive day, yyyy-mm-dd")
|
||||
endDay = flag.String("end-day", "", "inclusive day, yyyy-mm-dd")
|
||||
apply = flag.Bool("apply", false, "write identity resolutions; default is dry-run")
|
||||
configPath = flag.String("config", envDefault("STATISTICS_CONFIG", "/app/config.yaml"), "statistics-service YAML used when no DSN flag/env is set")
|
||||
dsn = flag.String("statistics-dsn", firstNonEmptyEnv("STATISTICS_MYSQL_DSN", "MYSQL_DSN"), "statistics MySQL DSN")
|
||||
pause = flag.Duration("pause", 250*time.Millisecond, "pause between bounded day queries")
|
||||
slicePause = flag.Duration("slice-pause", 50*time.Millisecond, "pause after each one-hour raw scan")
|
||||
queryLimit = flag.Duration("query-timeout", 30*time.Second, "timeout for each bounded query or day transaction")
|
||||
maxOpenConn = flag.Int("max-open-conns", 1, "small connection cap for production-safe backfill")
|
||||
receivedCutoff = flag.Int64("received-before-ms", time.Now().UTC().UnixMilli(), "only use raw events received at or before this fixed cutoff")
|
||||
allowWindowFallback = flag.Bool("allow-window-fallback", false, "allow cross-day unique device inference; unsafe by default for legacy shared devices")
|
||||
)
|
||||
flag.Parse()
|
||||
|
||||
apps := normalizeApps(*appsValue)
|
||||
dsnValue := strings.TrimSpace(*dsn)
|
||||
if dsnValue == "" {
|
||||
cfg, err := statisticsconfig.Load(strings.TrimSpace(*configPath))
|
||||
if err != nil {
|
||||
log.Fatalf("load statistics config: %v", err)
|
||||
}
|
||||
dsnValue = strings.TrimSpace(cfg.MySQLDSN)
|
||||
}
|
||||
if len(apps) == 0 || dsnValue == "" {
|
||||
log.Fatal("apps and statistics MySQL DSN are required")
|
||||
}
|
||||
window, err := resolveWindow(*startDay, *endDay, *statTZ)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
db, err := sql.Open("mysql", dsnValue)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
if *maxOpenConn < 1 {
|
||||
*maxOpenConn = 1
|
||||
}
|
||||
db.SetMaxOpenConns(*maxOpenConn)
|
||||
db.SetMaxIdleConns(1)
|
||||
db.SetConnMaxLifetime(5 * time.Minute)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), *queryLimit)
|
||||
if err := db.PingContext(ctx); err != nil {
|
||||
cancel()
|
||||
log.Fatal(err)
|
||||
}
|
||||
cancel()
|
||||
fmt.Printf("identity backfill cutoff received_before_ms=%d (%s)\n", *receivedCutoff, time.UnixMilli(*receivedCutoff).UTC().Format(time.RFC3339))
|
||||
|
||||
for _, app := range apps {
|
||||
if err := backfillApp(db, app, *statTZ, window, *receivedCutoff, *allowWindowFallback, *apply, *pause, *slicePause, *queryLimit); err != nil {
|
||||
log.Fatalf("app=%s: %v", app, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func backfillApp(db *sql.DB, app, statTZ string, window replayWindow, receivedCutoff int64, allowWindowFallback, apply bool, pause, slicePause, queryLimit time.Duration) error {
|
||||
windowDeviceUsers := map[string]deviceResolution{}
|
||||
dayDeviceUsers := map[string]map[string]deviceResolution{}
|
||||
for day := window.start; !day.After(window.end); day = day.AddDate(0, 0, 1) {
|
||||
dayText := day.Format("2006-01-02")
|
||||
dayDeviceUsers[dayText] = map[string]deviceResolution{}
|
||||
// One-hour slices cap every index range even on the peak day. All
|
||||
// authenticated event names participate: an account that only emitted
|
||||
// page_open must still disqualify a shared device from "unique" mapping.
|
||||
for hour := day; hour.Before(day.AddDate(0, 0, 1)); hour = hour.Add(time.Hour) {
|
||||
if err := loadLoggedIdentitySlice(db, app, hour.UnixMilli(), hour.Add(time.Hour).UnixMilli(), receivedCutoff,
|
||||
dayDeviceUsers[dayText], windowDeviceUsers, queryLimit); err != nil {
|
||||
return fmt.Errorf("load logged identities day=%s hour=%s: %w", dayText, hour.Format("15:04"), err)
|
||||
}
|
||||
pauseFor(slicePause)
|
||||
}
|
||||
pauseFor(pause)
|
||||
}
|
||||
|
||||
for day := window.start; !day.After(window.end); day = day.AddDate(0, 0, 1) {
|
||||
dayText := day.Format("2006-01-02")
|
||||
resolutions, summary, err := planDay(db, app, statTZ, dayText, dayDeviceUsers[dayText], windowDeviceUsers, allowWindowFallback, queryLimit)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
mode := "dry-run"
|
||||
if apply {
|
||||
mode = "applied"
|
||||
if err := applyDay(db, app, statTZ, dayText, resolutions, queryLimit); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
fmt.Printf("%s app=%s stat_tz=%s day=%s anonymous=%d resolved=%d ambiguous=%d unresolved=%d\n",
|
||||
mode, app, statTZ, summary.day, summary.anonymous, summary.resolved, summary.ambiguous, summary.unresolved)
|
||||
pauseFor(pause)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func loadLoggedIdentitySlice(db *sql.DB, app string, startMS, endMS, receivedCutoff int64, dayUsers, windowUsers map[string]deviceResolution, queryLimit time.Duration) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), queryLimit)
|
||||
defer cancel()
|
||||
rows, err := db.QueryContext(ctx, `
|
||||
SELECT device_id, user_id
|
||||
FROM app_tracking_events FORCE INDEX (idx_app_tracking_time_window)
|
||||
WHERE app_code = ? AND occurred_at_ms >= ? AND occurred_at_ms < ? AND received_at_ms <= ?
|
||||
AND device_id <> '' AND user_id > 0
|
||||
`, app, startMS, endMS, receivedCutoff)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var deviceID string
|
||||
var userID int64
|
||||
if err := rows.Scan(&deviceID, &userID); err != nil {
|
||||
return err
|
||||
}
|
||||
deviceID = strings.TrimSpace(deviceID)
|
||||
if deviceID == "" || userID <= 0 {
|
||||
continue
|
||||
}
|
||||
addDeviceUser(dayUsers, deviceID, userID)
|
||||
addDeviceUser(windowUsers, deviceID, userID)
|
||||
}
|
||||
return rows.Err()
|
||||
}
|
||||
|
||||
func addDeviceUser(target map[string]deviceResolution, deviceID string, userID int64) {
|
||||
current, exists := target[deviceID]
|
||||
if !exists {
|
||||
target[deviceID] = deviceResolution{userID: userID}
|
||||
return
|
||||
}
|
||||
if current.ambiguous || current.userID == userID {
|
||||
return
|
||||
}
|
||||
current.ambiguous = true
|
||||
current.userID = 0
|
||||
target[deviceID] = current
|
||||
}
|
||||
|
||||
func planDay(db *sql.DB, app, statTZ, day string, dayUsers, windowUsers map[string]deviceResolution, allowWindowFallback bool, queryLimit time.Duration) ([]dayResolution, daySummary, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), queryLimit)
|
||||
defer cancel()
|
||||
rows, err := db.QueryContext(ctx, `
|
||||
SELECT device_id
|
||||
FROM stat_social_user_day
|
||||
WHERE app_code = ? AND stat_tz = ? AND stat_day = ?
|
||||
AND user_id = 0 AND device_id <> ''
|
||||
ORDER BY device_id
|
||||
`, app, statTZ, day)
|
||||
if err != nil {
|
||||
return nil, daySummary{}, fmt.Errorf("load anonymous devices day=%s: %w", day, err)
|
||||
}
|
||||
defer rows.Close()
|
||||
summary := daySummary{day: day}
|
||||
resolutions := []dayResolution{}
|
||||
for rows.Next() {
|
||||
var deviceID string
|
||||
if err := rows.Scan(&deviceID); err != nil {
|
||||
return nil, daySummary{}, err
|
||||
}
|
||||
summary.anonymous++
|
||||
identity, ok := dayUsers[deviceID]
|
||||
switch {
|
||||
case ok && (identity.ambiguous || identity.userID <= 0):
|
||||
summary.ambiguous++
|
||||
resolutions = append(resolutions, dayResolution{deviceID: deviceID, state: "ambiguous", source: legacyAmbiguousResolutionSource})
|
||||
case ok:
|
||||
summary.resolved++
|
||||
resolutions = append(resolutions, dayResolution{deviceID: deviceID, userID: identity.userID, state: "resolved", source: legacyDayResolutionSource})
|
||||
default:
|
||||
if !allowWindowFallback {
|
||||
summary.unresolved++
|
||||
continue
|
||||
}
|
||||
identity, ok = windowUsers[deviceID]
|
||||
if !ok {
|
||||
summary.unresolved++
|
||||
continue
|
||||
}
|
||||
if identity.ambiguous || identity.userID <= 0 {
|
||||
summary.ambiguous++
|
||||
resolutions = append(resolutions, dayResolution{deviceID: deviceID, state: "ambiguous", source: legacyAmbiguousResolutionSource})
|
||||
continue
|
||||
}
|
||||
// Cross-day unique fallback repairs the exact retention gap where day 1
|
||||
// was anonymous and the same device first authenticated on day 2. It is
|
||||
// never used if the device showed more than one account in the window.
|
||||
summary.resolved++
|
||||
resolutions = append(resolutions, dayResolution{deviceID: deviceID, userID: identity.userID, state: "resolved", source: legacyWindowResolutionSource})
|
||||
}
|
||||
}
|
||||
return resolutions, summary, rows.Err()
|
||||
}
|
||||
|
||||
func applyDay(db *sql.DB, app, statTZ, day string, resolutions []dayResolution, queryLimit time.Duration) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), queryLimit)
|
||||
defer cancel()
|
||||
tx, err := db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
// Only rows produced by this legacy job are replaced. Realtime client-session
|
||||
// evidence has higher authority and must survive an idempotent historical rerun.
|
||||
if _, err := tx.ExecContext(ctx, `
|
||||
DELETE FROM stat_social_identity_day
|
||||
WHERE app_code = ? AND stat_tz = ? AND stat_day = ?
|
||||
AND resolution_source IN (?, ?, ?)
|
||||
`, app, statTZ, day, legacyDayResolutionSource, legacyWindowResolutionSource, legacyAmbiguousResolutionSource); err != nil {
|
||||
return err
|
||||
}
|
||||
stmt, err := tx.PrepareContext(ctx, `
|
||||
INSERT INTO stat_social_identity_day (
|
||||
app_code, stat_tz, stat_day, device_id, resolved_user_id,
|
||||
resolution_state, resolution_source, updated_at_ms
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||||
ON DUPLICATE KEY UPDATE
|
||||
resolved_user_id = IF(resolution_source = 'client_session', resolved_user_id, VALUES(resolved_user_id)),
|
||||
resolution_state = IF(resolution_source = 'client_session', resolution_state, VALUES(resolution_state)),
|
||||
resolution_source = IF(resolution_source = 'client_session', resolution_source, VALUES(resolution_source)),
|
||||
updated_at_ms = IF(resolution_source = 'client_session', updated_at_ms, VALUES(updated_at_ms))
|
||||
`)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer stmt.Close()
|
||||
nowMS := time.Now().UTC().UnixMilli()
|
||||
for _, resolution := range resolutions {
|
||||
if _, err := stmt.ExecContext(ctx, app, statTZ, day, resolution.deviceID, resolution.userID,
|
||||
resolution.state, resolution.source, nowMS); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return tx.Commit()
|
||||
}
|
||||
|
||||
func resolveWindow(startDay, endDay, statTZ string) (replayWindow, error) {
|
||||
var loc *time.Location
|
||||
switch strings.TrimSpace(statTZ) {
|
||||
case "UTC":
|
||||
loc = time.UTC
|
||||
case "Asia/Shanghai":
|
||||
loc = time.FixedZone("Asia/Shanghai", 8*60*60)
|
||||
default:
|
||||
return replayWindow{}, fmt.Errorf("unsupported timezone %q", statTZ)
|
||||
}
|
||||
start, err := time.ParseInLocation("2006-01-02", strings.TrimSpace(startDay), loc)
|
||||
if err != nil {
|
||||
return replayWindow{}, fmt.Errorf("parse start-day: %w", err)
|
||||
}
|
||||
end, err := time.ParseInLocation("2006-01-02", strings.TrimSpace(endDay), loc)
|
||||
if err != nil {
|
||||
return replayWindow{}, fmt.Errorf("parse end-day: %w", err)
|
||||
}
|
||||
if end.Before(start) {
|
||||
return replayWindow{}, fmt.Errorf("end-day is before start-day")
|
||||
}
|
||||
if days := int(end.Sub(start).Hours()/24) + 1; days > 31 {
|
||||
return replayWindow{}, fmt.Errorf("window has %d days; maximum is 31", days)
|
||||
}
|
||||
return replayWindow{start: start, end: end, loc: loc}, nil
|
||||
}
|
||||
|
||||
func normalizeApps(value string) []string {
|
||||
seen := map[string]struct{}{}
|
||||
apps := []string{}
|
||||
for _, item := range strings.Split(value, ",") {
|
||||
app := appcode.Normalize(item)
|
||||
if app == "" {
|
||||
continue
|
||||
}
|
||||
if _, ok := seen[app]; ok {
|
||||
continue
|
||||
}
|
||||
seen[app] = struct{}{}
|
||||
apps = append(apps, app)
|
||||
}
|
||||
sort.Strings(apps)
|
||||
return apps
|
||||
}
|
||||
|
||||
func pauseFor(pause time.Duration) {
|
||||
if pause > 0 {
|
||||
time.Sleep(pause)
|
||||
}
|
||||
}
|
||||
|
||||
func envDefault(key, fallback string) string {
|
||||
if value := strings.TrimSpace(os.Getenv(key)); value != "" {
|
||||
return value
|
||||
}
|
||||
return fallback
|
||||
}
|
||||
|
||||
func firstNonEmptyEnv(keys ...string) string {
|
||||
for _, key := range keys {
|
||||
if value := strings.TrimSpace(os.Getenv(key)); value != "" {
|
||||
return value
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
@ -28,7 +28,9 @@ CREATE TABLE IF NOT EXISTS app_tracking_events (
|
||||
user_id BIGINT NOT NULL DEFAULT 0 COMMENT '服务端识别的用户 ID，匿名为 0',
|
||||
device_id VARCHAR(128) NOT NULL DEFAULT '' COMMENT '设备 ID',
|
||||
identity_key VARCHAR(160) GENERATED ALWAYS AS (CASE WHEN user_id > 0 THEN CONCAT('u:', user_id) ELSE CONCAT('d:', device_id) END) VIRTUAL COMMENT '漏斗/留存去重身份，登录用户优先 user_id，匿名用户回退 device_id',
|
||||
session_id VARCHAR(160) NOT NULL DEFAULT '' COMMENT '会话 ID',
|
||||
session_id VARCHAR(160) NOT NULL DEFAULT '' COMMENT '兼容会话 ID',
|
||||
client_session_id VARCHAR(160) NOT NULL DEFAULT '' COMMENT 'App 进程会话 ID，登录前后保持不变',
|
||||
auth_session_id VARCHAR(160) NOT NULL DEFAULT '' COMMENT '服务端认证会话 ID',
|
||||
platform VARCHAR(32) NOT NULL DEFAULT '' COMMENT '平台',
|
||||
app_version VARCHAR(64) NOT NULL DEFAULT '' COMMENT 'App 版本',
|
||||
language VARCHAR(32) NOT NULL DEFAULT '' COMMENT '语言',
|
||||
@ -205,6 +207,49 @@ CREATE TABLE IF NOT EXISTS stat_social_user_day (
|
||||
KEY idx_stat_social_user_day_filter (app_code, stat_tz, stat_day, user_role, recharge_user)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='Social BI P0-P5 用户日级读模型';
|
||||
|
||||
CREATE TABLE IF NOT EXISTS stat_social_identity_session_user (
|
||||
app_code VARCHAR(32) NOT NULL,
|
||||
device_id VARCHAR(128) COLLATE utf8mb4_bin NOT NULL,
|
||||
client_session_id VARCHAR(160) COLLATE utf8mb4_bin NOT NULL,
|
||||
user_id BIGINT NOT NULL,
|
||||
first_seen_at_ms BIGINT NOT NULL,
|
||||
updated_at_ms BIGINT NOT NULL,
|
||||
PRIMARY KEY (app_code, device_id, client_session_id, user_id),
|
||||
KEY idx_social_identity_session_user (app_code, user_id, first_seen_at_ms)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='Social 客户端会话登录账号观测表';
|
||||
|
||||
CREATE TABLE IF NOT EXISTS stat_social_identity_device_lock (
|
||||
app_code VARCHAR(32) NOT NULL,
|
||||
device_id VARCHAR(128) COLLATE utf8mb4_bin NOT NULL,
|
||||
created_at_ms BIGINT NOT NULL,
|
||||
PRIMARY KEY (app_code, device_id)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='Social 身份解析设备级事务锁';
|
||||
|
||||
CREATE TABLE IF NOT EXISTS stat_social_identity_session_day (
|
||||
app_code VARCHAR(32) NOT NULL,
|
||||
stat_tz VARCHAR(64) NOT NULL,
|
||||
stat_day DATE NOT NULL,
|
||||
device_id VARCHAR(128) COLLATE utf8mb4_bin NOT NULL,
|
||||
client_session_id VARCHAR(160) COLLATE utf8mb4_bin NOT NULL,
|
||||
first_seen_at_ms BIGINT NOT NULL,
|
||||
updated_at_ms BIGINT NOT NULL,
|
||||
PRIMARY KEY (app_code, device_id, client_session_id, stat_tz, stat_day),
|
||||
KEY idx_social_identity_session_day (app_code, stat_tz, stat_day, device_id, client_session_id)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='Social 匿名会话活跃日观测表';
|
||||
|
||||
CREATE TABLE IF NOT EXISTS stat_social_identity_day (
|
||||
app_code VARCHAR(32) NOT NULL,
|
||||
stat_tz VARCHAR(64) NOT NULL,
|
||||
stat_day DATE NOT NULL,
|
||||
device_id VARCHAR(128) COLLATE utf8mb4_bin NOT NULL,
|
||||
resolved_user_id BIGINT NOT NULL DEFAULT 0,
|
||||
resolution_state VARCHAR(16) NOT NULL DEFAULT 'unresolved',
|
||||
resolution_source VARCHAR(32) NOT NULL DEFAULT 'client_session',
|
||||
updated_at_ms BIGINT NOT NULL,
|
||||
PRIMARY KEY (app_code, stat_tz, stat_day, device_id),
|
||||
KEY idx_social_identity_day_user (app_code, stat_tz, stat_day, resolved_user_id)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='Social 设备日级 canonical 身份解析表';
|
||||
|
||||
CREATE TABLE IF NOT EXISTS stat_recharge_day_payers (
|
||||
app_code VARCHAR(32) NOT NULL,
|
||||
stat_tz VARCHAR(64) NOT NULL DEFAULT 'UTC' COMMENT '统计时区，UTC 或 Asia/Shanghai',
|
||||
|
||||
@ -296,6 +296,8 @@ type appTrackingEventRequest struct {
|
||||
UserID int64 `json:"user_id"`
|
||||
DeviceID string `json:"device_id"`
|
||||
SessionID string `json:"session_id"`
|
||||
ClientSessionID string `json:"client_session_id"`
|
||||
AuthSessionID string `json:"auth_session_id"`
|
||||
Platform string `json:"platform"`
|
||||
AppVersion string `json:"app_version"`
|
||||
Language string `json:"language"`
|
||||
@ -385,6 +387,8 @@ func (s *queryHTTPServer) appEvents(w http.ResponseWriter, r *http.Request) {
|
||||
UserID: item.UserID,
|
||||
DeviceID: item.DeviceID,
|
||||
SessionID: item.SessionID,
|
||||
ClientSessionID: item.ClientSessionID,
|
||||
AuthSessionID: item.AuthSessionID,
|
||||
Platform: item.Platform,
|
||||
AppVersion: item.AppVersion,
|
||||
Language: item.Language,
|
||||
|
||||
@ -27,6 +27,32 @@ func TestGiftDebitConsumptionOwnersDoNotOverlap(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestCanonicalizeSocialUserDaysMergesIdentityWithoutLosingDeviceCount(t *testing.T) {
|
||||
if socialRetentionLookaheadDays([]string{SocialSectionActive}) != 0 ||
|
||||
socialRetentionLookaheadDays([]string{SocialSectionNewUsers}) != 1 ||
|
||||
socialRetentionLookaheadDays([]string{SocialSectionActive, SocialSectionRetention}) != 30 {
|
||||
t.Fatal("social retention lookahead must stay section-bounded")
|
||||
}
|
||||
rows := []socialRequirementUserDay{
|
||||
{StatDay: "2026-07-01", Subject: "d:device-A", DeviceID: "device-A", CountryID: 999, RegionID: 999, UserRole: "user", AppFirstOpen: 1, Active: 1, AppStayMS: 60_000, RoomStayMS: 20_000},
|
||||
{StatDay: "2026-07-01", Subject: "u:123", UserID: 123, DirectUser: true, DeviceID: "device-A", CountryID: 86, RegionID: 210, UserRole: "host", AppFirstOpen: 1, Active: 1, AppStayMS: 30_000, RoomOnlineMS: 45_000},
|
||||
}
|
||||
resolved := map[socialIdentityDeviceDayKey]int64{
|
||||
{StatDay: "2026-07-01", DeviceID: "device-A"}: 123,
|
||||
}
|
||||
got := canonicalizeSocialUserDays(rows, resolved, 0, 0, []int64{210})
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("expected one canonical user-day, got %+v", got)
|
||||
}
|
||||
row := got[0]
|
||||
if row.Subject != "u:123" || row.UserID != 123 || !row.DirectUser || row.CountryID != 86 || row.RegionID != 210 || row.UserRole != "host" || row.Active != 1 {
|
||||
t.Fatalf("canonical identity/dimension mismatch: %+v", row)
|
||||
}
|
||||
if row.AppFirstOpen != 1 || row.AppStayMS != 90_000 || row.RoomStayMS != 45_000 {
|
||||
t.Fatalf("canonical metric merge mismatch: %+v", row)
|
||||
}
|
||||
}
|
||||
|
||||
func TestQueryOverviewUsesActivityLuckyGiftDayStats(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
_, file, _, _ := runtime.Caller(0)
|
||||
@ -134,6 +160,8 @@ func TestConsumeAppTrackingEventsStoresRawEventsAndDeduplicates(t *testing.T) {
|
||||
UserID: 42,
|
||||
DeviceID: "dev-1",
|
||||
SessionID: "sess-1",
|
||||
ClientSessionID: "client-sess-1",
|
||||
AuthSessionID: "auth-sess-1",
|
||||
Platform: "ios",
|
||||
AppVersion: "1.2.3",
|
||||
Language: "en-US",
|
||||
@ -172,25 +200,130 @@ func TestConsumeAppTrackingEventsStoresRawEventsAndDeduplicates(t *testing.T) {
|
||||
}
|
||||
|
||||
var count int64
|
||||
var statDay, screen, identityKey, propertiesJSON string
|
||||
var statDay, screen, identityKey, clientSessionID, authSessionID, propertiesJSON string
|
||||
var userID, countryID, regionID, durationMS int64
|
||||
var success bool
|
||||
if err := repository.db.QueryRowContext(ctx, `
|
||||
SELECT COUNT(*), MAX(CAST(stat_day AS CHAR)), MAX(screen), MAX(identity_key), MAX(user_id), MAX(country_id), MAX(region_id),
|
||||
SELECT COUNT(*), MAX(CAST(stat_day AS CHAR)), MAX(screen), MAX(identity_key),
|
||||
MAX(client_session_id), MAX(auth_session_id), MAX(user_id), MAX(country_id), MAX(region_id),
|
||||
MAX(duration_ms), MAX(success), COALESCE(MAX(CAST(properties_json AS CHAR)), '{}')
|
||||
FROM app_tracking_events
|
||||
WHERE app_code = 'lalu' AND event_id = 'evt-banner-1'
|
||||
`).Scan(&count, &statDay, &screen, &identityKey, &userID, &countryID, ®ionID, &durationMS, &success, &propertiesJSON); err != nil {
|
||||
`).Scan(&count, &statDay, &screen, &identityKey, &clientSessionID, &authSessionID, &userID, &countryID, ®ionID, &durationMS, &success, &propertiesJSON); err != nil {
|
||||
t.Fatalf("query app tracking event: %v", err)
|
||||
}
|
||||
if count != 1 || statDay != "2026-07-02" || screen != "home" || identityKey != "u:42" || userID != 42 || countryID != 86 || regionID != 210 || durationMS != 12 || !success {
|
||||
t.Fatalf("stored event mismatch: count=%d statDay=%s screen=%s identityKey=%s userID=%d countryID=%d regionID=%d durationMS=%d success=%v", count, statDay, screen, identityKey, userID, countryID, regionID, durationMS, success)
|
||||
if count != 1 || statDay != "2026-07-02" || screen != "home" || identityKey != "u:42" || clientSessionID != "client-sess-1" || authSessionID != "auth-sess-1" || userID != 42 || countryID != 86 || regionID != 210 || durationMS != 12 || !success {
|
||||
t.Fatalf("stored event mismatch: count=%d statDay=%s screen=%s identityKey=%s clientSessionID=%s authSessionID=%s userID=%d countryID=%d regionID=%d durationMS=%d success=%v", count, statDay, screen, identityKey, clientSessionID, authSessionID, userID, countryID, regionID, durationMS, success)
|
||||
}
|
||||
if DecodeJSON(propertiesJSON)["slot"] != "top" {
|
||||
t.Fatalf("properties_json mismatch: %s", propertiesJSON)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSocialIdentityCanonicalizesAnonymousAndLoggedSession(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_social_identity_test",
|
||||
})
|
||||
repository, err := Open(ctx, statsSchema.DSN)
|
||||
if err != nil {
|
||||
t.Fatalf("open repository: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = repository.Close() })
|
||||
|
||||
start := time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC)
|
||||
events := []AppTrackingEvent{
|
||||
// Both pre-login rows are stored as d:device-A. The authenticated event
|
||||
// establishes the client-session bridge without rewriting those rows.
|
||||
{AppCode: "lalu", EventID: "identity:first-open", EventName: "app_first_open", DeviceID: "device-A", ClientSessionID: "client-session-A", Success: true, OccurredAtMS: start.UnixMilli()},
|
||||
{AppCode: "lalu", EventID: "identity:anonymous-session", EventName: "app_session", DeviceID: "device-A", ClientSessionID: "client-session-A", DurationMS: 120_000, Success: true, OccurredAtMS: start.Add(time.Minute).UnixMilli()},
|
||||
{AppCode: "lalu", EventID: "identity:login", EventName: "home_view", UserID: 123, DeviceID: "device-A", ClientSessionID: "client-session-A", AuthSessionID: "auth-session-A", CountryID: 86, RegionID: 210, Success: true, OccurredAtMS: start.Add(2 * time.Minute).UnixMilli()},
|
||||
{AppCode: "lalu", EventID: "identity:d1", EventName: "home_view", UserID: 123, DeviceID: "device-A", ClientSessionID: "client-session-A", AuthSessionID: "auth-session-A", CountryID: 86, RegionID: 999, Success: true, OccurredAtMS: start.Add(24*time.Hour + time.Minute).UnixMilli()},
|
||||
}
|
||||
if _, err := repository.ConsumeAppTrackingEvents(appcode.WithContext(ctx, "lalu"), events); err != nil {
|
||||
t.Fatalf("consume identity events: %v", err)
|
||||
}
|
||||
|
||||
result, err := repository.QuerySocialRequirements(ctx, SocialRequirementsQuery{
|
||||
AppCode: "lalu",
|
||||
StartMS: start.UnixMilli(),
|
||||
EndMS: start.Add(24 * time.Hour).UnixMilli(),
|
||||
RegionIDs: []int64{210},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("query canonical social requirements: %v", err)
|
||||
}
|
||||
active := socialTestSection(t, result, SocialSectionActive)
|
||||
if socialMetricInt(t, active.Total, "active_users") != 1 || math.Abs(socialMetricFloat(t, active.Total, "avg_app_stay_min")-2) > 0.0001 {
|
||||
t.Fatalf("anonymous and logged rows were not canonicalized: %+v", active.Total.Metrics)
|
||||
}
|
||||
newUsers := socialTestSection(t, result, SocialSectionNewUsers)
|
||||
if socialMetricInt(t, newUsers.Total, "app_download_users") != 1 {
|
||||
t.Fatalf("device download count changed after identity merge: %+v", newUsers.Total.Metrics)
|
||||
}
|
||||
retention := socialTestSection(t, result, SocialSectionRetention)
|
||||
if socialMetricInt(t, retention.Total, "active_d1_active_base_users") != 1 ||
|
||||
socialMetricInt(t, retention.Total, "active_d1_active_users") != 1 ||
|
||||
math.Abs(socialMetricFloat(t, retention.Total, "active_d1_active_rate")-1) > 0.0001 {
|
||||
t.Fatalf("canonical retention mismatch: %+v", retention.Total.Metrics)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSocialIdentityMarksConcurrentAccountSwitchAmbiguous(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_identity_switch",
|
||||
})
|
||||
repository, err := Open(ctx, statsSchema.DSN)
|
||||
if err != nil {
|
||||
t.Fatalf("open repository: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = repository.Close() })
|
||||
|
||||
start := time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC)
|
||||
if _, err := repository.ConsumeAppTrackingEvents(appcode.WithContext(ctx, "lalu"), []AppTrackingEvent{{
|
||||
AppCode: "lalu", EventID: "switch:anonymous", EventName: "home_view", DeviceID: "shared-device", ClientSessionID: "shared-process-session", Success: true, OccurredAtMS: start.UnixMilli(),
|
||||
}}); err != nil {
|
||||
t.Fatalf("consume anonymous event: %v", err)
|
||||
}
|
||||
|
||||
// Deliberately race two accounts in one process session. The device lock
|
||||
// must serialize the current reads so the last writer cannot leave a falsely
|
||||
// resolved user behind.
|
||||
errors := make(chan error, 2)
|
||||
for userID := int64(1); userID <= 2; userID++ {
|
||||
userID := userID
|
||||
go func() {
|
||||
_, consumeErr := repository.ConsumeAppTrackingEvents(appcode.WithContext(ctx, "lalu"), []AppTrackingEvent{{
|
||||
AppCode: "lalu", EventID: "switch:user:" + strconv.FormatInt(userID, 10), EventName: "home_view", UserID: userID, DeviceID: "shared-device", ClientSessionID: "shared-process-session", Success: true, OccurredAtMS: start.Add(time.Duration(userID) * time.Minute).UnixMilli(),
|
||||
}})
|
||||
errors <- consumeErr
|
||||
}()
|
||||
}
|
||||
for index := 0; index < 2; index++ {
|
||||
if err := <-errors; err != nil {
|
||||
t.Fatalf("consume concurrent account event: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
var resolvedUserID int64
|
||||
var state, source string
|
||||
if err := repository.db.QueryRowContext(ctx, `
|
||||
SELECT resolved_user_id, resolution_state, resolution_source
|
||||
FROM stat_social_identity_day
|
||||
WHERE app_code = 'lalu' AND stat_tz = 'UTC' AND stat_day = '2026-07-01' AND device_id = 'shared-device'
|
||||
`).Scan(&resolvedUserID, &state, &source); err != nil {
|
||||
t.Fatalf("query identity resolution: %v", err)
|
||||
}
|
||||
if resolvedUserID != 0 || state != "ambiguous" || source != "client_session" {
|
||||
t.Fatalf("account switch must remain ambiguous: user=%d state=%s source=%s", resolvedUserID, state, source)
|
||||
}
|
||||
}
|
||||
|
||||
func TestQueryAppTrackingFunnelAggregatesStepsAndD1Cohorts(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
_, file, _, _ := runtime.Caller(0)
|
||||
@ -355,6 +488,7 @@ func TestQuerySocialRequirementsAggregatesP0ToP5(t *testing.T) {
|
||||
{AppCode: "lalu", EventID: "social:follow-user:1001", EventName: "follow_host", UserID: user1, DeviceID: "dev-1001", CountryID: 86, RegionID: 210, Success: true, OccurredAtMS: start.Add(7 * time.Minute).UnixMilli()},
|
||||
{AppCode: "lalu", EventID: "social:follow-room:1001", EventName: "follow_room", UserID: user1, DeviceID: "dev-1001", CountryID: 86, RegionID: 210, Success: true, OccurredAtMS: start.Add(8 * time.Minute).UnixMilli()},
|
||||
{AppCode: "lalu", EventID: "social:d1-active:1001", EventName: "home_view", UserID: user1, DeviceID: "dev-1001", CountryID: 86, RegionID: 210, Success: true, OccurredAtMS: start.Add(24*time.Hour + time.Minute).UnixMilli()},
|
||||
{AppCode: "lalu", EventID: "social:d1-room:1001", EventName: "room_join_success", UserID: user1, DeviceID: "dev-1001", CountryID: 86, RegionID: 210, Success: true, OccurredAtMS: start.Add(24*time.Hour + 2*time.Minute).UnixMilli()},
|
||||
} {
|
||||
if _, err := repository.ConsumeAppTrackingEvents(appcode.WithContext(ctx, "lalu"), []AppTrackingEvent{event}); err != nil {
|
||||
t.Fatalf("consume app event %s: %v", event.EventName, err)
|
||||
|
||||
@ -5,6 +5,7 @@ import (
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@ -122,7 +123,8 @@ func (r *Repository) Migrate(ctx context.Context) error {
|
||||
target_type VARCHAR(64) NOT NULL DEFAULT '', target_id VARCHAR(128) NOT NULL DEFAULT '',
|
||||
user_id BIGINT NOT NULL DEFAULT 0, device_id VARCHAR(128) NOT NULL DEFAULT '',
|
||||
identity_key VARCHAR(160) GENERATED ALWAYS AS (CASE WHEN user_id > 0 THEN CONCAT('u:', user_id) ELSE CONCAT('d:', device_id) END) VIRTUAL,
|
||||
session_id VARCHAR(160) NOT NULL DEFAULT '', platform VARCHAR(32) NOT NULL DEFAULT '',
|
||||
session_id VARCHAR(160) NOT NULL DEFAULT '', client_session_id VARCHAR(160) NOT NULL DEFAULT '',
|
||||
auth_session_id VARCHAR(160) NOT NULL DEFAULT '', platform VARCHAR(32) NOT NULL DEFAULT '',
|
||||
app_version VARCHAR(64) NOT NULL DEFAULT '', language VARCHAR(32) NOT NULL DEFAULT '',
|
||||
timezone VARCHAR(64) NOT NULL DEFAULT '', country_id BIGINT NOT NULL DEFAULT 0,
|
||||
region_id BIGINT NOT NULL DEFAULT 0, duration_ms BIGINT NOT NULL DEFAULT 0,
|
||||
@ -216,6 +218,34 @@ func (r *Repository) Migrate(ctx context.Context) error {
|
||||
KEY idx_stat_social_user_day_region (app_code, stat_tz, stat_day, region_id),
|
||||
KEY idx_stat_social_user_day_filter (app_code, stat_tz, stat_day, user_role, recharge_user)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||||
`CREATE TABLE IF NOT EXISTS stat_social_identity_session_user (
|
||||
app_code VARCHAR(32) NOT NULL, device_id VARCHAR(128) COLLATE utf8mb4_bin NOT NULL,
|
||||
client_session_id VARCHAR(160) COLLATE utf8mb4_bin NOT NULL,
|
||||
user_id BIGINT NOT NULL, first_seen_at_ms BIGINT NOT NULL, updated_at_ms BIGINT NOT NULL,
|
||||
PRIMARY KEY (app_code, device_id, client_session_id, user_id),
|
||||
KEY idx_social_identity_session_user (app_code, user_id, first_seen_at_ms)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||||
`CREATE TABLE IF NOT EXISTS stat_social_identity_device_lock (
|
||||
app_code VARCHAR(32) NOT NULL, device_id VARCHAR(128) COLLATE utf8mb4_bin NOT NULL,
|
||||
created_at_ms BIGINT NOT NULL,
|
||||
PRIMARY KEY (app_code, device_id)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||||
`CREATE TABLE IF NOT EXISTS stat_social_identity_session_day (
|
||||
app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL, stat_day DATE NOT NULL,
|
||||
device_id VARCHAR(128) COLLATE utf8mb4_bin NOT NULL,
|
||||
client_session_id VARCHAR(160) COLLATE utf8mb4_bin NOT NULL,
|
||||
first_seen_at_ms BIGINT NOT NULL, updated_at_ms BIGINT NOT NULL,
|
||||
PRIMARY KEY (app_code, device_id, client_session_id, stat_tz, stat_day),
|
||||
KEY idx_social_identity_session_day (app_code, stat_tz, stat_day, device_id, client_session_id)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||||
`CREATE TABLE IF NOT EXISTS stat_social_identity_day (
|
||||
app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL, stat_day DATE NOT NULL,
|
||||
device_id VARCHAR(128) COLLATE utf8mb4_bin NOT NULL, resolved_user_id BIGINT NOT NULL DEFAULT 0,
|
||||
resolution_state VARCHAR(16) NOT NULL DEFAULT 'unresolved',
|
||||
resolution_source VARCHAR(32) NOT NULL DEFAULT 'client_session', updated_at_ms BIGINT NOT NULL,
|
||||
PRIMARY KEY (app_code, stat_tz, stat_day, device_id),
|
||||
KEY idx_social_identity_day_user (app_code, stat_tz, stat_day, resolved_user_id)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||||
`CREATE TABLE IF NOT EXISTS stat_recharge_day_payers (
|
||||
app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0,
|
||||
region_id BIGINT NOT NULL DEFAULT 0,
|
||||
@ -381,6 +411,10 @@ func (r *Repository) Migrate(ctx context.Context) error {
|
||||
}
|
||||
alterStatements := []string{
|
||||
`ALTER TABLE app_tracking_events ADD COLUMN identity_key VARCHAR(160) GENERATED ALWAYS AS (CASE WHEN user_id > 0 THEN CONCAT('u:', user_id) ELSE CONCAT('d:', device_id) END) VIRTUAL AFTER device_id`,
|
||||
// INSTANT only updates metadata for these defaulted columns. MySQL 8.4
|
||||
// rejects an explicit LOCK=NONE with INSTANT, so retain the default lock.
|
||||
`ALTER TABLE app_tracking_events ADD COLUMN client_session_id VARCHAR(160) NOT NULL DEFAULT '', ALGORITHM=INSTANT`,
|
||||
`ALTER TABLE app_tracking_events ADD COLUMN auth_session_id VARCHAR(160) NOT NULL DEFAULT '', ALGORITHM=INSTANT`,
|
||||
`ALTER TABLE stat_app_day_country ADD COLUMN paid_users BIGINT NOT NULL DEFAULT 0 AFTER active_users`,
|
||||
`ALTER TABLE stat_app_day_country ADD COLUMN game_refund BIGINT NOT NULL DEFAULT 0 AFTER game_payout`,
|
||||
`ALTER TABLE stat_app_day_country ADD COLUMN lucky_gift_payout BIGINT NOT NULL DEFAULT 0 AFTER lucky_gift_turnover`,
|
||||
@ -775,6 +809,8 @@ type AppTrackingEvent struct {
|
||||
UserID int64
|
||||
DeviceID string
|
||||
SessionID string
|
||||
ClientSessionID string
|
||||
AuthSessionID string
|
||||
Platform string
|
||||
AppVersion string
|
||||
Language string
|
||||
@ -1664,6 +1700,28 @@ func (r *Repository) ConsumeSelfGameH5Event(ctx context.Context, event SelfGameH
|
||||
}
|
||||
|
||||
func (r *Repository) ConsumeAppTrackingEvents(ctx context.Context, events []AppTrackingEvent) (AppTrackingReportResult, error) {
|
||||
var lastResult AppTrackingReportResult
|
||||
var lastErr error
|
||||
for attempt := 0; attempt < 5; attempt++ {
|
||||
lastResult, lastErr = r.consumeAppTrackingEventsOnce(ctx, events)
|
||||
if lastErr == nil || !isRetryableMySQLError(lastErr) {
|
||||
return lastResult, lastErr
|
||||
}
|
||||
// Identity rows add one transactional lock per device. Concurrent batches
|
||||
// can still deadlock with unrelated aggregate rows, so retry the entire
|
||||
// idempotent batch just like owner-fact consumers do.
|
||||
timer := time.NewTimer(time.Duration(attempt+1) * 50 * time.Millisecond)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
timer.Stop()
|
||||
return AppTrackingReportResult{}, ctx.Err()
|
||||
case <-timer.C:
|
||||
}
|
||||
}
|
||||
return lastResult, lastErr
|
||||
}
|
||||
|
||||
func (r *Repository) consumeAppTrackingEventsOnce(ctx context.Context, events []AppTrackingEvent) (AppTrackingReportResult, error) {
|
||||
if r == nil || r.db == nil {
|
||||
return AppTrackingReportResult{}, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||||
}
|
||||
@ -1672,12 +1730,23 @@ func (r *Repository) ConsumeAppTrackingEvents(ctx context.Context, events []AppT
|
||||
if !apptracking.ValidBatchSize(len(events)) {
|
||||
return AppTrackingReportResult{}, xerr.New(xerr.InvalidArgument, "app tracking events batch is invalid")
|
||||
}
|
||||
// Device locks live until this batch commits. A stable device/event order
|
||||
// makes two overlapping batches acquire them consistently and removes the
|
||||
// common A->B / B->A deadlock without changing the idempotent response.
|
||||
orderedEvents := append([]AppTrackingEvent(nil), events...)
|
||||
sort.SliceStable(orderedEvents, func(i, j int) bool {
|
||||
leftDevice, rightDevice := strings.TrimSpace(orderedEvents[i].DeviceID), strings.TrimSpace(orderedEvents[j].DeviceID)
|
||||
if leftDevice != rightDevice {
|
||||
return leftDevice < rightDevice
|
||||
}
|
||||
return strings.TrimSpace(orderedEvents[i].EventID) < strings.TrimSpace(orderedEvents[j].EventID)
|
||||
})
|
||||
tx, err := r.db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return AppTrackingReportResult{}, err
|
||||
}
|
||||
defer func() { _ = tx.Rollback() }()
|
||||
for _, event := range events {
|
||||
for _, event := range orderedEvents {
|
||||
normalized, propertiesJSON, err := normalizeAppTrackingEvent(ctx, event, nowMS)
|
||||
if err != nil {
|
||||
return AppTrackingReportResult{}, err
|
||||
@ -1685,12 +1754,13 @@ func (r *Repository) ConsumeAppTrackingEvents(ctx context.Context, events []AppT
|
||||
affected, err := insertUnique(ctx, tx, `
|
||||
INSERT IGNORE INTO app_tracking_events (
|
||||
app_code, event_id, event_name, event_type, screen, target_type, target_id,
|
||||
user_id, device_id, session_id, platform, app_version, language, timezone,
|
||||
user_id, device_id, session_id, client_session_id, auth_session_id, platform, app_version, language, timezone,
|
||||
country_id, region_id, duration_ms, success, error_code, properties_json,
|
||||
occurred_at_ms, received_at_ms, stat_day
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CAST(? AS JSON), ?, ?, ?)
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CAST(? AS JSON), ?, ?, ?)
|
||||
`, normalized.AppCode, normalized.EventID, normalized.EventName, normalized.EventType, normalized.Screen,
|
||||
normalized.TargetType, normalized.TargetID, normalized.UserID, normalized.DeviceID, normalized.SessionID,
|
||||
normalized.ClientSessionID, normalized.AuthSessionID,
|
||||
normalized.Platform, normalized.AppVersion, normalized.Language, normalized.Timezone, normalized.CountryID,
|
||||
normalized.RegionID, normalized.DurationMS, normalized.Success, normalized.ErrorCode, propertiesJSON,
|
||||
normalized.OccurredAtMS, nowMS, statDay(normalized.OccurredAtMS))
|
||||
@ -1699,6 +1769,11 @@ func (r *Repository) ConsumeAppTrackingEvents(ctx context.Context, events []AppT
|
||||
}
|
||||
if affected > 0 {
|
||||
result.Stored++
|
||||
// 身份观察和原始埋点、旧宽表必须在同一事务提交；否则页面可能短暂看到
|
||||
// 新 d:/u: 行却仍使用旧解析关系，造成 DAU 与留存随请求时点抖动。
|
||||
if err := applySocialTrackingIdentity(ctx, tx, normalized, nowMS); err != nil {
|
||||
return AppTrackingReportResult{}, err
|
||||
}
|
||||
for _, scope := range statDayScopesFromContext(ctx, normalized.OccurredAtMS) {
|
||||
delta, ok := socialDeltaForAppTrackingEvent(normalized, scope, nowMS)
|
||||
if !ok {
|
||||
@ -2370,6 +2445,14 @@ func normalizeAppTrackingEvent(ctx context.Context, event AppTrackingEvent, rece
|
||||
if err != nil {
|
||||
return AppTrackingEvent{}, "", err
|
||||
}
|
||||
event.ClientSessionID, err = normalizeAppTrackingText(event.ClientSessionID, apptracking.MaxSessionIDBytes)
|
||||
if err != nil {
|
||||
return AppTrackingEvent{}, "", err
|
||||
}
|
||||
event.AuthSessionID, err = normalizeAppTrackingText(event.AuthSessionID, apptracking.MaxSessionIDBytes)
|
||||
if err != nil {
|
||||
return AppTrackingEvent{}, "", err
|
||||
}
|
||||
event.Platform, err = normalizeAppTrackingText(strings.ToLower(event.Platform), apptracking.MaxPlatformBytes)
|
||||
if err != nil {
|
||||
return AppTrackingEvent{}, "", err
|
||||
|
||||
@ -0,0 +1,230 @@
|
||||
package mysql
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"strings"
|
||||
|
||||
"hyapp/pkg/appcode"
|
||||
)
|
||||
|
||||
const missingClientSessionID = "__missing_client_session__"
|
||||
|
||||
type socialIdentityDayKey struct {
|
||||
StatTZ string
|
||||
StatDay string
|
||||
}
|
||||
|
||||
// applySocialTrackingIdentity records only identity evidence. It deliberately
|
||||
// never moves or rewrites stat_social_user_day rows: that table contains additive
|
||||
// money and duration fields which cannot be safely undone when events arrive out
|
||||
// of order. Query-time canonicalization consumes the small day-resolution table
|
||||
// and leaves the financial read model untouched.
|
||||
func applySocialTrackingIdentity(ctx context.Context, tx *sql.Tx, event AppTrackingEvent, nowMS int64) error {
|
||||
app := appcode.Normalize(event.AppCode)
|
||||
deviceID := strings.TrimSpace(event.DeviceID)
|
||||
clientSessionID := strings.TrimSpace(event.ClientSessionID)
|
||||
if app == "" || deviceID == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
if event.UserID <= 0 {
|
||||
// Only events that actually create a d: row may participate in the
|
||||
// device-day resolution. page_open and failed actions otherwise introduce
|
||||
// phantom unresolved sessions that permanently block a valid login merge.
|
||||
contributesSocialRow := false
|
||||
for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) {
|
||||
if _, ok := socialDeltaForAppTrackingEvent(event, scope, nowMS); ok {
|
||||
contributesSocialRow = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !contributesSocialRow {
|
||||
return nil
|
||||
}
|
||||
} else if clientSessionID == "" {
|
||||
// Old authenticated clients still produce a direct u: fact, but without
|
||||
// the process session they cannot link any earlier anonymous row. Return
|
||||
// before taking a device lock on this high-volume compatibility path.
|
||||
return nil
|
||||
}
|
||||
if err := lockSocialIdentityDevice(ctx, tx, app, deviceID, nowMS); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if event.UserID <= 0 {
|
||||
// Old clients do not send client_session_id. A single unresolved sentinel
|
||||
// per device/day blocks a different resolved session from swallowing those
|
||||
// anonymous events; conservative under-count correction is safer than a
|
||||
// false account merge on shared devices.
|
||||
if clientSessionID == "" {
|
||||
clientSessionID = missingClientSessionID
|
||||
}
|
||||
for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) {
|
||||
affected, err := insertUnique(ctx, tx, `
|
||||
INSERT IGNORE INTO stat_social_identity_session_day (
|
||||
app_code, stat_tz, stat_day, device_id, client_session_id, first_seen_at_ms, updated_at_ms
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?)
|
||||
`, app, scope.tz, scope.day, deviceID, clientSessionID, event.OccurredAtMS, nowMS)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if affected > 0 {
|
||||
if err := recomputeSocialIdentityDay(ctx, tx, app, scope.tz, scope.day, deviceID, nowMS); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
inserted, err := insertUnique(ctx, tx, `
|
||||
INSERT IGNORE INTO stat_social_identity_session_user (
|
||||
app_code, device_id, client_session_id, user_id, first_seen_at_ms, updated_at_ms
|
||||
) VALUES (?, ?, ?, ?, ?, ?)
|
||||
`, app, deviceID, clientSessionID, normalizeID(event.UserID), event.OccurredAtMS, nowMS)
|
||||
if err != nil || inserted == 0 {
|
||||
return err
|
||||
}
|
||||
|
||||
// A session can cross a Beijing/UTC midnight. When its first authenticated
|
||||
// event arrives, recompute every earlier anonymous day observed for that same
|
||||
// device+client session so D1/D3/D7 retention sees one canonical account.
|
||||
rows, err := tx.QueryContext(ctx, `
|
||||
SELECT stat_tz, DATE_FORMAT(stat_day, '%Y-%m-%d')
|
||||
FROM stat_social_identity_session_day
|
||||
WHERE app_code = ? AND device_id = ? AND client_session_id = ?
|
||||
ORDER BY stat_tz, stat_day
|
||||
FOR UPDATE
|
||||
`, app, deviceID, clientSessionID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
days := make([]socialIdentityDayKey, 0, 2)
|
||||
for rows.Next() {
|
||||
var key socialIdentityDayKey
|
||||
if err := rows.Scan(&key.StatTZ, &key.StatDay); err != nil {
|
||||
_ = rows.Close()
|
||||
return err
|
||||
}
|
||||
days = append(days, key)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
_ = rows.Close()
|
||||
return err
|
||||
}
|
||||
if err := rows.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, day := range days {
|
||||
if err := recomputeSocialIdentityDay(ctx, tx, app, day.StatTZ, day.StatDay, deviceID, nowMS); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func recomputeSocialIdentityDay(ctx context.Context, tx *sql.Tx, app, statTZ, statDay, deviceID string, nowMS int64) error {
|
||||
rows, err := tx.QueryContext(ctx, `
|
||||
SELECT d.client_session_id, u.user_id
|
||||
FROM stat_social_identity_session_day d
|
||||
LEFT JOIN stat_social_identity_session_user u
|
||||
ON u.app_code = d.app_code
|
||||
AND u.device_id = d.device_id
|
||||
AND u.client_session_id = d.client_session_id
|
||||
WHERE d.app_code = ? AND d.stat_tz = ? AND d.stat_day = ? AND d.device_id = ?
|
||||
ORDER BY d.client_session_id, u.user_id
|
||||
FOR UPDATE
|
||||
`, app, normalizeStatTZ(statTZ), statDay, deviceID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
sessionUsers := map[string]map[int64]struct{}{}
|
||||
for rows.Next() {
|
||||
var sessionID string
|
||||
var userID sql.NullInt64
|
||||
if err := rows.Scan(&sessionID, &userID); err != nil {
|
||||
_ = rows.Close()
|
||||
return err
|
||||
}
|
||||
if sessionUsers[sessionID] == nil {
|
||||
sessionUsers[sessionID] = map[int64]struct{}{}
|
||||
}
|
||||
if userID.Valid && userID.Int64 > 0 {
|
||||
sessionUsers[sessionID][userID.Int64] = struct{}{}
|
||||
}
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
_ = rows.Close()
|
||||
return err
|
||||
}
|
||||
if err := rows.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
state, resolvedUserID := "unresolved", int64(0)
|
||||
resolutionSource := "client_session"
|
||||
ambiguous := false
|
||||
unresolved := false
|
||||
for _, users := range sessionUsers {
|
||||
if len(users) == 0 {
|
||||
unresolved = true
|
||||
continue
|
||||
}
|
||||
if len(users) > 1 {
|
||||
ambiguous = true
|
||||
continue
|
||||
}
|
||||
for userID := range users {
|
||||
if resolvedUserID == 0 {
|
||||
resolvedUserID = userID
|
||||
} else if resolvedUserID != userID {
|
||||
ambiguous = true
|
||||
}
|
||||
}
|
||||
}
|
||||
if ambiguous {
|
||||
state, resolvedUserID = "ambiguous", 0
|
||||
} else if !unresolved && len(sessionUsers) > 0 && resolvedUserID > 0 {
|
||||
state = "resolved"
|
||||
} else {
|
||||
resolvedUserID = 0
|
||||
}
|
||||
if len(sessionUsers) == 1 {
|
||||
if _, onlyMissingSession := sessionUsers[missingClientSessionID]; onlyMissingSession {
|
||||
// A legacy client cannot establish an exact session bridge. Keep that
|
||||
// evidence conservative, but let a later bounded historical replay
|
||||
// replace it instead of giving it permanent client-session authority.
|
||||
resolutionSource = "missing_client_session"
|
||||
}
|
||||
}
|
||||
|
||||
_, err = tx.ExecContext(ctx, `
|
||||
INSERT INTO stat_social_identity_day (
|
||||
app_code, stat_tz, stat_day, device_id, resolved_user_id,
|
||||
resolution_state, resolution_source, updated_at_ms
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||||
ON DUPLICATE KEY UPDATE
|
||||
resolved_user_id = VALUES(resolved_user_id),
|
||||
resolution_state = VALUES(resolution_state),
|
||||
resolution_source = VALUES(resolution_source),
|
||||
updated_at_ms = VALUES(updated_at_ms)
|
||||
`, app, normalizeStatTZ(statTZ), statDay, deviceID, resolvedUserID, state, resolutionSource, nowMS)
|
||||
return err
|
||||
}
|
||||
|
||||
func lockSocialIdentityDevice(ctx context.Context, tx *sql.Tx, app, deviceID string, nowMS int64) error {
|
||||
if _, err := tx.ExecContext(ctx, `
|
||||
INSERT IGNORE INTO stat_social_identity_device_lock (app_code, device_id, created_at_ms)
|
||||
VALUES (?, ?, ?)
|
||||
`, appcode.Normalize(app), strings.TrimSpace(deviceID), nowMS); err != nil {
|
||||
return err
|
||||
}
|
||||
var lockedDeviceID string
|
||||
return tx.QueryRowContext(ctx, `
|
||||
SELECT device_id
|
||||
FROM stat_social_identity_device_lock
|
||||
WHERE app_code = ? AND device_id = ?
|
||||
FOR UPDATE
|
||||
`, appcode.Normalize(app), strings.TrimSpace(deviceID)).Scan(&lockedDeviceID)
|
||||
}
|
||||
@ -101,6 +101,9 @@ type socialRequirementUserDay struct {
|
||||
StatDay string
|
||||
Subject string
|
||||
UserID int64
|
||||
DirectUser bool
|
||||
DeviceID string
|
||||
CountryID int64
|
||||
RegionID int64
|
||||
UserRole string
|
||||
AppFirstOpen int64
|
||||
@ -235,13 +238,19 @@ func (r *Repository) QuerySocialRequirements(ctx context.Context, query SocialRe
|
||||
startDay := statDayIn(startMS, statTZ)
|
||||
endDay := statDayIn(endMS-1, statTZ)
|
||||
days := socialDayRange(startDay, endDay)
|
||||
// 留存指标需要读取 cohort 日之后的 D30 行；这些仍是本服务已沉淀的用户日宽表，不回查 owner service。
|
||||
rows, err := r.querySocialUserDays(ctx, app, statTZ, startDay, addStatDays(endDay, 30), query.CountryID, query.RegionID, query.RegionIDs)
|
||||
sections := socialRequestedSections(query.Section)
|
||||
lookaheadDays := socialRetentionLookaheadDays(sections)
|
||||
// 只有 P0 次留或 P2 留存需要读取查询结束日之后的观察行；单独查看活跃、
|
||||
// 营收、送礼、游戏时不再固定多扫 30 天的 50 列用户日宽表。
|
||||
allRows, err := r.querySocialUserDays(ctx, app, statTZ, startDay, addStatDays(endDay, lookaheadDays), 0, 0, nil)
|
||||
if err != nil {
|
||||
return SocialRequirements{}, err
|
||||
}
|
||||
byDay, bySubjectDay := indexSocialRows(rows)
|
||||
sections := socialRequestedSections(query.Section)
|
||||
rows := filterSocialRequirementUserDays(allRows, query.CountryID, query.RegionID, query.RegionIDs)
|
||||
byDay, _ := indexSocialRows(rows)
|
||||
// 区域只筛 cohort 基准日；D+N 观察日保留完整 canonical 活跃索引。
|
||||
// 用户次日跨区仍算回来，不能因为目标日维度不同被误判为流失。
|
||||
_, bySubjectDay := indexSocialRows(allRows)
|
||||
var gameBreakdown []SocialRequirementGame
|
||||
for _, section := range sections {
|
||||
if section == SocialSectionGame {
|
||||
@ -269,7 +278,10 @@ func (r *Repository) QuerySocialRequirements(ctx context.Context, query SocialRe
|
||||
for _, day := range days {
|
||||
agg := aggregateSocialRows(byDay[day], section, query.UserRole, query.PayerType, query.NewUserType)
|
||||
totalAgg.add(agg)
|
||||
retention := socialRetentionMetrics(day, byDay[day], bySubjectDay, section, query.UserRole, query.PayerType, query.NewUserType)
|
||||
retention := map[string]retentionMetric{}
|
||||
if socialSectionNeedsRetention(section) {
|
||||
retention = socialRetentionMetrics(day, byDay[day], bySubjectDay, section, query.UserRole, query.PayerType, query.NewUserType)
|
||||
}
|
||||
mergeRetentionMetrics(totalRetention, retention)
|
||||
sectionRows = append(sectionRows, SocialRequirementRow{
|
||||
StatDay: day,
|
||||
@ -326,26 +338,31 @@ func socialAverageRow(rows []SocialRequirementRow) SocialRequirementRow {
|
||||
return SocialRequirementRow{StatDay: "average", Label: "平均值", Metrics: metrics}
|
||||
}
|
||||
|
||||
func socialRetentionLookaheadDays(sections []string) int {
|
||||
lookahead := 0
|
||||
for _, section := range sections {
|
||||
switch section {
|
||||
case SocialSectionRetention:
|
||||
return 30
|
||||
case SocialSectionNewUsers:
|
||||
lookahead = 1
|
||||
}
|
||||
}
|
||||
return lookahead
|
||||
}
|
||||
|
||||
func socialSectionNeedsRetention(section string) bool {
|
||||
return section == SocialSectionRetention || section == SocialSectionNewUsers
|
||||
}
|
||||
|
||||
func (r *Repository) querySocialUserDays(ctx context.Context, app string, statTZ string, startDay string, endDay string, countryID int64, regionID int64, regionIDs []int64) ([]socialRequirementUserDay, error) {
|
||||
args := []any{appcode.Normalize(app), normalizeStatTZ(statTZ), startDay, endDay}
|
||||
conditions := []string{"app_code = ?", "stat_tz = ?", "stat_day BETWEEN ? AND ?"}
|
||||
if countryID > 0 {
|
||||
conditions = append(conditions, "country_id = ?")
|
||||
args = append(args, countryID)
|
||||
}
|
||||
if regionID > 0 {
|
||||
conditions = append(conditions, "region_id = ?")
|
||||
args = append(args, regionID)
|
||||
} else if ids := normalizePositiveIDs(regionIDs); len(ids) > 0 {
|
||||
placeholders := make([]string, 0, len(ids))
|
||||
for _, id := range ids {
|
||||
placeholders = append(placeholders, "?")
|
||||
args = append(args, id)
|
||||
}
|
||||
conditions = append(conditions, "region_id IN ("+strings.Join(placeholders, ",")+")")
|
||||
}
|
||||
// 匿名行通常没有国家/区域，而登录后的 u: 行才带用户维度。必须先在完整
|
||||
// app+day 小宽表范围内完成 canonical 合并，再应用维度过滤；提前在 SQL 里
|
||||
// 过滤会丢掉同一用户登录前的活跃，并让区域 DAU 再次出现 d/u 口径偏差。
|
||||
rows, err := r.db.QueryContext(ctx, `
|
||||
SELECT DATE_FORMAT(stat_day, '%Y-%m-%d'), subject_key, user_id, region_id, user_role,
|
||||
SELECT DATE_FORMAT(stat_day, '%Y-%m-%d'), subject_key, user_id, device_id, country_id, region_id, user_role,
|
||||
app_first_open, registered_success, active_user, room_join_success, room_join_fail,
|
||||
mic_up_success, gift_sent, normal_gift_sent, lucky_gift_sent, super_lucky_gift_sent,
|
||||
recharge_user, first_recharge_user, repeat_recharge_user, game_bet_user, self_game_bet_user,
|
||||
@ -367,7 +384,7 @@ func (r *Repository) querySocialUserDays(ctx context.Context, app string, statTZ
|
||||
out := []socialRequirementUserDay{}
|
||||
for rows.Next() {
|
||||
var item socialRequirementUserDay
|
||||
if err := rows.Scan(&item.StatDay, &item.Subject, &item.UserID, &item.RegionID, &item.UserRole,
|
||||
if err := rows.Scan(&item.StatDay, &item.Subject, &item.UserID, &item.DeviceID, &item.CountryID, &item.RegionID, &item.UserRole,
|
||||
&item.AppFirstOpen, &item.Registered, &item.Active, &item.RoomJoinOK, &item.RoomJoinNG,
|
||||
&item.MicUp, &item.Gift, &item.NormalGift, &item.LuckyGift, &item.SuperGift,
|
||||
&item.Recharge, &item.FirstRecharge, &item.RepeatRecharge, &item.GameBet, &item.SelfGameBet,
|
||||
@ -380,19 +397,210 @@ func (r *Repository) querySocialUserDays(ctx context.Context, app string, statTZ
|
||||
&item.OtherConsumeCoin, &item.OutputCoin); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// 在房时长口径：客户端 room_stay 埋点（杀进程/闪退整段丢失）和服务端进退房事实 room_online_ms
|
||||
// 双轨并存，取 GREATEST 避免切换期数据断层；埋点按 app 版本灰度下线后该口径自然收敛到服务端事实。
|
||||
if item.RoomStayMS < item.RoomOnlineMS {
|
||||
item.RoomStayMS = item.RoomOnlineMS
|
||||
}
|
||||
// 在房时长下界修正：上麦必在房内，麦时长（服务端事实，权威）是房时长的天然下界。
|
||||
// 服务端在房事实上线后保留该钳制作为兜底（如回灌历史区间尚无服务端事实）。
|
||||
if item.RoomStayMS < item.MicStayMS {
|
||||
item.RoomStayMS = item.MicStayMS
|
||||
}
|
||||
item.DirectUser = item.UserID > 0
|
||||
out = append(out, item)
|
||||
}
|
||||
return out, rows.Err()
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
identityByDeviceDay, err := r.querySocialIdentityDays(ctx, app, statTZ, startDay, endDay)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return canonicalizeSocialUserDays(out, identityByDeviceDay, countryID, regionID, regionIDs), nil
|
||||
}
|
||||
|
||||
type socialIdentityDeviceDayKey struct {
|
||||
StatDay string
|
||||
DeviceID string
|
||||
}
|
||||
|
||||
func (r *Repository) querySocialIdentityDays(ctx context.Context, app, statTZ, startDay, endDay string) (map[socialIdentityDeviceDayKey]int64, error) {
|
||||
rows, err := r.db.QueryContext(ctx, `
|
||||
SELECT DATE_FORMAT(stat_day, '%Y-%m-%d'), device_id, resolved_user_id
|
||||
FROM stat_social_identity_day
|
||||
WHERE app_code = ? AND stat_tz = ? AND stat_day BETWEEN ? AND ?
|
||||
AND resolution_state = 'resolved' AND resolved_user_id > 0
|
||||
`, appcode.Normalize(app), normalizeStatTZ(statTZ), startDay, endDay)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
resolved := map[socialIdentityDeviceDayKey]int64{}
|
||||
for rows.Next() {
|
||||
var day, deviceID string
|
||||
var userID int64
|
||||
if err := rows.Scan(&day, &deviceID, &userID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if deviceID = strings.TrimSpace(deviceID); deviceID != "" && userID > 0 {
|
||||
resolved[socialIdentityDeviceDayKey{StatDay: day, DeviceID: deviceID}] = userID
|
||||
}
|
||||
}
|
||||
return resolved, rows.Err()
|
||||
}
|
||||
|
||||
// canonicalizeSocialUserDays keeps stat_social_user_day as the immutable source
|
||||
// of financial/duration facts and only rewrites identity in memory. This avoids
|
||||
// destructive primary-key moves and keeps a replay idempotent even when late
|
||||
// events or an account switch changes the device-day resolution.
|
||||
func canonicalizeSocialUserDays(rows []socialRequirementUserDay, resolved map[socialIdentityDeviceDayKey]int64, countryID, regionID int64, regionIDs []int64) []socialRequirementUserDay {
|
||||
merged := map[string]socialRequirementUserDay{}
|
||||
firstOpenDevices := map[string]map[string]struct{}{}
|
||||
for _, row := range rows {
|
||||
if row.UserID <= 0 {
|
||||
if userID := resolved[socialIdentityDeviceDayKey{StatDay: row.StatDay, DeviceID: strings.TrimSpace(row.DeviceID)}]; userID > 0 {
|
||||
row.UserID = userID
|
||||
row.Subject = "u:" + strconv.FormatInt(userID, 10)
|
||||
}
|
||||
}
|
||||
key := row.StatDay + "\x00" + row.Subject
|
||||
if row.AppFirstOpen > 0 {
|
||||
deviceID := strings.TrimSpace(row.DeviceID)
|
||||
if deviceID == "" {
|
||||
// Old u: rows may not retain a device. Use the physical subject as a
|
||||
// stable one-count fallback; known d/u rows still dedupe by device ID.
|
||||
deviceID = "subject:" + row.Subject
|
||||
}
|
||||
if firstOpenDevices[key] == nil {
|
||||
firstOpenDevices[key] = map[string]struct{}{}
|
||||
}
|
||||
firstOpenDevices[key][deviceID] = struct{}{}
|
||||
}
|
||||
if current, ok := merged[key]; ok {
|
||||
merged[key] = mergeSocialRequirementUserDay(current, row)
|
||||
} else {
|
||||
merged[key] = row
|
||||
}
|
||||
}
|
||||
out := make([]socialRequirementUserDay, 0, len(merged))
|
||||
for key, row := range merged {
|
||||
row.AppFirstOpen = int64(len(firstOpenDevices[key]))
|
||||
// Client room_stay and authoritative server room_online are parallel
|
||||
// measurements, not additive copies. Clamp only after d:/u: fields have
|
||||
// been summed so the same interval is not counted twice.
|
||||
if row.RoomStayMS < row.RoomOnlineMS {
|
||||
row.RoomStayMS = row.RoomOnlineMS
|
||||
}
|
||||
if row.RoomStayMS < row.MicStayMS {
|
||||
row.RoomStayMS = row.MicStayMS
|
||||
}
|
||||
out = append(out, row)
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool {
|
||||
if out[i].StatDay != out[j].StatDay {
|
||||
return out[i].StatDay < out[j].StatDay
|
||||
}
|
||||
return out[i].Subject < out[j].Subject
|
||||
})
|
||||
return filterSocialRequirementUserDays(out, countryID, regionID, regionIDs)
|
||||
}
|
||||
|
||||
func mergeSocialRequirementUserDay(current, next socialRequirementUserDay) socialRequirementUserDay {
|
||||
if current.UserID == 0 {
|
||||
current.UserID = next.UserID
|
||||
}
|
||||
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.
|
||||
if next.DirectUser && !current.DirectUser {
|
||||
current.CountryID, current.RegionID, current.UserRole = next.CountryID, next.RegionID, next.UserRole
|
||||
current.DirectUser = true
|
||||
} else if next.DirectUser == current.DirectUser {
|
||||
if current.CountryID == 0 && current.RegionID == 0 && (next.CountryID > 0 || next.RegionID > 0) {
|
||||
current.CountryID, current.RegionID = next.CountryID, next.RegionID
|
||||
}
|
||||
if normalizeSocialUserRole(next.UserRole) == "host" {
|
||||
current.UserRole = next.UserRole
|
||||
}
|
||||
}
|
||||
// canonicalizeSocialUserDays replaces this flag with the distinct known
|
||||
// device count after all rows merge; MAX is only the empty-device fallback.
|
||||
current.AppFirstOpen = maxSocialFlag(current.AppFirstOpen, next.AppFirstOpen)
|
||||
current.Registered = maxSocialFlag(current.Registered, next.Registered)
|
||||
current.Active = maxSocialFlag(current.Active, next.Active)
|
||||
current.RoomJoinOK = maxSocialFlag(current.RoomJoinOK, next.RoomJoinOK)
|
||||
current.RoomJoinNG = maxSocialFlag(current.RoomJoinNG, next.RoomJoinNG)
|
||||
current.MicUp = maxSocialFlag(current.MicUp, next.MicUp)
|
||||
current.Gift = maxSocialFlag(current.Gift, next.Gift)
|
||||
current.NormalGift = maxSocialFlag(current.NormalGift, next.NormalGift)
|
||||
current.LuckyGift = maxSocialFlag(current.LuckyGift, next.LuckyGift)
|
||||
current.SuperGift = maxSocialFlag(current.SuperGift, next.SuperGift)
|
||||
current.Recharge = maxSocialFlag(current.Recharge, next.Recharge)
|
||||
current.FirstRecharge = maxSocialFlag(current.FirstRecharge, next.FirstRecharge)
|
||||
current.RepeatRecharge = maxSocialFlag(current.RepeatRecharge, next.RepeatRecharge)
|
||||
current.GameBet = maxSocialFlag(current.GameBet, next.GameBet)
|
||||
current.SelfGameBet = maxSocialFlag(current.SelfGameBet, next.SelfGameBet)
|
||||
current.ProbabilityBet = maxSocialFlag(current.ProbabilityBet, next.ProbabilityBet)
|
||||
current.Chat = maxSocialFlag(current.Chat, next.Chat)
|
||||
current.FollowUser = maxSocialFlag(current.FollowUser, next.FollowUser)
|
||||
current.FollowRoom = maxSocialFlag(current.FollowRoom, next.FollowRoom)
|
||||
current.GiftPanel = maxSocialFlag(current.GiftPanel, next.GiftPanel)
|
||||
current.ProbabilityPanel = maxSocialFlag(current.ProbabilityPanel, next.ProbabilityPanel)
|
||||
current.AppStayMS += next.AppStayMS
|
||||
current.RoomStayMS += next.RoomStayMS
|
||||
current.RoomOnlineMS += next.RoomOnlineMS
|
||||
current.MicStayMS += next.MicStayMS
|
||||
current.RechargeCount += next.RechargeCount
|
||||
current.RechargeUSDMinor += next.RechargeUSDMinor
|
||||
current.GoogleRechargeUSDMinor += next.GoogleRechargeUSDMinor
|
||||
current.ThirdPartyRechargeUSDMinor += next.ThirdPartyRechargeUSDMinor
|
||||
current.CoinSellerRechargeUSDMinor += next.CoinSellerRechargeUSDMinor
|
||||
current.FirstRechargeUSDMinor += next.FirstRechargeUSDMinor
|
||||
current.RepeatRechargeUSDMinor += next.RepeatRechargeUSDMinor
|
||||
current.NormalGiftCoin += next.NormalGiftCoin
|
||||
current.LuckyGiftCoin += next.LuckyGiftCoin
|
||||
current.SuperGiftCoin += next.SuperGiftCoin
|
||||
current.LuckyGiftPayoutCoin += next.LuckyGiftPayoutCoin
|
||||
current.SuperGiftPayoutCoin += next.SuperGiftPayoutCoin
|
||||
current.GameBetCoin += next.GameBetCoin
|
||||
current.GameBetCount += next.GameBetCount
|
||||
current.SelfGameBetCoin += next.SelfGameBetCoin
|
||||
current.SelfGameBetCount += next.SelfGameBetCount
|
||||
current.ProbabilityBetCoin += next.ProbabilityBetCoin
|
||||
current.ProbabilityBetCount += next.ProbabilityBetCount
|
||||
current.OtherConsumeCoin += next.OtherConsumeCoin
|
||||
current.OutputCoin += next.OutputCoin
|
||||
return current
|
||||
}
|
||||
|
||||
func maxSocialFlag(left, right int64) int64 {
|
||||
if right > left {
|
||||
return right
|
||||
}
|
||||
return left
|
||||
}
|
||||
|
||||
func filterSocialRequirementUserDays(rows []socialRequirementUserDay, countryID, regionID int64, regionIDs []int64) []socialRequirementUserDay {
|
||||
allowedRegions := map[int64]struct{}{}
|
||||
if regionID <= 0 {
|
||||
for _, id := range normalizePositiveIDs(regionIDs) {
|
||||
allowedRegions[id] = struct{}{}
|
||||
}
|
||||
}
|
||||
if countryID <= 0 && regionID <= 0 && len(allowedRegions) == 0 {
|
||||
return rows
|
||||
}
|
||||
out := make([]socialRequirementUserDay, 0, len(rows))
|
||||
for _, row := range rows {
|
||||
if countryID > 0 && row.CountryID != countryID {
|
||||
continue
|
||||
}
|
||||
if regionID > 0 && row.RegionID != regionID {
|
||||
continue
|
||||
}
|
||||
if len(allowedRegions) > 0 {
|
||||
if _, ok := allowedRegions[row.RegionID]; !ok {
|
||||
continue
|
||||
}
|
||||
}
|
||||
out = append(out, row)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func indexSocialRows(rows []socialRequirementUserDay) (map[string][]socialRequirementUserDay, map[string]map[string]socialRequirementUserDay) {
|
||||
@ -414,7 +622,9 @@ func aggregateSocialRows(rows []socialRequirementUserDay, section string, role s
|
||||
if !socialRowMatchesSection(row, section, role, payerType, newUserType) {
|
||||
continue
|
||||
}
|
||||
agg.AppFirstOpen += flagValue(row.AppFirstOpen)
|
||||
// 下载/首次打开是设备口径；身份归并只消除用户行为的 d/u 重复，
|
||||
// 不能把同一账号在两台设备上的两次安装压成 1。
|
||||
agg.AppFirstOpen += normalizeID(row.AppFirstOpen)
|
||||
agg.Registered += flagValue(row.Registered)
|
||||
if flagValue(row.Registered) > 0 && normalizeSocialUserRole(row.UserRole) == "host" {
|
||||
agg.NewHostUsers++
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user