1package daemon23// This file contains the concrete adapters that keep replication tenant4// scoped. The replication package stays independent from daemon ownership,5// filesystem layout, and community implementation details.67import (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"2223 "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)3334type tenantReplicationProvider struct{ tenant *Tenant }3536func (t *Tenant) ReplicationBackupProvider() replication.BackupProvider {37 return tenantReplicationProvider{tenant: t}38}3940func (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{}, admissionErr47 }48 defer done()49 state := replication.BackupState{}50 policy, err := json.Marshal(p.tenant.Policy())51 if err != nil {52 return state, err53 }54 state.Config = policy55 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, err62 }63 state.Identity = identity64 state.Database, err = snapshotTables(ctx, p.tenant.store)65 if err != nil {66 return state, err67 }68 state.Blobs, err = snapshotFiles(filepath.Join(p.tenant.meta.Paths.Root, "blobs"))69 if err != nil {70 return state, err71 }72 state.Git, err = snapshotFiles(p.tenant.meta.Paths.Git)73 if err != nil {74 return state, err75 }76 return state, nil77}7879func (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{}, err83 }84 archive := replication.BackupArchive{Format: replication.BackupFormat, Created: now, State: state}85 for _, table := range state.Database {86 if table.Name != "events" {87 continue88 }89 for _, row := range table.Rows {90 for i, column := range table.Columns {91 if column != "raw" || i >= len(row) {92 continue93 }94 var raw string95 if err := json.Unmarshal(row[i], &raw); err != nil {96 return replication.BackupArchive{}, err97 }98 var item event.Event99 if err := json.Unmarshal([]byte(raw), &item); err != nil {100 return replication.BackupArchive{}, err101 }102 archive.Events = append(archive.Events, item)103 }104 }105 }106 return replication.SealBackup(archive)107}108109func (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 admissionErr116 }117 defer done()118 oldBlobs, err := snapshotFiles(filepath.Join(p.tenant.meta.Paths.Root, "blobs"))119 if err != nil {120 return err121 }122 oldGit, err := snapshotFiles(p.tenant.meta.Paths.Git)123 if err != nil {124 return err125 }126 oldTables, err := snapshotTables(ctx, p.tenant.store)127 if err != nil {128 return err129 }130 restored := false131 defer func() {132 if restored {133 return134 }135 // Restore is a single maintenance operation. Any failure after the136 // 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 err146 }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 err151 }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 err157 }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 := true184 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 = fresh208 if p.tenant.customViews != nil {209 p.tenant.customViews.records = fresh210 }211 }212 } else {213 var next policy.Policy214 if err := json.Unmarshal(state.Config, &next); err != nil {215 return err216 }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 err223 }224 next = envelope.Policy225 if next.Owner == "" {226 next.Owner = p.tenant.Policy().Owner227 }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 = next235 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 = fresh242 if p.tenant.customViews != nil {243 p.tenant.customViews.records = fresh244 }245 }246 }247 if len(state.Blobs) > 0 {248 for _, object := range state.Blobs {249 if object.Type == "" {250 continue251 }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 err254 }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 = true268 return nil269}270271func (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 err283 }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 err287 }288 if _, err := tx.ExecContext(ctx, `DELETE FROM replication_pending_until WHERE until<=?`, now); err != nil {289 return err290 }291 if _, err := tx.ExecContext(ctx, `DELETE FROM replication_pr_refs WHERE until>0 AND until<=?`, now); err != nil {292 return err293 }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 err296 }); err != nil {297 return err298 }299 if p.tenant.git != nil {300 return p.tenant.git.Reload(ctx)301 }302 return nil303}304305// restoreRetainedState keeps the bind.ws relay's auxiliary state even when306// the current schema has no first-class feature for one of those records. The307// small native tables are deliberately explicit: a restore must either write308// 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 err331 }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 err334 }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 err337 }338 for _, encoded := range retained.ListHistory {339 var item event.Event340 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 err348 }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 err356 }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 err364 }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 err372 }373 // pending_events references events, so retain the marker only when the374 // event is already present. The durable until table above preserves it375 // 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 err378 }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 err386 }387 }388 return nil389 })390}391392func snapshotTables(ctx context.Context, store *storage.Store) ([]replication.BackupTable, error) {393 var out []replication.BackupTable394 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 err398 }399 defer rows.Close()400 for rows.Next() {401 var name, definition string402 if err := rows.Scan(&name, &definition); err != nil {403 return err404 }405 if excludedSnapshotTable(name, definition) {406 continue407 }408 table, err := snapshotTableTx(ctx, tx, name)409 if err != nil {410 return err411 }412 out = append(out, table)413 }414 return rows.Err()415 })416 return out, err417}418419func 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 true424 }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}428429func snapshotTable(ctx context.Context, db *sql.DB, name string) (replication.BackupTable, error) {430 return snapshotTableDB(ctx, db, name)431}432433func 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{}, err440 }441 defer rows.Close()442 columns, err := rows.Columns()443 if err != nil {444 return replication.BackupTable{}, err445 }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, err455 }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, err461 }462 encoded[i] = raw463 }464 table.Rows = append(table.Rows, encoded)465 }466 return table, rows.Err()467}468469func 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{}, err476 }477 defer rows.Close()478 cols, err := rows.Columns()479 if err != nil {480 return replication.BackupTable{}, err481 }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, err491 }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, err497 }498 encoded[i] = raw499 }500 out.Rows = append(out.Rows, encoded)501 }502 return out, rows.Err()503}504505func 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 err509 }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 err516 }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 err533 }534 args[i] = value535 }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 err545 }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 err551 }552 }553 return nil554 })555}556557func 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}563564func unmarshalDBValue(raw json.RawMessage) (any, error) {565 var marker map[string]string566 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 any572 if err := decoder.Decode(&value); err != nil {573 return nil, err574 }575 if number, ok := value.(json.Number); ok {576 if integer, err := strconv.ParseInt(string(number), 10, 64); err == nil {577 return integer, nil578 }579 return strconv.ParseFloat(string(number), 64)580 }581 return value, nil582}583584func snapshotFiles(root string) ([]replication.BackupObject, error) {585 if root == "" {586 return nil, nil587 }588 var out []replication.BackupObject589 err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error {590 if err != nil {591 return err592 }593 if info.IsDir() {594 return nil595 }596 body, err := os.ReadFile(path)597 if err != nil {598 return err599 }600 sum := sha256.Sum256(body)601 rel, err := filepath.Rel(root, path)602 if err != nil {603 return err604 }605 out = append(out, replication.BackupObject{Name: rel, SHA256: hex.EncodeToString(sum[:]), Content: body})606 return nil607 })608 if os.IsNotExist(err) {609 return nil, nil610 }611 return out, err612}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 nil622 }623 return walkErr624 }625 if info.IsDir() || path == root {626 return nil627 }628 rel, err := filepath.Rel(root, path)629 if err != nil {630 return err631 }632 if _, ok := allowed[filepath.Clean(rel)]; !ok {633 return os.Remove(path)634 }635 return nil636 }); err != nil && !os.IsNotExist(err) {637 return err638 }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 err650 }651 if err := os.WriteFile(path, object.Content, 0600); err != nil {652 return err653 }654 }655 return nil656}657func safeIdentifier(value string) bool {658 if value == "" {659 return false660 }661 for _, r := range value {662 if r != '_' && (r < 'a' || r > 'z') && (r < 'A' || r > 'Z') && (r < '0' || r > '9') {663 return false664 }665 }666 return true667}668669type tenantDiscovery struct {670 tenant *Tenant671 transport *replication.NostrTransport672 seeds []string673}674675func (t *Tenant) ReplicationDiscovery(seeds []string) replication.Discovery {676 dialer := replication.WebsocketDialer{}677 if t != nil && t.app != nil {678 dialer.AllowPrivate = t.app.cfg.AllowPrivateRelays679 dialer.MaxMessageBytes = t.app.cfg.MaxMessageBytes680 }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}686687func (d tenantDiscovery) DiscoverReadRelays(ctx context.Context, pubkey string) ([]string, error) {688 return d.discover(ctx, pubkey, true)689}690691func (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, nil702 }703 if d.transport == nil || len(d.seeds) == 0 {704 return nil, nil705 }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 continue710 }711 for _, item := range events {712 if err := event.Validate(item); err != nil {713 continue714 }715 if _, err := d.tenant.store.Save(ctx, item, storage.SaveOptions{Now: time.Now().Unix()}); err != nil && err != storage.ErrDuplicate && err != storage.ErrReplaced {716 continue717 }718 stored := localDirectory{store: d.tenant.store}719 if read {720 return stored.ReadRelays(pubkey), nil721 }722 return stored.WriteRelays(pubkey), nil723 }724 }725 return nil, nil726}727func intPtr(value int) *int { return &value }728729func (t *Tenant) PrepareReplicationCallbacks(ctx context.Context, e event.Event) ([]storage.Intent, error) {730 if t == nil || !t.Policy().Features.Push {731 return nil, nil732 }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, err737 }738 defer rows.Close()739 registrations := make([]replication.PushRegistration, 0)740 for rows.Next() {741 var raw string742 if err := rows.Scan(&raw); err != nil {743 return nil, err744 }745 var registration event.Event746 if err := json.Unmarshal([]byte(raw), ®istration); err != nil {747 continue748 }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}756757// callbackPolicy keeps host approval separate from the tenant owner's758// registration list. A tenant can opt in an origin only after the operator has759// 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}764765// validatePushRegistration applies the source relay's admission rules before766// the registration is persisted. Callback workers revalidate later, but an767// 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 nil771 }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 err780 } 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.RawMessage793 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 err809 }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 nil821}822823func pushFilterField(key string) bool {824 if key == "ids" || key == "authors" || key == "kinds" || key == "since" || key == "until" {825 return true826 }827 return len(key) == 2 && key[0] == '#'828}829830func (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}