| 1 | package replication |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "strings" |
| 9 | "time" |
| 10 | |
| 11 | "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 12 | "github.com/FelineStateMachine/tinyrelay/internal/storage" |
| 13 | "github.com/FelineStateMachine/tinyrelay/internal/work" |
| 14 | ) |
| 15 | |
| 16 | // JobKind selects the durable replication operation. |
| 17 | type JobKind string |
| 18 | |
| 19 | const ( |
| 20 | JobPull JobKind = "pull" |
| 21 | JobPush JobKind = "push" |
| 22 | JobImport JobKind = "import" |
| 23 | JobMirror JobKind = "mirror" |
| 24 | JobDump JobKind = "dump" |
| 25 | JobBackup JobKind = "backup" |
| 26 | ) |
| 27 | |
| 28 | // JobSpec describes one immediate or recurring replication job. Every is an |
| 29 | // interval in hours; zero runs once. |
| 30 | type JobSpec struct { |
| 31 | ID string `json:"id"` |
| 32 | Kind JobKind `json:"kind"` |
| 33 | Relays []string `json:"relays,omitempty"` |
| 34 | Filter string `json:"filter,omitempty"` |
| 35 | Every int `json:"every,omitempty"` |
| 36 | DiscoverPubKey string `json:"discoverPubKey,omitempty"` |
| 37 | } |
| 38 | |
| 39 | // JobRunner executes a validated job specification. |
| 40 | type JobRunner interface { |
| 41 | Run(ctx context.Context, spec JobSpec) error |
| 42 | } |
| 43 | |
| 44 | // JobHandler decodes and validates durable job intents before delegating them. |
| 45 | type JobHandler struct { |
| 46 | runner JobRunner |
| 47 | } |
| 48 | |
| 49 | func NewJobHandler(runner JobRunner) *JobHandler { return &JobHandler{runner: runner} } |
| 50 | |
| 51 | func (h *JobHandler) Handle(ctx context.Context, intent work.Intent) error { |
| 52 | if h == nil || h.runner == nil { |
| 53 | return errors.New("replication: no job runner configured" ) |
| 54 | } |
| 55 | var spec JobSpec |
| 56 | if err := json.Unmarshal([]byte(intent.Payload), &spec); err != nil { |
| 57 | return fmt.Errorf("replication: decode job: %w" , err) |
| 58 | } |
| 59 | if err := ValidateJob(spec); err != nil { |
| 60 | return err |
| 61 | } |
| 62 | return h.runner.Run(ctx, spec) |
| 63 | } |
| 64 | |
| 65 | func ValidateJob(spec JobSpec) error { |
| 66 | switch spec.Kind { |
| 67 | case JobPull, JobPush, JobImport, JobMirror, JobDump, JobBackup: |
| 68 | default: |
| 69 | return fmt.Errorf("replication: unsupported job kind %q" , spec.Kind) |
| 70 | } |
| 71 | if spec.ID == "" { |
| 72 | return errors.New("replication: job id is required" ) |
| 73 | } |
| 74 | if spec.Every < 0 { |
| 75 | return errors.New("replication: job interval cannot be negative" ) |
| 76 | } |
| 77 | if spec.Filter != "" { |
| 78 | if _, err := event.ParseFilter([]byte(spec.Filter)); err != nil { |
| 79 | return fmt.Errorf("replication: invalid job filter: %w" , err) |
| 80 | } |
| 81 | } |
| 82 | if spec.Kind == JobPull || spec.Kind == JobPush { |
| 83 | if len(spec.Relays) == 0 && !(spec.Kind == JobPull && spec.DiscoverPubKey != "" ) { |
| 84 | return errors.New("replication: relay job needs a target" ) |
| 85 | } |
| 86 | for _, relay := range spec.Relays { |
| 87 | if !safeRelayURL(relay) { |
| 88 | return fmt.Errorf("replication: invalid relay target %q" , relay) |
| 89 | } |
| 90 | } |
| 91 | } |
| 92 | return nil |
| 93 | } |
| 94 | |
| 95 | func EnqueueJob(ctx context.Context, queue *work.Queue, spec JobSpec, at time.Time) (string, error) { |
| 96 | if err := ValidateJob(spec); err != nil { |
| 97 | return "" , err |
| 98 | } |
| 99 | payload, err := json.Marshal(spec) |
| 100 | if err != nil { |
| 101 | return "" , fmt.Errorf("replication: encode job: %w" , err) |
| 102 | } |
| 103 | id := fmt.Sprintf("%s/%d" , spec.ID, at.Unix()) |
| 104 | // The schedule timestamp makes each recurring execution a distinct |
| 105 | // durable intent. The work table intentionally deduplicates identical |
| 106 | // (kind,event,target) tuples, so a stable target would drop later runs. |
| 107 | return queue.Enqueue(ctx, work.Intent{ID: id, Kind: "job" , EventID: spec.ID, Target: string(spec.Kind) + ":" + id, Payload: string(payload), NextAt: at}) |
| 108 | } |
| 109 | |
| 110 | func CancelJob(ctx context.Context, queue *work.Queue, id string, now time.Time) error { |
| 111 | if strings.TrimSpace(id) == "" { |
| 112 | return errors.New("replication: job id is required" ) |
| 113 | } |
| 114 | return queue.Cancel(ctx, id, now) |
| 115 | } |
| 116 | |
| 117 | // RescheduleJob makes a recurring job eligible again after a successful run. |
| 118 | // It is separate from completion so the completed intent remains an audit |
| 119 | // record and a fresh run receives an independent claim token. |
| 120 | func RescheduleJob(ctx context.Context, queue *work.Queue, spec JobSpec, at time.Time) (string, error) { |
| 121 | if spec.Every <= 0 { |
| 122 | return "" , errors.New("replication: only recurring jobs can be rescheduled" ) |
| 123 | } |
| 124 | return EnqueueJob(ctx, queue, spec, at.Add(time.Duration(spec.Every)*time.Hour)) |
| 125 | } |
| 126 | |
| 127 | func IntentForJob(spec JobSpec, at time.Time) (storage.Intent, error) { |
| 128 | if err := ValidateJob(spec); err != nil { |
| 129 | return storage.Intent{}, err |
| 130 | } |
| 131 | payload, err := json.Marshal(spec) |
| 132 | if err != nil { |
| 133 | return storage.Intent{}, fmt.Errorf("replication: encode job intent: %w" , err) |
| 134 | } |
| 135 | return storage.Intent{Kind: "job" , EventID: spec.ID, Target: string(spec.Kind), Payload: string(payload)}, nil |
| 136 | } |