fathom/internal/store/blob.go

package store

import (
	"fmt"
	"os"
	"sync"
)

// BlobFile is a shard's append-only sidecar for large text.
//
// Prompts, tool payloads, and model responses are the bulk of a trace by bytes
// and are needed only when someone opens that one trace. Keeping them out of
// the columnar chunks means a scan over a day of spans touches kilobytes of
// timestamps and dictionary codes instead of megabytes of prose. The columnar
// side stores only an offset and a length.
type BlobFile struct {
	path     string
	mu       sync.Mutex
	f        *os.File
	size     int64
	readOnly bool
}

// OpenBlobFileReadOnly opens a blob sidecar for reading only.
//
// A missing sidecar is not an error here: a shard whose rows all had empty
// input and output never creates one, and a reader must not fail on that.
func OpenBlobFileReadOnly(path string) (*BlobFile, error) {
	f, err := os.Open(path)
	if os.IsNotExist(err) {
		return &BlobFile{path: path, readOnly: true}, nil
	}
	if err != nil {
		return nil, err
	}
	st, err := f.Stat()
	if err != nil {
		f.Close()
		return nil, err
	}
	return &BlobFile{path: path, f: f, size: st.Size(), readOnly: true}, nil
}

func OpenBlobFile(path string) (*BlobFile, error) {
	f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0o644)
	if err != nil {
		return nil, err
	}
	st, err := f.Stat()
	if err != nil {
		f.Close()
		return nil, err
	}
	return &BlobFile{path: path, f: f, size: st.Size()}, nil
}

// Append writes s and returns a reference to it. An empty string gets the zero
// reference and costs no bytes, which matters because most spans carry neither
// input nor output.
func (b *BlobFile) Append(s string) (BlobRef, error) {
	if s == "" {
		return BlobRef{}, nil
	}
	b.mu.Lock()
	defer b.mu.Unlock()

	if b.readOnly {
		return BlobRef{}, fmt.Errorf("blob file %s is open read-only", b.path)
	}

	off := b.size
	n, err := b.f.WriteString(s)
	if err != nil {
		// A partial write leaves the file longer than our accounting. Trust the
		// file: a later Append must not hand out an offset that overlaps these
		// bytes, even though this blob is now unreferenced garbage.
		b.size += int64(n)
		return BlobRef{}, err
	}
	b.size += int64(n)
	return BlobRef{Off: off, Len: uint32(n)}, nil
}

// Read resolves a reference. Callers get an error rather than a panic on a
// reference that points past the file, since a damaged chunk can produce one.
func (b *BlobFile) Read(ref BlobRef) (string, error) {
	if ref.Len == 0 {
		return "", nil
	}
	b.mu.Lock()
	size := b.size
	b.mu.Unlock()

	if b.f == nil {
		return "", fmt.Errorf("blob file %s was never created", b.path)
	}
	if ref.Off < 0 || ref.Off+int64(ref.Len) > size {
		return "", fmt.Errorf("blob ref %d+%d runs past the %d-byte file %s", ref.Off, ref.Len, size, b.path)
	}
	buf := make([]byte, ref.Len)
	if _, err := b.f.ReadAt(buf, ref.Off); err != nil {
		return "", err
	}
	return string(buf), nil
}

// Sync flushes blob bytes. The shard calls this before syncing its own chunk,
// so a chunk on disk never references blob bytes that are not.
func (b *BlobFile) Sync() error {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.f == nil || b.readOnly {
		return nil
	}
	return b.f.Sync()
}

func (b *BlobFile) Size() int64 {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.size
}

func (b *BlobFile) Close() error {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.f == nil {
		return nil
	}
	return b.f.Close()
}