teraWeb/backend/internal/logstorage/clear_port.go

package logstorage

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"regexp"

	"tera-clone/services/tera-api/internal/storage/atomicfile"
)

var canonicalPort = regexp.MustCompile(`^CH(?:0[1-9]|[1-3][0-9]|4[0-8])$`)

// A small, authoritative deletion checkpoint is committed before compaction.
// A crash between segment rewrites cannot resurrect cleared records, or reuse
// their sequence numbers. This is not part of the rebuildable index.
type portClearState struct {
	Version      int               `json:"version"`
	Revision     uint64            `json:"revision"`
	NextSequence uint64            `json:"next_sequence"`
	Cutoffs      map[string]uint64 `json:"cutoffs"`
}

func (s *Store) portClearPath() string { return filepath.Join(s.root, "port", "cleared-ports-v1.json") }

func (s *Store) loadPortClearsLocked() error {
	info, err := os.Lstat(s.portClearPath())
	if os.IsNotExist(err) {
		return nil
	}
	if err != nil {
		return err
	}
	if !info.Mode().IsRegular() || info.Size() > 8192 {
		return errors.New("invalid port log deletion checkpoint")
	}
	data, err := os.ReadFile(s.portClearPath())
	if err != nil {
		return err
	}
	var state portClearState
	if err := json.Unmarshal(data, &state); err != nil {
		return err
	}
	if state.Version != 1 || state.NextSequence == 0 || len(state.Cutoffs) > 48 {
		return errors.New("invalid port log deletion checkpoint")
	}
	for channel, cutoff := range state.Cutoffs {
		if !canonicalPort.MatchString(channel) || cutoff >= state.NextSequence {
			return errors.New("invalid port log deletion cutoff")
		}
	}
	s.portClears = state
	return nil
}

func (s *Store) portRecordCleared(event Envelope) bool {
	if event.Class != ClassPort || len(s.portClears.Cutoffs) == 0 {
		return false
	}
	var record struct {
		Channel string `json:"channel"`
	}
	if json.Unmarshal(event.Payload, &record) != nil {
		return false
	}
	cutoff, found := s.portClears.Cutoffs[record.Channel]
	return found && event.Sequence <= cutoff
}

func (s *Store) PortClearRevision() uint64 {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.portClears.Revision
}

// ClearPort removes one port or ALL ports' communication logs. Audit,
// system, and link-state classes are never deletion targets. The mutex gives
// the clear a precise boundary relative to concurrent appends.
func (s *Store) ClearPort(channel string) (int, error) {
	if channel != "ALL" && !canonicalPort.MatchString(channel) {
		return 0, errors.New("port log channel must be ALL or CH01 through CH48")
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	deleted := 0
	if err := s.walkClassLocked(ClassPort, false, func(event Envelope) error {
		var record struct {
			Channel string `json:"channel"`
		}
		if err := json.Unmarshal(event.Payload, &record); err != nil {
			return fmt.Errorf("invalid port log payload: %w", err)
		}
		if channel == "ALL" || record.Channel == channel {
			deleted++
		}
		return nil
	}); err != nil {
		return 0, err
	}
	next := portClearState{Version: 1, Revision: s.portClears.Revision + 1, NextSequence: s.classes[ClassPort].NextSequence, Cutoffs: map[string]uint64{}}
	for key, value := range s.portClears.Cutoffs {
		next.Cutoffs[key] = value
	}
	if channel == "ALL" {
		for i := 1; i <= 48; i++ {
			next.Cutoffs[fmt.Sprintf("CH%02d", i)] = next.NextSequence - 1
		}
	} else {
		next.Cutoffs[channel] = next.NextSequence - 1
	}
	data, err := json.Marshal(next)
	if err != nil {
		return 0, err
	}
	if err := atomicfile.Write(s.portClearPath(), append(data, '\n'), 0o600); err != nil {
		// A directory-sync failure can occur after rename. Reload the committed
		// checkpoint so subsequent reads still honor a completed rename.
		_ = s.loadPortClearsLocked()
		return 0, err
	}
	s.portClears = next
	if err := s.compactClearedPortsLocked(); err != nil {
		return deleted, err
	}
	return deleted, nil
}

func (s *Store) compactClearedPortsLocked() error {
	return s.compactClearedClassLocked(ClassPort, s.portRecordCleared)
}

func (s *Store) compactClearedClassLocked(class string, cleared func(Envelope) bool) (resultErr error) {
	// Even a partial compaction changes byte offsets. Never leave an old index
	// in memory after an I/O failure halfway through the segment list.
	defer func() {
		if err := s.rebuildLocked(); resultErr == nil {
			resultErr = err
		}
	}()
	paths, err := s.closedSegmentsLocked(class)
	if err != nil {
		return err
	}
	active := s.activePath(class)
	if _, err := os.Lstat(active); err == nil {
		paths = append(paths, active)
	} else if !os.IsNotExist(err) {
		return err
	}
	for _, path := range paths {
		var kept bytes.Buffer
		removed := 0
		_, err := inspectSegment(path, class, false, func(event Envelope) error {
			if cleared(event) {
				removed++
				return nil
			}
			return json.NewEncoder(&kept).Encode(event)
		})
		if err != nil {
			return err
		}
		if removed == 0 {
			continue
		}
		// The checkpoint already hides all targets. Each bounded segment rewrite
		// is atomic; remaining ports retain their original envelopes/checksums.
		if err := atomicfile.Write(path, kept.Bytes(), 0o600); err != nil {
			return err
		}
	}
	return nil
}