Firestore 回复整个文档变更集/历史记录

Question

我有两个云函数，当用户使用我的前端 Web 应用进行身份验证时，它会在用户对象上设置一个标志，这会触发第一个云函数执行：

exports.observeUserUpdate = functions.firestore
    .document("/data/{uid}")
    .onUpdate(async (snapshot) => {
      const user = snapshot.after.data() as User;

      if (user.scanRepositories) {
        await pubSubClient
            .topic("user-repositories-update")
            .publish(Buffer.from(JSON.stringify(user)));
      }
    });

这个函数会将一个主题发送到一个队列，我有一个订阅者将接收该消息并执行一些后端操作，修改用户下的嵌套文档。

我有第二个云函数，它正在侦听用户文档下的更改，但只对我感兴趣的更改进行操作：

exports.observeRepositoryUpdate = functions.firestore
    .document("/data/{uid}/repositories/{repositoryId}")
    .onUpdate(async (snapshot) => {
      const repository = snapshot.after.data() as Repository;

      if (repository.task == null || repository.task.length == 0) {
        return;
      }

      if (repository.task.some((task: Task) => task.run)) {
        await pubSubClient
            .topic("repository-update")
            .publish(Buffer.from(JSON.stringify(repository)));
      }
    });

当文档发生变化并满足特定条件时，它将触发另一个主题以供不同的后端系统处理。

这一切都非常简单且有道理，出现问题的地方是用户正在退出，当他们重新登录时，并且设置了一个标志，正确地第一个函数触发并且后端执行它应该做的事情，但是，作为第二个函数的一部分发送的所有消息都被“重播”给该主题的使用者。

例如，用户登录并为repositories 创建tasks。这些任务已正确执行，并且一切正常。用户然后注销，然后重新登录，这些相同的任务在不应该被第二个函数再次触发。

看起来第二个函数正在触发整个变更集/历史记录。

是我的数据结构错误，还是我的文档路径不正确？希望得到一些帮助和道歉，这太罗嗦了。

编辑：为了帮助说明这里发生的事情的服务器日志中的顺序

// User logs in, login task fires as expected
2021-10-28 12:12:27.489  INFO 60338 --- [pool-1-thread-2] s.f.g.listeners.consumer.UserConsumer    : User consumer triggered: User(uid=g1OQ0K1...)
2021-10-28 12:12:27.492  INFO 60338 --- [pool-1-thread-2] s.f.github.services.GitHubServiceImpl    : Finding repositories
2021-10-28 12:12:31.954  INFO 60338 --- [pool-1-thread-2] s.f.g.r.FirestoreRepositoryImpl          : Batch write for 131 repositories
2021-10-28 12:13:05.827  INFO 60338 --- [pool-1-thread-2] s.f.g.r.FirestoreRepositoryImpl          : Batch write for 9 languages
2021-10-28 12:13:05.829  INFO 60338 --- [pool-1-thread-2] s.f.g.r.FirestoreRepositoryImpl          : Batch write for 5 types
// Login task complete 

// User runs task on repository
2021-10-28 12:13:50.392  INFO 60338 --- [pool-1-thread-2] s.f.g.l.consumer.RepositoryConsumer      : Repository consumer triggered: RepositoryDto(id=39931...)
2021-10-28 12:13:51.237  INFO 60338 --- [pool-1-thread-2] s.f.g.s.PackageManagerServiceImpl        : Detecting package manage for google-photo-frame with language Java
2021-10-28 12:13:52.319  INFO 60338 --- [pool-1-thread-2] s.f.g.r.FirestoreRepositoryImpl          : Batch write for 1 package managers
// Task completes

// User logs out and back in, the RepositoryConsumer is triggered again when it shouldnt
2021-10-28 12:15:09.917  INFO 60338 --- [pool-1-thread-5] s.f.g.l.consumer.RepositoryConsumer      : Repository consumer triggered: RepositoryDto(id=39931...)
2021-10-28 12:15:10.807  INFO 60338 --- [pool-1-thread-5] s.f.g.s.PackageManagerServiceImpl        : Detecting package manage for google-photo-frame with language Java
2021-10-28 12:15:11.872  INFO 60338 --- [pool-1-thread-5] s.f.g.r.FirestoreRepositoryImpl          : Batch write for 1 repositories
2021-10-28 12:15:12.262  INFO 60338 --- [pool-1-thread-5] s.f.g.listeners.consumer.UserConsumer    : User consumer triggered: User(uid=g1OQ0K1...)
2021-10-28 12:15:12.262  INFO 60338 --- [pool-1-thread-5] s.f.github.services.GitHubServiceImpl    : Finding repositories
2021-10-28 12:12:31.954  INFO 60338 --- [pool-1-thread-2] s.f.g.r.FirestoreRepositoryImpl          : Batch write for 131 repositories
2021-10-28 12:13:05.827  INFO 60338 --- [pool-1-thread-2] s.f.g.r.FirestoreRepositoryImpl          : Batch write for 9 languages
2021-10-28 12:13:05.829  INFO 60338 --- [pool-1-thread-2] s.f.g.r.FirestoreRepositoryImpl          : Batch write for 5 types

Answer 1

如documentation中所述：

默认情况下，订阅会在消息被确认后立即丢弃它们。未确认的消息默认保留 7 天（可通过订阅的 message_retention_duration 属性配置）。

因此，如果消息已被第二个函数确认，则重播消息似​​乎很奇怪，但它也指出：

将订阅配置为保留已确认的消息（通过retain_acked_messages 属性）可让您重播之前发送到订阅的确认消息。订阅中的消息最多可以保留 7 天，无论它们是已确认的还是未确认的。换言之，订阅中最早消息的使用期限不会超过 7 天。

这可以解释当用户注销并再次登录时您的行为，任务再次被触发。

正如GitHub repo 中所述，它的行为取决于retain_acked_messagesflag：

# Whether to retain acknowledged messages. If true, acknowledged messages
# will not be expunged until they fall out of the RetentionDuration window.
  subscription.retain_acked_messages: false

在Seeking to a snapshot 部分也指出：

创建快照后，它会保留：

• 创建快照时源订阅中未确认的所有消息。
• 此后发布到该主题的任何消息。

您可以通过使用快照来重放这些未确认的消息以查找任何主题的订阅。

为避免这种行为，我建议将retain_acked_messagesflag 配置为最适合您，同时使用filters：

您可以使用过滤器重播来自订阅的消息。如果您使用带有过滤器的订阅来查找时间戳，则 Pub/Sub 服务只会重新传递与过滤器匹配的消息。 带有过滤器的订阅快照包含以下消息：

• 比快照更新的所有消息，包括与过滤器不匹配的消息。
• 早于快照的未确认消息。