openai_sticky_compat.go 6.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221
  1. package service
  2. import (
  3. "context"
  4. "crypto/sha256"
  5. "encoding/hex"
  6. "fmt"
  7. "strings"
  8. "sync/atomic"
  9. "time"
  10. "github.com/cespare/xxhash/v2"
  11. "github.com/gin-gonic/gin"
  12. )
  13. type openAILegacySessionHashContextKey struct{}
  14. var openAILegacySessionHashKey = openAILegacySessionHashContextKey{}
  15. var (
  16. openAIStickyLegacyReadFallbackTotal atomic.Int64
  17. openAIStickyLegacyReadFallbackHit atomic.Int64
  18. openAIStickyLegacyDualWriteTotal atomic.Int64
  19. )
  20. func openAIStickyCompatStats() (legacyReadFallbackTotal, legacyReadFallbackHit, legacyDualWriteTotal int64) {
  21. return openAIStickyLegacyReadFallbackTotal.Load(),
  22. openAIStickyLegacyReadFallbackHit.Load(),
  23. openAIStickyLegacyDualWriteTotal.Load()
  24. }
  25. // DeriveSessionHashFromSeed computes the current-format sticky-session hash
  26. // from an arbitrary seed string.
  27. func DeriveSessionHashFromSeed(seed string) string {
  28. currentHash, _ := deriveOpenAISessionHashes(seed)
  29. return currentHash
  30. }
  31. func deriveOpenAISessionHashes(sessionID string) (currentHash string, legacyHash string) {
  32. normalized := strings.TrimSpace(sessionID)
  33. if normalized == "" {
  34. return "", ""
  35. }
  36. currentHash = fmt.Sprintf("%016x", xxhash.Sum64String(normalized))
  37. sum := sha256.Sum256([]byte(normalized))
  38. legacyHash = hex.EncodeToString(sum[:])
  39. return currentHash, legacyHash
  40. }
  41. func withOpenAILegacySessionHash(ctx context.Context, legacyHash string) context.Context {
  42. if ctx == nil {
  43. return nil
  44. }
  45. trimmed := strings.TrimSpace(legacyHash)
  46. if trimmed == "" {
  47. return ctx
  48. }
  49. return context.WithValue(ctx, openAILegacySessionHashKey, trimmed)
  50. }
  51. func openAILegacySessionHashFromContext(ctx context.Context) string {
  52. if ctx == nil {
  53. return ""
  54. }
  55. value, _ := ctx.Value(openAILegacySessionHashKey).(string)
  56. return strings.TrimSpace(value)
  57. }
  58. func attachOpenAILegacySessionHashToGin(c *gin.Context, legacyHash string) {
  59. if c == nil || c.Request == nil {
  60. return
  61. }
  62. c.Request = c.Request.WithContext(withOpenAILegacySessionHash(c.Request.Context(), legacyHash))
  63. }
  64. func (s *OpenAIGatewayService) openAISessionHashReadOldFallbackEnabled() bool {
  65. if s == nil || s.cfg == nil {
  66. return true
  67. }
  68. return s.cfg.Gateway.OpenAIWS.SessionHashReadOldFallback
  69. }
  70. func (s *OpenAIGatewayService) openAISessionHashDualWriteOldEnabled() bool {
  71. if s == nil || s.cfg == nil {
  72. return true
  73. }
  74. return s.cfg.Gateway.OpenAIWS.SessionHashDualWriteOld
  75. }
  76. func (s *OpenAIGatewayService) openAISessionCacheKey(sessionHash string) string {
  77. normalized := strings.TrimSpace(sessionHash)
  78. if normalized == "" {
  79. return ""
  80. }
  81. return "openai:" + normalized
  82. }
  83. func (s *OpenAIGatewayService) openAILegacySessionCacheKey(ctx context.Context, sessionHash string) string {
  84. legacyHash := openAILegacySessionHashFromContext(ctx)
  85. if legacyHash == "" {
  86. return ""
  87. }
  88. legacyKey := "openai:" + legacyHash
  89. if legacyKey == s.openAISessionCacheKey(sessionHash) {
  90. return ""
  91. }
  92. return legacyKey
  93. }
  94. func (s *OpenAIGatewayService) openAIStickyLegacyTTL(ttl time.Duration) time.Duration {
  95. legacyTTL := ttl
  96. if legacyTTL <= 0 {
  97. legacyTTL = openaiStickySessionTTL
  98. }
  99. if legacyTTL > 10*time.Minute {
  100. return 10 * time.Minute
  101. }
  102. return legacyTTL
  103. }
  104. func (s *OpenAIGatewayService) getStickySessionAccountID(ctx context.Context, groupID *int64, sessionHash string) (int64, error) {
  105. if s == nil || s.cache == nil {
  106. return 0, nil
  107. }
  108. primaryKey := s.openAISessionCacheKey(sessionHash)
  109. if primaryKey == "" {
  110. return 0, nil
  111. }
  112. accountID, err := s.cache.GetSessionAccountID(ctx, derefGroupID(groupID), primaryKey)
  113. if err == nil && accountID > 0 {
  114. return accountID, nil
  115. }
  116. if !s.openAISessionHashReadOldFallbackEnabled() {
  117. return accountID, err
  118. }
  119. legacyKey := s.openAILegacySessionCacheKey(ctx, sessionHash)
  120. if legacyKey == "" {
  121. return accountID, err
  122. }
  123. openAIStickyLegacyReadFallbackTotal.Add(1)
  124. legacyAccountID, legacyErr := s.cache.GetSessionAccountID(ctx, derefGroupID(groupID), legacyKey)
  125. if legacyErr == nil && legacyAccountID > 0 {
  126. openAIStickyLegacyReadFallbackHit.Add(1)
  127. return legacyAccountID, nil
  128. }
  129. return accountID, err
  130. }
  131. func (s *OpenAIGatewayService) setStickySessionAccountID(ctx context.Context, groupID *int64, sessionHash string, accountID int64, ttl time.Duration) error {
  132. if s == nil || s.cache == nil || accountID <= 0 {
  133. return nil
  134. }
  135. primaryKey := s.openAISessionCacheKey(sessionHash)
  136. if primaryKey == "" {
  137. return nil
  138. }
  139. if err := s.cache.SetSessionAccountID(ctx, derefGroupID(groupID), primaryKey, accountID, ttl); err != nil {
  140. return err
  141. }
  142. if !s.openAISessionHashDualWriteOldEnabled() {
  143. return nil
  144. }
  145. legacyKey := s.openAILegacySessionCacheKey(ctx, sessionHash)
  146. if legacyKey == "" {
  147. return nil
  148. }
  149. if err := s.cache.SetSessionAccountID(ctx, derefGroupID(groupID), legacyKey, accountID, s.openAIStickyLegacyTTL(ttl)); err != nil {
  150. return err
  151. }
  152. openAIStickyLegacyDualWriteTotal.Add(1)
  153. return nil
  154. }
  155. func (s *OpenAIGatewayService) refreshStickySessionTTL(ctx context.Context, groupID *int64, sessionHash string, ttl time.Duration) error {
  156. if s == nil || s.cache == nil {
  157. return nil
  158. }
  159. primaryKey := s.openAISessionCacheKey(sessionHash)
  160. if primaryKey == "" {
  161. return nil
  162. }
  163. err := s.cache.RefreshSessionTTL(ctx, derefGroupID(groupID), primaryKey, ttl)
  164. if !s.openAISessionHashReadOldFallbackEnabled() && !s.openAISessionHashDualWriteOldEnabled() {
  165. return err
  166. }
  167. legacyKey := s.openAILegacySessionCacheKey(ctx, sessionHash)
  168. if legacyKey != "" {
  169. _ = s.cache.RefreshSessionTTL(ctx, derefGroupID(groupID), legacyKey, s.openAIStickyLegacyTTL(ttl))
  170. }
  171. return err
  172. }
  173. func (s *OpenAIGatewayService) deleteStickySessionAccountID(ctx context.Context, groupID *int64, sessionHash string) error {
  174. if s == nil || s.cache == nil {
  175. return nil
  176. }
  177. primaryKey := s.openAISessionCacheKey(sessionHash)
  178. if primaryKey == "" {
  179. return nil
  180. }
  181. err := s.cache.DeleteSessionAccountID(ctx, derefGroupID(groupID), primaryKey)
  182. if !s.openAISessionHashReadOldFallbackEnabled() && !s.openAISessionHashDualWriteOldEnabled() {
  183. return err
  184. }
  185. legacyKey := s.openAILegacySessionCacheKey(ctx, sessionHash)
  186. if legacyKey != "" {
  187. _ = s.cache.DeleteSessionAccountID(ctx, derefGroupID(groupID), legacyKey)
  188. }
  189. return err
  190. }