| 1 | package replication |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "io" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "strings" |
| 12 | "time" |
| 13 | |
| 14 | "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 15 | "github.com/FelineStateMachine/tinyrelay/internal/storage" |
| 16 | "github.com/FelineStateMachine/tinyrelay/internal/work" |
| 17 | ) |
| 18 | |
| 19 | // Config supplies a Service's store, routing policy, transports and optional |
| 20 | // application hooks. Store is required. DataDir defaults to the store's |
| 21 | // directory; Workers defaults to work.RunPool's worker count when Run is used. |
| 22 | // The caller retains ownership of the supplied transports and store. |
| 23 | type Config struct { |
| 24 | Store *storage.Store |
| 25 | Policy Policy |
| 26 | CurrentPolicy func() Policy |
| 27 | Directory RelayDirectory |
| 28 | // Owner identifies the tenant owner for owner-scoped management jobs. |
| 29 | // It is used by backfill when the UI leaves the relay list blank. |
| 30 | Owner func() string |
| 31 | Discovery Discovery |
| 32 | Delivery DeliveryTransport |
| 33 | DeliveryAllowed func(context.Context, storage.Intent) bool |
| 34 | Pull PullTransport |
| 35 | Push PushTransport |
| 36 | DataDir string |
| 37 | Ingest func(context.Context, event.Event, Origin) error |
| 38 | ExecuteExternal func(context.Context, JobSpec) error |
| 39 | BackupProvider BackupProvider |
| 40 | Workers int |
| 41 | } |
| 42 | |
| 43 | // Service owns replication job state and the durable queue built on Store. |
| 44 | type Service struct { |
| 45 | store *storage.Store |
| 46 | queue *work.Queue |
| 47 | config Config |
| 48 | } |
| 49 | |
| 50 | // NewService initializes replication job tables and returns a service. |
| 51 | func NewService(config Config) (*Service, error) { |
| 52 | if config.Store == nil { |
| 53 | return nil, errors.New("replication: store is required" ) |
| 54 | } |
| 55 | if config.DataDir == "" { |
| 56 | config.DataDir = filepath.Dir(config.Store.Path()) |
| 57 | } |
| 58 | if err := createJobStateTable(config.Store); err != nil { |
| 59 | return nil, err |
| 60 | } |
| 61 | return &Service{store: config.Store, queue: work.New(config.Store), config: config}, nil |
| 62 | } |
| 63 | |
| 64 | // Store returns the service's supplied store. |
| 65 | func (s *Service) Store() *storage.Store { return s.store } |
| 66 | |
| 67 | // Queue returns the service's durable replication queue. |
| 68 | func (s *Service) Queue() *work.Queue { return s.queue } |
| 69 | |
| 70 | func (s *Service) Prepare(e event.Event, origin Origin) []storage.Intent { |
| 71 | directory := s.config.Directory |
| 72 | if directory == nil { |
| 73 | policy := s.config.Policy |
| 74 | if s.config.CurrentPolicy != nil { |
| 75 | policy = s.config.CurrentPolicy() |
| 76 | } |
| 77 | if !eligible(e, origin, policy) { |
| 78 | return nil |
| 79 | } |
| 80 | payload, _ := json.Marshal(IntentPayload{Origin: origin}) |
| 81 | return []storage.Intent{{Kind: "delivery-discovery" , EventID: e.ID, Target: e.PubKey, Payload: string(payload)}} |
| 82 | } |
| 83 | policy := s.config.Policy |
| 84 | if s.config.CurrentPolicy != nil { |
| 85 | policy = s.config.CurrentPolicy() |
| 86 | } |
| 87 | intents := PrepareIntents(context.Background(), e, origin, policy, directory) |
| 88 | if len(intents) == 0 && eligible(e, origin, policy) { |
| 89 | payload, _ := json.Marshal(IntentPayload{Origin: origin}) |
| 90 | return []storage.Intent{{Kind: "delivery-discovery" , EventID: e.ID, Target: e.PubKey, Payload: string(payload)}} |
| 91 | } |
| 92 | return intents |
| 93 | } |
| 94 | |
| 95 | func (s *Service) resolveDiscovery(ctx context.Context, e event.Event) RelayDirectory { |
| 96 | if s.config.Discovery == nil { |
| 97 | return nil |
| 98 | } |
| 99 | writes, err := s.config.Discovery.DiscoverRelays(ctx, e.PubKey) |
| 100 | if err != nil { |
| 101 | return nil |
| 102 | } |
| 103 | directory := staticDirectory{writes: writes, reads: map[string][]string{}} |
| 104 | if discovery, ok := s.config.Discovery.(ReadDiscovery); ok { |
| 105 | for _, recipient := range event.TagValues(e, "p" ) { |
| 106 | reads, readErr := discovery.DiscoverReadRelays(ctx, recipient) |
| 107 | if readErr == nil { |
| 108 | directory.reads[recipient] = reads |
| 109 | } |
| 110 | } |
| 111 | } |
| 112 | return directory |
| 113 | } |
| 114 | |
| 115 | type staticDirectory struct { |
| 116 | writes []string |
| 117 | reads map[string][]string |
| 118 | } |
| 119 | |
| 120 | func (d staticDirectory) WriteRelays(string) []string { return d.writes } |
| 121 | func (d staticDirectory) ReadRelays(pubkey string) []string { return d.reads[pubkey] } |
| 122 | |
| 123 | func (s *Service) AddJob(ctx context.Context, spec JobSpec) error { |
| 124 | if _, err := EnqueueJob(ctx, s.queue, spec, time.Now()); err != nil { |
| 125 | return err |
| 126 | } |
| 127 | return s.saveJobState(ctx, spec) |
| 128 | } |
| 129 | |
| 130 | func (s *Service) ListJobs(ctx context.Context) ([]JobSpec, error) { |
| 131 | rows, err := s.store.DB().QueryContext(ctx, `SELECT payload FROM replication_jobs WHERE state!='cancelled' ORDER BY created_at,id` ) |
| 132 | if err != nil { |
| 133 | return nil, fmt.Errorf("replication: list jobs: %w" , err) |
| 134 | } |
| 135 | defer rows.Close() |
| 136 | jobs := make([]JobSpec, 0) |
| 137 | for rows.Next() { |
| 138 | var raw string |
| 139 | if err := rows.Scan(&raw); err != nil { |
| 140 | return nil, err |
| 141 | } |
| 142 | var spec JobSpec |
| 143 | if json.Unmarshal([]byte(raw), &spec) == nil { |
| 144 | jobs = append(jobs, spec) |
| 145 | } |
| 146 | } |
| 147 | return jobs, rows.Err() |
| 148 | } |
| 149 | |
| 150 | func (s *Service) RemoveJob(ctx context.Context, id string) error { |
| 151 | if strings.TrimSpace(id) == "" { |
| 152 | return errors.New("replication: job id is required" ) |
| 153 | } |
| 154 | _, err := s.store.DB().ExecContext(ctx, `UPDATE replication_jobs SET state='cancelled',updated_at=? WHERE id=?` , time.Now().Unix(), id) |
| 155 | if err != nil { |
| 156 | return err |
| 157 | } |
| 158 | _, err = s.store.DB().ExecContext(ctx, `UPDATE work_intents SET state='cancelled',updated_at=?,claim_token='',claim_until=0 WHERE event_id=? AND kind='job'` , time.Now().Unix(), id) |
| 159 | return err |
| 160 | } |
| 161 | |
| 162 | func (s *Service) RunJob(ctx context.Context, id string) error { |
| 163 | if _, err := s.store.DB().ExecContext(ctx, `UPDATE replication_jobs SET state='pending',updated_at=? WHERE id=? AND state!='cancelled'` , time.Now().Unix(), id); err != nil { |
| 164 | return err |
| 165 | } |
| 166 | if _, err := s.store.DB().ExecContext(ctx, `UPDATE work_intents SET state='pending',next_at=?,updated_at=?,claim_token='',claim_until=0 WHERE event_id=? AND kind='job' AND state!='completed' AND state!='cancelled'` , time.Now().Unix(), time.Now().Unix(), id); err != nil { |
| 167 | return fmt.Errorf("replication: run job: %w" , err) |
| 168 | } |
| 169 | if err := s.saveJobState(ctx, s.jobSpec(ctx, id)); err != nil { |
| 170 | return err |
| 171 | } |
| 172 | return nil |
| 173 | } |
| 174 | |
| 175 | // Handlers returns the replication-owned durable work handlers. The daemon |
| 176 | // composes these with tenant handlers before starting its worker pool. |
| 177 | func (s *Service) Handlers() map[string]work.Handler { |
| 178 | handlers := map[string]work.Handler{"job" : s.handleJob, "delivery-discovery" : s.handleDiscovery} |
| 179 | if s.config.Delivery != nil { |
| 180 | delivery := NewDeliveryHandler(s.store, s.config.Delivery) |
| 181 | handlers["delivery" ] = func(ctx context.Context, intent work.Intent) error { |
| 182 | storageIntent := storage.Intent{Kind: intent.Kind, EventID: intent.EventID, Target: intent.Target, Payload: intent.Payload} |
| 183 | if s.config.DeliveryAllowed != nil && !s.config.DeliveryAllowed(ctx, storageIntent) { |
| 184 | return nil |
| 185 | } |
| 186 | if s.config.DeliveryAllowed == nil && !s.deliveryAllowed(ctx, storageIntent) { |
| 187 | return nil |
| 188 | } |
| 189 | return delivery(ctx, storageIntent) |
| 190 | } |
| 191 | } |
| 192 | return handlers |
| 193 | } |
| 194 | |
| 195 | // Run is a convenience for replication-only callers. Applications that own |
| 196 | // additional durable work should compose Handlers and call work.RunPool. |
| 197 | func (s *Service) Run(ctx context.Context) error { |
| 198 | return work.RunPool(ctx, s.queue, s.Handlers(), work.PoolOptions{Workers: s.config.Workers}) |
| 199 | } |
| 200 | |
| 201 | func (s *Service) handleDiscovery(ctx context.Context, intent work.Intent) error { |
| 202 | if s.config.Discovery == nil { |
| 203 | return errors.New("replication: relay discovery is not configured" ) |
| 204 | } |
| 205 | result, err := s.store.Query(ctx, event.Filter{IDs: []string{intent.EventID}, Tags: map[string][]string{}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: 1}) |
| 206 | if err != nil { |
| 207 | return err |
| 208 | } |
| 209 | if len(result.Events) == 0 { |
| 210 | return nil |
| 211 | } |
| 212 | policy := s.config.Policy |
| 213 | if s.config.CurrentPolicy != nil { |
| 214 | policy = s.config.CurrentPolicy() |
| 215 | } |
| 216 | directory := s.resolveDiscovery(ctx, result.Events[0]) |
| 217 | if directory == nil { |
| 218 | return errors.New("replication: no relay routing evidence yet" ) |
| 219 | } |
| 220 | intents := PrepareIntents(ctx, result.Events[0], OriginClient, policy, directory) |
| 221 | if len(intents) == 0 { |
| 222 | return errors.New("replication: no relay routing evidence yet" ) |
| 223 | } |
| 224 | for _, delivery := range intents { |
| 225 | if _, err := s.queue.Enqueue(ctx, work.Intent{Kind: delivery.Kind, EventID: delivery.EventID, Target: delivery.Target, Payload: delivery.Payload}); err != nil { |
| 226 | return err |
| 227 | } |
| 228 | } |
| 229 | return nil |
| 230 | } |
| 231 | |
| 232 | func (s *Service) deliveryAllowed(ctx context.Context, intent storage.Intent) bool { |
| 233 | policy := s.config.Policy |
| 234 | if s.config.CurrentPolicy != nil { |
| 235 | policy = s.config.CurrentPolicy() |
| 236 | } |
| 237 | result, err := s.store.Query(ctx, event.Filter{IDs: []string{intent.EventID}, Tags: map[string][]string{}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: 1}) |
| 238 | if err != nil || len(result.Events) == 0 { |
| 239 | return false |
| 240 | } |
| 241 | directory := s.config.Directory |
| 242 | if directory == nil && s.config.Discovery != nil { |
| 243 | relays, discoverErr := s.config.Discovery.DiscoverRelays(ctx, result.Events[0].PubKey) |
| 244 | if discoverErr != nil { |
| 245 | return false |
| 246 | } |
| 247 | directory = staticDirectory{writes: relays, reads: map[string][]string{}} |
| 248 | } |
| 249 | if directory == nil { |
| 250 | return false |
| 251 | } |
| 252 | return containsIntent(PrepareIntents(ctx, result.Events[0], OriginClient, policy, directory), intent.Target) |
| 253 | } |
| 254 | |
| 255 | func containsIntent(intents []storage.Intent, target string) bool { |
| 256 | for _, intent := range intents { |
| 257 | if intent.Target == target { |
| 258 | return true |
| 259 | } |
| 260 | } |
| 261 | return false |
| 262 | } |
| 263 | |
| 264 | func (s *Service) handleJob(ctx context.Context, intent work.Intent) error { |
| 265 | var spec JobSpec |
| 266 | if err := json.Unmarshal([]byte(intent.Payload), &spec); err != nil { |
| 267 | return err |
| 268 | } |
| 269 | if err := ValidateJob(spec); err != nil { |
| 270 | return err |
| 271 | } |
| 272 | started := time.Now().Unix() |
| 273 | if _, err := s.store.DB().ExecContext(ctx, `UPDATE replication_jobs SET phase='running',error='',started_at=?,updated_at=? WHERE id=?` , started, started, spec.ID); err != nil { |
| 274 | return fmt.Errorf("replication: mark job running: %w" , err) |
| 275 | } |
| 276 | var runErr error |
| 277 | if s.config.ExecuteExternal != nil { |
| 278 | runErr = s.config.ExecuteExternal(ctx, spec) |
| 279 | } else { |
| 280 | switch spec.Kind { |
| 281 | case JobPull: |
| 282 | runErr = s.pull(ctx, spec) |
| 283 | case JobPush: |
| 284 | runErr = s.push(ctx, spec) |
| 285 | case JobImport: |
| 286 | runErr = s.importFile(ctx, spec) |
| 287 | case JobDump: |
| 288 | runErr = s.dumpFile(ctx, spec) |
| 289 | case JobBackup: |
| 290 | runErr = s.backupFile(ctx, spec) |
| 291 | case JobMirror: |
| 292 | runErr = s.mirror(ctx, spec) |
| 293 | default: |
| 294 | runErr = fmt.Errorf("replication: no local executor for %s" , spec.Kind) |
| 295 | } |
| 296 | } |
| 297 | if runErr != nil { |
| 298 | if _, updateErr := s.store.DB().ExecContext(ctx, `UPDATE replication_jobs SET phase='failed',error=?,finished_at=?,updated_at=? WHERE id=?` , runErr.Error(), time.Now().Unix(), time.Now().Unix(), spec.ID); updateErr != nil { |
| 299 | return errors.Join(runErr, fmt.Errorf("replication: mark job failed: %w" , updateErr)) |
| 300 | } |
| 301 | return runErr |
| 302 | } |
| 303 | if _, err := s.store.DB().ExecContext(ctx, `UPDATE replication_jobs SET phase='complete',error='',finished_at=?,updated_at=? WHERE id=?` , time.Now().Unix(), time.Now().Unix(), spec.ID); err != nil { |
| 304 | return fmt.Errorf("replication: mark job complete: %w" , err) |
| 305 | } |
| 306 | if spec.Every > 0 { |
| 307 | _, err := EnqueueJob(ctx, s.queue, spec, NextRun(spec, time.Now())) |
| 308 | return err |
| 309 | } |
| 310 | return nil |
| 311 | } |
| 312 | |
| 313 | func (s *Service) pull(ctx context.Context, spec JobSpec) error { |
| 314 | if s.config.Pull == nil { |
| 315 | return errors.New("replication: pull transport is not configured" ) |
| 316 | } |
| 317 | filter, err := parseJobFilter(spec.Filter) |
| 318 | if err != nil { |
| 319 | return err |
| 320 | } |
| 321 | relays := append([]string(nil), spec.Relays...) |
| 322 | if len(relays) == 0 && spec.DiscoverPubKey != "" { |
| 323 | discovery, ok := s.config.Discovery.(ReadDiscovery) |
| 324 | if !ok { |
| 325 | return errors.New("replication: discovery is not configured" ) |
| 326 | } |
| 327 | relays, err = discovery.DiscoverReadRelays(ctx, spec.DiscoverPubKey) |
| 328 | if err != nil { |
| 329 | return fmt.Errorf("replication: discover inbox relays: %w" , err) |
| 330 | } |
| 331 | } |
| 332 | var total SyncStats |
| 333 | for _, target := range relays { |
| 334 | stats, err := PullOnceWithIngest(ctx, s.store, s.config.Pull, target, filter, time.Now().Unix(), s.config.Ingest) |
| 335 | total.Stored += stats.Stored |
| 336 | total.Duplicates += stats.Duplicates |
| 337 | total.Rejected += stats.Rejected |
| 338 | if err != nil { |
| 339 | return err |
| 340 | } |
| 341 | } |
| 342 | return s.recordJobStats(ctx, spec.ID, total, 0, len(relays)) |
| 343 | } |
| 344 | |
| 345 | func (s *Service) push(ctx context.Context, spec JobSpec) error { |
| 346 | if s.config.Push == nil { |
| 347 | return errors.New("replication: push transport is not configured" ) |
| 348 | } |
| 349 | filter, err := parseJobFilter(spec.Filter) |
| 350 | if err != nil { |
| 351 | return err |
| 352 | } |
| 353 | var total SyncStats |
| 354 | var lastCursor int64 |
| 355 | for _, target := range spec.Relays { |
| 356 | cursor := s.loadCursor(ctx, spec.ID, target) |
| 357 | stats, nextCursor, err := PushAfter(ctx, s.store, s.config.Push, target, cursor, filter, time.Now().Unix()) |
| 358 | total.Sent += stats.Sent |
| 359 | total.Refused += stats.Refused |
| 360 | lastCursor = nextCursor |
| 361 | if err != nil { |
| 362 | return err |
| 363 | } |
| 364 | if err := s.saveCursor(ctx, spec.ID, target, nextCursor); err != nil { |
| 365 | return err |
| 366 | } |
| 367 | } |
| 368 | return s.recordJobStats(ctx, spec.ID, total, lastCursor, len(spec.Relays)) |
| 369 | } |
| 370 | |
| 371 | func (s *Service) importFile(ctx context.Context, spec JobSpec) error { |
| 372 | if len(spec.Relays) != 1 { |
| 373 | return errors.New("replication: import job needs one local file target" ) |
| 374 | } |
| 375 | path, err := safeArtifactPath(s.config.DataDir, spec.Relays[0]) |
| 376 | if err != nil { |
| 377 | return err |
| 378 | } |
| 379 | file, err := os.Open(path) |
| 380 | if err != nil { |
| 381 | return err |
| 382 | } |
| 383 | defer file.Close() |
| 384 | if s.config.Ingest != nil { |
| 385 | return readJSONLEvents(ctx, file, func(item event.Event) error { return s.config.Ingest(ctx, item, OriginImport) }) |
| 386 | } |
| 387 | return ImportJSONL(ctx, s.store, file, time.Now().Unix()) |
| 388 | } |
| 389 | |
| 390 | func (s *Service) dumpFile(ctx context.Context, spec JobSpec) error { |
| 391 | name := spec.ID + ".jsonl" |
| 392 | if len(spec.Relays) == 1 { |
| 393 | name = spec.Relays[0] |
| 394 | } |
| 395 | path, err := safeArtifactPath(s.config.DataDir, name) |
| 396 | if err != nil { |
| 397 | return err |
| 398 | } |
| 399 | return writeArtifact(path, func(output io.Writer) error { return WriteDump(ctx, s.store, output, time.Now().Unix()) }) |
| 400 | } |
| 401 | |
| 402 | func (s *Service) backupFile(ctx context.Context, spec JobSpec) error { |
| 403 | name := spec.ID + ".backup.json" |
| 404 | if len(spec.Relays) == 1 { |
| 405 | name = spec.Relays[0] |
| 406 | } |
| 407 | path, err := safeArtifactPath(s.config.DataDir, name) |
| 408 | if err != nil { |
| 409 | return err |
| 410 | } |
| 411 | if s.config.BackupProvider != nil { |
| 412 | archive, err := CreateBackupWithProvider(ctx, s.store, s.config.BackupProvider, time.Now().Unix()) |
| 413 | if err != nil { |
| 414 | return err |
| 415 | } |
| 416 | return writeArtifact(path, func(output io.Writer) error { return json.NewEncoder(output).Encode(archive) }) |
| 417 | } |
| 418 | return writeArtifact(path, func(output io.Writer) error { return WriteBackup(ctx, s.store, output, time.Now().Unix()) }) |
| 419 | } |
| 420 | |
| 421 | func (s *Service) mirror(ctx context.Context, spec JobSpec) error { |
| 422 | // A mirror job uses the same durable pull cursor path for relay content; |
| 423 | // hosts with site/blob storage can additionally supply ExecuteExternal. |
| 424 | if s.config.Pull == nil { |
| 425 | return errors.New("replication: mirror transport is not configured" ) |
| 426 | } |
| 427 | return s.pull(ctx, spec) |
| 428 | } |
| 429 | |
| 430 | func createJobStateTable(store *storage.Store) error { |
| 431 | _, err := store.DB().Exec(`CREATE TABLE IF NOT EXISTS replication_jobs (id TEXT PRIMARY KEY, payload TEXT NOT NULL, state TEXT NOT NULL, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, phase TEXT NOT NULL DEFAULT 'pending', error TEXT NOT NULL DEFAULT '', stored INTEGER NOT NULL DEFAULT 0, sent INTEGER NOT NULL DEFAULT 0, skipped INTEGER NOT NULL DEFAULT 0, refused INTEGER NOT NULL DEFAULT 0, cursor INTEGER NOT NULL DEFAULT 0, rounds INTEGER NOT NULL DEFAULT 0, started_at INTEGER NOT NULL DEFAULT 0, finished_at INTEGER NOT NULL DEFAULT 0); CREATE TABLE IF NOT EXISTS replication_job_cursors (job_id TEXT NOT NULL, target TEXT NOT NULL, cursor INTEGER NOT NULL DEFAULT 0, PRIMARY KEY(job_id,target))` ) |
| 432 | return err |
| 433 | } |
| 434 | |
| 435 | func (s *Service) saveJobState(ctx context.Context, spec JobSpec) error { |
| 436 | raw, err := json.Marshal(spec) |
| 437 | if err != nil { |
| 438 | return err |
| 439 | } |
| 440 | now := time.Now().Unix() |
| 441 | _, err = s.store.DB().ExecContext(ctx, `INSERT INTO replication_jobs(id,payload,state,created_at,updated_at) VALUES(?,?, 'pending',?,?) ON CONFLICT(id) DO UPDATE SET payload=excluded.payload,state='pending',updated_at=excluded.updated_at` , spec.ID, string(raw), now, now) |
| 442 | return err |
| 443 | } |
| 444 | |
| 445 | func (s *Service) jobSpec(ctx context.Context, id string) JobSpec { |
| 446 | var raw string |
| 447 | _ = s.store.DB().QueryRowContext(ctx, `SELECT payload FROM replication_jobs WHERE id=?` , id).Scan(&raw) |
| 448 | var spec JobSpec |
| 449 | _ = json.Unmarshal([]byte(raw), &spec) |
| 450 | return spec |
| 451 | } |
| 452 | |
| 453 | func (s *Service) loadCursor(ctx context.Context, id, target string) int64 { |
| 454 | var cursor int64 |
| 455 | _ = s.store.DB().QueryRowContext(ctx, `SELECT cursor FROM replication_job_cursors WHERE job_id=? AND target=?` , id, target).Scan(&cursor) |
| 456 | return cursor |
| 457 | } |
| 458 | |
| 459 | func (s *Service) saveCursor(ctx context.Context, id, target string, cursor int64) error { |
| 460 | _, err := s.store.DB().ExecContext(ctx, `INSERT INTO replication_job_cursors(job_id,target,cursor) VALUES(?,?,?) ON CONFLICT(job_id,target) DO UPDATE SET cursor=excluded.cursor` , id, target, cursor) |
| 461 | return err |
| 462 | } |
| 463 | |
| 464 | func (s *Service) recordJobStats(ctx context.Context, id string, stats SyncStats, cursor int64, rounds int) error { |
| 465 | _, err := s.store.DB().ExecContext(ctx, `UPDATE replication_jobs SET stored=stored+?,sent=sent+?,skipped=skipped+?,refused=refused+?,cursor=?,rounds=rounds+?,updated_at=? WHERE id=?` , stats.Stored, stats.Sent, stats.Duplicates, stats.Refused+stats.Rejected, cursor, rounds, time.Now().Unix(), id) |
| 466 | if err != nil { |
| 467 | return fmt.Errorf("replication: record job stats: %w" , err) |
| 468 | } |
| 469 | return nil |
| 470 | } |
| 471 | |
| 472 | func parseJobFilter(raw string) (event.Filter, error) { |
| 473 | if raw == "" { |
| 474 | return event.Filter{Tags: map[string][]string{}}, nil |
| 475 | } |
| 476 | if filter, err := event.ParseFilter([]byte(raw)); err == nil { |
| 477 | return filter, nil |
| 478 | } else { |
| 479 | return event.Filter{}, fmt.Errorf("replication: invalid job filter: %w" , err) |
| 480 | } |
| 481 | } |
| 482 | |
| 483 | func safeArtifactPath(root, name string) (string, error) { |
| 484 | if root == "" || filepath.IsAbs(name) { |
| 485 | return "" , errors.New("replication: invalid artifact path" ) |
| 486 | } |
| 487 | path := filepath.Join(root, name) |
| 488 | rel, err := filepath.Rel(root, path) |
| 489 | if err != nil || rel == ".." || strings.HasPrefix(rel, ".." +string(filepath.Separator)) { |
| 490 | return "" , errors.New("replication: artifact path escapes tenant data directory" ) |
| 491 | } |
| 492 | return path, nil |
| 493 | } |