feat: extracted

models.py

@@ -0,0 +1,116 @@
+import hashlib
+import json
+from enum import Enum
+from sqlite3 import Row
+from typing import List, Optional
+
+from pydantic import BaseModel, Field
+from secp256k1 import PublicKey
+
+
+class NostrRelay(BaseModel):
+    id: str
+    wallet: str
+    name: str
+    currency: str
+    tip_options: Optional[str]
+    tip_wallet: Optional[str]
+
+    @classmethod
+    def from_row(cls, row: Row) -> "NostrRelay":
+        return cls(**dict(row))
+
+
+class NostrEvent(BaseModel):
+    id: str
+    pubkey: str
+    created_at: int
+    kind: int
+    tags: List[List[str]] = []
+    content: str = ""
+    sig: str
+
+    def serialize(self) -> List:
+        return [0, self.pubkey, self.created_at, self.kind, self.tags, self.content]
+
+    def serialize_json(self) -> str:
+        e = self.serialize()
+        return json.dumps(e, separators=(",", ":"))
+
+    @property
+    def event_id(self) -> str:
+        data = self.serialize_json()
+        id = hashlib.sha256(data.encode()).hexdigest()
+        return id
+
+    def check_signature(self):
+        event_id = self.event_id
+        if self.id != event_id:
+            raise ValueError(
+                f"Invalid event id. Expected: '{event_id}' got '{self.id}'"
+            )
+        try:
+            pub_key = PublicKey(bytes.fromhex("02" + self.pubkey), True)
+        except Exception:
+            raise ValueError(
+                f"Invalid public key: '{self.pubkey}' for event '{self.id}'"
+            )
+
+        valid_signature = pub_key.schnorr_verify(
+            bytes.fromhex(event_id), bytes.fromhex(self.sig), None, raw=True
+        )
+        if not valid_signature:
+            raise ValueError(f"Invalid signature: '{self.sig}' for event '{self.id}'")
+
+    @classmethod
+    def from_row(cls, row: Row) -> "NostrEvent":
+        return cls(**dict(row))
+
+
+class NostrFilter(BaseModel):
+    subscription_id: Optional[str]
+
+    ids: List[str] = []
+    authors: List[str] = []
+    kinds: List[int] = []
+    e: List[str] = Field([], alias="#e")
+    p: List[str] = Field([], alias="#p")
+    since: Optional[int]
+    until: Optional[int]
+    limit: Optional[int]
+
+    def matches(self, e: NostrEvent) -> bool:
+        # todo: starts with
+        if len(self.ids) != 0 and e.id not in self.ids:
+            return False
+        if len(self.authors) != 0 and e.pubkey not in self.authors:
+            return False
+        if len(self.kinds) != 0 and e.kind not in self.kinds:
+            return False
+
+        if self.since and e.created_at < self.since:
+            return False
+        if self.until and self.until > 0 and e.created_at > self.until:
+            return False
+
+        found_e_tag = self.tag_in_list(e.tags, "e")
+        found_p_tag = self.tag_in_list(e.tags, "p")
+        if not found_e_tag or not found_p_tag:
+            return False
+
+        return True
+
+    def tag_in_list(self, event_tags, tag_name):
+        tag_values = [t[1] for t in event_tags if t[0] == tag_name]
+        if len(tag_values) == 0:
+            return True
+        common_tags = [t for t in tag_values if t in self.e]
+        if len(common_tags) == 0:
+            return False
+
+
+
+class NostrEventType(str, Enum):
+    EVENT = "EVENT"
+    REQ = "REQ"
+    CLOSE = "CLOSE"