Streams.scala

postsByAuthor
  .join(usersByKey,
        (posts: Set[Post], author: User) =>
          posts.map(DenormalisedPost(_, author, DenormalisedPost.Interactions(Set.empty, 0))))
  .toStream
  .flatMapValues(identity)
  .groupBy((_, denormalisedPost) => denormalisedPost.post.id)
  .reduce((first, second) => if (first.post.updatedOn.isAfter(second.post.updatedOn)) first else second)
  .leftJoin(likesByKey,
            (denormalisedPost: DenormalisedPost, likes: Set[Like]) =>
              Option(likes).fold(denormalisedPost)(denormalisedPost.lens(_.interactions.likes).set(_)))
  .leftJoin(commentCountByKey,
            (denormalisedPost: DenormalisedPost, commentCount: Int) =>
              denormalisedPost.lens(_.interactions.comments).set(commentCount))
  .toStream
  .to("denormalised-posts")