golangLAKEHOUSE/internal/pathway/persistor.go

// persistor.go — JSONL append-only persistence for pathway memory.
//
// Each event is one JSON line. Append is O(1) (open append, write,
// close — Go's *os.File default fsync policy is "rely on OS" which
// is fine here; correctness on power-loss is best-effort, not
// transactional). Replay reads the file once at startup.
//
// Corruption recovery: malformed lines log a warn (counted in
// Replay's return) but do not stop the load. Partial state is
// better than no state for an agent substrate.
//
// What's NOT here:
//   - Compaction. JSONL grows linearly with mutations; below 100K
//     traces this is fine. Compaction will land when needed and
//     will emit a snapshot file + tail JSONL.
//   - fsync per write. We rely on the OS's eventual fsync; trace
//     loss on hard crash is acceptable for the substrate's
//     "remember most things" guarantee.

package pathway

import (
	"bufio"
	"encoding/json"
	"errors"
	"fmt"
	"io/fs"
	"log/slog"
	"os"
	"path/filepath"
)

// Persistor wraps a single JSONL file. Construct with NewPersistor;
// it does NOT load on construction — callers must call Store.Load()
// to replay.
type Persistor struct {
	path string
}

// NewPersistor returns a persistor for the given file path. The
// parent directory is created on demand. The file is created lazily
// on first Append.
func NewPersistor(path string) (*Persistor, error) {
	if path == "" {
		return nil, errors.New("pathway: persistor path is empty")
	}
	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
		return nil, fmt.Errorf("pathway: create dir: %w", err)
	}
	return &Persistor{path: path}, nil
}

// Path returns the underlying file path. Useful for tests + logs.
func (p *Persistor) Path() string { return p.path }

// Append writes one event to the JSONL log. Each call opens the
// file in append mode, writes one line, and closes — simple but
// correct. A pooled persistent fd is a future optimization if
// profiling shows append-rate matters.
func (p *Persistor) Append(e event) error {
	line, err := json.Marshal(e)
	if err != nil {
		return fmt.Errorf("pathway: marshal event: %w", err)
	}
	f, err := os.OpenFile(p.path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
	if err != nil {
		return fmt.Errorf("pathway: open log: %w", err)
	}
	defer f.Close()
	if _, err := f.Write(line); err != nil {
		return fmt.Errorf("pathway: write event: %w", err)
	}
	if _, err := f.Write([]byte{'\n'}); err != nil {
		return fmt.Errorf("pathway: write newline: %w", err)
	}
	return nil
}

// Replay reads the log line-by-line and invokes apply for each
// event. Returns the count of events successfully applied. A
// missing file is NOT an error (means "no prior state"); a
// partially-corrupt file logs warns and continues.
func (p *Persistor) Replay(apply func(event) error) (int, error) {
	f, err := os.Open(p.path)
	if errors.Is(err, fs.ErrNotExist) {
		return 0, nil
	}
	if err != nil {
		return 0, fmt.Errorf("pathway: open log: %w", err)
	}
	defer f.Close()

	scanner := bufio.NewScanner(f)
	// Big buffer for unusually long content — 1 MiB per line cap.
	buf := make([]byte, 0, 64*1024)
	scanner.Buffer(buf, 1<<20)

	applied := 0
	skipped := 0
	lineNo := 0
	for scanner.Scan() {
		lineNo++
		raw := scanner.Bytes()
		if len(raw) == 0 {
			continue
		}
		var e event
		if err := json.Unmarshal(raw, &e); err != nil {
			slog.Warn("pathway: replay skipped malformed line",
				"path", p.path, "line", lineNo, "err", err.Error())
			skipped++
			continue
		}
		if err := apply(e); err != nil {
			slog.Warn("pathway: replay event apply failed",
				"path", p.path, "line", lineNo, "op", e.Op, "err", err.Error())
			skipped++
			continue
		}
		applied++
	}
	if err := scanner.Err(); err != nil {
		return applied, fmt.Errorf("pathway: scan log: %w", err)
	}
	if skipped > 0 {
		slog.Info("pathway: replay completed with skips",
			"path", p.path, "applied", applied, "skipped", skipped)
	}
	return applied, nil
}