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) }