browser/platform/service/log_body_store.go

package service

import (
	"bufio"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/url"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync"

	"browser.local/platform/domain"
	"browser.local/platform/validator"
)

type LogBodyStore interface {
	AppendBatch(streamID string, record domain.LogBatchRecord) error
	GetBatch(streamID string, firstSeq uint64) (domain.LogBatchRecord, bool, error)
	Query(streamID string, afterSeq uint64, limit int) ([]domain.LogEntry, uint64, error)
}

type MemoryLogBodyStore struct {
	mu      sync.Mutex
	entries map[string][]domain.LogEntry
	batches map[string]map[uint64]domain.LogBatchRecord
}

func NewMemoryLogBodyStore() *MemoryLogBodyStore {
	return &MemoryLogBodyStore{
		entries: map[string][]domain.LogEntry{},
		batches: map[string]map[uint64]domain.LogBatchRecord{},
	}
}

func (store *MemoryLogBodyStore) AppendBatch(streamID string, record domain.LogBatchRecord) error {
	store.mu.Lock()
	defer store.mu.Unlock()

	records := store.batches[streamID]
	if records == nil {
		records = map[uint64]domain.LogBatchRecord{}
		store.batches[streamID] = records
	}
	if existing, exists := records[record.FirstSeq]; exists {
		if existing.LastSeq == record.LastSeq && existing.Checksum == record.Checksum {
			return nil
		}
		return validationError("log batch conflicts with acknowledged range")
	}
	records[record.FirstSeq] = domain.CopyLogBatchRecord(record)
	store.entries[streamID] = append(store.entries[streamID], domain.CopyLogEntries(record.Entries)...)
	return nil
}

func (store *MemoryLogBodyStore) GetBatch(streamID string, firstSeq uint64) (domain.LogBatchRecord, bool, error) {
	store.mu.Lock()
	defer store.mu.Unlock()

	record, exists := store.batches[streamID][firstSeq]
	if !exists {
		return domain.LogBatchRecord{}, false, nil
	}
	return domain.CopyLogBatchRecord(record), true, nil
}

func (store *MemoryLogBodyStore) Query(streamID string, afterSeq uint64, limit int) ([]domain.LogEntry, uint64, error) {
	store.mu.Lock()
	defer store.mu.Unlock()

	entries := domain.CopyLogEntries(store.entries[streamID])
	sort.SliceStable(entries, func(i, j int) bool { return entries[i].Seq < entries[j].Seq })
	selected := make([]domain.LogEntry, 0, limit)
	nextSeq := afterSeq
	for _, entry := range entries {
		if entry.Seq <= afterSeq {
			continue
		}
		if len(selected) >= limit {
			break
		}
		selected = append(selected, entry)
		nextSeq = entry.Seq
	}
	return selected, nextSeq, nil
}

func (store *MemoryLogBodyStore) LatestSeq(streamID string) (uint64, error) {
	store.mu.Lock()
	defer store.mu.Unlock()
	var latest uint64
	for _, entry := range store.entries[streamID] {
		if entry.Seq > latest {
			latest = entry.Seq
		}
	}
	return latest, nil
}

type FileLogBodyStore struct {
	mu      sync.Mutex
	rootDir string
	memory  *MemoryLogBodyStore
}

func NewFileLogBodyStore(rootDir string) (*FileLogBodyStore, error) {
	rootDir = strings.TrimSpace(rootDir)
	if rootDir == "" {
		return nil, fmt.Errorf("log directory is required")
	}
	store := &FileLogBodyStore{
		rootDir: rootDir,
		memory:  NewMemoryLogBodyStore(),
	}
	if err := os.MkdirAll(rootDir, 0o700); err != nil {
		return nil, fmt.Errorf("create log directory: %w", err)
	}
	if err := store.load(); err != nil {
		return nil, err
	}
	return store, nil
}

func (store *FileLogBodyStore) RootDir() string {
	return store.rootDir
}

func (store *FileLogBodyStore) AppendBatch(streamID string, record domain.LogBatchRecord) error {
	store.mu.Lock()
	defer store.mu.Unlock()

	if _, exists, err := store.memory.GetBatch(streamID, record.FirstSeq); err != nil {
		return err
	} else if exists {
		return store.memory.AppendBatch(streamID, record)
	}
	streamDir := store.streamDir(streamID)
	if err := os.MkdirAll(streamDir, 0o700); err != nil {
		return fmt.Errorf("create log stream directory: %w", err)
	}
	segmentPath := store.segmentPath(streamID, record.FirstSeq)
	if _, err := os.Stat(segmentPath); err == nil {
		return validationError("log batch conflicts with acknowledged range")
	} else if !os.IsNotExist(err) {
		return fmt.Errorf("stat log segment: %w", err)
	}
	var body strings.Builder
	encoder := json.NewEncoder(&body)
	for _, entry := range record.Entries {
		if err := encoder.Encode(domain.CopyLogEntry(entry)); err != nil {
			return fmt.Errorf("write log segment: %w", err)
		}
	}
	if err := writeAtomicFile(segmentPath, []byte(body.String()), 0o600); err != nil {
		return fmt.Errorf("persist log segment: %w", err)
	}
	return store.memory.AppendBatch(streamID, record)
}

func (store *FileLogBodyStore) GetBatch(streamID string, firstSeq uint64) (domain.LogBatchRecord, bool, error) {
	return store.memory.GetBatch(streamID, firstSeq)
}

func (store *FileLogBodyStore) Query(streamID string, afterSeq uint64, limit int) ([]domain.LogEntry, uint64, error) {
	return store.memory.Query(streamID, afterSeq, limit)
}

func (store *FileLogBodyStore) LatestSeq(streamID string) (uint64, error) {
	return store.memory.LatestSeq(streamID)
}

func (store *FileLogBodyStore) load() error {
	entries, err := os.ReadDir(store.rootDir)
	if err != nil {
		return fmt.Errorf("read log root: %w", err)
	}
	for _, entry := range entries {
		if !entry.IsDir() {
			continue
		}
		streamID, err := url.PathUnescape(entry.Name())
		if err != nil {
			return fmt.Errorf("decode log stream directory: %w", err)
		}
		if err := store.loadStream(streamID, filepath.Join(store.rootDir, entry.Name())); err != nil {
			return err
		}
	}
	return nil
}

func (store *FileLogBodyStore) loadStream(streamID string, streamDir string) error {
	segments, err := os.ReadDir(streamDir)
	if err != nil {
		return fmt.Errorf("read log stream directory: %w", err)
	}
	sort.SliceStable(segments, func(i, j int) bool { return segments[i].Name() < segments[j].Name() })
	for _, segment := range segments {
		if segment.IsDir() || !strings.HasPrefix(segment.Name(), "segment-") || !strings.HasSuffix(segment.Name(), ".jsonl") {
			continue
		}
		segmentPath := filepath.Join(streamDir, segment.Name())
		record, err := readLogSegment(segmentPath)
		if err != nil {
			return err
		}
		if len(record.Entries) == 0 {
			continue
		}
		if err := store.memory.AppendBatch(streamID, record); err != nil {
			return err
		}
	}
	return nil
}

func (store *FileLogBodyStore) streamDir(streamID string) string {
	return filepath.Join(store.rootDir, url.PathEscape(streamID))
}

func (store *FileLogBodyStore) segmentPath(streamID string, firstSeq uint64) string {
	return filepath.Join(store.streamDir(streamID), fmt.Sprintf("segment-%020d.jsonl", firstSeq))
}

func readLogSegment(path string) (domain.LogBatchRecord, error) {
	file, err := os.Open(path)
	if err != nil {
		return domain.LogBatchRecord{}, fmt.Errorf("open log segment: %w", err)
	}
	defer file.Close()

	entries := []domain.LogEntry{}
	reader := bufio.NewReader(file)
	for {
		line, readErr := reader.ReadBytes('\n')
		if len(line) == 0 && errors.Is(readErr, io.EOF) {
			break
		}
		var entry domain.LogEntry
		if err := json.Unmarshal(line, &entry); err != nil {
			return domain.LogBatchRecord{}, fmt.Errorf("decode log segment %s: %w", filepath.Base(path), err)
		}
		entries = append(entries, domain.CopyLogEntry(entry))
		if readErr == nil {
			continue
		}
		if errors.Is(readErr, io.EOF) {
			break
		}
		return domain.LogBatchRecord{}, fmt.Errorf("read log segment %s: %w", filepath.Base(path), readErr)
	}
	sort.SliceStable(entries, func(i, j int) bool { return entries[i].Seq < entries[j].Seq })
	if len(entries) == 0 {
		return domain.LogBatchRecord{}, nil
	}
	checksum, err := validator.LogEntriesChecksum(entries)
	if err != nil {
		return domain.LogBatchRecord{}, fmt.Errorf("checksum log segment %s: %w", filepath.Base(path), err)
	}
	return domain.LogBatchRecord{
		Checksum: checksum,
		FirstSeq: entries[0].Seq,
		LastSeq:  entries[len(entries)-1].Seq,
		Entries:  entries,
	}, nil
}