package connector_inbound

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

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

type preparedClaimedEvent struct {
	ev           *connector_events.ConnectorEvent
	prep         preparedEvent
	details      map[string]json.RawMessage
	batchReadErr error
}

type pendingBatchGroup struct {
	key    string
	spec   connector_inbound_routing.RouteExecutionSpec
	events []preparedClaimedEvent
}

func (w *Worker) shouldBatchEvent(ev *connector_events.ConnectorEvent, prep preparedEvent) bool {
	return prep.spec.InvocationMode == connector_inbound_routing.InvocationModeBatchCollection
}

func (w *Worker) shouldCoalesceLatestBySource(prep preparedEvent) bool {
	return prep.spec.InvocationMode == connector_inbound_routing.InvocationModeLatestBySource ||
		prep.spec.InvocationMode == connector_inbound_routing.InvocationModeLatestBySourceFullRecord ||
		prep.spec.InvocationMode == connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayload ||
		prep.spec.InvocationMode == connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayloadSettled
}

func (w *Worker) coalesceLatestBySource(ctx context.Context, ev *connector_events.ConnectorEvent, prep preparedEvent) (*connector_events.ConnectorEvent, preparedEvent, int, error) {
	if !w.shouldCoalesceLatestBySource(prep) {
		return ev, prep, 0, nil
	}
	if prep.spec.InvocationMode == connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayloadSettled &&
		w.cfg.ClaimPath == ClaimPathWorkItem {
		payloads, err := settledProjectionPayloadInputs(ctx, w.events, ev, prep)
		if err != nil {
			return ev, prep, 0, w.failWithSyncState(ctx, ev, prep.objectKind, "load settled projection payloads", err)
		}
		prep.coalescedPayloads = payloads
	} else if preservesCoalescedPayloads(prep.spec) {
		prep.coalescedPayloads = []coalescedPayloadInput{coalescedPayloadInputForEvent(ev, prep)}
	}
	coalescer, ok := w.events.(sourceLaneCoalescingEventsService)
	if !ok {
		return ev, prep, 0, nil
	}
	// ClaimNextCoalescedSibling CAS-claims `next` to 'processing' and marks the
	// prior same-source row 'processed' in one tx. If prepareEvent over that
	// just-claimed sibling panics, the sibling would strand in 'processing' (the
	// caller's recover only sees the anchor, already 'processed' here). Track the
	// in-flight sibling and finalize it before re-raising; the worker's
	// per-iteration backstop then logs and keeps the goroutine alive (it finalizes
	// the anchor, already terminal).
	var inflightSibling *connector_events.ConnectorEvent
	defer func() {
		if r := recover(); r != nil {
			if inflightSibling != nil {
				_ = w.failWithSyncState(ctx, inflightSibling, "", "coalesce sibling panic recovery", fmt.Errorf("%w: %v", errWorkerPanic, r))
			}
			panic(r)
		}
	}()
	coalesced := 0
	sourceScoped := prep.spec.InvocationMode == connector_inbound_routing.InvocationModeLatestBySourceFullRecordPreservePayloadSettled
	for {
		var next *connector_events.ConnectorEvent
		var found bool
		var err error
		if sourceScoped {
			if scoped, ok := w.events.(sourceScopedLaneCoalescingEventsService); ok {
				next, found, err = scoped.ClaimNextCoalescedSiblingSourceScoped(ctx, ev.ID, ev.ClaimToken, w.cfg.WorkerID)
			} else {
				next, found, err = coalescer.ClaimNextCoalescedSibling(ctx, ev.ID, ev.ClaimToken, w.cfg.WorkerID)
			}
		} else {
			next, found, err = coalescer.ClaimNextCoalescedSibling(ctx, ev.ID, ev.ClaimToken, w.cfg.WorkerID)
		}
		if err != nil {
			return ev, prep, coalesced, w.failWithSyncState(ctx, ev, prep.objectKind, "coalesce source lane", err)
		}
		if !found {
			return ev, prep, coalesced, nil
		}
		coalesced++
		// `next` is now the only same-source row in 'processing' (the prior `ev`
		// was marked 'processed' by the claim above). Track it until prepareEvent
		// succeeds; on the next loop turn it is superseded, and on a clean return
		// the caller takes over tracking via `cev = *cevPtr`.
		inflightSibling = next
		nextPrep, prepErr := w.prepareEvent(ctx, next)
		if prepErr != nil {
			return next, preparedEvent{}, coalesced, prepErr
		}
		if len(prep.coalescedPayloads) > 0 {
			nextPrep.coalescedPayloads = append(prep.coalescedPayloads, coalescedPayloadInputForEvent(next, nextPrep))
		}
		ev = next
		prep = nextPrep
	}
}

func (w *Worker) appendPendingBatch(ctx context.Context, pending *pendingBatchGroup, item preparedClaimedEvent) (*pendingBatchGroup, error) {
	spec := item.prep.spec
	key := batchGroupKey(item.ev, item.prep.objectKind, spec)
	var flushErr error
	if pending != nil && (pending.key != key || len(pending.events) >= batchGroupLimit(w.cfg.BatchSize, pending.spec)) {
		flushErr = w.executeBatchCollection(ctx, pending)
		pending = nil
	}
	if pending == nil {
		pending = &pendingBatchGroup{key: key, spec: spec}
	}
	pending.events = append(pending.events, item)
	return pending, flushErr
}

func batchGroupKey(ev *connector_events.ConnectorEvent, objectKind string, spec connector_inbound_routing.RouteExecutionSpec) string {
	return ev.InstalledPackID + "\x00" +
		ev.EventKind + "\x00" +
		fmt.Sprint(ev.RouteVersion.Int64) + "\x00" +
		objectKind + "\x00" +
		spec.FlowDeveloperName + "\x00" +
		spec.BatchInputName + "\x00" +
		spec.BatchResultOutputName + "\x00" +
		fmt.Sprint(spec.MaxBatchSize) + "\x00" +
		ev.CompanionPackInstallID + "\x00" +
		ev.CompanionFlowVersionID
}

func batchGroupLimit(workerBatchSize int, spec connector_inbound_routing.RouteExecutionSpec) int {
	limit := spec.MaxBatchSize
	if limit <= 0 || limit > workerBatchSize {
		limit = workerBatchSize
	}
	if limit <= 0 {
		return 1
	}
	return limit
}

func (w *Worker) flushPendingBatch(ctx context.Context, pending *pendingBatchGroup) (*pendingBatchGroup, error) {
	if pending == nil || len(pending.events) == 0 {
		return nil, nil
	}
	return nil, w.executeBatchCollection(ctx, pending)
}

func sameDeliveryBatch(group *pendingBatchGroup, ev *connector_events.ConnectorEvent) bool {
	if group == nil || ev == nil || ev.DeliveryId == "" {
		return false
	}
	for _, item := range group.events {
		if item.ev != nil && item.ev.DeliveryId == ev.DeliveryId {
			return true
		}
	}
	return false
}

func sameDeliveryEvent(a, b *connector_events.ConnectorEvent) bool {
	return a != nil && b != nil && a.DeliveryId != "" && a.DeliveryId == b.DeliveryId
}

func (w *Worker) deferPendingBatchSameDelivery(ctx context.Context, pending *pendingBatchGroup, ev *connector_events.ConnectorEvent) bool {
	if !sameDeliveryBatch(pending, ev) {
		return false
	}
	if len(pending.events) >= batchGroupLimit(w.cfg.BatchSize, pending.spec) || !ev.RouteVersion.Valid || ev.RouteVersion.Int64 == 0 {
		return true
	}
	env, err := DecodeQueueEnvelope(ev.Payload)
	if err != nil {
		return true
	}
	spec, err := w.resolver.ResolveExecutionSpecByVersion(ctx, ev.InstalledPackID, ev.EventKind, ev.RouteVersion.Int64)
	if err != nil {
		return true
	}
	return spec.InvocationMode != connector_inbound_routing.InvocationModeBatchCollection ||
		spec.FlowExecutionFilter != nil ||
		batchGroupKey(ev, env.ObjectKind, spec) != pending.key
}

func (w *Worker) executeBatchCollection(ctx context.Context, group *pendingBatchGroup) (err error) {
	if group == nil || len(group.events) == 0 {
		return nil
	}
	execTiming := &connectorInboundExecutionTiming{
		startedAt:      time.Now(),
		flowPath:       "batch_collection",
		batchGroupSize: len(group.events),
	}
	ctx = withConnectorInboundExecutionTiming(ctx, execTiming)
	status := "success"
	defer func() {
		if err != nil {
			status = "error"
		}
		w.logBatchStageTiming(ctx, group, status, err)
	}()
	// Lease heartbeat over every group member's (eventID, claimToken): a slow batch
	// flow keeps all members' leases fresh so stale-reclaim cannot steal and
	// re-execute any of them. Stopped deterministically before any member finalizes.
	// A member released mid-loop (same-delivery ordering) simply gets ErrLeaseLost
	// on its next renewal — a harmless no-op.
	stopHeartbeat := w.startLeaseHeartbeat(leaseRefsForGroup(group))
	defer func() {
		heartbeatStopStarted := time.Now()
		stopHeartbeat()
		execTiming.heartbeatStopMS += connectorInboundTimingMillis(heartbeatStopStarted)
	}()

	spec := group.spec
	batchInputs := make([]any, 0, len(group.events))
	traceItems := make([]constants.ConnectorInboundTraceItem, 0, len(group.events))
	byID := make(map[string]preparedClaimedEvent, len(group.events))
	blockedDeliveries := make(map[string]bool)
	admittedDeliveries := make(map[string]bool)
	var first error
	for _, item := range group.events {
		if item.ev.DeliveryId != "" && blockedDeliveries[item.ev.DeliveryId] {
			finalizeStarted := time.Now()
			if rerr := w.releaseClaim(item.ev, "batch input deferred by same-delivery ordering", errors.New("earlier same-delivery batch member did not reach terminal success")); rerr != nil && first == nil {
				first = rerr
			}
			execTiming.finalizeMS += connectorInboundTimingMillis(finalizeStarted)
			continue
		}
		if item.batchReadErr != nil {
			finalizeStarted := time.Now()
			if item.ev.DeliveryId != "" && admittedDeliveries[item.ev.DeliveryId] {
				blockedDeliveries[item.ev.DeliveryId] = true
				if rerr := w.releaseClaim(item.ev, "batch read outcome deferred by same-delivery ordering", item.batchReadErr); rerr != nil && first == nil {
					first = rerr
				}
				execTiming.finalizeMS += connectorInboundTimingMillis(finalizeStarted)
				continue
			}
			if errors.Is(item.batchReadErr, ErrSourceRecordInactive) {
				if ferr := w.markConfirmedInactiveSkipped(ctx, item.ev, item.prep); ferr != nil {
					if first == nil {
						first = ferr
					}
					if item.ev.DeliveryId != "" {
						blockedDeliveries[item.ev.DeliveryId] = true
					}
				}
			} else {
				if ferr := w.failWithSyncState(ctx, item.ev, item.prep.objectKind, "batch read", item.batchReadErr); ferr != nil && first == nil {
					first = ferr
				}
				if item.ev.DeliveryId != "" {
					blockedDeliveries[item.ev.DeliveryId] = true
				}
			}
			execTiming.finalizeMS += connectorInboundTimingMillis(finalizeStarted)
			continue
		}
		buildInputsStarted := time.Now()
		inputs, err := w.buildFlowInputs(ctx, item.ev, item.prep.objectKind, item.prep.sourceID, item.prep.userPayload, item.details, item.prep.requiresEnrichmentRecord, item.prep.coalescedPayloads)
		execTiming.buildInputsMS += connectorInboundTimingMillis(buildInputsStarted)
		if err != nil {
			finalizeStarted := time.Now()
			if errors.Is(err, ErrBatchReadRecordShape) {
				if ferr := w.failWithSyncState(ctx, item.ev, item.prep.objectKind, "batch read", err); ferr != nil && first == nil {
					first = ferr
				}
			} else {
				if ferr := w.failTerminal(ctx, item.ev, "build flow inputs", err); ferr != nil && first == nil {
					first = ferr
				}
			}
			if item.ev.DeliveryId != "" {
				blockedDeliveries[item.ev.DeliveryId] = true
			}
			execTiming.finalizeMS += connectorInboundTimingMillis(finalizeStarted)
			continue
		}
		batchInputs = append(batchInputs, inputs)
		traceItems = append(traceItems, w.connectorInboundExecutionTraceItem(item.ev, item.prep, inputs))
		byID[item.ev.ID] = item
		if item.ev.DeliveryId != "" {
			admittedDeliveries[item.ev.DeliveryId] = true
		}
	}
	if len(batchInputs) == 0 {
		return first
	}

	ctx = context.WithValue(ctx, constants.ContextKeyConnectorSyncOrigin, w.cfg.Namespace)
	ctx = w.withConnectorInboundBatchExecutionTrace(ctx, traceItems)
	user := w.flow.SystemUser(ctx)
	inputs := map[string]any{spec.BatchInputName: batchInputs}
	var outcome *FlowDependencyOutcome
	var result *FlowResult
	var deadlineHit bool
	if group.events[0].ev.CompanionFlowVersionID != "" {
		companionVersionID := group.events[0].ev.CompanionFlowVersionID
		if checkErr := w.checkBatchCompanionIncarnation(ctx, group); checkErr != nil {
			return checkErr
		}
		ctx = constants.WithConnectorInboundCompanionFlowExecution(ctx, companionVersionID)
		flowCtx, cancelFlow := w.flowDeadlineContext(ctx)
		flowStarted := time.Now()
		outcome, err = w.safeExecuteFlowVersionOutcome(flowCtx, spec.FlowDeveloperName, companionVersionID, inputs, user)
		ctx = withFlowAttemptOutcome(ctx, outcome)
		result = flowResultFromOutcome(outcome)
		flowElapsed := time.Since(flowStarted)
		execTiming.flowCallMS = int64(flowElapsed / time.Millisecond)
		stagetiming.Add(ctx, stagetiming.StageFlowExec, flowElapsed)
		deadlineHit = isFlowDeadlineExceeded(flowCtx, err)
		cancelFlow()
	} else {
		flowCtx, cancelFlow := w.flowDeadlineContext(ctx)
		flowStarted := time.Now()
		outcome, err = w.safeExecuteFlowOutcome(flowCtx, spec.FlowDeveloperName, inputs, user)
		ctx = withFlowAttemptOutcome(ctx, outcome)
		result = flowResultFromOutcome(outcome)
		flowElapsed := time.Since(flowStarted)
		execTiming.flowCallMS = int64(flowElapsed / time.Millisecond)
		stagetiming.Add(ctx, stagetiming.StageFlowExec, flowElapsed)
		deadlineHit = isFlowDeadlineExceeded(flowCtx, err)
		cancelFlow()
	}
	if outcome != nil && outcome.Dependency != nil && err == nil && result != nil && result.IsSuccess {
		// A batch outcome has no unambiguous event selector to which the
		// dependency can be bound. Fail through the bounded failure budget
		// instead of releasing the batch forever without persisting a wakeable
		// dependency.
		return w.finalizeBatchFailureOrdered(ctx, group, byID, "batch identity convergence dependency", errors.New("per-event durable dependency output is required"))
	}
	if err != nil || result == nil || !result.IsSuccess {
		// A FlowDeadline timeout re-drives the WHOLE batch via
		// finalizeBatchRetriableOrdered (ReleaseClaimForRetry, no RetryCount burn) —
		// the batch mirror of the single-event deadline path — so a hung batch flow
		// kept alive by the lease heartbeat cannot pin its lane forever.
		if deadlineHit {
			return w.finalizeBatchRetriableOrdered(ctx, group, byID, "batch flow exec deadline", errFlowDeadlineExceeded)
		}
		// A retriable flow fault (typed cause or legacy IsRetriable result) re-drives
		// the whole batch with a backoff via ReleaseClaimForRetry
		// instead of MarkFailed, mirroring the single-event releaseClaimForRetry
		// path so a concurrent create-vs-adopt loser self-heals rather than
		// dead-lettering on the genuine-failure budget. The success+retriable case
		// below (#2128 missing-dependency) deliberately stays on MarkFailed.
		if flowOutcomeIsRetriable(outcome) {
			return w.finalizeBatchRetriableOrdered(ctx, group, byID, "batch flow exec", errors.New(flowRetriableMessage(spec.FlowDeveloperName, result)))
		}
		msg := flowFailureMessage(spec.FlowDeveloperName, result, err)
		return w.finalizeBatchFailureOrdered(ctx, group, byID, "batch flow exec", errors.New(msg))
	}
	if flowOutcomeIsRetriable(outcome) {
		msg := flowRetriableMessage(spec.FlowDeveloperName, result)
		return w.finalizeBatchFailureOrdered(ctx, group, byID, "batch flow exec", errors.New(msg))
	}

	results, err := batchEventResults(result.OutputValues, spec.BatchResultOutputName)
	if err != nil {
		return w.finalizeBatchFailureOrdered(ctx, group, byID, "batch flow result", err)
	}
	blockedDeliveries = make(map[string]bool)
	for _, item := range group.events {
		if _, ok := byID[item.ev.ID]; !ok {
			continue
		}
		if item.ev.DeliveryId != "" && blockedDeliveries[item.ev.DeliveryId] {
			finalizeStarted := time.Now()
			if rerr := w.releaseClaimForRetry(item.ev, "batch result deferred by same-delivery ordering", errors.New("earlier same-delivery batch member did not reach terminal success"), ctx); rerr != nil && first == nil {
				first = rerr
			}
			execTiming.finalizeMS += connectorInboundTimingMillis(finalizeStarted)
			continue
		}
		eventID := item.ev.ID
		outcome, ok := results[eventID]
		if !ok {
			err := fmt.Errorf("batch flow output %q missing result for eventID %q", spec.BatchResultOutputName, eventID)
			finalizeStarted := time.Now()
			if ferr := w.failWithSyncState(ctx, item.ev, item.prep.objectKind, "batch flow result", err); ferr != nil && first == nil {
				first = ferr
			}
			if item.ev.DeliveryId != "" {
				blockedDeliveries[item.ev.DeliveryId] = true
			}
			execTiming.finalizeMS += connectorInboundTimingMillis(finalizeStarted)
			continue
		}
		if outcome.success && !outcome.retriable {
			finalizeStarted := time.Now()
			if ferr := w.markProcessed(item.ev, ctx); ferr != nil {
				if first == nil {
					first = ferr
				}
				if item.ev.DeliveryId != "" {
					blockedDeliveries[item.ev.DeliveryId] = true
				}
			}
			execTiming.finalizeMS += connectorInboundTimingMillis(finalizeStarted)
			continue
		}
		msg := outcome.message
		if msg == "" {
			msg = "batch flow returned unsuccessful event result"
		}
		if outcome.retriable {
			msg = "retriable: " + msg
		}
		finalizeStarted := time.Now()
		if ferr := w.failWithSyncState(ctx, item.ev, item.prep.objectKind, "batch flow result", errors.New(msg)); ferr != nil && first == nil {
			first = ferr
		}
		if item.ev.DeliveryId != "" {
			blockedDeliveries[item.ev.DeliveryId] = true
		}
		execTiming.finalizeMS += connectorInboundTimingMillis(finalizeStarted)
	}
	return first
}

func (w *Worker) finalizeBatchFailureOrdered(ctx context.Context, group *pendingBatchGroup, byID map[string]preparedClaimedEvent, stage string, cause error) error {
	blockedDeliveries := make(map[string]bool)
	var first error
	for _, item := range group.events {
		if _, ok := byID[item.ev.ID]; !ok {
			continue
		}
		if item.ev.DeliveryId != "" && blockedDeliveries[item.ev.DeliveryId] {
			if rerr := w.releaseClaimForRetry(item.ev, stage+" deferred by same-delivery ordering", errors.New("earlier same-delivery batch member did not reach terminal success"), ctx); rerr != nil && first == nil {
				first = rerr
			}
			continue
		}
		if ferr := w.failWithSyncState(ctx, item.ev, item.prep.objectKind, stage, cause); ferr != nil && first == nil {
			first = ferr
		}
		if item.ev.DeliveryId != "" {
			blockedDeliveries[item.ev.DeliveryId] = true
		}
	}
	return first
}

// finalizeBatchRetriableOrdered re-drives every batch member for a backed-off
// retry via ReleaseClaimForRetry (events sets 'failed' + NextAttemptAt, no
// RetryCount increment) instead of MarkFailed. It is the batch counterpart of
// the single-event releaseClaimForRetry path, used for a batch-level retriable
// flow fault (IsSuccess=false, IsRetriable=true — the shape a FlowRetriableFault
// produces) so a concurrent create-vs-adopt loser self-heals rather than
// dead-lettering on the genuine-failure budget before the winner finalizes.
//
// The same-delivery ordering / blockedDeliveries handling is identical to
// finalizeBatchFailureOrdered: a non-terminal member blocks its later
// same-delivery siblings, which are released (no budget burn) to be re-evaluated
// on the next claim.
func (w *Worker) finalizeBatchRetriableOrdered(ctx context.Context, group *pendingBatchGroup, byID map[string]preparedClaimedEvent, stage string, cause error) error {
	blockedDeliveries := make(map[string]bool)
	var first error
	for _, item := range group.events {
		if _, ok := byID[item.ev.ID]; !ok {
			continue
		}
		if item.ev.DeliveryId != "" && blockedDeliveries[item.ev.DeliveryId] {
			if rerr := w.releaseClaimForRetry(item.ev, stage+" deferred by same-delivery ordering", errors.New("earlier same-delivery batch member did not reach terminal success"), ctx); rerr != nil && first == nil {
				first = rerr
			}
			continue
		}
		if ferr := w.releaseClaimForRetry(item.ev, stage, cause, ctx); ferr != nil && first == nil {
			first = ferr
		}
		if item.ev.DeliveryId != "" {
			blockedDeliveries[item.ev.DeliveryId] = true
		}
	}
	return first
}

type batchEventOutcome struct {
	success   bool
	retriable bool
	message   string
}

func batchEventResults(outputs map[string]any, outputName string) (map[string]batchEventOutcome, error) {
	if outputs == nil {
		return nil, fmt.Errorf("batch flow returned no output values")
	}
	raw, ok := outputs[outputName]
	if !ok {
		return nil, fmt.Errorf("batch flow output %q is missing", outputName)
	}
	items, err := batchEventResultItems(raw, outputName)
	if err != nil {
		return nil, err
	}
	out := make(map[string]batchEventOutcome, len(items))
	for i, item := range items {
		eventID, _ := item["eventID"].(string)
		if eventID == "" {
			return nil, fmt.Errorf("batch flow output %q item %d missing eventID", outputName, i)
		}
		success := boolValue(item, "success")
		if v, ok := item["isSuccess"]; ok {
			success = boolFromAny(v)
		}
		out[eventID] = batchEventOutcome{
			success:   success,
			retriable: boolValue(item, "retriable") || boolValue(item, "isRetriable"),
			message:   stringValue(item, "message"),
		}
	}
	return out, nil
}

func batchEventResultItems(raw any, outputName string) ([]map[string]any, error) {
	switch typed := raw.(type) {
	case []any:
		items := make([]map[string]any, 0, len(typed))
		for i, rawItem := range typed {
			item, err := batchEventResultItem(rawItem, outputName, i)
			if err != nil {
				return nil, err
			}
			items = append(items, item)
		}
		return items, nil
	case []map[string]any:
		return typed, nil
	case []models.SObject:
		items := make([]map[string]any, 0, len(typed))
		for _, item := range typed {
			items = append(items, map[string]any(item))
		}
		return items, nil
	default:
		return nil, fmt.Errorf("batch flow output %q must be a collection", outputName)
	}
}

func batchEventResultItem(raw any, outputName string, index int) (map[string]any, error) {
	switch item := raw.(type) {
	case map[string]any:
		return item, nil
	case models.SObject:
		return map[string]any(item), nil
	default:
		return nil, fmt.Errorf("batch flow output %q item %d must be an object", outputName, index)
	}
}

func boolValue(values map[string]any, key string) bool {
	v, ok := values[key]
	if !ok {
		return false
	}
	return boolFromAny(v)
}

func boolFromAny(v any) bool {
	b, _ := v.(bool)
	return b
}

func stringValue(values map[string]any, key string) string {
	v, _ := values[key].(string)
	return v
}