package ingest_test import ( "context" "crypto/sha256" "encoding/hex" "errors" "io" "log/slog" "os" "path/filepath" "testing" "git.vakhrushev.me/av/healthlog/internal/archive" "git.vakhrushev.me/av/healthlog/internal/ingest" "git.vakhrushev.me/av/healthlog/internal/store" ) func TestAcceptStoresBodyVerbatim(t *testing.T) { svc, arch, st := newService(t) // Незнакомая метрика и неизвестная форма точки — не повод отвергать пакет. body := []byte(`{"data":{"metrics":[{"name":"новая_метрика","units":"?","data":[{"чего_то":1,"date":"2026-07-31 12:00:00 +0300"}]}]}}`) res, err := svc.Accept(context.Background(), body, ingest.Meta{ AutomationName: "всё подряд", Aggregation: "None", Period: "Since Last Sync", }) if err != nil { t.Fatalf("Accept: %v", err) } if res.Bytes != int64(len(body)) { t.Errorf("bytes = %d, ожидалось %d", res.Bytes, len(body)) } sum := sha256.Sum256(body) if res.SHA256 != hex.EncodeToString(sum[:]) { t.Errorf("sha256 = %q", res.SHA256) } // Тело в архиве — байт в байт то, что прислали. r, err := arch.Open(res.RawPath) if err != nil { t.Fatalf("Open: %v", err) } defer func() { _ = r.Close() }() got, err := io.ReadAll(r) if err != nil { t.Fatalf("чтение архива: %v", err) } if string(got) != string(body) { t.Errorf("в архиве %q, прислано %q", got, body) } n, err := st.CountDeliveries(context.Background()) if err != nil { t.Fatalf("CountDeliveries: %v", err) } if n != 1 { t.Errorf("доставок в БД: %d, ожидалась 1", n) } } // Повтор того же тела пока принимается — отсев идентичных доставок отложен // (docs/plan.md). Проверяем, что повтор не ломается и не затирает первую. func TestAcceptAllowsRepeatedBody(t *testing.T) { svc, _, st := newService(t) body := []byte(`{"data":{"metrics":[]}}`) first, err := svc.Accept(context.Background(), body, ingest.Meta{}) if err != nil { t.Fatalf("первый Accept: %v", err) } second, err := svc.Accept(context.Background(), body, ingest.Meta{}) if err != nil { t.Fatalf("второй Accept: %v", err) } if first.DeliveryID == second.DeliveryID { t.Error("у двух доставок совпали идентификаторы") } if first.SHA256 != second.SHA256 { t.Error("у одинаковых тел разошёлся sha256") } n, err := st.CountDeliveries(context.Background()) if err != nil { t.Fatalf("CountDeliveries: %v", err) } if n != 2 { t.Errorf("доставок в БД: %d, ожидалось 2", n) } } func TestAcceptRejectsMalformed(t *testing.T) { cases := map[string]string{ "пустое тело": ``, "не JSON": `не json`, "обрезанный JSON": `{"data":{"metrics":[`, "нет data": `{"metrics":[]}`, "data не объект": `{"data":[]}`, } for name, body := range cases { t.Run(name, func(t *testing.T) { svc, _, st := newService(t) _, err := svc.Accept(context.Background(), []byte(body), ingest.Meta{}) if !errors.Is(err, ingest.ErrMalformed) { t.Fatalf("ошибка = %v, ожидалась ErrMalformed", err) } // Отвергнутая доставка не оставляет следа в учёте. n, err := st.CountDeliveries(context.Background()) if err != nil { t.Fatalf("CountDeliveries: %v", err) } if n != 0 { t.Errorf("доставок в БД: %d, ожидалось 0", n) } }) } } // Ответ отдаётся ДО свёртки: принятая доставка ждёт разбора в очереди, а не // приезжает разобранной. Это смена контракта, и она проверяется явно. func TestAcceptОставляетДоставкуВОчереди(t *testing.T) { svc, _, st := newService(t) ctx := context.Background() body, err := os.ReadFile(filepath.Join("..", "hae", "testdata", "minute.json")) if err != nil { t.Fatalf("фикстура: %v", err) } if _, err := svc.Accept(ctx, body, ingest.Meta{Aggregation: "Minutes"}); err != nil { t.Fatalf("Accept: %v", err) } d, err := st.LastDelivery(ctx) if err != nil { t.Fatalf("LastDelivery: %v", err) } if d.ParseStatus != store.ParsePending { t.Errorf("parse_status = %q, ожидался %q", d.ParseStatus, store.ParsePending) } n, err := st.CountBuckets(ctx) if err != nil { t.Fatalf("CountBuckets: %v", err) } if n != 0 { t.Errorf("объектов %d: свёртка произошла внутри приёма", n) } } // Сигнал уходит после того, как доставка учтена: воркер, разбуженный раньше, // потратил бы проход впустую. func TestAcceptБудитСвёрткуПослеУчёта(t *testing.T) { dir := t.TempDir() st, arch := newDeps(t, dir) var seen int64 notify := func() { n, err := st.CountDeliveries(context.Background()) if err != nil { t.Errorf("CountDeliveries: %v", err) } seen = n } svc := ingest.New(arch, st, notify, slog.New(slog.DiscardHandler)) if _, err := svc.Accept(context.Background(), []byte(`{"data":{"metrics":[]}}`), ingest.Meta{}); err != nil { t.Fatalf("Accept: %v", err) } if seen != 1 { t.Errorf("на момент сигнала доставок в учёте %d, ожидалась 1", seen) } } // Нулевой сигнал — законный вход (свёрткой заведует вызывающий), и приём от // него не падает. Паника здесь наступила бы ПОСЛЕ записи тела и учёта, то есть // отправитель получил бы отказ по сохранённой доставке. func TestAcceptБезСигналаНеПадает(t *testing.T) { dir := t.TempDir() st, arch := newDeps(t, dir) svc := ingest.New(arch, st, nil, slog.New(slog.DiscardHandler)) if _, err := svc.Accept(context.Background(), []byte(`{"data":{"metrics":[]}}`), ingest.Meta{}); err != nil { t.Fatalf("Accept: %v", err) } } // Обрыв соединения после записи тела не должен оставлять тело без учёта: // доставка, не попавшая в журнал, восстанавливается только пересборкой с // ручной подменой базы. func TestAcceptУчитываетДоставкуПослеОбрываСоединения(t *testing.T) { svc, _, st := newService(t) ctx, cancel := context.WithCancel(context.Background()) cancel() res, err := svc.Accept(ctx, []byte(`{"data":{"metrics":[]}}`), ingest.Meta{}) if err != nil { t.Fatalf("Accept на отменённом контексте: %v", err) } status, err := st.DeliveryStatus(context.Background(), res.DeliveryID) if err != nil { t.Fatalf("DeliveryStatus: %v", err) } if status != store.ParsePending { t.Errorf("parse_status = %q, ожидался %q", status, store.ParsePending) } } // Учёта нет, а тело есть: приём кладёт тело на диск раньше строки в базе, и // отказ на вставке оставляет тело в архиве. Такое тело подберёт пересборка. func TestAcceptПриОтказеУчётаОставляетТелоВАрхиве(t *testing.T) { dir := t.TempDir() st, arch := newDeps(t, dir) svc := ingest.New(arch, st, nil, slog.New(slog.DiscardHandler)) // База закрыта — учесть доставку нечем. if err := st.Close(); err != nil { t.Fatalf("закрытие базы: %v", err) } if _, err := svc.Accept(context.Background(), []byte(`{"data":{"metrics":[]}}`), ingest.Meta{}); err == nil { t.Fatal("приём не заметил, что доставка не учтена") } entries, err := filepath.Glob(filepath.Join(arch.Root(), "*", "*", "*", "*.json.gz")) if err != nil { t.Fatalf("обход архива: %v", err) } if len(entries) != 1 { t.Errorf("тел в архиве %d, ожидалось 1: тело потеряно вместе с учётом", len(entries)) } } func newDeps(t *testing.T, dir string) (*store.Store, *archive.Archive) { t.Helper() st, err := store.Open(filepath.Join(dir, "healthlog.db")) if err != nil { t.Fatalf("store.Open: %v", err) } t.Cleanup(func() { _ = st.Close() }) arch, err := archive.New(filepath.Join(dir, "raw")) if err != nil { t.Fatalf("archive.New: %v", err) } return st, arch } func newService(t *testing.T) (*ingest.Service, *archive.Archive, *store.Store) { t.Helper() st, arch := newDeps(t, t.TempDir()) return ingest.New(arch, st, nil, slog.New(slog.DiscardHandler)), arch, st }