diff --git a/.changelog/28572.txt b/.changelog/28572.txt
new file mode 100644
index 00000000000..f0dedabebe1
--- /dev/null
+++ b/.changelog/28572.txt
@@ -0,0 +1,3 @@
+```release-note:bug
+resource/aws_sns_topic_subscription: Fix `filter_policy_scope` update from `MessageAttributes` to `MessageBody` with nested objects in `filter_policy`
+```
\ No newline at end of file
diff --git a/internal/service/sns/topic_subscription.go b/internal/service/sns/topic_subscription.go
index 1a258ac3643..a5f8502dfa5 100644
--- a/internal/service/sns/topic_subscription.go
+++ b/internal/service/sns/topic_subscription.go
@@ -263,6 +263,21 @@ func resourceTopicSubscriptionDelete(ctx context.Context, d *schema.ResourceData
 }
 
 func putSubscriptionAttributes(ctx context.Context, conn *sns.SNS, arn string, attributes map[string]string) error {
+	// Filter policy order matters
+	filterPolicyScope, ok := attributes[SubscriptionAttributeNameFilterPolicyScope]
+
+	if ok {
+		delete(attributes, SubscriptionAttributeNameFilterPolicyScope)
+	}
+
+	// MessageBody is backwards-compatible so it should always be applied first
+	if filterPolicyScope == SubscriptionFilterPolicyScopeMessageBody {
+		err := putSubscriptionAttribute(ctx, conn, arn, SubscriptionAttributeNameFilterPolicyScope, filterPolicyScope)
+		if err != nil {
+			return err
+		}
+	}
+
 	for name, value := range attributes {
 		err := putSubscriptionAttribute(ctx, conn, arn, name, value)
 
@@ -271,6 +286,16 @@ func putSubscriptionAttributes(ctx context.Context, conn *sns.SNS, arn string, a
 		}
 	}
 
+	// MessageAttributes isn't compatible with nested policies, so it should always be last
+	// in case the update also includes a change from a nested policy to a flat policy
+	if filterPolicyScope == SubscriptionFilterPolicyScopeMessageAttributes {
+		err := putSubscriptionAttribute(ctx, conn, arn, SubscriptionAttributeNameFilterPolicyScope, filterPolicyScope)
+
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
diff --git a/internal/service/sns/topic_subscription_test.go b/internal/service/sns/topic_subscription_test.go
index b1a7edd8a4d..84e185dbf3d 100644
--- a/internal/service/sns/topic_subscription_test.go
+++ b/internal/service/sns/topic_subscription_test.go
@@ -279,6 +279,56 @@ func TestAccSNSTopicSubscription_filterPolicyScope(t *testing.T) {
 					"endpoint_auto_confirms",
 				},
 			},
+			// Test transition from MessageAttributes to nested MessageBody ...
+			{
+				Config: testAccTopicSubscriptionConfig_filterPolicyScope(rName, strconv.Quote("MessageAttributes")),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckTopicSubscriptionExists(ctx, resourceName, &attributes),
+					resource.TestCheckResourceAttr(resourceName, "filter_policy_scope", "MessageAttributes"),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+				ImportStateVerifyIgnore: []string{
+					"confirmation_timeout_in_minutes",
+					"endpoint_auto_confirms",
+				},
+			},
+			{
+				Config: testAccTopicSubscriptionConfig_nestedFilterPolicyScope(rName, strconv.Quote("MessageBody"), true),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckTopicSubscriptionExists(ctx, resourceName, &attributes),
+					resource.TestCheckResourceAttr(resourceName, "filter_policy_scope", "MessageBody"),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+				ImportStateVerifyIgnore: []string{
+					"confirmation_timeout_in_minutes",
+					"endpoint_auto_confirms",
+				},
+			},
+			// ... and transition from nested MessageBody back to flat MessageAttributes
+			{
+				Config: testAccTopicSubscriptionConfig_filterPolicyScope(rName, strconv.Quote("MessageAttributes")),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckTopicSubscriptionExists(ctx, resourceName, &attributes),
+					resource.TestCheckResourceAttr(resourceName, "filter_policy_scope", "MessageAttributes"),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+				ImportStateVerifyIgnore: []string{
+					"confirmation_timeout_in_minutes",
+					"endpoint_auto_confirms",
+				},
+			},
 		},
 	})
 }
@@ -787,7 +837,11 @@ resource "aws_sns_topic_subscription" "test" {
 `, rName, policy)
 }
 
-func testAccTopicSubscriptionConfig_filterPolicyScope(rName, scope string) string {
+func testAccTopicSubscriptionConfig_nestedFilterPolicyScope(rName, scope string, nested bool) string {
+	filterPolicy := `jsonencode({"key1"=["value1"]})`
+	if nested {
+		filterPolicy = `jsonencode({"key2"={"key1"=["value1"]}})`
+	}
 	return fmt.Sprintf(`
 resource "aws_sns_topic" "test" {
   name = %[1]q
@@ -803,10 +857,13 @@ resource "aws_sns_topic_subscription" "test" {
   topic_arn           = aws_sns_topic.test.arn
   protocol            = "sqs"
   endpoint            = aws_sqs_queue.test.arn
-  filter_policy       = jsonencode({ key1 = ["value1"] })
-  filter_policy_scope = %[2]s
+  filter_policy       = %[2]s
+  filter_policy_scope = %[3]s
+}
+`, rName, filterPolicy, scope)
 }
-`, rName, scope)
+func testAccTopicSubscriptionConfig_filterPolicyScope(rName, scope string) string {
+	return testAccTopicSubscriptionConfig_nestedFilterPolicyScope(rName, scope, false)
 }
 
 func testAccTopicSubscriptionConfig_filterPolicyScope_policyNotSet(rName string) string {