From a6502c0888613bd0377a25e43de8ae306c4de4d7 Mon Sep 17 00:00:00 2001 From: bndw Date: Fri, 13 Feb 2026 17:41:13 -0800 Subject: feat: add SQLite storage layer with binary-first event persistence Storage implementation: - Concrete type with constructor (consumer-side interfaces) - Event storage: protobuf + zstd-compressed canonical JSON - Schema: events, deletions, replaceable_events, auth_challenges, rate_limits - WAL mode, STRICT typing, optimized indexes - Methods: StoreEvent, GetEvent, GetEventWithCanonical, DeleteEvent Dependencies: - modernc.org/sqlite v1.45.0 (pure Go SQLite driver) - github.com/klauspost/compress v1.18.4 (zstd compression) 366 lines, 10 tests passing --- internal/storage/events.go | 216 ++++++++++++++++++++++++++++++++ internal/storage/events_test.go | 264 +++++++++++++++++++++++++++++++++++++++ internal/storage/storage.go | 150 ++++++++++++++++++++++ internal/storage/storage_test.go | 70 +++++++++++ 4 files changed, 700 insertions(+) create mode 100644 internal/storage/events.go create mode 100644 internal/storage/events_test.go create mode 100644 internal/storage/storage.go create mode 100644 internal/storage/storage_test.go (limited to 'internal/storage') diff --git a/internal/storage/events.go b/internal/storage/events.go new file mode 100644 index 0000000..d74fc7e --- /dev/null +++ b/internal/storage/events.go @@ -0,0 +1,216 @@ +package storage + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + + "github.com/klauspost/compress/zstd" + "google.golang.org/protobuf/proto" + + pb "northwest.io/nostr-grpc/api/nostr/v1" +) + +var ( + // ErrEventNotFound is returned when an event ID is not found in storage. + ErrEventNotFound = errors.New("event not found") + + // ErrEventExists is returned when attempting to store a duplicate event. + ErrEventExists = errors.New("event already exists") +) + +// EventData holds the data needed to store an event. +type EventData struct { + Event *pb.Event // Protobuf event + CanonicalJSON []byte // Uncompressed canonical JSON (will be compressed on storage) +} + +// StoreEvent stores an event in the database. +// Returns ErrEventExists if the event ID already exists. +func (s *Storage) StoreEvent(ctx context.Context, data *EventData) error { + // Serialize protobuf + eventBytes, err := proto.Marshal(data.Event) + if err != nil { + return fmt.Errorf("failed to marshal event: %w", err) + } + + // Compress canonical JSON + compressedJSON, err := compressJSON(data.CanonicalJSON) + if err != nil { + return fmt.Errorf("failed to compress canonical JSON: %w", err) + } + + // Serialize tags to JSON + tagsJSON, err := marshalTags(data.Event.Tags) + if err != nil { + return fmt.Errorf("failed to marshal tags: %w", err) + } + + // Insert event + query := ` + INSERT INTO events (id, event_data, canonical_json, pubkey, kind, created_at, content, tags, sig) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ` + + _, err = s.db.ExecContext(ctx, query, + data.Event.Id, + eventBytes, + compressedJSON, + data.Event.Pubkey, + data.Event.Kind, + data.Event.CreatedAt, + data.Event.Content, + tagsJSON, + data.Event.Sig, + ) + + if err != nil { + // Check for unique constraint violation (duplicate ID) + if isDuplicateError(err) { + return ErrEventExists + } + return fmt.Errorf("failed to insert event: %w", err) + } + + return nil +} + +// GetEvent retrieves an event by ID. +// Returns ErrEventNotFound if the event doesn't exist. +func (s *Storage) GetEvent(ctx context.Context, id string) (*pb.Event, error) { + query := ` + SELECT event_data, canonical_json + FROM events + WHERE id = ? AND deleted = 0 + ` + + var eventBytes, compressedJSON []byte + err := s.db.QueryRowContext(ctx, query, id).Scan(&eventBytes, &compressedJSON) + + if err == sql.ErrNoRows { + return nil, ErrEventNotFound + } + if err != nil { + return nil, fmt.Errorf("failed to query event: %w", err) + } + + // Deserialize protobuf + event := &pb.Event{} + if err := proto.Unmarshal(eventBytes, event); err != nil { + return nil, fmt.Errorf("failed to unmarshal event: %w", err) + } + + return event, nil +} + +// GetEventWithCanonical retrieves an event by ID with its canonical JSON. +// The canonical_json field will be populated in the returned event. +func (s *Storage) GetEventWithCanonical(ctx context.Context, id string) (*pb.Event, error) { + query := ` + SELECT event_data, canonical_json + FROM events + WHERE id = ? AND deleted = 0 + ` + + var eventBytes, compressedJSON []byte + err := s.db.QueryRowContext(ctx, query, id).Scan(&eventBytes, &compressedJSON) + + if err == sql.ErrNoRows { + return nil, ErrEventNotFound + } + if err != nil { + return nil, fmt.Errorf("failed to query event: %w", err) + } + + // Deserialize protobuf + event := &pb.Event{} + if err := proto.Unmarshal(eventBytes, event); err != nil { + return nil, fmt.Errorf("failed to unmarshal event: %w", err) + } + + // Decompress canonical JSON + canonicalJSON, err := decompressJSON(compressedJSON) + if err != nil { + return nil, fmt.Errorf("failed to decompress canonical JSON: %w", err) + } + + event.CanonicalJson = canonicalJSON + + return event, nil +} + +// DeleteEvent marks an event as deleted (soft delete). +func (s *Storage) DeleteEvent(ctx context.Context, eventID string) error { + query := `UPDATE events SET deleted = 1 WHERE id = ?` + result, err := s.db.ExecContext(ctx, query, eventID) + if err != nil { + return fmt.Errorf("failed to delete event: %w", err) + } + + rows, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("failed to get rows affected: %w", err) + } + + if rows == 0 { + return ErrEventNotFound + } + + return nil +} + +// compressJSON compresses JSON bytes using zstd. +func compressJSON(data []byte) ([]byte, error) { + encoder, err := zstd.NewWriter(nil, zstd.WithEncoderLevel(zstd.SpeedDefault)) + if err != nil { + return nil, err + } + defer encoder.Close() + + return encoder.EncodeAll(data, nil), nil +} + +// decompressJSON decompresses zstd-compressed JSON bytes. +func decompressJSON(data []byte) ([]byte, error) { + decoder, err := zstd.NewReader(nil) + if err != nil { + return nil, err + } + defer decoder.Close() + + return decoder.DecodeAll(data, nil) +} + +// marshalTags converts protobuf tags to JSON string. +func marshalTags(tags []*pb.Tag) (string, error) { + if len(tags) == 0 { + return "[]", nil + } + + // Convert to [][]string for JSON serialization + tagArrays := make([][]string, len(tags)) + for i, tag := range tags { + tagArrays[i] = tag.Values + } + + data, err := json.Marshal(tagArrays) + if err != nil { + return "", err + } + + return string(data), nil +} + +// isDuplicateError checks if the error is a unique constraint violation. +func isDuplicateError(err error) bool { + if err == nil { + return false + } + // SQLite error messages for UNIQUE constraint violations + msg := err.Error() + return msg == "UNIQUE constraint failed: events.id" || + msg == "constraint failed: UNIQUE constraint failed: events.id" || + msg == "constraint failed: UNIQUE constraint failed: events.id (1555)" +} diff --git a/internal/storage/events_test.go b/internal/storage/events_test.go new file mode 100644 index 0000000..4393404 --- /dev/null +++ b/internal/storage/events_test.go @@ -0,0 +1,264 @@ +package storage + +import ( + "context" + "testing" + + pb "northwest.io/nostr-grpc/api/nostr/v1" +) + +func TestStoreEvent(t *testing.T) { + store, err := New(":memory:") + if err != nil { + t.Fatalf("failed to create storage: %v", err) + } + defer store.Close() + + ctx := context.Background() + + // Create test event + event := &pb.Event{ + Id: "test123", + Pubkey: "pubkey123", + CreatedAt: 1234567890, + Kind: 1, + Tags: []*pb.Tag{{Values: []string{"e", "event1"}}}, + Content: "Hello, Nostr!", + Sig: "sig123", + } + + canonicalJSON := []byte(`[0,"pubkey123",1234567890,1,[["e","event1"]],"Hello, Nostr!"]`) + + data := &EventData{ + Event: event, + CanonicalJSON: canonicalJSON, + } + + // Store event + err = store.StoreEvent(ctx, data) + if err != nil { + t.Fatalf("failed to store event: %v", err) + } + + // Verify event was stored + retrieved, err := store.GetEvent(ctx, "test123") + if err != nil { + t.Fatalf("failed to retrieve event: %v", err) + } + + if retrieved.Id != event.Id { + t.Errorf("expected ID %s, got %s", event.Id, retrieved.Id) + } + if retrieved.Content != event.Content { + t.Errorf("expected content %s, got %s", event.Content, retrieved.Content) + } +} + +func TestStoreEventDuplicate(t *testing.T) { + store, err := New(":memory:") + if err != nil { + t.Fatalf("failed to create storage: %v", err) + } + defer store.Close() + + ctx := context.Background() + + event := &pb.Event{ + Id: "duplicate123", + Pubkey: "pubkey123", + CreatedAt: 1234567890, + Kind: 1, + Content: "test", + Sig: "sig123", + } + + data := &EventData{ + Event: event, + CanonicalJSON: []byte(`[0,"pubkey123",1234567890,1,[],"test"]`), + } + + // Store first time + err = store.StoreEvent(ctx, data) + if err != nil { + t.Fatalf("failed to store event first time: %v", err) + } + + // Try to store again + err = store.StoreEvent(ctx, data) + if err != ErrEventExists { + t.Errorf("expected ErrEventExists, got %v", err) + } +} + +func TestGetEvent(t *testing.T) { + store, err := New(":memory:") + if err != nil { + t.Fatalf("failed to create storage: %v", err) + } + defer store.Close() + + ctx := context.Background() + + // Test non-existent event + _, err = store.GetEvent(ctx, "nonexistent") + if err != ErrEventNotFound { + t.Errorf("expected ErrEventNotFound, got %v", err) + } +} + +func TestGetEventWithCanonical(t *testing.T) { + store, err := New(":memory:") + if err != nil { + t.Fatalf("failed to create storage: %v", err) + } + defer store.Close() + + ctx := context.Background() + + canonicalJSON := []byte(`[0,"pubkey123",1234567890,1,[],"test"]`) + + event := &pb.Event{ + Id: "canonical123", + Pubkey: "pubkey123", + CreatedAt: 1234567890, + Kind: 1, + Content: "test", + Sig: "sig123", + } + + data := &EventData{ + Event: event, + CanonicalJSON: canonicalJSON, + } + + err = store.StoreEvent(ctx, data) + if err != nil { + t.Fatalf("failed to store event: %v", err) + } + + // Retrieve with canonical JSON + retrieved, err := store.GetEventWithCanonical(ctx, "canonical123") + if err != nil { + t.Fatalf("failed to retrieve event: %v", err) + } + + if string(retrieved.CanonicalJson) != string(canonicalJSON) { + t.Errorf("canonical JSON mismatch:\nexpected: %s\ngot: %s", + canonicalJSON, retrieved.CanonicalJson) + } +} + +func TestDeleteEvent(t *testing.T) { + store, err := New(":memory:") + if err != nil { + t.Fatalf("failed to create storage: %v", err) + } + defer store.Close() + + ctx := context.Background() + + event := &pb.Event{ + Id: "delete123", + Pubkey: "pubkey123", + CreatedAt: 1234567890, + Kind: 1, + Content: "to be deleted", + Sig: "sig123", + } + + data := &EventData{ + Event: event, + CanonicalJSON: []byte(`[0,"pubkey123",1234567890,1,[],"to be deleted"]`), + } + + // Store event + err = store.StoreEvent(ctx, data) + if err != nil { + t.Fatalf("failed to store event: %v", err) + } + + // Delete event + err = store.DeleteEvent(ctx, "delete123") + if err != nil { + t.Fatalf("failed to delete event: %v", err) + } + + // Verify event is no longer retrievable + _, err = store.GetEvent(ctx, "delete123") + if err != ErrEventNotFound { + t.Errorf("expected ErrEventNotFound after deletion, got %v", err) + } + + // Try deleting non-existent event + err = store.DeleteEvent(ctx, "nonexistent") + if err != ErrEventNotFound { + t.Errorf("expected ErrEventNotFound, got %v", err) + } +} + +func TestCompressDecompressJSON(t *testing.T) { + original := []byte(`{"key":"value","array":[1,2,3],"nested":{"a":"b"}}`) + + compressed, err := compressJSON(original) + if err != nil { + t.Fatalf("compression failed: %v", err) + } + + // Verify compression reduces size (for larger data) + if len(compressed) >= len(original) { + t.Logf("Note: compressed size (%d) >= original (%d) - normal for small data", + len(compressed), len(original)) + } + + decompressed, err := decompressJSON(compressed) + if err != nil { + t.Fatalf("decompression failed: %v", err) + } + + if string(decompressed) != string(original) { + t.Errorf("decompressed data doesn't match original:\nexpected: %s\ngot: %s", + original, decompressed) + } +} + +func TestMarshalTags(t *testing.T) { + tests := []struct { + name string + tags []*pb.Tag + expected string + }{ + { + name: "empty tags", + tags: nil, + expected: "[]", + }, + { + name: "single tag", + tags: []*pb.Tag{ + {Values: []string{"e", "event123"}}, + }, + expected: `[["e","event123"]]`, + }, + { + name: "multiple tags", + tags: []*pb.Tag{ + {Values: []string{"e", "event123", "wss://relay.example.com"}}, + {Values: []string{"p", "pubkey456"}}, + }, + expected: `[["e","event123","wss://relay.example.com"],["p","pubkey456"]]`, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result, err := marshalTags(tt.tags) + if err != nil { + t.Fatalf("marshalTags failed: %v", err) + } + + if result != tt.expected { + t.Errorf("expected %s, got %s", tt.expected, result) + } + }) + } +} diff --git a/internal/storage/storage.go b/internal/storage/storage.go new file mode 100644 index 0000000..64fc4c6 --- /dev/null +++ b/internal/storage/storage.go @@ -0,0 +1,150 @@ +package storage + +import ( + "context" + "database/sql" + "fmt" + + _ "modernc.org/sqlite" // Pure Go SQLite driver +) + +// Storage provides event persistence using SQLite. +// Consumers should define their own interface based on their needs. +type Storage struct { + db *sql.DB +} + +// New creates a new Storage instance and initializes the database schema. +// The dbPath should be a file path or ":memory:" for in-memory database. +func New(dbPath string) (*Storage, error) { + db, err := sql.Open("sqlite", dbPath) + if err != nil { + return nil, fmt.Errorf("failed to open database: %w", err) + } + + // Configure SQLite for optimal performance + pragmas := []string{ + "PRAGMA journal_mode=WAL", // Write-Ahead Logging for concurrency + "PRAGMA synchronous=NORMAL", // Balance safety and performance + "PRAGMA cache_size=10000", // 10000 pages (~40MB cache) + "PRAGMA temp_store=MEMORY", // Temp tables in memory + "PRAGMA mmap_size=268435456", // 256MB memory-mapped I/O + "PRAGMA page_size=4096", // Standard page size + "PRAGMA foreign_keys=ON", // Enforce foreign key constraints + "PRAGMA busy_timeout=5000", // Wait up to 5s for locks + "PRAGMA auto_vacuum=INCREMENTAL", // Reclaim space incrementally + } + + for _, pragma := range pragmas { + if _, err := db.Exec(pragma); err != nil { + db.Close() + return nil, fmt.Errorf("failed to set pragma %q: %w", pragma, err) + } + } + + s := &Storage{db: db} + + // Initialize schema + if err := s.initSchema(context.Background()); err != nil { + db.Close() + return nil, fmt.Errorf("failed to initialize schema: %w", err) + } + + return s, nil +} + +// Close closes the database connection. +func (s *Storage) Close() error { + return s.db.Close() +} + +// initSchema creates all necessary tables and indexes. +func (s *Storage) initSchema(ctx context.Context) error { + schema := ` + -- Main events table + CREATE TABLE IF NOT EXISTS events ( + -- Primary event data + id TEXT PRIMARY KEY, + event_data BLOB NOT NULL, -- Protobuf binary + canonical_json BLOB NOT NULL, -- zstd compressed canonical JSON + + -- Denormalized fields for efficient querying + pubkey TEXT NOT NULL, + kind INTEGER NOT NULL, + created_at INTEGER NOT NULL, -- Unix timestamp + content TEXT, -- For full-text search (optional) + tags TEXT, -- JSON text for tag queries (use json_* functions) + sig TEXT NOT NULL, + + -- Metadata + deleted INTEGER DEFAULT 0, -- STRICT mode: use INTEGER for boolean + received_at INTEGER DEFAULT (unixepoch()) + ) STRICT; + + -- Critical indexes for Nostr query patterns + CREATE INDEX IF NOT EXISTS idx_pubkey_created + ON events(pubkey, created_at DESC) + WHERE deleted = 0; + + CREATE INDEX IF NOT EXISTS idx_kind_created + ON events(kind, created_at DESC) + WHERE deleted = 0; + + CREATE INDEX IF NOT EXISTS idx_created + ON events(created_at DESC) + WHERE deleted = 0; + + -- For tag queries (#e, #p, etc) + CREATE INDEX IF NOT EXISTS idx_tags + ON events(tags) + WHERE deleted = 0; + + -- Deletion events (NIP-09) + CREATE TABLE IF NOT EXISTS deletions ( + event_id TEXT PRIMARY KEY, -- ID of deletion event + deleted_event_id TEXT NOT NULL, -- ID of event being deleted + pubkey TEXT NOT NULL, -- Who requested deletion + created_at INTEGER NOT NULL, + FOREIGN KEY (deleted_event_id) REFERENCES events(id) + ) STRICT; + + CREATE INDEX IF NOT EXISTS idx_deleted_event + ON deletions(deleted_event_id); + + -- Replaceable events tracking (NIP-16, NIP-33) + CREATE TABLE IF NOT EXISTS replaceable_events ( + kind INTEGER NOT NULL, + pubkey TEXT NOT NULL, + d_tag TEXT NOT NULL DEFAULT '', -- For parameterized replaceable events (empty string for non-parameterized) + current_event_id TEXT NOT NULL, + created_at INTEGER NOT NULL, + PRIMARY KEY (kind, pubkey, d_tag), + FOREIGN KEY (current_event_id) REFERENCES events(id) + ) STRICT; + + -- Auth challenges (NIP-42) + CREATE TABLE IF NOT EXISTS auth_challenges ( + challenge TEXT PRIMARY KEY, + created_at INTEGER NOT NULL, + expires_at INTEGER NOT NULL, + used INTEGER DEFAULT 0 -- STRICT mode: use INTEGER for boolean + ) STRICT; + + -- Rate limiting + CREATE TABLE IF NOT EXISTS rate_limits ( + pubkey TEXT PRIMARY KEY, + event_count INTEGER DEFAULT 0, + window_start INTEGER NOT NULL, + last_reset INTEGER DEFAULT (unixepoch()) + ) STRICT; + ` + + _, err := s.db.ExecContext(ctx, schema) + return err +} + +// DB returns the underlying *sql.DB for advanced usage. +// This allows consumers to execute custom queries if needed. +func (s *Storage) DB() *sql.DB { + return s.db +} diff --git a/internal/storage/storage_test.go b/internal/storage/storage_test.go new file mode 100644 index 0000000..f2fe401 --- /dev/null +++ b/internal/storage/storage_test.go @@ -0,0 +1,70 @@ +package storage + +import ( + "testing" +) + +func TestNew(t *testing.T) { + // Test in-memory database + store, err := New(":memory:") + if err != nil { + t.Fatalf("failed to create storage: %v", err) + } + defer store.Close() + + // Verify database is accessible + if store.DB() == nil { + t.Fatal("DB() returned nil") + } + + // Verify schema was created by checking if tables exist + var count int + query := `SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name IN ('events', 'deletions', 'replaceable_events')` + err = store.DB().QueryRow(query).Scan(&count) + if err != nil { + t.Fatalf("failed to query tables: %v", err) + } + + if count != 3 { + t.Errorf("expected 3 main tables, got %d", count) + } +} + +func TestNewFileDatabase(t *testing.T) { + // Test file-based database + dbPath := t.TempDir() + "/test.db" + store, err := New(dbPath) + if err != nil { + t.Fatalf("failed to create file-based storage: %v", err) + } + defer store.Close() + + // Verify WAL mode is enabled + var walMode string + err = store.DB().QueryRow("PRAGMA journal_mode").Scan(&walMode) + if err != nil { + t.Fatalf("failed to query journal mode: %v", err) + } + + if walMode != "wal" { + t.Errorf("expected WAL mode, got %s", walMode) + } +} + +func TestClose(t *testing.T) { + store, err := New(":memory:") + if err != nil { + t.Fatalf("failed to create storage: %v", err) + } + + err = store.Close() + if err != nil { + t.Errorf("failed to close storage: %v", err) + } + + // Verify database is closed by attempting a query + err = store.DB().Ping() + if err == nil { + t.Error("expected error when pinging closed database") + } +} -- cgit v1.2.3