| 1 | package daemon |
| 2 | |
| 3 | // Long task browsing: the NIP-90 job requests the caller may see, each with |
| 4 | // its newest feedback status and its result when one exists, and one |
| 5 | // request with its feedback timeline and results. Both read through the |
| 6 | // tenant gate with the caller's session, like the other browse methods. |
| 7 | |
| 8 | import ( |
| 9 | "context" |
| 10 | "encoding/json" |
| 11 | "errors" |
| 12 | "fmt" |
| 13 | "strconv" |
| 14 | "time" |
| 15 | |
| 16 | "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 17 | "github.com/FelineStateMachine/tinyrelay/internal/relay" |
| 18 | "github.com/FelineStateMachine/tinyrelay/internal/storage" |
| 19 | ) |
| 20 | |
| 21 | // jobScan bounds how many stored requests or answers one listing walks. |
| 22 | const jobScan = 1000 |
| 23 | |
| 24 | var jobStates = []string{"open" , "done" , "all" } |
| 25 | |
| 26 | // jobRequestKinds lists the kinds classified as long-task requests. |
| 27 | var jobRequestKinds = func() []int { |
| 28 | kinds := make([]int, 0, event.KIND_JOB_REQUEST_MAX-event.KIND_JOB_REQUEST_MIN+1) |
| 29 | for kind := event.KIND_JOB_REQUEST_MIN; kind <= event.KIND_JOB_REQUEST_MAX; kind++ { |
| 30 | if event.IsJobRequest(kind) { |
| 31 | kinds = append(kinds, kind) |
| 32 | } |
| 33 | } |
| 34 | return kinds |
| 35 | }() |
| 36 | |
| 37 | type jobBrowseRequest struct { |
| 38 | Limit int `json:"limit"` |
| 39 | Cursor string `json:"cursor"` |
| 40 | State string `json:"state"` |
| 41 | Mine bool `json:"mine"` |
| 42 | ID string `json:"id"` |
| 43 | Event string `json:"event"` |
| 44 | } |
| 45 | |
| 46 | // jobInput is one i tag of a request. |
| 47 | type jobInput struct { |
| 48 | Data string `json:"data"` |
| 49 | Type string `json:"type"` |
| 50 | Relay string `json:"relay,omitempty"` |
| 51 | Marker string `json:"marker,omitempty"` |
| 52 | } |
| 53 | |
| 54 | // jobParam is one param tag of a request. |
| 55 | type jobParam struct { |
| 56 | Key string `json:"key"` |
| 57 | Value string `json:"value"` |
| 58 | } |
| 59 | |
| 60 | // jobItem is one request as a browser sees it, settled against its answers. |
| 61 | type jobItem struct { |
| 62 | ID string `json:"id"` |
| 63 | Kind int `json:"kind"` |
| 64 | Requester string `json:"requester"` |
| 65 | Providers []string `json:"providers"` |
| 66 | Inputs []jobInput `json:"inputs"` |
| 67 | Output string `json:"output,omitempty"` |
| 68 | Params []jobParam `json:"params"` |
| 69 | Bid string `json:"bid,omitempty"` |
| 70 | Relays []string `json:"relays"` |
| 71 | Encrypted bool `json:"encrypted,omitempty"` |
| 72 | Content string `json:"content"` |
| 73 | CreatedAt int64 `json:"created_at"` |
| 74 | Expires int64 `json:"expires,omitempty"` |
| 75 | State string `json:"state"` |
| 76 | Status string `json:"status,omitempty"` |
| 77 | Feedback *jobAnswer `json:"feedback,omitempty"` |
| 78 | Result *jobAnswer `json:"result,omitempty"` |
| 79 | Event event.Event `json:"event"` |
| 80 | } |
| 81 | |
| 82 | // jobAnswer is one feedback or result event from a service provider. |
| 83 | type jobAnswer struct { |
| 84 | ID string `json:"id"` |
| 85 | Kind int `json:"kind"` |
| 86 | Provider string `json:"provider"` |
| 87 | Status string `json:"status,omitempty"` |
| 88 | Info string `json:"info,omitempty"` |
| 89 | Amount string `json:"amount,omitempty"` |
| 90 | Invoice string `json:"invoice,omitempty"` |
| 91 | Content string `json:"content"` |
| 92 | CreatedAt int64 `json:"created_at"` |
| 93 | } |
| 94 | |
| 95 | func jobBrowseMethod(method string) bool { |
| 96 | return method == "browsejobs" || method == "browsejob" |
| 97 | } |
| 98 | |
| 99 | func (t *Tenant) executeJobs(ctx context.Context, actor, method string, params []json.RawMessage) (any, error) { |
| 100 | if !t.Policy().Features.Jobs { |
| 101 | return nil, errors.New("restricted: long tasks are switched off on this relay" ) |
| 102 | } |
| 103 | q := jobBrowseRequest{} |
| 104 | if len(params) > 0 { |
| 105 | if err := json.Unmarshal(params[0], &q); err != nil { |
| 106 | return nil, fmt.Errorf("invalid: job parameters: %w" , err) |
| 107 | } |
| 108 | } |
| 109 | if q.Limit <= 0 || q.Limit > 100 { |
| 110 | q.Limit = 50 |
| 111 | } |
| 112 | switch method { |
| 113 | case "browsejobs" : |
| 114 | return t.browseJobs(ctx, actor, q) |
| 115 | case "browsejob" : |
| 116 | if q.ID == "" { |
| 117 | q.ID = q.Event |
| 118 | } |
| 119 | return t.browseJob(ctx, actor, q.ID) |
| 120 | } |
| 121 | return nil, errors.New("unsupported: browse operation" ) |
| 122 | } |
| 123 | |
| 124 | // browseJobs lists the requests the caller may see, newest first. A request |
| 125 | // is done once a result exists or its newest feedback reports success or an |
| 126 | // error; every other request is open. Expired requests are not served. |
| 127 | func (t *Tenant) browseJobs(ctx context.Context, actor string, q jobBrowseRequest) (any, error) { |
| 128 | if q.State == "" { |
| 129 | q.State = "all" |
| 130 | } |
| 131 | if !containsString(jobStates, q.State) { |
| 132 | return nil, errors.New("invalid: job state must be open, done or all" ) |
| 133 | } |
| 134 | if q.Mine && actor == "" { |
| 135 | return nil, errors.New("auth-required: sign in to list your own requests" ) |
| 136 | } |
| 137 | cursor, err := parseCollaborationCursor(q.Cursor) |
| 138 | if err != nil { |
| 139 | return nil, errors.New("invalid: job cursor" ) |
| 140 | } |
| 141 | session := browseSession(t, actor) |
| 142 | now := time.Now().Unix() |
| 143 | filter := event.Filter{Kinds: jobRequestKinds, Tags: map[string][]string{}} |
| 144 | if q.Mine { |
| 145 | filter.Authors = []string{actor} |
| 146 | } |
| 147 | items := make([]jobItem, 0, q.Limit) |
| 148 | next := "" |
| 149 | for scanned := 0; scanned < jobScan && next == "" ; scanned += 100 { |
| 150 | page, err := t.store.Query(ctx, filter, storage.QueryOptions{Now: now, Access: storage.Access{PubKeys: session.PubKeys}, Limit: 100, Before: cursor}) |
| 151 | if err != nil { |
| 152 | return nil, err |
| 153 | } |
| 154 | visible := make([]jobItem, 0, len(page.Events)) |
| 155 | for _, row := range page.Events { |
| 156 | cursor = &storage.EventCursor{CreatedAt: row.CreatedAt, ID: row.ID} |
| 157 | if t.gate.CanSee(ctx, row, session, nil) { |
| 158 | visible = append(visible, jobItemFrom(row)) |
| 159 | } |
| 160 | } |
| 161 | answers, err := t.jobAnswers(ctx, visible, session, now) |
| 162 | if err != nil { |
| 163 | return nil, err |
| 164 | } |
| 165 | for i := range visible { |
| 166 | jobSettle(&visible[i], answers[visible[i].ID]) |
| 167 | if q.State != "all" && visible[i].State != q.State { |
| 168 | continue |
| 169 | } |
| 170 | if len(items) == q.Limit { |
| 171 | next = jobCursor(items[len(items)-1]) |
| 172 | break |
| 173 | } |
| 174 | items = append(items, visible[i]) |
| 175 | } |
| 176 | if !page.More || len(page.Events) == 0 { |
| 177 | break |
| 178 | } |
| 179 | if next == "" && len(items) == q.Limit { |
| 180 | next = jobCursor(items[len(items)-1]) |
| 181 | } |
| 182 | } |
| 183 | return map[string]any{"items" : items, "next_cursor" : next}, nil |
| 184 | } |
| 185 | |
| 186 | // browseJob returns one request with its feedback, oldest first, and its |
| 187 | // results, newest first. |
| 188 | func (t *Tenant) browseJob(ctx context.Context, actor, id string) (any, error) { |
| 189 | if len(id) != 64 { |
| 190 | return nil, errors.New("invalid: job id" ) |
| 191 | } |
| 192 | session := browseSession(t, actor) |
| 193 | rows, err := t.Query(ctx, []event.Filter{{IDs: []string{id}, Kinds: jobRequestKinds, Limit: intPtr(1)}}, session) |
| 194 | if err != nil { |
| 195 | return nil, err |
| 196 | } |
| 197 | if len(rows) != 1 { |
| 198 | return nil, errors.New("not found: job request" ) |
| 199 | } |
| 200 | item := jobItemFrom(rows[0]) |
| 201 | answers, err := t.jobAnswers(ctx, []jobItem{item}, session, time.Now().Unix()) |
| 202 | if err != nil { |
| 203 | return nil, err |
| 204 | } |
| 205 | jobSettle(&item, answers[item.ID]) |
| 206 | feedback, results := []jobAnswer{}, []jobAnswer{} |
| 207 | for _, row := range answers[item.ID] { |
| 208 | if row.Kind == event.KIND_JOB_FEEDBACK { |
| 209 | feedback = append([]jobAnswer{jobAnswerFrom(row)}, feedback...) |
| 210 | } else { |
| 211 | results = append(results, jobAnswerFrom(row)) |
| 212 | } |
| 213 | } |
| 214 | return map[string]any{"item" : item, "feedback" : feedback, "results" : results}, nil |
| 215 | } |
| 216 | |
| 217 | func jobItemFrom(e event.Event) jobItem { |
| 218 | item := jobItem{ID: e.ID, Kind: e.Kind, Requester: e.PubKey, Providers: event.TagValues(e, "p" ), Inputs: []jobInput{}, Params: []jobParam{}, Relays: []string{}, Output: event.Tag(e, "output" ), Bid: event.Tag(e, "bid" ), Content: e.Content, CreatedAt: e.CreatedAt, Expires: event.Expiration(e), State: "open" , Event: e} |
| 219 | for _, tag := range e.Tags { |
| 220 | switch { |
| 221 | case len(tag) >= 3 && tag[0] == "i" : |
| 222 | input := jobInput{Data: tag[1], Type: tag[2]} |
| 223 | if len(tag) > 3 { |
| 224 | input.Relay = tag[3] |
| 225 | } |
| 226 | if len(tag) > 4 { |
| 227 | input.Marker = tag[4] |
| 228 | } |
| 229 | item.Inputs = append(item.Inputs, input) |
| 230 | case len(tag) >= 3 && tag[0] == "param" : |
| 231 | item.Params = append(item.Params, jobParam{Key: tag[1], Value: tag[2]}) |
| 232 | case len(tag) >= 2 && tag[0] == "relays" : |
| 233 | item.Relays = append(item.Relays, tag[1:]...) |
| 234 | case len(tag) >= 1 && tag[0] == "encrypted" : |
| 235 | item.Encrypted = true |
| 236 | } |
| 237 | } |
| 238 | return item |
| 239 | } |
| 240 | |
| 241 | func jobAnswerFrom(row event.Event) jobAnswer { |
| 242 | answer := jobAnswer{ID: row.ID, Kind: row.Kind, Provider: row.PubKey, Content: row.Content, CreatedAt: row.CreatedAt} |
| 243 | for _, tag := range row.Tags { |
| 244 | switch { |
| 245 | case len(tag) >= 2 && tag[0] == "status" && answer.Status == "" : |
| 246 | answer.Status = tag[1] |
| 247 | if len(tag) > 2 { |
| 248 | answer.Info = tag[2] |
| 249 | } |
| 250 | case len(tag) >= 2 && tag[0] == "amount" && answer.Amount == "" : |
| 251 | answer.Amount = tag[1] |
| 252 | if len(tag) > 2 { |
| 253 | answer.Invoice = tag[2] |
| 254 | } |
| 255 | } |
| 256 | } |
| 257 | return answer |
| 258 | } |
| 259 | |
| 260 | // jobAnswers finds the feedback and results naming the given requests in |
| 261 | // one query, newest first, keeping only what the session may see. |
| 262 | func (t *Tenant) jobAnswers(ctx context.Context, items []jobItem, session relay.Session, now int64) (map[string][]event.Event, error) { |
| 263 | answers := make(map[string][]event.Event, len(items)) |
| 264 | if len(items) == 0 { |
| 265 | return answers, nil |
| 266 | } |
| 267 | ids := make([]string, 0, len(items)) |
| 268 | kinds := []int{event.KIND_JOB_FEEDBACK} |
| 269 | wanted := make(map[string]int, len(items)) |
| 270 | for _, item := range items { |
| 271 | ids = append(ids, item.ID) |
| 272 | wanted[item.ID] = event.JobResultKind(item.Kind) |
| 273 | if !containsInt(kinds, wanted[item.ID]) { |
| 274 | kinds = append(kinds, wanted[item.ID]) |
| 275 | } |
| 276 | } |
| 277 | rows, err := t.store.Query(ctx, event.Filter{Kinds: kinds, Tags: map[string][]string{"e" : ids}}, storage.QueryOptions{Now: now, Access: storage.Access{PubKeys: session.PubKeys}, Limit: jobScan}) |
| 278 | if err != nil { |
| 279 | return nil, err |
| 280 | } |
| 281 | sortEventsNewestFirst(rows.Events) |
| 282 | for _, row := range rows.Events { |
| 283 | if !t.gate.CanSee(ctx, row, session, nil) { |
| 284 | continue |
| 285 | } |
| 286 | for _, id := range event.TagValues(row, "e" ) { |
| 287 | resultKind, ok := wanted[id] |
| 288 | if ok && (row.Kind == event.KIND_JOB_FEEDBACK || row.Kind == resultKind) { |
| 289 | answers[id] = append(answers[id], row) |
| 290 | } |
| 291 | } |
| 292 | } |
| 293 | return answers, nil |
| 294 | } |
| 295 | |
| 296 | // jobSettle fills the newest feedback, the newest result and the state of |
| 297 | // one request from its answers, which arrive newest first. |
| 298 | func jobSettle(item *jobItem, rows []event.Event) { |
| 299 | for _, row := range rows { |
| 300 | answer := jobAnswerFrom(row) |
| 301 | if row.Kind == event.KIND_JOB_FEEDBACK { |
| 302 | if item.Feedback == nil { |
| 303 | item.Feedback = &answer |
| 304 | item.Status = answer.Status |
| 305 | } |
| 306 | } else if item.Result == nil { |
| 307 | item.Result = &answer |
| 308 | } |
| 309 | } |
| 310 | item.State = "open" |
| 311 | if item.Result != nil || item.Status == "success" || item.Status == "error" { |
| 312 | item.State = "done" |
| 313 | } |
| 314 | } |
| 315 | |
| 316 | func jobCursor(item jobItem) string { |
| 317 | return strconv.FormatInt(item.CreatedAt, 10) + ":" + item.ID |
| 318 | } |