opencode/yao
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Refactor file, text, and URL addition processes to improve request handling and validation

Browse source 
- 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.
This commit is contained in:
Max 2025-08-14 14:51:48 +08:00
parent 2ebf73fce0
commit 3bef5a6222
4 changed files with 172 additions and 121 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

35
openapi/kb/addfile.go
View file

@ -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,
})
}

95
openapi/kb/addtext.go
View file

@ -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,
})
}

131
openapi/kb/addurl.go
View file

@ -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,
})
}

32
openapi/kb/document.go
View file

@ -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})
}