Weknora/Enterprise-Knowledge-Base
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 支持数据库迁移脏状态自动恢复和Neo4j连接重试机制

Browse Source
This commit is contained in:
wizardchen
2025-12-08 17:40:10 +08:00
parent 8eb737ae28
commit c67a5d3949
9 changed files with 829 additions and 474 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+3
View File
@@ -54,6 +54,9 @@ REDIS_PREFIX=stream:
# 当使用本地存储时，文件保存的基础目录路径
LOCAL_STORAGE_BASE_DIR=./data/files
# 是否自动恢复脏数据
AUTO_RECOVER_DIRTY=true
TENANT_AES_KEY=weknorarag-api-key-secret-secret
# 是否开启知识图谱构建和检索（构建阶段需调用大模型，耗时较长）
docker-compose.dev.yml
-2
View File
@@ -12,8 +12,6 @@ services:
- POSTGRES_DB=${DB_NAME}
volumes:
- postgres-data-dev:/var/lib/postgresql/data
- ./migrations/paradedb/00-init-db.sql:/docker-entrypoint-initdb.d/00-init-db.sql
- ./migrations/paradedb/01-migrate-to-paradedb.sql:/docker-entrypoint-initdb.d/01-migrate-to-paradedb.sql
networks:
- WeKnora-network-dev
healthcheck:
docker-compose.yml
+1 -2
View File
@@ -59,6 +59,7 @@ services:
- DOCREADER_ADDR=docreader:50051
- STORAGE_TYPE=${STORAGE_TYPE:-}
- LOCAL_STORAGE_BASE_DIR=${LOCAL_STORAGE_BASE_DIR:-}
- AUTO_RECOVER_DIRTY=${AUTO_RECOVER_DIRTY:-true}
- MINIO_ENDPOINT=minio:9000
- MINIO_ACCESS_KEY_ID=${MINIO_ACCESS_KEY_ID:-minioadmin}
- MINIO_SECRET_ACCESS_KEY=${MINIO_SECRET_ACCESS_KEY:-minioadmin}
@@ -151,8 +152,6 @@ services:
- POSTGRES_DB=${DB_NAME}
volumes:
- postgres-data:/var/lib/postgresql/data
- ./migrations/paradedb/00-init-db.sql:/docker-entrypoint-initdb.d/00-init-db.sql
- ./migrations/paradedb/01-migrate-to-paradedb.sql:/docker-entrypoint-initdb.d/01-migrate-to-paradedb.sql
networks:
- WeKnora-network
healthcheck:
internal/container/container.go
+35 -10
View File
@@ -279,9 +279,13 @@ func initDatabase(cfg *config.Config) (*gorm.DB, error) {
// Run database migrations automatically (optional, can be disabled via env var)
// To disable auto-migration, set AUTO_MIGRATE=false
// To enable auto-recovery from dirty state, set AUTO_RECOVER_DIRTY=true
if os.Getenv("AUTO_MIGRATE") != "false" {
logger.Infof(context.Background(), "Running database migrations...")
if err := database.RunMigrations(migrateDSN); err != nil {
autoRecover := os.Getenv("AUTO_RECOVER_DIRTY") == "true"
if err := database.RunMigrationsWithOptions(migrateDSN, database.MigrationOptions{
AutoRecoverDirty: autoRecover,
}); err != nil {
// Log warning but don't fail startup - migrations might be handled externally
logger.Warnf(context.Background(), "Database migration failed: %v", err)
logger.Warnf(
@@ -539,21 +543,42 @@ func initOllamaService() (*ollama.OllamaService, error) {
}
func initNeo4jClient() (neo4j.Driver, error) {
ctx := context.Background()
if strings.ToLower(os.Getenv("NEO4J_ENABLE")) != "true" {
logger.Debugf(context.Background(), "NOT SUPPORT RETRIEVE GRAPH")
logger.Debugf(ctx, "NOT SUPPORT RETRIEVE GRAPH")
return nil, nil
}
uri := os.Getenv("NEO4J_URI")
username := os.Getenv("NEO4J_USERNAME")
password := os.Getenv("NEO4J_PASSWORD")
driver, err := neo4j.NewDriver(uri, neo4j.BasicAuth(username, password, ""))
if err != nil {
return nil, err
// Retry configuration
maxRetries := 30 // Max retry attempts
retryInterval := 2 * time.Second // Wait between retries
var driver neo4j.Driver
var err error
for attempt := 1; attempt <= maxRetries; attempt++ {
driver, err = neo4j.NewDriver(uri, neo4j.BasicAuth(username, password, ""))
if err != nil {
logger.Warnf(ctx, "Failed to create Neo4j driver (attempt %d/%d): %v", attempt, maxRetries, err)
time.Sleep(retryInterval)
continue
}
err = driver.VerifyAuthentication(ctx, nil)
if err == nil {
if attempt > 1 {
logger.Infof(ctx, "Successfully connected to Neo4j after %d attempts", attempt)
}
return driver, nil
}
logger.Warnf(ctx, "Failed to verify Neo4j authentication (attempt %d/%d): %v", attempt, maxRetries, err)
driver.Close(ctx)
time.Sleep(retryInterval)
}
err = driver.VerifyAuthentication(context.Background(), nil)
if err != nil {
return nil, err
}
return driver, nil
return nil, fmt.Errorf("failed to connect to Neo4j after %d attempts: %w", maxRetries, err)
}
internal/database/migration.go
+129 -35
View File
@@ -14,11 +14,28 @@ import (
// RunMigrations executes all pending database migrations
// This should be called during application startup
func RunMigrations(dsn string) error {
return RunMigrationsWithOptions(dsn, MigrationOptions{AutoRecoverDirty: false})
}
// MigrationOptions configures migration behavior
type MigrationOptions struct {
// AutoRecoverDirty when true, automatically attempts to recover from dirty state
// by forcing to the previous version and retrying the migration
AutoRecoverDirty bool
}
// RunMigrationsWithOptions executes all pending database migrations with custom options
func RunMigrationsWithOptions(dsn string, opts MigrationOptions) error {
ctx := context.Background()
logger.Infof(ctx, "Starting database migration...")
// Use versioned migrations directory
migrationsPath := "file://migrations/versioned"
m, err := migrate.New(migrationsPath, dsn)
if err != nil {
logger.Errorf(ctx, "Failed to create migrate instance: %v", err)
return fmt.Errorf("failed to create migrate instance: %w", err)
}
defer m.Close()
@@ -26,57 +43,93 @@ func RunMigrations(dsn string) error {
// Check current version and dirty state before migration
oldVersion, oldDirty, versionErr := m.Version()
if versionErr != nil && versionErr != migrate.ErrNilVersion {
// If we can't get version and it's not because database is empty, return error
logger.Errorf(ctx, "Failed to get migration version: %v", versionErr)
return fmt.Errorf("failed to get migration version: %w", versionErr)
}
// If database is in dirty state, return detailed error with fix instructions
if oldDirty {
// Calculate the version to force to (usually the previous version)
forceVersion := int(oldVersion) - 1
if oldVersion == 0 || forceVersion < 0 {
forceVersion = 0
}
return fmt.Errorf(
"database is in dirty state at version %d. This usually means a migration failed partway through. "+
"To fix this:\n"+
"1. Check if the migration partially applied changes and manually fix if needed\n"+
"2. Use the force command to set the version to the last successful migration (usually %d):\n"+
" ./scripts/migrate.sh force %d\n"+
" Or if using make: make migrate-force version=%d\n"+
"3. After fixing, restart the application to retry the migration",
oldVersion,
forceVersion,
forceVersion,
forceVersion,
)
if versionErr == migrate.ErrNilVersion {
logger.Infof(ctx, "Database has no migration history, will start from version 0")
} else {
logger.Infof(ctx, "Current migration version: %d, dirty: %v", oldVersion, oldDirty)
}
// Run all pending migrations
if err := m.Up(); err != nil && err != migrate.ErrNoChange {
// Check if error is due to dirty state (in case it became dirty during migration)
currentVersion, currentDirty, versionCheckErr := m.Version()
if versionCheckErr == nil && currentDirty {
// If database is in dirty state, try to recover or return error
if oldDirty {
logger.Warnf(ctx, "Database is in dirty state at version %d", oldVersion)
if opts.AutoRecoverDirty {
logger.Infof(ctx, "AutoRecoverDirty is enabled, attempting recovery...")
if err := recoverFromDirtyState(ctx, m, oldVersion); err != nil {
return err
}
// Update oldVersion after recovery
oldVersion, _, _ = m.Version()
} else {
// Calculate the version to force to (usually the previous version)
forceVersion := currentVersion - 1
if currentVersion == 0 {
forceVersion := int(oldVersion) - 1
if oldVersion == 0 || forceVersion < 0 {
forceVersion = 0
}
return fmt.Errorf(
"migration failed and database is now in dirty state at version %d. "+
"database is in dirty state at version %d. This usually means a migration failed partway through. "+
"To fix this:\n"+
"1. Check if the migration partially applied changes and manually fix if needed\n"+
"2. Use the force command to set the version to the last successful migration (usually %d):\n"+
" ./scripts/migrate.sh force %d\n"+
" Or if using make: make migrate-force version=%d\n"+
"3. After fixing, restart the application to retry the migration",
currentVersion,
"3. After fixing, restart the application to retry the migration\n"+
"Or enable AutoRecoverDirty option to automatically retry",
oldVersion,
forceVersion,
forceVersion,
forceVersion,
)
}
return fmt.Errorf("failed to run migrations: %w", err)
}
// Run all pending migrations
logger.Infof(ctx, "Running pending migrations...")
if err := m.Up(); err != nil && err != migrate.ErrNoChange {
logger.Errorf(ctx, "Migration failed: %v", err)
// Check if error is due to dirty state (in case it became dirty during migration)
currentVersion, currentDirty, versionCheckErr := m.Version()
if versionCheckErr == nil && currentDirty {
logger.Warnf(ctx, "Migration caused dirty state at version %d", currentVersion)
if opts.AutoRecoverDirty {
logger.Infof(ctx, "Attempting to recover from dirty state...")
// Try to recover and retry
if recoverErr := recoverFromDirtyState(ctx, m, currentVersion); recoverErr != nil {
return recoverErr
}
// Retry migration after recovery
logger.Infof(ctx, "Retrying migration after recovery...")
if retryErr := m.Up(); retryErr != nil && retryErr != migrate.ErrNoChange {
logger.Errorf(ctx, "Migration failed after recovery attempt: %v", retryErr)
return fmt.Errorf("migration failed after recovery attempt: %w", retryErr)
}
} else {
// Calculate the version to force to (usually the previous version)
forceVersion := currentVersion - 1
if currentVersion == 0 {
forceVersion = 0
}
return fmt.Errorf(
"migration failed and database is now in dirty state at version %d. "+
"To fix this:\n"+
"1. Check if the migration partially applied changes and manually fix if needed\n"+
"2. Use the force command to set the version to the last successful migration (usually %d):\n"+
" ./scripts/migrate.sh force %d\n"+
" Or if using make: make migrate-force version=%d\n"+
"3. After fixing, restart the application to retry the migration\n"+
"Or enable AutoRecoverDirty option to automatically retry",
currentVersion,
forceVersion,
forceVersion,
forceVersion,
)
}
} else {
return fmt.Errorf("failed to run migrations: %w", err)
}
}
// Get current version after migration
@@ -86,18 +139,59 @@ func RunMigrations(dsn string) error {
}
if oldVersion != version {
logger.Infof(context.Background(), "Database migrated from version %d to %d", oldVersion, version)
logger.Infof(ctx, "Database migrated from version %d to %d", oldVersion, version)
} else {
logger.Infof(context.Background(), "Database is up to date (version: %d)", version)
logger.Infof(ctx, "Database is up to date (version: %d)", version)
}
if dirty {
logger.Warnf(context.Background(), "Database is in dirty state! Manual intervention may be required.")
logger.Warnf(ctx, "Database is in dirty state! Manual intervention may be required.")
}
return nil
}
// recoverFromDirtyState attempts to recover from a dirty migration state
// by forcing to the previous version and allowing the migration to be retried
func recoverFromDirtyState(ctx context.Context, m *migrate.Migrate, dirtyVersion uint) error {
// Special case: if dirty at version 0 (init migration), we cannot go back further
// The only option is to force to version 0 and retry, but this requires the migration to be idempotent
if dirtyVersion == 0 {
logger.Warnf(ctx, "Database is in dirty state at version 0 (init migration). "+
"This is the initial migration, cannot rollback further. "+
"Will attempt to clear dirty flag and retry. "+
"Note: This only works if the init migration uses IF NOT EXISTS clauses.")
// Force to version -1 (no version) to allow re-running version 0
// This effectively tells migrate that no migrations have been applied
if err := m.Force(-1); err != nil {
return fmt.Errorf(
"failed to recover from dirty state at version 0. "+
"Manual intervention required:\n"+
"1. Check what was partially created in the database\n"+
"2. Either drop all created objects and retry, or\n"+
"3. Manually complete the migration and run: ./scripts/migrate.sh force 0\n"+
"Error: %w", err)
}
logger.Infof(ctx, "Cleared migration state, will retry from version 0")
return nil
}
forceVersion := int(dirtyVersion) - 1
logger.Warnf(ctx, "Database is in dirty state at version %d, attempting auto-recovery by forcing to version %d",
dirtyVersion, forceVersion)
// Force to previous version to clear dirty state
if err := m.Force(forceVersion); err != nil {
return fmt.Errorf("failed to force migration version during recovery: %w", err)
}
logger.Infof(ctx, "Successfully forced migration to version %d, migration will be retried", forceVersion)
return nil
}
// GetMigrationVersion returns the current migration version
func GetMigrationVersion() (uint, bool, error) {
dbURL := fmt.Sprintf(
migrations/versioned/000000_init.down.sql
+62
View File
@@ -0,0 +1,62 @@
-- Drop indexes for embeddings
DROP INDEX IF EXISTS embeddings_unique_source;
DROP INDEX IF EXISTS embeddings_search_idx;
-- Drop embeddings table
DROP TABLE IF EXISTS embeddings;
-- Drop indexes for chunks
DROP INDEX IF EXISTS idx_chunks_tenant_kg;
DROP INDEX IF EXISTS idx_chunks_parent_id;
DROP INDEX IF EXISTS idx_chunks_chunk_type;
-- Drop chunks table
DROP TABLE IF EXISTS chunks;
-- Drop indexes for messages
DROP INDEX IF EXISTS idx_messages_session_id;
-- Drop messages table
DROP TABLE IF EXISTS messages;
-- Drop indexes for sessions
DROP INDEX IF EXISTS idx_sessions_tenant_id;
-- Drop sessions table
DROP TABLE IF EXISTS sessions;
-- Drop indexes for knowledges
DROP INDEX IF EXISTS idx_knowledges_tenant_id;
DROP INDEX IF EXISTS idx_knowledges_base_id;
DROP INDEX IF EXISTS idx_knowledges_parse_status;
DROP INDEX IF EXISTS idx_knowledges_enable_status;
-- Drop knowledges table
DROP TABLE IF EXISTS knowledges;
-- Drop indexes for knowledge_bases
DROP INDEX IF EXISTS idx_knowledge_bases_tenant_id;
-- Drop knowledge_bases table
DROP TABLE IF EXISTS knowledge_bases;
-- Drop indexes for models
DROP INDEX IF EXISTS idx_models_type;
DROP INDEX IF EXISTS idx_models_source;
-- Drop models table
DROP TABLE IF EXISTS models;
-- Drop indexes for tenants
DROP INDEX IF EXISTS idx_tenants_api_key;
DROP INDEX IF EXISTS idx_tenants_status;
-- Drop tenants table
DROP TABLE IF EXISTS tenants;
-- Note: Extensions are not dropped as they may be used by other databases/schemas
-- If you want to drop extensions, uncomment the following lines:
-- DROP EXTENSION IF EXISTS pg_search;
-- DROP EXTENSION IF EXISTS pg_trgm;
-- DROP EXTENSION IF EXISTS vector;
-- DROP EXTENSION IF EXISTS "uuid-ossp";
migrations/versioned/000000_init.up.sql
+281
View File
@@ -0,0 +1,281 @@
-- Migration: 000000_init
-- Description: Initialize database schema
DO $$ BEGIN RAISE NOTICE '[Migration 000000] Starting initial database setup...'; END $$;
-- Create extensions
DO $$ BEGIN RAISE NOTICE '[Migration 000000] Creating extensions...'; END $$;
CREATE EXTENSION IF NOT EXISTS "uuid-ossp";
CREATE EXTENSION IF NOT EXISTS vector;
CREATE EXTENSION IF NOT EXISTS pg_trgm;
CREATE EXTENSION IF NOT EXISTS pg_search;
-- Create tenant table
DO $$ BEGIN RAISE NOTICE '[Migration 000000] Creating table: tenants'; END $$;
CREATE TABLE IF NOT EXISTS tenants (
id SERIAL PRIMARY KEY,
name VARCHAR(255) NOT NULL,
description TEXT,
api_key VARCHAR(64) NOT NULL,
retriever_engines JSONB NOT NULL DEFAULT '[]',
status VARCHAR(50) DEFAULT 'active',
business VARCHAR(255) NOT NULL,
storage_quota BIGINT NOT NULL DEFAULT 10737418240, -- 默认10GB配额(Bytes)
storage_used BIGINT NOT NULL DEFAULT 0, -- 已使用的存储空间(Bytes)
agent_config JSONB DEFAULT NULL,
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
deleted_at TIMESTAMP WITH TIME ZONE
);
COMMENT ON COLUMN tenants.agent_config IS 'Tenant-level agent configuration in JSON format';
-- Set the starting value for tenants id sequence (only if current value is less than 10000)
DO $$
DECLARE
current_val BIGINT;
BEGIN
SELECT last_value INTO current_val FROM tenants_id_seq;
IF current_val < 10000 THEN
ALTER SEQUENCE tenants_id_seq RESTART WITH 10000;
RAISE NOTICE '[Migration 000000] Set tenants_id_seq to start at 10000';
ELSE
RAISE NOTICE '[Migration 000000] tenants_id_seq already at % (>= 10000), skipping', current_val;
END IF;
EXCEPTION
WHEN undefined_table THEN
-- Sequence doesn't exist yet, will be created with table
RAISE NOTICE '[Migration 000000] tenants_id_seq not found, will be created with table';
END $$;
-- Add indexes
CREATE INDEX IF NOT EXISTS idx_tenants_api_key ON tenants(api_key);
CREATE INDEX IF NOT EXISTS idx_tenants_status ON tenants(status);
-- Create model table
DO $$ BEGIN RAISE NOTICE '[Migration 000000] Creating table: models'; END $$;
CREATE TABLE IF NOT EXISTS models (
id VARCHAR(64) PRIMARY KEY DEFAULT uuid_generate_v4(),
tenant_id INTEGER NOT NULL,
name VARCHAR(255) NOT NULL,
type VARCHAR(50) NOT NULL,
source VARCHAR(50) NOT NULL,
description TEXT,
parameters JSONB NOT NULL,
is_default BOOLEAN NOT NULL DEFAULT false,
status VARCHAR(50) NOT NULL DEFAULT 'active',
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
deleted_at TIMESTAMP WITH TIME ZONE
);
-- Add indexes for models
CREATE INDEX IF NOT EXISTS idx_models_type ON models(type);
CREATE INDEX IF NOT EXISTS idx_models_source ON models(source);
-- Create knowledge_base table
DO $$ BEGIN RAISE NOTICE '[Migration 000000] Creating table: knowledge_bases'; END $$;
CREATE TABLE IF NOT EXISTS knowledge_bases (
id VARCHAR(36) PRIMARY KEY DEFAULT uuid_generate_v4(),
name VARCHAR(255) NOT NULL,
description TEXT,
tenant_id INTEGER NOT NULL,
chunking_config JSONB NOT NULL DEFAULT '{"chunk_size": 512, "chunk_overlap": 50, "split_markers": ["\n\n", "\n", "。"], "keep_separator": true}',
image_processing_config JSONB NOT NULL DEFAULT '{"enable_multimodal": false, "model_id": ""}',
embedding_model_id VARCHAR(64) NOT NULL,
summary_model_id VARCHAR(64) NOT NULL,
rerank_model_id VARCHAR(64) NOT NULL,
cos_config JSONB NOT NULL DEFAULT '{}',
vlm_config JSONB NOT NULL DEFAULT '{}',
extract_config JSONB NULL DEFAULT NULL,
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
deleted_at TIMESTAMP WITH TIME ZONE
);
-- Add indexes for knowledge_bases
CREATE INDEX IF NOT EXISTS idx_knowledge_bases_tenant_id ON knowledge_bases(tenant_id);
-- Create knowledge table
DO $$ BEGIN RAISE NOTICE '[Migration 000000] Creating table: knowledges'; END $$;
CREATE TABLE IF NOT EXISTS knowledges (
id VARCHAR(36) PRIMARY KEY DEFAULT uuid_generate_v4(),
tenant_id INTEGER NOT NULL,
knowledge_base_id VARCHAR(36) NOT NULL,
type VARCHAR(50) NOT NULL,
title VARCHAR(255) NOT NULL,
description TEXT,
source VARCHAR(128) NOT NULL,
parse_status VARCHAR(50) NOT NULL DEFAULT 'unprocessed',
enable_status VARCHAR(50) NOT NULL DEFAULT 'enabled',
embedding_model_id VARCHAR(64),
file_name VARCHAR(255),
file_type VARCHAR(50),
file_size BIGINT,
file_path TEXT,
file_hash VARCHAR(64),
storage_size BIGINT NOT NULL DEFAULT 0, -- 存储大小(Byte)
metadata JSONB,
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
processed_at TIMESTAMP WITH TIME ZONE,
error_message TEXT,
deleted_at TIMESTAMP WITH TIME ZONE
);
-- Add indexes for knowledge
CREATE INDEX IF NOT EXISTS idx_knowledges_tenant_id ON knowledges(tenant_id);
CREATE INDEX IF NOT EXISTS idx_knowledges_base_id ON knowledges(knowledge_base_id);
CREATE INDEX IF NOT EXISTS idx_knowledges_parse_status ON knowledges(parse_status);
CREATE INDEX IF NOT EXISTS idx_knowledges_enable_status ON knowledges(enable_status);
-- Create session table
DO $$ BEGIN RAISE NOTICE '[Migration 000000] Creating table: sessions'; END $$;
CREATE TABLE IF NOT EXISTS sessions (
id VARCHAR(36) PRIMARY KEY DEFAULT uuid_generate_v4(),
tenant_id INTEGER NOT NULL,
title VARCHAR(255),
description TEXT,
knowledge_base_id VARCHAR(36),
max_rounds INTEGER NOT NULL DEFAULT 5,
enable_rewrite BOOLEAN NOT NULL DEFAULT true,
fallback_strategy VARCHAR(255) NOT NULL DEFAULT 'fixed',
fallback_response TEXT NOT NULL DEFAULT '很抱歉，我暂时无法回答这个问题。',
keyword_threshold FLOAT NOT NULL DEFAULT 0.5,
vector_threshold FLOAT NOT NULL DEFAULT 0.5,
rerank_model_id VARCHAR(64),
embedding_top_k INTEGER NOT NULL DEFAULT 10,
rerank_top_k INTEGER NOT NULL DEFAULT 10,
rerank_threshold FLOAT NOT NULL DEFAULT 0.65,
summary_model_id VARCHAR(64),
summary_parameters JSONB NOT NULL DEFAULT '{}',
agent_config JSONB DEFAULT NULL,
context_config JSONB DEFAULT NULL,
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
deleted_at TIMESTAMP WITH TIME ZONE
);
COMMENT ON COLUMN sessions.agent_config IS 'Session-level agent configuration in JSON format';
COMMENT ON COLUMN sessions.context_config IS 'LLM context management configuration (separate from message storage)';
-- Create Index for sessions
CREATE INDEX IF NOT EXISTS idx_sessions_tenant_id ON sessions(tenant_id);
-- Create message table
DO $$ BEGIN RAISE NOTICE '[Migration 000000] Creating table: messages'; END $$;
CREATE TABLE IF NOT EXISTS messages (
id VARCHAR(36) PRIMARY KEY DEFAULT uuid_generate_v4(),
request_id VARCHAR(36) NOT NULL,
session_id VARCHAR(36) NOT NULL,
role VARCHAR(50) NOT NULL,
content TEXT NOT NULL,
knowledge_references JSONB NOT NULL DEFAULT '[]',
agent_steps JSONB DEFAULT NULL,
is_completed BOOLEAN NOT NULL DEFAULT false,
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
deleted_at TIMESTAMP WITH TIME ZONE
);
COMMENT ON COLUMN messages.agent_steps IS 'Agent execution steps (reasoning process and tool calls)';
-- Create Index for messages
CREATE INDEX IF NOT EXISTS idx_messages_session_id ON messages(session_id);
-- Create chunks table
DO $$ BEGIN RAISE NOTICE '[Migration 000000] Creating table: chunks'; END $$;
CREATE TABLE IF NOT EXISTS chunks (
id VARCHAR(36) PRIMARY KEY DEFAULT uuid_generate_v4(),
tenant_id INTEGER NOT NULL,
knowledge_base_id VARCHAR(36) NOT NULL,
knowledge_id VARCHAR(36) NOT NULL,
content TEXT NOT NULL,
chunk_index INTEGER NOT NULL,
is_enabled BOOLEAN NOT NULL DEFAULT true,
start_at INTEGER NOT NULL,
end_at INTEGER NOT NULL,
pre_chunk_id VARCHAR(36),
next_chunk_id VARCHAR(36),
chunk_type VARCHAR(20) NOT NULL DEFAULT 'text',
parent_chunk_id VARCHAR(36),
image_info TEXT,
relation_chunks JSONB,
indirect_relation_chunks JSONB,
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
deleted_at TIMESTAMP WITH TIME ZONE
);
CREATE INDEX IF NOT EXISTS idx_chunks_tenant_kg ON chunks(tenant_id, knowledge_id);
CREATE INDEX IF NOT EXISTS idx_chunks_parent_id ON chunks(parent_chunk_id);
CREATE INDEX IF NOT EXISTS idx_chunks_chunk_type ON chunks(chunk_type);
-- Create embeddings table
DO $$ BEGIN RAISE NOTICE '[Migration 000000] Creating table: embeddings'; END $$;
CREATE TABLE IF NOT EXISTS embeddings (
id SERIAL PRIMARY KEY,
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
source_id VARCHAR(64) NOT NULL,
source_type INTEGER NOT NULL,
chunk_id VARCHAR(64),
knowledge_id VARCHAR(64),
knowledge_base_id VARCHAR(64),
content TEXT,
dimension INTEGER NOT NULL,
embedding halfvec
);
DO $$ BEGIN RAISE NOTICE '[Migration 000000] Creating indexes for embeddings (this may take a while)...'; END $$;
CREATE UNIQUE INDEX IF NOT EXISTS embeddings_unique_source ON embeddings(source_id, source_type);
-- Create BM25 search index (check if exists first)
DO $$
BEGIN
IF NOT EXISTS (SELECT 1 FROM pg_indexes WHERE indexname = 'embeddings_search_idx') THEN
CREATE INDEX embeddings_search_idx ON embeddings
USING bm25 (id, knowledge_base_id, content, knowledge_id, chunk_id)
WITH (
key_field = 'id',
text_fields = '{
"content": {
"tokenizer": {"type": "chinese_lindera"}
}
}'
);
RAISE NOTICE '[Migration 000000] Created BM25 index embeddings_search_idx';
ELSE
RAISE NOTICE '[Migration 000000] BM25 index embeddings_search_idx already exists';
END IF;
END $$;
-- Create HNSW indexes for vector search (check if exists first)
DO $$
BEGIN
IF NOT EXISTS (SELECT 1 FROM pg_indexes WHERE indexname = 'embeddings_embedding_idx' OR indexname LIKE 'embeddings_embedding%3584%') THEN
CREATE INDEX embeddings_embedding_idx_3584 ON embeddings
USING hnsw ((embedding::halfvec(3584)) halfvec_cosine_ops)
WITH (m = 16, ef_construction = 64)
WHERE (dimension = 3584);
RAISE NOTICE '[Migration 000000] Created HNSW index for dimension 3584';
ELSE
RAISE NOTICE '[Migration 000000] HNSW index for dimension 3584 already exists';
END IF;
END $$;
DO $$
BEGIN
IF NOT EXISTS (SELECT 1 FROM pg_indexes WHERE indexname = 'embeddings_embedding_idx_798' OR indexname LIKE 'embeddings_embedding%798%') THEN
CREATE INDEX embeddings_embedding_idx_798 ON embeddings
USING hnsw ((embedding::halfvec(798)) halfvec_cosine_ops)
WITH (m = 16, ef_construction = 64)
WHERE (dimension = 798);
RAISE NOTICE '[Migration 000000] Created HNSW index for dimension 798';
ELSE
RAISE NOTICE '[Migration 000000] HNSW index for dimension 798 already exists';
END IF;
END $$;
DO $$ BEGIN RAISE NOTICE '[Migration 000000] Initial database setup completed successfully!'; END $$;
migrations/versioned/000001_agent.up.sql
+313 -423
View File
@@ -1,10 +1,17 @@
BEGIN;
-- Migration: 000001_agent
-- Description: Add user authentication, agent config, MCP services and other enhancements
-- Create users table
DO $$ BEGIN RAISE NOTICE '[Migration 000001] Starting agent and authentication migration...'; END $$;
-- ============================================================================
-- Section 1: User Authentication Tables
-- ============================================================================
DO $$ BEGIN RAISE NOTICE '[Migration 000001] Creating table: users'; END $$;
CREATE TABLE IF NOT EXISTS users (
id VARCHAR(36) PRIMARY KEY DEFAULT uuid_generate_v4(),
username VARCHAR(100) NOT NULL UNIQUE,
email VARCHAR(255) NOT NULL UNIQUE,
username VARCHAR(100) NOT NULL,
email VARCHAR(255) NOT NULL,
password_hash VARCHAR(255) NOT NULL,
avatar VARCHAR(500),
tenant_id INTEGER,
@@ -14,6 +21,19 @@ CREATE TABLE IF NOT EXISTS users (
deleted_at TIMESTAMP WITH TIME ZONE
);
-- Add unique constraints if not exists
DO $$
BEGIN
IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'users_username_key') THEN
ALTER TABLE users ADD CONSTRAINT users_username_key UNIQUE (username);
RAISE NOTICE '[Migration 000001] Added unique constraint on users.username';
END IF;
IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'users_email_key') THEN
ALTER TABLE users ADD CONSTRAINT users_email_key UNIQUE (email);
RAISE NOTICE '[Migration 000001] Added unique constraint on users.email';
END IF;
END $$;
COMMENT ON TABLE users IS 'User accounts in the system';
COMMENT ON COLUMN users.id IS 'Unique identifier of the user';
COMMENT ON COLUMN users.username IS 'Username of the user';
@@ -29,23 +49,20 @@ CREATE INDEX IF NOT EXISTS idx_users_email ON users(email);
CREATE INDEX IF NOT EXISTS idx_users_tenant_id ON users(tenant_id);
CREATE INDEX IF NOT EXISTS idx_users_deleted_at ON users(deleted_at);
-- Add foreign key constraint for tenant_id (only if not exists)
-- Add foreign key constraint for tenant_id
DO $$
BEGIN
IF NOT EXISTS (
SELECT 1 FROM pg_constraint
WHERE conname = 'fk_users_tenant'
) THEN
ALTER TABLE users
ADD CONSTRAINT fk_users_tenant
IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'fk_users_tenant') THEN
ALTER TABLE users ADD CONSTRAINT fk_users_tenant
FOREIGN KEY (tenant_id) REFERENCES tenants(id) ON DELETE SET NULL;
RAISE NOTICE 'Added foreign key constraint fk_users_tenant';
ELSE
RAISE NOTICE 'Foreign key constraint fk_users_tenant already exists';
RAISE NOTICE '[Migration 000001] Added foreign key constraint fk_users_tenant';
END IF;
END $$;
-- Create auth_tokens table
-- Add can_access_all_tenants column to users
ALTER TABLE users ADD COLUMN IF NOT EXISTS can_access_all_tenants BOOLEAN NOT NULL DEFAULT FALSE;
DO $$ BEGIN RAISE NOTICE '[Migration 000001] Creating table: auth_tokens'; END $$;
CREATE TABLE IF NOT EXISTS auth_tokens (
id VARCHAR(36) PRIMARY KEY DEFAULT uuid_generate_v4(),
user_id VARCHAR(36) NOT NULL,
@@ -71,335 +88,174 @@ CREATE INDEX IF NOT EXISTS idx_auth_tokens_token ON auth_tokens(token);
CREATE INDEX IF NOT EXISTS idx_auth_tokens_token_type ON auth_tokens(token_type);
CREATE INDEX IF NOT EXISTS idx_auth_tokens_expires_at ON auth_tokens(expires_at);
-- Add foreign key constraint (only if not exists)
-- Add foreign key constraint for auth_tokens
DO $$
BEGIN
IF NOT EXISTS (
SELECT 1 FROM pg_constraint
WHERE conname = 'fk_auth_tokens_user'
) THEN
ALTER TABLE auth_tokens
ADD CONSTRAINT fk_auth_tokens_user
IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'fk_auth_tokens_user') THEN
ALTER TABLE auth_tokens ADD CONSTRAINT fk_auth_tokens_user
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE;
RAISE NOTICE 'Added foreign key constraint fk_auth_tokens_user';
ELSE
RAISE NOTICE 'Foreign key constraint fk_auth_tokens_user already exists';
RAISE NOTICE '[Migration 000001] Added foreign key constraint fk_auth_tokens_user';
END IF;
END $$;
-- Add is_temporary column to knowledge_bases
ALTER TABLE knowledge_bases
ADD COLUMN IF NOT EXISTS is_temporary BOOLEAN NOT NULL DEFAULT false;
-- ============================================================================
-- Section 2: Tenant Configuration Enhancements
-- ============================================================================
COMMENT ON COLUMN knowledge_bases.is_temporary IS 'Whether this knowledge base is temporary (ephemeral) and should be hidden from UI';
DO $$ BEGIN RAISE NOTICE '[Migration 000001] Adding tenant configuration columns...'; END $$;
-- Add context_config column to tenants
ALTER TABLE tenants
ADD COLUMN IF NOT EXISTS context_config JSONB;
ALTER TABLE tenants ADD COLUMN IF NOT EXISTS context_config JSONB;
COMMENT ON COLUMN tenants.context_config IS 'Global Context configuration for this tenant (default for all sessions)';
-- Add conversation_config column to tenants
ALTER TABLE tenants
ADD COLUMN IF NOT EXISTS conversation_config JSONB;
ALTER TABLE tenants ADD COLUMN IF NOT EXISTS conversation_config JSONB;
COMMENT ON COLUMN tenants.conversation_config IS 'Global Conversation configuration for this tenant (default for normal mode sessions)';
-- For tenants table: add agent_config
-- Add web_search_config column to tenants
ALTER TABLE tenants ADD COLUMN IF NOT EXISTS web_search_config JSONB DEFAULT NULL;
COMMENT ON COLUMN tenants.web_search_config IS 'Web search configuration for the tenant';
-- Ensure agent_config exists and is JSONB type
DO $$
BEGIN
IF NOT EXISTS (
SELECT 1
FROM information_schema.columns
WHERE table_name = 'tenants'
AND column_name = 'agent_config'
) THEN
ALTER TABLE tenants
ADD COLUMN agent_config JSONB DEFAULT NULL;
COMMENT ON COLUMN tenants.agent_config IS 'Tenant-level agent configuration in JSON format';
RAISE NOTICE 'Added agent_config column to tenants table';
ELSE
-- If field exists but type is JSON, convert to JSONB
IF EXISTS (
SELECT 1
FROM information_schema.columns
WHERE table_name = 'tenants'
AND column_name = 'agent_config'
AND data_type = 'json'
) THEN
ALTER TABLE tenants
ALTER COLUMN agent_config TYPE JSONB USING agent_config::jsonb;
RAISE NOTICE 'Converted tenants.agent_config from JSON to JSONB';
ELSE
RAISE NOTICE 'agent_config column already exists in tenants table';
END IF;
END IF;
END $$;
-- For sessions table: add agent_config
DO $$
BEGIN
IF NOT EXISTS (
SELECT 1
FROM information_schema.columns
WHERE table_name = 'sessions'
AND column_name = 'agent_config'
) THEN
ALTER TABLE sessions
ADD COLUMN agent_config JSONB DEFAULT NULL;
COMMENT ON COLUMN sessions.agent_config IS 'Session-level agent configuration in JSON format';
RAISE NOTICE 'Added agent_config column to sessions table';
ELSE
RAISE NOTICE 'agent_config column already exists in sessions table';
END IF;
END $$;
-- For sessions table: add context_config
DO $$
BEGIN
IF NOT EXISTS (
SELECT 1
FROM information_schema.columns
WHERE table_name = 'sessions'
AND column_name = 'context_config'
) THEN
ALTER TABLE sessions
ADD COLUMN context_config JSONB DEFAULT NULL;
COMMENT ON COLUMN sessions.context_config IS 'LLM context management configuration (separate from message storage)';
RAISE NOTICE 'Added context_config column to sessions table';
ELSE
RAISE NOTICE 'context_config column already exists in sessions table';
END IF;
END $$;
-- For messages table: add agent_steps
DO $$
BEGIN
IF NOT EXISTS (
SELECT 1
FROM information_schema.columns
WHERE table_name = 'messages'
AND column_name = 'agent_steps'
) THEN
ALTER TABLE messages
ADD COLUMN agent_steps JSONB DEFAULT NULL;
COMMENT ON COLUMN messages.agent_steps IS 'Agent execution steps (reasoning process and tool calls)';
RAISE NOTICE 'Added agent_steps column to messages table';
ELSE
RAISE NOTICE 'agent_steps column already exists in messages table';
END IF;
END $$;
-- Add GIN indexes for JSON fields
DO $$
BEGIN
-- For tenants.agent_config
IF NOT EXISTS (
SELECT 1 FROM pg_indexes
WHERE tablename = 'tenants'
AND indexname = 'idx_tenants_agent_config'
) THEN
IF EXISTS (
SELECT 1
FROM information_schema.columns
WHERE table_name = 'tenants'
AND column_name = 'agent_config'
AND data_type = 'jsonb'
) THEN
CREATE INDEX idx_tenants_agent_config ON tenants USING GIN (agent_config);
RAISE NOTICE 'Created index idx_tenants_agent_config';
ELSE
RAISE NOTICE 'Skipped index creation for tenants.agent_config (not JSONB type)';
END IF;
ELSE
RAISE NOTICE 'Index idx_tenants_agent_config already exists';
END IF;
-- For sessions.agent_config
IF NOT EXISTS (
SELECT 1 FROM pg_indexes
WHERE tablename = 'sessions'
AND indexname = 'idx_sessions_agent_config'
) THEN
CREATE INDEX idx_sessions_agent_config ON sessions USING GIN (agent_config);
RAISE NOTICE 'Created index idx_sessions_agent_config';
ELSE
RAISE NOTICE 'Index idx_sessions_agent_config already exists';
END IF;
-- For sessions.context_config
IF NOT EXISTS (
SELECT 1 FROM pg_indexes
WHERE tablename = 'sessions'
AND indexname = 'idx_sessions_context_config'
) THEN
CREATE INDEX idx_sessions_context_config ON sessions USING GIN (context_config);
RAISE NOTICE 'Created index idx_sessions_context_config';
ELSE
RAISE NOTICE 'Index idx_sessions_context_config already exists';
END IF;
-- For messages.agent_steps
IF NOT EXISTS (
SELECT 1 FROM pg_indexes
WHERE tablename = 'messages'
AND indexname = 'idx_messages_agent_steps'
) THEN
CREATE INDEX idx_messages_agent_steps ON messages USING GIN (agent_steps);
RAISE NOTICE 'Created index idx_messages_agent_steps';
ELSE
RAISE NOTICE 'Index idx_messages_agent_steps already exists';
END IF;
END $$;
COMMIT;
BEGIN;
WITH referenced_models AS (
SELECT embedding_model_id AS model_id FROM knowledge_bases WHERE deleted_at IS NULL AND embedding_model_id != ''
UNION
SELECT summary_model_id FROM knowledge_bases WHERE deleted_at IS NULL AND summary_model_id != ''
UNION
SELECT rerank_model_id FROM knowledge_bases WHERE deleted_at IS NULL AND rerank_model_id != ''
UNION
SELECT vlm_config ->> 'model_id'
FROM knowledge_bases
WHERE deleted_at IS NULL
AND vlm_config ->> 'model_id' IS NOT NULL
AND vlm_config ->> 'model_id' != ''
UNION
SELECT embedding_model_id FROM knowledges WHERE deleted_at IS NULL AND embedding_model_id IS NOT NULL AND embedding_model_id != ''
)
UPDATE models m
SET deleted_at = CURRENT_TIMESTAMP
WHERE m.deleted_at IS NULL
AND m.is_default = FALSE
AND m.tenant_id != 0
AND m.id NOT IN (SELECT model_id FROM referenced_models);
COMMIT;
BEGIN;
-- Create mcp_services table
CREATE TABLE IF NOT EXISTS mcp_services (
id VARCHAR(36) PRIMARY KEY,
tenant_id INTEGER NOT NULL,
name VARCHAR(255) NOT NULL,
description TEXT,
enabled BOOLEAN DEFAULT true,
transport_type VARCHAR(50) NOT NULL,
url VARCHAR(512) NOT NULL,
headers JSONB,
auth_config JSONB,
advanced_config JSONB,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
deleted_at TIMESTAMP
);
-- Create indexes
CREATE INDEX IF NOT EXISTS idx_mcp_services_tenant_id ON mcp_services(tenant_id);
CREATE INDEX IF NOT EXISTS idx_mcp_services_enabled ON mcp_services(enabled);
CREATE INDEX IF NOT EXISTS idx_mcp_services_deleted_at ON mcp_services(deleted_at);
-- Add comment to table
COMMENT ON TABLE mcp_services IS 'MCP service configurations';
-- Create trigger for updated_at
CREATE OR REPLACE FUNCTION update_mcp_services_updated_at()
RETURNS TRIGGER AS $$
BEGIN
NEW.updated_at = CURRENT_TIMESTAMP;
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER trigger_mcp_services_updated_at
BEFORE UPDATE ON mcp_services
FOR EACH ROW
EXECUTE FUNCTION update_mcp_services_updated_at();
COMMIT;
BEGIN;
-- Add web_search_config column to tenants table
DO $$
BEGIN
IF NOT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'tenants' AND column_name = 'web_search_config'
WHERE table_name = 'tenants' AND column_name = 'agent_config'
) THEN
ALTER TABLE tenants
ADD COLUMN web_search_config JSONB DEFAULT NULL;
ALTER TABLE tenants ADD COLUMN agent_config JSONB DEFAULT NULL;
RAISE NOTICE '[Migration 000001] Added agent_config column to tenants table';
ELSE
-- If field exists but type is JSON, convert to JSONB
IF EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'tenants' AND column_name = 'agent_config' AND data_type = 'json'
) THEN
ALTER TABLE tenants ALTER COLUMN agent_config TYPE JSONB USING agent_config::jsonb;
RAISE NOTICE '[Migration 000001] Converted tenants.agent_config from JSON to JSONB';
END IF;
END IF;
END $$;
COMMENT ON COLUMN tenants.agent_config IS 'Tenant-level agent configuration in JSON format';
-- ============================================================================
-- Section 3: Session Configuration Enhancements
-- ============================================================================
DO $$ BEGIN RAISE NOTICE '[Migration 000001] Adding session configuration columns...'; END $$;
-- Add agent_config column to sessions
ALTER TABLE sessions ADD COLUMN IF NOT EXISTS agent_config JSONB DEFAULT NULL;
COMMENT ON COLUMN sessions.agent_config IS 'Session-level agent configuration in JSON format';
-- Add context_config column to sessions
ALTER TABLE sessions ADD COLUMN IF NOT EXISTS context_config JSONB DEFAULT NULL;
COMMENT ON COLUMN sessions.context_config IS 'LLM context management configuration (separate from message storage)';
-- ============================================================================
-- Section 4: Messages Enhancements
-- ============================================================================
DO $$ BEGIN RAISE NOTICE '[Migration 000001] Adding messages enhancements...'; END $$;
-- Add agent_steps column to messages
ALTER TABLE messages ADD COLUMN IF NOT EXISTS agent_steps JSONB DEFAULT NULL;
COMMENT ON COLUMN messages.agent_steps IS 'Agent execution steps (reasoning process and tool calls)';
-- ============================================================================
-- Section 5: Knowledge Base Enhancements
-- ============================================================================
DO $$ BEGIN RAISE NOTICE '[Migration 000001] Adding knowledge base enhancements...'; END $$;
-- Add is_temporary column to knowledge_bases
ALTER TABLE knowledge_bases ADD COLUMN IF NOT EXISTS is_temporary BOOLEAN NOT NULL DEFAULT false;
COMMENT ON COLUMN knowledge_bases.is_temporary IS 'Whether this knowledge base is temporary (ephemeral) and should be hidden from UI';
-- Add type and faq_config columns
ALTER TABLE knowledge_bases ADD COLUMN IF NOT EXISTS type VARCHAR(32) NOT NULL DEFAULT 'document';
ALTER TABLE knowledge_bases ADD COLUMN IF NOT EXISTS faq_config JSONB;
-- Add question_generation_config column
ALTER TABLE knowledge_bases ADD COLUMN IF NOT EXISTS question_generation_config JSONB NULL;
-- Update existing rows with default type
UPDATE knowledge_bases SET type = 'document' WHERE type IS NULL OR type = '';
-- Drop rerank_model_id column if exists (moved to session level)
ALTER TABLE knowledge_bases DROP COLUMN IF EXISTS rerank_model_id;
-- ============================================================================
-- Section 6: Knowledges Enhancements
-- ============================================================================
DO $$ BEGIN RAISE NOTICE '[Migration 000001] Adding knowledges enhancements...'; END $$;
-- Add tag_id column
ALTER TABLE knowledges ADD COLUMN IF NOT EXISTS tag_id VARCHAR(36);
CREATE INDEX IF NOT EXISTS idx_knowledges_tag ON knowledges(tag_id);
-- Add summary_status column
ALTER TABLE knowledges ADD COLUMN IF NOT EXISTS summary_status VARCHAR(32) DEFAULT 'none';
CREATE INDEX IF NOT EXISTS idx_knowledges_summary_status ON knowledges(summary_status);
-- ============================================================================
-- Section 7: Chunks Enhancements
-- ============================================================================
DO $$ BEGIN RAISE NOTICE '[Migration 000001] Adding chunks enhancements...'; END $$;
-- Add metadata column
ALTER TABLE chunks ADD COLUMN IF NOT EXISTS metadata JSONB;
-- Add tag_id column
ALTER TABLE chunks ADD COLUMN IF NOT EXISTS tag_id VARCHAR(36);
CREATE INDEX IF NOT EXISTS idx_chunks_tag ON chunks(tag_id);
-- Add status field to track chunk processing state
ALTER TABLE chunks ADD COLUMN IF NOT EXISTS status INT NOT NULL DEFAULT 0;
-- Add content_hash field for quick content matching
ALTER TABLE chunks ADD COLUMN IF NOT EXISTS content_hash VARCHAR(64);
CREATE INDEX IF NOT EXISTS idx_chunks_content_hash ON chunks(content_hash);
-- ============================================================================
-- Section 8: Embeddings Enhancements
-- ============================================================================
DO $$ BEGIN RAISE NOTICE '[Migration 000001] Adding embeddings enhancements...'; END $$;
-- Add is_enabled column
ALTER TABLE embeddings ADD COLUMN IF NOT EXISTS is_enabled BOOLEAN DEFAULT TRUE;
CREATE INDEX IF NOT EXISTS idx_embeddings_is_enabled ON embeddings(is_enabled);
-- Add index for knowledge_base_id
CREATE INDEX IF NOT EXISTS idx_embeddings_knowledge_base_id ON embeddings(knowledge_base_id);
-- Reindex BM25 search index (idempotent - will rebuild if exists)
DO $$
BEGIN
IF EXISTS (SELECT 1 FROM pg_indexes WHERE indexname = 'embeddings_search_idx') THEN
REINDEX INDEX embeddings_search_idx;
RAISE NOTICE '[Migration 000001] Reindexed embeddings_search_idx';
END IF;
END $$;
-- Add or update comment
COMMENT ON COLUMN tenants.web_search_config IS 'Web search configuration for the tenant';
-- ============================================================================
-- Section 9: Models Enhancements
-- ============================================================================
COMMIT;
DO $$ BEGIN RAISE NOTICE '[Migration 000001] Adding models enhancements...'; END $$;
BEGIN;
-- Add stdio_config and env_vars columns
ALTER TABLE mcp_services
ADD COLUMN IF NOT EXISTS stdio_config JSONB,
ADD COLUMN IF NOT EXISTS env_vars JSONB;
-- Make url column optional
ALTER TABLE mcp_services
ALTER COLUMN url DROP NOT NULL;
-- Add check constraint
ALTER TABLE mcp_services
ADD CONSTRAINT chk_mcp_transport_config CHECK (
(transport_type = 'stdio' AND stdio_config IS NOT NULL) OR
(transport_type != 'stdio' AND url IS NOT NULL)
);
COMMIT;
BEGIN;
-- Add is_builtin column to models table
ALTER TABLE models
ADD COLUMN IF NOT EXISTS is_builtin BOOLEAN NOT NULL DEFAULT false;
-- Add index for is_builtin field
-- Add is_builtin column
ALTER TABLE models ADD COLUMN IF NOT EXISTS is_builtin BOOLEAN NOT NULL DEFAULT false;
CREATE INDEX IF NOT EXISTS idx_models_is_builtin ON models(is_builtin);
COMMIT;
-- ============================================================================
-- Section 10: Knowledge Tags Table
-- ============================================================================
BEGIN;
ALTER TABLE knowledge_bases
ADD COLUMN IF NOT EXISTS type VARCHAR(32) NOT NULL DEFAULT 'document',
ADD COLUMN IF NOT EXISTS faq_config JSONB;
UPDATE knowledge_bases
SET type = 'document'
WHERE type IS NULL OR type = '';
ALTER TABLE chunks
ADD COLUMN IF NOT EXISTS metadata JSONB;
COMMIT;
BEGIN;
-- Tag table (per knowledge base)
DO $$ BEGIN RAISE NOTICE '[Migration 000001] Creating table: knowledge_tags'; END $$;
CREATE TABLE IF NOT EXISTS knowledge_tags (
id VARCHAR(36) PRIMARY KEY,
tenant_id INTEGER NOT NULL,
@@ -412,123 +268,157 @@ CREATE TABLE IF NOT EXISTS knowledge_tags (
deleted_at TIMESTAMPTZ
);
CREATE UNIQUE INDEX IF NOT EXISTS idx_knowledge_tags_kb_name
ON knowledge_tags(tenant_id, knowledge_base_id, name);
CREATE UNIQUE INDEX IF NOT EXISTS idx_knowledge_tags_kb_name ON knowledge_tags(tenant_id, knowledge_base_id, name);
CREATE INDEX IF NOT EXISTS idx_knowledge_tags_kb ON knowledge_tags(tenant_id, knowledge_base_id);
CREATE INDEX IF NOT EXISTS idx_knowledge_tags_kb
ON knowledge_tags(tenant_id, knowledge_base_id);
-- ============================================================================
-- Section 11: MCP Services Table
-- ============================================================================
-- Tag reference on knowledges
ALTER TABLE knowledges
ADD COLUMN IF NOT EXISTS tag_id VARCHAR(36);
DO $$ BEGIN RAISE NOTICE '[Migration 000001] Creating table: mcp_services'; END $$;
CREATE TABLE IF NOT EXISTS mcp_services (
id VARCHAR(36) PRIMARY KEY,
tenant_id INTEGER NOT NULL,
name VARCHAR(255) NOT NULL,
description TEXT,
enabled BOOLEAN DEFAULT true,
transport_type VARCHAR(50) NOT NULL,
url VARCHAR(512),
headers JSONB,
auth_config JSONB,
advanced_config JSONB,
stdio_config JSONB,
env_vars JSONB,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
deleted_at TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_knowledges_tag
ON knowledges(tag_id);
CREATE INDEX IF NOT EXISTS idx_mcp_services_tenant_id ON mcp_services(tenant_id);
CREATE INDEX IF NOT EXISTS idx_mcp_services_enabled ON mcp_services(enabled);
CREATE INDEX IF NOT EXISTS idx_mcp_services_deleted_at ON mcp_services(deleted_at);
-- Tag reference on chunks
ALTER TABLE chunks
ADD COLUMN IF NOT EXISTS tag_id VARCHAR(36);
COMMENT ON TABLE mcp_services IS 'MCP service configurations';
CREATE INDEX IF NOT EXISTS idx_chunks_tag
ON chunks(tag_id);
-- Create or replace trigger function for updated_at
CREATE OR REPLACE FUNCTION update_mcp_services_updated_at()
RETURNS TRIGGER AS $$
BEGIN
NEW.updated_at = CURRENT_TIMESTAMP;
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
COMMIT;
-- Create trigger if not exists
DO $$
BEGIN
IF NOT EXISTS (SELECT 1 FROM pg_trigger WHERE tgname = 'trigger_mcp_services_updated_at') THEN
CREATE TRIGGER trigger_mcp_services_updated_at
BEFORE UPDATE ON mcp_services
FOR EACH ROW
EXECUTE FUNCTION update_mcp_services_updated_at();
RAISE NOTICE '[Migration 000001] Created trigger trigger_mcp_services_updated_at';
END IF;
END $$;
BEGIN;
-- ============================================================================
-- Section 12: GIN Indexes for JSONB Fields
-- ============================================================================
-- Add is_enabled column to embeddings table
ALTER TABLE embeddings
ADD COLUMN IF NOT EXISTS is_enabled BOOLEAN DEFAULT TRUE;
DO $$ BEGIN RAISE NOTICE '[Migration 000001] Creating GIN indexes for JSONB fields...'; END $$;
-- Create index for is_enabled column
CREATE INDEX IF NOT EXISTS idx_embeddings_is_enabled
ON embeddings(is_enabled);
DO $$
BEGIN
-- For tenants.agent_config
IF NOT EXISTS (SELECT 1 FROM pg_indexes WHERE indexname = 'idx_tenants_agent_config') THEN
IF EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'tenants' AND column_name = 'agent_config' AND data_type = 'jsonb'
) THEN
CREATE INDEX idx_tenants_agent_config ON tenants USING GIN (agent_config);
RAISE NOTICE '[Migration 000001] Created index idx_tenants_agent_config';
END IF;
END IF;
-- For sessions.agent_config
IF NOT EXISTS (SELECT 1 FROM pg_indexes WHERE indexname = 'idx_sessions_agent_config') THEN
CREATE INDEX idx_sessions_agent_config ON sessions USING GIN (agent_config);
RAISE NOTICE '[Migration 000001] Created index idx_sessions_agent_config';
END IF;
-- For sessions.context_config
IF NOT EXISTS (SELECT 1 FROM pg_indexes WHERE indexname = 'idx_sessions_context_config') THEN
CREATE INDEX idx_sessions_context_config ON sessions USING GIN (context_config);
RAISE NOTICE '[Migration 000001] Created index idx_sessions_context_config';
END IF;
-- For messages.agent_steps
IF NOT EXISTS (SELECT 1 FROM pg_indexes WHERE indexname = 'idx_messages_agent_steps') THEN
CREATE INDEX idx_messages_agent_steps ON messages USING GIN (agent_steps);
RAISE NOTICE '[Migration 000001] Created index idx_messages_agent_steps';
END IF;
END $$;
COMMIT;
-- ============================================================================
-- Section 13: Data Migrations
-- ============================================================================
BEGIN;
DO $$ BEGIN RAISE NOTICE '[Migration 000001] Running data migrations...'; END $$;
-- Create index for knowledge_base_id
CREATE INDEX IF NOT EXISTS idx_embeddings_knowledge_base_id
ON embeddings(knowledge_base_id);
-- Clean up unreferenced models (soft delete)
DO $$
DECLARE
affected_rows INTEGER;
BEGIN
WITH referenced_models AS (
SELECT embedding_model_id AS model_id FROM knowledge_bases WHERE deleted_at IS NULL AND embedding_model_id != ''
UNION
SELECT summary_model_id FROM knowledge_bases WHERE deleted_at IS NULL AND summary_model_id != ''
UNION
SELECT vlm_config ->> 'model_id'
FROM knowledge_bases
WHERE deleted_at IS NULL
AND vlm_config ->> 'model_id' IS NOT NULL
AND vlm_config ->> 'model_id' != ''
UNION
SELECT embedding_model_id FROM knowledges WHERE deleted_at IS NULL AND embedding_model_id IS NOT NULL AND embedding_model_id != ''
)
UPDATE models m
SET deleted_at = CURRENT_TIMESTAMP
WHERE m.deleted_at IS NULL
AND m.is_default = FALSE
AND m.tenant_id != 0
AND m.id NOT IN (SELECT model_id FROM referenced_models WHERE model_id IS NOT NULL);
GET DIAGNOSTICS affected_rows = ROW_COUNT;
IF affected_rows > 0 THEN
RAISE NOTICE '[Migration 000001] Soft deleted % unreferenced models', affected_rows;
END IF;
END $$;
COMMIT;
-- Update models tenant_id from knowledge_bases references
DO $$
DECLARE
affected_rows INTEGER;
BEGIN
WITH tenant_source AS (
SELECT kb.embedding_model_id AS model_id, kb.tenant_id
FROM knowledge_bases kb
WHERE kb.tenant_id IS NOT NULL AND kb.embedding_model_id IS NOT NULL AND kb.embedding_model_id <> ''
UNION
SELECT kb.summary_model_id AS model_id, kb.tenant_id
FROM knowledge_bases kb
WHERE kb.tenant_id IS NOT NULL AND kb.summary_model_id IS NOT NULL AND kb.summary_model_id <> ''
)
UPDATE models m
SET tenant_id = ts.tenant_id
FROM tenant_source ts
WHERE m.id = ts.model_id AND m.tenant_id = 0;
GET DIAGNOSTICS affected_rows = ROW_COUNT;
IF affected_rows > 0 THEN
RAISE NOTICE '[Migration 000001] Updated tenant_id for % models', affected_rows;
END IF;
END $$;
REINDEX INDEX embeddings_search_idx;
BEGIN;
-- Add status field to track chunk processing state
ALTER TABLE chunks
ADD COLUMN IF NOT EXISTS status INT NOT NULL DEFAULT 0;
-- Add content_hash field for quick content matching
ALTER TABLE chunks
ADD COLUMN IF NOT EXISTS content_hash VARCHAR(64);
-- Create index on content_hash
CREATE INDEX IF NOT EXISTS idx_chunks_content_hash
ON chunks(content_hash);
COMMIT;
BEGIN;
WITH tenant_source AS (
SELECT
kb.embedding_model_id AS model_id,
kb.tenant_id
FROM knowledge_bases kb
WHERE kb.tenant_id IS NOT NULL
AND kb.embedding_model_id IS NOT NULL
AND kb.embedding_model_id <> ''
UNION
SELECT
kb.summary_model_id AS model_id,
kb.tenant_id
FROM knowledge_bases kb
WHERE kb.tenant_id IS NOT NULL
AND kb.summary_model_id IS NOT NULL
AND kb.summary_model_id <> ''
UNION
SELECT
kb.rerank_model_id AS model_id,
kb.tenant_id
FROM knowledge_bases kb
WHERE kb.tenant_id IS NOT NULL
AND kb.rerank_model_id IS NOT NULL
AND kb.rerank_model_id <> ''
)
UPDATE models m
SET tenant_id = ts.tenant_id
FROM tenant_source ts
WHERE m.id = ts.model_id
AND m.tenant_id = 0;
COMMIT;
BEGIN;
ALTER TABLE knowledge_bases
DROP COLUMN IF EXISTS rerank_model_id;
COMMIT;
BEGIN;
ALTER TABLE users
ADD COLUMN IF NOT EXISTS can_access_all_tenants BOOLEAN NOT NULL DEFAULT FALSE;
COMMIT;
ALTER TABLE knowledge_bases
ADD COLUMN IF NOT EXISTS question_generation_config JSONB NULL ;
ALTER TABLE knowledges ADD COLUMN IF NOT EXISTS summary_status VARCHAR(32) DEFAULT 'none';
-- Add index for efficient querying
CREATE INDEX IF NOT EXISTS idx_knowledges_summary_status ON knowledges(summary_status);
DO $$ BEGIN RAISE NOTICE '[Migration 000001] Migration completed successfully!'; END $$;
scripts/migrate.sh
+5 -2
View File
@@ -96,10 +96,13 @@ case "$1" in
if [ -z "$2" ]; then
echo "Error: Version number is required"
echo "Usage: $0 force <version>"
echo "Note: Use -1 to reset to no version (allows re-running all migrations)"
exit 1
fi
echo "Forcing migration version to $2..."
migrate -path ${MIGRATIONS_DIR} -database ${DB_URL} force $2
VERSION="$2"
echo "Forcing migration version to $VERSION..."
# Use env to pass the command, avoiding shell flag parsing issues with negative numbers
env migrate -path "${MIGRATIONS_DIR}" -database "${DB_URL}" force -- "$VERSION"
;;
goto)
if [ -z "$2" ]; then