browser/platform/service/log_body_store.go

package service

import (
	"bufio"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/url"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"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
}

type logSegmentRef struct {
	firstSeq uint64
	path     string
}

func NewFileLogBodyStore(rootDir string) (*FileLogBodyStore, error) {
	rootDir = strings.TrimSpace(rootDir)
	if rootDir == "" {
		return nil, fmt.Errorf("log directory is required")
	}
	store := &FileLogBodyStore{
		rootDir: rootDir,
	}
	if err := os.MkdirAll(rootDir, 0o700); err != nil {
		return nil, fmt.Errorf("create log directory: %w", 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()

	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 {
		existing, err := readLogSegment(segmentPath)
		if err != nil {
			return err
		}
		if sameLogBatchRecord(existing, record) {
			return 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 nil
}

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

	record, err := readLogSegment(store.segmentPath(streamID, firstSeq))
	if errors.Is(err, os.ErrNotExist) {
		return domain.LogBatchRecord{}, false, nil
	}
	if err != nil {
		return domain.LogBatchRecord{}, false, err
	}
	if len(record.Entries) == 0 {
		return domain.LogBatchRecord{}, false, nil
	}
	return record, true, nil
}

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

	if limit <= 0 {
		return nil, afterSeq, nil
	}
	segments, err := store.listSegments(streamID)
	if err != nil {
		return nil, 0, err
	}
	selected := make([]domain.LogEntry, 0, limit)
	nextSeq := afterSeq
	for index := firstLogSegmentIndex(segments, afterSeq); index < len(segments) && len(selected) < limit; index++ {
		record, err := readLogSegment(segments[index].path)
		if err != nil {
			return nil, 0, err
		}
		for _, entry := range record.Entries {
			if entry.Seq <= afterSeq {
				continue
			}
			if len(selected) >= limit {
				break
			}
			selected = append(selected, domain.CopyLogEntry(entry))
			nextSeq = entry.Seq
		}
	}
	return selected, nextSeq, nil
}

func (store *FileLogBodyStore) LatestSeq(streamID string) (uint64, error) {
	store.mu.Lock()
	defer store.mu.Unlock()

	segments, err := store.listSegments(streamID)
	if err != nil {
		return 0, err
	}
	for index := len(segments) - 1; index >= 0; index-- {
		record, err := readLogSegment(segments[index].path)
		if err != nil {
			return 0, err
		}
		if len(record.Entries) > 0 {
			return record.LastSeq, nil
		}
	}
	return 0, 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 (store *FileLogBodyStore) listSegments(streamID string) ([]logSegmentRef, error) {
	streamDir := store.streamDir(streamID)
	entries, err := os.ReadDir(streamDir)
	if errors.Is(err, os.ErrNotExist) {
		return nil, nil
	}
	if err != nil {
		return nil, fmt.Errorf("read log stream directory: %w", err)
	}
	segments := make([]logSegmentRef, 0, len(entries))
	for _, entry := range entries {
		if entry.IsDir() {
			continue
		}
		firstSeq, ok, err := logSegmentFirstSeq(entry.Name())
		if err != nil {
			return nil, err
		}
		if !ok {
			continue
		}
		segments = append(segments, logSegmentRef{firstSeq: firstSeq, path: filepath.Join(streamDir, entry.Name())})
	}
	sort.SliceStable(segments, func(i, j int) bool { return segments[i].firstSeq < segments[j].firstSeq })
	return segments, nil
}

func logSegmentFirstSeq(name string) (uint64, bool, error) {
	if !strings.HasPrefix(name, "segment-") || !strings.HasSuffix(name, ".jsonl") {
		return 0, false, nil
	}
	firstSeq, err := strconv.ParseUint(strings.TrimSuffix(strings.TrimPrefix(name, "segment-"), ".jsonl"), 10, 64)
	if err != nil {
		return 0, true, fmt.Errorf("parse log segment %s: %w", name, err)
	}
	return firstSeq, true, nil
}

func firstLogSegmentIndex(segments []logSegmentRef, afterSeq uint64) int {
	index := sort.Search(len(segments), func(i int) bool { return segments[i].firstSeq > afterSeq })
	if index < len(segments) && afterSeq != ^uint64(0) && segments[index].firstSeq == afterSeq+1 {
		return index
	}
	if index > 0 {
		return index - 1
	}
	return index
}

func sameLogBatchRecord(left domain.LogBatchRecord, right domain.LogBatchRecord) bool {
	if left.FirstSeq != right.FirstSeq || left.LastSeq != right.LastSeq {
		return false
	}
	return left.Checksum == right.Checksum
}

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
}