fn-registry agent
78 lines
2.5 KiB
Go
78 lines
2.5 KiB
Go
// Package datafactory opens and migrates data_factory.db. It is consumed by
|
|
// projects/fn_monitoring/apps/sqlite_api to expose REST + WS endpoints over
|
|
// the data_factory schema (nodes, connections, runs, databases).
|
|
//
|
|
// The C++ app in apps/data_factory (main.cpp) does NOT depend on this
|
|
// package. The Go subpackage lives in apps/data_factory/datafactory/ so it
|
|
// does not collide with main.cpp at the Go-toolchain level.
|
|
//
|
|
// Migrations are read from disk at runtime (apps/data_factory/migrations/
|
|
// relative to FN_REGISTRY_ROOT, or via an explicit migrationsDir argument).
|
|
// This keeps the SQL as the single source of truth — the C++ side reads
|
|
// the same files via its own bridge.
|
|
package datafactory
|
|
|
|
import (
|
|
"database/sql"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"sort"
|
|
"strings"
|
|
|
|
_ "github.com/mattn/go-sqlite3"
|
|
)
|
|
|
|
// Open opens data_factory.db at dbPath (creating the file and the parent
|
|
// directory if needed) and applies every *.sql under migrationsDir in
|
|
// lexical order. Idempotent — re-running on an already-migrated DB is a
|
|
// no-op (duplicate-column / already-exists errors are tolerated).
|
|
// Returns a RW *sql.DB; callers are responsible for Close().
|
|
func Open(dbPath, migrationsDir string) (*sql.DB, error) {
|
|
if err := os.MkdirAll(filepath.Dir(dbPath), 0o755); err != nil {
|
|
return nil, fmt.Errorf("datafactory: mkdir %s: %w", filepath.Dir(dbPath), err)
|
|
}
|
|
dsn := dbPath + "?_journal_mode=WAL&_busy_timeout=5000&_foreign_keys=on"
|
|
db, err := sql.Open("sqlite3", dsn)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("datafactory: open %s: %w", dbPath, err)
|
|
}
|
|
if err := db.Ping(); err != nil {
|
|
db.Close()
|
|
return nil, fmt.Errorf("datafactory: ping %s: %w", dbPath, err)
|
|
}
|
|
if err := applyMigrations(db, migrationsDir); err != nil {
|
|
db.Close()
|
|
return nil, fmt.Errorf("datafactory: migrate: %w", err)
|
|
}
|
|
return db, nil
|
|
}
|
|
|
|
func applyMigrations(conn *sql.DB, dir string) error {
|
|
entries, err := os.ReadDir(dir)
|
|
if err != nil {
|
|
return fmt.Errorf("read migrations dir %s: %w", dir, err)
|
|
}
|
|
names := make([]string, 0, len(entries))
|
|
for _, e := range entries {
|
|
if !e.IsDir() && strings.HasSuffix(e.Name(), ".sql") {
|
|
names = append(names, e.Name())
|
|
}
|
|
}
|
|
sort.Strings(names)
|
|
for _, n := range names {
|
|
b, err := os.ReadFile(filepath.Join(dir, n))
|
|
if err != nil {
|
|
return fmt.Errorf("%s: read: %w", n, err)
|
|
}
|
|
if _, err := conn.Exec(string(b)); err != nil {
|
|
if strings.Contains(err.Error(), "duplicate column") ||
|
|
strings.Contains(err.Error(), "already exists") {
|
|
continue
|
|
}
|
|
return fmt.Errorf("%s: %w", n, err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|