1package replication23import (4 "context"5 "encoding/json"6 "errors"7 "fmt"8 "net/url"9 "os"10 "path/filepath"11 "strings"12 "time"1314 "github.com/FelineStateMachine/tinyrelay/internal/event"15)1617var managementMethods = []string{"supportedmethods", "pullfrom", "pullstatus", "listjobs", "deliverystatus", "addjob", "removejob", "runjob", "backfill", "listdumps", "deletedump", "dumpnow", "backupnow", "listbackups", "deletebackup"}1819type JobStatus struct {20 Spec JobSpec `json:"spec"`21 Phase string `json:"phase"`22 Error string `json:"error,omitempty"`23 Stored, Sent, Skipped, Refused, Cursor, Rounds, Started, Finished int6424}2526func (s *Service) Methods() []string { return append([]string(nil), managementMethods...) }2728// Execute accepts decoded arguments for compatibility with existing daemon callers.29func (s *Service) Execute(ctx context.Context, method string, values ...any) (any, error) {30 params := make([]json.RawMessage, len(values))31 for i, value := range values {32 raw, err := json.Marshal(value)33 if err != nil {34 return nil, err35 }36 params[i] = raw37 }38 return s.ExecuteRaw(ctx, method, params)39}4041// ExecuteRaw accepts raw JSON arguments, matching NIP-86 and HTTP adapters.42func (s *Service) ExecuteRaw(ctx context.Context, method string, params []json.RawMessage) (any, error) {43 switch method {44 case "supportedmethods":45 return s.Methods(), nil46 case "listjobs":47 return s.ListJobStatus(ctx)48 case "pullstatus":49 return s.pullStatuses(ctx)50 case "deliverystatus":51 return s.deliveryStatuses(ctx)52 case "addjob":53 if len(params) != 1 {54 return nil, errors.New("replication: addjob expects one object")55 }56 var spec JobSpec57 if err := json.Unmarshal(params[0], &spec); err != nil {58 return nil, err59 }60 if err := s.AddJob(ctx, spec); err != nil {61 return nil, err62 }63 return spec, nil64 case "removejob", "runjob":65 id, err := rawString(params)66 if err != nil {67 return nil, err68 }69 if method == "removejob" {70 return true, s.RemoveJob(ctx, id)71 }72 return true, s.RunJob(ctx, id)73 case "pullfrom":74 return s.enqueuePull(ctx, params)75 case "backfill":76 return s.enqueueBackfill(ctx, params)77 case "listdumps":78 return s.listArtifacts(".jsonl")79 case "listbackups":80 return s.listArtifacts(".backup.json")81 case "deletedump":82 return s.deleteArtifact(ctx, params, ".jsonl")83 case "deletebackup":84 return s.deleteArtifact(ctx, params, ".backup.json")85 case "dumpnow":86 return s.enqueueArtifact(ctx, JobDump, params, ".jsonl")87 case "backupnow":88 return s.enqueueArtifact(ctx, JobBackup, params, ".backup.json")89 default:90 return nil, fmt.Errorf("replication: unsupported management method %q", method)91 }92}9394func (s *Service) ListJobStatus(ctx context.Context) ([]JobStatus, error) {95 rows, err := s.store.DB().QueryContext(ctx, `SELECT payload,phase,error,stored,sent,skipped,refused,cursor,rounds,started_at,finished_at FROM replication_jobs WHERE state!='cancelled' ORDER BY created_at,id`)96 if err != nil {97 return nil, err98 }99 defer rows.Close()100 out := []JobStatus{}101 for rows.Next() {102 var raw string103 var j JobStatus104 if err := rows.Scan(&raw, &j.Phase, &j.Error, &j.Stored, &j.Sent, &j.Skipped, &j.Refused, &j.Cursor, &j.Rounds, &j.Started, &j.Finished); err != nil {105 return nil, err106 }107 if err := json.Unmarshal([]byte(raw), &j.Spec); err != nil {108 return nil, err109 }110 out = append(out, j)111 }112 return out, rows.Err()113}114func (s *Service) pullStatuses(ctx context.Context) ([]JobStatus, error) {115 all, err := s.ListJobStatus(ctx)116 if err != nil {117 return nil, err118 }119 out := all[:0]120 for _, j := range all {121 if j.Spec.Kind == JobPull {122 out = append(out, j)123 }124 }125 return out, nil126}127128type workStatus struct {129 ID string `json:"id"`130 Target string `json:"target"`131 State string `json:"state"`132 Attempts int `json:"attempts"`133 Error string `json:"error,omitempty"`134}135136func (s *Service) deliveryStatuses(ctx context.Context) ([]workStatus, error) {137 rows, err := s.queue.List(ctx, "pending,running")138 if err != nil {139 return nil, err140 }141 out := []workStatus{}142 for _, r := range rows {143 if r.Kind == "delivery" {144 out = append(out, workStatus{r.ID, r.Target, r.State, r.Attempts, r.LastError})145 }146 }147 return out, nil148}149func (s *Service) enqueuePull(ctx context.Context, p []json.RawMessage) (JobSpec, error) {150 if len(p) < 1 || len(p) > 2 {151 return JobSpec{}, errors.New("replication: pullfrom expects a relay URL and optional filter")152 }153 var target string154 if err := json.Unmarshal(p[0], &target); err != nil {155 return JobSpec{}, err156 }157 target, err := normalizePullURL(target)158 if err != nil {159 return JobSpec{}, err160 }161 spec := JobSpec{ID: fmt.Sprintf("pull-%d", time.Now().UnixNano()), Kind: JobPull, Relays: []string{target}}162 if len(p) == 2 && string(p[1]) != "null" {163 var filter map[string]json.RawMessage164 if err := json.Unmarshal(p[1], &filter); err != nil {165 return JobSpec{}, errors.New("replication: pullfrom filter must be an object")166 }167 raw, err := json.Marshal(filter)168 if err != nil {169 return JobSpec{}, err170 }171 spec.Filter = string(raw)172 }173 return spec, s.AddJob(ctx, spec)174}175func (s *Service) enqueueBackfill(ctx context.Context, p []json.RawMessage) (JobSpec, error) {176 if len(p) > 1 {177 return JobSpec{}, errors.New("replication: backfill expects an optional relay list or object")178 }179 var relays []string180 var err error181 if len(p) == 1 {182 if err = json.Unmarshal(p[0], &relays); err != nil {183 var plan struct {184 Relays []string `json:"relays"`185 }186 if objectErr := json.Unmarshal(p[0], &plan); objectErr != nil {187 return JobSpec{}, errors.New("replication: backfill expects a relay list or object")188 }189 relays = plan.Relays190 }191 }192 owner := ""193 if s.config.Owner != nil {194 owner = strings.TrimSpace(s.config.Owner())195 }196 if len(relays) == 0 && owner != "" && s.config.Directory != nil {197 if list, ok := s.config.Directory.(RelayListDiscovery); ok {198 relays = list.RelayList(owner)199 } else {200 relays = append([]string(nil), s.config.Directory.ReadRelays(owner)...)201 for _, relay := range s.config.Directory.WriteRelays(owner) {202 if !containsRelay(relays, relay) {203 relays = append(relays, relay)204 }205 }206 }207 }208 if len(relays) == 0 {209 return JobSpec{}, errors.New("replication: no relay list (kind 10002) is stored here; give relays to fetch from")210 }211 for i, relay := range relays {212 relays[i], err = normalizePullURL(relay)213 if err != nil {214 return JobSpec{}, err215 }216 }217 filter := ""218 if owner != "" {219 raw, marshalErr := json.Marshal(event.Filter{Authors: []string{owner}, Tags: map[string][]string{}})220 if marshalErr != nil {221 return JobSpec{}, fmt.Errorf("replication: encode backfill filter: %w", marshalErr)222 }223 filter = string(raw)224 }225 spec := JobSpec{ID: fmt.Sprintf("backfill-%d", time.Now().UnixNano()), Kind: JobPull, Relays: relays, Filter: filter}226 return spec, s.AddJob(ctx, spec)227}228229func containsRelay(relays []string, candidate string) bool {230 for _, relay := range relays {231 if relay == candidate {232 return true233 }234 }235 return false236}237238func normalizePullURL(raw string) (string, error) {239 value := strings.TrimRight(strings.TrimSpace(raw), "/")240 u, err := url.Parse(value)241 if err != nil || u.Host == "" || u.User != nil || u.Fragment != "" {242 return "", errors.New("replication: invalid relay URL")243 }244 switch u.Scheme {245 case "http":246 u.Scheme = "ws"247 case "https":248 u.Scheme = "wss"249 case "ws", "wss":250 default:251 return "", errors.New("replication: invalid relay URL")252 }253 value = strings.TrimRight(u.String(), "/")254 if !safeRelayURL(value) {255 return "", errors.New("replication: invalid relay URL")256 }257 return value, nil258}259func (s *Service) listArtifacts(suffix string) ([]string, error) {260 entries, err := os.ReadDir(s.config.DataDir)261 if err != nil {262 return nil, err263 }264 out := []string{}265 for _, e := range entries {266 if !e.IsDir() && strings.HasSuffix(e.Name(), suffix) {267 out = append(out, e.Name())268 }269 }270 return out, nil271}272func (s *Service) deleteArtifact(ctx context.Context, p []json.RawMessage, suffix string) (bool, error) {273 name, err := rawString(p)274 if err != nil {275 return false, err276 }277 if !strings.HasSuffix(name, suffix) {278 return false, errors.New("replication: invalid artifact name")279 }280 path, err := safeArtifactPath(s.config.DataDir, name)281 if err != nil {282 return false, err283 }284 if err := os.Remove(path); err != nil {285 return false, err286 }287 _, err = s.store.DB().ExecContext(ctx, `INSERT INTO audit(at,actor,action,detail) VALUES(?,?,?,?)`, time.Now().Unix(), "owner", "delete_artifact", name)288 return err == nil, err289}290func (s *Service) enqueueArtifact(ctx context.Context, kind JobKind, p []json.RawMessage, suffix string) (JobSpec, error) {291 name := fmt.Sprintf("%d%s", time.Now().Unix(), suffix)292 if len(p) == 1 {293 var given string294 if err := json.Unmarshal(p[0], &given); err != nil {295 return JobSpec{}, err296 }297 if filepath.Base(given) != given {298 return JobSpec{}, errors.New("replication: invalid artifact name")299 }300 name = given301 }302 spec := JobSpec{ID: fmt.Sprintf("%s-%d", kind, time.Now().UnixNano()), Kind: kind, Relays: []string{name}}303 return spec, s.AddJob(ctx, spec)304}305func rawString(p []json.RawMessage) (string, error) {306 if len(p) != 1 {307 return "", errors.New("replication: expected one string")308 }309 var value string310 if err := json.Unmarshal(p[0], &value); err != nil || strings.TrimSpace(value) == "" {311 return "", errors.New("replication: expected nonempty string")312 }313 return value, nil314}