mirror of
https://github.com/CJackHwang/ds2api.git
synced 2026-05-04 16:35:27 +08:00
fix: 修复会话管理相关问题并拆分文件
1. 修复无限循环问题 - DeleteAllSessions/DeleteAllSessionsForToken 添加无进度检测 - 连续 3 轮删除失败则退出循环 - DeleteAllSessionsForToken 添加 cursor 推进逻辑 2. 修复字段语义不准确 - TotalCount 重命名为 FirstPageCount - 明确该值仅统计第一页，多页账户需关注 HasMore 3. 修复 defer 执行顺序问题 - 合并两个 defer，确保先删除会话再释放账号 - 使用同步删除避免并发截断风险 4. 文件拆分 - 新建 client_session_delete.go 处理会话删除 - client_session.go 专注于会话查询
This commit is contained in:
latticeon
@@ -35,19 +35,22 @@ func (h *Handler) ChatCompletions(w http.ResponseWriter, r *http.Request) {
|
||||
writeOpenAIError(w, status, detail)
|
||||
return
|
||||
}
|
||||
defer h.Auth.Release(a)
|
||||
|
||||
// 自动删除会话功能
|
||||
if h.Store.AutoDeleteSessions() && a.DeepSeekToken != "" {
|
||||
defer func() {
|
||||
defer func() {
|
||||
// 自动删除会话（同步）
|
||||
// 必须在 Release 之前同步删除，否则：
|
||||
// 1. 异步删除时账号已被 Release
|
||||
// 2. 新请求可能获取到同一账号并开始使用
|
||||
// 3. 异步删除仍在进行，会截断新请求正在使用的会话
|
||||
if h.Store.AutoDeleteSessions() && a.DeepSeekToken != "" {
|
||||
deleted, err := h.DS.DeleteAllSessionsForToken(context.Background(), a.DeepSeekToken)
|
||||
if err != nil {
|
||||
config.Logger.Warn("[auto_delete_sessions] failed", "account", a.AccountID, "error", err)
|
||||
} else {
|
||||
config.Logger.Debug("[auto_delete_sessions] deleted", "account", a.AccountID, "count", deleted)
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
h.Auth.Release(a)
|
||||
}()
|
||||
|
||||
r = r.WithContext(auth.WithAuth(r.Context(), a))
|
||||
|
||||
|
||||
@@ -128,7 +128,7 @@ func (h *Handler) testAccount(ctx context.Context, acc config.Account, model, me
|
||||
// 获取会话数量
|
||||
sessionStats, sessionErr := h.DS.GetSessionCountForToken(ctx, token)
|
||||
if sessionErr == nil && sessionStats != nil {
|
||||
result["session_count"] = sessionStats.TotalCount
|
||||
result["session_count"] = sessionStats.FirstPageCount
|
||||
}
|
||||
|
||||
if strings.TrimSpace(message) == "" {
|
||||
|
||||
@@ -22,12 +22,12 @@ type SessionInfo struct {
|
||||
|
||||
// SessionStats 会话统计结果
|
||||
type SessionStats struct {
|
||||
AccountID string // 账号标识 (email 或 mobile)
|
||||
TotalCount int // 总会话数量
|
||||
PinnedCount int // 置顶会话数量
|
||||
HasMore bool // 是否还有更多
|
||||
Success bool // 请求是否成功
|
||||
ErrorMessage string // 错误信息
|
||||
AccountID string // 账号标识 (email 或 mobile)
|
||||
FirstPageCount int // 第一页会话数量（当 HasMore 为 true 时，真实总数可能更大）
|
||||
PinnedCount int // 置顶会话数量
|
||||
HasMore bool // 是否还有更多页
|
||||
Success bool // 请求是否成功
|
||||
ErrorMessage string // 错误信息
|
||||
}
|
||||
|
||||
// GetSessionCount 获取单个账号的会话数量
|
||||
@@ -63,7 +63,7 @@ func (c *Client) GetSessionCount(ctx context.Context, a *auth.RequestAuth, maxAt
|
||||
chatSessions, _ := bizData["chat_sessions"].([]any)
|
||||
hasMore, _ := bizData["has_more"].(bool)
|
||||
|
||||
stats.TotalCount = len(chatSessions)
|
||||
stats.FirstPageCount = len(chatSessions)
|
||||
stats.HasMore = hasMore
|
||||
stats.Success = true
|
||||
|
||||
@@ -126,9 +126,9 @@ func (c *Client) GetSessionCountForToken(ctx context.Context, token string) (*Se
|
||||
hasMore, _ := bizData["has_more"].(bool)
|
||||
|
||||
stats := &SessionStats{
|
||||
TotalCount: len(chatSessions),
|
||||
HasMore: hasMore,
|
||||
Success: true,
|
||||
FirstPageCount: len(chatSessions),
|
||||
HasMore: hasMore,
|
||||
Success: true,
|
||||
}
|
||||
|
||||
// 统计置顶会话数量
|
||||
@@ -252,185 +252,3 @@ func floatFromMap(m map[string]any, key string) float64 {
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
// DeleteSessionResult 删除会话结果
|
||||
type DeleteSessionResult struct {
|
||||
SessionID string // 会话 ID
|
||||
Success bool // 是否成功
|
||||
ErrorMessage string // 错误信息
|
||||
}
|
||||
|
||||
// DeleteSession 删除单个会话
|
||||
func (c *Client) DeleteSession(ctx context.Context, a *auth.RequestAuth, sessionID string, maxAttempts int) (*DeleteSessionResult, error) {
|
||||
if maxAttempts <= 0 {
|
||||
maxAttempts = c.maxRetries
|
||||
}
|
||||
|
||||
result := &DeleteSessionResult{
|
||||
SessionID: sessionID,
|
||||
}
|
||||
|
||||
if sessionID == "" {
|
||||
result.ErrorMessage = "session_id is required"
|
||||
return result, errors.New(result.ErrorMessage)
|
||||
}
|
||||
|
||||
attempts := 0
|
||||
refreshed := false
|
||||
|
||||
for attempts < maxAttempts {
|
||||
headers := c.authHeaders(a.DeepSeekToken)
|
||||
|
||||
payload := map[string]any{
|
||||
"chat_session_id": sessionID,
|
||||
}
|
||||
|
||||
resp, status, err := c.postJSONWithStatus(ctx, c.regular, DeepSeekDeleteSessionURL, headers, payload)
|
||||
if err != nil {
|
||||
config.Logger.Warn("[delete_session] request error", "error", err, "session_id", sessionID)
|
||||
attempts++
|
||||
continue
|
||||
}
|
||||
|
||||
code := intFrom(resp["code"])
|
||||
if status == http.StatusOK && code == 0 {
|
||||
result.Success = true
|
||||
return result, nil
|
||||
}
|
||||
|
||||
msg, _ := resp["msg"].(string)
|
||||
result.ErrorMessage = fmt.Sprintf("status=%d, code=%d, msg=%s", status, code, msg)
|
||||
config.Logger.Warn("[delete_session] failed", "status", status, "code", code, "msg", msg, "session_id", sessionID)
|
||||
|
||||
if a.UseConfigToken {
|
||||
if isTokenInvalid(status, code, msg) && !refreshed {
|
||||
if c.Auth.RefreshToken(ctx, a) {
|
||||
refreshed = true
|
||||
continue
|
||||
}
|
||||
}
|
||||
if c.Auth.SwitchAccount(ctx, a) {
|
||||
refreshed = false
|
||||
attempts++
|
||||
continue
|
||||
}
|
||||
}
|
||||
attempts++
|
||||
}
|
||||
|
||||
result.Success = false
|
||||
result.ErrorMessage = "delete session failed after retries"
|
||||
return result, errors.New(result.ErrorMessage)
|
||||
}
|
||||
|
||||
// DeleteSessionForToken 直接使用 token 删除会话（直通模式）
|
||||
func (c *Client) DeleteSessionForToken(ctx context.Context, token string, sessionID string) (*DeleteSessionResult, error) {
|
||||
result := &DeleteSessionResult{
|
||||
SessionID: sessionID,
|
||||
}
|
||||
|
||||
if sessionID == "" {
|
||||
result.ErrorMessage = "session_id is required"
|
||||
return result, errors.New(result.ErrorMessage)
|
||||
}
|
||||
|
||||
headers := c.authHeaders(token)
|
||||
payload := map[string]any{
|
||||
"chat_session_id": sessionID,
|
||||
}
|
||||
|
||||
resp, status, err := c.postJSONWithStatus(ctx, c.regular, DeepSeekDeleteSessionURL, headers, payload)
|
||||
if err != nil {
|
||||
result.ErrorMessage = err.Error()
|
||||
return result, err
|
||||
}
|
||||
|
||||
code := intFrom(resp["code"])
|
||||
if status != http.StatusOK || code != 0 {
|
||||
msg, _ := resp["msg"].(string)
|
||||
result.ErrorMessage = fmt.Sprintf("request failed: status=%d, code=%d, msg=%s", status, code, msg)
|
||||
return result, fmt.Errorf(result.ErrorMessage)
|
||||
}
|
||||
|
||||
result.Success = true
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// DeleteAllSessions 删除所有会话（谨慎使用）
|
||||
func (c *Client) DeleteAllSessions(ctx context.Context, a *auth.RequestAuth) (int, error) {
|
||||
deleted := 0
|
||||
cursor := ""
|
||||
|
||||
for {
|
||||
sessions, hasMore, err := c.FetchSessionPage(ctx, a, cursor)
|
||||
if err != nil {
|
||||
return deleted, err
|
||||
}
|
||||
|
||||
for _, session := range sessions {
|
||||
_, err := c.DeleteSession(ctx, a, session.ID, 1)
|
||||
if err == nil {
|
||||
deleted++
|
||||
}
|
||||
}
|
||||
|
||||
if !hasMore || len(sessions) == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
return deleted, nil
|
||||
}
|
||||
|
||||
// DeleteAllSessionsForToken 直接使用 token 删除所有会话（直通模式）
|
||||
func (c *Client) DeleteAllSessionsForToken(ctx context.Context, token string) (int, error) {
|
||||
deleted := 0
|
||||
cursor := ""
|
||||
|
||||
for {
|
||||
// 获取会话列表
|
||||
headers := c.authHeaders(token)
|
||||
params := url.Values{}
|
||||
params.Set("lte_cursor.pinned", "false")
|
||||
if cursor != "" {
|
||||
params.Set("lte_cursor", cursor)
|
||||
}
|
||||
reqURL := DeepSeekFetchSessionURL + "?" + params.Encode()
|
||||
|
||||
resp, status, err := c.getJSONWithStatus(ctx, c.regular, reqURL, headers)
|
||||
if err != nil {
|
||||
return deleted, err
|
||||
}
|
||||
|
||||
code := intFrom(resp["code"])
|
||||
if status != http.StatusOK || code != 0 {
|
||||
msg, _ := resp["msg"].(string)
|
||||
return deleted, fmt.Errorf("fetch sessions failed: status=%d, code=%d, msg=%s", status, code, msg)
|
||||
}
|
||||
|
||||
data, _ := resp["data"].(map[string]any)
|
||||
bizData, _ := data["biz_data"].(map[string]any)
|
||||
chatSessions, _ := bizData["chat_sessions"].([]any)
|
||||
hasMore, _ := bizData["has_more"].(bool)
|
||||
|
||||
// 删除每个会话
|
||||
for _, s := range chatSessions {
|
||||
if m, ok := s.(map[string]any); ok {
|
||||
sessionID := stringFromMap(m, "id")
|
||||
if sessionID == "" {
|
||||
continue
|
||||
}
|
||||
_, err := c.DeleteSessionForToken(ctx, token, sessionID)
|
||||
if err == nil {
|
||||
deleted++
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if !hasMore || len(chatSessions) == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
return deleted, nil
|
||||
}
|
||||
|
||||
238
internal/deepseek/client_session_delete.go
Normal file
238
internal/deepseek/client_session_delete.go
Normal file
@@ -0,0 +1,238 @@
|
||||
package deepseek
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/url"
|
||||
|
||||
"ds2api/internal/auth"
|
||||
"ds2api/internal/config"
|
||||
)
|
||||
|
||||
// DeleteSessionResult 删除会话结果
|
||||
type DeleteSessionResult struct {
|
||||
SessionID string // 会话 ID
|
||||
Success bool // 是否成功
|
||||
ErrorMessage string // 错误信息
|
||||
}
|
||||
|
||||
// DeleteSession 删除单个会话
|
||||
func (c *Client) DeleteSession(ctx context.Context, a *auth.RequestAuth, sessionID string, maxAttempts int) (*DeleteSessionResult, error) {
|
||||
if maxAttempts <= 0 {
|
||||
maxAttempts = c.maxRetries
|
||||
}
|
||||
|
||||
result := &DeleteSessionResult{
|
||||
SessionID: sessionID,
|
||||
}
|
||||
|
||||
if sessionID == "" {
|
||||
result.ErrorMessage = "session_id is required"
|
||||
return result, errors.New(result.ErrorMessage)
|
||||
}
|
||||
|
||||
attempts := 0
|
||||
refreshed := false
|
||||
|
||||
for attempts < maxAttempts {
|
||||
headers := c.authHeaders(a.DeepSeekToken)
|
||||
|
||||
payload := map[string]any{
|
||||
"chat_session_id": sessionID,
|
||||
}
|
||||
|
||||
resp, status, err := c.postJSONWithStatus(ctx, c.regular, DeepSeekDeleteSessionURL, headers, payload)
|
||||
if err != nil {
|
||||
config.Logger.Warn("[delete_session] request error", "error", err, "session_id", sessionID)
|
||||
attempts++
|
||||
continue
|
||||
}
|
||||
|
||||
code := intFrom(resp["code"])
|
||||
if status == http.StatusOK && code == 0 {
|
||||
result.Success = true
|
||||
return result, nil
|
||||
}
|
||||
|
||||
msg, _ := resp["msg"].(string)
|
||||
result.ErrorMessage = fmt.Sprintf("status=%d, code=%d, msg=%s", status, code, msg)
|
||||
config.Logger.Warn("[delete_session] failed", "status", status, "code", code, "msg", msg, "session_id", sessionID)
|
||||
|
||||
if a.UseConfigToken {
|
||||
if isTokenInvalid(status, code, msg) && !refreshed {
|
||||
if c.Auth.RefreshToken(ctx, a) {
|
||||
refreshed = true
|
||||
continue
|
||||
}
|
||||
}
|
||||
if c.Auth.SwitchAccount(ctx, a) {
|
||||
refreshed = false
|
||||
attempts++
|
||||
continue
|
||||
}
|
||||
}
|
||||
attempts++
|
||||
}
|
||||
|
||||
result.Success = false
|
||||
result.ErrorMessage = "delete session failed after retries"
|
||||
return result, errors.New(result.ErrorMessage)
|
||||
}
|
||||
|
||||
// DeleteSessionForToken 直接使用 token 删除会话（直通模式）
|
||||
func (c *Client) DeleteSessionForToken(ctx context.Context, token string, sessionID string) (*DeleteSessionResult, error) {
|
||||
result := &DeleteSessionResult{
|
||||
SessionID: sessionID,
|
||||
}
|
||||
|
||||
if sessionID == "" {
|
||||
result.ErrorMessage = "session_id is required"
|
||||
return result, errors.New(result.ErrorMessage)
|
||||
}
|
||||
|
||||
headers := c.authHeaders(token)
|
||||
payload := map[string]any{
|
||||
"chat_session_id": sessionID,
|
||||
}
|
||||
|
||||
resp, status, err := c.postJSONWithStatus(ctx, c.regular, DeepSeekDeleteSessionURL, headers, payload)
|
||||
if err != nil {
|
||||
result.ErrorMessage = err.Error()
|
||||
return result, err
|
||||
}
|
||||
|
||||
code := intFrom(resp["code"])
|
||||
if status != http.StatusOK || code != 0 {
|
||||
msg, _ := resp["msg"].(string)
|
||||
result.ErrorMessage = fmt.Sprintf("request failed: status=%d, code=%d, msg=%s", status, code, msg)
|
||||
return result, fmt.Errorf(result.ErrorMessage)
|
||||
}
|
||||
|
||||
result.Success = true
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// DeleteAllSessions 删除所有会话（谨慎使用）
|
||||
func (c *Client) DeleteAllSessions(ctx context.Context, a *auth.RequestAuth) (int, error) {
|
||||
const maxNoProgress = 3 // 最大无进度次数
|
||||
|
||||
deleted := 0
|
||||
cursor := ""
|
||||
noProgressCount := 0
|
||||
|
||||
for {
|
||||
sessions, hasMore, err := c.FetchSessionPage(ctx, a, cursor)
|
||||
if err != nil {
|
||||
return deleted, err
|
||||
}
|
||||
|
||||
deletedThisRound := 0
|
||||
for _, session := range sessions {
|
||||
_, err := c.DeleteSession(ctx, a, session.ID, 1)
|
||||
if err == nil {
|
||||
deleted++
|
||||
deletedThisRound++
|
||||
}
|
||||
}
|
||||
|
||||
// 无进度检测：如果连续多轮没有成功删除任何会话，退出循环
|
||||
if deletedThisRound == 0 {
|
||||
noProgressCount++
|
||||
if noProgressCount >= maxNoProgress {
|
||||
config.Logger.Warn("[delete_all_sessions] exiting due to no progress", "deleted", deleted)
|
||||
break
|
||||
}
|
||||
} else {
|
||||
noProgressCount = 0
|
||||
}
|
||||
|
||||
if !hasMore || len(sessions) == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
return deleted, nil
|
||||
}
|
||||
|
||||
// DeleteAllSessionsForToken 直接使用 token 删除所有会话（直通模式）
|
||||
func (c *Client) DeleteAllSessionsForToken(ctx context.Context, token string) (int, error) {
|
||||
const maxNoProgress = 3 // 最大无进度次数
|
||||
|
||||
deleted := 0
|
||||
cursor := ""
|
||||
noProgressCount := 0
|
||||
|
||||
for {
|
||||
// 获取会话列表
|
||||
headers := c.authHeaders(token)
|
||||
params := url.Values{}
|
||||
params.Set("lte_cursor.pinned", "false")
|
||||
if cursor != "" {
|
||||
params.Set("lte_cursor", cursor)
|
||||
}
|
||||
reqURL := DeepSeekFetchSessionURL + "?" + params.Encode()
|
||||
|
||||
resp, status, err := c.getJSONWithStatus(ctx, c.regular, reqURL, headers)
|
||||
if err != nil {
|
||||
return deleted, err
|
||||
}
|
||||
|
||||
code := intFrom(resp["code"])
|
||||
if status != http.StatusOK || code != 0 {
|
||||
msg, _ := resp["msg"].(string)
|
||||
return deleted, fmt.Errorf("fetch sessions failed: status=%d, code=%d, msg=%s", status, code, msg)
|
||||
}
|
||||
|
||||
data, _ := resp["data"].(map[string]any)
|
||||
bizData, _ := data["biz_data"].(map[string]any)
|
||||
chatSessions, _ := bizData["chat_sessions"].([]any)
|
||||
hasMore, _ := bizData["has_more"].(bool)
|
||||
|
||||
// 删除每个会话
|
||||
deletedThisRound := 0
|
||||
for _, s := range chatSessions {
|
||||
if m, ok := s.(map[string]any); ok {
|
||||
sessionID := stringFromMap(m, "id")
|
||||
if sessionID == "" {
|
||||
continue
|
||||
}
|
||||
_, err := c.DeleteSessionForToken(ctx, token, sessionID)
|
||||
if err == nil {
|
||||
deleted++
|
||||
deletedThisRound++
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 无进度检测：如果连续多轮没有成功删除任何会话，退出循环
|
||||
if deletedThisRound == 0 {
|
||||
noProgressCount++
|
||||
if noProgressCount >= maxNoProgress {
|
||||
config.Logger.Warn("[delete_all_sessions_for_token] exiting due to no progress", "deleted", deleted)
|
||||
break
|
||||
}
|
||||
} else {
|
||||
noProgressCount = 0
|
||||
}
|
||||
|
||||
if !hasMore || len(chatSessions) == 0 {
|
||||
break
|
||||
}
|
||||
|
||||
// 推进 cursor：从最后一个会话中提取 cursor（通常基于 updated_at 或 id）
|
||||
if len(chatSessions) > 0 {
|
||||
if lastSession, ok := chatSessions[len(chatSessions)-1].(map[string]any); ok {
|
||||
// 尝试从响应中获取 cursor 字段，或使用最后会话的 ID/updated_at
|
||||
if nextCursor, ok := bizData["cursor"].(string); ok && nextCursor != "" {
|
||||
cursor = nextCursor
|
||||
} else if nextCursor := stringFromMap(lastSession, "id"); nextCursor != "" {
|
||||
cursor = nextCursor
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return deleted, nil
|
||||
}
|
||||
Reference in New Issue
Block a user