av
41f2f05f2f
- POST /api/v1/ingest: токен, лимит тела, gzip, durable-запись тела в сырой архив, учёт доставки в SQLite - код ответа отражает доставку, а не разбор: битый JSON — 400, непонятое содержимое — 200, данные уже сохранены и доразберутся позже - сохраняется полный набор заголовков запроса с вычисткой секретов по имени и по совпадению значения с токеном
130 lines
4.2 KiB
Go
130 lines
4.2 KiB
Go
// Package archive — сырой архив принятых тел запросов.
|
|
//
|
|
// Архив, а не витрина, является источником истины: тело каждого запроса
|
|
// ложится на диск как есть до любого разбора, и витрина в любой момент
|
|
// пересобирается из него. Отсюда требование к записи — durability: пока файл
|
|
// не лёг на диск, доставка не считается принятой.
|
|
package archive
|
|
|
|
import (
|
|
"compress/gzip"
|
|
"fmt"
|
|
"io"
|
|
"os"
|
|
"path/filepath"
|
|
"time"
|
|
)
|
|
|
|
// Archive — каталог сырых тел, разложенных по дате приёма.
|
|
type Archive struct {
|
|
root string
|
|
}
|
|
|
|
// New создаёт архив в указанном каталоге (каталог создаётся при отсутствии).
|
|
func New(root string) (*Archive, error) {
|
|
if err := os.MkdirAll(root, 0o755); err != nil {
|
|
return nil, fmt.Errorf("create archive dir %q: %w", root, err)
|
|
}
|
|
return &Archive{root: root}, nil
|
|
}
|
|
|
|
// Root возвращает корневой каталог архива.
|
|
func (a *Archive) Root() string { return a.root }
|
|
|
|
// Write сохраняет тело запроса и возвращает путь относительно корня архива.
|
|
//
|
|
// Запись durable: временный файл → fsync → атомарный rename → fsync каталога.
|
|
// Без этого «сохранили — значит приняли» было бы обещанием, которое не
|
|
// переживает потерю питания.
|
|
func (a *Archive) Write(id string, at time.Time, body []byte) (string, error) {
|
|
rel := filepath.Join(at.UTC().Format("2006/01/02"), id+".json.gz")
|
|
full := filepath.Join(a.root, rel)
|
|
|
|
dir := filepath.Dir(full)
|
|
if err := os.MkdirAll(dir, 0o755); err != nil {
|
|
return "", fmt.Errorf("create archive dir %q: %w", dir, err)
|
|
}
|
|
|
|
tmp := full + ".tmp"
|
|
if err := writeGzip(tmp, body); err != nil {
|
|
_ = os.Remove(tmp)
|
|
return "", err
|
|
}
|
|
if err := os.Rename(tmp, full); err != nil {
|
|
_ = os.Remove(tmp)
|
|
return "", fmt.Errorf("rename archive file: %w", err)
|
|
}
|
|
if err := syncDir(dir); err != nil {
|
|
return "", err
|
|
}
|
|
return rel, nil
|
|
}
|
|
|
|
// Open открывает сохранённое тело для чтения (распакованным). Нужен для
|
|
// пересборки витрины из архива.
|
|
func (a *Archive) Open(rel string) (io.ReadCloser, error) {
|
|
f, err := os.Open(filepath.Join(a.root, rel))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("open archive file %q: %w", rel, err)
|
|
}
|
|
gz, err := gzip.NewReader(f)
|
|
if err != nil {
|
|
_ = f.Close()
|
|
return nil, fmt.Errorf("gunzip archive file %q: %w", rel, err)
|
|
}
|
|
return readCloser{gz: gz, f: f}, nil
|
|
}
|
|
|
|
type readCloser struct {
|
|
gz *gzip.Reader
|
|
f *os.File
|
|
}
|
|
|
|
func (r readCloser) Read(p []byte) (int, error) { return r.gz.Read(p) } //nolint:wrapcheck // io.Reader: ошибку отдаём как есть
|
|
|
|
func (r readCloser) Close() error {
|
|
err := r.gz.Close()
|
|
if cerr := r.f.Close(); err == nil {
|
|
err = cerr
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("close archive file: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func writeGzip(path string, body []byte) error {
|
|
f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o644)
|
|
if err != nil {
|
|
return fmt.Errorf("create archive file: %w", err)
|
|
}
|
|
defer func() { _ = f.Close() }()
|
|
|
|
gz := gzip.NewWriter(f)
|
|
if _, err := gz.Write(body); err != nil {
|
|
return fmt.Errorf("gzip archive body: %w", err)
|
|
}
|
|
if err := gz.Close(); err != nil {
|
|
return fmt.Errorf("flush gzip: %w", err)
|
|
}
|
|
if err := f.Sync(); err != nil {
|
|
return fmt.Errorf("fsync archive file: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// syncDir фиксирует появление нового имени в каталоге: без этого rename может
|
|
// не пережить внезапную перезагрузку.
|
|
func syncDir(dir string) error {
|
|
d, err := os.Open(dir)
|
|
if err != nil {
|
|
return fmt.Errorf("open archive dir: %w", err)
|
|
}
|
|
defer func() { _ = d.Close() }()
|
|
|
|
if err := d.Sync(); err != nil {
|
|
return fmt.Errorf("fsync archive dir: %w", err)
|
|
}
|
|
return nil
|
|
}
|