1package storage23import (4 "context"5 "database/sql"6 "errors"7 "fmt"8 "net/url"9 "os"10 "path/filepath"11 "sync"12 "time"1314 _ "modernc.org/sqlite"15)1617// Rejection is a stable Save error that callers can match with errors.Is.18type Rejection string1920func (e Rejection) Error() string { return string(e) }2122const (23 ErrDuplicate Rejection = "duplicate: already have this event"24 ErrReplaced Rejection = "duplicate: a newer version exists"25 ErrDeleted Rejection = "blocked: event was previously deleted by its author"26 ErrVanished Rejection = "blocked: this pubkey asked for its events to be deleted"27)2829// Observer receives operation name, outcome and duration for instrumented30// Store operations.31type Observer interface {32 ObserveStorage(operation, outcome string, duration time.Duration)33}3435// Store is a tenant's SQLite database. A Store must be closed when its owner36// is done with it.37type Store struct {38 db *sql.DB39 path string40 observerMu sync.RWMutex41 observer Observer42}4344// Open creates or opens path, initializes the tenant schema and returns a45// Store. The parent directory is created with mode 0700 and the database file46// with mode 0600.47func Open(ctx context.Context, path string) (*Store, error) {48 if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil {49 return nil, fmt.Errorf("create database directory: %w", err)50 }51 f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0600)52 if err != nil {53 return nil, fmt.Errorf("create database: %w", err)54 }55 if err := f.Close(); err != nil {56 return nil, fmt.Errorf("close database file: %w", err)57 }58 u := url.URL{Scheme: "file", Path: path}59 q := u.Query()60 for _, pragma := range []string{"journal_mode(WAL)", "synchronous(FULL)", "foreign_keys(ON)", "busy_timeout(5000)"} {61 q.Add("_pragma", pragma)62 }63 u.RawQuery = q.Encode()64 db, err := sql.Open("sqlite", u.String())65 if err != nil {66 return nil, fmt.Errorf("open database: %w", err)67 }68 // One connection serializes tenant database operations. Transaction hooks69 // use their supplied transaction while it holds this connection.70 db.SetMaxOpenConns(1)71 db.SetMaxIdleConns(1)72 s := &Store{db: db, path: path}73 if _, err := db.ExecContext(ctx, schema); err != nil {74 return nil, errors.Join(fmt.Errorf("initialize tenant schema: %w", err), db.Close())75 }76 return s, nil77}7879// DB returns the database used by tenant services for their schemas, queries80// and transactions. Event persistence uses Save or SaveTx to apply storage's81// replacement, deletion and indexing rules.82func (s *Store) DB() *sql.DB { return s.db }8384// Path returns the database file path supplied to Open.85func (s *Store) Path() string { return s.path }8687// Close releases the database connection.88func (s *Store) Close() error { return s.db.Close() }8990// SetObserver replaces the observer used for operation timing and outcomes.91func (s *Store) SetObserver(o Observer) {92 s.observerMu.Lock()93 defer s.observerMu.Unlock()94 s.observer = o95}9697func (s *Store) observe(operation string, started time.Time, err error) {98 s.observerMu.RLock()99 defer s.observerMu.RUnlock()100 if s.observer == nil {101 return102 }103 outcome := "ok"104 if err != nil {105 outcome = "error"106 }107 s.observer.ObserveStorage(operation, outcome, time.Since(started))108}109110// WithTx opens a transaction, calls fn and commits after fn succeeds. It rolls111// back on failure. The callback uses the supplied transaction for database112// work because it holds the store's sole connection until completion.113func (s *Store) WithTx(ctx context.Context, fn func(*sql.Tx) error) (err error) {114 started := time.Now()115 defer func() { s.observe("commit", started, err) }()116 tx, err := s.db.BeginTx(ctx, nil)117 if err != nil {118 return fmt.Errorf("begin tenant transaction: %w", err)119 }120 defer func() {121 rollbackErr := tx.Rollback()122 if rollbackErr != nil && !errors.Is(rollbackErr, sql.ErrTxDone) {123 err = errors.Join(err, rollbackErr)124 }125 }()126 if err := fn(tx); err != nil {127 return err128 }129 if err := tx.Commit(); err != nil {130 return fmt.Errorf("commit tenant transaction: %w", err)131 }132 return nil133}134135func exists(ctx context.Context, tx *sql.Tx, query string, args ...any) (bool, error) {136 var value int137 err := tx.QueryRowContext(ctx, query, args...).Scan(&value)138 if errors.Is(err, sql.ErrNoRows) {139 return false, nil140 }141 return err == nil, err142}