Netpulse_SasS/server/internal/store/retention_policy.go

package store

import (
	"context"
	"errors"
	"fmt"
	"regexp"
	"sort"
	"time"

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

// Строки зберігання даних: скільки живе кожен вид того, що система
// накопичує, і чим саме воно прибирається.
//
// ЧОМУ ЦЕ ПЕРЕЛІК, А НЕ ОДНА ЦИФРА
//
// Спільний строк на все — найпоширеніша й найдорожча помилка в цій
// задачі. Сирі лічильники портів цінні тиждень: далі на них ніхто не
// дивиться, бо є п'ятихвилинні згортки з тими самими висновками й у
// сто разів меншим обсягом. Журнал аудиту цінний роками й важить
// копійки. Одна ручка на обидва означає або гори марних лічильників,
// або знищений журнал — залежно від того, під що її налаштували.
//
// ЧОМУ ГІПЕРТАБЛИЦІ ПРИБИРАЄМО НЕ СВОЇМ КОДОМ
//
// DELETE у Postgres не звільняє місця на диску: рядок позначається
// мертвим, сторінка лишається файлу, і повернути її здатен лише
// VACUUM FULL — а він блокує таблицю цілком і вимагає стільки ж
// вільного місця, скільки вона займає. Тобто на переповненому томі,
// саме тоді, коли це потрібно, він не спрацює взагалі.
//
// drop_chunks у TimescaleDB працює інакше: чанк — окрема таблиця, і
// його зносить DROP TABLE. Місце повертається файловій системі
// негайно. Заради цього TimescaleDB і взято; писати замість нього свій
// цикл DELETE означало б платити за базу, якою не користуєшся.
//
// Для звичайних таблиць (прогони команд, завдання збору, прогони
// пошуку) вибору немає — там DELETE партіями. Про його межу сказано
// людині прямо, у самій формі: місце звільняється всередині таблиці й
// перевикористовується, а не повертається на диск.
//
// ЧОГО ТУТ НЕМАЄ
//
// Версій конфігів. Їхня політика (0037, ncm_retention.go) влаштована не
// за віком, а як «останні N АБО молодші за M днів» із захистом
// останньої версії хоста. Звести це до однієї цифри в добах означало б
// утратити рівно ту гарантію, заради якої воно написане. Сторінка
// показує цю політику поруч із рештою й відправляє міняти її туди, де
// вона живе: двох ручок до одних даних бути не має.

// RetentionKindInfo — те, що словник знає про вид даних.
//
// Словник живе в Go, а не в браузері й не в базі, з тієї ж причини, що
// й словник дій аудиту: перелік видів — властивість того, що продукт
// НАКОПИЧУЄ. Копія в TypeScript розійшлася б із цією на першому ж
// новому рядку, і побачив би це лише той, хто відкриє обидва файли
// поруч.
type RetentionKindInfo struct {
	Kind  string `json:"kind"`
	Label string `json:"label"`
	Group string `json:"group"`

	// Note — від чого залежить обсяг цього виду. Читає це людина, яка
	// вперше вирішує, скільки тут ставити.
	Note string `json:"note"`

	// MinDays — нижче ставити не можна, MinReason каже чому. Не примха:
	// для сирих даних під згортками це межа, за якою згортки не
	// встигають порахуватись, і графік за минулий місяць зникає
	// назавжди.
	MinDays   int    `json:"min_days"`
	MinReason string `json:"min_reason,omitempty"`

	// SourceKind — вид, із якого цей рахується. Згортка не має жити
	// менше за своє джерело: інакше видалення сирих даних лишає діру,
	// яку вже нічим не заповнити.
	SourceKind string `json:"source_kind,omitempty"`

	// SizeAlso — таблиці, які зникають РАЗОМ із цією за каскадом і тому
	// враховуються в розмірі. Для прогону команд це його цілі: саме в
	// них лежать транскрипти, тобто вся вага.
	SizeAlso []string `json:"-"`
}

// retentionCatalog — усі види даних, якими керує форма.
//
// Порядок фіксований і йде від найгустішого до найрідшого: людина, яка
// відкрила сторінку заради місця на диску, має першими побачити рядки,
// де воно справді витрачається.
//
// Джерело завжди стоїть перед своєю згорткою — це перевіряє init().
var retentionCatalog = []RetentionKindInfo{
	{
		Kind: "icmp_raw", Group: "Доступність",
		Label:     "Сирі виміри ICMP",
		Note:      "Рядок на кожен хост на кожному такті опитування — найчастіший запис у системі. 500 хостів раз на 30 секунд дають близько 1,4 млн рядків на добу.",
		MinDays:   2,
		MinReason: "п'ятихвилинні згортки рахуються з відставанням у три години, і менший строк знищив би сирі дані до того, як їх встигнуть згорнути",
	},
	{
		Kind: "icmp_5m", Group: "Доступність",
		Label: "Доступність за 5 хвилин", SourceKind: "icmp_raw",
		Note:      "Те, з чого малюються графіки за тиждень і місяць. У десятки разів менше за сирі виміри й достатнє для будь-якого питання, старшого за добу.",
		MinDays:   7,
		MinReason: "годинні згортки рахуються з відставанням у дві доби",
	},
	{
		Kind: "icmp_1h", Group: "Доступність",
		Label: "Доступність за годину", SourceKind: "icmp_5m",
		Note:      "База для звітів про доступність за місяць і рік. Найдешевший ряд у системі: 24 рядки на хост на добу.",
		MinDays:   30,
		MinReason: "місячні звіти читають саме звідси",
	},
	{
		Kind: "ifc_raw", Group: "Порти",
		Label:     "Сирі лічильники портів",
		Note:      "Найбільший обсяг у перерахунку на хост: множник — кількість портів. Комутатор на 24 порти пише 24 рядки на такт, і 500 таких хостів дають десятки мільйонів рядків на добу.",
		MinDays:   2,
		MinReason: "п'ятихвилинні згортки рахуються з відставанням у три години",
	},
	{
		Kind: "ifc_5m", Group: "Порти",
		Label: "Трафік портів за 5 хвилин", SourceKind: "ifc_raw",
		Note:      "Звідси беруться графіки навантаження за тиждень і місяць, зокрема максимуми — те, за чим планують розширення каналу.",
		MinDays:   7,
		MinReason: "годинні згортки рахуються з відставанням у дві доби",
	},
	{
		Kind: "ifc_1h", Group: "Порти",
		Label: "Трафік портів за годину", SourceKind: "ifc_5m",
		Note:      "Довга пам'ять про завантаження каналів: питання «як воно було торік у листопаді» відповідається лише звідси.",
		MinDays:   30,
		MinReason: "довгі звіти читають саме звідси",
	},
	{
		Kind: "metrics_raw", Group: "Метрики",
		Label:     "Сирі метрики",
		Note:      "Усе, що збирають плагіни: завантаження процесора, пам'ять, температура, будь-яка OID. Один ряд на кожну пару «хост + метрика», і кількість рядів залежить від шаблонів, а не від нас.",
		MinDays:   2,
		MinReason: "п'ятихвилинні згортки рахуються з відставанням у три години",
	},
	{
		Kind: "metrics_5m", Group: "Метрики",
		Label: "Метрики за 5 хвилин", SourceKind: "metrics_raw",
		Note:      "Те, що показують графіки хоста за будь-який період, довший за добу.",
		MinDays:   7,
		MinReason: "годинні згортки рахуються з відставанням у дві доби",
	},
	{
		Kind: "metrics_1h", Group: "Метрики",
		Label: "Метрики за годину", SourceKind: "metrics_5m",
		Note:      "Найдовша пам'ять про метрики й найдешевша.",
		MinDays:   30,
		MinReason: "довгі звіти читають саме звідси",
	},
	{
		Kind: "syslog", Group: "Події з мережі",
		Label:     "Syslog",
		Note:      "Обсяг не залежить від наших налаштувань узагалі: скільки надішле залізо, стільки й ляже. Одна балакуча коробка з увімкненим debug здатна перевершити всю решту телеметрії разом.",
		MinDays:   3,
		MinReason: "стиснення вмикається на третю добу, і менший строк робить його марним",
	},
	{
		Kind: "traps", Group: "Події з мережі",
		Label:   "SNMP-трапи",
		Note:    "Рідші за syslog, але саме вони запускають позачерговий збір конфігу — і саме їх шукають, з'ясовуючи, чому конфіг зібрався о третій ночі.",
		MinDays: 7,
	},
	{
		Kind: "device_status", Group: "Стан",
		Label:     "Історія станів хостів",
		Note:      "Рядок на кожну зміну «вгору/вниз», а не на такт. Мала в спокійній мережі й велика в тій, що блимає, — тобто росте найшвидше там, де на неї найбільше дивляться.",
		MinDays:   7,
		MinReason: "звіти про доступність за тиждень будуються саме на ній",
	},
	{
		Kind: "link_status", Group: "Стан",
		Label:   "Історія станів лінків",
		Note:    "Те саме для зв'язків на мапі. Обсяг залежить від кількості намальованих ліній і від стабільності мережі.",
		MinDays: 7,
	},
	{
		Kind: "alerts_history", Group: "Алерти",
		Label:     "Історія алертів",
		Note:      "Закриті алерти переїжджають сюди фоном, щоб активна таблиця лишалась маленькою. Росте назавжди: рядок на кожну аварію кожного хоста.",
		MinDays:   7,
		MinReason: "розбір «що в нас падало цього тижня» читає саме її",
	},
	{
		Kind: "notifications", Group: "Алерти",
		Label:   "Журнал доставки сповіщень",
		Note:    "Рядок на кожну спробу відправити повідомлення в кожен канал. Потрібен, щоб відповісти на «чому мені не прийшло», і рівно на стільки, скільки це питання ще ставлять.",
		MinDays: 7,
	},
	{
		Kind: "command_runs", Group: "Прогони",
		Label:     "Прогони команд і їхні транскрипти",
		Note:      "Найтовщі рядки серед звичайних таблиць: вивід кожної команди на кожному хості плюс повна стенограма сесії. Один прогін по дільниці — це кілобайти тексту на кожен зі ста хостів.",
		MinDays:   7,
		MinReason: "звіт про прогін зазвичай читають у ті самі дні, коли його запустили",
		SizeAlso:  []string{"ncm.command_targets"},
	},
	{
		Kind: "ncm_jobs", Group: "Прогони",
		Label:   "Завдання збору конфігів",
		Note:    "Рядок на кожну спробу зняти конфіг разом із транскриптом сесії для діагностики. Добовий розклад на 500 хостів — 500 рядків із текстом щодня.",
		MinDays: 7,
	},
	{
		Kind: "discovery_runs", Group: "Прогони",
		Label:   "Прогони пошуку сусідів",
		Note:    "Найменша таблиця цієї групи: запускають її руками й нечасто.",
		MinDays: 7,
	},
	{
		Kind: "agent_health", Group: "Службове",
		Label:   "Самометрики зондів",
		Note:    "Пам'ять і завантаження самих зондів. Цінні кілька днів — рівно доти, доки з'ясовують, чому зонд не встигав.",
		MinDays: 3,
	},
	{
		Kind: "login_attempts", Group: "Безпека",
		Label:     "Спроби входу",
		Note:      "І вдалі, і невдалі. Це доказова база, а не телеметрія: строк тут визначають вимоги до організації, а не місце — важить вона копійки.",
		MinDays:   30,
		MinReason: "менший строк робить безглуздим саме питання «хто заходив минулого місяця»",
	},
	{
		Kind: "audit_log", Group: "Безпека",
		Label:     "Журнал аудиту",
		Note:      "Хто що зробив у системі. Стискається після року (0050) і саме тому важить мало навіть за роки. Видаляти його заради місця немає сенсу: у ньому не мегабайти, у ньому відповіді.",
		MinDays:   365,
		MinReason: "рік — межа, у яку вміщається практично будь-який розбір; коротший строк знищує докази раніше, ніж по них приходять",
	},
}

var retentionByKind = func() map[string]RetentionKindInfo {
	m := make(map[string]RetentionKindInfo, len(retentionCatalog))
	for _, k := range retentionCatalog {
		m[k.Kind] = k
	}
	return m
}()

func init() {
	seen := map[string]bool{}
	for _, k := range retentionCatalog {
		if seen[k.Kind] {
			panic("retentionCatalog: дубль виду даних " + k.Kind)
		}
		// Джерело має бути описане РАНІШЕ за свою згортку: інакше
		// перевірка «згортка не живе менше за джерело» мовчки
		// пропустила б найважливішу умову всієї форми.
		if k.SourceKind != "" && !seen[k.SourceKind] {
			panic("retentionCatalog: джерело " + k.SourceKind + " має стояти перед " + k.Kind)
		}
		seen[k.Kind] = true
	}
}

// RetentionCatalog — словник видів даних для форми.
func RetentionCatalog() []RetentionKindInfo { return retentionCatalog }

// RetentionRow — вид даних разом із тим, що з ним зараз відбувається.
type RetentionRow struct {
	RetentionKindInfo

	Relation  string `json:"relation"`
	Mechanism string `json:"mechanism"`

	// KeepDays — збережений намір, LiveDays — те, що НАСПРАВДІ стоїть у
	// TimescaleDB. Два поля, а не одне, бо вони можуть розійтись:
	// політику знімають руками під час обслуговування бази, і pg_dump
	// не везе фонових задач TimescaleDB взагалі. Мовчазна розбіжність
	// між екраном і базою — саме той стан, у якому людина впевнена, що
	// дані прибираються, а вони ростуть.
	KeepDays *int `json:"keep_days"`
	LiveDays *int `json:"live_days,omitempty"`

	UpdatedAt *time.Time `json:"updated_at,omitempty"`
	UpdatedBy string     `json:"updated_by,omitempty"`

	Bytes           int64 `json:"bytes"`
	CompressedBytes int64 `json:"compressed_bytes"`
	Chunks          int   `json:"chunks"`

	// Колонка часу. Не експортується: у JSON вона нічого не пояснює, а
	// потрібна лише запитам цього пакета.
	timeColumn string
}

// RetentionPreview — що зникне, якщо застосувати запропонований строк.
//
// Той самий підхід, що й у повного видалення хоста: спершу покажи, що
// зникне. Різниця лише в тому, що тут зникає не об'єкт, а хвіст
// історії, і побачити його інакше ніяк — у жодному переліку його немає.
type RetentionPreview struct {
	Kind  string `json:"kind"`
	Label string `json:"label"`

	FromDays *int `json:"from_days"`
	ToDays   *int `json:"to_days"`

	// Rows — точна кількість. -1 означає «не порахували»: запит
	// зупинили за часом. Нуль тут був би брехнею найгіршого ґатунку —
	// «нічого не зникне» перед видаленням мільйонів рядків.
	Rows   int64 `json:"rows"`
	Bytes  int64 `json:"bytes"`
	Chunks int   `json:"chunks"`

	// Exact — чи точна цифра місця. Для гіпертаблиць так: зникають цілі
	// чанки, і їхній розмір відомий до байта. Для звичайних таблиць ні:
	// рахується частка від розміру таблиці.
	Exact bool `json:"exact"`

	// Warning — те, що людина має прочитати ДО натискання.
	Warning string `json:"warning,omitempty"`
}

// relationRe — межа, за яку не пускаємо ім'я таблиці в текст запиту.
//
// Імена приходять із core.retention_settings, тобто з нашої ж бази, і
// це майже гарантія. «Майже» тут замало: наступного разу рядок туди
// покладе міграція, написана поспіхом, а вставка імені в текст запиту —
// рівно те місце, де недбалість перетворюється на довільний SQL.
// Параметром ім'я таблиці передати не можна, тож межа стоїть тут.
var relationRe = regexp.MustCompile(`^[a-z_]+\.[a-z0-9_]+$`)

// identRe — те саме для окремого ідентифікатора (колонки часу).
var identRe = regexp.MustCompile(`^[a-z][a-z0-9_]*$`)

// ErrRetentionInvalid — запропоновані строки суперечать одне одному.
//
// Окремо від ErrInvalid: це не «ви ввели букву замість числа», а
// «разом ці числа означають втрату даних». Обробник відповідає на них
// однаково, а от текст помилки має різну природу, і плутати їх у
// журналі не варто.
var ErrRetentionInvalid = errors.New("строки зберігання суперечать одне одному")

// RetentionSettings читає всі види даних разом із фактичним станом.
//
// Ходить через bg-пул: строки — рівня інсталяції, колонки tenant_id у
// них немає за побудовою (див. 0064).
func (s *Store) RetentionSettings(ctx context.Context) ([]RetentionRow, error) {
	rows, err := s.bg.Query(ctx, `
		SELECT r.kind, r.relation, r.mechanism, r.time_column, r.keep_days,
		       r.updated_at, COALESCE(u.email, ''),
		       -- Строк переводиться в доби прямо в базі: інтервал
		       -- Postgres не має однозначного представлення в Go, а
		       -- форма все одно оперує добами.
		       floor(EXTRACT(EPOCH FROM core.retention_current(r.relation)) / 86400)::int
		FROM core.retention_settings r
		LEFT JOIN core.users u ON u.id = r.updated_by
	`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	byKind := map[string]RetentionRow{}
	for rows.Next() {
		var r RetentionRow
		if err := rows.Scan(&r.Kind, &r.Relation, &r.Mechanism, &r.timeColumn,
			&r.KeepDays, &r.UpdatedAt, &r.UpdatedBy, &r.LiveDays); err != nil {
			return nil, err
		}
		// Політика на кілька годин після округлення вниз дала б нуль,
		// а нуль на екрані читається як «видаляти все». Таких політик
		// у продукті немає, але показати їх треба чесно найменшим
		// значенням, яке взагалі має сенс.
		if r.LiveDays != nil && *r.LiveDays < 1 {
			one := 1
			r.LiveDays = &one
		}
		byKind[r.Kind] = r
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}

	// Розміри окремим запитом і без фатальності: сторінка без цифр
	// розміру ще корисна («який строк стоїть»), а сторінка, яка не
	// відкрилась через збій у системному представленні, — ні.
	sizes, _ := s.relationSizes(ctx)

	out := make([]RetentionRow, 0, len(retentionCatalog))
	for _, info := range retentionCatalog {
		r, ok := byKind[info.Kind]
		if !ok {
			// Вид є в словнику, а рядка в базі немає: міграція, яка
			// його заводить, ще не накотилась. Показувати такий рядок
			// означало б обіцяти ручку, якої немає.
			continue
		}
		r.RetentionKindInfo = info
		if sz, ok := sizes[r.Relation]; ok {
			r.Bytes, r.CompressedBytes, r.Chunks = sz.total, sz.compressed, sz.chunks
		}
		for _, extra := range info.SizeAlso {
			if e, ok := sizes[extra]; ok {
				r.Bytes += e.total
			}
		}
		out = append(out, r)
	}
	return out, nil
}

// ValidateRetention перевіряє запропонований набір строків цілком.
//
// Цілком, а не по одному полю: половина умов тут — про стосунки між
// видами («згортка не живе менше за джерело»), і перевірити їх,
// дивлячись на одне число, неможливо. Саме тому форма шле весь набір, а
// не те, що змінилось.
func ValidateRetention(want map[string]*int) error {
	for kind, days := range want {
		info, ok := retentionByKind[kind]
		if !ok {
			return fmt.Errorf("%w: невідомий вид даних %q", ErrInvalid, kind)
		}
		if days == nil {
			continue
		}
		if *days < 1 || *days > 36500 {
			return fmt.Errorf("%w: строк для «%s» має бути від 1 до 36500 діб",
				ErrInvalid, info.Label)
		}
		if *days < info.MinDays {
			return fmt.Errorf("%w: «%s» не можна тримати менше за %d діб — %s",
				ErrRetentionInvalid, info.Label, info.MinDays, info.MinReason)
		}
	}

	// Найлегше місце для непомітної помилки в усій задачі: строки
	// виглядають розумно кожен окремо, а разом означають «сирі дані
	// видалено, згорток не порахували» — тобто діру в графіках за
	// минулий місяць, яку вже нічим не закрити.
	//
	// Ключове слово тут — «нічим»: сирих даних, з яких згортку рахують,
	// уже немає, і перерахувати її неможливо в принципі.
	for kind, days := range want {
		info := retentionByKind[kind]
		if info.SourceKind == "" {
			continue
		}
		src, ok := want[info.SourceKind]
		if !ok {
			continue
		}
		// nil — «не видаляти», тобто нескінченність. Згортка, яку не
		// видаляють, не може бути коротшою за що завгодно.
		if days == nil {
			continue
		}
		if src == nil || *src > *days {
			return fmt.Errorf(
				"%w: «%s» не може зникати раніше за «%s» — інакше в графіках за старі періоди лишиться діра, яку вже нічим не заповнити",
				ErrRetentionInvalid, info.Label, retentionByKind[info.SourceKind].Label)
		}
	}
	return nil
}

// PreviewRetention рахує, що зникне прямо зараз від запропонованих
// строків.
//
// Рахується ДО збереження й тими самими межами, що й саме видалення.
// Другого місця з умовами немає навмисно: розбіжність між обіцянкою у
// формі й тим, що станеться вночі, коштувала б довіри до обох.
func (s *Store) PreviewRetention(ctx context.Context, want map[string]*int) ([]RetentionPreview, error) {
	current, err := s.RetentionSettings(ctx)
	if err != nil {
		return nil, err
	}

	out := make([]RetentionPreview, 0, len(want))
	for _, row := range current {
		days, ok := want[row.Kind]
		if !ok || days == nil {
			// Строк знімають зовсім — зникати нічому.
			continue
		}
		// Строк подовжують або лишають: видалене не повернеться, а
		// нового під видалення не додається.
		if row.KeepDays != nil && *days >= *row.KeepDays {
			continue
		}

		p := RetentionPreview{
			Kind: row.Kind, Label: row.Label,
			FromDays: row.KeepDays, ToDays: days,
		}
		if err := s.previewOne(ctx, row, *days, &p); err != nil {
			return nil, err
		}
		out = append(out, p)
	}

	sort.Slice(out, func(i, j int) bool { return out[i].Bytes > out[j].Bytes })
	return out, nil
}

// previewTimeout — скільки чекаємо на точний підрахунок рядків.
//
// Не про продуктивність, а про поведінку форми: count(*) по хвосту
// гіпертаблиці на великій інсталяції може йти хвилини, і людина за цей
// час вирішить, що сторінка зависла. П'ятнадцять секунд — межа, після
// якої чесніше сказати «не порахували», ніж мовчати далі.
const previewTimeout = 15 * time.Second

func (s *Store) previewOne(ctx context.Context, row RetentionRow, days int, p *RetentionPreview) error {
	if !relationRe.MatchString(row.Relation) {
		return fmt.Errorf("неприйнятне ім'я таблиці %q", row.Relation)
	}
	if row.Mechanism == "timescale" {
		p.Exact = true
		p.Warning = "Чанки зносяться цілком, і місце повертається на диск одразу."
		return s.previewTimescale(ctx, row, days, p)
	}
	p.Exact = false
	p.Warning = "Місце звільниться всередині таблиці й буде перевикористане наступними рядками, але на диск не повернеться."
	return s.previewBatch(ctx, row, days, p)
}

// previewTimescale рахує чанки, які drop_chunks знесе прямо зараз.
//
// Саме чанки, а не «рядки, старші за строк»: це різні множини.
// drop_chunks зносить чанк лише тоді, коли ВЕСЬ його діапазон вийшов за
// строк, тож частина рядків, старших за строк, ще поживе в чанку, який
// закриється завтра. Рахувати за рядками означало б обіцяти більше, ніж
// станеться, — а людина, звіривши цифри після, вирішила б, що
// прибирання не працює.
func (s *Store) previewTimescale(ctx context.Context, row RetentionRow, days int, p *RetentionPreview) error {
	var boundary *time.Time
	// Розміри чанків беруться ОДНИМ викликом chunks_detailed_size, а не
	// викликом на кожен чанк: функція щоразу повертає весь перелік, і
	// виклик усередині з'єднання перетворив би сотню чанків на сотню
	// повних обходів.
	err := s.bg.QueryRow(ctx, `
		WITH ht AS (
			SELECT core.retention_hypertable($1) AS name
		),
		doomed AS (
			SELECT c.chunk_schema, c.chunk_name, c.range_end
			FROM ht
			JOIN timescaledb_information.chunks c
			  ON c.hypertable_schema = split_part(ht.name, '.', 1)
			 AND c.hypertable_name   = split_part(ht.name, '.', 2)
			WHERE c.range_end <= now() - make_interval(days => $2::int)
		),
		sizes AS (
			SELECT d.chunk_schema, d.chunk_name, d.total_bytes
			FROM ht, LATERAL chunks_detailed_size(ht.name::regclass) d
		)
		SELECT count(*)::int,
		       COALESCE(sum(s.total_bytes), 0)::bigint,
		       max(dm.range_end)
		FROM doomed dm
		LEFT JOIN sizes s
		  ON s.chunk_schema = dm.chunk_schema AND s.chunk_name = dm.chunk_name
	`, row.Relation, days).Scan(&p.Chunks, &p.Bytes, &boundary)
	if err != nil {
		return err
	}
	if p.Chunks == 0 || boundary == nil {
		return nil
	}

	// Точна кількість рядків — окремим запитом зі своєю стелею часу.
	// Окремим саме тому, що він може не встигнути, а вже пораховані
	// байти від цього втрачати не можна.
	p.Rows = s.countBefore(ctx, row.Relation, row.timeCol(), *boundary)
	return nil
}

func (s *Store) previewBatch(ctx context.Context, row RetentionRow, days int, p *RetentionPreview) error {
	col := row.timeCol()
	if !identRe.MatchString(col) {
		return fmt.Errorf("неприйнятне ім'я колонки %q", col)
	}

	rels := append([]string{row.Relation}, row.SizeAlso...)
	var (
		rows      int64
		reltuples float64
		total     int64
	)
	// to_regclass, а не ::regclass: другий кидає помилку на таблиці,
	// якої немає, і збій виглядав би як «попередній перегляд зламався»,
	// хоча означав би «супутньої таблиці вже не існує».
	q := fmt.Sprintf(`
		SELECT (SELECT count(*) FROM %s WHERE %s < now() - make_interval(days => $1::int)),
		       (SELECT COALESCE(sum(c.reltuples), 0)::float8
		          FROM unnest($2::text[]) t
		          JOIN pg_class c ON c.oid = to_regclass(t)),
		       (SELECT COALESCE(sum(pg_total_relation_size(to_regclass(t))), 0)::bigint
		          FROM unnest($2::text[]) t
		         WHERE to_regclass(t) IS NOT NULL)
	`, row.Relation, col)
	if err := s.bg.QueryRow(ctx, q, days, rels).Scan(&rows, &reltuples, &total); err != nil {
		return err
	}
	p.Rows = rows

	// Оцінка місця через середній рядок. Точної відповіді тут немає й
	// бути не може: рядок із транскриптом на 40 КБ і рядок з помилкою
	// на 30 байтів відрізняються на три порядки, а знати заздалегідь,
	// яких серед старих більше, нема звідки. Саме тому Exact = false.
	if reltuples > 0 && rows > 0 {
		share := float64(rows) / reltuples
		if share > 1 {
			share = 1
		}
		p.Bytes = int64(float64(total) * share)
	}
	return nil
}

// timeCol — колонка часу з бази, зі значенням за замовчуванням.
//
// Значення за замовчуванням лишається в коді: рядок, заведений
// майбутньою міграцією без цього поля, має поводитись як гіпертаблиця,
// а не падати посеред сторінки.
func (r RetentionRow) timeCol() string {
	if r.timeColumn != "" {
		return r.timeColumn
	}
	return "ts"
}

// countBefore — точна кількість рядків, старших за межу.
//
// Повертає -1, якщо не встигли за previewTimeout. Саме -1, а не 0:
// «не порахували» і «нічого немає» — різні відповіді, і плутати їх
// перед видаленням не можна.
func (s *Store) countBefore(ctx context.Context, relation, col string, before time.Time) int64 {
	if !relationRe.MatchString(relation) || !identRe.MatchString(col) {
		return -1
	}

	ctx, cancel := context.WithTimeout(ctx, previewTimeout+5*time.Second)
	defer cancel()

	var n int64 = -1
	_ = pgx.BeginFunc(ctx, s.bg, func(tx pgx.Tx) error {
		if _, err := tx.Exec(ctx, fmt.Sprintf(`SET LOCAL statement_timeout = %d`,
			previewTimeout.Milliseconds())); err != nil {
			return err
		}
		var got int64
		if err := tx.QueryRow(ctx, fmt.Sprintf(`SELECT count(*) FROM %s WHERE %s < $1`,
			relation, col), before).Scan(&got); err != nil {
			return err
		}
		n = got
		return nil
	})
	return n
}

// SaveRetention зберігає строки й одразу приводить політики у
// відповідність.
//
// Одним викликом, а не «зберегли, а вночі застосуємо»: людина, яка
// натиснула «зберегти» й побачила успіх, має підстави вважати, що строк
// діє. Розрив між наміром і станом бази — це саме та мовчазна помилка,
// від якої в RetentionRow заведено окреме поле LiveDays.
func (s *Store) SaveRetention(ctx context.Context, want map[string]*int, userID string) (int, error) {
	if err := ValidateRetention(want); err != nil {
		return 0, err
	}

	kinds := make([]string, 0, len(want))
	for k := range want {
		kinds = append(kinds, k)
	}
	sort.Strings(kinds)

	var changed int
	err := pgx.BeginFunc(ctx, s.bg, func(tx pgx.Tx) error {
		for _, kind := range kinds {
			if _, err := tx.Exec(ctx, `
				UPDATE core.retention_settings
				SET keep_days = $2, updated_at = now(), updated_by = $3
				WHERE kind = $1 AND keep_days IS DISTINCT FROM $2
			`, kind, want[kind], nullUUID(userID)); err != nil {
				return err
			}
		}
		// Політики накладає функція з правами власника:
		// add_retention_policy вимагає ownership гіпертаблиці, а
		// застосунок власником не є й не має ним бути (0063).
		return tx.QueryRow(ctx, `SELECT core.apply_retention_policies()`).Scan(&changed)
	})
	return changed, err
}

// SyncRetentionPolicies звіряє політики TimescaleDB зі збереженими
// строками й виправляє розбіжності.
//
// Потрібне не для збереження — там політика накладається одразу, — а
// для відновлення після сторонньої правки: політику знімають руками під
// час обслуговування бази, і pg_dump не везе фонових задач TimescaleDB
// взагалі. Такт, що звіряє це раз на годину, коштує один запит і рятує
// від тихого зростання аж до наступного перегляду сторінки.
func (s *Store) SyncRetentionPolicies(ctx context.Context) (int, error) {
	var n int
	err := s.bg.QueryRow(ctx, `SELECT core.apply_retention_policies()`).Scan(&n)
	return n, err
}

// ---------------------------------------------------------------------
// Пакетне прибирання звичайних таблиць
// ---------------------------------------------------------------------

// retentionDeleteBatch — скільки рядків прибирати за одну транзакцію.
//
// Не про швидкість. Один DELETE на річну історію прогонів тримав би
// рядки заблокованими стільки, скільки триває видалення, і рівно
// стільки ж ріс би в WAL. П'ять тисяч рядків — це частки секунди на
// транзакцію, тобто такт, якого ніхто не помічає. Те саме міркування,
// що й у telemetryBatch при повному видаленні хоста.
const retentionDeleteBatch = 5000

// retentionSweepCap — стеля на один вид даних за один прохід.
//
// Перший прохід на інсталяції, яка рік нічого не прибирала, інакше йшов
// би годинами й тримав би такт зайнятим. Залишок піде наступного разу —
// поспішати нікуди, а от лишити систему без прибиральника на пів дня
// не можна.
const retentionSweepCap = 200_000

// SweepStatKind — підсумок прибирання одного виду даних.
type SweepStatKind struct {
	Kind    string
	Deleted int64
	// More — стеля проходу вичерпана, лишилось іще. Окремим полем, бо
	// без нього «прибрано 200 000» невідрізнимо від «прибрано все».
	More bool
	// Due — скільки рядків підпадало під строк на початок проходу.
	//
	// Потрібне заради однієї-єдиної ситуації, і вона того варта: якщо
	// Due > 0, а Deleted = 0, то DELETE відпрацював і не зачепив
	// нічого. Це підпис RLS — з'єднання відкрито роллю без BYPASSRLS і
	// без заданого кабінету, тож політика ховає від видалення все
	// (див. коментар до Store.bg і 0063). Без цієї пари чисел збій
	// виглядав би як повна тиша: помилки немає, прибирання немає, диск
	// росте.
	Due int64
}

// SweepDataRetention прибирає звичайні таблиці за їхніми строками.
//
// Гіпертаблиць не чіпає взагалі: там працює політика TimescaleDB, і
// дублювати її своїм DELETE означало б робити ту саму роботу вдруге,
// повільніше й без звільнення місця на диску.
func (s *Store) SweepDataRetention(ctx context.Context) ([]SweepStatKind, error) {
	type target struct {
		kind, relation, col string
		days                int
	}

	rows, err := s.bg.Query(ctx, `
		SELECT kind, relation, time_column, keep_days
		FROM core.retention_settings
		WHERE mechanism = 'batch' AND keep_days IS NOT NULL
		ORDER BY kind
	`)
	if err != nil {
		return nil, err
	}
	var targets []target
	for rows.Next() {
		var t target
		if err := rows.Scan(&t.kind, &t.relation, &t.col, &t.days); err != nil {
			rows.Close()
			return nil, err
		}
		targets = append(targets, t)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return nil, err
	}

	var out []SweepStatKind
	for _, t := range targets {
		if ctx.Err() != nil {
			break
		}
		if !relationRe.MatchString(t.relation) || !identRe.MatchString(t.col) {
			return out, fmt.Errorf("неприйнятне ім'я %s.%s", t.relation, t.col)
		}

		// Партія за партією, кожна — своя транзакція. Перерваний
		// посеред роботи прохід лишає зрозумілий стан: частину старого
		// прибрано, решта піде наступного разу.
		q := fmt.Sprintf(`
			DELETE FROM %s
			WHERE id IN (
				SELECT id FROM %s
				WHERE %s < now() - make_interval(days => $1::int)
				ORDER BY %s
				LIMIT $2
			)
		`, t.relation, t.relation, t.col, t.col)

		stat := SweepStatKind{Kind: t.kind}
		if err := s.bg.QueryRow(ctx, fmt.Sprintf(
			`SELECT count(*) FROM %s WHERE %s < now() - make_interval(days => $1::int)`,
			t.relation, t.col), t.days).Scan(&stat.Due); err != nil {
			return out, fmt.Errorf("підрахунок застарілого в %s: %w", t.relation, err)
		}
		if stat.Due == 0 {
			continue
		}

		for stat.Deleted < retentionSweepCap {
			if ctx.Err() != nil {
				break
			}
			tag, err := s.bg.Exec(ctx, q, t.days, retentionDeleteBatch)
			if err != nil {
				return out, fmt.Errorf("прибирання %s: %w", t.relation, err)
			}
			n := tag.RowsAffected()
			stat.Deleted += n
			if n < retentionDeleteBatch {
				break
			}
		}
		stat.More = stat.Deleted >= retentionSweepCap
		// Рядок звіту заводиться й тоді, коли не прибрано нічого:
		// «підпадало 40 000, прибрано 0» — це не порожній результат, а
		// діагноз, і мовчати про нього не можна.
		out = append(out, stat)
	}
	return out, nil
}

// ---------------------------------------------------------------------
// Аудит
// ---------------------------------------------------------------------

// RetentionAuditMeta збирає зміни строків для журналу.
//
// У журнал має поїхати не «зберегли форму», а що саме змінилось: через
// рік питання буде не «хто відкривав сторінку», а «хто скоротив історію
// алертів до тижня». Мітки поруч із ключами з тієї ж причини, що й
// імена хостів у записі про масове видалення: через рік ключ
// alerts_history нічого не скаже.
func RetentionAuditMeta(before []RetentionRow, want map[string]*int) map[string]any {
	prev := map[string]*int{}
	for _, r := range before {
		prev[r.Kind] = r.KeepDays
	}

	kinds := make([]string, 0, len(want))
	for k := range want {
		kinds = append(kinds, k)
	}
	sort.Strings(kinds)

	changes := []map[string]any{}
	for _, k := range kinds {
		old, had := prev[k]
		if !had || samePtr(old, want[k]) {
			continue
		}
		changes = append(changes, map[string]any{
			"kind":  k,
			"label": retentionByKind[k].Label,
			"from":  daysLabel(old),
			"to":    daysLabel(want[k]),
			// Скорочення строку — саме те, після чого дані зникають.
			// Окремою позначкою, щоб у журналі це було видно без
			// порівняння двох чисел очима.
			"shortened": shortened(old, want[k]),
		})
	}
	return map[string]any{"changes": changes}
}

func samePtr(a, b *int) bool {
	if a == nil || b == nil {
		return a == nil && b == nil
	}
	return *a == *b
}

func shortened(old, next *int) bool {
	if next == nil {
		return false
	}
	return old == nil || *next < *old
}

func daysLabel(d *int) string {
	if d == nil {
		return "не видаляти"
	}
	return fmt.Sprintf("%d діб", *d)
}