replicator/replicator/filetree.go

package replicator

import (
	"context"
	"fmt"
	"io"
	"io/fs"
	"io/ioutil"
	"os"
	"path"
	"path/filepath"
	"strconv"
	"strings"
	"time"
)

type FileTree string

func NewFileTree(root string) (FileTree, error) {
	return FileTree(root), nil
}

func (tree FileTree) KeysSince(ctx context.Context, t time.Time) (chan KeyVersion, *error) {
	result := make(chan KeyVersion)
	var final error
	go func() {
		defer close(result)
		if err := filepath.Walk(path.Dir(tree.realpath(Key{})), func(p string, info fs.FileInfo, err error) error {
			if err != nil {
				return err
			}
			if info.IsDir() {
				return ctx.Err()
			}
			key := tree.toKey(p)
			version, err := tree.getVersion(key)
			if err != nil {
				return err
			}
			select {
			case result <- KeyVersion{Key: key, Version: version}:
			case <-ctx.Done():
			}
			return ctx.Err()
		}); err != nil && !os.IsNotExist(err) {
			final = err
		}
	}()
	return result, &final
}

func (tree FileTree) Get(ctx context.Context, key Key) (ValueVersion, error) {
	version, err := tree.getVersion(key)
	if version == nil {
		return ValueVersion{}, err
	}

	f, err := os.Open(tree.realpath(key))
	if err != nil {
		return ValueVersion{}, err
	}
	defer f.Close()
	b, _ := io.ReadAll(f)

	return ValueVersion{
		Value:   b,
		Version: version,
	}, nil
}

func (tree FileTree) Set(ctx context.Context, key Key, value Value, version Version) error {
	v, err := tree.fromVersion(version)
	if err != nil {
		return err
	}

	if oldv, err := tree._getVersion(key); err != nil {
		return err
	} else if v.Before(oldv) {
		return nil // conflict
	}

	f2, err := ioutil.TempFile(os.TempDir(), "*.*")
	if err != nil {
		return err
	}
	defer os.Remove(f2.Name())

	if n, err := f2.Write(value); err != nil {
		return err
	} else if n != len(value) {
		return fmt.Errorf("truncated write")
	}

	if err := f2.Close(); err != nil {
		return err
	}

	if err := os.Chtimes(f2.Name(), time.Time{}, v); err != nil {
		return err
	}

	os.MkdirAll(path.Dir(tree.realpath(key)), os.ModePerm)
	return os.Rename(f2.Name(), tree.realpath(key))
}

func (tree FileTree) Del(ctx context.Context, key Key, version Version) error {
	if version != nil {
		v, err := tree.fromVersion(version)
		if err != nil {
			return err
		}

		if oldv, err := tree._getVersion(key); err != nil {
			return err
		} else if v.Before(oldv) {
			return nil // conflict
		}
	}

	return os.Remove(tree.realpath(key))
}

func (tree FileTree) realpath(key Key) string {
	namespace := ""
	tmp := path.Clean("/" + key.Namespace)
	for path.Dir(tmp) != tmp {
		_, base := path.Split(tmp)
		namespace = path.Join(base+".d", namespace)
		tmp = path.Dir(tmp)
	}
	return path.Join(string(tree), namespace, key.Key+".bin")
}

func (tree FileTree) getVersion(key Key) (Version, error) {
	v, err := tree._getVersion(key)
	if v == (time.Time{}) {
		return nil, err
	}
	return TimeAsVersion(v), err
}

func (tree FileTree) _getVersion(key Key) (time.Time, error) {
	info, err := os.Stat(tree.realpath(key))
	if os.IsNotExist(err) {
		return time.Time{}, nil
	}
	if err != nil {
		return time.Time{}, err
	}
	return info.ModTime(), nil
}

func (tree FileTree) fromVersion(v Version) (time.Time, error) {
	if len(v) == 0 {
		return time.Time{}, nil
	}
	n, err := strconv.ParseInt(string(v), 10, 64)
	if err != nil {
		return time.Time{}, err
	}
	return time.Unix(0, n), nil
}

func (tree FileTree) toKey(p string) Key {
	p = path.Clean(p)
	p = strings.TrimPrefix(p, path.Clean(string(tree)))
	p = strings.TrimPrefix(p, "/")
	p = strings.TrimSuffix(p, ".bin")
	p = strings.ReplaceAll(p, ".d/", "/")
	namespace := path.Dir(p)
	if namespace == "/" || namespace == "." {
		namespace = ""
	}
	key := path.Base(p)
	if key == "" {
		key = namespace
		namespace = ""
	}
	return Key{
		Namespace: namespace,
		Key:       key,
	}
}