package database import ( "bufio" "context" "fmt" "log" "os" "path/filepath" "regexp" "strings" "time" "github.com/jackc/pgx/v5/pgxpool" ) var SchemaFile string func init() { // Try multiple locations for schema.sql // 1. Relative to working directory (for local dev) // 2. Relative to binary (for containerized deployment) locations := []string{ "database/schema/schema.sql", "/app/database/schema/schema.sql", "../database/schema/schema.sql", "../../database/schema/schema.sql", } var content []byte var err error for _, path := range locations { content, err = os.ReadFile(path) if err == nil { log.Printf("INFO: Loaded schema from %s", path) break } } if content == nil { // Last resort: try to find it relative to this file using runtime.Caller // This will work in development but not in containers absPath, _ := filepath.Abs("../../database/schema/schema.sql") content, err = os.ReadFile(absPath) if err != nil { log.Fatalf("Failed to read schema.sql from any location. Tried: %v. Last error: %v", locations, err) } } SchemaFile = string(content) } const ( // Pre-computed FNV-1a hash of "bookhoard:schema:init" // Generated using: generateLockID("bookhoard:schema:init") schemaInitLockID = 7804706162000639061 ) // Hash calculation function (for reference/testing) func generateLockID(key string) int64 { hash := uint64(14695981039346656037) // FNV offset basis for _, c := range key { hash ^= uint64(c) hash *= 1099511628211 // FNV prime } return int64(hash) } // parseTableNames extracts all table names from CREATE TABLE statements // Uses stream scanning for performance and memory efficiency // Handles both "CREATE TABLE" and "CREATE TABLE IF NOT EXISTS" formats func parseTableNames() ([]string, error) { scanner := bufio.NewScanner(strings.NewReader(SchemaFile)) pattern := regexp.MustCompile(`CREATE TABLE (?:IF NOT EXISTS )?(?:\w+\.)?(\w+)`) tables := make(map[string]bool) for scanner.Scan() { line := scanner.Text() matches := pattern.FindStringSubmatch(line) if len(matches) > 1 { tableName := matches[1] tables[tableName] = true } } if err := scanner.Err(); err != nil { return nil, fmt.Errorf("failed to scan schema.sql: %w", err) } // Convert map to unique slice result := make([]string, 0, len(tables)) for table := range tables { result = append(result, table) } return result, nil } // verifyTables checks all expected tables exist in database func verifyTables(ctx context.Context, db *pgxpool.Pool, expectedTables []string) error { rows, err := db.Query(ctx, ` SELECT table_name FROM information_schema.tables WHERE table_schema = 'public' AND table_type = 'BASE TABLE' `) if err != nil { return fmt.Errorf("failed to query existing tables: %w", err) } defer rows.Close() existingTables := make(map[string]bool) for rows.Next() { var tableName string if err := rows.Scan(&tableName); err != nil { return fmt.Errorf("failed to scan table name: %w", err) } existingTables[tableName] = true } if err := rows.Err(); err != nil { return fmt.Errorf("error iterating tables: %w", err) } // Check all expected tables exist var missing []string for _, expected := range expectedTables { if !existingTables[expected] { missing = append(missing, expected) } } if len(missing) > 0 { return fmt.Errorf("missing tables: %s", strings.Join(missing, ", ")) } return nil } // verifyFunctions checks all expected functions exist in database func verifyFunctions(ctx context.Context, db *pgxpool.Pool, expectedFunctions []string) error { rows, err := db.Query(ctx, ` SELECT routine_name FROM information_schema.routines WHERE routine_schema = 'public' AND routine_type = 'FUNCTION' `) if err != nil { return fmt.Errorf("failed to query existing functions: %w", err) } defer rows.Close() existingFunctions := make(map[string]bool) for rows.Next() { var functionName string if err := rows.Scan(&functionName); err != nil { return fmt.Errorf("failed to scan function name: %w", err) } existingFunctions[functionName] = true } if err := rows.Err(); err != nil { return fmt.Errorf("error iterating functions: %w", err) } // Check all expected functions exist var missing []string for _, expected := range expectedFunctions { if !existingFunctions[expected] { missing = append(missing, expected) } } if len(missing) > 0 { return fmt.Errorf("missing functions: %s", strings.Join(missing, ", ")) } return nil } // executeSchema runs the entire schema.sql in a single transaction func executeSchema(ctx context.Context, db *pgxpool.Pool) error { tx, err := db.Begin(ctx) if err != nil { return fmt.Errorf("failed to start transaction: %w", err) } defer tx.Rollback(ctx) _, err = tx.Exec(ctx, SchemaFile) if err != nil { log.Printf("HINT: Run manually to debug: psql -h localhost -U postgres -d bookhoard -f database/schema/schema.sql") return fmt.Errorf("schema execution failed: %w", err) } if err := tx.Commit(ctx); err != nil { return fmt.Errorf("failed to commit schema transaction: %w", err) } return nil } // Initialize ensures that database schema is up-to-date // Main entry point - runs idempotently on every startup with paranoid verification func Initialize(ctx context.Context, db *pgxpool.Pool) error { log.Println("🔐 Acquiring PostgreSQL advisory lock for schema initialization...") // Get database connection conn, err := db.Acquire(ctx) if err != nil { return fmt.Errorf("failed to acquire database connection: %w", err) } defer conn.Release() // Get advisory lock with timeout (prevents indefinite hangs) log.Printf("DEBUG: Attempting to acquire advisory lock %d...", schemaInitLockID) lockCtx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() var lockAcquired bool err = conn.QueryRow(lockCtx, "SELECT pg_try_advisory_lock($1)", schemaInitLockID).Scan(&lockAcquired) if err != nil { return fmt.Errorf("failed to acquire advisory lock: %w", err) } if !lockAcquired { log.Println("⏳ Another instance is initializing schema, waiting...") // Use pg_advisory_lock instead (blocks until available or timeout) _, err = conn.Exec(lockCtx, "SELECT pg_advisory_lock($1)", schemaInitLockID) if err != nil { return fmt.Errorf("timeout waiting for advisory lock: %w", err) } } log.Println("✅ Advisory lock acquired") defer func() { // Release lock when done _, err = conn.Exec(ctx, "SELECT pg_advisory_unlock($1)", schemaInitLockID) if err != nil { log.Printf("WARNING: Failed to release advisory lock: %v", err) } else { log.Println("🔓 Advisory lock released") } }() // Parse schema.sql to extract expected table names log.Println("📋 Parsing schema.sql for expected tables...") expectedTables, err := parseTableNames() if err != nil { return fmt.Errorf("failed to parse schema.sql: %w", err) } log.Printf("DEBUG: Found %d expected tables in schema.sql", len(expectedTables)) // Critical functions that must exist for app to function criticalFunctions := []string{ "update_updated_at_column", "detect_format_group", "convert_progress", "detect_conflict", "merge_progress", "bulk_update_progress_from_koreader", } // Execute schema in a single transaction log.Println("🔧 Executing schema.sql in transaction...") err = executeSchema(ctx, db) if err != nil { log.Printf("ERROR: Schema execution failed: %v", err) return fmt.Errorf("schema execution failed: %w", err) } log.Println("✅ Schema executed successfully") // Verify all expected tables exist log.Println("🔍 Verifying all expected tables exist...") err = verifyTables(ctx, db, expectedTables) if err != nil { log.Printf("ERROR: Schema verification failed: %v", err) return fmt.Errorf("schema verification failed: %w", err) } log.Println("✅ All expected tables verified") // Verify critical functions exist log.Println("🔍 Verifying critical functions exist...") err = verifyFunctions(ctx, db, criticalFunctions) if err != nil { log.Printf("ERROR: Function verification failed: %v", err) return fmt.Errorf("function verification failed: %w", err) } log.Println("✅ All critical functions verified") log.Printf("✅ Database schema initialization complete (%d tables, %d functions verified)", len(expectedTables), len(criticalFunctions)) return nil }