healthlog/internal/ingest/ingest.go

// Package ingest — use-case приёма пакета от Health Auto Export, общий для
// HTTP и (позже) CLI-импорта.
package ingest

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"time"

	"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
}

// recordTimeout — сколько отводится записи учёта доставки.
//
// Учёт ведётся на контексте, переживающем обрыв соединения (см. Accept),
// поэтому собственный дедлайн обязателен: без него отказ базы держал бы
// обработчик неограниченно.
const recordTimeout = 10 * time.Second

// Notify — «есть работа»: сигнал тому, кто сворачивает принятое.
//
// Функцией, а не интерфейсом: сигнал ничего не несёт и ничего не возвращает,
// а приёму незачем знать, кто именно свернёт доставку.
type Notify func()

// Service принимает пакеты: сохраняет тело в архив, учитывает доставку и будит
// свёртку.
//
// Сворачивать сам он не умеет намеренно. Свёртка широкой доставки идёт
// секундами, а `WriteTimeout` в Go ставится до вызова обработчика — то есть
// синхронная свёртка тратила бы бюджет ответа и обрывала бы соединение молча,
// с записью `status_code=200` в журнале доступа.
type Service struct {
	arch   *archive.Archive
	store  *store.Store
	notify Notify
	log    *slog.Logger
}

// New собирает use-case приёма.
//
// Нулевой notify означает «о свёртке заботится вызывающий» и приводится к
// пустой функции здесь же, один раз: проверка на nil в месте вызова рано или
// поздно окажется забытой, а паника там наступила бы ПОСЛЕ того, как тело уже
// записано и доставка учтена, — то есть отправитель получил бы отказ по
// сохранённой доставке.
func New(arch *archive.Archive, st *store.Store, notify Notify, log *slog.Logger) *Service {
	if notify == nil {
		notify = func() {}
	}
	return &Service{arch: arch, store: st, notify: notify, log: log.With("capability", "ingest")}
}

// Accept принимает тело пакета: проверяет форму, кладёт в сырой архив, заводит
// запись о доставке и будит свёртку.
//
// Порядок важен: сначала тело оказывается на диске, и только потом появляется
// учётная запись. Обратный порядок дал бы учтённую доставку без данных.
//
// Возврат означает «сохранено и учтено», а не «разобрано»: доставка уезжает в
// очередь свёртки статусом `pending`, и её исход появится позже.
//
// Это единственный логирующий чекпоинт приёма — транспорт исход не логирует.
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[:]),
	}
	// Метка приёма выводится ИЗ идентификатора, а не берётся вторым обращением
	// к часам. Источник обязан быть один: у тела, лежащего в архиве без учётной
	// записи, метку восстанавливают из ULID, и два разных источника разошлись бы
	// на границе секунды — а от порядка журнала зависит наследование слоя.
	// Ошибка здесь означает, что наш же генератор выдал неразбираемый
	// идентификатор. Второго источника времени тут быть не может — он разошёлся
	// бы с меткой, которую пересборка восстанавливает из ULID; поэтому отказ, а
	// не подмена. Тело на диск ещё не легло, так что доставка не теряется.
	receivedAt, err := ident.TimeOf(res.DeliveryID)
	if err != nil {
		s.log.ErrorContext(ctx, "delivery failed", "error", err, "delivery_id", res.DeliveryID)
		return Result{}, fmt.Errorf("метка приёма из идентификатора: %w", err)
	}

	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

	// Учёт ведётся на контексте, ПЕРЕЖИВАЮЩЕМ обрыв соединения. Тело к этому
	// моменту уже на диске (arch.Write контекста не берёт), и отказ вставки
	// из-за ушедшего клиента оставил бы тело сиротой: доставки в журнале нет,
	// а вернуть её может только пересборка с ручной подменой базы. Проверка
	// формы выше остаётся отменяемой — там отмена уместна.
	recordCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), recordTimeout)
	defer cancel()

	err = s.store.CreateDelivery(recordCtx, 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 {
		// Тело уже на диске — данные не потеряны, но учёта нет. Такое тело
		// подберёт пересборка (`healthlog reindex`), заведя запись заново.
		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", clip(meta.AutomationName),
		"aggregation", clip(meta.Aggregation),
		"period", clip(meta.Period))

	// Сигнал идёт последним — после того, как строка учёта закоммичена: иначе
	// воркер мог бы проснуться раньше, чем увидит доставку, и потратить проход
	// впустую. Потеря сигнала отказом не является: доставка числится `pending`,
	// и её подберёт следующий сигнал, тик воркера или старт сервиса.
	s.notify()

	return res, nil
}

// maxAttrLen — сколько байт значения заголовка попадает в лог.
//
// Заголовки контролирует отправитель целиком, а `MaxHeaderBytes` у Go — мегабайт
// на запрос: без границы одна доставка выдавливает из ротации логов всю недавнюю
// историю, включая записи, по которым эту же доставку потом разыскивают. Та же
// граница по той же причине стоит на именах метрик и секций.
const maxAttrLen = 128

// clip обрезает значение, пришедшее от отправителя, до пригодного для лога.
func clip(s string) string {
	if len(s) <= maxAttrLen {
		return s
	}
	// Обрезка названа в самом значении: молча укороченное имя автоматизации
	// выглядит как другое имя.
	return s[:maxAttrLen] + "…(обрезано)"
}

// 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
}