package runtimeworker

import (
	"context"
	"database/sql"
	"errors"
	"testing"
	"time"

	eventfeature "github.com/nexuscrm/events"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/records"
)

type recordEventObservationSourceStub struct{}

func (*recordEventObservationSourceStub) ObserveWork(context.Context) (eventfeature.OutboxWorkState, error) {
	return eventfeature.OutboxWorkState{QueueDepth: 7, Failed: 2, OldestAge: 3 * time.Second}, nil
}

func TestRecordEventProcessorMeasuresAdmissionFence(t *testing.T) {
	processor := &recordEventProcessor{}
	wantErr := errors.New("fence rejected")
	fence := processor.measureAdmissionFence(func(context.Context) error {
		time.Sleep(time.Millisecond)
		return wantErr
	})
	if err := fence(t.Context()); !errors.Is(err, wantErr) {
		t.Fatalf("measured fence error = %v, want %v", err, wantErr)
	}
	diagnostics := processor.WorkDiagnostics()
	if diagnostics.admissionFenceAttempts != 1 || diagnostics.admissionFenceDuration < time.Millisecond {
		t.Fatalf("admission diagnostics = %+v", diagnostics)
	}
	worker, err := newMetadataWorker(processor, 1, time.Second, 0, 0)
	if err != nil {
		t.Fatal(err)
	}
	snapshot := worker.snapshot()
	if snapshot.AdmissionFenceAttempts != 1 || snapshot.AdmissionFenceDuration < time.Millisecond {
		t.Fatalf("worker admission snapshot = %+v", snapshot)
	}
}

func TestRecordEventProcessorReusesAndReleasesObservationSource(t *testing.T) {
	var opens, releases int
	processor := &recordEventProcessor{observationFactory: func() (recordEventObservationSource, func() error, error) {
		opens++
		return &recordEventObservationSourceStub{}, func() error {
			releases++
			return nil
		}, nil
	}}
	for range 2 {
		state, err := processor.Observe(t.Context())
		if err != nil {
			t.Fatal(err)
		}
		if state.queueDepth != 7 || state.poison != 2 || state.oldestAge != 3*time.Second {
			t.Fatalf("observation state = %+v", state)
		}
	}
	if opens != 1 {
		t.Fatalf("observation source opens = %d, want one per lifecycle", opens)
	}
	if err := processor.closeObservation(); err != nil {
		t.Fatal(err)
	}
	if releases != 1 {
		t.Fatalf("observation source releases = %d, want one", releases)
	}
	if _, err := processor.Observe(t.Context()); err != nil {
		t.Fatal(err)
	}
	if opens != 2 {
		t.Fatalf("observation source opens after close = %d, want reopened source", opens)
	}
	if err := processor.closeObservation(); err != nil {
		t.Fatal(err)
	}
}

type rollupDeltaObservationSourceStub struct {
	state records.RuntimeRollupDeltaState
}

func (s *rollupDeltaObservationSourceStub) Observe(context.Context) (records.RuntimeRollupDeltaState, error) {
	return s.state, nil
}

func testRollupDeltaOwnership() persistence.RuntimeWorkOwnership {
	return persistence.RuntimeWorkOwnership{
		WorkFamily:          records.RuntimeRollupDeltaWorkFamily,
		DeploymentID:        "rollup-observation-test",
		WorkerReleaseDigest: "sha256:dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd",
		OwnershipEpoch:      4,
	}
}

func testRollupDeltaScope() persistence.RuntimeWorkScope {
	return persistence.RuntimeWorkScope{
		OrgID: "org-test", AppInstallationID: "installation-test", RuntimeCapabilityID: "runtime.records.mutate.v1",
	}
}

func testRollupDeltaAdmissionPersistence(
	t *testing.T,
	db *sql.DB,
	owner persistence.RuntimeWorkOwnership,
	scope persistence.RuntimeWorkScope,
) *records.PersistenceService {
	t.Helper()
	ownership, err := persistence.NewRuntimeWorkOwnershipRepository(db, owner)
	if err != nil {
		t.Fatal(err)
	}
	service := &records.PersistenceService{}
	if err := service.ConfigureRuntimeRollupDeltaWorker(
		db,
		ownership,
		scope,
		func(ctx context.Context, _ records.RollupDeltaDefinitionBinding) (context.Context, error) {
			return ctx, nil
		},
		3,
	); err != nil {
		t.Fatal(err)
	}
	return service
}

func TestRollupDeltaProcessorReusesAndReleasesObservationSource(t *testing.T) {
	var opens, releases int
	processor := &rollupDeltaProcessor{observationFactory: func() (rollupDeltaObservationSource, func() error, error) {
		opens++
		return &rollupDeltaObservationSourceStub{state: records.RuntimeRollupDeltaState{
				PendingCount: 7, FailedCount: 2, OldestAge: 3 * time.Second,
			}}, func() error {
				releases++
				return nil
			}, nil
	}}
	for range 2 {
		state, err := processor.Observe(t.Context())
		if err != nil {
			t.Fatal(err)
		}
		if state.queueDepth != 7 || state.poison != 2 || state.oldestAge != 3*time.Second {
			t.Fatalf("observation state = %+v", state)
		}
	}
	if opens != 1 {
		t.Fatalf("observation source opens = %d, want one per lifecycle", opens)
	}
	if err := processor.closeObservation(); err != nil {
		t.Fatal(err)
	}
	if releases != 1 {
		t.Fatalf("observation source releases = %d, want one", releases)
	}
}