Method SyncUsersChunked

controller/rpc/user.go:45–88

(stream grpc.ClientStreamingServer[common.UsersChunk, common.Empty])

}

func (s *Service) SyncUsersChunked(stream grpc.ClientStreamingServer[common.UsersChunk, common.Empty]) error {
	chunks := make(map[uint64][]*common.User)
	var (
		lastIndex uint64
		sawLast   bool
	)

	for {
		chunk, err := stream.Recv()
		if errors.Is(err, io.EOF) {
			break
		}
		if err != nil {
			return status.Errorf(codes.Internal, "failed to receive chunk: %v", err)
		}

		chunks[chunk.GetIndex()] = append(chunks[chunk.GetIndex()], chunk.GetUsers()...)

		if chunk.GetLast() {
			sawLast = true
			lastIndex = chunk.GetIndex()
			break
		}
	}

	users, err := controller.BuildUsersFromChunks(chunks, lastIndex, sawLast)
	if err != nil {
		return status.Error(codes.InvalidArgument, err.Error())
	}

	// Large chunk: update in-memory then restart (no API calls).
	if len(users) > 100 {
		if err := s.Backend().UpdateUsersAndRestart(stream.Context(), users); err != nil {
			return status.Errorf(codes.Internal, "failed to update users: %v", err)
		}
	} else {
		// Small chunk: update via API without restart.
		if err := s.Backend().UpdateUsers(stream.Context(), users); err != nil {
			return status.Errorf(codes.Internal, "failed to update users: %v", err)
		}
	}

	return stream.SendAndClose(&common.Empty{})
}