Archived
152 lines
3.9 KiB
Go
152 lines
3.9 KiB
Go
package main
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
)
|
|
|
|
var ErrTimestampNotNewer = errors.New("timestamp is not newer than existing data")
|
|
|
|
// Storage handles atomic file operations with per-file locking.
|
|
type Storage struct {
|
|
dataDir string
|
|
locks sync.Map // map[string]*sync.RWMutex, keyed by filename
|
|
}
|
|
|
|
// NewStorage creates a new Storage instance.
|
|
func NewStorage(dataDir string) *Storage {
|
|
return &Storage{
|
|
dataDir: dataDir,
|
|
}
|
|
}
|
|
|
|
// getLock returns the mutex for a given filename.
|
|
func (s *Storage) getLock(filename string) *sync.RWMutex {
|
|
lock, _ := s.locks.LoadOrStore(filename, &sync.RWMutex{})
|
|
return lock.(*sync.RWMutex)
|
|
}
|
|
|
|
// GetTimestamp returns the stored timestamp for a filename, or 0 if no data exists.
|
|
func (s *Storage) GetTimestamp(filename string) (int64, error) {
|
|
lock := s.getLock(filename)
|
|
lock.RLock()
|
|
defer lock.RUnlock()
|
|
|
|
_, ts, err := s.findExistingFile(filename)
|
|
return ts, err
|
|
}
|
|
|
|
// findExistingFile finds the existing file for a filename and extracts the timestamp.
|
|
// Returns empty path and 0 timestamp if no file exists.
|
|
func (s *Storage) findExistingFile(filename string) (string, int64, error) {
|
|
pattern := filepath.Join(s.dataDir, filename+"_*.dat")
|
|
matches, err := filepath.Glob(pattern)
|
|
if err != nil {
|
|
return "", 0, fmt.Errorf("failed to glob for existing files: %w", err)
|
|
}
|
|
|
|
if len(matches) == 0 {
|
|
return "", 0, nil
|
|
}
|
|
|
|
// Use the first match (there should only be one per filename)
|
|
path := matches[0]
|
|
ts, err := s.extractTimestamp(path, filename)
|
|
if err != nil {
|
|
return "", 0, err
|
|
}
|
|
|
|
return path, ts, nil
|
|
}
|
|
|
|
// extractTimestamp extracts the timestamp from a path like "datadir/filename_1234567890.dat"
|
|
func (s *Storage) extractTimestamp(path string, filename string) (int64, error) {
|
|
base := filepath.Base(path)
|
|
base = strings.TrimSuffix(base, ".dat")
|
|
|
|
prefix := filename + "_"
|
|
if !strings.HasPrefix(base, prefix) {
|
|
return 0, fmt.Errorf("invalid filename format: %s", path)
|
|
}
|
|
|
|
tsStr := strings.TrimPrefix(base, prefix)
|
|
ts, err := strconv.ParseInt(tsStr, 10, 64)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("invalid timestamp in filename %s: %w", path, err)
|
|
}
|
|
|
|
return ts, nil
|
|
}
|
|
|
|
// WriteIfNewer atomically writes data if the timestamp is newer than the existing data.
|
|
// Returns ErrTimestampNotNewer if the timestamp is not newer.
|
|
func (s *Storage) WriteIfNewer(filename string, timestamp int64, data []byte) error {
|
|
lock := s.getLock(filename)
|
|
lock.Lock()
|
|
defer lock.Unlock()
|
|
|
|
existingPath, existingTS, err := s.findExistingFile(filename)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if timestamp <= existingTS {
|
|
return ErrTimestampNotNewer
|
|
}
|
|
|
|
return s.writeAtomic(filename, timestamp, data, existingPath)
|
|
}
|
|
|
|
// writeAtomic writes data atomically using temp file + rename (caller must hold lock).
|
|
func (s *Storage) writeAtomic(filename string, timestamp int64, data []byte, oldPath string) error {
|
|
finalPath := s.filePath(filename, timestamp)
|
|
tempPath := finalPath + ".tmp"
|
|
|
|
f, err := os.Create(tempPath)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to create temp file: %w", err)
|
|
}
|
|
|
|
// Write data only (timestamp is in filename)
|
|
if _, err := f.Write(data); err != nil {
|
|
f.Close()
|
|
os.Remove(tempPath)
|
|
return fmt.Errorf("failed to write data: %w", err)
|
|
}
|
|
|
|
// Sync to disk
|
|
if err := f.Sync(); err != nil {
|
|
f.Close()
|
|
os.Remove(tempPath)
|
|
return fmt.Errorf("failed to sync file: %w", err)
|
|
}
|
|
|
|
if err := f.Close(); err != nil {
|
|
os.Remove(tempPath)
|
|
return fmt.Errorf("failed to close temp file: %w", err)
|
|
}
|
|
|
|
// Atomic rename
|
|
if err := os.Rename(tempPath, finalPath); err != nil {
|
|
os.Remove(tempPath)
|
|
return fmt.Errorf("failed to rename temp file: %w", err)
|
|
}
|
|
|
|
// Remove old file if it exists
|
|
if oldPath != "" {
|
|
os.Remove(oldPath)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// filePath returns the file path for a given filename and timestamp.
|
|
func (s *Storage) filePath(filename string, timestamp int64) string {
|
|
return filepath.Join(s.dataDir, fmt.Sprintf("%s_%d.dat", filename, timestamp))
|
|
}
|