1package daemon23// Long task browsing: the NIP-90 job requests the caller may see, each with4// its newest feedback status and its result when one exists, and one5// request with its feedback timeline and results. Both read through the6// tenant gate with the caller's session, like the other browse methods.78import (9 "context"10 "encoding/json"11 "errors"12 "fmt"13 "strconv"14 "time"1516 "github.com/FelineStateMachine/tinyrelay/internal/event"17 "github.com/FelineStateMachine/tinyrelay/internal/relay"18 "github.com/FelineStateMachine/tinyrelay/internal/storage"19)2021// jobScan bounds how many stored requests or answers one listing walks.22const jobScan = 10002324var jobStates = []string{"open", "done", "all"}2526// jobRequestKinds lists the kinds classified as long-task requests.27var 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 kinds35}()3637type 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}4546// jobInput is one i tag of a request.47type 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}5354// jobParam is one param tag of a request.55type jobParam struct {56 Key string `json:"key"`57 Value string `json:"value"`58}5960// jobItem is one request as a browser sees it, settled against its answers.61type 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}8182// jobAnswer is one feedback or result event from a service provider.83type 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}9495func jobBrowseMethod(method string) bool {96 return method == "browsejobs" || method == "browsejob"97}9899func (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 = 50111 }112 switch method {113 case "browsejobs":114 return t.browseJobs(ctx, actor, q)115 case "browsejob":116 if q.ID == "" {117 q.ID = q.Event118 }119 return t.browseJob(ctx, actor, q.ID)120 }121 return nil, errors.New("unsupported: browse operation")122}123124// browseJobs lists the requests the caller may see, newest first. A request125// is done once a result exists or its newest feedback reports success or an126// error; every other request is open. Expired requests are not served.127func (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, err153 }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, err164 }165 for i := range visible {166 jobSettle(&visible[i], answers[visible[i].ID])167 if q.State != "all" && visible[i].State != q.State {168 continue169 }170 if len(items) == q.Limit {171 next = jobCursor(items[len(items)-1])172 break173 }174 items = append(items, visible[i])175 }176 if !page.More || len(page.Events) == 0 {177 break178 }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}, nil184}185186// browseJob returns one request with its feedback, oldest first, and its187// results, newest first.188func (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, err196 }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, err204 }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}, nil215}216217func 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 = true236 }237 }238 return item239}240241func 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 answer258}259260// jobAnswers finds the feedback and results naming the given requests in261// one query, newest first, keeping only what the session may see.262func (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, nil266 }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, err280 }281 sortEventsNewestFirst(rows.Events)282 for _, row := range rows.Events {283 if !t.gate.CanSee(ctx, row, session, nil) {284 continue285 }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, nil294}295296// jobSettle fills the newest feedback, the newest result and the state of297// one request from its answers, which arrive newest first.298func 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 = &answer304 item.Status = answer.Status305 }306 } else if item.Result == nil {307 item.Result = &answer308 }309 }310 item.State = "open"311 if item.Result != nil || item.Status == "success" || item.Status == "error" {312 item.State = "done"313 }314}315316func jobCursor(item jobItem) string {317 return strconv.FormatInt(item.CreatedAt, 10) + ":" + item.ID318}