hyapp-server/services/user-service/internal/storage/mysql/auth/sessions.go

package auth

import (
	"context"
	"crypto/subtle"
	"database/sql"
	"strings"

	"hyapp/pkg/appcode"
	"hyapp/pkg/xerr"
	authdomain "hyapp/services/user-service/internal/domain/auth"
)

const sessionProjection = `app_code, session_id, user_id, refresh_token_hash, token_family_id, generation, parent_session_id, rotated_to_session_id, rotation_at_ms, rotation_request_id, device_id, expires_at_ms, last_heartbeat_at_ms, last_heartbeat_request_id, revoked_at_ms, revoked_reason, revoked_request_id, revoked_by, created_at_ms, updated_at_ms`

func (r *Repository) CreateSession(ctx context.Context, session authdomain.Session) error {
	// refresh token 只保存 hash，原文不会写入数据库。
	_, err := r.db.ExecContext(ctx, `
		INSERT INTO auth_sessions (app_code, session_id, user_id, refresh_token_hash, token_family_id, generation, parent_session_id, rotated_to_session_id, rotation_at_ms, rotation_request_id, device_id, expires_at_ms, last_heartbeat_at_ms, last_heartbeat_request_id, revoked_at_ms, revoked_reason, revoked_request_id, revoked_by, created_at_ms, updated_at_ms)
		VALUES (?, ?, ?, ?, ?, ?, ?, '', 0, '', ?, ?, ?, ?, NULL, '', '', '', ?, ?)
	`, appcode.Normalize(session.AppCode), session.SessionID, session.UserID, session.RefreshTokenHash, normalizedFamilyID(session), session.Generation, session.ParentSessionID, session.DeviceID, session.ExpiresAtMs, session.LastHeartbeatAtMs, session.LastHeartbeatRequestID, session.CreatedAtMs, session.UpdatedAtMs)
	if err != nil {
		// session_id 或 refresh hash 冲突统一映射成 auth 领域冲突。
		return mapAuthDuplicateError(err)
	}

	return nil
}

// FindSessionByRefreshHash 返回已轮换/吊销行在内的完整 session；调用方必须继续做状态判定。
func (r *Repository) FindSessionByRefreshHash(ctx context.Context, refreshTokenHash string) (authdomain.Session, error) {
	return r.findSessionByRefreshHash(ctx, refreshTokenHash)
}

// FindActiveSessionByRefreshHash 查找未过期且未吊销的 session。

func (r *Repository) FindActiveSessionByRefreshHash(ctx context.Context, refreshTokenHash string, nowMs int64) (authdomain.Session, error) {
	// 查到 session 后再判断 revoked/expired，给调用方稳定 reason。
	session, err := r.findSessionByRefreshHash(ctx, refreshTokenHash)
	if err != nil {
		return authdomain.Session{}, err
	}
	if session.RevokedAtMs > 0 {
		// 已吊销 session 不能继续 refresh。
		return authdomain.Session{}, xerr.New(xerr.SessionRevoked, "session revoked")
	}
	if session.ExpiresAtMs <= nowMs {
		// 过期 session 与吊销 session 使用不同 reason。
		return authdomain.Session{}, xerr.New(xerr.SessionExpired, "session expired")
	}

	return session, nil
}

// FindActiveSessionByID 查找未过期且未吊销的 session。

func (r *Repository) FindActiveSessionByID(ctx context.Context, sessionID string, nowMs int64) (authdomain.Session, error) {
	// onboarding 完成后只需要重签 access token，不能要求客户端再提交 refresh token 原文。
	session, err := r.findSessionByID(ctx, sessionID)
	if err != nil {
		return authdomain.Session{}, err
	}
	if session.RevokedAtMs > 0 {
		// 已吊销 session 不能继续签发任何 access token。
		return authdomain.Session{}, xerr.New(xerr.SessionRevoked, "session revoked")
	}
	if session.ExpiresAtMs <= nowMs {
		// session 过期后只能重新登录，不能用完成注册接口续命。
		return authdomain.Session{}, xerr.New(xerr.SessionExpired, "session expired")
	}

	return session, nil
}

// FindLatestActiveSessionByUser 返回用户当前最新的未过期未吊销 session，供后台按用户重签 access token。
// 与 FindActiveSessionByID 不同：这里没有可信 sid 输入，用“最近创建的 active session”近似客户端当前会话；
// refresh 轮换后旧 session 已吊销，最新 created_at_ms 即客户端实际持有的会话。

func (r *Repository) FindLatestActiveSessionByUser(ctx context.Context, userID int64, nowMs int64) (authdomain.Session, error) {
	// idx_auth_sessions_user_id(app_code, user_id) 先收敛到单用户，再在少量行上过滤排序即可。
	session, err := scanSession(r.db.QueryRowContext(ctx, `
		SELECT `+sessionProjection+`
		FROM auth_sessions
		WHERE app_code = ? AND user_id = ? AND revoked_at_ms IS NULL AND expires_at_ms > ?
		ORDER BY created_at_ms DESC, session_id DESC
		LIMIT 1
	`, appcode.FromContext(ctx), userID, nowMs))
	if err == sql.ErrNoRows {
		// “无活跃会话”是后台要明确展示的业务事实，用 NotFound 与其他存储错误区分。
		return authdomain.Session{}, xerr.New(xerr.NotFound, "active session not found")
	}
	if err != nil {
		return authdomain.Session{}, err
	}
	return session, nil
}

// RotateRefreshSession 在一个 MySQL 事务内串行化同一 refresh token 的首次轮换、grace 重试和 replay 撤销。
func (r *Repository) RotateRefreshSession(ctx context.Context, command authdomain.RefreshRotationCommand) (authdomain.RefreshRotationResult, error) {
	appCode := appcode.Normalize(command.AppCode)
	if appCode == "" {
		appCode = appcode.FromContext(ctx)
	}
	if command.RefreshTokenHash == "" || command.NowMs <= 0 || command.GracePeriodMs <= 0 || command.DenyUntilMs <= command.NowMs || strings.TrimSpace(command.RefreshRequestID) == "" {
		return authdomain.RefreshRotationResult{}, xerr.New(xerr.InvalidArgument, "invalid refresh rotation command")
	}

	tx, err := r.db.BeginTx(ctx, nil)
	if err != nil {
		return authdomain.RefreshRotationResult{}, err
	}
	defer tx.Rollback()

	// 唯一 refresh hash + FOR UPDATE 保证多实例并发只能有一个 winner；幂等密文和 lineage 与轮换同提交。
	source, err := scanSession(tx.QueryRowContext(ctx, `
		SELECT `+sessionProjection+`
		FROM auth_sessions
		WHERE app_code = ? AND refresh_token_hash = ?
		FOR UPDATE
	`, appCode, command.RefreshTokenHash))
	if err == sql.ErrNoRows {
		return authdomain.RefreshRotationResult{}, xerr.New(xerr.SessionRevoked, "session revoked")
	}
	if err != nil {
		return authdomain.RefreshRotationResult{}, err
	}

	if source.RevokedAtMs == 0 && source.ExpiresAtMs <= command.NowMs {
		// 自然过期的当前代不是 token reuse，保持 SESSION_EXPIRED 让客户端重新登录。
		return authdomain.RefreshRotationResult{}, xerr.New(xerr.SessionExpired, "session expired")
	}
	if source.RevokedAtMs > 0 && source.RevokedReason != "REFRESH_ROTATED" && source.RevokedReason != "REFRESH_TOKEN_REUSE" {
		// logout、风控和后台封禁都不允许进入 grace，也不能覆盖其原始撤销原因。
		return authdomain.RefreshRotationResult{}, xerr.New(xerr.SessionRevoked, "session revoked")
	}
	if source.RevokedReason == "REFRESH_ROTATED" && (strings.TrimSpace(source.RotatedToSessionID) == "" || source.RotationAtMs <= 0) {
		// 旧 binary 没有写 parent→child lineage/outcome，时间或设备启发式都可能把并发显式登录误认成 child，
		// 造成古老 token 永久踢掉未来新会话。无法证明 lineage 时只拒绝旧 token，不扩大任何撤销范围。
		return authdomain.RefreshRotationResult{}, xerr.New(xerr.SessionRevoked, "session revoked")
	}
	if subtle.ConstantTimeCompare([]byte(source.DeviceID), []byte(strings.TrimSpace(command.DeviceID))) != 1 {
		// token 原文在另一设备出现就是重放；family 查询仍同时限定 app/user/device，不能误伤其他设备登录态。
		result, reuseErr := r.revokeRefreshFamilyForReuse(ctx, tx, source, command)
		if reuseErr != nil {
			return authdomain.RefreshRotationResult{}, reuseErr
		}
		if err := tx.Commit(); err != nil {
			return authdomain.RefreshRotationResult{}, err
		}
		return result, nil
	}

	if source.RevokedAtMs == 0 {
		result, rotateErr := r.createRefreshRotation(ctx, tx, source, command)
		if rotateErr != nil {
			return authdomain.RefreshRotationResult{}, rotateErr
		}
		if err := tx.Commit(); err != nil {
			return authdomain.RefreshRotationResult{}, err
		}
		return result, nil
	}

	if source.RevokedReason == "REFRESH_TOKEN_REUSE" {
		// 上次 denylist 写失败时，后续请求仍返回整族 sid，让 service 可以继续补偿 gateway denylist。
		result, reuseErr := r.revokeRefreshFamilyForReuse(ctx, tx, source, command)
		if reuseErr != nil {
			return authdomain.RefreshRotationResult{}, reuseErr
		}
		if err := tx.Commit(); err != nil {
			return authdomain.RefreshRotationResult{}, err
		}
		return result, nil
	}

	child, childErr := scanSession(tx.QueryRowContext(ctx, `
		SELECT `+sessionProjection+`
		FROM auth_sessions
		WHERE app_code = ? AND session_id = ?
		FOR UPDATE
	`, appCode, source.RotatedToSessionID))
	if childErr != nil && childErr != sql.ErrNoRows {
		return authdomain.RefreshRotationResult{}, childErr
	}
	var outcomeRequestID string
	var outcomeChildID string
	var outcomeCiphertext []byte
	var outcomeCreatedAtMs int64
	var outcomeExpiresAtMs int64
	outcomeErr := tx.QueryRowContext(ctx, `
		SELECT child_session_id, refresh_request_id, token_ciphertext, created_at_ms, expires_at_ms
		FROM auth_refresh_outcomes
		WHERE app_code = ? AND parent_session_id = ?
		FOR UPDATE
	`, appCode, source.SessionID).Scan(&outcomeChildID, &outcomeRequestID, &outcomeCiphertext, &outcomeCreatedAtMs, &outcomeExpiresAtMs)
	if outcomeErr != nil && outcomeErr != sql.ErrNoRows {
		return authdomain.RefreshRotationResult{}, outcomeErr
	}

	graceValid := childErr == nil && outcomeErr == nil &&
		child.RevokedAtMs == 0 && child.ExpiresAtMs > command.NowMs &&
		child.ParentSessionID == source.SessionID && child.TokenFamilyID == normalizedFamilyID(source) && child.Generation == source.Generation+1 &&
		outcomeChildID == child.SessionID && outcomeCreatedAtMs == source.RotationAtMs &&
		command.NowMs <= outcomeExpiresAtMs
	if graceValid {
		// 窗口由首次 winner 与 token pair 同事务固化；滚动配置变更时 retry 实例必须信任该 persisted deadline，
		// 不能再用自身当前 grace 配置缩短或扩张，否则 30s/10s 混部会把合法重试误判为 reuse。
		if err := tx.Commit(); err != nil {
			return authdomain.RefreshRotationResult{}, err
		}
		return authdomain.RefreshRotationResult{
			Status:           authdomain.RefreshRotationGraceHit,
			SourceSession:    source,
			ActiveSession:    child,
			RefreshRequestID: outcomeRequestID,
			TokenCiphertext:  append([]byte(nil), outcomeCiphertext...),
		}, nil
	}

	// child 已再次轮换（第二上一代）、outcome 过窗或 lineage 不完整都属于不可安全恢复的 reuse。
	result, reuseErr := r.revokeRefreshFamilyForReuse(ctx, tx, source, command)
	if reuseErr != nil {
		return authdomain.RefreshRotationResult{}, reuseErr
	}
	if err := tx.Commit(); err != nil {
		return authdomain.RefreshRotationResult{}, err
	}
	return result, nil
}

func (r *Repository) createRefreshRotation(ctx context.Context, tx *sql.Tx, source authdomain.Session, command authdomain.RefreshRotationCommand) (authdomain.RefreshRotationResult, error) {
	child := command.NewSession
	familyID := normalizedFamilyID(source)
	if child.SessionID == "" || child.RefreshTokenHash == "" || len(command.TokenCiphertext) == 0 || len(command.TokenCiphertext) > 4096 ||
		child.AppCode != source.AppCode || child.UserID != source.UserID || child.DeviceID != source.DeviceID ||
		child.TokenFamilyID != familyID || child.ParentSessionID != source.SessionID || child.Generation != source.Generation+1 ||
		command.OutcomeExpiresAtMs != command.NowMs+command.GracePeriodMs {
		return authdomain.RefreshRotationResult{}, xerr.New(xerr.InvalidArgument, "invalid refresh rotation lineage")
	}
	if source.ParentSessionID != "" {
		// source 成为新 parent 后，它的 parent 已是第二上一代；事务内删除旧密文，确保每个 family 最多保留 immediate predecessor 的一份结果。
		if _, err := tx.ExecContext(ctx, `
			DELETE FROM auth_refresh_outcomes
			WHERE app_code = ? AND parent_session_id = ?
		`, source.AppCode, source.ParentSessionID); err != nil {
			return authdomain.RefreshRotationResult{}, err
		}
	}

	_, err := tx.ExecContext(ctx, `
		UPDATE auth_sessions
		SET token_family_id = ?, rotated_to_session_id = ?, rotation_at_ms = ?, rotation_request_id = ?,
		    revoked_at_ms = ?, revoked_reason = 'REFRESH_ROTATED', revoked_request_id = ?, revoked_by = 'refresh_token', updated_at_ms = ?
		WHERE app_code = ? AND session_id = ? AND revoked_at_ms IS NULL
	`, familyID, child.SessionID, command.NowMs, command.RefreshRequestID, command.NowMs, command.RefreshRequestID, command.NowMs, source.AppCode, source.SessionID)
	if err != nil {
		return authdomain.RefreshRotationResult{}, err
	}
	if err := insertSession(ctx, tx, child); err != nil {
		return authdomain.RefreshRotationResult{}, err
	}
	_, err = tx.ExecContext(ctx, `
		INSERT INTO auth_refresh_outcomes (app_code, parent_session_id, child_session_id, refresh_request_id, token_ciphertext, created_at_ms, expires_at_ms)
		VALUES (?, ?, ?, ?, ?, ?, ?)
	`, source.AppCode, source.SessionID, child.SessionID, command.RefreshRequestID, command.TokenCiphertext, command.NowMs, command.OutcomeExpiresAtMs)
	if err != nil {
		return authdomain.RefreshRotationResult{}, mapAuthDuplicateError(err)
	}
	// 正常 rotation 的旧 sid 也有尚未过期的 access JWT；与 child/outcome 同事务写 durable job，
	// 覆盖 Redis 写成功后进程崩溃、Redis 重启等旧 JWT 重新放行风险。
	if err := enqueueSessionDenylistJobs(ctx, tx, source.AppCode, []string{source.SessionID}, "REFRESH_ROTATED", command.NowMs, command.DenyUntilMs); err != nil {
		return authdomain.RefreshRotationResult{}, err
	}
	source.TokenFamilyID = familyID
	source.RotatedToSessionID = child.SessionID
	source.RotationAtMs = command.NowMs
	source.RotationRequestID = command.RefreshRequestID
	source.RevokedAtMs = command.NowMs
	source.RevokedReason = "REFRESH_ROTATED"
	return authdomain.RefreshRotationResult{
		Status:           authdomain.RefreshRotationCreated,
		SourceSession:    source,
		ActiveSession:    child,
		RefreshRequestID: command.RefreshRequestID,
		TokenCiphertext:  append([]byte(nil), command.TokenCiphertext...),
	}, nil
}

func (r *Repository) revokeRefreshFamilyForReuse(ctx context.Context, tx *sql.Tx, source authdomain.Session, command authdomain.RefreshRotationCommand) (authdomain.RefreshRotationResult, error) {
	sessionIDs, err := lockProvenSessionFamily(ctx, tx, source)
	if err != nil {
		return authdomain.RefreshRotationResult{}, err
	}
	if len(sessionIDs) == 0 {
		// source 已由 refresh hash 锁定，正常 schema 下至少包含自身；空结果视为存储不一致，不能静默成功。
		return authdomain.RefreshRotationResult{}, xerr.New(xerr.Unavailable, "refresh token family is unavailable")
	}
	result, err := revokeProvenSessionFamily(ctx, tx, source, command.NowMs, "REFRESH_TOKEN_REUSE", command.RefreshRequestID, "refresh_replay")
	if err != nil {
		return authdomain.RefreshRotationResult{}, err
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return authdomain.RefreshRotationResult{}, err
	}
	// family 已撤销后任何历史 token pair 都不能再被恢复；与撤销同事务清掉全部相关密文。
	if err := deleteProvenSessionFamilyOutcomes(ctx, tx, source); err != nil {
		return authdomain.RefreshRotationResult{}, err
	}
	if err := enqueueSessionDenylistJobs(ctx, tx, source.AppCode, sessionIDs, "REFRESH_TOKEN_REUSE", command.NowMs, command.DenyUntilMs); err != nil {
		return authdomain.RefreshRotationResult{}, err
	}
	return authdomain.RefreshRotationResult{
		Status:             authdomain.RefreshRotationReuseDetected,
		SourceSession:      source,
		FamilySessionIDs:   sessionIDs,
		FamilyNewlyRevoked: affected > 0,
	}, nil
}

// RevokeSession 按 session_id 或 refresh token hash 吊销 session。

func (r *Repository) RevokeSession(ctx context.Context, sessionID string, refreshTokenHash string, revokedAtMs int64) (bool, error) {
	if sessionID == "" && refreshTokenHash != "" {
		// 只传 refresh token 时先反查 session_id。
		session, err := r.findSessionByRefreshHash(ctx, refreshTokenHash)
		if err != nil {
			return false, err
		}
		sessionID = session.SessionID
	}

	return r.revokeSessionByID(ctx, sessionID, revokedAtMs, "USER_LOGOUT", "", "user_logout")
}

// RevokeSessionFamily 让显式 logout 与并发 refresh 收敛到同一个 family；不同设备拥有不同 family，不会被连带退出。
func (r *Repository) RevokeSessionFamily(ctx context.Context, sessionID string, refreshTokenHash string, revokedAtMs int64, denyUntilMs int64, requestID string) (string, []string, bool, error) {
	if revokedAtMs <= 0 || denyUntilMs <= revokedAtMs {
		return "", nil, false, xerr.New(xerr.InvalidArgument, "invalid session family denylist deadline")
	}
	tx, err := r.db.BeginTx(ctx, nil)
	if err != nil {
		return "", nil, false, err
	}
	defer tx.Rollback()

	var source authdomain.Session
	if strings.TrimSpace(sessionID) != "" {
		source, err = scanSession(tx.QueryRowContext(ctx, `
			SELECT `+sessionProjection+` FROM auth_sessions WHERE app_code = ? AND session_id = ? FOR UPDATE
		`, appcode.FromContext(ctx), strings.TrimSpace(sessionID)))
	} else {
		source, err = scanSession(tx.QueryRowContext(ctx, `
			SELECT `+sessionProjection+` FROM auth_sessions WHERE app_code = ? AND refresh_token_hash = ? FOR UPDATE
		`, appcode.FromContext(ctx), refreshTokenHash))
	}
	if err == sql.ErrNoRows {
		return "", nil, false, tx.Commit()
	}
	if err != nil {
		return "", nil, false, err
	}

	sessionIDs, err := lockProvenSessionFamily(ctx, tx, source)
	if err != nil {
		return "", nil, false, err
	}
	result, err := revokeProvenSessionFamily(ctx, tx, source, revokedAtMs, "USER_LOGOUT", requestID, "user_logout")
	if err != nil {
		return "", nil, false, err
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return "", nil, false, err
	}
	if err := deleteProvenSessionFamilyOutcomes(ctx, tx, source); err != nil {
		return "", nil, false, err
	}
	if err := enqueueSessionDenylistJobs(ctx, tx, source.AppCode, sessionIDs, "USER_LOGOUT", revokedAtMs, denyUntilMs); err != nil {
		return "", nil, false, err
	}
	if err := tx.Commit(); err != nil {
		return "", nil, false, err
	}
	return source.AppCode, sessionIDs, affected > 0, nil
}

func lockProvenSessionFamily(ctx context.Context, tx *sql.Tx, source authdomain.Session) ([]string, error) {
	familyID := strings.TrimSpace(source.TokenFamilyID)
	if familyID == "" {
		// 迁移前 session 没有可证明 lineage；source 已由调用方 FOR UPDATE 锁定，只处理自身，
		// 绝不能按 user/device/time 猜测 successor，否则古老 token 可撤销未来显式登录。
		return []string{source.SessionID}, nil
	}
	// 禁止写成 token_family_id OR session_id：大账号会让优化器退回 user_id 索引并扫描/锁等待数万行。
	// 启动已强校验 family 索引，因此显式 FORCE INDEX，避免统计信息过旧时又选到高基数 user_id 索引；
	// user/device 仅做数据污染防线。
	rows, err := tx.QueryContext(ctx, `
		SELECT session_id
		FROM auth_sessions FORCE INDEX (idx_auth_sessions_token_family)
		WHERE app_code = ? AND token_family_id = ? AND user_id = ? AND device_id = ?
		ORDER BY generation, created_at_ms, session_id
		FOR UPDATE
	`, source.AppCode, familyID, source.UserID, source.DeviceID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var sessionIDs []string
	for rows.Next() {
		var sessionID string
		if err := rows.Scan(&sessionID); err != nil {
			return nil, err
		}
		sessionIDs = append(sessionIDs, sessionID)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return sessionIDs, nil
}

func revokeProvenSessionFamily(ctx context.Context, tx *sql.Tx, source authdomain.Session, revokedAtMs int64, reason string, requestID string, revokedBy string) (sql.Result, error) {
	familyID := strings.TrimSpace(source.TokenFamilyID)
	if familyID == "" {
		// legacy 无 lineage 时仅更新已锁定 source；session_id 主键路径为单行更新。
		return tx.ExecContext(ctx, `
			UPDATE auth_sessions
			SET revoked_at_ms = ?, revoked_reason = ?, revoked_request_id = ?, revoked_by = ?, updated_at_ms = ?
			WHERE app_code = ? AND session_id = ? AND revoked_at_ms IS NULL
		`, revokedAtMs, reason, requestID, revokedBy, revokedAtMs, source.AppCode, source.SessionID)
	}
	return tx.ExecContext(ctx, `
		UPDATE auth_sessions FORCE INDEX (idx_auth_sessions_token_family)
		SET revoked_at_ms = ?, revoked_reason = ?, revoked_request_id = ?, revoked_by = ?, updated_at_ms = ?
		WHERE app_code = ? AND token_family_id = ? AND user_id = ? AND device_id = ? AND revoked_at_ms IS NULL
	`, revokedAtMs, reason, requestID, revokedBy, revokedAtMs, source.AppCode, familyID, source.UserID, source.DeviceID)
}

func deleteProvenSessionFamilyOutcomes(ctx context.Context, tx *sql.Tx, source authdomain.Session) error {
	familyID := strings.TrimSpace(source.TokenFamilyID)
	if familyID == "" {
		// 无 lineage 时最多删除 source 自己作为 predecessor 的结果，不通过任何启发式扩大范围。
		_, err := tx.ExecContext(ctx, `
			DELETE FROM auth_refresh_outcomes
			WHERE app_code = ? AND parent_session_id = ?
		`, source.AppCode, source.SessionID)
		return err
	}
	_, err := tx.ExecContext(ctx, `
		DELETE outcome
		FROM auth_sessions session FORCE INDEX (idx_auth_sessions_token_family)
		STRAIGHT_JOIN auth_refresh_outcomes outcome
		  ON session.app_code = outcome.app_code AND session.session_id = outcome.parent_session_id
		WHERE session.app_code = ? AND session.token_family_id = ? AND session.user_id = ? AND session.device_id = ?
	`, source.AppCode, familyID, source.UserID, source.DeviceID)
	return err
}

func (r *Repository) RevokeSessionWithReason(ctx context.Context, sessionID string, revokedAtMs int64, reason string, requestID string, revokedBy string) (bool, error) {
	return r.revokeSessionByID(ctx, sessionID, revokedAtMs, reason, requestID, revokedBy)
}

func (r *Repository) RevokeUserDeviceSessionsForRisk(ctx context.Context, sourceSessionID string, userID int64, revokedAtMs int64, reason string, requestID string, revokedBy string) ([]string, error) {
	// IP 风控是异步执行；客户端可能已经用源 session 的 refresh token 轮换出了新 session。
	// 这里先锁定源 session，拿到设备和创建时间，再圈定同一用户、同一设备、同一登录链路之后产生的未过期 session，避免误伤同账号其他设备。
	tx, err := r.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()

	var deviceID string
	var sourceUserID int64
	var sourceCreatedAtMs int64
	err = tx.QueryRowContext(ctx, `
		SELECT user_id, device_id, created_at_ms
		FROM auth_sessions
		WHERE app_code = ? AND session_id = ?
		FOR UPDATE
	`, appcode.FromContext(ctx), sourceSessionID).Scan(&sourceUserID, &deviceID, &sourceCreatedAtMs)
	if err == sql.ErrNoRows {
		// 风控任务来自登录成功后的持久 session；若源 session 已被异常清理，按空结果幂等结束。
		return nil, tx.Commit()
	}
	if err != nil {
		return nil, err
	}
	if sourceUserID != userID {
		// job.user_id 和 session.user_id 不一致说明数据链路被污染，不能扩大吊销范围。
		return nil, xerr.New(xerr.SessionRevoked, "risk session owner mismatch")
	}

	rows, err := tx.QueryContext(ctx, `
		SELECT session_id
		FROM auth_sessions
		WHERE app_code = ?
		  AND user_id = ?
		  AND device_id = ?
		  AND created_at_ms >= ?
		  AND expires_at_ms > ?
		FOR UPDATE
	`, appcode.FromContext(ctx), userID, deviceID, sourceCreatedAtMs, revokedAtMs)
	if err != nil {
		return nil, err
	}
	var sessionIDs []string
	for rows.Next() {
		var sessionID string
		if err := rows.Scan(&sessionID); err != nil {
			_ = rows.Close()
			return nil, err
		}
		sessionIDs = append(sessionIDs, sessionID)
	}
	if err := rows.Close(); err != nil {
		return nil, err
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	if len(sessionIDs) == 0 {
		return sessionIDs, tx.Commit()
	}

	for _, sessionID := range sessionIDs {
		// 已经被 refresh 轮换的 session 不覆盖原撤销原因，但仍返回给上层写 denylist，避免旧 access token 继续通过 gateway。
		if _, err := tx.ExecContext(ctx, `
			UPDATE auth_sessions
			SET revoked_at_ms = ?, revoked_reason = ?, revoked_request_id = ?, revoked_by = ?, updated_at_ms = ?
			WHERE app_code = ? AND session_id = ? AND revoked_at_ms IS NULL
		`, revokedAtMs, reason, requestID, revokedBy, revokedAtMs, appcode.FromContext(ctx), sessionID); err != nil {
			return nil, err
		}
	}

	return sessionIDs, tx.Commit()
}

func (r *Repository) UpdateSessionHeartbeat(ctx context.Context, sessionID string, userID int64, heartbeatAtMs int64, requestID string) (authdomain.Session, error) {
	// App 心跳绑定服务端 session，不能只按 user_id 宽泛刷新，避免旧 token 或跨设备请求续住错误会话。
	result, err := r.db.ExecContext(ctx, `
		UPDATE auth_sessions
		SET last_heartbeat_at_ms = ?, last_heartbeat_request_id = ?, updated_at_ms = ?
		WHERE app_code = ? AND session_id = ? AND user_id = ? AND revoked_at_ms IS NULL AND expires_at_ms > ?
	`, heartbeatAtMs, requestID, heartbeatAtMs, appcode.FromContext(ctx), sessionID, userID, heartbeatAtMs)
	if err != nil {
		return authdomain.Session{}, err
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return authdomain.Session{}, err
	}
	if affected == 0 {
		// 未命中只说明 session 不可用或不属于当前用户；对外统一成 session revoked，避免泄漏会话存在性。
		return authdomain.Session{}, xerr.New(xerr.SessionRevoked, "session revoked")
	}

	return r.findSessionByID(ctx, sessionID)
}

func (r *Repository) revokeSessionByID(ctx context.Context, sessionID string, revokedAtMs int64, reason string, requestID string, revokedBy string) (bool, error) {
	result, err := r.db.ExecContext(ctx, `
		UPDATE auth_sessions
		SET revoked_at_ms = ?, revoked_reason = ?, revoked_request_id = ?, revoked_by = ?, updated_at_ms = ?
		WHERE app_code = ? AND session_id = ? AND revoked_at_ms IS NULL
	`, revokedAtMs, reason, requestID, revokedBy, revokedAtMs, appcode.FromContext(ctx), sessionID)
	if err != nil {
		return false, err
	}

	affected, err := result.RowsAffected()
	if err != nil {
		return false, err
	}

	// affected=false 表示没有可吊销 session，调用方按幂等结果处理。
	return affected > 0, nil
}

// FindThirdPartyIdentity 查找 provider + subject 绑定。

func (r *Repository) findSessionByRefreshHash(ctx context.Context, refreshTokenHash string) (authdomain.Session, error) {
	// refresh token hash 是唯一索引，最多命中一条 session。
	session, err := scanSession(r.db.QueryRowContext(ctx, `
		SELECT `+sessionProjection+`
		FROM auth_sessions
		WHERE app_code = ? AND refresh_token_hash = ?
	`, appcode.FromContext(ctx), refreshTokenHash))
	if err == sql.ErrNoRows {
		// 找不到 hash 按 revoked 返回，避免暴露 token 是否存在。
		return authdomain.Session{}, xerr.New(xerr.SessionRevoked, "session revoked")
	}
	if err != nil {
		return authdomain.Session{}, err
	}
	return session, nil
}

func (r *Repository) findSessionByID(ctx context.Context, sessionID string) (authdomain.Session, error) {
	// session_id 来自已校验 access token 的 sid claim，仍必须回查服务端 session 状态。
	session, err := scanSession(r.db.QueryRowContext(ctx, `
		SELECT `+sessionProjection+`
		FROM auth_sessions
		WHERE app_code = ? AND session_id = ?
	`, appcode.FromContext(ctx), sessionID))
	if err == sql.ErrNoRows {
		// 不暴露 session_id 是否存在，统一按 revoked 处理。
		return authdomain.Session{}, xerr.New(xerr.SessionRevoked, "session revoked")
	}
	if err != nil {
		return authdomain.Session{}, err
	}
	return session, nil
}

func insertSession(ctx context.Context, tx *sql.Tx, session authdomain.Session) error {
	// insertSession 只在已有事务中使用，保证与 session 替换或三方注册同提交。
	_, err := tx.ExecContext(ctx, `
		INSERT INTO auth_sessions (app_code, session_id, user_id, refresh_token_hash, token_family_id, generation, parent_session_id, rotated_to_session_id, rotation_at_ms, rotation_request_id, device_id, expires_at_ms, last_heartbeat_at_ms, last_heartbeat_request_id, revoked_at_ms, revoked_reason, revoked_request_id, revoked_by, created_at_ms, updated_at_ms)
		VALUES (?, ?, ?, ?, ?, ?, ?, '', 0, '', ?, ?, ?, ?, NULL, '', '', '', ?, ?)
	`, appcode.Normalize(session.AppCode), session.SessionID, session.UserID, session.RefreshTokenHash, normalizedFamilyID(session), session.Generation, session.ParentSessionID, session.DeviceID, session.ExpiresAtMs, session.LastHeartbeatAtMs, session.LastHeartbeatRequestID, session.CreatedAtMs, session.UpdatedAtMs)
	if err != nil {
		return mapAuthDuplicateError(err)
	}

	return nil
}

type sessionScanner interface {
	Scan(dest ...any) error
}

func scanSession(scanner sessionScanner) (authdomain.Session, error) {
	var session authdomain.Session
	var revokedAt sql.Null[int64]
	err := scanner.Scan(
		&session.AppCode,
		&session.SessionID,
		&session.UserID,
		&session.RefreshTokenHash,
		&session.TokenFamilyID,
		&session.Generation,
		&session.ParentSessionID,
		&session.RotatedToSessionID,
		&session.RotationAtMs,
		&session.RotationRequestID,
		&session.DeviceID,
		&session.ExpiresAtMs,
		&session.LastHeartbeatAtMs,
		&session.LastHeartbeatRequestID,
		&revokedAt,
		&session.RevokedReason,
		&session.RevokedRequestID,
		&session.RevokedBy,
		&session.CreatedAtMs,
		&session.UpdatedAtMs,
	)
	if err != nil {
		return authdomain.Session{}, err
	}
	if revokedAt.Valid {
		session.RevokedAtMs = revokedAt.V
	}
	return session, nil
}

func normalizedFamilyID(session authdomain.Session) string {
	if familyID := strings.TrimSpace(session.TokenFamilyID); familyID != "" {
		return familyID
	}
	return session.SessionID
}