reindex: пересборка витрины проигрыванием журнала

- `healthlog reindex` собирает витрину из журнала (тела архива + учёт доставок) в ОТДЕЛЬНЫЙ файл базы, строго по `(received_at, id)`; рабочую базу читает без наката миграций и не трогает вовсе. Подмену делает человек при остановленном сервисе: переименование поверх открытого дескриптора портит базу молча. - Журналом считается архив, а не таблица доставок: тело без учётной записи заводится заново (метка из ULID, размер и хеш по распакованному телу), запись без тела переносится, но не сворачивается. Оракул сходимости встроен — два отпечатка и «объектов было/стало»; пустой журнал успехом не считается. - Прогон живого архива переехал на новый пакет: второго проигрывателя журнала в проекте не осталось, а его утверждение о ключе сна перестало быть константой, протухающей с каждой доставкой.
2026-08-02 09:07:46 +03:00
parent 84bcbbea5c
commit 5ae0c5ff81
36 changed files with 4452 additions and 258 deletions
@@ -105,6 +105,62 @@ func (s *Store) LastDelivery(ctx context.Context) (Delivery, error) {
 	return d, nil
 }

+// ListDeliveries возвращает учёт доставок в порядке журнала — `(received_at,
+// id)`, тем же, в котором их проигрывает пересборка.
+//
+// Отдаются только **факты журнала**: то, что пришло вместе с доставкой.
+// Производные от разбора поля (`parse_status`, `points`, `derived_layer`,
+// `uncovered_sections`) сюда не попадают намеренно — перенос их в пересобранную
+// базу сделал бы витрину функцией предыдущего прогона. Особенно `derived_layer`:
+// доставка, чей повторный разбор отказал, отдала бы в наследование слой
+// прежнего разбора, и следующая доставка той же автоматизации унаследовала бы
+// его молча.
+func (s *Store) ListDeliveries(ctx context.Context) ([]Delivery, error) {
+	const q = `
+		SELECT id, received_at, automation_name, automation_id, aggregation,
+		       period, session_id, bytes, sha256, raw_path, headers
+		FROM delivery ORDER BY received_at, id`
+
+	rows, err := s.db.QueryContext(ctx, q)
+	if err != nil {
+		return nil, fmt.Errorf("select deliveries: %w", err)
+	}
+	defer func() { _ = rows.Close() }()
+
+	var out []Delivery
+	for rows.Next() {
+		var d Delivery
+		var receivedAt string
+		if err := rows.Scan(&d.ID, &receivedAt, &d.AutomationName, &d.AutomationID,
+			&d.Aggregation, &d.Period, &d.SessionID, &d.Bytes, &d.SHA256,
+			&d.RawPath, &d.Headers); err != nil {
+			return nil, fmt.Errorf("scan delivery: %w", err)
+		}
+		d.ReceivedAt, err = ParseTime(receivedAt)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, d)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("select deliveries: %w", err)
+	}
+	return out, nil
+}
+
+// DeliveryStatus возвращает статус разбора доставки.
+func (s *Store) DeliveryStatus(ctx context.Context, id string) (string, error) {
+	var status string
+	err := s.db.GetContext(ctx, &status, `SELECT parse_status FROM delivery WHERE id = ?`, id)
+	if errors.Is(err, sql.ErrNoRows) {
+		return "", ErrNotFound
+	}
+	if err != nil {
+		return "", fmt.Errorf("select parse status: %w", err)
+	}
+	return status, nil
+}
+
 // FinishParse записывает исход разбора доставки.
 //
 // Слой сохраняется здесь же, потому что он нужен следующей доставке той же
@@ -0,0 +1,110 @@
+package store_test
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"path/filepath"
+	"testing"
+	"time"
+
+	_ "modernc.org/sqlite" // чистый Go-драйвер SQLite, без cgo
+
+	"git.vakhrushev.me/av/healthlog/internal/store"
+)
+
+// Пересборка читает рабочую базу, пока в неё может писать сервис. Обычное
+// открытие накатывает миграции безусловно, а миграции меняют и данные — та, что
+// ввела частичный разбор, переписала parse_status у всех строк. Значит утилите
+// нужен путь чтения, который базу не трогает.
+func TestOpenForReadНеПишетВБазу(t *testing.T) {
+	t.Parallel()
+
+	path := filepath.Join(t.TempDir(), "healthlog.db")
+	seed(t, path)
+
+	ro, err := store.OpenForRead(path)
+	if err != nil {
+		t.Fatalf("OpenForRead: %v", err)
+	}
+	defer func() { _ = ro.Close() }()
+
+	ctx := context.Background()
+	got, err := ro.ListDeliveries(ctx)
+	if err != nil {
+		t.Fatalf("ListDeliveries: %v", err)
+	}
+	if len(got) != 2 {
+		t.Fatalf("доставок %d, ожидалось 2", len(got))
+	}
+	// Порядок журнала — (received_at, id), тот же, в котором проигрывает
+	// пересборка.
+	if got[0].ID != "01hzzzzzzzzzzzzzzzzzzzzzz1" || got[1].ID != "01hzzzzzzzzzzzzzzzzzzzzzz0" {
+		t.Errorf("порядок %q, %q — не по времени приёма", got[0].ID, got[1].ID)
+	}
+	// Заголовки переносятся дословно: восстановить их неоткуда, в архиве их нет.
+	if got[0].Headers != `{"x-test":["1"]}` {
+		t.Errorf("заголовки %q не дошли дословно", got[0].Headers)
+	}
+
+	if err := ro.CreateDelivery(ctx, store.Delivery{
+		ID: "01hzzzzzzzzzzzzzzzzzzzzzz2", ReceivedAt: store.Now(),
+		Bytes: 1, SHA256: "-", RawPath: "x", ParseStatus: store.ParsePending,
+	}); err == nil {
+		t.Error("запись в базу, открытую на чтение, удалась")
+	}
+}
+
+// Расхождение версии схемы — отказ, а не повод мигрировать: иначе свежий бинарь
+// молча меняет схему под работающим старым сервисом.
+func TestOpenForReadОтвергаетЧужуюВерсиюСхемы(t *testing.T) {
+	t.Parallel()
+
+	path := filepath.Join(t.TempDir(), "healthlog.db")
+	seed(t, path)
+
+	// Откатываем учёт миграций мимо store: имитируем базу, к которой бинарь
+	// новее.
+	db, err := sql.Open("sqlite", "file:"+path)
+	if err != nil {
+		t.Fatalf("sql.Open: %v", err)
+	}
+	_, err = db.Exec(`DELETE FROM goose_db_version
+		WHERE version_id = (SELECT max(version_id) FROM goose_db_version)`)
+	if err != nil {
+		t.Fatalf("откат версии: %v", err)
+	}
+	_ = db.Close()
+
+	if _, err := store.OpenForRead(path); !errors.Is(err, store.ErrSchemaMismatch) {
+		t.Errorf("OpenForRead дал %v, ожидался ErrSchemaMismatch", err)
+	}
+}
+
+func seed(t *testing.T, path string) {
+	t.Helper()
+
+	st, err := store.Open(path)
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	defer func() { _ = st.Close() }()
+
+	base := time.Date(2026, 8, 1, 12, 0, 0, 0, time.UTC)
+	// Второй идентификатор меньше первого, а приехал он раньше: так проверяется,
+	// что порядок берётся из времени приёма, а не из имени.
+	rows := []store.Delivery{
+		{ID: "01hzzzzzzzzzzzzzzzzzzzzzz1", ReceivedAt: base},
+		{ID: "01hzzzzzzzzzzzzzzzzzzzzzz0", ReceivedAt: base.Add(time.Second)},
+	}
+	for _, d := range rows {
+		d.Bytes = 1
+		d.SHA256 = "-"
+		d.RawPath = d.ID + ".json.gz"
+		d.ParseStatus = store.ParsePending
+		d.Headers = `{"x-test":["1"]}`
+		if err := st.CreateDelivery(context.Background(), d); err != nil {
+			t.Fatalf("CreateDelivery: %v", err)
+		}
+	}
+}
@@ -5,9 +5,12 @@ package store
 import (
 	"context"
 	"embed"
+	"errors"
 	"fmt"
 	"io/fs"
 	"net/url"
+	"strconv"
+	"strings"
 	"time"

 	"github.com/jmoiron/sqlx"
@@ -38,6 +41,70 @@ func Open(dbPath string) (*Store, error) {
 	return &Store{db: db}, nil
 }

+// ErrSchemaMismatch — версия схемы базы не та, которую знает бинарь.
+var ErrSchemaMismatch = errors.New("версия схемы базы не совпадает с версией бинаря")
+
+// OpenForRead открывает базу только для чтения и **без наката миграций**.
+//
+// Обычный Open мигрирует безусловно, а миграции здесь меняют не только схему, но
+// и данные: та, что ввела частичный разбор, переписала parse_status у всех строк.
+// Значит утилита, которой достаточно прочитать учёт, обычным открытием нарушала
+// бы обещание «рабочую базу не трогаем», — и хуже: свежий бинарь мигрировал бы
+// схему под работающим старым сервисом, который держит запросы к прежней.
+//
+// Расхождение версий — отказ с указанием обеих, а не повод мигрировать.
+func OpenForRead(dbPath string) (*Store, error) {
+	db, err := sqlx.Connect("sqlite", readOnlyDSN(dbPath))
+	if err != nil {
+		return nil, fmt.Errorf("open sqlite %q read-only: %w", dbPath, err)
+	}
+
+	want, err := latestMigration()
+	if err != nil {
+		_ = db.Close()
+		return nil, err
+	}
+	var got int64
+	if err := db.Get(&got, `SELECT max(version_id) FROM goose_db_version`); err != nil {
+		_ = db.Close()
+		return nil, fmt.Errorf("read schema version: %w", err)
+	}
+	if got != want {
+		_ = db.Close()
+		return nil, fmt.Errorf("%w: база %d, бинарь %d", ErrSchemaMismatch, got, want)
+	}
+	return &Store{db: db}, nil
+}
+
+// latestMigration — номер последней миграции, вшитой в бинарь.
+func latestMigration() (int64, error) {
+	entries, err := fs.ReadDir(migrationsFS, "migrations")
+	if err != nil {
+		return 0, fmt.Errorf("read migrations dir: %w", err)
+	}
+	var top int64
+	for _, e := range entries {
+		name := e.Name()
+		// Неразобранное имя — отказ, а не пропуск: страж «версия схемы не та»,
+		// молча не заметивший миграцию, перестаёт страховать, не сказав об этом.
+		idx := strings.IndexByte(name, '_')
+		if idx <= 0 {
+			return 0, fmt.Errorf("имя миграции %q не вида NNNNN_*.sql", name)
+		}
+		v, err := strconv.ParseInt(name[:idx], 10, 64)
+		if err != nil {
+			return 0, fmt.Errorf("имя миграции %q не вида NNNNN_*.sql", name)
+		}
+		if v > top {
+			top = v
+		}
+	}
+	if top == 0 {
+		return 0, errors.New("миграций не найдено")
+	}
+	return top, nil
+}
+
 // Close закрывает соединение с БД.
 func (s *Store) Close() error {
 	if err := s.db.Close(); err != nil {
@@ -63,6 +130,18 @@ func dsn(path string) string {
 	return "file:" + path + "?" + q.Encode()
 }

+// readOnlyDSN — подключение только для чтения.
+//
+// `journal_mode` здесь не задаётся: сменить его на read-only соединении нельзя,
+// а читать базу в режиме WAL это не мешает. `_txlock=immediate` тоже не нужен —
+// он лечит повышение блокировки с чтения на запись, которого здесь не бывает.
+func readOnlyDSN(path string) string {
+	q := url.Values{}
+	q.Add("mode", "ro")
+	q.Add("_pragma", "busy_timeout(5000)")
+	return "file:" + path + "?" + q.Encode()
+}
+
 // migrate накатывает миграции.
 //
 // Через Provider, а не через пакетные функции: goose.SetBaseFS и