- Try multiple locations for schema.sql file - Support both local dev and containerized deployment paths - Add informative logging when schema is loaded - Prevent runtime.Caller issues in containers Locations checked: - database/schema/schema.sql (working directory) - /app/database/schema/schema.sql (container) - ../database/schema/schema.sql (relative) - ../../database/schema/schema.sql (relative) This fixes the 'no such file or directory' error in production containers.
295 lines
8.3 KiB
Go
295 lines
8.3 KiB
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",
|
|
}
|
|
|
|
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
|
|
}
|