tinyrelayfile
tinyrelay/internal/daemon/replication_services.go/
internal/daemon/replication_services.go raw
1package daemon
2
3// This file contains the concrete adapters that keep replication tenant
4// scoped. The replication package stays independent from daemon ownership,
5// filesystem layout, and community implementation details.
6
7import (
8 "bytes"
9 "context"
10 "crypto/sha256"
11 "database/sql"
12 "encoding/base64"
13 "encoding/hex"
14 "encoding/json"
15 "errors"
16 "fmt"
17 "os"
18 "path/filepath"
19 "strconv"
20 "strings"
21 "time"
22
23 "github.com/FelineStateMachine/tinyrelay/internal/configport"
24 "github.com/FelineStateMachine/tinyrelay/internal/event"
25 gitrelay "github.com/FelineStateMachine/tinyrelay/tinygit"
26 "github.com/FelineStateMachine/tinyrelay/internal/policy"
27 "github.com/FelineStateMachine/tinyrelay/internal/records"
28 "github.com/FelineStateMachine/tinyrelay/internal/relay"
29 "github.com/FelineStateMachine/tinyrelay/internal/replication"
30 "github.com/FelineStateMachine/tinyrelay/internal/storage"
31 "github.com/FelineStateMachine/tinyrelay/internal/work"
32)
33
34type tenantReplicationProvider struct{ tenant *Tenant }
35
36func (t *Tenant) ReplicationBackupProvider() replication.BackupProvider {
37 return tenantReplicationProvider{tenant: t}
38}
39
40func (p tenantReplicationProvider) Snapshot(ctx context.Context) (replication.BackupState, error) {
41 if p.tenant == nil {
42 return replication.BackupState{}, errors.New("daemon: nil tenant")
43 }
44 ctx, done, admissionErr := p.tenant.beginMaintenance(ctx)
45 if admissionErr != nil {
46 return replication.BackupState{}, admissionErr
47 }
48 defer done()
49 state := replication.BackupState{}
50 policy, err := json.Marshal(p.tenant.Policy())
51 if err != nil {
52 return state, err
53 }
54 state.Config = policy
55 pubkey := ""
56 if p.tenant.records != nil {
57 pubkey = p.tenant.records.PublicKey()
58 }
59 identity, err := json.Marshal(map[string]string{"pubkey": pubkey, "relay": p.tenant.RelayURL()})
60 if err != nil {
61 return state, err
62 }
63 state.Identity = identity
64 state.Database, err = snapshotTables(ctx, p.tenant.store)
65 if err != nil {
66 return state, err
67 }
68 state.Blobs, err = snapshotFiles(filepath.Join(p.tenant.meta.Paths.Root, "blobs"))
69 if err != nil {
70 return state, err
71 }
72 state.Git, err = snapshotFiles(p.tenant.meta.Paths.Git)
73 if err != nil {
74 return state, err
75 }
76 return state, nil
77}
78
79func (p tenantReplicationProvider) SnapshotArchive(ctx context.Context, now int64) (replication.BackupArchive, error) {
80 state, err := p.Snapshot(ctx)
81 if err != nil {
82 return replication.BackupArchive{}, err
83 }
84 archive := replication.BackupArchive{Format: replication.BackupFormat, Created: now, State: state}
85 for _, table := range state.Database {
86 if table.Name != "events" {
87 continue
88 }
89 for _, row := range table.Rows {
90 for i, column := range table.Columns {
91 if column != "raw" || i >= len(row) {
92 continue
93 }
94 var raw string
95 if err := json.Unmarshal(row[i], &raw); err != nil {
96 return replication.BackupArchive{}, err
97 }
98 var item event.Event
99 if err := json.Unmarshal([]byte(raw), &item); err != nil {
100 return replication.BackupArchive{}, err
101 }
102 archive.Events = append(archive.Events, item)
103 }
104 }
105 }
106 return replication.SealBackup(archive)
107}
108
109func (p tenantReplicationProvider) Restore(ctx context.Context, state replication.BackupState) error {
110 if p.tenant == nil {
111 return errors.New("daemon: nil tenant")
112 }
113 ctx, done, admissionErr := p.tenant.beginMaintenance(ctx)
114 if admissionErr != nil {
115 return admissionErr
116 }
117 defer done()
118 oldBlobs, err := snapshotFiles(filepath.Join(p.tenant.meta.Paths.Root, "blobs"))
119 if err != nil {
120 return err
121 }
122 oldGit, err := snapshotFiles(p.tenant.meta.Paths.Git)
123 if err != nil {
124 return err
125 }
126 oldTables, err := snapshotTables(ctx, p.tenant.store)
127 if err != nil {
128 return err
129 }
130 restored := false
131 defer func() {
132 if restored {
133 return
134 }
135 // Restore is a single maintenance operation. Any failure after the
136 // first write must leave both SQLite and tenant files as they were.
137 _ = restoreTables(context.Background(), p.tenant.store, oldTables)
138 _ = restoreFiles(filepath.Join(p.tenant.meta.Paths.Root, "blobs"), oldBlobs)
139 _ = restoreFiles(p.tenant.meta.Paths.Git, oldGit)
140 if p.tenant.git != nil {
141 _ = p.tenant.git.Reload(context.WithoutCancel(ctx))
142 }
143 }()
144 if err := restoreTables(ctx, p.tenant.store, state.Database); err != nil {
145 return err
146 }
147 if err := restoreFiles(filepath.Join(p.tenant.meta.Paths.Root, "blobs"), state.Blobs); err != nil {
148 _ = restoreTables(ctx, p.tenant.store, oldTables)
149 _ = restoreFiles(filepath.Join(p.tenant.meta.Paths.Root, "blobs"), oldBlobs)
150 return err
151 }
152 if err := restoreFiles(p.tenant.meta.Paths.Git, state.Git); err != nil {
153 _ = restoreTables(ctx, p.tenant.store, oldTables)
154 _ = restoreFiles(filepath.Join(p.tenant.meta.Paths.Root, "blobs"), oldBlobs)
155 _ = restoreFiles(p.tenant.meta.Paths.Git, oldGit)
156 return err
157 }
158 if len(state.GitSQL) > 0 && string(state.GitSQL) != "null" {
159 if p.tenant.git == nil {
160 _ = restoreTables(ctx, p.tenant.store, oldTables)
161 _ = restoreFiles(filepath.Join(p.tenant.meta.Paths.Root, "blobs"), oldBlobs)
162 _ = restoreFiles(filepath.Join(p.tenant.meta.Paths.Git), oldGit)
163 return errors.New("daemon: Git backup migration unavailable")
164 }
165 if _, err := gitrelay.ValidateGitBackup(state.GitSQL); err != nil {
166 _ = restoreTables(ctx, p.tenant.store, oldTables)
167 _ = restoreFiles(filepath.Join(p.tenant.meta.Paths.Root, "blobs"), oldBlobs)
168 _ = restoreFiles(filepath.Join(p.tenant.meta.Paths.Git), oldGit)
169 return fmt.Errorf("daemon: validate legacy Git backup: %w", err)
170 }
171 if err := p.tenant.git.ImportGitBackup(ctx, state.GitSQL); err != nil {
172 _ = restoreTables(ctx, p.tenant.store, oldTables)
173 _ = restoreFiles(filepath.Join(p.tenant.meta.Paths.Root, "blobs"), oldBlobs)
174 _ = restoreFiles(filepath.Join(p.tenant.meta.Paths.Git), oldGit)
175 return fmt.Errorf("daemon: import legacy Git backup: %w", err)
176 }
177 }
178 if len(state.Config) > 0 {
179 var envelope struct {
180 Format string `json:"format"`
181 }
182 if json.Unmarshal(state.Config, &envelope) == nil && strings.HasPrefix(envelope.Format, "bind.ws/relay-config/") {
183 legacyConfig := true
184 if p.tenant.config == nil {
185 return errors.New("daemon: config importer unavailable for legacy backup")
186 }
187 parsed, err := configport.Parse(state.Config)
188 if err != nil {
189 return fmt.Errorf("daemon: legacy config: %w", err)
190 }
191 var identity struct {
192 Owner string `json:"owner"`
193 }
194 if len(state.Identity) > 0 {
195 if err := json.Unmarshal(state.Identity, &identity); err != nil {
196 return fmt.Errorf("daemon: legacy identity: %w", err)
197 }
198 }
199 if _, err := p.tenant.config.ApplyWithOptions(ctx, parsed, configport.ApplyOptions{MigrationOwner: identity.Owner}); err != nil {
200 return fmt.Errorf("daemon: legacy config apply: %w", err)
201 }
202 fresh, err := records.New(ctx, records.Config{Store: p.tenant.store, Community: p.tenant.community, Policy: p.tenant.Policy, RelayURL: p.tenant.RelayURL(), GroupID: p.tenant.meta.Name, OnGenerated: p.tenant.generatedRecord, DeliverNotification: p.tenant.deliverNotification, PushNotification: p.tenant.enqueuePush, SetPolicy: func(next policy.Policy) error { return p.tenant.applyPolicy(context.Background(), next) }})
203 if err != nil {
204 return fmt.Errorf("daemon: refresh records after legacy config: %w", err)
205 }
206 if legacyConfig {
207 p.tenant.records = fresh
208 if p.tenant.customViews != nil {
209 p.tenant.customViews.records = fresh
210 }
211 }
212 } else {
213 var next policy.Policy
214 if err := json.Unmarshal(state.Config, &next); err != nil {
215 return err
216 }
217 if next.Owner == "" {
218 var envelope struct {
219 Policy policy.Policy `json:"policy"`
220 }
221 if err := json.Unmarshal(state.Config, &envelope); err != nil {
222 return err
223 }
224 next = envelope.Policy
225 if next.Owner == "" {
226 next.Owner = p.tenant.Policy().Owner
227 }
228 }
229 p.tenant.community.SetOwner(next.Owner)
230 if p.tenant.router != nil {
231 p.tenant.replacePolicy(next)
232 } else {
233 p.tenant.mu.Lock()
234 p.tenant.policy = next
235 p.tenant.mu.Unlock()
236 }
237 fresh, err := records.New(ctx, records.Config{Store: p.tenant.store, Community: p.tenant.community, Policy: p.tenant.Policy, RelayURL: p.tenant.RelayURL(), GroupID: p.tenant.meta.Name, OnGenerated: p.tenant.generatedRecord, DeliverNotification: p.tenant.deliverNotification, PushNotification: p.tenant.enqueuePush, SetPolicy: func(next policy.Policy) error { return p.tenant.applyPolicy(context.Background(), next) }})
238 if err != nil {
239 return fmt.Errorf("daemon: refresh records identity: %w", err)
240 }
241 p.tenant.records = fresh
242 if p.tenant.customViews != nil {
243 p.tenant.customViews.records = fresh
244 }
245 }
246 }
247 if len(state.Blobs) > 0 {
248 for _, object := range state.Blobs {
249 if object.Type == "" {
250 continue
251 }
252 if _, err := p.tenant.store.DB().ExecContext(ctx, `INSERT OR REPLACE INTO blobs(sha256,size,type,uploader,uploaded) VALUES(?,?,?,?,?)`, object.SHA256, len(object.Content), object.Type, object.Uploader, object.Uploaded); err != nil {
253 return err
254 }
255 }
256 }
257 if len(state.Community) > 0 {
258 if err := restoreRetainedState(ctx, p.tenant.store, state.Community); err != nil {
259 return fmt.Errorf("daemon: restore retained state: %w", err)
260 }
261 }
262 if p.tenant.git != nil {
263 if err := p.tenant.git.Reload(ctx); err != nil {
264 return fmt.Errorf("daemon: reload Git after restore: %w", err)
265 }
266 }
267 restored = true
268 return nil
269}
270
271func (p tenantReplicationProvider) FinalizeRestore(ctx context.Context) error {
272 if p.tenant == nil {
273 return errors.New("daemon: nil tenant")
274 }
275 if err := p.tenant.store.WithTx(ctx, func(tx *sql.Tx) error {
276 now := time.Now().Unix()
277 for _, statement := range []string{
278 `CREATE TABLE IF NOT EXISTS replication_pending_until(id TEXT PRIMARY KEY NOT NULL, until INTEGER NOT NULL)`,
279 `CREATE TABLE IF NOT EXISTS replication_pr_refs(repo TEXT NOT NULL, ref TEXT NOT NULL, until INTEGER NOT NULL, PRIMARY KEY(repo,ref))`,
280 } {
281 if _, err := tx.ExecContext(ctx, statement); err != nil {
282 return err
283 }
284 }
285 if _, err := tx.ExecContext(ctx, `DELETE FROM pending_events WHERE id IN (SELECT id FROM replication_pending_until WHERE until<=?)`, now); err != nil {
286 return err
287 }
288 if _, err := tx.ExecContext(ctx, `DELETE FROM replication_pending_until WHERE until<=?`, now); err != nil {
289 return err
290 }
291 if _, err := tx.ExecContext(ctx, `DELETE FROM replication_pr_refs WHERE until>0 AND until<=?`, now); err != nil {
292 return err
293 }
294 _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO pending_events(id,reason) SELECT p.id,'backup' FROM replication_pending_until p JOIN events e ON e.id=p.id WHERE p.until>?`, now)
295 return err
296 }); err != nil {
297 return err
298 }
299 if p.tenant.git != nil {
300 return p.tenant.git.Reload(ctx)
301 }
302 return nil
303}
304
305// restoreRetainedState keeps the bind.ws relay's auxiliary state even when
306// the current schema has no first-class feature for one of those records. The
307// small native tables are deliberately explicit: a restore must either write
308// every record or return an error, never silently discard a family.
309func restoreRetainedState(ctx context.Context, store *storage.Store, rawState []byte) error {
310 var retained struct {
311 ListHistory []string `json:"listHistory"`
312 Hidden []string `json:"hidden"`
313 Hosted []string `json:"hosted"`
314 Pending []struct {
315 ID string `json:"id"`
316 Until int64 `json:"until"`
317 } `json:"pending"`
318 PRRefs []struct {
319 Repo string `json:"repo"`
320 Ref string `json:"ref"`
321 Until int64 `json:"until"`
322 } `json:"prRefs"`
323 }
324 if err := json.Unmarshal(rawState, &retained); err != nil {
325 return fmt.Errorf("invalid state: %w", err)
326 }
327 now := time.Now().Unix()
328 return store.WithTx(ctx, func(tx *sql.Tx) error {
329 if _, err := tx.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS replication_hosted_events(id TEXT PRIMARY KEY NOT NULL)`); err != nil {
330 return err
331 }
332 if _, err := tx.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS replication_pending_until(id TEXT PRIMARY KEY NOT NULL, until INTEGER NOT NULL)`); err != nil {
333 return err
334 }
335 if _, err := tx.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS replication_pr_refs(repo TEXT NOT NULL, ref TEXT NOT NULL, until INTEGER NOT NULL, PRIMARY KEY(repo,ref))`); err != nil {
336 return err
337 }
338 for _, encoded := range retained.ListHistory {
339 var item event.Event
340 if err := json.Unmarshal([]byte(encoded), &item); err != nil {
341 return fmt.Errorf("list history event: %w", err)
342 }
343 if err := event.Validate(item); err != nil {
344 return fmt.Errorf("list history event validation: %w", err)
345 }
346 if _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO list_history(owner,kind,d,event_id,created_at,saved_at,expires,raw) VALUES(?,?,?,?,?,?,?,?)`, item.PubKey, item.Kind, event.Tag(item, "d"), item.ID, item.CreatedAt, now, event.Expiration(item), encoded); err != nil {
347 return err
348 }
349 }
350 for _, id := range retained.Hidden {
351 if id == "" {
352 return errors.New("hidden event id is empty")
353 }
354 if _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO hidden_events(id,reason) VALUES(?,?)`, id, "backup"); err != nil {
355 return err
356 }
357 }
358 for _, id := range retained.Hosted {
359 if id == "" {
360 return errors.New("hosted event id is empty")
361 }
362 if _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO replication_hosted_events(id) VALUES(?)`, id); err != nil {
363 return err
364 }
365 }
366 for _, row := range retained.Pending {
367 if row.ID == "" || row.Until < 0 {
368 return errors.New("invalid pending event state")
369 }
370 if _, err := tx.ExecContext(ctx, `INSERT OR REPLACE INTO replication_pending_until(id,until) VALUES(?,?)`, row.ID, row.Until); err != nil {
371 return err
372 }
373 // pending_events references events, so retain the marker only when the
374 // event is already present. The durable until table above preserves it
375 // for the subsequent event import.
376 if _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO pending_events(id,reason) SELECT id,'backup' FROM events WHERE id=?`, row.ID); err != nil {
377 return err
378 }
379 }
380 for _, row := range retained.PRRefs {
381 if row.Repo == "" || row.Ref == "" || row.Until < 0 {
382 return errors.New("invalid pull request reference state")
383 }
384 if _, err := tx.ExecContext(ctx, `INSERT OR REPLACE INTO replication_pr_refs(repo,ref,until) VALUES(?,?,?)`, row.Repo, row.Ref, row.Until); err != nil {
385 return err
386 }
387 }
388 return nil
389 })
390}
391
392func snapshotTables(ctx context.Context, store *storage.Store) ([]replication.BackupTable, error) {
393 var out []replication.BackupTable
394 err := store.WithTx(ctx, func(tx *sql.Tx) error {
395 rows, err := tx.QueryContext(ctx, `SELECT name,sql FROM sqlite_schema WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name`)
396 if err != nil {
397 return err
398 }
399 defer rows.Close()
400 for rows.Next() {
401 var name, definition string
402 if err := rows.Scan(&name, &definition); err != nil {
403 return err
404 }
405 if excludedSnapshotTable(name, definition) {
406 continue
407 }
408 table, err := snapshotTableTx(ctx, tx, name)
409 if err != nil {
410 return err
411 }
412 out = append(out, table)
413 }
414 return rows.Err()
415 })
416 return out, err
417}
418
419func excludedSnapshotTable(name, definition string) bool {
420 lower := strings.ToLower(name + " " + definition)
421 for _, part := range []string{"work_intents", "replication_job_cursors", "browser_sessions", "auth_challenges", "push_queue", "push_delivered", "delivery_queue"} {
422 if strings.Contains(lower, part) {
423 return true
424 }
425 }
426 return strings.Contains(lower, "virtual table") || strings.HasSuffix(name, "_content") || strings.HasSuffix(name, "_data") || strings.HasSuffix(name, "_idx") || strings.HasSuffix(name, "_docsize") || strings.HasSuffix(name, "_config")
427}
428
429func snapshotTable(ctx context.Context, db *sql.DB, name string) (replication.BackupTable, error) {
430 return snapshotTableDB(ctx, db, name)
431}
432
433func snapshotTableTx(ctx context.Context, tx *sql.Tx, name string) (replication.BackupTable, error) {
434 if !safeIdentifier(name) {
435 return replication.BackupTable{}, errors.New("daemon: unsafe table name")
436 }
437 rows, err := tx.QueryContext(ctx, `SELECT * FROM "`+name+`"`)
438 if err != nil {
439 return replication.BackupTable{}, err
440 }
441 defer rows.Close()
442 columns, err := rows.Columns()
443 if err != nil {
444 return replication.BackupTable{}, err
445 }
446 table := replication.BackupTable{Name: name, Columns: columns}
447 for rows.Next() {
448 values := make([]any, len(columns))
449 destinations := make([]any, len(columns))
450 for i := range values {
451 destinations[i] = &values[i]
452 }
453 if err := rows.Scan(destinations...); err != nil {
454 return table, err
455 }
456 encoded := make([]json.RawMessage, len(values))
457 for i, value := range values {
458 raw, err := marshalDBValue(value)
459 if err != nil {
460 return table, err
461 }
462 encoded[i] = raw
463 }
464 table.Rows = append(table.Rows, encoded)
465 }
466 return table, rows.Err()
467}
468
469func snapshotTableDB(ctx context.Context, db *sql.DB, name string) (replication.BackupTable, error) {
470 if !safeIdentifier(name) {
471 return replication.BackupTable{}, errors.New("daemon: unsafe table name")
472 }
473 rows, err := db.QueryContext(ctx, "SELECT * FROM \""+name+"\"")
474 if err != nil {
475 return replication.BackupTable{}, err
476 }
477 defer rows.Close()
478 cols, err := rows.Columns()
479 if err != nil {
480 return replication.BackupTable{}, err
481 }
482 out := replication.BackupTable{Name: name, Columns: cols}
483 for rows.Next() {
484 values := make([]any, len(cols))
485 dest := make([]any, len(cols))
486 for i := range values {
487 dest[i] = &values[i]
488 }
489 if err := rows.Scan(dest...); err != nil {
490 return out, err
491 }
492 encoded := make([]json.RawMessage, len(values))
493 for i, value := range values {
494 raw, err := json.Marshal(value)
495 if err != nil {
496 return out, err
497 }
498 encoded[i] = raw
499 }
500 out.Rows = append(out.Rows, encoded)
501 }
502 return out, rows.Err()
503}
504
505func restoreTables(ctx context.Context, store *storage.Store, tables []replication.BackupTable) error {
506 return store.WithTx(ctx, func(tx *sql.Tx) error {
507 if _, err := tx.ExecContext(ctx, `PRAGMA defer_foreign_keys=ON`); err != nil {
508 return err
509 }
510 for _, table := range tables {
511 if !safeIdentifier(table.Name) {
512 return errors.New("daemon: unsafe table name")
513 }
514 if _, err := tx.ExecContext(ctx, "DELETE FROM \""+table.Name+"\""); err != nil {
515 return err
516 }
517 for _, row := range table.Rows {
518 if len(row) != len(table.Columns) {
519 return errors.New("daemon: malformed table row")
520 }
521 columns := make([]string, len(table.Columns))
522 marks := make([]string, len(table.Columns))
523 args := make([]any, len(row))
524 for i, column := range table.Columns {
525 if !safeIdentifier(column) {
526 return errors.New("daemon: unsafe column name")
527 }
528 columns[i] = "\"" + column + "\""
529 marks[i] = "?"
530 value, err := unmarshalDBValue(row[i])
531 if err != nil {
532 return err
533 }
534 args[i] = value
535 }
536 query := "INSERT INTO \"" + table.Name + "\" (" + strings.Join(columns, ",") + ") VALUES (" + strings.Join(marks, ",") + ")"
537 if _, err := tx.ExecContext(ctx, query, args...); err != nil {
538 return fmt.Errorf("restore %s: %w", table.Name, err)
539 }
540 }
541 }
542 if _, err := tx.ExecContext(ctx, `DELETE FROM search`); err == nil {
543 if _, err := tx.ExecContext(ctx, `INSERT INTO search(rowid,content) SELECT seq,json_extract(raw,'$.content') FROM events WHERE json_extract(raw,'$.content')<>''`); err != nil {
544 return err
545 }
546 }
547 if _, err := tx.ExecContext(ctx, `UPDATE replication_jobs SET state='pending',phase='pending',error='',stored=0,sent=0,skipped=0,refused=0,cursor=0,rounds=0,started_at=0,finished_at=0,updated_at=strftime('%s','now')`); err != nil {
548 // Older tenant databases may not have the replication job table yet.
549 if !strings.Contains(err.Error(), "no such table") {
550 return err
551 }
552 }
553 return nil
554 })
555}
556
557func marshalDBValue(value any) (json.RawMessage, error) {
558 if bytesValue, ok := value.([]byte); ok {
559 return json.Marshal(map[string]string{"$bytes": base64.StdEncoding.EncodeToString(bytesValue)})
560 }
561 return json.Marshal(value)
562}
563
564func unmarshalDBValue(raw json.RawMessage) (any, error) {
565 var marker map[string]string
566 if err := json.Unmarshal(raw, &marker); err == nil && marker["$bytes"] != "" {
567 return base64.StdEncoding.DecodeString(marker["$bytes"])
568 }
569 decoder := json.NewDecoder(bytes.NewReader(raw))
570 decoder.UseNumber()
571 var value any
572 if err := decoder.Decode(&value); err != nil {
573 return nil, err
574 }
575 if number, ok := value.(json.Number); ok {
576 if integer, err := strconv.ParseInt(string(number), 10, 64); err == nil {
577 return integer, nil
578 }
579 return strconv.ParseFloat(string(number), 64)
580 }
581 return value, nil
582}
583
584func snapshotFiles(root string) ([]replication.BackupObject, error) {
585 if root == "" {
586 return nil, nil
587 }
588 var out []replication.BackupObject
589 err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
590 if err != nil {
591 return err
592 }
593 if info.IsDir() {
594 return nil
595 }
596 body, err := os.ReadFile(path)
597 if err != nil {
598 return err
599 }
600 sum := sha256.Sum256(body)
601 rel, err := filepath.Rel(root, path)
602 if err != nil {
603 return err
604 }
605 out = append(out, replication.BackupObject{Name: rel, SHA256: hex.EncodeToString(sum[:]), Content: body})
606 return nil
607 })
608 if os.IsNotExist(err) {
609 return nil, nil
610 }
611 return out, err
612}
613func restoreFiles(root string, objects []replication.BackupObject) error {
614 allowed := make(map[string]struct{}, len(objects))
615 for _, object := range objects {
616 allowed[filepath.Clean(object.Name)] = struct{}{}
617 }
618 if err := filepath.Walk(root, func(path string, info os.FileInfo, walkErr error) error {
619 if walkErr != nil {
620 if os.IsNotExist(walkErr) {
621 return nil
622 }
623 return walkErr
624 }
625 if info.IsDir() || path == root {
626 return nil
627 }
628 rel, err := filepath.Rel(root, path)
629 if err != nil {
630 return err
631 }
632 if _, ok := allowed[filepath.Clean(rel)]; !ok {
633 return os.Remove(path)
634 }
635 return nil
636 }); err != nil && !os.IsNotExist(err) {
637 return err
638 }
639 for _, object := range objects {
640 if object.Name == "" || filepath.IsAbs(object.Name) || strings.HasPrefix(filepath.Clean(object.Name), ".."+string(filepath.Separator)) {
641 return errors.New("daemon: unsafe backup file")
642 }
643 sum := sha256.Sum256(object.Content)
644 if hex.EncodeToString(sum[:]) != object.SHA256 {
645 return errors.New("daemon: backup file checksum mismatch")
646 }
647 path := filepath.Join(root, object.Name)
648 if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil {
649 return err
650 }
651 if err := os.WriteFile(path, object.Content, 0600); err != nil {
652 return err
653 }
654 }
655 return nil
656}
657func safeIdentifier(value string) bool {
658 if value == "" {
659 return false
660 }
661 for _, r := range value {
662 if r != '_' && (r < 'a' || r > 'z') && (r < 'A' || r > 'Z') && (r < '0' || r > '9') {
663 return false
664 }
665 }
666 return true
667}
668
669type tenantDiscovery struct {
670 tenant *Tenant
671 transport *replication.NostrTransport
672 seeds []string
673}
674
675func (t *Tenant) ReplicationDiscovery(seeds []string) replication.Discovery {
676 dialer := replication.WebsocketDialer{}
677 if t != nil && t.app != nil {
678 dialer.AllowPrivate = t.app.cfg.AllowPrivateRelays
679 dialer.MaxMessageBytes = t.app.cfg.MaxMessageBytes
680 }
681 return tenantDiscovery{tenant: t, transport: &replication.NostrTransport{Dialer: dialer}, seeds: append([]string(nil), seeds...)}
682}
683func (d tenantDiscovery) DiscoverRelays(ctx context.Context, pubkey string) ([]string, error) {
684 return d.discover(ctx, pubkey, false)
685}
686
687func (d tenantDiscovery) DiscoverReadRelays(ctx context.Context, pubkey string) ([]string, error) {
688 return d.discover(ctx, pubkey, true)
689}
690
691func (d tenantDiscovery) discover(ctx context.Context, pubkey string, read bool) ([]string, error) {
692 if d.tenant == nil {
693 return nil, errors.New("daemon: nil tenant")
694 }
695 directory := localDirectory{store: d.tenant.store}
696 local := directory.WriteRelays(pubkey)
697 if read {
698 local = directory.ReadRelays(pubkey)
699 }
700 if len(local) > 0 {
701 return local, nil
702 }
703 if d.transport == nil || len(d.seeds) == 0 {
704 return nil, nil
705 }
706 for _, seed := range d.seeds {
707 events, err := d.transport.Query(ctx, seed, event.Filter{Authors: []string{pubkey}, Kinds: []int{10002}, Tags: map[string][]string{}, Limit: intPtr(1)})
708 if err != nil {
709 continue
710 }
711 for _, item := range events {
712 if err := event.Validate(item); err != nil {
713 continue
714 }
715 if _, err := d.tenant.store.Save(ctx, item, storage.SaveOptions{Now: time.Now().Unix()}); err != nil && err != storage.ErrDuplicate && err != storage.ErrReplaced {
716 continue
717 }
718 stored := localDirectory{store: d.tenant.store}
719 if read {
720 return stored.ReadRelays(pubkey), nil
721 }
722 return stored.WriteRelays(pubkey), nil
723 }
724 }
725 return nil, nil
726}
727func intPtr(value int) *int { return &value }
728
729func (t *Tenant) PrepareReplicationCallbacks(ctx context.Context, e event.Event) ([]storage.Intent, error) {
730 if t == nil || !t.Policy().Features.Push {
731 return nil, nil
732 }
733 approved := t.callbackPolicy()
734 rows, err := t.store.DB().QueryContext(ctx, `SELECT raw FROM events WHERE kind=? ORDER BY created_at DESC`, event.KIND_PUSH_REGISTRATION)
735 if err != nil {
736 return nil, err
737 }
738 defer rows.Close()
739 registrations := make([]replication.PushRegistration, 0)
740 for rows.Next() {
741 var raw string
742 if err := rows.Scan(&raw); err != nil {
743 return nil, err
744 }
745 var registration event.Event
746 if err := json.Unmarshal([]byte(raw), &registration); err != nil {
747 continue
748 }
749 parsed, parseErr := replication.ParsePushRegistration(registration, approved, t.RelayURL())
750 if parseErr == nil {
751 registrations = append(registrations, parsed)
752 }
753 }
754 return replication.PrepareCallbackIntents(ctx, e, registrations), rows.Err()
755}
756
757// callbackPolicy keeps host approval separate from the tenant owner's
758// registration list. A tenant can opt in an origin only after the operator has
759// placed it in PushCallbackOrigins as well.
760func (t *Tenant) callbackPolicy() replication.CallbackPolicy {
761 p := t.Policy()
762 return replication.CallbackPolicy{HostOrigins: append([]string(nil), t.app.cfg.PushCallbackOrigins...), OwnerOrigins: append([]string(nil), p.PushCallbacks...)}
763}
764
765// validatePushRegistration applies the source relay's admission rules before
766// the registration is persisted. Callback workers revalidate later, but an
767// invalid registration must never become durable state.
768func (t *Tenant) validatePushRegistration(ctx context.Context, e event.Event, s relay.Session) error {
769 if e.Kind != event.KIND_PUSH_REGISTRATION {
770 return nil
771 }
772 if !t.Policy().Features.Push {
773 return errors.New("unsupported: relay push is switched off")
774 }
775 if !containsString(s.PubKeys, e.PubKey) {
776 return errors.New("auth-required: push registration must be authenticated as its author")
777 }
778 if role, err := t.community.Role(ctx, e.PubKey); err != nil {
779 return err
780 } else if role == "" {
781 return errors.New("restricted: push registrations are limited to relay members")
782 }
783 if len(event.TagValues(e, "d")) != 1 || event.Tag(e, "d") == "" || len(event.Tag(e, "d")) > 256 {
784 return errors.New("invalid: push registration needs one d tag")
785 }
786 if len(event.TagValues(e, "relay")) == 0 || len(event.TagValues(e, "relay")) > 4 || len(event.TagValues(e, "callback")) != 1 {
787 return errors.New("invalid: push registration needs relay and callback tags")
788 }
789 for _, name := range []string{"filter", "ignore"} {
790 values := event.TagValues(e, name)
791 for _, raw := range values {
792 var object map[string]json.RawMessage
793 if err := json.Unmarshal([]byte(raw), &object); err != nil {
794 return errors.New("invalid: push filter is not JSON")
795 }
796 if object == nil {
797 return errors.New("invalid: push filter must be an object")
798 }
799 for key := range object {
800 if !pushFilterField(key) {
801 return errors.New("invalid: unknown push filter field")
802 }
803 }
804 }
805 }
806 parsed, err := replication.ParsePushRegistration(e, t.callbackPolicy(), t.RelayURL())
807 if err != nil {
808 return err
809 }
810 for _, filter := range append(parsed.Filters, parsed.Ignore...) {
811 for _, kind := range filter.Kinds {
812 if kind < 0 || kind > 65535 {
813 return errors.New("invalid: push filter value out of range")
814 }
815 }
816 if filter.Since != nil && *filter.Since < 0 || filter.Until != nil && *filter.Until < 0 {
817 return errors.New("invalid: push filter value out of range")
818 }
819 }
820 return nil
821}
822
823func pushFilterField(key string) bool {
824 if key == "ids" || key == "authors" || key == "kinds" || key == "since" || key == "until" {
825 return true
826 }
827 return len(key) == 2 && key[0] == '#'
828}
829
830func (t *Tenant) ReplicationCallbackHandler() work.Handler {
831 return replication.NewCallbackHandlerWithEventVisibility(t.store, replication.CallbackClient{}, func() replication.CallbackPolicy {
832 return t.callbackPolicy()
833 }, func() bool { return t.Policy().Features.Push }, func(ctx context.Context, pubkey string) bool {
834 return t.Policy().Reads == "open" || func() bool { role, err := t.community.Role(ctx, pubkey); return err == nil && role != "" }()
835 }, func(ctx context.Context, pubkey string, delivered event.Event) bool {
836 return t.gate.CanSee(ctx, delivered, relay.Session{PubKeys: []string{pubkey}, RelayURL: t.RelayURL()}, nil)
837 })
838}

About

ownerc01503e8035f
visibilitypublic
headd730679639bf

Maintainers

  • c01503e8035f owner
  • ba1b5beed5b5 agent hermes

Clone

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

Refs