Netpulse_SasS/server/internal/store/store.go

// Package store — доступ до PostgreSQL/TimescaleDB.
//
// Головне правило шару: жоден запит не виконується без tenant_id.
// Ізоляція забезпечується двома незалежними механізмами, і це навмисно:
//
//  1. RLS у БД — політика tenant_isolation на кожній звичайній таблиці
//     з колонкою tenant_id. Вмикається через SET LOCAL app.tenant_id.
//  2. Явний предикат tenant_id у кожному запиті цього пакета.
//
// Дублювання потрібне, бо RLS НЕ працює на гіпертаблях: TimescaleDB
// не поєднує row level security зі стисненням. Саме туди йде вся
// телеметрія, тому для неї другий механізм — єдиний.
//
// Чи діє перший механізм — залежить не від цього пакета, а від того,
// якою роллю відкрито з'єднання. Роль із BYPASSRLS (а bootstrap-роль
// образу Postgres є суперкористувачем, тобто саме такою) обходить усі
// політики, і тоді з двох рубежів лишається один.
//
// Роль без BYPASSRLS заводить 0063, а видає їй пароль netpulse-migrate —
// на чистій базі одразу після накочування схеми. Тобто інсталяція,
// зроблена сьогодні, працює під політиками з першої секунди, і цей
// пакет ніколи не бачить суперкористувача. Інсталяції, старші за 0063,
// переходять окремо й покроково: deploy/RLS-EXISTING-INSTALL.md.
package store

import (
	"context"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/netpulse/netpulse/server/internal/gitstore"
)

type Store struct {
	pool *pgxpool.Pool

	// bg — пул для запитів, які за побудовою ходять поверх усіх
	// кабінетів. Ніколи не nil: без окремого DSN тут лежить той самий
	// pool, і тоді все поводиться рівно як до 0063.
	//
	// Що сюди можна класти — питання не зручності, а переліку. Дозволено
	// РІВНО три види запитів, і всі вони мають одну спільну рису: вони
	// з'ясовують, кого обробляти, а не обробляють:
	//
	//  1. Запити-шукачі черг: ClaimConfigJobs, ClaimCommandJobs,
	//     ClaimIdentifyRequests, DuePolicies, FetchEvents, ActiveRules,
	//     MirrorTargets, PendingRefDeletes, TenantsWithRetention,
	//     TenantIDs. Один UPDATE ... SKIP LOCKED ... RETURNING tenant_id
	//     на всю інсталяцію; розкласти його по кабінетах означає завести
	//     голодування (пояснення в 0063).
	//  2. Закриття тих самих одиниць роботи за первинним ключем, коли
	//     тенант у виклик уже не доходить: FinishConfigJob,
	//     ReapStuckJobs, FinishRefDelete тощо.
	//  3. Прибирання поверх кабінетів: PruneEvents, ArchiveResolved,
	//     видалення протермінованих квитків.
	//  4. Шляхи, які з'ясовують САМ кабінет, і тому виконуються до того,
	//     як його стало відомо: вхід людини й сесії (users.go),
	//     машинні токени (apitokens.go), автентифікація й реєстрація
	//     зонда (agents.go, enrollment.go), публічна панель за токеном
	//     (dashboards_public.go), квиток на завантаження (downloads.go),
	//     приймання телеметрії (telemetry.go).
	//
	//     Це не поступка зручності. RLS захищає доступ до даних
	//     кабінету — він за побудовою не може захистити запит, який
	//     питає «а який це кабінет?»: політика вимагає відповіді на те
	//     саме питання, яке запит ставить. Спроба лишити їх під RLS дає
	//     не безпеку, а 403 на вході: `core.tenants` під роллю без
	//     контексту віддає нуль рядків, і людина не може увійти в
	//     систему, де в неї є права.
	//
	//     Їхній захист — не політика, а предикат: хеш токена, user_id,
	//     термін дії квитка. Він був єдиним і до 0063, і лишається ним.
	//
	//     Окремо телеметрія: `ts.series` — єдина таблиця під RLS серед
	//     цілей запису. Ряди, що вже існують, писалися б і з основного
	//     пулу; впав би саме НОВИЙ ряд — новий хост, новий порт, новий
	//     рядок прототипу. Тобто поломка чекала б не перемикання, а
	//     наступної зміни в мережі.
	//
	// Усе інше — через InTenantTx. Перелік фактичних місць видно одним
	// grep-ом по `s.bg.`, і саме тому пул названий коротко: довге ім'я
	// в цьому проєкті було б непомітним серед s.pool.
	bg *pgxpool.Pool

	// Версіювання конфігів. Не обов'язкове: без нього NCM працює як і
	// раніше — тіла лежать зашифрованими в базі, а замість sha коміту
	// пишеться хеш вмісту. Порожнє значення тут — робочий стан, а не
	// недороблена ініціалізація.
	git *gitstore.Store
}

// UseGit під'єднує сховище версій. Окремим викликом, а не параметром
// New: конструктор потрібен і там, де Git не потрібен зовсім, — у
// міграторі, в утилітах командного рядка, у тестах.
func (s *Store) UseGit(g *gitstore.Store) { s.git = g }

func New(ctx context.Context, dsn string) (*Store, error) {
	cfg, err := pgxpool.ParseConfig(dsn)
	if err != nil {
		return nil, fmt.Errorf("розбір DSN: %w", err)
	}
	cfg.MaxConns = 16
	cfg.MinConns = 2
	cfg.MaxConnLifetime = time.Hour
	cfg.MaxConnIdleTime = 10 * time.Minute

	pool, err := pgxpool.NewWithConfig(ctx, cfg)
	if err != nil {
		return nil, err
	}
	if err := pool.Ping(ctx); err != nil {
		pool.Close()
		return nil, fmt.Errorf("ping: %w", err)
	}
	return &Store{pool: pool, bg: pool}, nil
}

// UseWorkerDSN відкриває друге з'єднання — для запитів поверх кабінетів.
//
// Окремим викликом, а не параметром New, з тієї ж причини, що й UseGit:
// конструктор потрібен і там, де другої ролі немає зовсім, — у тестах, у
// netpulse-secret, у netpulse-user. Порожній dsn — робочий стан: фонові
// запити просто йдуть основним пулом, як до 0063.
//
// Робочий він рівно доти, доки основний пул відкритий роллю з BYPASSRLS.
// Порожній dsn воркера ПРИ основному пулі на netpulse_app — стан, у
// якому інтерфейс працює, а фонові такти мовчки не знаходять нічого;
// цей пакет його не розрізняє й розрізняти не може, тому пару перевіряє
// netpulse-migrate і не дає стеку піднятись у такому вигляді.
//
// Два пули, а не SET ROLE на одному: SET ROLE живе в межах з'єднання, а
// з'єднання повертається в pgxpool і дістається наступному запиту. Тобто
// одна забута команда RESET перетворює запит користувача на запит з
// BYPASSRLS — рівно та помилка, проти якої вся ця робота. Окремий пул
// такої помилки не допускає взагалі: у нього інший пароль.
//
// Пул навмисно маленький. Фонових тактів десяток, найчастіший — раз на
// 5 секунд; усе, що росте з кількістю кабінетів, іде основним пулом.
func (s *Store) UseWorkerDSN(ctx context.Context, dsn string) error {
	if dsn == "" {
		return nil
	}
	cfg, err := pgxpool.ParseConfig(dsn)
	if err != nil {
		return fmt.Errorf("розбір DSN воркера: %w", err)
	}
	cfg.MaxConns = 4
	cfg.MinConns = 1
	cfg.MaxConnLifetime = time.Hour
	cfg.MaxConnIdleTime = 10 * time.Minute

	pool, err := pgxpool.NewWithConfig(ctx, cfg)
	if err != nil {
		return err
	}
	if err := pool.Ping(ctx); err != nil {
		pool.Close()
		return fmt.Errorf("ping воркера: %w", err)
	}
	s.bg = pool
	return nil
}

func (s *Store) Close() {
	if s.bg != nil && s.bg != s.pool {
		s.bg.Close()
	}
	s.pool.Close()
}

func (s *Store) Pool() *pgxpool.Pool { return s.pool }

// WorkerPool — пул для фонових тактів. Ним беруть з'єднання під
// advisory-блокування такту: блокування має жити стільки ж, скільки
// сам такт, а такт увесь ходить поверх кабінетів.
func (s *Store) WorkerPool() *pgxpool.Pool { return s.bg }

// InTenantTx виконує fn у транзакції з виставленим app.tenant_id.
//
// SET LOCAL, а не SET: значення живе рівно до кінця транзакції й не
// протікає на наступний запит, який візьме те саме з'єднання з пулу.
// Протікання тут означало б показ чужих даних, тому це не оптимізація,
// а вимога.
func (s *Store) InTenantTx(ctx context.Context, tenantID string, fn func(pgx.Tx) error) error {
	return s.inTenantTxOn(ctx, s.pool, tenantID, fn)
}

// InTenantTxBG — те саме, але пулом воркера.
//
// Потрібне рівно там, де дія тенант-скопована за змістом, а її окремий
// крок за побудовою крос-тенантний. Зараз такий випадок один:
// CreateUser бере глобального користувача через ON CONFLICT (username),
// і цей рядок може належати чужому кабінету — під політиками читання
// RETURNING його не побачить, і замість «підхопили наявного» вийде
// помилка унікальності (пояснення в 0063, розділ 6).
//
// app.tenant_id виставляється й тут, хоча BYPASSRLS його й не питає:
// значення читає не лише RLS, а й майбутні тригери й будь-хто, хто
// загляне в current_setting. Транзакція, що не вміє назвати свій
// кабінет, — це транзакція, яку не розслідувати.
func (s *Store) InTenantTxBG(ctx context.Context, tenantID string, fn func(pgx.Tx) error) error {
	return s.inTenantTxOn(ctx, s.bg, tenantID, fn)
}

func (s *Store) inTenantTxOn(ctx context.Context, pool *pgxpool.Pool, tenantID string, fn func(pgx.Tx) error) error {
	tx, err := pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback(ctx) }()

	if _, err := tx.Exec(ctx, "SELECT set_config('app.tenant_id', $1, true)", tenantID); err != nil {
		return fmt.Errorf("встановити app.tenant_id: %w", err)
	}
	if err := fn(tx); err != nil {
		return err
	}
	return tx.Commit(ctx)
}

// nullUUID перетворює порожній рядок на NULL — protobuf не має
// nullable-рядків, тому "" з дроту означає «не задано».
func nullUUID(s string) any {
	if s == "" {
		return nil
	}
	return s
}

func nullString(s string) any {
	if s == "" {
		return nil
	}
	return s
}