Logging-System/server/store.go

package server

import (
	"bufio"
	"encoding/json"
	"os"
	"sync"

	"logging-system/logsys"
	"logging-system/redact"
)

// Record 是带序号的日志，序号即查询游标。
type Record struct {
	Seq int64 `json:"seq"`
	logsys.Entry
}

// Store 是线程安全的内存日志表，可选追加到 JSONL 文件。
type Store struct {
	mu      sync.Mutex
	records []Record
	nextSeq int64
	limit   int
	path    string
	file    *os.File
	redact  *redact.Redactor
	subs    map[int]chan Record
	nextSub int
}

// OpenStore 打开存储。path 为空则只放内存。limit <= 0 时保留最近 100000 条。
func OpenStore(path string, limit int, redactor *redact.Redactor) (*Store, error) {
	if limit <= 0 {
		limit = 100000
	}
	s := &Store{
		limit:  limit,
		path:   path,
		redact: redactor,
		subs:   map[int]chan Record{},
	}
	if path == "" {
		return s, nil
	}
	if err := s.load(); err != nil {
		return nil, err
	}
	f, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644)
	if err != nil {
		return nil, err
	}
	s.file = f
	return s, nil
}

func (s *Store) load() error {
	f, err := os.Open(s.path)
	if os.IsNotExist(err) {
		return nil
	}
	if err != nil {
		return err
	}
	defer f.Close()
	sc := bufio.NewScanner(f)
	sc.Buffer(make([]byte, 0, 64*1024), 2<<20)
	for sc.Scan() {
		var rec Record
		if err := json.Unmarshal(sc.Bytes(), &rec); err != nil {
			continue
		}
		s.records = append(s.records, rec)
		if rec.Seq > s.nextSeq {
			s.nextSeq = rec.Seq
		}
	}
	if len(s.records) > s.limit {
		s.records = s.records[len(s.records)-s.limit:]
	}
	return sc.Err()
}

// Append 写入一批日志并通知订阅者。慢订阅者丢帧，不阻塞写入。
func (s *Store) Append(entries []logsys.Entry) []Record {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := make([]Record, 0, len(entries))
	for _, e := range entries {
		if s.redact != nil {
			e.Message = s.redact.Apply(e.Message)
			e.Error = s.redact.Apply(e.Error)
			e.Fields = redactFields(e.Fields, s.redact)
		}
		s.nextSeq++
		rec := Record{Seq: s.nextSeq, Entry: e}
		s.records = append(s.records, rec)
		if s.file != nil {
			if data, err := json.Marshal(rec); err == nil {
				_, _ = s.file.Write(append(data, '\n'))
			}
		}
		out = append(out, rec)
		for _, ch := range s.subs {
			select {
			case ch <- rec:
			default:
			}
		}
	}
	if len(s.records) > s.limit {
		s.records = s.records[len(s.records)-s.limit:]
	}
	return out
}

// Query 按条件过滤。cursor 为已见过的最大序号，只返回更大的序号。
func (s *Store) Query(q Query) (records []Record, next int64) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if q.Limit <= 0 {
		q.Limit = 100
	}
	if q.Limit > 1000 {
		q.Limit = 1000
	}
	for _, rec := range s.records {
		if rec.Seq <= q.Cursor {
			continue
		}
		if q.TraceID != "" && rec.TraceID != q.TraceID {
			continue
		}
		if q.Level != "" && rec.Level != q.Level {
			continue
		}
		if q.Service != "" && rec.Service != q.Service {
			continue
		}
		if q.Text != "" && !contains(rec, q.Text) {
			continue
		}
		records = append(records, rec)
		next = rec.Seq
		if len(records) >= q.Limit {
			break
		}
	}
	if next == 0 {
		next = q.Cursor
	}
	return records, next
}

// Query 是检索条件。
type Query struct {
	TraceID string
	Level   string
	Service string
	Text    string
	Cursor  int64
	Limit   int
}

func contains(rec Record, text string) bool {
	if substr(rec.Message, text) || substr(rec.Error, text) || substr(rec.Code, text) || substr(rec.TraceID, text) {
		return true
	}
	for k, v := range rec.Fields {
		if substr(k, text) || substr(toString(v), text) {
			return true
		}
	}
	return false
}

func substr(s, sub string) bool {
	return len(sub) > 0 && len(s) >= len(sub) && (s == sub || indexOf(s, sub) >= 0)
}

func indexOf(s, sub string) int {
	for i := 0; i+len(sub) <= len(s); i++ {
		if s[i:i+len(sub)] == sub {
			return i
		}
	}
	return -1
}

func toString(v any) string {
	s, _ := v.(string)
	return s
}

// Subscribe 从 after 序号之后回放，并继续接收新日志。
func (s *Store) Subscribe(after int64) (<-chan Record, func()) {
	s.mu.Lock()
	defer s.mu.Unlock()
	var replay []Record
	for _, rec := range s.records {
		if rec.Seq > after {
			replay = append(replay, rec)
		}
	}
	buf := len(replay) + 256
	ch := make(chan Record, buf)
	for _, rec := range replay {
		ch <- rec
	}
	id := s.nextSub
	s.nextSub++
	s.subs[id] = ch
	return ch, func() {
		s.mu.Lock()
		delete(s.subs, id)
		s.mu.Unlock()
	}
}

// Close 关闭落盘文件。
func (s *Store) Close() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.file == nil {
		return nil
	}
	err := s.file.Close()
	s.file = nil
	return err
}

func redactFields(in map[string]any, r *redact.Redactor) map[string]any {
	if len(in) == 0 {
		return in
	}
	out := make(map[string]any, len(in))
	for k, v := range in {
		if s, ok := v.(string); ok {
			out[k] = r.Apply(s)
			continue
		}
		out[k] = v
	}
	return out
}