From 7ae6ea1f08379d2b2893d1cecf8326738ad5efc9 Mon Sep 17 00:00:00 2001 From: Jonathan Zernik Date: Thu, 9 Apr 2020 17:49:53 -0700 Subject: [PATCH] Upload squeaks to server (#13) * Include address in squeak database. * Add tables to db for uploader * Include address in squeak database. * Add tables to db for uploader * Add uploader class. * Fix foreign key relation in client db. * Change uploader to check all squeaks based on address and block number. * Start and stop uploader thread in client. --- squeaknode/client/clientsqueaknode.py | 22 +++++-- squeaknode/client/hub_store.py | 55 ++++++++++++++++ squeaknode/client/main.py | 3 +- squeaknode/client/schema.sql | 18 ++++++ squeaknode/client/squeak_store.py | 91 +++++++++++++++++++++------ squeaknode/client/uploader.py | 56 +++++++++++++++++ 6 files changed, 221 insertions(+), 24 deletions(-) create mode 100644 squeaknode/client/hub_store.py create mode 100644 squeaknode/client/uploader.py diff --git a/squeaknode/client/clientsqueaknode.py b/squeaknode/client/clientsqueaknode.py index e65882d0..3c7599e2 100644 --- a/squeaknode/client/clientsqueaknode.py +++ b/squeaknode/client/clientsqueaknode.py @@ -1,12 +1,15 @@ import logging +import threading from squeak.core.signing import CSigningKey +from squeak.core.signing import CSqueakAddress from squeaknode.client.squeak_store import SqueakStore from squeaknode.common.blockchain_client import BlockchainClient from squeaknode.common.lightning_client import LightningClient from squeaknode.common.squeak_maker import SqueakMaker from squeaknode.client.db import SQLiteDBFactory +from squeaknode.client.uploader import Uploader logger = logging.getLogger(__name__) @@ -26,10 +29,22 @@ class SqueakNodeClient(object): self.blockchain_client = blockchain_client self.lightning_client = lightning_client self.signing_key = signing_key + self.address = CSqueakAddress.from_verifying_key(signing_key.get_verifying_key()) self.squeak_store = SqueakStore(db_factory) + self.hub_store = None + + # Event is set when the client stops + self.stopped = threading.Event() + self.uploader = Uploader(self.hub_store, self.squeak_store, self.address, self.stopped) + + def start(self): + self.uploader.start() + + def stop(self): + self.stopped.set() def get_address(self): - pass + return self.address def make_squeak(self, content): if self.signing_key is None: @@ -38,11 +53,8 @@ class SqueakNodeClient(object): squeak_maker = SqueakMaker(self.signing_key, self.blockchain_client) squeak = squeak_maker.make_squeak(content) logger.info('Made squeak: {}'.format(squeak)) - self.add_squeak(squeak) - return squeak - - def add_squeak(self, squeak): self.squeak_store.save_squeak(squeak) + return squeak def get_squeak(self, squeak_hash): return self.squeak_store.get_squeak(squeak_hash) diff --git a/squeaknode/client/hub_store.py b/squeaknode/client/hub_store.py new file mode 100644 index 00000000..e3b7af64 --- /dev/null +++ b/squeaknode/client/hub_store.py @@ -0,0 +1,55 @@ +import logging + +from squeak.core import CSqueak +from squeak.core import CheckSqueak +from squeak.core import HASH_LENGTH +from squeak.core import MakeSqueakFromStr +from squeak.core import CSqueakEncContent +from squeak.core.signing import CSigningKey +from squeak.core.signing import CSigningKey +from squeak.core.script import CScript + +from squeaknode.client.db import get_db +from squeaknode.client.db import close_db +from squeaknode.client.db import initialize_db +from squeaknode.client.db import SQLiteDBFactory + + +logger = logging.getLogger(__name__) + + +class HubStore(object): + """Network node that handles client commands. + """ + + def __init__( + self, + db_factory: SQLiteDBFactory, + ) -> None: + self.db_factory = db_factory + + def save_hub(self, hub): + host, port = hub + with self.db_factory.make_conn() as conn: + conn.execute( + "INSERT INTO hub (host, port) VALUES (?, ?)", + (host, port), + ) + + def get_hubs(self): + with self.db_factory.make_conn() as conn: + hub_rows = ( + conn + .execute( + "SELECT host, port" + " FROM hub h", + ) + .fetchall() + ) + hubs = [] + for hub_row in hub_rows: + hubs.append( + hub_row['host'], + hub_row['port'], + ) + return hubs diff --git a/squeaknode/client/main.py b/squeaknode/client/main.py index c5b15218..f0fb6553 100644 --- a/squeaknode/client/main.py +++ b/squeaknode/client/main.py @@ -141,12 +141,13 @@ def run_client(config): signal.signal(signal.SIGTERM, sigterm_handler) print("Starting client...") + node.start() try: while True: time.sleep(1) finally: print("Shutting down...") - close_db(db) + node.stop() if __name__ == '__main__': diff --git a/squeaknode/client/schema.sql b/squeaknode/client/schema.sql index 89fcd2c7..8828b00c 100644 --- a/squeaknode/client/schema.sql +++ b/squeaknode/client/schema.sql @@ -18,6 +18,24 @@ CREATE TABLE squeak ( nNonce INTEGER NOT NULL, encContent BLOB NOT NULL, scriptSig BLOB NOT NULL, + address TEXT NOT NULL, vchDataKey TEXT, content BLOB ); + + +CREATE TABLE hub ( + host TEXT NOT NULL, + port TEXT NOT NULL, + created TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY(host, port) +); + + +CREATE TABLE upload ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + created TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + squeakHash TEXT NOT NULL, + complete INTEGER NOT NULL, + FOREIGN KEY (squeakHash) REFERENCES [squeak] (hash) ON UPDATE NO ACTION ON DELETE CASCADE +); diff --git a/squeaknode/client/squeak_store.py b/squeaknode/client/squeak_store.py index ea49da1e..8bf07b0c 100644 --- a/squeaknode/client/squeak_store.py +++ b/squeaknode/client/squeak_store.py @@ -28,11 +28,33 @@ class SqueakStore(object): ) -> None: self.db_factory = db_factory + def _row_to_squeak(self, squeak_row): + squeak = CSqueak( + nVersion=squeak_row['nVersion'], + hashEncContent=squeak_row['hashEncContent'], + hashReplySqk=squeak_row['hashReplySqk'], + hashBlock=squeak_row['hashBlock'], + nBlockHeight=squeak_row['nBlockHeight'], + scriptPubKey=CScript(squeak_row['scriptPubKey']), + hashDataKey=squeak_row['hashDataKey'], + vchIv=squeak_row['vchIv'], + nTime=squeak_row['nTime'], + nNonce=squeak_row['nNonce'], + encContent=CSqueakEncContent(squeak_row['encContent']), + scriptSig=CScript(squeak_row['scriptSig']), + vchDataKey=squeak_row['vchDataKey'], + ) + try: + CheckSqueak(squeak) + return squeak + except: + return None + def save_squeak(self, squeak): CheckSqueak(squeak) with self.db_factory.make_conn() as conn: conn.execute( - "INSERT INTO squeak (hash, nVersion, hashEncContent, hashReplySqk, hashBlock, nBlockHeight, scriptPubKey, hashDataKey, vchIv, nTime, nNonce, encContent, scriptSig, vchDataKey, content) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + "INSERT INTO squeak (hash, nVersion, hashEncContent, hashReplySqk, hashBlock, nBlockHeight, scriptPubKey, hashDataKey, vchIv, nTime, nNonce, encContent, scriptSig, address, vchDataKey, content) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ( squeak.GetHash(), squeak.nVersion, @@ -47,6 +69,7 @@ class SqueakStore(object): squeak.nNonce, bytes(squeak.encContent.vchEncContent), bytes(squeak.scriptSig), + str(squeak.GetAddress()), squeak.vchDataKey, "", ), @@ -66,26 +89,58 @@ class SqueakStore(object): ) if squeak_row is None: return None - squeak = CSqueak( - nVersion=squeak_row['nVersion'], - hashEncContent=squeak_row['hashEncContent'], - hashReplySqk=squeak_row['hashReplySqk'], - hashBlock=squeak_row['hashBlock'], - nBlockHeight=squeak_row['nBlockHeight'], - scriptPubKey=CScript(squeak_row['scriptPubKey']), - hashDataKey=squeak_row['hashDataKey'], - vchIv=squeak_row['vchIv'], - nTime=squeak_row['nTime'], - nNonce=squeak_row['nNonce'], - encContent=CSqueakEncContent(squeak_row['encContent']), - scriptSig=CScript(squeak_row['scriptSig']), - vchDataKey=squeak_row['vchDataKey'], - ) - CheckSqueak(squeak) - return squeak + return self._row_to_squeak(squeak_row) def delete_squeak(self): pass def unlock_squeak(self): pass + + def get_squeaks_to_upload(self, address, min_block=0): + """Get all squeaks that need to be uploaded. + """ + with self.db_factory.make_conn() as conn: + squeak_row = ( + conn + .execute( + "SELECT s.hash, nVersion, hashEncContent, hashReplySqk, hashBlock, nBlockHeight, scriptPubKey, hashDataKey, vchIv, nTime, nNonce, encContent, scriptSig, vchDataKey" + " FROM squeak s" + " WHERE s.address = ?", + (address,), + ) + .fetchone() + ) + if squeak_rows is None: + return None + squeaks = [] + for squeak_row in squeak_rows: + squeak = self._row_to_squeak(squeak_row) + squeaks.append(squeak) + return squeaks + + def mark_squeak_uploaded(self, squeak_hash): + """Mark the given squeak as one that needs to be uploaded. + """ + with self.db_factory.make_conn() as conn: + conn.execute( + "INSERT INTO upload (squeakHash, complete) VALUES (?, ?)", + (squeak_hash, 1), + ) + + def is_squeak_uploaded(self, squeak_hash): + """True if the squeak has already been uploaded. + """ + with self.db_factory.make_conn() as conn: + upload_row = ( + conn + .execute( + "SELECT u.squeakHash, complete" + " FROM upload u" + " WHERE u.squeakHash = ?", + (squeak_hash,), + ) + .fetchone() + ) + complete = upload_row['complete'] + return complete == 1 diff --git a/squeaknode/client/uploader.py b/squeaknode/client/uploader.py new file mode 100644 index 00000000..4831e9ac --- /dev/null +++ b/squeaknode/client/uploader.py @@ -0,0 +1,56 @@ +import logging +import threading + +from squeak.core.signing import CSigningKey +from squeak.core.signing import CSqueakAddress + +from squeaknode.client.squeak_store import SqueakStore +from squeaknode.common.blockchain_client import BlockchainClient +from squeaknode.common.lightning_client import LightningClient +from squeaknode.common.squeak_maker import SqueakMaker +from squeaknode.client.db import SQLiteDBFactory +from squeaknode.client.hub_store import HubStore + + +logger = logging.getLogger(__name__) + + +class Uploader(object): + """Uploads squeaks to the hubs. + """ + + def __init__( + self, + hub_store: HubStore, + squeak_store: SqueakStore, + address: CSqueakAddress, + stopped: threading.Event, + ) -> None: + self.hub_store = hub_store + self.squeak_store = squeak_store + self.address = address + self.stopped = stopped + + def upload_squeak(self, squeak): + squeak_hash = squeak.GetHash() + if self.squeak_store.is_squeak_uploaded(squeak_hash): + # TODO: Upload squeak here. + ## + logger.info("Upload the squeak here.") + self.squeak_store.mark_squeak_uploaded(squeak_hash) + + def upload_squeaks(self): + for squeak in self.squeak_store.get_squeaks_to_upload(self.address): + self.upload_squeak(squeak) + + def start(self): + while True: + try: + self.upload_squeaks() + except: + self.stopped.set() + + # Check if client is stopped. + is_stopped = self.stopped.wait(10) + if is_stopped: + break