go-synapse-backupper/pkg/adapters/storage/local/local.go

package local

import (
	"errors"
	"os"
	"path/filepath"
	"strings"
	"sync"

	"git.tswf.io/infra/go-synapse-backupper/pkg/domain"
)

// NewLocalSink creates a new local filesystem sink that writes into the given directory.
func NewLocalSink(dir string) domain.Sink {
	return &localSink{dir: dir}
}

type localSink struct {
	dir string
}

func (l *localSink) Begin(key string) (domain.SinkTx, error) {
	tmpPath := filepath.Join(l.dir, key+".tmp")
	finalPath := filepath.Join(l.dir, key)

	file, err := os.Create(tmpPath)
	if err != nil {
		return nil, err
	}

	return &localSinkTx{
		file:      file,
		tmpPath:   tmpPath,
		finalPath: finalPath,
	}, nil
}

func (l *localSink) List(prefix string) ([]string, error) {
	entries, err := os.ReadDir(l.dir)
	if err != nil {
		return nil, err
	}

	keys := make([]string, 0, len(entries))
	for _, entry := range entries {
		if entry.IsDir() {
			continue
		}
		name := entry.Name()
		if prefix != "" && !strings.HasPrefix(name, prefix) {
			continue
		}
		keys = append(keys, name)
	}

	return keys, nil
}

func (l *localSink) Remove(key string) error {
	path := filepath.Join(l.dir, key)
	return os.Remove(path)
}

type localSinkTx struct {
	file      *os.File
	tmpPath   string
	finalPath string
	committed bool
	aborted   bool
	mu        sync.Mutex
}

func (t *localSinkTx) Write(p []byte) (int, error) {
	t.mu.Lock()
	defer t.mu.Unlock()

	if t.committed || t.aborted {
		return 0, errors.New("transaction already finished")
	}

	return t.file.Write(p)
}

func (t *localSinkTx) Commit() error {
	t.mu.Lock()
	defer t.mu.Unlock()

	if t.committed || t.aborted {
		return nil
	}

	t.committed = true

	if err := t.file.Sync(); err != nil {
		return err
	}
	if err := t.file.Close(); err != nil {
		return err
	}

	if err := os.Rename(t.tmpPath, t.finalPath); err != nil {
		return err
	}

	parent, err := os.Open(filepath.Dir(t.tmpPath))
	if err != nil {
		return err
	}
	defer parent.Close()

	if err := parent.Sync(); err != nil {
		return err
	}

	return nil
}

func (t *localSinkTx) Abort() error {
	t.mu.Lock()
	defer t.mu.Unlock()

	if t.committed || t.aborted {
		return nil
	}

	t.aborted = true

	_ = t.file.Close()
	_ = os.Remove(t.tmpPath)

	return nil
}