feat: Ingest SinkEvents as SinkingTxs (#21)

* feat: fetch sink events from retroshades

* test: mock sink event source

* test: patch mercury key

* build: update dependencies

* feat: add contract_id field to SinkingTx

* feat: sink event loader (scaffolding)

* fix: take tx_order from the 2^16 to 2^20 range

* feat: dedicated impact project loader

* fix: apply foreign key constraint in sqlite

* test: interleaved sink events and classic txs

* feat(cli): add sink events loader to CLI

* perf: move test mappers to migrations.env

Author: aolieman

poetry.lock

@@ -27,7 +27,7 @@ def dump_table(db_model: type[ScBase] | str, output_path: Path | None = None) ->
     ndjson_table = table_to_ndjson(db_table)
     if output_path is not None:
         with output_path.open('w') as ofile:
-            ofile.write(ndjson_table)
+            ofile.write(ndjson_table + "\n")
     else:
         return ndjson_table
 

sc_audit/backup/dump.py

@@ -70,7 +70,7 @@ def df_from_ndjson(json_path: Path) -> pd.DataFrame:
     table_name = json_path.stem
     db_table = ScBase.metadata.tables[table_name]
     with open(json_path, 'r') as infile:
-        ndjson = infile.read()
+        ndjson = infile.read().rstrip("\n")
 
     # turn json lines into a valid array
     json_array = "[" + ndjson.replace("\n", ",") + "]"

sc_audit/backup/restore.py

@@ -16,11 +16,14 @@
 from sc_audit.loader.__main__ import catch_up_from_sources
 from sc_audit.loader.distribution_outflows import load_distribution_txs
 from sc_audit.loader.get_latest import get_latest_attr
+from sc_audit.loader.impact_projects import load_impact_projects
 from sc_audit.loader.minted_blocks import load_minted_blocks
 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_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.views.inventory import view_inventory
 from sc_audit.views.retirement import view_retirements
 from sc_audit.views.sink_status import view_sinking_txs
@@ -149,6 +152,13 @@ def load():
     pass
 
 
+@load.command(name="impact-projects")
+def db_load_impact_projects():
+    """Load impact projects into the DB"""
+    num_impact_projects = load_impact_projects()
+    click.echo(f"Loaded {num_impact_projects} impact projects")
+
+
 @load.command(name="minted-blocks")
 def db_load_minted_blocks():
     """Load minted blocks into the DB"""
@@ -171,6 +181,12 @@ def db_load_sinking_txs():
     sink_cursor = get_latest_attr('sink_tx')
     num_sinking_txs = load_sinking_txs(cursor=sink_cursor) # type: ignore[arg-type]
     click.echo(f"Loaded {num_sinking_txs} sinking transactions")
+    try:
+        num_sink_events = load_sink_events(cursor=sink_cursor)  # type: ignore[arg-type]
+        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

@@ -33,7 +33,7 @@ def get_default_db_path(db_name="sc-audit.sqlite3") -> Path:
 
 
 class Settings(BaseSettings):
-    model_config = SettingsConfigDict(env_prefix='SC_')
+    model_config = SettingsConfigDict(env_prefix='SC_', env_file=".env")
 
     DBAPI_URL: str = f"sqlite+pysqlite:///{get_default_db_path()}"
     TABLE_PREFIX: DbTablePrefix = None
@@ -50,6 +50,10 @@ class Settings(BaseSettings):
     FIRST_MINT_CURSOR: int = 164806777139924992
     FIRST_DIST_CURSOR: int = 164810659791396865
 
+    RETROSHADES_URL: HttpUrl = HttpUrl("https://api.mercurydata.app/retroshadesv1")
+    RETROSHADES_MD5: str = "571c515522665c5da4d18f7ccbf8eb3a"
+    MERCURY_KEY: str = ""
+
     @computed_field
     @property
     def CARBON_ASSET(self) -> Asset:

sc_audit/config.py

@@ -26,6 +26,7 @@ class SinkingTxBase(MappedAsDataclass, kw_only=True):
 
     hash: Mapped[hashpk]
     created_at: Mapped[dt.datetime]
+    contract_id: Mapped[strkey | None]
     funder: Mapped[strkey]
     recipient: Mapped[strkey]
     carbon_amount: Mapped[kgdecimal]

sc_audit/db_schema/sink.py

@@ -1,25 +1,31 @@
 """
 Load all database records in the order:
-  1. Retirements
-  2. Sinking Transactions
-  3. Minted Blocks
-  4. Retirement from Block
-  5. Sink Statuses
+    1. Impact Projects
+    2. Distribution Outflows
+    3. Sinking Transactions
+    4. Minted Blocks
+    5. Retirements
+    6. Retirement from Block
+    7. Sink Statuses
 
 Loading from scratch may fail if a Horizon instance with pruned history is selected.
 You may select another instance with the SC_HORIZON_URL env variable.
 
 Author: Alex Olieman <https://keybase.io/alioli>
 """
 import datetime as dt
+import sys
 
 from sc_audit.loader.distribution_outflows import load_distribution_txs
 from sc_audit.loader.get_latest import get_latest_attr
+from sc_audit.loader.impact_projects import load_impact_projects
 from sc_audit.loader.minted_blocks import load_minted_blocks
 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_status import load_sink_statuses
 from sc_audit.loader.sinking_txs import load_sinking_txs
+from sc_audit.sources.sink_events import MercuryError
 
 
 def catch_up_from_sources():
@@ -39,10 +45,19 @@ def catch_up_from_sources():
     ) # type: ignore[return-value]
 
     print("Started catch-up from data sources...")
+    num_impact_projects = load_impact_projects()
+    print(f"Loaded {num_impact_projects} impact projects")
     num_distribution_txs = load_distribution_txs(cursor=dist_cursor)
     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")
+    try:
+        num_sink_events = load_sink_events(cursor=sink_cursor)  # type: ignore[arg-type]
+        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")
     num_retirements = load_retirements(from_date=retirement_date)

sc_audit/loader/__main__.py

@@ -5,6 +5,7 @@
 """
 
 from decimal import Decimal
+
 from sc_audit.config import settings
 from sc_audit.db_schema.distribution import DistributionTx
 from sc_audit.loader.utils import parse_iso_datetime

sc_audit/loader/distribution_outflows.py

@@ -0,0 +1,27 @@
+"""
+Load impact projects into the DB.
+
+Author: Alex Olieman <https://keybase.io/alioli>
+"""
+
+from sqlalchemy import select
+from sc_audit.db_schema.base import intpk
+from sc_audit.db_schema.impact_project import VcsProject, get_vcs_project
+from sc_audit.session_manager import Session
+
+
+def load_impact_projects() -> int:
+    """
+    Load impact projects from our own data schema into the DB.
+    """
+    number_loaded = 0
+
+    with Session.begin() as session:
+        existing_vcs_projects: set[intpk] = set(session.scalars(select(VcsProject.id)).all())
+        if 1360 not in existing_vcs_projects:
+            vcs_1360 = get_vcs_project(1360)
+            session.add(vcs_1360)
+        
+        number_loaded = len(session.new)
+
+    return number_loaded

sc_audit/loader/impact_projects.py

@@ -16,7 +16,7 @@
 from sqlalchemy import select
 
 from sc_audit.config import settings
-from sc_audit.db_schema.impact_project import UnknownVcsProject, VcsProject, get_vcs_project
+from sc_audit.db_schema.impact_project import UnknownVcsProject, get_vcs_project
 from sc_audit.db_schema.mint import MintedBlock, verra_carbon_pool
 from sc_audit.db_schema.retirement import Retirement
 from sc_audit.loader.utils import VcsSerialNumber, decode_hash_memo, parse_iso_datetime
@@ -36,21 +36,16 @@ def load_minted_blocks(cursor: int=settings.FIRST_MINT_CURSOR) -> int:
     pristine_blocks: list[MintedBlock] = []
     carbon_pool = index_carbon_pool()
     with Session.begin() as session:
-        existing_vcs_projects: set[int] = set(session.scalars(select(VcsProject.id)).all())
         for mint_tx in get_minting_transactions(cursor):
-            # ensure that the related VCS Project exists
+            # ensure that the related VCS Project is available
             vcs_project_id = get_vcs_project_id(mint_tx)
-            if vcs_project_id not in existing_vcs_projects:
-                vcs_project = get_vcs_project(vcs_project_id)
-                if vcs_project:
-                    existing_vcs_projects.add(vcs_project_id)
-                    session.add(vcs_project)
-                else:
-                    raise UnknownVcsProject(
-                        f"VCS project {vcs_project_id} needs to be loaded before related transactions"
-                        " can be stored.",
-                        vcs_id=vcs_project_id
-                    )
+            vcs_project = get_vcs_project(vcs_project_id)
+            if not vcs_project:
+                raise UnknownVcsProject(
+                    f"VCS project {vcs_project_id} needs to be loaded before related transactions"
+                    " can be stored.",
+                    vcs_id=vcs_project_id
+                )
             # check whether the block is pristine
             serial_hash = decode_hash_memo(mint_tx['transaction']['memo'])
             if serial_hash in carbon_pool:
@@ -91,7 +86,7 @@ def load_minted_blocks(cursor: int=settings.FIRST_MINT_CURSOR) -> int:
     return len(retired_blocks) + len(pristine_blocks)
 
 
-def get_vcs_project_id(sinking_tx) -> Literal[1360]:
+def get_vcs_project_id(mint_tx) -> Literal[1360]:
     # TODO: support multiple VCS projects
     return 1360