feat(sense): establish M1 offline intake skeleton

Sense/internal/reconcile/reconciler.go

@@ -0,0 +1,134 @@
+// Package reconcile converges MediaMTX paths from the database desired state.
+package reconcile
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/url"
+	"time"
+
+	"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)
+}
+
+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 {
+	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()
+	}
+	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)
+			}
+		}
+	}
+}