yovision/Sense/internal/orphan/orphan.go

// Package orphan reports MediaMTX configuration paths that do not match the
// current Sense ledger. Only paths with durable Sense ownership evidence can
// ever enter the controlled cleanup set.
package orphan

import (
	"context"
	"crypto/rand"
	"errors"
	"fmt"
	"math/big"
	"regexp"
	"sort"
	"strings"
	"time"

	"yovision/sense/internal/metrics"
	"yovision/sense/internal/store"
)

const (
	defaultLeaseDuration    = 30 * time.Second
	defaultOperationTimeout = 20 * time.Second
	reportTTL               = 15 * time.Minute
	maxCleanupItems         = 128
	crockford               = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
)

var (
	ErrLeaseHeld       = errors.New("orphan operation lease held")
	ErrConfirmation    = errors.New("orphan cleanup confirmation invalid")
	ErrSnapshotExpired = errors.New("orphan scan snapshot expired")
	ErrSafetyBlocked   = errors.New("orphan cleanup safety gate blocked")
	operatorIDPattern  = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$`)
)

type Runtime interface {
	ListPathNames(context.Context) ([]string, error)
	DeletePath(context.Context, string) error
}

type Manager struct {
	repository       store.OrphanRepository
	runtime          Runtime
	instanceID       string
	metrics          *metrics.Registry
	now              func() time.Time
	leaseDuration    time.Duration
	operationTimeout time.Duration
}

type CleanupResult struct {
	ScanID  string `json:"scan_id"`
	Deleted int    `json:"deleted"`
	Failed  int    `json:"failed"`
}

func New(
	repository store.OrphanRepository,
	runtime Runtime,
	instanceID string,
	registry *metrics.Registry,
) *Manager {
	return &Manager{
		repository: repository, runtime: runtime, instanceID: instanceID, metrics: registry,
		now: time.Now, leaseDuration: defaultLeaseDuration, operationTimeout: defaultOperationTimeout,
	}
}

func (m *Manager) Report(ctx context.Context) (store.OrphanScan, error) {
	now := m.now().UTC()
	token, err := randomToken()
	if err != nil {
		return store.OrphanScan{}, err
	}
	acquired, err := m.repository.AcquireOperationalLease(
		ctx, store.OperationalLeaseOrphanScan, m.instanceID, token, now, m.leaseDuration,
	)
	if err != nil {
		m.observeScan(0, 0, 0, err)
		return store.OrphanScan{}, err
	}
	if !acquired {
		return store.OrphanScan{}, ErrLeaseHeld
	}
	saved := false
	defer func() {
		if saved {
			return
		}
		releaseCtx, cancel := context.WithTimeout(context.Background(), time.Second)
		defer cancel()
		_ = m.repository.ReleaseOperationalLease(
			releaseCtx, store.OperationalLeaseOrphanScan, m.instanceID, token, m.now().UTC(),
		)
	}()

	operationCtx, cancel := context.WithTimeout(ctx, m.operationTimeout)
	paths, err := m.runtime.ListPathNames(operationCtx)
	cancel()
	if err != nil {
		m.observeScan(0, 0, 0, err)
		return store.OrphanScan{}, fmt.Errorf("list MediaMTX paths: %w", err)
	}
	ownership, err := m.repository.ListMediaPathOwnership(ctx)
	if err != nil {
		m.observeScan(0, 0, 0, err)
		return store.OrphanScan{}, err
	}
	findings, stale, unowned := classify(paths, ownership)
	allowed, reason := safetyGate(stale, len(paths))
	completedAt := m.now().UTC()
	acquired, err = m.repository.AcquireOperationalLease(
		ctx, store.OperationalLeaseOrphanScan, m.instanceID, token, completedAt, m.leaseDuration,
	)
	if err != nil || !acquired {
		if err == nil {
			err = store.ErrOperationalLeaseLost
		}
		m.observeScan(len(paths), stale, unowned, err)
		return store.OrphanScan{}, err
	}
	id, err := newScanID(completedAt)
	if err != nil {
		m.observeScan(0, 0, 0, err)
		return store.OrphanScan{}, err
	}
	scan := store.OrphanScan{
		ID: id, InstanceID: m.instanceID, ObservedCount: len(paths),
		OwnedStaleCount: stale, UnownedCount: unowned,
		SafetyAllowed: allowed, SafetyReason: reason,
		CompletedAt: completedAt, ExpiresAt: completedAt.Add(reportTTL), Findings: findings,
	}
	if err := m.repository.SaveOrphanScan(ctx, scan, m.instanceID, token); err != nil {
		m.observeScan(len(paths), stale, unowned, err)
		return store.OrphanScan{}, err
	}
	saved = true
	m.observeScan(len(paths), stale, unowned, nil)
	return scan, nil
}

func classify(
	paths []string,
	ownership []store.MediaPathOwnership,
) ([]store.OrphanFinding, int, int) {
	history := make(map[string]store.MediaPathOwnership, len(ownership))
	for _, value := range ownership {
		history[value.PathName] = value
	}
	unique := make(map[string]struct{}, len(paths))
	for _, path := range paths {
		unique[path] = struct{}{}
	}
	ordered := make([]string, 0, len(unique))
	for path := range unique {
		ordered = append(ordered, path)
	}
	sort.Strings(ordered)
	findings := make([]store.OrphanFinding, 0)
	stale, unowned := 0, 0
	for _, path := range ordered {
		value, known := history[path]
		switch {
		case known && value.CurrentClaim:
			continue
		case known:
			stale++
			findings = append(findings, store.OrphanFinding{
				PathName: path, Classification: store.OrphanOwnedStale, DeviceID: value.DeviceID,
			})
		default:
			unowned++
			findings = append(findings, store.OrphanFinding{
				PathName: path, Classification: store.OrphanUnowned,
			})
		}
	}
	return findings, stale, unowned
}

func safetyGate(candidates, observed int) (bool, string) {
	switch {
	case candidates <= 0:
		return false, "no_candidates"
	case observed <= 0:
		return false, "empty_inventory"
	case candidates > maxCleanupItems:
		return false, "scope_exceeded"
	case candidates*100 > observed*10:
		return false, "ratio_exceeded"
	default:
		return true, "allowed"
	}
}

func (m *Manager) Apply(
	ctx context.Context,
	scanID, actorID, confirmation string,
) (CleanupResult, error) {
	result := CleanupResult{ScanID: scanID}
	if !operatorIDPattern.MatchString(actorID) || confirmation != "DELETE "+scanID {
		m.blocked("scope_changed")
		return result, ErrConfirmation
	}
	now := m.now().UTC()
	token, err := randomToken()
	if err != nil {
		return result, err
	}
	acquired, err := m.repository.AcquireOperationalLease(
		ctx, store.OperationalLeaseOrphanCleanup, m.instanceID, token, now, m.leaseDuration,
	)
	if err != nil {
		return result, err
	}
	if !acquired {
		return result, ErrLeaseHeld
	}
	defer func() {
		releaseCtx, cancel := context.WithTimeout(context.Background(), time.Second)
		defer cancel()
		_ = m.repository.ReleaseOperationalLease(
			releaseCtx, store.OperationalLeaseOrphanCleanup, m.instanceID, token, m.now().UTC(),
		)
	}()

	scan, err := m.repository.GetOrphanScan(ctx, scanID)
	if err != nil {
		m.blocked("scope_changed")
		return result, err
	}
	if !scan.ExpiresAt.After(now) {
		m.blocked("snapshot_expired")
		return result, ErrSnapshotExpired
	}
	if !scan.SafetyAllowed {
		m.blocked(scan.SafetyReason)
		return result, fmt.Errorf("%w: %s", ErrSafetyBlocked, scan.SafetyReason)
	}
	operationCtx, cancel := context.WithTimeout(ctx, m.operationTimeout)
	paths, err := m.runtime.ListPathNames(operationCtx)
	cancel()
	if err != nil {
		return result, fmt.Errorf("refresh MediaMTX path inventory: %w", err)
	}
	ownership, err := m.repository.ListMediaPathOwnership(ctx)
	if err != nil {
		return result, err
	}
	candidates := cleanupCandidates(scan, paths, ownership)
	if len(candidates) == 0 {
		return result, nil
	}
	allowed, reason := safetyGate(len(candidates), len(uniqueNames(paths)))
	if !allowed {
		m.blocked(reason)
		return result, fmt.Errorf("%w: %s", ErrSafetyBlocked, reason)
	}
	var failures []error
	for _, path := range candidates {
		now = m.now().UTC()
		acquired, err = m.repository.AcquireOperationalLease(
			ctx, store.OperationalLeaseOrphanCleanup, m.instanceID, token, now, m.leaseDuration,
		)
		if err != nil || !acquired {
			if err == nil {
				err = ErrLeaseHeld
			}
			failures = append(failures, err)
			break
		}
		itemCtx, itemCancel := context.WithTimeout(ctx, m.operationTimeout)
		deleteErr := m.runtime.DeletePath(itemCtx, path)
		itemCancel()
		status, code := "deleted", ""
		if deleteErr != nil {
			status, code = "failed", "media_error"
			result.Failed++
			failures = append(failures, fmt.Errorf("delete owned stale path: %w", deleteErr))
		} else {
			result.Deleted++
		}
		if err := m.repository.RecordOrphanCleanup(
			ctx, scanID, path, actorID, status, code, m.now().UTC(),
		); err != nil {
			failures = append(failures, err)
			break
		}
	}
	if m.metrics != nil {
		m.metrics.ObserveOrphanCleanup(result.Deleted, result.Failed)
	}
	return result, errors.Join(failures...)
}

func cleanupCandidates(
	scan store.OrphanScan,
	paths []string,
	ownership []store.MediaPathOwnership,
) []string {
	runtime := uniqueNames(paths)
	history := make(map[string]store.MediaPathOwnership, len(ownership))
	for _, value := range ownership {
		history[value.PathName] = value
	}
	values := make([]string, 0)
	for _, finding := range scan.Findings {
		if finding.Classification != store.OrphanOwnedStale || finding.Deleted {
			continue
		}
		_, present := runtime[finding.PathName]
		owner, known := history[finding.PathName]
		if present && known && !owner.CurrentClaim && owner.DeviceID == finding.DeviceID {
			values = append(values, finding.PathName)
		}
	}
	sort.Strings(values)
	return values
}

func uniqueNames(values []string) map[string]struct{} {
	result := make(map[string]struct{}, len(values))
	for _, value := range values {
		result[value] = struct{}{}
	}
	return result
}

func (m *Manager) Run(ctx context.Context, interval time.Duration, report func(error)) {
	run := func() {
		_, err := m.Report(ctx)
		if err != nil && !errors.Is(err, ErrLeaseHeld) && ctx.Err() == nil && report != nil {
			report(err)
		}
	}
	run()
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			run()
		}
	}
}

func (m *Manager) observeScan(observed, stale, unowned int, err error) {
	if m.metrics != nil {
		m.metrics.ObserveOrphanScan(observed, stale, unowned, err)
	}
}

func (m *Manager) blocked(reason string) {
	if m.metrics != nil {
		m.metrics.ObserveOrphanCleanupBlocked(reason)
	}
}

func randomToken() (string, error) {
	value := make([]byte, 16)
	if _, err := rand.Read(value); err != nil {
		return "", errors.New("generate fencing token")
	}
	const alphabet = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
	result := make([]byte, 22)
	number := new(big.Int).SetBytes(value)
	base := big.NewInt(int64(len(alphabet)))
	remainder := new(big.Int)
	for index := len(result) - 1; index >= 0; index-- {
		number.QuoRem(number, base, remainder)
		result[index] = alphabet[remainder.Int64()]
	}
	return string(result), nil
}

func newScanID(now time.Time) (string, error) {
	value := make([]byte, 16)
	milliseconds := uint64(now.UTC().UnixMilli())
	value[0], value[1], value[2] = byte(milliseconds>>40), byte(milliseconds>>32), byte(milliseconds>>24)
	value[3], value[4], value[5] = byte(milliseconds>>16), byte(milliseconds>>8), byte(milliseconds)
	if _, err := rand.Read(value[6:]); err != nil {
		return "", errors.New("generate orphan scan identifier")
	}
	number := new(big.Int).SetBytes(value)
	base := big.NewInt(32)
	remainder := new(big.Int)
	encoded := make([]byte, 26)
	for index := len(encoded) - 1; index >= 0; index-- {
		number.QuoRem(number, base, remainder)
		encoded[index] = crockford[remainder.Int64()]
	}
	return "scan_" + string(encoded), nil
}

func Confirmation(scanID string) string {
	return strings.Join([]string{"DELETE", scanID}, " ")
}