feat: implement SC-9 add_indexer across contract, API, and SDKs (#1178)

Wire the existing Soroban add_indexer function through the Django ingest API, Python/TypeScript SDK clients, and CLI. Closes #1114.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: NanaKhadija1980j

django-backend/.env.example

@@ -77,6 +77,7 @@ SOROSCAN_CONTRACT_ID=CCAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA
 
 # Optional indexer signing key.
 INDEXER_SECRET_KEY=
+ADMIN_SECRET_KEY=
 
 TESTNET_RPC_URL=https://soroban-testnet.stellar.org
 MAINNET_RPC_URL=https://mainnet.stellar.validationcloud.io/v1/public

django-backend/soroscan/ingest/serializers.py

@@ -567,6 +567,12 @@ class RecordEventRequestSerializer(serializers.Serializer):
     )
 
 
+class AddIndexerRequestSerializer(serializers.Serializer):
+    """Serializer for SC-9: authorize an indexer on the SoroScan contract."""
+
+    indexer_address = serializers.CharField(
+        max_length=56,
+        help_text="Stellar address of the indexer to authorize",
 class StructuredEventRequestSerializer(RecordEventRequestSerializer):
     """SC-38 request payload for versioned, idempotent contract events."""
 

django-backend/soroscan/ingest/stellar_client.py

@@ -144,6 +144,25 @@ def _bytes_to_sc_val(self, data: bytes) -> SCVal:
             bytes=SCBytes(data),
         )
 
+    def _get_admin_keypair(self) -> Optional[Keypair]:
+        admin_secret = getattr(settings, "ADMIN_SECRET_KEY", "") or self.secret_key
+        if not admin_secret:
+            return None
+        return Keypair.from_secret(admin_secret)
+
+    def _submit_contract_transaction(
+        self,
+        function_name: str,
+        parameters: list[SCVal],
+        signer: Keypair,
+    ) -> TransactionResult:
+        """Simulate, prepare, and submit a Soroban contract write transaction."""
+        try:
+            account = self.server.load_account(signer.public_key)
+            tx_builder = TransactionBuilder(
+                source_account=account,
+                network_passphrase=self.network_passphrase,
+                base_fee=100000,
     def _simulate_contract_read(
         self,
         function_name: str,
@@ -169,6 +188,60 @@ def _simulate_contract_read(
             simulate_response = self.server.simulate_transaction(tx)
 
             if simulate_response.error:
+                return TransactionResult(
+                    success=False,
+                    tx_hash="",
+                    status="simulation_failed",
+                    error=simulate_response.error,
+                )
+
+            prepared_tx = self.server.prepare_transaction(tx, simulate_response)
+            prepared_tx.sign(signer)
+            send_response = self.server.send_transaction(prepared_tx)
+
+            logger.info(
+                "Contract transaction submitted: %s (%s)",
+                send_response.hash,
+                function_name,
+            )
+
+            return TransactionResult(
+                success=send_response.status == "PENDING",
+                tx_hash=send_response.hash,
+                status=send_response.status,
+                result_xdr=getattr(send_response, "result_xdr", None),
+            )
+        except Exception as exc:
+            logger.exception("Failed to submit contract transaction: %s", function_name)
+            return TransactionResult(
+                success=False,
+                tx_hash="",
+                status="error",
+                error=str(exc),
+            )
+
+    def add_indexer(self, indexer_address: str) -> TransactionResult:
+        """
+        Submit an add_indexer transaction to the SoroScan contract (SC-9).
+
+        The configured admin keypair must sign the transaction.
+        """
+        admin_keypair = self._get_admin_keypair()
+        if not admin_keypair:
+            return TransactionResult(
+                success=False,
+                tx_hash="",
+                status="error",
+                error="No admin keypair configured",
+            )
+
+        return self._submit_contract_transaction(
+            function_name="add_indexer",
+            parameters=[
+                self._address_to_sc_val(admin_keypair.public_key),
+                self._address_to_sc_val(indexer_address),
+            ],
+            signer=admin_keypair,
                 return False, simulate_response.error
 
             results = getattr(simulate_response, "results", None) or []

django-backend/soroscan/ingest/tests/test_stellar_client.py

@@ -177,3 +177,31 @@ def test_get_total_events(self, client):
         result = client.get_total_events()
 
         assert result is None
+
+    def test_add_indexer_no_admin_keypair(self, mock_server, hex_contract_id):
+        client = SorobanClient(
+            rpc_url="https://soroban-testnet.stellar.org",
+            contract_id=hex_contract_id,
+            secret_key="",
+        )
+        result = client.add_indexer("G" + "A" * 54)
+        assert result.success is False
+        assert result.error == "No admin keypair configured"
+
+    def test_add_indexer_success(self, client, hex_contract_id):
+        mock_account = MagicMock()
+        mock_account.sequence = 1
+        client.server = MagicMock()
+        client.server.load_account.return_value = mock_account
+
+        mock_simulate_response = MagicMock()
+        mock_simulate_response.error = None
+        client.server.simulate_transaction.return_value = mock_simulate_response
+        client.server.prepare_transaction.return_value = MagicMock()
+        client.server.send_transaction.return_value = MagicMock(
+            status="PENDING", hash="addindexer123"
+        )
+
+        result = client.add_indexer("G" + "A" * 54)
+        assert result.success is True
+        assert result.tx_hash == "addindexer123"

django-backend/soroscan/ingest/tests/test_views.py

@@ -403,6 +403,28 @@ def test_record_event_validation_error(self, authenticated_client):
         assert response.status_code == status.HTTP_400_BAD_REQUEST
         assert "contract_id" in response.data
 
+    @responses.activate
+    def test_add_indexer_success(self, authenticated_client):
+        responses.add(
+            responses.POST,
+            "https://soroban-testnet.stellar.org/",
+            json={"status": "PENDING", "hash": "indexeradd123"},
+            status=200,
+        )
+
+        url = reverse("add-indexer")
+        data = {"indexer_address": "G" + "A" * 54}
+        response = authenticated_client.post(url, data, format="json")
+
+        assert response.status_code in [status.HTTP_202_ACCEPTED, status.HTTP_400_BAD_REQUEST]
+
+    def test_add_indexer_validation_error(self, authenticated_client):
+        url = reverse("add-indexer")
+        response = authenticated_client.post(url, {}, format="json")
+
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert "indexer_address" in response.data
+
 
 @pytest.mark.django_db
 class TestWebhookPingEndpoint:

django-backend/soroscan/ingest/urls.py

@@ -29,6 +29,7 @@
     health_check,
     networks_view,
     record_event_view,
+    add_indexer_view,
     is_indexer_view,
     get_admin_view,
     record_structured_event_view,
@@ -93,6 +94,7 @@
     ),
     path("", include(router.urls)),
     path("record/", record_event_view, name="record-event"),
+    path("indexers/add/", add_indexer_view, name="add-indexer"),
     path("indexers/check/", is_indexer_view, name="is-indexer"),
     path("contract/admin/", get_admin_view, name="get-admin"),
     path("record/structured/", record_structured_event_view, name="record-structured-event"),

django-backend/soroscan/ingest/views.py

@@ -67,6 +67,7 @@
     OrganizationCorsSerializer,
     OrganizationCostSnapshotSerializer,
     RecordEventRequestSerializer,
+    AddIndexerRequestSerializer,
     StructuredEventRequestSerializer,
     TeamMemberAddSerializer,
     TeamSerializer,
@@ -1063,6 +1064,77 @@ def record_structured_event_view(request):
 
 
 @extend_schema(
+    request=AddIndexerRequestSerializer,
+    responses={
+        202: inline_serializer(
+            name="AddIndexerResponse",
+            fields={
+                "status": serializers.CharField(),
+                "tx_hash": serializers.CharField(),
+                "transaction_status": serializers.CharField(),
+            },
+        ),
+        400: inline_serializer(
+            name="AddIndexerFailed",
+            fields={
+                "status": serializers.CharField(),
+                "error": serializers.CharField(),
+                "transaction_status": serializers.CharField(),
+            },
+        ),
+    },
+)
+@api_view(["POST"])
+@permission_classes([IsAuthenticated])
+@throttle_classes([IngestRateThrottle, AnonRateThrottle, UserRateThrottle])
+def add_indexer_view(request):
+    """
+    Authorize an indexer address on the SoroScan contract (SC-9).
+
+    Request body:
+    {
+        "indexer_address": "GABC..."
+    }
+    """
+    serializer = AddIndexerRequestSerializer(data=request.data)
+
+    if not serializer.is_valid():
+        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+    indexer_address = serializer.validated_data["indexer_address"]
+
+    try:
+        client = SorobanClient()
+        result = client.add_indexer(indexer_address=indexer_address)
+
+        if result.success:
+            return Response(
+                {
+                    "status": "submitted",
+                    "tx_hash": result.tx_hash,
+                    "transaction_status": result.status,
+                },
+                status=status.HTTP_202_ACCEPTED,
+            )
+
+        return Response(
+            {
+                "status": "failed",
+                "error": result.error,
+                "transaction_status": result.status,
+            },
+            status=status.HTTP_400_BAD_REQUEST,
+        )
+
+    except Exception as e:
+        logger.exception(
+            "Failed to add indexer",
+            extra={"indexer_address": indexer_address},
+        )
+        return Response(
+            {"status": "error", "error": str(e)},
+            status=status.HTTP_500_INTERNAL_SERVER_ERROR,
+        )
     parameters=[
         OpenApiParameter(
             name="indexer_address",

django-backend/soroscan/settings.py

@@ -460,6 +460,7 @@ def _load_software_version() -> str:
 )
 SOROSCAN_CONTRACT_ID = env("SOROSCAN_CONTRACT_ID", default="")
 INDEXER_SECRET_KEY = env("INDEXER_SECRET_KEY", default="")
+ADMIN_SECRET_KEY = env("ADMIN_SECRET_KEY", default=INDEXER_SECRET_KEY)
 
 # Available Soroban networks exposed via GET /api/ingest/networks/.
 # Override individual RPC URLs via the corresponding env vars if needed.

sdk/python/soroscan/__init__.py

@@ -27,6 +27,8 @@
     ContractStats,
     EventEntry,
     PaginatedResponse,
+    AddIndexerRequest,
+    AddIndexerResponse,
     IsIndexerResponse,
     GetAdminResponse,
     RecordEventsBatchRequest,
@@ -54,6 +56,8 @@
     "IsIndexerResponse",
     "GetAdminResponse",
     "EventEntry",
+    "AddIndexerRequest",
+    "AddIndexerResponse",
     "RecordEventsBatchRequest",
     "RecordEventsBatchResponse",
     "SoroScanError",

sdk/python/soroscan/cli.py

@@ -141,6 +141,16 @@ def _handle_contracts(args: argparse.Namespace) -> int:
     return 0
 
 
+def _handle_indexers(args: argparse.Namespace) -> int:
+    with _build_client(args) as client:
+        result = client.add_indexer(args.indexer_address)
+    if args.output == "json":
+        _print_json(result)
+    else:
+        _print_table(
+            [result],
+            ["status", "tx_hash", "transaction_status", "error"],
+        )
 def _handle_record_event(args: argparse.Namespace) -> int:
     """Submit a single event to the SoroScan contract (SC-10)."""
     with _build_client(args) as client:
@@ -261,6 +271,13 @@ def build_parser() -> argparse.ArgumentParser:
     record.add_argument("--output", choices=["table", "json"], default="table")
     record.set_defaults(func=_handle_record_event)
 
+    indexers = subcommands.add_parser("indexers", help="Manage Soroban contract indexers (SC-9)")
+    indexer_subcommands = indexers.add_subparsers(dest="indexer_command", required=True)
+    indexers_add = indexer_subcommands.add_parser("add", help="Authorize an indexer address")
+    indexers_add.add_argument("indexer_address", help="Stellar address of the indexer")
+    indexers_add.add_argument("--output", choices=["table", "json"], default="table")
+    indexers_add.set_defaults(func=_handle_indexers)
+
     return parser