| 1 | package replication |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "net/url" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "strings" |
| 12 | "time" |
| 13 | |
| 14 | "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 15 | ) |
| 16 | |
| 17 | var managementMethods = []string{"supportedmethods" , "pullfrom" , "pullstatus" , "listjobs" , "deliverystatus" , "addjob" , "removejob" , "runjob" , "backfill" , "listdumps" , "deletedump" , "dumpnow" , "backupnow" , "listbackups" , "deletebackup" } |
| 18 | |
| 19 | type 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 int64 |
| 24 | } |
| 25 | |
| 26 | func (s *Service) Methods() []string { return append([]string(nil), managementMethods...) } |
| 27 | |
| 28 | // Execute accepts decoded arguments for compatibility with existing daemon callers. |
| 29 | func (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, err |
| 35 | } |
| 36 | params[i] = raw |
| 37 | } |
| 38 | return s.ExecuteRaw(ctx, method, params) |
| 39 | } |
| 40 | |
| 41 | // ExecuteRaw accepts raw JSON arguments, matching NIP-86 and HTTP adapters. |
| 42 | func (s *Service) ExecuteRaw(ctx context.Context, method string, params []json.RawMessage) (any, error) { |
| 43 | switch method { |
| 44 | case "supportedmethods" : |
| 45 | return s.Methods(), nil |
| 46 | 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 JobSpec |
| 57 | if err := json.Unmarshal(params[0], &spec); err != nil { |
| 58 | return nil, err |
| 59 | } |
| 60 | if err := s.AddJob(ctx, spec); err != nil { |
| 61 | return nil, err |
| 62 | } |
| 63 | return spec, nil |
| 64 | case "removejob" , "runjob" : |
| 65 | id, err := rawString(params) |
| 66 | if err != nil { |
| 67 | return nil, err |
| 68 | } |
| 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 | } |
| 93 | |
| 94 | func (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, err |
| 98 | } |
| 99 | defer rows.Close() |
| 100 | out := []JobStatus{} |
| 101 | for rows.Next() { |
| 102 | var raw string |
| 103 | var j JobStatus |
| 104 | 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, err |
| 106 | } |
| 107 | if err := json.Unmarshal([]byte(raw), &j.Spec); err != nil { |
| 108 | return nil, err |
| 109 | } |
| 110 | out = append(out, j) |
| 111 | } |
| 112 | return out, rows.Err() |
| 113 | } |
| 114 | func (s *Service) pullStatuses(ctx context.Context) ([]JobStatus, error) { |
| 115 | all, err := s.ListJobStatus(ctx) |
| 116 | if err != nil { |
| 117 | return nil, err |
| 118 | } |
| 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, nil |
| 126 | } |
| 127 | |
| 128 | type 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 | } |
| 135 | |
| 136 | func (s *Service) deliveryStatuses(ctx context.Context) ([]workStatus, error) { |
| 137 | rows, err := s.queue.List(ctx, "pending,running" ) |
| 138 | if err != nil { |
| 139 | return nil, err |
| 140 | } |
| 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, nil |
| 148 | } |
| 149 | func (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 string |
| 154 | if err := json.Unmarshal(p[0], &target); err != nil { |
| 155 | return JobSpec{}, err |
| 156 | } |
| 157 | target, err := normalizePullURL(target) |
| 158 | if err != nil { |
| 159 | return JobSpec{}, err |
| 160 | } |
| 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.RawMessage |
| 164 | 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{}, err |
| 170 | } |
| 171 | spec.Filter = string(raw) |
| 172 | } |
| 173 | return spec, s.AddJob(ctx, spec) |
| 174 | } |
| 175 | func (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 []string |
| 180 | var err error |
| 181 | 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.Relays |
| 190 | } |
| 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{}, err |
| 215 | } |
| 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 | } |
| 228 | |
| 229 | func containsRelay(relays []string, candidate string) bool { |
| 230 | for _, relay := range relays { |
| 231 | if relay == candidate { |
| 232 | return true |
| 233 | } |
| 234 | } |
| 235 | return false |
| 236 | } |
| 237 | |
| 238 | func 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, nil |
| 258 | } |
| 259 | func (s *Service) listArtifacts(suffix string) ([]string, error) { |
| 260 | entries, err := os.ReadDir(s.config.DataDir) |
| 261 | if err != nil { |
| 262 | return nil, err |
| 263 | } |
| 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, nil |
| 271 | } |
| 272 | func (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, err |
| 276 | } |
| 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, err |
| 283 | } |
| 284 | if err := os.Remove(path); err != nil { |
| 285 | return false, err |
| 286 | } |
| 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, err |
| 289 | } |
| 290 | func (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 string |
| 294 | if err := json.Unmarshal(p[0], &given); err != nil { |
| 295 | return JobSpec{}, err |
| 296 | } |
| 297 | if filepath.Base(given) != given { |
| 298 | return JobSpec{}, errors.New("replication: invalid artifact name" ) |
| 299 | } |
| 300 | name = given |
| 301 | } |
| 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 | } |
| 305 | func rawString(p []json.RawMessage) (string, error) { |
| 306 | if len(p) != 1 { |
| 307 | return "" , errors.New("replication: expected one string" ) |
| 308 | } |
| 309 | var value string |
| 310 | if err := json.Unmarshal(p[0], &value); err != nil || strings.TrimSpace(value) == "" { |
| 311 | return "" , errors.New("replication: expected nonempty string" ) |
| 312 | } |
| 313 | return value, nil |
| 314 | } |