package runtimeworker

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"

	recordmutation "github.com/nexuscrm/backend/internal/runtimeassembly/recordmutation"
	database "github.com/nexuscrm/db"
	eventfeature "github.com/nexuscrm/events"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/records"
)

type recordEventObservationSource interface {
	ObserveWork(context.Context) (eventfeature.OutboxWorkState, error)
}

type recordEventObservationFactory func() (recordEventObservationSource, func() error, error)

type recordEventProcessor struct {
	events             *eventfeature.OutboxService
	retention          time.Duration
	observationFactory recordEventObservationFactory

	observationMu      sync.Mutex
	observationSource  recordEventObservationSource
	releaseObservation func() error

	admissionFenceAttempts atomic.Uint64
	admissionFenceNanos    atomic.Uint64
}

func (p *recordEventProcessor) measureAdmissionFence(fence func(context.Context) error) func(context.Context) error {
	return func(ctx context.Context) error {
		started := time.Now()
		err := fence(ctx)
		p.admissionFenceAttempts.Add(1)
		p.admissionFenceNanos.Add(uint64(time.Since(started)))
		return err
	}
}

func (p *recordEventProcessor) WorkDiagnostics() metadataWorkDiagnostics {
	if p == nil {
		return metadataWorkDiagnostics{}
	}
	return metadataWorkDiagnostics{
		admissionFenceAttempts: p.admissionFenceAttempts.Load(),
		admissionFenceDuration: time.Duration(p.admissionFenceNanos.Load()),
	}
}

func (p *recordEventProcessor) CheckReady(ctx context.Context) error {
	if p == nil || p.events == nil {
		return errors.New("runtime record-event worker: processor is unavailable")
	}
	return p.events.CheckDurableRecordEventReady(ctx)
}

func (p *recordEventProcessor) Observe(ctx context.Context) (metadataWorkState, error) {
	if p == nil || p.observationFactory == nil {
		return metadataWorkState{}, errors.New("runtime record-event worker: observation source is unavailable")
	}
	p.observationMu.Lock()
	defer p.observationMu.Unlock()
	if p.observationSource == nil {
		source, release, err := p.observationFactory()
		if err != nil {
			return metadataWorkState{}, fmt.Errorf("runtime record-event worker: open observation source: %w", err)
		}
		if source == nil || release == nil {
			if release != nil {
				_ = release()
			}
			return metadataWorkState{}, errors.New("runtime record-event worker: observation source is incomplete")
		}
		p.observationSource = source
		p.releaseObservation = release
	}
	state, err := p.observationSource.ObserveWork(ctx)
	return metadataWorkState{queueDepth: state.QueueDepth, poison: state.Failed, oldestAge: state.OldestAge}, err
}

func (p *recordEventProcessor) closeObservation() error {
	if p == nil {
		return nil
	}
	p.observationMu.Lock()
	defer p.observationMu.Unlock()
	release := p.releaseObservation
	p.observationSource = nil
	p.releaseObservation = nil
	if release == nil {
		return nil
	}
	return release()
}

func newRecordEventObservationFactory(
	openIndependentDB func(database.PoolOptions) (*sql.DB, error),
	scope persistence.RuntimeWorkScope,
	claimLease time.Duration,
	trackers ...*workerAuxiliaryPoolTracker,
) recordEventObservationFactory {
	return func() (recordEventObservationSource, func() error, error) {
		if openIndependentDB == nil {
			return nil, nil, errors.New("independent database opener is unavailable")
		}
		observationDB, release, err := openWorkerObservationPool(
			openIndependentDB, optionalWorkerAuxiliaryPoolTracker(trackers),
		)
		if err != nil {
			return nil, nil, err
		}
		repository, err := persistence.NewScopedRecordEventOutboxRepository(observationDB, scope, claimLease)
		if err != nil {
			_ = release()
			return nil, nil, err
		}
		return eventfeature.NewOutboxService(repository, nil, nil), release, nil
	}
}

func (p *recordEventProcessor) HasWork(ctx context.Context) (bool, error) {
	if p == nil || p.events == nil {
		return false, errors.New("runtime record-event worker: processor is unavailable")
	}
	return p.events.HasWork(ctx)
}

func (p *recordEventProcessor) Maintain(ctx context.Context, limit int) error {
	if p == nil || p.events == nil || p.retention < time.Second {
		return errors.New("runtime record-event worker: bounded processed-row retention is unavailable")
	}
	_, err := p.events.CleanupProcessedBatch(ctx, p.retention, limit)
	return err
}

func (p *recordEventProcessor) ProcessOne(ctx context.Context, _ int) (int, bool, error) {
	found, err := p.events.ProcessNext(ctx)
	if err != nil || !found {
		return 0, found, err
	}
	return 1, true, nil
}

func (p *recordEventProcessor) ClaimBatch(ctx context.Context, limit int) ([]metadataClaimedWork, error) {
	if p == nil || p.events == nil {
		return nil, errors.New("runtime record-event worker: processor is unavailable")
	}
	claimed, err := p.events.ClaimDurableRecordEventBatch(ctx, limit)
	if err != nil {
		return nil, err
	}
	work := make([]metadataClaimedWork, 0, len(claimed))
	for _, item := range claimed {
		item := item
		work = append(work, metadataClaimedWork{run: func(effectCtx context.Context) error {
			return p.events.ProcessDurableRecordEventClaim(effectCtx, item)
		}})
	}
	return work, nil
}

type rollupDeltaProcessor struct {
	runtime            *recordmutation.RecordsMutationRuntime
	observationFactory rollupDeltaObservationFactory

	observationMu      sync.Mutex
	observationSource  rollupDeltaObservationSource
	releaseObservation func() error
}

type rollupDeltaObservationSource interface {
	Observe(context.Context) (records.RuntimeRollupDeltaState, error)
}

type rollupDeltaObservationFactory func() (rollupDeltaObservationSource, func() error, error)

func (p *rollupDeltaProcessor) CheckReady(ctx context.Context) error {
	if p == nil || p.runtime == nil || p.runtime.Persistence == nil || p.runtime.Rollup == nil {
		return errors.New("runtime rollup-delta worker: processor is unavailable")
	}
	return p.runtime.Persistence.CheckRuntimeRollupDeltaReady(ctx)
}

func (p *rollupDeltaProcessor) Observe(ctx context.Context) (metadataWorkState, error) {
	if p == nil || p.observationFactory == nil {
		return metadataWorkState{}, errors.New("runtime rollup-delta worker: observation source is unavailable")
	}
	p.observationMu.Lock()
	defer p.observationMu.Unlock()
	if p.observationSource == nil {
		source, release, err := p.observationFactory()
		if err != nil {
			return metadataWorkState{}, fmt.Errorf("runtime rollup-delta worker: open observation source: %w", err)
		}
		if source == nil || release == nil {
			if release != nil {
				_ = release()
			}
			return metadataWorkState{}, errors.New("runtime rollup-delta worker: observation source is incomplete")
		}
		p.observationSource = source
		p.releaseObservation = release
	}
	state, err := p.observationSource.Observe(ctx)
	return metadataWorkState{queueDepth: state.PendingCount, poison: state.FailedCount, oldestAge: state.OldestAge}, err
}

func (p *rollupDeltaProcessor) closeObservation() error {
	if p == nil {
		return nil
	}
	p.observationMu.Lock()
	defer p.observationMu.Unlock()
	release := p.releaseObservation
	p.observationSource = nil
	p.releaseObservation = nil
	if release == nil {
		return nil
	}
	return release()
}

func newRollupDeltaObservationFactory(
	openIndependentDB func(database.PoolOptions) (*sql.DB, error),
	expected persistence.RuntimeWorkOwnership,
	scope persistence.RuntimeWorkScope,
	trackers ...*workerAuxiliaryPoolTracker,
) rollupDeltaObservationFactory {
	return func() (rollupDeltaObservationSource, func() error, error) {
		if openIndependentDB == nil {
			return nil, nil, errors.New("independent database opener is unavailable")
		}
		observationDB, release, err := openWorkerObservationPool(
			openIndependentDB, optionalWorkerAuxiliaryPoolTracker(trackers),
		)
		if err != nil {
			return nil, nil, err
		}
		store, err := records.NewRuntimeRollupDeltaObservationStore(observationDB, expected, scope)
		if err != nil {
			_ = release()
			return nil, nil, err
		}
		return store, release, nil
	}
}

// HasWork uses the bounded business-store hint, never the hourly diagnostic
// aggregate. The metadata worker fails open on hint errors, and ProcessOne
// remains the fenced claim authority for every positive result.
func (p *rollupDeltaProcessor) HasWork(ctx context.Context) (bool, error) {
	if p == nil || p.runtime == nil || p.runtime.Persistence == nil {
		return false, errors.New("runtime rollup-delta worker: processor is unavailable")
	}
	return p.runtime.Persistence.HasRuntimeRollupDeltaWork(ctx)
}

func (*rollupDeltaProcessor) Maintain(context.Context, int) error { return nil }

func (p *rollupDeltaProcessor) ProcessOne(ctx context.Context, limit int) (int, bool, error) {
	if p == nil || p.runtime == nil || p.runtime.Persistence == nil || p.runtime.Rollup == nil {
		return 0, false, errors.New("runtime rollup-delta worker: processor is unavailable")
	}
	processed, err := p.runtime.Persistence.ProcessRollupDeltaBatch(ctx, limit, p.runtime.Rollup.RecalculateCountRollupValues)
	return processed, processed > 0, err
}