fix: gracefully handle sorocarbon memos longer than 64 bytes

Author: aolieman

sc_audit/loader/sink_events.py

@@ -11,6 +11,7 @@
 from sc_audit.config import settings
 from sc_audit.db_schema.impact_project import get_vcs_project
 from sc_audit.db_schema.sink import SinkingTx
+from sc_audit.loader.utils import truncate_sorocarbon_memo
 from sc_audit.session_manager import Session
 from sc_audit.sources.sink_events import get_sink_events
 
@@ -45,6 +46,9 @@ def load_sink_events(cursor: int=settings.FIRST_SINK_CURSOR) -> int:
             # with the TOID (paging_token). Not done yet, because it requires a data migration.
 
             memo = sink_event.memo_text or None
+            if memo:
+                memo = truncate_sorocarbon_memo(memo)
+                
             session.add(
                 SinkingTx(
                     hash=sink_event.transaction,

sc_audit/loader/sink_invocations.py

@@ -7,6 +7,7 @@
 from sc_audit.config import settings
 from sc_audit.db_schema.sink import SinkingTx
 from sc_audit.loader.sink_events import try_project_id
+from sc_audit.loader.utils import truncate_sorocarbon_memo
 from sc_audit.session_manager import Session
 from sc_audit.sources.sink_invocations import InvocationSource
 
@@ -24,6 +25,9 @@ def load_sink_invocations(cursor: int=settings.FIRST_SINK_CURSOR) -> int:
         for sink_invoke in InvocationSource.get_sink_invocations(cursor):
             vcs_project_id = try_project_id(sink_invoke.project_id)
             memo = sink_invoke.memo_text or None
+            if memo:
+                memo = truncate_sorocarbon_memo(memo)
+                
             session.add(
                 SinkingTx(
                     hash=sink_invoke.tx_hash,

sc_audit/loader/utils.py

@@ -13,6 +13,9 @@ def parse_iso_datetime(iso_datetime: str) -> datetime.datetime:
 def decode_hash_memo(b64_hash_memo: str) -> str:
     return base64.b64decode(b64_hash_memo).hex()
 
+def truncate_sorocarbon_memo(text_memo: str) -> str:
+    return text_memo.encode()[:64].decode()
+
 
 class VcsSerialNumber(BaseModel):
     batch_number: int

tests/data_fixtures/events.py

@@ -105,7 +105,7 @@
     "recipient": "GAN4FQZL5P7B7OGW4KMYA6B2V34W7E6C7NL7K4P6ZNNWQ6UZD6I7GABC",
     "amount": 109999,
     "project_id": "VCS1360",
-    "memo_text": "testing",
+    "memo_text": "testing a memo that is longer than 64 bytes 🏠🏠 truncated | this will be sliced off",
     "ledger": 50189018,
     "timestamp": 1717875054,
     "contract_id": "CBW45IZ3W5BBDIKTIXQEAOR3TAHPCFIAVQMD4NO2YPX2FA4LKGLJLWYL",

tests/data_fixtures/invocations.py

@@ -59,7 +59,7 @@ def get_items() -> list[SinkInvocation]:
         recipient='GC53JCXZHW3SVNRE4CT6XFP46WX4ACFQU32P4PR3CU43OB7AKKMFXZ6Y',
         amount=4444444,
         project_id='VCS1360',
-        memo_text='tri',
+        memo_text='tri testing a memo that is longer than 64 bytes 🏠 truncated | this will be sliced off',
         processed_at=dt.datetime(2025, 8, 24, 12, 30, 34, 920249, tzinfo=gmt_zoneinfo),
         schema_name='carbon_sink_v1',
         successful=True,

tests/test_event.py

@@ -119,7 +119,7 @@ def test_load_sink_events(self, mock_session):
                 'early support',
                 'offset tokens',
                 'gift',
-                'testing'
+                'testing a memo that is longer than 64 bytes 🏠🏠 truncated |'
             ]
 
     def test_load_sink_txs_and_events(self, mock_http, mock_session):

tests/test_invocation.py

@@ -86,7 +86,7 @@ def test_invoke_list_full(self):
         assert tri.tx_hash == "e7d68ab7c57e38e2e103b81926db97731eb9ca861de5282eaf1bcc97e1605deb"
         assert tri.amount == 4444444
         assert tri.ton_amount == Decimal("0.444")
-        assert tri.memo_text == "tri"
+        assert tri.memo_text == "tri testing a memo that is longer than 64 bytes 🏠 truncated | this will be sliced off"
 
     def test_returns_the_same_when_repeated(self):
         res_1 = InvocationSource.get_sink_invocations(cursor=0)
@@ -138,7 +138,9 @@ def test_load_sink_invocations(self, mock_session):
             )
 
             memos = [tx.memo_value for tx in loaded_invocations]
-            assert memos == ["one", "two", "tri"]
+            assert memos == [
+                "one", "two", "tri testing a memo that is longer than 64 bytes 🏠 truncated |"
+            ]
 
     def test_load_sink_txs_and_invocations(self, mock_http, mock_session):
         sink_loader.load_sinking_txs(cursor=999)