bookhoard/internal/database/schema.go

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",
		// Integration tests run with CWD=cmd/server/tests, three levels
		// below the repo root — two up only reaches cmd/.
		"../../../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"
	// (FNV offset basis 14695981039346656037, FNV prime 1099511628211)
	schemaInitLockID = 7804706162000639061
)

// 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()
		// Skip SQL comments: a doc line like "-- ... declared in CREATE
		// TABLE above" would otherwise register a phantom table and fail
		// startup verification.
		if strings.HasPrefix(strings.TrimSpace(line), "--") {
			continue
		}
		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
}

// schemaMigrationIdentity is the synthetic identity the schema
// migration transaction declares for RLS (plan appendix A11). The
// schema's dedup DO block SELECTs/DELETEs media_items, and with the
// visibility policies live (FORCE applies to the owner) that SELECT
// needs app.user_id set even in this maintenance tx. The identity has
// NO library_visibility rows, so the policy grants it full visibility
// — appropriate for the server's own schema pass, and unreachable by
// requests (this tx only ever runs at startup, on the raw pool).
const schemaMigrationIdentity = "00000000-0000-0000-0000-000000000001"

// 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)

	// Declare the migration identity BEFORE the file executes (A11):
	// policy'd tables are queried by schema.sql itself on re-runs.
	if _, err := tx.Exec(ctx, `SELECT set_config('app.user_id', $1, true)`, schemaMigrationIdentity); err != nil {
		return fmt.Errorf("schema execution failed (declare migration identity): %w", err)
	}

	_, 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
}