First Commit

internal/index/index.go

@@ -0,0 +1,123 @@
+package index
+
+import (
+	"bufio"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"sort"
+	"sync"
+	"time"
+)
+
+type opType string
+
+const (
+	OpPut opType = "put"
+	OpDel opType = "del"
+)
+
+type record struct {
+	Op       opType    `json:"op"`
+	Hash     string    `json:"hash"`
+	Bytes    int64     `json:"bytes,omitempty"`
+	StoredAt time.Time `json:"stored_at,omitempty"`
+	Private  bool      `json:"private,omitempty"`
+}
+
+type Entry struct {
+	Hash     string    `json:"hash"`
+	Bytes    int64     `json:"bytes"`
+	StoredAt time.Time `json:"stored_at"`
+	Private  bool      `json:"private"`
+}
+
+type Index struct {
+	path string
+	mu   sync.Mutex
+}
+
+func New(baseDir string) *Index {
+	return &Index{path: filepath.Join(baseDir, "index.jsonl")}
+}
+
+func (i *Index) AppendPut(e Entry) error {
+	i.mu.Lock()
+	defer i.mu.Unlock()
+	return appendRec(i.path, record{
+		Op:       OpPut,
+		Hash:     e.Hash,
+		Bytes:    e.Bytes,
+		StoredAt: e.StoredAt,
+		Private:  e.Private,
+	})
+}
+
+func (i *Index) AppendDelete(hash string) error {
+	i.mu.Lock()
+	defer i.mu.Unlock()
+	return appendRec(i.path, record{Op: OpDel, Hash: hash})
+}
+
+func appendRec(path string, r record) error {
+	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+		return err
+	}
+	f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	enc := json.NewEncoder(f)
+	return enc.Encode(r)
+}
+
+func (i *Index) Snapshot() ([]Entry, error) {
+	i.mu.Lock()
+	defer i.mu.Unlock()
+
+	f, err := os.Open(i.path)
+	if os.IsNotExist(err) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	defer f.Close()
+
+	sc := bufio.NewScanner(f)
+	sc.Buffer(make([]byte, 0, 64*1024), 4*1024*1024)
+
+	type state struct {
+		Entry   Entry
+		Deleted bool
+	}
+	m := make(map[string]state)
+	for sc.Scan() {
+		var rec record
+		if err := json.Unmarshal(sc.Bytes(), &rec); err != nil {
+			continue
+		}
+		switch rec.Op {
+		case OpPut:
+			m[rec.Hash] = state{Entry: Entry{
+				Hash: rec.Hash, Bytes: rec.Bytes, StoredAt: rec.StoredAt, Private: rec.Private,
+			}}
+		case OpDel:
+			s := m[rec.Hash]
+			s.Deleted = true
+			m[rec.Hash] = s
+		}
+	}
+	if err := sc.Err(); err != nil {
+		return nil, err
+	}
+	var out []Entry
+	for _, s := range m {
+		if !s.Deleted && s.Entry.Hash != "" {
+			out = append(out, s.Entry)
+		}
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].StoredAt.After(out[j].StoredAt) })
+	return out, nil
+}