279 lines
7.0 KiB
Go
279 lines
7.0 KiB
Go
package storage
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
)
|
|
|
|
// ── PVC Store ──────────────────────────────
|
|
|
|
// PVCStore implements ObjectStore using the local filesystem.
|
|
// Used with Kubernetes PersistentVolumeClaims (CephFS, NFS, local PV)
|
|
// or plain Docker volume mounts.
|
|
type PVCStore struct {
|
|
basePath string
|
|
|
|
// Content-type cache: key → content type.
|
|
// Populated on Put, used by Get. PVC doesn't store metadata
|
|
// alongside files, so we keep MIME types in memory. On restart,
|
|
// Get falls back to "application/octet-stream" — the DB has the
|
|
// authoritative content_type for serving to clients.
|
|
mu sync.RWMutex
|
|
mimeMap map[string]string
|
|
}
|
|
|
|
// NewPVC creates a PVCStore rooted at basePath.
|
|
// Returns an error if the path does not exist or is not writable.
|
|
func NewPVC(basePath string) (*PVCStore, error) {
|
|
// Ensure base path exists
|
|
if err := os.MkdirAll(basePath, 0o755); err != nil {
|
|
return nil, fmt.Errorf("storage/pvc: cannot create base path %q: %w", basePath, err)
|
|
}
|
|
|
|
// Create well-known subdirectories
|
|
for _, sub := range []string{"attachments", "processing"} {
|
|
if err := os.MkdirAll(filepath.Join(basePath, sub), 0o755); err != nil {
|
|
return nil, fmt.Errorf("storage/pvc: cannot create %s dir: %w", sub, err)
|
|
}
|
|
}
|
|
|
|
s := &PVCStore{
|
|
basePath: basePath,
|
|
mimeMap: make(map[string]string),
|
|
}
|
|
|
|
// Validate writability
|
|
if err := s.Healthy(context.Background()); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return s, nil
|
|
}
|
|
|
|
func (s *PVCStore) Backend() string { return "pvc" }
|
|
|
|
// Put writes data to the filesystem at the resolved key path.
|
|
func (s *PVCStore) Put(ctx context.Context, key string, r io.Reader, size int64, contentType string) error {
|
|
if err := validateKey(key); err != nil {
|
|
return err
|
|
}
|
|
|
|
absPath := s.resolve(key)
|
|
|
|
// Ensure parent directory exists
|
|
if err := os.MkdirAll(filepath.Dir(absPath), 0o755); err != nil {
|
|
return fmt.Errorf("storage/pvc: mkdir for %q: %w", key, err)
|
|
}
|
|
|
|
// Write to temp file first, then rename (atomic on same filesystem)
|
|
tmp, err := os.CreateTemp(filepath.Dir(absPath), ".upload-*")
|
|
if err != nil {
|
|
return fmt.Errorf("storage/pvc: create temp for %q: %w", key, err)
|
|
}
|
|
tmpName := tmp.Name()
|
|
defer func() {
|
|
tmp.Close()
|
|
// Clean up temp file on failure
|
|
if err != nil {
|
|
os.Remove(tmpName)
|
|
}
|
|
}()
|
|
|
|
written, err := io.Copy(tmp, r)
|
|
if err != nil {
|
|
return fmt.Errorf("storage/pvc: write %q: %w", key, err)
|
|
}
|
|
if err = tmp.Close(); err != nil {
|
|
return fmt.Errorf("storage/pvc: close %q: %w", key, err)
|
|
}
|
|
|
|
// Verify size if provided (0 means unknown)
|
|
if size > 0 && written != size {
|
|
os.Remove(tmpName)
|
|
return fmt.Errorf("storage/pvc: size mismatch for %q: expected %d, got %d", key, size, written)
|
|
}
|
|
|
|
// Atomic rename
|
|
if err = os.Rename(tmpName, absPath); err != nil {
|
|
return fmt.Errorf("storage/pvc: rename %q: %w", key, err)
|
|
}
|
|
|
|
// Cache content type
|
|
if contentType != "" {
|
|
s.mu.Lock()
|
|
s.mimeMap[key] = contentType
|
|
s.mu.Unlock()
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// Get returns a reader for the file at key.
|
|
func (s *PVCStore) Get(ctx context.Context, key string) (io.ReadCloser, int64, string, error) {
|
|
if err := validateKey(key); err != nil {
|
|
return nil, 0, "", err
|
|
}
|
|
|
|
absPath := s.resolve(key)
|
|
|
|
info, err := os.Stat(absPath)
|
|
if err != nil {
|
|
if os.IsNotExist(err) {
|
|
return nil, 0, "", ErrNotFound
|
|
}
|
|
return nil, 0, "", fmt.Errorf("storage/pvc: stat %q: %w", key, err)
|
|
}
|
|
|
|
f, err := os.Open(absPath)
|
|
if err != nil {
|
|
return nil, 0, "", fmt.Errorf("storage/pvc: open %q: %w", key, err)
|
|
}
|
|
|
|
// Look up cached content type
|
|
s.mu.RLock()
|
|
ct := s.mimeMap[key]
|
|
s.mu.RUnlock()
|
|
if ct == "" {
|
|
ct = "application/octet-stream"
|
|
}
|
|
|
|
return f, info.Size(), ct, nil
|
|
}
|
|
|
|
// Delete removes the file at key.
|
|
func (s *PVCStore) Delete(ctx context.Context, key string) error {
|
|
if err := validateKey(key); err != nil {
|
|
return err
|
|
}
|
|
|
|
absPath := s.resolve(key)
|
|
|
|
err := os.Remove(absPath)
|
|
if err != nil && !os.IsNotExist(err) {
|
|
return fmt.Errorf("storage/pvc: delete %q: %w", key, err)
|
|
}
|
|
|
|
// Clean up cache
|
|
s.mu.Lock()
|
|
delete(s.mimeMap, key)
|
|
s.mu.Unlock()
|
|
|
|
return nil
|
|
}
|
|
|
|
// DeletePrefix removes all files and directories under the given prefix.
|
|
func (s *PVCStore) DeletePrefix(ctx context.Context, prefix string) error {
|
|
if err := validateKey(prefix); err != nil {
|
|
return err
|
|
}
|
|
|
|
absPath := s.resolve(prefix)
|
|
|
|
// Check if the path exists
|
|
if _, err := os.Stat(absPath); os.IsNotExist(err) {
|
|
return nil // Nothing to delete
|
|
}
|
|
|
|
if err := os.RemoveAll(absPath); err != nil {
|
|
return fmt.Errorf("storage/pvc: delete prefix %q: %w", prefix, err)
|
|
}
|
|
|
|
// Clean up cache entries with this prefix
|
|
s.mu.Lock()
|
|
for k := range s.mimeMap {
|
|
if strings.HasPrefix(k, prefix) {
|
|
delete(s.mimeMap, k)
|
|
}
|
|
}
|
|
s.mu.Unlock()
|
|
|
|
return nil
|
|
}
|
|
|
|
// Exists checks if a file exists at key.
|
|
func (s *PVCStore) Exists(ctx context.Context, key string) (bool, error) {
|
|
if err := validateKey(key); err != nil {
|
|
return false, err
|
|
}
|
|
|
|
_, err := os.Stat(s.resolve(key))
|
|
if err == nil {
|
|
return true, nil
|
|
}
|
|
if os.IsNotExist(err) {
|
|
return false, nil
|
|
}
|
|
return false, fmt.Errorf("storage/pvc: exists %q: %w", key, err)
|
|
}
|
|
|
|
// Healthy verifies the storage path is writable.
|
|
func (s *PVCStore) Healthy(ctx context.Context) error {
|
|
probe := filepath.Join(s.basePath, ".health-probe")
|
|
if err := os.WriteFile(probe, []byte("ok"), 0o644); err != nil {
|
|
return fmt.Errorf("storage/pvc: not writable: %w", err)
|
|
}
|
|
os.Remove(probe)
|
|
return nil
|
|
}
|
|
|
|
// Stats returns file count and total size under the base path.
|
|
func (s *PVCStore) Stats(ctx context.Context) (*StorageStats, error) {
|
|
stats := &StorageStats{
|
|
Backend: "pvc",
|
|
Path: s.basePath,
|
|
Configured: true,
|
|
}
|
|
|
|
// Check health
|
|
if err := s.Healthy(ctx); err != nil {
|
|
stats.Healthy = false
|
|
return stats, nil
|
|
}
|
|
stats.Healthy = true
|
|
|
|
// Walk attachments directory for counts
|
|
attDir := filepath.Join(s.basePath, "attachments")
|
|
err := filepath.Walk(attDir, func(path string, info os.FileInfo, err error) error {
|
|
if err != nil {
|
|
return nil // skip errors (e.g. permission denied)
|
|
}
|
|
if !info.IsDir() {
|
|
stats.TotalFiles++
|
|
stats.TotalBytes += info.Size()
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil && !os.IsNotExist(err) {
|
|
return stats, fmt.Errorf("storage/pvc: stats walk: %w", err)
|
|
}
|
|
|
|
return stats, nil
|
|
}
|
|
|
|
// ── Helpers ────────────────────────────────
|
|
|
|
// resolve converts a storage key to an absolute filesystem path.
|
|
func (s *PVCStore) resolve(key string) string {
|
|
return filepath.Join(s.basePath, filepath.FromSlash(key))
|
|
}
|
|
|
|
// validateKey ensures the key is safe (no path traversal, no absolute paths).
|
|
func validateKey(key string) error {
|
|
if key == "" {
|
|
return fmt.Errorf("storage: empty key")
|
|
}
|
|
if strings.HasPrefix(key, "/") || strings.HasPrefix(key, "\\") {
|
|
return fmt.Errorf("storage: absolute key not allowed: %q", key)
|
|
}
|
|
cleaned := filepath.Clean(key)
|
|
if strings.Contains(cleaned, "..") {
|
|
return fmt.Errorf("storage: path traversal not allowed: %q", key)
|
|
}
|
|
return nil
|
|
}
|