// UpdateShardReplicationFields implements topo.Server.
func (s *Server) UpdateShardReplicationFields(ctx context.Context, cell, keyspace, shard string, updateFunc func(*topodatapb.ShardReplication) error) error {
	var sri *topo.ShardReplicationInfo
	var version int64
	var err error

	for {
		if sri, version, err = s.getShardReplication(cell, keyspace, shard); err != nil {
			if err == topo.ErrNoNode {
				// Pass an empty struct to the update func, as specified in topo.Server.
				sri = topo.NewShardReplicationInfo(&topodatapb.ShardReplication{}, cell, keyspace, shard)
				version = -1
			} else {
				return err
			}
		}
		if err = updateFunc(sri.ShardReplication); err != nil {
			return err
		}
		if version == -1 {
			if _, err = s.createShardReplication(sri); err != topo.ErrNodeExists {
				return err
			}
		} else {
			if _, err = s.updateShardReplication(sri, version); err != topo.ErrBadVersion {
				return err
			}
		}
	}
}

// GetShardReplication is part of the topo.Server interface
func (zkts *Server) GetShardReplication(ctx context.Context, cell, keyspace, shard string) (*topo.ShardReplicationInfo, error) {
	zkPath := shardReplicationPath(cell, keyspace, shard)
	data, _, err := zkts.zconn.Get(zkPath)
	if err != nil {
		return nil, convertError(err)
	}

	sr := &topodatapb.ShardReplication{}
	if err = json.Unmarshal([]byte(data), sr); err != nil {
		return nil, fmt.Errorf("bad ShardReplication data %v", err)
	}

	return topo.NewShardReplicationInfo(sr, cell, keyspace, shard), nil
}

func (zkts *Server) GetShardReplication(cell, keyspace, shard string) (*topo.ShardReplicationInfo, error) {
	zkPath := shardReplicationPath(cell, keyspace, shard)
	data, _, err := zkts.zconn.Get(zkPath)
	if err != nil {
		if zookeeper.IsError(err, zookeeper.ZNONODE) {
			err = topo.ErrNoNode
		}
		return nil, err
	}

	sr := &topo.ShardReplication{}
	if err = json.Unmarshal([]byte(data), sr); err != nil {
		return nil, fmt.Errorf("bad ShardReplication data %v", err)
	}

	return topo.NewShardReplicationInfo(sr, cell, keyspace, shard), nil
}

// GetShardReplication should return all the nodes in a shard,
// but instead we cheat for this test and just return all the
// tablets in the cell.
func (ft *fakeTopo) GetShardReplication(ctx context.Context, cell, keyspace, shard string) (*topo.ShardReplicationInfo, error) {
	if ft.expectGetTabletsByCell {
		return nil, fmt.Errorf("unexpected GetShardReplication")
	}

	ft.mu.RLock()
	defer ft.mu.RUnlock()
	nodes := make([]*topodatapb.ShardReplication_Node, 0, 1)
	for alias, tablet := range ft.tablets {
		if tablet.Alias.Cell == cell {
			nodes = append(nodes, &topodatapb.ShardReplication_Node{
				TabletAlias: &alias,
			})
		}
	}
	return topo.NewShardReplicationInfo(&topodatapb.ShardReplication{
		Nodes: nodes,
	}, cell, keyspace, shard), nil
}

func (s *Server) getShardReplication(cellName, keyspace, shard string) (*topo.ShardReplicationInfo, int64, error) {
	cell, err := s.getCell(cellName)
	if err != nil {
		return nil, -1, err
	}

	resp, err := cell.Get(shardReplicationFilePath(keyspace, shard), false /* sort */, false /* recursive */)
	if err != nil {
		return nil, -1, convertError(err)
	}
	if resp.Node == nil {
		return nil, -1, ErrBadResponse
	}

	value := &topodatapb.ShardReplication{}
	if err := json.Unmarshal([]byte(resp.Node.Value), value); err != nil {
		return nil, -1, fmt.Errorf("bad shard replication data (%v): %q", err, resp.Node.Value)
	}

	return topo.NewShardReplicationInfo(value, cellName, keyspace, shard), int64(resp.Node.ModifiedIndex), nil
}