// Package archive — сырой архив принятых тел запросов. // // Архив, а не витрина, является источником истины: тело каждого запроса // ложится на диск как есть до любого разбора, и витрина в любой момент // пересобирается из него. Отсюда требование к записи — durability: пока файл // не лёг на диск, доставка не считается принятой. package archive import ( "compress/gzip" "fmt" "io" "io/fs" "os" "path/filepath" "sort" "strings" "time" ) // bodyExt — расширение файла тела. Всё, что ему не соответствует, телом не // является: остаток `*.json.gz.tmp` от прерванной записи, чужой файл, каталог. const bodyExt = ".json.gz" // 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 } // Existing открывает уже существующий архив и каталога не создаёт. // // Отличие от New не косметическое: приёму каталог создать надо, а пересборке — // нельзя. Созданный на лету пустой каталог превращает запуск не из той // директории в успешный прогон по пустому журналу, а пустая витрина совпадает // по отпечатку с пустой витриной, то есть выглядит идеальной сходимостью. func Existing(root string) (*Archive, error) { fi, err := os.Stat(root) if err != nil { return nil, fmt.Errorf("open archive dir %q: %w", root, err) } if !fi.IsDir() { return nil, fmt.Errorf("archive dir %q: не каталог", root) } 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 } // Listing — что нашлось в архиве. type Listing struct { // Bodies — относительные пути тел, отсортированные лексикографически. // Порядок здесь только для воспроизводимости перечисления: журнал // упорядочивает не он, а время приёма. Bodies []string // Skipped — файлы, телом не являющиеся. Считаются, а не выбрасываются: // молчаливый пропуск означал бы «тело есть, а в отчёте его нет». Skipped []string } // List перечисляет тела архива. // // Ошибку чтения каталога отдаёт наружу, а не превращает в пустой список, и // каталога не создаёт. Различие принципиально для пересборки: нечитаемый или // отсутствующий каталог означает «неизвестно, есть ли тела», а не «тел нет», — // а пустой журнал даёт пустую витрину, чей отпечаток совпадает с отпечатком // любой другой пустой витрины, то есть выглядит идеальной сходимостью. func (a *Archive) List() (Listing, error) { var out Listing err := filepath.WalkDir(a.root, func(path string, d fs.DirEntry, err error) error { if err != nil { // Отказ чтения каталога прекращает обход целиком: пропустить его // значило бы молча потерять сутки журнала. return fmt.Errorf("walk archive %q: %w", path, err) } if d.IsDir() { return nil } rel, err := filepath.Rel(a.root, path) if err != nil { return fmt.Errorf("relative path %q: %w", path, err) } if !strings.HasSuffix(d.Name(), bodyExt) { out.Skipped = append(out.Skipped, rel) return nil } out.Bodies = append(out.Bodies, rel) return nil }) if err != nil { return Listing{}, err //nolint:wrapcheck // ошибка уже обёрнута внутри обхода } sort.Strings(out.Bodies) sort.Strings(out.Skipped) return out, nil } // BodyID возвращает идентификатор доставки по относительному пути тела. func BodyID(rel string) string { return strings.TrimSuffix(filepath.Base(rel), bodyExt) } // 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 }