package peerstream

import (
	"context"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/golang/protobuf/proto"
	"github.com/hashicorp/consul/ipaddr"
	"github.com/hashicorp/consul/lib/retry"
	"github.com/hashicorp/go-hclog"
	"github.com/hashicorp/go-memdb"

	"github.com/hashicorp/consul/acl"
	"github.com/hashicorp/consul/agent/cache"
	"github.com/hashicorp/consul/agent/connect"
	"github.com/hashicorp/consul/agent/consul/autopilotevents"
	"github.com/hashicorp/consul/agent/consul/state"
	"github.com/hashicorp/consul/agent/consul/stream"
	"github.com/hashicorp/consul/agent/structs"
	"github.com/hashicorp/consul/agent/submatview"
	"github.com/hashicorp/consul/api"
	"github.com/hashicorp/consul/proto/pbcommon"
	"github.com/hashicorp/consul/proto/pbpeering"
	"github.com/hashicorp/consul/proto/pbpeerstream"
	"github.com/hashicorp/consul/proto/pbservice"
)

type MaterializedViewStore interface {
	Get(ctx context.Context, req submatview.Request) (submatview.Result, error)
	Notify(ctx context.Context, req submatview.Request, cID string, ch chan<- cache.UpdateEvent) error
}

type SubscriptionBackend interface {
	Subscriber
}

// subscriptionManager handlers requests to subscribe to events from an events publisher.
type subscriptionManager struct {
	logger               hclog.Logger
	config               Config
	trustDomain          string
	viewStore            MaterializedViewStore
	backend              SubscriptionBackend
	getStore             func() StateStore
	serviceSubReady      <-chan struct{}
	trustBundlesSubReady <-chan struct{}
	serverAddrsSubReady  <-chan struct{}
}

// TODO(peering): Maybe centralize so that there is a single manager per datacenter, rather than per peering.
func newSubscriptionManager(
	ctx context.Context,
	logger hclog.Logger,
	config Config,
	trustDomain string,
	backend SubscriptionBackend,
	getStore func() StateStore,
	remoteSubTracker *resourceSubscriptionTracker,
) *subscriptionManager {
	logger = logger.Named("subscriptions")
	store := submatview.NewStore(logger.Named("viewstore"))
	go store.Run(ctx)

	return &subscriptionManager{
		logger:               logger,
		config:               config,
		trustDomain:          trustDomain,
		viewStore:            store,
		backend:              backend,
		getStore:             getStore,
		serviceSubReady:      remoteSubTracker.SubscribedChan(pbpeerstream.TypeURLExportedService),
		trustBundlesSubReady: remoteSubTracker.SubscribedChan(pbpeerstream.TypeURLPeeringTrustBundle),
		serverAddrsSubReady:  remoteSubTracker.SubscribedChan(pbpeerstream.TypeURLPeeringServerAddresses),
	}
}

// subscribe returns a channel that will contain updates to exported service instances for a given peer.
func (m *subscriptionManager) subscribe(ctx context.Context, peerID, peerName, partition string) <-chan cache.UpdateEvent {
	var (
		updateCh       = make(chan cache.UpdateEvent, 1)
		publicUpdateCh = make(chan cache.UpdateEvent, 1)
	)

	state := newSubscriptionState(peerName, partition)
	state.publicUpdateCh = publicUpdateCh
	state.updateCh = updateCh

	// Wrap our bare state store queries in goroutines that emit events.
	go m.notifyExportedServicesForPeerID(ctx, state, peerID)
	go m.notifyServerAddrUpdates(ctx, state.updateCh)
	if m.config.ConnectEnabled {
		go m.notifyMeshGatewaysForPartition(ctx, state, state.partition)
		// If connect is enabled, watch for updates to CA roots.
		go m.notifyRootCAUpdatesForPartition(ctx, state.updateCh, state.partition)
	}

	// This goroutine is the only one allowed to manipulate protected
	// subscriptionManager fields.
	go m.handleEvents(ctx, state, updateCh)

	return publicUpdateCh
}

func (m *subscriptionManager) handleEvents(ctx context.Context, state *subscriptionState, updateCh <-chan cache.UpdateEvent) {
	for {
		// TODO(peering): exponential backoff

		select {
		case <-ctx.Done():
			return
		case update := <-updateCh:
			if err := m.handleEvent(ctx, state, update); err != nil {
				m.logger.Error("Failed to handle update from watch",
					"id", update.CorrelationID, "error", err,
				)
				continue
			}
		}
	}
}

func (m *subscriptionManager) handleEvent(ctx context.Context, state *subscriptionState, u cache.UpdateEvent) error {
	if u.Err != nil {
		return fmt.Errorf("received error event: %w", u.Err)
	}

	switch {
	case u.CorrelationID == subExportedServiceList:
		// Everything starts with the exported service list coming from
		// our state store watchset loop.
		evt, ok := u.Result.(*structs.ExportedServiceList)
		if !ok {
			return fmt.Errorf("invalid type for response: %T", u.Result)
		}

		state.exportList = evt

		pending := &pendingPayload{}
		m.syncNormalServices(ctx, state, evt.Services)
		if m.config.ConnectEnabled {
			m.syncDiscoveryChains(ctx, state, pending, evt.ListAllDiscoveryChains())
		}

		err := pending.Add(
			exportedServiceListID,
			subExportedServiceList,
			pbpeerstream.ExportedServiceListFromStruct(evt),
		)
		if err != nil {
			return err
		}

		state.sendPendingEvents(ctx, m.logger, pending)

		// cleanup event versions too
		state.cleanupEventVersions(m.logger)

	case strings.HasPrefix(u.CorrelationID, subExportedService):
		csn, ok := u.Result.(*pbservice.IndexedCheckServiceNodes)
		if !ok {
			return fmt.Errorf("invalid type for response: %T", u.Result)
		}

		// TODO(peering): is it safe to edit these protobufs in place?

		// Clear this raft index before exporting.
		csn.Index = 0

		// Ensure that connect things are scrubbed so we don't mix-and-match
		// with the synthetic entries that point to mesh gateways.
		filterConnectReferences(csn)

		// Flatten health checks
		for _, instance := range csn.Nodes {
			instance.Checks = flattenChecks(
				instance.Node.Node,
				instance.Service.ID,
				instance.Service.Service,
				instance.Service.EnterpriseMeta,
				instance.Checks,
			)
		}

		// Scrub raft indexes
		for _, instance := range csn.Nodes {
			instance.Node.RaftIndex = nil
			instance.Service.RaftIndex = nil
			// skip checks since we just generated one from scratch
		}

		id := servicePayloadIDPrefix + strings.TrimPrefix(u.CorrelationID, subExportedService)

		// Just ferry this one directly along to the destination.
		pending := &pendingPayload{}
		if err := pending.Add(id, u.CorrelationID, csn); err != nil {
			return err
		}
		state.sendPendingEvents(ctx, m.logger, pending)

	case strings.HasPrefix(u.CorrelationID, subMeshGateway):
		csn, ok := u.Result.(*pbservice.IndexedCheckServiceNodes)
		if !ok {
			return fmt.Errorf("invalid type for response: %T", u.Result)
		}

		partition := strings.TrimPrefix(u.CorrelationID, subMeshGateway)

		if !m.config.ConnectEnabled {
			return nil // ignore event
		}

		if !acl.EqualPartitions(partition, state.partition) {
			return nil // ignore event
		}

		// Clear this raft index before exporting.
		csn.Index = 0

		// Flatten health checks
		for _, instance := range csn.Nodes {
			instance.Checks = flattenChecks(
				instance.Node.Node,
				instance.Service.ID,
				instance.Service.Service,
				instance.Service.EnterpriseMeta,
				instance.Checks,
			)
		}

		// Scrub raft indexes
		for _, instance := range csn.Nodes {
			instance.Node.RaftIndex = nil
			instance.Service.RaftIndex = nil
			// skip checks since we just generated one from scratch

			// Remove connect things like native mode.
			if instance.Service.Connect != nil || instance.Service.Proxy != nil {
				instance.Service.Connect = nil
				instance.Service.Proxy = nil

				// VirtualIPs assigned in this cluster won't make sense on the importing side
				delete(instance.Service.TaggedAddresses, structs.TaggedAddressVirtualIP)
			}
		}

		state.meshGateway = csn

		pending := &pendingPayload{}

		if state.exportList != nil {
			// Trigger public events for all synthetic discovery chain replies.
			for chainName, info := range state.connectServices {
				m.collectPendingEventForDiscoveryChain(ctx, state, pending, chainName, info)
			}
		}

		// TODO(peering): should we ship this down verbatim to the consumer?
		state.sendPendingEvents(ctx, m.logger, pending)

	case u.CorrelationID == subCARoot:
		roots, ok := u.Result.(*pbpeering.PeeringTrustBundle)
		if !ok {
			return fmt.Errorf("invalid type for response: %T", u.Result)
		}
		pending := &pendingPayload{}
		if err := pending.Add(caRootsPayloadID, u.CorrelationID, roots); err != nil {
			return err
		}

		state.sendPendingEvents(ctx, m.logger, pending)

	case u.CorrelationID == subServerAddrs:
		addrs, ok := u.Result.(*pbpeering.PeeringServerAddresses)
		if !ok {
			return fmt.Errorf("invalid type for response: %T", u.Result)
		}
		pending := &pendingPayload{}
		if err := pending.Add(serverAddrsPayloadID, u.CorrelationID, addrs); err != nil {
			return err
		}

		state.sendPendingEvents(ctx, m.logger, pending)
	default:
		return fmt.Errorf("unknown correlation ID: %s", u.CorrelationID)
	}
	return nil
}

func filterConnectReferences(orig *pbservice.IndexedCheckServiceNodes) {
	newNodes := make([]*pbservice.CheckServiceNode, 0, len(orig.Nodes))
	for i := range orig.Nodes {
		csn := orig.Nodes[i]

		if csn.Service.Kind != string(structs.ServiceKindTypical) {
			continue // skip non-typical services
		}

		if strings.HasSuffix(csn.Service.Service, syntheticProxyNameSuffix) {
			// Skip things that might LOOK like a proxy so we don't get a
			// collision with the ones we generate.
			continue
		}

		// Remove connect things like native mode.
		if csn.Service.Connect != nil || csn.Service.Proxy != nil {
			csn = proto.Clone(csn).(*pbservice.CheckServiceNode)
			csn.Service.Connect = nil
			csn.Service.Proxy = nil

			// VirtualIPs assigned in this cluster won't make sense on the importing side
			delete(csn.Service.TaggedAddresses, structs.TaggedAddressVirtualIP)
		}

		newNodes = append(newNodes, csn)
	}
	orig.Nodes = newNodes
}

func (m *subscriptionManager) notifyRootCAUpdatesForPartition(
	ctx context.Context,
	updateCh chan<- cache.UpdateEvent,
	partition string,
) {
	// Wait until this is subscribed-to.
	select {
	case <-m.trustBundlesSubReady:
	case <-ctx.Done():
		return
	}

	var idx uint64
	// TODO(peering): retry logic; fail past a threshold
	for {
		var err error
		// Typically, this function will block inside `m.subscribeCARoots` and only return on error.
		// Errors are logged and the watch is retried.
		idx, err = m.subscribeCARoots(ctx, idx, updateCh, partition)
		if errors.Is(err, stream.ErrSubForceClosed) {
			m.logger.Trace("subscription force-closed due to an ACL change or snapshot restore, will attempt resume")
		} else if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
			m.logger.Warn("failed to subscribe to CA roots, will attempt resume", "error", err.Error())
		} else {
			m.logger.Trace(err.Error())
		}

		select {
		case <-ctx.Done():
			return
		default:
		}
	}
}

const subCARoot = "roots"

// subscribeCARoots subscribes to state.EventTopicCARoots for changes to CA roots.
// Upon receiving an event it will send the payload in updateCh.
func (m *subscriptionManager) subscribeCARoots(
	ctx context.Context,
	idx uint64,
	updateCh chan<- cache.UpdateEvent,
	partition string,
) (uint64, error) {
	// following code adapted from connectca/watch_roots.go
	sub, err := m.backend.Subscribe(&stream.SubscribeRequest{
		Topic:   state.EventTopicCARoots,
		Subject: stream.SubjectNone,
		Token:   "", // using anonymous token for now
		Index:   idx,
	})
	if err != nil {
		return 0, fmt.Errorf("failed to subscribe to CA Roots events: %w", err)
	}
	defer sub.Unsubscribe()

	for {
		event, err := sub.Next(ctx)
		switch {
		case errors.Is(err, stream.ErrSubForceClosed):
			// If the subscription was closed because the state store was abandoned (e.g.
			// following a snapshot restore) reset idx to ensure we don't skip over the
			// new store's events.
			select {
			case <-m.getStore().AbandonCh():
				idx = 0
			default:
			}
			return idx, err
		case errors.Is(err, context.Canceled):
			return 0, err
		case errors.Is(err, context.DeadlineExceeded):
			return 0, err
		case err != nil:
			return idx, fmt.Errorf("failed to read next event: %w", err)
		}

		// Note: this check isn't strictly necessary because the event publishing
		// machinery will ensure the index increases monotonically, but it can be
		// tricky to faithfully reproduce this in tests (e.g. the EventPublisher
		// garbage collects topic buffers and snapshots aggressively when streams
		// disconnect) so this avoids a bunch of confusing setup code.
		if event.Index <= idx {
			continue
		}

		idx = event.Index

		// We do not send framing events (e.g. EndOfSnapshot, NewSnapshotToFollow)
		// because we send a full list of roots on every event, rather than expecting
		// clients to maintain a state-machine in the way they do for service health.
		if event.IsFramingEvent() {
			continue
		}

		payload, ok := event.Payload.(state.EventPayloadCARoots)
		if !ok {
			return 0, fmt.Errorf("unexpected event payload type: %T", payload)
		}

		var rootPems []string
		for _, root := range payload.CARoots {
			rootPems = append(rootPems, root.RootCert)
		}

		updateCh <- cache.UpdateEvent{
			CorrelationID: subCARoot,
			Result: &pbpeering.PeeringTrustBundle{
				TrustDomain:       m.trustDomain,
				RootPEMs:          rootPems,
				ExportedPartition: partition,
				// TODO(peering): revisit decision not to validate datacenter in RBAC
			},
		}
	}
}

func (m *subscriptionManager) syncNormalServices(
	ctx context.Context,
	state *subscriptionState,
	services []structs.ServiceName,
) {
	// seen contains the set of exported service names and is used to reconcile the list of watched services.
	seen := make(map[structs.ServiceName]struct{})

	// Ensure there is a subscription for each service exported to the peer.
	for _, svc := range services {
		seen[svc] = struct{}{}

		if _, ok := state.watchedServices[svc]; ok {
			// Exported service is already being watched, nothing to do.
			continue
		}

		notifyCtx, cancel := context.WithCancel(ctx)
		if err := m.NotifyStandardService(notifyCtx, svc, state.updateCh); err != nil {
			cancel()
			m.logger.Error("failed to subscribe to service", "service", svc.String())
			continue
		}

		state.watchedServices[svc] = cancel
	}

	// For every subscription without an exported service, call the associated cancel fn.
	for svc, cancel := range state.watchedServices {
		if _, ok := seen[svc]; !ok {
			cancel()
			delete(state.watchedServices, svc)
		}
	}
}

func (m *subscriptionManager) syncDiscoveryChains(
	ctx context.Context,
	state *subscriptionState,
	pending *pendingPayload,
	chainsByName map[structs.ServiceName]structs.ExportedDiscoveryChainInfo,
) {
	// if it was newly added, then try to emit an UPDATE event
	for chainName, info := range chainsByName {
		if oldInfo, ok := state.connectServices[chainName]; ok && info.Equal(oldInfo) {
			continue
		}

		state.connectServices[chainName] = info

		m.collectPendingEventForDiscoveryChain(ctx, state, pending, chainName, info)
	}

	// if it was dropped, try to emit an DELETE event
	for chainName := range state.connectServices {
		if _, ok := chainsByName[chainName]; ok {
			continue
		}

		delete(state.connectServices, chainName)

		if state.meshGateway != nil {
			// Only need to clean this up if we know we may have ever sent it in the first place.
			proxyName := generateProxyNameForDiscoveryChain(chainName)
			err := pending.Add(
				discoveryChainPayloadIDPrefix+chainName.String(),
				subExportedService+proxyName.String(),
				&pbservice.IndexedCheckServiceNodes{},
			)
			if err != nil {
				m.logger.Error("failed to send event for discovery chain", "service", chainName.String(), "error", err)
				continue
			}
		}
	}
}

func (m *subscriptionManager) collectPendingEventForDiscoveryChain(
	ctx context.Context,
	state *subscriptionState,
	pending *pendingPayload,
	chainName structs.ServiceName,
	info structs.ExportedDiscoveryChainInfo,
) {
	if _, ok := state.connectServices[chainName]; !ok {
		return // not found
	}

	if state.exportList == nil || state.meshGateway == nil {
		return // skip because we don't have the data to do it yet
	}

	// Emit event with fake data
	proxyName := generateProxyNameForDiscoveryChain(chainName)

	err := pending.Add(
		discoveryChainPayloadIDPrefix+chainName.String(),
		subExportedService+proxyName.String(),
		createDiscoChainHealth(
			state.peerName,
			m.config.Datacenter,
			m.trustDomain,
			chainName,
			info,
			state.meshGateway,
		),
	)
	if err != nil {
		m.logger.Error("failed to send event for discovery chain", "service", chainName.String(), "error", err)
	}
}

func createDiscoChainHealth(
	peerName string,
	datacenter, trustDomain string,
	sn structs.ServiceName,
	info structs.ExportedDiscoveryChainInfo,
	pb *pbservice.IndexedCheckServiceNodes,
) *pbservice.IndexedCheckServiceNodes {
	fakeProxyName := sn.Name + syntheticProxyNameSuffix

	var peerMeta *pbservice.PeeringServiceMeta
	{
		spiffeID := connect.SpiffeIDService{
			Host:       trustDomain,
			Partition:  sn.PartitionOrDefault(),
			Namespace:  sn.NamespaceOrDefault(),
			Datacenter: datacenter,
			Service:    sn.Name,
		}
		mainSpiffeIDString := spiffeID.URI().String()

		sni := connect.PeeredServiceSNI(
			sn.Name,
			sn.NamespaceOrDefault(),
			sn.PartitionOrDefault(),
			peerName,
			trustDomain,
		)

		gwSpiffeID := connect.SpiffeIDMeshGateway{
			Host:       trustDomain,
			Partition:  sn.PartitionOrDefault(),
			Datacenter: datacenter,
		}

		// Create common peer meta.
		//
		// TODO(peering): should this be replicated by service and not by instance?
		peerMeta = &pbservice.PeeringServiceMeta{
			SNI: []string{sni},
			SpiffeID: []string{
				mainSpiffeIDString,
				// Always include the gateway id here to facilitate error-free
				// L4/L7 upgrade/downgrade scenarios.
				gwSpiffeID.URI().String(),
			},
			Protocol: info.Protocol,
		}

		if !structs.IsProtocolHTTPLike(info.Protocol) {
			for _, target := range info.TCPTargets {
				targetSpiffeID := connect.SpiffeIDService{
					Host:       trustDomain,
					Partition:  target.Partition,
					Namespace:  target.Namespace,
					Datacenter: target.Datacenter,
					Service:    target.Service,
				}
				targetSpiffeIDString := targetSpiffeID.URI().String()
				if targetSpiffeIDString != mainSpiffeIDString {
					peerMeta.SpiffeID = append(peerMeta.SpiffeID, targetSpiffeIDString)
				}
			}
		}
	}

	newNodes := make([]*pbservice.CheckServiceNode, 0, len(pb.Nodes))
	for i := range pb.Nodes {
		gwNode := pb.Nodes[i].Node
		gwService := pb.Nodes[i].Service
		gwChecks := pb.Nodes[i].Checks

		pbEntMeta := pbcommon.NewEnterpriseMetaFromStructs(sn.EnterpriseMeta)

		fakeProxyID := fakeProxyName
		destServiceID := sn.Name
		if gwService.ID != "" {
			// This is only going to be relevant if multiple mesh gateways are
			// on the same exporting node.
			fakeProxyID = fmt.Sprintf("%s-instance-%d", fakeProxyName, i)
			destServiceID = fmt.Sprintf("%s-instance-%d", sn.Name, i)
		}

		csn := &pbservice.CheckServiceNode{
			Node: gwNode,
			Service: &pbservice.NodeService{
				Kind:           string(structs.ServiceKindConnectProxy),
				Service:        fakeProxyName,
				ID:             fakeProxyID,
				EnterpriseMeta: pbEntMeta,
				PeerName:       structs.DefaultPeerKeyword,
				Proxy: &pbservice.ConnectProxyConfig{
					DestinationServiceName: sn.Name,
					DestinationServiceID:   destServiceID,
				},
				// direct
				Address:         gwService.Address,
				TaggedAddresses: gwService.TaggedAddresses,
				Port:            gwService.Port,
				SocketPath:      gwService.SocketPath,
				Weights:         gwService.Weights,
				Connect: &pbservice.ServiceConnect{
					PeerMeta: peerMeta,
				},
			},
			Checks: flattenChecks(gwNode.Node, fakeProxyID, fakeProxyName, pbEntMeta, gwChecks),
		}
		newNodes = append(newNodes, csn)
	}

	return &pbservice.IndexedCheckServiceNodes{
		Index: 0,
		Nodes: newNodes,
	}
}

func flattenChecks(
	nodeName string,
	serviceID string,
	serviceName string,
	entMeta *pbcommon.EnterpriseMeta,
	checks []*pbservice.HealthCheck,
) []*pbservice.HealthCheck {
	if len(checks) == 0 {
		return nil
	}

	healthStatus := api.HealthPassing
	for _, chk := range checks {
		if chk.Status != api.HealthPassing {
			healthStatus = chk.Status
		}
	}

	if serviceID == "" {
		serviceID = serviceName
	}

	return []*pbservice.HealthCheck{
		{
			CheckID:        serviceID + ":overall-check",
			Name:           "overall-check",
			Status:         healthStatus,
			Node:           nodeName,
			ServiceID:      serviceID,
			ServiceName:    serviceName,
			EnterpriseMeta: entMeta,
			PeerName:       structs.DefaultPeerKeyword,
		},
	}
}

const (
	subExportedServiceList = "exported-service-list"
	subExportedService     = "exported-service:"
	subMeshGateway         = "mesh-gateway:"
)

// NotifyStandardService will notify the given channel when there are updates
// to the requested service of the same name in the catalog.
func (m *subscriptionManager) NotifyStandardService(
	ctx context.Context,
	svc structs.ServiceName,
	updateCh chan<- cache.UpdateEvent,
) error {
	sr := newExportedStandardServiceRequest(m.logger, svc, m.backend)
	return m.viewStore.Notify(ctx, sr, subExportedService+svc.String(), updateCh)
}

// syntheticProxyNameSuffix is the suffix to add to synthetic proxies we
// replicate to route traffic to an exported discovery chain through the mesh
// gateways.
//
// This name was chosen to match existing "sidecar service" generation logic
// and similar logic in the Service Identity synthetic ACL policies.
const syntheticProxyNameSuffix = "-sidecar-proxy"

func generateProxyNameForDiscoveryChain(sn structs.ServiceName) structs.ServiceName {
	return structs.NewServiceName(sn.Name+syntheticProxyNameSuffix, &sn.EnterpriseMeta)
}

const subServerAddrs = "server-addrs"

func (m *subscriptionManager) notifyServerAddrUpdates(
	ctx context.Context,
	updateCh chan<- cache.UpdateEvent,
) {
	// Wait until server address updates are subscribed-to.
	select {
	case <-m.serverAddrsSubReady:
	case <-ctx.Done():
		return
	}

	configNotifyCh := m.notifyMeshConfigUpdates(ctx)

	// Intentionally initialized to empty values.
	// These are set after the first mesh config entry update arrives.
	var queryCtx context.Context
	cancel := func() {}

	useGateways := false
	for {
		select {
		case <-ctx.Done():
			cancel()
			return

		case event := <-configNotifyCh:
			entry, ok := event.Result.(*structs.MeshConfigEntry)
			if event.Result != nil && !ok {
				m.logger.Error(fmt.Sprintf("saw unexpected type %T for mesh config entry: falling back to pushing direct server addresses", event.Result))
			}
			if entry != nil && entry.Peering != nil && entry.Peering.PeerThroughMeshGateways {
				useGateways = true
			} else {
				useGateways = false
			}

			// Cancel and re-set watches based on the updated config entry.
			cancel()

			queryCtx, cancel = context.WithCancel(ctx)

			if useGateways {
				go m.notifyServerMeshGatewayAddresses(queryCtx, updateCh)
			} else {
				go m.ensureServerAddrSubscription(queryCtx, updateCh)
			}
		}
	}
}

func (m *subscriptionManager) notifyMeshConfigUpdates(ctx context.Context) <-chan cache.UpdateEvent {
	const meshConfigWatch = "mesh-config-entry"

	notifyCh := make(chan cache.UpdateEvent, 1)
	go m.syncViaBlockingQuery(ctx, meshConfigWatch, func(ctx_ context.Context, store StateStore, ws memdb.WatchSet) (interface{}, error) {
		_, rawEntry, err := store.ConfigEntry(ws, structs.MeshConfig, structs.MeshConfigMesh, acl.DefaultEnterpriseMeta())
		if err != nil {
			return nil, fmt.Errorf("failed to get mesh config entry: %w", err)

		}
		return rawEntry, nil
	}, meshConfigWatch, notifyCh)

	return notifyCh
}

func (m *subscriptionManager) notifyServerMeshGatewayAddresses(ctx context.Context, updateCh chan<- cache.UpdateEvent) {
	m.syncViaBlockingQuery(ctx, "mesh-gateways", func(ctx context.Context, store StateStore, ws memdb.WatchSet) (interface{}, error) {
		_, nodes, err := store.ServiceDump(ws, structs.ServiceKindMeshGateway, true, acl.DefaultEnterpriseMeta(), structs.DefaultPeerKeyword)
		if err != nil {
			return nil, fmt.Errorf("failed to watch mesh gateways services for servers: %w", err)
		}

		var gatewayAddrs []string
		for _, csn := range nodes {
			_, addr, port := csn.BestAddress(true)
			gatewayAddrs = append(gatewayAddrs, ipaddr.FormatAddressPort(addr, port))
		}
		if len(gatewayAddrs) == 0 {
			return nil, errors.New("configured to peer through mesh gateways but no mesh gateways are registered")
		}

		// We may return an empty list if there are no gateway addresses.
		return &pbpeering.PeeringServerAddresses{
			Addresses: gatewayAddrs,
		}, nil
	}, subServerAddrs, updateCh)
}

func (m *subscriptionManager) ensureServerAddrSubscription(ctx context.Context, updateCh chan<- cache.UpdateEvent) {
	waiter := &retry.Waiter{
		MinFailures: 1,
		Factor:      500 * time.Millisecond,
		MaxWait:     60 * time.Second,
		Jitter:      retry.NewJitter(100),
	}

	logger := m.logger.With("queryType", "server-addresses")

	var idx uint64
	for {
		var err error

		idx, err = m.subscribeServerAddrs(ctx, idx, updateCh)
		if err == nil {
			waiter.Reset()
		} else if errors.Is(err, stream.ErrSubForceClosed) {
			logger.Trace("subscription force-closed due to an ACL change or snapshot restore, will attempt resume")

		} else if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
			logger.Warn("failed to subscribe to server addresses, will attempt resume", "error", err.Error())

		} else if err != nil {
			logger.Trace(err.Error())
			return
		}
		if err := waiter.Wait(ctx); err != nil {
			return
		}
	}
}

func (m *subscriptionManager) subscribeServerAddrs(
	ctx context.Context,
	idx uint64,
	updateCh chan<- cache.UpdateEvent,
) (uint64, error) {
	// following code adapted from serverdiscovery/watch_servers.go
	sub, err := m.backend.Subscribe(&stream.SubscribeRequest{
		Topic:   autopilotevents.EventTopicReadyServers,
		Subject: stream.SubjectNone,
		Token:   "", // using anonymous token for now
		Index:   idx,
	})
	if err != nil {
		return 0, fmt.Errorf("failed to subscribe to ReadyServers events: %w", err)
	}
	defer sub.Unsubscribe()

	for {
		event, err := sub.Next(ctx)
		switch {
		case errors.Is(err, context.Canceled):
			return 0, err
		case err != nil:
			return idx, err
		}

		// We do not send framing events (e.g. EndOfSnapshot, NewSnapshotToFollow)
		// because we send a full list of ready servers on every event, rather than expecting
		// clients to maintain a state-machine in the way they do for service health.
		if event.IsFramingEvent() {
			continue
		}

		// Note: this check isn't strictly necessary because the event publishing
		// machinery will ensure the index increases monotonically, but it can be
		// tricky to faithfully reproduce this in tests (e.g. the EventPublisher
		// garbage collects topic buffers and snapshots aggressively when streams
		// disconnect) so this avoids a bunch of confusing setup code.
		if event.Index <= idx {
			continue
		}

		idx = event.Index

		payload, ok := event.Payload.(autopilotevents.EventPayloadReadyServers)
		if !ok {
			return 0, fmt.Errorf("unexpected event payload type: %T", payload)
		}

		var serverAddrs = make([]string, 0, len(payload))

		for _, srv := range payload {
			if srv.ExtGRPCPort == 0 {
				continue
			}
			grpcAddr := srv.Address + ":" + strconv.Itoa(srv.ExtGRPCPort)
			serverAddrs = append(serverAddrs, grpcAddr)
		}
		if len(serverAddrs) == 0 {
			m.logger.Warn("did not find any server addresses with external gRPC ports to publish")
			continue
		}

		u := cache.UpdateEvent{
			CorrelationID: subServerAddrs,
			Result: &pbpeering.PeeringServerAddresses{
				Addresses: serverAddrs,
			},
		}

		select {
		case <-ctx.Done():
			return 0, ctx.Err()
		case updateCh <- u:
		}
	}
}