feat(go-api) sessions message update (#16517)

### Summary ``` /api/v1/chats/<chat_id>/sessions/<session_id>/messages/<msg_id> DELETE /api/v1/chats/<chat_id>/sessions/<session_id>/messages/<msg_id>/feedback PUT ``` Migrates the chat session message delete and feedback APIs to the Go server, matching the Python behavior for authorization, session ownership checks, message/reference updates, and feedback validation.
2026-07-20 22:51:06 +08:00 · 2026-07-02 10:33:27 +08:00
parent 5bc4753d1e
commit 0de69e5bba
7 changed files with 1657 additions and 30 deletions
--- a/internal/engine/elasticsearch/chunk.go
+++ b/internal/engine/elasticsearch/chunk.go
@@ -271,6 +271,93 @@ func (e *elasticsearchEngine) UpdateChunks(ctx context.Context, condition map[st
 	return e.updateChunksByQuery(ctx, fullIndexName, condition, newValue)
 }

+// AdjustChunkPagerank atomically adjusts pagerank_fea and clamps it to
+// [minWeight, maxWeight].
+func (e *elasticsearchEngine) AdjustChunkPagerank(ctx context.Context, indexName, chunkID, kbID string, delta, minWeight, maxWeight float64) error {
+	if indexName == "" {
+		return fmt.Errorf("index name cannot be empty")
+	}
+	if chunkID == "" {
+		return fmt.Errorf("chunk id cannot be empty")
+	}
+	script := `
+		if (ctx._source.kb_id == null || !ctx._source.kb_id.equals(params.kb_id)) {
+			ctx.op = 'noop';
+		} else {
+			double current = 0.0;
+			if (ctx._source.containsKey(params.field) && ctx._source[params.field] != null) {
+				Object currentValue = ctx._source[params.field];
+				if (currentValue instanceof Number) {
+					current = ((Number)currentValue).doubleValue();
+				} else {
+					try {
+						current = Double.parseDouble(currentValue.toString());
+					} catch (Exception e) {
+						current = 0.0;
+					}
+				}
+			}
+			double next = current + params.delta;
+			if (next < params.min_weight) {
+				next = params.min_weight;
+			}
+			if (next > params.max_weight) {
+				next = params.max_weight;
+			}
+			if (next <= 0.0) {
+				ctx._source.remove(params.field);
+			} else {
+				ctx._source[params.field] = next;
+			}
+		}
+	`
+	body, err := json.Marshal(map[string]interface{}{
+		"script": map[string]interface{}{
+			"source": script,
+			"lang":   "painless",
+			"params": map[string]interface{}{
+				"field":      common.PAGERANK_FLD,
+				"kb_id":      kbID,
+				"delta":      delta,
+				"min_weight": minWeight,
+				"max_weight": maxWeight,
+			},
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("failed to marshal pagerank adjust request: %w", err)
+	}
+	retryOnConflict := 3
+	req := esapi.UpdateRequest{
+		Index:           indexName,
+		DocumentID:      chunkID,
+		Body:            bytes.NewReader(body),
+		RetryOnConflict: &retryOnConflict,
+	}
+	res, err := req.Do(ctx, e.client)
+	if err != nil {
+		return fmt.Errorf("failed to adjust chunk pagerank: %w", err)
+	}
+	defer res.Body.Close()
+	if res.IsError() {
+		if res.StatusCode == http.StatusNotFound {
+			return fmt.Errorf("%w: %s", types.ErrDocumentNotFound, chunkID)
+		}
+		bodyBytes, _ := io.ReadAll(res.Body)
+		return fmt.Errorf("elasticsearch pagerank adjust error: %s, body: %s", res.Status(), string(bodyBytes))
+	}
+	var updateResp struct {
+		Result string `json:"result"`
+	}
+	if err := json.NewDecoder(res.Body).Decode(&updateResp); err != nil {
+		return fmt.Errorf("failed to decode pagerank adjust response: %w", err)
+	}
+	if updateResp.Result == "noop" {
+		return fmt.Errorf("chunk %s does not belong to dataset %s", chunkID, kbID)
+	}
+	return nil
+}
+
 func (e *elasticsearchEngine) updateSingleMemoryMessage(ctx context.Context, indexName, messageDocID string, newValue map[string]interface{}) error {
 	doc := mapMemoryMessageESUpdateFields(newValue)
 	delete(doc, "id")
--- a/internal/engine/infinity/chunk.go
+++ b/internal/engine/infinity/chunk.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"hash/fnv"
 	"os"
 	"path/filepath"
 	"ragflow/internal/common"
@@ -30,6 +31,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"

 	infinity "github.com/infiniflow/infinity-go-sdk"
 	"go.uber.org/zap"
@@ -38,6 +40,13 @@ import (
 // ChinesePunctRegex splits on comma, semicolon, Chinese punctuations, and newlines
 var ChinesePunctRegex = regexp.MustCompile(`[,，;；、\r\n]+`)

+const (
+	pagerankAdjustRetryCount = 2
+	pagerankAdjustLockCount  = 256
+)
+
+var pagerankAdjustLocks [pagerankAdjustLockCount]sync.Mutex
+
 // CreateChunkStore creates a chunk table in Infinity
 // baseName is the table name prefix (e.g., "ragflow_<tenant_id>")
 // The full table name is built as "{baseName}_{datasetID}"
@@ -518,6 +527,97 @@ func (e *infinityEngine) UpdateChunks(ctx context.Context, condition map[string]
 	return nil
 }

+// AdjustChunkPagerank adjusts pagerank_fea and clamps it to [minWeight, maxWeight].
+func (e *infinityEngine) AdjustChunkPagerank(ctx context.Context, baseName, chunkID, datasetID string, delta, minWeight, maxWeight float64) error {
+	if baseName == "" {
+		return fmt.Errorf("index name cannot be empty")
+	}
+	if chunkID == "" {
+		return fmt.Errorf("chunk id cannot be empty")
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	if e.client == nil || e.client.conn == nil {
+		return fmt.Errorf("Infinity client not initialized")
+	}
+
+	tableName := buildChunkTableName(baseName, datasetID)
+	lock := pagerankAdjustLock(tableName + ":" + chunkID)
+	lock.Lock()
+	defer lock.Unlock()
+
+	db, err := e.client.conn.GetDatabase(e.client.dbName)
+	if err != nil {
+		return fmt.Errorf("failed to get database: %w", err)
+	}
+	table, err := db.GetTable(tableName)
+	if err != nil {
+		return fmt.Errorf("failed to get table %s: %w", tableName, err)
+	}
+
+	var lastErr error
+	filter := fmt.Sprintf("id = '%s'", escapeFilterValue(chunkID))
+	for attempt := 0; attempt <= pagerankAdjustRetryCount; attempt++ {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		result, err := table.Output([]string{common.PAGERANK_FLD, "row_id()"}).Filter(filter).ToResult()
+		if err != nil {
+			lastErr = err
+			continue
+		}
+		qr, ok := result.(*infinity.QueryResult)
+		if !ok {
+			return fmt.Errorf("unexpected query result type: %T", result)
+		}
+
+		rowID, ok := firstQueryInt64(qr.Data, "ROW_ID", "row_id()", "row_id")
+		if !ok {
+			return fmt.Errorf("%w: %s", types.ErrDocumentNotFound, chunkID)
+		}
+
+		currentWeight := 0.0
+		if currentValue, ok := firstQueryValue(qr.Data, common.PAGERANK_FLD); ok {
+			if weight, ok := coerceToFloat(currentValue); ok {
+				currentWeight = weight
+			}
+		}
+		nextWeight := currentWeight + delta
+		if nextWeight < minWeight {
+			nextWeight = minWeight
+		}
+		if nextWeight > maxWeight {
+			nextWeight = maxWeight
+		}
+		if floatsEqual(currentWeight, nextWeight) {
+			return nil
+		}
+
+		updateFilter := fmt.Sprintf("_row_id = %d AND %s = %s", rowID, common.PAGERANK_FLD, formatFilterFloat(currentWeight))
+		if _, err := table.Update(updateFilter, map[string]interface{}{common.PAGERANK_FLD: nextWeight}); err != nil {
+			lastErr = err
+			continue
+		}
+		verifyResult, err := table.Output([]string{common.PAGERANK_FLD}).Filter(filter).ToResult()
+		if err != nil {
+			lastErr = err
+			continue
+		}
+		verifyQR, ok := verifyResult.(*infinity.QueryResult)
+		if !ok {
+			return fmt.Errorf("unexpected query result type: %T", verifyResult)
+		}
+		if currentValue, ok := firstQueryValue(verifyQR.Data, common.PAGERANK_FLD); ok {
+			if actualWeight, ok := coerceToFloat(currentValue); ok && floatsEqual(actualWeight, nextWeight) {
+				return nil
+			}
+		}
+		lastErr = fmt.Errorf("pagerank update conflict")
+	}
+	return fmt.Errorf("failed to adjust chunk pagerank: %w", lastErr)
+}
+
 // DeleteChunks deletes chunks from a dataset table
 // Table name format: {baseName}_{datasetID}
 // condition specifies which chunks to delete
@@ -1976,6 +2076,77 @@ func escapeFilterValue(s string) string {
 	return strings.ReplaceAll(s, "'", "''")
 }

+func firstQueryValue(data map[string][]interface{}, columns ...string) (interface{}, bool) {
+	for _, column := range columns {
+		values, ok := data[column]
+		if ok && len(values) > 0 {
+			return values[0], true
+		}
+	}
+	return nil, false
+}
+
+func firstQueryInt64(data map[string][]interface{}, columns ...string) (int64, bool) {
+	value, ok := firstQueryValue(data, columns...)
+	if !ok {
+		return 0, false
+	}
+	switch v := value.(type) {
+	case int:
+		return int64(v), true
+	case int8:
+		return int64(v), true
+	case int16:
+		return int64(v), true
+	case int32:
+		return int64(v), true
+	case int64:
+		return v, true
+	case uint:
+		if uint64(v) <= ^uint64(0)>>1 {
+			return int64(v), true
+		}
+	case uint8:
+		return int64(v), true
+	case uint16:
+		return int64(v), true
+	case uint32:
+		return int64(v), true
+	case uint64:
+		if v <= ^uint64(0)>>1 {
+			return int64(v), true
+		}
+	case float32:
+		return int64(v), true
+	case float64:
+		return int64(v), true
+	case string:
+		parsed, err := strconv.ParseInt(v, 10, 64)
+		if err == nil {
+			return parsed, true
+		}
+	}
+	return 0, false
+}
+
+func pagerankAdjustLock(key string) *sync.Mutex {
+	hash := fnv.New32a()
+	_, _ = hash.Write([]byte(key))
+	return &pagerankAdjustLocks[hash.Sum32()%pagerankAdjustLockCount]
+}
+
+func formatFilterFloat(value float64) string {
+	return strconv.FormatFloat(value, 'f', -1, 64)
+}
+
+func floatsEqual(a, b float64) bool {
+	diff := a - b
+	if diff < 0 {
+		diff = -diff
+	}
+	return diff < 1e-9
+}
+
 // equivalentConditionToStr converts a condition map to an Infinity filter string
 func equivalentConditionToStr(condition map[string]interface{}) string {
 	if len(condition) == 0 {