init

2026-09-10 12:30:42 +02:00
commit 9843fe0c50
79 changed files with 16318 additions and 0 deletions
@@ -0,0 +1,621 @@
+package journal
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"reflect"
+	"regexp"
+	"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 {
+	switch path {
+	case "accounts.finance", "categories.finance", "tags.finance", "merchants.finance":
+		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 []string{"accounts.finance", "categories.finance", "tags.finance", "merchants.finance"} {
+		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{}, 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.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)
+}