94 lines
3.0 KiB
Python
94 lines
3.0 KiB
Python
import json
|
|
import time
|
|
import hashlib
|
|
from typing import List, Dict, Any, Optional
|
|
|
|
class ContractRecord(dict):
|
|
"""A lightweight wrapper for contract payloads stored in the ledger."""
|
|
def __init__(self, contract_type: str, payload: Dict[str, Any]):
|
|
super().__init__(payload)
|
|
self["type"] = contract_type
|
|
|
|
def to_json(self) -> str:
|
|
return json.dumps(self, sort_keys=True)
|
|
|
|
|
|
class Block:
|
|
def __init__(self, index: int, previous_hash: str, data: List[ContractRecord], timestamp: Optional[float] = None):
|
|
self.index = index
|
|
self.timestamp = timestamp if timestamp is not None else time.time()
|
|
self.previous_hash = previous_hash
|
|
self.data = data
|
|
self.hash = self.compute_hash()
|
|
|
|
def compute_hash(self) -> str:
|
|
payload = {
|
|
"index": self.index,
|
|
"timestamp": self.timestamp,
|
|
"previous_hash": self.previous_hash,
|
|
"data": [d for d in self.data],
|
|
}
|
|
serialized = json.dumps(payload, sort_keys=True, default=str)
|
|
return hashlib.sha256(serialized.encode("utf-8")).hexdigest()
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"index": self.index,
|
|
"timestamp": self.timestamp,
|
|
"previous_hash": self.previous_hash,
|
|
"hash": self.hash,
|
|
"data": [d for d in self.data],
|
|
}
|
|
|
|
|
|
class Ledger:
|
|
def __init__(self):
|
|
self.blocks: List[Block] = [] # append-only
|
|
self._init_genesis()
|
|
|
|
def _init_genesis(self):
|
|
genesis = Block(index=0, previous_hash="0" * 64, data=[])
|
|
self.blocks.append(genesis)
|
|
|
|
@property
|
|
def head(self) -> Block:
|
|
return self.blocks[-1]
|
|
|
|
def head_hash(self) -> str:
|
|
return self.head.hash
|
|
|
|
def add_record(self, contract_type: str, payload: Dict[str, Any]) -> Block:
|
|
record = ContractRecord(contract_type, payload)
|
|
prev_hash = "0" * 64 if self.head.index == 0 else self.head.hash
|
|
new_block = Block(
|
|
index=self.head.index + 1,
|
|
previous_hash=prev_hash,
|
|
data=[record],
|
|
)
|
|
self.blocks.append(new_block)
|
|
return new_block
|
|
|
|
def to_json(self) -> str:
|
|
return json.dumps([b.to_dict() for b in self.blocks], sort_keys=True, default=str)
|
|
|
|
|
|
def delta_sync(our_ledger: Ledger, remote_head_hash: str) -> List[Dict[str, Any]]:
|
|
# Return blocks that the remote hasn't seen yet, given the remote head hash.
|
|
# If remote_head_hash is unknown, return entire chain except genesis for safety.
|
|
blocks = []
|
|
start_index = 1 # skip genesis for delta
|
|
if remote_head_hash:
|
|
# locate the block with this hash
|
|
idx = None
|
|
for i, b in enumerate(our_ledger.blocks):
|
|
if b.hash == remote_head_hash:
|
|
idx = i
|
|
break
|
|
if idx is None:
|
|
start_index = 1
|
|
else:
|
|
start_index = idx + 1
|
|
for b in our_ledger.blocks[start_index:]:
|
|
blocks.append(b.to_dict())
|
|
return blocks
|