feat: added idempotency + recovery checks in fulfillment

db/schema.sql

@@ -21,6 +21,8 @@ CREATE TABLE orders (
   status            order_status,
 
   -- References to the same order in Stripe and Printful.
+  -- `stripe_id` has an implicit unique index for efficient
+  -- queries.
   stripe_id         TEXT UNIQUE NOT NULL,
   printful_id       TEXT,
 
@@ -60,5 +62,6 @@ CREATE TABLE order_items (
 CREATE TABLE stripe_checkouts (
   -- The Stripe checkout session used for idempotency protection
   id TEXT PRIMARY KEY,
-  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+  is_processed BOOLEAN NOT NULL DEFAULT FALSE,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
 );

src/handlers/checkout.py

@@ -10,12 +10,11 @@
 
 from lib.types import StripeCheckout
 
-
 db = Database(url=os.getenv("DATABASE_URL"))
 stripe.api_key = get_api_key()
 
 
-def begin_fulfillment(session_id: str, event_type: str):
+def begin_fulfillment(session_id: str):
     """
     Begin fulfilling a checkout session on successful payment,
     using an idempotent operation as required by Stripe.
@@ -46,4 +45,4 @@ def process_webhook_request(payload: Any, signature: Any):
         event.type == "checkout.session.completed"
         or event.type == "checkout.session.async_payment_succeeded"
     ):
-        begin_fulfillment(event.data.object["id"], event.type)
+        begin_fulfillment(event.data.object["id"])

src/handlers/fulfillment.py

@@ -1,26 +1,86 @@
+import os
+
+import requests
 import stripe
 from lib.stripe import get_api_key
-from lib.types import StripeCheckout
-
+from lib.types import Order, OrderStatus, StripeCheckout
+from lib.db import Database
+from lib.printful import (
+    PRINTFUL_STATUS_MAP,
+    PrintfulClient,
+    PrintfulItem,
+    PrintfulRecipient,
+)
+from lib.logs import Logs
 
+db = Database(url=os.getenv("DATABASE_URL"))
+printful = PrintfulClient(api_key=os.getenv("PRINTFUL_API_KEY"))
+log = Logs(log_group=os.environ["LOG_GROUP"])
 stripe.api_key = get_api_key()
 
 
 def process_fulfillment(checkout: StripeCheckout):
     """
     Fulfills a successful checkout session.
     """
-    # TODO: Make sure fulfillment hasn't already been
-    # performed for this checkout.
-
+    # Fetch the items purchased and the receipt link.
     session = stripe.checkout.Session.retrieve(
         checkout.id,
-        expand=["line_items"],
+        expand=["line_items", "payment_intent.latest_charge"],
     )
 
     if session.payment_status == "unpaid":
         return
 
-    # TODO: Perform fulfillment of the line items
-    # TODO: Record/save fulfillment status for this
-    pass
+    log.info(session.line_items.data)
+    order = Order(
+        email=session.customer_details.email,
+        stripe_id=checkout.id,
+        receipt_url=session.payment_intent.latest_charge.receipt_url,
+        price=session.amount_total,
+        items=[
+            # TODO: Think about the items a little more.
+        ],
+    )
+
+    order, created = db.create_order(order)
+    if not created and order.printful_id is not None:
+        # Order was already fully fulfilled on a previous attempt.
+        return
+
+    shipping = session.shipping_details
+    recipient = PrintfulRecipient(
+        name=shipping.name,
+        address1=shipping.address.line1,
+        address2=shipping.address.line2,
+        city=shipping.address.city,
+        state_code=shipping.address.state,
+        country_code=shipping.address.country,
+        zip=shipping.address.postal_code,
+        email=session.customer_details.email,
+    )
+    items = [
+        PrintfulItem(product_id=int(item.product_id), quantity=item.quantity)
+        for item in order.items
+    ]
+
+    # Perform a lookup so we avoid creating duplicate errors if we already processed
+    # this in a previous call.
+    try:
+        result = printful.get_order_by_external_id(order.id)
+    except requests.HTTPError as e:
+        if e.response.status_code == 404:
+            result = printful.create_order(
+                recipient=recipient,
+                items=items,
+                external_id=checkout.id,
+            )
+        else:
+            raise e
+
+    printful_order = result["result"]
+    db.update_order(
+        order.id,
+        printful_id=str(printful_order["id"]),
+        status=PRINTFUL_STATUS_MAP.get(printful_order["status"], OrderStatus.pending),
+    )

src/handlers/printful.py

@@ -15,17 +15,6 @@
 notify = Notify(topic_arn=os.environ["SNS_TOPIC_ARN"], phone=os.environ["NOTIFY_PHONE"])
 queue = Queue(queue_url=os.environ["PRINTFUL_QUEUE_URL"])
 
-# Capture the states of an order from Printful, and for some of these, such as "onhold,"
-# we prefer to deal with them in the application-level, rather than persist such states
-# to the database.
-PRINTFUL_STATUS_MAP = {
-    "pending": OrderStatus.pending,
-    "inprocess": OrderStatus.pending,
-    "onhold": OrderStatus.pending,
-    "inreview": OrderStatus.pending,
-    "fulfilled": OrderStatus.fulfilled,
-    "failed": OrderStatus.failed,
-}
 
 SUPPORTED_EVENTS = {
     "package_shipped",

src/lib/db.py

@@ -1,7 +1,7 @@
 from datetime import datetime
 from typing import Optional
 
-from sqlalchemy import create_engine
+from sqlalchemy import create_engine, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -45,15 +45,24 @@ def upsert_shipment(self, shipment: Shipment) -> Shipment:
             session.refresh(shipment)
             return shipment
 
-    def create_order(self, order: Order) -> Order:
+    def create_order(self, order: Order) -> tuple[Order, bool]:
         """
         Write an order to the database, including its items.
+        Returns the order and whether it was newly created.
         """
         with Session(self.engine) as session:
-            session.add(order)
-            session.commit()
-            session.refresh(order)
-            return order
+            try:
+                session.add(order)
+                session.commit()
+                session.refresh(order)
+                return order, True
+            except IntegrityError:
+                session.rollback()
+                existing = session.execute(
+                    select(Order).where(Order.stripe_id == order.stripe_id)
+                ).scalar_one()
+
+                return existing, False
 
     def record_stripe_checkout(self, checkout: StripeCheckout) -> bool:
         """

src/lib/printful.py

@@ -2,6 +2,20 @@
 from typing import Optional
 import requests
 
+from lib.types import OrderStatus
+
+# Capture the states of an order from Printful, and for some of these, such as "onhold,"
+# we prefer to deal with them in the application-level, rather than persist such states
+# to the database.
+PRINTFUL_STATUS_MAP = {
+    "pending": OrderStatus.pending,
+    "inprocess": OrderStatus.pending,
+    "onhold": OrderStatus.pending,
+    "inreview": OrderStatus.pending,
+    "fulfilled": OrderStatus.fulfilled,
+    "failed": OrderStatus.failed,
+}
+
 
 @dataclass
 class PrintfulItem:
@@ -70,3 +84,9 @@ def get_order(self, order_id: int):
         Send a request to get the order with `order_id`.
         """
         return self._request("GET", f"/orders/{order_id}")
+
+    def get_order_by_external_id(self, external_id: str):
+        """
+        Send a request to get the order with `external_id`.
+        """
+        return self._request("GET", f"/orders/@{external_id}")

src/lib/types.py

@@ -61,4 +61,5 @@ class StripeCheckout(Base):
     __tablename__ = "stripe_checkouts"
 
     id: Mapped[str] = mapped_column(primary_key=True)
+    is_processed: Mapped[bool] = mapped_column(default=False)
     created_at: Mapped[datetime] = mapped_column(default=datetime.now)