134 lines
4.7 KiB
Python
134 lines
4.7 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import sqlite3
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Iterable, Mapping
|
|
|
|
|
|
def _canonical_json(value: Any) -> str:
|
|
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
|
|
|
|
|
|
def _hash_chain(*parts: str) -> str:
|
|
digest = hashlib.sha256()
|
|
for part in parts:
|
|
digest.update(part.encode())
|
|
return digest.hexdigest()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DeltaRecord:
|
|
site_id: str
|
|
sequence: int
|
|
kind: str
|
|
payload: dict[str, Any]
|
|
author_did: str
|
|
prev_hash: str
|
|
hash: str
|
|
|
|
|
|
class OfflineDeltaStore:
|
|
def __init__(self, db_path: str | Path = ":memory:") -> None:
|
|
self.db_path = str(db_path)
|
|
self.conn = sqlite3.connect(self.db_path)
|
|
self.conn.row_factory = sqlite3.Row
|
|
self._ensure_schema()
|
|
|
|
def _ensure_schema(self) -> None:
|
|
self.conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS deltas (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
site_id TEXT NOT NULL,
|
|
sequence INTEGER NOT NULL,
|
|
kind TEXT NOT NULL,
|
|
payload TEXT NOT NULL,
|
|
author_did TEXT NOT NULL,
|
|
prev_hash TEXT NOT NULL,
|
|
hash TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
self.conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_deltas_site_sequence ON deltas(site_id, sequence)")
|
|
self.conn.commit()
|
|
|
|
def latest_sequence(self, site_id: str) -> int:
|
|
row = self.conn.execute("SELECT MAX(sequence) AS sequence FROM deltas WHERE site_id = ?", (site_id,)).fetchone()
|
|
return int(row["sequence"] or 0)
|
|
|
|
def latest_hash(self, site_id: str) -> str:
|
|
row = self.conn.execute("SELECT hash FROM deltas WHERE site_id = ? ORDER BY sequence DESC LIMIT 1", (site_id,)).fetchone()
|
|
return row["hash"] if row else "GENESIS"
|
|
|
|
def append(self, site_id: str, kind: str, payload: Mapping[str, Any], author_did: str) -> DeltaRecord:
|
|
sequence = self.latest_sequence(site_id) + 1
|
|
prev_hash = self.latest_hash(site_id)
|
|
payload_json = _canonical_json(payload)
|
|
hash_value = _hash_chain(site_id, str(sequence), kind, payload_json, author_did, prev_hash)
|
|
self.conn.execute(
|
|
"""
|
|
INSERT INTO deltas (site_id, sequence, kind, payload, author_did, prev_hash, hash)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(site_id, sequence, kind, payload_json, author_did, prev_hash, hash_value),
|
|
)
|
|
self.conn.commit()
|
|
return DeltaRecord(site_id, sequence, kind, dict(payload), author_did, prev_hash, hash_value)
|
|
|
|
def records(self, site_id: str) -> list[DeltaRecord]:
|
|
rows = self.conn.execute("SELECT * FROM deltas WHERE site_id = ? ORDER BY sequence ASC", (site_id,)).fetchall()
|
|
return [
|
|
DeltaRecord(
|
|
site_id=row["site_id"],
|
|
sequence=row["sequence"],
|
|
kind=row["kind"],
|
|
payload=json.loads(row["payload"]),
|
|
author_did=row["author_did"],
|
|
prev_hash=row["prev_hash"],
|
|
hash=row["hash"],
|
|
)
|
|
for row in rows
|
|
]
|
|
|
|
def delta_since(self, site_id: str, sequence: int) -> list[DeltaRecord]:
|
|
rows = self.conn.execute(
|
|
"SELECT * FROM deltas WHERE site_id = ? AND sequence > ? ORDER BY sequence ASC",
|
|
(site_id, sequence),
|
|
).fetchall()
|
|
return [
|
|
DeltaRecord(
|
|
site_id=row["site_id"],
|
|
sequence=row["sequence"],
|
|
kind=row["kind"],
|
|
payload=json.loads(row["payload"]),
|
|
author_did=row["author_did"],
|
|
prev_hash=row["prev_hash"],
|
|
hash=row["hash"],
|
|
)
|
|
for row in rows
|
|
]
|
|
|
|
def replay(self, site_id: str) -> dict[str, Any]:
|
|
state: dict[str, Any] = {}
|
|
for record in self.records(site_id):
|
|
if record.kind == "set":
|
|
state.update(record.payload)
|
|
elif record.kind == "increment":
|
|
for key, value in record.payload.items():
|
|
state[key] = state.get(key, 0) + value
|
|
else:
|
|
state[record.kind] = record.payload
|
|
return state
|
|
|
|
def verify_chain(self, site_id: str) -> bool:
|
|
previous = "GENESIS"
|
|
for record in self.records(site_id):
|
|
recalculated = _hash_chain(record.site_id, str(record.sequence), record.kind, _canonical_json(record.payload), record.author_did, previous)
|
|
if recalculated != record.hash:
|
|
return False
|
|
previous = record.hash
|
|
return True
|