// Package persistence provides data persistence implementations for the domain layer. // This file contains the indexed log storage system for O(1) log retrieval. package persistence import ( "bufio" "context" "encoding/json" "fmt" "os" "path/filepath" "regexp" "sort" "strings" "sync" "time" "github.com/router-for-me/CLIProxyAPI/v6/internal/domain/errors" "github.com/router-for-me/CLIProxyAPI/v6/internal/domain/ports" ) // LogIndexEntry represents a single entry in the log index type LogIndexEntry struct { RequestID string `json:"request_id"` Filename string `json:"filename"` Timestamp time.Time `json:"timestamp"` Method string `json:"method"` URL string `json:"url"` StatusCode int `json:"status_code"` Size int64 `json:"size"` Offset int64 `json:"offset"` // Byte offset in file for O(1) access Tags map[string]string `json:"tags"` // Optional tags for filtering } // LogIndex provides O(1) lookup for log entries by various criteria type LogIndex struct { mu sync.RWMutex // Primary index: RequestID -> Entry byRequestID map[string]*LogIndexEntry // Secondary index: Timestamp -> []Entries (for time range queries) byTimestamp *TimeIndex // Secondary index: Method -> []Entries byMethod map[string][]*LogIndexEntry // Secondary index: StatusCode -> []Entries byStatusCode map[int][]*LogIndexEntry // File index: Filename -> []Entries byFile map[string][]*LogIndexEntry // Index file path indexPath string // Dirty flag for persistence dirty bool // Last persisted time lastPersisted time.Time } // TimeIndex provides efficient time range queries type TimeIndex struct { // Sorted list of timestamps for binary search timestamps []time.Time // Map from timestamp to entries entries map[int64][]*LogIndexEntry } // NewTimeIndex creates a new TimeIndex func NewTimeIndex() *TimeIndex { return &TimeIndex{ timestamps: make([]time.Time, 0), entries: make(map[int64][]*LogIndexEntry), } } // Add adds an entry to the time index func (ti *TimeIndex) Add(entry *LogIndexEntry) { ts := entry.Timestamp.Unix() // Check if we already have this timestamp if _, exists := ti.entries[ts]; !exists { // Insert timestamp in sorted order idx := sort.Search(len(ti.timestamps), func(i int) bool { return ti.timestamps[i].Unix() >= ts }) if idx < len(ti.timestamps) && ti.timestamps[idx].Unix() == ts { // Already exists } else { ti.timestamps = append(ti.timestamps, time.Time{}) copy(ti.timestamps[idx+1:], ti.timestamps[idx:]) ti.timestamps[idx] = entry.Timestamp } } ti.entries[ts] = append(ti.entries[ts], entry) } // QueryRange returns all entries within the given time range func (ti *TimeIndex) QueryRange(start, end time.Time) []*LogIndexEntry { result := make([]*LogIndexEntry, 0) startIdx := sort.Search(len(ti.timestamps), func(i int) bool { return !ti.timestamps[i].Before(start) }) endIdx := sort.Search(len(ti.timestamps), func(i int) bool { return ti.timestamps[i].After(end) }) for i := startIdx; i < endIdx && i < len(ti.timestamps); i++ { ts := ti.timestamps[i].Unix() result = append(result, ti.entries[ts]...) } return result } // NewLogIndex creates a new LogIndex func NewLogIndex(indexPath string) *LogIndex { return &LogIndex{ byRequestID: make(map[string]*LogIndexEntry), byTimestamp: NewTimeIndex(), byMethod: make(map[string][]*LogIndexEntry), byStatusCode: make(map[int][]*LogIndexEntry), byFile: make(map[string][]*LogIndexEntry), indexPath: indexPath, lastPersisted: time.Now(), } } // LoadLogIndex loads a LogIndex from disk func LoadLogIndex(indexPath string) (*LogIndex, error) { idx := NewLogIndex(indexPath) // Check if index file exists if _, err := os.Stat(indexPath); os.IsNotExist(err) { // No existing index, return empty return idx, nil } // Load index from file data, err := os.ReadFile(indexPath) if err != nil { return nil, errors.Wrap(errors.InternalError, "failed to read index file", err) } // Parse index entries var entries []*LogIndexEntry if err := json.Unmarshal(data, &entries); err != nil { // Try line-delimited JSON format entries = idx.parseLineDelimitedJSON(data) } // Add all entries to index for _, entry := range entries { idx.addEntry(entry) } return idx, nil } // parseLineDelimitedJSON parses line-delimited JSON format func (idx *LogIndex) parseLineDelimitedJSON(data []byte) []*LogIndexEntry { entries := make([]*LogIndexEntry, 0) scanner := bufio.NewScanner(strings.NewReader(string(data))) for scanner.Scan() { line := strings.TrimSpace(scanner.Text()) if line == "" { continue } var entry LogIndexEntry if err := json.Unmarshal([]byte(line), &entry); err != nil { continue } entries = append(entries, &entry) } return entries } // Add adds a log entry to the index func (idx *LogIndex) Add(entry *LogIndexEntry) { idx.mu.Lock() defer idx.mu.Unlock() idx.addEntry(entry) idx.dirty = true } // addEntry adds an entry without locking (internal use) func (idx *LogIndex) addEntry(entry *LogIndexEntry) { // Add to primary index idx.byRequestID[entry.RequestID] = entry // Add to time index idx.byTimestamp.Add(entry) // Add to method index idx.byMethod[entry.Method] = append(idx.byMethod[entry.Method], entry) // Add to status code index idx.byStatusCode[entry.StatusCode] = append(idx.byStatusCode[entry.StatusCode], entry) // Add to file index idx.byFile[entry.Filename] = append(idx.byFile[entry.Filename], entry) } // GetByRequestID retrieves a log entry by request ID (O(1)) func (idx *LogIndex) GetByRequestID(requestID string) (*LogIndexEntry, bool) { idx.mu.RLock() defer idx.mu.RUnlock() entry, exists := idx.byRequestID[requestID] return entry, exists } // QueryByTimeRange retrieves log entries within a time range func (idx *LogIndex) QueryByTimeRange(start, end time.Time) []*LogIndexEntry { idx.mu.RLock() defer idx.mu.RUnlock() return idx.byTimestamp.QueryRange(start, end) } // QueryByMethod retrieves log entries by HTTP method func (idx *LogIndex) QueryByMethod(method string) []*LogIndexEntry { idx.mu.RLock() defer idx.mu.RUnlock() return idx.byMethod[method] } // QueryByStatusCode retrieves log entries by status code func (idx *LogIndex) QueryByStatusCode(statusCode int) []*LogIndexEntry { idx.mu.RLock() defer idx.mu.RUnlock() return idx.byStatusCode[statusCode] } // QueryByFile retrieves log entries by filename func (idx *LogIndex) QueryByFile(filename string) []*LogIndexEntry { idx.mu.RLock() defer idx.mu.RUnlock() return idx.byFile[filename] } // GetAllEntries returns all indexed entries func (idx *LogIndex) GetAllEntries() []*LogIndexEntry { idx.mu.RLock() defer idx.mu.RUnlock() result := make([]*LogIndexEntry, 0, len(idx.byRequestID)) for _, entry := range idx.byRequestID { result = append(result, entry) } // Sort by timestamp sort.Slice(result, func(i, j int) bool { return result[i].Timestamp.Before(result[j].Timestamp) }) return result } // Persist saves the index to disk func (idx *LogIndex) Persist() error { idx.mu.Lock() defer idx.mu.Unlock() if !idx.dirty { return nil } // Get all entries entries := make([]*LogIndexEntry, 0, len(idx.byRequestID)) for _, entry := range idx.byRequestID { entries = append(entries, entry) } // Sort by timestamp for consistent output sort.Slice(entries, func(i, j int) bool { return entries[i].Timestamp.Before(entries[j].Timestamp) }) // Write as line-delimited JSON for append-friendly format file, err := os.Create(idx.indexPath) if err != nil { return errors.Wrap(errors.InternalError, "failed to create index file", err) } defer file.Close() encoder := json.NewEncoder(file) for _, entry := range entries { if err := encoder.Encode(entry); err != nil { return errors.Wrap(errors.InternalError, "failed to encode index entry", err) } } idx.dirty = false idx.lastPersisted = time.Now() return nil } // Remove removes an entry from the index func (idx *LogIndex) Remove(requestID string) bool { idx.mu.Lock() defer idx.mu.Unlock() entry, exists := idx.byRequestID[requestID] if !exists { return false } delete(idx.byRequestID, requestID) // Note: Removing from secondary indexes is complex and may leave stale entries. // For simplicity, we mark as dirty and will rebuild on next load if needed. // A full implementation would properly remove from all indexes. _ = entry idx.dirty = true return true } // Size returns the number of entries in the index func (idx *LogIndex) Size() int { idx.mu.RLock() defer idx.mu.RUnlock() return len(idx.byRequestID) } // IsDirty returns whether the index has unsaved changes func (idx *LogIndex) IsDirty() bool { idx.mu.RLock() defer idx.mu.RUnlock() return idx.dirty } // IndexedLogRepository wraps LogRepository with indexing capabilities type IndexedLogRepository struct { *LogRepository index *LogIndex indexPath string logDir string mu sync.RWMutex } // NewIndexedLogRepository creates a new indexed log repository func NewIndexedLogRepository(logDir string, cfg interface{}) (*IndexedLogRepository, error) { indexPath := filepath.Join(logDir, ".log_index.json") // Load or create index index, err := LoadLogIndex(indexPath) if err != nil { // Create new index if load fails index = NewLogIndex(indexPath) } // Create base repository (we'll use a minimal config) baseRepo := &LogRepository{ logDir: logDir, } return &IndexedLogRepository{ LogRepository: baseRepo, index: index, indexPath: indexPath, logDir: logDir, }, nil } // GetRequestLogByID retrieves a specific request log by ID (O(1) with index) func (r *IndexedLogRepository) GetRequestLogByID(ctx context.Context, requestID string) ([]byte, error) { r.mu.RLock() defer r.mu.RUnlock() // Try index first if entry, exists := r.index.GetByRequestID(requestID); exists { // Read from file at specific offset for efficiency return r.readLogAtOffset(entry.Filename, entry.Offset, entry.Size) } // Fall back to scanning (for entries not yet indexed) return r.LogRepository.GetRequestLogByID(ctx, requestID) } // readLogAtOffset reads a log entry from a file at a specific offset func (r *IndexedLogRepository) readLogAtOffset(filename string, offset, size int64) ([]byte, error) { filePath := filepath.Join(r.logDir, filename) file, err := os.Open(filePath) if err != nil { return nil, errors.Wrap(errors.InternalError, "failed to open log file", err) } defer file.Close() // Seek to offset if _, err := file.Seek(offset, 0); err != nil { return nil, errors.Wrap(errors.InternalError, "failed to seek in log file", err) } // Read the entry data := make([]byte, size) if _, err := file.Read(data); err != nil { return nil, errors.Wrap(errors.InternalError, "failed to read log entry", err) } return data, nil } // IndexLogFile indexes a single log file func (r *IndexedLogRepository) IndexLogFile(filename string) error { r.mu.Lock() defer r.mu.Unlock() filePath := filepath.Join(r.logDir, filename) file, err := os.Open(filePath) if err != nil { return errors.Wrap(errors.InternalError, "failed to open log file", err) } defer file.Close() // Parse log file and extract entries entries, err := r.parseLogFile(filename, file) if err != nil { return err } // Add entries to index for _, entry := range entries { r.index.Add(entry) } return nil } // parseLogFile parses a log file and extracts index entries func (r *IndexedLogRepository) parseLogFile(filename string, file *os.File) ([]*LogIndexEntry, error) { entries := make([]*LogIndexEntry, 0) scanner := bufio.NewScanner(file) var offset int64 var currentEntry *LogIndexEntry requestIDRegex := regexp.MustCompile(`-(\w+)\.log$`) requestID := "" if matches := requestIDRegex.FindStringSubmatch(filename); len(matches) > 1 { requestID = matches[1] } for scanner.Scan() { line := scanner.Text() lineLen := int64(len(line) + 1) // +1 for newline // Parse log entry header if strings.HasPrefix(line, "=== REQUEST INFO ===") { currentEntry = &LogIndexEntry{ Filename: filename, Offset: offset, RequestID: requestID, Tags: make(map[string]string), } } if currentEntry != nil { // Extract fields from log if strings.HasPrefix(line, "Method: ") { currentEntry.Method = strings.TrimPrefix(line, "Method: ") } if strings.HasPrefix(line, "URL: ") { currentEntry.URL = strings.TrimPrefix(line, "URL: ") } if strings.HasPrefix(line, "Timestamp: ") { tsStr := strings.TrimPrefix(line, "Timestamp: ") if ts, err := time.Parse(time.RFC3339Nano, tsStr); err == nil { currentEntry.Timestamp = ts } } if strings.HasPrefix(line, "Status: ") { fmt.Sscanf(strings.TrimPrefix(line, "Status: "), "%d", ¤tEntry.StatusCode) } } // End of entry if currentEntry != nil && line == "=== END ===" { currentEntry.Size = offset - currentEntry.Offset entries = append(entries, currentEntry) currentEntry = nil } offset += lineLen } return entries, scanner.Err() } // RebuildIndex rebuilds the entire index from log files func (r *IndexedLogRepository) RebuildIndex() error { r.mu.Lock() defer r.mu.Unlock() // Create new index r.index = NewLogIndex(r.indexPath) // List all log files entries, err := os.ReadDir(r.logDir) if err != nil { return errors.Wrap(errors.InternalError, "failed to list log directory", err) } // Index each file for _, entry := range entries { if entry.IsDir() { continue } name := entry.Name() if !strings.HasSuffix(name, ".log") { continue } if err := r.IndexLogFile(name); err != nil { // Log error but continue continue } } // Persist the new index return r.index.Persist() } // PersistIndex saves the index to disk func (r *IndexedLogRepository) PersistIndex() error { r.mu.RLock() defer r.mu.RUnlock() return r.index.Persist() } // GetIndexStats returns statistics about the index func (r *IndexedLogRepository) GetIndexStats() map[string]interface{} { r.mu.RLock() defer r.mu.RUnlock() return map[string]interface{}{ "total_entries": r.index.Size(), "is_dirty": r.index.IsDirty(), "index_path": r.indexPath, "last_persisted": r.index.lastPersisted, } } // QueryLogs queries logs using the index func (r *IndexedLogRepository) QueryLogs( ctx context.Context, start, end time.Time, method string, statusCode int, ) ([]*ports.LogFileInfo, error) { r.mu.RLock() defer r.mu.RUnlock() var entries []*LogIndexEntry // Use time range query as base if !start.IsZero() && !end.IsZero() { entries = r.index.QueryByTimeRange(start, end) } else if method != "" { entries = r.index.QueryByMethod(method) } else if statusCode > 0 { entries = r.index.QueryByStatusCode(statusCode) } else { entries = r.index.GetAllEntries() } // Convert to LogFileInfo result := make([]*ports.LogFileInfo, 0, len(entries)) for _, entry := range entries { result = append(result, &ports.LogFileInfo{ Name: entry.Filename, Size: entry.Size, Modified: entry.Timestamp, }) } return result, nil } // Ensure IndexedLogRepository implements the interface var _ ports.LogRepository = (*IndexedLogRepository)(nil)