todo

go.mod

@@ -0,0 +1,3 @@
+module replicator
+
+go 1.21.1

main.go

@@ -0,0 +1,17 @@
+package main
+
+import (
+	"context"
+	"os/signal"
+	"replicator/replicator"
+	"syscall"
+)
+
+func main() {
+	ctx, can := signal.NotifyContext(context.Background(), syscall.SIGINT)
+	defer can()
+
+	if err := replicator.Main(ctx); err != nil && ctx.Err() == nil {
+		panic(err)
+	}
+}

replicator/config.go

@@ -0,0 +1,20 @@
+package replicator
+
+import "net/url"
+
+type (
+	Config struct {
+		Local  Storage
+		Remote Storage
+	}
+
+	Storage struct {
+		Driver      url.URL
+		Full        Stream
+		Incremental Stream
+	}
+
+	Stream struct {
+		Disabled bool
+	}
+)

replicator/driver.go

@@ -0,0 +1,60 @@
+package replicator
+
+import (
+	"context"
+	"fmt"
+	"net/url"
+	"path"
+	"time"
+)
+
+type (
+	Driver interface {
+		KeysSince(context.Context, time.Time) (chan KeyVersion, *error)
+		Get(context.Context, Key) (ValueVersion, error)
+		Set(context.Context, Key, Value, Version) error
+		Del(context.Context, Key, Version) error
+	}
+
+	Key struct {
+		Namespace string
+		Key       string
+	}
+
+	Value []byte
+
+	Version []byte
+
+	KeyVersion struct {
+		Key     Key
+		Version []byte
+	}
+
+	ValueVersion struct {
+		Value   []byte
+		Version Version
+	}
+)
+
+func NewDriver(ctx context.Context, driver url.URL) (Driver, error) {
+	switch driver.Scheme {
+	case "filetree":
+		p := driver.Path
+		if driver.Host != "" {
+			p = path.Join(driver.Host, p)
+		}
+		return NewFileTree(p)
+	case "map":
+		return NewMap(), nil
+	default:
+		return nil, fmt.Errorf("unknown driver spec %s", driver.String())
+	}
+}
+
+func (version Version) ToTime() (time.Time, error) {
+	panic(nil) // TODO
+}
+
+func VersionFromTime(t time.Time) Version {
+	panic(nil) // TODO
+}

replicator/driver_test.go

@@ -0,0 +1,86 @@
+package replicator
+
+import (
+	"bytes"
+	"context"
+	"testing"
+	"time"
+)
+
+func TestDriverInterface(t *testing.T) {
+	var _ Driver = FileTree("")
+	var _ Driver = Map{}
+}
+
+func testDriver(t *testing.T, d Driver) {
+	ctx, can := context.WithTimeout(context.Background(), time.Second*2)
+	defer can()
+
+	key := Key{Namespace: "x/y", Key: "z"}
+	value := ValueVersion{Value: []byte(t.Name()), Version: []byte("1")}
+
+	t.Run("get does not exist", func(t *testing.T) {
+		v, err := d.Get(ctx, key)
+		if err != nil {
+			t.Errorf("getting 404 returned an err: %v", err)
+		}
+		if v.Value != nil || v.Version != nil {
+			t.Errorf("expected nil for 404 but got %q/%s", v.Value, v.Version)
+		}
+	})
+
+	t.Run("404 set get", func(t *testing.T) {
+		if err := d.Set(ctx, key, value.Value, value.Version); err != nil {
+			t.Fatal(err)
+		}
+
+		v, err := d.Get(ctx, key)
+		if err != nil {
+			t.Errorf("getting key returned an err: %v", err)
+		}
+		if !bytes.Equal(v.Value, value.Value) {
+			t.Errorf("value didnt match set-get: want %q, got %q", value.Value, v.Value)
+		}
+		if !bytes.Equal(v.Version, value.Version) {
+			t.Errorf("version didnt match set-get: want %q, got %q", value.Version, v.Version)
+		}
+
+		if err := d.Del(ctx, key, nil); err != nil {
+			t.Errorf("failed to clean up: %v", err)
+		}
+	})
+
+	t.Run("keys of nothing", func(t *testing.T) {
+		ch, err := d.KeysSince(ctx, time.Time{})
+		for got := range ch {
+			t.Error("expected nothing but got", got)
+		}
+		if *err != nil {
+			t.Error(*err)
+		}
+	})
+
+	t.Run("keys of one key", func(t *testing.T) {
+		if err := d.Set(ctx, key, value.Value, value.Version); err != nil {
+			t.Fatal(err)
+		}
+		defer d.Del(ctx, key, nil)
+
+		ch, err := d.KeysSince(ctx, time.Time{})
+		n := 0
+		for got := range ch {
+			n += 1
+			if got.Key != key {
+				t.Error(got)
+			} else if !bytes.Equal(got.Version, value.Version) {
+				t.Error(got)
+			}
+		}
+		if n == 0 {
+			t.Error("expected to find at least 1 key")
+		}
+		if *err != nil {
+			t.Error(*err)
+		}
+	})
+}

replicator/filetree.go

@@ -0,0 +1,184 @@
+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) ([]byte, error) {
+	v, err := tree._getVersion(key)
+	if v == (time.Time{}) {
+		return nil, err
+	}
+	return []byte(strconv.FormatInt(v.UnixNano(), 10)), 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 []byte) (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,
+	}
+}

replicator/filetree_test.go

@@ -0,0 +1,57 @@
+package replicator
+
+import (
+	"testing"
+)
+
+func TestFileTree(t *testing.T) {
+	d := t.TempDir()
+	tree, err := NewFileTree(d)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	testDriver(t, tree)
+}
+
+func TestFileTreeRealpath(t *testing.T) {
+	root := "/root"
+	base := "base"
+	cases := map[string]string{
+		"a/b": "/root/a.d/b.d/base.bin",
+		"a":   "/root/a.d/base.bin",
+		"":    "/root/base.bin",
+	}
+
+	for given, wantd := range cases {
+		want := wantd
+		t.Run(given, func(t *testing.T) {
+			key := Key{Key: base, Namespace: given}
+			got := FileTree(root).realpath(key)
+			if got != want {
+				t.Error(got)
+			}
+		})
+	}
+}
+
+func TestFileTreeToKey(t *testing.T) {
+	root := "/root"
+	cases := map[string]Key{
+		"/root/a.d/b.bin":     {Namespace: "a", Key: "b"},
+		"//root/a.d/b.bin":    {Namespace: "a", Key: "b"},
+		"/root/b.bin":         {Namespace: "", Key: "b"},
+		"/root/c.d/a.d/b.bin": {Namespace: "c/a", Key: "b"},
+		"/root/c/a.d/b.bin":   {Namespace: "c/a", Key: "b"},
+	}
+
+	for given, wantd := range cases {
+		want := wantd
+		t.Run(given, func(t *testing.T) {
+			got := FileTree(root).toKey(given)
+			if got != want {
+				t.Error(got)
+			}
+		})
+	}
+}

replicator/map.go

@@ -0,0 +1,84 @@
+package replicator
+
+import (
+	"context"
+	"sync"
+	"time"
+)
+
+type Map struct {
+	m    map[Key]ValueVersion
+	lock *sync.RWMutex
+}
+
+func NewMap() Map {
+	return Map{
+		m:    make(map[Key]ValueVersion),
+		lock: &sync.RWMutex{},
+	}
+}
+
+func (m Map) KeysSince(ctx context.Context, t time.Time) (chan KeyVersion, *error) {
+	result := make(chan KeyVersion)
+	var err error
+	go func() {
+		defer close(result)
+		m.lock.RLock()
+		defer m.lock.RUnlock()
+		for k := range m.m {
+			select {
+			case result <- KeyVersion{Key: k, Version: m.m[k].Version}:
+			case <-ctx.Done():
+				err = ctx.Err()
+				return
+			}
+		}
+	}()
+	return result, &err
+}
+
+func (m Map) Get(_ context.Context, k Key) (ValueVersion, error) {
+	m.lock.RLock()
+	defer m.lock.RUnlock()
+	return m.m[k], nil
+}
+
+func (m Map) Set(_ context.Context, key Key, value Value, version Version) error {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+
+	if version != nil {
+		if was, ok := m.m[key]; !ok {
+		} else if wasVersion, err := was.Version.ToTime(); err != nil {
+			return err
+		} else if wantVersion, err := version.ToTime(); err != nil {
+			return err
+		} else if wantVersion.Before(wasVersion) {
+			return nil // conflict
+		}
+	}
+
+	m.m[key] = ValueVersion{Value: value, Version: version}
+	return nil
+}
+
+func (m Map) Del(_ context.Context, k Key, v Version) error {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+
+	if v != nil {
+		if was, ok := m.m[k]; !ok {
+			return nil
+		} else if wasVersion, err := was.Version.ToTime(); err != nil {
+			return err
+		} else if wantVersion, err := v.ToTime(); err != nil {
+			return err
+		} else if wantVersion.Before(wasVersion) {
+			return nil // conflict
+		}
+	}
+
+	delete(m.m, k)
+
+	return nil
+}

replicator/map_test.go

@@ -0,0 +1,9 @@
+package replicator
+
+import (
+	"testing"
+)
+
+func TestMap(t *testing.T) {
+	testDriver(t, NewMap())
+}

replicator/replicator.go

@@ -0,0 +1,10 @@
+package replicator
+
+import (
+	"context"
+	"errors"
+)
+
+func Main(ctx context.Context) error {
+	return errors.New("not impl")
+}