feat: add schema initialization logic (Phase 2)

- Create internal/database/schema.go with full initialization logic - Parse table names from schema.sql using regex (handles both formats) - Execute schema in atomic transaction - Verify all expected tables exist - Verify all critical functions exist (6 functions) - PostgreSQL advisory locking with 30-second timeout - Self-healing from partial/corrupted state - Load schema.sql from filesystem at runtime Features: - Defensive regex handles IF NOT EXISTS and legacy CREATE TABLE - Lock timeout prevents indefinite hangs - Function verification ensures sync operations work - Clear error messages with debug hints
2026-02-10 16:48:28 -05:00
parent 6ffb2ef6bc
commit 9d1a1e0228
1 changed files with 273 additions and 0 deletions
@@ -0,0 +1,273 @@
+package database
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"log"
+	"os"
+	"path/filepath"
+	"regexp"
+	"runtime"
+	"strings"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+var SchemaFile string
+
+func init() {
+	// Get the path to schema.sql relative to this file
+	_, currentFile, _, _ := runtime.Caller(0)
+	baseDir := filepath.Dir(currentFile)
+	schemaPath := filepath.Join(baseDir, "..", "..", "database", "schema", "schema.sql")
+
+	content, err := os.ReadFile(schemaPath)
+	if err != nil {
+		log.Fatalf("Failed to read schema.sql: %v", 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
+}