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{}, Assets: []domain.Asset{}, 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.Asset: snap.data.Assets = append(snap.data.Assets, 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) }