feat(Flow): Provide an alternative for Mercury events using Obsrvr Flow extracted invocations (#28)

* feat(Flow): alternative contract invocation source

* refactor: remove emails from load_sink_events

* feat(Flow): hook up invocations loader

* fix(Mercury): quantize sink events to kg resolution

* test(Flow): test invocation source and loader

* test(Flow): fix data fixture persistent/detached state

* fix(Flow): postpone OBSRVR_FLOW_DB_URI evaluation

Author: aolieman

poetry.lock

@@ -21,6 +21,7 @@ alembic = "^1.16.4"
 click = "^8.2.1"
 pygithub = "^2.7.0"
 pydantic-settings = "^2.10.1"
+psycopg = {extras = ["binary"], version = "^3.2.9"}
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^8.4.1"

pyproject.toml

@@ -21,9 +21,11 @@
 from sc_audit.loader.retirement_from_block import load_retirement_from_block
 from sc_audit.loader.retirements import load_retirements
 from sc_audit.loader.sink_events import load_sink_events
+from sc_audit.loader.sink_invocations import load_sink_invocations
 from sc_audit.loader.sink_status import load_sink_statuses
 from sc_audit.loader.sinking_txs import load_sinking_txs
 from sc_audit.sources.sink_events import MercuryError
+from sc_audit.sources.sink_invocations import ObsrvrError
 from sc_audit.views.inventory import view_inventory
 from sc_audit.views.retirement import view_retirements
 from sc_audit.views.sink_status import view_sinking_txs
@@ -181,12 +183,21 @@ def db_load_sinking_txs():
     sink_cursor: int = get_latest_attr('sink_tx') # type: ignore[return-value]
     num_sinking_txs = load_sinking_txs(cursor=sink_cursor)
     click.echo(f"Loaded {num_sinking_txs} sinking transactions")
+
+    # TODO: separate classic and SinkContract txs
+    # attempt to load SinkContract txs from Obsrvr or Mercury
     try:
-        num_sink_events, _ = load_sink_events(cursor=sink_cursor)
-        print(f"Loaded {num_sink_events} sink events")
-    except MercuryError as exc:
-        click.echo(f"Couldn't load sink events from Mercury")
+        num_sink_invocations = load_sink_invocations(cursor=sink_cursor)
+        print(f"Loaded {num_sink_invocations} sink invocations")
+    except ObsrvrError as exc:
+        click.echo(f"Couldn't load sink invocations from Obsrvr")
         click.echo(repr(exc), err=True)
+        try:
+            num_sink_events = load_sink_events(cursor=sink_cursor)
+            print(f"Loaded {num_sink_events} sink events")
+        except MercuryError as exc:
+            click.echo(f"Couldn't load sink events from Mercury")
+            click.echo(repr(exc), err=True)
 
 
 @load.command(name="retirements")

sc_audit/cli.py

@@ -1,7 +1,7 @@
 from pathlib import Path
 from typing import Literal
 
-from pydantic import HttpUrl, computed_field
+from pydantic import HttpUrl, PostgresDsn, computed_field
 from pydantic_settings import BaseSettings, SettingsConfigDict
 from stellar_sdk import Asset
 
@@ -54,6 +54,9 @@ class Settings(BaseSettings):
     RETROSHADES_MD5: str = "3dd6e7b71adb9fda05a5b45a154611f3"
     MERCURY_KEY: str = ""
 
+    OBSRVR_FLOW_DB_URI: PostgresDsn | None = None
+    OBSRVR_FLOW_TABLE: str = "extracted_contract_invocations"
+
     @computed_field
     @property
     def CARBON_ASSET(self) -> Asset:

sc_audit/config.py

@@ -0,0 +1,5 @@
+from decimal import Decimal
+
+
+KG = Decimal("0.001")
+UNIT_IN_STROOPS = Decimal(10_000_000)

sc_audit/constants.py

@@ -23,9 +23,11 @@
 from sc_audit.loader.retirement_from_block import load_retirement_from_block
 from sc_audit.loader.retirements import load_retirements
 from sc_audit.loader.sink_events import load_sink_events
+from sc_audit.loader.sink_invocations import load_sink_invocations
 from sc_audit.loader.sink_status import load_sink_statuses
 from sc_audit.loader.sinking_txs import load_sinking_txs
 from sc_audit.sources.sink_events import MercuryError
+from sc_audit.sources.sink_invocations import ObsrvrError
 
 
 def catch_up_from_sources():
@@ -51,12 +53,20 @@ def catch_up_from_sources():
     print(f"Loaded {num_distribution_txs} distribution outflows")
     num_sinking_txs = load_sinking_txs(cursor=sink_cursor)
     print(f"Loaded {num_sinking_txs} sinking transactions")
+    
+    # attempt to load SinkContract txs from Obsrvr or Mercury
     try:
-        num_sink_events, _ = load_sink_events(cursor=sink_cursor)
-        print(f"Loaded {num_sink_events} sink events")
-    except MercuryError as exc:
-        print(f"Couldn't load sink events from Mercury")
+        num_sink_invocations = load_sink_invocations(cursor=sink_cursor)
+        print(f"Loaded {num_sink_invocations} sink invocations")
+    except ObsrvrError as exc:
+        print(f"Couldn't load sink invocations from Obsrvr")
         print(repr(exc), file=sys.stderr)
+        try:
+            num_sink_events = load_sink_events(cursor=sink_cursor)
+            print(f"Loaded {num_sink_events} sink events")
+        except MercuryError as exc:
+            print(f"Couldn't load sink events from Mercury")
+            print(repr(exc), file=sys.stderr)
 
     num_minting_txs = load_minted_blocks(cursor=mint_cursor)
     print(f"Loaded {num_minting_txs} minted blocks")

sc_audit/loader/__main__.py

@@ -15,7 +15,7 @@
 from sc_audit.sources.sink_events import get_sink_events
 
 
-def load_sink_events(cursor: int=settings.FIRST_SINK_CURSOR) -> tuple[int, list[tuple[str, str]]]:
+def load_sink_events(cursor: int=settings.FIRST_SINK_CURSOR) -> int:
     """
     Load (all) sink events from Mercury Retroshades into the DB.
 
@@ -29,17 +29,13 @@ def load_sink_events(cursor: int=settings.FIRST_SINK_CURSOR) -> tuple[int, list[
     separately.
     """
     number_loaded = 0
-    recipient_emails: dict[str, str] = {}
 
     with Session.begin() as session:
         last_ledger = 0
         # For each new ledger, reset tx_order to 2^16 to avoid collisions with SinkingTxs.
         # This is a pragmatic choice; ideally, the order should come from Retroshades.
         tx_order = first_tx_index = 2**16
         for sink_event in get_sink_events(cursor):
-            if sink_event.email:
-                recipient_emails[sink_event.recipient] = sink_event.email
-
             vcs_project_id = try_project_id(sink_event.project_id)
 
             # FIXME: the TOID should come from Retroshades if possible
@@ -78,7 +74,7 @@ def load_sink_events(cursor: int=settings.FIRST_SINK_CURSOR) -> tuple[int, list[
 
         number_loaded = len(session.new)
 
-    return number_loaded, list(recipient_emails.items())
+    return number_loaded
 
 
 @lru_cache(maxsize=32)