From 3bef5a6222aa4edc793c70648bccb2e79fd08f94 Mon Sep 17 00:00:00 2001 From: Max Date: Thu, 14 Aug 2025 14:51:48 +0800 Subject: [PATCH] Refactor file, text, and URL addition processes to improve request handling and validation - Updated AddFileProcess, AddTextProcess, and AddURLProcess functions to accept an optional job ID for async operations. - Enhanced error handling by requiring document IDs to be provided by the caller before processing. - Streamlined document creation and upsert operations, ensuring better validation and rollback mechanisms. - Implemented async processing capabilities in AddFileAsync, AddTextAsync, and AddURLAsync functions, returning job IDs and document IDs in responses. - Improved code maintainability and readability by encapsulating business logic and reducing dependencies on Gin context. --- openapi/kb/addfile.go | 35 ++++++++--- openapi/kb/addtext.go | 95 +++++++++++------------------- openapi/kb/addurl.go | 131 ++++++++++++++++++++++++++++------------- openapi/kb/document.go | 32 ++++++---- 4 files changed, 172 insertions(+), 121 deletions(-) diff --git a/openapi/kb/addfile.go b/openapi/kb/addfile.go index d4ec337c..a795a498 100644 --- a/openapi/kb/addfile.go +++ b/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 - // Use context.Background() for async operations to avoid Gin context expiration - handleAsyncWithRequest(c, &req, func(ctx context.Context, r *AddFileRequest) { - err := AddFileProcess(ctx, r) + // Generate document ID if not provided + if req.DocID == "" { + req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID) + } + + // 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, + }) } diff --git a/openapi/kb/addtext.go b/openapi/kb/addtext.go index a054fdd7..e253e2aa 100644 --- a/openapi/kb/addtext.go +++ b/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 - _, documentData, err := PrepareAddText(c, req) + // Use the business logic function + 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 - handleAsyncWithRequest(c, &req, func(ctx context.Context, r *AddTextRequest) { - err := ProcessAddTextRequest(ctx, r) + // Generate document ID if not provided + if req.DocID == "" { + 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, + }) } diff --git a/openapi/kb/addurl.go b/openapi/kb/addurl.go index c5da3db5..f9661cfb 100644 --- a/openapi/kb/addurl.go +++ b/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 -func addURLWithRequest(c *gin.Context, req *AddURLRequest) { - // Prepare request and database data - _, documentData, err := PrepareAddURL(c, req) - if err != nil { - errorResp := &response.ErrorResponse{ - Code: response.ErrInvalidRequest.Code, - ErrorDescription: err.Error(), - } - response.RespondWithError(c, response.StatusBadRequest, errorResp) - return +// AddURLProcess processes a URL addition request with business logic only +// This function is Gin-agnostic and can be used for both sync and async operations +func AddURLProcess(ctx context.Context, req *AddURLRequest, jobID ...string) error { + // Check if kb.Instance is available + if kb.Instance == nil { + return fmt.Errorf("knowledge base not initialized") + } + + // Validate request + if err := req.Validate(); err != nil { + 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{ - Code: response.ErrServerError.Code, - ErrorDescription: "Failed to get KB config: " + err.Error(), - } - response.RespondWithError(c, response.StatusInternalServerError, errorResp) - return + return fmt.Errorf("failed to get KB config: %w", err) } + // 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{ - Code: response.ErrServerError.Code, - ErrorDescription: "Failed to save document metadata: " + err.Error(), - } - response.RespondWithError(c, response.StatusInternalServerError, errorResp) - return + return fmt.Errorf("failed to save document metadata: %w", err) } // 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 { - log.Error("Failed to rollback document database record: %v", err) + if rollbackErr := config.RemoveDocument(req.DocID); rollbackErr != nil { + 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()}) - - errorResp := &response.ErrorResponse{ - Code: response.ErrServerError.Code, - ErrorDescription: "Failed to add URL: " + err.Error(), - } - response.RespondWithError(c, response.StatusInternalServerError, errorResp) - return + return fmt.Errorf("failed to add URL: %w", err) } // 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 - handleAsyncWithRequest(c, &req, func(ctx context.Context, r *AddURLRequest) { - // Temporary placeholder - would need ProcessAddURLRequest function - log.Info("Async URL processing placeholder for: %s", r.URL) + // Generate document ID if not provided + if req.DocID == "" { + req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID) + } + + // 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, }) } diff --git a/openapi/kb/document.go b/openapi/kb/document.go index 9bb62183..bc3caee7 100644 --- a/openapi/kb/document.go +++ b/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}) -}