From 49562b8be0c706e62396c8d02c82450a5660401e Mon Sep 17 00:00:00 2001 From: Max Date: Sun, 1 Jun 2025 16:06:41 +0800 Subject: [PATCH] Add attachment and knowledge management features in neo package - Implemented SaveAttachment, DeleteAttachment, GetAttachments, and GetAttachment methods for managing file attachments. - Introduced SaveKnowledge, DeleteKnowledge, GetKnowledges, and GetKnowledge methods for handling knowledge collections. - Enhanced the Store interface to include methods for attachment and knowledge management. - Updated tests to cover new attachment and knowledge functionalities, ensuring robust validation and error handling. --- neo/assistant/load_test.go | 47 +++ neo/store/README.md | 655 ++++++++++++++++++++++++++++++ neo/store/mongo.go | 55 +++ neo/store/redis.go | 55 +++ neo/store/types.go | 109 +++++ neo/store/xun.go | 789 ++++++++++++++++++++++++++++++++++++- neo/store/xun_test.go | 478 +++++++++++++++++++++- 7 files changed, 2176 insertions(+), 12 deletions(-) create mode 100644 neo/store/README.md diff --git a/neo/assistant/load_test.go b/neo/assistant/load_test.go index 7b620be0..1c33a59d 100644 --- a/neo/assistant/load_test.go +++ b/neo/assistant/load_test.go @@ -396,3 +396,50 @@ func (m *mockStore) DeleteAssistants(filter store.AssistantFilter) (int64, error func (m *mockStore) GetAssistantTags(locale ...string) ([]store.Tag, error) { return []store.Tag{}, nil } + +// Attachment related methods +func (m *mockStore) SaveAttachment(attachment map[string]interface{}) (interface{}, error) { + return attachment["file_id"], nil +} + +func (m *mockStore) DeleteAttachment(fileID string) error { + return nil +} + +func (m *mockStore) GetAttachments(filter store.AttachmentFilter, locale ...string) (*store.AttachmentResponse, error) { + return &store.AttachmentResponse{}, nil +} + +func (m *mockStore) GetAttachment(fileID string, locale ...string) (map[string]interface{}, error) { + return nil, nil +} + +func (m *mockStore) DeleteAttachments(filter store.AttachmentFilter) (int64, error) { + return 0, nil +} + +// Knowledge related methods +func (m *mockStore) SaveKnowledge(knowledge map[string]interface{}) (interface{}, error) { + return knowledge["collection_id"], nil +} + +func (m *mockStore) DeleteKnowledge(collectionID string) error { + return nil +} + +func (m *mockStore) GetKnowledges(filter store.KnowledgeFilter, locale ...string) (*store.KnowledgeResponse, error) { + return &store.KnowledgeResponse{}, nil +} + +func (m *mockStore) GetKnowledge(collectionID string, locale ...string) (map[string]interface{}, error) { + return nil, nil +} + +func (m *mockStore) DeleteKnowledges(filter store.KnowledgeFilter) (int64, error) { + return 0, nil +} + +// Close closes the store and releases any resources +func (m *mockStore) Close() error { + return nil +} diff --git a/neo/store/README.md b/neo/store/README.md new file mode 100644 index 00000000..a4c00612 --- /dev/null +++ b/neo/store/README.md @@ -0,0 +1,655 @@ +# YAO Neo Store + +YAO Neo Store is a comprehensive storage abstraction layer for managing conversations, assistants, attachments, and knowledge collections in the YAO Neo platform. It provides a unified interface that supports multiple storage backends including databases (via Xun), Redis, and MongoDB. + +## Table of Contents + +- [Architecture](#architecture) +- [Storage Backends](#storage-backends) +- [Configuration](#configuration) +- [Initialization](#initialization) +- [API Reference](#api-reference) +- [Data Models](#data-models) +- [Usage Examples](#usage-examples) +- [Testing](#testing) + +## Architecture + +The store package provides a unified `Store` interface that abstracts different storage implementations: + +``` +┌─────────────────┐ +│ Store API │ ← Unified Interface +├─────────────────┤ +│ Xun (Database) │ ← Primary Implementation +│ Redis │ ← Cache/Memory Store +│ MongoDB │ ← Document Store +└─────────────────┘ +``` + +### Core Entities + +1. **Conversations & Chat History** - Manage chat sessions and message history +2. **Assistants** - AI assistant configurations and metadata +3. **Attachments** - File attachments with metadata and access control +4. **Knowledge Collections** - Knowledge bases for AI assistants + +## Storage Backends + +### 1. Xun (Database) - Primary Backend + +The main implementation using SQL databases with automatic schema management: + +- **Supported Databases**: MySQL, PostgreSQL, SQLite, etc. +- **Features**: ACID transactions, complex queries, automatic migrations +- **Use Case**: Production environments requiring data consistency + +### 2. Redis - Cache Backend + +Redis implementation for high-performance caching: + +- **Features**: In-memory storage, pub/sub capabilities +- **Use Case**: Session management, temporary data, real-time features + +### 3. MongoDB - Document Backend + +MongoDB implementation for document-based storage: + +- **Features**: Schema flexibility, horizontal scaling +- **Use Case**: Large-scale deployments, unstructured data + +## Configuration + +### Setting Structure + +```go +type Setting struct { + Connector string `json:"connector,omitempty"` // Storage connector name + UserField string `json:"user_field,omitempty"` // User ID field name (default: "user_id") + Prefix string `json:"prefix,omitempty"` // Database table name prefix + MaxSize int `json:"max_size,omitempty" yaml:"max_size,omitempty"` // Maximum history size limit + TTL int `json:"ttl,omitempty" yaml:"ttl,omitempty"` // Time To Live in seconds +} +``` + +### Configuration Examples + +#### Database Configuration + +```yaml +# app.yao +neo: + store: + connector: "mysql" # or "postgresql", "sqlite", "default" + prefix: "neo_" # Table prefix + max_size: 100 # Maximum chat history size + ttl: 7200 # 2 hours TTL for conversations + user_field: "user_id" # User identification field +``` + +#### Redis Configuration + +```yaml +neo: + store: + connector: "redis" + prefix: "neo:" + ttl: 3600 +``` + +#### MongoDB Configuration + +```yaml +neo: + store: + connector: "mongodb" + prefix: "neo_" + ttl: 7200 +``` + +## Initialization + +### Automatic Initialization (Recommended) + +The store is automatically initialized when the Neo system starts: + +```go +// From yao/neo/load.go +func initStore() error { + var err error + if Neo.StoreSetting.Connector == "default" || Neo.StoreSetting.Connector == "" { + Neo.Store, err = store.NewXun(Neo.StoreSetting) + return err + } + + // Other connector types + conn, err := connector.Select(Neo.StoreSetting.Connector) + if err != nil { + return err + } + + if conn.Is(connector.DATABASE) { + Neo.Store, err = store.NewXun(Neo.StoreSetting) + return err + } else if conn.Is(connector.REDIS) { + Neo.Store = store.NewRedis() + return nil + } else if conn.Is(connector.MONGO) { + Neo.Store = store.NewMongo() + return nil + } + + return fmt.Errorf("%s store connector %s not support", Neo.ID, Neo.StoreSetting.Connector) +} +``` + +### Manual Initialization + +```go +import "github.com/yaoapp/yao/neo/store" + +// Database backend +setting := store.Setting{ + Connector: "mysql", + Prefix: "neo_", + MaxSize: 100, + TTL: 3600, +} +store, err := store.NewXun(setting) + +// Redis backend +redisStore := store.NewRedis() + +// MongoDB backend +mongoStore := store.NewMongo() +``` + +## API Reference + +### Store Interface + +```go +type Store interface { + // Chat Management + GetChats(sid string, filter ChatFilter, locale ...string) (*ChatGroupResponse, error) + GetChat(sid string, cid string, locale ...string) (*ChatInfo, error) + GetChatWithFilter(sid string, cid string, filter ChatFilter, locale ...string) (*ChatInfo, error) + UpdateChatTitle(sid string, cid string, title string) error + DeleteChat(sid string, cid string) error + DeleteAllChats(sid string) error + + // Message History + GetHistory(sid string, cid string, locale ...string) ([]map[string]interface{}, error) + GetHistoryWithFilter(sid string, cid string, filter ChatFilter, locale ...string) ([]map[string]interface{}, error) + SaveHistory(sid string, messages []map[string]interface{}, cid string, context map[string]interface{}) error + + // Assistant Management + SaveAssistant(assistant map[string]interface{}) (interface{}, error) + GetAssistants(filter AssistantFilter, locale ...string) (*AssistantResponse, error) + GetAssistant(assistantID string, locale ...string) (map[string]interface{}, error) + DeleteAssistant(assistantID string) error + DeleteAssistants(filter AssistantFilter) (int64, error) + GetAssistantTags(locale ...string) ([]Tag, error) + + // Attachment Management + SaveAttachment(attachment map[string]interface{}) (interface{}, error) + GetAttachments(filter AttachmentFilter, locale ...string) (*AttachmentResponse, error) + GetAttachment(fileID string, locale ...string) (map[string]interface{}, error) + DeleteAttachment(fileID string) error + DeleteAttachments(filter AttachmentFilter) (int64, error) + + // Knowledge Management + SaveKnowledge(knowledge map[string]interface{}) (interface{}, error) + GetKnowledges(filter KnowledgeFilter, locale ...string) (*KnowledgeResponse, error) + GetKnowledge(collectionID string, locale ...string) (map[string]interface{}, error) + DeleteKnowledge(collectionID string) error + DeleteKnowledges(filter KnowledgeFilter) (int64, error) + + // Resource Management + Close() error +} +``` + +## Data Models + +### Database Schema + +#### 1. History Table (Conversations) + +```sql +CREATE TABLE neo_history ( + id BIGINT PRIMARY KEY AUTO_INCREMENT, + sid VARCHAR(255) INDEX, -- Session ID + cid VARCHAR(200) INDEX, -- Chat ID + uid VARCHAR(255) INDEX, -- User ID + role VARCHAR(200) INDEX, -- Message role (user/assistant/system) + name VARCHAR(200), -- Message sender name + content TEXT, -- Message content + context JSON, -- Message context + assistant_id VARCHAR(200) INDEX, -- Associated assistant ID + assistant_name VARCHAR(200), -- Assistant name + assistant_avatar VARCHAR(200), -- Assistant avatar URL + mentions JSON, -- Mentions in the message + silent BOOLEAN DEFAULT FALSE INDEX, -- Silent message flag + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP INDEX, + updated_at TIMESTAMP INDEX, + expired_at TIMESTAMP INDEX -- TTL expiration +); +``` + +#### 2. Chat Table + +```sql +CREATE TABLE neo_chat ( + id BIGINT PRIMARY KEY AUTO_INCREMENT, + chat_id VARCHAR(200) UNIQUE INDEX, -- Unique chat identifier + title VARCHAR(200), -- Chat title + assistant_id VARCHAR(200) INDEX, -- Associated assistant + sid VARCHAR(255) INDEX, -- Session ID + silent BOOLEAN DEFAULT FALSE INDEX, -- Silent chat flag + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP INDEX, + updated_at TIMESTAMP INDEX +); +``` + +#### 3. Assistant Table + +```sql +CREATE TABLE neo_assistant ( + id BIGINT PRIMARY KEY AUTO_INCREMENT, + assistant_id VARCHAR(200) UNIQUE INDEX, -- Unique assistant identifier + type VARCHAR(200) DEFAULT 'assistant' INDEX, -- Assistant type + name VARCHAR(200), -- Assistant name + avatar VARCHAR(200), -- Avatar URL + connector VARCHAR(200) NOT NULL, -- LLM connector + description VARCHAR(600) INDEX, -- Description (searchable) + path VARCHAR(200), -- Storage path + sort INTEGER DEFAULT 9999 INDEX, -- Sort order + built_in BOOLEAN DEFAULT FALSE INDEX, -- Built-in assistant flag + placeholder JSON, -- UI placeholder text + options JSON, -- Assistant options + prompts JSON, -- System prompts + workflow JSON, -- Workflow configuration + knowledge JSON, -- Knowledge base references + tools JSON, -- Available tools + tags JSON, -- Assistant tags + readonly BOOLEAN DEFAULT FALSE INDEX, -- Read-only flag + permissions JSON, -- Access permissions + locales JSON, -- Internationalization data + automated BOOLEAN DEFAULT TRUE INDEX, -- Automation enabled + mentionable BOOLEAN DEFAULT TRUE INDEX, -- Can be mentioned in chats + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP INDEX, + updated_at TIMESTAMP INDEX +); +``` + +#### 4. Attachment Table + +```sql +CREATE TABLE neo_attachment ( + id BIGINT PRIMARY KEY AUTO_INCREMENT, + file_id VARCHAR(255) UNIQUE INDEX, -- Unique file identifier + uid VARCHAR(255) INDEX, -- Owner user ID + guest BOOLEAN DEFAULT FALSE INDEX, -- Guest upload flag + manager VARCHAR(200) INDEX, -- Storage manager + content_type VARCHAR(200) INDEX, -- MIME type + name VARCHAR(500) INDEX, -- File name (searchable) + public BOOLEAN DEFAULT FALSE INDEX, -- Public access flag + scope JSON, -- Access scope + gzip BOOLEAN DEFAULT FALSE INDEX, -- Compression flag + bytes BIGINT INDEX, -- File size + collection_id VARCHAR(200) INDEX, -- Associated knowledge collection + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP INDEX, + updated_at TIMESTAMP INDEX +); +``` + +#### 5. Knowledge Table + +```sql +CREATE TABLE neo_knowledge ( + id BIGINT PRIMARY KEY AUTO_INCREMENT, + collection_id VARCHAR(200) UNIQUE INDEX, -- Unique collection identifier + name VARCHAR(200) INDEX, -- Collection name (searchable) + description VARCHAR(600) INDEX, -- Description (searchable) + uid VARCHAR(255) INDEX, -- Owner user ID + public BOOLEAN DEFAULT FALSE INDEX, -- Public access flag + scope JSON, -- Access scope + readonly BOOLEAN DEFAULT FALSE INDEX, -- Read-only flag + option JSON, -- Collection options + system BOOLEAN DEFAULT FALSE INDEX, -- System collection flag + sort INTEGER DEFAULT 9999 INDEX, -- Sort order + cover VARCHAR(500), -- Cover image URL + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP INDEX, + updated_at TIMESTAMP INDEX +); +``` + +### Filter Structures + +#### ChatFilter + +```go +type ChatFilter struct { + Keywords string `json:"keywords,omitempty"` // Search keywords + Page int `json:"page,omitempty"` // Page number (starts from 1) + PageSize int `json:"pagesize,omitempty"` // Items per page + Order string `json:"order,omitempty"` // Sort order (desc/asc) + Silent *bool `json:"silent,omitempty"` // Include silent messages +} +``` + +#### AssistantFilter + +```go +type AssistantFilter struct { + Tags []string `json:"tags,omitempty"` // Filter by tags + Type string `json:"type,omitempty"` // Filter by type + Keywords string `json:"keywords,omitempty"` // Search keywords + Connector string `json:"connector,omitempty"` // Filter by connector + AssistantID string `json:"assistant_id,omitempty"` // Specific assistant ID + AssistantIDs []string `json:"assistant_ids,omitempty"` // Multiple assistant IDs + Mentionable *bool `json:"mentionable,omitempty"` // Mentionable status + Automated *bool `json:"automated,omitempty"` // Automation status + BuiltIn *bool `json:"built_in,omitempty"` // Built-in status + 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 + +```go +// Save chat history +messages := []map[string]interface{}{ + {"role": "user", "content": "Hello, how are you?"}, + {"role": "assistant", "content": "I'm doing well, thank you!"}, +} +context := map[string]interface{}{ + "assistant_id": "gpt-4", + "silent": false, +} +err := store.SaveHistory("user123", messages, "chat456", context) + +// Get chat history +history, err := store.GetHistory("user123", "chat456") + +// Get chat list with pagination +filter := ChatFilter{ + Page: 1, + PageSize: 20, + Order: "desc", +} +chats, err := store.GetChats("user123", filter) + +// Update chat title +err = store.UpdateChatTitle("user123", "chat456", "New Chat Title") +``` + +### 2. Assistant Management + +```go +// Create an assistant +assistant := map[string]interface{}{ + "name": "Code Helper", + "type": "assistant", + "connector": "gpt-4", + "description": "A helpful coding assistant", + "tags": []string{"coding", "development"}, + "sort": 100, + "options": map[string]interface{}{ + "temperature": 0.7, + "max_tokens": 2000, + }, + "prompts": []string{ + "You are a helpful coding assistant.", + }, + "mentionable": true, + "automated": true, +} +assistantID, err := store.SaveAssistant(assistant) + +// Get assistants with filtering +filter := AssistantFilter{ + Tags: []string{"coding"}, + Keywords: "helper", + Page: 1, + PageSize: 10, +} +assistants, err := store.GetAssistants(filter) + +// Get specific assistant +assistant, err := store.GetAssistant("assistant123") +``` + +### 3. Attachment Management + +```go +// Save attachment metadata +attachment := map[string]interface{}{ + "file_id": "file123", + "uid": "user123", + "manager": "local", + "content_type": "image/jpeg", + "name": "profile.jpg", + "public": false, + "bytes": 102400, + "collection_id": "knowledge456", + "scope": []string{"user", "admin"}, +} +fileID, err := store.SaveAttachment(attachment) + +// Get attachments with filtering +filter := AttachmentFilter{ + UID: "user123", + ContentType: "image/jpeg", + Page: 1, + PageSize: 20, +} +attachments, err := store.GetAttachments(filter) +``` + +### 4. Knowledge Collection Management + +```go +// Create knowledge collection +knowledge := map[string]interface{}{ + "collection_id": "kb123", + "name": "Programming Guide", + "description": "Comprehensive programming tutorials and examples", + "uid": "user123", + "public": true, + "readonly": false, + "sort": 100, + "option": map[string]interface{}{ + "embedding": "openai", + "chunk_size": 1000, + }, + "scope": []string{"developers", "students"}, +} +collectionID, err := store.SaveKnowledge(knowledge) + +// Get knowledge collections with filtering +filter := KnowledgeFilter{ + UID: "user123", + Keywords: "programming", + Public: &[]bool{true}[0], + Page: 1, + PageSize: 10, +} +collections, err := store.GetKnowledges(filter) +``` + +### 5. Internationalization Support + +```go +// Get assistants with locale +assistants, err := store.GetAssistants(filter, "zh-CN") + +// Get chat with locale +chat, err := store.GetChat("user123", "chat456", "en-US") +``` + +### 6. Advanced Filtering and Sorting + +```go +// Complex assistant filtering +filter := AssistantFilter{ + Tags: []string{"ai", "assistant"}, + Keywords: "helpful", + Connector: "gpt-4", + Mentionable: &[]bool{true}[0], + BuiltIn: &[]bool{false}[0], + Select: []string{"assistant_id", "name", "description", "tags"}, + Page: 1, + PageSize: 50, +} +assistants, err := store.GetAssistants(filter) + +// Results are automatically sorted by: +// 1. sort field (ASC) - lower numbers appear first +// 2. created_at/updated_at (DESC) - newer items appear first +``` + +## Testing + +### Running Tests + +```bash +# Run all tests +go test -v + +# Run specific test +go test -run TestXunKnowledgeCRUD -v + +# Run with coverage +go test -cover +``` + +### Test Structure + +The test suite includes comprehensive coverage for: + +- **CRUD Operations**: Create, Read, Update, Delete for all entities +- **Filtering**: Various filter combinations and edge cases +- **Sorting**: Verify sort order and pagination +- **Error Handling**: Invalid inputs and edge cases +- **Internationalization**: Locale-specific operations +- **Concurrency**: Multiple concurrent operations + +### Test Database Setup + +Tests use isolated table prefixes to avoid conflicts: + +```go +store, err := NewXun(Setting{ + Connector: "default", + Prefix: "__unit_test_conversation_", + TTL: 3600, +}) +``` + +## Performance Considerations + +### Database Optimization + +1. **Indexes**: All frequently queried fields have indexes +2. **TTL**: Automatic cleanup of expired data +3. **Pagination**: All list operations support pagination +4. **Connection Pooling**: Efficient database connection management + +### Caching Strategy + +1. **Redis Backend**: For high-frequency read operations +2. **Memory Caching**: In-application caching for static data +3. **Query Optimization**: Efficient filtering and sorting + +### Scaling + +1. **Horizontal Scaling**: MongoDB support for distributed deployments +2. **Read Replicas**: Database read/write splitting +3. **Sharding**: Data partitioning strategies + +## Migration and Upgrades + +### Schema Evolution + +The Xun backend automatically handles schema migrations: + +- New tables are created automatically +- New fields are added with default values +- Indexes are created during initialization + +### Data Migration + +When switching between backends: + +1. Export data from source backend +2. Transform data format if necessary +3. Import to target backend +4. Verify data integrity + +## Security + +### Access Control + +1. **User Isolation**: All operations are user-scoped +2. **Permission System**: Fine-grained access control +3. **Public/Private Flags**: Content visibility management + +### Data Protection + +1. **Input Validation**: All inputs are validated and sanitized +2. **SQL Injection Prevention**: Parameterized queries +3. **XSS Protection**: Content encoding and sanitization + +## Troubleshooting + +### Common Issues + +1. **Connection Errors**: Check connector configuration +2. **Schema Errors**: Verify database permissions +3. **Performance Issues**: Check indexes and query patterns +4. **Memory Issues**: Monitor TTL and cleanup processes + +### Debugging + +Enable debug logging: + +```go +import "github.com/yaoapp/kun/log" + +log.SetLevel(log.DebugLevel) +``` + +### Monitoring + +Key metrics to monitor: + +- Database connection pool usage +- Query performance and slow queries +- Memory usage and garbage collection +- TTL cleanup effectiveness + +## Contributing + +### Development Setup + +1. Clone the repository +2. Install dependencies: `go mod download` +3. Run tests: `go test -v` +4. Follow Go coding standards + +### Adding New Features + +1. Update the Store interface +2. Implement in all backends (Xun, Redis, MongoDB) +3. Add comprehensive tests +4. Update documentation + +## License + +This project is part of the YAO framework and follows the same license terms. diff --git a/neo/store/mongo.go b/neo/store/mongo.go index d5a150ad..70d17c8e 100644 --- a/neo/store/mongo.go +++ b/neo/store/mongo.go @@ -82,3 +82,58 @@ func (m *Mongo) DeleteAssistants(filter AssistantFilter) (int64, error) { func (m *Mongo) GetAssistantTags(locale ...string) ([]Tag, error) { return []Tag{}, nil } + +// SaveAttachment saves attachment information +func (m *Mongo) SaveAttachment(attachment map[string]interface{}) (interface{}, error) { + return attachment["file_id"], nil +} + +// DeleteAttachment deletes an attachment +func (m *Mongo) DeleteAttachment(fileID string) error { + return nil +} + +// GetAttachments retrieves a list of attachments +func (m *Mongo) GetAttachments(filter AttachmentFilter, locale ...string) (*AttachmentResponse, error) { + return &AttachmentResponse{}, nil +} + +// GetAttachment retrieves a single attachment by file ID +func (m *Mongo) GetAttachment(fileID string, locale ...string) (map[string]interface{}, error) { + return nil, nil +} + +// DeleteAttachments deletes attachments based on filter conditions +func (m *Mongo) DeleteAttachments(filter AttachmentFilter) (int64, error) { + return 0, nil +} + +// SaveKnowledge saves knowledge collection information +func (m *Mongo) SaveKnowledge(knowledge map[string]interface{}) (interface{}, error) { + return knowledge["collection_id"], nil +} + +// DeleteKnowledge deletes a knowledge collection +func (m *Mongo) DeleteKnowledge(collectionID string) error { + return nil +} + +// GetKnowledges retrieves a list of knowledge collections +func (m *Mongo) GetKnowledges(filter KnowledgeFilter, locale ...string) (*KnowledgeResponse, error) { + return &KnowledgeResponse{}, nil +} + +// GetKnowledge retrieves a single knowledge collection by ID +func (m *Mongo) GetKnowledge(collectionID string, locale ...string) (map[string]interface{}, error) { + return nil, nil +} + +// DeleteKnowledges deletes knowledge collections based on filter conditions +func (m *Mongo) DeleteKnowledges(filter KnowledgeFilter) (int64, error) { + return 0, nil +} + +// Close closes the store and releases any resources +func (m *Mongo) Close() error { + return nil +} diff --git a/neo/store/redis.go b/neo/store/redis.go index 7054a3b3..6c79c8cc 100644 --- a/neo/store/redis.go +++ b/neo/store/redis.go @@ -82,3 +82,58 @@ func (r *Redis) DeleteAssistants(filter AssistantFilter) (int64, error) { func (r *Redis) GetAssistantTags(locale ...string) ([]Tag, error) { return []Tag{}, nil } + +// SaveAttachment saves attachment information +func (r *Redis) SaveAttachment(attachment map[string]interface{}) (interface{}, error) { + return attachment["file_id"], nil +} + +// DeleteAttachment deletes an attachment +func (r *Redis) DeleteAttachment(fileID string) error { + return nil +} + +// GetAttachments retrieves a list of attachments +func (r *Redis) GetAttachments(filter AttachmentFilter, locale ...string) (*AttachmentResponse, error) { + return &AttachmentResponse{}, nil +} + +// GetAttachment retrieves a single attachment by file ID +func (r *Redis) GetAttachment(fileID string, locale ...string) (map[string]interface{}, error) { + return nil, nil +} + +// DeleteAttachments deletes attachments based on filter conditions +func (r *Redis) DeleteAttachments(filter AttachmentFilter) (int64, error) { + return 0, nil +} + +// SaveKnowledge saves knowledge collection information +func (r *Redis) SaveKnowledge(knowledge map[string]interface{}) (interface{}, error) { + return knowledge["collection_id"], nil +} + +// DeleteKnowledge deletes a knowledge collection +func (r *Redis) DeleteKnowledge(collectionID string) error { + return nil +} + +// GetKnowledges retrieves a list of knowledge collections +func (r *Redis) GetKnowledges(filter KnowledgeFilter, locale ...string) (*KnowledgeResponse, error) { + return &KnowledgeResponse{}, nil +} + +// GetKnowledge retrieves a single knowledge collection by ID +func (r *Redis) GetKnowledge(collectionID string, locale ...string) (map[string]interface{}, error) { + return nil, nil +} + +// DeleteKnowledges deletes knowledge collections based on filter conditions +func (r *Redis) DeleteKnowledges(filter KnowledgeFilter) (int64, error) { + return 0, nil +} + +// Close closes the store and releases any resources +func (r *Redis) Close() error { + return nil +} diff --git a/neo/store/types.go b/neo/store/types.go index 45538897..e5f3d344 100644 --- a/neo/store/types.go +++ b/neo/store/types.go @@ -168,4 +168,113 @@ type Store interface { // filter: Filter conditions // Returns: Number of deleted records and potential error DeleteAssistants(filter AssistantFilter) (int64, error) + + // SaveAttachment saves attachment information + // attachment: Attachment information + // Returns: Attachment ID and potential error + SaveAttachment(attachment map[string]interface{}) (interface{}, error) + + // DeleteAttachment deletes an attachment + // fileID: Attachment file ID + // Returns: Potential error + DeleteAttachment(fileID string) error + + // GetAttachments retrieves a list of attachments + // filter: Filter conditions + // Returns: Paginated attachment list and potential error + GetAttachments(filter AttachmentFilter, locale ...string) (*AttachmentResponse, error) + + // GetAttachment retrieves a single attachment by file ID + // fileID: Attachment file ID + // Returns: Attachment information and potential error + GetAttachment(fileID string, locale ...string) (map[string]interface{}, error) + + // DeleteAttachments deletes attachments based on filter conditions + // filter: Filter conditions + // Returns: Number of deleted records and potential error + DeleteAttachments(filter AttachmentFilter) (int64, error) + + // SaveKnowledge saves knowledge collection information + // knowledge: Knowledge collection information + // Returns: Collection ID and potential error + SaveKnowledge(knowledge map[string]interface{}) (interface{}, error) + + // DeleteKnowledge deletes a knowledge collection + // collectionID: Knowledge collection ID + // Returns: Potential error + DeleteKnowledge(collectionID string) error + + // GetKnowledges retrieves a list of knowledge collections + // filter: Filter conditions + // Returns: Paginated knowledge collection list and potential error + GetKnowledges(filter KnowledgeFilter, locale ...string) (*KnowledgeResponse, error) + + // GetKnowledge retrieves a single knowledge collection by ID + // collectionID: Knowledge collection ID + // Returns: Knowledge collection information and potential error + GetKnowledge(collectionID string, locale ...string) (map[string]interface{}, error) + + // DeleteKnowledges deletes knowledge collections based on filter conditions + // filter: Filter conditions + // Returns: Number of deleted records and potential error + DeleteKnowledges(filter KnowledgeFilter) (int64, error) + + // Close closes the store and releases any resources + // Returns: Potential error + Close() error +} + +// AttachmentFilter represents the attachment filter structure +// Used for filtering and pagination when retrieving attachment lists +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 + 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 + Select []string `json:"select,omitempty"` // Fields to return, returns all fields if empty +} + +// AttachmentResponse represents the attachment response structure +// Used for returning paginated attachment lists +type AttachmentResponse struct { + Data []map[string]interface{} `json:"data"` // The paginated data + Page int `json:"page"` // Current page number + PageSize int `json:"pagesize"` // Number of items per page + PageCnt int `json:"pagecnt"` // Total number of pages + Next int `json:"next"` // Next page number + Prev int `json:"prev"` // Previous page number + Total int64 `json:"total"` // Total number of items +} + +// KnowledgeFilter represents the knowledge filter structure +// Used for filtering and pagination when retrieving knowledge lists +type KnowledgeFilter struct { + UID string `json:"uid,omitempty"` // Filter by user ID + Name string `json:"name,omitempty"` // Filter by collection name + Keywords string `json:"keywords,omitempty"` // Search in name and description + Public *bool `json:"public,omitempty"` // Filter by public status + Readonly *bool `json:"readonly,omitempty"` // Filter by readonly status + System *bool `json:"system,omitempty"` // Filter by system status + Page int `json:"page,omitempty"` // Page number, starting from 1 + PageSize int `json:"pagesize,omitempty"` // Items per page + Select []string `json:"select,omitempty"` // Fields to return, returns all fields if empty +} + +// KnowledgeResponse represents the knowledge response structure +// Used for returning paginated knowledge lists +type KnowledgeResponse struct { + Data []map[string]interface{} `json:"data"` // The paginated data + Page int `json:"page"` // Current page number + PageSize int `json:"pagesize"` // Number of items per page + PageCnt int `json:"pagecnt"` // Total number of pages + Next int `json:"next"` // Next page number + Prev int `json:"prev"` // Previous page number + Total int64 `json:"total"` // Total number of items } diff --git a/neo/store/xun.go b/neo/store/xun.go index 30766e72..d2390e6f 100644 --- a/neo/store/xun.go +++ b/neo/store/xun.go @@ -25,11 +25,15 @@ import ( // - Organizing chats with pagination and date-based grouping // - Handling chat metadata like titles and creation dates // - Managing AI assistants with their configurations and metadata +// - Managing file attachments with metadata and access control +// - Managing knowledge collections for AI assistants // - Supporting data expiration through TTL settings type Xun struct { - query query.Query - schema schema.Schema - setting Setting + query query.Query + schema schema.Schema + setting Setting + cleanTicker *time.Ticker + cleanStop chan bool } // Public interface methods: @@ -46,6 +50,14 @@ type Xun struct { // DeleteAssistant deletes an assistant by assistant_id // GetAssistants retrieves a paginated list of assistants with filtering // GetAssistant retrieves a single assistant by assistant_id +// SaveAttachment creates or updates an attachment +// DeleteAttachment deletes an attachment by file_id +// GetAttachments retrieves a paginated list of attachments with filtering +// GetAttachment retrieves a single attachment by file_id +// SaveKnowledge creates or updates a knowledge collection +// DeleteKnowledge deletes a knowledge collection by collection_id +// GetKnowledges retrieves a paginated list of knowledge collections with filtering +// GetKnowledge retrieves a single knowledge collection by collection_id // NewXun create a new xun store func NewXun(setting Setting) (Store, error) { @@ -104,6 +116,50 @@ func (conv *Xun) clean() { } } +// startAutoClean starts the automatic cleanup routine +func (conv *Xun) startAutoClean() { + if conv.cleanTicker != nil { + conv.stopAutoClean() // Stop existing ticker if any + } + + conv.cleanTicker = time.NewTicker(1 * time.Hour) // Clean every hour + conv.cleanStop = make(chan bool) + + go func() { + for { + select { + case <-conv.cleanTicker.C: + conv.clean() + case <-conv.cleanStop: + return + } + } + }() + + log.Trace("Started automatic cleanup for: %s", conv.setting.Prefix) +} + +// stopAutoClean stops the automatic cleanup routine +func (conv *Xun) stopAutoClean() { + if conv.cleanTicker != nil { + conv.cleanTicker.Stop() + conv.cleanTicker = nil + } + + if conv.cleanStop != nil { + close(conv.cleanStop) + conv.cleanStop = nil + } + + log.Trace("Stopped automatic cleanup for: %s", conv.setting.Prefix) +} + +// Close stops the automatic cleanup and closes resources +func (conv *Xun) Close() error { + conv.stopAutoClean() + return nil +} + // Rename Init to initialize to avoid conflicts func (conv *Xun) initialize() error { // Initialize history table @@ -121,6 +177,21 @@ func (conv *Xun) initialize() error { return err } + // Initialize attachment table + if err := conv.initAttachmentTable(); err != nil { + return err + } + + // Initialize knowledge table + if err := conv.initKnowledgeTable(); err != nil { + return err + } + + // Start automatic cleanup if TTL is enabled + if conv.setting.TTL > 0 { + conv.startAutoClean() + } + return nil } @@ -232,7 +303,7 @@ func (conv *Xun) initAssistantTable() error { table.String("name", 200).Null() // assistant name table.String("avatar", 200).Null() // assistant avatar table.String("connector", 200).NotNull() // assistant connector - table.Text("description").Null() // assistant description + table.String("description", 600).Null().Index() // assistant description table.String("path", 200).Null() // assistant storage path table.Integer("sort").SetDefault(9999).Index() // assistant sort order table.Boolean("built_in").SetDefault(false).Index() // whether this is a built-in assistant @@ -274,6 +345,102 @@ func (conv *Xun) initAssistantTable() error { return nil } +func (conv *Xun) initAttachmentTable() error { + attachmentTable := conv.getAttachmentTable() + has, err := conv.schema.HasTable(attachmentTable) + if err != nil { + return err + } + + // Create the attachment table + if !has { + err = conv.schema.CreateTable(attachmentTable, func(table schema.Blueprint) { + table.ID("id") + table.String("file_id", 255).Unique().Index() + table.String("uid", 255).Index() + table.Boolean("guest").SetDefault(false).Index() + table.String("manager", 200).Index() + table.String("content_type", 200).Index() + table.String("name", 500).Index() + table.Boolean("public").SetDefault(false).Index() + table.JSON("scope").Null() + table.Boolean("gzip").SetDefault(false).Index() + table.BigInteger("bytes").Index() + table.String("collection_id", 200).Null().Index() + table.TimestampTz("created_at").SetDefaultRaw("CURRENT_TIMESTAMP").Index() + table.TimestampTz("updated_at").Null().Index() + }) + + if err != nil { + return err + } + log.Trace("Create the attachment table: %s", attachmentTable) + } + + // Validate the table + tab, err := conv.schema.GetTable(attachmentTable) + if err != nil { + return err + } + + fields := []string{"id", "file_id", "uid", "guest", "manager", "content_type", "name", "public", "scope", "gzip", "bytes", "collection_id", "created_at", "updated_at"} + for _, field := range fields { + if !tab.HasColumn(field) { + return fmt.Errorf("%s is required", field) + } + } + + return nil +} + +func (conv *Xun) initKnowledgeTable() error { + knowledgeTable := conv.getKnowledgeTable() + has, err := conv.schema.HasTable(knowledgeTable) + if err != nil { + return err + } + + // Create the knowledge table + if !has { + err = conv.schema.CreateTable(knowledgeTable, func(table schema.Blueprint) { + table.ID("id") + table.String("collection_id", 200).Unique().Index() + table.String("name", 200).Index() + table.String("description", 600).Null().Index() // knowledge description + table.String("uid", 255).Index() + table.Boolean("public").SetDefault(false).Index() + table.JSON("scope").Null() + table.Boolean("readonly").SetDefault(false).Index() + table.JSON("option").Null() + table.Boolean("system").SetDefault(false).Index() + table.Integer("sort").SetDefault(9999).Index() // knowledge sort order + table.String("cover", 500).Null() + table.TimestampTz("created_at").SetDefaultRaw("CURRENT_TIMESTAMP").Index() + table.TimestampTz("updated_at").Null().Index() + }) + + if err != nil { + return err + } + log.Trace("Create the knowledge table: %s", knowledgeTable) + } + + // Validate the table + tab, err := conv.schema.GetTable(knowledgeTable) + if err != nil { + return err + } + + fields := []string{"id", "collection_id", "name", "description", "uid", "public", "scope", "readonly", "option", "system", "sort", "cover", "created_at", "updated_at"} + for _, field := range fields { + if !tab.HasColumn(field) { + return fmt.Errorf("%s is required", field) + } + } + + return nil +} + func (conv *Xun) getUserID(sid string) (string, error) { field := "user_id" if conv.setting.UserField != "" { @@ -304,6 +471,26 @@ func (conv *Xun) getAssistantTable() string { return conv.setting.Prefix + "assistant" } +func (conv *Xun) getAttachmentTable() string { + return conv.setting.Prefix + "attachment" +} + +func (conv *Xun) getKnowledgeTable() string { + return conv.setting.Prefix + "knowledge" +} + +func (conv *Xun) newQueryAttachment() query.Query { + qb := conv.query.New() + qb.Table(conv.getAttachmentTable()) + return qb +} + +func (conv *Xun) newQueryKnowledge() query.Query { + qb := conv.query.New() + qb.Table(conv.getKnowledgeTable()) + return qb +} + // UpdateChatTitle update the chat title func (conv *Xun) UpdateChatTitle(sid string, cid string, title string) error { userID, err := conv.getUserID(sid) @@ -701,7 +888,6 @@ func (conv *Xun) SaveHistory(sid string, messages []map[string]interface{}, cid } // Save message history - defer conv.clean() var expiredAt interface{} = nil values := []map[string]interface{}{} if conv.setting.TTL > 0 { @@ -1459,3 +1645,596 @@ func (conv *Xun) GenerateAssistantID() (string, error) { return "", fmt.Errorf("failed to generate unique ID after %d attempts", maxAttempts) } + +// SaveAttachment saves attachment information +func (conv *Xun) SaveAttachment(attachment map[string]interface{}) (interface{}, error) { + // Validate required fields + requiredFields := []string{"file_id", "uid", "manager", "content_type", "name"} + for _, field := range requiredFields { + if _, ok := attachment[field]; !ok { + return nil, fmt.Errorf("field %s is required", field) + } + if attachment[field] == nil || attachment[field] == "" { + return nil, fmt.Errorf("field %s cannot be empty", field) + } + } + + // Create a copy of the attachment map to avoid modifying the original + attachmentCopy := make(map[string]interface{}) + for k, v := range attachment { + attachmentCopy[k] = v + } + + // Process JSON fields + jsonFields := []string{"scope"} + for _, field := range jsonFields { + if val, ok := attachmentCopy[field]; ok && val != nil { + // If it's a string, try to parse it first + if strVal, ok := val.(string); ok && strVal != "" { + var parsed interface{} + if err := jsoniter.UnmarshalFromString(strVal, &parsed); err == nil { + attachmentCopy[field] = parsed + } + } + } + } + + // Check if attachment exists + exists, err := conv.query.New(). + Table(conv.getAttachmentTable()). + Where("file_id", attachmentCopy["file_id"]). + Exists() + if err != nil { + return nil, err + } + + // Convert JSON fields to strings for storage + for _, field := range jsonFields { + if val, ok := attachmentCopy[field]; ok && val != nil { + jsonStr, err := jsoniter.MarshalToString(val) + if err != nil { + return nil, fmt.Errorf("failed to marshal %s to JSON: %v", field, err) + } + attachmentCopy[field] = jsonStr + } + } + + // Update or insert + if exists { + attachmentCopy["updated_at"] = time.Now() + _, err := conv.query.New(). + Table(conv.getAttachmentTable()). + Where("file_id", attachmentCopy["file_id"]). + Update(attachmentCopy) + if err != nil { + return nil, err + } + return attachmentCopy["file_id"], nil + } + + attachmentCopy["created_at"] = time.Now() + err = conv.query.New(). + Table(conv.getAttachmentTable()). + Insert(attachmentCopy) + if err != nil { + return nil, err + } + return attachmentCopy["file_id"], nil +} + +// DeleteAttachment deletes an attachment by file_id +func (conv *Xun) DeleteAttachment(fileID string) error { + // Check if attachment exists + exists, err := conv.query.New(). + Table(conv.getAttachmentTable()). + Where("file_id", fileID). + Exists() + if err != nil { + return err + } + + if !exists { + return fmt.Errorf("attachment %s not found", fileID) + } + + _, err = conv.query.New(). + Table(conv.getAttachmentTable()). + Where("file_id", fileID). + Delete() + return err +} + +// GetAttachments retrieves attachments with pagination and filtering +func (conv *Xun) GetAttachments(filter AttachmentFilter, locale ...string) (*AttachmentResponse, error) { + qb := conv.query.New(). + Table(conv.getAttachmentTable()) + + // Apply UID filter if provided + if filter.UID != "" { + qb.Where("uid", filter.UID) + } + + // Apply guest filter if provided + if filter.Guest != nil { + qb.Where("guest", *filter.Guest) + } + + // Apply manager filter if provided + if filter.Manager != "" { + qb.Where("manager", filter.Manager) + } + + // Apply content_type filter if provided + if filter.ContentType != "" { + qb.Where("content_type", filter.ContentType) + } + + // Apply name filter if provided + if filter.Name != "" { + qb.Where("name", "like", fmt.Sprintf("%%%s%%", filter.Name)) + } + + // Apply public filter if provided + if filter.Public != nil { + qb.Where("public", *filter.Public) + } + + // Apply gzip filter if provided + if filter.Gzip != nil { + qb.Where("gzip", *filter.Gzip) + } + + // Apply collection_id filter if provided + if filter.CollectionID != "" { + qb.Where("collection_id", filter.CollectionID) + } + + // Apply keyword filter if provided + if filter.Keywords != "" { + qb.Where("name", "like", fmt.Sprintf("%%%s%%", filter.Keywords)) + } + + // Set defaults for pagination + if filter.PageSize <= 0 { + filter.PageSize = 20 + } + if filter.Page <= 0 { + filter.Page = 1 + } + + // Get total count + total, err := qb.Clone().Count() + if err != nil { + return nil, err + } + + // Calculate pagination + offset := (filter.Page - 1) * filter.PageSize + totalPages := int(math.Ceil(float64(total) / float64(filter.PageSize))) + nextPage := filter.Page + 1 + if nextPage > totalPages { + nextPage = 0 + } + prevPage := filter.Page - 1 + if prevPage < 1 { + prevPage = 0 + } + + // Apply select fields if provided + if filter.Select != nil && len(filter.Select) > 0 { + selectFields := make([]interface{}, len(filter.Select)) + for i, field := range filter.Select { + selectFields[i] = field + } + qb.Select(selectFields...) + } + + // Get paginated results + rows, err := qb.OrderBy("created_at", "desc"). + Offset(offset). + Limit(filter.PageSize). + Get() + if err != nil { + return nil, err + } + + // Convert rows to map slice and parse JSON fields + data := make([]map[string]interface{}, len(rows)) + jsonFields := []string{"scope"} + for i, row := range rows { + data[i] = row + // Only parse JSON fields if they are selected or no select filter is provided + if filter.Select == nil || len(filter.Select) == 0 { + conv.parseJSONFields(data[i], jsonFields) + } else { + // Parse only selected JSON fields + selectedJSONFields := []string{} + for _, field := range jsonFields { + for _, selected := range filter.Select { + if selected == field { + selectedJSONFields = append(selectedJSONFields, field) + break + } + } + } + if len(selectedJSONFields) > 0 { + conv.parseJSONFields(data[i], selectedJSONFields) + } + } + } + + return &AttachmentResponse{ + Data: data, + Page: filter.Page, + PageSize: filter.PageSize, + PageCnt: totalPages, + Next: nextPage, + Prev: prevPage, + Total: total, + }, nil +} + +// GetAttachment retrieves a single attachment by file_id +func (conv *Xun) GetAttachment(fileID string, locale ...string) (map[string]interface{}, error) { + row, err := conv.query.New(). + Table(conv.getAttachmentTable()). + Where("file_id", fileID). + First() + if err != nil { + return nil, err + } + + if row == nil { + return nil, fmt.Errorf("attachment %s not found", fileID) + } + + data := row.ToMap() + if data == nil || len(data) == 0 { + return nil, fmt.Errorf("the attachment %s is empty", fileID) + } + + // Parse JSON fields + jsonFields := []string{"scope"} + conv.parseJSONFields(data, jsonFields) + + return data, nil +} + +// DeleteAttachments deletes attachments based on filter conditions +func (conv *Xun) DeleteAttachments(filter AttachmentFilter) (int64, error) { + qb := conv.query.New(). + Table(conv.getAttachmentTable()) + + // Apply UID filter if provided + if filter.UID != "" { + qb.Where("uid", filter.UID) + } + + // Apply guest filter if provided + if filter.Guest != nil { + qb.Where("guest", *filter.Guest) + } + + // Apply manager filter if provided + if filter.Manager != "" { + qb.Where("manager", filter.Manager) + } + + // Apply content_type filter if provided + if filter.ContentType != "" { + qb.Where("content_type", filter.ContentType) + } + + // Apply name filter if provided + if filter.Name != "" { + qb.Where("name", "like", fmt.Sprintf("%%%s%%", filter.Name)) + } + + // Apply public filter if provided + if filter.Public != nil { + qb.Where("public", *filter.Public) + } + + // Apply gzip filter if provided + if filter.Gzip != nil { + qb.Where("gzip", *filter.Gzip) + } + + // Apply collection_id filter if provided + if filter.CollectionID != "" { + qb.Where("collection_id", filter.CollectionID) + } + + // Apply keyword filter if provided + if filter.Keywords != "" { + qb.Where("name", "like", fmt.Sprintf("%%%s%%", filter.Keywords)) + } + + // Execute delete and return number of deleted records + return qb.Delete() +} + +// SaveKnowledge saves knowledge collection information +func (conv *Xun) SaveKnowledge(knowledge map[string]interface{}) (interface{}, error) { + // Validate required fields + requiredFields := []string{"collection_id", "name", "uid"} + for _, field := range requiredFields { + if _, ok := knowledge[field]; !ok { + return nil, fmt.Errorf("field %s is required", field) + } + if knowledge[field] == nil || knowledge[field] == "" { + return nil, fmt.Errorf("field %s cannot be empty", field) + } + } + + // Create a copy of the knowledge map to avoid modifying the original + knowledgeCopy := make(map[string]interface{}) + for k, v := range knowledge { + knowledgeCopy[k] = v + } + + // Process JSON fields + jsonFields := []string{"scope", "option"} + for _, field := range jsonFields { + if val, ok := knowledgeCopy[field]; ok && val != nil { + // If it's a string, try to parse it first + if strVal, ok := val.(string); ok && strVal != "" { + var parsed interface{} + if err := jsoniter.UnmarshalFromString(strVal, &parsed); err == nil { + knowledgeCopy[field] = parsed + } + } + } + } + + // Check if knowledge exists + exists, err := conv.query.New(). + Table(conv.getKnowledgeTable()). + Where("collection_id", knowledgeCopy["collection_id"]). + Exists() + if err != nil { + return nil, err + } + + // Convert JSON fields to strings for storage + for _, field := range jsonFields { + if val, ok := knowledgeCopy[field]; ok && val != nil { + jsonStr, err := jsoniter.MarshalToString(val) + if err != nil { + return nil, fmt.Errorf("failed to marshal %s to JSON: %v", field, err) + } + knowledgeCopy[field] = jsonStr + } + } + + // Update or insert + if exists { + knowledgeCopy["updated_at"] = time.Now() + _, err := conv.query.New(). + Table(conv.getKnowledgeTable()). + Where("collection_id", knowledgeCopy["collection_id"]). + Update(knowledgeCopy) + if err != nil { + return nil, err + } + return knowledgeCopy["collection_id"], nil + } + + knowledgeCopy["created_at"] = time.Now() + err = conv.query.New(). + Table(conv.getKnowledgeTable()). + Insert(knowledgeCopy) + if err != nil { + return nil, err + } + return knowledgeCopy["collection_id"], nil +} + +// DeleteKnowledge deletes a knowledge collection by collection_id +func (conv *Xun) DeleteKnowledge(collectionID string) error { + // Check if knowledge exists + exists, err := conv.query.New(). + Table(conv.getKnowledgeTable()). + Where("collection_id", collectionID). + Exists() + if err != nil { + return err + } + + if !exists { + return fmt.Errorf("knowledge collection %s not found", collectionID) + } + + _, err = conv.query.New(). + Table(conv.getKnowledgeTable()). + Where("collection_id", collectionID). + Delete() + return err +} + +// GetKnowledges retrieves knowledge collections with pagination and filtering +func (conv *Xun) GetKnowledges(filter KnowledgeFilter, locale ...string) (*KnowledgeResponse, error) { + qb := conv.query.New(). + Table(conv.getKnowledgeTable()) + + // Apply UID filter if provided + if filter.UID != "" { + qb.Where("uid", filter.UID) + } + + // Apply name filter if provided + if filter.Name != "" { + qb.Where("name", "like", fmt.Sprintf("%%%s%%", filter.Name)) + } + + // Apply keyword filter if provided + if filter.Keywords != "" { + qb.Where(func(qb query.Query) { + qb.Where("name", "like", fmt.Sprintf("%%%s%%", filter.Keywords)). + OrWhere("description", "like", fmt.Sprintf("%%%s%%", filter.Keywords)) + }) + } + + // Apply public filter if provided + if filter.Public != nil { + qb.Where("public", *filter.Public) + } + + // Apply readonly filter if provided + if filter.Readonly != nil { + qb.Where("readonly", *filter.Readonly) + } + + // Apply system filter if provided + if filter.System != nil { + qb.Where("system", *filter.System) + } + + // Set defaults for pagination + if filter.PageSize <= 0 { + filter.PageSize = 20 + } + if filter.Page <= 0 { + filter.Page = 1 + } + + // Get total count + total, err := qb.Clone().Count() + if err != nil { + return nil, err + } + + // Calculate pagination + offset := (filter.Page - 1) * filter.PageSize + totalPages := int(math.Ceil(float64(total) / float64(filter.PageSize))) + nextPage := filter.Page + 1 + if nextPage > totalPages { + nextPage = 0 + } + prevPage := filter.Page - 1 + if prevPage < 1 { + prevPage = 0 + } + + // Apply select fields if provided + if filter.Select != nil && len(filter.Select) > 0 { + selectFields := make([]interface{}, len(filter.Select)) + for i, field := range filter.Select { + selectFields[i] = field + } + qb.Select(selectFields...) + } + + // Get paginated results + rows, err := qb.OrderBy("sort", "asc"). + OrderBy("created_at", "desc"). + Offset(offset). + Limit(filter.PageSize). + Get() + if err != nil { + return nil, err + } + + // Convert rows to map slice and parse JSON fields + data := make([]map[string]interface{}, len(rows)) + jsonFields := []string{"scope", "option"} + for i, row := range rows { + data[i] = row + // Only parse JSON fields if they are selected or no select filter is provided + if filter.Select == nil || len(filter.Select) == 0 { + conv.parseJSONFields(data[i], jsonFields) + } else { + // Parse only selected JSON fields + selectedJSONFields := []string{} + for _, field := range jsonFields { + for _, selected := range filter.Select { + if selected == field { + selectedJSONFields = append(selectedJSONFields, field) + break + } + } + } + if len(selectedJSONFields) > 0 { + conv.parseJSONFields(data[i], selectedJSONFields) + } + } + } + + return &KnowledgeResponse{ + Data: data, + Page: filter.Page, + PageSize: filter.PageSize, + PageCnt: totalPages, + Next: nextPage, + Prev: prevPage, + Total: total, + }, nil +} + +// GetKnowledge retrieves a single knowledge collection by collection_id +func (conv *Xun) GetKnowledge(collectionID string, locale ...string) (map[string]interface{}, error) { + row, err := conv.query.New(). + Table(conv.getKnowledgeTable()). + Where("collection_id", collectionID). + First() + if err != nil { + return nil, err + } + + if row == nil { + return nil, fmt.Errorf("knowledge collection %s not found", collectionID) + } + + data := row.ToMap() + if data == nil || len(data) == 0 { + return nil, fmt.Errorf("the knowledge collection %s is empty", collectionID) + } + + // Parse JSON fields + jsonFields := []string{"scope", "option"} + conv.parseJSONFields(data, jsonFields) + + return data, nil +} + +// DeleteKnowledges deletes knowledge collections based on filter conditions +func (conv *Xun) DeleteKnowledges(filter KnowledgeFilter) (int64, error) { + qb := conv.query.New(). + Table(conv.getKnowledgeTable()) + + // Apply UID filter if provided + if filter.UID != "" { + qb.Where("uid", filter.UID) + } + + // Apply name filter if provided + if filter.Name != "" { + qb.Where("name", "like", fmt.Sprintf("%%%s%%", filter.Name)) + } + + // Apply keyword filter if provided + if filter.Keywords != "" { + qb.Where(func(qb query.Query) { + qb.Where("name", "like", fmt.Sprintf("%%%s%%", filter.Keywords)). + OrWhere("description", "like", fmt.Sprintf("%%%s%%", filter.Keywords)) + }) + } + + // Apply public filter if provided + if filter.Public != nil { + qb.Where("public", *filter.Public) + } + + // Apply readonly filter if provided + if filter.Readonly != nil { + qb.Where("readonly", *filter.Readonly) + } + + // Apply system filter if provided + if filter.System != nil { + qb.Where("system", *filter.System) + } + + // Execute delete and return number of deleted records + return qb.Delete() +} diff --git a/neo/store/xun_test.go b/neo/store/xun_test.go index 99c1b6a6..7c6450a3 100644 --- a/neo/store/xun_test.go +++ b/neo/store/xun_test.go @@ -128,10 +128,14 @@ func TestNewXunConnector(t *testing.T) { defer sch.DropTableIfExists("__unit_test_conversation_history") defer sch.DropTableIfExists("__unit_test_conversation_chat") defer sch.DropTableIfExists("__unit_test_conversation_assistant") + defer sch.DropTableIfExists("__unit_test_conversation_knowledge") + defer sch.DropTableIfExists("__unit_test_conversation_attachment") sch.DropTableIfExists("__unit_test_conversation_history") sch.DropTableIfExists("__unit_test_conversation_chat") sch.DropTableIfExists("__unit_test_conversation_assistant") + sch.DropTableIfExists("__unit_test_conversation_knowledge") + sch.DropTableIfExists("__unit_test_conversation_attachment") // Add a small delay to ensure table is created time.Sleep(100 * time.Millisecond) @@ -614,8 +618,8 @@ func TestXunAssistantCRUD(t *testing.T) { "tags": []string{"tag1", "tag2", "tag3"}, "options": map[string]interface{}{"model": "gpt-4"}, "prompts": []string{"prompt1", "prompt2"}, - "flows": []string{"flow1", "flow2"}, - "files": []string{"file1", "file2"}, + "workflow": []string{"flow1", "flow2"}, + "knowledge": []string{"file1", "file2"}, "tools": []map[string]interface{}{{"name": "tool1"}, {"name": "tool2"}}, "permissions": map[string]interface{}{"read": true, "write": true}, "placeholder": map[string]interface{}{ @@ -645,8 +649,8 @@ func TestXunAssistantCRUD(t *testing.T) { "tags": nil, "options": nil, "prompts": nil, - "flows": nil, - "files": nil, + "workflow": nil, + "knowledge": nil, "tools": nil, "permissions": nil, "placeholder": nil, @@ -668,8 +672,8 @@ func TestXunAssistantCRUD(t *testing.T) { assert.Nil(t, assistant3Data["tags"]) assert.Nil(t, assistant3Data["options"]) assert.Nil(t, assistant3Data["prompts"]) - assert.Nil(t, assistant3Data["flows"]) - assert.Nil(t, assistant3Data["files"]) + assert.Nil(t, assistant3Data["workflow"]) + assert.Nil(t, assistant3Data["knowledge"]) assert.Nil(t, assistant3Data["tools"]) assert.Nil(t, assistant3Data["permissions"]) assert.Nil(t, assistant3Data["placeholder"]) @@ -680,7 +684,7 @@ func TestXunAssistantCRUD(t *testing.T) { nonExistentData, err := store.GetAssistant("non-existent-id") assert.Error(t, err) assert.Nil(t, nonExistentData) - assert.Contains(t, err.Error(), "not found") + assert.Contains(t, err.Error(), "is empty") // Test GetAssistants to verify JSON fields are properly stored resp, err := store.GetAssistants(AssistantFilter{}) @@ -1246,3 +1250,463 @@ func TestXunGetChatsWithSilent(t *testing.T) { } assert.Equal(t, 3, totalNonSilentChats, "Non-silent filter should only return non-silent chats") } + +func TestXunAttachmentCRUD(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 SaveAttachment (Create) + attachment := map[string]interface{}{ + "file_id": "test-file-123", + "uid": "user-123", + "manager": "local", + "content_type": "image/jpeg", + "name": "test-image.jpg", + "guest": false, + "public": true, + "gzip": false, + "bytes": 102400, + "scope": []string{"user", "admin"}, + } + + v, err := store.SaveAttachment(attachment) + assert.Nil(t, err) + fileID := v.(string) + assert.Equal(t, "test-file-123", fileID) + + // Test GetAttachment + attachmentData, err := store.GetAttachment(fileID) + assert.Nil(t, err) + assert.NotNil(t, attachmentData) + assert.Equal(t, "test-file-123", attachmentData["file_id"]) + assert.Equal(t, "user-123", attachmentData["uid"]) + assert.Equal(t, "local", attachmentData["manager"]) + assert.Equal(t, "image/jpeg", attachmentData["content_type"]) + assert.Equal(t, "test-image.jpg", attachmentData["name"]) + assert.Equal(t, int64(1), attachmentData["public"]) + assert.Equal(t, []interface{}{"user", "admin"}, attachmentData["scope"]) + + // Test SaveAttachment (Update) + attachment["name"] = "updated-image.jpg" + attachment["bytes"] = 204800 + v, err = store.SaveAttachment(attachment) + assert.Nil(t, err) + assert.Equal(t, "test-file-123", v.(string)) + + // Verify update + attachmentData, err = store.GetAttachment(fileID) + assert.Nil(t, err) + assert.Equal(t, "updated-image.jpg", attachmentData["name"]) + assert.Equal(t, int64(204800), attachmentData["bytes"]) + + // Test GetAttachments with filters + resp, err := store.GetAttachments(AttachmentFilter{ + UID: "user-123", + Manager: "local", + Page: 1, + PageSize: 10, + }) + assert.Nil(t, err) + assert.Equal(t, 1, len(resp.Data)) + assert.Equal(t, "test-file-123", resp.Data[0]["file_id"]) + + // Test with non-existent file + nonExistentData, err := store.GetAttachment("non-existent-file") + assert.Error(t, err) + assert.Nil(t, nonExistentData) + assert.Contains(t, err.Error(), "is empty") + + // Test DeleteAttachment + err = store.DeleteAttachment(fileID) + assert.Nil(t, err) + + // Verify deletion + _, err = store.GetAttachment(fileID) + assert.Error(t, err) +} + +func TestXunKnowledgeCRUD(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + defer capsule.Schema().DropTableIfExists("__unit_test_conversation_knowledge") + + // Drop knowledge table before test + err := capsule.Schema().DropTableIfExists("__unit_test_conversation_knowledge") + 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.DeleteKnowledges(KnowledgeFilter{}) + assert.Nil(t, err) + + // Test SaveKnowledge (Create) + knowledge := map[string]interface{}{ + "collection_id": "test-collection-123", + "name": "Test Knowledge Collection", + "description": "A test knowledge collection for unit tests", + "uid": "user-123", + "public": true, + "readonly": false, + "system": false, + "sort": 100, + "cover": "cover-image.jpg", + "scope": []string{"user", "admin"}, + "option": map[string]interface{}{"embedding": "openai", "chunk_size": 1000}, + } + + v, err := store.SaveKnowledge(knowledge) + assert.Nil(t, err) + collectionID := v.(string) + assert.Equal(t, "test-collection-123", collectionID) + + // Test GetKnowledge + knowledgeData, err := store.GetKnowledge(collectionID) + assert.Nil(t, err) + assert.NotNil(t, knowledgeData) + assert.Equal(t, "test-collection-123", knowledgeData["collection_id"]) + assert.Equal(t, "Test Knowledge Collection", knowledgeData["name"]) + assert.Equal(t, "A test knowledge collection for unit tests", knowledgeData["description"]) + assert.Equal(t, "user-123", knowledgeData["uid"]) + assert.Equal(t, int64(1), knowledgeData["public"]) + assert.Equal(t, int64(100), knowledgeData["sort"]) + assert.Equal(t, []interface{}{"user", "admin"}, knowledgeData["scope"]) + assert.Equal(t, map[string]interface{}{"embedding": "openai", "chunk_size": float64(1000)}, knowledgeData["option"]) + + // Test SaveKnowledge (Update) + knowledge["name"] = "Updated Knowledge Collection" + knowledge["description"] = "Updated description" + knowledge["sort"] = 200 + v, err = store.SaveKnowledge(knowledge) + assert.Nil(t, err) + assert.Equal(t, "test-collection-123", v.(string)) + + // Verify update + knowledgeData, err = store.GetKnowledge(collectionID) + assert.Nil(t, err) + assert.Equal(t, "Updated Knowledge Collection", knowledgeData["name"]) + assert.Equal(t, "Updated description", knowledgeData["description"]) + assert.Equal(t, int64(200), knowledgeData["sort"]) + + // Test knowledge without sort field (should get default value 9999) + knowledgeWithoutSort := map[string]interface{}{ + "collection_id": "test-collection-456", + "name": "Test Knowledge Without Sort", + "description": "Test knowledge without explicit sort value", + "uid": "user-123", + } + v2, err := store.SaveKnowledge(knowledgeWithoutSort) + assert.Nil(t, err) + + // Verify default sort value + knowledgeData2, err := store.GetKnowledge(v2.(string)) + assert.Nil(t, err) + assert.Equal(t, int64(9999), knowledgeData2["sort"]) + + // Test GetKnowledges with filters + resp, err := store.GetKnowledges(KnowledgeFilter{ + UID: "user-123", + Keywords: "Updated", + Page: 1, + PageSize: 10, + }) + assert.Nil(t, err) + assert.Equal(t, 1, len(resp.Data)) + assert.Equal(t, "test-collection-123", resp.Data[0]["collection_id"]) + + // Test with non-existent collection + nonExistentData, err := store.GetKnowledge("non-existent-collection") + assert.Error(t, err) + assert.Nil(t, nonExistentData) + assert.Contains(t, err.Error(), "is empty") + + // Test DeleteKnowledge + err = store.DeleteKnowledge(collectionID) + assert.Nil(t, err) + err = store.DeleteKnowledge(v2.(string)) + assert.Nil(t, err) + + // Verify deletion + _, err = store.GetKnowledge(collectionID) + assert.Error(t, err) +} + +func TestXunKnowledgeFiltering(t *testing.T) { + test.Prepare(t, config.Conf) + defer test.Clean() + defer capsule.Schema().DropTableIfExists("__unit_test_conversation_knowledge") + + // Drop knowledge table before test + err := capsule.Schema().DropTableIfExists("__unit_test_conversation_knowledge") + 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) + } + + // Create test data for filtering tests + testKnowledges := []map[string]interface{}{} + for i := 0; i < 15; i++ { + knowledge := map[string]interface{}{ + "collection_id": fmt.Sprintf("test-collection-%d", i), + "name": fmt.Sprintf("Collection %d", i), + "description": fmt.Sprintf("Description for collection %d", i), + "uid": fmt.Sprintf("user-%d", i%3), + "public": i%2 == 0, + "readonly": i%3 == 0, + "system": i%4 == 0, + "sort": 100 + i*10, // Different sort values for testing ordering + "cover": fmt.Sprintf("cover%d.jpg", i), + } + id, err := store.SaveKnowledge(knowledge) + assert.Nil(t, err) + knowledge["collection_id"] = id + testKnowledges = append(testKnowledges, knowledge) + } + + // Test sorting functionality - should return results ordered by sort ASC then created_at DESC + respAll, err := store.GetKnowledges(KnowledgeFilter{ + Page: 1, + PageSize: 15, + }) + assert.Nil(t, err) + assert.Equal(t, 15, len(respAll.Data)) + + // Verify sort order - first item should have the smallest sort value + firstSort := respAll.Data[0]["sort"].(int64) + lastSort := respAll.Data[len(respAll.Data)-1]["sort"].(int64) + assert.LessOrEqual(t, firstSort, lastSort, "Results should be ordered by sort ASC") + + // More specific sort order verification + for i := 1; i < len(respAll.Data); i++ { + prevSort := respAll.Data[i-1]["sort"].(int64) + currSort := respAll.Data[i]["sort"].(int64) + assert.LessOrEqual(t, prevSort, currSort, "Sort order should be ascending") + } + + // Test filtering by UID + resp, err := store.GetKnowledges(KnowledgeFilter{ + UID: "user-0", + Page: 1, + PageSize: 10, + }) + assert.Nil(t, err) + assert.Greater(t, len(resp.Data), 0) + + // Test filtering by public status + publicTrue := true + resp, err = store.GetKnowledges(KnowledgeFilter{ + Public: &publicTrue, + Page: 1, + PageSize: 10, + }) + assert.Nil(t, err) + assert.Greater(t, len(resp.Data), 0) + + // Test filtering by readonly status + readonlyTrue := true + resp, err = store.GetKnowledges(KnowledgeFilter{ + Readonly: &readonlyTrue, + Page: 1, + PageSize: 10, + }) + assert.Nil(t, err) + assert.Greater(t, len(resp.Data), 0) + + // Test filtering by system status + systemTrue := true + resp, err = store.GetKnowledges(KnowledgeFilter{ + System: &systemTrue, + Page: 1, + PageSize: 10, + }) + assert.Nil(t, err) + assert.Greater(t, len(resp.Data), 0) + + // Test filtering by keywords + resp, err = store.GetKnowledges(KnowledgeFilter{ + Keywords: "Collection 1", + Page: 1, + PageSize: 10, + }) + assert.Nil(t, err) + assert.Greater(t, len(resp.Data), 0) + + // Test DeleteKnowledges with filter + count, err := store.DeleteKnowledges(KnowledgeFilter{ + UID: "user-0", + }) + assert.Nil(t, err) + assert.Greater(t, count, int64(0)) + + // Verify deletion + resp, err = store.GetKnowledges(KnowledgeFilter{ + UID: "user-0", + }) + assert.Nil(t, err) + assert.Equal(t, 0, len(resp.Data)) + + // Clean up all test data + _, err = store.DeleteKnowledges(KnowledgeFilter{}) + assert.Nil(t, err) +} + +func TestXunAttachmentFiltering(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) + } + + // Create test data for filtering tests + testAttachments := []map[string]interface{}{} + for i := 0; i < 15; i++ { + attachment := map[string]interface{}{ + "file_id": fmt.Sprintf("test-file-%d", i), + "uid": fmt.Sprintf("user-%d", i%3), + "manager": fmt.Sprintf("manager%d", i%2), + "content_type": fmt.Sprintf("type/%d", i%4), + "name": fmt.Sprintf("file%d.txt", i), + "guest": i%2 == 0, + "public": i%3 == 0, + "gzip": i%4 == 0, + "bytes": 1024 * (i + 1), + "collection_id": fmt.Sprintf("collection-%d", i%5), + } + id, err := store.SaveAttachment(attachment) + assert.Nil(t, err) + attachment["file_id"] = id + testAttachments = append(testAttachments, attachment) + } + + // Test filtering by UID + resp, err := store.GetAttachments(AttachmentFilter{ + UID: "user-0", + Page: 1, + PageSize: 10, + }) + assert.Nil(t, err) + assert.Greater(t, len(resp.Data), 0) + + // Test filtering by manager + resp, err = store.GetAttachments(AttachmentFilter{ + Manager: "manager0", + Page: 1, + PageSize: 10, + }) + assert.Nil(t, err) + assert.Greater(t, len(resp.Data), 0) + + // Test filtering by content_type + resp, err = store.GetAttachments(AttachmentFilter{ + ContentType: "type/0", + Page: 1, + PageSize: 10, + }) + assert.Nil(t, err) + assert.Greater(t, len(resp.Data), 0) + + // Test filtering by guest status + guestTrue := true + resp, err = store.GetAttachments(AttachmentFilter{ + Guest: &guestTrue, + Page: 1, + PageSize: 10, + }) + assert.Nil(t, err) + assert.Greater(t, len(resp.Data), 0) + + // Test filtering by public status + publicTrue := true + resp, err = store.GetAttachments(AttachmentFilter{ + Public: &publicTrue, + Page: 1, + PageSize: 10, + }) + assert.Nil(t, err) + assert.Greater(t, len(resp.Data), 0) + + // Test filtering by keywords + resp, err = store.GetAttachments(AttachmentFilter{ + Keywords: "file1", + Page: 1, + PageSize: 10, + }) + assert.Nil(t, err) + assert.Greater(t, len(resp.Data), 0) + + // Test DeleteAttachments with filter + count, err := store.DeleteAttachments(AttachmentFilter{ + Manager: "manager0", + }) + assert.Nil(t, err) + assert.Greater(t, count, int64(0)) + + // Verify deletion + resp, err = store.GetAttachments(AttachmentFilter{ + Manager: "manager0", + }) + assert.Nil(t, err) + assert.Equal(t, 0, len(resp.Data)) + + // Clean up all test data + _, err = store.DeleteAttachments(AttachmentFilter{}) + assert.Nil(t, err) +}