добавлен приём пакетов Health Auto Export
- POST /api/v1/ingest: токен, лимит тела, gzip, durable-запись тела в сырой архив, учёт доставки в SQLite - код ответа отражает доставку, а не разбор: битый JSON — 400, непонятое содержимое — 200, данные уже сохранены и доразберутся позже - сохраняется полный набор заголовков запроса с вычисткой секретов по имени и по совпадению значения с токеном
This commit is contained in:
@@ -0,0 +1,162 @@
|
||||
// Package ingest — use-case приёма пакета от Health Auto Export, общий для
|
||||
// HTTP и (позже) CLI-импорта.
|
||||
package ingest
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
|
||||
"git.vakhrushev.me/av/healthlog/internal/archive"
|
||||
"git.vakhrushev.me/av/healthlog/internal/ident"
|
||||
"git.vakhrushev.me/av/healthlog/internal/store"
|
||||
)
|
||||
|
||||
// ErrMalformed — тело не разбирается как JSON ожидаемой формы. Это отказ
|
||||
// доставки (обрыв, обрезанное тело), о нём отправителю сообщаем ошибкой.
|
||||
// Незнакомое СОДЕРЖИМОЕ (новая метрика, новая форма точки) сюда не относится:
|
||||
// такой пакет принимается, сохраняется и доразбирается позже.
|
||||
var ErrMalformed = errors.New("некорректный формат пакета")
|
||||
|
||||
// Meta — что рассказала о себе автоматизация Health Auto Export (свои
|
||||
// заголовки запроса).
|
||||
type Meta struct {
|
||||
AutomationName string
|
||||
AutomationID string
|
||||
Aggregation string
|
||||
Period string
|
||||
SessionID string
|
||||
// Headers — весь набор заголовков запроса, уже без секретов (вырезает
|
||||
// транспорт: только он знает список токенов). Документация HAE называет
|
||||
// пять заголовков, но что приложение шлёт на самом деле — неизвестно, и
|
||||
// именно из незадокументированного вышли самые полезные находки.
|
||||
Headers map[string][]string
|
||||
}
|
||||
|
||||
// Result — итог приёма.
|
||||
type Result struct {
|
||||
DeliveryID string
|
||||
Bytes int64
|
||||
SHA256 string
|
||||
RawPath string
|
||||
}
|
||||
|
||||
// Service принимает пакеты: сохраняет тело в архив и учитывает доставку.
|
||||
type Service struct {
|
||||
arch *archive.Archive
|
||||
store *store.Store
|
||||
log *slog.Logger
|
||||
}
|
||||
|
||||
// New собирает use-case приёма.
|
||||
func New(arch *archive.Archive, st *store.Store, log *slog.Logger) *Service {
|
||||
return &Service{arch: arch, store: st, log: log.With("capability", "ingest")}
|
||||
}
|
||||
|
||||
// Accept принимает тело пакета: проверяет форму, кладёт в сырой архив и
|
||||
// заводит запись о доставке.
|
||||
//
|
||||
// Порядок важен: сначала тело оказывается на диске, и только потом появляется
|
||||
// учётная запись. Обратный порядок дал бы учтённую доставку без данных.
|
||||
//
|
||||
// Это единственный логирующий чекпоинт приёма — транспорт исход не логирует.
|
||||
func (s *Service) Accept(ctx context.Context, body []byte, meta Meta) (Result, error) {
|
||||
if err := checkEnvelope(body); err != nil {
|
||||
// Некорректный ввод от отправителя — норма жизни, команде разбирать
|
||||
// нечего: DEBUG, а не ERROR.
|
||||
s.log.DebugContext(ctx, "delivery rejected", "error", err, "bytes", len(body))
|
||||
return Result{}, err
|
||||
}
|
||||
|
||||
sum := sha256.Sum256(body)
|
||||
res := Result{
|
||||
DeliveryID: ident.NewID(),
|
||||
Bytes: int64(len(body)),
|
||||
SHA256: hex.EncodeToString(sum[:]),
|
||||
}
|
||||
receivedAt := store.Now()
|
||||
|
||||
rawPath, err := s.arch.Write(res.DeliveryID, receivedAt, body)
|
||||
if err != nil {
|
||||
s.log.ErrorContext(ctx, "delivery failed", "error", err, "delivery_id", res.DeliveryID)
|
||||
return Result{}, fmt.Errorf("archive body: %w", err)
|
||||
}
|
||||
res.RawPath = rawPath
|
||||
|
||||
err = s.store.CreateDelivery(ctx, store.Delivery{
|
||||
ID: res.DeliveryID,
|
||||
ReceivedAt: receivedAt,
|
||||
Headers: encodeHeaders(meta.Headers),
|
||||
AutomationName: meta.AutomationName,
|
||||
AutomationID: meta.AutomationID,
|
||||
Aggregation: meta.Aggregation,
|
||||
Period: meta.Period,
|
||||
SessionID: meta.SessionID,
|
||||
Bytes: res.Bytes,
|
||||
SHA256: res.SHA256,
|
||||
RawPath: rawPath,
|
||||
ParseStatus: store.ParsePending,
|
||||
})
|
||||
if err != nil {
|
||||
// Тело уже на диске — данные не потеряны, но учёта нет. Разбор архива
|
||||
// на следующем шаге проекта такую доставку подберёт.
|
||||
s.log.ErrorContext(ctx, "delivery failed", "error", err, "delivery_id", res.DeliveryID, "raw_path", rawPath)
|
||||
return Result{}, fmt.Errorf("record delivery: %w", err)
|
||||
}
|
||||
|
||||
s.log.InfoContext(ctx, "delivery accepted",
|
||||
"delivery_id", res.DeliveryID,
|
||||
"bytes", res.Bytes,
|
||||
"raw_path", rawPath,
|
||||
"automation_name", meta.AutomationName,
|
||||
"aggregation", meta.Aggregation,
|
||||
"period", meta.Period)
|
||||
|
||||
return res, nil
|
||||
}
|
||||
|
||||
// encodeHeaders сериализует заголовки для хранения. Ключи json.Marshal
|
||||
// сортирует сам, поэтому запись стабильна и её удобно сравнивать между
|
||||
// доставками. Сбой сериализации не должен ронять приём: заголовки —
|
||||
// вспомогательные сведения, а не данные, ради которых всё затевалось.
|
||||
func encodeHeaders(h map[string][]string) string {
|
||||
if len(h) == 0 {
|
||||
return "{}"
|
||||
}
|
||||
b, err := json.Marshal(h)
|
||||
if err != nil {
|
||||
return "{}"
|
||||
}
|
||||
return string(b)
|
||||
}
|
||||
|
||||
// envelope — минимальная форма пакета Health Auto Export.
|
||||
type envelope struct {
|
||||
Data json.RawMessage `json:"data"`
|
||||
}
|
||||
|
||||
// checkEnvelope проверяет ровно то, что делает пакет доставленным: это JSON,
|
||||
// в нём есть объект data. Содержимое data не трогаем — его форму определяет
|
||||
// Apple, и незнакомая форма не повод отвергать доставку.
|
||||
func checkEnvelope(body []byte) error {
|
||||
if len(bytes.TrimSpace(body)) == 0 {
|
||||
return fmt.Errorf("%w: пустое тело", ErrMalformed)
|
||||
}
|
||||
|
||||
var env envelope
|
||||
if err := json.Unmarshal(body, &env); err != nil {
|
||||
return fmt.Errorf("%w: %v", ErrMalformed, err) //nolint:errorlint // причину наружу не раскрываем, она уходит в лог
|
||||
}
|
||||
if len(env.Data) == 0 {
|
||||
return fmt.Errorf("%w: нет объекта data", ErrMalformed)
|
||||
}
|
||||
if trimmed := bytes.TrimSpace(env.Data); len(trimmed) == 0 || trimmed[0] != '{' {
|
||||
return fmt.Errorf("%w: data не объект", ErrMalformed)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,141 @@
|
||||
package ingest_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"io"
|
||||
"log/slog"
|
||||
"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 newService(t *testing.T) (*ingest.Service, *archive.Archive, *store.Store) {
|
||||
t.Helper()
|
||||
dir := t.TempDir()
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
log := slog.New(slog.DiscardHandler)
|
||||
return ingest.New(arch, st, log), arch, st
|
||||
}
|
||||
Reference in New Issue
Block a user