Add attachment status management and filtering in neo package
- Introduced new status field in the attachment table to track processing states: uploading, uploaded, indexing, indexed, upload_failed, and index_failed. - Added progress and error fields to provide detailed information during the attachment processing workflow. - Implemented AttachmentFilter struct to support filtering attachments by status. - Updated SaveAttachment and GetAttachments methods to handle new status and progress information. - Enhanced tests to validate attachment status management and filtering functionality.
This commit is contained in:
Max
parent
49562b8be0
commit
e1f2899723
4 changed files with 242 additions and 1 deletions
|
|
@ -299,6 +299,9 @@ CREATE TABLE neo_attachment (
|
|||
gzip BOOLEAN DEFAULT FALSE INDEX, -- Compression flag
|
||||
bytes BIGINT INDEX, -- File size
|
||||
collection_id VARCHAR(200) INDEX, -- Associated knowledge collection
|
||||
status ENUM('uploading', 'uploaded', 'indexing', 'indexed', 'upload_failed', 'index_failed') DEFAULT 'uploading' INDEX, -- Processing status
|
||||
progress VARCHAR(200), -- Progress information (nullable)
|
||||
error VARCHAR(600), -- Error message (nullable)
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP INDEX,
|
||||
updated_at TIMESTAMP INDEX
|
||||
);
|
||||
|
|
@ -358,6 +361,26 @@ type AssistantFilter struct {
|
|||
}
|
||||
```
|
||||
|
||||
#### AttachmentFilter
|
||||
|
||||
```go
|
||||
type AttachmentFilter struct {
|
||||
UID string `json:"uid,omitempty"` // Filter by user ID
|
||||
Guest *bool `json:"guest,omitempty"` // Filter by guest status
|
||||
Manager string `json:"manager,omitempty"` // Filter by upload manager
|
||||
ContentType string `json:"content_type,omitempty"` // Filter by content type
|
||||
Name string `json:"name,omitempty"` // Filter by filename
|
||||
Public *bool `json:"public,omitempty"` // Filter by public status
|
||||
Gzip *bool `json:"gzip,omitempty"` // Filter by gzip compression
|
||||
CollectionID string `json:"collection_id,omitempty"` // Filter by knowledge collection ID
|
||||
Status string `json:"status,omitempty"` // Filter by processing status
|
||||
Keywords string `json:"keywords,omitempty"` // Search in filename
|
||||
Page int `json:"page,omitempty"` // Page number
|
||||
PageSize int `json:"pagesize,omitempty"` // Items per page
|
||||
Select []string `json:"select,omitempty"` // Fields to return
|
||||
}
|
||||
```
|
||||
|
||||
## Usage Examples
|
||||
|
||||
### 1. Chat Management
|
||||
|
|
@ -439,19 +462,65 @@ attachment := map[string]interface{}{
|
|||
"bytes": 102400,
|
||||
"collection_id": "knowledge456",
|
||||
"scope": []string{"user", "admin"},
|
||||
"status": "uploaded", // Status: uploading, uploaded, indexing, indexed, upload_failed, index_failed
|
||||
"progress": "Upload completed", // Progress information (optional)
|
||||
"error": nil, // Error message (optional, for failed statuses)
|
||||
}
|
||||
fileID, err := store.SaveAttachment(attachment)
|
||||
|
||||
// Update attachment status during processing workflow
|
||||
attachment["status"] = "indexing"
|
||||
attachment["progress"] = "Processing file for indexing..."
|
||||
_, err = store.SaveAttachment(attachment)
|
||||
|
||||
// Handle failed upload
|
||||
attachment["status"] = "upload_failed"
|
||||
attachment["progress"] = nil
|
||||
attachment["error"] = "Network connection timeout"
|
||||
_, err = store.SaveAttachment(attachment)
|
||||
|
||||
// Complete indexing
|
||||
attachment["status"] = "indexed"
|
||||
attachment["progress"] = "File indexed successfully"
|
||||
attachment["error"] = nil
|
||||
_, err = store.SaveAttachment(attachment)
|
||||
|
||||
// Get attachments with filtering
|
||||
filter := AttachmentFilter{
|
||||
UID: "user123",
|
||||
ContentType: "image/jpeg",
|
||||
Status: "indexed", // Filter by status
|
||||
Page: 1,
|
||||
PageSize: 20,
|
||||
}
|
||||
attachments, err := store.GetAttachments(filter)
|
||||
|
||||
// Get all failed uploads
|
||||
failedFilter := AttachmentFilter{
|
||||
UID: "user123",
|
||||
Status: "upload_failed",
|
||||
Page: 1,
|
||||
PageSize: 10,
|
||||
}
|
||||
failedUploads, err := store.GetAttachments(failedFilter)
|
||||
```
|
||||
|
||||
#### Attachment Status Workflow
|
||||
|
||||
The attachment system supports a complete file processing workflow with the following status values:
|
||||
|
||||
- **`uploading`** (default): File upload is in progress
|
||||
- **`uploaded`**: File upload completed successfully
|
||||
- **`indexing`**: File is being processed for search indexing
|
||||
- **`indexed`**: File has been indexed and is ready for use
|
||||
- **`upload_failed`**: File upload failed (check `error` field for details)
|
||||
- **`index_failed`**: File indexing failed (check `error` field for details)
|
||||
|
||||
#### Additional Fields
|
||||
|
||||
- **`progress`**: Human-readable progress information (string, nullable)
|
||||
- **`error`**: Error message for failed operations (string, nullable, max 600 characters)
|
||||
|
||||
### 4. Knowledge Collection Management
|
||||
|
||||
```go
|
||||
|
|
|
|||
|
|
@ -235,6 +235,7 @@ type AttachmentFilter struct {
|
|||
Public *bool `json:"public,omitempty"` // Filter by public status
|
||||
Gzip *bool `json:"gzip,omitempty"` // Filter by gzip compression
|
||||
CollectionID string `json:"collection_id,omitempty"` // Filter by knowledge collection ID
|
||||
Status string `json:"status,omitempty"` // Filter by processing status (uploading, uploaded, indexing, indexed, upload_failed, index_failed)
|
||||
Keywords string `json:"keywords,omitempty"` // Search in filename
|
||||
Page int `json:"page,omitempty"` // Page number, starting from 1
|
||||
PageSize int `json:"pagesize,omitempty"` // Items per page
|
||||
|
|
|
|||
|
|
@ -367,6 +367,9 @@ func (conv *Xun) initAttachmentTable() error {
|
|||
table.Boolean("gzip").SetDefault(false).Index()
|
||||
table.BigInteger("bytes").Index()
|
||||
table.String("collection_id", 200).Null().Index()
|
||||
table.Enum("status", []string{"uploading", "uploaded", "indexing", "indexed", "upload_failed", "index_failed"}).SetDefault("uploading").Index() // Status field enum
|
||||
table.String("progress", 200).Null() // Progress information
|
||||
table.String("error", 600).Null() // Error information
|
||||
table.TimestampTz("created_at").SetDefaultRaw("CURRENT_TIMESTAMP").Index()
|
||||
table.TimestampTz("updated_at").Null().Index()
|
||||
})
|
||||
|
|
@ -383,7 +386,7 @@ func (conv *Xun) initAttachmentTable() error {
|
|||
return err
|
||||
}
|
||||
|
||||
fields := []string{"id", "file_id", "uid", "guest", "manager", "content_type", "name", "public", "scope", "gzip", "bytes", "collection_id", "created_at", "updated_at"}
|
||||
fields := []string{"id", "file_id", "uid", "guest", "manager", "content_type", "name", "public", "scope", "gzip", "bytes", "collection_id", "status", "progress", "error", "created_at", "updated_at"}
|
||||
for _, field := range fields {
|
||||
if !tab.HasColumn(field) {
|
||||
return fmt.Errorf("%s is required", field)
|
||||
|
|
@ -1789,6 +1792,11 @@ func (conv *Xun) GetAttachments(filter AttachmentFilter, locale ...string) (*Att
|
|||
qb.Where("collection_id", filter.CollectionID)
|
||||
}
|
||||
|
||||
// Apply status filter if provided
|
||||
if filter.Status != "" {
|
||||
qb.Where("status", filter.Status)
|
||||
}
|
||||
|
||||
// Apply keyword filter if provided
|
||||
if filter.Keywords != "" {
|
||||
qb.Where("name", "like", fmt.Sprintf("%%%s%%", filter.Keywords))
|
||||
|
|
@ -1945,6 +1953,11 @@ func (conv *Xun) DeleteAttachments(filter AttachmentFilter) (int64, error) {
|
|||
qb.Where("collection_id", filter.CollectionID)
|
||||
}
|
||||
|
||||
// Apply status filter if provided
|
||||
if filter.Status != "" {
|
||||
qb.Where("status", filter.Status)
|
||||
}
|
||||
|
||||
// Apply keyword filter if provided
|
||||
if filter.Keywords != "" {
|
||||
qb.Where("name", "like", fmt.Sprintf("%%%s%%", filter.Keywords))
|
||||
|
|
|
|||
|
|
@ -1289,6 +1289,9 @@ func TestXunAttachmentCRUD(t *testing.T) {
|
|||
"gzip": false,
|
||||
"bytes": 102400,
|
||||
"scope": []string{"user", "admin"},
|
||||
"status": "uploaded",
|
||||
"progress": "100%",
|
||||
"error": nil,
|
||||
}
|
||||
|
||||
v, err := store.SaveAttachment(attachment)
|
||||
|
|
@ -1307,10 +1310,16 @@ func TestXunAttachmentCRUD(t *testing.T) {
|
|||
assert.Equal(t, "test-image.jpg", attachmentData["name"])
|
||||
assert.Equal(t, int64(1), attachmentData["public"])
|
||||
assert.Equal(t, []interface{}{"user", "admin"}, attachmentData["scope"])
|
||||
assert.Equal(t, "uploaded", attachmentData["status"])
|
||||
assert.Equal(t, "100%", attachmentData["progress"])
|
||||
assert.Nil(t, attachmentData["error"])
|
||||
|
||||
// Test SaveAttachment (Update)
|
||||
attachment["name"] = "updated-image.jpg"
|
||||
attachment["bytes"] = 204800
|
||||
attachment["status"] = "indexing"
|
||||
attachment["progress"] = "Processing..."
|
||||
attachment["error"] = "Connection timeout"
|
||||
v, err = store.SaveAttachment(attachment)
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "test-file-123", v.(string))
|
||||
|
|
@ -1320,6 +1329,9 @@ func TestXunAttachmentCRUD(t *testing.T) {
|
|||
assert.Nil(t, err)
|
||||
assert.Equal(t, "updated-image.jpg", attachmentData["name"])
|
||||
assert.Equal(t, int64(204800), attachmentData["bytes"])
|
||||
assert.Equal(t, "indexing", attachmentData["status"])
|
||||
assert.Equal(t, "Processing...", attachmentData["progress"])
|
||||
assert.Equal(t, "Connection timeout", attachmentData["error"])
|
||||
|
||||
// Test GetAttachments with filters
|
||||
resp, err := store.GetAttachments(AttachmentFilter{
|
||||
|
|
@ -1710,3 +1722,149 @@ func TestXunAttachmentFiltering(t *testing.T) {
|
|||
_, err = store.DeleteAttachments(AttachmentFilter{})
|
||||
assert.Nil(t, err)
|
||||
}
|
||||
|
||||
func TestXunAttachmentStatusFields(t *testing.T) {
|
||||
test.Prepare(t, config.Conf)
|
||||
defer test.Clean()
|
||||
defer capsule.Schema().DropTableIfExists("__unit_test_conversation_attachment")
|
||||
|
||||
// Drop attachment table before test
|
||||
err := capsule.Schema().DropTableIfExists("__unit_test_conversation_attachment")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Add a small delay to ensure table is created
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
store, err := NewXun(Setting{
|
||||
Connector: "default",
|
||||
Prefix: "__unit_test_conversation_",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Clean up any existing data
|
||||
_, err = store.DeleteAttachments(AttachmentFilter{})
|
||||
assert.Nil(t, err)
|
||||
|
||||
// Test all possible enum status values
|
||||
statusValues := []string{"uploading", "uploaded", "indexing", "indexed", "upload_failed", "index_failed"}
|
||||
|
||||
for i, status := range statusValues {
|
||||
// Create attachment with specific status
|
||||
attachment := map[string]interface{}{
|
||||
"file_id": fmt.Sprintf("test-file-%s-%d", status, i),
|
||||
"uid": "user-123",
|
||||
"manager": "local",
|
||||
"content_type": "image/jpeg",
|
||||
"name": fmt.Sprintf("test-%s.jpg", status),
|
||||
"guest": false,
|
||||
"public": true,
|
||||
"gzip": false,
|
||||
"bytes": 102400,
|
||||
"status": status,
|
||||
"progress": fmt.Sprintf("%s in progress", status),
|
||||
"error": nil,
|
||||
}
|
||||
|
||||
// Set error message for failed statuses
|
||||
if status == "upload_failed" || status == "index_failed" {
|
||||
attachment["error"] = fmt.Sprintf("%s error occurred", status)
|
||||
}
|
||||
|
||||
v, err := store.SaveAttachment(attachment)
|
||||
assert.Nil(t, err)
|
||||
fileID := v.(string)
|
||||
|
||||
// Verify the attachment was saved with correct status
|
||||
attachmentData, err := store.GetAttachment(fileID)
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, status, attachmentData["status"])
|
||||
assert.Equal(t, fmt.Sprintf("%s in progress", status), attachmentData["progress"])
|
||||
|
||||
if status == "upload_failed" || status == "index_failed" {
|
||||
assert.Equal(t, fmt.Sprintf("%s error occurred", status), attachmentData["error"])
|
||||
} else {
|
||||
assert.Nil(t, attachmentData["error"])
|
||||
}
|
||||
}
|
||||
|
||||
// Test default status value (should be "uploading")
|
||||
attachmentWithoutStatus := map[string]interface{}{
|
||||
"file_id": "test-file-default",
|
||||
"uid": "user-123",
|
||||
"manager": "local",
|
||||
"content_type": "image/jpeg",
|
||||
"name": "test-default.jpg",
|
||||
"guest": false,
|
||||
"public": true,
|
||||
"gzip": false,
|
||||
"bytes": 102400,
|
||||
// status not specified - should use default
|
||||
}
|
||||
|
||||
v, err := store.SaveAttachment(attachmentWithoutStatus)
|
||||
assert.Nil(t, err)
|
||||
fileID := v.(string)
|
||||
|
||||
// Verify default status
|
||||
attachmentData, err := store.GetAttachment(fileID)
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "uploading", attachmentData["status"]) // Should be default value
|
||||
assert.Nil(t, attachmentData["progress"]) // Should be null
|
||||
assert.Nil(t, attachmentData["error"]) // Should be null
|
||||
|
||||
// Test updating status workflow: uploading -> uploaded -> indexing -> indexed
|
||||
workflowAttachment := map[string]interface{}{
|
||||
"file_id": "test-file-workflow",
|
||||
"uid": "user-123",
|
||||
"manager": "local",
|
||||
"content_type": "text/plain",
|
||||
"name": "workflow-test.txt",
|
||||
"status": "uploading",
|
||||
"progress": "Starting upload...",
|
||||
}
|
||||
|
||||
v, err = store.SaveAttachment(workflowAttachment)
|
||||
assert.Nil(t, err)
|
||||
workflowFileID := v.(string)
|
||||
|
||||
// Update to uploaded
|
||||
workflowAttachment["status"] = "uploaded"
|
||||
workflowAttachment["progress"] = "Upload completed, starting indexing..."
|
||||
_, err = store.SaveAttachment(workflowAttachment)
|
||||
assert.Nil(t, err)
|
||||
|
||||
attachmentData, err = store.GetAttachment(workflowFileID)
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "uploaded", attachmentData["status"])
|
||||
assert.Equal(t, "Upload completed, starting indexing...", attachmentData["progress"])
|
||||
|
||||
// Update to indexing
|
||||
workflowAttachment["status"] = "indexing"
|
||||
workflowAttachment["progress"] = "Indexing in progress..."
|
||||
_, err = store.SaveAttachment(workflowAttachment)
|
||||
assert.Nil(t, err)
|
||||
|
||||
attachmentData, err = store.GetAttachment(workflowFileID)
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "indexing", attachmentData["status"])
|
||||
assert.Equal(t, "Indexing in progress...", attachmentData["progress"])
|
||||
|
||||
// Update to indexed (final state)
|
||||
workflowAttachment["status"] = "indexed"
|
||||
workflowAttachment["progress"] = "Indexing completed"
|
||||
_, err = store.SaveAttachment(workflowAttachment)
|
||||
assert.Nil(t, err)
|
||||
|
||||
attachmentData, err = store.GetAttachment(workflowFileID)
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "indexed", attachmentData["status"])
|
||||
assert.Equal(t, "Indexing completed", attachmentData["progress"])
|
||||
|
||||
// Clean up test data
|
||||
_, err = store.DeleteAttachments(AttachmentFilter{})
|
||||
assert.Nil(t, err)
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue