1package replication23import (4 "context"5 "encoding/json"6 "errors"7 "fmt"8 "strings"9 "time"1011 "github.com/FelineStateMachine/tinyrelay/internal/event"12 "github.com/FelineStateMachine/tinyrelay/internal/storage"13 "github.com/FelineStateMachine/tinyrelay/internal/work"14)1516// JobKind selects the durable replication operation.17type JobKind string1819const (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)2728// JobSpec describes one immediate or recurring replication job. Every is an29// interval in hours; zero runs once.30type 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}3839// JobRunner executes a validated job specification.40type JobRunner interface {41 Run(ctx context.Context, spec JobSpec) error42}4344// JobHandler decodes and validates durable job intents before delegating them.45type JobHandler struct {46 runner JobRunner47}4849func NewJobHandler(runner JobRunner) *JobHandler { return &JobHandler{runner: runner} }5051func (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 JobSpec56 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 err61 }62 return h.runner.Run(ctx, spec)63}6465func 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 nil93}9495func EnqueueJob(ctx context.Context, queue *work.Queue, spec JobSpec, at time.Time) (string, error) {96 if err := ValidateJob(spec); err != nil {97 return "", err98 }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 distinct105 // durable intent. The work table intentionally deduplicates identical106 // (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}109110func 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}116117// RescheduleJob makes a recurring job eligible again after a successful run.118// It is separate from completion so the completed intent remains an audit119// record and a fresh run receives an independent claim token.120func 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}126127func IntentForJob(spec JobSpec, at time.Time) (storage.Intent, error) {128 if err := ValidateJob(spec); err != nil {129 return storage.Intent{}, err130 }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)}, nil136}