package media

import (
	"context"
	"crypto/sha256"
	"errors"
	"fmt"
	"strings"
	"sync"
	"time"

	"gorm.io/gorm"
	"gorm.io/gorm/clause"

	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/credential"
	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/reconcile"
)

var (
	ErrNotFound           = errors.New("媒体路由不存在")
	ErrRuntimeUnavailable = errors.New("视频服务运行配置不可用")
)

type Service struct {
	db         *gorm.DB
	controller Controller
	process    *Supervisor
	config     RuntimeConfig
	now        func() time.Time
	mu         sync.Mutex
}

func NewService(db *gorm.DB, controller Controller, process *Supervisor, config RuntimeConfig) *Service {
	return &Service{db: db, controller: controller, process: process, config: config, now: time.Now}
}

func (s *Service) EnsureAllVerified(ctx context.Context) error {
	var ids []string
	if err := s.db.WithContext(ctx).Model(&admissionProfile{}).Where("verification_status = ?", "ready").Distinct().Pluck("device_id", &ids).Error; err != nil {
		return err
	}
	for _, id := range ids {
		if err := s.ensureDevice(ctx, id, false); err != nil {
			return err
		}
	}
	return nil
}

func (s *Service) EnsureDevice(ctx context.Context, deviceID string) error {
	return s.ensureDevice(ctx, deviceID, true)
}

func (s *Service) ensureDevice(ctx context.Context, deviceID string, reactivate bool) error {
	if strings.TrimSpace(deviceID) == "" {
		return errors.New("device id is required")
	}
	var profiles []admissionProfile
	if err := s.db.WithContext(ctx).Where("device_id = ? AND verification_status = ?", deviceID, "ready").Find(&profiles).Error; err != nil {
		return err
	}
	now := s.now().UTC()
	for _, profile := range profiles {
		id := profile.DeviceID + ":" + profile.Token
		digest := sha256.Sum256([]byte(id))
		route := Route{ID: id, DeviceID: profile.DeviceID, ProfileToken: profile.Token, Path: fmt.Sprintf("sense_%x", digest[:12]), Desired: DesiredRunning, Actual: "pending", Detail: "等待视频服务对账", Version: 1, UpdatedAt: now}
		if err := s.db.WithContext(ctx).Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "id"}}, DoNothing: true}).Create(&route).Error; err != nil {
			return err
		}
		var existing Route
		if err := s.db.WithContext(ctx).First(&existing, "id = ?", id).Error; err != nil {
			return err
		}
		if reactivate && existing.Desired != DesiredRunning {
			if err := s.db.WithContext(ctx).Model(&existing).Updates(map[string]any{"desired": DesiredRunning, "actual": "pending", "detail": "等待视频服务对账", "next_retry_at": nil, "version": existing.Version + 1, "updated_at": now}).Error; err != nil {
				return err
			}
		}
	}
	return nil
}

func (s *Service) Run(ctx context.Context) {
	_ = s.EnsureAllVerified(ctx)
	_ = s.ReconcileDue(ctx)
	interval := s.config.PollInterval
	if interval <= 0 {
		interval = 5 * time.Second
	}
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			_ = s.ReconcileDue(ctx)
		}
	}
}

func (s *Service) ReconcileDue(ctx context.Context) error {
	now := s.now().UTC()
	var routes []Route
	if err := s.db.WithContext(ctx).Where("desired = ? AND (next_retry_at IS NULL OR next_retry_at <= ?)", DesiredRunning, now).Order("updated_at ASC").Limit(128).Find(&routes).Error; err != nil {
		return err
	}
	var result error
	for _, route := range routes {
		var err error
		if route.FailureCount == 0 && (route.Actual == "waiting" || route.Actual == "ready") {
			_, err = s.Refresh(ctx, route.ID)
		} else {
			_, err = s.Reconcile(ctx, route.ID)
		}
		if err != nil {
			result = errors.Join(result, err)
		}
	}
	return result
}

func (s *Service) Refresh(ctx context.Context, id string) (RouteResponse, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	var route Route
	if err := s.db.WithContext(ctx).First(&route, "id = ?", id).Error; err != nil {
		return RouteResponse{}, ErrNotFound
	}
	if err := s.ensureControl(ctx); err != nil {
		return s.saveFailure(ctx, route, "process_unavailable", "MediaMTX 未启动或 Control API 未就绪", err)
	}
	status, err := s.controller.Status(ctx, route.Path)
	if err != nil {
		return s.saveFailure(ctx, route, "status_unavailable", "尚未取得媒体路径状态", err)
	}
	if !status.Exists {
		// A cold MediaMTX start begins with paths: {}; re-apply the route
		// after releasing the service lock.
		s.mu.Unlock()
		response, reconcileErr := s.Reconcile(ctx, id)
		s.mu.Lock()
		return response, reconcileErr
	}
	if status.Ready {
		return s.saveSuccess(ctx, route, "ready", true, status.Readers, "上游拉流正常")
	}
	return s.saveSuccess(ctx, route, "waiting", false, status.Readers, "等待播放器连接并按需拉流")
}

func (s *Service) Reconcile(ctx context.Context, id string) (RouteResponse, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	var route Route
	if err := s.db.WithContext(ctx).First(&route, "id = ?", id).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return RouteResponse{}, ErrNotFound
		}
		return RouteResponse{}, err
	}
	if route.Desired == DesiredStopped {
		if s.controller != nil {
			_ = s.controller.Delete(ctx, route.Path)
		}
		return s.saveSuccess(ctx, route, "stopped", false, 0, "媒体路径已停止")
	}
	if err := s.ensureControl(ctx); err != nil {
		return s.saveFailure(ctx, route, "process_unavailable", "MediaMTX 未启动或 Control API 未就绪", err)
	}
	var profile admissionProfile
	if err := s.db.WithContext(ctx).Where("device_id = ? AND token = ? AND verification_status = ?", route.DeviceID, route.ProfileToken, "ready").First(&profile).Error; err != nil {
		return s.saveFailure(ctx, route, "profile_unavailable", "已验证 Profile 不可用", err)
	}
	value, err := credential.Read(s.db.WithContext(ctx), route.DeviceID, credential.PurposeRTSP)
	if err != nil {
		return s.saveFailure(ctx, route, "credential_unavailable", "RTSP 凭据不可用", err)
	}
	if err = s.controller.Apply(ctx, Source{Path: route.Path, URI: profile.StreamURI, Username: value.Username, Password: value.Password}); err != nil {
		return s.saveFailure(ctx, route, "apply_failed", "媒体路径配置失败", err)
	}
	status, err := s.controller.Status(ctx, route.Path)
	if err != nil {
		return s.saveFailure(ctx, route, "status_unavailable", "尚未取得媒体路径状态", err)
	}
	if !status.Exists {
		return s.saveFailure(ctx, route, "path_missing", "媒体路径不存在", errors.New("MediaMTX path missing after apply"))
	}
	if status.Ready {
		return s.saveSuccess(ctx, route, "ready", true, status.Readers, "上游拉流正常")
	}
	return s.saveSuccess(ctx, route, "waiting", false, status.Readers, "等待播放器连接并按需拉流")
}

func (s *Service) ensureControl(ctx context.Context) error {
	if s.controller == nil {
		return ErrRuntimeUnavailable
	}
	if err := s.controller.Health(ctx); err == nil {
		if s.process != nil && !s.process.State().Owned {
			s.process.MarkExternal()
		}
		return nil
	}
	if s.process == nil {
		return ErrRuntimeUnavailable
	}
	if err := EnsureBaseConfig(s.config.ConfigPath, s.config.APIBase); err != nil {
		s.process.MarkFailed("MediaMTX 基础配置不可用")
		return err
	}
	if err := s.process.Start(ctx); err != nil {
		return err
	}
	deadline := s.now().Add(s.config.StartTimeout)
	for s.now().Before(deadline) {
		if err := s.controller.Health(ctx); err == nil {
			s.process.MarkReady()
			return nil
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(100 * time.Millisecond):
		}
	}
	s.process.MarkFailed("MediaMTX Control API 就绪超时")
	return errors.New("MediaMTX readiness timeout")
}

func (s *Service) saveSuccess(ctx context.Context, route Route, actual string, sourceReady bool, readers int, detail string) (RouteResponse, error) {
	if route.Actual == actual && route.SourceReady == sourceReady && route.Readers == readers && route.FailureCount == 0 && route.NextRetryAt == nil && route.LastErrorCode == "" && route.Detail == detail {
		return routeResponse(route), nil
	}
	now := s.now().UTC()
	updates := map[string]any{"actual": actual, "source_ready": sourceReady, "readers": readers, "failure_count": 0, "next_retry_at": nil, "last_error_code": "", "detail": detail, "version": route.Version + 1, "updated_at": now}
	if err := s.db.WithContext(ctx).Model(&route).Updates(updates).Error; err != nil {
		return RouteResponse{}, err
	}
	for key, value := range updates {
		switch key {
		case "actual":
			route.Actual = value.(string)
		case "source_ready":
			route.SourceReady = value.(bool)
		case "readers":
			route.Readers = value.(int)
		case "failure_count":
			route.FailureCount = value.(int)
		case "last_error_code":
			route.LastErrorCode = value.(string)
		case "detail":
			route.Detail = value.(string)
		case "version":
			route.Version = value.(int64)
		case "updated_at":
			route.UpdatedAt = value.(time.Time)
		}
	}
	route.NextRetryAt = nil
	return routeResponse(route), nil
}

func (s *Service) saveFailure(ctx context.Context, route Route, code, detail string, cause error) (RouteResponse, error) {
	now := s.now().UTC()
	failures := route.FailureCount + 1
	next := now.Add(reconcile.Backoff(failures))
	updates := map[string]any{"actual": code, "source_ready": false, "readers": 0, "failure_count": failures, "next_retry_at": &next, "last_error_code": code, "detail": detail, "version": route.Version + 1, "updated_at": now}
	if err := s.db.WithContext(ctx).Model(&route).Updates(updates).Error; err != nil {
		return RouteResponse{}, errors.Join(cause, err)
	}
	route.Actual, route.SourceReady, route.Readers, route.FailureCount, route.NextRetryAt, route.LastErrorCode, route.Detail = code, false, 0, failures, &next, code, detail
	route.Version, route.UpdatedAt = route.Version+1, now
	return routeResponse(route), cause
}

func (s *Service) StopRoute(ctx context.Context, id string) (RouteResponse, error) {
	var route Route
	if err := s.db.WithContext(ctx).First(&route, "id = ?", id).Error; err != nil {
		return RouteResponse{}, ErrNotFound
	}
	route.Desired = DesiredStopped
	if err := s.db.WithContext(ctx).Model(&route).Updates(map[string]any{"desired": DesiredStopped, "next_retry_at": nil, "version": route.Version + 1, "updated_at": s.now().UTC()}).Error; err != nil {
		return RouteResponse{}, err
	}
	return s.Reconcile(ctx, id)
}

func (s *Service) List(ctx context.Context) ([]RouteResponse, error) {
	var routes []Route
	if err := s.db.WithContext(ctx).Order("updated_at DESC").Find(&routes).Error; err != nil {
		return nil, err
	}
	items := make([]RouteResponse, 0, len(routes))
	for _, route := range routes {
		items = append(items, routeResponse(route))
	}
	return items, nil
}

func (s *Service) ProcessState() ProcessState {
	if s.process == nil {
		return ProcessState{Phase: "configuration_failed", Detail: "视频服务运行配置不可用", LastChanged: s.now().UTC()}
	}
	return s.process.State()
}