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

import (
	"context"
	"crypto/rand"
	"encoding/base64"
	"errors"
	"fmt"
	"net/url"
	"time"

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

const defaultBatchSize = 128

type Repository interface {
	ClaimDueReconcile(context.Context, store.ReconcileClaim) ([]store.ReconcileCandidate, error)
	RenewReconcileLease(context.Context, string, string, string, time.Time, time.Duration) (bool, error)
	CompleteReconcile(context.Context, string, int64, string, string, time.Time) error
	FailReconcile(context.Context, string, int, time.Time, string, string, string, time.Time) error
	ConvergenceSnapshot(context.Context) (store.ConvergenceSnapshot, 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
	instanceID       string
	leaseDuration    time.Duration
	operationTimeout time.Duration
	metrics          *metrics.Registry
}

type Options struct {
	InstanceID       string
	LeaseDuration    time.Duration
	OperationTimeout time.Duration
	Metrics          *metrics.Registry
}

func New(repository Repository, discovery onvif.Adapter, media MediaPaths) *Reconciler {
	return NewWithOptions(repository, discovery, media, Options{})
}

func NewWithOptions(
	repository Repository,
	discovery onvif.Adapter,
	media MediaPaths,
	options Options,
) *Reconciler {
	if options.InstanceID == "" {
		options.InstanceID = "single"
	}
	if options.LeaseDuration <= 0 {
		options.LeaseDuration = 30 * time.Second
	}
	if options.OperationTimeout <= 0 {
		options.OperationTimeout = 20 * time.Second
	}
	if options.OperationTimeout >= options.LeaseDuration {
		options.OperationTimeout = options.LeaseDuration / 2
	}
	return &Reconciler{
		repository: repository, discovery: discovery, media: media,
		now: time.Now, baseBackoff: time.Second, maxBackoff: time.Minute, batchSize: defaultBatchSize,
		instanceID: options.InstanceID, leaseDuration: options.LeaseDuration,
		operationTimeout: options.OperationTimeout, metrics: options.Metrics,
	}
}

func (r *Reconciler) RunOnce(ctx context.Context) (runErr error) {
	started := time.Now()
	defer func() {
		if r.metrics != nil {
			r.metrics.ObserveReconcileRun(runErr, time.Since(started))
		}
	}()
	now := r.now().UTC()
	token, err := newClaimToken()
	if err != nil {
		return err
	}
	candidates, err := r.repository.ClaimDueReconcile(ctx, store.ReconcileClaim{
		Owner: r.instanceID, Token: token, Now: now,
		LeaseDuration: r.leaseDuration, Limit: r.batchSize,
	})
	if err != nil {
		return fmt.Errorf("claim reconciliation candidates: %w", err)
	}
	var runErrors []error
	for _, candidate := range candidates {
		if err := ctx.Err(); err != nil {
			return err
		}
		itemNow := r.now().UTC()
		renewed, err := r.repository.RenewReconcileLease(
			ctx, candidate.Device.ID, r.instanceID, token, itemNow, r.leaseDuration,
		)
		if err != nil {
			r.observeItem("error")
			runErrors = append(runErrors, fmt.Errorf("renew reconcile device lease: %w", err))
			continue
		}
		if !renewed {
			r.observeItem("lease_lost")
			continue
		}
		itemCtx, cancel := context.WithTimeout(ctx, r.operationTimeout)
		err = r.reconcileOne(itemCtx, ctx, candidate, itemNow, r.instanceID, token)
		cancel()
		if errors.Is(err, store.ErrReconcileLeaseLost) {
			r.observeItem("lease_lost")
			continue
		}
		if err != nil {
			r.observeItem("error")
			runErrors = append(runErrors, fmt.Errorf("reconcile device %s: %w", candidate.Device.ID, err))
		} else {
			r.observeItem("success")
		}
	}
	if r.metrics != nil {
		snapshot, err := r.repository.ConvergenceSnapshot(ctx)
		if err != nil {
			runErrors = append(runErrors, fmt.Errorf("read convergence metrics: %w", err))
		} else {
			r.metrics.SetConvergence(snapshot.Total, snapshot.Unconverged)
		}
	}
	return errors.Join(runErrors...)
}

func (r *Reconciler) reconcileOne(
	ctx context.Context,
	persistCtx context.Context,
	candidate store.ReconcileCandidate,
	now time.Time,
	owner, token string,
) error {
	if candidate.Device.DesiredState == device.DesiredDisabled {
		if err := r.media.DeletePath(ctx, candidate.Device.PathName); err == nil {
			return r.repository.CompleteReconcile(
				persistCtx, candidate.Device.ID, candidate.Device.Generation, owner, token, now,
			)
		} else if persistCtx.Err() != nil {
			return persistCtx.Err()
		} else {
			return r.persistFailure(persistCtx, candidate, now, owner, token, 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.CompleteReconcile(
			persistCtx, candidate.Device.ID, candidate.Device.Generation, owner, token, now,
		)
	}
	if persistCtx.Err() != nil {
		return persistCtx.Err()
	}
	return r.persistFailure(persistCtx, candidate, now, owner, token, err)
}

func (r *Reconciler) persistFailure(
	ctx context.Context,
	candidate store.ReconcileCandidate,
	now time.Time,
	owner, token string,
	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.FailReconcile(
		ctx, candidate.Device.ID, failureCount, nextAttempt, errorCode, owner, token, now,
	); markErr != nil {
		return errors.Join(err, fmt.Errorf("persist reconcile failure: %w", markErr))
	}
	return err
}

func (r *Reconciler) observeItem(result string) {
	if r.metrics != nil {
		r.metrics.ObserveReconcileItem(result)
	}
}

func newClaimToken() (string, error) {
	value := make([]byte, 18)
	if _, err := rand.Read(value); err != nil {
		return "", errors.New("generate reconcile claim token")
	}
	return base64.RawURLEncoding.EncodeToString(value), nil
}

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)
			}
		}
	}
}