Date: 2025-01-15
Source: Perplexity Research Response
Status: Core architectural foundation - build this first
Sneakerware is not a feature - it's the architectural foundation. Building event-sourced, append-only logs with Ed25519 signatures from day one means:
User Action → Create Event → Event Log (Source of Truth)
↓
Materialized View (Fast Queries)
↓
Export/Import (Events)
Key Principle: Event log is primary, everything else is derived.
Cryptographic identity for this Rhizome instance (like SSB, Syncthing).
# models/instance_identity.py
class InstanceIdentity(db.Model):
"""Cryptographic identity for this Rhizome instance"""
id = Column(UUID, primary_key=True, default=uuid.uuid4)
# Ed25519 keypair (like SSB)
public_key = Column(String(64), nullable=False, unique=True)
private_key_encrypted = Column(Text, nullable=False) # Encrypted at rest
# Instance ID = hash of public key (like Syncthing)
instance_id = Column(String(64), nullable=False, unique=True, index=True)
# Human-readable name
instance_name = Column(String(100), default="My Rhizome")
# Trust relationships for federation
trusted_instances = Column(JSON, default=dict) # {instance_id: trust_level}
blocked_instances = Column(JSON, default=list)
created_at = Column(DateTime(timezone=True), default=datetime.utcnow)
@classmethod
def get_or_create_instance_identity(cls):
"""Singleton pattern - one identity per instance"""
identity = cls.query.first()
if not identity:
identity = cls.generate_new_identity()
db.session.add(identity)
db.session.commit()
return identity
@classmethod
def generate_new_identity(cls):
from cryptography.hazmat.primitives.asymmetric import ed25519
import hashlib
private_key = ed25519.Ed25519PrivateKey.generate()
public_key = private_key.public_key()
public_key_bytes = public_key.public_bytes(
encoding=serialization.Encoding.Raw,
format=serialization.PublicFormat.Raw
)
# Instance ID = SHA256(public_key)[:32]
instance_id = hashlib.sha256(public_key_bytes).hexdigest()[:32]
return cls(
public_key=public_key_bytes.hex(),
private_key_encrypted=encrypt_private_key(private_key),
instance_id=instance_id
)
def sign_data(self, data):
"""Sign any data with instance private key"""
private_key = decrypt_private_key(self.private_key_encrypted)
signature = private_key.sign(json.dumps(data, sort_keys=True).encode())
return signature.hex()
def verify_signature(self, data, signature, public_key_hex):
"""Verify signature from another instance"""
from cryptography.hazmat.primitives.asymmetric import ed25519
public_key = ed25519.Ed25519PublicKey.from_public_bytes(
bytes.fromhex(public_key_hex)
)
try:
public_key.verify(
bytes.fromhex(signature),
json.dumps(data, sort_keys=True).encode()
)
return True
except:
return False
Immutable, signed event log (SSB-inspired). Tamper-proof, USB-safe.
# models/rhizome_event.py
class RhizomeEvent(db.Model):
"""Immutable, signed event log (SSB-inspired)"""
id = Column(UUID, primary_key=True, default=uuid.uuid4)
# Append-only sequence number (per user)
user_id = Column(Integer, ForeignKey('user.id'), nullable=False)
sequence = Column(Integer, nullable=False) # Auto-increment per user
# Event type
event_type = Column(Enum('POST', 'COMMENT', 'LIKE', 'RESHARE', 'CIRCLE_UPDATE',
'TAG_ADD', 'STAR', name='event_types'), nullable=False)
# Event payload (immutable JSON)
payload = Column(JSON, nullable=False)
# Cryptographic chain
previous_hash = Column(String(64)) # SHA256 of previous event
content_hash = Column(String(64), nullable=False, unique=True, index=True) # SHA256 of this event
signature = Column(String(128), nullable=False) # Ed25519 signature
# Timestamps
created_at = Column(DateTime(timezone=True), default=datetime.utcnow)
# Metadata for sync
synced_to_instances = Column(JSON, default=list) # [instance_id1, instance_id2]
export_count = Column(Integer, default=0)
__table_args__ = (
Index('idx_user_sequence', 'user_id', 'sequence', unique=True),
Index('idx_content_hash', 'content_hash'),
)
@classmethod
def create_event(cls, user, event_type, payload):
"""Create new event in user's append-only feed"""
instance = InstanceIdentity.get_or_create_instance_identity()
# Get previous event to build chain
previous_event = cls.query.filter_by(user_id=user.id).order_by(cls.sequence.desc()).first()
sequence = (previous_event.sequence + 1) if previous_event else 1
previous_hash = previous_event.content_hash if previous_event else None
# Build event
event_data = {
'user_id': user.id,
'sequence': sequence,
'event_type': event_type.value,
'payload': payload,
'previous_hash': previous_hash,
'timestamp': datetime.utcnow().isoformat()
}
# Hash content
content_hash = hashlib.sha256(
json.dumps(event_data, sort_keys=True).encode()
).hexdigest()
# Sign event
signature = instance.sign_data(event_data)
event = cls(
user_id=user.id,
sequence=sequence,
event_type=event_type,
payload=payload,
previous_hash=previous_hash,
content_hash=content_hash,
signature=signature
)
db.session.add(event)
db.session.commit()
return event
def verify_chain(self):
"""Verify this event's chain integrity"""
if self.previous_hash:
previous = RhizomeEvent.query.filter_by(
user_id=self.user_id,
content_hash=self.previous_hash
).first()
if not previous or previous.sequence != self.sequence - 1:
return False
# Verify signature
instance = InstanceIdentity.get_or_create_instance_identity()
event_data = {
'user_id': self.user_id,
'sequence': self.sequence,
'event_type': self.event_type.value,
'payload': self.payload,
'previous_hash': self.previous_hash,
'timestamp': self.created_at.isoformat()
}
return instance.verify_signature(event_data, self.signature, instance.public_key)
Track what each instance has seen (enables incremental exports).
# models/sync_state.py
class SyncState(db.Model):
"""Vector clock for tracking sync state with other instances"""
id = Column(UUID, primary_key=True, default=uuid.uuid4)
# Local user
user_id = Column(Integer, ForeignKey('user.id'), nullable=False)
# Remote instance we're syncing with
remote_instance_id = Column(String(64), nullable=False)
# Vector clock: {user_id: last_seen_sequence}
vector_clock = Column(JSON, default=dict)
# Last sync metadata
last_sync_at = Column(DateTime(timezone=True))
last_sync_method = Column(String(20)) # 'usb', 'https', 'bluetooth'
updated_at = Column(DateTime(timezone=True), default=datetime.utcnow, onupdate=datetime.utcnow)
__table_args__ = (
Index('idx_user_remote', 'user_id', 'remote_instance_id', unique=True),
)
@classmethod
def get_delta_since_last_sync(cls, user_id, remote_instance_id):
"""Get events that remote instance hasn't seen yet"""
sync_state = cls.query.filter_by(
user_id=user_id,
remote_instance_id=remote_instance_id
).first()
if not sync_state:
# First sync - send everything
return RhizomeEvent.query.filter_by(user_id=user_id).all()
# Get events after last seen sequence
vector_clock = sync_state.vector_clock
new_events = []
for followed_user_id, last_sequence in vector_clock.items():
events = RhizomeEvent.query.filter(
RhizomeEvent.user_id == followed_user_id,
RhizomeEvent.sequence > last_sequence
).order_by(RhizomeEvent.sequence).all()
new_events.extend(events)
return new_events
@classmethod
def update_after_sync(cls, user_id, remote_instance_id, imported_events):
"""Update vector clock after successful sync"""
sync_state = cls.query.filter_by(
user_id=user_id,
remote_instance_id=remote_instance_id
).first()
if not sync_state:
sync_state = cls(user_id=user_id, remote_instance_id=remote_instance_id)
db.session.add(sync_state)
# Update vector clock
for event in imported_events:
current_seq = sync_state.vector_clock.get(str(event.user_id), 0)
sync_state.vector_clock[str(event.user_id)] = max(current_seq, event.sequence)
sync_state.last_sync_at = datetime.utcnow()
flag_modified(sync_state, 'vector_clock')
db.session.commit()
Handle USB export/import with signatures and verification.
See docs/implementation_plans/mvp_demo_roadmap.md Phase 0.4 for full implementation.
Key Features:
- JSON export with signatures (human-readable, inspectable)
- Export includes: all events, instance metadata, vector clock
- Import with verification (signatures, chain integrity)
- Deduplication (content_hash prevents duplicate imports)
- Trust model (user approval for unknown instances)
- Export size estimation (show before export)
- Progress tracking (for large exports)
- Partial import resume (resume from last successful event)
Crypto Service:
# services/crypto_service.py
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import serialization
def encrypt_private_key(private_key, password=None):
"""Encrypt private key for storage"""
if password is None:
password = get_or_create_encryption_key()
cipher = Fernet(password)
key_bytes = private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption()
)
return cipher.encrypt(key_bytes).decode()
def decrypt_private_key(encrypted_key, password=None):
"""Decrypt private key from storage"""
if password is None:
password = get_or_create_encryption_key()
cipher = Fernet(password)
key_bytes = cipher.decrypt(encrypted_key.encode())
return serialization.load_pem_private_key(key_bytes, password=None)
def get_or_create_encryption_key():
"""Get encryption key from environment or generate"""
key = os.getenv('RHIZOME_ENCRYPTION_KEY')
if not key:
key = Fernet.generate_key()
# Store in .env file
with open('.env', 'a') as f:
f.write(f'\nRHIZOME_ENCRYPTION_KEY={key.decode()}\n')
return key.encode() if isinstance(key, str) else key
JSON Encoder:
# utils/json_encoder.py
import json
from datetime import datetime
from decimal import Decimal
from uuid import UUID
class RhizomeJSONEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime):
return obj.isoformat()
elif isinstance(obj, UUID):
return str(obj)
elif isinstance(obj, Decimal):
return float(obj)
return super().default(obj)
# Usage
json.dumps(data, sort_keys=True, cls=RhizomeJSONEncoder)
Event Schema Validation:
# models/event_schemas.py
import jsonschema
EVENT_SCHEMAS = {
'STAR': {
'required': ['entry_id', 'entry_type', 'entry_uri', 'action'],
'properties': {
'entry_id': {'type': 'string'},
'entry_type': {'type': 'string'},
'entry_uri': {'type': 'string'},
'action': {'enum': ['star', 'unstar']}
}
},
'COMMENT': {
'required': ['content', 'target_uri'],
'properties': {
'content': {'type': 'string', 'maxLength': 5000},
'target_uri': {'type': 'string'},
'parent_comment_id': {'type': 'string', 'nullable': True}
}
},
# ... more schemas
}
def validate_event_payload(event_type, payload):
schema = EVENT_SCHEMAS.get(event_type)
if not schema:
raise ValueError(f"Unknown event type: {event_type}")
jsonschema.validate(payload, schema)
Batch Chain Verification:
def verify_chain_batch(events):
"""Verify chain for multiple events efficiently"""
events_by_hash = {e.content_hash: e for e in events}
for event in events:
if event.previous_hash:
previous = events_by_hash.get(event.previous_hash)
if not previous or previous.sequence != event.sequence - 1:
return False
# Verify signature
if not event.verify_signature():
return False
return True
# services/migration_service.py
def migrate_existing_data_to_events():
"""One-time migration: convert existing data to event log"""
instance = InstanceIdentity.get_or_create_instance_identity()
# Migrate all user actions to events
for user in User.query.all():
sequence = 1
previous_hash = None
# Migrate starred items
for entry in UserFeedEntry.query.filter_by(user_id=user.id, starred=True).all():
event = RhizomeEvent.create_event(
user=user,
event_type=EventType.STAR,
payload={'entry_id': entry.id}
)
sequence += 1
# Migrate comments (if any)
# Migrate likes (if any)
# etc.
┌─────────────────────────────────────────────────┐
│ USER ACTIONS │
│ (Star, Comment, Like, Share, Create Circle) │
└──────────────────┬──────────────────────────────┘
│
↓
┌─────────────────────────────────────────────────┐
│ RHIZOME EVENT LOG │
│ • Append-only │
│ • Ed25519 signed │
│ • SHA256 hash-chained │
│ • Vector clock timestamped │
└──────────┬────────────────────┬──────────────────┘
│ │
↓ ↓
┌──────────────────┐ ┌────────────────────────┐
│ MATERIALIZED │ │ EXPORT/IMPORT │
│ VIEWS │ │ (Sneakerware) │
│ • user_feed_ │ │ │
│ entries │ │ • USB (JSON) │
│ • starred │ │ • Bluetooth │
│ • circles │ │ • HTTPS │
│ (Fast queries) │ │ • Mesh network │
└──────────────────┘ └────────────────────────┘
Key: Event log is primary, everything else is derived.
Adjusted Estimates (Perplexity Feedback):
| Phase | Original Estimate | Adjusted Estimate | Notes |
|---|---|---|---|
| Phase 0 | 1 day (8-10 hours) | 1.5 days (12 hours) | Crypto + testing takes longer |
| Phase 1 | 4 days | 4 days | Reasonable |
| Phase 2 | 3 days | 3 days | Reasonable |
| Phase 3 | 2 days | 2 days | Reasonable |
| Total | 10 days | 10.5 days | Add 0.5 day buffer |
Key Recommendations:
- Test export/import on Day 1 (don't wait until Day 10)
- Document as you go (architecture decisions, patterns)
- Accept "good enough" for MVP (perfect is enemy of shipped)
- Build in buffer for Phase 0 (crypto + testing)
This architecture is the foundation for all RHIZOME features. Build it first, then build features on top.
← Back to Splash