Initial import

storage/codec/binary.go

@@ -0,0 +1,248 @@
+package codec
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"io"
+	"math"
+
+	"golang.org/x/crypto/ssh"
+)
+
+func init() {
+	Register("binray", func() Codec { return binaryCodec{order: binary.NativeEndian} })
+	Register("be", func() Codec { return binaryCodec{order: binary.BigEndian} })
+	Register("le", func() Codec { return binaryCodec{order: binary.LittleEndian} })
+	Register("ssh", func() Codec { return sshCodec{} })
+}
+
+type binaryCodec struct {
+	order   binary.ByteOrder
+	scratch [16]byte
+}
+
+func (binaryCodec) Type() string { return "binary" }
+
+func (codec binaryCodec) Encode(value any) ([]byte, error) {
+	switch value := value.(type) {
+	case bool:
+		if value {
+			return []byte{1}, nil
+		}
+		return []byte{0}, nil
+	case *bool:
+		if *value {
+			return []byte{1}, nil
+		}
+		return []byte{0}, nil
+	case int:
+		codec.order.PutUint64(codec.scratch[:], uint64(value))
+		return codec.scratch[:8], nil
+	case *int:
+		return codec.Encode(*value)
+	case int8:
+		return []byte{uint8(value)}, nil
+	case *int8:
+		return []byte{uint8(*value)}, nil
+	case int16:
+		codec.order.PutUint16(codec.scratch[:], uint16(value))
+		return codec.scratch[:2], nil
+	case *int16:
+		return codec.Encode(*value)
+	case int32:
+		codec.order.PutUint32(codec.scratch[:], uint32(value))
+		return codec.scratch[:4], nil
+	case *int32:
+		return codec.Encode(*value)
+	case int64:
+		codec.order.PutUint64(codec.scratch[:], uint64(value))
+		return codec.scratch[:8], nil
+	case *int64:
+		return codec.Encode(*value)
+	case uint:
+		codec.order.PutUint64(codec.scratch[:], uint64(value))
+		return codec.scratch[:8], nil
+	case *uint:
+		return codec.Encode(*value)
+	case uint8:
+		return []byte{value}, nil
+	case *uint8:
+		return []byte{*value}, nil
+	case uint16:
+		codec.order.PutUint16(codec.scratch[:], value)
+		return codec.scratch[:2], nil
+	case *uint16:
+		return codec.Encode(*value)
+	case uint32:
+		codec.order.PutUint32(codec.scratch[:], value)
+		return codec.scratch[:4], nil
+	case *uint32:
+		return codec.Encode(*value)
+	case uint64:
+		codec.order.PutUint64(codec.scratch[:], value)
+		return codec.scratch[:8], nil
+	case *uint64:
+		return codec.Encode(*value)
+	case float32:
+		codec.order.PutUint32(codec.scratch[:], math.Float32bits(value))
+		return codec.scratch[:4], nil
+	case *float32:
+		codec.order.PutUint32(codec.scratch[:], math.Float32bits(*value))
+		return codec.scratch[:4], nil
+	case float64:
+		codec.order.PutUint64(codec.scratch[:], math.Float64bits(value))
+		return codec.scratch[:8], nil
+	case *float64:
+		codec.order.PutUint64(codec.scratch[:], math.Float64bits(*value))
+		return codec.scratch[:8], nil
+	case complex64:
+		codec.order.PutUint32(codec.scratch[0:], math.Float32bits(real(value)))
+		codec.order.PutUint32(codec.scratch[4:], math.Float32bits(imag(value)))
+		return codec.scratch[:8], nil
+	case *complex64:
+		codec.order.PutUint32(codec.scratch[0:], math.Float32bits(real(*value)))
+		codec.order.PutUint32(codec.scratch[4:], math.Float32bits(imag(*value)))
+		return codec.scratch[:8], nil
+	case complex128:
+		codec.order.PutUint64(codec.scratch[0:], math.Float64bits(real(value)))
+		codec.order.PutUint64(codec.scratch[4:], math.Float64bits(imag(value)))
+		return codec.scratch[:16], nil
+	case *complex128:
+		codec.order.PutUint64(codec.scratch[0:], math.Float64bits(real(*value)))
+		codec.order.PutUint64(codec.scratch[4:], math.Float64bits(imag(*value)))
+		return codec.scratch[:16], nil
+	case string:
+		n := binary.PutUvarint(codec.scratch[:], uint64(len(value)))
+		return append(codec.scratch[:n], []byte(value)...), nil
+	case *string:
+		n := binary.PutUvarint(codec.scratch[:], uint64(len(*value)))
+		return append(codec.scratch[:n], []byte(*value)...), nil
+	case []byte:
+		n := binary.PutUvarint(codec.scratch[:], uint64(len(value)))
+		return append(codec.scratch[:n], value...), nil
+	case *[]byte:
+		n := binary.PutUvarint(codec.scratch[:], uint64(len(*value)))
+		return append(codec.scratch[:n], *value...), nil
+	default:
+		return nil, fmt.Errorf("codec: don't know how to binary encode %T", value)
+	}
+}
+
+func (codec binaryCodec) Decode(data []byte, value any) error {
+	switch value := value.(type) {
+	case *bool:
+		if len(data) < 1 {
+			return io.ErrUnexpectedEOF
+		}
+		*value = data[0] != 0
+
+	case *int:
+		if len(data) < 8 {
+			return io.ErrUnexpectedEOF
+		}
+		*value = int(codec.order.Uint64(data))
+	case *int8:
+		if len(data) < 1 {
+			return io.ErrUnexpectedEOF
+		}
+		*value = int8(data[0])
+	case *int16:
+		if len(data) < 2 {
+			return io.ErrUnexpectedEOF
+		}
+		*value = int16(codec.order.Uint16(data))
+	case *int32:
+		if len(data) < 4 {
+			return io.ErrUnexpectedEOF
+		}
+		*value = int32(codec.order.Uint32(data))
+	case *int64:
+		if len(data) < 8 {
+			return io.ErrUnexpectedEOF
+		}
+		*value = int64(codec.order.Uint64(data))
+	case *uint:
+		if len(data) < 8 {
+			return io.ErrUnexpectedEOF
+		}
+		*value = uint(codec.order.Uint64(data))
+	case *uint8:
+		if len(data) < 1 {
+			return io.ErrUnexpectedEOF
+		}
+		*value = data[0]
+	case *uint16:
+		if len(data) < 1 {
+			return io.ErrUnexpectedEOF
+		}
+		*value = codec.order.Uint16(data)
+	case *uint32:
+		if len(data) < 4 {
+			return io.ErrUnexpectedEOF
+		}
+		*value = codec.order.Uint32(data)
+	case *uint64:
+		if len(data) < 8 {
+			return io.ErrUnexpectedEOF
+		}
+		*value = codec.order.Uint64(data)
+	case *float32:
+		if len(data) < 4 {
+			return io.ErrUnexpectedEOF
+		}
+		*value = math.Float32frombits(codec.order.Uint32(data))
+	case *float64:
+		if len(data) < 8 {
+			return io.ErrUnexpectedEOF
+		}
+		*value = math.Float64frombits(codec.order.Uint64(data))
+	case *complex64:
+		if len(data) < 8 {
+			return io.ErrUnexpectedEOF
+		}
+		*value = complex(
+			math.Float32frombits(codec.order.Uint32(data[0:])),
+			math.Float32frombits(codec.order.Uint32(data[4:])),
+		)
+	case *complex128:
+		if len(data) < 16 {
+			return io.ErrUnexpectedEOF
+		}
+		*value = complex(
+			math.Float64frombits(codec.order.Uint64(data[0:])),
+			math.Float64frombits(codec.order.Uint64(data[8:])),
+		)
+	case *string:
+		r := bytes.NewReader(data)
+		n, err := binary.ReadUvarint(r)
+		if err != nil {
+			return err
+		}
+		if uint64(r.Len()) < n {
+			return io.ErrUnexpectedEOF
+		}
+		*value = string(data[len(data)-r.Len():])
+
+	case *[]byte:
+		r := bytes.NewReader(data)
+		n, err := binary.ReadUvarint(r)
+		if err != nil {
+			return err
+		}
+		if uint64(r.Len()) < n {
+			return io.ErrUnexpectedEOF
+		}
+		copy(*value, data[len(data)-r.Len():])
+
+	default:
+		return fmt.Errorf("codec: don't know how to binary decode %T", value)
+	}
+	return nil
+}
+
+type sshCodec struct{}
+
+func (sshCodec) Type() string                        { return "ssh" }
+func (sshCodec) Encode(value any) ([]byte, error)    { return ssh.Marshal(value), nil }
+func (sshCodec) Decode(data []byte, value any) error { return ssh.Unmarshal(data, value) }

storage/codec/codec.go

@@ -0,0 +1,66 @@
+package codec
+
+import (
+	"fmt"
+	"io"
+)
+
+type Codec interface {
+	Type() string
+	Encode(any) ([]byte, error)
+	Decode([]byte, any) error
+}
+
+type Stream interface {
+	Encode(any) error
+	Decode(any) error
+}
+
+var (
+	codecs  = make(map[string]func() Codec)
+	streams = make(map[string]func(io.Reader, io.Writer) Stream)
+)
+
+func Register(name string, create func() Codec) {
+	if _, dupe := codecs[name]; dupe {
+		panic(fmt.Sprintf("codec: duplicate codec %q registered", name))
+	}
+	codecs[name] = create
+}
+
+func New(name string) (Codec, error) {
+	if f, ok := codecs[name]; ok {
+		return f(), nil
+	}
+	return nil, fmt.Errorf("codec: no %q codec registered", name)
+}
+
+func Must(name string) Codec {
+	c, err := New(name)
+	if err != nil {
+		panic(err)
+	}
+	return c
+}
+
+func RegisterStream(name string, create func(io.Reader, io.Writer) Stream) {
+	if _, dupe := streams[name]; dupe {
+		panic(fmt.Sprintf("codec: duplicate stream %q registered", name))
+	}
+	streams[name] = create
+}
+
+func NewStream(name string, r io.Reader, w io.Writer) (Stream, error) {
+	if f, ok := streams[name]; ok {
+		return f(r, w), nil
+	}
+	return nil, fmt.Errorf("codec: no %q stream registered", name)
+}
+
+func MustStream(name string, r io.Reader, w io.Writer) Stream {
+	s, err := NewStream(name, r, w)
+	if err != nil {
+		panic(err)
+	}
+	return s
+}

storage/codec/default.go

@@ -0,0 +1,77 @@
+package codec
+
+import (
+	"encoding/json"
+	"io"
+
+	"github.com/goccy/go-yaml"
+)
+
+func init() {
+	Register("json", func() Codec { return jsonCodec{} })
+	Register("yaml", func() Codec { return yamlCodec{} })
+}
+
+type jsonCodec struct{}
+
+func (jsonCodec) Type() string                        { return "json" }
+func (jsonCodec) Encode(value any) ([]byte, error)    { return json.Marshal(value) }
+func (jsonCodec) Decode(data []byte, value any) error { return json.Unmarshal(data, value) }
+
+type jsonStream struct {
+	decoder *json.Decoder
+	encoder *json.Encoder
+}
+
+type JSONOption func(*json.Encoder)
+
+func NewJSONStream(r io.Reader, w io.Writer, options ...JSONOption) Stream {
+	s := &jsonStream{
+		decoder: json.NewDecoder(r),
+		encoder: json.NewEncoder(w),
+	}
+	for _, option := range options {
+		option(s.encoder)
+	}
+	return s
+}
+
+func JSONIndent(prefix, indent string) JSONOption {
+	return func(e *json.Encoder) {
+		e.SetIndent(prefix, indent)
+	}
+}
+
+func JSONEscapeHTML(on bool) JSONOption {
+	return func(e *json.Encoder) {
+		e.SetEscapeHTML(on)
+	}
+}
+
+func (s jsonStream) Encode(value any) error { return s.encoder.Encode(value) }
+func (s jsonStream) Decode(value any) error { return s.decoder.Decode(value) }
+
+type yamlCodec struct{}
+
+func (yamlCodec) Type() string                        { return "yaml" }
+func (yamlCodec) Encode(value any) ([]byte, error)    { return yaml.Marshal(value) }
+func (yamlCodec) Decode(data []byte, value any) error { return yaml.Unmarshal(data, value) }
+
+type yamlStream struct {
+	decoder *yaml.Decoder
+	encoder *yaml.Encoder
+}
+
+func NewYaMLStream(r io.Reader, w io.Writer, options ...yaml.EncodeOption) Stream {
+	s := &yamlStream{
+		decoder: yaml.NewDecoder(r),
+		encoder: yaml.NewEncoder(w),
+	}
+	for _, option := range options {
+		option(s.encoder)
+	}
+	return s
+}
+
+func (s yamlStream) Encode(value any) error { return s.encoder.Encode(value) }
+func (s yamlStream) Decode(value any) error { return s.decoder.Decode(value) }

storage/io.go

@@ -0,0 +1,225 @@
+package storage
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"syscall"
+)
+
+type ValueError struct {
+	Value any
+}
+
+func (err ValueError) Error() string {
+	return fmt.Sprintf("kv: can't store value of type %T", err.Value)
+}
+
+type readSeekerKV struct {
+	mu  sync.Mutex
+	r   io.ReadSeeker
+	c   io.Closer
+	sep rune
+}
+
+func (kv *readSeekerKV) scan(key string) (data string, ok bool, err error) {
+	kv.mu.Lock()
+
+	if _, err = kv.r.Seek(0, io.SeekStart); err != nil {
+		return
+	}
+
+	s := bufio.NewScanner(kv.r)
+	for s.Scan() {
+		line := s.Text()
+		if i := strings.IndexRune(line, kv.sep); i > -1 && line[:i] == key {
+			kv.mu.Unlock()
+			return line[i+1:], true, nil
+		}
+	}
+	err = s.Err()
+	kv.mu.Unlock()
+	return
+}
+
+func (kv *readSeekerKV) Has(key string) bool {
+	_, ok, err := kv.scan(key)
+	return ok && err == nil
+}
+
+func (kv *readSeekerKV) Get(key string) (value any, ok bool) {
+	value, ok, _ = kv.scan(key)
+	return
+}
+
+func (kv *readSeekerKV) Close() error {
+	if kv.c == nil {
+		return nil
+	}
+	return kv.c.Close()
+}
+
+func OpenKV(name string, sep rune) (KV, error) {
+	f, err := os.Open(name)
+	if err != nil {
+		return nil, err
+	}
+	return &readSeekerKV{
+		r:   f,
+		sep: sep,
+	}, nil
+}
+
+type readSeekWriterKV struct {
+	readSeekerKV
+	w io.WriteSeeker
+}
+
+type truncater interface {
+	Truncate(int64) error
+}
+
+func (kv *readSeekWriterKV) Set(key string, value any) error {
+	var line string
+	switch value := value.(type) {
+	case string:
+		line = key + string(kv.sep) + value + "\n"
+	case []string:
+		line = key + string(kv.sep) + strings.Join(value, string(kv.sep)) + "\n"
+	default:
+		return ValueError{Value: value}
+	}
+
+	kv.mu.Lock()
+	defer kv.mu.Unlock()
+
+	if _, err := kv.r.Seek(0, io.SeekStart); err != nil {
+		return err
+	}
+
+	var (
+		lines   []string
+		found   bool
+		scanner = bufio.NewScanner(kv.r)
+	)
+	for scanner.Scan() {
+		text := scanner.Text()
+		if i := strings.IndexRune(text, kv.sep); i > -1 {
+			if found = text[:i] == key; found {
+				lines = append(lines, line)
+			} else {
+				lines = append(lines, text)
+			}
+		} else {
+			lines = append(lines, line)
+		}
+	}
+	if !found {
+		lines = append(lines, line)
+	}
+
+	// Writing strategy: if it's a file, write to a new file and move it over.
+	if f, ok := kv.w.(*os.File); ok {
+		return kv.replaceFile(f, lines)
+	}
+
+	if t, ok := kv.w.(truncater); ok {
+		if err := t.Truncate(0); err != nil {
+			return err
+		}
+	}
+
+	if _, err := kv.w.Seek(0, io.SeekStart); err != nil {
+		return err
+	}
+
+	for _, line := range lines {
+		if _, err := io.WriteString(kv.w, line+"\n"); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (kv *readSeekWriterKV) replaceFile(f *os.File, lines []string) (err error) {
+	var (
+		prev = f.Name()
+		info os.FileInfo
+	)
+
+	if info, err = f.Stat(); err != nil {
+		return
+	}
+
+	var (
+		name = "." + filepath.Base(f.Name()) + ".*"
+		n    *os.File
+	)
+	if n, err = os.CreateTemp(filepath.Dir(f.Name()), name); err != nil {
+		return
+	}
+	name = n.Name()
+
+	// Replicate original file mode
+	if err = os.Chmod(name, info.Mode()); err != nil {
+		_ = n.Close()
+		_ = os.Remove(name)
+		return
+	}
+
+	// Replicate original file ownership
+	if stat, ok := info.Sys().(*syscall.Stat_t); ok {
+		// This may fail if we aren't allowed, ignore it.
+		_ = os.Chown(name, int(stat.Uid), int(stat.Gid))
+	}
+
+	// Write lines to tempfile.
+	for _, line := range lines {
+		if _, err = io.WriteString(n, line+"\n"); err != nil {
+			_ = n.Close()
+			_ = os.Remove(name)
+			return
+		}
+	}
+
+	if err = n.Close(); err != nil {
+		_ = os.Remove(name)
+		return
+	}
+
+	// Close original file and replace it.
+	_ = f.Close()
+
+	if err = os.Rename(name, prev); err != nil {
+		return
+	}
+
+	// Reopen file and replace our readers/writers/closers
+	if f, err = os.OpenFile(prev, os.O_APPEND|os.O_RDWR, info.Mode()|os.ModePerm); err != nil {
+		return
+	}
+	kv.r = f
+	kv.w = f
+	kv.c = f
+	return
+}
+
+func OpenWritableKV(name string, sep rune, perm os.FileMode) (WritableKV, error) {
+	f, err := os.OpenFile(name, os.O_CREATE|os.O_RDWR, perm)
+	if err != nil {
+		return nil, err
+	}
+	return &readSeekWriterKV{
+		readSeekerKV: readSeekerKV{
+			r:   f,
+			c:   f,
+			sep: sep,
+		},
+		w: f,
+	}, nil
+}

storage/io_test.go

@@ -0,0 +1,124 @@
+package storage
+
+import (
+	"errors"
+	"os"
+	"path/filepath"
+	"reflect"
+	"testing"
+)
+
+func TestKV(t *testing.T) {
+	kv, err := OpenKV(filepath.Join("testdata", "kv"), ':')
+	if err != nil {
+		t.Skip(err)
+		return
+	}
+
+	tests := []struct {
+		Key   string
+		Value any
+		OK    bool
+	}{
+		{"test", "data", true},
+		{"empty", "", true},
+		{"", "emptykey", true},
+		{"nonexistant", nil, false},
+		{"ignored line because not relevant", nil, false},
+	}
+	for _, test := range tests {
+		t.Run(test.Key, func(it *testing.T) {
+			value, ok := kv.Get(test.Key)
+			if ok != test.OK {
+				t.Errorf("expected ok %t, got %t", test.OK, ok)
+			}
+			if ok && !reflect.DeepEqual(value, test.Value) {
+				t.Errorf("expected value %v, got %v", test.Value, value)
+			}
+		})
+	}
+}
+
+func TestWritableKV(t *testing.T) {
+	orgfile := filepath.Join("testdata", "kv")
+	b, err := os.ReadFile(orgfile)
+	if err != nil {
+		t.Skip(err)
+	}
+
+	w, err := os.CreateTemp(os.TempDir(), "kv.*")
+	if err != nil {
+		t.Skip(err)
+		return
+	}
+	testfile := w.Name()
+	t.Logf("using a copy of %s as %s", orgfile, testfile)
+	defer os.Remove(testfile)
+
+	if _, err = w.Write(b); err != nil {
+		return
+	}
+	if err = w.Close(); err != nil {
+		return
+	}
+
+	kv, err := OpenWritableKV(testfile, ':', 0640)
+	if err != nil {
+		t.Skip(err)
+		return
+	}
+	defer kv.Close()
+
+	t.Run("write", func(t *testing.T) {
+		writeTests := []struct {
+			Key   string
+			Value any
+			Error error
+		}{
+			{"test", "newdata", nil},
+			{"strings", []string{"test", "data"}, nil},
+			{"int", 42, ValueError{Value: 42}},
+		}
+		for _, test := range writeTests {
+			t.Run(test.Key, func(it *testing.T) {
+				err := kv.Set(test.Key, test.Value)
+				if err != nil {
+					if test.Error == nil {
+						it.Errorf("unepxected error %q (%T)", err, err)
+					} else if !errors.Is(err, test.Error) {
+						it.Errorf("expected error %q, but got %q (%T)", test.Error, err, err)
+					}
+					return
+				} else if err == nil && test.Error != nil {
+					it.Errorf("expected error %q, but got nil", err)
+				}
+			})
+		}
+	})
+
+	t.Run("read", func(t *testing.T) {
+		readTests := []struct {
+			Key   string
+			Value any
+			OK    bool
+		}{
+			{"test", "newdata", true},
+			{"empty", "", true},
+			{"", "emptykey", true},
+			{"strings", []string{"test", "data"}, true},
+			{"nonexistant", nil, false},
+			{"ignored line because not relevant", nil, false},
+		}
+		for _, test := range readTests {
+			t.Run(test.Key, func(it *testing.T) {
+				value, ok := kv.Get(test.Key)
+				if ok != test.OK {
+					t.Errorf("expected ok %t, got %t", test.OK, ok)
+				}
+				if ok && !reflect.DeepEqual(value, test.Value) {
+					t.Errorf("expected value %v, got %v", test.Value, value)
+				}
+			})
+		}
+	})
+}

storage/kv.go

@@ -0,0 +1,61 @@
+package storage
+
+import (
+	"encoding/json"
+
+	"git.maze.io/maze/conduit/storage/codec"
+)
+
+// KV implements a key-value store.
+type KV interface {
+	// Has checks if the key is present.
+	Has(key string) bool
+
+	// Get a value by key.
+	Get(key string) (value []byte, ok bool)
+}
+
+// WritableKV can store key-value items.
+type WritableKV interface {
+	KV
+
+	// Close the key-value storage.
+	Close() error
+
+	// DeleteKey removes a key.
+	Delete(key string) error
+
+	// Set a value by key.
+	Set(key string, value []byte) error
+}
+
+type EncodedKV interface {
+	Has(key string) bool
+	Get(key string, value any) (ok bool, err error)
+}
+
+type EncodedWritableKV interface {
+	EncodedKV
+
+	Close() error
+
+	Delete(key string) error
+
+	Set(key string, value []byte) error
+}
+
+type encodedKV struct {
+	kv     KV
+	encode func(any) ([]byte, error)
+	decode func([]byte, any) error
+}
+
+func (kv encodedKV) Has(key)
+
+func NewEncodedKV(kv KV, codec codec.Codec) EncodedKV {
+	return &encodedKV{
+		kv:     kv,
+		encode: json.Marshal,
+		decode: json.Unmarshal,
+	}
+}

storage/testdata/kv

@@ -0,0 +1,4 @@
+test:data
+ignored line because not relevant
+empty:
+:emptykey