initial commit

pkg/fs/fs.go

@@ -0,0 +1,225 @@
+// Copyright (C) 2022 Marius Schellenberger
+
+package fs
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	stdfs "io/fs"
+	"net/http"
+	"os"
+	"path"
+	"path/filepath"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/go-logr/logr"
+)
+
+type FS struct {
+	mu      sync.RWMutex
+	log     logr.Logger
+	NoCache http.Handler
+	src     string
+	dst     string
+	mdf     *os.File
+	jenc    *json.Encoder
+	mm      map[string]*Metadata
+	cancel  func()
+}
+
+func NewFS(src, dst, metadata string, log logr.Logger) (fs *FS, err error) {
+	if !filepath.IsAbs(src) {
+		return nil, errors.New("src path is not absolute")
+	}
+	if !filepath.IsAbs(dst) {
+		return nil, errors.New("dst path is not absolute")
+	}
+	if !filepath.IsAbs(metadata) {
+		return nil, errors.New("metadata path is not absolute")
+	}
+	if src == dst {
+		return nil, errors.New("src and dst path can not be equal")
+	}
+	mdf, err := os.OpenFile(metadata, os.O_RDWR|os.O_CREATE, 0o640)
+	if err != nil {
+		return
+	}
+	fs = &FS{
+		log:     log,
+		NoCache: http.FileServer(http.Dir(src)),
+		src:     src,
+		dst:     dst,
+		mdf:     mdf,
+		mm:      make(map[string]*Metadata),
+		jenc:    json.NewEncoder(mdf),
+	}
+	err = json.NewDecoder(mdf).Decode(&fs.mm)
+	if err == io.EOF {
+		err = nil
+	}
+	for k := range fs.mm {
+		_, err = stat(fs.dst, k)
+		if errors.Is(err, os.ErrNotExist) {
+			delete(fs.mm, k)
+		}
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	fs.cancel = cancel
+	go fs.flusher(ctx)
+	return fs, err
+}
+
+func (fs *FS) Stat(name string) (stdfs.FileInfo, error) {
+	return stat(fs.src, name)
+}
+
+func (fs *FS) Open(name string) (http.File, error) {
+	return fs.open(name)
+}
+
+func (fs *FS) Preload(name string) {
+	f, err := fs.open(name)
+	if err != nil {
+		return
+	}
+	if f.md.Preload() {
+		fs.log.WithValues("file", name).Error(errors.New("preload is running"), "error starting preload")
+		return
+	}
+	unlock := func() { f.md.UnlockPreload() }
+	go f.Preload(unlock)
+}
+
+func skipLog(name string) bool {
+	return strings.HasSuffix(name, "index.html")
+}
+
+func (fs *FS) open(name string) (f *File, err error) {
+	log := fs.log.WithValues("file", name)
+	rp := filepath.Join(fs.src, filepath.FromSlash(path.Clean("/"+name)))
+	rf, err := os.Open(rp)
+	if err != nil {
+		if !skipLog(name) {
+			log.Error(err, "error opening source file")
+		}
+		return
+	}
+	rfi, err := rf.Stat()
+	if err != nil {
+		log.Error(err, "error stat source file")
+		return
+	}
+	if rfi.IsDir() {
+		return &File{f: rf}, nil
+	}
+	mp := filepath.Join(fs.dst, filepath.FromSlash(path.Clean("/"+name)))
+	var mf *os.File
+	i := strings.LastIndex(mp, "/")
+	if i > 0 {
+		dir := mp[:i]
+		_, err = os.Stat(dir)
+		if errors.Is(err, os.ErrNotExist) {
+			err = os.MkdirAll(dir, 0o755)
+			if err != nil {
+				log.Error(err, "error creating cache dir", "dir", mp)
+				return nil, err
+			}
+		}
+	}
+	_, err = os.Stat(mp)
+	truncate := errors.Is(err, os.ErrNotExist)
+	if !fs.isOpen(name) {
+		mf, err = os.OpenFile(mp, os.O_RDWR|os.O_CREATE, 0o644)
+		if err != nil {
+			log.Error(err, "error opening cache file")
+			return nil, err
+		}
+		if truncate {
+			err = mf.Truncate(rfi.Size())
+			if err != nil {
+				log.Error(err, "error truncating cache file")
+				return nil, err
+			}
+			err = mf.Sync()
+			if err != nil {
+				log.Error(err, "error syncing cache file")
+				return nil, err
+			}
+		}
+	}
+	return &File{
+		log: log,
+		f:   rf,
+		md:  fs.metadata(name, rfi.Size(), mf),
+	}, nil
+}
+
+func (fs *FS) isOpen(name string) bool {
+	fs.mu.RLock()
+	defer fs.mu.RUnlock()
+	md, ok := fs.mm[name]
+	return ok && md.f != nil
+}
+
+func (fs *FS) metadata(name string, size int64, f *os.File) (md *Metadata) {
+	fs.mu.Lock()
+	defer fs.mu.Unlock()
+	if md, ok := fs.mm[name]; ok {
+		if md.f == nil {
+			md.f = f
+		}
+		return md
+	}
+	md = &Metadata{Size: size, f: f}
+	fs.mm[name] = md
+	return
+}
+
+func (fs *FS) flusher(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(time.Minute * 5):
+		}
+		fs.flushMetadata()
+	}
+}
+
+func (fs *FS) flushMetadata() {
+	log := fs.log
+	fs.mu.Lock()
+	defer fs.mu.Unlock()
+	_, err := fs.mdf.Seek(0, io.SeekStart)
+	if err != nil {
+		log.Error(err, "failure seeking metadata file")
+		return
+	}
+	err = fs.mdf.Truncate(0)
+	if err != nil {
+		log.Error(err, "failure truncating metadata file")
+		return
+	}
+	err = fs.jenc.Encode(fs.mm)
+	if err != nil {
+		log.Error(err, "failure flushing metadata file")
+	}
+}
+
+func (fs *FS) Close() {
+	fs.cancel()
+	for _, md := range fs.mm {
+		md.f.Sync()
+		md.f.Close()
+	}
+	fs.flushMetadata()
+}
+
+func stat(prefix, name string) (stdfs.FileInfo, error) {
+	p := filepath.Join(prefix, filepath.FromSlash(path.Clean("/"+name)))
+	return os.Stat(p)
+}