622 lines
16 KiB
Go
622 lines
16 KiB
Go
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)
|
|
}
|