finance-duck/internal/journal/store.go

package journal

import (
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"os"
	"path/filepath"
	"reflect"
	"regexp"
	"slices"
	"sort"
	"strings"
	"sync"
	"syscall"

	"finance-duck/internal/domain"
)

var ErrConflict = errors.New("journal revision conflict")
var ErrClosed = errors.New("journal is closed")
var ErrImmutable = errors.New("existing transaction facts are immutable")
var monthlyPath = regexp.MustCompile(`^journal/([0-9]{4})/([0-9]{4})-(0[1-9]|1[0-2])\.finance$`)

const maxFileBytes = 64 << 20
const walName = ".commit"

type Store struct {
	mu     sync.Mutex
	dir    string
	lock   *os.File
	closed bool
}
type snapshot struct {
	data     domain.Dataset
	revision string
	docs     map[string]*document
	raw      map[string][]byte
}
type walEntry struct {
	Path   string `json:"path"`
	Before string `json:"before"`
	After  string `json:"after"`
	Stage  string `json:"stage"`
}
type manifest struct {
	Version  int        `json:"version"`
	Revision string     `json:"revision"`
	Files    []walEntry `json:"files"`
}

func Open(dir string) (*Store, error) {
	root, err := filepath.Abs(dir)
	if err != nil {
		return nil, err
	}
	if err = privateDir(root); err != nil {
		return nil, err
	}
	fd, err := syscall.Open(filepath.Join(root, ".lock"), syscall.O_RDWR|syscall.O_CREAT|syscall.O_NOFOLLOW|syscall.O_CLOEXEC|syscall.O_NONBLOCK, 0600)
	if err != nil {
		return nil, fmt.Errorf("journal lock: %w", err)
	}
	lock := os.NewFile(uintptr(fd), "journal lock")
	info, err := lock.Stat()
	if err != nil {
		lock.Close()
		return nil, err
	}
	if !info.Mode().IsRegular() {
		lock.Close()
		return nil, fmt.Errorf("journal lock must be a regular file")
	}
	if err = lock.Chmod(0600); err != nil {
		lock.Close()
		return nil, err
	}
	if err = syscall.Flock(fd, syscall.LOCK_EX|syscall.LOCK_NB); err != nil {
		lock.Close()
		return nil, fmt.Errorf("journal is already locked: %w", err)
	}
	s := &Store{dir: root, lock: lock}
	fail := func(err error) (*Store, error) { s.Close(); return nil, err }
	if err = s.recover(); err != nil {
		return fail(err)
	}
	snap, err := s.snapshot()
	if err != nil {
		return fail(err)
	}
	if len(snap.raw) == 0 {
		if _, err = s.Commit(snap.revision, domain.NewDataset()); err != nil {
			return fail(err)
		}
	}
	return s, nil
}
func (s *Store) Close() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.closed {
		return nil
	}
	s.closed = true
	err := syscall.Flock(int(s.lock.Fd()), syscall.LOCK_UN)
	closeErr := s.lock.Close()
	if err != nil {
		return err
	}
	return closeErr
}
func (s *Store) Load() (domain.Dataset, string, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.closed {
		return domain.Dataset{}, "", ErrClosed
	}
	if err := s.recover(); err != nil {
		return domain.Dataset{}, "", err
	}
	snap, err := s.snapshot()
	if err != nil {
		return domain.Dataset{}, "", err
	}
	return domain.Clone(snap.data), snap.revision, nil
}
func (s *Store) Commit(expectedRevision string, next domain.Dataset) (string, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.closed {
		return "", ErrClosed
	}
	if err := s.recover(); err != nil {
		return "", err
	}
	before, err := s.snapshot()
	if err != nil {
		return "", err
	}
	if before.revision != expectedRevision {
		return "", ErrConflict
	}
	if err = domain.Validate(next); err != nil {
		return "", err
	}
	existing := map[string]domain.Facts{}
	for _, t := range next.Transactions {
		existing[t.Facts.ID] = t.Facts
	}
	for _, t := range before.data.Transactions {
		f, ok := existing[t.Facts.ID]
		if !ok || !reflect.DeepEqual(f, t.Facts) {
			return "", fmt.Errorf("%w: %s", ErrImmutable, t.Facts.ID)
		}
	}
	output, err := renderFiles(next, before.docs)
	if err != nil {
		return "", err
	}
	for path, raw := range output {
		if len(raw) > maxFileBytes {
			return "", fmt.Errorf("%s: exceeds 64 MiB file limit", path)
		}
	}
	newRevision := revision(output)
	if newRevision == before.revision {
		return newRevision, nil
	}
	// Staging never changes canonical files. A synced manifest is the commit point:
	// once present, every reader/reopen finishes the entire validated generation.
	temp := filepath.Join(s.dir, ".prepare")
	if err = removePrivateTree(temp); err != nil {
		return "", err
	}
	if err = privateDir(temp); err != nil {
		return "", err
	}
	committed := false
	defer func() {
		if !committed {
			_ = removePrivateTree(temp)
		}
	}()
	m := manifest{Version: 1, Revision: before.revision, Files: []walEntry{}}
	paths := sortedPaths(output)
	for i, path := range paths {
		stage := fmt.Sprintf("%06d", i)
		if err = writeSynced(filepath.Join(temp, stage), output[path]); err != nil {
			return "", err
		}
		old := ""
		if raw, ok := before.raw[path]; ok {
			old = hash(raw)
		}
		m.Files = append(m.Files, walEntry{Path: path, Before: old, After: hash(output[path]), Stage: stage})
	}
	raw, err := json.Marshal(m)
	if err != nil {
		return "", err
	}
	if len(raw) > maxFileBytes {
		return "", fmt.Errorf("commit manifest exceeds 64 MiB file limit")
	}
	if err = writeSynced(filepath.Join(temp, "manifest.json"), raw); err != nil {
		return "", err
	}
	if err = syncDir(temp); err != nil {
		return "", err
	}
	// Detect edits made while the new generation was being prepared.
	current, err := s.readFiles()
	if err != nil {
		return "", err
	}
	if revision(current) != before.revision {
		return "", ErrConflict
	}
	if err = os.Rename(temp, filepath.Join(s.dir, walName)); err != nil {
		return "", err
	}
	committed = true
	if err = syncDir(s.dir); err != nil {
		return "", fmt.Errorf("commit pending recovery: %w", err)
	}
	if err = s.recover(); err != nil {
		return "", fmt.Errorf("commit pending recovery: %w", err)
	}
	return newRevision, nil
}
func hash(raw []byte) string { sum := sha256.Sum256(raw); return hex.EncodeToString(sum[:]) }
func sortedPaths[V any](m map[string]V) []string {
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	return keys
}
func revision(raw map[string][]byte) string {
	hashes := map[string]string{}
	for p, b := range raw {
		hashes[p] = hash(b)
	}
	return revisionHashes(hashes)
}
func revisionHashes(hashes map[string]string) string {
	h := sha256.New()
	for _, p := range sortedPaths(hashes) {
		io.WriteString(h, p)
		h.Write([]byte{0})
		io.WriteString(h, hashes[p])
		h.Write([]byte{0})
	}
	return hex.EncodeToString(h.Sum(nil))
}
func validPath(path string) bool {
	if slices.Contains(registryFiles, path) {
		return true
	}
	parts := monthlyPath.FindStringSubmatch(path)
	return len(parts) > 0 && parts[1] == parts[2] && parts[1] != "0000"
}
func privateDir(path string) error {
	for ancestor := filepath.Clean(path); ; ancestor = filepath.Dir(ancestor) {
		info, err := os.Lstat(ancestor)
		if err != nil && !errors.Is(err, os.ErrNotExist) {
			return err
		}
		if err == nil && (info.Mode()&os.ModeSymlink != 0 || !info.IsDir()) {
			return fmt.Errorf("%s: expected real directory, not symlink", ancestor)
		}
		if filepath.Dir(ancestor) == ancestor {
			break
		}
	}
	info, err := os.Lstat(path)
	if errors.Is(err, os.ErrNotExist) {
		if err = os.MkdirAll(path, 0700); err != nil {
			return err
		}
		info, err = os.Lstat(path)
	}
	if err != nil {
		return err
	}
	if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() {
		return fmt.Errorf("%s: expected real directory, not symlink", path)
	}
	return os.Chmod(path, 0700)
}
func syncDir(path string) error {
	f, err := os.Open(path)
	if err != nil {
		return err
	}
	defer f.Close()
	return f.Sync()
}
func readSecure(path string) ([]byte, error) {
	fd, err := syscall.Open(path, syscall.O_RDONLY|syscall.O_NOFOLLOW|syscall.O_CLOEXEC|syscall.O_NONBLOCK, 0)
	if err != nil {
		return nil, err
	}
	f := os.NewFile(uintptr(fd), path)
	defer f.Close()
	info, err := f.Stat()
	if err != nil {
		return nil, err
	}
	if !info.Mode().IsRegular() {
		return nil, fmt.Errorf("%s: not a regular file", path)
	}
	if info.Size() > maxFileBytes {
		return nil, fmt.Errorf("%s: exceeds 64 MiB file limit", path)
	}
	raw, err := io.ReadAll(io.LimitReader(f, maxFileBytes+1))
	if len(raw) > maxFileBytes {
		return nil, fmt.Errorf("%s: exceeds 64 MiB file limit", path)
	}
	return raw, err
}
func writeSynced(path string, raw []byte) error {
	fd, err := syscall.Open(path, syscall.O_WRONLY|syscall.O_CREAT|syscall.O_EXCL|syscall.O_NOFOLLOW|syscall.O_CLOEXEC, 0600)
	if err != nil {
		return err
	}
	f := os.NewFile(uintptr(fd), path)
	if _, err = f.Write(raw); err != nil {
		f.Close()
		return err
	}
	if err = f.Sync(); err != nil {
		f.Close()
		return err
	}
	return f.Close()
}
func removePrivateTree(path string) error {
	info, err := os.Lstat(path)
	if errors.Is(err, os.ErrNotExist) {
		return nil
	}
	if err != nil {
		return err
	}
	if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() {
		return fmt.Errorf("%s: unsafe staging directory", path)
	}
	return os.RemoveAll(path)
}
func (s *Store) readFiles() (map[string][]byte, error) {
	entries, err := os.ReadDir(s.dir)
	if err != nil {
		return nil, err
	}
	for _, entry := range entries {
		if strings.HasSuffix(entry.Name(), ".finance") && !validPath(entry.Name()) {
			return nil, fmt.Errorf("%s:1: unexpected registry filename", entry.Name())
		}
	}
	raw := map[string][]byte{}
	for _, path := range registryFiles {
		b, err := readSecure(filepath.Join(s.dir, path))
		if errors.Is(err, os.ErrNotExist) {
			continue
		}
		if err != nil {
			return nil, fmt.Errorf("%s:1: %w", path, err)
		}
		raw[path] = b
	}
	root := filepath.Join(s.dir, "journal")
	info, err := os.Lstat(root)
	if errors.Is(err, os.ErrNotExist) {
		return raw, nil
	}
	if err != nil {
		return nil, err
	}
	if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() {
		return nil, fmt.Errorf("journal:1: expected real directory")
	}
	err = filepath.WalkDir(root, func(path string, entry fs.DirEntry, walkErr error) error {
		if walkErr != nil {
			return walkErr
		}
		relative, err := filepath.Rel(s.dir, path)
		if err != nil {
			return err
		}
		relative = filepath.ToSlash(relative)
		if entry.Type()&os.ModeSymlink != 0 {
			return fmt.Errorf("%s:1: symbolic links are prohibited", relative)
		}
		if entry.IsDir() {
			return nil
		}
		if !strings.HasSuffix(relative, ".finance") {
			return nil
		}
		if !validPath(relative) {
			return fmt.Errorf("%s:1: expected journal/YYYY/YYYY-MM.finance", relative)
		}
		b, err := readSecure(path)
		if err != nil {
			return fmt.Errorf("%s:1: %w", relative, err)
		}
		raw[relative] = b
		return nil
	})
	return raw, err
}
func (s *Store) snapshot() (*snapshot, error) {
	raw, err := s.readFiles()
	if err != nil {
		return nil, err
	}
	snap, err := decodeSnapshot(raw)
	if err != nil {
		return nil, err
	}
	// Non-cooperating editors do not take our process lock. Do not publish a
	// mixed-generation read if files changed while parsing and validating.
	current, err := s.readFiles()
	if err != nil {
		return nil, err
	}
	if revision(current) != snap.revision {
		return nil, ErrConflict
	}
	return snap, nil
}
func decodeSnapshot(raw map[string][]byte) (*snapshot, error) {
	snap := &snapshot{raw: raw, docs: map[string]*document{}, revision: revision(raw), data: domain.Dataset{Accounts: []domain.Account{}, Categories: []domain.Category{}, Tags: []domain.Tag{}, Merchants: []domain.Merchant{}, Instruments: []domain.Instrument{}, Transactions: []domain.Transaction{}}}
	if len(raw) == 0 {
		snap.data = domain.NewDataset()
		return snap, nil
	}
	locations := map[string]string{}
	for _, path := range sortedPaths(raw) {
		doc, err := parseDocument(path, raw[path])
		if err != nil {
			return nil, err
		}
		snap.docs[path] = doc
		for _, p := range doc.pieces {
			if p.block == nil {
				continue
			}
			b := p.block
			location := fmt.Sprintf("%s:%d", path, b.line)
			if previous, ok := locations[b.id]; ok {
				return nil, fmt.Errorf("%s: duplicate ID %q (first at %s)", location, b.id, previous)
			}
			locations[b.id] = location
			expected := b.kind + "s.finance"
			if b.kind == "category" {
				expected = "categories.finance"
			}
			if b.kind == "transaction" {
				t := b.value.(domain.Transaction)
				if len(t.Facts.BookingDate) < 7 {
					return nil, fmt.Errorf("%s: invalid booking date", location)
				}
				month := t.Facts.BookingDate[:7]
				expected = "journal/" + month[:4] + "/" + month + ".finance"
			}
			if path != expected {
				return nil, fmt.Errorf("%s: %s block belongs in %s", location, b.kind, expected)
			}
			switch v := b.value.(type) {
			case domain.Account:
				snap.data.Accounts = append(snap.data.Accounts, v)
			case domain.Category:
				snap.data.Categories = append(snap.data.Categories, v)
			case domain.Tag:
				snap.data.Tags = append(snap.data.Tags, v)
			case domain.Merchant:
				snap.data.Merchants = append(snap.data.Merchants, v)
			case domain.Instrument:
				snap.data.Instruments = append(snap.data.Instruments, v)
			case domain.Transaction:
				snap.data.Transactions = append(snap.data.Transactions, v)
			}
		}
	}
	if err := domain.Validate(snap.data); err != nil {
		for _, id := range sortedPaths(locations) {
			if strings.Contains(err.Error(), fmt.Sprintf("%q", id)) {
				return nil, fmt.Errorf("%s: %w", locations[id], err)
			}
		}
		path := "categories.finance"
		if _, ok := raw[path]; !ok {
			path = sortedPaths(raw)[0]
		}
		return nil, fmt.Errorf("%s:1: %w", path, err)
	}
	return snap, nil
}
func (s *Store) recover() error {
	wal := filepath.Join(s.dir, walName)
	info, err := os.Lstat(wal)
	if errors.Is(err, os.ErrNotExist) {
		return nil
	}
	if err != nil {
		return err
	}
	if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() {
		return fmt.Errorf("unsafe recovery directory")
	}
	raw, err := readSecure(filepath.Join(wal, "manifest.json"))
	if err != nil {
		return err
	}
	var m manifest
	if err = decodeStrict(raw, &m); err != nil {
		return fmt.Errorf("recovery manifest: %w", err)
	}
	if m.Version != 1 || len(m.Files) == 0 {
		return fmt.Errorf("unsupported or empty recovery manifest")
	}
	staged := map[string][]byte{}
	seen := map[string]bool{}
	for i, e := range m.Files {
		if !validPath(e.Path) || e.Stage != fmt.Sprintf("%06d", i) || seen[e.Path] {
			return fmt.Errorf("unsafe recovery entry %q", e.Path)
		}
		seen[e.Path] = true
		b, err := readSecure(filepath.Join(wal, e.Stage))
		if err != nil {
			return err
		}
		if hash(b) != e.After {
			return fmt.Errorf("recovery checksum mismatch: %s", e.Path)
		}
		staged[e.Path] = b
	}
	if _, err = decodeSnapshot(staged); err != nil {
		return fmt.Errorf("invalid staged generation: %w", err)
	}
	current, err := s.readFiles()
	if err != nil {
		return err
	}
	baseline := map[string]string{}
	for p, b := range current {
		baseline[p] = hash(b)
	}
	for _, e := range m.Files {
		actual := baseline[e.Path]
		if actual != e.Before && actual != e.After {
			return fmt.Errorf("%w: %s edited during pending commit; staged data retained", ErrConflict, e.Path)
		}
		if e.Before == "" {
			delete(baseline, e.Path)
		} else {
			baseline[e.Path] = e.Before
		}
	}
	if revisionHashes(baseline) != m.Revision {
		return fmt.Errorf("%w: files added or removed during pending commit; staged data retained", ErrConflict)
	}
	for _, e := range m.Files {
		if b, ok := current[e.Path]; ok && hash(b) == e.After {
			continue
		}
		target := filepath.Join(s.dir, filepath.FromSlash(e.Path))
		parent := filepath.Dir(target)
		if parent != s.dir {
			if err = privateDir(filepath.Join(s.dir, "journal")); err != nil {
				return err
			}
			if err = privateDir(parent); err != nil {
				return err
			}
			if err = syncDir(filepath.Join(s.dir, "journal")); err != nil {
				return err
			}
			if err = syncDir(s.dir); err != nil {
				return err
			}
		}
		temporary := target + ".pending"
		if info, statErr := os.Lstat(temporary); statErr == nil {
			if !info.Mode().IsRegular() {
				return fmt.Errorf("unsafe pending file %s", temporary)
			}
			if err = os.Remove(temporary); err != nil {
				return err
			}
		} else if !errors.Is(statErr, os.ErrNotExist) {
			return statErr
		}
		if err = writeSynced(temporary, staged[e.Path]); err != nil {
			return err
		}
		if err = os.Rename(temporary, target); err != nil {
			return err
		}
		if err = syncDir(parent); err != nil {
			return err
		}
	}
	// Removing the manifest first would make a partially deleted WAL ambiguous.
	// Atomically retire the whole directory after every target and directory sync.
	retired := filepath.Join(s.dir, ".retired")
	if err = removePrivateTree(retired); err != nil {
		return err
	}
	if err = os.Rename(wal, retired); err != nil {
		return err
	}
	if err = syncDir(s.dir); err != nil {
		return err
	}
	return removePrivateTree(retired)
}