tinyrelayfile
tinyrelay/internal/storage/save.go/
internal/storage/save.go raw
1package storage
2
3import (
4 "context"
5 "crypto/sha256"
6 "database/sql"
7 "encoding/hex"
8 "encoding/json"
9 "fmt"
10 "strconv"
11 "strings"
12
13 "github.com/FelineStateMachine/tinyrelay/internal/event"
14)
15
16// Intent describes durable work to insert with an event save. The tuple of
17// Kind, EventID and Target determines its stable ID; Payload is retained for
18// the worker handling that kind.
19type Intent struct {
20 Kind string `json:"kind"`
21 EventID string `json:"event_id"`
22 Target string `json:"target"`
23 Payload string `json:"payload"`
24}
25
26// SaveOptions controls event persistence. Now is the Unix time used for
27// history and intent timestamps. SearchMode is empty for the default indexed
28// kinds, "full" for all public content and "off" to skip content indexing.
29// For stored events, Intents and BeforeCommit run in the event transaction.
30// Ephemeral events return before those hooks. A BeforeCommit error rolls back
31// the transaction owned by Save; SaveTx returns it to the transaction's owner.
32type SaveOptions struct {
33 Now int64
34 SearchMode string
35 Intents []Intent
36 BeforeCommit func(context.Context, *sql.Tx) error
37}
38
39// SaveResult reports the event sequence. Ephemeral events return without a
40// database row and set Ephemeral.
41type SaveResult struct {
42 Sequence int64
43 Ephemeral bool
44}
45
46// Save commits an event, its indexes and the supplied transaction hooks
47// together. Callers validate the event before saving. Duplicate, replaced,
48// deleted and vanished events return a Rejection. Ephemeral events return an
49// Ephemeral result without writing rows or running the hooks.
50func (s *Store) Save(ctx context.Context, e event.Event, opts SaveOptions) (SaveResult, error) {
51 var result SaveResult
52 err := s.WithTx(ctx, func(tx *sql.Tx) error {
53 var err error
54 result, err = SaveTx(ctx, tx, e, opts)
55 return err
56 })
57 return result, err
58}
59
60// SaveTx composes event persistence with membership or other tenant changes.
61// The caller owns commit/rollback; no success may be acknowledged before commit.
62func SaveTx(ctx context.Context, tx *sql.Tx, e event.Event, opts SaveOptions) (SaveResult, error) {
63 if event.IsEphemeral(e.Kind) {
64 return SaveResult{Ephemeral: true}, nil
65 }
66 var result SaveResult
67 err := func() error {
68 if err := checkSave(ctx, tx, e); err != nil {
69 return err
70 }
71 if err := replaceOrDelete(ctx, tx, e, opts.Now); err != nil {
72 return err
73 }
74 seq, err := insertEvent(ctx, tx, e, opts.SearchMode)
75 if err != nil {
76 return err
77 }
78 result.Sequence = seq
79 if err := AddIntents(ctx, tx, opts.Intents, opts.Now); err != nil {
80 return err
81 }
82 if opts.BeforeCommit != nil {
83 return opts.BeforeCommit(ctx, tx)
84 }
85 return nil
86 }()
87 return result, err
88}
89
90func checkSave(ctx context.Context, tx *sql.Tx, e event.Event) error {
91 checks := []struct {
92 query string
93 args []any
94 rejection Rejection
95 }{
96 {"SELECT 1 FROM events WHERE id=?", []any{e.ID}, ErrDuplicate},
97 {"SELECT 1 FROM vanished WHERE pubkey=? AND until>=?", []any{e.PubKey, e.CreatedAt}, ErrVanished},
98 }
99 for _, check := range checks {
100 has, err := exists(ctx, tx, check.query, check.args...)
101 if err != nil {
102 return fmt.Errorf("check event persistence: %w", err)
103 }
104 if has {
105 return check.rejection
106 }
107 }
108 return checkDeleted(ctx, tx, e)
109}
110
111func checkDeleted(ctx context.Context, tx *sql.Tx, e event.Event) error {
112 deleters := []string{e.PubKey}
113 if e.Kind == 1059 {
114 deleters = append(deleters, event.TagValues(e, "p")...)
115 }
116 list, err := json.Marshal(deleters)
117 if err != nil {
118 return fmt.Errorf("encode deletion principals: %w", err)
119 }
120 has, err := exists(ctx, tx, "SELECT 1 FROM deletions WHERE target_type='e' AND target=? AND author IN (SELECT value FROM json_each(?)) LIMIT 1", e.ID, string(list))
121 if err != nil {
122 return fmt.Errorf("check deletion tombstone: %w", err)
123 }
124 if has {
125 return ErrDeleted
126 }
127 if !event.IsAddressable(e.Kind) {
128 return nil
129 }
130 address := strconv.Itoa(e.Kind) + ":" + e.PubKey + ":" + event.Tag(e, "d")
131 has, err = exists(ctx, tx, "SELECT 1 FROM deletions WHERE author=? AND target_type='a' AND target=? AND until>=?", e.PubKey, address, e.CreatedAt)
132 if err != nil {
133 return fmt.Errorf("check address tombstone: %w", err)
134 }
135 if has {
136 return ErrDeleted
137 }
138 return nil
139}
140
141func replaceOrDelete(ctx context.Context, tx *sql.Tx, e event.Event, now int64) error {
142 if event.IsReplaceable(e.Kind) || event.IsAddressable(e.Kind) {
143 return replace(ctx, tx, e, now)
144 }
145 if e.Kind != 5 {
146 return nil
147 }
148 for _, tag := range e.Tags {
149 if len(tag) < 2 || (tag[0] != "e" && tag[0] != "a") {
150 continue
151 }
152 if err := applyDeletion(ctx, tx, e, tag); err != nil {
153 return err
154 }
155 }
156 return nil
157}
158
159func replace(ctx context.Context, tx *sql.Tx, e event.Event, now int64) error {
160 d := ""
161 if event.IsAddressable(e.Kind) {
162 d = event.Tag(e, "d")
163 }
164 has, err := exists(ctx, tx, "SELECT 1 FROM events WHERE pubkey=? AND kind=? AND d=? AND (created_at>? OR (created_at=? AND id<?)) LIMIT 1", e.PubKey, e.Kind, d, e.CreatedAt, e.CreatedAt, e.ID)
165 if err != nil {
166 return fmt.Errorf("check replaceable event: %w", err)
167 }
168 if has {
169 return ErrReplaced
170 }
171 if isListKind(e.Kind) {
172 _, err := tx.ExecContext(ctx, "INSERT OR IGNORE INTO list_history(owner,kind,d,event_id,created_at,saved_at,expires,raw) SELECT pubkey,kind,d,id,created_at,?,expires,raw FROM events WHERE pubkey=? AND kind=? AND d=?", now, e.PubKey, e.Kind, d)
173 if err != nil {
174 return fmt.Errorf("archive replaced list: %w", err)
175 }
176 }
177 if e.Kind == kindWikiArticle {
178 // A wiki page keeps its history: the version being replaced moves
179 // to wiki_revisions with the id of the event that replaced it.
180 _, err := tx.ExecContext(ctx, "INSERT OR IGNORE INTO wiki_revisions(event_id,author,d,created_at,raw,superseded_by) SELECT id,pubkey,d,created_at,raw,? FROM events WHERE pubkey=? AND kind=? AND d=?", e.ID, e.PubKey, e.Kind, d)
181 if err != nil {
182 return fmt.Errorf("archive replaced wiki revision: %w", err)
183 }
184 }
185 if e.Kind == event.KIND_AGENT_GRANT {
186 if _, err := tx.ExecContext(ctx, "INSERT OR IGNORE INTO agent_grant_revisions(event_id,author,d,created_at,raw,superseded_by) SELECT id,pubkey,d,created_at,raw,? FROM events WHERE pubkey=? AND kind=? AND d=?", e.ID, e.PubKey, e.Kind, d); err != nil {
187 return fmt.Errorf("archive replaced agent grant: %w", err)
188 }
189 }
190 _, err = tx.ExecContext(ctx, "DELETE FROM events WHERE pubkey=? AND kind=? AND d=?", e.PubKey, e.Kind, d)
191 if err != nil {
192 return fmt.Errorf("replace event: %w", err)
193 }
194 return nil
195}
196
197func applyDeletion(ctx context.Context, tx *sql.Tx, e event.Event, tag []string) error {
198 if tag[0] == "a" {
199 parts := strings.SplitN(tag[1], ":", 3)
200 if len(parts) != 3 || parts[1] != e.PubKey {
201 return nil
202 }
203 kind, err := strconv.Atoi(parts[0])
204 if err != nil {
205 return nil
206 }
207 if kind == event.KIND_AGENT_GRANT {
208 if err := archiveAgentGrantDeletion(ctx, tx, e, parts[1], parts[2]); err != nil {
209 return err
210 }
211 }
212 if _, err := tx.ExecContext(ctx, "DELETE FROM events WHERE kind=? AND pubkey=? AND d=? AND created_at<=?", kind, e.PubKey, parts[2], e.CreatedAt); err != nil {
213 return fmt.Errorf("delete address: %w", err)
214 }
215 if _, err := tx.ExecContext(ctx, "DELETE FROM list_history WHERE kind=? AND owner=? AND d=? AND created_at<=?", kind, e.PubKey, parts[2], e.CreatedAt); err != nil {
216 return fmt.Errorf("delete list history: %w", err)
217 }
218 if kind == kindWikiArticle {
219 if _, err := tx.ExecContext(ctx, "DELETE FROM wiki_revisions WHERE author=? AND d=? AND created_at<=?", e.PubKey, parts[2], e.CreatedAt); err != nil {
220 return fmt.Errorf("delete wiki revisions: %w", err)
221 }
222 }
223 } else {
224 if err := archiveAgentGrantEventDeletion(ctx, tx, e, tag[1]); err != nil {
225 return err
226 }
227 if _, err := tx.ExecContext(ctx, "DELETE FROM events WHERE id=? AND (pubkey=? OR (kind=1059 AND EXISTS(SELECT 1 FROM tags WHERE event_id=events.id AND name='p' AND value=?)))", tag[1], e.PubKey, e.PubKey); err != nil {
228 return fmt.Errorf("delete event: %w", err)
229 }
230 if _, err := tx.ExecContext(ctx, "DELETE FROM list_history WHERE owner=? AND event_id=?", e.PubKey, tag[1]); err != nil {
231 return fmt.Errorf("delete list version: %w", err)
232 }
233 if _, err := tx.ExecContext(ctx, "DELETE FROM wiki_revisions WHERE author=? AND event_id=?", e.PubKey, tag[1]); err != nil {
234 return fmt.Errorf("delete wiki revision: %w", err)
235 }
236 }
237 _, err := tx.ExecContext(ctx, "INSERT INTO deletions(author,target_type,target,until) VALUES(?,?,?,?) ON CONFLICT(author,target_type,target) DO UPDATE SET until=max(until,excluded.until)", e.PubKey, tag[0], tag[1], e.CreatedAt)
238 if err != nil {
239 return fmt.Errorf("persist deletion tombstone: %w", err)
240 }
241 return nil
242}
243
244func archiveAgentGrantDeletion(ctx context.Context, tx *sql.Tx, deletion event.Event, owner, d string) error {
245 _, err := tx.ExecContext(ctx, "INSERT OR IGNORE INTO agent_grant_revisions(event_id,author,d,created_at,raw,superseded_by) SELECT id,pubkey,d,created_at,raw,? FROM events WHERE pubkey=? AND kind=? AND d=? AND created_at<=?", deletion.ID, owner, event.KIND_AGENT_GRANT, d, deletion.CreatedAt)
246 if err != nil {
247 return fmt.Errorf("archive deleted agent grant: %w", err)
248 }
249 return nil
250}
251
252func archiveAgentGrantEventDeletion(ctx context.Context, tx *sql.Tx, deletion event.Event, target string) error {
253 _, err := tx.ExecContext(ctx, "INSERT OR IGNORE INTO agent_grant_revisions(event_id,author,d,created_at,raw,superseded_by) SELECT id,pubkey,d,created_at,raw,? FROM events WHERE id=? AND kind=? AND pubkey=?", deletion.ID, target, event.KIND_AGENT_GRANT, deletion.PubKey)
254 if err != nil {
255 return fmt.Errorf("archive deleted agent grant: %w", err)
256 }
257 return nil
258}
259
260func insertEvent(ctx context.Context, tx *sql.Tx, e event.Event, mode string) (int64, error) {
261 raw, err := event.Canonical(e)
262 if err != nil {
263 return 0, fmt.Errorf("encode event: %w", err)
264 }
265 d := ""
266 if event.IsAddressable(e.Kind) {
267 d = event.Tag(e, "d")
268 }
269 var seq int64
270 err = tx.QueryRowContext(ctx, "INSERT INTO events(id,pubkey,created_at,kind,d,expires,raw) VALUES(?,?,?,?,?,?,?) RETURNING seq", e.ID, e.PubKey, e.CreatedAt, e.Kind, d, event.Expiration(e), string(raw)).Scan(&seq)
271 if err != nil {
272 return 0, fmt.Errorf("insert event: %w", err)
273 }
274 for _, tag := range e.Tags {
275 if len(tag) < 2 || len(tag[0]) != 1 {
276 continue
277 }
278 if _, err := tx.ExecContext(ctx, "INSERT INTO tags(event_id,name,value) VALUES(?,?,?)", e.ID, tag[0], tag[1]); err != nil {
279 return 0, fmt.Errorf("index event tag: %w", err)
280 }
281 }
282 if e.Content != "" && indexed(e.Kind, mode) {
283 if _, err := tx.ExecContext(ctx, "INSERT INTO search(rowid,content) VALUES(?,?)", seq, e.Content); err != nil {
284 return 0, fmt.Errorf("index event content: %w", err)
285 }
286 }
287 return seq, nil
288}
289
290func indexed(kind int, mode string) bool {
291 if event.IsPrivate(kind) || mode == "off" {
292 return false
293 }
294 if mode == "full" {
295 return true
296 }
297 switch kind {
298 case 0, 1, 11, 1111, 9802, 30023, 30024, 30818:
299 return true
300 }
301 return false
302}
303
304func isListKind(kind int) bool { return kind == 3 || kind == 10002 || kind == 10003 || kind == 30003 }
305
306// kindWikiArticle is the NIP-54 article kind, whose replaced versions are
307// kept as revisions.
308const kindWikiArticle = 30818
309
310func AddIntents(ctx context.Context, tx *sql.Tx, intents []Intent, now int64) error {
311 for _, intent := range intents {
312 _, err := tx.ExecContext(ctx, "INSERT OR IGNORE INTO work_intents(id,kind,event_id,target,payload,next_at,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?)", IntentID(intent.Kind, intent.EventID, intent.Target), intent.Kind, intent.EventID, intent.Target, intent.Payload, now, now, now)
313 if err != nil {
314 return fmt.Errorf("enqueue %s intent: %w", intent.Kind, err)
315 }
316 }
317 return nil
318}
319
320// IntentID returns the stable identifier shared by transactional and queued
321// durable work. Payload changes do not create a second intent.
322func IntentID(kind, eventID, target string) string {
323 key := sha256.Sum256([]byte(kind + "\x00" + eventID + "\x00" + target))
324 return hex.EncodeToString(key[:])
325}

About

ownerc01503e8035f
visibilitypublic
headd730679639bf

Maintainers

  • c01503e8035f owner
  • ba1b5beed5b5 agent hermes

Clone

  • https://012.run/npub1cq2s86qrtadf36eqq3epm38wmhdynrrcwxf84eldk3f94dkv7ttsrulrz5/tinyrelay.git

Refs