ref(hc): Fix ArrayField for historic bad data (#57531)

Fixes
https://sentry.sentry.io/issues/4524783782/?query=is%3Aunresolved+outboxflusherror&referrer=issue-stream&statsPeriod=24h&stream_index=1

src/sentry/db/models/fields/array.py

@@ -58,5 +58,12 @@ class ArrayField(models.Field):
             except json.JSONDecodeError:
                 # This is to accommodate the erroneous exports pre 21.4.0
                 # See getsentry/sentry#23843 for more details
-                value = ast.literal_eval(value)
+                try:
+                    value = ast.literal_eval(value)
+                except ValueError:
+                    # this handles old database values using postgresql array format
+                    # see https://sentry.sentry.io/issues/4524783782/
+                    assert value[0] == "{" and value[-1] == "}", "Unexpected ArrayField format"
+                    assert "\\" not in value, "Unexpected ArrayField format"
+                    value = value[1:-1].split(",")
         return [self.of.to_python(x) for x in value]

tests/sentry/api/endpoints/test_organization_api_key_details.py

@@ -1,6 +1,8 @@
+from sentry.hybridcloud.models import ApiKeyReplica
 from sentry.models import ApiKey
+from sentry.silo import SiloMode
 from sentry.testutils.cases import APITestCase
-from sentry.testutils.silo import control_silo_test
+from sentry.testutils.silo import assume_test_silo_mode, control_silo_test
 
 DEFAULT_SCOPES = ["project:read", "event:read", "team:read", "org:read", "member:read"]
 
@@ -31,13 +33,38 @@ class OrganizationApiKeyDetailsPut(OrganizationApiKeyDetailsBase):
     method = "put"
 
     def test_update_api_key_details(self):
-        data = {"label": "New Label", "allowed_origins": "sentry.io"}
+        data = {
+            "label": "New Label",
+            "allowed_origins": "sentry.io",
+            "scope_list": ["a", "b", "c", "d"],
+        }
         self.get_success_response(self.organization.slug, self.api_key.id, **data)
 
         api_key = ApiKey.objects.get(id=self.api_key.id, organization_id=self.organization.id)
 
         assert api_key.label == "New Label"
         assert api_key.allowed_origins == "sentry.io"
+        assert api_key.get_scopes() == ["a", "b", "c", "d"]
+
+    def test_update_api_key_details_legacy_data(self):
+        # Some old api keys have this psql special format string
+        self.api_key.scope_list = "{event:read,member:read,org:read,project:read,team:read}"
+        self.api_key.save()
+
+        with assume_test_silo_mode(SiloMode.REGION):
+            assert ApiKeyReplica.objects.get(apikey_id=self.api_key.id).get_scopes() == [
+                "event:read",
+                "member:read",
+                "org:read",
+                "project:read",
+                "team:read",
+            ]
+
+        data = {"scope_list": ["a", "b", "c", "d"]}
+        self.get_success_response(self.organization.slug, self.api_key.id, **data)
+
+        api_key = ApiKey.objects.get(id=self.api_key.id, organization_id=self.organization.id)
+        assert api_key.get_scopes() == ["a", "b", "c", "d"]
 
 
 @control_silo_test(stable=True)

tests/sentry/hybrid_cloud/test_replica.py

@@ -1,3 +1,4 @@
+from sentry.hybridcloud.models import ApiKeyReplica
 from sentry.models import (
     AuthIdentity,
     AuthIdentityReplica,
@@ -62,6 +63,28 @@ def test_replicate_auth_provider():
     )
 
 
+@django_db_all(transaction=True)
+@all_silo_test(stable=True)
+def test_replicate_api_key():
+    org = Factories.create_organization()
+    with assume_test_silo_mode(SiloMode.CONTROL):
+        api_key = Factories.create_api_key(org, scope_list=["a", "b"])
+
+    with assume_test_silo_mode(SiloMode.REGION):
+        replicated = ApiKeyReplica.objects.get(apikey_id=api_key.id)
+
+    assert replicated.get_scopes() == api_key.get_scopes()
+
+    with assume_test_silo_mode(SiloMode.CONTROL):
+        api_key.scope_list = ["a", "b", "c"]
+        api_key.save()
+
+    with assume_test_silo_mode(SiloMode.REGION):
+        replicated = ApiKeyReplica.objects.get(apikey_id=api_key.id)
+
+    assert replicated.get_scopes() == api_key.get_scopes()
+
+
 @django_db_all(transaction=True)
 @all_silo_test(stable=True)
 def test_replicate_auth_identity():