Set value db fields in cic_cache to handle biiig numbers

Get that "ussd_menu.json" out there.

See merge request grassrootseconomics/cic-internal-integration!50

apps/cic-cache/cic_cache/db/migrations/default/versions/63b629f14a85_.py

@@ -29,8 +29,8 @@ def upgrade():
             sa.Column('source_token', sa.String(42), nullable=False),
             sa.Column('destination_token', sa.String(42), nullable=False),
             sa.Column('success', sa.Boolean, nullable=False),
-            sa.Column('from_value', sa.BIGINT(), nullable=False),
-            sa.Column('to_value', sa.BIGINT(), nullable=False),
+            sa.Column('from_value', sa.NUMERIC(), nullable=False),
+            sa.Column('to_value', sa.NUMERIC(), nullable=False),
             sa.Column('date_block', sa.DateTime, nullable=False),
             )
     op.create_table(

apps/cic-cache/cic_cache/runnable/tracker.py

@@ -312,7 +312,7 @@ class Tracker:
         session.close()
 
 (provider, w3) = web3_constructor()
-trust = config.get('CIC_TRUST_ADDRESS', []).split(",")
+trust = config.get('CIC_TRUST_ADDRESS', "").split(",")
 chain_spec = args.i
 
 try:

apps/cic-cache/config/database.ini

@@ -1,5 +1,5 @@
 [database]
-NAME=cic-eth
+NAME=cic_cache
 USER=postgres
 PASSWORD=
 HOST=localhost

apps/cic-eth/cic_eth/admin/ctrl.py

@@ -8,6 +8,7 @@ from cic_registry import zero_address
 
 # local imports
 from cic_eth.db.enum import LockEnum
+from cic_eth.db.models.base import SessionBase
 from cic_eth.db.models.lock import Lock
 from cic_eth.error import LockedError
 
@@ -116,9 +117,10 @@ def unlock_queue(chained_input, chain_str, address=zero_address):
 
 @celery_app.task()
 def check_lock(chained_input, chain_str, lock_flags, address=None):
-    r = Lock.check(chain_str, lock_flags, address=zero_address)
+    session = SessionBase.create_session()
+    r = Lock.check(chain_str, lock_flags, address=zero_address, session=session)
     if address != None:
-        r |= Lock.check(chain_str, lock_flags, address=address)
+        r |= Lock.check(chain_str, lock_flags, address=address, session=session)
     if r > 0:
         logg.debug('lock check {} has match {} for {}'.format(lock_flags, r, address))
         raise LockedError(r)

apps/cic-eth/cic_eth/admin/debug.py

@@ -1,12 +1,25 @@
+# standard imports
 import datetime
 
+# external imports
 import celery
 
+# local imports
+from cic_eth.db.models.debug import Debug
+from cic_eth.db.models.base import SessionBase
+from cic_eth.task import CriticalSQLAlchemyTask
+
 celery_app = celery.current_app
 
 
-@celery_app.task()
-def out_tmp(tag, txt):
-    f = open('/tmp/err.{}.txt'.format(tag), "w")
-    f.write(txt)
-    f.close()
+@celery_app.task(base=CriticalSQLAlchemyTask)
+def alert(chained_input, tag, txt):
+    session = SessionBase.create_session()
+
+    o = Debug(tag, txt)
+    session.add(o)
+    session.commit()
+
+    session.close()
+    
+    return chained_input

apps/cic-eth/cic_eth/api/api_admin.py

@@ -457,6 +457,7 @@ class AdminApi:
         tx_unpacked = unpack_signed_raw_tx(bytes.fromhex(tx['signed_tx'][2:]), chain_spec.chain_id())
         tx['gas_price'] = tx_unpacked['gasPrice']
         tx['gas_limit'] = tx_unpacked['gas']
+        tx['data'] = tx_unpacked['data']
 
         s = celery.signature(
             'cic_eth.queue.tx.get_state_log',

apps/cic-eth/cic_eth/api/api_task.py

@@ -489,8 +489,9 @@ class Api:
                 ],
             queue=self.queue,
             )
+        s_local.link(s_brief)
         if self.callback_param != None:
-            s_assemble.link(self.callback_success).on_error(self.callback_error)
+            s_brief.link(self.callback_success).on_error(self.callback_error)
 
         t = None
         if external_task != None:
@@ -515,11 +516,10 @@ class Api:
             c = celery.chain(s_external_get, s_external_process)
             t = celery.chord([s_local, c])(s_brief)
         else:
-            t = s_local.apply_sync()
+            t = s_local.apply_async(queue=self.queue)
 
         return t
 
-
     def ping(self, r):
         """A noop callback ping for testing purposes.
 

apps/cic-eth/cic_eth/callbacks/tcp.py

@@ -20,5 +20,10 @@ def tcp(self, result, destination, status_code):
     (host, port) = destination.split(':')
     logg.debug('tcp callback to {} {}'.format(host, port))
     s.connect((host, int(port)))
-    s.send(json.dumps(result).encode('utf-8'))
+    data = {
+            'root_id': self.request.root_id,
+            'status': status_code,
+            'result': result,
+            }
+    s.send(json.dumps(data).encode('utf-8'))
     s.close()

apps/cic-eth/cic_eth/db/migrations/default/versions/ec40ac0974c1_add_chain_syncer.py

@@ -0,0 +1,28 @@
+"""Add chain syncer
+
+Revision ID: ec40ac0974c1
+Revises: 6ac7a1dadc46
+Create Date: 2021-02-23 06:10:19.246304
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from chainsyncer.db.migrations.sqlalchemy import (
+        chainsyncer_upgrade,
+        chainsyncer_downgrade,
+        )
+
+
+# revision identifiers, used by Alembic.
+revision = 'ec40ac0974c1'
+down_revision = '6ac7a1dadc46'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    chainsyncer_upgrade(0, 0, 1)
+
+
+def downgrade():
+    chainsyncer_downgrade(0, 0, 1)

apps/cic-eth/cic_eth/db/migrations/default/versions/f738d9962fdf_debug_output.py

@@ -0,0 +1,32 @@
+"""debug output
+
+Revision ID: f738d9962fdf
+Revises: ec40ac0974c1
+Create Date: 2021-03-04 08:32:43.281214
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'f738d9962fdf'
+down_revision = 'ec40ac0974c1'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+            'debug',
+            sa.Column('id', sa.Integer, primary_key=True),
+            sa.Column('tag', sa.String, nullable=False),
+            sa.Column('description', sa.String, nullable=False),
+            sa.Column('date_created', sa.DateTime, nullable=False),
+            )
+    pass
+
+
+def downgrade():
+    op.drop_table('debug')
+    pass

apps/cic-eth/cic_eth/db/migrations/postgresql/versions/ec40ac0974c1_add_chain_syncer.py

@@ -0,0 +1,28 @@
+"""Add chain syncer
+
+Revision ID: ec40ac0974c1
+Revises: 6ac7a1dadc46
+Create Date: 2021-02-23 06:10:19.246304
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from chainsyncer.db.migrations.sqlalchemy import (
+        chainsyncer_upgrade,
+        chainsyncer_downgrade,
+        )
+
+
+# revision identifiers, used by Alembic.
+revision = 'ec40ac0974c1'
+down_revision = '6ac7a1dadc46'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    chainsyncer_upgrade(0, 0, 1)
+
+
+def downgrade():
+    chainsyncer_downgrade(0, 0, 1)

apps/cic-eth/cic_eth/db/migrations/postgresql/versions/f738d9962fdf_debug_output.py

@@ -0,0 +1,32 @@
+"""debug output
+
+Revision ID: f738d9962fdf
+Revises: ec40ac0974c1
+Create Date: 2021-03-04 08:32:43.281214
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'f738d9962fdf'
+down_revision = 'ec40ac0974c1'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+            'debug',
+            sa.Column('id', sa.Integer, primary_key=True),
+            sa.Column('tag', sa.String, nullable=False),
+            sa.Column('description', sa.String, nullable=False),
+            sa.Column('date_created', sa.DateTime, nullable=False),
+            )
+    pass
+
+
+def downgrade():
+    op.drop_table('debug')
+    pass

apps/cic-eth/cic_eth/db/models/base.py

@@ -114,7 +114,6 @@ class SessionBase(Model):
 
     @staticmethod
     def release_session(session=None):
-        session.flush()
         session_key = str(id(session))
         if SessionBase.localsessions.get(session_key) != None:
             logg.debug('destroying session {}'.format(session_key))

apps/cic-eth/cic_eth/db/models/debug.py

@@ -0,0 +1,23 @@
+# standard imports
+import datetime
+import logging
+
+# external imports
+from sqlalchemy import Column, String, DateTime
+
+# local imports
+from .base import SessionBase
+
+
+class Debug(SessionBase):
+
+    __tablename__ = 'debug'
+
+    date_created = Column(DateTime, default=datetime.datetime.utcnow)
+    tag = Column(String)
+    description = Column(String)
+
+
+    def __init__(self, tag, description):
+        self.tag = tag
+        self.description = description

apps/cic-eth/cic_eth/db/models/lock.py

@@ -55,11 +55,9 @@ class Lock(SessionBase):
         :returns: New flag state of entry
         :rtype: number
         """
-        localsession = session
-        if localsession == None:
-            localsession = SessionBase.create_session()
+        session = SessionBase.bind_session(session)
 
-        q = localsession.query(Lock)
+        q = session.query(Lock)
         #q = q.join(TxCache, isouter=True)
         q = q.filter(Lock.address==address)
         q = q.filter(Lock.blockchain==chain_str)
@@ -71,7 +69,8 @@ class Lock(SessionBase):
             lock.address = address
             lock.blockchain = chain_str
             if tx_hash != None:
-                q = localsession.query(Otx)
+                session.flush()
+                q = session.query(Otx)
                 q = q.filter(Otx.tx_hash==tx_hash)
                 otx = q.first()
                 if otx != None:
@@ -80,12 +79,11 @@ class Lock(SessionBase):
         lock.flags |= flags
         r = lock.flags
 
-        localsession.add(lock)
-        localsession.commit()
+        session.add(lock)
+        session.commit()
+
+        SessionBase.release_session(session)
 
-        if session == None:
-            localsession.close()
-        
         return r
 
 
@@ -110,11 +108,9 @@ class Lock(SessionBase):
         :returns: New flag state of entry
         :rtype: number
         """
-        localsession = session
-        if localsession == None:
-            localsession = SessionBase.create_session()
+        session = SessionBase.bind_session(session)
 
-        q = localsession.query(Lock)
+        q = session.query(Lock)
         #q = q.join(TxCache, isouter=True)
         q = q.filter(Lock.address==address)
         q = q.filter(Lock.blockchain==chain_str)
@@ -124,14 +120,13 @@ class Lock(SessionBase):
         if lock != None:
             lock.flags &= ~flags
             if lock.flags == 0:
-                localsession.delete(lock)
+                session.delete(lock)
             else:
-                localsession.add(lock)
+                session.add(lock)
                 r = lock.flags
-            localsession.commit()
+            session.commit()
 
-        if session == None:
-            localsession.close()
+        SessionBase.release_session(session)
 
         return r
 
@@ -156,22 +151,20 @@ class Lock(SessionBase):
         :rtype: number
         """
 
-        localsession = session
-        if localsession == None:
-            localsession = SessionBase.create_session()
+        session = SessionBase.bind_session(session)
 
-        q = localsession.query(Lock)
+        q = session.query(Lock)
         #q = q.join(TxCache, isouter=True)
         q = q.filter(Lock.address==address)
         q = q.filter(Lock.blockchain==chain_str)
         q = q.filter(Lock.flags.op('&')(flags)==flags)
         lock = q.first()
-        if session == None:
-            localsession.close()
        
         r = 0
         if lock != None:
             r = lock.flags & flags
+
+        SessionBase.release_session(session)
         return r
 
 

apps/cic-eth/cic_eth/db/models/nonce.py

@@ -21,12 +21,9 @@ class Nonce(SessionBase):
 
     @staticmethod
     def get(address, session=None):
-        localsession = session
-        if localsession == None:
-            localsession = SessionBase.create_session()
+        session = SessionBase.bind_session(session)
 
-
-        q = localsession.query(Nonce)
+        q = session.query(Nonce)
         q = q.filter(Nonce.address_hex==address)
         nonce = q.first()
 
@@ -34,28 +31,29 @@ class Nonce(SessionBase):
         if nonce != None:
             nonce_value = nonce.nonce;
 
-        if session == None:
-            localsession.close()
+        SessionBase.release_session(session)
 
         return nonce_value
 
 
     @staticmethod
-    def __get(conn, address):
-        r = conn.execute("SELECT nonce FROM nonce WHERE address_hex = '{}'".format(address))
+    def __get(session, address):
+        r = session.execute("SELECT nonce FROM nonce WHERE address_hex = '{}'".format(address))
         nonce = r.fetchone()
+        session.flush()
         if nonce == None:
             return None
         return nonce[0]
 
 
     @staticmethod
-    def __set(conn, address, nonce):
-        conn.execute("UPDATE nonce set nonce = {} WHERE address_hex = '{}'".format(nonce, address))
+    def __set(session, address, nonce):
+        session.execute("UPDATE nonce set nonce = {} WHERE address_hex = '{}'".format(nonce, address))
+        session.flush()
 
 
     @staticmethod
-    def next(address, initial_if_not_exists=0):
+    def next(address, initial_if_not_exists=0, session=None):
         """Generate next nonce for the given address.
 
         If there is no previous nonce record for the address, the nonce may be initialized to a specified value, or 0 if no value has been given.
@@ -67,20 +65,32 @@ class Nonce(SessionBase):
         :returns: Nonce
         :rtype: number
         """
-        conn = Nonce.engine.connect()
+        session = SessionBase.bind_session(session)
+        
+        SessionBase.release_session(session)
+
+        session.begin_nested()
+        #conn = Nonce.engine.connect()
         if Nonce.transactional:
-            conn.execute('BEGIN')
-            conn.execute('LOCK TABLE nonce IN SHARE ROW EXCLUSIVE MODE')
-        nonce = Nonce.__get(conn, address)
+            #session.execute('BEGIN')
+            session.execute('LOCK TABLE nonce IN SHARE ROW EXCLUSIVE MODE')
+            session.flush()
+        nonce = Nonce.__get(session, address)
         logg.debug('get nonce {} for address {}'.format(nonce, address))
         if nonce == None:
             nonce = initial_if_not_exists
-            conn.execute("INSERT INTO nonce (nonce, address_hex) VALUES ({}, '{}')".format(nonce, address))
+            session.execute("INSERT INTO nonce (nonce, address_hex) VALUES ({}, '{}')".format(nonce, address))
+            session.flush()
             logg.debug('setting default nonce to {} for address {}'.format(nonce, address))
-        Nonce.__set(conn, address, nonce+1)
-        if Nonce.transactional:
-            conn.execute('COMMIT')
-        conn.close()
+        Nonce.__set(session, address, nonce+1)
+        #if Nonce.transactional:
+            #session.execute('COMMIT')
+            #session.execute('UNLOCK TABLE nonce')
+        #conn.close()
+        session.commit()
+#        session.commit()
+
+        SessionBase.release_session(session)
         return nonce
 
 

apps/cic-eth/cic_eth/db/models/otx.py

@@ -2,7 +2,7 @@
 import datetime
 import logging
 
-# third-party imports
+# external imports
 from sqlalchemy import Column, Enum, String, Integer, DateTime, Text, or_, ForeignKey
 from sqlalchemy.ext.hybrid import hybrid_property, hybrid_method
 
@@ -79,6 +79,13 @@ class Otx(SessionBase):
         return r
 
 
+    def __status_not_set(self, status):
+        r = not(self.status & status)
+        if r:
+            logg.warning('status bit {} not set on {}'.format(status.name, self.tx_hash))
+        return r
+
+
     def set_block(self, block, session=None):
         """Set block number transaction was mined in.
 
@@ -320,6 +327,32 @@ class Otx(SessionBase):
         SessionBase.release_session(session)
 
 
+    def dequeue(self, session=None):
+        """Marks that a process to execute send attempt is underway
+
+        Only manipulates object, does not transaction or commit to backend.
+
+        :raises cic_eth.db.error.TxStateChangeError: State change represents a sequence of events that should not exist.
+        """
+        if self.__status_not_set(StatusBits.QUEUED):
+            return
+
+        session = SessionBase.bind_session(session)
+
+        if self.status & StatusBits.FINAL:
+            raise TxStateChangeError('SENDFAIL cannot be set on an entry with FINAL state set ({})'.format(status_str(self.status)))
+        if self.status & StatusBits.IN_NETWORK:
+            raise TxStateChangeError('SENDFAIL cannot be set on an entry with IN_NETWORK state set ({})'.format(status_str(self.status)))
+
+        self.__reset_status(StatusBits.QUEUED, session)
+
+        if self.tracing:
+            self.__state_log(session=session)
+
+        SessionBase.release_session(session)
+
+
+
     def minefail(self, block, session=None):
         """Marks that transaction was mined but code execution did not succeed.
 
@@ -373,18 +406,6 @@ class Otx(SessionBase):
         else:
             self.__set_status(StatusEnum.OBSOLETED, session)
 
-
-#        if confirmed:
-#            if self.status != StatusEnum.OBSOLETED:
-#                logg.warning('CANCELLED must follow OBSOLETED, but had {}'.format(StatusEnum(self.status).name))
-#                #raise TxStateChangeError('CANCELLED must follow OBSOLETED, but had {}'.format(StatusEnum(self.status).name))
-#            self.__set_status(StatusEnum.CANCELLED, session)
-#        elif self.status != StatusEnum.OBSOLETED:
-#            if self.status > StatusEnum.SENT:
-#                logg.warning('OBSOLETED must follow PENDING, SENDFAIL or SENT, but had {}'.format(StatusEnum(self.status).name))
-#                #raise TxStateChangeError('OBSOLETED must follow PENDING, SENDFAIL or SENT, but had {}'.format(StatusEnum(self.status).name))
-        #    self.__set_status(StatusEnum.OBSOLETED, session)
-
         if self.tracing:
             self.__state_log(session=session)
 

apps/cic-eth/cic_eth/db/models/role.py

@@ -40,12 +40,14 @@ class AccountRole(SessionBase):
 
         session = SessionBase.bind_session(session)
 
-        role = AccountRole.get_role(tag, session)
+        role = AccountRole.__get_role(tag, session)
     
         r = zero_address
         if role != None:
             r = role.address_hex
 
+        session.flush()
+
         SessionBase.release_session(session)
 
         return r
@@ -63,6 +65,8 @@ class AccountRole(SessionBase):
         session = SessionBase.bind_session(session)
         
         role = AccountRole.__get_role(tag, session)
+
+        session.flush()
         
         SessionBase.release_session(session)
 
@@ -74,7 +78,6 @@ class AccountRole(SessionBase):
         q = session.query(AccountRole)
         q = q.filter(AccountRole.tag==tag)
         r = q.first()
-        session.flush()
         return r
 
 
@@ -93,10 +96,12 @@ class AccountRole(SessionBase):
         """
         session = SessionBase.bind_session(session)
         
-        role = AccountRole.get_role(tag, session)
+        role = AccountRole.__get_role(tag, session)
         if role == None:
             role = AccountRole(tag)
         role.address_hex = address_hex
+
+        session.flush()
         
         SessionBase.release_session(session)
 

apps/cic-eth/cic_eth/db/models/tx.py

@@ -85,26 +85,27 @@ class TxCache(SessionBase):
         :param tx_hash_new: tx hash to associate the copied entry with
         :type tx_hash_new: str, 0x-hex
         """
-        localsession = SessionBase.bind_session(session)
+        session = SessionBase.bind_session(session)
         
-        q = localsession.query(TxCache)
+        q = session.query(TxCache)
         q = q.join(Otx)
         q = q.filter(Otx.tx_hash==tx_hash_original)
         txc = q.first()
 
         if txc == None:
-            SessionBase.release_session(localsession)
+            SessionBase.release_session(session)
             raise NotLocalTxError('original {}'.format(tx_hash_original))
         if txc.block_number != None:
-            SessionBase.release_session(localsession)
+            SessionBase.release_session(session)
             raise TxStateChangeError('cannot clone tx cache of confirmed tx {}'.format(tx_hash_original))
 
-        q = localsession.query(Otx)
+        session.flush()
+        q = session.query(Otx)
         q = q.filter(Otx.tx_hash==tx_hash_new)
         otx = q.first()
 
         if otx == None:
-            SessionBase.release_session(localsession)
+            SessionBase.release_session(session)
             raise NotLocalTxError('new {}'.format(tx_hash_new))
 
         txc_new = TxCache(
@@ -115,18 +116,21 @@ class TxCache(SessionBase):
                 txc.destination_token_address,
                 int(txc.from_value),
                 int(txc.to_value),
+                session=session,
                 )
-        localsession.add(txc_new)
-        localsession.commit()
+        session.add(txc_new)
+        session.commit()
 
-        SessionBase.release_session(localsession)
+        SessionBase.release_session(session)
 
 
     def __init__(self, tx_hash, sender, recipient, source_token_address, destination_token_address, from_value, to_value, block_number=None, tx_index=None, session=None):
-        localsession = SessionBase.bind_session(session)
-        tx = localsession.query(Otx).filter(Otx.tx_hash==tx_hash).first()
+        session = SessionBase.bind_session(session)
+        q = session.query(Otx)
+        q = q.filter(Otx.tx_hash==tx_hash)
+        tx = q.first()
         if tx == None:
-            SessionBase.release_session(localsession)
+            SessionBase.release_session(session)
             raise FileNotFoundError('outgoing transaction record unknown {} (add a Tx first)'.format(tx_hash))
         self.otx_id = tx.id
 
@@ -143,5 +147,5 @@ class TxCache(SessionBase):
         self.date_updated = self.date_created
         self.date_checked = self.date_created
 
-        SessionBase.release_session(localsession)
+        SessionBase.release_session(session)
 

apps/cic-eth/cic_eth/eth/account.py

@@ -8,6 +8,7 @@ from cic_registry import CICRegistry
 from cic_registry.chain import ChainSpec
 from erc20_single_shot_faucet import Faucet
 from cic_registry import zero_address
+from hexathon import strip_0x
 
 # local import
 from cic_eth.eth import RpcClient
@@ -21,6 +22,7 @@ from cic_eth.db.models.role import AccountRole
 from cic_eth.db.models.tx import TxCache
 from cic_eth.eth.util import unpack_signed_raw_tx
 from cic_eth.error import RoleMissingError
+from cic_eth.task import CriticalSQLAlchemyTask
 
 #logg = logging.getLogger(__name__)
 logg = logging.getLogger()
@@ -34,6 +36,7 @@ class AccountTxFactory(TxFactory):
             self,
             address,
             chain_spec,
+            session=None,
             ):
         """Register an Ethereum account address with the on-chain account registry
 
@@ -56,7 +59,7 @@ class AccountTxFactory(TxFactory):
             'gas': gas,
             'gasPrice': self.gas_price,
             'chainId': chain_spec.chain_id(),
-            'nonce': self.next_nonce(),
+            'nonce': self.next_nonce(session=session),
             'value': 0,
             })
         return tx_add
@@ -66,6 +69,7 @@ class AccountTxFactory(TxFactory):
             self,
             address,
             chain_spec,
+            session=None,
         ):
         """Trigger the on-chain faucet to disburse tokens to the provided Ethereum account
 
@@ -86,7 +90,7 @@ class AccountTxFactory(TxFactory):
             'gas': gas,
             'gasPrice': self.gas_price,
             'chainId': chain_spec.chain_id(),
-            'nonce': self.next_nonce(),
+            'nonce': self.next_nonce(session=session),
             'value': 0,
             })
         return tx_add
@@ -101,11 +105,12 @@ def unpack_register(data):
     :returns: Parsed parameters
     :rtype: dict
     """
-    f = data[2:10]
+    data = strip_0x(data)
+    f = data[:8]
     if f != '0a3b0a4f':
         raise ValueError('Invalid account index register data ({})'.format(f))
 
-    d = data[10:]
+    d = data[8:]
     return {
         'to': web3.Web3.toChecksumAddress('0x' + d[64-40:64]),
         }
@@ -120,17 +125,19 @@ def unpack_gift(data):
     :returns: Parsed parameters
     :rtype: dict
     """
-    f = data[2:10]
+    data = strip_0x(data)
+    f = data[:8]
     if f != '63e4bff4':
-        raise ValueError('Invalid account index register data ({})'.format(f))
+        raise ValueError('Invalid gift data ({})'.format(f))
 
-    d = data[10:]
+    d = data[8:]
     return {
         'to': web3.Web3.toChecksumAddress('0x' + d[64-40:64]),
         }
      
 
-@celery_app.task()
+# TODO: Separate out nonce initialization task
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def create(password, chain_str):
     """Creates and stores a new ethereum account in the keystore.
 
@@ -149,9 +156,13 @@ def create(password, chain_str):
     logg.debug('created account {}'.format(a))
 
     # Initialize nonce provider record for account
+    # TODO: this can safely be set to zero, since we are randomly creating account
     n = c.w3.eth.getTransactionCount(a, 'pending')
     session = SessionBase.create_session()
-    o = session.query(Nonce).filter(Nonce.address_hex==a).first()
+    q = session.query(Nonce)
+    q = q.filter(Nonce.address_hex==a)
+    o = q.first()
+    session.flush()
     if o == None:
         o = Nonce()
         o.address_hex = a
@@ -162,7 +173,7 @@ def create(password, chain_str):
     return a
 
 
-@celery_app.task(bind=True, throws=(RoleMissingError,))
+@celery_app.task(bind=True, throws=(RoleMissingError,), base=CriticalSQLAlchemyTask)
 def register(self, account_address, chain_str, writer_address=None):
     """Creates a transaction to add the given address to the accounts index.
 
@@ -180,21 +191,22 @@ def register(self, account_address, chain_str, writer_address=None):
 
     session = SessionBase.create_session()
     if writer_address == None:
-        writer_address = AccountRole.get_address('ACCOUNTS_INDEX_WRITER', session)
-    session.close()
+        writer_address = AccountRole.get_address('ACCOUNTS_INDEX_WRITER', session=session)
 
     if writer_address == zero_address:
+        session.close()
         raise RoleMissingError(account_address)
 
-
     logg.debug('adding account address {} to index; writer {}'.format(account_address, writer_address))
     queue = self.request.delivery_info['routing_key']
 
     c = RpcClient(chain_spec, holder_address=writer_address)
     txf = AccountTxFactory(writer_address, c)
 
-    tx_add = txf.add(account_address, chain_spec)
-    (tx_hash_hex, tx_signed_raw_hex) = sign_and_register_tx(tx_add, chain_str, queue, 'cic_eth.eth.account.cache_account_data')
+    tx_add = txf.add(account_address, chain_spec, session=session)
+    
+    (tx_hash_hex, tx_signed_raw_hex) = sign_and_register_tx(tx_add, chain_str, queue, 'cic_eth.eth.account.cache_account_data', session=session)
+    session.close()
 
     gas_budget = tx_add['gas'] * tx_add['gasPrice']
 
@@ -211,7 +223,7 @@ def register(self, account_address, chain_str, writer_address=None):
     return account_address
 
 
-@celery_app.task(bind=True)
+@celery_app.task(bind=True, base=CriticalSQLAlchemyTask)
 def gift(self, account_address, chain_str):
     """Creates a transaction to invoke the faucet contract for the given address.
 
@@ -230,12 +242,14 @@ def gift(self, account_address, chain_str):
     c = RpcClient(chain_spec, holder_address=account_address)
     txf = AccountTxFactory(account_address, c)
 
-    tx_add = txf.gift(account_address, chain_spec)
-    (tx_hash_hex, tx_signed_raw_hex) = sign_and_register_tx(tx_add, chain_str, queue, 'cic_eth.eth.account.cache_gift_data')
+    session = SessionBase.create_session()
+    tx_add = txf.gift(account_address, chain_spec, session=session)
+    (tx_hash_hex, tx_signed_raw_hex) = sign_and_register_tx(tx_add, chain_str, queue, 'cic_eth.eth.account.cache_gift_data', session=session)
+    session.close()
 
     gas_budget = tx_add['gas'] * tx_add['gasPrice']
 
-    logg.debug('register user tx {}'.format(tx_hash_hex))
+    logg.debug('gift user tx {}'.format(tx_hash_hex))
     s = create_check_gas_and_send_task(
             [tx_signed_raw_hex],
             chain_str,
@@ -326,7 +340,7 @@ def cache_gift_data(
     return (tx_hash_hex, cache_id)
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def cache_account_data(
     tx_hash_hex,
     tx_signed_raw_hex,

apps/cic-eth/cic_eth/eth/factory.py

@@ -32,10 +32,10 @@ class TxFactory:
         logg.debug('txfactory instance address {} gas price'.format(self.address, self.gas_price))
         
 
-    def next_nonce(self):
+    def next_nonce(self, session=None):
         """Returns the current cached nonce value, and increments it for next transaction.
 
         :returns: Nonce
         :rtype: number
         """
-        return self.nonce_oracle.next()
+        return self.nonce_oracle.next(session=session)

apps/cic-eth/cic_eth/eth/nonce.py

@@ -14,10 +14,10 @@ class NonceOracle():
         self.default_nonce = default_nonce
 
 
-    def next(self):
+    def next(self, session=None):
         """Get next unique nonce.
 
         :returns: Nonce
         :rtype: number
         """
-        return Nonce.next(self.address, self.default_nonce)
+        return Nonce.next(self.address, self.default_nonce, session=session)

apps/cic-eth/cic_eth/eth/request.py

@@ -1,194 +0,0 @@
-# standard imports
-import logging
-
-# third-party imports
-import web3
-import celery
-from erc20_approval_escrow import TransferApproval
-from cic_registry import CICRegistry
-from cic_registry.chain import ChainSpec
-
-# local imports
-from cic_eth.db.models.tx import TxCache
-from cic_eth.db.models.base import SessionBase
-from cic_eth.eth import RpcClient
-from cic_eth.eth.factory import TxFactory
-from cic_eth.eth.task import sign_and_register_tx
-from cic_eth.eth.util import unpack_signed_raw_tx
-from cic_eth.eth.task import create_check_gas_and_send_task
-from cic_eth.error import TokenCountError
-
-celery_app = celery.current_app
-logg = logging.getLogger()
-
-contract_function_signatures = {
-        'request': 'b0addede',
-        }
-
-
-class TransferRequestTxFactory(TxFactory):
-    """Factory for creating Transfer request transactions using the TransferApproval contract backend
-    """
-    def request(
-            self,
-            token_address,
-            beneficiary_address,
-            amount,
-            chain_spec,
-            ):
-        """Create a new TransferApproval.request transaction
-
-        :param token_address: Token to create transfer request for
-        :type token_address: str, 0x-hex
-        :param beneficiary_address: Beneficiary of token transfer
-        :type beneficiary_address: str, 0x-hex
-        :param amount: Amount of tokens to transfer
-        :type amount: number
-        :param chain_spec: Chain spec
-        :type chain_spec: cic_registry.chain.ChainSpec
-        :returns: Transaction in standard Ethereum format
-        :rtype: dict
-        """
-        transfer_approval = CICRegistry.get_contract(chain_spec, 'TransferApproval', 'TransferAuthorization')
-        fn = transfer_approval.function('createRequest')
-        tx_approval_buildable = fn(beneficiary_address, token_address, amount)
-        transfer_approval_gas = transfer_approval.gas('createRequest')
-
-        tx_approval = tx_approval_buildable.buildTransaction({
-            'from': self.address,
-            'gas': transfer_approval_gas,
-            'gasPrice': self.gas_price,
-            'chainId': chain_spec.chain_id(),
-            'nonce': self.next_nonce(),
-            })
-        return tx_approval
-
-
-def unpack_transfer_approval_request(data):
-    """Verifies that a transaction is an "TransferApproval.request" transaction, and extracts call parameters from it.
-
-    :param data: Raw input data from Ethereum transaction.
-    :type data: str, 0x-hex
-    :raises ValueError: Function signature does not match AccountRegister.add
-    :returns: Parsed parameters
-    :rtype: dict
-    """
-    f = data[2:10]
-    if f != contract_function_signatures['request']:
-        raise ValueError('Invalid transfer request data ({})'.format(f))
-
-    d = data[10:]
-    return {
-        'to': web3.Web3.toChecksumAddress('0x' + d[64-40:64]),
-        'token': web3.Web3.toChecksumAddress('0x' + d[128-40:128]),
-        'amount': int(d[128:], 16)
-        }
-
-
-@celery_app.task(bind=True)
-def transfer_approval_request(self, tokens, holder_address, receiver_address, value, chain_str):
-    """Creates a new transfer approval
-
-    :param tokens: Token to generate transfer request for
-    :type tokens: list with single token spec as dict
-    :param holder_address: Address to generate transfer on behalf of
-    :type holder_address: str, 0x-hex
-    :param receiver_address: Address to transfser tokens to
-    :type receiver_address: str, 0x-hex
-    :param value: Amount of tokens to transfer
-    :type value: number
-    :param chain_spec: Chain spec string representation
-    :type chain_spec: str
-    :raises cic_eth.error.TokenCountError: More than one token in tokens argument
-    :returns: Raw signed transaction
-    :rtype: list with transaction as only element
-    """
-
-    if len(tokens) != 1:
-        raise TokenCountError
-
-    chain_spec = ChainSpec.from_chain_str(chain_str)
-
-    queue = self.request.delivery_info['routing_key']
-
-    t = tokens[0]
-
-    c = RpcClient(holder_address)
-
-    txf = TransferRequestTxFactory(holder_address, c)
-
-    tx_transfer = txf.request(t['address'], receiver_address, value, chain_spec)
-    (tx_hash_hex, tx_signed_raw_hex) = sign_and_register_tx(tx_transfer, chain_str, queue, 'cic_eth.eth.request.otx_cache_transfer_approval_request')
-
-    gas_budget = tx_transfer['gas'] * tx_transfer['gasPrice']
-
-    s = create_check_gas_and_send_task(
-             [tx_signed_raw_hex],
-             chain_str,
-             holder_address,
-             gas_budget,
-             [tx_hash_hex],
-             queue,
-            )
-    s.apply_async()
-    return [tx_signed_raw_hex]
-
-
-@celery_app.task()
-def otx_cache_transfer_approval_request(
-        tx_hash_hex,
-        tx_signed_raw_hex,
-        chain_str,
-        ):
-    """Generates and commits transaction cache metadata for an TransferApproval.request transaction
-
-    :param tx_hash_hex: Transaction hash
-    :type tx_hash_hex: str, 0x-hex
-    :param tx_signed_raw_hex: Raw signed transaction
-    :type tx_signed_raw_hex: str, 0x-hex
-    :param chain_str: Chain spec string representation
-    :type chain_str: str
-    :returns: Transaction hash and id of cache element in storage backend, respectively
-    :rtype: tuple
-    """
-    chain_spec = ChainSpec.from_chain_str(chain_str)
-    tx_signed_raw_bytes = bytes.fromhex(tx_signed_raw_hex[2:])
-    tx = unpack_signed_raw_tx(tx_signed_raw_bytes, chain_spec.chain_id())
-    logg.debug('in otx acche transfer approval request')
-    (txc, cache_id) = cache_transfer_approval_request_data(tx_hash_hex, tx)
-    return txc
-
-
-@celery_app.task()
-def cache_transfer_approval_request_data(
-    tx_hash_hex,
-    tx,
-        ):
-    """Helper function for otx_cache_transfer_approval_request
-
-    :param tx_hash_hex: Transaction hash
-    :type tx_hash_hex: str, 0x-hex
-    :param tx: Signed raw transaction
-    :type tx: str, 0x-hex
-    :returns: Transaction hash and id of cache element in storage backend, respectively
-    :rtype: tuple
-    """
-    tx_data = unpack_transfer_approval_request(tx['data'])
-    logg.debug('tx approval request data {}'.format(tx_data))
-    logg.debug('tx approval request {}'.format(tx))
-
-    session = SessionBase.create_session()
-    tx_cache = TxCache(
-        tx_hash_hex,
-        tx['from'],
-        tx_data['to'],
-        tx_data['token'],
-        tx_data['token'],
-        tx_data['amount'],
-        tx_data['amount'],
-            )
-    session.add(tx_cache)
-    session.commit()
-    cache_id = tx_cache.id
-    session.close()
-    return (tx_hash_hex, cache_id)

apps/cic-eth/cic_eth/eth/task.py

@@ -33,7 +33,7 @@ def sign_tx(tx, chain_str):
     return (tx_hash_hex, tx_transfer_signed['raw'],)
 
 
-def sign_and_register_tx(tx, chain_str, queue, cache_task=None):
+def sign_and_register_tx(tx, chain_str, queue, cache_task=None, session=None):
     """Signs the provided transaction, and adds it to the transaction queue cache (with status PENDING).
 
     :param tx: Standard ethereum transaction data
@@ -44,6 +44,7 @@ def sign_and_register_tx(tx, chain_str, queue, cache_task=None):
     :type queue: str
     :param cache_task: Cache task to call with signed transaction. If None, no task will be called.
     :type cache_task: str
+    :raises: sqlalchemy.exc.DatabaseError
     :returns: Tuple; Transaction hash, signed raw transaction data
     :rtype: tuple
     """
@@ -51,25 +52,13 @@ def sign_and_register_tx(tx, chain_str, queue, cache_task=None):
 
     logg.debug('adding queue tx {}'.format(tx_hash_hex))
 
-#    s = celery.signature(
-#        'cic_eth.queue.tx.create',
-#        [
-#            tx['nonce'],
-#            tx['from'],
-#            tx_hash_hex,
-#            tx_signed_raw_hex,
-#            chain_str,
-#            ],
-#        queue=queue,
-#        )
-
-    # TODO: consider returning this as a signature that consequtive tasks can be linked to
     queue_create(
         tx['nonce'],
         tx['from'],
         tx_hash_hex,
         tx_signed_raw_hex,
         chain_str,
+        session=session,
     )        
 
     if cache_task != None:

apps/cic-eth/cic_eth/eth/token.py

@@ -8,6 +8,8 @@ import web3
 from cic_registry import CICRegistry
 from cic_registry import zero_address
 from cic_registry.chain import ChainSpec
+from hexathon import strip_0x
+from chainlib.status import Status as TxStatus
 
 # platform imports
 from cic_eth.db.models.tx import TxCache
@@ -19,6 +21,10 @@ from cic_eth.eth.task import create_check_gas_and_send_task
 from cic_eth.eth.factory import TxFactory
 from cic_eth.eth.util import unpack_signed_raw_tx
 from cic_eth.ext.address import translate_address
+from cic_eth.task import (
+        CriticalSQLAlchemyTask,
+        CriticalWeb3Task,
+    )
 
 celery_app = celery.current_app
 logg = logging.getLogger()
@@ -40,6 +46,7 @@ class TokenTxFactory(TxFactory):
             spender_address,
             amount,
             chain_spec,
+            session=None,
             ):
         """Create an ERC20 "approve" transaction
 
@@ -67,7 +74,7 @@ class TokenTxFactory(TxFactory):
             'gas': source_token_gas,
             'gasPrice': self.gas_price,
             'chainId': chain_spec.chain_id(),
-            'nonce': self.next_nonce(),
+            'nonce': self.next_nonce(session=session),
             })
         return tx_approve
 
@@ -78,6 +85,7 @@ class TokenTxFactory(TxFactory):
         receiver_address,
         value,
         chain_spec,
+        session=None,
         ):
         """Create an ERC20 "transfer" transaction
 
@@ -106,7 +114,7 @@ class TokenTxFactory(TxFactory):
                     'gas': source_token_gas,
                     'gasPrice': self.gas_price,
                     'chainId': chain_spec.chain_id(),
-                    'nonce': self.next_nonce(),
+                    'nonce': self.next_nonce(session=session),
                 })
         return tx_transfer
 
@@ -120,11 +128,12 @@ def unpack_transfer(data):
     :returns: Parsed parameters
     :rtype: dict
     """
-    f = data[2:10]
+    data = strip_0x(data)
+    f = data[:8]
     if f != contract_function_signatures['transfer']:
         raise ValueError('Invalid transfer data ({})'.format(f))
 
-    d = data[10:]
+    d = data[8:]
     return {
         'to': web3.Web3.toChecksumAddress('0x' + d[64-40:64]),
         'amount': int(d[64:], 16)
@@ -140,11 +149,12 @@ def unpack_transferfrom(data):
     :returns: Parsed parameters
     :rtype: dict
     """
-    f = data[2:10]
+    data = strip_0x(data)
+    f = data[:8]
     if f != contract_function_signatures['transferfrom']:
         raise ValueError('Invalid transferFrom data ({})'.format(f))
 
-    d = data[10:]
+    d = data[8:]
     return {
         'from': web3.Web3.toChecksumAddress('0x' + d[64-40:64]),
         'to': web3.Web3.toChecksumAddress('0x' + d[128-40:128]),
@@ -161,18 +171,19 @@ def unpack_approve(data):
     :returns: Parsed parameters
     :rtype: dict
     """
-    f = data[2:10]
+    data = strip_0x(data)
+    f = data[:8]
     if f != contract_function_signatures['approve']:
         raise ValueError('Invalid approval data ({})'.format(f))
 
-    d = data[10:]
+    d = data[8:]
     return {
         'to': web3.Web3.toChecksumAddress('0x' + d[64-40:64]),
         'amount': int(d[64:], 16)
         }
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalWeb3Task)
 def balance(tokens, holder_address, chain_str):
     """Return token balances for a list of tokens for given address
 
@@ -199,7 +210,7 @@ def balance(tokens, holder_address, chain_str):
     return tokens
 
 
-@celery_app.task(bind=True)
+@celery_app.task(bind=True, base=CriticalSQLAlchemyTask)
 def transfer(self, tokens, holder_address, receiver_address, value, chain_str):
     """Transfer ERC20 tokens between addresses
 
@@ -235,9 +246,11 @@ def transfer(self, tokens, holder_address, receiver_address, value, chain_str):
     c = RpcClient(chain_spec, holder_address=holder_address)
 
     txf = TokenTxFactory(holder_address, c)
-
-    tx_transfer = txf.transfer(t['address'], receiver_address, value, chain_spec)
-    (tx_hash_hex, tx_signed_raw_hex) = sign_and_register_tx(tx_transfer, chain_str, queue, cache_task='cic_eth.eth.token.otx_cache_transfer')
+    
+    session = SessionBase.create_session()
+    tx_transfer = txf.transfer(t['address'], receiver_address, value, chain_spec, session=session)
+    (tx_hash_hex, tx_signed_raw_hex) = sign_and_register_tx(tx_transfer, chain_str, queue, cache_task='cic_eth.eth.token.otx_cache_transfer', session=session)
+    session.close()
     
     gas_budget = tx_transfer['gas'] * tx_transfer['gasPrice']
 
@@ -253,7 +266,7 @@ def transfer(self, tokens, holder_address, receiver_address, value, chain_str):
     return tx_hash_hex
 
 
-@celery_app.task(bind=True)
+@celery_app.task(bind=True, base=CriticalSQLAlchemyTask)
 def approve(self, tokens, holder_address, spender_address, value, chain_str):
     """Approve ERC20 transfer on behalf of holder address
 
@@ -290,8 +303,10 @@ def approve(self, tokens, holder_address, spender_address, value, chain_str):
 
     txf = TokenTxFactory(holder_address, c)
 
-    tx_transfer = txf.approve(t['address'], spender_address, value, chain_spec)
-    (tx_hash_hex, tx_signed_raw_hex) = sign_and_register_tx(tx_transfer, chain_str, queue, cache_task='cic_eth.eth.token.otx_cache_approve')
+    session = SessionBase.create_session()
+    tx_transfer = txf.approve(t['address'], spender_address, value, chain_spec, session=session)
+    (tx_hash_hex, tx_signed_raw_hex) = sign_and_register_tx(tx_transfer, chain_str, queue, cache_task='cic_eth.eth.token.otx_cache_approve', session=session)
+    session.close()
     
     gas_budget = tx_transfer['gas'] * tx_transfer['gasPrice']
 
@@ -307,7 +322,7 @@ def approve(self, tokens, holder_address, spender_address, value, chain_str):
     return tx_hash_hex
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalWeb3Task)
 def resolve_tokens_by_symbol(token_symbols, chain_str):
     """Returns contract addresses of an array of ERC20 token symbols
 
@@ -330,7 +345,7 @@ def resolve_tokens_by_symbol(token_symbols, chain_str):
     return tokens
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def otx_cache_transfer(
         tx_hash_hex,
         tx_signed_raw_hex,
@@ -354,7 +369,7 @@ def otx_cache_transfer(
     return txc
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def cache_transfer_data(
     tx_hash_hex,
     tx,
@@ -390,7 +405,7 @@ def cache_transfer_data(
     return (tx_hash_hex, cache_id)
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def otx_cache_approve(
         tx_hash_hex,
         tx_signed_raw_hex,
@@ -414,7 +429,7 @@ def otx_cache_approve(
     return txc
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def cache_approve_data(
     tx_hash_hex,
     tx,
@@ -450,6 +465,7 @@ def cache_approve_data(
     return (tx_hash_hex, cache_id)
 
 
+# TODO: Move to dedicated metadata package
 class ExtendedTx:
 
     _default_decimals = 6
@@ -470,6 +486,8 @@ class ExtendedTx:
         self.destination_token_symbol = ''
         self.source_token_decimals = ExtendedTx._default_decimals
         self.destination_token_decimals = ExtendedTx._default_decimals
+        self.status = TxStatus.PENDING.name
+        self.status_code = TxStatus.PENDING.value
 
 
     def set_actors(self, sender, recipient, trusted_declarator_addresses=None):
@@ -497,10 +515,18 @@ class ExtendedTx:
         self.destination_token_value = destination_value
 
 
+    def set_status(self, n):
+        if n:
+            self.status = TxStatus.ERROR.name
+        else:
+            self.status = TxStatus.SUCCESS.name
+        self.status_code = n
+
+
     def to_dict(self):
         o = {}
         for attr in dir(self):
-            if attr[0] == '_' or attr in ['set_actors', 'set_tokens', 'to_dict']:
+            if attr[0] == '_' or attr in ['set_actors', 'set_tokens', 'set_status', 'to_dict']:
                 continue
             o[attr] = getattr(self, attr)
         return o 

apps/cic-eth/cic_eth/eth/tx.py

@@ -32,6 +32,10 @@ from cic_eth.eth.nonce import NonceOracle
 from cic_eth.error import AlreadyFillingGasError
 from cic_eth.eth.util import tx_hex_string
 from cic_eth.admin.ctrl import lock_send
+from cic_eth.task import (
+        CriticalSQLAlchemyTask,
+        CriticalWeb3Task,
+        )
 
 celery_app = celery.current_app
 logg = logging.getLogger()
@@ -40,7 +44,7 @@ MAX_NONCE_ATTEMPTS = 3
 
 
 # TODO this function is too long
-@celery_app.task(bind=True, throws=(OutOfGasError))
+@celery_app.task(bind=True, throws=(OutOfGasError), base=CriticalSQLAlchemyTask)
 def check_gas(self, tx_hashes, chain_str, txs=[], address=None, gas_required=None):
     """Check the gas level of the sender address of a transaction.
 
@@ -65,7 +69,6 @@ def check_gas(self, tx_hashes, chain_str, txs=[], address=None, gas_required=Non
         for i in range(len(tx_hashes)):
             o = get_tx(tx_hashes[i])
             txs.append(o['signed_tx'])
-            logg.debug('ooooo {}'.format(o))
             if address == None:
                 address = o['address']
 
@@ -131,7 +134,7 @@ def check_gas(self, tx_hashes, chain_str, txs=[], address=None, gas_required=Non
 
 
 # TODO: chain chainable transactions that use hashes as inputs may be chained to this function to output signed txs instead.
-@celery_app.task(bind=True)
+@celery_app.task(bind=True, base=CriticalSQLAlchemyTask)
 def hashes_to_txs(self, tx_hashes):
     """Return a list of raw signed transactions from the local transaction queue corresponding to a list of transaction hashes.
 
@@ -174,12 +177,7 @@ class ParityNodeHandler:
     def handle(self, exception, tx_hash_hex, tx_hex):
         meth = self.handle_default
         if isinstance(exception, (ValueError)):
-#            s_debug = celery.signature(
-#                'cic_eth.admin.debug.out_tmp',
-#                [tx_hash_hex, '{}: {}'.format(tx_hash_hex, exception)],
-#                queue=queue,
-#                )
-#            s_debug.apply_async()
+            
             earg = exception.args[0]
             if earg['code'] == -32010:
                 logg.debug('skipping lock for code {}'.format(earg['code']))
@@ -187,14 +185,15 @@ class ParityNodeHandler:
             elif earg['code'] == -32602:
                 meth = self.handle_invalid_encoding
             else:
+                # TODO: move to status log db comment field
                 meth = self.handle_invalid
         elif isinstance(exception, (requests.exceptions.ConnectionError)):
             meth = self.handle_connection
-        (t, e_fn, message) = meth(tx_hash_hex, tx_hex)
+        (t, e_fn, message) = meth(tx_hash_hex, tx_hex, str(exception))
         return (t, e_fn, '{} {}'.format(message, exception))
            
 
-    def handle_connection(self, tx_hash_hex, tx_hex):
+    def handle_connection(self, tx_hash_hex, tx_hex, debugstr=None):
         s_set_sent = celery.signature(
             'cic_eth.queue.tx.set_sent_status',
             [
@@ -207,7 +206,7 @@ class ParityNodeHandler:
         return (t, TemporaryTxError, 'Sendfail {}'.format(tx_hex_string(tx_hex, self.chain_spec.chain_id())))
 
 
-    def handle_invalid_encoding(self, tx_hash_hex, tx_hex):
+    def handle_invalid_encoding(self, tx_hash_hex, tx_hex, debugstr=None):
         tx_bytes = bytes.fromhex(tx_hex[2:])
         tx = unpack_signed_raw_tx(tx_bytes, self.chain_spec.chain_id())
         s_lock = celery.signature(
@@ -254,7 +253,7 @@ class ParityNodeHandler:
         return (t, PermanentTxError, 'Reject invalid encoding {}'.format(tx_hex_string(tx_hex, self.chain_spec.chain_id())))
 
 
-    def handle_invalid_parameters(self, tx_hash_hex, tx_hex):
+    def handle_invalid_parameters(self, tx_hash_hex, tx_hex, debugstr=None):
         s_sync = celery.signature(
             'cic_eth.eth.tx.sync_tx',
             [
@@ -267,7 +266,7 @@ class ParityNodeHandler:
         return (t, PermanentTxError, 'Reject invalid parameters {}'.format(tx_hex_string(tx_hex, self.chain_spec.chain_id())))
 
 
-    def handle_invalid(self, tx_hash_hex, tx_hex):
+    def handle_invalid(self, tx_hash_hex, tx_hex, debugstr=None):
         tx_bytes = bytes.fromhex(tx_hex[2:])
         tx = unpack_signed_raw_tx(tx_bytes, self.chain_spec.chain_id())
         s_lock = celery.signature(
@@ -285,6 +284,16 @@ class ParityNodeHandler:
             [],
             queue=self.queue,
             )
+        s_debug = celery.signature(
+            'cic_eth.admin.debug.alert',
+            [
+                tx_hash_hex,
+                tx_hash_hex,
+                debugstr,
+                ],
+            queue=queue,
+            )
+        s_set_reject.link(s_debug)
         s_lock.link(s_set_reject)
         t = s_lock.apply_async()
         return (t, PermanentTxError, 'Reject invalid {}'.format(tx_hex_string(tx_hex, self.chain_spec.chain_id())))
@@ -313,7 +322,8 @@ class ParityNodeHandler:
         return (t, PermanentTxError, 'Fubar {}'.format(tx_hex_string(tx_hex, self.chain_spec.chain_id())))
 
 
-@celery_app.task(bind=True)
+# TODO: A lock should be introduced to ensure that the send status change and the transaction send is atomic.
+@celery_app.task(bind=True, base=CriticalWeb3Task)
 def send(self, txs, chain_str):
     """Send transactions to the network.
 
@@ -351,13 +361,6 @@ def send(self, txs, chain_str):
 
     c = RpcClient(chain_spec)
     r = None
-    try:
-        r = c.w3.eth.send_raw_transaction(tx_hex)
-    except Exception as e:
-        raiser = ParityNodeHandler(chain_spec, queue)
-        (t, e, m) = raiser.handle(e, tx_hash_hex, tx_hex)
-        raise e(m)
-
     s_set_sent = celery.signature(
         'cic_eth.queue.tx.set_sent_status',
         [
@@ -366,6 +369,14 @@ def send(self, txs, chain_str):
             ],
             queue=queue,
         )
+    try:
+        r = c.w3.eth.send_raw_transaction(tx_hex)
+    except requests.exceptions.ConnectionError as e:
+        raise(e)
+    except Exception as e:
+        raiser = ParityNodeHandler(chain_spec, queue)
+        (t, e, m) = raiser.handle(e, tx_hash_hex, tx_hex)
+        raise e(m)
     s_set_sent.apply_async()
 
     tx_tail = txs[1:]
@@ -380,7 +391,9 @@ def send(self, txs, chain_str):
     return r.hex()
 
 
-@celery_app.task(bind=True, throws=(AlreadyFillingGasError))
+# TODO: if this method fails the nonce will be out of sequence. session needs to be extended to include the queue create, so that nonce is rolled back if the second sql query fails. Better yet, split each state change into separate tasks.
+# TODO: method is too long, factor out code for clarity
+@celery_app.task(bind=True, throws=(web3.exceptions.TransactionNotFound,), base=CriticalWeb3Task)
 def refill_gas(self, recipient_address, chain_str):
     """Executes a native token transaction to fund the recipient's gas expenditures.
 
@@ -402,8 +415,8 @@ def refill_gas(self, recipient_address, chain_str):
     q = q.filter(TxCache.from_value!=0)
     q = q.filter(TxCache.recipient==recipient_address)
     c = q.count()
-    session.close()
     if c > 0:
+        session.close()
         raise AlreadyFillingGasError(recipient_address)
 
     queue = self.request.delivery_info['routing_key']
@@ -413,7 +426,7 @@ def refill_gas(self, recipient_address, chain_str):
     logg.debug('refill gas from provider address {}'.format(c.gas_provider()))
     default_nonce = c.w3.eth.getTransactionCount(c.gas_provider(), 'pending')
     nonce_generator = NonceOracle(c.gas_provider(), default_nonce)
-    nonce = nonce_generator.next()
+    nonce = nonce_generator.next(session=session)
     gas_price = c.gas_price()
     gas_limit = c.default_gas_limit
     refill_amount = c.refill_amount()
@@ -442,7 +455,9 @@ def refill_gas(self, recipient_address, chain_str):
         tx_hash_hex,
         tx_send_gas_signed['raw'],
         chain_str,
+        session=session,
         )
+    session.close()
 
     s_tx_cache = celery.signature(
             'cic_eth.eth.tx.cache_gas_refill_data',
@@ -485,8 +500,8 @@ def resend_with_higher_gas(self, txold_hash_hex, chain_str, gas=None, default_fa
     q = session.query(Otx)
     q = q.filter(Otx.tx_hash==txold_hash_hex)
     otx = q.first()
-    session.close()
     if otx == None:
+        session.close()
         raise NotLocalTxError(txold_hash_hex)
 
     chain_spec = ChainSpec.from_chain_str(chain_str)
@@ -521,8 +536,10 @@ def resend_with_higher_gas(self, txold_hash_hex, chain_str, gas=None, default_fa
         tx_hash_hex,
         tx_signed_raw_hex,
         chain_str,
+        session=session,
             )
-    TxCache.clone(txold_hash_hex, tx_hash_hex)
+    TxCache.clone(txold_hash_hex, tx_hash_hex, session=session)
+    session.close()
 
     s = create_check_gas_and_send_task(
             [tx_signed_raw_hex],
@@ -537,8 +554,15 @@ def resend_with_higher_gas(self, txold_hash_hex, chain_str, gas=None, default_fa
     return tx_hash_hex
 
 
-@celery_app.task(bind=True, throws=(web3.exceptions.TransactionNotFound,))
+@celery_app.task(bind=True, throws=(web3.exceptions.TransactionNotFound,), base=CriticalWeb3Task)
 def sync_tx(self, tx_hash_hex, chain_str):
+    """Force update of network status of a simgle transaction
+
+    :param tx_hash_hex: Transaction hash
+    :type tx_hash_hex: str, 0x-hex
+    :param chain_str: Chain spec string representation
+    :type chain_str: str
+    """
 
     queue = self.request.delivery_info['routing_key']
 
@@ -621,7 +645,7 @@ def resume_tx(self, txpending_hash_hex, chain_str):
     return txpending_hash_hex
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def otx_cache_parse_tx(
         tx_hash_hex,
         tx_signed_raw_hex,
@@ -648,7 +672,7 @@ def otx_cache_parse_tx(
     return txc
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def cache_gas_refill_data(
         tx_hash_hex,
         tx,

apps/cic-eth/cic_eth/ext/tx.py

@@ -149,6 +149,9 @@ def tx_collate(tx_batches, chain_str, offset, limit, newest_first=True):
     txs_by_block = {}
     chain_spec = ChainSpec.from_chain_str(chain_str)
 
+    if isinstance(tx_batches, dict):
+        tx_batches = [tx_batches]
+
     for b in tx_batches:
         for v in b.values():
             tx = None

apps/cic-eth/cic_eth/queue/balance.py

@@ -14,6 +14,7 @@ from cic_eth.db.enum import (
         StatusBits,
         dead,
         )
+from cic_eth.task import CriticalSQLAlchemyTask
 
 celery_app = celery.current_app
 
@@ -35,7 +36,7 @@ def __balance_outgoing_compatible(token_address, holder_address, chain_str):
     return delta
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def balance_outgoing(tokens, holder_address, chain_str):
     """Retrieve accumulated value of unprocessed transactions sent from the given address.
 
@@ -73,7 +74,7 @@ def __balance_incoming_compatible(token_address, receiver_address, chain_str):
     return delta
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def balance_incoming(tokens, receipient_address, chain_str):
     """Retrieve accumulated value of unprocessed transactions to be received by the given address.
 

apps/cic-eth/cic_eth/queue/time.py

@@ -10,6 +10,7 @@ from cic_registry.chain import ChainSpec
 from cic_eth.eth.rpc import RpcClient
 from cic_eth.db.models.otx import Otx
 from cic_eth.error import NotLocalTxError
+from cic_eth.task import CriticalSQLAlchemyAndWeb3Task
 
 celery_app = celery.current_app
 
@@ -17,7 +18,7 @@ logg = logging.getLogger()
 
 
 # TODO: This method does not belong in the _queue_ module, it operates across queue and network
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyAndWeb3Task)
 def tx_times(tx_hash, chain_str):
     chain_spec = ChainSpec.from_chain_str(chain_str)
     c = RpcClient(chain_spec)

apps/cic-eth/cic_eth/queue/tx.py

@@ -26,9 +26,11 @@ from cic_eth.db.enum import (
         is_alive,
         dead,
         )
+from cic_eth.task import CriticalSQLAlchemyTask
 from cic_eth.eth.util import unpack_signed_raw_tx # TODO: should not be in same sub-path as package that imports queue.tx
 from cic_eth.error import NotLocalTxError
 from cic_eth.error import LockedError
+from cic_eth.db.enum import status_str
 
 celery_app = celery.current_app
 #logg = celery_app.log.get_default_logger()
@@ -86,7 +88,7 @@ def create(nonce, holder_address, tx_hash, signed_tx, chain_str, obsolete_predec
 
 
 # TODO: Replace set_* with single task for set status
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def set_sent_status(tx_hash, fail=False):
     """Used to set the status after a send attempt
 
@@ -118,7 +120,7 @@ def set_sent_status(tx_hash, fail=False):
     return tx_hash
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def set_final_status(tx_hash, block=None, fail=False):
     """Used to set the status of an incoming transaction result. 
 
@@ -174,7 +176,7 @@ def set_final_status(tx_hash, block=None, fail=False):
     return tx_hash
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def set_cancel(tx_hash, manual=False):
     """Used to set the status when a transaction is cancelled.
 
@@ -206,7 +208,7 @@ def set_cancel(tx_hash, manual=False):
     return tx_hash
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def set_rejected(tx_hash):
     """Used to set the status when the node rejects sending a transaction to network
 
@@ -232,7 +234,7 @@ def set_rejected(tx_hash):
     return tx_hash
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def set_fubar(tx_hash):
     """Used to set the status when an unexpected error occurs.
 
@@ -258,7 +260,7 @@ def set_fubar(tx_hash):
     return tx_hash
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def set_manual(tx_hash):
     """Used to set the status when queue is manually changed
 
@@ -284,7 +286,7 @@ def set_manual(tx_hash):
     return tx_hash
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def set_ready(tx_hash):
     """Used to mark a transaction as ready to be sent to network
 
@@ -310,7 +312,25 @@ def set_ready(tx_hash):
     return tx_hash
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
+def set_dequeue(tx_hash):
+    session = SessionBase.create_session()
+    o = session.query(Otx).filter(Otx.tx_hash==tx_hash).first()
+    if o == None:
+        session.close()
+        raise NotLocalTxError('queue does not contain tx hash {}'.format(tx_hash))
+
+    session.flush()
+
+    o.dequeue(session=session)
+    session.commit()
+    session.close()
+
+    return tx_hash
+
+
+
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def set_waitforgas(tx_hash):
     """Used to set the status when a transaction must be deferred due to gas refill
 
@@ -336,7 +356,7 @@ def set_waitforgas(tx_hash):
     return tx_hash
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def get_state_log(tx_hash):
 
     logs = []
@@ -355,7 +375,7 @@ def get_state_log(tx_hash):
     return logs
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def get_tx_cache(tx_hash):
     """Returns an aggregate dictionary of outgoing transaction data and metadata
 
@@ -386,7 +406,7 @@ def get_tx_cache(tx_hash):
         'tx_hash': otx.tx_hash,
         'signed_tx': otx.signed_tx,
         'nonce': otx.nonce,
-        'status': StatusEnum(otx.status).name,
+        'status': status_str(otx.status),
         'status_code': otx.status,
         'source_token': txc.source_token_address,
         'destination_token': txc.destination_token_address,
@@ -404,7 +424,7 @@ def get_tx_cache(tx_hash):
     return tx
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def get_lock(address=None):
     """Retrieve all active locks
 
@@ -442,7 +462,7 @@ def get_lock(address=None):
     return locks
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def get_tx(tx_hash):
     """Retrieve a transaction queue record by transaction hash
 
@@ -453,7 +473,9 @@ def get_tx(tx_hash):
     :rtype: dict
     """
     session = SessionBase.create_session()
-    tx = session.query(Otx).filter(Otx.tx_hash==tx_hash).first()
+    q = session.query(Otx)
+    q = q.filter(Otx.tx_hash==tx_hash)
+    tx = q.first()
     if tx == None:
         session.close()
         raise NotLocalTxError('queue does not contain tx hash {}'.format(tx_hash))
@@ -469,7 +491,7 @@ def get_tx(tx_hash):
     return o
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def get_nonce_tx(nonce, sender, chain_id):
     """Retrieve all transactions for address with specified nonce
 
@@ -652,7 +674,7 @@ def get_upcoming_tx(status=StatusEnum.READYSEND, recipient=None, before=None, ch
     return txs
 
 
-@celery_app.task()
+@celery_app.task(base=CriticalSQLAlchemyTask)
 def get_account_tx(address, as_sender=True, as_recipient=True, counterpart=None):
     """Returns all local queue transactions for a given Ethereum address
 

apps/cic-eth/cic_eth/registry.py

@@ -0,0 +1,86 @@
+# standard imports
+import logging
+import copy
+
+# external imports
+from cic_registry import CICRegistry
+from eth_token_index import TokenUniqueSymbolIndex
+from eth_accounts_index import AccountRegistry
+from chainlib.chain import ChainSpec
+from cic_registry.chain import ChainRegistry
+from cic_registry.helper.declarator import DeclaratorOracleAdapter
+
+logg = logging.getLogger(__name__)
+
+
+class TokenOracle:
+
+    def __init__(self, conn, chain_spec, registry):
+        self.tokens = []
+        self.chain_spec = chain_spec
+        self.registry = registry
+
+        token_registry_contract = CICRegistry.get_contract(chain_spec, 'TokenRegistry', 'Registry')
+        self.getter = TokenUniqueSymbolIndex(conn, token_registry_contract.address())
+
+
+    def get_tokens(self):
+        token_count = self.getter.count()
+        if token_count == len(self.tokens):
+            return self.tokens
+
+        for i in range(len(self.tokens), token_count):
+            token_address = self.getter.get_index(i)
+            t = self.registry.get_address(self.chain_spec, token_address)
+            token_symbol = t.symbol()
+            self.tokens.append(t)
+
+            logg.debug('adding token idx {} symbol {} address {}'.format(i, token_symbol, token_address))
+
+        return copy.copy(self.tokens)
+
+
+class AccountsOracle:
+
+    def __init__(self, conn, chain_spec, registry):
+        self.accounts = []
+        self.chain_spec = chain_spec
+        self.registry = registry
+
+        accounts_registry_contract = CICRegistry.get_contract(chain_spec, 'AccountRegistry', 'Registry')
+        self.getter = AccountRegistry(conn, accounts_registry_contract.address())
+
+
+    def get_accounts(self):
+        accounts_count = self.getter.count()
+        if accounts_count == len(self.accounts):
+            return self.accounts
+
+        for i in range(len(self.accounts), accounts_count):
+            account = self.getter.get_index(i)
+            self.accounts.append(account)
+            logg.debug('adding account {}'.format(account))
+
+        return copy.copy(self.accounts)
+
+
+def init_registry(config, w3):
+    chain_spec = ChainSpec.from_chain_str(config.get('CIC_CHAIN_SPEC'))
+    CICRegistry.init(w3, config.get('CIC_REGISTRY_ADDRESS'), chain_spec)
+    CICRegistry.add_path(config.get('ETH_ABI_DIR'))
+
+    chain_registry = ChainRegistry(chain_spec)
+    CICRegistry.add_chain_registry(chain_registry, True)
+
+    declarator = CICRegistry.get_contract(chain_spec, 'AddressDeclarator', interface='Declarator')
+    trusted_addresses_src = config.get('CIC_TRUST_ADDRESS')
+    if trusted_addresses_src == None:
+        raise ValueError('At least one trusted address must be declared in CIC_TRUST_ADDRESS')
+    trusted_addresses = trusted_addresses_src.split(',')
+    for address in trusted_addresses:
+        logg.info('using trusted address {}'.format(address))
+
+    oracle = DeclaratorOracleAdapter(declarator.contract, trusted_addresses)
+    chain_registry.add_oracle(oracle, 'naive_erc20_oracle')
+
+    return CICRegistry

apps/cic-eth/cic_eth/runnable/daemons/dispatcher.py

@@ -21,14 +21,21 @@ import cic_eth
 from cic_eth.eth import RpcClient
 from cic_eth.db import SessionBase
 from cic_eth.db.enum import StatusEnum
+from cic_eth.db.enum import StatusBits
 from cic_eth.db.enum import LockEnum
 from cic_eth.db import dsn_from_config
-from cic_eth.queue.tx import get_upcoming_tx
+from cic_eth.queue.tx import (
+        get_upcoming_tx,
+        set_dequeue,
+        )
 from cic_eth.admin.ctrl import lock_send
 from cic_eth.sync.error import LoopDone
 from cic_eth.eth.tx import send as task_tx_send
-from cic_eth.error import PermanentTxError
-from cic_eth.error import TemporaryTxError
+from cic_eth.error import (
+        PermanentTxError,
+        TemporaryTxError,
+        NotLocalTxError,
+        )
 from cic_eth.eth.util import unpack_signed_raw_tx_hex
 
 logging.basicConfig(level=logging.WARNING)
@@ -109,6 +116,11 @@ class DispatchSyncer:
         for k in txs.keys():
             tx_raw = txs[k]
             tx = unpack_signed_raw_tx_hex(tx_raw, self.chain_spec.chain_id())
+            
+            try:
+                set_dequeue(tx['hash'])
+            except NotLocalTxError as e:
+                logg.warning('dispatcher was triggered with non-local tx {}'.format(tx['hash']))
 
             s_check = celery.signature(
                 'cic_eth.admin.ctrl.check_lock',
@@ -129,12 +141,13 @@ class DispatchSyncer:
                     )
             s_check.link(s_send)
             t = s_check.apply_async()
+            logg.info('processed {}'.format(k))
 
 
     def loop(self, w3, interval):
         while run:
             txs = {}
-            typ = StatusEnum.READYSEND
+            typ = StatusBits.QUEUED
             utxs = get_upcoming_tx(typ, chain_id=self.chain_id)
             for k in utxs.keys():
                 txs[k] = utxs[k]

apps/cic-eth/cic_eth/runnable/daemons/filters/__init__.py

@@ -2,3 +2,4 @@ from .callback import CallbackFilter
 from .tx import TxFilter
 from .gas import GasFilter
 from .register import RegistrationFilter
+from .transferauth import TransferAuthFilter

apps/cic-eth/cic_eth/runnable/daemons/filters/callback.py

@@ -5,37 +5,46 @@ import logging
 import web3
 import celery
 from cic_registry.error import UnknownContractError
+from chainlib.status import Status as TxStatus
+from chainlib.eth.address import to_checksum
+from chainlib.eth.constant import ZERO_ADDRESS
+from hexathon import strip_0x
 
 # local imports
 from .base import SyncFilter
-from cic_eth.eth.token import unpack_transfer
-from cic_eth.eth.token import unpack_transferfrom
+from cic_eth.eth.token import (
+        unpack_transfer,
+        unpack_transferfrom,
+        )
+from cic_eth.eth.account import unpack_gift
 from cic_eth.eth.token import ExtendedTx
 from .base import SyncFilter
 
-logg = logging.getLogger()
+logg = logging.getLogger(__name__)
 
-transfer_method_signature = '0xa9059cbb' # keccak256(transfer(address,uint256))
-transferfrom_method_signature = '0x23b872dd' # keccak256(transferFrom(address,address,uint256))
-giveto_method_signature = '0x63e4bff4' # keccak256(giveTo(address))
+transfer_method_signature = 'a9059cbb' # keccak256(transfer(address,uint256))
+transferfrom_method_signature = '23b872dd' # keccak256(transferFrom(address,address,uint256))
+giveto_method_signature = '63e4bff4' # keccak256(giveTo(address))
 
 
 class CallbackFilter(SyncFilter):
 
     trusted_addresses = []
 
-    def __init__(self, method, queue):
+    def __init__(self, chain_spec, method, queue):
         self.queue = queue
         self.method = method
+        self.chain_spec = chain_spec
 
 
     def call_back(self, transfer_type, result):
+        logg.debug('result {}'.format(result))
         s = celery.signature(
             self.method,
             [
                 result,
                 transfer_type,
-                int(rcpt.status == 0),
+                int(result['status_code'] == 0),
             ],
             queue=self.queue,
             )
@@ -50,58 +59,85 @@ class CallbackFilter(SyncFilter):
 #            )
 #        s_translate.link(s)
 #        s_translate.apply_async()
-        s.apply_async()
+        t = s.apply_async()
+        return s
 
 
-    def parse_data(self, tx, rcpt):
-        transfer_type = 'transfer'
+    def parse_data(self, tx):
+        transfer_type = None
         transfer_data = None
-        method_signature = tx.input[:10]
+        logg.debug('have payload {}'.format(tx.payload))
+        method_signature = tx.payload[:8]
 
+        logg.debug('tx status {}'.format(tx.status))
         if method_signature == transfer_method_signature:
-            transfer_data = unpack_transfer(tx.input)
+            transfer_data = unpack_transfer(tx.payload)
             transfer_data['from'] = tx['from']
             transfer_data['token_address'] = tx['to']
 
         elif method_signature == transferfrom_method_signature:
             transfer_type = 'transferfrom'
-            transfer_data = unpack_transferfrom(tx.input)
+            transfer_data = unpack_transferfrom(tx.payload)
             transfer_data['token_address'] = tx['to']
 
         # TODO: do not rely on logs here
         elif method_signature == giveto_method_signature:
             transfer_type = 'tokengift'
-            transfer_data = unpack_gift(tx.input)
-            for l in rcpt.logs:
-                if l.topics[0].hex() == '0x45c201a59ac545000ead84f30b2db67da23353aa1d58ac522c48505412143ffa':
-                    transfer_data['value'] = web3.Web3.toInt(hexstr=l.data)
-                    token_address_bytes = l.topics[2][32-20:]
-                    transfer_data['token_address'] = web3.Web3.toChecksumAddress(token_address_bytes.hex())
-                    transfer_data['from'] = rcpt.to
+            transfer_data = unpack_gift(tx.payload)
+            transfer_data['from'] = tx.inputs[0]
+            transfer_data['value'] = 0
+            transfer_data['token_address'] = ZERO_ADDRESS
+            for l in tx.logs:
+                topics = l['topics']
+                logg.debug('topixx {}'.format(topics))
+                if strip_0x(topics[0]) == '45c201a59ac545000ead84f30b2db67da23353aa1d58ac522c48505412143ffa':
+                    transfer_data['value'] = web3.Web3.toInt(hexstr=strip_0x(l['data']))
+                    #token_address_bytes = topics[2][32-20:]
+                    token_address = strip_0x(topics[2])[64-40:]
+                    transfer_data['token_address'] = to_checksum(token_address)
+
+        logg.debug('resolved method {}'.format(transfer_type))
+
+        if transfer_data != None:
+            transfer_data['status'] = tx.status
 
         return (transfer_type, transfer_data)
 
 
-    def filter(self, w3, tx, rcpt, chain_spec, session=None):
-        logg.debug('applying callback filter "{}:{}"'.format(self.queue, self.method))
-        chain_str = str(chain_spec)
-
-        transfer_data = self.parse_data(tx, rcpt)
+    def filter(self, conn, block, tx, db_session=None):
+        chain_str = str(self.chain_spec)
 
         transfer_data = None
-        if len(tx.input) < 10:
-            logg.debug('callbacks filter data length not sufficient for method signature in tx {}, skipping'.format(tx['hash']))
+        transfer_type = None
+        try:
+            (transfer_type, transfer_data) = self.parse_data(tx)
+        except TypeError:
+            logg.debug('invalid method data length for tx {}'.format(tx.hash))
             return
 
-        logg.debug('checking callbacks filter input {}'.format(tx.input[:10]))
+        if len(tx.payload) < 8:
+            logg.debug('callbacks filter data length not sufficient for method signature in tx {}, skipping'.format(tx.hash))
+            return
+
+        logg.debug('checking callbacks filter input {}'.format(tx.payload[:8]))
 
         if transfer_data != None:
+            logg.debug('wtfoo {}'.format(transfer_data))
             token_symbol = None
             result = None
             try:
-                tokentx = ExtendedTx(self.chain_spec)
+                tokentx = ExtendedTx(tx.hash, self.chain_spec)
                 tokentx.set_actors(transfer_data['from'], transfer_data['to'], self.trusted_addresses)
                 tokentx.set_tokens(transfer_data['token_address'], transfer_data['value'])
-                self.call_back(tokentx.to_dict())
+                if transfer_data['status'] == 0:
+                    tokentx.set_status(1)
+                else:
+                    tokentx.set_status(0)
+                t = self.call_back(transfer_type, tokentx.to_dict())
+                logg.info('callback success task id {} tx {}'.format(t, tx.hash))
             except UnknownContractError:
-                logg.debug('callback filter {}:{} skipping "transfer" method on unknown contract {} tx {}'.format(tc.queue, tc.method, transfer_data['to'], tx.hash.hex()))
+                logg.debug('callback filter {}:{} skipping "transfer" method on unknown contract {} tx {}'.format(tc.queue, tc.method, transfer_data['to'], tx.hash))
+
+
+    def __str__(self):
+        return 'cic-eth callbacks'

apps/cic-eth/cic_eth/runnable/daemons/filters/gas.py

@@ -1,8 +1,9 @@
 # standard imports
 import logging
 
-# third-party imports
+# external imports
 from cic_registry.chain import ChainSpec
+from hexathon import add_0x
 
 # local imports
 from cic_eth.db.enum import StatusBits
@@ -13,20 +14,19 @@ from cic_eth.queue.tx import get_paused_txs
 from cic_eth.eth.task import create_check_gas_and_send_task
 from .base import SyncFilter
 
-logg = logging.getLogger()
+logg = logging.getLogger(__name__)
 
 
 class GasFilter(SyncFilter):
 
-    def __init__(self, gas_provider, queue=None):
+    def __init__(self, chain_spec, queue=None):
         self.queue = queue
-        self.gas_provider = gas_provider
+        self.chain_spec = chain_spec
 
 
-    def filter(self, w3, tx, rcpt, chain_str, session=None):
-        logg.debug('applying gas filter')
-        tx_hash_hex = tx.hash.hex()
-        if tx['value'] > 0:
+    def filter(self, conn, block, tx, session):
+        tx_hash_hex = add_0x(tx.hash)
+        if tx.value > 0:
             logg.debug('gas refill tx {}'.format(tx_hash_hex))
             session = SessionBase.bind_session(session)
             q = session.query(TxCache.recipient)
@@ -35,23 +35,26 @@ class GasFilter(SyncFilter):
             r = q.first()
 
             if r == None:
-                logg.warning('unsolicited gas refill tx {}'.format(tx_hash_hex))
+                logg.debug('unsolicited gas refill tx {}'.format(tx_hash_hex))
                 SessionBase.release_session(session)
                 return
 
-            chain_spec = ChainSpec.from_chain_str(chain_str)
-            txs = get_paused_txs(StatusBits.GAS_ISSUES, r[0], chain_spec.chain_id(), session=session)
+            txs = get_paused_txs(StatusBits.GAS_ISSUES, r[0], self.chain_spec.chain_id(), session=session)
 
             SessionBase.release_session(session)
 
+            logg.info('resuming gas-in-waiting txs for {}'.format(r[0]))
             if len(txs) > 0:
-                logg.info('resuming gas-in-waiting txs for {}: {}'.format(r[0], txs.keys()))
                 s = create_check_gas_and_send_task(
                         list(txs.values()),
-                        str(chain_str),
+                        str(self.chain_spec),
                         r[0],
                         0,
                         tx_hashes_hex=list(txs.keys()),
                         queue=self.queue,
                 )
                 s.apply_async()
+
+
+    def __str__(self):
+        return 'eic-eth gasfilter'

apps/cic-eth/cic_eth/runnable/daemons/filters/register.py

@@ -4,36 +4,48 @@ import logging
 # third-party imports
 import celery
 from chainlib.eth.address import to_checksum
+from hexathon import (
+        add_0x,
+        strip_0x,
+        )
 
 # local imports
 from .base import SyncFilter
 
-logg = logging.getLogger()
+logg = logging.getLogger(__name__)
 
 account_registry_add_log_hash = '0x5ed3bdd47b9af629827a8d129aa39c870b10c03f0153fe9ddb8e84b665061acd' # keccak256(AccountAdded(address,uint256))
 
 
 class RegistrationFilter(SyncFilter):
 
-    def __init__(self, queue):
+    def __init__(self, chain_spec, queue):
+        self.chain_spec = chain_spec
         self.queue = queue
 
 
-    def filter(self, w3, tx, rcpt, chain_spec, session=None):
-        logg.debug('applying registration filter')
+    def filter(self, conn, block, tx, db_session=None): 
         registered_address = None
-        for l in rcpt['logs']:
-            event_topic_hex = l['topics'][0].hex()
+        logg.debug('register filter checking log {}'.format(tx.logs))
+        for l in tx.logs:
+            event_topic_hex = l['topics'][0]
             if event_topic_hex == account_registry_add_log_hash:
-                address_bytes = l.topics[1][32-20:]
-                address = to_checksum(address_bytes.hex())
+                # TODO: use abi conversion method instead
+
+                address_hex = strip_0x(l['topics'][1])[64-40:]
+                address = to_checksum(add_0x(address_hex))
                 logg.debug('request token gift to {}'.format(address))
                 s = celery.signature(
                     'cic_eth.eth.account.gift',
                     [
                         address,
-                        str(chain_spec),
+                        str(self.chain_spec),
                         ],
                     queue=self.queue,
                     )
                 s.apply_async()
+
+
+    def __str__(self):
+        return 'cic-eth account registration'
+

apps/cic-eth/cic_eth/runnable/daemons/filters/transferauth.py

@@ -0,0 +1,89 @@
+# standard imports
+import logging
+
+# external imports
+import celery
+from hexathon import (
+        strip_0x,
+        add_0x,
+        )
+from chainlib.eth.address import to_checksum
+from .base import SyncFilter
+
+
+logg = logging.getLogger(__name__)
+
+transfer_request_signature = 'ed71262a'
+
+def unpack_create_request(data):
+
+    data = strip_0x(data)
+    cursor = 0
+    f = data[cursor:cursor+8]
+    cursor += 8
+
+    if f != transfer_request_signature:
+        raise ValueError('Invalid create request data ({})'.format(f))
+
+    o = {}
+    o['sender'] = data[cursor+24:cursor+64]
+    cursor += 64
+    o['recipient'] = data[cursor+24:cursor+64]
+    cursor += 64
+    o['token'] = data[cursor+24:cursor+64]
+    cursor += 64
+    o['value'] = int(data[cursor:], 16)
+    return o
+
+
+class TransferAuthFilter(SyncFilter):
+
+    def __init__(self, registry, chain_spec, queue=None):
+        self.queue = queue
+        self.chain_spec = chain_spec
+        self.transfer_request_contract = registry.get_contract(self.chain_spec, 'TransferAuthorization')
+
+
+    def filter(self, conn, block, tx, session): #rcpt, chain_str, session=None):
+
+        if tx.payload == None:
+            logg.debug('no payload')
+            return False
+
+        payloadlength = len(tx.payload)
+        if payloadlength != 8+256:
+            logg.debug('{} below minimum length for a transfer auth call'.format(payloadlength))
+            logg.debug('payload {}'.format(tx.payload))
+            return False
+
+        recipient = tx.inputs[0]
+        if recipient != self.transfer_request_contract.address():
+            logg.debug('not our transfer auth contract address {}'.format(recipient))
+            return False
+
+        o = unpack_create_request(tx.payload) 
+           
+        sender = add_0x(to_checksum(o['sender']))
+        recipient = add_0x(to_checksum(recipient))
+        token = add_0x(to_checksum(o['token']))
+        s = celery.signature(
+            'cic_eth.eth.token.approve',
+            [
+                [
+                    {
+                        'address': token,
+                        },
+                    ],
+                sender,
+                recipient,
+                o['value'],
+                str(self.chain_spec),
+                ],
+            queue=self.queue,
+            )
+        t = s.apply_async()
+        return True
+
+
+    def __str__(self):
+        return 'cic-eth transfer auth filter'

apps/cic-eth/cic_eth/runnable/daemons/filters/tx.py

@@ -3,39 +3,47 @@ import logging
 
 # third-party imports
 import celery
+from hexathon import (
+        add_0x,
+        )
 
 # local imports
 from cic_eth.db.models.otx import Otx
-from cic_eth.db.models.base import SessionBase
+from chainsyncer.db.models.base import SessionBase
+from chainlib.status import Status
 from .base import SyncFilter
 
-logg = logging.getLogger()
+logg = logging.getLogger(__name__)
 
 
 class TxFilter(SyncFilter):
 
-    def __init__(self, queue):
+    def __init__(self, chain_spec, queue):
         self.queue = queue
+        self.chain_spec = chain_spec
 
 
-    def filter(self, w3, tx, rcpt, chain_spec, session=None):
-        session = SessionBase.bind_session(session)
-        logg.debug('applying tx filter')
-        tx_hash_hex = tx.hash.hex()
-        otx = Otx.load(tx_hash_hex, session=session)
-        SessionBase.release_session(session)
+    def filter(self, conn, block, tx, db_session=None):
+        db_session = SessionBase.bind_session(db_session)
+        tx_hash_hex = tx.hash
+        otx = Otx.load(add_0x(tx_hash_hex), session=db_session)
         if otx == None:
             logg.debug('tx {} not found locally, skipping'.format(tx_hash_hex))
             return None
-        logg.info('otx found {}'.format(otx.tx_hash))
+        logg.info('local tx match {}'.format(otx.tx_hash))
+        SessionBase.release_session(db_session)
         s = celery.signature(
                 'cic_eth.queue.tx.set_final_status',
                 [
-                    tx_hash_hex,
-                    rcpt.blockNumber,
-                    rcpt.status == 0,
+                    add_0x(tx_hash_hex),
+                    tx.block.number,
+                    tx.status == Status.ERROR,
                     ],
                 queue=self.queue,
                 )
         t = s.apply_async()
         return t
+
+
+    def __str__(self):
+        return 'cic-eth erc20 transfer filter'

apps/cic-eth/cic_eth/runnable/daemons/manager.py

@@ -1,207 +0,0 @@
-# standard imports
-import os
-import sys
-import logging
-import time
-import argparse
-import sys
-import re
-
-# third-party imports
-import confini
-import celery
-import rlp
-import web3
-from web3 import HTTPProvider, WebsocketProvider
-from cic_registry import CICRegistry
-from cic_registry.chain import ChainSpec
-from cic_registry import zero_address
-from cic_registry.chain import ChainRegistry
-from cic_registry.error import UnknownContractError
-from cic_bancor.bancor import BancorRegistryClient
-
-# local imports
-import cic_eth
-from cic_eth.eth import RpcClient
-from cic_eth.db import SessionBase
-from cic_eth.db import Otx
-from cic_eth.db import TxConvertTransfer
-from cic_eth.db.models.tx import TxCache
-from cic_eth.db.enum import StatusEnum
-from cic_eth.db import dsn_from_config
-from cic_eth.queue.tx import get_paused_txs
-from cic_eth.sync import Syncer
-from cic_eth.sync.error import LoopDone
-from cic_eth.db.error import UnknownConvertError
-from cic_eth.eth.util import unpack_signed_raw_tx
-from cic_eth.eth.task import create_check_gas_and_send_task
-from cic_eth.sync.backend import SyncerBackend
-from cic_eth.eth.token import unpack_transfer
-from cic_eth.eth.token import unpack_transferfrom
-from cic_eth.eth.account import unpack_gift
-from cic_eth.runnable.daemons.filters import (
-        CallbackFilter,
-        GasFilter,
-        TxFilter,
-        RegistrationFilter,
-        )
-
-logging.basicConfig(level=logging.WARNING)
-logg = logging.getLogger()
-logging.getLogger('websockets.protocol').setLevel(logging.CRITICAL)
-logging.getLogger('web3.RequestManager').setLevel(logging.CRITICAL)
-logging.getLogger('web3.providers.WebsocketProvider').setLevel(logging.CRITICAL)
-logging.getLogger('web3.providers.HTTPProvider').setLevel(logging.CRITICAL)
-
-
-config_dir = os.path.join('/usr/local/etc/cic-eth')
-
-argparser = argparse.ArgumentParser(description='daemon that monitors transactions in new blocks')
-argparser.add_argument('-c', type=str, default=config_dir, help='config root to use')
-argparser.add_argument('-i', '--chain-spec', type=str, dest='i', help='chain spec')
-argparser.add_argument('--abi-dir', dest='abi_dir', type=str, help='Directory containing bytecode and abi')
-argparser.add_argument('--env-prefix', default=os.environ.get('CONFINI_ENV_PREFIX'), dest='env_prefix', type=str, help='environment prefix for variables to overwrite configuration')
-argparser.add_argument('-q', type=str, default='cic-eth', help='celery queue to submit transaction tasks to')
-argparser.add_argument('-v', help='be verbose', action='store_true')
-argparser.add_argument('-vv', help='be more verbose', action='store_true')
-argparser.add_argument('mode', type=str, help='sync mode: (head|history)', default='head')
-args = argparser.parse_args(sys.argv[1:])
-
-if args.v == True:
-    logging.getLogger().setLevel(logging.INFO)
-elif args.vv == True:
-    logging.getLogger().setLevel(logging.DEBUG)
-
-config_dir = os.path.join(args.c)
-os.makedirs(config_dir, 0o777, True)
-config = confini.Config(config_dir, args.env_prefix)
-config.process()
-# override args
-args_override = {
-        'ETH_ABI_DIR': getattr(args, 'abi_dir'),
-        'CIC_CHAIN_SPEC': getattr(args, 'i'),
-        }
-config.dict_override(args_override, 'cli flag')
-config.censor('PASSWORD', 'DATABASE')
-config.censor('PASSWORD', 'SSL')
-logg.debug('config loaded from {}:\n{}'.format(config_dir, config))
-
-app = celery.Celery(backend=config.get('CELERY_RESULT_URL'),  broker=config.get('CELERY_BROKER_URL'))
-
-queue = args.q
-
-chain_spec = ChainSpec.from_chain_str(config.get('CIC_CHAIN_SPEC'))
-
-
-re_websocket = re.compile('^wss?://')
-re_http = re.compile('^https?://')
-blockchain_provider = config.get('ETH_PROVIDER')
-if re.match(re_websocket, blockchain_provider) != None:
-    blockchain_provider = WebsocketProvider(blockchain_provider)
-elif re.match(re_http, blockchain_provider) != None:
-    blockchain_provider = HTTPProvider(blockchain_provider)
-else:
-    raise ValueError('unknown provider url {}'.format(blockchain_provider))
-
-def web3_constructor():
-    w3 = web3.Web3(blockchain_provider)
-    return (blockchain_provider, w3)
-RpcClient.set_constructor(web3_constructor)
-
-c = RpcClient(chain_spec)
-CICRegistry.init(c.w3, config.get('CIC_REGISTRY_ADDRESS'), chain_spec)
-CICRegistry.add_path(config.get('ETH_ABI_DIR'))
-chain_registry = ChainRegistry(chain_spec)
-CICRegistry.add_chain_registry(chain_registry, True)
-
-declarator = CICRegistry.get_contract(chain_spec, 'AddressDeclarator', interface='Declarator')
-
-
-dsn = dsn_from_config(config)
-SessionBase.connect(dsn, pool_size=1, debug=config.true('DATABASE_DEBUG'))
-
-
-def main():
-    global chain_spec, c, queue
-
-    if config.get('ETH_ACCOUNT_ACCOUNTS_INDEX_WRITER') != None:
-        CICRegistry.add_role(chain_spec, config.get('ETH_ACCOUNT_ACCOUNTS_INDEX_WRITER'), 'AccountRegistry', True)
-
-    syncers = []
-    block_offset = c.w3.eth.blockNumber
-    chain = str(chain_spec)
-
-    if SyncerBackend.first(chain):
-        from cic_eth.sync.history import HistorySyncer
-        backend = SyncerBackend.initial(chain, block_offset)
-        syncer = HistorySyncer(backend) 
-        syncers.append(syncer)
-
-    if args.mode == 'head':
-        from cic_eth.sync.head import HeadSyncer
-        block_sync = SyncerBackend.live(chain, block_offset+1)
-        syncers.append(HeadSyncer(block_sync))
-    elif args.mode == 'history':
-        from cic_eth.sync.history import HistorySyncer
-        backends = SyncerBackend.resume(chain, block_offset+1)
-        for backend in backends:
-            syncers.append(HistorySyncer(backend))
-        if len(syncers) == 0:
-            logg.info('found no unsynced history. terminating')
-            sys.exit(0)
-    else:
-        sys.stderr.write("unknown mode '{}'\n".format(args.mode))
-        sys.exit(1)
-
-#    bancor_registry_contract = CICRegistry.get_contract(chain_spec, 'BancorRegistry', interface='Registry')
-#    bancor_chain_registry = CICRegistry.get_chain_registry(chain_spec)
-#    bancor_registry = BancorRegistryClient(c.w3, bancor_chain_registry, config.get('ETH_ABI_DIR'))
-#    bancor_registry.load() 
- 
-    trusted_addresses_src = config.get('CIC_TRUST_ADDRESS')
-    if trusted_addresses_src == None:
-        logg.critical('At least one trusted address must be declared in CIC_TRUST_ADDRESS')
-        sys.exit(1)
-    trusted_addresses = trusted_addresses_src.split(',')
-    for address in trusted_addresses:
-        logg.info('using trusted address {}'.format(address))
-    CallbackFilter.trusted_addresses = trusted_addresses
-
-    callback_filters = []
-    for cb in config.get('TASKS_TRANSFER_CALLBACKS', '').split(','):
-        task_split = cb.split(':')
-        task_queue = queue
-        if len(task_split) > 1:
-            task_queue = task_split[0]
-        callback_filter = CallbackFilter(task_split[1], task_queue)
-        callback_filters.append(callback_filter)
-
-    tx_filter = TxFilter(queue)
-
-    registration_filter = RegistrationFilter(queue)
-
-    gas_filter = GasFilter(c.gas_provider(), queue)
-
-    i = 0
-    for syncer in syncers:
-        logg.debug('running syncer index {}'.format(i))
-        syncer.filter.append(gas_filter.filter)
-        syncer.filter.append(registration_filter.filter)
-        # TODO: the two following filter functions break the filter loop if return uuid. Pro: less code executed. Con: Possibly unintuitive flow break
-        syncer.filter.append(tx_filter.filter)
-        #syncer.filter.append(convert_filter)
-        for cf in callback_filters:
-            syncer.filter.append(cf.filter)
-
-        try:
-            syncer.loop(int(config.get('SYNCER_LOOP_INTERVAL')))
-        except LoopDone as e:
-            sys.stderr.write("sync '{}' done at block {}\n".format(args.mode, e))
-
-        i += 1
-
-    sys.exit(0)
-
-
-if __name__ == '__main__':
-    main()

apps/cic-eth/cic_eth/runnable/daemons/server.py

@@ -55,62 +55,25 @@ SessionBase.connect(dsn)
 celery_app = celery.Celery(backend=config.get('CELERY_RESULT_URL'),  broker=config.get('CELERY_BROKER_URL'))
 queue = args.q
 
-re_transfer_approval_request = r'^/transferrequest/?'
+re_something = r'^/something/?'
 
 chain_spec = ChainSpec.from_chain_str(config.get('CIC_CHAIN_SPEC'))
 
 
-def process_transfer_approval_request(session, env):
-    r = re.match(re_transfer_approval_request, env.get('PATH_INFO'))
+def process_something(session, env):
+    r = re.match(re_something, env.get('PATH_INFO'))
     if not r:
         return None
 
-    if env.get('CONTENT_TYPE') != 'application/json':
-        raise AttributeError('content type')
+    #if env.get('CONTENT_TYPE') != 'application/json':
+    #    raise AttributeError('content type')
 
-    if env.get('REQUEST_METHOD') != 'POST':
-        raise AttributeError('method')
+    #if env.get('REQUEST_METHOD') != 'POST':
+    #    raise AttributeError('method')
 
-    post_data = json.load(env.get('wsgi.input'))
-    token_address = web3.Web3.toChecksumAddress(post_data['token_address'])
-    holder_address = web3.Web3.toChecksumAddress(post_data['holder_address'])
-    beneficiary_address = web3.Web3.toChecksumAddress(post_data['beneficiary_address'])
-    value = int(post_data['value'])
-
-    logg.debug('transfer approval request token {} to {} from {} value {}'.format(
-        token_address,
-        beneficiary_address,
-        holder_address,
-        value,
-        )
-        )
-
-    s = celery.signature(
-        'cic_eth.eth.request.transfer_approval_request',
-        [
-            [
-                {
-                    'address': token_address,
-                    },
-                ],
-            holder_address,
-            beneficiary_address,
-            value,
-            config.get('CIC_CHAIN_SPEC'),
-            ],
-        queue=queue,
-     )
-    t = s.apply_async()
-    r = t.get()
-    tx_raw_bytes = bytes.fromhex(r[0][2:])
-    tx = unpack_signed_raw_tx(tx_raw_bytes, chain_spec.chain_id())
-    for r in t.collect():
-        logg.debug('result {}'.format(r))
-
-    if not t.successful():
-        raise RuntimeError(tx['hash'])
-
-    return ('text/plain', tx['hash'].encode('utf-8'),)
+    #post_data = json.load(env.get('wsgi.input'))
+    
+    #return ('text/plain', 'foo'.encode('utf-8'),)
 
 
 # uwsgi application
@@ -125,7 +88,7 @@ def application(env, start_response):
 
     session = SessionBase.create_session()
     for handler in [
-            process_transfer_approval_request,
+            process_something,
             ]:
         try:
             r = handler(session, env)

apps/cic-eth/cic_eth/runnable/daemons/tasker.py

@@ -26,7 +26,6 @@ from cic_eth.eth import bancor
 from cic_eth.eth import token
 from cic_eth.eth import tx
 from cic_eth.eth import account
-from cic_eth.eth import request
 from cic_eth.admin import debug
 from cic_eth.admin import ctrl
 from cic_eth.eth.rpc import RpcClient
@@ -40,6 +39,7 @@ from cic_eth.callbacks import redis
 from cic_eth.db.models.base import SessionBase
 from cic_eth.db.models.otx import Otx
 from cic_eth.db import dsn_from_config
+from cic_eth.ext import tx
 
 logging.basicConfig(level=logging.WARNING)
 logg = logging.getLogger()

apps/cic-eth/cic_eth/runnable/daemons/tracker.py

@@ -0,0 +1,172 @@
+# standard imports
+import os
+import sys
+import logging
+import time
+import argparse
+import sys
+import re
+
+# third-party imports
+import confini
+import celery
+import rlp
+import web3
+from web3 import HTTPProvider, WebsocketProvider
+import cic_base.config
+import cic_base.log
+import cic_base.argparse
+import cic_base.rpc
+from cic_registry import CICRegistry
+from chainlib.chain import ChainSpec
+from cic_registry import zero_address
+from cic_registry.chain import ChainRegistry
+from cic_registry.error import UnknownContractError
+from chainlib.eth.connection import HTTPConnection
+from chainlib.eth.block import (
+        block_latest,
+        )
+from hexathon import (
+        strip_0x,
+        )
+from chainsyncer.backend import SyncerBackend
+from chainsyncer.driver import (
+        HeadSyncer,
+        HistorySyncer,
+        )
+from chainsyncer.db.models.base import SessionBase
+
+# local imports
+from cic_eth.registry import init_registry
+from cic_eth.eth import RpcClient
+from cic_eth.db import Otx
+from cic_eth.db import TxConvertTransfer
+from cic_eth.db.models.tx import TxCache
+from cic_eth.db.enum import StatusEnum
+from cic_eth.db import dsn_from_config
+from cic_eth.queue.tx import get_paused_txs
+#from cic_eth.sync import Syncer
+#from cic_eth.sync.error import LoopDone
+from cic_eth.db.error import UnknownConvertError
+from cic_eth.eth.util import unpack_signed_raw_tx
+from cic_eth.eth.task import create_check_gas_and_send_task
+from cic_eth.eth.token import unpack_transfer
+from cic_eth.eth.token import unpack_transferfrom
+from cic_eth.eth.account import unpack_gift
+from cic_eth.runnable.daemons.filters import (
+        CallbackFilter,
+        GasFilter,
+        TxFilter,
+        RegistrationFilter,
+        TransferAuthFilter,
+        )
+
+script_dir = os.path.realpath(os.path.dirname(__file__))
+
+logg = cic_base.log.create()
+argparser = cic_base.argparse.create(script_dir, cic_base.argparse.full_template)
+#argparser = cic_base.argparse.add(argparser, add_traffic_args, 'traffic')
+args = cic_base.argparse.parse(argparser, logg)
+config = cic_base.config.create(args.c, args, args.env_prefix)
+
+config.add(args.y, '_KEYSTORE_FILE', True)
+
+config.add(args.q, '_CELERY_QUEUE', True)
+
+cic_base.config.log(config)
+
+
+dsn = dsn_from_config(config)
+SessionBase.connect(dsn, pool_size=1, debug=config.true('DATABASE_DEBUG'))
+
+
+def main():
+    # parse chain spec object
+    chain_spec = ChainSpec.from_chain_str(config.get('CIC_CHAIN_SPEC'))
+            
+    # connect to celery
+    celery.Celery(broker=config.get('CELERY_BROKER_URL'), backend=config.get('CELERY_RESULT_URL'))
+
+    # set up registry
+    w3 = cic_base.rpc.create(config.get('ETH_PROVIDER')) # replace with HTTPConnection when registry has been so refactored
+    registry = init_registry(config, w3)
+
+    # Connect to blockchain with chainlib
+    conn = HTTPConnection(config.get('ETH_PROVIDER'))
+
+    o = block_latest()
+    r = conn.do(o)
+    block_offset = int(strip_0x(r), 16) + 1
+
+    logg.debug('starting at block {}'.format(block_offset))
+
+    syncers = []
+
+    #if SyncerBackend.first(chain_spec):
+    #    backend = SyncerBackend.initial(chain_spec, block_offset)
+    syncer_backends = SyncerBackend.resume(chain_spec, block_offset)
+
+    if len(syncer_backends) == 0:
+        logg.info('found no backends to resume')
+        syncer_backends.append(SyncerBackend.initial(chain_spec, block_offset))
+    else:
+        for syncer_backend in syncer_backends:
+            logg.info('resuming sync session {}'.format(syncer_backend))
+
+    syncer_backends.append(SyncerBackend.live(chain_spec, block_offset+1))
+
+    for syncer_backend in syncer_backends:
+        try:
+            syncers.append(HistorySyncer(syncer_backend))
+            logg.info('Initializing HISTORY syncer on backend {}'.format(syncer_backend))
+        except AttributeError:
+            logg.info('Initializing HEAD syncer on backend {}'.format(syncer_backend))
+            syncers.append(HeadSyncer(syncer_backend))
+
+    trusted_addresses_src = config.get('CIC_TRUST_ADDRESS')
+    if trusted_addresses_src == None:
+        logg.critical('At least one trusted address must be declared in CIC_TRUST_ADDRESS')
+        sys.exit(1)
+    trusted_addresses = trusted_addresses_src.split(',')
+    for address in trusted_addresses:
+        logg.info('using trusted address {}'.format(address))
+    CallbackFilter.trusted_addresses = trusted_addresses
+
+    callback_filters = []
+    for cb in config.get('TASKS_TRANSFER_CALLBACKS', '').split(','):
+        task_split = cb.split(':')
+        task_queue = config.get('_CELERY_QUEUE')
+        if len(task_split) > 1:
+            task_queue = task_split[0]
+        callback_filter = CallbackFilter(chain_spec, task_split[1], task_queue)
+        callback_filters.append(callback_filter)
+
+    tx_filter = TxFilter(chain_spec, config.get('_CELERY_QUEUE'))
+
+    registration_filter = RegistrationFilter(chain_spec, config.get('_CELERY_QUEUE'))
+
+    gas_filter = GasFilter(chain_spec, config.get('_CELERY_QUEUE'))
+
+    transfer_auth_filter = TransferAuthFilter(registry, chain_spec, config.get('_CELERY_QUEUE'))
+
+    i = 0
+    for syncer in syncers:
+        logg.debug('running syncer index {}'.format(i))
+        syncer.add_filter(gas_filter)
+        syncer.add_filter(registration_filter)
+        # TODO: the two following filter functions break the filter loop if return uuid. Pro: less code executed. Con: Possibly unintuitive flow break
+        syncer.add_filter(tx_filter)
+        syncer.add_filter(transfer_auth_filter)
+        for cf in callback_filters:
+            syncer.add_filter(cf)
+
+        r = syncer.loop(int(config.get('SYNCER_LOOP_INTERVAL')), conn)
+        sys.stderr.write("sync {} done at block {}\n".format(syncer, r))
+
+        i += 1
+
+    sys.exit(0)
+
+
+if __name__ == '__main__':
+    main()

apps/cic-eth/cic_eth/runnable/view.py

@@ -18,12 +18,16 @@ import web3
 from cic_registry import CICRegistry
 from cic_registry.chain import ChainSpec
 from cic_registry.chain import ChainRegistry
+from hexathon import add_0x
 
 # local imports
 from cic_eth.api import AdminApi
 from cic_eth.eth.rpc import RpcClient
-from cic_eth.db.enum import StatusEnum
-from cic_eth.db.enum import LockEnum
+from cic_eth.db.enum import (
+    StatusEnum,
+    status_str,
+    LockEnum,
+)
 
 logging.basicConfig(level=logging.WARNING)
 logg = logging.getLogger()
@@ -36,8 +40,8 @@ default_abi_dir = '/usr/share/local/cic/solidity/abi'
 default_config_dir = os.path.join('/usr/local/etc/cic-eth')
 
 argparser = argparse.ArgumentParser()
-argparser.add_argument('-p', '--provider', dest='p', default='http://localhost:8545', type=str, help='Web3 provider url (http only)')
-argparser.add_argument('-r', '--registry-address', type=str, help='CIC registry address')
+argparser.add_argument('-p', '--provider', dest='p', type=str, help='Web3 provider url (http only)')
+argparser.add_argument('-r', '--registry-address', dest='r', type=str, help='CIC registry address')
 argparser.add_argument('-f', '--format', dest='f', default='terminal', type=str, help='Output format')
 argparser.add_argument('-c', type=str, default=default_config_dir, help='config root to use')
 argparser.add_argument('-i', '--chain-spec', dest='i', type=str, help='chain spec')
@@ -61,12 +65,16 @@ config.process()
 args_override = {
         'ETH_PROVIDER': getattr(args, 'p'),
         'CIC_CHAIN_SPEC': getattr(args, 'i'),
+        'CIC_REGISTRY_ADDRESS': getattr(args, 'r'),
         }
 # override args
+config.dict_override(args_override, 'cli args')
 config.censor('PASSWORD', 'DATABASE')
 config.censor('PASSWORD', 'SSL')
 logg.debug('config loaded from {}:\n{}'.format(config_dir, config))
 
+config.add(add_0x(args.query), '_QUERY', True)
+
 re_websocket = re.compile('^wss?://')
 re_http = re.compile('^https?://')
 blockchain_provider = config.get('ETH_PROVIDER')
@@ -114,7 +122,7 @@ def render_tx(o, **kwargs):
     
     for v in o.get('status_log', []):
         d = datetime.datetime.fromisoformat(v[0])
-        e = StatusEnum(v[1]).name
+        e = status_str(v[1])
         content += '{}: {}\n'.format(d, e)
 
     return content
@@ -148,21 +156,20 @@ def render_lock(o, **kwargs):
 
 # TODO: move each command to submodule
 def main():
-    logg.debug('len {}'.format(len(args.query)))
     txs  = []
     renderer = render_tx
-    if len(args.query) > 66:
-        txs = [admin_api.tx(chain_spec, tx_raw=args.query)]
-    elif len(args.query) > 42:
-        txs = [admin_api.tx(chain_spec, tx_hash=args.query)]
-    elif len(args.query) == 42:
-        txs = admin_api.account(chain_spec, args.query, include_recipient=False)
+    if len(config.get('_QUERY')) > 66:
+        txs = [admin_api.tx(chain_spec, tx_raw=config.get('_QUERY'))]
+    elif len(config.get('_QUERY')) > 42:
+        txs = [admin_api.tx(chain_spec, tx_hash=config.get('_QUERY'))]
+    elif len(config.get('_QUERY')) == 42:
+        txs = admin_api.account(chain_spec, config.get('_QUERY'), include_recipient=False)
         renderer = render_account
-    elif len(args.query) >= 4 and args.query[:4] == 'lock':
+    elif len(config.get('_QUERY')) >= 4 and config.get('_QUERY')[:4] == 'lock':
         txs = admin_api.get_lock()
         renderer = render_lock
     else:
-        raise ValueError('cannot parse argument {}'.format(args.query))
+        raise ValueError('cannot parse argument {}'.format(config.get('_QUERY')))
 
     if len(txs) == 0:
         logg.info('no matches found')

apps/cic-eth/cic_eth/sync/mined.py

@@ -100,7 +100,7 @@ class MinedSyncer(Syncer):
             logg.debug('got blocks {}'.format(e))
             for block in e:
                 block_number = self.process(c.w3, block.hex())
-                logg.info('processed block {} {}'.format(block_number, block.hex()))
+                logg.debug('processed block {} {}'.format(block_number, block.hex()))
             self.bc_cache.disconnect()
             if len(e) > 0:
                 time.sleep(self.yield_delay)

apps/cic-eth/cic_eth/task.py

@@ -0,0 +1,33 @@
+# import
+import requests
+
+# external imports
+import celery
+import sqlalchemy
+
+
+class CriticalTask(celery.Task):
+    retry_jitter = True
+    retry_backoff = True
+    retry_backoff_max = 8
+
+
+class CriticalSQLAlchemyTask(CriticalTask):
+    autoretry_for = (
+        sqlalchemy.exc.DatabaseError,
+        sqlalchemy.exc.TimeoutError,
+        ) 
+
+
+class CriticalWeb3Task(CriticalTask):
+    autoretry_for = (
+        requests.exceptions.ConnectionError,
+        )
+
+
+class CriticalSQLAlchemyAndWeb3Task(CriticalTask):
+    autoretry_for = (
+        sqlalchemy.exc.DatabaseError,
+        sqlalchemy.exc.TimeoutError,
+        requests.exceptions.ConnectionError,
+        )

apps/cic-eth/cic_eth/version.py

@@ -10,7 +10,7 @@ version = (
         0,
         10,
         0,
-        'alpha.30',
+        'alpha.38',
         )
 
 version_object = semver.VersionInfo(

apps/cic-eth/config/docker/database.ini

@@ -6,4 +6,4 @@ HOST=localhost
 PORT=63432
 ENGINE=postgresql
 DRIVER=psycopg2
-DEBUG=1
+DEBUG=0

apps/cic-eth/docker/Dockerfile

@@ -16,7 +16,7 @@ ARG root_requirement_file='requirements.txt'
 #RUN apk add linux-headers
 #RUN apk add libffi-dev
 RUN apt-get update && \
-	apt install -y gcc gnupg libpq-dev wget make g++ gnupg bash procps
+	apt install -y gcc gnupg libpq-dev wget make g++ gnupg bash procps git
 
 # Copy shared requirements from top of mono-repo
 RUN echo "copying root req file ${root_requirement_file}"
@@ -42,7 +42,6 @@ COPY cic-eth/config/ /usr/local/etc/cic-eth/
 COPY cic-eth/cic_eth/db/migrations/ /usr/local/share/cic-eth/alembic/
 COPY cic-eth/crypto_dev_signer_config/ /usr/local/etc/crypto-dev-signer/
 
-RUN apt-get install -y git && \
-	git clone https://gitlab.com/grassrootseconomics/cic-contracts.git && \
+RUN git clone https://gitlab.com/grassrootseconomics/cic-contracts.git && \
 	mkdir -p /usr/local/share/cic/solidity && \
 	cp -R cic-contracts/abis /usr/local/share/cic/solidity/abi

apps/cic-eth/docker/start_manager.sh

@@ -1,5 +0,0 @@
-#!/bin/bash
-
-. ./db.sh
-
-/usr/local/bin/cic-eth-managerd $@

apps/cic-eth/docker/start_tracker.sh

@@ -0,0 +1,5 @@
+#!/bin/bash
+
+. ./db.sh
+
+/usr/local/bin/cic-eth-trackerd $@

apps/cic-eth/requirements.txt

@@ -1,15 +1,16 @@
+cic-base~=0.1.1a10
 web3==5.12.2
 celery==4.4.7
-crypto-dev-signer~=0.4.13rc2
-confini~=0.3.6b1
-cic-registry~=0.5.3a20
+crypto-dev-signer~=0.4.13rc4
+confini~=0.3.6rc3
+cic-registry~=0.5.3a22
 cic-bancor~=0.0.6
 redis==3.5.3
 alembic==1.4.2
 websockets==8.1
 requests~=2.24.0
 eth_accounts_index~=0.0.10a10
-erc20-approval-escrow~=0.3.0a5
+erc20-transfer-authorization~=0.3.0a10
 erc20-single-shot-faucet~=0.2.0a6
 rlp==2.0.1
 uWSGI==2.0.19.1
@@ -18,5 +19,6 @@ eth-gas-proxy==0.0.1a4
 websocket-client==0.57.0
 moolb~=0.1.1b2
 eth-address-index~=0.1.0a8
-chainlib~=0.0.1a16
+chainlib~=0.0.1a19
 hexathon~=0.0.1a3
+chainsyncer~=0.0.1a19

apps/cic-eth/setup.cfg

@@ -45,7 +45,7 @@ scripts =
 console_scripts =
 	# daemons
 	cic-eth-taskerd = cic_eth.runnable.daemons.tasker:main
-	cic-eth-managerd = cic_eth.runnable.daemons.manager:main
+	cic-eth-trackerd = cic_eth.runnable.daemons.tracker:main
 	cic-eth-dispatcherd = cic_eth.runnable.daemons.dispatcher:main
 	cic-eth-retrierd = cic_eth.runnable.daemons.retry:main
 	# tools	

apps/cic-eth/tests/fixtures_celery.py

@@ -13,13 +13,13 @@ def celery_includes():
     return [
         'cic_eth.eth.bancor',
         'cic_eth.eth.token',
-        'cic_eth.eth.request',
         'cic_eth.eth.tx',
         'cic_eth.ext.tx',
         'cic_eth.queue.tx',
         'cic_eth.queue.balance',
         'cic_eth.admin.ctrl',
         'cic_eth.admin.nonce',
+        'cic_eth.admin.debug',
         'cic_eth.eth.account',
         'cic_eth.callbacks.noop',
         'cic_eth.callbacks.http',

apps/cic-eth/tests/tasks/test_debug.py

@@ -0,0 +1,29 @@
+# external imports
+import celery
+
+# local imports
+from cic_eth.db.models.debug import Debug
+
+
+def test_debug_alert(
+        init_database,
+        celery_session_worker,
+        ):
+
+    s = celery.signature(
+            'cic_eth.admin.debug.alert',
+            [
+                'foo',
+                'bar',
+                'baz',
+                ],
+            queue=None,
+            )
+    t = s.apply_async()
+    r = t.get()
+    assert r == 'foo'
+
+    q = init_database.query(Debug)
+    q = q.filter(Debug.tag=='bar')
+    o = q.first()
+    assert o.description == 'baz'

apps/cic-eth/tests/tasks/test_gas_tasks.py

@@ -37,7 +37,7 @@ def test_refill_gas(
     eth_empty_accounts,
         ):
 
-    provider_address = AccountRole.get_address('GAS_GIFTER')
+    provider_address = AccountRole.get_address('GAS_GIFTER', init_database)
     receiver_address = eth_empty_accounts[0]
 
     c = init_rpc
@@ -93,7 +93,7 @@ def test_refill_deduplication(
     eth_empty_accounts,
         ):
 
-    provider_address = AccountRole.get_address('ETH_GAS_PROVIDER_ADDRESS')
+    provider_address = AccountRole.get_address('ETH_GAS_PROVIDER_ADDRESS', init_database)
     receiver_address = eth_empty_accounts[0]
 
     c = init_rpc

apps/cic-eth/tests/tasks/test_states.py

@@ -27,14 +27,14 @@ def test_states_initial(
     tx = {
         'from': init_w3.eth.accounts[0],
         'to': init_w3.eth.accounts[1],
-        'nonce': 42,
+        'nonce': 13,
         'gas': 21000,
         'gasPrice': 1000000,
         'value': 128,
-        'chainId': 666,
+        'chainId': 42,
         'data': '',
         }
-    (tx_hash_hex, tx_raw_signed_hex) = sign_and_register_tx(tx, 'Foo:666', None)
+    (tx_hash_hex, tx_raw_signed_hex) = sign_and_register_tx(tx, 'foo:bar:42', None)
 
     otx = init_database.query(Otx).filter(Otx.tx_hash==tx_hash_hex).first()
     assert otx.status == StatusEnum.PENDING.value
@@ -43,7 +43,7 @@ def test_states_initial(
             'cic_eth.eth.tx.check_gas',
             [
                 [tx_hash_hex],
-                'Foo:666',
+                'foo:bar:42',
                 [tx_raw_signed_hex],
                 init_w3.eth.accounts[0],
                 8000000,
@@ -67,7 +67,7 @@ def test_states_initial(
             'cic_eth.eth.tx.check_gas',
             [
                 [tx_hash_hex],
-                'Foo:666',
+                'foo:bar:42',
                 [tx_raw_signed_hex],
                 init_w3.eth.accounts[0],
                 8000000,
@@ -94,14 +94,14 @@ def test_states_failed(
     tx = {
         'from': init_w3.eth.accounts[0],
         'to': init_w3.eth.accounts[1],
-        'nonce': 42,
+        'nonce': 13,
         'gas': 21000,
         'gasPrice': 1000000,
         'value': 128,
-        'chainId': 666,
+        'chainId': 42,
         'data': '',
         }
-    (tx_hash_hex, tx_raw_signed_hex) = sign_and_register_tx(tx, 'Foo:666', None)
+    (tx_hash_hex, tx_raw_signed_hex) = sign_and_register_tx(tx, 'foo:bar:42', None)
 
     otx = init_database.query(Otx).filter(Otx.tx_hash==tx_hash_hex).first()
     otx.sendfail(session=init_database)
@@ -112,7 +112,7 @@ def test_states_failed(
             'cic_eth.eth.tx.check_gas',
             [
                 [tx_hash_hex],
-                'Foo:666',
+                'foo:bar:42',
                 [tx_raw_signed_hex],
                 init_w3.eth.accounts[0],
                 8000000,

apps/cic-eth/tests/tasks/test_transfer_approval.py

@@ -1,76 +0,0 @@
-# standard imports
-import logging
-import time
-
-# third-party imports
-from erc20_approval_escrow import TransferApproval
-import celery
-import sha3
-
-# local imports
-from cic_eth.eth.token import TokenTxFactory
-
-logg = logging.getLogger()
-
-
-# BUG: transaction receipt only found sometimes
-def test_transfer_approval(
-    default_chain_spec,
-    transfer_approval,
-    bancor_tokens,
-    w3_account_roles,
-    eth_empty_accounts,
-    cic_registry,
-    init_database,
-    celery_session_worker,
-    init_eth_tester,
-    init_w3,
-        ):
-
-    s = celery.signature(
-            'cic_eth.eth.request.transfer_approval_request',
-            [
-                [
-                    {
-                        'address': bancor_tokens[0],
-                        },
-                    ],
-                w3_account_roles['eth_account_sarafu_owner'],
-                eth_empty_accounts[0],
-                1024,
-                str(default_chain_spec),
-                ],
-         )
-
-    s_send = celery.signature(
-            'cic_eth.eth.tx.send',
-            [
-                str(default_chain_spec),
-                ],
-            
-            )
-    s.link(s_send)
-    t = s.apply_async()
-
-    tx_signed_raws = t.get()
-    for r in t.collect():
-        logg.debug('result {}'.format(r))
-
-    assert t.successful()
-
-    init_eth_tester.mine_block()
-
-    h = sha3.keccak_256()
-    tx_signed_raw = tx_signed_raws[0]
-    tx_signed_raw_bytes = bytes.fromhex(tx_signed_raw[2:])
-    h.update(tx_signed_raw_bytes)
-    tx_hash = h.digest()
-    rcpt = init_w3.eth.getTransactionReceipt(tx_hash)
-
-    assert rcpt.status == 1
-
-    a = TransferApproval(init_w3, transfer_approval)
-    assert a.last_serial() == 1
-
-    logg.debug('requests {}'.format(a.requests(1)['serial']))
-

apps/cic-eth/tests/unit/api/test_callback.py

@@ -67,7 +67,7 @@ def test_callback_tcp(
             logg.debug('recived {} '.format(data))
             o = json.loads(echo)
             try:
-                assert o == data
+                assert o['result'] == data
             except Exception as e:
                 self.exception = e
 
@@ -130,7 +130,7 @@ def test_callback_redis(
             o = json.loads(echo['data'])
             logg.debug('recived {} '.format(o))
             try:
-                assert o == data
+                assert o['result'] == data
             except Exception as e:
                 self.exception = e
 

apps/cic-eth/tests/unit/db/test_debug.py

@@ -0,0 +1,16 @@
+# local imports
+from cic_eth.db.models.debug import Debug
+
+
+def test_debug(
+        init_database,
+        ):
+
+    o = Debug('foo', 'bar')
+    init_database.add(o)
+    init_database.commit()
+
+    q = init_database.query(Debug)
+    q = q.filter(Debug.tag=='foo')
+    o = q.first()
+    assert o.description == 'bar'

apps/cic-eth/tests/unit/db/test_role.py

@@ -9,18 +9,18 @@ def test_db_role(
     foo = AccountRole.set('foo', eth_empty_accounts[0])
     init_database.add(foo)
     init_database.commit()
-    assert AccountRole.get_address('foo') == eth_empty_accounts[0]
+    assert AccountRole.get_address('foo', init_database) == eth_empty_accounts[0]
 
     bar = AccountRole.set('bar', eth_empty_accounts[1])
     init_database.add(bar)
     init_database.commit()
-    assert AccountRole.get_address('bar') == eth_empty_accounts[1]
+    assert AccountRole.get_address('bar', init_database) == eth_empty_accounts[1]
 
     foo = AccountRole.set('foo', eth_empty_accounts[2])
     init_database.add(foo)
     init_database.commit()
-    assert AccountRole.get_address('foo') == eth_empty_accounts[2]
-    assert AccountRole.get_address('bar') == eth_empty_accounts[1]
+    assert AccountRole.get_address('foo', init_database) == eth_empty_accounts[2]
+    assert AccountRole.get_address('bar', init_database) == eth_empty_accounts[1]
 
     tag = AccountRole.role_for(eth_empty_accounts[2])
     assert tag == 'foo'

apps/cic-eth/tests/unit/db/test_tx.py

@@ -26,7 +26,7 @@ def test_set(
                 'data': '',
                 'chainId': 1,
                 }
-        (tx_hash, tx_signed) = sign_tx(tx_def, 'Foo:1')
+        (tx_hash, tx_signed) = sign_tx(tx_def, 'foo:bar:1')
         otx = Otx(
             tx_def['nonce'],
             tx_def['from'],
@@ -82,7 +82,7 @@ def test_clone(
                 'data': '',
                 'chainId': 1,
                 }
-        (tx_hash, tx_signed) = sign_tx(tx_def, 'Foo:1')
+        (tx_hash, tx_signed) = sign_tx(tx_def, 'foo:bar:1')
         otx = Otx(
             tx_def['nonce'],
             tx_def['from'],

apps/cic-eth/tests/unit/eth/test_raw.py

@@ -14,11 +14,11 @@ def test_unpack(
         'gas': 21000,
         'gasPrice': 200000000,
         'data': '0x',
-        'chainId': 8995,
+        'chainId': 42,
             }
 
-    (tx_hash, tx_raw) = sign_tx(tx, 'Foo:8995')
+    (tx_hash, tx_raw) = sign_tx(tx, 'foo:bar:42')
 
-    tx_recovered = unpack_signed_raw_tx(bytes.fromhex(tx_raw[2:]), 8995)
+    tx_recovered = unpack_signed_raw_tx(bytes.fromhex(tx_raw[2:]), 42)
 
     assert tx_hash == tx_recovered['hash']

apps/cic-eth/tests/unit/sync/test_head.py

@@ -1,43 +0,0 @@
-# standard imports
-import logging
-
-# local imports
-from cic_eth.sync.head import HeadSyncer
-from cic_eth.sync.backend import SyncerBackend
-
-logg = logging.getLogger()
-
-
-def test_head(
-    init_rpc,
-    init_database,
-    init_eth_tester,
-    mocker,
-    eth_empty_accounts,
-        ):
-
-    #backend = SyncBackend(eth_empty_accounts[0], 'foo')
-    block_number = init_rpc.w3.eth.blockNumber
-    backend = SyncerBackend.live('foo:666', block_number)
-    syncer = HeadSyncer(backend)
-
-    #init_eth_tester.mine_block()
-    nonce = init_rpc.w3.eth.getTransactionCount(init_rpc.w3.eth.accounts[0], 'pending') 
-    logg.debug('nonce {}'.format(nonce))
-    tx = {
-        'from': init_rpc.w3.eth.accounts[0],
-        'to': eth_empty_accounts[0],
-        'value': 404,
-        'gas': 21000,
-        'gasPrice': init_rpc.w3.eth.gasPrice,
-        'nonce': nonce,
-            }
-    tx_hash_one = init_rpc.w3.eth.sendTransaction(tx)
-
-    block_number = init_rpc.w3.eth.blockNumber
-    backend.set(block_number, 0)
-    b = syncer.get(init_rpc.w3)
-
-    tx = init_rpc.w3.eth.getTransactionByBlock(b[0], 0)
-
-    assert tx.hash.hex() == tx_hash_one.hex()

apps/cic-eth/tests/unit/sync/test_history.py

@@ -1,194 +0,0 @@
-# standard imports
-import logging
-
-# third-party imports
-import pytest
-from web3.exceptions import BlockNotFound
-from cic_registry import CICRegistry
-
-# local imports
-from cic_eth.sync.history import HistorySyncer
-from cic_eth.sync.head import HeadSyncer
-#from cic_eth.sync import Syncer
-from cic_eth.db.models.otx import OtxSync
-from cic_eth.db.models.base import SessionBase
-from cic_eth.sync.backend import SyncerBackend
-
-logg = logging.getLogger()
-
-class FinishedError(Exception):
-    pass
-
-       
-class DebugFilter:
-
-    def __init__(self, address):
-        self.txs = []
-        self.monitor_to_address = address
-
-    def filter(self, w3, tx, rcpt, chain_spec):
-        logg.debug('sync filter {}'.format(tx['hash'].hex()))
-        if tx['to'] == self.monitor_to_address:
-            self.txs.append(tx)
-        # hack workaround, latest block hash not found in eth_tester for some reason
-        if len(self.txs) == 2:
-            raise FinishedError('intentionally finished on tx {}'.format(tx))
-
-
-def test_history(
-    init_rpc,
-    init_database,
-    init_eth_tester,
-    #celery_session_worker,
-    eth_empty_accounts,
-        ):
-
-    nonce = init_rpc.w3.eth.getTransactionCount(init_rpc.w3.eth.accounts[0], 'pending') 
-    logg.debug('nonce {}'.format(nonce))
-    tx = {
-        'from': init_rpc.w3.eth.accounts[0],
-        'to': eth_empty_accounts[0],
-        'value': 404,
-        'gas': 21000,
-        'gasPrice': init_rpc.w3.eth.gasPrice,
-        'nonce': nonce,
-            }
-    tx_hash_one = init_rpc.w3.eth.sendTransaction(tx)
-
-    nonce = init_rpc.w3.eth.getTransactionCount(init_rpc.w3.eth.accounts[0], 'pending')
-    logg.debug('nonce {}'.format(nonce))
-    tx = {
-        'from': init_rpc.w3.eth.accounts[1],
-        'to': eth_empty_accounts[0],
-        'value': 404,
-        'gas': 21000,
-        'gasPrice': init_rpc.w3.eth.gasPrice,
-        'nonce': nonce,
-            }
-    tx_hash_two = init_rpc.w3.eth.sendTransaction(tx)
-    init_eth_tester.mine_block()
-
-    block_number = init_rpc.w3.eth.blockNumber
-
-    live_syncer = SyncerBackend.live('foo:666', 0)
-    HeadSyncer(live_syncer)
-
-    history_syncers = SyncerBackend.resume('foo:666', block_number)
-
-    for history_syncer in history_syncers:
-        logg.info('history syncer start {} target {}'.format(history_syncer.start(), history_syncer.target()))
-
-    backend = history_syncers[0]
-
-    syncer = HistorySyncer(backend)
-    fltr = DebugFilter(eth_empty_accounts[0])
-    syncer.filter.append(fltr.filter)
-
-    logg.debug('have txs {} {}'.format(tx_hash_one.hex(), tx_hash_two.hex()))
-
-    try:
-        syncer.loop(0.1)
-    except FinishedError:
-        pass
-    except BlockNotFound as e:
-        logg.error('the last block given in loop does not seem to exist :/ {}'.format(e))
-
-    check_hashes = []
-    for h in fltr.txs:
-        check_hashes.append(h['hash'].hex())
-    assert tx_hash_one.hex() in check_hashes
-    assert tx_hash_two.hex() in check_hashes
-
-
-def test_history_multiple(
-    init_rpc,
-    init_database,
-    init_eth_tester,
-    #celery_session_worker,
-    eth_empty_accounts,
-        ):
-
-    block_number = init_rpc.w3.eth.blockNumber
-    live_syncer = SyncerBackend.live('foo:666', block_number)
-    HeadSyncer(live_syncer)
-
-    nonce = init_rpc.w3.eth.getTransactionCount(init_rpc.w3.eth.accounts[0], 'pending') 
-    logg.debug('nonce {}'.format(nonce))
-    tx = {
-        'from': init_rpc.w3.eth.accounts[0],
-        'to': eth_empty_accounts[0],
-        'value': 404,
-        'gas': 21000,
-        'gasPrice': init_rpc.w3.eth.gasPrice,
-        'nonce': nonce,
-            }
-    tx_hash_one = init_rpc.w3.eth.sendTransaction(tx)
-
-
-    init_eth_tester.mine_block()
-    block_number = init_rpc.w3.eth.blockNumber
-    history_syncers = SyncerBackend.resume('foo:666', block_number)
-    for history_syncer in history_syncers:
-        logg.info('halfway history syncer start {} target {}'.format(history_syncer.start(), history_syncer.target()))
-    live_syncer = SyncerBackend.live('foo:666', block_number)
-    HeadSyncer(live_syncer)
-
-    nonce = init_rpc.w3.eth.getTransactionCount(init_rpc.w3.eth.accounts[0], 'pending')
-    logg.debug('nonce {}'.format(nonce))
-    tx = {
-        'from': init_rpc.w3.eth.accounts[1],
-        'to': eth_empty_accounts[0],
-        'value': 404,
-        'gas': 21000,
-        'gasPrice': init_rpc.w3.eth.gasPrice,
-        'nonce': nonce,
-            }
-    tx_hash_two = init_rpc.w3.eth.sendTransaction(tx)
-
-    init_eth_tester.mine_block()
-    block_number = init_rpc.w3.eth.blockNumber
-    history_syncers = SyncerBackend.resume('foo:666', block_number)
-    live_syncer = SyncerBackend.live('foo:666', block_number)
-    HeadSyncer(live_syncer)
-
-    for history_syncer in history_syncers:
-        logg.info('history syncer start {} target {}'.format(history_syncer.start(), history_syncer.target()))
-
-    assert len(history_syncers) == 2
-
-    backend = history_syncers[0]
-    syncer = HistorySyncer(backend)
-    fltr = DebugFilter(eth_empty_accounts[0])
-    syncer.filter.append(fltr.filter)
-    try:
-        syncer.loop(0.1)
-    except FinishedError:
-        pass
-    except BlockNotFound as e:
-        logg.error('the last block given in loop does not seem to exist :/ {}'.format(e))
-
-    check_hashes = []
-    for h in fltr.txs:
-        check_hashes.append(h['hash'].hex())
-    assert tx_hash_one.hex() in check_hashes
-
-
-    backend = history_syncers[1]
-    syncer = HistorySyncer(backend)
-    fltr = DebugFilter(eth_empty_accounts[0])
-    syncer.filter.append(fltr.filter)
-    try:
-        syncer.loop(0.1)
-    except FinishedError:
-        pass
-    except BlockNotFound as e:
-        logg.error('the last block given in loop does not seem to exist :/ {}'.format(e))
-
-    check_hashes = []
-    for h in fltr.txs:
-        check_hashes.append(h['hash'].hex())
-    assert tx_hash_two.hex() in check_hashes
-
-    history_syncers = SyncerBackend.resume('foo:666', block_number)
-
-    assert len(history_syncers) == 0

apps/cic-eth/tests/unit/sync/test_syncer_scratch.py

@@ -1,79 +0,0 @@
-# third-party imports
-import pytest
-
-# local imports
-from cic_eth.db.models.sync import BlockchainSync
-from cic_eth.sync.backend import SyncerBackend
-
-
-def test_scratch(
-    init_database,
-        ):
-
-    with pytest.raises(ValueError):
-        s = SyncerBackend('Testchain:666', 13)
-
-    syncer = SyncerBackend.live('Testchain:666', 13)
-
-    s = SyncerBackend('Testchain:666', syncer.object_id)
-
-
-
-def test_live(
-    init_database,
-    ):
-
-    s = SyncerBackend.live('Testchain:666', 13)
-
-    s.connect()
-    assert s.db_object.target() == None
-    s.disconnect()
-
-    assert s.get() == (13, 0)
-
-    s.set(14, 1)
-    assert s.get() == (14, 1)
-
-
-def test_resume(
-    init_database,
-    ):
-
-    live = SyncerBackend.live('Testchain:666', 13)
-    live.set(13, 2)
-
-    resumes = SyncerBackend.resume('Testchain:666', 26)
-
-    assert len(resumes) == 1
-    resume = resumes[0]
-
-    assert resume.get() == (13, 2)
-
-    resume.set(13, 4)
-    assert resume.get() == (13, 4)
-    assert resume.start() == (13, 2)
-    assert resume.target() == 26
-
-
-def test_unsynced(
-    init_database,
-    ):
-
-    live = SyncerBackend.live('Testchain:666', 13)
-    live.set(13, 2)
-
-    resumes = SyncerBackend.resume('Testchain:666', 26)
-    live = SyncerBackend.live('Testchain:666', 26)
-    resumes[0].set(18, 12)
-
-    resumes = SyncerBackend.resume('Testchain:666', 42)
-
-    assert len(resumes) == 2
-
-    assert resumes[0].start() == (13, 2)
-    assert resumes[0].get() == (18, 12)
-    assert resumes[0].target() == 26
-
-    assert resumes[1].start() == (26, 0)
-    assert resumes[1].get() == (26, 0)
-    assert resumes[1].target() == 42

apps/cic-eth/tests/unit/util/test_unpack.py

@@ -7,24 +7,22 @@ def test_unpack(
         ):
 
     tx = {
-        'nonce': 42,
+        'nonce': 13,
         'from': init_w3_conn.eth.accounts[0],
         'to': init_w3_conn.eth.accounts[1],
         'data': '0xdeadbeef',
         'value': 1024,
         'gas': 23000,
         'gasPrice': 1422521,
-        'chainId': 1337,
+        'chainId': 42,
             }
 
-    (tx_hash, tx_signed) = sign_tx(tx, 'Foo:1337')
+    (tx_hash, tx_signed) = sign_tx(tx, 'foo:bar:42')
 
-    tx_unpacked = unpack_signed_raw_tx_hex(tx_signed, 1337)
+    tx_unpacked = unpack_signed_raw_tx_hex(tx_signed, 42)
 
     for k in tx.keys():
         assert tx[k] == tx_unpacked[k]
 
-    tx_str = tx_hex_string(tx_signed, 1337)
-    assert tx_str == 'tx nonce 42 from 0x7E5F4552091A69125d5DfCb7b8C2659029395Bdf to 0x2B5AD5c4795c026514f8317c7a215E218DcCD6cF hash 0xe5aba32b1a7255d035faccb70cd8bb92c8c4a2f6bbea3f655bc5a8b802bbaa91'
-
-
+    tx_str = tx_hex_string(tx_signed, 42)
+    assert tx_str == 'tx nonce 13 from 0x7E5F4552091A69125d5DfCb7b8C2659029395Bdf to 0x2B5AD5c4795c026514f8317c7a215E218DcCD6cF hash 0x23ba3c2b400fbddcacc77d99644bfb17ac4653a69bfa46e544801fbd841b8f1e'

apps/cic-meta/.gitignore

@@ -1,5 +1,6 @@
 node_modules
 dist
 dist-web
+dist-server
 scratch
 tests

apps/cic-meta/CHANGELOG

@@ -1,6 +1,7 @@
 * 0.0.7-pending
 	- Add immutable content
 	- Add db lock on server
+	- Add ArgPair and KeyStore to src exports
 * 0.0.6
 	- Add server build
 * 0.0.5

apps/cic-meta/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "cic-client-meta",
-	"version": "0.0.7-alpha.2",
+	"version": "0.0.7-alpha.3",
 	"description": "Signed CRDT metadata graphs for the CIC network",
 	"main": "dist/index.js",
 	"types": "dist/index.d.ts",
@@ -32,6 +32,12 @@
 		"webpack-cli": "^4.2.0"
 	},
 	"author": "Louis Holbrook <dev@holbrook.no>",
+	"contributors": [
+		{
+			"name": "Spencer Ofwiti",
+			"email": "maxspencer56@gmail.com"
+		}
+	],
 	"license": "GPL-3.0-or-later",
 	"engines": {
 		"node": "~15.3.0"

apps/cic-meta/scripts/server/server.ts

@@ -101,6 +101,18 @@ function parseDigest(url) {
 
 async function processRequest(req, res) {
 	let digest = undefined;
+	const headers = {
+		"Access-Control-Allow-Origin": "*",
+		"Access-Control-Allow-Methods": "OPTIONS, POST, GET, PUT",
+		"Access-Control-Max-Age": 2592000, // 30 days
+		"Access-Control-Allow-Headers": 'Access-Control-Allow-Origin, Content-Type, x-cic-automerge'
+	};
+
+	if (req.method === "OPTIONS") {
+		res.writeHead(200, headers);
+		res.end();
+		return;
+	}
 
 	if (!['PUT', 'GET', 'POST'].includes(req.method)) {
 		res.writeHead(405, {"Content-Type": "text/plain"});
@@ -201,6 +213,7 @@ async function processRequest(req, res) {
 
 		const responseContentLength = (new TextEncoder().encode(content)).length;
 		res.writeHead(200, {
+			"Access-Control-Allow-Origin": "*",
 			"Content-Type": contentType,
 			"Content-Length": responseContentLength,
 		});

apps/cic-meta/src/index.ts

@@ -1,5 +1,5 @@
-export { PGPSigner, PGPKeyStore } from './auth';
-export { Envelope, Syncable } from './sync';
+export { PGPSigner, PGPKeyStore, Signer, KeyStore } from './auth';
+export { ArgPair,  Envelope, Syncable } from './sync';
 export { User } from './assets/user';
 export { Phone } from './assets/phone';
 export { Config } from './config';

apps/cic-meta/src/sync.ts

@@ -93,7 +93,7 @@ function orderDict(src) {
 	return dst;
 }
 
-class Syncable implements JSONSerializable, Authoritative {
+class Syncable implements JSONSerializable, Authoritative, Signable {
 
 	id:		string
 	timestamp:	number

apps/cic-ussd/.config/app.ini

@@ -7,10 +7,8 @@ PASSWORD_PEPPER=QYbzKff6NhiQzY3ygl2BkiKOpER8RE/Upqs/5aZWW+I=
 SERVICE_CODE=*483*46#
 
 [ussd]
-MENU_FILE=/usr/local/lib/python3.8/site-packages/cic_ussd/db/ussd_menu.json
+MENU_FILE=/usr/src/data/ussd_menu.json
 
 [statemachine]
 STATES=/usr/src/cic-ussd/states/
 TRANSITIONS=/usr/src/cic-ussd/transitions/
-
-

apps/cic-ussd/.config/cic.ini

@@ -1,2 +1,5 @@
 [cic]
-chain_spec = Bloxberg:8995
+engine = evm
+common_name = bloxberg
+network_id = 8996
+meta_url = http://localhost:63380

apps/cic-ussd/.config/database.ini

@@ -1,7 +1,7 @@
 [database]
 NAME=cic_ussd
 USER=postgres
-PASSWORD=password
+PASSWORD=
 HOST=localhost
 PORT=5432
 ENGINE=postgresql

apps/cic-ussd/.config/pgp.ini

@@ -0,0 +1,5 @@
+[pgp]
+export_dir = /usr/src/pgp/keys/
+keys_path = /usr/src/secrets/
+private_keys = privatekeys_meta.asc
+passphrase =

apps/cic-ussd/.config/test/app.ini

@@ -7,8 +7,8 @@ PASSWORD_PEPPER=QYbzKff6NhiQzY3ygl2BkiKOpER8RE/Upqs/5aZWW+I=
 SERVICE_CODE=*483*46#
 
 [ussd]
-MENU_FILE=cic_ussd/db/ussd_menu.json
+MENU_FILE=/usr/local/lib/python3.8/site-packages/cic_ussd/db/ussd_menu.json
 
 [statemachine]
-STATES=states/
-TRANSITIONS=transitions/
+STATES=/usr/src/cic-ussd/states/
+TRANSITIONS=/usr/src/cic-ussd/transitions/

apps/cic-ussd/.config/test/cic.ini

@@ -1,2 +1,5 @@
 [cic]
-chain_spec = Bloxberg:8995
+engine = evm
+common_name = bloxberg
+network_id = 8996
+meta_url = http://localhost:63380

apps/cic-ussd/.config/test/pgp.ini

@@ -0,0 +1,5 @@
+[pgp]
+export_dir = /usr/src/pgp/keys/
+keys_path = /usr/src/secrets/
+private_keys = privatekeys_meta.asc
+passphrase =

apps/cic-ussd/cic_ussd/account.py

@@ -0,0 +1,49 @@
+# standard imports
+import json
+
+# third-party imports
+from cic_eth.api import Api
+from cic_types.models.person import Person
+from cic_types.processor import generate_metadata_pointer
+
+# local imports
+from cic_ussd.chain import Chain
+from cic_ussd.db.models.user import User
+from cic_ussd.metadata import blockchain_address_to_metadata_pointer
+from cic_ussd.redis import get_cached_data
+
+
+def define_account_tx_metadata(user: User):
+    # get sender metadata
+    identifier = blockchain_address_to_metadata_pointer(
+        blockchain_address=user.blockchain_address
+    )
+    key = generate_metadata_pointer(
+        identifier=identifier,
+        cic_type='cic.person'
+    )
+    account_metadata = get_cached_data(key=key)
+
+    if account_metadata:
+        account_metadata = json.loads(account_metadata)
+        person = Person()
+        deserialized_person = person.deserialize(metadata=account_metadata)
+        given_name = deserialized_person.given_name
+        family_name = deserialized_person.family_name
+        phone_number = deserialized_person.tel
+
+        return f'{given_name} {family_name} {phone_number}'
+    else:
+        phone_number = user.phone_number
+        return phone_number
+
+
+def retrieve_account_statement(blockchain_address: str):
+    chain_str = Chain.spec.__str__()
+    cic_eth_api = Api(
+        chain_str=chain_str,
+        callback_queue='cic-ussd',
+        callback_task='cic_ussd.tasks.callback_handler.process_statement_callback',
+        callback_param=blockchain_address
+    )
+    result = cic_eth_api.list(address=blockchain_address, limit=9)

apps/cic-ussd/cic_ussd/accounts.py

@@ -1,39 +0,0 @@
-# standard imports
-import logging
-from collections import deque
-
-# third-party imports
-from cic_eth.api import Api
-
-# local imports
-from cic_ussd.transactions import from_wei
-
-logg = logging.getLogger()
-
-
-class BalanceManager:
-
-    def __init__(self, address: str, chain_str: str, token_symbol: str):
-        """
-        :param address: Ethereum address of account whose balance is being queried
-        :type address: str, 0x-hex
-        :param chain_str: The chain name and network id.
-        :type chain_str: str
-        :param token_symbol: ERC20 token symbol of whose balance is being queried
-        :type token_symbol: str
-        """
-        self.address = address
-        self.chain_str = chain_str
-        self.token_symbol = token_symbol
-
-    def get_operational_balance(self) -> float:
-        """This question queries cic-eth for an account's balance
-        :return: The current balance of the account as reflected on the blockchain.
-        :rtype: int
-        """
-        cic_eth_api = Api(chain_str=self.chain_str, callback_task=None)
-        balance_request_task = cic_eth_api.balance(address=self.address, token_symbol=self.token_symbol)
-        balance_request_task_results = balance_request_task.collect()
-        balance_result = deque(balance_request_task_results, maxlen=1).pop()
-        balance = from_wei(value=balance_result[-1])
-        return balance

apps/cic-ussd/cic_ussd/balance.py

@@ -0,0 +1,90 @@
+# standard imports
+import json
+import logging
+from typing import Union
+
+# third-party imports
+import celery
+from cic_eth.api import Api
+
+# local imports
+from cic_ussd.error import CachedDataNotFoundError
+from cic_ussd.redis import create_cached_data_key, get_cached_data
+from cic_ussd.conversions import from_wei
+
+logg = logging.getLogger()
+
+
+class BalanceManager:
+
+    def __init__(self, address: str, chain_str: str, token_symbol: str):
+        """
+        :param address: Ethereum address of account whose balance is being queried
+        :type address: str, 0x-hex
+        :param chain_str: The chain name and network id.
+        :type chain_str: str
+        :param token_symbol: ERC20 token symbol of whose balance is being queried
+        :type token_symbol: str
+        """
+        self.address = address
+        self.chain_str = chain_str
+        self.token_symbol = token_symbol
+
+    def get_balances(self, asynchronous: bool = False) -> Union[celery.Task, dict]:
+        """
+        This function queries cic-eth for an account's balances, It provides a means to receive the balance either
+        asynchronously or synchronously depending on the provided value for teh asynchronous parameter. It returns a
+        dictionary containing network, outgoing and incoming balances.
+        :param asynchronous: Boolean value checking whether to return balances asynchronously
+        :type asynchronous: bool
+        :return:
+        :rtype:
+        """
+        if asynchronous:
+            cic_eth_api = Api(
+                chain_str=self.chain_str,
+                callback_queue='cic-ussd',
+                callback_task='cic_ussd.tasks.callback_handler.process_balances_callback',
+                callback_param=''
+            )
+            cic_eth_api.balance(address=self.address, token_symbol=self.token_symbol)
+        else:
+            cic_eth_api = Api(chain_str=self.chain_str)
+            balance_request_task = cic_eth_api.balance(
+                address=self.address,
+                token_symbol=self.token_symbol)
+            return balance_request_task.get()[0]
+
+
+def compute_operational_balance(balances: dict) -> float:
+    """This function calculates the right balance given incoming and outgoing
+    :param balances:
+    :type balances:
+    :return:
+    :rtype:
+    """
+    incoming_balance = balances.get('balance_incoming')
+    outgoing_balance = balances.get('balance_outgoing')
+    network_balance = balances.get('balance_network')
+
+    operational_balance = (network_balance + incoming_balance) - outgoing_balance
+    return from_wei(value=operational_balance)
+
+
+def get_cached_operational_balance(blockchain_address: str):
+    """
+    :param blockchain_address:
+    :type blockchain_address:
+    :return:
+    :rtype:
+    """
+    key = create_cached_data_key(
+        identifier=bytes.fromhex(blockchain_address[2:]),
+        salt='cic.balances_data'
+    )
+    cached_balance = get_cached_data(key=key)
+    if cached_balance:
+        operational_balance = compute_operational_balance(balances=json.loads(cached_balance))
+        return operational_balance
+    else:
+        raise CachedDataNotFoundError('Cached operational balance not found.')

apps/cic-ussd/cic_ussd/chain.py

@@ -0,0 +1,10 @@
+# local imports
+
+# third-party imports
+from chainlib.chain import ChainSpec
+
+# local imports
+
+
+class Chain:
+    spec: ChainSpec = None

apps/cic-ussd/cic_ussd/conversions.py

@@ -0,0 +1,41 @@
+# standard imports
+import decimal
+
+# third-party imports
+
+# local imports
+
+
+def truncate(value: float, decimals: int):
+    """This function truncates a value to a specified number of decimals places.
+    :param value: The value to be truncated.
+    :type value: float
+    :param decimals: The number of decimals for the value to be truncated to
+    :type decimals: int
+    :return: The truncated value.
+    :rtype: int
+    """
+    decimal.getcontext().rounding = decimal.ROUND_DOWN
+    contextualized_value = decimal.Decimal(value)
+    return round(contextualized_value, decimals)
+
+
+def from_wei(value: int) -> float:
+    """This function converts values in Wei to a token in the cic network.
+    :param value: Value in Wei
+    :type value: int
+    :return: SRF equivalent of value in Wei
+    :rtype: float
+    """
+    value = float(value) / 1e+6
+    return truncate(value=value, decimals=2)
+
+
+def to_wei(value: int) -> int:
+    """This functions converts values from a token in the cic network to Wei.
+    :param value: Value in SRF
+    :type value: int
+    :return: Wei equivalent of value in SRF
+    :rtype: int
+    """
+    return int(value * 1e+6)

apps/cic-ussd/cic_ussd/db/ussd_menu.json

@@ -1,213 +1,237 @@
 {
     "ussd_menu": {
         "1": {
-            "description": "The self signup process has been initiated and the account is being created",
-            "display_key": "ussd.kenya.account_creation_prompt",
-            "name": "account_creation_prompt",
-            "parent": null
-        },
-        "2": {
-            "description": "Start menu. This is the entry point for users to select their preferred language",
+            "description": "Entry point for users to select their preferred language.",
             "display_key": "ussd.kenya.initial_language_selection",
             "name": "initial_language_selection",
             "parent": null
         },
-        "3": {
-            "description": "PIN setup entry menu",
+        "2": {
+            "description": "Entry point for users to enter a pin to secure their account.",
             "display_key": "ussd.kenya.initial_pin_entry",
             "name": "initial_pin_entry",
-            "parent": "initial_language_selection"
+            "parent": null
         },
-        "4": {
-            "description": "Confirm new PIN menu",
+        "3": {
+            "description": "Pin confirmation entry menu.",
             "display_key": "ussd.kenya.initial_pin_confirmation",
             "name": "initial_pin_confirmation",
             "parent": "initial_pin_entry"
         },
+        "4": {
+            "description": "The signup process has been initiated and the account is being created.",
+            "display_key": "ussd.kenya.account_creation_prompt",
+            "name": "account_creation_prompt",
+            "parent": null
+        },
         "5": {
-            "description": "Start menu. This is the entry point for activated users",
+            "description": "Entry point for activated users.",
             "display_key": "ussd.kenya.start",
             "name": "start",
             "parent": null
         },
         "6": {
-            "description": "Send Token recipient entry",
+            "description": "Given name entry menu.",
+            "display_key": "ussd.kenya.enter_given_name",
+            "name": "enter_given_name",
+            "parent": "metadata_management"
+        },
+        "7": {
+            "description": "Family name entry menu.",
+            "display_key": "ussd.kenya.enter_family_name",
+            "name": "enter_family_name",
+            "parent": "metadata_management"
+        },
+        "8": {
+            "description": "Gender entry menu.",
+            "display_key": "ussd.kenya.enter_gender",
+            "name": "enter_gender",
+            "parent": "metadata_management"
+        },
+        "9": {
+            "description": "Age entry menu.",
+            "display_key": "ussd.kenya.enter_gender",
+            "name": "enter_gender",
+            "parent": "metadata_management"
+        },
+        "10": {
+            "description": "Location entry menu.",
+            "display_key": "ussd.kenya.enter_location",
+            "name": "enter_location",
+            "parent": "metadata_management"
+        },
+        "11": {
+            "description": "Products entry menu.",
+            "display_key": "ussd.kenya.enter_products",
+            "name": "enter_products",
+            "parent": "metadata_management"
+        },
+        "12": {
+            "description": "Entry point for activated users.",
+            "display_key": "ussd.kenya.start",
+            "name": "start",
+            "parent": null
+        },
+        "13": {
+            "description": "Send Token recipient entry.",
             "display_key": "ussd.kenya.enter_transaction_recipient",
             "name": "enter_transaction_recipient",
             "parent": "start"
         },
-        "7": {
-            "description": "Send Token amount prompt menu",
+        "14": {
+            "description": "Send Token amount prompt menu.",
             "display_key": "ussd.kenya.enter_transaction_amount",
             "name": "enter_transaction_amount",
             "parent": "start"
         },
-        "8": {
-            "description": "PIN entry for authorization to send token",
+        "15": {
+            "description": "Pin entry for authorization to send token.",
             "display_key": "ussd.kenya.transaction_pin_authorization",
             "name": "transaction_pin_authorization",
             "parent": "start"
         },
-        "9": {
-            "description": "Terminal of a menu flow where an SMS is expected after.",
-            "display_key": "ussd.kenya.complete",
-            "name": "complete",
-            "parent": null
-        },
-        "10": {
-            "description": "Help menu",
-            "display_key": "ussd.kenya.help",
-            "name": "help",
+        "16": {
+            "description": "Manage account menu.",
+            "display_key": "ussd.kenya.account_management",
+            "name": "account_management",
             "parent": "start"
         },
-        "11": {
-            "description": "Manage account menu",
-            "display_key": "ussd.kenya.profile_management",
-            "name": "profile_management",
+        "17": {
+            "description": "Manage metadata menu.",
+            "display_key": "ussd.kenya.metadata_management",
+            "name": "metadata_management",
             "parent": "start"
         },
-        "12": {
-            "description": "Manage business directory info",
+        "18": {
+            "description": "Manage user's preferred language menu.",
             "display_key": "ussd.kenya.select_preferred_language",
             "name": "select_preferred_language",
             "parent": "account_management"
         },
-        "13": {
-            "description": "About business directory info",
+        "19": {
+            "description": "Retrieve mini-statement menu.",
             "display_key": "ussd.kenya.mini_statement_pin_authorization",
             "name": "mini_statement_pin_authorization",
             "parent": "account_management"
         },
-        "14": {
-            "description": "Change business directory info",
+        "20": {
+            "description": "Manage user's pin menu.",
             "display_key": "ussd.kenya.enter_current_pin",
             "name": "enter_current_pin",
             "parent": "account_management"
         },
-        "15": {
-            "description": "New PIN entry menu",
+        "21": {
+            "description": "New pin entry menu.",
             "display_key": "ussd.kenya.enter_new_pin",
             "name": "enter_new_pin",
             "parent": "account_management"
         },
-        "16": {
-            "description": "First name entry menu",
-            "display_key": "ussd.kenya.enter_first_name",
-            "name": "enter_first_name",
-            "parent": "profile_management"
-        },
-        "17": {
-            "description": "Last name entry menu",
-            "display_key": "ussd.kenya.enter_last_name",
-            "name": "enter_last_name",
-            "parent": "profile_management"
-        },
-        "18": {
-            "description": "Gender entry menu",
-            "display_key": "ussd.kenya.enter_gender",
-            "name": "enter_gender",
-            "parent": "profile_management"
-        },
-        "19": {
-            "description": "Location entry menu",
-            "display_key": "ussd.kenya.enter_location",
-            "name": "enter_location",
-            "parent": "profile_management"
-        },
-        "20": {
-            "description": "Business profile entry menu",
-            "display_key": "ussd.kenya.enter_business_profile",
-            "name": "enter_business_profile",
-            "parent": "profile_management"
-        },
-        "21": {
-            "description": "Menu to display a user's entire profile",
-            "display_key": "ussd.kenya.display_user_profile_data",
-            "name": "display_user_profile_data",
-            "parent": "profile_management"
-        },
         "22": {
-            "description": "Pin authorization to change name",
-            "display_key": "ussd.kenya.name_management_pin_authorization",
-            "name": "name_management_pin_authorization",
-            "parent": "profile_management"
+            "description": "Pin entry menu.",
+            "display_key": "ussd.kenya.standard_pin_authorization",
+            "name": "standard_pin_authorization",
+            "parent": "start"
         },
         "23": {
-            "description": "Pin authorization to change gender",
-            "display_key": "ussd.kenya.gender_management_pin_authorization",
-            "name": "gender_management_pin_authorization",
-            "parent": "profile_management"
-        },
-        "24": {
-            "description": "Pin authorization to change location",
-            "display_key": "ussd.kenya.location_management_pin_authorization",
-            "name": "location_management_pin_authorization",
-            "parent": "profile_management"
-        },
-        "26": {
-            "description": "Pin authorization to display user's profile",
-            "display_key": "ussd.kenya.view_profile_pin_authorization",
-            "name": "view_profile_pin_authorization",
-            "parent": "profile_management"
-        },
-        "27": {
-            "description": "Exit menu",
+            "description": "Exit menu.",
             "display_key": "ussd.kenya.exit",
             "name": "exit",
             "parent": null
         },
-        "28": {
-            "description": "Invalid menu option",
+        "24": {
+            "description": "Invalid menu option.",
             "display_key": "ussd.kenya.exit_invalid_menu_option",
             "name": "exit_invalid_menu_option",
             "parent": null
         },
-        "29": {
-            "description": "PIN policy violation",
+        "25": {
+            "description": "Pin policy violation.",
             "display_key": "ussd.kenya.exit_invalid_pin",
             "name": "exit_invalid_pin",
             "parent": null
         },
-        "30": {
-            "description": "PIN mismatch. New PIN and the new PIN confirmation do not match",
+        "26": {
+            "description": "Pin mismatch. New pin and the new pin confirmation do not match",
             "display_key": "ussd.kenya.exit_pin_mismatch",
             "name": "exit_pin_mismatch",
             "parent": null
         },
-        "31": {
-            "description": "Ussd PIN Blocked Menu",
+        "27": {
+            "description": "Ussd pin blocked Menu",
             "display_key": "ussd.kenya.exit_pin_blocked",
             "name": "exit_pin_blocked",
             "parent": null
         },
-        "32": {
-            "description": "Key params missing in request",
+        "28": {
+            "description": "Key params missing in request.",
             "display_key": "ussd.kenya.exit_invalid_request",
             "name": "exit_invalid_request",
             "parent": null
         },
-        "33": {
-            "description": "The user did not select a choice",
+        "29": {
+            "description": "The user did not select a choice.",
             "display_key": "ussd.kenya.exit_invalid_input",
             "name": "exit_invalid_input",
             "parent": null
         },
-        "34": {
+        "30": {
+            "description": "Exit following unsuccessful transaction due to insufficient account balance.",
+            "display_key": "ussd.kenya.exit_insufficient_balance",
+            "name": "exit_insufficient_balance",
+            "parent": null
+        },
+        "31": {
             "description": "Exit following a successful transaction.",
             "display_key": "ussd.kenya.exit_successful_transaction",
             "name": "exit_successful_transaction",
             "parent": null
         },
+        "32": {
+            "description": "End of a menu flow.",
+            "display_key": "ussd.kenya.complete",
+            "name": "complete",
+            "parent": null
+        },
+        "33": {
+            "description": "Pin entry menu to view account balances.",
+            "display_key": "ussd.kenya.account_balances_pin_authorization",
+            "name": "account_balances_pin_authorization",
+            "parent": "account_management"
+        },
+        "34": {
+            "description": "Pin entry menu to view account statement.",
+            "display_key": "ussd.kenya.account_statement_pin_authorization",
+            "name": "account_statement_pin_authorization",
+            "parent": "account_management"
+        },
         "35": {
-            "description": "Manage account menu",
-            "display_key": "ussd.kenya.account_management",
-            "name": "account_management",
-            "parent": "start"
+            "description": "Menu to display account balances.",
+            "display_key": "ussd.kenya.account_balances",
+            "name": "account_balances",
+            "parent": "account_management"
         },
         "36": {
-            "description": "Exit following insufficient balance to perform a transaction.",
-            "display_key": "ussd.kenya.exit_insufficient_balance",
-            "name": "exit_insufficient_balance",
+            "description": "Menu to display first set of transactions in statement.",
+            "display_key": "ussd.kenya.first_transaction_set",
+            "name": "first_transaction_set",
+            "parent": null
+        },
+        "37": {
+            "description": "Menu to display middle set of transactions in statement.",
+            "display_key": "ussd.kenya.middle_transaction_set",
+            "name": "middle_transaction_set",
+            "parent": null
+        },
+        "38": {
+            "description": "Menu to display last set of transactions in statement.",
+            "display_key": "ussd.kenya.last_transaction_set",
+            "name": "last_transaction_set",
+            "parent": null
+        },
+        "39": {
+            "description": "Menu to instruct users to call the office.",
+            "display_key": "ussd.key.help",
+            "name": "help",
             "parent": null
         }
     }

apps/cic-ussd/cic_ussd/error.py

@@ -17,3 +17,17 @@ class ActionDataNotFoundError(OSError):
     """Raised when action data matching a specific task uuid is not found in the redis cache"""
     pass
 
+
+class UserMetadataNotFoundError(OSError):
+    """Raised when metadata is expected but not available in cache."""
+    pass
+
+
+class UnsupportedMethodError(OSError):
+    """Raised when the method passed to the make request function is unsupported."""
+    pass
+
+
+class CachedDataNotFoundError(OSError):
+    """Raised when the method passed to the make request function is unsupported."""
+    pass

apps/cic-ussd/cic_ussd/metadata/__init__.py

@@ -0,0 +1,43 @@
+# standard imports
+
+# third-party imports
+import requests
+from chainlib.eth.address import to_checksum
+from hexathon import add_0x
+
+# local imports
+from cic_ussd.error import UnsupportedMethodError
+
+
+def make_request(method: str, url: str, data: any = None, headers: dict = None):
+    """
+    :param method:
+    :type method:
+    :param url:
+    :type url:
+    :param data:
+    :type data:
+    :param headers:
+    :type headers:
+    :return:
+    :rtype:
+    """
+    if method == 'GET':
+        result = requests.get(url=url)
+    elif method == 'POST':
+        result = requests.post(url=url, data=data, headers=headers)
+    elif method == 'PUT':
+        result = requests.put(url=url, data=data, headers=headers)
+    else:
+        raise UnsupportedMethodError(f'Unsupported method: {method}')
+    return result
+
+
+def blockchain_address_to_metadata_pointer(blockchain_address: str):
+    """
+    :param blockchain_address:
+    :type blockchain_address:
+    :return:
+    :rtype:
+    """
+    return bytes.fromhex(blockchain_address[2:])

apps/cic-ussd/cic_ussd/metadata/signer.py

@@ -0,0 +1,63 @@
+# standard imports
+import json
+import logging
+from typing import Optional
+from urllib.request import Request, urlopen
+
+# third-party imports
+import gnupg
+
+# local imports
+
+logg = logging.getLogger()
+
+
+class Signer:
+    """
+    :cvar gpg_path:
+    :type gpg_path:
+    :cvar gpg_passphrase:
+    :type gpg_passphrase:
+    :cvar key_file_path:
+    :type key_file_path:
+
+    """
+    gpg_path: str = None
+    gpg_passphrase: str = None
+    key_file_path: str = None
+
+    def __init__(self):
+        self.gpg = gnupg.GPG(gnupghome=self.gpg_path)
+
+        # parse key file data
+        key_file = open(self.key_file_path, 'r')
+        self.key_data = key_file.read()
+        key_file.close()
+
+    def get_operational_key(self):
+        """
+        :return:
+        :rtype:
+        """
+        # import key data into keyring
+        self.gpg.import_keys(key_data=self.key_data)
+        gpg_keys = self.gpg.list_keys()
+        key_algorithm = gpg_keys[0].get('algo')
+        key_id = gpg_keys[0].get("keyid")
+        logg.info(f'using signing key: {key_id}, algorithm: {key_algorithm}')
+        return gpg_keys[0]
+
+    def sign_digest(self, data: bytes):
+        """
+        :param data:
+        :type data:
+        :return:
+        :rtype:
+        """
+        data = json.loads(data)
+        digest = data['digest']
+        key_id = self.get_operational_key().get('keyid')
+        signature = self.gpg.sign(digest, passphrase=self.gpg_passphrase, keyid=key_id)
+        return str(signature)
+
+

apps/cic-ussd/cic_ussd/metadata/user.py

@@ -0,0 +1,102 @@
+# standard imports
+import json
+import logging
+import os
+
+# third-party imports
+import requests
+from cic_types.models.person import generate_metadata_pointer, Person
+
+# local imports
+from cic_ussd.chain import Chain
+from cic_ussd.metadata import make_request
+from cic_ussd.metadata.signer import Signer
+from cic_ussd.redis import cache_data
+
+logg = logging.getLogger()
+
+
+class UserMetadata:
+    """
+    :cvar base_url:
+    :type base_url:
+    """
+    base_url = None
+
+    def __init__(self, identifier: bytes):
+        """
+        :param identifier:
+        :type identifier:
+        """
+        self. headers = {
+            'X-CIC-AUTOMERGE': 'server',
+            'Content-Type': 'application/json'
+        }
+        self.identifier = identifier
+        self.metadata_pointer = generate_metadata_pointer(
+                identifier=self.identifier,
+                cic_type='cic.person'
+        )
+        if self.base_url:
+            self.url = os.path.join(self.base_url, self.metadata_pointer)
+
+    def create(self, data: dict):
+        try:
+            data = json.dumps(data).encode('utf-8')
+            result = make_request(method='POST', url=self.url, data=data, headers=self.headers)
+            metadata = result.content
+            self.edit(data=metadata, engine='pgp')
+            logg.info(f'Get sign material response status: {result.status_code}')
+            result.raise_for_status()
+        except requests.exceptions.HTTPError as error:
+            raise RuntimeError(error)
+
+    def edit(self, data: bytes, engine: str):
+        """
+        :param data:
+        :type data:
+        :param engine:
+        :type engine:
+        :return:
+        :rtype:
+        """
+        cic_meta_signer = Signer()
+        signature = cic_meta_signer.sign_digest(data=data)
+        algorithm = cic_meta_signer.get_operational_key().get('algo')
+        formatted_data = {
+            'm': data.decode('utf-8'),
+            's': {
+                'engine': engine,
+                'algo': algorithm,
+                'data': signature,
+                'digest': json.loads(data).get('digest'),
+            }
+        }
+        formatted_data = json.dumps(formatted_data).encode('utf-8')
+
+        try:
+            result = make_request(method='PUT', url=self.url, data=formatted_data, headers=self.headers)
+            logg.info(f'Signed content submission status: {result.status_code}.')
+            result.raise_for_status()
+        except requests.exceptions.HTTPError as error:
+            raise RuntimeError(error)
+
+    def query(self):
+        result = make_request(method='GET', url=self.url)
+        status = result.status_code
+        logg.info(f'Get latest data status: {status}')
+        try:
+            if status == 200:
+                response_data = result.content
+                data = json.loads(response_data.decode())
+
+                # validate data
+                person = Person()
+                deserialized_person = person.deserialize(metadata=json.loads(data))
+
+                cache_data(key=self.metadata_pointer, data=json.dumps(deserialized_person.serialize()))
+            elif status == 404:
+                logg.info('The data is not available and might need to be added.')
+            result.raise_for_status()
+        except requests.exceptions.HTTPError as error:
+            raise RuntimeError(error)

apps/cic-ussd/cic_ussd/operations.py

@@ -5,7 +5,6 @@ import logging
 # third party imports
 import celery
 import i18n
-import phonenumbers
 from cic_eth.api.api_task import Api
 from tinydb.table import Document
 from typing import Optional
@@ -239,7 +238,7 @@ def persist_session_to_db_task(external_session_id: str, queue: str):
     :type queue: str
     """
     s_persist_session_to_db = celery.signature(
-        'cic_ussd.tasks.ussd.persist_session_to_db',
+        'cic_ussd.tasks.ussd_session.persist_session_to_db',
         [external_session_id]
     )
     s_persist_session_to_db.apply_async(queue=queue)
@@ -453,37 +452,3 @@ def save_to_in_memory_ussd_session_data(queue: str, session_data: dict, ussd_ses
     )
     persist_session_to_db_task(external_session_id=external_session_id, queue=queue)
 
-
-def process_phone_number(phone_number: str, region: str):
-    """This function parses any phone number for the provided region
-    :param phone_number: A string with a phone number.
-    :type phone_number: str
-    :param region: Caller defined region
-    :type region: str
-    :return: The parsed phone number value based on the defined region
-    :rtype: str
-    """
-    if not isinstance(phone_number, str):
-        try:
-            phone_number = str(int(phone_number))
-
-        except ValueError:
-            pass
-
-    phone_number_object = phonenumbers.parse(phone_number, region)
-    parsed_phone_number = phonenumbers.format_number(phone_number_object, phonenumbers.PhoneNumberFormat.E164)
-
-    return parsed_phone_number
-
-
-def get_user_by_phone_number(phone_number: str) -> Optional[User]:
-    """This function queries the database for a user based on the provided phone number.
-    :param phone_number: A valid phone number.
-    :type phone_number: str
-    :return: A user object matching a given phone number
-    :rtype: User|None
-    """
-    # consider adding region to user's metadata
-    phone_number = process_phone_number(phone_number=phone_number, region='KE')
-    user = User.session.query(User).filter_by(phone_number=phone_number).first()
-    return user

apps/cic-ussd/cic_ussd/phone_number.py

@@ -0,0 +1,43 @@
+# standard imports
+from typing import Optional
+
+# third-party imports
+import phonenumbers
+
+# local imports
+from cic_ussd.db.models.user import User
+
+
+def process_phone_number(phone_number: str, region: str):
+    """This function parses any phone number for the provided region
+    :param phone_number: A string with a phone number.
+    :type phone_number: str
+    :param region: Caller defined region
+    :type region: str
+    :return: The parsed phone number value based on the defined region
+    :rtype: str
+    """
+    if not isinstance(phone_number, str):
+        try:
+            phone_number = str(int(phone_number))
+
+        except ValueError:
+            pass
+
+    phone_number_object = phonenumbers.parse(phone_number, region)
+    parsed_phone_number = phonenumbers.format_number(phone_number_object, phonenumbers.PhoneNumberFormat.E164)
+
+    return parsed_phone_number
+
+
+def get_user_by_phone_number(phone_number: str) -> Optional[User]:
+    """This function queries the database for a user based on the provided phone number.
+    :param phone_number: A valid phone number.
+    :type phone_number: str
+    :return: A user object matching a given phone number
+    :rtype: User|None
+    """
+    # consider adding region to user's metadata
+    phone_number = process_phone_number(phone_number=phone_number, region='KE')
+    user = User.session.query(User).filter_by(phone_number=phone_number).first()
+    return user

apps/cic-ussd/cic_ussd/processor.py

@@ -1,17 +1,26 @@
 # standard imports
 import logging
+import json
+import re
 from typing import Optional
 
 # third party imports
+import celery
+from cic_types.models.person import Person
 from tinydb.table import Document
 
 # local imports
-from cic_ussd.accounts import BalanceManager
+from cic_ussd.account import define_account_tx_metadata, retrieve_account_statement
+from cic_ussd.balance import BalanceManager, compute_operational_balance, get_cached_operational_balance
+from cic_ussd.chain import Chain
 from cic_ussd.db.models.user import AccountStatus, User
 from cic_ussd.db.models.ussd_session import UssdSession
 from cic_ussd.menu.ussd_menu import UssdMenu
+from cic_ussd.metadata import blockchain_address_to_metadata_pointer
+from cic_ussd.phone_number import get_user_by_phone_number
+from cic_ussd.redis import cache_data, create_cached_data_key, get_cached_data
 from cic_ussd.state_machine import UssdStateMachine
-from cic_ussd.transactions import to_wei, from_wei
+from cic_ussd.conversions import to_wei, from_wei
 from cic_ussd.translation import translation_for
 
 logg = logging.getLogger(__name__)
@@ -57,17 +66,17 @@ def process_exit_insufficient_balance(display_key: str, user: User, ussd_session
     :rtype: str
     """
     # get account balance
-    balance_manager = BalanceManager(address=user.blockchain_address,
-                                     chain_str=UssdStateMachine.chain_str,
-                                     token_symbol='SRF')
-    balance = balance_manager.get_operational_balance()
+    operational_balance = get_cached_operational_balance(blockchain_address=user.blockchain_address)
 
     # compile response data
     user_input = ussd_session.get('user_input').split('*')[-1]
     transaction_amount = to_wei(value=int(user_input))
     token_symbol = 'SRF'
+
     recipient_phone_number = ussd_session.get('session_data').get('recipient_phone_number')
-    tx_recipient_information = recipient_phone_number
+    recipient = get_user_by_phone_number(phone_number=recipient_phone_number)
+
+    tx_recipient_information = define_account_tx_metadata(user=recipient)
 
     return translation_for(
         key=display_key,
@@ -75,7 +84,7 @@ def process_exit_insufficient_balance(display_key: str, user: User, ussd_session
         amount=from_wei(transaction_amount),
         token_symbol=token_symbol,
         recipient_information=tx_recipient_information,
-        token_balance=balance
+        token_balance=operational_balance
     )
 
 
@@ -93,9 +102,9 @@ def process_exit_successful_transaction(display_key: str, user: User, ussd_sessi
     transaction_amount = to_wei(int(ussd_session.get('session_data').get('transaction_amount')))
     token_symbol = 'SRF'
     recipient_phone_number = ussd_session.get('session_data').get('recipient_phone_number')
-    sender_phone_number = user.phone_number
-    tx_recipient_information = recipient_phone_number
-    tx_sender_information = sender_phone_number
+    recipient = get_user_by_phone_number(phone_number=recipient_phone_number)
+    tx_recipient_information = define_account_tx_metadata(user=recipient)
+    tx_sender_information = define_account_tx_metadata(user=user)
 
     return translation_for(
         key=display_key,
@@ -122,9 +131,10 @@ def process_transaction_pin_authorization(user: User, display_key: str, ussd_ses
     """
     # compile response data
     recipient_phone_number = ussd_session.get('session_data').get('recipient_phone_number')
-    tx_recipient_information = recipient_phone_number
-    tx_sender_information = user.phone_number
-    logg.debug('Requires integration with cic-meta to get user name.')
+    recipient = get_user_by_phone_number(phone_number=recipient_phone_number)
+    tx_recipient_information = define_account_tx_metadata(user=recipient)
+    tx_sender_information = define_account_tx_metadata(user=user)
+
     token_symbol = 'SRF'
     user_input = ussd_session.get('user_input').split('*')[-1]
     transaction_amount = to_wei(value=int(user_input))
@@ -139,6 +149,122 @@ def process_transaction_pin_authorization(user: User, display_key: str, ussd_ses
     )
 
 
+def process_account_balances(user: User, display_key: str, ussd_session: dict):
+    """
+    :param user:
+    :type user:
+    :param display_key:
+    :type display_key:
+    :param ussd_session:
+    :type ussd_session:
+    :return:
+    :rtype:
+    """
+    # retrieve cached balance
+    operational_balance = get_cached_operational_balance(blockchain_address=user.blockchain_address)
+
+    logg.debug('Requires call to retrieve tax and bonus amounts')
+    tax = ''
+    bonus = ''
+
+    return translation_for(
+        key=display_key,
+        preferred_language=user.preferred_language,
+        operational_balance=operational_balance,
+        tax=tax,
+        bonus=bonus,
+        token_symbol='SRF'
+    )
+
+
+def format_transactions(transactions: list, preferred_language: str):
+    
+    formatted_transactions = ''
+    if len(transactions) > 0:
+        for transaction in transactions:
+            recipient_phone_number = transaction.get('recipient_phone_number')
+            sender_phone_number = transaction.get('sender_phone_number')
+            value = transaction.get('to_value')
+            timestamp = transaction.get('timestamp')
+            action_tag = transaction.get('action_tag')
+            token_symbol = 'SRF'
+
+            if action_tag == 'SENT' or action_tag == 'ULITUMA':
+                formatted_transactions += f'{action_tag} {value} {token_symbol} {recipient_phone_number} {timestamp}.\n'
+            else:
+                formatted_transactions += f'{action_tag} {value} {token_symbol} {sender_phone_number} {timestamp}. \n'
+        return formatted_transactions
+    else:
+        if preferred_language == 'en':
+            formatted_transactions = 'Empty'
+        else:
+            formatted_transactions = 'Hamna historia'
+        return formatted_transactions
+
+
+def process_account_statement(user: User, display_key: str, ussd_session: dict):
+    """
+    :param user:
+    :type user:
+    :param display_key:
+    :type display_key:
+    :param ussd_session:
+    :type ussd_session:
+    :return:
+    :rtype:
+    """
+    # retrieve cached statement
+    identifier = blockchain_address_to_metadata_pointer(blockchain_address=user.blockchain_address)
+    key = create_cached_data_key(identifier=identifier, salt='cic.statement')
+    transactions = get_cached_data(key=key)
+
+    first_transaction_set = []
+    middle_transaction_set = []
+    last_transaction_set = []
+
+    transactions = json.loads(transactions)
+
+    if len(transactions) > 6:
+        last_transaction_set += transactions[6:]
+        middle_transaction_set += transactions[3:][:3]
+        first_transaction_set += transactions[:3]
+    # there are probably much cleaner and operational inexpensive ways to do this so find them
+    elif 4 < len(transactions) < 7:
+        middle_transaction_set += transactions[3:]
+        first_transaction_set += transactions[:3]
+    else:
+        first_transaction_set += transactions[:3]
+
+    if display_key == 'ussd.kenya.first_transaction_set':
+        return translation_for(
+            key=display_key,
+            preferred_language=user.preferred_language,
+            first_transaction_set=format_transactions(
+                transactions=first_transaction_set,
+                preferred_language=user.preferred_language
+            )
+        )
+    elif display_key == 'ussd.kenya.middle_transaction_set':
+        return translation_for(
+            key=display_key,
+            preferred_language=user.preferred_language,
+            middle_transaction_set=format_transactions(
+                transactions=middle_transaction_set,
+                preferred_language=user.preferred_language
+            )
+        )
+
+    elif display_key == 'ussd.kenya.last_transaction_set':
+        return translation_for(
+            key=display_key,
+            preferred_language=user.preferred_language,
+            last_transaction_set=format_transactions(
+                transactions=last_transaction_set,
+                preferred_language=user.preferred_language
+            )
+        )
+
+
 def process_start_menu(display_key: str, user: User):
     """This function gets data on an account's balance and token in order to append it to the start of the start menu's
     title. It passes said arguments to the translation function and returns the appropriate corresponding text from the
@@ -150,16 +276,41 @@ def process_start_menu(display_key: str, user: User):
     :return: Corresponding translation text response
     :rtype: str
     """
-    balance_manager = BalanceManager(address=user.blockchain_address,
-                                     chain_str=UssdStateMachine.chain_str,
+    chain_str = Chain.spec.__str__()
+    blockchain_address = user.blockchain_address
+    balance_manager = BalanceManager(address=blockchain_address,
+                                     chain_str=chain_str,
                                      token_symbol='SRF')
-    balance = balance_manager.get_operational_balance()
+
+    # get balances synchronously for display on start menu
+    balances_data = balance_manager.get_balances()
+
+    key = create_cached_data_key(
+        identifier=bytes.fromhex(blockchain_address[2:]),
+        salt='cic.balances_data'
+    )
+    cache_data(key=key, data=json.dumps(balances_data))
+
+    # get operational balance
+    operational_balance = compute_operational_balance(balances=balances_data)
+
+    # retrieve and cache account's metadata
+    s_query_user_metadata = celery.signature(
+        'cic_ussd.tasks.metadata.query_user_metadata',
+        [blockchain_address]
+    )
+    s_query_user_metadata.apply_async(queue='cic-ussd')
+
+    # retrieve and cache account's statement
+    retrieve_account_statement(blockchain_address=blockchain_address)
+
+    # TODO [Philip]: figure out how to get token symbol from a metadata layer of sorts.
     token_symbol = 'SRF'
-    logg.debug("Requires integration to determine user's balance and token.")
+
     return translation_for(
         key=display_key,
         preferred_language=user.preferred_language,
-        account_balance=balance,
+        account_balance=operational_balance,
         account_token_name=token_symbol
     )
 
@@ -241,5 +392,11 @@ def custom_display_text(
         return process_exit_successful_transaction(display_key=display_key, user=user, ussd_session=ussd_session)
     elif menu_name == 'start':
         return process_start_menu(display_key=display_key, user=user)
+    elif 'pin_authorization' in menu_name:
+        return process_pin_authorization(display_key=display_key, user=user)
+    elif menu_name == 'account_balances':
+        return process_account_balances(display_key=display_key, user=user, ussd_session=ussd_session)
+    elif 'transaction_set' in menu_name:
+        return process_account_statement(display_key=display_key, user=user, ussd_session=ussd_session)
     else:
         return translation_for(key=display_key, preferred_language=user.preferred_language)

apps/cic-ussd/cic_ussd/redis.py

@@ -1,6 +1,52 @@
+# standard imports
+import hashlib
+import logging
+
 # third-party imports
 from redis import Redis
 
+logg = logging.getLogger()
+
 
 class InMemoryStore:
     cache: Redis = None
+
+
+def cache_data(key: str, data: str):
+    """
+    :param key:
+    :type key:
+    :param data:
+    :type data:
+    :return:
+    :rtype:
+    """
+    cache = InMemoryStore.cache
+    cache.set(name=key, value=data)
+    cache.persist(name=key)
+
+
+def get_cached_data(key: str):
+    """
+    :param key:
+    :type key:
+    :return:
+    :rtype:
+    """
+    cache = InMemoryStore.cache
+    return cache.get(name=key)
+
+
+def create_cached_data_key(identifier: bytes, salt: str):
+    """
+    :param identifier:
+    :type identifier:
+    :param salt:
+    :type salt:
+    :return:
+    :rtype:
+    """
+    hash_object = hashlib.new("sha256")
+    hash_object.update(identifier)
+    hash_object.update(salt.encode(encoding="utf-8"))
+    return hash_object.digest().hex()

apps/cic-ussd/cic_ussd/runnable/server.py

@@ -12,14 +12,18 @@ import redis
 
 # third-party imports
 from confini import Config
+from chainlib.chain import ChainSpec
 from urllib.parse import quote_plus
 
 # local imports
+from cic_ussd.chain import Chain
 from cic_ussd.db import dsn_from_config
 from cic_ussd.db.models.base import SessionBase
 from cic_ussd.encoder import PasswordEncoder
 from cic_ussd.files.local_files import create_local_file_data_stores, json_file_parser
 from cic_ussd.menu.ussd_menu import UssdMenu
+from cic_ussd.metadata.signer import Signer
+from cic_ussd.metadata.user import UserMetadata
 from cic_ussd.operations import (define_response_with_content,
                                  process_menu_interaction_requests,
                                  define_multilingual_responses)
@@ -59,6 +63,7 @@ config.censor('PASSWORD', 'DATABASE')
 # define log levels
 if args.vv:
     logging.getLogger().setLevel(logging.DEBUG)
+    logging.getLogger('sqlalchemy.engine').setLevel(logging.DEBUG)
 elif args.v:
     logging.getLogger().setLevel(logging.INFO)
 
@@ -92,6 +97,14 @@ InMemoryStore.cache = redis.StrictRedis(host=config.get('REDIS_HOSTNAME'),
                                         decode_responses=True)
 InMemoryUssdSession.redis_cache = InMemoryStore.cache
 
+# define metadata URL
+UserMetadata.base_url = config.get('CIC_META_URL')
+
+# define signer values
+Signer.gpg_path = config.get('PGP_EXPORT_DIR')
+Signer.gpg_passphrase = config.get('PGP_PASSPHRASE')
+Signer.key_file_path = f"{config.get('PGP_KEYS_PATH')}{config.get('PGP_PRIVATE_KEYS')}"
+
 # initialize celery app
 celery.Celery(backend=config.get('CELERY_RESULT_URL'), broker=config.get('CELERY_BROKER_URL'))
 
@@ -99,7 +112,13 @@ celery.Celery(backend=config.get('CELERY_RESULT_URL'), broker=config.get('CELERY
 states = json_file_parser(filepath=config.get('STATEMACHINE_STATES'))
 transitions = json_file_parser(filepath=config.get('STATEMACHINE_TRANSITIONS'))
 
-UssdStateMachine.chain_str = config.get('CIC_CHAIN_SPEC')
+chain_spec = ChainSpec(
+    common_name=config.get('CIC_COMMON_NAME'),
+    engine=config.get('CIC_ENGINE'),
+    network_id=config.get('CIC_NETWORK_ID')
+)
+
+Chain.spec = chain_spec
 UssdStateMachine.states = states
 UssdStateMachine.transitions = transitions
 
@@ -152,7 +171,8 @@ def application(env, start_response):
             return []
 
         # handle menu interaction requests
-        response = process_menu_interaction_requests(chain_str=config.get('CIC_CHAIN_SPEC'),
+        chain_str = chain_spec.__str__()
+        response = process_menu_interaction_requests(chain_str=chain_str,
                                                      external_session_id=external_session_id,
                                                      phone_number=phone_number,
                                                      queue=args.q,

apps/cic-ussd/cic_ussd/runnable/tasker.py

@@ -12,6 +12,8 @@ from confini import Config
 # local imports
 from cic_ussd.db import dsn_from_config
 from cic_ussd.db.models.base import SessionBase
+from cic_ussd.metadata.signer import Signer
+from cic_ussd.metadata.user import UserMetadata
 from cic_ussd.redis import InMemoryStore
 from cic_ussd.session.ussd_session import UssdSession as InMemoryUssdSession
 
@@ -59,6 +61,14 @@ InMemoryStore.cache = redis.StrictRedis(host=config.get('REDIS_HOSTNAME'),
                                         decode_responses=True)
 InMemoryUssdSession.redis_cache = InMemoryStore.cache
 
+# define metadata URL
+UserMetadata.base_url = config.get('CIC_META_URL')
+
+# define signer values
+Signer.gpg_path = config.get('PGP_EXPORT_DIR')
+Signer.gpg_passphrase = config.get('PGP_PASSPHRASE')
+Signer.key_file_path = f"{config.get('PGP_KEYS_PATH')}{config.get('PGP_PRIVATE_KEYS')}"
+
 # set up celery
 current_app = celery.Celery(__name__)
 

apps/cic-ussd/cic_ussd/state_machine/logic/transaction.py

@@ -1,14 +1,18 @@
 # standard imports
+import json
 import logging
 from typing import Tuple
 
 # third party imports
+import celery
 
 # local imports
-from cic_ussd.accounts import BalanceManager
+from cic_ussd.balance import BalanceManager, compute_operational_balance
+from cic_ussd.chain import Chain
 from cic_ussd.db.models.user import AccountStatus, User
-from cic_ussd.operations import get_user_by_phone_number, save_to_in_memory_ussd_session_data
-from cic_ussd.state_machine.state_machine import UssdStateMachine
+from cic_ussd.operations import save_to_in_memory_ussd_session_data
+from cic_ussd.phone_number import get_user_by_phone_number
+from cic_ussd.redis import create_cached_data_key, get_cached_data
 from cic_ussd.transactions import OutgoingTransactionProcessor
 
 
@@ -27,22 +31,7 @@ def is_valid_recipient(state_machine_data: Tuple[str, dict, User]) -> bool:
     recipient = get_user_by_phone_number(phone_number=user_input)
     is_not_initiator = user_input != user.phone_number
     has_active_account_status = user.get_account_status() == AccountStatus.ACTIVE.name
-    logg.debug('This section requires implementation of checks for user roles and authorization status of an account.')
-    return is_not_initiator and has_active_account_status
-
-
-def is_valid_token_agent(state_machine_data: Tuple[str, dict, User]) -> bool:
-    """This function checks that a user exists, is not the initiator of the transaction, has an active account status
-    and is authorized to perform exchange transactions.
-    :param state_machine_data: A tuple containing user input, a ussd session and user object.
-    :type state_machine_data: tuple
-    :return: A user's validity
-    :rtype: bool
-    """
-    user_input, ussd_session, user = state_machine_data
-    # is_token_agent = AccountRole.TOKEN_AGENT.value in user.get_user_roles()
-    logg.debug('This section requires implementation of user roles and authorization to facilitate exchanges.')
-    return is_valid_recipient(state_machine_data=state_machine_data)
+    return is_not_initiator and has_active_account_status and recipient is not None
 
 
 def is_valid_transaction_amount(state_machine_data: Tuple[str, dict, User]) -> bool:
@@ -70,10 +59,17 @@ def has_sufficient_balance(state_machine_data: Tuple[str, dict, User]) -> bool:
     """
     user_input, ussd_session, user = state_machine_data
     balance_manager = BalanceManager(address=user.blockchain_address,
-                                     chain_str=UssdStateMachine.chain_str,
+                                     chain_str=Chain.spec.__str__(),
                                      token_symbol='SRF')
-    balance = balance_manager.get_operational_balance()
-    return int(user_input) <= balance
+    # get cached balance
+    key = create_cached_data_key(
+        identifier=bytes.fromhex(user.blockchain_address[2:]),
+        salt='cic.balances_data'
+    )
+    cached_balance = get_cached_data(key=key)
+    operational_balance = compute_operational_balance(balances=json.loads(cached_balance))
+
+    return int(user_input) <= operational_balance
 
 
 def save_recipient_phone_to_session_data(state_machine_data: Tuple[str, dict, User]):
@@ -88,6 +84,25 @@ def save_recipient_phone_to_session_data(state_machine_data: Tuple[str, dict, Us
     save_to_in_memory_ussd_session_data(queue='cic-ussd', session_data=session_data, ussd_session=ussd_session)
 
 
+def retrieve_recipient_metadata(state_machine_data: Tuple[str, dict, User]):
+    """
+    :param state_machine_data:
+    :type state_machine_data:
+    :return:
+    :rtype:
+    """
+    user_input, ussd_session, user = state_machine_data
+
+    recipient = get_user_by_phone_number(phone_number=user_input)
+    blockchain_address = recipient.blockchain_address
+    # retrieve and cache account's metadata
+    s_query_user_metadata = celery.signature(
+        'cic_ussd.tasks.metadata.query_user_metadata',
+        [blockchain_address]
+    )
+    s_query_user_metadata.apply_async(queue='cic-ussd')
+
+
 def save_transaction_amount_to_session_data(state_machine_data: Tuple[str, dict, User]):
     """This function saves the phone number corresponding the intended recipients blockchain account.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
@@ -113,7 +128,8 @@ def process_transaction_request(state_machine_data: Tuple[str, dict, User]):
     to_address = recipient.blockchain_address
     from_address = user.blockchain_address
     amount = int(ussd_session.get('session_data').get('transaction_amount'))
-    outgoing_tx_processor = OutgoingTransactionProcessor(chain_str=UssdStateMachine.chain_str,
+    chain_str = Chain.spec.__str__()
+    outgoing_tx_processor = OutgoingTransactionProcessor(chain_str=chain_str,
                                                          from_address=from_address,
                                                          to_address=to_address)
     outgoing_tx_processor.process_outgoing_transfer_transaction(amount=amount)

apps/cic-ussd/cic_ussd/state_machine/logic/user.py

@@ -1,10 +1,20 @@
 # standard imports
+import json
 import logging
 from typing import Tuple
 
+# third-party imports
+import celery
+from cic_types.models.person import Person, generate_metadata_pointer
+from cic_types.models.person import generate_vcard_from_contact_data, manage_identity_data
+
 # local imports
+from cic_ussd.chain import Chain
 from cic_ussd.db.models.user import User
+from cic_ussd.error import UserMetadataNotFoundError
+from cic_ussd.metadata import blockchain_address_to_metadata_pointer
 from cic_ussd.operations import save_to_in_memory_ussd_session_data
+from cic_ussd.redis import get_cached_data
 
 logg = logging.getLogger(__file__)
 
@@ -42,7 +52,29 @@ def update_account_status_to_active(state_machine_data: Tuple[str, dict, User]):
     User.session.commit()
 
 
-def save_profile_attribute_to_session_data(state_machine_data: Tuple[str, dict, User]):
+def process_gender_user_input(user: User, user_input: str):
+    """
+    :param user:
+    :type user:
+    :param user_input:
+    :type user_input:
+    :return:
+    :rtype:
+    """
+    if user.preferred_language == 'en':
+        if user_input == '1':
+            gender = 'Male'
+        else:
+            gender = 'Female'
+    else:
+        if user_input == '1':
+            gender = 'Mwanaume'
+        else:
+            gender = 'Mwanamke'
+    return gender
+
+
+def save_metadata_attribute_to_session_data(state_machine_data: Tuple[str, dict, User]):
     """This function saves first name data to the ussd session in the redis cache.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: tuple
@@ -54,16 +86,17 @@ def save_profile_attribute_to_session_data(state_machine_data: Tuple[str, dict,
 
     # define session data key from current state
     key = ''
-    if 'first_name' in current_state:
-        key = 'first_name'
-    elif 'last_name' in current_state:
-        key = 'last_name'
+    if 'given_name' in current_state:
+        key = 'given_name'
+    elif 'family_name' in current_state:
+        key = 'family_name'
     elif 'gender' in current_state:
         key = 'gender'
+        user_input = process_gender_user_input(user=user, user_input=user_input)
     elif 'location' in current_state:
         key = 'location'
-    elif 'business_profile' in current_state:
-        key = 'business_profile'
+    elif 'products' in current_state:
+        key = 'products'
 
     # check if there is existing session data
     if ussd_session.get('session_data'):
@@ -76,14 +109,120 @@ def save_profile_attribute_to_session_data(state_machine_data: Tuple[str, dict,
     save_to_in_memory_ussd_session_data(queue='cic-ussd', session_data=session_data, ussd_session=ussd_session)
 
 
-def persist_profile_data(state_machine_data: Tuple[str, dict, User]):
-    """This function persists elements of the user profile stored in session data
+def format_user_metadata(metadata: dict, user: User):
+    """
+    :param metadata:
+    :type metadata:
+    :param user:
+    :type user:
+    :return:
+    :rtype:
+    """
+    gender = metadata.get('gender')
+    given_name = metadata.get('given_name')
+    family_name = metadata.get('family_name')
+    location = {
+        "area_name": metadata.get('location')
+    }
+    products = []
+    if metadata.get('products'):
+        products = metadata.get('products').split(',')
+    phone_number = user.phone_number
+    date_registered = int(user.created.replace().timestamp())
+    blockchain_address = user.blockchain_address
+    chain_spec = f'{Chain.spec.common_name()}:{Chain.spec.network_id()}'
+    identities = manage_identity_data(
+        blockchain_address=blockchain_address,
+        blockchain_type=Chain.spec.engine(),
+        chain_spec=chain_spec
+    )
+    return {
+        "date_registered": date_registered,
+        "gender": gender,
+        "identities": identities,
+        "location": location,
+        "products": products,
+        "vcard": generate_vcard_from_contact_data(
+            family_name=family_name,
+            given_name=given_name,
+            tel=phone_number
+        )
+    }
+
+
+def save_complete_user_metadata(state_machine_data: Tuple[str, dict, User]):
+    """This function persists elements of the user metadata stored in session data
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: tuple
     """
     user_input, ussd_session, user = state_machine_data
 
     # get session data
-    profile_data = ussd_session.get('session_data')
-    logg.debug('This section requires implementation of user metadata.')
+    metadata = ussd_session.get('session_data')
 
+    # format metadata appropriately
+    user_metadata = format_user_metadata(metadata=metadata, user=user)
+
+    blockchain_address = user.blockchain_address
+    s_create_user_metadata = celery.signature(
+        'cic_ussd.tasks.metadata.create_user_metadata',
+        [blockchain_address, user_metadata]
+    )
+    s_create_user_metadata.apply_async(queue='cic-ussd')
+
+
+def edit_user_metadata_attribute(state_machine_data: Tuple[str, dict, User]):
+    user_input, ussd_session, user = state_machine_data
+    blockchain_address = user.blockchain_address
+    key = generate_metadata_pointer(
+        identifier=blockchain_address_to_metadata_pointer(blockchain_address=user.blockchain_address),
+        cic_type='cic.person'
+    )
+    user_metadata = get_cached_data(key=key)
+
+    if not user_metadata:
+        raise UserMetadataNotFoundError(f'Expected user metadata but found none in cache for key: {blockchain_address}')
+
+    given_name = ussd_session.get('session_data').get('given_name')
+    family_name = ussd_session.get('session_data').get('family_name')
+    gender = ussd_session.get('session_data').get('gender')
+    location = ussd_session.get('session_data').get('location')
+    products = ussd_session.get('session_data').get('products')
+
+    # validate user metadata
+    person = Person()
+    user_metadata = json.loads(user_metadata)
+    deserialized_person = person.deserialize(metadata=user_metadata)
+
+    # edit specific metadata attribute
+    if given_name:
+        deserialized_person.given_name = given_name
+    elif family_name:
+        deserialized_person.family_name = family_name
+    elif gender:
+        deserialized_person.gender = gender
+    elif location:
+        # get existing location metadata:
+        location_data = user_metadata.get('location')
+        location_data['area_name'] = location
+        deserialized_person.location = location_data
+    elif products:
+        deserialized_person.products = products
+
+    edited_metadata = deserialized_person.serialize()
+
+    s_edit_user_metadata = celery.signature(
+        'cic_ussd.tasks.metadata.edit_user_metadata',
+        [blockchain_address, edited_metadata, 'pgp']
+    )
+    s_edit_user_metadata.apply_async(queue='cic-ussd')
+
+
+def get_user_metadata(state_machine_data: Tuple[str, dict, User]):
+    user_input, ussd_session, user = state_machine_data
+    blockchain_address = user.blockchain_address
+    s_get_user_metadata = celery.signature(
+        'cic_ussd.tasks.metadata.query_user_metadata',
+        [blockchain_address]
+    )
+    s_get_user_metadata.apply_async(queue='cic-ussd')

apps/cic-ussd/cic_ussd/state_machine/logic/validator.py

@@ -3,55 +3,30 @@ import logging
 import re
 from typing import Tuple
 
+# third-party imports
+from cic_types.models.person import generate_metadata_pointer
+
 # local imports
 from cic_ussd.db.models.user import User
+from cic_ussd.metadata import blockchain_address_to_metadata_pointer
+from cic_ussd.redis import get_cached_data
 
 logg = logging.getLogger()
 
 
-def has_complete_profile_data(state_machine_data: Tuple[str, dict, User]):
+def has_cached_user_metadata(state_machine_data: Tuple[str, dict, User]):
     """This function checks whether the attributes of the user's metadata constituting a profile are filled out.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: str
     """
     user_input, ussd_session, user = state_machine_data
-    logg.debug('This section requires implementation of user metadata.')
-
-
-def has_empty_username_data(state_machine_data: Tuple[str, dict, User]):
-    """This function checks whether the aspects of the user's name metadata is filled out.
-    :param state_machine_data: A tuple containing user input, a ussd session and user object.
-    :type state_machine_data: str
-    """
-    user_input, ussd_session, user = state_machine_data
-    logg.debug('This section requires implementation of user metadata.')
-
-
-def has_empty_gender_data(state_machine_data: Tuple[str, dict, User]):
-    """This function checks whether the aspects of the user's gender metadata is filled out.
-    :param state_machine_data: A tuple containing user input, a ussd session and user object.
-    :type state_machine_data: str
-    """
-    user_input, ussd_session, user = state_machine_data
-    logg.debug('This section requires implementation of user metadata.')
-
-
-def has_empty_location_data(state_machine_data: Tuple[str, dict, User]):
-    """This function checks whether the aspects of the user's location metadata is filled out.
-    :param state_machine_data: A tuple containing user input, a ussd session and user object.
-    :type state_machine_data: str
-    """
-    user_input, ussd_session, user = state_machine_data
-    logg.debug('This section requires implementation of user metadata.')
-
-
-def has_empty_business_profile_data(state_machine_data: Tuple[str, dict, User]):
-    """This function checks whether the aspects of the user's business profile metadata is filled out.
-    :param state_machine_data: A tuple containing user input, a ussd session and user object.
-    :type state_machine_data: str
-    """
-    user_input, ussd_session, user = state_machine_data
-    logg.debug('This section requires implementation of user metadata.')
+    # check for user metadata in cache
+    key = generate_metadata_pointer(
+        identifier=blockchain_address_to_metadata_pointer(blockchain_address=user.blockchain_address),
+        cic_type='cic.person'
+    )
+    user_metadata = get_cached_data(key=key)
+    return user_metadata is not None
 
 
 def is_valid_name(state_machine_data: Tuple[str, dict, User]):
@@ -66,3 +41,18 @@ def is_valid_name(state_machine_data: Tuple[str, dict, User]):
         return True
     else:
         return False
+
+
+def is_valid_gender_selection(state_machine_data: Tuple[str, dict, User]):
+    """
+    :param state_machine_data:
+    :type state_machine_data:
+    :return:
+    :rtype:
+    """
+    user_input, ussd_session, user = state_machine_data
+    selection_matcher = "^[1-2]$"
+    if re.match(selection_matcher, user_input):
+        return True
+    else:
+        return False

apps/cic-ussd/cic_ussd/state_machine/state_machine.py

@@ -14,15 +14,11 @@ class UssdStateMachine(Machine):
     menu  as well as providing a means for navigating through these states based on different user inputs.
     It defines different helper functions that co-ordinate with the stakeholder components of the ussd menu: i.e  the
     User, UssdSession, UssdMenu to facilitate user interaction with ussd menu.
-
-    :cvar chain_str: The chain name and network id.
-    :type chain_str: str
     :cvar states: A list of pre-defined states.
     :type states: list
     :cvar transitions: A list of pre-defined transitions.
     :type transitions: list
     """
-    chain_str = None
     states = []
     transitions = []
 

apps/cic-ussd/cic_ussd/tasks/__init__.py

@@ -10,6 +10,7 @@ import celery
 
 celery_app = celery.current_app
 # export external celery task modules
-from .foo import log_it_plz
-from .ussd import persist_session_to_db
-from .callback_handler import process_account_creation_callback
+from .logger import *
+from .ussd_session import *
+from .callback_handler import *
+from .metadata import *