Spaces:
Running on Zero
Running on Zero
File size: 1,595 Bytes
31c93b1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 | from __future__ import annotations
import base64
import secrets
import time
from dataclasses import dataclass
from typing import Any, Literal
EventType = Literal[
"community.created",
"community.member.joined",
"community.member.left",
"community.member.invited",
"community.member.removed",
"community.member.revoked",
"community.member.promoted",
"community.member.demoted",
"community.policy.updated",
"node.manifest.updated",
"market.post.created",
"market.post.updated",
"market.post.expired",
"chat.message.sent",
"chat.message.delivered",
"chat.message.read",
"rag.document.ingested",
"file.advertised",
"file.cid.advertised",
"file.cid.unpinned",
"federation.peer.added",
"federation.peer.removed",
]
_ALL_EVENT_TYPES: frozenset[str] = frozenset(EventType.__args__) # type: ignore[attr-defined]
def new_ulid() -> str:
"""Generate a sortable unique ID (ULID-compatible 26-char string)."""
ts = int(time.time() * 1000)
ts_bytes = ts.to_bytes(10, "big")
rand_bytes = secrets.token_bytes(10)
raw = ts_bytes + rand_bytes # 20 bytes
encoded = base64.b32encode(raw).decode("ascii") # 32 chars
return encoded[:26]
@dataclass(frozen=True)
class Event:
schema_version: int # always 1
event_id: str # ULID
event_type: EventType
community_id: str
author: str # full node_id
lamport: int
payload: dict[str, Any]
issued_at: str # RFC 3339 UTC
signature: str # "ed25519:<b64url>" or ""
|