Merge pull request #1109 from trheyi/main

Refactor file, text, and URL addition processes to improve request ha…

openapi/kb/addfile.go

@@ -15,7 +15,7 @@ import (
 
 // AddFileProcess processes a file addition request with business logic only
 // This function is Gin-agnostic and can be used for both sync and async operations
-func AddFileProcess(ctx context.Context, req *AddFileRequest) error {
+func AddFileProcess(ctx context.Context, req *AddFileRequest, jobID ...string) error {
 	// Check if kb.Instance is available
 	if kb.Instance == nil {
 		return fmt.Errorf("knowledge base not initialized")
@@ -49,9 +49,9 @@ func AddFileProcess(ctx context.Context, req *AddFileRequest) error {
 		return fmt.Errorf("failed to get file info: %w", err)
 	}
 
-	// Generate document ID if not provided
+	// DocID should be generated by the caller before calling this function
 	if req.DocID == "" {
-		req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID)
+		return fmt.Errorf("document ID is required")
 	}
 
 	// Get KB config
@@ -74,6 +74,11 @@ func AddFileProcess(ctx context.Context, req *AddFileRequest) error {
 		"size":           int64(fileInfo.Bytes),
 	}
 
+	// Add job_id if provided (for async operations)
+	if len(jobID) > 0 && jobID[0] != "" {
+		documentData["job_id"] = jobID[0]
+	}
+
 	// Add base request fields
 	req.BaseUpsertRequest.AddBaseFields(documentData)
 
@@ -162,6 +167,11 @@ func AddFile(c *gin.Context) {
 		return
 	}
 
+	// Generate document ID if not provided
+	if req.DocID == "" {
+		req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID)
+	}
+
 	// Process the request
 	addFileWithRequest(c, &req)
 }
@@ -207,12 +217,23 @@ func AddFileAsync(c *gin.Context) {
 		return
 	}
 
-	// Handle async processing with parsed request
+	// Generate document ID if not provided
-	// Use context.Background() for async operations to avoid Gin context expiration
+	if req.DocID == "" {
-	handleAsyncWithRequest(c, &req, func(ctx context.Context, r *AddFileRequest) {
+		req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID)
-		err := AddFileProcess(ctx, r)
+	}
+
+	// Create and run job
+	job := NewJob()
+	jobID := job.Run(func() {
+		err := AddFileProcess(context.Background(), &req, job.ID)
 		if err != nil {
 			log.Error("Async file processing failed: %v", err)
 		}
 	})
+
+	// Return job_id and doc_id
+	response.RespondWithSuccess(c, response.StatusCreated, gin.H{
+		"job_id": jobID,
+		"doc_id": req.DocID,
+	})
 }

openapi/kb/addtext.go

@@ -12,9 +12,9 @@ import (
 	"github.com/yaoapp/yao/openapi/response"
 )
 
-// ProcessAddTextRequest processes a text addition request with business logic only
+// AddTextProcess processes a text addition request with business logic only
 // This function is Gin-agnostic and can be used for both sync and async operations
-func ProcessAddTextRequest(ctx context.Context, req *AddTextRequest) error {
+func AddTextProcess(ctx context.Context, req *AddTextRequest, jobID ...string) error {
 	// Check if kb.Instance is available
 	if kb.Instance == nil {
 		return fmt.Errorf("knowledge base not initialized")
@@ -25,9 +25,9 @@ func ProcessAddTextRequest(ctx context.Context, req *AddTextRequest) error {
 		return err
 	}
 
-	// Generate document ID if not provided
+	// DocID should be generated by the caller before calling this function
 	if req.DocID == "" {
-		req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID)
+		return fmt.Errorf("document ID is required")
 	}
 
 	// Get KB config
@@ -47,6 +47,11 @@ func ProcessAddTextRequest(ctx context.Context, req *AddTextRequest) error {
 		"size":          int64(len(req.Text)),
 	}
 
+	// Add job_id if provided (for async operations)
+	if len(jobID) > 0 && jobID[0] != "" {
+		documentData["job_id"] = jobID[0]
+	}
+
 	// Use title from metadata if available
 	if req.Metadata != nil {
 		if title, ok := req.Metadata["title"].(string); ok && title != "" {
@@ -89,70 +94,19 @@ func ProcessAddTextRequest(ctx context.Context, req *AddTextRequest) error {
 	return nil
 }
 
-// addTextWithRequest processes a text addition with pre-parsed request
+// addTextWithRequest processes a text addition with pre-parsed request using Gin context
 func addTextWithRequest(c *gin.Context, req *AddTextRequest) {
-	// Prepare request and database data
+	// Use the business logic function
-	_, documentData, err := PrepareAddText(c, req)
+	err := AddTextProcess(c.Request.Context(), req)
 	if err != nil {
 		errorResp := &response.ErrorResponse{
-			Code:             response.ErrInvalidRequest.Code,
+			Code:             response.ErrServerError.Code,
 			ErrorDescription: err.Error(),
 		}
-		response.RespondWithError(c, response.StatusBadRequest, errorResp)
-		return
-	}
-
-	// Get KB config
-	config, err := kb.GetConfig()
-	if err != nil {
-		errorResp := &response.ErrorResponse{
-			Code:             response.ErrServerError.Code,
-			ErrorDescription: "Failed to get KB config: " + err.Error(),
-		}
 		response.RespondWithError(c, response.StatusInternalServerError, errorResp)
 		return
 	}
 
-	// First create database record
-	_, err = config.CreateDocument(maps.MapStrAny(documentData))
-	if err != nil {
-		errorResp := &response.ErrorResponse{
-			Code:             response.ErrServerError.Code,
-			ErrorDescription: "Failed to save document metadata: " + err.Error(),
-		}
-		response.RespondWithError(c, response.StatusInternalServerError, errorResp)
-		return
-	}
-
-	// Convert request to UpsertOptions
-	upsertOptions, err := getUpsertOptions(c, &req.BaseUpsertRequest)
-	if err != nil {
-		// Rollback: remove the database record
-		if err := config.RemoveDocument(req.DocID); err != nil {
-			log.Error("Failed to rollback document database record: %v", err)
-		}
-		return
-	}
-
-	// Perform upsert operation with text
-	_, err = kb.Instance.AddText(c.Request.Context(), req.Text, upsertOptions)
-	if err != nil {
-		// Update status to error and return error response
-		config.UpdateDocument(req.DocID, maps.MapStrAny{"status": "error", "error_message": err.Error()})
-
-		errorResp := &response.ErrorResponse{
-			Code:             response.ErrServerError.Code,
-			ErrorDescription: "Failed to add text: " + err.Error(),
-		}
-		response.RespondWithError(c, response.StatusInternalServerError, errorResp)
-		return
-	}
-
-	// Update status to completed after successful processing
-	if err := config.UpdateDocument(req.DocID, maps.MapStrAny{"status": "completed"}); err != nil {
-		log.Error("Failed to update document status to completed: %v", err)
-	}
-
 	// Return success response
 	result := gin.H{
 		"message":       "Text added successfully",
@@ -192,6 +146,11 @@ func AddText(c *gin.Context) {
 		return
 	}
 
+	// Generate document ID if not provided
+	if req.DocID == "" {
+		req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID)
+	}
+
 	// Process the request
 	addTextWithRequest(c, &req)
 }
@@ -231,11 +190,23 @@ func AddTextAsync(c *gin.Context) {
 		return
 	}
 
-	// Handle async processing with parsed request
+	// Generate document ID if not provided
-	handleAsyncWithRequest(c, &req, func(ctx context.Context, r *AddTextRequest) {
+	if req.DocID == "" {
-		err := ProcessAddTextRequest(ctx, r)
+		req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID)
+	}
+
+	// Create and run job
+	job := NewJob()
+	jobID := job.Run(func() {
+		err := AddTextProcess(context.Background(), &req, job.ID)
 		if err != nil {
 			log.Error("Async text processing failed: %v", err)
 		}
 	})
+
+	// Return job_id and doc_id
+	response.RespondWithSuccess(c, response.StatusCreated, gin.H{
+		"job_id": jobID,
+		"doc_id": req.DocID,
+	})
 }

openapi/kb/addurl.go

@@ -2,71 +2,87 @@ package kb
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/gin-gonic/gin"
+	"github.com/yaoapp/gou/graphrag/utils"
 	"github.com/yaoapp/kun/log"
 	"github.com/yaoapp/kun/maps"
 	"github.com/yaoapp/yao/kb"
 	"github.com/yaoapp/yao/openapi/response"
 )
 
-// addURLWithRequest processes a URL addition with pre-parsed request
+// AddURLProcess processes a URL addition request with business logic only
-func addURLWithRequest(c *gin.Context, req *AddURLRequest) {
+// This function is Gin-agnostic and can be used for both sync and async operations
-	// Prepare request and database data
+func AddURLProcess(ctx context.Context, req *AddURLRequest, jobID ...string) error {
-	_, documentData, err := PrepareAddURL(c, req)
+	// Check if kb.Instance is available
-	if err != nil {
+	if kb.Instance == nil {
-		errorResp := &response.ErrorResponse{
+		return fmt.Errorf("knowledge base not initialized")
-			Code:             response.ErrInvalidRequest.Code,
+	}
-			ErrorDescription: err.Error(),
+
-		}
+	// Validate request
-		response.RespondWithError(c, response.StatusBadRequest, errorResp)
+	if err := req.Validate(); err != nil {
-		return
+		return err
+	}
+
+	// DocID should be generated by the caller before calling this function
+	if req.DocID == "" {
+		return fmt.Errorf("document ID is required")
 	}
 
 	// Get KB config
 	config, err := kb.GetConfig()
 	if err != nil {
-		errorResp := &response.ErrorResponse{
+		return fmt.Errorf("failed to get KB config: %w", err)
-			Code:             response.ErrServerError.Code,
-			ErrorDescription: "Failed to get KB config: " + err.Error(),
-		}
-		response.RespondWithError(c, response.StatusInternalServerError, errorResp)
-		return
 	}
 
+	// Prepare document data for database
+	documentData := map[string]interface{}{
+		"document_id":   req.DocID,
+		"collection_id": req.CollectionID,
+		"name":          "URL Document",
+		"type":          "url",
+		"status":        "pending",
+		"url":           req.URL,
+	}
+
+	// Add job_id if provided (for async operations)
+	if len(jobID) > 0 && jobID[0] != "" {
+		documentData["job_id"] = jobID[0]
+	}
+
+	// Use title from metadata if available
+	if req.Metadata != nil {
+		if title, ok := req.Metadata["title"].(string); ok && title != "" {
+			documentData["name"] = title
+		}
+	}
+
+	// Add base request fields
+	req.BaseUpsertRequest.AddBaseFields(documentData)
+
 	// First create database record
 	_, err = config.CreateDocument(maps.MapStrAny(documentData))
 	if err != nil {
-		errorResp := &response.ErrorResponse{
+		return fmt.Errorf("failed to save document metadata: %w", err)
-			Code:             response.ErrServerError.Code,
-			ErrorDescription: "Failed to save document metadata: " + err.Error(),
-		}
-		response.RespondWithError(c, response.StatusInternalServerError, errorResp)
-		return
 	}
 
 	// Convert request to UpsertOptions
-	upsertOptions, err := getUpsertOptions(c, &req.BaseUpsertRequest)
+	upsertOptions, err := req.BaseUpsertRequest.ToUpsertOptions()
 	if err != nil {
 		// Rollback: remove the database record
-		if err := config.RemoveDocument(req.DocID); err != nil {
+		if rollbackErr := config.RemoveDocument(req.DocID); rollbackErr != nil {
-			log.Error("Failed to rollback document database record: %v", err)
+			log.Error("Failed to rollback document database record: %v", rollbackErr)
 		}
-		return
+		return fmt.Errorf("failed to convert request to upsert options: %w", err)
 	}
 
 	// Perform upsert operation with URL
-	_, err = kb.Instance.AddURL(c.Request.Context(), req.URL, upsertOptions)
+	_, err = kb.Instance.AddURL(ctx, req.URL, upsertOptions)
 	if err != nil {
-		// Update status to error and return error response
+		// Update status to error
 		config.UpdateDocument(req.DocID, maps.MapStrAny{"status": "error", "error_message": err.Error()})
-
+		return fmt.Errorf("failed to add URL: %w", err)
-		errorResp := &response.ErrorResponse{
-			Code:             response.ErrServerError.Code,
-			ErrorDescription: "Failed to add URL: " + err.Error(),
-		}
-		response.RespondWithError(c, response.StatusInternalServerError, errorResp)
-		return
 	}
 
 	// Update status to completed after successful processing
@@ -74,6 +90,22 @@ func addURLWithRequest(c *gin.Context, req *AddURLRequest) {
 		log.Error("Failed to update document status to completed: %v", err)
 	}
 
+	return nil
+}
+
+// addURLWithRequest processes a URL addition with pre-parsed request using Gin context
+func addURLWithRequest(c *gin.Context, req *AddURLRequest) {
+	// Use the business logic function
+	err := AddURLProcess(c.Request.Context(), req)
+	if err != nil {
+		errorResp := &response.ErrorResponse{
+			Code:             response.ErrServerError.Code,
+			ErrorDescription: err.Error(),
+		}
+		response.RespondWithError(c, response.StatusInternalServerError, errorResp)
+		return
+	}
+
 	// Return success response
 	result := gin.H{
 		"message":       "URL added successfully",
@@ -114,6 +146,11 @@ func AddURL(c *gin.Context) {
 		return
 	}
 
+	// Generate document ID if not provided
+	if req.DocID == "" {
+		req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID)
+	}
+
 	// Process the request
 	addURLWithRequest(c, &req)
 }
@@ -153,9 +190,23 @@ func AddURLAsync(c *gin.Context) {
 		return
 	}
 
-	// Handle async processing with parsed request
+	// Generate document ID if not provided
-	handleAsyncWithRequest(c, &req, func(ctx context.Context, r *AddURLRequest) {
+	if req.DocID == "" {
-		// Temporary placeholder - would need ProcessAddURLRequest function
+		req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID)
-		log.Info("Async URL processing placeholder for: %s", r.URL)
+	}
+
+	// Create and run job
+	job := NewJob()
+	jobID := job.Run(func() {
+		err := AddURLProcess(context.Background(), &req, job.ID)
+		if err != nil {
+			log.Error("Async URL processing failed: %v", err)
+		}
+	})
+
+	// Return job_id and doc_id
+	response.RespondWithSuccess(c, response.StatusCreated, gin.H{
+		"job_id": jobID,
+		"doc_id": req.DocID,
 	})
 }

openapi/kb/document.go

@@ -1,7 +1,6 @@
 package kb
 
 import (
-	"context"
 	"net/http"
 
 	"github.com/gin-gonic/gin"
@@ -12,6 +11,26 @@ import (
 	"github.com/yaoapp/yao/openapi/response"
 )
 
+// SimpleJob represents a simple job for async operations
+// TODO: replace with proper job system later
+type SimpleJob struct {
+	ID string
+}
+
+// NewJob creates a new simple job
+func NewJob() *SimpleJob {
+	return &SimpleJob{
+		ID: uuid.New().String(),
+	}
+}
+
+// Run executes the job function asynchronously and returns job ID
+func (j *SimpleJob) Run(fn func()) string {
+	// temporary solution to handle async operations ( TODO: use job queue )
+	go fn()
+	return j.ID
+}
+
 // Document Management Handlers
 
 // ListDocuments lists documents with pagination
@@ -132,14 +151,3 @@ func validateFileAndGetPath(c *gin.Context, req *AddFileRequest) (string, string
 
 	return path, contentType, nil
 }
-
-// handleAsyncWithRequest handles async processing for handlers that need parsed request data
-func handleAsyncWithRequest[T any](c *gin.Context, req T, handler func(context.Context, T)) {
-	jobid := uuid.New().String()
-
-	// temporary solution to handle async operations ( TODO: use job queue )
-	// Use context.Background() to avoid Gin context expiration in async operations
-	go func() { handler(context.Background(), req) }()
-
-	response.RespondWithSuccess(c, response.StatusCreated, gin.H{"job_id": jobid})
-}