// NewExecutor creates an Executor and registers a callback on the
// system config.
func NewExecutor(db client.DB, gossip *gossip.Gossip, leaseMgr *LeaseManager, metaRegistry *metric.Registry, stopper *stop.Stopper) *Executor {
	exec := &Executor{
		db:       db,
		reCache:  parser.NewRegexpCache(512),
		leaseMgr: leaseMgr,

		latency: metaRegistry.Latency("sql.latency"),
	}
	exec.systemConfigCond = sync.NewCond(&exec.systemConfigMu)

	gossipUpdateC := gossip.RegisterSystemConfigChannel()
	stopper.RunWorker(func() {
		for {
			select {
			case <-gossipUpdateC:
				cfg := gossip.GetSystemConfig()
				exec.updateSystemConfig(cfg)
			case <-stopper.ShouldStop():
				return
			}
		}
	})

	return exec
}

// storeDescFromGossip retrieves a StoreDescriptor from the specified
// store gossip key. Returns an error if the gossip doesn't exist
// or is not a StoreDescriptor.
func storeDescFromGossip(key string, g *gossip.Gossip) (*proto.StoreDescriptor, error) {
	storeDesc := &proto.StoreDescriptor{}
	if err := g.GetInfoProto(key, storeDesc); err != nil {
		return nil, err
	}
	return storeDesc, nil
}

// newReplicateQueue returns a new instance of replicateQueue.
func newReplicateQueue(store *Store, g *gossip.Gossip, allocator Allocator, clock *hlc.Clock,
	options AllocatorOptions) *replicateQueue {
	rq := &replicateQueue{
		allocator:  allocator,
		clock:      clock,
		updateChan: make(chan struct{}, 1),
	}
	rq.baseQueue = makeBaseQueue("replicate", rq, store, g, queueConfig{
		maxSize:              replicateQueueMaxSize,
		needsLease:           true,
		acceptsUnsplitRanges: false,
	})

	if g != nil { // gossip is nil for some unittests
		// Register a gossip callback to signal queue that replicas in
		// purgatory might be retried due to new store gossip.
		g.RegisterCallback(gossip.MakePrefixPattern(gossip.KeyStorePrefix), func(_ string, _ roachpb.Value) {
			select {
			case rq.updateChan <- struct{}{}:
			default:
			}
		})
	}

	return rq
}

// computeSplitKeys returns an array of keys at which the supplied
// range should be split, as computed by intersecting the range with
// accounting and zone config map boundaries.
func computeSplitKeys(g *gossip.Gossip, rng *Range) []proto.Key {
	// Now split the range into pieces by intersecting it with the
	// boundaries of the config map.
	splitKeys := proto.KeySlice{}
	for _, configKey := range []string{gossip.KeyConfigAccounting, gossip.KeyConfigZone} {
		info, err := g.GetInfo(configKey)
		if err != nil {
			log.Errorf("unable to fetch %s config from gossip: %s", configKey, err)
			continue
		}
		configMap := info.(PrefixConfigMap)
		splits, err := configMap.SplitRangeByPrefixes(rng.Desc().StartKey, rng.Desc().EndKey)
		if err != nil {
			log.Errorf("unable to split %s by prefix map %s", rng, configMap)
			continue
		}
		// Gather new splits.
		for _, split := range splits {
			if split.end.Less(rng.Desc().EndKey) {
				splitKeys = append(splitKeys, split.end)
			}
		}
	}

	// Sort and unique the combined split keys from intersections with
	// both the accounting and zone config maps.
	sort.Sort(splitKeys)
	var unique []proto.Key
	for i, key := range splitKeys {
		if i == 0 || !key.Equal(splitKeys[i-1]) {
			unique = append(unique, key)
		}
	}
	return unique
}

// RefreshLeases starts a goroutine that refreshes the lease manager
// leases for tables received in the latest system configuration via gossip.
func (m *LeaseManager) RefreshLeases(s *stop.Stopper, db *client.DB, gossip *gossip.Gossip) {
	s.RunWorker(func() {
		descKeyPrefix := keys.MakeTablePrefix(uint32(sqlbase.DescriptorTable.ID))
		gossipUpdateC := gossip.RegisterSystemConfigChannel()
		for {
			select {
			case <-gossipUpdateC:
				cfg, _ := gossip.GetSystemConfig()
				if m.testingKnobs.GossipUpdateEvent != nil {
					m.testingKnobs.GossipUpdateEvent(cfg)
				}
				// Read all tables and their versions
				if log.V(2) {
					log.Info("received a new config; will refresh leases")
				}

				// Loop through the configuration to find all the tables.
				for _, kv := range cfg.Values {
					if !bytes.HasPrefix(kv.Key, descKeyPrefix) {
						continue
					}
					// Attempt to unmarshal config into a table/database descriptor.
					var descriptor sqlbase.Descriptor
					if err := kv.Value.GetProto(&descriptor); err != nil {
						log.Warningf("%s: unable to unmarshal descriptor %v", kv.Key, kv.Value)
						continue
					}
					switch union := descriptor.Union.(type) {
					case *sqlbase.Descriptor_Table:
						table := union.Table
						if err := table.Validate(); err != nil {
							log.Errorf("%s: received invalid table descriptor: %v", kv.Key, table)
							continue
						}
						if log.V(2) {
							log.Infof("%s: refreshing lease table: %d (%s), version: %d",
								kv.Key, table.ID, table.Name, table.Version)
						}
						// Try to refresh the table lease to one >= this version.
						if t := m.findTableState(table.ID, false /* create */, nil); t != nil {
							if err := t.purgeOldLeases(
								db, table.Deleted(), table.Version, m.LeaseStore); err != nil {
								log.Warningf("error purging leases for table %d(%s): %s",
									table.ID, table.Name, err)
							}
						}
					case *sqlbase.Descriptor_Database:
						// Ignore.
					}
				}
				if m.testingKnobs.TestingLeasesRefreshedEvent != nil {
					m.testingKnobs.TestingLeasesRefreshedEvent(cfg)
				}

			case <-s.ShouldStop():
				return
			}
		}
	})
}

// NewStorePool creates a StorePool and registers the store updating callback
// with gossip.
func NewStorePool(
	g *gossip.Gossip,
	clock *hlc.Clock,
	rpcContext *rpc.Context,
	reservationsEnabled bool,
	timeUntilStoreDead time.Duration,
	stopper *stop.Stopper,
) *StorePool {
	sp := &StorePool{
		clock:               clock,
		timeUntilStoreDead:  timeUntilStoreDead,
		rpcContext:          rpcContext,
		reservationsEnabled: reservationsEnabled,
		failedReservationsTimeout: envutil.EnvOrDefaultDuration("failed_reservation_timeout",
			defaultFailedReservationsTimeout),
		declinedReservationsTimeout: envutil.EnvOrDefaultDuration("declined_reservation_timeout",
			defaultDeclinedReservationsTimeout),
		reserveRPCTimeout: envutil.EnvOrDefaultDuration("reserve_rpc_timeout",
			defaultReserveRPCTimeout),
	}
	sp.mu.stores = make(map[roachpb.StoreID]*storeDetail)
	heap.Init(&sp.mu.queue)
	storeRegex := gossip.MakePrefixPattern(gossip.KeyStorePrefix)
	g.RegisterCallback(storeRegex, sp.storeGossipUpdate)
	sp.start(stopper)

	return sp
}

// NewExecutor creates an Executor and registers a callback on the
// system config.
func NewExecutor(db client.DB, gossip *gossip.Gossip, leaseMgr *LeaseManager, stopper *stop.Stopper) *Executor {
	registry := metric.NewRegistry()
	exec := &Executor{
		db:       db,
		reCache:  parser.NewRegexpCache(512),
		leaseMgr: leaseMgr,

		registry:      registry,
		latency:       registry.Latency("latency"),
		txnBeginCount: registry.Counter("transaction.begincount"),
		selectCount:   registry.Counter("select.count"),
		updateCount:   registry.Counter("update.count"),
		insertCount:   registry.Counter("insert.count"),
		deleteCount:   registry.Counter("delete.count"),
		ddlCount:      registry.Counter("ddl.count"),
		miscCount:     registry.Counter("misc.count"),
	}
	exec.systemConfigCond = sync.NewCond(&exec.systemConfigMu)

	gossipUpdateC := gossip.RegisterSystemConfigChannel()
	stopper.RunWorker(func() {
		for {
			select {
			case <-gossipUpdateC:
				cfg := gossip.GetSystemConfig()
				exec.updateSystemConfig(cfg)
			case <-stopper.ShouldStop():
				return
			}
		}
	})

	return exec
}

// NewStorePool creates a StorePool and registers the store updating callback
// with gossip.
func NewStorePool(
	g *gossip.Gossip,
	clock *hlc.Clock,
	rpcContext *rpc.Context,
	reservationsEnabled bool,
	timeUntilStoreDead time.Duration,
	stopper *stop.Stopper,
) *StorePool {
	sp := &StorePool{
		clock:               clock,
		timeUntilStoreDead:  timeUntilStoreDead,
		rpcContext:          rpcContext,
		reservationsEnabled: reservationsEnabled,
		failedReservationsTimeout: envutil.EnvOrDefaultDuration("COCKROACH_FAILED_RESERVATION_TIMEOUT",
			defaultFailedReservationsTimeout),
		declinedReservationsTimeout: envutil.EnvOrDefaultDuration("COCKROACH_DECLINED_RESERVATION_TIMEOUT",
			defaultDeclinedReservationsTimeout),
		reserveRPCTimeout: envutil.EnvOrDefaultDuration("COCKROACH_RESERVE_RPC_TIMEOUT",
			defaultReserveRPCTimeout),
		resolver: GossipAddressResolver(g),
	}
	sp.mu.stores = make(map[roachpb.StoreID]*storeDetail)
	heap.Init(&sp.mu.queue)
	storeRegex := gossip.MakePrefixPattern(gossip.KeyStorePrefix)
	g.RegisterCallback(storeRegex, sp.storeGossipUpdate)
	deadReplicasRegex := gossip.MakePrefixPattern(gossip.KeyDeadReplicasPrefix)
	g.RegisterCallback(deadReplicasRegex, sp.deadReplicasGossipUpdate)
	sp.start(stopper)

	return sp
}

func newStoreGossiper(g *gossip.Gossip) *storeGossiper {
	sg := &storeGossiper{
		g: g,
	}
	g.RegisterCallback(gossip.MakePrefixPattern(gossip.KeyStorePrefix), func(_ string, _ []byte) { sg.wg.Done() })
	return sg
}

// lookupZoneConfig returns the zone config matching the range.
func lookupZoneConfig(g *gossip.Gossip, repl *Replica) (config.ZoneConfig, error) {
	zoneMap, err := g.GetZoneConfig()
	if err != nil {
		return config.ZoneConfig{}, util.Errorf("unable to lookup zone config for range %s: %s", repl, err)
	}
	prefixConfig := zoneMap.MatchByPrefix(repl.Desc().StartKey)
	return *prefixConfig.Config.GetValue().(*config.ZoneConfig), nil
}

// lookupZoneConfig returns the zone config matching the range.
func lookupZoneConfig(g *gossip.Gossip, rng *Range) (proto.ZoneConfig, error) {
	zoneMap, err := g.GetInfo(gossip.KeyConfigZone)
	if err != nil || zoneMap == nil {
		return proto.ZoneConfig{}, util.Errorf("unable to lookup zone config for range %s: %s", rng, err)
	}
	prefixConfig := zoneMap.(PrefixConfigMap).MatchByPrefix(rng.Desc().StartKey)
	return *prefixConfig.Config.(*proto.ZoneConfig), nil
}

// RefreshLeases starts a goroutine that refreshes the lease manager
// leases for tables received in the latest system configuration via gossip.
func (m *LeaseManager) RefreshLeases(s *stop.Stopper, db *client.DB, gossip *gossip.Gossip) {
	s.RunWorker(func() {
		descKeyPrefix := keys.MakeTablePrefix(uint32(DescriptorTable.ID))
		gossipUpdateC := gossip.RegisterSystemConfigChannel()
		for {
			select {
			case <-gossipUpdateC:
				cfg := *gossip.GetSystemConfig()
				m.updateSystemConfig(cfg)

				// Read all tables and their versions
				if log.V(2) {
					log.Info("received a new config %v", cfg)
				}

				// Loop through the configuration to find all the tables.
				for _, kv := range cfg.Values {
					if !bytes.HasPrefix(kv.Key, descKeyPrefix) {
						continue
					}
					// Attempt to unmarshal config into a table/database descriptor.
					var descriptor Descriptor
					if err := kv.Value.GetProto(&descriptor); err != nil {
						log.Warningf("%s: unable to unmarshal descriptor %v", kv.Key, kv.Value)
						continue
					}
					switch union := descriptor.Union.(type) {
					case *Descriptor_Table:
						table := union.Table
						if err := table.Validate(); err != nil {
							log.Errorf("%s: received invalid table descriptor: %v", kv.Key, table)
							continue
						}
						if log.V(2) {
							log.Infof("%s: refreshing lease table: %d, version: %d",
								kv.Key, table.ID, table.Version)
						}
						// Try to refresh the table lease to one >= this version.
						if err := m.refreshLease(db, table.ID, table.Version); err != nil {
							log.Warningf("%s: %v", kv.Key, err)
						}

					case *Descriptor_Database:
						// Ignore.
					}
				}

			case <-s.ShouldStop():
				return
			}
		}
	})
}

// storeDescFromGossip retrieves a StoreDescriptor from the specified
// capacity gossip key. Returns an error if the gossip doesn't exist
// or is not a StoreDescriptor.
func storeDescFromGossip(key string, g *gossip.Gossip) (*proto.StoreDescriptor, error) {
	info, err := g.GetInfo(key)

	if err != nil {
		return nil, err
	}
	storeDesc, ok := info.(proto.StoreDescriptor)
	if !ok {
		return nil, fmt.Errorf("gossiped info is not a StoreDescriptor: %+v", info)
	}
	return &storeDesc, nil
}

// newStoreGossiper creates a store gossiper for use by tests. It adds the
// callback to gossip.
func newStoreGossiper(g *gossip.Gossip) *storeGossiper {
	sg := &storeGossiper{
		g:           g,
		storeKeyMap: make(map[string]struct{}),
	}
	g.RegisterCallback(gossip.MakePrefixPattern(gossip.KeyStorePrefix), func(key string, _ []byte) {
		sg.mu.Lock()
		defer sg.mu.Unlock()
		if _, ok := sg.storeKeyMap[key]; ok {
			sg.wg.Done()
		}
	})
	return sg
}

// NewStoreGossiper creates a store gossiper for use by tests. It adds the
// callback to gossip.
func NewStoreGossiper(g *gossip.Gossip) *StoreGossiper {
	sg := &StoreGossiper{
		g:           g,
		storeKeyMap: make(map[string]struct{}),
	}
	sg.cond = sync.NewCond(&sg.mu)
	g.RegisterCallback(gossip.MakePrefixPattern(gossip.KeyStorePrefix), func(key string, _ roachpb.Value) {
		sg.mu.Lock()
		defer sg.mu.Unlock()
		delete(sg.storeKeyMap, key)
		sg.cond.Broadcast()
	})
	return sg
}

// NewStorePool creates a StorePool and registers the store updating callback
// with gossip.
func NewStorePool(g *gossip.Gossip, timeUntilStoreDead time.Duration, stopper *stop.Stopper) *StorePool {
	sp := &StorePool{
		timeUntilStoreDead: timeUntilStoreDead,
		stores:             make(map[roachpb.StoreID]*storeDetail),
	}
	heap.Init(&sp.queue)

	storeRegex := gossip.MakePrefixPattern(gossip.KeyStorePrefix)
	g.RegisterCallback(storeRegex, sp.storeGossipUpdate)

	sp.start(stopper)

	return sp
}

// newExecutor creates an Executor and registers a callback on the
// system config.
func newExecutor(db client.DB, gossip *gossip.Gossip, clock *hlc.Clock) *Executor {
	exec := &Executor{
		db:      db,
		reCache: parser.NewRegexpCache(512),
	}
	gossip.RegisterSystemConfigCallback(exec.updateSystemConfig)
	return exec
}

func gossipStores(g *gossip.Gossip, stores []*proto.StoreDescriptor, t *testing.T) {
	var wg sync.WaitGroup
	wg.Add(len(stores))
	g.RegisterCallback(gossip.MakePrefixPattern(gossip.KeyCapacityPrefix), func(_ string, _ bool) { wg.Done() })

	for _, s := range stores {
		keyMaxCapacity := gossip.MakeCapacityKey(s.Node.NodeID, s.StoreID)
		// Gossip store descriptor.
		err := g.AddInfo(keyMaxCapacity, *s, 0)
		if err != nil {
			t.Fatal(err)
		}
	}

	// Wait for all gossip callbacks to be invoked.
	wg.Wait()
}

// newExecutor creates an Executor and registers a callback on the
// system config.
func newExecutor(db client.DB, gossip *gossip.Gossip, leaseMgr *LeaseManager) *Executor {
	exec := &Executor{
		db:       db,
		reCache:  parser.NewRegexpCache(512),
		leaseMgr: leaseMgr,
	}
	exec.systemConfigCond = sync.NewCond(&exec.systemConfigMu)
	gossip.RegisterSystemConfigCallback(exec.updateSystemConfig)
	return exec
}

// newReplicaSlice creates a replicaSlice from the replicas listed in the range
// descriptor and using gossip to lookup node descriptors. Replicas on nodes
// that are not gossipped are omitted from the result.
func newReplicaSlice(gossip *gossip.Gossip, desc *proto.RangeDescriptor) replicaSlice {
	replicas := make(replicaSlice, 0, len(desc.Replicas))
	for _, r := range desc.Replicas {
		nd, err := gossip.GetNodeDescriptor(r.NodeID)
		if err != nil {
			if log.V(1) {
				log.Infof("node %d is not gossiped: %v", r.NodeID, err)
			}
			continue
		}
		replicas = append(replicas, replicaInfo{
			Replica:  r,
			NodeDesc: nd,
		})
	}
	return replicas
}