Catch invalid phone return value in verify script

Fix missing history syncer in cic-cache-tracker

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

.gitlab-ci.yml

@@ -5,6 +5,7 @@ include:
   - local: 'apps/cic-ussd/.gitlab-ci.yml'
   - local: 'apps/cic-notify/.gitlab-ci.yml'
   - local: 'apps/cic-meta/.gitlab-ci.yml'
+  - local: 'apps/cic-cache/.gitlab-ci.yml'
 
 stages:
   - build

.gitmodules

@@ -1,3 +0,0 @@
-[submodule "apps/cic-cache"]
-	path = apps/cic-cache
-	url = git@gitlab.com:grassrootseconomics/cic-cache.git

apps/cic-cache/.config/bancor.ini

@@ -0,0 +1,2 @@
+[bancor]
+dir =

apps/cic-cache/.config/cic.ini

@@ -0,0 +1,2 @@
+[cic]
+registry_address = 

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

@@ -0,0 +1,8 @@
+[database]
+NAME=cic_cache
+USER=postgres
+PASSWORD=
+HOST=localhost
+PORT=5432
+ENGINE=postgresql
+DRIVER=psycopg2

apps/cic-cache/.config/eth.ini

@@ -0,0 +1,6 @@
+[eth]
+provider = ws://localhost:8545
+#ttp_provider = http://localhost:8545
+#provider = http://localhost:8545
+gas_provider_address = 
+#chain_id = 

apps/cic-cache/.config/test/bancor.ini

@@ -0,0 +1,2 @@
+[bancor]
+dir =

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

@@ -0,0 +1,2 @@
+[cic]
+registry_address = 

apps/cic-cache/.config/test/database.ini

@@ -0,0 +1,8 @@
+[database]
+NAME=cic-cache-test
+USER=postgres
+PASSWORD=
+HOST=localhost
+PORT=5432
+ENGINE=sqlite
+DRIVER=pysqlite

apps/cic-cache/.config/test/eth.ini

@@ -0,0 +1,5 @@
+[eth]
+#ws_provider = ws://localhost:8546
+#ttp_provider = http://localhost:8545
+provider = http://localhost:8545
+#chain_id = 

apps/cic-cache/.coveragerc

@@ -0,0 +1,5 @@
+[report]
+omit = 
+	.venv/*
+	scripts/*
+	cic_cache/db/postgres/*

apps/cic-cache/.envrc_example

@@ -0,0 +1,7 @@
+set -a
+CICTEST_DATABASE_ENGINE=postgresql
+CICTEST_DATABASE_DRIVER=psycopg2
+CICTEST_DATABASE_HOST=localhost
+CICTEST_DATABASE_PORT=5432
+CICTEST_DATABASE_NAME=cic-eth-test
+set +a

apps/cic-cache/.gitignore

@@ -0,0 +1,8 @@
+.envrc
+.envrc_dev
+.venv
+__pycache__
+*.pyc
+_build
+doc/**/*.png
+doc/**/html

apps/cic-cache/.gitlab-ci.yml

@@ -0,0 +1,22 @@
+.cic_cache_variables:
+    variables:
+        APP_NAME: cic-cache
+        DOCKERFILE_PATH: $APP_NAME/docker/Dockerfile
+
+.cic_cache_changes_target:
+    rules:
+        - changes:
+            - $CONTEXT/$APP_NAME/*
+
+build-mr-cic-cache:
+    extends:
+        - .cic_cache_changes_target
+        - .py_build_merge_request
+        - .cic_cache_variables
+
+build-push-cic-cache:
+    extends:
+        - .py_build_push
+        - .cic_cache_variables
+        
+

apps/cic-cache/CHANGELOG

@@ -0,0 +1,13 @@
+- 0.1.2
+	* Revert to alembic migrations
+- 0.1.1
+	* Add missing modules to setup 
+- 0.1.0
+	* Remove old APIs
+	* Add bloom filter output APIs for all txs and per-account txs
+- 0.0.2
+	* UWSGI server endpoint example
+	* OpenAPI spec
+	* stored procedures, test fixture for database schema
+- 0.0.1
+	* Add json translators of transaction_list and balances stored procedure queries

apps/cic-cache/cic_cache/__init__.py

@@ -0,0 +1 @@
+from .cache import BloomCache

apps/cic-cache/cic_cache/api.py

@@ -0,0 +1,73 @@
+"""API for cic-cache celery tasks
+
+.. moduleauthor:: Louis Holbrook <dev@holbrook.no>
+
+"""
+# standard imports
+import logging
+
+# third-party imports 
+import celery
+
+
+app = celery.current_app
+
+logg = logging.getLogger(__name__)
+
+
+class Api:
+    """Creates task chains to perform well-known CIC operations.
+
+    Each method that sends tasks returns details about the root task. The root task uuid can be provided in the callback, to enable to caller to correlate the result with individual calls. It can also be used to independently poll the completion of a task chain.
+
+    :param callback_param: Static value to pass to callback
+    :type callback_param: str
+    :param callback_task: Callback task that executes callback_param call. (Must be included by the celery worker)
+    :type callback_task: string
+    :param queue: Name of worker queue to submit tasks to
+    :type queue: str
+    """
+    def __init__(self, queue='cic-cache', callback_param=None, callback_task='cic_cache.callbacks.noop.noop', callback_queue=None):
+        self.callback_param = callback_param
+        self.callback_task = callback_task
+        self.queue = queue
+        logg.info('api using queue {}'.format(self.queue))
+        self.callback_success = None
+        self.callback_error = None
+        if callback_queue == None:
+            callback_queue=self.queue
+
+        if callback_param != None:
+            self.callback_success = celery.signature(
+                    callback_task,
+                    [
+                        callback_param,
+                        0,
+                        ],
+                    queue=callback_queue,
+                    )
+            self.callback_error = celery.signature(
+                    callback_task,
+                    [
+                        callback_param,
+                        1,
+                        ],
+                    queue=callback_queue,
+                    )       
+
+    def list(self, offset, limit, address=None):
+        s = celery.signature(
+        'cic_cache.tasks.tx.tx_filter',
+        [
+            0,
+            100,
+            address,
+            ],
+            queue=None
+        )
+        if self.callback_param != None:
+            s.link(self.callback_success).on_error(self.callback_error)
+
+        t = s.apply_async()
+
+        return t

apps/cic-cache/cic_cache/cache.py

@@ -0,0 +1,89 @@
+# standard imports
+import logging
+
+# third-party imports
+import moolb
+
+# local imports
+from cic_cache.db import list_transactions_mined
+from cic_cache.db import list_transactions_account_mined
+
+logg = logging.getLogger()
+
+
+class BloomCache:
+
+    def __init__(self, session):
+        self.session = session
+
+
+    @staticmethod
+    def __get_filter_size(n):
+        n = 8192 * 8
+        logg.warning('filter size hardcoded to {}'.format(n))
+        return n
+
+
+    def load_transactions(self, offset, limit):
+        """Retrieves a list of transactions from cache and creates a bloom filter pointing to blocks and transactions.
+
+        Block and transaction numbers are serialized as 32-bit big-endian numbers. The input to the second bloom filter is the concatenation of the serialized block number and transaction index.
+
+        For example, if the block number is 13 and the transaction index is 42, the input are:
+
+        block filter:       0x0d000000
+        block+tx filter:    0x0d0000002a0000000
+
+        :param offset: Offset in data set to return transactions from
+        :type offset: int
+        :param limit: Max number of transactions to retrieve
+        :type limit: int
+        :return: Lowest block, bloom filter for blocks, bloom filter for blocks|tx
+        :rtype: tuple
+        """
+        rows = list_transactions_mined(self.session, offset, limit) 
+
+        f_block = moolb.Bloom(BloomCache.__get_filter_size(limit), 3)
+        f_blocktx = moolb.Bloom(BloomCache.__get_filter_size(limit), 3)
+        highest_block = -1
+        lowest_block = -1
+        for r in rows:
+            if highest_block == -1:
+                highest_block = r[0]
+            lowest_block = r[0]
+            block = r[0].to_bytes(4, byteorder='big')
+            tx = r[1].to_bytes(4, byteorder='big')
+            f_block.add(block)
+            f_blocktx.add(block + tx)
+            logg.debug('added block {} tx {} lo {} hi {}'.format(r[0], r[1], lowest_block, highest_block))
+        return (lowest_block, highest_block, f_block.to_bytes(), f_blocktx.to_bytes(),)
+
+
+    def load_transactions_account(self, address, offset, limit):
+        """Same as load_transactions(...), but only retrieves transactions where the specified account address is sender or recipient.
+
+        :param address: Address to retrieve transactions for.
+        :type address: str, 0x-hex
+        :param offset: Offset in data set to return transactions from
+        :type offset: int
+        :param limit: Max number of transactions to retrieve
+        :type limit: int
+        :return: Lowest block, bloom filter for blocks, bloom filter for blocks|tx
+        :rtype: tuple
+        """
+        rows = list_transactions_account_mined(self.session, address, offset, limit) 
+
+        f_block = moolb.Bloom(BloomCache.__get_filter_size(limit), 3)
+        f_blocktx = moolb.Bloom(BloomCache.__get_filter_size(limit), 3)
+        highest_block = -1;
+        lowest_block = -1;
+        for r in rows:
+            if highest_block == -1:
+                highest_block = r[0]
+            lowest_block = r[0]
+            block = r[0].to_bytes(4, byteorder='big')
+            tx = r[1].to_bytes(4, byteorder='big')
+            f_block.add(block)
+            f_blocktx.add(block + tx)
+            logg.debug('added block {} tx {} lo {} hi {}'.format(r[0], r[1], lowest_block, highest_block))
+        return (lowest_block, highest_block, f_block.to_bytes(), f_blocktx.to_bytes(),)

apps/cic-cache/cic_cache/db/__init__.py

@@ -0,0 +1,35 @@
+# standard imports
+import logging
+
+# local imports
+from .list import list_transactions_mined
+from .list import list_transactions_account_mined
+from .list import add_transaction
+
+logg = logging.getLogger()
+
+
+def dsn_from_config(config):
+    scheme = config.get('DATABASE_ENGINE')
+    if config.get('DATABASE_DRIVER') != None:
+        scheme += '+{}'.format(config.get('DATABASE_DRIVER'))
+
+    dsn = ''
+    if config.get('DATABASE_ENGINE') == 'sqlite':
+        dsn = '{}:///{}'.format(
+                scheme,
+                config.get('DATABASE_NAME'),    
+            )
+
+    else:
+        dsn = '{}://{}:{}@{}:{}/{}'.format(
+                scheme,
+                config.get('DATABASE_USER'),
+                config.get('DATABASE_PASSWORD'),
+                config.get('DATABASE_HOST'),
+                config.get('DATABASE_PORT'),
+                config.get('DATABASE_NAME'),    
+            )
+    logg.debug('parsed dsn from config: {}'.format(dsn))
+    return dsn
+

apps/cic-cache/cic_cache/db/list.py

@@ -0,0 +1,79 @@
+# standard imports
+import logging
+import datetime
+
+# third-party imports
+from cic_cache.db.models.base import SessionBase
+
+logg = logging.getLogger()
+
+
+def list_transactions_mined(
+        session,
+        offset,
+        limit,
+        ):
+    """Executes db query to return all confirmed transactions according to the specified offset and limit.
+
+    :param offset: Offset in data set to return transactions from
+    :type offset: int
+    :param limit: Max number of transactions to retrieve
+    :type limit: int
+    :result: Result set
+    :rtype: SQLAlchemy.ResultProxy
+    """
+    s = "SELECT block_number, tx_index FROM tx ORDER BY block_number DESC, tx_index DESC LIMIT {} OFFSET {}".format(limit, offset)
+    r = session.execute(s)
+    return r
+
+
+def list_transactions_account_mined(
+        session,
+        address,
+        offset,
+        limit,
+        ):
+    """Same as list_transactions_mined(...), but only retrieves transaction where the specified account address is sender or recipient.
+
+    :param address: Address to retrieve transactions for.
+    :type address: str, 0x-hex
+    :param offset: Offset in data set to return transactions from
+    :type offset: int
+    :param limit: Max number of transactions to retrieve
+    :type limit: int
+    :result: Result set
+    :rtype: SQLAlchemy.ResultProxy
+    """
+    s = "SELECT block_number, tx_index FROM tx WHERE sender = '{}' OR recipient = '{}' ORDER BY block_number DESC, tx_index DESC LIMIT {} OFFSET {}".format(address, address, limit, offset)
+    r = session.execute(s)
+    return r
+
+
+def add_transaction(
+        session, tx_hash,
+        block_number,
+        tx_index,
+        sender,
+        receiver,
+        source_token,
+        destination_token,
+        from_value,
+        to_value,
+        success,
+        timestamp,
+        ):
+    date_block = datetime.datetime.fromtimestamp(timestamp)
+    s = "INSERT INTO tx (tx_hash, block_number, tx_index, sender, recipient, source_token, destination_token, from_value, to_value, success, date_block) VALUES ('{}', {}, {}, '{}', '{}', '{}', '{}', {}, {}, {}, '{}')".format(
+            tx_hash,
+            block_number,
+            tx_index,
+            sender,
+            receiver,
+            source_token,
+            destination_token,
+            from_value,
+            to_value,
+            success,
+            date_block,
+            )
+    session.execute(s)

apps/cic-cache/cic_cache/db/migrations/default/README

@@ -0,0 +1 @@
+Generic single-database configuration.

apps/cic-cache/cic_cache/db/migrations/default/alembic.ini

@@ -27,16 +27,17 @@ script_location = .
 # sourceless = false
 
 # version location specification; this defaults
-# to migrations/versions.  When using multiple version
+# to ./versions.  When using multiple version
 # directories, initial revisions must be specified with --version-path
-# version_locations = %(here)s/bar %(here)s/bat migrations/versions
+# version_locations = %(here)s/bar %(here)s/bat ./versions
 
 # the output encoding used when revision files
 # are written from script.py.mako
 # output_encoding = utf-8
 
 #sqlalchemy.url = driver://user:pass@localhost/dbname
-sqlalchemy.url = postgresql+psycopg2://postgres@localhost:5432/cic-eth
+sqlalchemy.url = postgresql+psycopg2://postgres@localhost:5432/cic_cache
+
 
 [post_write_hooks]
 # post_write_hooks defines scripts or Python functions that are run

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

@@ -0,0 +1,52 @@
+"""Base tables
+
+Revision ID: 63b629f14a85
+Revises: 
+Create Date: 2020-12-04 08:16:00.412189
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '63b629f14a85'
+down_revision = None
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+            'tx',
+            sa.Column('id', sa.Integer, primary_key=True),
+            sa.Column('date_registered', sa.DateTime, nullable=False, server_default=sa.func.current_timestamp()),
+            sa.Column('block_number', sa.Integer, nullable=False),
+            sa.Column('tx_index', sa.Integer, nullable=False),
+            sa.Column('tx_hash', sa.String(66), nullable=False),
+            sa.Column('sender', sa.String(42), nullable=False),
+            sa.Column('recipient', sa.String(42), nullable=False),
+            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.NUMERIC(), nullable=False),
+            sa.Column('to_value', sa.NUMERIC(), nullable=False),
+            sa.Column('date_block', sa.DateTime, nullable=False),
+            )
+    op.create_table(
+            'tx_sync',
+            sa.Column('id', sa.Integer, primary_key=True),
+            sa.Column('tx', sa.String(66), nullable=False),
+            )
+
+    op.execute("INSERT INTO tx_sync (tx) VALUES('0x0000000000000000000000000000000000000000000000000000000000000000');")
+
+    op.create_index('sender_token_idx', 'tx', ['sender', 'source_token'])
+    op.create_index('recipient_token_idx', 'tx', ['recipient', 'destination_token'])
+            
+
+def downgrade():
+    op.drop_index('recipient_token_idx')
+    op.drop_index('sender_token_idx')
+    op.drop_table('tx_sync')
+    op.drop_table('tx')

apps/cic-cache/cic_cache/db/migrations/default/versions/6604de4203e2_add_chain_syncer.py

@@ -0,0 +1,28 @@
+"""Add chain syncer
+
+Revision ID: 6604de4203e2
+Revises: 63b629f14a85
+Create Date: 2021-04-01 08:10:29.156243
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from chainsyncer.db.migrations.sqlalchemy import (
+        chainsyncer_upgrade,
+        chainsyncer_downgrade,
+        )
+
+
+# revision identifiers, used by Alembic.
+revision = '6604de4203e2'
+down_revision = '63b629f14a85'
+branch_labels = None
+depends_on = None
+
+def upgrade():
+    chainsyncer_upgrade(0, 0, 1)
+
+
+def downgrade():
+    chainsyncer_downgrade(0, 0, 1)
+

apps/cic-cache/cic_cache/db/models/base.py

@@ -0,0 +1,102 @@
+# stanard imports
+import logging
+
+# third-party imports
+from sqlalchemy import Column, Integer
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+logg = logging.getLogger()
+
+Model = declarative_base(name='Model')
+
+
+class SessionBase(Model):
+    """The base object for all SQLAlchemy enabled models. All other models must extend this.
+    """
+    __abstract__ = True
+ 
+    id = Column(Integer, primary_key=True)
+
+    engine = None
+    """Database connection engine of the running aplication"""
+    sessionmaker = None
+    """Factory object responsible for creating sessions from the  connection pool"""
+    transactional = True
+    """Whether the database backend supports query transactions. Should be explicitly set by initialization code"""
+    poolable = True
+    """Whether the database backend supports connection pools. Should be explicitly set by initialization code"""
+    procedural = True
+    """Whether the database backend supports stored procedures"""
+    localsessions = {}
+    """Contains dictionary of sessions initiated by db model components"""
+
+
+    @staticmethod
+    def create_session():
+        """Creates a new database session.
+        """
+        return SessionBase.sessionmaker()
+
+
+    @staticmethod
+    def _set_engine(engine):
+        """Sets the database engine static property
+        """
+        SessionBase.engine = engine
+        SessionBase.sessionmaker = sessionmaker(bind=SessionBase.engine)
+
+
+    @staticmethod
+    def connect(dsn, debug=False):
+        """Create new database connection engine and connect to database backend.
+
+        :param dsn: DSN string defining connection.
+        :type dsn: str
+        """
+        e = None
+        if SessionBase.poolable:
+            e = create_engine(
+                    dsn,
+                    max_overflow=50,
+                    pool_pre_ping=True,
+                    pool_size=20,
+                    pool_recycle=10,
+                    echo=debug,
+                )
+        else:
+            e = create_engine(
+                    dsn,
+                    echo=debug,
+                    )
+
+        SessionBase._set_engine(e)
+
+
+    @staticmethod
+    def disconnect():
+        """Disconnect from database and free resources.
+        """
+        SessionBase.engine.dispose()
+        SessionBase.engine = None
+
+
+    @staticmethod
+    def bind_session(session=None):
+        localsession = session
+        if localsession == None:
+            localsession = SessionBase.create_session()
+            localsession_key = str(id(localsession))
+            logg.debug('creating new session {}'.format(localsession_key))
+            SessionBase.localsessions[localsession_key] = localsession
+        return localsession
+
+
+    @staticmethod
+    def release_session(session=None):
+        session_key = str(id(session))
+        if SessionBase.localsessions.get(session_key) != None:
+            logg.debug('destroying session {}'.format(session_key))
+            session.commit()
+            session.close()

apps/cic-cache/cic_cache/runnable/daemons/filters/__init__.py

@@ -0,0 +1 @@
+from .erc20 import *

apps/cic-cache/cic_cache/runnable/daemons/filters/base.py

@@ -0,0 +1,2 @@
+class SyncFilter:
+    pass

apps/cic-cache/cic_cache/runnable/daemons/filters/erc20.py

@@ -0,0 +1,72 @@
+# standard imports
+import logging
+
+# external imports
+from chainlib.eth.erc20 import ERC20
+from chainlib.eth.address import (
+        to_checksum_address,
+        )
+from chainlib.eth.error import RequestMismatchException
+from chainlib.status import Status
+from cic_eth_registry.erc20 import ERC20Token
+from cic_eth_registry.error import (
+        NotAContractError,
+        ContractMismatchError,
+        )
+
+# local imports
+from .base import SyncFilter
+from cic_cache import db as cic_cache_db
+
+logg = logging.getLogger().getChild(__name__)
+
+
+class ERC20TransferFilter(SyncFilter):
+
+    def __init__(self, chain_spec):
+        self.chain_spec = chain_spec
+
+
+    # TODO: Verify token in declarator / token index
+    def filter(self, conn, block, tx, db_session=None):
+        logg.debug('filter {} {}'.format(block, tx))
+        token = None
+        try:
+            token = ERC20Token(self.chain_spec, conn, tx.inputs[0])
+        except NotAContractError:
+            logg.debug('not a contract {}'.format(tx.inputs[0]))
+            return False
+        except ContractMismatchError:
+            logg.debug('not an erc20 token  {}'.format(tx.inputs[0]))
+            return False
+
+        transfer_data = None
+        try:
+            transfer_data = ERC20.parse_transfer_request(tx.payload)
+        except RequestMismatchException:
+            logg.debug('erc20 match but not a transfer, skipping')
+            return False
+
+        token_sender = tx.outputs[0]
+        token_recipient = transfer_data[0]
+        token_value = transfer_data[1]
+
+        logg.debug('matched erc20 token transfer {} ({}) to {} value {}'.format(token.name, token.address, transfer_data[0], transfer_data[1]))
+
+        cic_cache_db.add_transaction(
+                db_session,
+                tx.hash,
+                block.number,
+                tx.index,
+                to_checksum_address(token_sender),
+                to_checksum_address(token_recipient),
+                token.address,
+                token.address,
+                token_value,
+                token_value,
+                tx.status == Status.SUCCESS,
+                block.timestamp,
+                )
+        db_session.flush()
+
+        return True

apps/cic-cache/cic_cache/runnable/daemons/server.py

@@ -0,0 +1,141 @@
+# standard imports
+import os
+import re
+import logging
+import argparse
+import json
+import base64
+
+# third-party imports
+import confini
+
+# local imports
+from cic_cache import BloomCache
+from cic_cache.db import dsn_from_config
+from cic_cache.db.models.base import SessionBase
+
+logging.basicConfig(level=logging.WARNING)
+logg = logging.getLogger()
+
+rootdir = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
+dbdir = os.path.join(rootdir, 'cic_cache', 'db')
+migrationsdir = os.path.join(dbdir, 'migrations')
+
+config_dir = os.path.join('/usr/local/etc/cic-cache')
+
+argparser = argparse.ArgumentParser()
+argparser.add_argument('-c', type=str, default=config_dir, help='config file')
+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('-v', action='store_true', help='be verbose')
+argparser.add_argument('-vv', action='store_true', help='be more verbose')
+args = argparser.parse_args()
+
+if args.vv:
+    logging.getLogger().setLevel(logging.DEBUG)
+elif args.v:
+    logging.getLogger().setLevel(logging.INFO)
+
+config = confini.Config(args.c, args.env_prefix)
+config.process()
+config.censor('PASSWORD', 'DATABASE')
+config.censor('PASSWORD', 'SSL')
+logg.debug('config:\n{}'.format(config))
+
+dsn = dsn_from_config(config)
+SessionBase.connect(dsn, config.true('DATABASE_DEBUG'))
+
+re_transactions_all_bloom = r'/tx/(\d+)?/?(\d+)/?'
+re_transactions_account_bloom = r'/tx/user/((0x)?[a-fA-F0-9]+)/?(\d+)?/?(\d+)/?'
+
+DEFAULT_LIMIT = 100
+
+
+def process_transactions_account_bloom(session, env):
+    r = re.match(re_transactions_account_bloom, env.get('PATH_INFO'))
+    if not r:
+        return None
+
+    address = r[1]
+    if r[2] == None:
+        address = '0x' + address
+    offset = DEFAULT_LIMIT
+    if r.lastindex > 2:
+        offset = r[3]
+    limit = 0
+    if r.lastindex > 3:
+        limit = r[4]
+
+    c = BloomCache(session)
+    (lowest_block, highest_block, bloom_filter_block, bloom_filter_tx) = c.load_transactions_account(address, offset, limit)
+
+    o = {
+        'alg': 'sha256',
+        'low': lowest_block,
+        'high': highest_block,
+        'block_filter': base64.b64encode(bloom_filter_block).decode('utf-8'),
+        'blocktx_filter': base64.b64encode(bloom_filter_tx).decode('utf-8'),
+        'filter_rounds': 3,
+            }
+
+    j = json.dumps(o)
+
+    return ('application/json', j.encode('utf-8'),)
+
+
+def process_transactions_all_bloom(session, env):
+    r = re.match(re_transactions_all_bloom, env.get('PATH_INFO'))
+    if not r:
+        return None
+
+    offset = DEFAULT_LIMIT
+    if r.lastindex > 0:
+        offset = r[1]
+    limit = 0
+    if r.lastindex > 1:
+        limit = r[2]
+
+    c = BloomCache(session)
+    (lowest_block, highest_block, bloom_filter_block, bloom_filter_tx) = c.load_transactions(offset, limit)
+
+    o = {
+        'alg': 'sha256',
+        'low': lowest_block,
+        'high': highest_block,
+        'block_filter': base64.b64encode(bloom_filter_block).decode('utf-8'),
+        'blocktx_filter': base64.b64encode(bloom_filter_tx).decode('utf-8'),
+        'filter_rounds': 3,
+            }
+
+    j = json.dumps(o)
+
+    return ('application/json', j.encode('utf-8'),)
+
+
+# uwsgi application
+def application(env, start_response):
+
+    headers = []
+    content = b''
+
+    session = SessionBase.create_session()
+    for handler in [
+            process_transactions_all_bloom,
+            process_transactions_account_bloom,
+            ]:
+        r = handler(session, env)
+        if r != None:
+            (mime_type, content) = r
+            break
+    session.close()
+
+    headers.append(('Content-Length', str(len(content))),)
+    headers.append(('Access-Control-Allow-Origin', '*',));
+
+    if len(content) == 0:
+        headers.append(('Content-Type', 'text/plain, charset=UTF-8',))
+        start_response('404 Looked everywhere, sorry', headers)
+    else:
+        headers.append(('Content-Type', mime_type,))
+        start_response('200 OK', headers)
+
+    return [content]

apps/cic-cache/cic_cache/runnable/daemons/tasker.py

@@ -0,0 +1,98 @@
+# standard imports
+import logging
+import os
+import sys
+import argparse
+
+# third-party imports
+import celery
+import confini
+
+# local imports
+from cic_cache.db import dsn_from_config
+from cic_cache.db.models.base import SessionBase
+from cic_cache.tasks.tx import *
+
+logging.basicConfig(level=logging.WARNING)
+logg = logging.getLogger()
+
+config_dir = os.path.join('/usr/local/etc/cic-cache')
+
+
+argparser = argparse.ArgumentParser()
+argparser.add_argument('-c', type=str, default=config_dir, help='config file')
+argparser.add_argument('-q', type=str, default='cic-cache', help='queue name for worker tasks')
+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('-v', action='store_true', help='be verbose')
+argparser.add_argument('-vv', action='store_true', help='be more verbose')
+
+args = argparser.parse_args()
+
+if args.vv:
+    logging.getLogger().setLevel(logging.DEBUG)
+elif args.v:
+    logging.getLogger().setLevel(logging.INFO)
+
+config = confini.Config(args.c, args.env_prefix)
+config.process()
+
+# connect to database
+dsn = dsn_from_config(config)
+SessionBase.connect(dsn)
+
+# verify database connection with minimal sanity query
+#session = SessionBase.create_session()
+#session.execute('select version_num from alembic_version')
+#session.close()
+
+# set up celery
+current_app = celery.Celery(__name__)
+
+broker = config.get('CELERY_BROKER_URL')
+if broker[:4] == 'file':
+    bq = tempfile.mkdtemp()
+    bp = tempfile.mkdtemp()
+    current_app.conf.update({
+            'broker_url': broker,
+            'broker_transport_options': {
+                'data_folder_in': bq,
+                'data_folder_out': bq,
+                'data_folder_processed': bp,
+            },
+            },
+            )
+    logg.warning('celery broker dirs queue i/o {} processed {}, will NOT be deleted on shutdown'.format(bq, bp))
+else:
+    current_app.conf.update({
+        'broker_url': broker,
+        })
+
+result = config.get('CELERY_RESULT_URL')
+if result[:4] == 'file':
+    rq = tempfile.mkdtemp()
+    current_app.conf.update({
+        'result_backend': 'file://{}'.format(rq),
+        })
+    logg.warning('celery backend store dir {} created, will NOT be deleted on shutdown'.format(rq))
+else:
+    current_app.conf.update({
+        'result_backend': result,
+        })
+
+
+def main():
+    argv = ['worker']
+    if args.vv:
+        argv.append('--loglevel=DEBUG')
+    elif args.v:
+        argv.append('--loglevel=INFO')
+    argv.append('-Q')
+    argv.append(args.q)
+    argv.append('-n')
+    argv.append(args.q)
+
+    current_app.worker_main(argv)
+
+
+if __name__ == '__main__':
+    main()

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

@@ -0,0 +1,112 @@
+# 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 cic_base.config
+import cic_base.log
+import cic_base.argparse
+import cic_base.rpc
+from cic_eth_registry import CICRegistry
+from cic_eth_registry.error import UnknownContractError
+from chainlib.chain import ChainSpec
+from chainlib.eth.constant import ZERO_ADDRESS
+from chainlib.connection import RPCConnection
+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_cache.db import dsn_from_config
+from cic_cache.runnable.daemons.filters import (
+        ERC20TransferFilter,
+        )
+
+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)
+
+cic_base.config.log(config)
+
+dsn = dsn_from_config(config)
+
+SessionBase.connect(dsn, debug=config.true('DATABASE_DEBUG'))
+
+chain_spec = ChainSpec.from_chain_str(config.get('CIC_CHAIN_SPEC'))
+
+#RPCConnection.register_location(config.get('ETH_PROVIDER'), chain_spec, 'default')
+cic_base.rpc.setup(chain_spec, config.get('ETH_PROVIDER'))
+
+
+def main():
+    # Connect to blockchain with chainlib
+    rpc = RPCConnection.connect(chain_spec, 'default')
+
+    o = block_latest()
+    r = rpc.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')
+        syncers.append(SyncerBackend.initial(chain_spec, block_offset))
+    else:
+        for syncer_backend in syncer_backends:
+            logg.info('resuming sync session {}'.format(syncer_backend))
+
+    for syncer_backend in syncer_backends:
+        syncers.append(HistorySyncer(syncer_backend))
+
+    syncer_backend = SyncerBackend.live(chain_spec, block_offset+1)
+    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))
+
+    erc20_transfer_filter = ERC20TransferFilter(chain_spec)
+
+    i = 0
+    for syncer in syncers:
+        logg.debug('running syncer index {}'.format(i))
+        syncer.add_filter(erc20_transfer_filter)
+        r = syncer.loop(int(config.get('SYNCER_LOOP_INTERVAL')), rpc)
+        sys.stderr.write("sync {} done at block {}\n".format(syncer, r))
+
+        i += 1
+
+
+if __name__ == '__main__':
+    main()

apps/cic-cache/cic_cache/tasks/tx.py

@@ -0,0 +1,38 @@
+# third-party imports
+import celery
+
+# local imports
+from cic_cache.cache import BloomCache
+from cic_cache.db.models.base import SessionBase
+
+celery_app = celery.current_app
+
+
+@celery_app.task(bind=True)
+def tx_filter(self, offset, limit, address=None, encoding='hex'):
+    queue = self.request.delivery_info.get('routing_key')
+
+    session = SessionBase.create_session()
+
+    c = BloomCache(session)
+    b = None
+    if address == None:
+        (lowest_block, highest_block, bloom_filter_block, bloom_filter_tx) = c.load_transactions(offset, limit)
+    else:
+        (lowest_block, highest_block, bloom_filter_block, bloom_filter_tx) = c.load_transactions_account(address, offset, limit)
+
+    session.close()
+
+    o = {
+        'alg': 'sha256',
+        'low': lowest_block,
+        'high': highest_block,
+        'block_filter': bloom_filter_block.hex(), 
+        'blocktx_filter': bloom_filter_tx.hex(),
+        'filter_rounds': 3,
+            }
+
+    return o
+
+
+

apps/cic-cache/cic_cache/version.py

@@ -0,0 +1,18 @@
+import os
+import semver
+
+version = (
+        0,
+        2,
+        0,
+        'alpha.2',
+        )
+
+version_object = semver.VersionInfo(
+        major=version[0],
+        minor=version[1],
+        patch=version[2],
+        prerelease=version[3],
+        )
+
+version_string = str(version_object)

apps/cic-cache/config/bancor.ini

@@ -0,0 +1,2 @@
+[bancor]
+dir =

apps/cic-cache/config/celery.ini

@@ -0,0 +1,3 @@
+[celery]
+broker_url = redis:///
+result_url = redis:///

apps/cic-cache/config/cic.ini

@@ -0,0 +1,4 @@
+[cic]
+registry_address = 
+chain_spec =
+trust_address = 

apps/cic-cache/config/database.ini

@@ -0,0 +1,9 @@
+[database]
+NAME=cic_cache
+USER=postgres
+PASSWORD=
+HOST=localhost
+PORT=5432
+ENGINE=postgresql
+DRIVER=psycopg2
+DEBUG=

apps/cic-cache/config/docker/bancor.ini

@@ -0,0 +1,3 @@
+[bancor]
+registry_address = 
+dir = /usr/local/share/bancor

apps/cic-cache/config/docker/celery.ini

@@ -0,0 +1,3 @@
+[celery]
+broker_url = redis://localhost:63379
+result_url = redis://localhost:63379

apps/cic-cache/config/docker/cic.ini

@@ -0,0 +1,4 @@
+[cic]
+chain_spec =
+registry_address =  
+trust_address = 0xEb3907eCad74a0013c259D5874AE7f22DcBcC95C

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

@@ -0,0 +1,9 @@
+[database]
+NAME=cic_cache
+USER=grassroots
+PASSWORD=
+HOST=localhost
+PORT=63432
+ENGINE=postgresql
+DRIVER=psycopg2
+DEBUG=0

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

@@ -0,0 +1,2 @@
+[eth]
+provider = ws://localhost:63546

apps/cic-cache/config/docker/syncer.ini

@@ -0,0 +1,2 @@
+[syncer]
+loop_interval = 1

apps/cic-cache/config/eth.ini

@@ -0,0 +1,2 @@
+[eth]
+provider = ws://localhost:8545

apps/cic-cache/config/syncer.ini

@@ -0,0 +1,2 @@
+[syncer]
+loop_interval = 5

apps/cic-cache/config/test/bancor.ini

@@ -0,0 +1,2 @@
+[bancor]
+dir =

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

@@ -0,0 +1,2 @@
+[cic]
+registry_address = 

apps/cic-cache/config/test/database.ini

@@ -0,0 +1,9 @@
+[database]
+NAME=cic-cache-test
+USER=postgres
+PASSWORD=
+HOST=localhost
+PORT=5432
+ENGINE=sqlite
+DRIVER=pysqlite
+DEBUG=

apps/cic-cache/config/test/eth.ini

@@ -0,0 +1,5 @@
+[eth]
+#ws_provider = ws://localhost:8546
+#ttp_provider = http://localhost:8545
+provider = http://localhost:8545
+#chain_id = 

apps/cic-cache/db/psycopg2/db.sql

@@ -0,0 +1,22 @@
+CREATE TABLE tx (
+	id SERIAL PRIMARY KEY,
+	date_registered TIMESTAMP NOT NULL default CURRENT_TIMESTAMP,
+	block_number INTEGER NOT NULL,
+	tx_index INTEGER NOT NULL,
+	tx_hash VARCHAR(66) NOT NULL,
+	sender VARCHAR(42) NOT NULL, 
+	recipient VARCHAR(42) NOT NULL, 
+	source_token VARCHAR(42) NOT NULL,
+	destination_token VARCHAR(42) NOT NULL,
+	from_value BIGINT NOT NULL,
+	to_value BIGINT NOT NULL,
+	success BOOLEAN NOT NULL, 
+	date_block TIMESTAMP NOT NULL
+);
+
+CREATE TABLE tx_sync (
+	id SERIAL PRIMARY KEY,
+	tx VARCHAR(66) NOT NULL
+);
+
+INSERT INTO tx_sync (tx) VALUES('0x0000000000000000000000000000000000000000000000000000000000000000');

apps/cic-cache/db/pysqlite/db.sql

@@ -0,0 +1,23 @@
+CREATE TABLE tx (
+	id SERIAL PRIMARY KEY,
+	date_registered DATETIME NOT NULL default CURRENT_DATE, 
+	block_number INTEGER NOT NULL,
+	tx_index INTEGER NOT NULL,
+	tx_hash VARCHAR(66) NOT NULL,
+	sender VARCHAR(42) NOT NULL, 
+	recipient VARCHAR(42) NOT NULL, 
+	source_token VARCHAR(42) NOT NULL,
+	destination_token VARCHAR(42) NOT NULL,
+	from_value INTEGER NOT NULL,
+	to_value INTEGER NOT NULL,
+	success BOOLEAN NOT NULL, 
+	date_block DATETIME NOT NULL,
+	CHECK (success IN (0, 1))
+);
+
+CREATE TABLE tx_sync (
+	id SERIAL PRIMARY_KEY,
+	tx VARCHAR(66) NOT NULL
+);
+
+INSERT INTO tx_sync (tx) VALUES('0x0000000000000000000000000000000000000000000000000000000000000000');

apps/cic-cache/doc/openapi/server.yml

@@ -0,0 +1,102 @@
+openapi: "3.0.3"
+info:
+  title: Grassroots Economics CIC Cache
+  description: Cache of processed transaction data from Ethereum blockchain and worker queues
+  termsOfService: bzz://grassrootseconomics.eth/terms
+  contact:
+    name: Grassroots Economics
+    url: https://www.grassrootseconomics.org
+    email: will@grassecon.org
+  license:
+    name: GPLv3
+  version: 0.1.0
+
+paths:
+  /tx/{offset}/{limit}:
+    description: Bloom filter for batch of latest transactions
+    get:
+      tags:
+        - transactions
+      description:
+        Retrieve transactions
+      operationId: tx.get
+      responses:
+        200:
+          description: Transaction query successful.
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/BlocksBloom"
+
+
+    parameters:
+    - name: offset
+      in: path
+      schema:
+        type: integer
+        format: int32
+    - name: limit
+      in: path
+      schema:
+        type: integer
+        format: int32
+
+
+  /tx/{address}/{offset}/{limit}:
+    description: Bloom filter for batch of latest transactions by account
+    get:
+      tags:
+        - transactions
+      description:
+        Retrieve transactions
+      operationId: tx.get
+      responses:
+        200:
+          description: Transaction query successful. 
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/BlocksBloom"
+
+
+    parameters:
+    - name: address
+      in: path
+      required: true
+      schema:
+        type: string
+    - name: offset
+      in: path
+      schema:
+        type: integer
+        format: int32
+    - name: limit
+      in: path
+      schema:
+        type: integer
+        format: int32
+
+components:
+  schemas:
+    BlocksBloom:
+      type: object
+      properties:
+        low:
+          type: int
+          format: int32
+          description: The lowest block number included in the filter
+        block_filter:
+          type: string
+          format: byte
+          description: Block number filter
+        blocktx_filter:
+          type: string
+          format: byte
+          description: Block and tx index filter
+        alg:
+          type: string
+          description: Hashing algorithm (currently only using sha256)
+        filter_rounds:
+          type: int
+          format: int32
+          description: Number of hash rounds used to create the filter

apps/cic-cache/docker/Dockerfile

@@ -0,0 +1,53 @@
+FROM python:3.8.6-slim-buster
+
+#COPY --from=0 /usr/local/share/cic/solidity/ /usr/local/share/cic/solidity/
+
+WORKDIR /usr/src/cic-cache
+
+ARG pip_extra_index_url_flag='--index https://pypi.org/simple --extra-index-url https://pip.grassrootseconomics.net:8433'
+ARG root_requirement_file='requirements.txt'
+
+#RUN apk update && \
+#	apk add gcc musl-dev gnupg libpq
+#RUN apk add postgresql-dev
+#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 git
+
+# Copy shared requirements from top of mono-repo
+RUN echo "copying root req file ${root_requirement_file}"
+RUN pip install $pip_extra_index_url_flag cic-base[full_graph]==0.1.2a58
+
+COPY cic-cache/requirements.txt ./
+COPY cic-cache/setup.cfg \
+	 cic-cache/setup.py \
+ 	 ./
+COPY cic-cache/cic_cache/ ./cic_cache/
+COPY cic-cache/scripts/ ./scripts/
+COPY cic-cache/test_requirements.txt ./
+RUN pip install $pip_extra_index_url_flag -r test_requirements.txt
+RUN pip install $pip_extra_index_url_flag .
+RUN pip install .[server]
+
+COPY cic-cache/tests/ ./tests/
+#COPY db/ cic-cache/db
+#RUN apk add postgresql-client
+
+# ini files in config directory defines the configurable parameters for the application
+# they can all be overridden by environment variables
+# to generate a list of environment variables from configuration, use: confini-dump -z <dir> (executable provided by confini package)
+COPY cic-cache/config/ /usr/local/etc/cic-cache/
+
+# for db migrations
+RUN git clone https://github.com/vishnubob/wait-for-it.git /usr/local/bin/wait-for-it/
+COPY cic-cache/cic_cache/db/migrations/ /usr/local/share/cic-cache/alembic/
+
+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
+
+# Tracker
+# ENTRYPOINT ["/usr/local/bin/cic-cache-tracker", "-vv"]
+# Server
+# ENTRYPOINT [ "/usr/local/bin/uwsgi", "--wsgi-file", "/usr/local/lib/python3.8/site-packages/cic_cache/runnable/server.py", "--http", ":80", "--pyargv", "-vv" ]

apps/cic-cache/examples/bloom_client/parse.js

@@ -0,0 +1,40 @@
+let xmlhttprequest = require('xhr2');
+let moolb = require('moolb');
+
+let xhr = new xmlhttprequest();
+xhr.responseType = 'json';
+xhr.open('GET', 'http://localhost:5555/tx/0/100');
+xhr.addEventListener('load', (e) => {
+
+	d = xhr.response;
+
+	b_one = Buffer.from(d.block_filter, 'base64');
+	b_two = Buffer.from(d.blocktx_filter, 'base64');
+
+	for (let i = 0; i < 8192; i++) {
+		if (b_two[i] > 0) {
+			console.debug('value on', i, b_two[i]);
+		}
+	}
+	console.log(b_one, b_two);
+
+	let f_block = moolb.fromBytes(b_one, d.filter_rounds);
+	let f_blocktx = moolb.fromBytes(b_two, d.filter_rounds);
+	let a = new ArrayBuffer(8);
+	let w = new DataView(a);
+	for (let i = 410000; i < 430000; i++) {
+		w.setInt32(0, i);
+		let r = new Uint8Array(a.slice(0, 4));
+		if (f_block.check(r)) {
+			for (let j = 0; j < 200; j++) {
+				w = new DataView(a);
+				w.setInt32(4, j);
+				r = new Uint8Array(a);
+				if (f_blocktx.check(r)) {
+					console.log('true', i, j);
+				}
+			}
+		}
+	}
+});
+let r = xhr.send();

apps/cic-cache/requirements.txt

@@ -0,0 +1,13 @@
+cic-base~=0.1.2a66
+alembic==1.4.2
+confini~=0.3.6rc3
+uwsgi==2.0.19.1
+moolb~=0.1.0
+cic-eth-registry~=0.5.4a13
+SQLAlchemy==1.3.20
+semver==2.13.0
+psycopg2==2.8.6
+celery==4.4.7
+redis==3.5.3
+chainlib~=0.0.2a10
+chainsyncer[sql]~=0.0.2a1

apps/cic-cache/scripts/migrate.py

@@ -0,0 +1,56 @@
+#!/usr/bin/python
+import os
+import argparse
+import logging
+
+import alembic
+from alembic.config import Config as AlembicConfig
+import confini
+
+from cic_cache.db import dsn_from_config
+
+logging.basicConfig(level=logging.WARNING)
+logg = logging.getLogger()
+
+# BUG: the dbdir doesn't work after script install
+rootdir = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
+dbdir = os.path.join(rootdir, 'cic_cache', 'db')
+migrationsdir = os.path.join(dbdir, 'migrations')
+
+config_dir = os.path.join('/usr/local/etc/cic-cache')
+
+argparser = argparse.ArgumentParser()
+argparser.add_argument('-c', type=str, default=config_dir, help='config file')
+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('--migrations-dir', dest='migrations_dir', default=migrationsdir, type=str, help='path to alembic migrations directory')
+argparser.add_argument('-v', action='store_true', help='be verbose')
+argparser.add_argument('-vv', action='store_true', help='be more verbose')
+args = argparser.parse_args()
+
+if args.vv:
+    logging.getLogger().setLevel(logging.DEBUG)
+elif args.v:
+    logging.getLogger().setLevel(logging.INFO)
+
+config = confini.Config(args.c, args.env_prefix)
+config.process()
+config.censor('PASSWORD', 'DATABASE')
+config.censor('PASSWORD', 'SSL')
+logg.debug('config:\n{}'.format(config))
+
+migrations_dir = os.path.join(args.migrations_dir, config.get('DATABASE_ENGINE'))
+if not os.path.isdir(migrations_dir):
+    logg.debug('migrations dir for engine {} not found, reverting to default'.format(config.get('DATABASE_ENGINE')))
+    migrations_dir = os.path.join(args.migrations_dir, 'default')
+
+# connect to database
+dsn = dsn_from_config(config)
+
+
+logg.info('using migrations dir {}'.format(migrations_dir))
+logg.info('using db {}'.format(dsn))
+ac = AlembicConfig(os.path.join(migrations_dir, 'alembic.ini'))
+ac.set_main_option('sqlalchemy.url', dsn)
+ac.set_main_option('script_location', migrations_dir)
+
+alembic.command.upgrade(ac, 'head')

apps/cic-cache/setup.cfg

@@ -0,0 +1,41 @@
+[metadata]
+name = cic-cache
+description = CIC Cache API and server
+author = Louis Holbrook
+author_email = dev@holbrook.no
+url = https://gitlab.com/grassrootseconomics/cic-eth
+keywords =
+	cic
+	cryptocurrency
+	ethereum
+classifiers =
+	Programming Language :: Python :: 3
+	Operating System :: OS Independent
+	Development Status :: 3 - Alpha
+	Environment :: No Input/Output (Daemon)
+	Intended Audience :: Developers
+	License :: OSI Approved :: GNU General Public License v3 or later (GPLv3+)
+	Topic :: Internet
+#	Topic :: Blockchain :: EVM
+license = GPL3
+licence_files =
+	LICENSE.txt
+
+[options]
+python_requires = >= 3.6
+packages = 
+	cic_cache
+	cic_cache.tasks
+	cic_cache.db
+	cic_cache.db.models
+	cic_cache.runnable
+	cic_cache.runnable.daemons
+	cic_cache.runnable.daemons.filters
+scripts =
+	./scripts/migrate.py
+
+[options.entry_points]
+console_scripts =
+	cic-cache-trackerd = cic_cache.runnable.daemons.tracker:main
+	cic-cache-serverd = cic_cache.runnable.daemons.server:main
+	cic-cache-taskerd = cic_cache.runnable.daemons.tasker:main

apps/cic-cache/setup.py

@@ -0,0 +1,60 @@
+from setuptools import setup
+
+import configparser
+import os
+import time
+
+from cic_cache.version import (
+        version_object,
+        version_string
+        )
+
+class PleaseCommitFirstError(Exception):
+    pass
+
+def git_hash():
+    import subprocess
+    git_diff = subprocess.run(['git', 'diff'], capture_output=True)
+    if len(git_diff.stdout) > 0:
+        raise PleaseCommitFirstError()
+    git_hash = subprocess.run(['git', 'rev-parse', 'HEAD'], capture_output=True)
+    git_hash_brief = git_hash.stdout.decode('utf-8')[:8]
+    return git_hash_brief
+
+version_string = str(version_object)
+
+try:
+    version_git = git_hash()
+    version_string += '+build.{}'.format(version_git)
+except FileNotFoundError:
+    time_string_pair = str(time.time()).split('.')
+    version_string += '+build.{}{:<09d}'.format(
+            time_string_pair[0],
+            int(time_string_pair[1]),
+            )
+print('final version string will be {}'.format(version_string))
+
+requirements = []
+f = open('requirements.txt', 'r')
+while True:
+    l = f.readline()
+    if l == '':
+        break
+    requirements.append(l.rstrip())
+f.close()
+
+test_requirements = []
+f = open('test_requirements.txt', 'r')
+while True:
+    l = f.readline()
+    if l == '':
+        break
+    test_requirements.append(l.rstrip())
+f.close()
+
+
+setup(
+    version=version_string,
+    install_requires=requirements,
+    tests_require=test_requirements,
+        )

apps/cic-cache/test_requirements.txt

@@ -0,0 +1,6 @@
+pytest==6.0.1
+pytest-cov==2.10.1
+pytest-mock==3.3.1
+pysqlite3==0.4.3
+sqlparse==0.4.1
+pytest-celery==0.0.0a1

apps/cic-cache/tests/conftest.py

@@ -0,0 +1,86 @@
+# standard imports
+import os
+import sys
+import datetime
+
+# third-party imports
+import pytest
+
+# local imports
+from cic_cache import db
+
+script_dir = os.path.dirname(os.path.realpath(__file__))
+root_dir = os.path.dirname(script_dir)
+sys.path.insert(0, root_dir)
+
+# fixtures
+from tests.fixtures_config import *
+from tests.fixtures_database import *
+from tests.fixtures_celery import *
+
+
+@pytest.fixture(scope='session')
+def balances_dict_fields():
+    return {
+            'out_pending': 0,
+            'out_synced': 1,
+            'out_confirmed': 2,
+            'in_pending': 3,
+            'in_synced': 4,
+            'in_confirmed': 5,
+    }
+
+
+@pytest.fixture(scope='function')
+def txs(
+        init_database,
+        list_defaults,
+        list_actors,
+        list_tokens,
+        ):
+
+    session = init_database
+
+    tx_number = 13 
+    tx_hash_first = '0x' + os.urandom(32).hex()
+    val = 15000
+    nonce = 1
+    dt = datetime.datetime.utcnow()
+    db.add_transaction(
+        session,
+        tx_hash_first,
+        list_defaults['block'],
+        tx_number,
+        list_actors['alice'],
+        list_actors['bob'],
+        list_tokens['foo'],
+        list_tokens['foo'],
+        1024,
+        2048,
+        True,
+        dt.timestamp(),
+            )
+
+
+    tx_number = 42
+    tx_hash_second = '0x' + os.urandom(32).hex()
+    tx_signed_second = '0x' + os.urandom(128).hex()
+    nonce = 1
+    dt -= datetime.timedelta(hours=1)
+    db.add_transaction(
+        session,
+        tx_hash_second,
+        list_defaults['block']-1,
+        tx_number,
+        list_actors['diane'],
+        list_actors['alice'],
+        list_tokens['foo'],
+        list_tokens['foo'],
+        1024,
+        2048,
+        False,
+        dt.timestamp(),
+        )
+    
+    session.commit()
+

apps/cic-cache/tests/fixtures_celery.py

@@ -0,0 +1,48 @@
+# third-party imports
+import pytest
+import tempfile
+import logging
+import shutil
+
+logg = logging.getLogger(__name__)
+
+
+# celery fixtures
+@pytest.fixture(scope='session')
+def celery_includes():
+    return [
+        'cic_cache.tasks.tx',
+    ]
+
+
+@pytest.fixture(scope='session')
+def celery_config():
+    bq = tempfile.mkdtemp()
+    bp = tempfile.mkdtemp()
+    rq = tempfile.mkdtemp()
+    logg.debug('celery broker queue {} processed {}'.format(bq, bp))
+    logg.debug('celery backend store {}'.format(rq))
+    yield {
+        'broker_url': 'filesystem://',
+        'broker_transport_options': {
+            'data_folder_in': bq,
+            'data_folder_out': bq,
+            'data_folder_processed': bp,
+            },
+        'result_backend': 'file://{}'.format(rq),
+            }
+    logg.debug('cleaning up celery filesystem backend files {} {} {}'.format(bq, bp, rq))
+    shutil.rmtree(bq)
+    shutil.rmtree(bp)
+    shutil.rmtree(rq)
+
+
+@pytest.fixture(scope='session')
+def celery_worker_parameters():
+    return {
+#            'queues': ('cic-cache'),
+            }
+
+@pytest.fixture(scope='session')
+def celery_enable_logging():
+    return True

apps/cic-cache/tests/fixtures_config.py

@@ -0,0 +1,20 @@
+# standard imports
+import os
+import logging
+
+# third-party imports
+import pytest
+import confini
+
+script_dir = os.path.dirname(os.path.realpath(__file__))
+root_dir = os.path.dirname(script_dir)
+logg = logging.getLogger(__file__)
+
+
+@pytest.fixture(scope='session')
+def load_config():
+    config_dir = os.path.join(root_dir, '.config/test')
+    conf = confini.Config(config_dir, 'CICTEST')
+    conf.process()
+    logg.debug('config {}'.format(conf))
+    return conf

apps/cic-cache/tests/fixtures_database.py

@@ -0,0 +1,118 @@
+# standard imports
+import os
+import logging
+import re
+
+# third-party imports
+import pytest
+import sqlparse
+
+# local imports
+from cic_cache.db.models.base import SessionBase
+from cic_cache.db import dsn_from_config
+
+logg = logging.getLogger(__file__)
+
+
+@pytest.fixture(scope='function')
+def database_engine(
+    load_config,
+        ):
+    if load_config.get('DATABASE_ENGINE') == 'sqlite':
+        SessionBase.transactional = False
+        SessionBase.poolable = False
+        try:
+            os.unlink(load_config.get('DATABASE_NAME'))
+        except FileNotFoundError:
+            pass
+    dsn = dsn_from_config(load_config)
+    SessionBase.connect(dsn)
+    return dsn
+
+
+# TODO: use alembic instead to migrate db, here we have to keep separate schema than migration script in script/migrate.py
+@pytest.fixture(scope='function')
+def init_database(
+        load_config,
+        database_engine,
+        ):
+
+    rootdir = os.path.dirname(os.path.dirname(__file__))
+    schemadir = os.path.join(rootdir, 'db', load_config.get('DATABASE_DRIVER'))
+
+    if load_config.get('DATABASE_ENGINE') == 'sqlite':
+        rconn = SessionBase.engine.raw_connection()
+        f = open(os.path.join(schemadir, 'db.sql'))
+        s = f.read()
+        f.close()
+        rconn.executescript(s)
+
+    else:
+        rconn = SessionBase.engine.raw_connection()
+        rcursor = rconn.cursor()
+
+        #rcursor.execute('DROP FUNCTION IF EXISTS public.transaction_list')
+        #rcursor.execute('DROP FUNCTION IF EXISTS public.balances')
+
+        f = open(os.path.join(schemadir, 'db.sql'))
+        s = f.read()
+        f.close()
+        r = re.compile(r'^[A-Z]', re.MULTILINE)
+        for l in sqlparse.parse(s):
+            strl = str(l)
+            # we need to check for empty query lines, as sqlparse doesn't do that on its own (and psycopg complains when it gets them)
+            if not re.search(r, strl):
+                logg.warning('skipping parsed query line {}'.format(strl))
+                continue
+            rcursor.execute(strl)
+        rconn.commit()
+
+        rcursor.execute('SET search_path TO public')
+
+# this doesn't work when run separately, no idea why
+# functions have been manually added to original schema from cic-eth 
+#        f = open(os.path.join(schemadir, 'proc_transaction_list.sql'))
+#        s = f.read()
+#        f.close()
+#        rcursor.execute(s)
+# 
+#        f = open(os.path.join(schemadir, 'proc_balances.sql'))
+#        s = f.read()
+#        f.close()
+#        rcursor.execute(s)
+
+        rcursor.close()
+
+    session = SessionBase.create_session()
+    yield session
+    session.commit()
+    session.close()
+
+
+@pytest.fixture(scope='function')
+def list_tokens(
+        ):
+    return {
+            'foo': '0x' + os.urandom(20).hex(),
+            'bar': '0x' + os.urandom(20).hex(),
+        }
+
+
+@pytest.fixture(scope='function')
+def list_actors(
+        ):
+    return {
+            'alice': '0x' + os.urandom(20).hex(),
+            'bob': '0x' + os.urandom(20).hex(),
+            'charlie': '0x' + os.urandom(20).hex(),
+            'diane': '0x' + os.urandom(20).hex(),
+            }
+
+
+@pytest.fixture(scope='function')
+def list_defaults(
+        ):
+
+    return {
+        'block': 420000,
+    }

apps/cic-cache/tests/test_cache.py

@@ -0,0 +1,35 @@
+# standard imports
+import os
+import datetime
+import logging
+import json
+
+# third-party imports
+import pytest
+
+# local imports
+from cic_cache import BloomCache
+
+logg = logging.getLogger()
+
+
+def test_cache(
+        init_database,
+        list_defaults,
+        list_actors,
+        list_tokens,
+        txs,
+        ):
+
+    session = init_database
+
+    c = BloomCache(session)
+    b = c.load_transactions(0, 100)
+
+    assert b[0] == list_defaults['block'] - 1
+
+    c = BloomCache(session)
+    c.load_transactions_account(list_actors['alice'],0, 100)
+
+    assert b[0] == list_defaults['block'] - 1
+

apps/cic-cache/tests/test_task.py

@@ -0,0 +1,27 @@
+# standard imports
+import logging
+
+# third-party imports
+import celery
+
+# local imports
+from cic_cache.api import Api
+
+logg = logging.getLogger()
+
+
+def test_task(
+        init_database,
+        list_defaults,
+        list_actors,
+        list_tokens,
+        txs,
+        celery_session_worker,
+        ):
+
+    api = Api(queue=None)
+    t = api.list(0, 100)
+    r = t.get()
+    logg.debug('r {}'.format(r))
+
+    assert r['low'] == list_defaults['block'] - 1

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

@@ -4,18 +4,23 @@ import logging
 
 # third-party imports
 import celery
-from cic_registry import zero_address
+from chainlib.eth.constant import ZERO_ADDRESS
+from chainlib.chain import ChainSpec
 
 # 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.task import (
+        CriticalSQLAlchemyTask,
+        )
 from cic_eth.error import LockedError
 
 celery_app = celery.current_app
 logg = logging.getLogger()
 
-@celery_app.task()
-def lock(chained_input, chain_str, address=zero_address, flags=LockEnum.ALL, tx_hash=None):
+@celery_app.task(base=CriticalSQLAlchemyTask)
+def lock(chained_input, chain_spec_dict, address=ZERO_ADDRESS, flags=LockEnum.ALL, tx_hash=None):
     """Task wrapper to set arbitrary locks
 
     :param chain_str: Chain spec string representation
@@ -27,13 +32,14 @@ def lock(chained_input, chain_str, address=zero_address, flags=LockEnum.ALL, tx_
     :returns: New lock state for address
     :rtype: number
     """
+    chain_str = str(ChainSpec.from_dict(chain_spec_dict))
     r = Lock.set(chain_str, flags, address=address, tx_hash=tx_hash)
     logg.debug('Locked {} for {}, flag now {}'.format(flags, address, r))
     return chained_input
 
 
-@celery_app.task()
-def unlock(chained_input, chain_str, address=zero_address, flags=LockEnum.ALL):
+@celery_app.task(base=CriticalSQLAlchemyTask)
+def unlock(chained_input, chain_spec_dict, address=ZERO_ADDRESS, flags=LockEnum.ALL):
     """Task wrapper to reset arbitrary locks
 
     :param chain_str: Chain spec string representation
@@ -45,13 +51,14 @@ def unlock(chained_input, chain_str, address=zero_address, flags=LockEnum.ALL):
     :returns: New lock state for address
     :rtype: number
     """
+    chain_str = str(ChainSpec.from_dict(chain_spec_dict))
     r = Lock.reset(chain_str, flags, address=address)
     logg.debug('Unlocked {} for {}, flag now {}'.format(flags, address, r))
     return chained_input
 
 
-@celery_app.task()
-def lock_send(chained_input, chain_str, address=zero_address, tx_hash=None):
+@celery_app.task(base=CriticalSQLAlchemyTask)
+def lock_send(chained_input, chain_spec_dict, address=ZERO_ADDRESS, tx_hash=None):
     """Task wrapper to set send lock
 
     :param chain_str: Chain spec string representation
@@ -61,13 +68,14 @@ def lock_send(chained_input, chain_str, address=zero_address, tx_hash=None):
     :returns: New lock state for address
     :rtype: number
     """
+    chain_str = str(ChainSpec.from_dict(chain_spec_dict))
     r = Lock.set(chain_str, LockEnum.SEND, address=address, tx_hash=tx_hash)
     logg.debug('Send locked for {}, flag now {}'.format(address, r))
     return chained_input
 
 
-@celery_app.task()
-def unlock_send(chained_input, chain_str, address=zero_address):
+@celery_app.task(base=CriticalSQLAlchemyTask)
+def unlock_send(chained_input, chain_spec_dict, address=ZERO_ADDRESS):
     """Task wrapper to reset send lock
 
     :param chain_str: Chain spec string representation
@@ -77,13 +85,14 @@ def unlock_send(chained_input, chain_str, address=zero_address):
     :returns: New lock state for address
     :rtype: number
     """
+    chain_str = str(ChainSpec.from_dict(chain_spec_dict))
     r = Lock.reset(chain_str, LockEnum.SEND, address=address)
     logg.debug('Send unlocked for {}, flag now {}'.format(address, r))
     return chained_input
 
 
-@celery_app.task()
-def lock_queue(chained_input, chain_str, address=zero_address, tx_hash=None):
+@celery_app.task(base=CriticalSQLAlchemyTask)
+def lock_queue(chained_input, chain_spec_dict, address=ZERO_ADDRESS, tx_hash=None):
     """Task wrapper to set queue direct lock
 
     :param chain_str: Chain spec string representation
@@ -93,13 +102,14 @@ def lock_queue(chained_input, chain_str, address=zero_address, tx_hash=None):
     :returns: New lock state for address
     :rtype: number
     """
+    chain_str = str(ChainSpec.from_dict(chain_spec_dict))
     r = Lock.set(chain_str, LockEnum.QUEUE, address=address, tx_hash=tx_hash)
     logg.debug('Queue direct locked for {}, flag now {}'.format(address, r))
     return chained_input
 
 
-@celery_app.task()
-def unlock_queue(chained_input, chain_str, address=zero_address):
+@celery_app.task(base=CriticalSQLAlchemyTask)
+def unlock_queue(chained_input, chain_spec_dict, address=ZERO_ADDRESS):
     """Task wrapper to reset queue direct lock
 
     :param chain_str: Chain spec string representation
@@ -109,17 +119,23 @@ def unlock_queue(chained_input, chain_str, address=zero_address):
     :returns: New lock state for address
     :rtype: number
     """
+    chain_str = str(ChainSpec.from_dict(chain_spec_dict))
     r = Lock.reset(chain_str, LockEnum.QUEUE, address=address)
     logg.debug('Queue direct unlocked for {}, flag now {}'.format(address, r))
     return chained_input
 
 
-@celery_app.task()
-def check_lock(chained_input, chain_str, lock_flags, address=None):
-    r = Lock.check(chain_str, lock_flags, address=zero_address)
+@celery_app.task(base=CriticalSQLAlchemyTask)
+def check_lock(chained_input, chain_spec_dict, lock_flags, address=None):
+    chain_str = str(ChainSpec.from_dict(chain_spec_dict))
+    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))
+        session.close()
         raise LockedError(r)
+    session.flush()
+    session.close()
     return chained_input

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/admin/nonce.py

@@ -1,25 +1,26 @@
 # standard imports
 import logging
 
-# third-party imports
+# external imports
 import celery
-from cic_registry.chain import ChainSpec
+from chainlib.chain import ChainSpec
+from chainlib.eth.tx import unpack
+from chainqueue.query import get_tx
+from chainqueue.state import set_cancel
+from chainqueue.db.models.otx import Otx
+from chainqueue.db.models.tx import TxCache
 
 # local imports
 from cic_eth.db.models.base import SessionBase
-from cic_eth.db.models.otx import Otx
-from cic_eth.db.models.tx import TxCache
 from cic_eth.db.models.nonce import Nonce
-from cic_eth.admin.ctrl import lock_send
-from cic_eth.admin.ctrl import unlock_send
-from cic_eth.admin.ctrl import lock_queue
-from cic_eth.admin.ctrl import unlock_queue
-from cic_eth.queue.tx import get_tx
-from cic_eth.queue.tx import set_cancel
-from cic_eth.queue.tx import create as queue_create
-from cic_eth.eth.util import unpack_signed_raw_tx
-from cic_eth.eth.task import sign_tx
-from cic_eth.eth.task import create_check_gas_and_send_task
+from cic_eth.admin.ctrl import (
+        lock_send,
+        unlock_send,
+        lock_queue,
+        unlock_queue,
+        )
+from cic_eth.queue.tx import queue_create
+from cic_eth.eth.gas import create_check_gas_task
 
 celery_app = celery.current_app
 logg = logging.getLogger()
@@ -45,8 +46,8 @@ def shift_nonce(self, chain_str, tx_hash_orig_hex, delta=1):
 
     chain_spec = ChainSpec.from_chain_str(chain_str)
     tx_brief = get_tx(tx_hash_orig_hex)
-    tx_raw = bytes.fromhex(tx_brief['signed_tx'][2:])
-    tx = unpack_signed_raw_tx(tx_raw, chain_spec.chain_id())
+    tx_raw = bytes.fromhex(strip_0x(tx_brief['signed_tx'][2:]))
+    tx = unpack(tx_raw, chain_spec)
     nonce = tx_brief['nonce']
     address = tx['from']
 
@@ -66,8 +67,8 @@ def shift_nonce(self, chain_str, tx_hash_orig_hex, delta=1):
     tx_hashes = []
     txs = []
     for otx in otxs:
-        tx_raw = bytes.fromhex(otx.signed_tx[2:])
-        tx_new = unpack_signed_raw_tx(tx_raw, chain_spec.chain_id())
+        tx_raw = bytes.fromhex(strip_0x(otx.signed_tx))
+        tx_new = unpack(tx_raw, chain_spec)
 
         tx_previous_hash_hex = tx_new['hash']
         tx_previous_nonce = tx_new['nonce']

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

@@ -1,33 +1,50 @@
 # standard imports
 import logging
+import sys
 
-# third-party imports
+# external imports
 import celery
-import web3
-from cic_registry import zero_address
-from cic_registry import zero_content
-from cic_registry import CICRegistry
-from crypto_dev_signer.eth.web3ext import Web3 as Web3Ext
-from cic_registry.error import UnknownContractError
+from chainlib.eth.constant import (
+        ZERO_ADDRESS,
+        )
+from cic_eth_registry import CICRegistry
+from cic_eth_registry.error import UnknownContractError
+from chainlib.eth.address import to_checksum_address
+from chainlib.eth.contract import code
+from chainlib.eth.tx import (
+        transaction,
+        receipt,
+        unpack,
+        )
+from chainlib.hash import keccak256_hex_to_hex
+from hexathon import (
+        strip_0x,
+        add_0x,
+        )
+from chainlib.eth.gas import balance
+from chainqueue.db.enum import (
+        StatusEnum,
+        StatusBits,
+        is_alive,
+        is_error_status,
+        status_str,
+    )
+from chainqueue.error import TxStateChangeError
 
 # local imports
 from cic_eth.db.models.base import SessionBase
 from cic_eth.db.models.role import AccountRole
-from cic_eth.db.models.otx import Otx
-from cic_eth.db.models.tx import TxCache
 from cic_eth.db.models.nonce import Nonce
-from cic_eth.db.enum import StatusEnum
 from cic_eth.error import InitializationError
-from cic_eth.db.error import TxStateChangeError
-from cic_eth.eth.rpc import RpcClient
-from cic_eth.queue.tx import get_tx
-from cic_eth.eth.util import unpack_signed_raw_tx
+from cic_eth.queue.query import get_tx
 
 app = celery.current_app
 
 #logg = logging.getLogger(__file__)
 logg = logging.getLogger()
 
+local_fail = StatusBits.LOCAL_ERROR | StatusBits.NODE_ERROR | StatusBits.UNKNOWN_ERROR
+
 
 class AdminApi:
     """Provides an interface to view and manipulate existing transaction tasks and system runtime settings.
@@ -37,19 +54,20 @@ class AdminApi:
     :param queue: Name of worker queue to submit tasks to
     :type queue: str
     """
-    def __init__(self, rpc_client, queue='cic-eth'):
-        self.rpc_client = rpc_client
-        self.w3 = rpc_client.w3
+    def __init__(self, rpc, queue='cic-eth', call_address=ZERO_ADDRESS):
+        self.rpc = rpc
         self.queue = queue
+        self.call_address = call_address
 
 
     def unlock(self, chain_spec, address, flags=None):
         s_unlock = celery.signature(
             'cic_eth.admin.ctrl.unlock',
             [
-                str(chain_spec),
-                flags,
+                None,
+                chain_spec.asdict(),
                 address,
+                flags,
                 ],
             queue=self.queue,
             )
@@ -60,9 +78,10 @@ class AdminApi:
         s_lock = celery.signature(
             'cic_eth.admin.ctrl.lock',
             [
-                str(chain_spec),
-                flags,
+                None,
+                chain_spec.asdict(),
                 address,
+                flags,
                 ],
             queue=self.queue,
             )
@@ -71,14 +90,14 @@ class AdminApi:
 
     def get_lock(self):
         s_lock = celery.signature(
-            'cic_eth.queue.tx.get_lock',
+            'cic_eth.queue.lock.get_lock',
             [],
             queue=self.queue,
             )
-        return s_lock.apply_async().get()
+        return s_lock.apply_async()
 
 
-    def tag_account(self, tag, address_hex):
+    def tag_account(self, tag, address_hex, chain_spec):
         """Persistently associate an address with a plaintext tag.
 
         Some tags are known by the system and is used to resolve addresses to use for certain transactions. 
@@ -89,20 +108,37 @@ class AdminApi:
         :type address_hex: str, 0x-hex
         :raises ValueError: Invalid checksum address
         """
-        if not web3.Web3.isChecksumAddress(address_hex):
-            raise ValueError('invalid address')
-        session = SessionBase.create_session()
-        role = AccountRole.set(tag, address_hex) 
-        session.add(role)
-        session.commit()
-        session.close()
+        s_tag = celery.signature(
+            'cic_eth.eth.account.set_role',
+            [
+                tag,
+                address_hex,
+                chain_spec.asdict(),
+                ],
+            queue=self.queue,
+            )
+        return s_tag.apply_async()
 
 
-    def resend(self, tx_hash_hex, chain_str, in_place=True, unlock=False):
+    def have_account(self, address_hex, chain_spec):
+        s_have = celery.signature(
+            'cic_eth.eth.account.have',
+            [
+                address_hex,
+                chain_spec.asdict(),
+                ],
+            queue=self.queue,
+            )
+        return s_have.apply_async()
+
+
+    def resend(self, tx_hash_hex, chain_spec, in_place=True, unlock=False):
+
         logg.debug('resend {}'.format(tx_hash_hex))
         s_get_tx_cache = celery.signature(
-            'cic_eth.queue.tx.get_tx_cache',
+            'cic_eth.queue.query.get_tx_cache',
             [
+                chain_spec.asdict(),
                 tx_hash_hex,
                 ],
             queue=self.queue,
@@ -110,41 +146,50 @@ class AdminApi:
 
         # TODO: This check should most likely be in resend task itself
         tx_dict = s_get_tx_cache.apply_async().get()
-        if tx_dict['status'] in [StatusEnum.REVERTED, StatusEnum.SUCCESS, StatusEnum.CANCELLED, StatusEnum.OBSOLETED]: 
+        #if tx_dict['status'] in [StatusEnum.REVERTED, StatusEnum.SUCCESS, StatusEnum.CANCELLED, StatusEnum.OBSOLETED]: 
+        if not is_alive(getattr(StatusEnum, tx_dict['status']).value):
             raise TxStateChangeError('Cannot resend mined or obsoleted transaction'.format(txold_hash_hex))
-
-        s = None
-        if in_place:
-            s = celery.signature(
-                'cic_eth.eth.tx.resend_with_higher_gas',
-                [
-                    tx_hash_hex,
-                    chain_str,
-                    None,
-                    1.01,
-                    ],
-                queue=self.queue,
-                )
-        else:
+        
+        if not in_place:
             raise NotImplementedError('resend as new not yet implemented')
 
+        s = celery.signature(
+            'cic_eth.eth.gas.resend_with_higher_gas',
+            [
+                chain_spec.asdict(),
+                None,
+                1.01,
+                ],
+            queue=self.queue,
+            )
+
+        s_manual = celery.signature(
+            'cic_eth.queue.state.set_manual',
+            [
+                tx_hash_hex,
+                ],
+            queue=self.queue,
+            )
+        s_manual.link(s)
+
         if unlock:
             s_gas = celery.signature(
                 'cic_eth.admin.ctrl.unlock_send',
                 [
-                    chain_str,
+                    chain_spec.asdict(),
                     tx_dict['sender'],
                 ],
                 queue=self.queue,
                 )
             s.link(s_gas)
 
-        return s.apply_async()
+        return s_manual.apply_async()
                         
     def check_nonce(self, address):
         s = celery.signature(
-                'cic_eth.queue.tx.get_account_tx',
+                'cic_eth.queue.query.get_account_tx',
                 [
+                    chain_spec.asdict(),
                     address,
                     True,
                     False,
@@ -156,28 +201,37 @@ class AdminApi:
         blocking_tx = None
         blocking_nonce = None
         nonce_otx = 0
+        last_nonce = -1
         for k in txs.keys():
             s_get_tx = celery.signature(
-                    'cic_eth.queue.tx.get_tx',
+                    'cic_eth.queue.query.get_tx',
                     [
+                    chain_spec.asdict(),
                         k,
                         ],
                     queue=self.queue,
                     )
             tx = s_get_tx.apply_async().get()
             #tx = get_tx(k)
-            logg.debug('checking nonce {}'.format(tx['nonce']))
-            if tx['status'] in [StatusEnum.REJECTED, StatusEnum.FUBAR]:
-                blocking_tx = k
-                blocking_nonce = tx['nonce']
+            logg.debug('checking nonce {} (previous {})'.format(tx['nonce'], last_nonce))
             nonce_otx = tx['nonce']
+            if not is_alive(tx['status']) and tx['status'] & local_fail > 0:
+                logg.info('permanently errored {} nonce {} status {}'.format(k, nonce_otx, status_str(tx['status'])))
+                blocking_tx = k
+                blocking_nonce = nonce_otx
+            elif nonce_otx - last_nonce > 1:
+                logg.error('nonce gap; {} followed {} for account {}'.format(nonce_otx, last_nonce, tx['from']))
+                blocking_tx = k
+                blocking_nonce = nonce_otx
+                break
+            last_nonce = nonce_otx
 
         #nonce_cache = Nonce.get(address)
-        nonce_w3 = self.w3.eth.getTransactionCount(address, 'pending') 
+        #nonce_w3 = self.w3.eth.getTransactionCount(address, 'pending') 
         
         return {
             'nonce': {
-                'network': nonce_w3,
+                #'network': nonce_cache,
                 'queue': nonce_otx,
                 #'cache': nonce_cache,
                 'blocking': blocking_nonce,
@@ -188,10 +242,11 @@ class AdminApi:
             }
 
 
-    def fix_nonce(self, address, nonce):
+    def fix_nonce(self, address, nonce, chain_spec):
         s = celery.signature(
-                'cic_eth.queue.tx.get_account_tx',
+                'cic_eth.queue.query.get_account_tx',
                 [
+                    chain_spec.asdict(),
                     address,
                     True,
                     False,
@@ -209,7 +264,7 @@ class AdminApi:
         s_nonce = celery.signature(
                 'cic_eth.admin.nonce.shift_nonce',
                 [
-                    str(self.rpc_client.chain_spec),
+                    self.rpc.chain_spec.asdict(),
                     tx_hash_hex, 
                 ],
                 queue=self.queue
@@ -217,66 +272,83 @@ class AdminApi:
         return s_nonce.apply_async()
 
 
-    # TODO: this is a stub, complete all checks
-    def ready(self):
-        """Checks whether all required initializations have been performed.
-
-        :raises cic_eth.error.InitializationError: At least one setting pre-requisite has not been met.
-        :raises KeyError: An address provided for initialization is not known by the keystore.
-        """
-        addr = AccountRole.get_address('ETH_GAS_PROVIDER_ADDRESS')
-        if addr == zero_address:
-            raise InitializationError('missing account ETH_GAS_PROVIDER_ADDRESS')
-
-        self.w3.eth.sign(addr, text='666f6f')
+#    # TODO: this is a stub, complete all checks
+#    def ready(self):
+#        """Checks whether all required initializations have been performed.
+#
+#        :raises cic_eth.error.InitializationError: At least one setting pre-requisite has not been met.
+#        :raises KeyError: An address provided for initialization is not known by the keystore.
+#        """
+#        addr = AccountRole.get_address('ETH_GAS_PROVIDER_ADDRESS')
+#        if addr == ZERO_ADDRESS:
+#            raise InitializationError('missing account ETH_GAS_PROVIDER_ADDRESS')
+#
+#        self.w3.eth.sign(addr, text='666f6f')
 
 
-    def account(self, chain_spec, address, cols=['tx_hash', 'sender', 'recipient', 'nonce', 'block', 'tx_index', 'status', 'network_status', 'date_created'], include_sender=True, include_recipient=True):
+    def account(self, chain_spec, address, include_sender=True, include_recipient=True, renderer=None, w=sys.stdout):
         """Lists locally originated transactions for the given Ethereum address.
 
         Performs a synchronous call to the Celery task responsible for performing the query.
 
         :param address: Ethereum address to return transactions for
         :type address: str, 0x-hex
-        :param cols: Data columns to include
-        :type cols: list of str
         """
+        last_nonce = -1
         s = celery.signature(
-                'cic_eth.queue.tx.get_account_tx',
-                [address],
+                'cic_eth.queue.query.get_account_tx',
+                [
+                    chain_spec.asdict(),
+                    address,
+                    ],
                 queue=self.queue,
                 )
         txs = s.apply_async().get()
 
         tx_dict_list = []
         for tx_hash in txs.keys():
+            errors = []
             s = celery.signature(
-                    'cic_eth.queue.tx.get_tx_cache',
-                    [tx_hash],
+                    'cic_eth.queue.query.get_tx_cache',
+                    [
+                        chain_spec.asdict(),
+                        tx_hash,
+                        ],
                     queue=self.queue,
                     )
             tx_dict = s.apply_async().get()
-            if tx_dict['sender'] == address and not include_sender:
-                logg.debug('skipping sender tx {}'.format(tx_dict['tx_hash']))
-                continue
+            if tx_dict['sender'] == address:
+                if tx_dict['nonce'] - last_nonce > 1:
+                    logg.error('nonce gap; {} followed {} for address {} tx {}'.format(tx_dict['nonce'], last_nonce, tx_dict['sender'], tx_hash))
+                    errors.append('nonce')
+                elif tx_dict['nonce'] == last_nonce:
+                    logg.info('nonce {} duplicate for address {} in tx {}'.format(tx_dict['nonce'], tx_dict['sender'], tx_hash))
+                last_nonce = tx_dict['nonce']
+                if not include_sender:
+                    logg.debug('skipping sender tx {}'.format(tx_dict['tx_hash']))
+                    continue
             elif tx_dict['recipient'] == address and not include_recipient:
                 logg.debug('skipping recipient tx {}'.format(tx_dict['tx_hash']))
                 continue
 
-            logg.debug(tx_dict)
             o = {
                 'nonce': tx_dict['nonce'], 
                 'tx_hash': tx_dict['tx_hash'],
                 'status': tx_dict['status'],
                 'date_updated': tx_dict['date_updated'],
+                'errors': errors,
                     }
-            tx_dict_list.append(o)
+            if renderer != None:
+                r = renderer(o)
+                w.write(r + '\n')
+            else:
+                tx_dict_list.append(o)
 
         return tx_dict_list
 
 
     # TODO: Add exception upon non-existent tx aswell as invalid tx data to docstring 
-    def tx(self, chain_spec, tx_hash=None, tx_raw=None):
+    def tx(self, chain_spec, tx_hash=None, tx_raw=None, registry=None, renderer=None, w=sys.stdout):
         """Output local and network details about a given transaction with local origin.
 
         If the transaction hash is given, the raw trasnaction data will be retrieved from the local transaction queue backend. Otherwise the raw transaction data must be provided directly. Only one of transaction hash and transaction data can be passed.
@@ -291,46 +363,57 @@ class AdminApi:
         :return: Transaction details
         :rtype: dict
         """
+        problems = []
+
         if tx_hash != None and tx_raw != None:
             ValueError('Specify only one of hash or raw tx')
 
         if tx_raw != None:
-            tx_hash = self.w3.keccak(hexstr=tx_raw).hex()
+            tx_hash = add_0x(keccak256_hex_to_hex(tx_raw))
+            #tx_hash = self.w3.keccak(hexstr=tx_raw).hex()
 
         s = celery.signature(
-            'cic_eth.queue.tx.get_tx_cache',
-            [tx_hash],
+            'cic_eth.queue.query.get_tx_cache',
+            [
+                chain_spec.asdict(),
+                tx_hash,
+                ],
             queue=self.queue,
             )
     
-        tx = s.apply_async().get()
+        t = s.apply_async()
+        tx = t.get()
   
         source_token = None
-        if tx['source_token'] != zero_address:
+        if tx['source_token'] != ZERO_ADDRESS:
             try:
-                source_token = CICRegistry.get_address(chain_spec, tx['source_token']).contract
+                source_token = registry.by_address(tx['source_token'])
+                #source_token = CICRegistry.get_address(chain_spec, tx['source_token']).contract
             except UnknownContractError:
-                source_token_contract = self.w3.eth.contract(abi=CICRegistry.abi('ERC20'), address=tx['source_token'])
-                source_token = CICRegistry.add_token(chain_spec, source_token_contract)
+                #source_token_contract = self.w3.eth.contract(abi=CICRegistry.abi('ERC20'), address=tx['source_token'])
+                #source_token = CICRegistry.add_token(chain_spec, source_token_contract)
                 logg.warning('unknown source token contract {}'.format(tx['source_token']))
 
         destination_token = None
-        if tx['source_token'] != zero_address:
+        if tx['source_token'] != ZERO_ADDRESS:
             try:
-                destination_token = CICRegistry.get_address(chain_spec, tx['destination_token'])
+                #destination_token = CICRegistry.get_address(chain_spec, tx['destination_token'])
+                destination_token = registry.by_address(tx['destination_token'])
             except UnknownContractError:
-                destination_token_contract = self.w3.eth.contract(abi=CICRegistry.abi('ERC20'), address=tx['source_token'])
-                destination_token = CICRegistry.add_token(chain_spec, destination_token_contract)
+                #destination_token_contract = self.w3.eth.contract(abi=CICRegistry.abi('ERC20'), address=tx['source_token'])
+                #destination_token = CICRegistry.add_token(chain_spec, destination_token_contract)
                 logg.warning('unknown destination token contract {}'.format(tx['destination_token']))
 
         tx['sender_description'] = 'Custodial account'
         tx['recipient_description'] = 'Custodial account'
 
-        c = RpcClient(chain_spec)
-        if len(c.w3.eth.getCode(tx['sender'])) > 0:
+        o = code(tx['sender'])
+        r = self.rpc.do(o)
+        if len(strip_0x(r, allow_empty=True)) > 0:
             try:
-                sender_contract = CICRegistry.get_address(chain_spec, tx['sender'])
-                tx['sender_description'] = 'Contract {}'.format(sender_contract.identifier())
+                #sender_contract = CICRegistry.get_address(chain_spec, tx['sender'])
+                sender_contract = registry.by_address(tx['sender'], sender_address=self.call_address)
+                tx['sender_description'] = 'Contract at {}'.format(tx['sender']) #sender_contract)
             except UnknownContractError:
                 tx['sender_description'] = 'Unknown contract'
             except KeyError as e:
@@ -340,7 +423,7 @@ class AdminApi:
                     'cic_eth.eth.account.have',
                     [
                         tx['sender'],
-                        str(chain_spec),
+                        chain_spec.asdict(),
                         ],
                     queue=self.queue,
                     )
@@ -353,7 +436,7 @@ class AdminApi:
                     'cic_eth.eth.account.role',
                     [
                         tx['sender'],
-                        str(chain_spec),
+                        chain_spec.asdict(),
                         ],
                     queue=self.queue,
                     )
@@ -362,11 +445,13 @@ class AdminApi:
                 if role != None:
                     tx['sender_description'] = role
 
-
-        if len(c.w3.eth.getCode(tx['recipient'])) > 0:
+        o = code(tx['recipient'])
+        r = self.rpc.do(o)
+        if len(strip_0x(r, allow_empty=True)) > 0:
             try:
-                recipient_contract = CICRegistry.get_address(chain_spec, tx['recipient'])
-                tx['recipient_description'] = 'Contract {}'.format(recipient_contract.identifier())
+                #recipient_contract = CICRegistry.by_address(tx['recipient'])
+                recipient_contract = registry.by_address(tx['recipient'])
+                tx['recipient_description'] = 'Contract at {}'.format(tx['recipient']) #recipient_contract)
             except UnknownContractError as e:
                 tx['recipient_description'] = 'Unknown contract'
             except KeyError as e:
@@ -376,7 +461,7 @@ class AdminApi:
                     'cic_eth.eth.account.have',
                     [
                         tx['recipient'],
-                        str(chain_spec),
+                        chain_spec.asdict(),
                         ],
                     queue=self.queue,
                     )
@@ -389,7 +474,7 @@ class AdminApi:
                     'cic_eth.eth.account.role',
                     [
                         tx['recipient'],
-                        str(chain_spec),
+                        chain_spec.asdict(),
                         ],
                     queue=self.queue,
                     )
@@ -406,35 +491,52 @@ class AdminApi:
             tx['destination_token_symbol'] = destination_token.symbol()
             tx['recipient_token_balance'] = source_token.function('balanceOf')(tx['recipient']).call()
 
-        tx['network_status'] = 'Not submitted'
+        # TODO: this can mean either not subitted or culled, need to check other txs with same nonce to determine which
+        tx['network_status'] = 'Not in node' 
 
+        r = None
         try:
-            c.w3.eth.getTransaction(tx_hash)
-            tx['network_status'] = 'Mempool'
-        except web3.exceptions.TransactionNotFound:
-            pass
+            o = transaction(tx_hash)
+            r = self.rpc.do(o)
+            if r != None:
+                tx['network_status'] = 'Mempool'
+        except Exception as e:
+            logg.warning('(too permissive exception handler, please fix!) {}'.format(e))
 
-        try:
-            r = c.w3.eth.getTransactionReceipt(tx_hash)
-            if r.status == 1:
-                tx['network_status'] = 'Confirmed'
-                tx['block'] = r.blockNumber
-                tx['tx_index'] = r.transactionIndex
-            else:
-                tx['network_status'] = 'Reverted'
-        except web3.exceptions.TransactionNotFound:
-            pass
+        if r != None:
+            try:
+                o = receipt(tx_hash)
+                r = self.rpc.do(o)
+                logg.debug('h {} o {}'.format(tx_hash, o))
+                if int(strip_0x(r['status'])) == 1:
+                    tx['network_status'] = 'Confirmed'
+                else:
+                    tx['network_status'] = 'Reverted'
+                tx['network_block_number'] = r.blockNumber
+                tx['network_tx_index'] = r.transactionIndex
+                if tx['block_number'] == None:
+                    problems.append('Queue is missing block number {} for mined tx'.format(r.blockNumber))
+            except Exception as e:
+                logg.warning('too permissive exception handler, please fix!')
+                pass
 
-        tx['sender_gas_balance'] = c.w3.eth.getBalance(tx['sender'])
-        tx['recipient_gas_balance'] = c.w3.eth.getBalance(tx['recipient'])
+        o = balance(tx['sender'])
+        r = self.rpc.do(o)
+        tx['sender_gas_balance'] = r
 
-        tx_unpacked = unpack_signed_raw_tx(bytes.fromhex(tx['signed_tx'][2:]), chain_spec.chain_id())
+        o = balance(tx['recipient'])
+        r = self.rpc.do(o)
+        tx['recipient_gas_balance'] = r
+
+        tx_unpacked = unpack(bytes.fromhex(strip_0x(tx['signed_tx'])), chain_spec)
         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',
+            'cic_eth.queue.state.get_state_log',
             [
+                chain_spec.asdict(),
                 tx_hash,
                 ],
             queue=self.queue,
@@ -442,4 +544,14 @@ class AdminApi:
         t = s.apply_async()
         tx['status_log'] = t.get()
 
-        return tx
+        if len(problems) > 0:
+            sys.stderr.write('\n')
+            for p in problems:
+                sys.stderr.write('!!!{}\n'.format(p))
+
+        if renderer == None:
+            return tx
+
+        r = renderer(tx)
+        w.write(r + '\n')
+        return None

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

@@ -6,11 +6,12 @@
 # standard imports
 import logging
 
-# third-party imports 
+# external imports 
 import celery
-from cic_registry.chain import ChainSpec
-from cic_registry import CICRegistry
-from cic_eth.eth.factory import TxFactory
+from cic_eth_registry import CICRegistry
+from chainlib.chain import ChainSpec
+
+# local imports
 from cic_eth.db.enum import LockEnum
 
 app = celery.current_app
@@ -30,13 +31,13 @@ class Api:
     :param queue: Name of worker queue to submit tasks to
     :type queue: str
     """
-    def __init__(self, chain_str, queue='cic-eth', callback_param=None, callback_task='cic_eth.callbacks.noop', callback_queue=None):
+    def __init__(self, chain_str, queue='cic-eth', callback_param=None, callback_task='cic_eth.callbacks.noop.noop', callback_queue=None):
         self.chain_str = chain_str
         self.chain_spec = ChainSpec.from_chain_str(chain_str)
         self.callback_param = callback_param
         self.callback_task = callback_task
         self.queue = queue
-        logg.info('api using queue {}'.format(self.queue))
+        logg.debug('api using queue {}'.format(self.queue))
         self.callback_success = None
         self.callback_error = None
         if callback_queue == None:
@@ -79,18 +80,26 @@ class Api:
         :returns: uuid of root task
         :rtype: celery.Task
         """
+        raise NotImplementedError('out of service until new DEX migration is done')
         s_check = celery.signature(
                 'cic_eth.admin.ctrl.check_lock',
                 [
                     [from_token_symbol, to_token_symbol],
-                    self.chain_str,
+                    self.chain_spec.asdict(),
                     LockEnum.QUEUE,
                     from_address,
                     ],
                 queue=self.queue,
                 )
+        s_nonce = celery.signature(
+                'cic_eth.eth.nonce.reserve_nonce',
+                [
+                    self.chain_spec.asdict(),
+                    ],
+                queue=self.queue,
+                )
         s_tokens = celery.signature(
-                'cic_eth.eth.token.resolve_tokens_by_symbol',
+                'cic_eth.eth.erc20.resolve_tokens_by_symbol',
                 [
                     self.chain_str,
                     ],
@@ -103,11 +112,12 @@ class Api:
                     target_return,
                     minimum_return,
                     to_address,
-                    self.chain_str,
+                    self.chain_spec.asdict(),
                     ],
                 queue=self.queue,
                 )
-        s_check.link(s_tokens)
+        s_nonce.link(s_tokens)
+        s_check.link(s_nonce)
         if self.callback_param != None:
             s_convert.link(self.callback_success)
             s_tokens.link(s_convert).on_error(self.callback_error)
@@ -134,20 +144,28 @@ class Api:
         :returns: uuid of root task
         :rtype: celery.Task
         """
+        raise NotImplementedError('out of service until new DEX migration is done')
         s_check = celery.signature(
                 'cic_eth.admin.ctrl.check_lock',
                 [
                     [from_token_symbol, to_token_symbol],
-                    self.chain_str,
+                    self.chain_spec.asdict(),
                     LockEnum.QUEUE,
                     from_address,
                     ],
                 queue=self.queue,
                 )
-        s_tokens = celery.signature(
-                'cic_eth.eth.token.resolve_tokens_by_symbol',
+        s_nonce = celery.signature(
+                'cic_eth.eth.nonce.reserve_nonce',
                 [
-                    self.chain_str,
+                    self.chain_spec.asdict(),
+                    ],
+                queue=self.queue,
+                )
+        s_tokens = celery.signature(
+                'cic_eth.eth.erc20.resolve_tokens_by_symbol',
+                [
+                    self.chain_spec.asdict(),
                     ],
                 queue=self.queue,
                 )
@@ -158,11 +176,12 @@ class Api:
                     target_return,
                     minimum_return,
                     from_address,
-                    self.chain_str,
+                    self.chain_spec.asdict(),
                     ],
                 queue=self.queue,
                 )
-        s_check.link(s_tokens)
+        s_nonce.link(s_tokens)
+        s_check.link(s_nonce)
         if self.callback_param != None:
             s_convert.link(self.callback_success)
             s_tokens.link(s_convert).on_error(self.callback_error)
@@ -191,30 +210,39 @@ class Api:
                 'cic_eth.admin.ctrl.check_lock',
                 [
                     [token_symbol],
-                    self.chain_str,
+                    self.chain_spec.asdict(),
                     LockEnum.QUEUE,
                     from_address,
                     ],
                 queue=self.queue,
                 )
-        s_tokens = celery.signature(
-                'cic_eth.eth.token.resolve_tokens_by_symbol',
+        s_nonce = celery.signature(
+                'cic_eth.eth.nonce.reserve_nonce',
                 [
-                    self.chain_str,
+                    self.chain_spec.asdict(),
+                    from_address,
+                    ],
+                queue=self.queue,
+                )
+        s_tokens = celery.signature(
+                'cic_eth.eth.erc20.resolve_tokens_by_symbol',
+                [
+                    self.chain_spec.asdict(),
                     ],
                 queue=self.queue,
                 )
         s_transfer = celery.signature(
-                'cic_eth.eth.token.transfer',
+                'cic_eth.eth.erc20.transfer',
                 [
                     from_address,
                     to_address,
                     value,
-                    self.chain_str,
+                    self.chain_spec.asdict(),
                     ],
                 queue=self.queue,
                 )
-        s_check.link(s_tokens)
+        s_nonce.link(s_tokens)
+        s_check.link(s_nonce)
         if self.callback_param != None:
             s_transfer.link(self.callback_success)
             s_tokens.link(s_transfer).on_error(self.callback_error)
@@ -225,89 +253,15 @@ class Api:
         return t
 
 
-    def transfer_request(self, from_address, to_address, spender_address, value, token_symbol):
-        """Executes a chain of celery tasks that issues a transfer request of ERC20 tokens from one address to another.
-
-        :param from_address: Ethereum address of sender
-        :type from_address: str, 0x-hex
-        :param to_address: Ethereum address of recipient
-        :type to_address: str, 0x-hex
-        :param spender_address: Ethereum address that is executing transfer (typically an escrow contract)
-        :type spender_address: str, 0x-hex
-        :param value: Estimated return from conversion
-        :type  value: int
-        :param token_symbol: ERC20 token symbol of token to send
-        :type token_symbol: str
-        :returns: uuid of root task
-        :rtype: celery.Task
-        """
-        s_check = celery.signature(
-                'cic_eth.admin.ctrl.check_lock',
-                [
-                    [token_symbol],
-                    self.chain_str,
-                    LockEnum.QUEUE,
-                    from_address,
-                    ],
-                queue=self.queue,
-                )
-        s_tokens_transfer_approval = celery.signature(
-                'cic_eth.eth.token.resolve_tokens_by_symbol',
-                [
-                    self.chain_str,
-                    ],
-                queue=self.queue,
-                )
-        s_tokens_approve = celery.signature(
-                'cic_eth.eth.token.resolve_tokens_by_symbol',
-                [
-                    self.chain_str,
-                    ],
-                queue=self.queue,
-                )
-        s_approve = celery.signature(
-                'cic_eth.eth.token.approve',
-                [
-                    from_address,
-                    spender_address,
-                    value,
-                    self.chain_str,
-                    ],
-                queue=self.queue,
-                )
-        s_transfer_approval = celery.signature(
-                'cic_eth.eth.request.transfer_approval_request',
-                [
-                    from_address,
-                    to_address,
-                    value,
-                    self.chain_str,
-                    ],
-                queue=self.queue,
-                )
-        # TODO: make approve and transfer_approval chainable so callback can be part of the full chain
-        if self.callback_param != None:
-            s_transfer_approval.link(self.callback_success)
-            s_tokens_approve.link(s_approve)
-            s_tokens_transfer_approval.link(s_transfer_approval).on_error(self.callback_error)
-        else:
-            s_tokens_approve.link(s_approve)
-            s_tokens_transfer_approval.link(s_transfer_approval)
-
-        g = celery.group(s_tokens_approve, s_tokens_transfer_approval) #s_tokens.apply_async(queue=self.queue)
-        s_check.link(g)
-        t = s_check.apply_async()
-        #t = s_tokens.apply_async(queue=self.queue)
-        return t
-
-
-    def balance(self, address, token_symbol):
+    def balance(self, address, token_symbol, include_pending=True):
         """Calls the provided callback with the current token balance of the given address.
 
         :param address: Ethereum address of holder
         :type address: str, 0x-hex
         :param token_symbol: ERC20 token symbol of token to send
         :type token_symbol: str
+        :param include_pending: If set, will include transactions that have not yet been fully processed
+        :type include_pending: bool
         :returns: uuid of root task
         :rtype: celery.Task
         """
@@ -315,29 +269,66 @@ class Api:
             logg.warning('balance pointlessly called with no callback url')
 
         s_tokens = celery.signature(
-                'cic_eth.eth.token.resolve_tokens_by_symbol',
+                'cic_eth.eth.erc20.resolve_tokens_by_symbol',
                 [
                     [token_symbol],
-                    self.chain_str,
+                    self.chain_spec.asdict(),
                     ],
                 queue=self.queue,
                 )
         s_balance = celery.signature(
-                'cic_eth.eth.token.balance',
+                'cic_eth.eth.erc20.balance',
                 [
                     address,
-                    self.chain_str,
+                    self.chain_spec.asdict(),
                     ],
                 queue=self.queue,
                 )
-        
-        if self.callback_param != None:
-            s_balance.link(self.callback_success)
-            s_tokens.link(s_balance).on_error(self.callback_error)
-        else:
-            s_tokens.link(s_balance)
+        s_result = celery.signature(
+                'cic_eth.queue.balance.assemble_balances',
+                [],
+                queue=self.queue,
+                )
 
-        t = s_tokens.apply_async(queue=self.queue)
+        last_in_chain = s_balance
+        if include_pending:
+            s_balance_incoming = celery.signature(
+                    'cic_eth.queue.balance.balance_incoming',
+                    [
+                        address,
+                        self.chain_spec.asdict(),
+                        ],
+                    queue=self.queue,
+                    )
+            s_balance_outgoing = celery.signature(
+                    'cic_eth.queue.balance.balance_outgoing',
+                    [
+                        address,
+                        self.chain_spec.asdict(),
+                        ],
+                    queue=self.queue,
+                    )
+            s_balance.link(s_balance_incoming)
+            s_balance_incoming.link(s_balance_outgoing)
+            last_in_chain = s_balance_outgoing
+
+            one = celery.chain(s_tokens, s_balance)
+            two = celery.chain(s_tokens, s_balance_incoming)
+            three = celery.chain(s_tokens, s_balance_outgoing)
+
+            t = None
+            if self.callback_param != None:
+                s_result.link(self.callback_success).on_error(self.callback_error)
+                t = celery.chord([one, two, three])(s_result)
+            else:
+                t = celery.chord([one, two, three])(s_result)
+        else:
+            # TODO: Chord is inefficient with only one chain, but assemble_balances must be able to handle different structures in order to avoid chord
+            one = celery.chain(s_tokens, s_balance)
+            if self.callback_param != None:
+                s_result.link(self.callback_success).on_error(self.callback_error)
+            t = celery.chord([one])(s_result)
+    
         return t
 
 
@@ -355,7 +346,7 @@ class Api:
                 'cic_eth.admin.ctrl.check_lock',
                 [
                     password,
-                    self.chain_str,
+                    self.chain_spec.asdict(),
                     LockEnum.CREATE,
                     ],
                 queue=self.queue,
@@ -363,7 +354,7 @@ class Api:
         s_account = celery.signature(
                 'cic_eth.eth.account.create',
                 [
-                    self.chain_str,
+                    self.chain_spec.asdict(),
                     ],
                 queue=self.queue,
                 )
@@ -372,14 +363,23 @@ class Api:
             s_account.link(self.callback_success)
 
         if register:
-            s_register = celery.signature(
-                'cic_eth.eth.account.register', 
+            s_nonce = celery.signature(
+                'cic_eth.eth.nonce.reserve_nonce',
                 [
-                    self.chain_str,
+                    self.chain_spec.asdict(),
+                    'ACCOUNT_REGISTRY_WRITER',
                     ],
                 queue=self.queue,
                 )
-            s_account.link(s_register)
+            s_register = celery.signature(
+                'cic_eth.eth.account.register', 
+                [
+                    self.chain_spec.asdict(),
+                    ],
+                queue=self.queue,
+                )
+            s_nonce.link(s_register)
+            s_account.link(s_nonce)
 
         t = s_check.apply_async(queue=self.queue)
         return t
@@ -397,19 +397,28 @@ class Api:
                 'cic_eth.admin.ctrl.check_lock',
                 [
                     address,
-                    self.chain_str,
+                    self.chain_spec.asdict(),
                     LockEnum.QUEUE,
                     ],
                 queue=self.queue,
                 )
-        s_refill = celery.signature(
-                'cic_eth.eth.tx.refill_gas',
+        s_nonce = celery.signature(
+                'cic_eth.eth.nonce.reserve_nonce',
                 [
-                    self.chain_str,
+                    self.chain_spec.asdict(),
+                    'GAS_GIFTER',
                     ],
                 queue=self.queue,
                 )
-        s_check.link(s_refill) 
+        s_refill = celery.signature(
+                'cic_eth.eth.gas.refill_gas',
+                [
+                    self.chain_spec.asdict(),
+                    ],
+                queue=self.queue,
+                )
+        s_nonce.link(s_refill) 
+        s_check.link(s_nonce)
         if self.callback_param != None:
             s_refill.link(self.callback_success)
 
@@ -417,6 +426,74 @@ class Api:
         return t
 
 
+    def list(self, address, limit=10, external_task=None, external_queue=None):
+        """Retrieve an aggregate list of latest transactions of internal and (optionally) external origin in reverse chronological order.
+
+        The array of transactions returned have the same dict layout as those passed by the callback filter in cic_eth/runnable/manager
+
+        If the external task is defined, this task will be used to query external transactions. If this is not defined, no external transactions will be included. The task must accept (offset, limit, address) as input parameters, and return a bloom filter that will be used to retrieve transaction data for the matching transactions. See cic_eth.ext.tx.list_tx_by_bloom for details on the bloom filter dat format.
+
+        :param address: Ethereum address to list transactions for
+        :type address: str, 0x-hex
+        :param limit: Amount of results to return
+        :type limit: number
+        :param external_task: Celery task providing external transactions
+        :type external_task: str
+        :param external_queue: Celery task queue providing exernal transactions task
+        :type external_queue: str
+        :returns: List of transactions
+        :rtype: list of dict
+        """
+        offset = 0
+        s_local = celery.signature(
+            'cic_eth.queue.query.get_account_tx',
+            [
+                self.chain_spec.asdict(),
+                address,
+                ],
+            queue=self.queue,
+            )
+
+        s_brief = celery.signature(
+            'cic_eth.ext.tx.tx_collate',
+            [
+                self.chain_spec.asdict(),
+                offset,
+                limit
+                ],
+            queue=self.queue,
+            )
+        s_local.link(s_brief)
+        if self.callback_param != None:
+            s_brief.link(self.callback_success).on_error(self.callback_error)
+
+        t = None
+        if external_task != None:
+            s_external_get = celery.signature(
+                external_task,
+                [
+                    address,
+                    offset,
+                    limit,
+                    ],
+                queue=external_queue,
+                )
+
+            s_external_process = celery.signature(
+                'cic_eth.ext.tx.list_tx_by_bloom',
+                [
+                    address,
+                    self.chain_spec.asdict(),
+                    ],
+                queue=self.queue,
+                    )
+            c = celery.chain(s_external_get, s_external_process)
+            t = celery.chord([s_local, c])(s_brief)
+        else:
+            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/noop.py

@@ -18,4 +18,4 @@ def noop(self, result, param, status_code):
     :rtype: bool
     """
     logg.info('noop callback {} {} {}'.format(result, param, status_code))
-    return True
+    return result

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

@@ -18,7 +18,11 @@ logg = celery_app.log.get_default_logger()
 def redis(self, result, destination, status_code):
     (host, port, db, channel) = destination.split(':')
     r = redis_interface.Redis(host=host, port=port, db=db)
-    s = json.dumps(result)
+    data = {
+            'root_id': self.request.root_id,
+            'status': status_code,
+            'result': result,
+            }
     logg.debug('redis callback on host {} port {} db {} channel {}'.format(host, port, db, channel))
-    r.publish(channel, s)
+    r.publish(channel, json.dumps(data))
     r.close()

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/__init__.py

@@ -11,10 +11,6 @@ logg = logging.getLogger()
 # an Engine, which the Session will use for connection
 # resources
 
-# TODO: Remove the package exports, all models should be imported using full path
-from .models.otx import Otx
-from .models.convert import TxConvertTransfer
-
 
 def dsn_from_config(config):
     """Generate a dsn string from the provided config dict.

apps/cic-eth/cic_eth/db/enum.py

@@ -1,6 +1,29 @@
 # standard imports
 import enum
 
+
+@enum.unique
+class StatusBits(enum.IntEnum):
+    """Individual bit flags that are combined to define the state and legacy of a queued transaction
+
+    """
+    QUEUED = 0x01 # transaction should be sent to network
+    IN_NETWORK = 0x08 # transaction is in network
+   
+    DEFERRED = 0x10 # an attempt to send the transaction to network has failed
+    GAS_ISSUES = 0x20 # transaction is pending sender account gas funding
+
+    LOCAL_ERROR = 0x100 # errors that originate internally from the component
+    NODE_ERROR = 0x200 # errors originating in the node (invalid RLP input...)
+    NETWORK_ERROR = 0x400 # errors that originate from the network (REVERT)
+    UNKNOWN_ERROR = 0x800 # unclassified errors (the should not occur)
+
+    FINAL = 0x1000 # transaction processing has completed
+    OBSOLETE = 0x2000 # transaction has been replaced by a different transaction with higher fee
+    MANUAL = 0x8000 # transaction processing has been manually overridden
+
+
+@enum.unique
 class StatusEnum(enum.IntEnum):
     """
 
@@ -22,21 +45,27 @@ class StatusEnum(enum.IntEnum):
         * SUCCESS: THe transaction was successfully mined. (Block number will be set)
 
     """
-    PENDING=-9
-    SENDFAIL=-8
-    RETRY=-7
-    READYSEND=-6
-    OBSOLETED=-2
-    WAITFORGAS=-1
-    SENT=0
-    FUBAR=1
-    CANCELLED=2
-    OVERRIDDEN=3
-    REJECTED=7
-    REVERTED=8
-    SUCCESS=9
+    PENDING = 0
+
+    SENDFAIL = StatusBits.DEFERRED | StatusBits.LOCAL_ERROR
+    RETRY = StatusBits.QUEUED | StatusBits.DEFERRED 
+    READYSEND = StatusBits.QUEUED
+
+    OBSOLETED = StatusBits.OBSOLETE | StatusBits.IN_NETWORK
+
+    WAITFORGAS = StatusBits.GAS_ISSUES
+
+    SENT = StatusBits.IN_NETWORK
+    FUBAR = StatusBits.FINAL | StatusBits.UNKNOWN_ERROR
+    CANCELLED = StatusBits.IN_NETWORK | StatusBits.FINAL | StatusBits.OBSOLETE
+    OVERRIDDEN = StatusBits.FINAL | StatusBits.OBSOLETE | StatusBits.MANUAL
+
+    REJECTED = StatusBits.NODE_ERROR | StatusBits.FINAL
+    REVERTED = StatusBits.IN_NETWORK | StatusBits.FINAL | StatusBits.NETWORK_ERROR
+    SUCCESS = StatusBits.IN_NETWORK | StatusBits.FINAL 
 
 
+@enum.unique
 class LockEnum(enum.IntEnum):
     """
     STICKY: When set, reset is not possible
@@ -48,4 +77,81 @@ class LockEnum(enum.IntEnum):
     CREATE=2
     SEND=4
     QUEUE=8
+    QUERY=16
     ALL=int(0xfffffffffffffffe)
+
+
+def status_str(v, bits_only=False):
+    """Render a human-readable string describing the status
+
+    If the bit field exactly matches a StatusEnum value, the StatusEnum label will be returned.
+
+    If a StatusEnum cannot be matched, the string will be postfixed with "*", unless explicitly instructed to return bit field labels only.
+
+    :param v: Status bit field
+    :type v: number
+    :param bits_only: Only render individual bit labels.
+    :type bits_only: bool
+    :returns: Status string
+    :rtype: str
+    """
+    s = ''
+    if not bits_only:
+        try:
+            s = StatusEnum(v).name
+            return s
+        except ValueError:
+            pass
+
+    if v == 0:
+        return 'NONE'
+
+    for i in range(16):
+        b = (1 << i)
+        if (b & 0xffff) & v:
+            n = StatusBits(b).name
+            if len(s) > 0:
+                s += ','
+            s += n
+    if not bits_only:
+        s += '*'
+    return s
+
+
+def all_errors():
+    """Bit mask of all error states
+
+    :returns: Error flags
+    :rtype: number
+    """
+    return StatusBits.LOCAL_ERROR | StatusBits.NODE_ERROR | StatusBits.NETWORK_ERROR | StatusBits.UNKNOWN_ERROR
+
+
+def is_error_status(v):
+    """Check if value is an error state
+
+    :param v: Status bit field
+    :type v: number
+    :returns: True if error
+    :rtype: bool
+    """
+    return bool(v & all_errors())
+
+
+def dead():
+    """Bit mask defining whether a transaction is still likely to be processed on the network.
+
+    :returns: Bit mask
+    :rtype: number
+    """
+    return StatusBits.FINAL | StatusBits.OBSOLETE
+
+
+def is_alive(v):
+    """Check if transaction is still likely to be processed on the network.
+
+    The contingency of "likely" refers to the case a transaction has been obsoleted after sent to the network, but the network still confirms the obsoleted transaction. The return value of this method will not change as a result of this, BUT the state itself will (as the FINAL bit will be set).
+
+    :returns: 
+    """
+    return bool(v & dead() == 0)

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

@@ -0,0 +1,29 @@
+"""Add chainqueue
+
+Revision ID: 0ec0d6d1e785
+Revises: 
+Create Date: 2021-04-02 18:30:55.398388
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+from chainqueue.db.migrations.sqlalchemy import (
+        chainqueue_upgrade,
+        chainqueue_downgrade,
+        )
+
+# revision identifiers, used by Alembic.
+revision = '0ec0d6d1e785'
+down_revision = None
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    chainqueue_upgrade(0, 0, 1)
+
+
+def downgrade():
+    chainqueue_downgrade(0, 0, 1)
+

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

@@ -0,0 +1,29 @@
+"""Roles
+
+Revision ID: 1f1b3b641d08
+Revises: 9c420530eeb2
+Create Date: 2021-04-02 18:40:27.787631
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '1f1b3b641d08'
+down_revision = '9c420530eeb2'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        'account_role',
+        sa.Column('id', sa.Integer, primary_key=True),
+        sa.Column('tag', sa.Text, nullable=False, unique=True),
+        sa.Column('address_hex', sa.String(42), nullable=False),
+        )
+
+
+def downgrade():
+    op.drop_table('account_role')

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

@@ -1,35 +0,0 @@
-"""Add new syncer table
-
-Revision ID: 2a07b543335e
-Revises: a2e2aab8f331
-Create Date: 2020-12-27 09:35:44.017981
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '2a07b543335e'
-down_revision = 'a2e2aab8f331'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'blockchain_sync',
-            sa.Column('id', sa.Integer, primary_key=True),
-            sa.Column('blockchain', sa.String, nullable=False),
-            sa.Column('block_start', sa.Integer, nullable=False, default=0),
-            sa.Column('tx_start', sa.Integer, nullable=False, default=0),
-            sa.Column('block_cursor', sa.Integer, nullable=False, default=0),
-            sa.Column('tx_cursor', sa.Integer, nullable=False, default=0),
-            sa.Column('block_target', sa.Integer, nullable=True),
-            sa.Column('date_created', sa.DateTime, nullable=False),
-            sa.Column('date_updated', sa.DateTime),
-            )
-
-
-def downgrade():
-    op.drop_table('blockchain_sync')

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

@@ -1,29 +0,0 @@
-"""Add nonce index
-
-Revision ID: 49b348246d70
-Revises: 52c7c59cd0b1
-Create Date: 2020-12-19 09:45:36.186446
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '49b348246d70'
-down_revision = '52c7c59cd0b1'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-        'nonce',
-        sa.Column('id', sa.Integer, primary_key=True),
-        sa.Column('address_hex', sa.String(42), nullable=False, unique=True),
-        sa.Column('nonce', sa.Integer, nullable=False),
-        )
-
-
-def downgrade():
-    op.drop_table('nonce')

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

@@ -1,31 +0,0 @@
-"""Add account roles
-
-Revision ID: 52c7c59cd0b1
-Revises: 9c4bd7491015
-Create Date: 2020-12-19 07:21:38.249237
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '52c7c59cd0b1'
-down_revision = '9c4bd7491015'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'account_role',
-            sa.Column('id', sa.Integer, primary_key=True),
-            sa.Column('tag', sa.Text, nullable=False, unique=True),
-            sa.Column('address_hex', sa.String(42), nullable=False),
-            )
-    pass
-
-
-def downgrade():
-    op.drop_table('account_role')
-    pass

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

@@ -0,0 +1,30 @@
+"""DEbug
+
+Revision ID: 5ca4b77ce205
+Revises: 75d4767b3031
+Create Date: 2021-04-02 18:42:12.257244
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '5ca4b77ce205'
+down_revision = '75d4767b3031'
+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),
+            )
+
+
+def downgrade():
+    op.drop_table('debug')

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

@@ -1,30 +0,0 @@
-"""Add otx state log
-
-Revision ID: 6ac7a1dadc46
-Revises: 89e1e9baa53c
-Create Date: 2021-01-30 13:59:49.022373
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '6ac7a1dadc46'
-down_revision = '89e1e9baa53c'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'otx_state_log',
-            sa.Column('id', sa.Integer, primary_key=True),
-            sa.Column('otx_id', sa.Integer, sa.ForeignKey('otx.id'), nullable=False),
-            sa.Column('date', sa.DateTime, nullable=False),
-            sa.Column('status', sa.Integer, nullable=False),
-            )
-
-
-def downgrade():
-    op.drop_table('otx_state_log')

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

@@ -1,31 +0,0 @@
-"""Add attempts and version log for otx
-
-Revision ID: 71708e943dbd
-Revises: 7e8d7626e38f
-Create Date: 2020-09-26 14:41:19.298651
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '71708e943dbd'
-down_revision = '7e8d7626e38f'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'otx_attempts',
-            sa.Column('id', sa.Integer, primary_key=True),
-            sa.Column('otx_id', sa.Integer, sa.ForeignKey('otx.id'), nullable=False),
-            sa.Column('date', sa.DateTime, nullable=False),
-            )
-    pass
-
-
-def downgrade():
-    op.drop_table('otx_attempts')
-    pass

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

@@ -1,8 +1,8 @@
-"""Add account lock
+"""Lock
 
-Revision ID: 89e1e9baa53c
-Revises: 2a07b543335e
-Create Date: 2021-01-27 19:57:36.793882
+Revision ID: 75d4767b3031
+Revises: 1f1b3b641d08
+Create Date: 2021-04-02 18:41:20.864265
 
 """
 from alembic import op
@@ -10,8 +10,8 @@ import sqlalchemy as sa
 
 
 # revision identifiers, used by Alembic.
-revision = '89e1e9baa53c'
-down_revision = '2a07b543335e'
+revision = '75d4767b3031'
+down_revision = '1f1b3b641d08'
 branch_labels = None
 depends_on = None
 
@@ -24,10 +24,11 @@ def upgrade():
             sa.Column('blockchain', sa.String),
             sa.Column("flags", sa.BIGINT(), nullable=False, default=0),
             sa.Column("date_created", sa.DateTime, nullable=False),
-            sa.Column("otx_id", sa.Integer, nullable=True),
+            sa.Column("otx_id", sa.Integer, sa.ForeignKey('otx.id'), nullable=True),
             )
     op.create_index('idx_chain_address', 'lock', ['blockchain', 'address'], unique=True)
 
+
 def downgrade():
     op.drop_index('idx_chain_address')
     op.drop_table('lock')

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

@@ -1,31 +0,0 @@
-"""add blocknumber pointer
-
-Revision ID: 7cb65b893934
-Revises: 8593fa1ca0f4
-Create Date: 2020-09-24 19:29:13.543648
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '7cb65b893934'
-down_revision = '8593fa1ca0f4'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'watcher_state',
-            sa.Column('block_number', sa.Integer)
-            )
-    conn = op.get_bind()
-    conn.execute('INSERT INTO watcher_state (block_number) VALUES (0);')
-    pass
-
-
-def downgrade():
-    op.drop_table('watcher_state')
-    pass

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

@@ -1,45 +0,0 @@
-"""Add block sync
-
-Revision ID: 7e8d7626e38f
-Revises: cd2052be6db2
-Create Date: 2020-09-26 11:12:27.818524
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '7e8d7626e38f'
-down_revision = 'cd2052be6db2'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'block_sync',
-            sa.Column('id', sa.Integer, primary_key=True),
-            sa.Column('blockchain', sa.String, nullable=False, unique=True),
-            sa.Column('block_height_backlog', sa.Integer, nullable=False, default=0),
-            sa.Column('tx_height_backlog', sa.Integer, nullable=False, default=0),
-            sa.Column('block_height_session', sa.Integer, nullable=False, default=0),
-            sa.Column('tx_height_session', sa.Integer, nullable=False, default=0),
-            sa.Column('block_height_head', sa.Integer, nullable=False, default=0),
-            sa.Column('tx_height_head', sa.Integer, nullable=False, default=0),
-            sa.Column('date_created', sa.DateTime, nullable=False),
-            sa.Column('date_updated', sa.DateTime),
-            )
-    op.drop_table('watcher_state')
-    pass
-
-
-def downgrade():
-    op.drop_table('block_sync')
-    op.create_table(
-            'watcher_state',
-            sa.Column('block_number', sa.Integer)
-            )
-    conn = op.get_bind()
-    conn.execute('INSERT INTO watcher_state (block_number) VALUES (0);')
-    pass

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

@@ -1,35 +0,0 @@
-"""Add transaction queue
-
-Revision ID: 8593fa1ca0f4
-Revises: 
-Create Date: 2020-09-22 21:56:42.117047
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '8593fa1ca0f4'
-down_revision = None
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.create_table(
-            'otx',
-            sa.Column('id', sa.Integer, primary_key=True),
-            sa.Column('date_created', sa.DateTime, nullable=False),
-            sa.Column('nonce', sa.Integer, nullable=False),
-            sa.Column('tx_hash', sa.String(66), nullable=False),
-            sa.Column('signed_tx', sa.Text, nullable=False),
-            sa.Column('status', sa.Integer, nullable=False, default=-9),
-            sa.Column('block', sa.Integer),
-            )
-    op.create_index('idx_otx_tx', 'otx', ['tx_hash'], unique=True)
-
-
-def downgrade():
-    op.drop_index('idx_otx_tx')
-    op.drop_table('otx')

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

@@ -0,0 +1,39 @@
+"""Nonce
+
+Revision ID: 9c420530eeb2
+Revises: b125cbf81e32
+Create Date: 2021-04-02 18:38:56.459334
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '9c420530eeb2'
+down_revision = 'b125cbf81e32'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        'nonce',
+        sa.Column('id', sa.Integer, primary_key=True),
+        sa.Column('address_hex', sa.String(42), nullable=False, unique=True),
+        sa.Column('nonce', sa.Integer, nullable=False),
+        )
+
+    op.create_table(
+        'nonce_task_reservation',
+        sa.Column('id', sa.Integer, primary_key=True),
+        sa.Column('address_hex', sa.String(42), nullable=False),
+        sa.Column('nonce', sa.Integer, nullable=False),
+        sa.Column('key', sa.String, nullable=False),
+        sa.Column('date_created', sa.DateTime, nullable=False),
+        )
+
+
+def downgrade():
+    op.drop_table('nonce_task_reservation')
+    op.drop_table('nonce')

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

@@ -1,26 +0,0 @@
-"""Rename block sync table
-
-Revision ID: 9c4bd7491015
-Revises: 9daa16518a91
-Create Date: 2020-10-15 23:45:56.306898
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '9c4bd7491015'
-down_revision = '9daa16518a91'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.rename_table('block_sync', 'otx_sync')
-    pass
-
-
-def downgrade():
-    op.rename_table('otx_sync', 'block_sync')
-    pass

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

@@ -1,30 +0,0 @@
-"""add tx sync state
-
-Revision ID: 9daa16518a91
-Revises: e3b5330ee71c
-Create Date: 2020-10-10 14:43:18.699276
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = '9daa16518a91'
-down_revision = 'e3b5330ee71c'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-#    op.create_table(
-#            'tx_sync',
-#            sa.Column('tx', sa.String(66), nullable=False),
-#            )
-#    op.execute("INSERT INTO tx_sync VALUES('0x0000000000000000000000000000000000000000000000000000000000000000')")
-    pass
-
-
-def downgrade():
-#    op.drop_table('tx_sync')
-    pass

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

@@ -1,34 +0,0 @@
-"""Add date accessed to txcache
-
-Revision ID: a2e2aab8f331
-Revises: 49b348246d70
-Create Date: 2020-12-24 18:58:06.137812
-
-"""
-from alembic import op
-import sqlalchemy as sa
-
-
-# revision identifiers, used by Alembic.
-revision = 'a2e2aab8f331'
-down_revision = '49b348246d70'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    op.add_column(
-        'tx_cache',
-        sa.Column(
-            'date_checked',
-            sa.DateTime,
-            nullable=False
-        )
-    )
-    pass
-
-
-def downgrade():
-    # drop does not work withs qlite
-    #op.drop_column('tx_cache', 'date_checked')
-    pass