Bridge-and-Join-s/internal/lkgateway/pgmove.go

package lkgateway

// База bj на своём сервере PostgreSQL организации (06.10.2026). Установщик кладёт базу bj на этот же сервер —
// так и остаётся по умолчанию. Если у организации свой сервер PostgreSQL (с резервным копированием и
// администратором баз), базу можно держать там: шаг «База bj» проверяет подключение и права, переносит данные
// (pg_dump этой базы → psql в пустую базу на том сервере), сверяет число строк в каждой таблице, переключает
// bj и перезапускает его. Прежняя база на этом сервере не удаляется — остаётся запасной копией.
// Только PostgreSQL: другие СУБД bj для своей базы не поддерживает.

import (
	"bufio"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"net/url"
	"os"
	"os/exec"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/jackc/pgx/v5"
)

// pgTarget — сервер PostgreSQL организации для базы bj (из формы).
type pgTarget struct{ Host, Port, DB, User, Password, SSL string }

func pgTargetFromForm(r *http.Request) (pgTarget, string) {
	t := pgTarget{
		Host: strings.TrimSpace(r.FormValue("pg_host")), Port: strings.TrimSpace(r.FormValue("pg_port")),
		DB: strings.TrimSpace(r.FormValue("pg_db")), User: strings.TrimSpace(r.FormValue("pg_user")),
		Password: r.FormValue("pg_password"), SSL: r.FormValue("pg_ssl"),
	}
	if t.Port == "" {
		t.Port = "5432"
	}
	switch t.SSL {
	case "disable", "require", "verify-full":
	default:
		t.SSL = "prefer"
	}
	switch {
	case t.Host == "" || strings.ContainsAny(t.Host, "/ @"):
		return t, "укажите адрес сервера PostgreSQL"
	case pgLocalHost(t.Host):
		return t, "это адрес этого же сервера — база bj и так здесь; укажите сервер PostgreSQL организации"
	case t.DB == "" || t.User == "":
		return t, "укажите базу и пользователя"
	}
	if p, err := strconv.Atoi(t.Port); err != nil || p < 1 || p > 65535 {
		return t, "порт — число от 1 до 65535"
	}
	return t, ""
}

// dsn — строка подключения; withPass=false — без пароля (для pg_dump/psql пароль идёт через PGPASSWORD,
// чтобы не светиться в списке процессов).
func (t pgTarget) dsn(withPass bool) string {
	u := url.URL{Scheme: "postgres", Host: net.JoinHostPort(t.Host, t.Port), Path: "/" + t.DB}
	if withPass && t.Password != "" {
		u.User = url.UserPassword(t.User, t.Password)
	} else {
		u.User = url.User(t.User)
	}
	u.RawQuery = "sslmode=" + t.SSL
	return u.String()
}

func pgLocalHost(h string) bool {
	h = strings.Trim(strings.ToLower(h), "[]")
	return h == "" || h == "localhost" || h == "127.0.0.1" || h == "::1" || strings.HasPrefix(h, "/")
}

// pgDSNLocal — база по этой строке подключения на этом же сервере.
func pgDSNLocal(dsn string) bool {
	u, err := url.Parse(strings.TrimSpace(dsn))
	return err != nil || pgLocalHost(u.Hostname())
}

// pgSplitDSN — строка подключения без пароля и пароль отдельно (для внешних программ).
func pgSplitDSN(dsn string) (string, string) {
	u, err := url.Parse(strings.TrimSpace(dsn))
	if err != nil || u.User == nil {
		return dsn, ""
	}
	pw, _ := u.User.Password()
	u.User = url.User(u.User.Username())
	return u.String(), pw
}

// pgInfo — что известно о базе: версия сервера, права, таблицы.
type pgInfo struct {
	Version   string
	Major     int
	CanCreate bool // CREATE в базе: создать схемы bj
	PublicOK  bool // CREATE в схеме public (с PostgreSQL 15 по умолчанию есть только у владельца базы)
	Tables    []string
	Rows      map[string]int64 // для сверки после переноса
}

func pgInspect(ctx context.Context, dsn string, counts bool) (pgInfo, error) {
	info := pgInfo{Rows: map[string]int64{}}
	cctx, cancel := context.WithTimeout(ctx, 15*time.Second)
	defer cancel()
	conn, err := pgx.Connect(cctx, dsn)
	if err != nil {
		return info, err
	}
	defer conn.Close(context.Background())
	var num int
	if err := conn.QueryRow(cctx, "SELECT current_setting('server_version'), current_setting('server_version_num')::int, has_database_privilege(current_user, current_database(), 'CREATE'), COALESCE(has_schema_privilege(current_user, 'public', 'CREATE'), false)").
		Scan(&info.Version, &num, &info.CanCreate, &info.PublicOK); err != nil {
		return info, err
	}
	info.Major = num / 10000
	rows, err := conn.Query(cctx, "SELECT quote_ident(table_schema)||'.'||quote_ident(table_name) FROM information_schema.tables WHERE table_type='BASE TABLE' AND table_schema NOT IN ('pg_catalog','information_schema') AND table_schema NOT LIKE 'pg_toast%' ORDER BY 1")
	if err != nil {
		return info, err
	}
	for rows.Next() {
		var n string
		if rows.Scan(&n) == nil {
			info.Tables = append(info.Tables, n)
		}
	}
	rows.Close()
	if counts {
		for _, t := range info.Tables {
			var c int64
			if err := conn.QueryRow(ctx, "SELECT count(*) FROM "+t).Scan(&c); err != nil {
				return info, fmt.Errorf("подсчёт строк %s: %w", t, err)
			}
			info.Rows[t] = c
		}
	}
	return info, nil
}

// pgPublicUsed — в базе bj есть таблицы в схеме public (тогда на новом сервере нужно право создавать в ней).
func pgPublicUsed(tables []string) bool {
	for _, t := range tables {
		if strings.HasPrefix(t, "public.") {
			return true
		}
	}
	return false
}

// pgCheckTarget — годится ли сервер организации: подключение, версия не старше нашей, права, пустая база.
func pgCheckTarget(ctx context.Context, src pgInfo, t pgTarget) (pgInfo, string) {
	dst, err := pgInspect(ctx, t.dsn(true), false)
	if err != nil {
		return dst, "не удалось подключиться к " + t.Host + ":" + t.Port + ", база " + t.DB + ": " + pgErrRu(err)
	}
	switch {
	case src.Major > 0 && dst.Major < src.Major:
		return dst, fmt.Sprintf("на сервере организации PostgreSQL %s, а база bj сейчас на %s — нужен PostgreSQL %d или новее", dst.Version, src.Version, src.Major)
	case !dst.CanCreate:
		return dst, "у пользователя " + t.User + " нет права создавать объекты в базе " + t.DB + " (GRANT CREATE ON DATABASE " + t.DB + " TO " + t.User + ") — проще всего сделать его владельцем базы"
	case pgPublicUsed(src.Tables) && !dst.PublicOK:
		return dst, "у пользователя " + t.User + " нет права создавать таблицы в схеме public базы " + t.DB + " (с PostgreSQL 15 оно есть только у владельца базы) — сделайте его владельцем базы: ALTER DATABASE " + t.DB + " OWNER TO " + t.User
	case len(dst.Tables) > 0:
		return dst, fmt.Sprintf("в базе %s уже есть таблицы (%d, например %s) — для bj нужна пустая база", t.DB, len(dst.Tables), dst.Tables[0])
	}
	return dst, ""
}

func pgErrRu(err error) string {
	s := err.Error()
	switch {
	case strings.Contains(s, "password authentication failed"):
		return "неверный пользователь или пароль"
	case strings.Contains(s, "does not exist") && strings.Contains(s, "database"):
		return "такой базы нет — создайте пустую базу"
	case strings.Contains(s, "no pg_hba.conf entry"):
		return "сервер не пускает с адреса bj — администратору баз нужно разрешить подключение в pg_hba.conf"
	case strings.Contains(s, "connection refused"), strings.Contains(s, "i/o timeout"), strings.Contains(s, "no route to host"):
		return "сервер не отвечает на этом адресе и порту (" + s + ")"
	}
	return s
}

// pgCopy — данные базы bj на сервер организации: pg_dump (простой SQL без владельцев и прав) → psql.
func pgCopy(ctx context.Context, srcDSN string, t pgTarget) error {
	src, srcPass := pgSplitDSN(srcDSN)
	dump := exec.CommandContext(ctx, "pg_dump", "--no-owner", "--no-privileges", "--format=plain", "--dbname="+src)
	dump.Env = append(os.Environ(), "PGPASSWORD="+srcPass)
	load := exec.CommandContext(ctx, "psql", "--quiet", "--no-psqlrc", "-v", "ON_ERROR_STOP=1", "--dbname="+t.dsn(false))
	load.Env = append(os.Environ(), "PGPASSWORD="+t.Password) // без PGOPTIONS: lc_messages — только суперпользователю
	pipe, err := dump.StdoutPipe()
	if err != nil {
		return err
	}
	load.Stdin = pipe
	var dumpErr, loadErr strings.Builder
	dump.Stderr, load.Stderr, load.Stdout = &dumpErr, &loadErr, io.Discard
	if err := load.Start(); err != nil {
		return fmt.Errorf("psql: %w", err)
	}
	if err := dump.Start(); err != nil {
		_ = load.Process.Kill()
		return fmt.Errorf("pg_dump: %w", err)
	}
	de := dump.Wait()
	le := load.Wait()
	if de != nil {
		return fmt.Errorf("pg_dump: %v %s", de, lastLine(dumpErr.String()))
	}
	if le != nil {
		return fmt.Errorf("загрузка на сервер организации: %v %s", le, lastLine(loadErr.String()))
	}
	return nil
}

func lastLine(s string) string {
	sc := bufio.NewScanner(strings.NewReader(s))
	last := ""
	for sc.Scan() {
		if l := strings.TrimSpace(sc.Text()); l != "" {
			last = l
		}
	}
	return last
}

// pgCompare — расхождения числа строк (пусто — всё сошлось).
func pgCompare(src, dst pgInfo) []string {
	var bad []string
	for _, t := range src.Tables {
		if d, ok := dst.Rows[t]; !ok {
			bad = append(bad, t+": нет на новом сервере")
		} else if d != src.Rows[t] {
			bad = append(bad, fmt.Sprintf("%s: было %d строк, стало %d", t, src.Rows[t], d))
		}
	}
	sort.Strings(bad)
	return bad
}

func (h *setupHandlers) registerPGMove(mux *http.ServeMux) {
	mux.HandleFunc("/admin/setup/postgres/move", h.pgMove)
}

// pgMove — POST /admin/setup/postgres/move: check=1 — только проверить сервер организации; иначе перенести
// и переключиться. Ход — строками JSON (как подготовка DEPO): этапы и итог.
func (h *setupHandlers) pgMove(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "method", http.StatusMethodNotAllowed)
		return
	}
	w.Header().Set("Content-Type", "application/x-ndjson; charset=utf-8")
	w.Header().Set("Cache-Control", "no-store")
	w.Header().Set("X-Accel-Buffering", "no")
	fl, _ := w.(http.Flusher)
	enc := json.NewEncoder(w)
	send := func(v any) {
		_ = enc.Encode(v)
		if fl != nil {
			fl.Flush()
		}
	}
	step := func(key, state, text string) { send(map[string]string{"step": key, "state": state, "text": text}) }
	done := func(ok bool, msg string) {
		send(map[string]any{"done": true, "ok": ok, "msg": msg, "go": "/admin/wizard?step=2&flash=" + url.QueryEscape(msg)})
	}
	ctx, cancel := context.WithTimeout(r.Context(), 30*time.Minute)
	defer cancel()

	cur := h.rc.Snapshot().Postgres.DSN
	t, bad := pgTargetFromForm(r)
	if bad != "" {
		done(false, bad)
		return
	}
	if !pgDSNLocal(cur) {
		done(false, "база bj уже не на этом сервере — перенос делается с базы на этом сервере")
		return
	}
	step("src", "run", "База bj на этом сервере")
	src, err := pgInspect(ctx, cur, false)
	if err != nil {
		step("src", "err", pgErrRu(err))
		done(false, "база bj на этом сервере недоступна: "+pgErrRu(err))
		return
	}
	step("src", "ok", fmt.Sprintf("PostgreSQL %s, таблиц: %d", src.Version, len(src.Tables)))
	step("dst", "run", "Подключение к серверу организации "+t.Host+":"+t.Port+", база "+t.DB)
	dst, why := pgCheckTarget(ctx, src, t)
	if why != "" {
		step("dst", "err", why)
		done(false, why)
		return
	}
	step("dst", "ok", "PostgreSQL "+dst.Version+", база пустая, права есть")
	if r.FormValue("check") != "" {
		done(true, "Сервер организации подходит: PostgreSQL "+dst.Version+", база "+t.DB+" пустая, права есть. Можно переносить.")
		return
	}

	step("copy", "run", "Копирование данных bj (pg_dump → psql)")
	start := time.Now()
	if err := pgCopy(ctx, cur, t); err != nil {
		step("copy", "err", err.Error())
		done(false, "данные не перенесены: "+err.Error()+". bj продолжает работать с базой на этом сервере; на сервере организации очистите базу "+t.DB+" и повторите")
		return
	}
	step("copy", "ok", "за "+humanDur(time.Since(start)))
	step("verify", "run", "Сверка: число строк в каждой таблице")
	srcN, err1 := pgInspect(ctx, cur, true)
	dstN, err2 := pgInspect(ctx, t.dsn(true), true)
	if err1 != nil || err2 != nil {
		e := err1
		if e == nil {
			e = err2
		}
		step("verify", "err", e.Error())
		done(false, "сверка не выполнена: "+e.Error()+". bj продолжает работать с базой на этом сервере")
		return
	}
	if diff := pgCompare(srcN, dstN); len(diff) > 0 {
		// пока шло копирование, bj мог дописать строки — это не повод переключаться на неполную базу
		step("verify", "err", strings.Join(diff, "; "))
		done(false, "после копирования данные не совпали ("+strings.Join(diff, "; ")+"). bj продолжает работать с базой на этом сервере; очистите базу "+t.DB+" на сервере организации и повторите в спокойное время")
		return
	}
	var total int64
	for _, n := range srcN.Rows {
		total += n
	}
	step("verify", "ok", fmt.Sprintf("таблиц %d, строк %d — совпадает", len(srcN.Tables), total))
	step("switch", "run", "Переключение bj на сервер организации и перезапуск")
	if err := h.rc.UpdatePostgres(PostgresSettings{DSN: t.dsn(true), Local: cur}); err != nil {
		step("switch", "err", err.Error())
		done(false, "настройки не сохранены: "+err.Error())
		return
	}
	step("switch", "ok", "bj перезапускается — через полминуты он работает с базой на "+t.Host)
	log.Printf("postgres: база bj перенесена на %s:%s/%s (%d таблиц, %d строк); прежняя база на этом сервере сохранена", t.Host, t.Port, t.DB, len(srcN.Tables), total)
	done(true, "База bj перенесена на сервер организации "+t.Host+", база "+t.DB+". Прежняя база на этом сервере сохранена как запасная копия.")
	go func() { // перезапуск после ответа: страница успевает показать итог
		time.Sleep(2 * time.Second)
		rctx, rcancel := context.WithTimeout(context.Background(), 2*time.Minute)
		defer rcancel()
		if _, err := restartUnit(rctx, "bj-server"); err != nil {
			log.Printf("postgres: перезапуск после переноса: %v", err)
		}
	}()
}