Endorser protocol updates - refactor to use event bus

Endorser.md

@@ -1,8 +1,8 @@
 # Transaction Endorser Support
 
-Note that the ACA-Py transaciton support is in the process of code refactor and cleanup.  The following documents the current state, but is subject to change.
+Note that the ACA-Py transaction support is in the process of code refactor and cleanup.  The following documents the current state, but is subject to change.
 
-ACA-Py supports an [Endorser Protocol](https://github.com/hyperledger/aries-rfcs/pull/586), that allows an un-privieged agent (an "Author") to request another agent (the "Endorser") to sign their transactions so they can write these transactions to the ledger.
+ACA-Py supports an [Endorser Protocol](https://github.com/hyperledger/aries-rfcs/pull/586), that allows an un-privileged agent (an "Author") to request another agent (the "Endorser") to sign their transactions so they can write these transactions to the ledger.  This is required on Indy ledgers, where new agents will typically be granted only "Author" privileges.
 
 Transaction Endorsement is built into the protocols for Schema, Credential Definition and Revocation, and endorsements can be explicitely requested, or ACA-Py can be configured to automate the endorsement workflow.
 

aries_cloudagent/core/plugin_registry.py

@@ -6,6 +6,7 @@
 from typing import Sequence
 
 from ..config.injection_context import InjectionContext
+from ..core.event_bus import EventBus
 from ..utils.classloader import ClassLoader, ModuleLoadError
 
 from .error import ProtocolDefinitionValidationError
@@ -201,6 +202,9 @@ async def init_context(self, context: InjectionContext):
             else:
                 await self.load_protocols(context, plugin)
 
+        # register event handlers for each protocol, if providedf
+        self.register_protocol_events(context)
+
     async def load_protocol_version(
         self,
         context: InjectionContext,
@@ -280,6 +284,36 @@ async def register_admin_routes(self, app):
                 if mod and hasattr(mod, "register"):
                     await mod.register(app)
 
+    def register_protocol_events(self, context: InjectionContext):
+        """Call route register_events methods on the current context."""
+        event_bus = context.inject_or(EventBus)
+        if not event_bus:
+            LOGGER.error("No event bus in context")
+            return
+        for plugin in self._plugins.values():
+            definition = ClassLoader.load_module("definition", plugin.__name__)
+            if definition:
+                # Load plugin routes that are in a versioned package.
+                for plugin_version in definition.versions:
+                    try:
+                        mod = ClassLoader.load_module(
+                            f"{plugin.__name__}.{plugin_version['path']}.routes"
+                        )
+                    except ModuleLoadError as e:
+                        LOGGER.error("Error loading admin routes: %s", e)
+                        continue
+                    if mod and hasattr(mod, "register_events"):
+                        mod.register_events(event_bus)
+            else:
+                # Load plugin routes that aren't in a versioned package.
+                try:
+                    mod = ClassLoader.load_module(f"{plugin.__name__}.routes")
+                except ModuleLoadError as e:
+                    LOGGER.error("Error loading admin routes: %s", e)
+                    continue
+                if mod and hasattr(mod, "register_events"):
+                    mod.register_events(event_bus)
+
     def post_process_routes(self, app):
         """Call route binary file response OpenAPI fixups if applicable."""
         for plugin in self._plugins.values():

aries_cloudagent/ledger/base.py

@@ -5,13 +5,9 @@
 from abc import ABC, abstractmethod, ABCMeta
 from enum import Enum
 from hashlib import sha256
-from time import time
 from typing import Sequence, Tuple, Union
 
 from ..indy.issuer import IndyIssuer
-from ..storage.base import StorageRecord
-from ..messaging.credential_definitions.util import CRED_DEF_SENT_RECORD_TYPE
-from ..messaging.schemas.util import SCHEMA_SENT_RECORD_TYPE
 from ..utils import sentinel
 from ..wallet.did_info import DIDInfo
 
@@ -276,58 +272,6 @@ async def get_revoc_reg_entry(
     ) -> Tuple[dict, int]:
         """Get revocation registry entry by revocation registry ID and timestamp."""
 
-    async def add_schema_non_secrets_record(self, schema_id: str, issuer_did: str):
-        """
-        Write the wallet non-secrets record for a schema (already written to the ledger).
-
-        Args:
-            schema_id: The schema id (or stringified sequence number)
-            issuer_did: The DID of the issuer
-
-        """
-        schema_id_parts = schema_id.split(":")
-        schema_tags = {
-            "schema_id": schema_id,
-            "schema_issuer_did": issuer_did,
-            "schema_name": schema_id_parts[-2],
-            "schema_version": schema_id_parts[-1],
-            "epoch": str(int(time())),
-        }
-        record = StorageRecord(SCHEMA_SENT_RECORD_TYPE, schema_id, schema_tags)
-        storage = self.get_indy_storage()
-        await storage.add_record(record)
-
-    async def add_cred_def_non_secrets_record(
-        self, schema_id: str, issuer_did: str, credential_definition_id: str
-    ):
-        """
-        Write the wallet non-secrets record for cred def (already written to the ledger).
-
-        Note that the cred def private key signing informtion must already exist in the
-        wallet.
-
-        Args:
-            schema_id: The schema id (or stringified sequence number)
-            issuer_did: The DID of the issuer
-            credential_definition_id: The credential definition id
-
-        """
-        schema_id_parts = schema_id.split(":")
-        cred_def_tags = {
-            "schema_id": schema_id,
-            "schema_issuer_did": schema_id_parts[0],
-            "schema_name": schema_id_parts[-2],
-            "schema_version": schema_id_parts[-1],
-            "issuer_did": issuer_did,
-            "cred_def_id": credential_definition_id,
-            "epoch": str(int(time())),
-        }
-        record = StorageRecord(
-            CRED_DEF_SENT_RECORD_TYPE, credential_definition_id, cred_def_tags
-        )
-        storage = self.get_indy_storage()
-        await storage.add_record(record)
-
 
 class Role(Enum):
     """Enum for indy roles."""

aries_cloudagent/ledger/indy.py

@@ -500,9 +500,6 @@ async def create_and_send_schema(
                 else:
                     raise
 
-            # Add non-secrets record
-            await self.add_schema_non_secrets_record(schema_id, public_info.did)
-
         return schema_id, schema_def
 
     async def check_existing_schema(
@@ -730,11 +727,6 @@ async def create_and_send_credential_definition(
             if not write_ledger:
                 return (credential_definition_id, {"signed_txn": resp}, novel)
 
-            # Add non-secrets record
-            await self.add_cred_def_non_secrets_record(
-                schema_id, public_info.did, credential_definition_id
-            )
-
         return (credential_definition_id, json.loads(credential_definition_json), novel)
 
     async def get_credential_definition(self, credential_definition_id: str) -> dict: