yovision/Sense/internal/reconcile/reconciler.go

// Package reconcile converges MediaMTX paths from the database desired state.
package reconcile

import (
	"context"
	"errors"
	"fmt"
	"net/url"
	"time"

	"yovision/sense/internal/device"
	"yovision/sense/internal/onvif"
	"yovision/sense/internal/store"
)

const defaultBatchSize = 128

type Repository interface {
	ListDueReconcile(ctx context.Context, now time.Time, limit int) ([]store.ReconcileCandidate, error)
	MarkReconciled(ctx context.Context, id string, generation int64, now time.Time) error
	MarkReconcileFailure(ctx context.Context, id string, failureCount int, nextAttempt time.Time, errorCode string, now time.Time) error
}

type MediaPaths interface {
	EnsurePath(ctx context.Context, name, source string) (bool, error)
	DeletePath(ctx context.Context, name string) error
}

type Reconciler struct {
	repository  Repository
	discovery   onvif.Adapter
	media       MediaPaths
	now         func() time.Time
	baseBackoff time.Duration
	maxBackoff  time.Duration
	batchSize   int
}

func New(repository Repository, discovery onvif.Adapter, media MediaPaths) *Reconciler {
	return &Reconciler{
		repository: repository, discovery: discovery, media: media,
		now: time.Now, baseBackoff: time.Second, maxBackoff: time.Minute, batchSize: defaultBatchSize,
	}
}

func (r *Reconciler) RunOnce(ctx context.Context) error {
	now := r.now().UTC()
	candidates, err := r.repository.ListDueReconcile(ctx, now, r.batchSize)
	if err != nil {
		return fmt.Errorf("list reconciliation candidates: %w", err)
	}
	var runErrors []error
	for _, candidate := range candidates {
		if err := ctx.Err(); err != nil {
			return err
		}
		if err := r.reconcileOne(ctx, candidate, now); err != nil {
			runErrors = append(runErrors, fmt.Errorf("reconcile device %s: %w", candidate.Device.ID, err))
		}
	}
	return errors.Join(runErrors...)
}

func (r *Reconciler) reconcileOne(ctx context.Context, candidate store.ReconcileCandidate, now time.Time) error {
	if candidate.Device.DesiredState == device.DesiredDisabled {
		if err := r.media.DeletePath(ctx, candidate.Device.PathName); err == nil {
			return r.repository.MarkReconciled(ctx, candidate.Device.ID, candidate.Device.Generation, now)
		} else if ctx.Err() != nil {
			return ctx.Err()
		} else {
			return r.persistFailure(ctx, candidate, now, err)
		}
	}
	result, err := r.discovery.Probe(ctx, onvif.Target{
		EndpointRef: candidate.Device.EndpointRef, CredentialRef: candidate.Device.CredentialRef,
	})
	if err == nil {
		err = validateStreamURI(result.StreamURI)
	}
	if err == nil {
		_, err = r.media.EnsurePath(ctx, candidate.Device.PathName, result.StreamURI)
	}
	if err == nil {
		return r.repository.MarkReconciled(ctx, candidate.Device.ID, candidate.Device.Generation, now)
	}
	if ctx.Err() != nil {
		return ctx.Err()
	}
	return r.persistFailure(ctx, candidate, now, err)
}

func (r *Reconciler) persistFailure(
	ctx context.Context, candidate store.ReconcileCandidate, now time.Time, err error,
) error {
	failureCount := candidate.FailureCount + 1
	nextAttempt := now.Add(r.backoff(failureCount))
	errorCode := string(onvif.CodeOf(err))
	var onvifError *onvif.Error
	if !errors.As(err, &onvifError) {
		errorCode = "media_error"
	}
	if markErr := r.repository.MarkReconcileFailure(
		ctx, candidate.Device.ID, failureCount, nextAttempt, errorCode, now,
	); markErr != nil {
		return errors.Join(err, fmt.Errorf("persist reconcile failure: %w", markErr))
	}
	return err
}

func validateStreamURI(value string) error {
	parsed, err := url.Parse(value)
	if err != nil || parsed.Host == "" || (parsed.Scheme != "rtsp" && parsed.Scheme != "rtsps") {
		return &onvif.Error{Code: onvif.ErrorInvalidReply, Err: fmt.Errorf("stream URI has invalid scheme or host")}
	}
	return nil
}

func (r *Reconciler) backoff(failureCount int) time.Duration {
	if failureCount <= 1 {
		return r.baseBackoff
	}
	value := r.baseBackoff
	for step := 1; step < failureCount; step++ {
		if value >= r.maxBackoff/2 {
			return r.maxBackoff
		}
		value *= 2
	}
	if value > r.maxBackoff {
		return r.maxBackoff
	}
	return value
}

func (r *Reconciler) Run(ctx context.Context, interval time.Duration, report func(error)) {
	if err := r.RunOnce(ctx); err != nil && ctx.Err() == nil && report != nil {
		report(err)
	}
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			if err := r.RunOnce(ctx); err != nil && ctx.Err() == nil && report != nil {
				report(err)
			}
		}
	}
}