go-sip/internal/store/cleanup_rejected.go

package store

import (
	"bytes"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"os"
	"path/filepath"
	"strings"

	"git.ipao.vip/rogee/go-sip/internal/contract"
)

// CleanupRejectedFeedback covers structural rejection without an admitted call.
// Admitted rejected calls use CleanupCall so their inbox and receipt stay atomic.
func (s *Store) CleanupRejectedFeedback(dispatcherID, eventType, eventID string) (result CleanupResult, cleanupErr error) {
	if eventType != "call.execute" && eventType != "task.control" && eventType != "sip.config" {
		return result, errors.New("unsupported rejected feedback type")
	}
	s.cleanupMu.Lock()
	defer s.cleanupMu.Unlock()
	tx, err := s.db.Begin()
	if err != nil {
		return result, err
	}
	defer tx.Rollback()
	event, body, ready, err := confirmedCleanupFeedback(tx, dispatcherID, eventType, eventID)
	if err != nil {
		return result, err
	}
	if !ready {
		result.RetainedReason = "rejection_not_confirmed"
		return result, nil
	}
	if event.Payload.Status != "rejected" {
		result.RetainedReason = "feedback_is_not_rejection"
		return result, nil
	}
	if eventType == "call.execute" {
		var state string
		err := tx.QueryRow(`SELECT status FROM dispatcher_inbox WHERE dispatcher_id=? AND event_id=?`, dispatcherID, eventID).Scan(&state)
		if err == nil {
			if state != "rejected" {
				return result, errors.New("rejection conflicts with a retained execution")
			}
			result.RetainedReason = "admitted_call_cleanup_required"
			return result, nil
		}
		if !errors.Is(err, sql.ErrNoRows) {
			return result, err
		}
	}
	result, err = s.cleanupRejectionFiles(event, body)
	if err != nil {
		return result, err
	}
	deleted, err := tx.Exec(`DELETE FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type=? AND event_id=? AND confirmed=1`, dispatcherID, eventType, eventID)
	if err != nil {
		return result, fmt.Errorf("delete confirmed rejection receipt after original files: %w", err)
	}
	n, err := deleted.RowsAffected()
	if err != nil {
		return result, err
	}
	if err := tx.Commit(); err != nil {
		return result, err
	}
	result.DeletedRows = int(n)
	result.DataBytes = int64(len(body))
	return result, nil
}

// The caller holds the cleanup write lock and the receipt's DB transaction.
// This excludes RejectData from adding another original before receipt deletion.
func (s *Store) cleanupRejectionFiles(event cleanupFeedback, body []byte) (result CleanupResult, cleanupErr error) {
	dir := filepath.Join(filepath.Dir(s.path), "rejected-data")
	stat, err := os.Lstat(dir)
	if errors.Is(err, os.ErrNotExist) {
		return result, nil
	}
	if err != nil {
		return result, err
	}
	if !stat.IsDir() || stat.Mode()&os.ModeSymlink != 0 || stat.Mode().Perm()&0077 != 0 {
		return result, errors.New("rejected evidence directory is not private and physical")
	}
	files, err := os.ReadDir(dir)
	if err != nil {
		return result, err
	}
	var failures []error
	receiptSum := sha256.Sum256(body)
	receiptKey := hex.EncodeToString(receiptSum[:])
	for _, file := range files {
		name := file.Name()
		parts := strings.Split(strings.TrimSuffix(name, ".json"), ".")
		indexed := len(parts) == 2
		if indexed && parts[0] != receiptKey {
			continue
		}
		if len(parts) != 1 && !indexed {
			continue
		}
		stem := parts[len(parts)-1]
		digest, hashErr := hex.DecodeString(stem)
		if !strings.HasSuffix(name, ".json") || hashErr != nil || len(digest) != sha256.Size || hex.EncodeToString(digest) != stem {
			// Unrecognized/temporary artifacts are outside this cleanup. Do not
			// open them, derive a source identity, or delete them.
			continue
		}
		path := filepath.Join(dir, name)
		stat, err := os.Lstat(path)
		if err != nil {
			failures = append(failures, fmt.Errorf("inspect rejection evidence %s: %w", stem, err))
			continue
		}
		if !stat.Mode().IsRegular() {
			failures = append(failures, fmt.Errorf("rejection evidence %s is not a regular file; completion proof retained", stem))
			continue
		}
		encoded, err := os.ReadFile(path)
		if err != nil {
			failures = append(failures, fmt.Errorf("read rejection evidence %s: %w", stem, err))
			continue
		}
		var evidence struct {
			Original      []byte          `json:"Original"`
			InvalidSource []byte          `json:"InvalidSource"`
			Feedback      json.RawMessage `json:"Feedback"`
		}
		decode := json.NewDecoder(bytes.NewReader(encoded))
		decode.DisallowUnknownFields()
		decodeErr := decode.Decode(&evidence)
		var header cleanupFeedback
		feedbackErr := json.Unmarshal(evidence.Feedback, &header)
		var original struct {
			EventID      string          `json:"event_id"`
			EventType    string          `json:"event_type"`
			DispatcherID string          `json:"dispatcher_id"`
			TenantID     int64           `json:"tenant_id"`
			Payload      json.RawMessage `json:"payload"`
		}
		originalErr := json.Unmarshal(evidence.Original, &original)
		candidate := indexed || (feedbackErr == nil && header.DispatcherID == event.DispatcherID && header.EventType == event.EventType && header.EventID == event.EventID) ||
			(originalErr == nil && original.DispatcherID == event.DispatcherID && original.EventType == event.EventType && original.EventID == event.EventID)
		if !candidate {
			if decodeErr != nil || feedbackErr != nil || originalErr != nil {
				slog.Warn("Unidentifiable rejection evidence retained", "file_sha256", stem, "reason", "unknown_source_identity")
				failures = append(failures, fmt.Errorf("legacy rejection evidence %s cannot be associated; completion proof retained", stem))
			}
			continue
		}
		fail := func(reason string) {
			failures = append(failures, fmt.Errorf("rejection evidence %s: %s", stem, reason))
		}
		if decodeErr != nil {
			fail("invalid evidence structure")
			continue
		}
		if err := decode.Decode(new(any)); !errors.Is(err, io.EOF) {
			fail("trailing evidence data")
			continue
		}
		actualDigest := sha256.Sum256(encoded)
		if !bytes.Equal(actualDigest[:], digest) || stat.Mode().Perm()&0077 != 0 {
			fail("evidence digest or private permissions differ")
			continue
		}
		if feedbackErr != nil || originalErr != nil || !bytes.Equal(evidence.Feedback, body) {
			fail("feedback differs from its confirmed rejection")
			continue
		}
		if original.EventID != event.EventID || original.EventType != event.EventType || original.DispatcherID != event.DispatcherID || (event.EventType != "sip.config" && original.TenantID != event.TenantID) {
			fail("original identity differs from its rejection")
			continue
		}
		if event.EventType != "sip.config" && (original.TenantID <= 0 || len(evidence.Original) == 0) {
			fail("original tenant identity is incomplete")
			continue
		}
		if contract.ValidateCurrent("mq", evidence.Original) == nil {
			var payload map[string]json.RawMessage
			if err := json.Unmarshal(original.Payload, &payload); err != nil {
				fail("original direction cannot be verified")
				continue
			}
			if _, receipt := payload["status"]; receipt {
				fail("original is an outbound receipt, not a command")
				continue
			}
		}
		if s.removeRejectedFile != nil {
			err = s.removeRejectedFile(path)
		} else {
			err = os.Remove(path)
		}
		if err != nil && !errors.Is(err, os.ErrNotExist) {
			fail("original file deletion failed: " + err.Error())
			continue
		}
		if err == nil {
			result.DeletedFiles++
			result.FileBytes += stat.Size()
		}
		slog.Info("Confirmed rejection original removed", "dispatcher_id", event.DispatcherID, "event_type", event.EventType, "event_id", event.EventID, "tenant_id", event.TenantID, "file_sha256", stem, "file_bytes", stat.Size())
	}
	// Even a retry with no remaining matched files persists earlier unlink
	// operations before allowing the DB receipt transaction to commit.
	if err := syncPrivateDirectory(dir); err != nil {
		failures = append(failures, fmt.Errorf("persist rejected original deletion: %w", err))
	}
	return result, errors.Join(failures...)
}