package connector_inbound

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

	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/events"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/workqueue"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/constants"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/models"
)

const companionClaimGeneration int64 = 17

// capturingVersionFlowInvoker captures ExecuteFlowVersion calls and distinguishes
// them from ExecuteFlow calls — used by gate-2 and gate-3 tests.
type capturingVersionFlowInvoker struct {
	executeFlowCalls        int
	executeFlowVersionCalls int
	lastVersionID           string
	result                  *FlowResult
}

func (c *capturingVersionFlowInvoker) ExecuteFlow(_ context.Context, _ string, _ map[string]any, _ *models.UserSession) (*FlowResult, error) {
	c.executeFlowCalls++
	return c.result, nil
}

func (c *capturingVersionFlowInvoker) ExecuteFlowVersion(_ context.Context, _, companionFlowVersionID string, _ map[string]any, _ *models.UserSession) (*FlowResult, error) {
	c.executeFlowVersionCalls++
	c.lastVersionID = companionFlowVersionID
	return c.result, nil
}

func (c *capturingVersionFlowInvoker) SystemUser(_ context.Context) *models.UserSession {
	return &models.UserSession{ID: "system-user"}
}

// makeCompanionEvent constructs a ConnectorEvent with non-empty companion snapshot.
func makeCompanionEvent(id, namespace, packInstallID, companionPackInstallID, companionFlowVersionID string) *connector_events.ConnectorEvent {
	ev := noopEvent(id, "companion.created")
	ev.Namespace = namespace
	ev.InstalledPackID = packInstallID
	ev.CompanionPackInstallID = companionPackInstallID
	ev.CompanionFlowVersionID = companionFlowVersionID
	return ev
}

// companionTerminalIdentityService materializes the WorkItem row returned by
// ClaimNextWorkItem from the worker-generated row-CAS input, then records every
// terminal completion. Companion tests use it to prove terminal calls reuse the
// actual claim identity rather than a static fixture lease.
type companionTerminalIdentityService struct {
	*completeWorkItemServiceFake

	mu              sync.Mutex
	claimInput      connector_workqueue.ClaimWorkItemInput
	claimedWorkItem *connector_workqueue.WorkItem
	completeCalls   []companionCompleteWorkItemCall
}

type companionCompleteWorkItemCall struct {
	workID        string
	leaseToken    string
	terminalState string
}

type companionCheckResult struct {
	present bool
	err     error
}

// scriptedCompanionChecker makes every expected incarnation result explicit.
// An additional call fails the test path instead of silently reusing the last
// answer, so companion lifecycle tests stay coupled to the worker's true
// prepare/execute call graph.
type scriptedCompanionChecker struct {
	mu      sync.Mutex
	results []companionCheckResult
	calls   int
}

func (c *scriptedCompanionChecker) check(_ context.Context, _ string) (bool, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	if c.calls >= len(c.results) {
		c.calls++
		return false, fmt.Errorf("unexpected companion checker call %d", c.calls)
	}
	result := c.results[c.calls]
	c.calls++
	return result.present, result.err
}

func (c *scriptedCompanionChecker) assertCalls(t *testing.T, want int) {
	t.Helper()
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.calls != want {
		t.Fatalf("companion checker calls = %d, want %d", c.calls, want)
	}
}

func (s *companionTerminalIdentityService) ClaimNextWorkItem(ctx context.Context, in connector_workqueue.ClaimWorkItemInput) (*connector_workqueue.WorkItem, bool, error) {
	work, claimed, err := s.completeWorkItemServiceFake.ClaimNextWorkItem(ctx, in)
	if err != nil || !claimed || work == nil {
		return work, claimed, err
	}

	claimedWorkItem := *work
	claimedWorkItem.InstalledPackID = in.InstalledPackID
	claimedWorkItem.Generation = in.Generation
	claimedWorkItem.State = connector_workqueue.WorkStateProcessing
	claimedWorkItem.LeaseOwner = in.OwnerID
	claimedWorkItem.LeaseToken = in.LeaseToken
	claimedWorkItem.LeaseUntil = sql.NullTime{Time: in.LeaseUntil, Valid: !in.LeaseUntil.IsZero()}

	s.mu.Lock()
	s.claimInput = in
	s.claimedWorkItem = &claimedWorkItem
	s.mu.Unlock()
	return &claimedWorkItem, true, nil
}

func (s *companionTerminalIdentityService) CompleteWorkItem(ctx context.Context, workID, leaseToken, terminalState string, now time.Time) (bool, error) {
	s.mu.Lock()
	s.completeCalls = append(s.completeCalls, companionCompleteWorkItemCall{
		workID:        workID,
		leaseToken:    leaseToken,
		terminalState: terminalState,
	})
	s.mu.Unlock()
	return s.completeWorkItemServiceFake.CompleteWorkItem(ctx, workID, leaseToken, terminalState, now)
}

func (s *companionTerminalIdentityService) claimIdentity() (connector_workqueue.ClaimWorkItemInput, connector_workqueue.WorkItem, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.claimedWorkItem == nil {
		return connector_workqueue.ClaimWorkItemInput{}, connector_workqueue.WorkItem{}, false
	}
	return s.claimInput, *s.claimedWorkItem, true
}

func (s *companionTerminalIdentityService) completedClaims() []companionCompleteWorkItemCall {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := make([]companionCompleteWorkItemCall, len(s.completeCalls))
	copy(out, s.completeCalls)
	return out
}

func assertCompanionClaimIdentity(t *testing.T, service *companionTerminalIdentityService, ev *connector_events.ConnectorEvent) (connector_workqueue.ClaimWorkItemInput, connector_workqueue.WorkItem) {
	t.Helper()

	in, work, ok := service.claimIdentity()
	if !ok {
		t.Fatal("expected ClaimNextWorkItem to return a materialized WorkItem")
	}
	if in.InstalledPackID != ev.InstalledPackID || in.Generation != companionClaimGeneration ||
		in.OwnerID != "scheduler-owner" || in.FencingToken != "scheduler-fence" {
		t.Fatalf("claim input = %+v, want companion pack/generation/owner/fence identity", in)
	}
	if in.LeaseToken == "" || in.LeaseToken == "seed-companion-lease" {
		t.Fatalf("claim lease token = %q, want a generated token distinct from the seed", in.LeaseToken)
	}
	if in.LeaseUntil.IsZero() || !in.LeaseUntil.After(in.Now) {
		t.Fatalf("claim lease interval = now %s until %s, want a future lease", in.Now, in.LeaseUntil)
	}
	if work.WorkID == "" || work.EventID != ev.ID ||
		work.InstalledPackID != in.InstalledPackID || work.Generation != in.Generation ||
		work.State != connector_workqueue.WorkStateProcessing ||
		work.LeaseOwner != in.OwnerID || work.LeaseToken != in.LeaseToken ||
		!work.LeaseUntil.Valid || !work.LeaseUntil.Time.Equal(in.LeaseUntil) {
		t.Fatalf("claimed WorkItem = %+v, want materialized ClaimWorkItemInput identity", work)
	}
	return in, work
}

func assertCompanionCompletionIdentity(t *testing.T, service *companionTerminalIdentityService, claimed connector_workqueue.WorkItem, terminalState string) {
	t.Helper()

	calls := service.completedClaims()
	if len(calls) != 1 {
		t.Fatalf("CompleteWorkItem calls = %v, want one", calls)
	}
	call := calls[0]
	if call.workID != claimed.WorkID || call.leaseToken != claimed.LeaseToken || call.terminalState != terminalState {
		t.Fatalf("CompleteWorkItem = %+v, want work=%q lease=%q state=%q", call, claimed.WorkID, claimed.LeaseToken, terminalState)
	}
}

func assertCompanionFailureIdentity(t *testing.T, claims *fakeWorkItemClaims, claimed connector_workqueue.WorkItem, eventID string) connector_workqueue.FailWorkItemInput {
	t.Helper()

	if len(claims.failInputs) != 1 {
		t.Fatalf("WorkItem failures = %v, want one terminal failure", claims.failInputs)
	}
	failure := claims.failInputs[0]
	if failure.WorkID != claimed.WorkID || failure.EventID != eventID || failure.LeaseToken != claimed.LeaseToken {
		t.Fatalf("WorkItem failure identity = %+v, want work=%q event=%q lease=%q", failure, claimed.WorkID, eventID, claimed.LeaseToken)
	}
	return failure
}

// newCompanionWorkItemWorker keeps companion-execution tests on the mandatory
// WorkItem executor. The Event remains immutable input; terminal ownership is
// asserted through the generated WorkItem claim identity.
func newCompanionWorkItemWorker(
	t *testing.T,
	ev *connector_events.ConnectorEvent,
	batchRead BatchReadClient,
	flow FlowInvoker,
	resolver connector_inbound_routing.RouteResolver,
) (*Worker, *companionTerminalIdentityService, *fakeSyncState) {
	t.Helper()
	work := liveWorkItem("work-"+ev.ID, ev.ID, connector_workqueue.WorkTypeConnectorEvent, -1, "seed-companion-lease")
	work.InstalledPackID = "seed-companion-pack"
	work.LeaseOwner = "seed-companion-owner"
	work.State = connector_workqueue.WorkStatePending
	resolvedGeneration := companionClaimGeneration
	claims := &fakeWorkItemClaims{work: work, resolveGeneration: &resolvedGeneration}
	service := &companionTerminalIdentityService{
		completeWorkItemServiceFake: newCompleteWorkItemServiceFake(claims, &fakeWorkItemEventLoader{event: ev}, nil),
	}
	syncstate := &fakeSyncState{}
	w, err := NewWorkItemWorkerForTest(context.Background(),
		WorkerConfig{
			Namespace:                     ev.Namespace,
			PackInstallID:                 ev.InstalledPackID,
			WorkerID:                      "w1",
			WorkItemSchedulerOwnerID:      "scheduler-owner",
			WorkItemSchedulerFencingToken: "scheduler-fence",
		},
		service, syncstate, batchRead, flow, resolver,
	)
	if err != nil {
		t.Fatalf("NewWorkItemWorkerForTest: %v", err)
	}
	return w, service, syncstate
}

// ── gate 2: worker uses getFlowVersionById (ExecuteFlowVersion), never by-name ─

// TestWorker_CompanionEvent_UsesExecuteFlowVersion (gate 2 §7):
// When CompanionFlowVersionID is set, processOneWithDrain calls ExecuteFlowVersion
// with the frozen version id, never ExecuteFlow (by name).
func TestWorker_CompanionEvent_UsesExecuteFlowVersion(t *testing.T) {
	const (
		companionPackInstallID = "companion-pack-install-v1"
		frozenVersionID        = "flow-version-v1"
	)
	ev := makeCompanionEvent("ev-gate2", "connector-ns", "pack-install-1", companionPackInstallID, frozenVersionID)
	flow := &capturingVersionFlowInvoker{result: &FlowResult{IsSuccess: true}}
	resolver := &fakeRouteResolver{flowName: "companion-flow"}
	w, service, syncstate := newCompanionWorkItemWorker(t, ev, nil, flow, resolver)
	claims := service.claims
	checker := &scriptedCompanionChecker{results: []companionCheckResult{{present: true}, {present: true}}}
	w.SetCompanionInstalledChecker(checker.check)

	claimed, err := w.processOne(context.Background())
	if err != nil {
		t.Fatalf("processOne: %v", err)
	}
	if !claimed {
		t.Fatal("expected claimed=true")
	}
	if claims.claimCalls != 1 || claims.claimByEventCalls != 0 {
		t.Fatalf("WorkItem claims = direct %d, by-event %d; want 1, 0", claims.claimCalls, claims.claimByEventCalls)
	}
	_, claimedWork := assertCompanionClaimIdentity(t, service, ev)
	assertCompanionCompletionIdentity(t, service, claimedWork, connector_workqueue.WorkStateProcessed)
	if len(claims.failInputs) != 0 || claims.retryCalls != 0 || len(syncstate.errors) != 0 {
		t.Fatalf("WorkItem failures/retries/SyncState errors = %v/%d/%v, want none", claims.failInputs, claims.retryCalls, syncstate.errors)
	}
	if flow.executeFlowCalls != 0 {
		t.Errorf("ExecuteFlow calls = %d, want 0 (cross-pack must use ExecuteFlowVersion)", flow.executeFlowCalls)
	}
	if flow.executeFlowVersionCalls != 1 {
		t.Errorf("ExecuteFlowVersion calls = %d, want 1", flow.executeFlowVersionCalls)
	}
	if flow.lastVersionID != frozenVersionID {
		t.Errorf("ExecuteFlowVersion called with version=%q, want %q", flow.lastVersionID, frozenVersionID)
	}
	checker.assertCalls(t, 2)
}

// ── gate 3: drain-time incarnation re-check → failTerminal when gone ─

// TestWorker_CompanionEvent_RebirthCausesFailTerminal (gate 3 §7):
// When checkCompanionInstalled returns false, processOneWithDrain calls failTerminal,
// does NOT execute the flow.
func TestWorker_CompanionEvent_RebirthCausesFailTerminal(t *testing.T) {
	ev := makeCompanionEvent("ev-gate3", "connector-ns", "pack-install-1", "stale-companion-install", "flow-v1")
	flow := &capturingVersionFlowInvoker{result: &FlowResult{IsSuccess: true}}
	resolver := &fakeRouteResolver{flowName: "companion-flow"}
	batchReadCalls := 0
	batchRead := &capturingBatchRead{fn: func(_, _ string, _ []string) {
		batchReadCalls++
	}}

	w, service, _ := newCompanionWorkItemWorker(t, ev, batchRead, flow, resolver)
	claims := service.claims
	// Incarnation absent — companion was reborn.
	checker := &scriptedCompanionChecker{results: []companionCheckResult{{present: false}}}
	w.SetCompanionInstalledChecker(checker.check)

	claimed, err := w.processOne(context.Background())
	// processOne returns (true, err) on failTerminal — err is the cause.
	if !claimed {
		t.Fatal("expected claimed=true (event was claimed before failTerminal)")
	}
	if err == nil {
		t.Fatal("expected error from failTerminal, got nil")
	}
	if flow.executeFlowVersionCalls != 0 || flow.executeFlowCalls != 0 {
		t.Errorf("flow should not have been invoked, got ExecuteFlowVersion=%d ExecuteFlow=%d",
			flow.executeFlowVersionCalls, flow.executeFlowCalls)
	}
	if batchReadCalls != 0 {
		t.Fatalf("batchRead calls = %d, want 0 before companion incarnation check passes", batchReadCalls)
	}
	if claims.claimCalls != 1 || claims.claimByEventCalls != 0 {
		t.Fatalf("WorkItem claims = direct %d, by-event %d; want 1, 0", claims.claimCalls, claims.claimByEventCalls)
	}
	_, claimedWork := assertCompanionClaimIdentity(t, service, ev)
	failure := assertCompanionFailureIdentity(t, claims, claimedWork, ev.ID)
	if failure.FailureClass != connector_workqueue.FailureClassInboundExecution || failure.FailureSummary != connector_workqueue.FailureSummaryInbound {
		t.Fatalf("WorkItem failure classification = %q/%q, want inbound execution/inbound", failure.FailureClass, failure.FailureSummary)
	}
	if len(claims.completeCalls) != 0 || claims.retryCalls != 0 {
		t.Fatalf("WorkItem terminal/retry calls = %v/%d, want none after fail", claims.completeCalls, claims.retryCalls)
	}
	checker.assertCalls(t, 1)
}

// ── gate 7: same-pack/legacy path unchanged ─

// TestWorker_SamePack_UsesExecuteFlow (gate 7 §7):
// When CompanionFlowVersionID is empty (same-pack/legacy), processOneWithDrain
// calls ExecuteFlow (unchanged path), never ExecuteFlowVersion.
func TestWorker_SamePack_UsesExecuteFlow(t *testing.T) {
	ev := noopEvent("ev-gate7", "noop.created") // CompanionFlowVersionID = ""
	flow := &capturingVersionFlowInvoker{result: &FlowResult{IsSuccess: true}}
	resolver := &fakeRouteResolver{flowName: "same-pack-flow"}

	w, service, _ := newCompanionWorkItemWorker(t, ev, nil, flow, resolver)
	claims := service.claims
	checker := &scriptedCompanionChecker{}
	w.SetCompanionInstalledChecker(checker.check)

	claimed, err := w.processOne(context.Background())
	if err != nil {
		t.Fatalf("processOne: %v", err)
	}
	if !claimed {
		t.Fatal("expected claimed=true")
	}
	if flow.executeFlowCalls != 1 {
		t.Errorf("ExecuteFlow calls = %d, want 1 (same-pack must use ExecuteFlow)", flow.executeFlowCalls)
	}
	if flow.executeFlowVersionCalls != 0 {
		t.Errorf("ExecuteFlowVersion calls = %d, want 0", flow.executeFlowVersionCalls)
	}
	if claims.claimCalls != 1 || claims.claimByEventCalls != 0 {
		t.Fatalf("WorkItem claims = direct %d, by-event %d; want 1, 0", claims.claimCalls, claims.claimByEventCalls)
	}
	_, claimedWork := assertCompanionClaimIdentity(t, service, ev)
	assertCompanionCompletionIdentity(t, service, claimedWork, connector_workqueue.WorkStateProcessed)
	if len(claims.failInputs) != 0 || claims.retryCalls != 0 {
		t.Fatalf("WorkItem failures/retries = %v/%d, want none", claims.failInputs, claims.retryCalls)
	}
	checker.assertCalls(t, 0)
}

// ── gate 3b: companion marker stamped on ctx AFTER re-check ─

// TestWorker_CompanionEvent_StampsCompanionMarkerAfterRecheck (gate 3b §7):
// verifies that ctx carries the companion version marker when ExecuteFlowVersion is
// called (marker is stamped after the incarnation re-check passes).
func TestWorker_CompanionEvent_StampsCompanionMarkerAfterRecheck(t *testing.T) {
	const frozenVersionID = "flow-version-marker-test"
	ev := makeCompanionEvent("ev-marker", "connector-ns", "pack-install-1", "companion-install-v1", frozenVersionID)

	var ctxAtExecution context.Context
	flow := &ctxCapturingVersionInvoker{result: &FlowResult{IsSuccess: true}, capturedCtx: &ctxAtExecution}
	resolver := &fakeRouteResolver{flowName: "companion-flow"}

	w, service, _ := newCompanionWorkItemWorker(t, ev, nil, flow, resolver)
	claims := service.claims
	checker := &scriptedCompanionChecker{results: []companionCheckResult{{present: true}, {present: true}}}
	w.SetCompanionInstalledChecker(checker.check)

	_, err := w.processOne(context.Background())
	if err != nil {
		t.Fatalf("processOne: %v", err)
	}
	if ctxAtExecution == nil {
		t.Fatal("ctxAtExecution not captured")
	}
	if !constants.ConnectorInboundCompanionExecutionAuthorizesVersion(ctxAtExecution, frozenVersionID) {
		t.Errorf("ctx at ExecuteFlowVersion does not authorize frozen version %q; companion marker was not stamped", frozenVersionID)
	}
	if claims.claimCalls != 1 || claims.claimByEventCalls != 0 {
		t.Fatalf("WorkItem claims = direct %d, by-event %d; want 1, 0", claims.claimCalls, claims.claimByEventCalls)
	}
	_, claimedWork := assertCompanionClaimIdentity(t, service, ev)
	assertCompanionCompletionIdentity(t, service, claimedWork, connector_workqueue.WorkStateProcessed)
	if len(claims.failInputs) != 0 || claims.retryCalls != 0 {
		t.Fatalf("WorkItem failures/retries = %v/%d, want none", claims.failInputs, claims.retryCalls)
	}
	checker.assertCalls(t, 2)
}

// ctxCapturingVersionInvoker captures the context at ExecuteFlowVersion call time.
type ctxCapturingVersionInvoker struct {
	capturedCtx *context.Context
	result      *FlowResult
}

func (c *ctxCapturingVersionInvoker) ExecuteFlow(_ context.Context, _ string, _ map[string]any, _ *models.UserSession) (*FlowResult, error) {
	return c.result, nil
}

func (c *ctxCapturingVersionInvoker) ExecuteFlowVersion(ctx context.Context, _, _ string, _ map[string]any, _ *models.UserSession) (*FlowResult, error) {
	*c.capturedCtx = ctx
	return c.result, nil
}

func (c *ctxCapturingVersionInvoker) SystemUser(_ context.Context) *models.UserSession {
	return &models.UserSession{ID: "system-user"}
}

// ── gate 8: subflow protection — companion marker rejects subflow ─

// TestCompanionMarker_AuthorizesExactVersionOnly (gate 8 §7):
// ConnectorInboundCompanionExecutionAuthorizesVersion returns true for the exact
// frozen version id and false for any other (subflow protection).
func TestCompanionMarker_AuthorizesExactVersionOnly(t *testing.T) {
	const frozenID = "flow-root-id"
	const subflowID = "flow-subflow-id"

	ctx := constants.WithConnectorInboundCompanionFlowExecution(context.Background(), frozenID)

	if !constants.ConnectorInboundCompanionExecutionAuthorizesVersion(ctx, frozenID) {
		t.Error("expected true for the frozen root flow id")
	}
	if constants.ConnectorInboundCompanionExecutionAuthorizesVersion(ctx, subflowID) {
		t.Error("expected false for a different (subflow) id — subflow protection must hold")
	}
	if constants.ConnectorInboundCompanionExecutionAuthorizesVersion(ctx, "") {
		t.Error("expected false for empty id (fail-safe)")
	}
	// Without the marker: always false.
	plainCtx := context.Background()
	if constants.ConnectorInboundCompanionExecutionAuthorizesVersion(plainCtx, frozenID) {
		t.Error("expected false when ctx has no companion marker")
	}
}

// ── gate 3c: DB error on re-check → failTerminal ─

// TestWorker_CompanionEvent_ReCheckDBError_FailsTerminal (gate 3c):
// When checkCompanionInstalled returns a DB error, processOneWithDrain
// calls failTerminal (not a retriable fail).
func TestWorker_CompanionEvent_ReCheckDBError_FailsTerminal(t *testing.T) {
	ev := makeCompanionEvent("ev-gate3c", "connector-ns", "pack-install-1", "companion-install", "flow-v1")
	flow := &capturingVersionFlowInvoker{result: &FlowResult{IsSuccess: true}}
	resolver := &fakeRouteResolver{flowName: "companion-flow"}

	w, service, _ := newCompanionWorkItemWorker(t, ev, nil, flow, resolver)
	claims := service.claims
	dbErr := errors.New("transient DB error")
	checker := &scriptedCompanionChecker{results: []companionCheckResult{{present: false, err: dbErr}}}
	w.SetCompanionInstalledChecker(checker.check)

	claimed, err := w.processOne(context.Background())
	if !claimed {
		t.Fatal("expected claimed=true")
	}
	if err == nil {
		t.Fatal("expected error from failTerminal on DB error, got nil")
	}
	if flow.executeFlowVersionCalls != 0 {
		t.Errorf("flow should not have been invoked on DB error, got %d calls", flow.executeFlowVersionCalls)
	}
	if claims.claimCalls != 1 || claims.claimByEventCalls != 0 {
		t.Fatalf("WorkItem claims = direct %d, by-event %d; want 1, 0", claims.claimCalls, claims.claimByEventCalls)
	}
	_, claimedWork := assertCompanionClaimIdentity(t, service, ev)
	failure := assertCompanionFailureIdentity(t, claims, claimedWork, ev.ID)
	if failure.FailureClass != connector_workqueue.FailureClassInboundExecution || failure.FailureSummary != connector_workqueue.FailureSummaryInbound {
		t.Fatalf("WorkItem failure classification = %q/%q, want inbound execution/inbound", failure.FailureClass, failure.FailureSummary)
	}
	if len(claims.completeCalls) != 0 || claims.retryCalls != 0 {
		t.Fatalf("WorkItem terminal/retry calls = %v/%d, want none after fail", claims.completeCalls, claims.retryCalls)
	}
	checker.assertCalls(t, 1)
}

// fakeRouteResolver used in companion tests — returns the configured flowName.
type fakeRouteResolver struct {
	flowName string
}

func (r *fakeRouteResolver) ResolveFlow(_ context.Context, _, _, _ string) (int64, string, error) {
	return 1, r.flowName, nil
}

func (r *fakeRouteResolver) ResolveFlowByVersion(_ context.Context, _, _ string, _ int64) (string, error) {
	return r.flowName, nil
}

func (r *fakeRouteResolver) ResolveExecutionSpecByVersion(_ context.Context, _, _ string, _ int64) (connector_inbound_routing.RouteExecutionSpec, error) {
	return connector_inbound_routing.NormalizeRouteExecutionSpec(connector_inbound_routing.RouteExecutionSpec{FlowDeveloperName: r.flowName})
}