fix(payment-methods): delete correct QR S3 keys, clean up replaced QR objects, return presigned URL

src/handlers/payment_methods_handlers.py

@@ -7,7 +7,7 @@
 
 import copy
 import os
-from typing import Any, Dict
+from typing import Any, Dict, Optional
 
 import boto3
 from botocore.exceptions import ClientError
@@ -147,8 +147,14 @@ def _validate_qr_upload_inputs(payment_method_name: str, s3_key: str, caller_id:
         raise AppError(ErrorCode.FORBIDDEN, "Invalid S3 key - access denied")
 
 
-def _update_payment_method_qr_url(caller_id: str, payment_method_name: str, s3_key: str) -> Dict[str, Any]:
-    """Update payment method with QR code S3 key and return updated method."""
+def _update_payment_method_qr_url(
+    caller_id: str, payment_method_name: str, s3_key: str
+) -> tuple[Dict[str, Any], Optional[str]]:
+    """Update payment method with QR code S3 key.
+
+    Returns (updated_method, previous_qr_key) so the caller can clean up the
+    replaced S3 object.
+    """
     account_id_key = f"ACCOUNT#{caller_id}"
     response = tables.accounts.get_item(Key={"accountId": account_id_key}, ConsistentRead=True)
 
@@ -159,8 +165,10 @@ def _update_payment_method_qr_url(caller_id: str, payment_method_name: str, s3_k
     existing_methods = list(preferences.get("paymentMethods", []))
 
     method_updated = None
+    previous_qr_key: Optional[str] = None
     for method in existing_methods:
         if method.get("name") == payment_method_name:
+            previous_qr_key = method.get("qrCodeUrl")
             method["qrCodeUrl"] = s3_key
             method_updated = method
             break
@@ -171,22 +179,24 @@ def _update_payment_method_qr_url(caller_id: str, payment_method_name: str, s3_k
     preferences["paymentMethods"] = existing_methods
     _save_preferences(account_id_key, response, preferences)
 
-    return dict(method_updated)
+    return dict(method_updated), previous_qr_key
 
 
 def confirm_qr_upload(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
     """
-    Confirm QR code upload and generate pre-signed GET URL.
+    Confirm QR code upload.
 
     AppSync Lambda resolver for confirmPaymentMethodQRCodeUpload mutation.
-    Validates S3 object exists, updates DynamoDB, returns pre-signed GET URL.
+    Validates S3 object exists, updates DynamoDB, and returns payment method with S3 key.
+    The AppSync field resolver for PaymentMethod.qrCodeUrl resolves the key to a
+    pre-signed GET URL.
 
     Args:
         event: AppSync event with identity and arguments
         context: Lambda context
 
     Returns:
-        PaymentMethod with name and qrCodeUrl (pre-signed GET URL)
+        PaymentMethod with name and qrCodeUrl (S3 key)
 
     Raises:
         AppError: If S3 object doesn't exist or update fails
@@ -204,7 +214,11 @@ def confirm_qr_upload(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
 
         bucket_name = get_required_env("EXPORTS_BUCKET")
         _validate_s3_object_exists(bucket_name, s3_key)
-        _update_payment_method_qr_url(caller_id, payment_method_name, s3_key)
+        _updated_method, previous_qr_key = _update_payment_method_qr_url(caller_id, payment_method_name, s3_key)
+
+        # Delete the replaced QR object so re-uploads don't orphan S3 objects
+        if previous_qr_key and previous_qr_key != s3_key:
+            _delete_qr_from_s3_storage(previous_qr_key, caller_id, payment_method_name, logger)
 
         logger.info(
             "Confirmed QR code upload",

src/utils/payment_methods.py

@@ -486,11 +486,19 @@ def _find_and_remove_method(
 
 def _delete_qr_if_exists(logger: Any, account_id: str, name: str, method_to_delete: Dict[str, Any]) -> None:
     """Delete QR code from S3 if it exists."""
-    if method_to_delete.get("qrCodeUrl"):
-        try:
+    stored_qr_key = method_to_delete.get("qrCodeUrl")
+    if not stored_qr_key:
+        return
+
+    try:
+        if stored_qr_key.startswith(f"{QR_CODE_S3_PREFIX}/"):
+            # Stored value is the actual S3 key (UUID-based uploads store the key directly)
+            delete_qr_by_key(stored_qr_key)
+        else:
+            # Legacy slug-based key or URL value
             delete_qr_from_s3(account_id, name)
-        except Exception as e:
-            logger.warning("Failed to delete QR code, continuing with method deletion", error=str(e))
+    except Exception as e:
+        logger.warning("Failed to delete QR code, continuing with method deletion", error=str(e))
 
 
 def delete_payment_method(account_id: str, name: str) -> None:

tests/unit/test_payment_methods.py

@@ -609,6 +609,39 @@ def test_delete_method_with_qr(
             s3_bucket.head_object(Bucket=bucket_name, Key=f"payment-qr-codes/{sample_account_id}/venmo.png")
         assert exc_info.value.response["Error"]["Code"] == "404"
 
+    def test_delete_method_with_uuid_qr_key(
+        self, dynamodb_tables: Dict[str, Any], sample_account: Dict[str, Any], s3_bucket: Any, sample_account_id: str
+    ) -> None:
+        """Test deleting a method deletes the stored UUID-based S3 key, not a slug-derived key."""
+        uuid_s3_key = f"payment-qr-codes/{sample_account_id}/{'a' * 32}.png"
+        slug_s3_key = f"payment-qr-codes/{sample_account_id}/venmo.png"
+
+        # Create method whose stored qrCodeUrl is a UUID-based key (as written by confirm_qr_upload)
+        tables_dict = dynamodb_tables
+        accounts_table = tables_dict["accounts"]
+        accounts_table.put_item(
+            Item={
+                "accountId": f"ACCOUNT#{sample_account_id}",
+                "preferences": {"paymentMethods": [{"name": "Venmo", "qrCodeUrl": uuid_s3_key}]},
+            }
+        )
+
+        # Upload the real object at the UUID key, plus a decoy at the legacy slug key
+        bucket_name = os.environ.get("EXPORTS_BUCKET")
+        s3_bucket.put_object(Bucket=bucket_name, Key=uuid_s3_key, Body=b"fake-qr-image")
+        s3_bucket.put_object(Bucket=bucket_name, Key=slug_s3_key, Body=b"decoy")
+
+        # Delete method
+        payment_methods.delete_payment_method(sample_account_id, "Venmo")
+
+        # The stored UUID key object must be deleted
+        with pytest.raises(ClientError) as exc_info:
+            s3_bucket.head_object(Bucket=bucket_name, Key=uuid_s3_key)
+        assert exc_info.value.response["Error"]["Code"] == "404"
+
+        # The unrelated slug-key object must be left alone
+        s3_bucket.head_object(Bucket=bucket_name, Key=slug_s3_key)
+
     def test_delete_nonexistent_method(
         self, dynamodb_tables: Dict[str, Any], sample_account: Dict[str, Any], sample_account_id: str
     ) -> None:

tests/unit/test_payment_methods_handlers.py

@@ -177,6 +177,92 @@ def test_confirm_upload_success(
         # Returns S3 key (field resolver will generate presigned URL)
         assert result["qrCodeUrl"] == s3_key
 
+    def test_confirm_upload_deletes_replaced_qr_object(
+        self, dynamodb_tables: Dict[str, Any], s3_bucket: Any, sample_account: Dict[str, Any], sample_account_id: str
+    ) -> None:
+        """Test re-uploading a QR code deletes the previous S3 object (no orphan)."""
+        from src.utils.dynamodb import tables
+
+        create_payment_method(sample_account_id, "Venmo")
+
+        # Previous upload: old UUID-based object stored on the method
+        old_s3_key = f"payment-qr-codes/{sample_account_id}/{'a' * 32}.png"
+        new_s3_key = f"payment-qr-codes/{sample_account_id}/{'b' * 32}.png"
+        bucket_name = os.environ.get("EXPORTS_BUCKET", "test-exports-bucket")
+        s3_bucket.put_object(Bucket=bucket_name, Key=old_s3_key, Body=b"old-qr-data")
+        s3_bucket.put_object(Bucket=bucket_name, Key=new_s3_key, Body=b"new-qr-data")
+
+        account_id_key = f"ACCOUNT#{sample_account_id}"
+        response = tables.accounts.get_item(Key={"accountId": account_id_key})
+        methods = response["Item"]["preferences"]["paymentMethods"]
+        for m in methods:
+            if m["name"] == "Venmo":
+                m["qrCodeUrl"] = old_s3_key
+        preferences = response["Item"].get("preferences", {})
+        preferences["paymentMethods"] = methods
+        tables.accounts.update_item(
+            Key={"accountId": account_id_key},
+            UpdateExpression="SET preferences = :prefs",
+            ExpressionAttributeValues={":prefs": preferences},
+        )
+
+        event = {
+            "identity": {"sub": sample_account_id},
+            "arguments": {"paymentMethodName": "Venmo", "s3Key": new_s3_key},
+        }
+
+        result = confirm_qr_upload(event, None)
+
+        assert result["qrCodeUrl"] is not None
+        # New object still exists
+        s3_bucket.head_object(Bucket=bucket_name, Key=new_s3_key)
+        # Old object was deleted
+        with pytest.raises(ClientError) as exc_info:
+            s3_bucket.head_object(Bucket=bucket_name, Key=old_s3_key)
+        assert exc_info.value.response["Error"]["Code"] == "404"
+        # Stored qrCodeUrl now points to the new key
+        response = tables.accounts.get_item(Key={"accountId": account_id_key})
+        methods = response["Item"]["preferences"]["paymentMethods"]
+        venmo = next(m for m in methods if m["name"] == "Venmo")
+        assert venmo["qrCodeUrl"] == new_s3_key
+
+    def test_confirm_upload_same_key_keeps_object(
+        self, dynamodb_tables: Dict[str, Any], s3_bucket: Any, sample_account: Dict[str, Any], sample_account_id: str
+    ) -> None:
+        """Test confirming with the same key as stored does not delete the object."""
+        from src.utils.dynamodb import tables
+
+        create_payment_method(sample_account_id, "Venmo")
+
+        s3_key = f"payment-qr-codes/{sample_account_id}/{'c' * 32}.png"
+        bucket_name = os.environ.get("EXPORTS_BUCKET", "test-exports-bucket")
+        s3_bucket.put_object(Bucket=bucket_name, Key=s3_key, Body=b"fake-qr-data")
+
+        account_id_key = f"ACCOUNT#{sample_account_id}"
+        response = tables.accounts.get_item(Key={"accountId": account_id_key})
+        methods = response["Item"]["preferences"]["paymentMethods"]
+        for m in methods:
+            if m["name"] == "Venmo":
+                m["qrCodeUrl"] = s3_key
+        preferences = response["Item"].get("preferences", {})
+        preferences["paymentMethods"] = methods
+        tables.accounts.update_item(
+            Key={"accountId": account_id_key},
+            UpdateExpression="SET preferences = :prefs",
+            ExpressionAttributeValues={":prefs": preferences},
+        )
+
+        event = {
+            "identity": {"sub": sample_account_id},
+            "arguments": {"paymentMethodName": "Venmo", "s3Key": s3_key},
+        }
+
+        result = confirm_qr_upload(event, None)
+
+        assert result["qrCodeUrl"] is not None
+        # Object was NOT deleted when the confirmed key matches the stored key
+        s3_bucket.head_object(Bucket=bucket_name, Key=s3_key)
+
     def test_confirm_upload_nonexistent_s3_object(
         self, dynamodb_tables: Dict[str, Any], s3_bucket: Any, sample_account: Dict[str, Any], sample_account_id: str
     ) -> None: