From c32ceaa920d83f649efd00091b4b168173a530a1 Mon Sep 17 00:00:00 2001
From: ymq <ymq@opencomputing.net>
Date: Sat, 12 Sep 2026 08:04:17 +0800
Subject: [PATCH] =?UTF-8?q?fix(rag=5Fclient):=20params=E8=AF=BB=E5=86=99?=
 =?UTF-8?q?=E5=88=97=E4=B8=8D=E4=B8=80=E8=87=B4=E8=87=B4=E9=87=8D=E5=A4=8D?=
 =?UTF-8?q?=E8=A1=8C=E2=80=94=E2=80=94=5Fload=5Fcached=5Fkey=E6=8C=89id?=
 =?UTF-8?q?=E6=9F=A5=E4=BD=86=E5=86=99=E5=85=A5id=3DgetID()=E9=9A=8F?=
 =?UTF-8?q?=E6=9C=BA=E5=80=BC,=E7=BC=93=E5=AD=98=E6=B0=B8=E8=BF=9Cmiss?=
 =?UTF-8?q?=E2=86=92=E6=AF=8F=E6=AC=A1=E5=8F=91key=E6=96=B0=E5=A2=9E?=
 =?UTF-8?q?=E4=B8=80=E8=A1=8C(=E5=AE=9E=E6=B5=8Buser-01=E5=A0=8612?=
 =?UTF-8?q?=E8=A1=8C);ON=20DUPLICATE=20KEY=20UPDATE=E5=9B=A0params=5Fname?=
 =?UTF-8?q?=E6=97=A0=E5=94=AF=E4=B8=80=E7=B4=A2=E5=BC=95=E6=B0=B8=E4=B8=8D?=
 =?UTF-8?q?=E8=A7=A6=E5=8F=91=E3=80=82=E4=BF=AE:=E8=AF=BB=E6=94=B9?=
 =?UTF-8?q?=E6=8C=89params=5Fname+=E8=AF=BB-=E6=94=B9-=E5=86=99=E5=B9=82?=
 =?UTF-8?q?=E7=AD=89(=E5=AD=98=E5=9C=A8=E5=88=99=E6=9B=B4=E6=96=B0?=
 =?UTF-8?q?=E9=A6=96=E8=A1=8C=E5=88=A0=E5=A4=9A=E4=BD=99=E8=A1=8C)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 pipeline_service/rag_client.py | 25 ++++++++++++++++++++-----
 1 file changed, 20 insertions(+), 5 deletions(-)

diff --git a/pipeline_service/rag_client.py b/pipeline_service/rag_client.py
index fbd45d9..767e8c4 100644
--- a/pipeline_service/rag_client.py
+++ b/pipeline_service/rag_client.py
@@ -119,7 +119,7 @@ async def _ensure_downapp(sor):
 async def _load_cached_key(sor, user_id):
     pname = _KEY_PARAM_PREFIX + str(user_id)
     recs = await sor.sqlExe(
-        "SELECT params_value FROM params WHERE id=${n}$", {"n": pname})
+        "SELECT params_value FROM params WHERE params_name=${n}$ LIMIT 1", {"n": pname})
     await sor.sqlExe("COMMIT", {})
     if not recs:
         return ""
@@ -160,10 +160,25 @@ async def get_owner_apikey(owner_id):
         # RC4 加密存 params（与平台密码字段约定一致）
         pname = _KEY_PARAM_PREFIX + str(owner_id)
         enc = _key_encode(key)
-        await sor.sqlExe(
-            "INSERT INTO params (id, params_name, params_value) VALUES (${i}$, ${n}$, ${v}$) "
-            "ON DUPLICATE KEY UPDATE params_value=${v}$",
-            {"i": getID(), "n": pname, "v": enc})
+        # 幂等按 params_name 读-改-写：params 表 params_name 无唯一索引，
+        # 不能用 INSERT...ON DUPLICATE KEY UPDATE（永不触发→重复插行）。
+        # 旧 bug：_load_cached_key 按 id 查、这里按 getID() 随机 id 写，缓存永远 miss
+        # →每次发 key 都新增一行（实测 user-01 堆了 12 行）。2026-09-12 修。
+        exists = await sor.sqlExe(
+            "SELECT id FROM params WHERE params_name=${n}$ ORDER BY id LIMIT 1", {"n": pname})
+        await sor.sqlExe("COMMIT", {})
+        if exists:
+            # 收敛历史重复行：更新首行，删多余行
+            await sor.sqlExe(
+                "UPDATE params SET params_value=${v}$ WHERE id=${i}$",
+                {"v": enc, "i": exists[0].id})
+            await sor.sqlExe(
+                "DELETE FROM params WHERE params_name=${n}$ AND id<>${i}$",
+                {"n": pname, "i": exists[0].id})
+        else:
+            await sor.sqlExe(
+                "INSERT INTO params (id, params_name, params_value) VALUES (${i}$, ${n}$, ${v}$)",
+                {"i": getID(), "n": pname, "v": enc})
         await sor.sqlExe("COMMIT", {})
         _key_cache[owner_id] = key
         return key, ""