1package replication23import (4 "context"5 "encoding/json"6 "errors"7 "fmt"8 "io"9 "os"10 "path/filepath"11 "strings"12 "time"1314 "github.com/FelineStateMachine/tinyrelay/internal/event"15 "github.com/FelineStateMachine/tinyrelay/internal/storage"16 "github.com/FelineStateMachine/tinyrelay/internal/work"17)1819// Config supplies a Service's store, routing policy, transports and optional20// application hooks. Store is required. DataDir defaults to the store's21// directory; Workers defaults to work.RunPool's worker count when Run is used.22// The caller retains ownership of the supplied transports and store.23type Config struct {24 Store *storage.Store25 Policy Policy26 CurrentPolicy func() Policy27 Directory RelayDirectory28 // 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() string31 Discovery Discovery32 Delivery DeliveryTransport33 DeliveryAllowed func(context.Context, storage.Intent) bool34 Pull PullTransport35 Push PushTransport36 DataDir string37 Ingest func(context.Context, event.Event, Origin) error38 ExecuteExternal func(context.Context, JobSpec) error39 BackupProvider BackupProvider40 Workers int41}4243// Service owns replication job state and the durable queue built on Store.44type Service struct {45 store *storage.Store46 queue *work.Queue47 config Config48}4950// NewService initializes replication job tables and returns a service.51func 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, err60 }61 return &Service{store: config.Store, queue: work.New(config.Store), config: config}, nil62}6364// Store returns the service's supplied store.65func (s *Service) Store() *storage.Store { return s.store }6667// Queue returns the service's durable replication queue.68func (s *Service) Queue() *work.Queue { return s.queue }6970func (s *Service) Prepare(e event.Event, origin Origin) []storage.Intent {71 directory := s.config.Directory72 if directory == nil {73 policy := s.config.Policy74 if s.config.CurrentPolicy != nil {75 policy = s.config.CurrentPolicy()76 }77 if !eligible(e, origin, policy) {78 return nil79 }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.Policy84 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 intents93}9495func (s *Service) resolveDiscovery(ctx context.Context, e event.Event) RelayDirectory {96 if s.config.Discovery == nil {97 return nil98 }99 writes, err := s.config.Discovery.DiscoverRelays(ctx, e.PubKey)100 if err != nil {101 return nil102 }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] = reads109 }110 }111 }112 return directory113}114115type staticDirectory struct {116 writes []string117 reads map[string][]string118}119120func (d staticDirectory) WriteRelays(string) []string { return d.writes }121func (d staticDirectory) ReadRelays(pubkey string) []string { return d.reads[pubkey] }122123func (s *Service) AddJob(ctx context.Context, spec JobSpec) error {124 if _, err := EnqueueJob(ctx, s.queue, spec, time.Now()); err != nil {125 return err126 }127 return s.saveJobState(ctx, spec)128}129130func (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 string139 if err := rows.Scan(&raw); err != nil {140 return nil, err141 }142 var spec JobSpec143 if json.Unmarshal([]byte(raw), &spec) == nil {144 jobs = append(jobs, spec)145 }146 }147 return jobs, rows.Err()148}149150func (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 err157 }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 err160}161162func (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 err165 }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 err171 }172 return nil173}174175// Handlers returns the replication-owned durable work handlers. The daemon176// composes these with tenant handlers before starting its worker pool.177func (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 nil185 }186 if s.config.DeliveryAllowed == nil && !s.deliveryAllowed(ctx, storageIntent) {187 return nil188 }189 return delivery(ctx, storageIntent)190 }191 }192 return handlers193}194195// Run is a convenience for replication-only callers. Applications that own196// additional durable work should compose Handlers and call work.RunPool.197func (s *Service) Run(ctx context.Context) error {198 return work.RunPool(ctx, s.queue, s.Handlers(), work.PoolOptions{Workers: s.config.Workers})199}200201func (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 err208 }209 if len(result.Events) == 0 {210 return nil211 }212 policy := s.config.Policy213 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 err227 }228 }229 return nil230}231232func (s *Service) deliveryAllowed(ctx context.Context, intent storage.Intent) bool {233 policy := s.config.Policy234 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 false240 }241 directory := s.config.Directory242 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 false246 }247 directory = staticDirectory{writes: relays, reads: map[string][]string{}}248 }249 if directory == nil {250 return false251 }252 return containsIntent(PrepareIntents(ctx, result.Events[0], OriginClient, policy, directory), intent.Target)253}254255func containsIntent(intents []storage.Intent, target string) bool {256 for _, intent := range intents {257 if intent.Target == target {258 return true259 }260 }261 return false262}263264func (s *Service) handleJob(ctx context.Context, intent work.Intent) error {265 var spec JobSpec266 if err := json.Unmarshal([]byte(intent.Payload), &spec); err != nil {267 return err268 }269 if err := ValidateJob(spec); err != nil {270 return err271 }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 error277 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 runErr302 }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 err309 }310 return nil311}312313func (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 err320 }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 SyncStats333 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.Stored336 total.Duplicates += stats.Duplicates337 total.Rejected += stats.Rejected338 if err != nil {339 return err340 }341 }342 return s.recordJobStats(ctx, spec.ID, total, 0, len(relays))343}344345func (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 err352 }353 var total SyncStats354 var lastCursor int64355 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.Sent359 total.Refused += stats.Refused360 lastCursor = nextCursor361 if err != nil {362 return err363 }364 if err := s.saveCursor(ctx, spec.ID, target, nextCursor); err != nil {365 return err366 }367 }368 return s.recordJobStats(ctx, spec.ID, total, lastCursor, len(spec.Relays))369}370371func (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 err378 }379 file, err := os.Open(path)380 if err != nil {381 return err382 }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}389390func (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 err398 }399 return writeArtifact(path, func(output io.Writer) error { return WriteDump(ctx, s.store, output, time.Now().Unix()) })400}401402func (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 err410 }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 err415 }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}420421func (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}429430func 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 err433}434435func (s *Service) saveJobState(ctx context.Context, spec JobSpec) error {436 raw, err := json.Marshal(spec)437 if err != nil {438 return err439 }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 err443}444445func (s *Service) jobSpec(ctx context.Context, id string) JobSpec {446 var raw string447 _ = s.store.DB().QueryRowContext(ctx, `SELECT payload FROM replication_jobs WHERE id=?`, id).Scan(&raw)448 var spec JobSpec449 _ = json.Unmarshal([]byte(raw), &spec)450 return spec451}452453func (s *Service) loadCursor(ctx context.Context, id, target string) int64 {454 var cursor int64455 _ = s.store.DB().QueryRowContext(ctx, `SELECT cursor FROM replication_job_cursors WHERE job_id=? AND target=?`, id, target).Scan(&cursor)456 return cursor457}458459func (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 err462}463464func (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 nil470}471472func parseJobFilter(raw string) (event.Filter, error) {473 if raw == "" {474 return event.Filter{Tags: map[string][]string{}}, nil475 }476 if filter, err := event.ParseFilter([]byte(raw)); err == nil {477 return filter, nil478 } else {479 return event.Filter{}, fmt.Errorf("replication: invalid job filter: %w", err)480 }481}482483func 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, nil493}