Stripe Webhook - 设计模式

Posted 2023-03-21

【中文标题】Stripe Webhook - 设计模式

【英文标题】：Stripe Webhooks - Design Pattern

【发布时间】：2021-06-30 15:21:16

【问题描述】：

我有一个处理 Stripe webhook 的设计，但我不确定这是否是处理它们的最佳方式。

订阅产品 Webhook 事件可以按任意顺序到达，并且可以有重复的事件 我正在使用 Flask，并在端点 (/webhooks) 中进行事件处理

当端点收到事件时，它会拉取与客户 ID 关联的所有事件。

event = stripe.Webhook.construct_event(
            payload, sig_header, endpoint_secret
        )

customer_id = event['data']['object']['customer']

#Filtering down to only the event types I need. I could do this from the Stripe dashboard as well
all_events_dict = json.loads(str(stripe.Event.list(related_object=customer_id, types=['...'])))

我有一些处理事件的逻辑，并将它们添加到数据库中，这样我就不会重新处理事件

events_to_process = []

    for event in all_events_dict['data']:
        event_id = event['id']

        if event_id not in events_to_process and not StripeEvent.query.filter_by(stripe_event_id=event_id).first():
            event_type = event['type']
            event_created = event['created']

            events_to_process.append(event_id)

            stripe_event = StripeEvent(stripe_event_id=event_id,
                                        event_type=event_type,
                                        event_created=event_created,
                                        event_json=str(event))
            db.session.add(stripe_event)
        
         db.session.commit()

问题 #1： 即使我正在检查事件是否已经存在（使用 StripeEvent.query），但我仍然经常收到 stripe_event_id 的 IntegrityError（重复键值违反唯一约束） ，主键。我认为这是因为 Stripe 有很多请求同时触发 /webhooks，因此有多个事件使其大致同时通过条件。

问题 #2： 我经常访问数据库，我认为是 Quadtratic N^2？对于每个事件，我为每个事件查询一次数据库。所以如果有 10 个事件，我会查询数据库 100 次。不确定这是否可以避免。

接下来我要为事件对象创建变量。我这样做，而不仅仅是遍历 all_events_dict，因为我希望能够控制事件的处理顺序。

customer_created = next((item for item in all_events_dict['data'] if item['type'] == 'customer.created' and item['id'] in events_to_process), None)

subscription_created = next((item for item in all_events_dict['data'] if item['type'] == 'customer.subscription.created'
                                and item['id'] in events_to_process and not Subscription.query.filter_by(id=item['data']['object']['id']).first()), None)

...

我依次浏览了我创建的变量

    if subscription_created:
        subscription_id = subscription_created['data']['object']['id']

        current_subscription = Subscription.query.filter_by(id=subscription_id).first()
        if not current_subscription:
            subscription = Subscription()
            
            # The processing logic is abstracted as function in the data model
            subscription.add_subscription(subscription_created)

            db.session.add(subscription)
            db.session.commit()

        else:
            current_subscription.update_subscription(subscription_created)

            db.session.commit()

    if subscription_updated:
        subscription_id = subscription_updated['data']['object']['id']
        event_created = subscription_updated['created']

        current_subscription = Subscription.query.filter_by(id=subscription_id).first()

        if current_subscription and event_created > current_subscription.last_updated:
            current_subscription.update_subscription(subscription_updated)

            db.session.commit()

问题 #4： 重复事件可能导致我在上面问题 #1 中描述的相同 IntegrityError。

问题 #5： 如果我按照步骤 1-3 中的描述处理事件（为每个事件提取每个事件），那么这可以正常工作。但是，subscription.updated 可以随时出现，包括在customer.subscription.created 事件之前。这意味着我必须添加逻辑以使用subscription.updated 事件将订阅添加到数据库。我认为这适用于订阅，但如果我将其扩展到其他事件类型，我认为在某些情况下....updated 事件不包含我需要的所有数据。

这些是我看到的一些问题，但总的来说，我正在寻找有关处理 Stripe 事件的设计的任何反馈。我可以完成上述工作，但似乎很笨拙。任何建议表示赞赏。

【问题讨论】：

这是全1功能吗？

都在路由中：@payments_bp.route('/webhooks', methods=['POST']) \n def webhooks():

我没有投反对票，但您的大部分问题可能更适合代码审查网站。您不应该在条带将事件发布到您的端点的同时获取事件。只需让 stripe 发送您订阅的每个事件并在收到它们时处理它们。

【参考方案1】：

这需要更多关注 - 这里的“理想”架构或模式取决于您的业务细节以及您使用/需要的确切信息。

但总的来说：您应该推迟后续处理。当您收到事件时不要查询任何内容，您应该检查签名，将此事件排队到其他地方并respond with a successful delivery response as quick as possible。

通过确认接收并将事件排队以进行异步处理，您可以更好地控制重复数据删除和同级事件。