package connector_delivery

import (
	"context"
	"encoding/json"
	"errors"
	"sync"
	"testing"
	"time"

	connector_oauth "github.com/pingcap-inc/nexuscrm-connectors/features/connector/oauth"
)

func TestDispatchCallsProviderOnceAndReplaysAcceptedResult(t *testing.T) {
	repository := newMemoryRepository()
	provider := &fakeProvider{result: DispatchResult{
		Outcome: DispatchAccepted, RemoteReceiptID: "message-1",
		RemoteConversationID: "thread-1", ResponseStatus: 200,
	}}
	service := newTestService(t, repository, provider)
	request := testRequest()

	first, err := service.Dispatch(context.Background(), request)
	if err != nil {
		t.Fatal(err)
	}
	second, err := service.Dispatch(context.Background(), request)
	if err != nil {
		t.Fatal(err)
	}
	if first.State != StateAccepted || second.State != StateAccepted ||
		second.RemoteReceiptID != "message-1" || provider.callCount() != 1 {
		t.Fatalf("first=%#v second=%#v calls=%d", first, second, provider.callCount())
	}
	completed, err := service.Complete(context.Background(), request.RequestID, request.UserID)
	if err != nil || completed.State != StateCompleted {
		t.Fatalf("Complete() = %#v, %v", completed, err)
	}
	replayed, err := service.Dispatch(context.Background(), request)
	if err != nil || replayed.State != StateCompleted || provider.callCount() != 1 {
		t.Fatalf("completed replay = %#v, %v; calls=%d", replayed, err, provider.callCount())
	}
}

func TestLookupReturnsFrozenAcceptedAndCompletedResultsWithoutProviderCall(t *testing.T) {
	repository := newMemoryRepository()
	provider := &fakeProvider{result: DispatchResult{
		Outcome: DispatchAccepted, RemoteReceiptID: "message-1",
		RemoteConversationID: "thread-1", ResponseStatus: 200,
	}}
	service := newTestService(t, repository, provider)
	request := testRequest()
	accepted, err := service.Dispatch(context.Background(), request)
	if err != nil {
		t.Fatal(err)
	}

	lookedUp, err := service.Lookup(context.Background(), "  "+request.RequestID+"  ", "  "+request.UserID+"  ")
	if err != nil || lookedUp.State != StateAccepted || lookedUp.RemoteReceiptID != accepted.RemoteReceiptID ||
		lookedUp.RemoteConversationID != accepted.RemoteConversationID ||
		!sameJSON(lookedUp.CompletionSnapshot, request.CompletionSnapshot) || provider.callCount() != 1 {
		t.Fatalf("Lookup() = %#v, %v; provider calls=%d", lookedUp, err, provider.callCount())
	}
	if _, err := service.Complete(context.Background(), request.RequestID, request.UserID); err != nil {
		t.Fatal(err)
	}
	completed, err := service.Lookup(context.Background(), request.RequestID, request.UserID)
	if err != nil || completed.State != StateCompleted || provider.callCount() != 1 {
		t.Fatalf("completed Lookup() = %#v, %v; provider calls=%d", completed, err, provider.callCount())
	}
}

func TestLookupHidesForeignAndMissingRequests(t *testing.T) {
	repository := newMemoryRepository()
	service := newTestService(t, repository, &fakeProvider{})
	request, payload, err := normalizeRequest(testRequest())
	if err != nil {
		t.Fatal(err)
	}
	repository.jobs[request.RequestID] = Job{
		RequestID: request.RequestID, ConnectedPrincipalID: "principal-1",
		PrincipalGeneration: 7, OriginatingUserID: request.UserID,
		State: StateReserved, PayloadSHA256: payload.SHA256, PayloadEncodedSize: payload.EncodedSize,
		SourceObject: request.SourceObject, SourceRecordID: request.SourceRecordID,
		CompletionSnapshot: request.CompletionSnapshot,
	}

	foreign, foreignErr := service.Lookup(context.Background(), request.RequestID, "user-2")
	missing, missingErr := service.Lookup(context.Background(), "missing-request", "user-2")
	if !errors.Is(foreignErr, ErrDeliveryNotFound) || !errors.Is(missingErr, ErrDeliveryNotFound) ||
		foreign.RequestID != "" || len(foreign.CompletionSnapshot) != 0 ||
		missing.RequestID != "" || len(missing.CompletionSnapshot) != 0 {
		t.Fatalf("foreign=%#v/%v missing=%#v/%v", foreign, foreignErr, missing, missingErr)
	}
}

func TestLookupRejectsInvalidInput(t *testing.T) {
	service := newTestService(t, newMemoryRepository(), &fakeProvider{})
	for _, input := range [][2]string{{"", "user-1"}, {"request-1", "  "}} {
		if _, err := service.Lookup(context.Background(), input[0], input[1]); !errors.Is(err, ErrInvalidInput) {
			t.Fatalf("Lookup(%q, %q) error = %v", input[0], input[1], err)
		}
	}
}

func TestCompleteRejectsDuplicateKeyDurableSnapshot(t *testing.T) {
	repository := newMemoryRepository()
	provider := &fakeProvider{result: DispatchResult{
		Outcome: DispatchAccepted, RemoteReceiptID: "message-1", ResponseStatus: 200,
	}}
	service := newTestService(t, repository, provider)
	request := testRequest()
	if _, err := service.Dispatch(context.Background(), request); err != nil {
		t.Fatal(err)
	}
	repository.mu.Lock()
	job := repository.jobs[request.RequestID]
	job.CompletionSnapshot = json.RawMessage(`{"taskId":"old","taskId":"task-1","emailMessageId":"email-1"}`)
	repository.jobs[request.RequestID] = job
	repository.mu.Unlock()

	result, err := service.Complete(context.Background(), request.RequestID, request.UserID)
	if !errors.Is(err, ErrStateInvalid) || result.RequestID != "" || len(result.CompletionSnapshot) != 0 {
		t.Fatalf("Complete() = %#v, %v", result, err)
	}
	repository.mu.Lock()
	state := repository.jobs[request.RequestID].State
	repository.mu.Unlock()
	if state != StateAccepted || provider.callCount() != 1 {
		t.Fatalf("state=%q provider calls=%d", state, provider.callCount())
	}
}

func TestConcurrentDuplicateReturnsInProgressWithoutSecondProviderCall(t *testing.T) {
	repository := newMemoryRepository()
	entered := make(chan struct{})
	release := make(chan struct{})
	provider := &fakeProvider{
		entered: entered, release: release,
		result: DispatchResult{Outcome: DispatchAccepted, RemoteReceiptID: "message-1", ResponseStatus: 200},
	}
	service := newTestService(t, repository, provider)
	request := testRequest()
	firstDone := make(chan error, 1)
	go func() {
		_, err := service.Dispatch(context.Background(), request)
		firstDone <- err
	}()
	<-entered

	result, err := service.Dispatch(context.Background(), request)
	if !errors.Is(err, ErrDispatchInProgress) || result.State != StateDispatching {
		t.Fatalf("duplicate Dispatch() = %#v, %v", result, err)
	}
	close(release)
	if err := <-firstDone; err != nil {
		t.Fatal(err)
	}
	if provider.callCount() != 1 {
		t.Fatalf("provider calls = %d, want 1", provider.callCount())
	}
}

func TestExpiredDispatchBecomesUncertainAndNeverCallsProvider(t *testing.T) {
	repository := newMemoryRepository()
	provider := &fakeProvider{}
	service := newTestService(t, repository, provider)
	now := service.now()
	request := testRequest()
	payloadRequest, payload, err := normalizeRequest(request)
	if err != nil {
		t.Fatal(err)
	}
	repository.jobs[request.RequestID] = Job{
		RequestID: request.RequestID, ConnectedPrincipalID: "principal-1",
		PrincipalGeneration: 7, OriginatingUserID: request.UserID,
		State: StateDispatching, PayloadSHA256: payload.SHA256,
		PayloadEncodedSize: payload.EncodedSize, SourceObject: request.SourceObject,
		SourceRecordID: request.SourceRecordID, CompletionSnapshot: payloadRequest.CompletionSnapshot,
		ClaimToken: "dead-claim", Fence: 1,
		LeaseExpiresAt:    ptrTime(now.Add(-time.Second)),
		DispatchStartedAt: ptrTime(now.Add(-time.Minute)),
	}
	repository.payloads[request.RequestID] = payload

	result, err := service.Dispatch(context.Background(), request)
	if !errors.Is(err, ErrDeliveryUncertain) || result.State != StateUncertain ||
		result.FailureCode != "dispatch_lease_expired" {
		t.Fatalf("Dispatch() = %#v, %v", result, err)
	}
	if provider.callCount() != 0 {
		t.Fatalf("provider calls = %d, want 0", provider.callCount())
	}
}

func TestAcceptedProviderWithPersistenceFailureCannotBeRedispatched(t *testing.T) {
	repository := newMemoryRepository()
	repository.failFinish = true
	provider := &fakeProvider{result: DispatchResult{
		Outcome: DispatchAccepted, RemoteReceiptID: "message-1", ResponseStatus: 200,
	}}
	service := newTestService(t, repository, provider)
	clock := service.now()
	service.now = func() time.Time { return clock }
	request := testRequest()

	_, err := service.Dispatch(context.Background(), request)
	if !errors.Is(err, ErrDeliveryUncertain) {
		t.Fatalf("first Dispatch() error = %v", err)
	}
	if _, err := service.Dispatch(context.Background(), request); !errors.Is(err, ErrDispatchInProgress) {
		t.Fatalf("active replay error = %v", err)
	}
	clock = clock.Add(DefaultDispatchLease + time.Second)
	result, err := service.Dispatch(context.Background(), request)
	if !errors.Is(err, ErrDeliveryUncertain) || result.State != StateUncertain {
		t.Fatalf("expired replay = %#v, %v", result, err)
	}
	if provider.callCount() != 1 {
		t.Fatalf("provider calls = %d, want 1", provider.callCount())
	}
}

func TestReconnectCannotRetargetReservedRequest(t *testing.T) {
	repository := newMemoryRepository()
	provider := &fakeProvider{}
	auth := &fakeOAuth{principal: connector_oauth.ConnectedPrincipal{
		ID: "principal-1", UserID: "user-1", ConnectorKey: "gmail",
		Generation: 8, State: connector_oauth.PrincipalConnected,
	}}
	service, err := newService(repository, auth, staticProviderResolver{provider: provider})
	if err != nil {
		t.Fatal(err)
	}
	request, payload, _ := normalizeRequest(testRequest())
	repository.jobs[request.RequestID] = Job{
		RequestID: request.RequestID, ConnectedPrincipalID: "principal-1",
		PrincipalGeneration: 7, OriginatingUserID: request.UserID, State: StateReserved,
		PayloadSHA256: payload.SHA256, PayloadEncodedSize: payload.EncodedSize,
		SourceObject: request.SourceObject, SourceRecordID: request.SourceRecordID,
		CompletionSnapshot: request.CompletionSnapshot,
	}
	repository.payloads[request.RequestID] = payload
	repository.connectedGeneration = 8

	result, err := service.Dispatch(context.Background(), request)
	if !errors.Is(err, ErrDeliveryFailed) || result.State != StateFailed ||
		result.FailureCode != "connection_generation_changed" {
		t.Fatalf("Dispatch() = %#v, %v", result, err)
	}
	if provider.callCount() != 0 || auth.resolveCalls != 0 {
		t.Fatalf("provider calls=%d token resolves=%d", provider.callCount(), auth.resolveCalls)
	}
}

func TestRequestIDWithChangedPayloadIsRejected(t *testing.T) {
	repository := newMemoryRepository()
	service := newTestService(t, repository, &fakeProvider{
		result: DispatchResult{Outcome: DispatchAccepted, RemoteReceiptID: "message-1", ResponseStatus: 200},
	})
	request := testRequest()
	if _, err := service.Dispatch(context.Background(), request); err != nil {
		t.Fatal(err)
	}
	request.Payload = []byte("different")
	if _, err := service.Dispatch(context.Background(), request); !errors.Is(err, ErrRequestConflict) {
		t.Fatalf("conflicting Dispatch() error = %v", err)
	}
}

func TestRequestIDWithDuplicateKeyCompletionSnapshotIsRejected(t *testing.T) {
	provider := &fakeProvider{
		result: DispatchResult{Outcome: DispatchAccepted, RemoteReceiptID: "message-1", ResponseStatus: 200},
	}
	service := newTestService(t, newMemoryRepository(), provider)
	request := testRequest()
	if _, err := service.Dispatch(context.Background(), request); err != nil {
		t.Fatal(err)
	}
	request.CompletionSnapshot = json.RawMessage(`{"taskId":"old","taskId":"task-1","emailMessageId":"email-1"}`)
	if _, err := service.Dispatch(context.Background(), request); !errors.Is(err, ErrInvalidInput) {
		t.Fatalf("duplicate-key Dispatch() error = %v", err)
	}
	if provider.callCount() != 1 {
		t.Fatalf("provider calls = %d, want 1", provider.callCount())
	}
}

func TestSameJSONRejectsDuplicateObjectKeys(t *testing.T) {
	for _, value := range []json.RawMessage{
		json.RawMessage(`{"taskId":"old","taskId":"new"}`),
		json.RawMessage(`{"completion":{"taskId":"old","taskId":"new"}}`),
		json.RawMessage(`[{"taskId":"old","taskId":"new"}]`),
	} {
		if sameJSON(value, value) {
			t.Fatalf("sameJSON(%s) accepted duplicate object key", value)
		}
	}
	if !sameJSON(json.RawMessage(`{"taskId":"task-1","emailMessageId":"email-1"}`),
		json.RawMessage(`{ "emailMessageId": "email-1", "taskId": "task-1" }`)) {
		t.Fatal("sameJSON rejected equivalent unique-key snapshots")
	}
}

func TestProviderFailureClassification(t *testing.T) {
	tests := []struct {
		name      string
		result    DispatchResult
		err       error
		wantState string
		wantErr   error
	}{
		{name: "definite rejection", err: &ProviderError{Kind: ProviderErrorRejected, Code: "invalid_recipient"}, wantState: StateFailed, wantErr: ErrDeliveryFailed},
		{name: "unknown error", err: errors.New("connection reset"), wantState: StateUncertain, wantErr: ErrDeliveryUncertain},
		{name: "typed nil", err: (*ProviderError)(nil), wantState: StateUncertain, wantErr: ErrDeliveryUncertain},
		{name: "server response", result: DispatchResult{Outcome: DispatchRejected, ResponseStatus: 503}, wantState: StateUncertain, wantErr: ErrDeliveryUncertain},
		{name: "inconsistent accepted response", result: DispatchResult{Outcome: DispatchAccepted, RemoteReceiptID: "message-1", ResponseStatus: 401}, wantState: StateUncertain, wantErr: ErrDeliveryUncertain},
		{name: "accepted without receipt", result: DispatchResult{Outcome: DispatchAccepted, ResponseStatus: 200}, wantState: StateUncertain, wantErr: ErrDeliveryUncertain},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			repository := newMemoryRepository()
			provider := &fakeProvider{result: tt.result, err: tt.err}
			service := newTestService(t, repository, provider)
			result, err := service.Dispatch(context.Background(), testRequest())
			if !errors.Is(err, tt.wantErr) || result.State != tt.wantState || provider.callCount() != 1 {
				t.Fatalf("Dispatch() = %#v, %v; calls=%d", result, err, provider.callCount())
			}
		})
	}
}

func TestDispatchContinuesOneStagedOAuthRefresh(t *testing.T) {
	repository := newMemoryRepository()
	provider := &fakeProvider{result: DispatchResult{
		Outcome: DispatchAccepted, RemoteReceiptID: "message-1", ResponseStatus: 200,
	}}
	auth := &fakeOAuth{
		principal: connector_oauth.ConnectedPrincipal{ID: "principal-1", UserID: "user-1", ConnectorKey: "gmail",
			Generation: 7, State: connector_oauth.PrincipalConnected},
		resolveErrors: []error{connector_oauth.ErrEffectResultStaged, nil},
		resolveTokens: []string{"", "refreshed-access-token"},
	}
	repository.connectedGeneration = 7
	service, err := newService(repository, auth, staticProviderResolver{provider: provider})
	if err != nil {
		t.Fatal(err)
	}
	result, err := service.Dispatch(context.Background(), testRequest())
	if err != nil || result.State != StateAccepted || auth.resolveCalls != 2 || provider.callCount() != 1 {
		t.Fatalf("Dispatch = %#v, %v; resolves=%d provider=%d", result, err, auth.resolveCalls, provider.callCount())
	}
}

func TestTransientOAuthFailureReleasesDispatchForRetry(t *testing.T) {
	repository := newMemoryRepository()
	provider := &fakeProvider{result: DispatchResult{
		Outcome: DispatchAccepted, RemoteReceiptID: "message-1", ResponseStatus: 200,
	}}
	auth := &fakeOAuth{
		principal: connector_oauth.ConnectedPrincipal{ID: "principal-1", UserID: "user-1", ConnectorKey: "gmail",
			Generation: 7, State: connector_oauth.PrincipalConnected},
		resolveErrors: []error{errors.New("token store unavailable"), nil},
		resolveTokens: []string{"", "access-token"},
	}
	repository.connectedGeneration = 7
	service, err := newService(repository, auth, staticProviderResolver{provider: provider})
	if err != nil {
		t.Fatal(err)
	}
	service.random = func() (string, error) { return "claim-1", nil }

	first, err := service.Dispatch(context.Background(), testRequest())
	if !errors.Is(err, ErrCredentialUnavailable) || first.State != StateReserved ||
		first.FailureCode != "" || provider.callCount() != 0 || auth.resolveCalls != 1 {
		t.Fatalf("first Dispatch = %#v, %v; provider=%d resolves=%d", first, err, provider.callCount(), auth.resolveCalls)
	}
	second, err := service.Dispatch(context.Background(), testRequest())
	if err != nil || second.State != StateAccepted || provider.callCount() != 1 || auth.resolveCalls != 2 {
		t.Fatalf("second Dispatch = %#v, %v; provider=%d resolves=%d", second, err, provider.callCount(), auth.resolveCalls)
	}
}

func TestCanceledOAuthResolutionReleasesDispatch(t *testing.T) {
	repository := newMemoryRepository()
	repository.releaseContextCheck = func(ctx context.Context) error {
		if ctx.Err() != nil {
			return errors.New("release context is canceled")
		}
		deadline, ok := ctx.Deadline()
		remaining := time.Until(deadline)
		if !ok || remaining <= 0 || remaining > defaultCompletionTimeout {
			return errors.New("release context deadline is invalid")
		}
		return nil
	}
	provider := &fakeProvider{}
	auth := &fakeOAuth{
		principal: connector_oauth.ConnectedPrincipal{ID: "principal-1", UserID: "user-1", ConnectorKey: "gmail",
			Generation: 7, State: connector_oauth.PrincipalConnected},
		resolveErr: context.Canceled,
	}
	repository.connectedGeneration = 7
	service, err := newService(repository, auth, staticProviderResolver{provider: provider})
	if err != nil {
		t.Fatal(err)
	}
	service.random = func() (string, error) { return "claim-1", nil }
	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	result, err := service.Dispatch(ctx, testRequest())
	if !errors.Is(err, ErrCredentialUnavailable) || result.State != StateReserved ||
		provider.callCount() != 0 || auth.resolveCalls != 1 {
		t.Fatalf("Dispatch = %#v, %v; provider=%d resolves=%d", result, err, provider.callCount(), auth.resolveCalls)
	}
}

func TestReleasedClaimCannotFinishRetriedDispatch(t *testing.T) {
	repository := newMemoryRepository()
	repository.connectedGeneration = 7
	request, payload, err := normalizeRequest(testRequest())
	if err != nil {
		t.Fatal(err)
	}
	principal := connector_oauth.ConnectedPrincipal{
		ID: "principal-1", UserID: "user-1", ConnectorKey: "gmail",
		Generation: 7, State: connector_oauth.PrincipalConnected,
	}
	now := time.Now().UTC().Truncate(time.Microsecond)
	if _, err := repository.Reserve(context.Background(), request, principal, payload, now); err != nil {
		t.Fatal(err)
	}
	_, firstClaim, err := repository.BeginDispatch(context.Background(), request.RequestID, request.UserID,
		"claim-1", now, now.Add(DefaultDispatchLease))
	if err != nil {
		t.Fatal(err)
	}
	released, err := repository.ReleaseDispatch(context.Background(), *firstClaim, now.Add(time.Second))
	if err != nil || released.State != StateReserved || released.Fence != 1 {
		t.Fatalf("ReleaseDispatch = %#v, %v", released, err)
	}
	_, secondClaim, err := repository.BeginDispatch(context.Background(), request.RequestID, request.UserID,
		"claim-2", now.Add(2*time.Second), now.Add(DefaultDispatchLease))
	if err != nil || secondClaim.Fence != 2 {
		t.Fatalf("second BeginDispatch = %#v, %v", secondClaim, err)
	}
	terminal := terminalDispatch{State: StateAccepted, RemoteReceiptID: "message-1", ResponseStatus: 200}
	if _, err := repository.FinishDispatch(context.Background(), *firstClaim, terminal, now.Add(3*time.Second)); !errors.Is(err, ErrOwnershipLost) {
		t.Fatalf("stale FinishDispatch error = %v", err)
	}
	finished, err := repository.FinishDispatch(context.Background(), *secondClaim, terminal, now.Add(3*time.Second))
	if err != nil || finished.State != StateAccepted || finished.Fence != 2 {
		t.Fatalf("current FinishDispatch = %#v, %v", finished, err)
	}
}

func TestUnconfirmedOAuthFailureReleaseCannotRedispatch(t *testing.T) {
	repository := newMemoryRepository()
	repository.failRelease = true
	provider := &fakeProvider{}
	auth := &fakeOAuth{
		principal: connector_oauth.ConnectedPrincipal{ID: "principal-1", UserID: "user-1", ConnectorKey: "gmail",
			Generation: 7, State: connector_oauth.PrincipalConnected},
		resolveErr: errors.New("token store unavailable"),
	}
	repository.connectedGeneration = 7
	service, err := newService(repository, auth, staticProviderResolver{provider: provider})
	if err != nil {
		t.Fatal(err)
	}
	service.random = func() (string, error) { return "claim-1", nil }

	first, err := service.Dispatch(context.Background(), testRequest())
	if !errors.Is(err, ErrCredentialUnavailable) || first.State != StateDispatching || provider.callCount() != 0 {
		t.Fatalf("first Dispatch = %#v, %v; provider=%d", first, err, provider.callCount())
	}
	second, err := service.Dispatch(context.Background(), testRequest())
	if !errors.Is(err, ErrDispatchInProgress) || second.State != StateDispatching || provider.callCount() != 0 {
		t.Fatalf("second Dispatch = %#v, %v; provider=%d", second, err, provider.callCount())
	}
}

func TestDeterministicOAuthFailureIsTerminal(t *testing.T) {
	repository := newMemoryRepository()
	provider := &fakeProvider{}
	auth := &fakeOAuth{
		principal: connector_oauth.ConnectedPrincipal{ID: "principal-1", UserID: "user-1", ConnectorKey: "gmail",
			Generation: 7, State: connector_oauth.PrincipalConnected},
		resolveErr: connector_oauth.ErrReconnectRequired,
	}
	repository.connectedGeneration = 7
	service, err := newService(repository, auth, staticProviderResolver{provider: provider})
	if err != nil {
		t.Fatal(err)
	}
	result, err := service.Dispatch(context.Background(), testRequest())
	if !errors.Is(err, ErrDeliveryFailed) || result.State != StateFailed ||
		result.FailureCode != "reconnect_required" || provider.callCount() != 0 {
		t.Fatalf("Dispatch = %#v, %v; provider=%d", result, err, provider.callCount())
	}
}

func TestDispatchFailsClosedForEmptyAccessToken(t *testing.T) {
	for _, test := range []struct {
		name  string
		token string
	}{
		{name: "empty"},
		{name: "whitespace", token: " \t\n"},
	} {
		t.Run(test.name, func(t *testing.T) {
			repository := newMemoryRepository()
			provider := &fakeProvider{}
			auth := &fakeOAuth{
				principal: connector_oauth.ConnectedPrincipal{ID: "principal-1", UserID: "user-1", ConnectorKey: "gmail",
					Generation: 7, State: connector_oauth.PrincipalConnected},
				resolveTokens: []string{test.token},
			}
			repository.connectedGeneration = 7
			service, err := newService(repository, auth, staticProviderResolver{provider: provider})
			if err != nil {
				t.Fatal(err)
			}

			result, err := service.Dispatch(context.Background(), testRequest())
			if !errors.Is(err, ErrDeliveryFailed) || result.State != StateFailed ||
				result.FailureCode != "credential_unavailable" || provider.callCount() != 0 || auth.resolveCalls != 1 {
				t.Fatalf("Dispatch = %#v, %v; provider=%d resolves=%d", result, err, provider.callCount(), auth.resolveCalls)
			}
			replayed, err := service.Dispatch(context.Background(), testRequest())
			if !errors.Is(err, ErrDeliveryFailed) || replayed.State != StateFailed ||
				provider.callCount() != 0 || auth.resolveCalls != 1 {
				t.Fatalf("replay = %#v, %v; provider=%d resolves=%d", replayed, err, provider.callCount(), auth.resolveCalls)
			}
		})
	}
}

func TestMalformedReservedJobFailsClosedWithoutProviderCall(t *testing.T) {
	repository := newMemoryRepository()
	provider := &fakeProvider{}
	service := newTestService(t, repository, provider)
	request, payload, err := normalizeRequest(testRequest())
	if err != nil {
		t.Fatal(err)
	}
	dispatched := service.now().Add(-time.Minute)
	repository.jobs[request.RequestID] = Job{
		RequestID: request.RequestID, ConnectedPrincipalID: "principal-1", PrincipalGeneration: 7,
		OriginatingUserID: request.UserID, State: StateReserved, PayloadSHA256: payload.SHA256,
		PayloadEncodedSize: payload.EncodedSize, SourceObject: request.SourceObject,
		SourceRecordID: request.SourceRecordID, CompletionSnapshot: request.CompletionSnapshot,
		Fence: 1, DispatchStartedAt: &dispatched,
	}
	repository.payloads[request.RequestID] = payload
	result, err := service.Dispatch(context.Background(), request)
	if !errors.Is(err, ErrStateInvalid) || result.RequestID != "" || provider.callCount() != 0 {
		t.Fatalf("Dispatch = %#v, %v; provider=%d", result, err, provider.callCount())
	}
}

func TestCanceledRequestPersistsKnownProviderAcceptance(t *testing.T) {
	repository := newMemoryRepository()
	entered := make(chan struct{})
	provider := &fakeProvider{
		entered: entered, waitForContext: true,
		result: DispatchResult{Outcome: DispatchAccepted, RemoteReceiptID: "message-1", ResponseStatus: 200},
	}
	service := newTestService(t, repository, provider)
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct {
		result Result
		err    error
	}, 1)
	go func() {
		result, err := service.Dispatch(ctx, testRequest())
		done <- struct {
			result Result
			err    error
		}{result: result, err: err}
	}()
	<-entered
	cancel()
	outcome := <-done
	if outcome.err != nil || outcome.result.State != StateAccepted || provider.callCount() != 1 {
		t.Fatalf("Dispatch = %#v, %v; provider=%d", outcome.result, outcome.err, provider.callCount())
	}
}

func TestProviderEffectContextReservesTerminationAndCompletionWindows(t *testing.T) {
	service := newTestService(t, newMemoryRepository(), &fakeProvider{})
	now := service.now().UTC()
	leaseExpiresAt := now.Add(service.lease)
	providerBudget := service.lease - 2*service.completionTimeout
	for _, test := range []struct {
		name  string
		bound time.Duration
		valid bool
	}{
		{name: "zero"},
		{name: "inside", bound: providerBudget - time.Nanosecond, valid: true},
		{name: "exact", bound: providerBudget},
		{name: "outside", bound: providerBudget + time.Nanosecond},
		{name: "maximum", bound: time.Duration(1<<63 - 1)},
	} {
		t.Run(test.name, func(t *testing.T) {
			provider := &fakeProvider{terminationBound: test.bound, terminationBoundSet: true}
			ctx, cancel, err := service.providerEffectContext(context.Background(), provider, &leaseExpiresAt)
			if !test.valid {
				if err == nil {
					cancel()
					t.Fatal("providerEffectContext unexpectedly succeeded")
				}
				return
			}
			if err != nil {
				t.Fatal(err)
			}
			defer cancel()
			deadline, ok := ctx.Deadline()
			want := leaseExpiresAt.Add(-test.bound).Add(-2 * service.completionTimeout)
			if !ok || !deadline.Equal(want) || !deadline.Before(leaseExpiresAt) {
				t.Fatalf("deadline = %v/%v, want %v", deadline, ok, want)
			}
		})
	}
}

func TestInvalidProviderTerminationBoundFailsBeforeDispatch(t *testing.T) {
	repository := newMemoryRepository()
	provider := &fakeProvider{terminationBoundSet: true}
	service := newTestService(t, repository, provider)
	result, err := service.Dispatch(context.Background(), testRequest())
	if !errors.Is(err, ErrDeliveryFailed) || result.State != StateFailed ||
		result.FailureCode != "provider_contract_invalid" || provider.callCount() != 0 {
		t.Fatalf("Dispatch = %#v, %v; provider=%d", result, err, provider.callCount())
	}
}

func newTestService(t *testing.T, repository *memoryRepository, provider *fakeProvider) *Service {
	t.Helper()
	auth := &fakeOAuth{principal: connector_oauth.ConnectedPrincipal{
		ID: "principal-1", UserID: "user-1", ConnectorKey: "gmail",
		Generation: 7, State: connector_oauth.PrincipalConnected,
	}}
	repository.connectedGeneration = 7
	service, err := newService(repository, auth, staticProviderResolver{provider: provider})
	if err != nil {
		t.Fatal(err)
	}
	fixedNow := time.Now().UTC().Truncate(time.Microsecond)
	service.now = func() time.Time { return fixedNow }
	service.random = func() (string, error) { return "claim-1", nil }
	return service
}

func testRequest() Request {
	return Request{
		RequestID: "request-1", UserID: "user-1", ConnectorKey: "gmail",
		ExpectedGeneration: 7,
		Payload:            []byte("immutable MIME"), SourceObject: "Contact",
		SourceRecordID:     "contact-1",
		CompletionSnapshot: json.RawMessage(`{ "taskId": "task-1", "emailMessageId": "email-1" }`),
	}
}

type fakeOAuth struct {
	principal     connector_oauth.ConnectedPrincipal
	resolveErr    error
	resolveErrors []error
	resolveTokens []string
	resolveCalls  int
}

func (f *fakeOAuth) Principal(context.Context, string, string) (*connector_oauth.ConnectedPrincipal, error) {
	principal := f.principal
	return &principal, nil
}

func (f *fakeOAuth) ResolveAccessToken(context.Context, connector_oauth.ResolveAccessTokenRequest) (string, error) {
	index := f.resolveCalls
	f.resolveCalls++
	if index < len(f.resolveErrors) || index < len(f.resolveTokens) {
		var token string
		var err error
		if index < len(f.resolveTokens) {
			token = f.resolveTokens[index]
		}
		if index < len(f.resolveErrors) {
			err = f.resolveErrors[index]
		}
		return token, err
	}
	return "access-token", f.resolveErr
}

type staticProviderResolver struct{ provider Provider }

func (r staticProviderResolver) Provider(string) (Provider, bool) {
	return r.provider, r.provider != nil
}

type fakeProvider struct {
	mu                  sync.Mutex
	calls               int
	result              DispatchResult
	err                 error
	entered             chan struct{}
	release             chan struct{}
	waitForContext      bool
	terminationBound    time.Duration
	terminationBoundSet bool
}

func (p *fakeProvider) EffectTerminationBound() time.Duration {
	if !p.terminationBoundSet && p.terminationBound == 0 {
		return time.Second
	}
	return p.terminationBound
}

func (p *fakeProvider) Dispatch(ctx context.Context, request DispatchRequest) (DispatchResult, error) {
	if request.AccessToken == "" || len(request.Payload) == 0 {
		return DispatchResult{}, errors.New("missing dispatch input")
	}
	p.mu.Lock()
	p.calls++
	p.mu.Unlock()
	if p.entered != nil {
		close(p.entered)
	}
	if p.waitForContext {
		<-ctx.Done()
		return p.result, p.err
	}
	if p.release != nil {
		select {
		case <-p.release:
		case <-ctx.Done():
			return p.result, ctx.Err()
		}
	}
	return p.result, p.err
}

func (p *fakeProvider) callCount() int {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.calls
}

type memoryRepository struct {
	mu                  sync.Mutex
	jobs                map[string]Job
	payloads            map[string]payloadRecord
	connectedGeneration int64
	releaseContextCheck func(context.Context) error
	failRelease         bool
	failFinish          bool
}

func newMemoryRepository() *memoryRepository {
	return &memoryRepository{jobs: map[string]Job{}, payloads: map[string]payloadRecord{}}
}

func (r *memoryRepository) Reserve(_ context.Context, request Request, principal connector_oauth.ConnectedPrincipal, payload payloadRecord, now time.Time) (Job, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if existing, ok := r.jobs[request.RequestID]; ok {
		if err := validateJobShape(existing); err != nil {
			return Job{}, err
		}
		stored := r.payloads[request.RequestID]
		if existing.OriginatingUserID != request.UserID || existing.ConnectedPrincipalID != principal.ID ||
			existing.PrincipalGeneration != request.ExpectedGeneration || existing.PayloadSHA256 != payload.SHA256 || existing.PayloadEncodedSize != payload.EncodedSize ||
			existing.SourceObject != request.SourceObject || existing.SourceRecordID != request.SourceRecordID ||
			!sameJSON(existing.CompletionSnapshot, request.CompletionSnapshot) || stored != payload {
			return Job{}, ErrRequestConflict
		}
		return cloneJob(existing), nil
	}
	if principal.State != connector_oauth.PrincipalConnected || principal.Generation != r.connectedGeneration ||
		principal.Generation != request.ExpectedGeneration {
		return Job{}, connector_oauth.ErrGenerationMismatch
	}
	job := Job{
		RequestID: request.RequestID, ConnectedPrincipalID: principal.ID,
		PrincipalGeneration: request.ExpectedGeneration, OriginatingUserID: request.UserID,
		State: StateReserved, PayloadSHA256: payload.SHA256, PayloadEncodedSize: payload.EncodedSize,
		SourceObject: request.SourceObject, SourceRecordID: request.SourceRecordID,
		CompletionSnapshot: append(json.RawMessage(nil), request.CompletionSnapshot...),
		CreatedDate:        now, LastModifiedDate: now,
	}
	r.jobs[job.RequestID] = job
	r.payloads[job.RequestID] = payload
	return cloneJob(job), nil
}

func (r *memoryRepository) Lookup(_ context.Context, requestID, userID string) (Job, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	job, ok := r.jobs[requestID]
	if !ok || job.OriginatingUserID != userID {
		return Job{}, ErrDeliveryNotFound
	}
	if err := validateJobShape(job); err != nil {
		return Job{}, err
	}
	return cloneJob(job), nil
}

func (r *memoryRepository) BeginDispatch(_ context.Context, requestID, userID, claimToken string, now, leaseUntil time.Time) (Job, *dispatchClaim, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	now, leaseUntil = schemaTime(now), schemaTime(leaseUntil)
	job := r.jobs[requestID]
	if err := validateJobShape(job); err != nil {
		return Job{}, nil, err
	}
	if job.OriginatingUserID != userID {
		return Job{}, nil, ErrRequestConflict
	}
	switch job.State {
	case StateDispatching:
		if job.LeaseExpiresAt != nil && job.LeaseExpiresAt.After(now) {
			return cloneJob(job), nil, ErrDispatchInProgress
		}
		job.State, job.FailureCode = StateUncertain, "dispatch_lease_expired"
		job.ClaimToken, job.LeaseExpiresAt, job.CompletedAt = "", nil, ptrTime(now)
		r.jobs[requestID] = job
		return cloneJob(job), nil, nil
	case StateAccepted, StateCompleted, StateFailed, StateUncertain:
		return cloneJob(job), nil, nil
	case StateReserved:
	default:
		return Job{}, nil, ErrRequestConflict
	}
	if job.PrincipalGeneration != r.connectedGeneration {
		job.State, job.FailureCode, job.CompletedAt = StateFailed, "connection_generation_changed", ptrTime(now)
		r.jobs[requestID] = job
		return cloneJob(job), nil, nil
	}
	job.State, job.ClaimToken, job.LeaseExpiresAt = StateDispatching, claimToken, ptrTime(leaseUntil)
	job.DispatchStartedAt, job.Fence = ptrTime(now), job.Fence+1
	r.jobs[requestID] = job
	claim := &dispatchClaim{Job: cloneJob(job), ClaimToken: claimToken, Fence: job.Fence}
	return cloneJob(job), claim, nil
}

func (r *memoryRepository) ReleaseDispatch(ctx context.Context, claim dispatchClaim, now time.Time) (Job, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.releaseContextCheck != nil {
		if err := r.releaseContextCheck(ctx); err != nil {
			return Job{}, err
		}
	}
	if r.failRelease {
		return Job{}, errors.New("persistence unavailable")
	}
	job := r.jobs[claim.Job.RequestID]
	if err := validateJobShape(job); err != nil {
		return Job{}, err
	}
	if job.State != StateDispatching || job.ClaimToken != claim.ClaimToken || job.Fence != claim.Fence ||
		job.LeaseExpiresAt == nil || claim.Job.LeaseExpiresAt == nil ||
		!job.LeaseExpiresAt.Equal(*claim.Job.LeaseExpiresAt) || job.DispatchStartedAt == nil ||
		claim.Job.DispatchStartedAt == nil || !job.DispatchStartedAt.Equal(*claim.Job.DispatchStartedAt) {
		return Job{}, ErrOwnershipLost
	}
	job.State, job.ClaimToken, job.LeaseExpiresAt = StateReserved, "", nil
	job.DispatchStartedAt, job.LastModifiedDate = nil, now
	r.jobs[job.RequestID] = job
	return cloneJob(job), nil
}

func (r *memoryRepository) FinishDispatch(_ context.Context, claim dispatchClaim, terminal terminalDispatch, now time.Time) (Job, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.failFinish {
		return Job{}, errors.New("persistence unavailable")
	}
	job := r.jobs[claim.Job.RequestID]
	if err := validateJobShape(job); err != nil {
		return Job{}, err
	}
	if err := validateTerminalDispatch(terminal); err != nil {
		return Job{}, err
	}
	if job.State != StateDispatching || job.ClaimToken != claim.ClaimToken || job.Fence != claim.Fence {
		return Job{}, ErrOwnershipLost
	}
	job.State, job.ClaimToken, job.LeaseExpiresAt = terminal.State, "", nil
	job.RemoteReceiptID, job.RemoteConversationID = terminal.RemoteReceiptID, terminal.RemoteConversationID
	job.FailureCode = terminal.FailureCode
	if terminal.ResponseStatus != 0 {
		job.ResponseStatus = &terminal.ResponseStatus
	}
	if terminal.State == StateAccepted {
		job.AcceptedAt = ptrTime(now)
	} else {
		job.CompletedAt = ptrTime(now)
	}
	r.jobs[job.RequestID] = job
	return cloneJob(job), nil
}

func (r *memoryRepository) Complete(_ context.Context, requestID, userID string, now time.Time) (Job, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	job := r.jobs[requestID]
	if err := validateJobShape(job); err != nil {
		return Job{}, err
	}
	if job.OriginatingUserID != userID {
		return Job{}, ErrRequestConflict
	}
	if job.State == StateCompleted {
		return cloneJob(job), nil
	}
	if job.State != StateAccepted {
		return cloneJob(job), ErrCompletionInvalid
	}
	job.State, job.CompletedAt = StateCompleted, ptrTime(now)
	r.jobs[requestID] = job
	return cloneJob(job), nil
}

func cloneJob(job Job) Job {
	job.CompletionSnapshot = append(json.RawMessage(nil), job.CompletionSnapshot...)
	return job
}

func ptrTime(value time.Time) *time.Time { return &value }