| 1 | package storage |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "encoding/json" |
| 7 | "fmt" |
| 8 | "os" |
| 9 | |
| 10 | "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 11 | ) |
| 12 | |
| 13 | // Stats reports event counts, database size and the current WAL size. |
| 14 | type Stats struct { |
| 15 | Events int64 `json:"events"` |
| 16 | Bytes int64 `json:"bytes"` |
| 17 | Oldest int64 `json:"oldest"` |
| 18 | Newest int64 `json:"newest"` |
| 19 | WALBytes int64 `json:"wal_bytes"` |
| 20 | } |
| 21 | |
| 22 | func (s *Store) Stats(ctx context.Context) (Stats, error) { |
| 23 | var result Stats |
| 24 | err := s.db.QueryRowContext(ctx, "SELECT count(*),coalesce(min(created_at),0),coalesce(max(created_at),0) FROM events WHERE kind<>?" , event.KIND_PUSH_REGISTRATION).Scan(&result.Events, &result.Oldest, &result.Newest) |
| 25 | if err != nil { |
| 26 | return result, fmt.Errorf("read storage stats: %w" , err) |
| 27 | } |
| 28 | var pages, size int64 |
| 29 | if err := s.db.QueryRowContext(ctx, "PRAGMA page_count" ).Scan(&pages); err != nil { |
| 30 | return result, err |
| 31 | } |
| 32 | if err := s.db.QueryRowContext(ctx, "PRAGMA page_size" ).Scan(&size); err != nil { |
| 33 | return result, err |
| 34 | } |
| 35 | result.Bytes = pages * size |
| 36 | if info, err := os.Stat(s.path + "-wal" ); err == nil { |
| 37 | result.WALBytes = info.Size() |
| 38 | } else if !os.IsNotExist(err) { |
| 39 | return result, err |
| 40 | } |
| 41 | return result, nil |
| 42 | } |
| 43 | |
| 44 | // Vanish records a deletion horizon and removes the author's events and |
| 45 | // related history through until, inclusive. |
| 46 | func (s *Store) Vanish(ctx context.Context, pubkey string, until int64) error { |
| 47 | return s.WithTx(ctx, func(tx *sql.Tx) error { |
| 48 | statements := []struct { |
| 49 | sql string |
| 50 | args []any |
| 51 | }{ |
| 52 | {"INSERT INTO vanished(pubkey,until) VALUES(?,?) ON CONFLICT(pubkey) DO UPDATE SET until=max(until,excluded.until)" , []any{pubkey, until}}, |
| 53 | {"DELETE FROM events WHERE pubkey=? AND created_at<=?" , []any{pubkey, until}}, |
| 54 | {"DELETE FROM events WHERE kind=1059 AND id IN(SELECT event_id FROM tags WHERE name='p' AND value=?)" , []any{pubkey}}, |
| 55 | {"DELETE FROM list_history WHERE owner=?" , []any{pubkey}}, |
| 56 | {"DELETE FROM wiki_revisions WHERE author=?" , []any{pubkey}}, |
| 57 | {"DELETE FROM agent_grant_revisions WHERE author=?" , []any{pubkey}}, |
| 58 | } |
| 59 | for _, q := range statements { |
| 60 | if _, err := tx.ExecContext(ctx, q.sql, q.args...); err != nil { |
| 61 | return fmt.Errorf("vanish events: %w" , err) |
| 62 | } |
| 63 | } |
| 64 | return nil |
| 65 | }) |
| 66 | } |
| 67 | |
| 68 | // SweepExpired removes expired events and history and returns the next event |
| 69 | // expiration timestamp, or zero when none remain. |
| 70 | func (s *Store) SweepExpired(ctx context.Context, now int64) (int64, error) { |
| 71 | var next int64 |
| 72 | err := s.WithTx(ctx, func(tx *sql.Tx) error { |
| 73 | for _, table := range []string{"events" , "list_history" } { |
| 74 | if _, err := tx.ExecContext(ctx, "DELETE FROM " +table+" WHERE expires>0 AND expires<=?" , now); err != nil { |
| 75 | return fmt.Errorf("sweep expiration: %w" , err) |
| 76 | } |
| 77 | } |
| 78 | return tx.QueryRowContext(ctx, "SELECT coalesce(min(expires),0) FROM events WHERE expires>0" ).Scan(&next) |
| 79 | }) |
| 80 | return next, err |
| 81 | } |
| 82 | |
| 83 | // DeleteEvent removes one event by ID and reports whether a row was removed. |
| 84 | func (s *Store) DeleteEvent(ctx context.Context, id string) (bool, error) { |
| 85 | result, err := s.db.ExecContext(ctx, "DELETE FROM events WHERE id=?" , id) |
| 86 | if err != nil { |
| 87 | return false, fmt.Errorf("delete stored event: %w" , err) |
| 88 | } |
| 89 | n, err := result.RowsAffected() |
| 90 | return n > 0, err |
| 91 | } |
| 92 | |
| 93 | // EraseAuthor removes an author's events and retained list and wiki history. |
| 94 | func (s *Store) EraseAuthor(ctx context.Context, pubkey string) (int64, error) { |
| 95 | var n int64 |
| 96 | err := s.WithTx(ctx, func(tx *sql.Tx) error { |
| 97 | result, err := tx.ExecContext(ctx, "DELETE FROM events WHERE pubkey=?" , pubkey) |
| 98 | if err != nil { |
| 99 | return err |
| 100 | } |
| 101 | n, err = result.RowsAffected() |
| 102 | if err != nil { |
| 103 | return err |
| 104 | } |
| 105 | if _, err = tx.ExecContext(ctx, "DELETE FROM list_history WHERE owner=?" , pubkey); err != nil { |
| 106 | return err |
| 107 | } |
| 108 | if _, err = tx.ExecContext(ctx, "DELETE FROM wiki_revisions WHERE author=?" , pubkey); err != nil { |
| 109 | return err |
| 110 | } |
| 111 | _, err = tx.ExecContext(ctx, "DELETE FROM agent_grant_revisions WHERE author=?" , pubkey) |
| 112 | return err |
| 113 | }) |
| 114 | return n, err |
| 115 | } |
| 116 | |
| 117 | // GetSetting decodes a JSON setting into dest. |
| 118 | func (s *Store) GetSetting(ctx context.Context, key string, dest any) error { |
| 119 | var raw string |
| 120 | if err := s.db.QueryRowContext(ctx, "SELECT value FROM settings WHERE key=?" , key).Scan(&raw); err != nil { |
| 121 | return err |
| 122 | } |
| 123 | if err := json.Unmarshal([]byte(raw), dest); err != nil { |
| 124 | return fmt.Errorf("decode setting %s: %w" , key, err) |
| 125 | } |
| 126 | return nil |
| 127 | } |
| 128 | |
| 129 | // PutSetting encodes value as JSON and upserts it under key. |
| 130 | func (s *Store) PutSetting(ctx context.Context, key string, value any) error { |
| 131 | return s.WithTx(ctx, func(tx *sql.Tx) error { return PutSetting(ctx, tx, key, value) }) |
| 132 | } |
| 133 | |
| 134 | // DeleteSetting removes one settings row. A missing row is not an error. |
| 135 | func (s *Store) DeleteSetting(ctx context.Context, key string) error { |
| 136 | _, err := s.db.ExecContext(ctx, "DELETE FROM settings WHERE key=?" , key) |
| 137 | return err |
| 138 | } |
| 139 | |
| 140 | // PruneSettings removes rows under prefix whose keys are not in keep, so |
| 141 | // per-filter state cannot accumulate as filters come and go. |
| 142 | func (s *Store) PruneSettings(ctx context.Context, prefix string, keep map[string]struct{}) error { |
| 143 | rows, err := s.db.QueryContext(ctx, "SELECT key FROM settings WHERE substr(key,1,?)=?" , len(prefix), prefix) |
| 144 | if err != nil { |
| 145 | return err |
| 146 | } |
| 147 | var stale []string |
| 148 | for rows.Next() { |
| 149 | var key string |
| 150 | if err := rows.Scan(&key); err != nil { |
| 151 | rows.Close() |
| 152 | return err |
| 153 | } |
| 154 | if _, ok := keep[key]; !ok { |
| 155 | stale = append(stale, key) |
| 156 | } |
| 157 | } |
| 158 | rows.Close() |
| 159 | if err := rows.Err(); err != nil { |
| 160 | return err |
| 161 | } |
| 162 | for _, key := range stale { |
| 163 | if err := s.DeleteSetting(ctx, key); err != nil { |
| 164 | return err |
| 165 | } |
| 166 | } |
| 167 | return nil |
| 168 | } |
| 169 | |
| 170 | // PutSetting writes one setting using a caller-owned transaction. |
| 171 | func PutSetting(ctx context.Context, tx *sql.Tx, key string, value any) error { |
| 172 | raw, err := json.Marshal(value) |
| 173 | if err != nil { |
| 174 | return fmt.Errorf("encode setting %s: %w" , key, err) |
| 175 | } |
| 176 | _, err = tx.ExecContext(ctx, "INSERT INTO settings(key,value) VALUES(?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value" , key, string(raw)) |
| 177 | if err != nil { |
| 178 | return fmt.Errorf("save setting %s: %w" , key, err) |
| 179 | } |
| 180 | return nil |
| 181 | } |