1package daemon23// Long task browsing: the job requests the caller may see, each settled4// against its answers, and one request with its answer timeline. Both read5// through the tenant gate with the caller's session, like the other browse6// methods.78import (9 "context"10 "encoding/json"11 "errors"12 "fmt"13 "strconv"14 "strings"15 "time"1617 "github.com/FelineStateMachine/tinyrelay/internal/event"18 "github.com/FelineStateMachine/tinyrelay/internal/gates"19 "github.com/FelineStateMachine/tinyrelay/internal/relay"20 "github.com/FelineStateMachine/tinyrelay/internal/storage"21)2223// jobScan bounds how many stored requests or answers one listing walks.24const jobScan = 10002526// jobStates lists the state filters a listing accepts: one of the four27// settled states, or all.28var jobStates = []string{"open", "done", "failed", "cancelled", "all"}2930// jobRequestKinds lists the kinds classified as long-task requests.31var jobRequestKinds = []int{event.KIND_JOB_REQUEST}3233// jobAnswerKinds lists every kind that answers or ends a request.34var jobAnswerKinds = []int{event.KIND_JOB_ACCEPTED, event.KIND_JOB_PROGRESS, event.KIND_JOB_RESULT, event.KIND_JOB_CANCEL, event.KIND_JOB_ERROR}3536type jobBrowseRequest struct {37 Limit int `json:"limit"`38 Cursor string `json:"cursor"`39 State string `json:"state"`40 Mine bool `json:"mine"`41 ID string `json:"id"`42 Event string `json:"event"`43}4445// jobItem is one request as a browser sees it, settled against its answers.46type jobItem struct {47 ID string `json:"id"`48 Room string `json:"room"`49 Requester string `json:"requester"`50 Assignees []string `json:"assignees"`51 Subject string `json:"subject,omitempty"`52 Content string `json:"content"`53 CreatedAt int64 `json:"created_at"`54 Expires int64 `json:"expires,omitempty"`55 Root string `json:"root,omitempty"`56 State string `json:"state"`57 Status string `json:"status"`58 Progress *jobAnswer `json:"progress,omitempty"`59 Result *jobAnswer `json:"result,omitempty"`60 Error *jobAnswer `json:"error,omitempty"`61 CancelledAt int64 `json:"cancelled_at,omitempty"`62 Event event.Event `json:"event"`63}6465// jobAnswer is one accepted, progress, result, cancel or error event.66type jobAnswer struct {67 ID string `json:"id"`68 Kind int `json:"kind"`69 Provider string `json:"provider"`70 Content string `json:"content"`71 CreatedAt int64 `json:"created_at"`72 Artifacts []jobArtifact `json:"artifacts,omitempty"`73}7475// jobArtifact is one artifact a result references: an event (e), an76// addressable event (a) or a URL (r).77type jobArtifact struct {78 Type string `json:"type"`79 Value string `json:"value"`80}8182func jobBrowseMethod(method string) bool {83 return method == "browsejobs" || method == "browsejob"84}8586func (t *Tenant) executeJobs(ctx context.Context, actor, method string, params []json.RawMessage) (any, error) {87 if !t.Policy().Features.Jobs {88 return nil, errors.New("restricted: long tasks are switched off on this relay")89 }90 q := jobBrowseRequest{}91 if len(params) > 0 {92 if err := json.Unmarshal(params[0], &q); err != nil {93 return nil, fmt.Errorf("invalid: job parameters: %w", err)94 }95 }96 if q.Limit <= 0 || q.Limit > 100 {97 q.Limit = 5098 }99 switch method {100 case "browsejobs":101 return t.browseJobs(ctx, actor, q)102 case "browsejob":103 if q.ID == "" {104 q.ID = q.Event105 }106 return t.browseJob(ctx, actor, q.ID)107 }108 return nil, errors.New("unsupported: browse operation")109}110111// browseJobs lists the requests the caller may see, newest first, each112// settled against its answers. Expired requests are not served.113func (t *Tenant) browseJobs(ctx context.Context, actor string, q jobBrowseRequest) (any, error) {114 if q.State == "" {115 q.State = "all"116 }117 if !containsString(jobStates, q.State) {118 return nil, errors.New("invalid: job state must be open, done, failed, cancelled or all")119 }120 if q.Mine && actor == "" {121 return nil, errors.New("auth-required: sign in to list your own requests")122 }123 cursor, err := parseCollaborationCursor(q.Cursor)124 if err != nil {125 return nil, errors.New("invalid: job cursor")126 }127 session := browseSession(t, actor)128 now := time.Now().Unix()129 filter := event.Filter{Kinds: jobRequestKinds, Tags: map[string][]string{}}130 if q.Mine {131 filter.Authors = []string{actor}132 }133 items := make([]jobItem, 0, q.Limit)134 next := ""135 for scanned := 0; scanned < jobScan && next == ""; scanned += 100 {136 page, err := t.store.Query(ctx, filter, storage.QueryOptions{Now: now, Access: storage.Access{PubKeys: session.PubKeys}, Limit: 100, Before: cursor})137 if err != nil {138 return nil, err139 }140 visible := make([]jobItem, 0, len(page.Events))141 for _, row := range page.Events {142 cursor = &storage.EventCursor{CreatedAt: row.CreatedAt, ID: row.ID}143 if t.gate.CanSee(ctx, row, session, nil) {144 visible = append(visible, jobItemFrom(row))145 }146 }147 answers, err := t.jobAnswers(ctx, visible, session, now)148 if err != nil {149 return nil, err150 }151 for i := range visible {152 jobSettle(&visible[i], answers[visible[i].ID])153 if q.State != "all" && visible[i].State != q.State {154 continue155 }156 if len(items) == q.Limit {157 next = jobCursor(items[len(items)-1])158 break159 }160 items = append(items, visible[i])161 }162 if !page.More || len(page.Events) == 0 {163 break164 }165 if next == "" && len(items) == q.Limit {166 next = jobCursor(items[len(items)-1])167 }168 }169 return map[string]any{"items": items, "next_cursor": next}, nil170}171172// browseJob returns one request with its answers, oldest first.173func (t *Tenant) browseJob(ctx context.Context, actor, id string) (any, error) {174 if len(id) != 64 {175 return nil, errors.New("invalid: job id")176 }177 session := browseSession(t, actor)178 rows, err := t.Query(ctx, []event.Filter{{IDs: []string{id}, Kinds: jobRequestKinds, Limit: intPtr(1)}}, session)179 if err != nil {180 return nil, err181 }182 if len(rows) != 1 {183 return nil, errors.New("not found: job request")184 }185 item := jobItemFrom(rows[0])186 answers, err := t.jobAnswers(ctx, []jobItem{item}, session, time.Now().Unix())187 if err != nil {188 return nil, err189 }190 jobSettle(&item, answers[item.ID])191 timeline := make([]jobAnswer, 0, len(answers[item.ID]))192 for i := len(answers[item.ID]) - 1; i >= 0; i-- {193 timeline = append(timeline, jobAnswerFrom(answers[item.ID][i]))194 }195 return map[string]any{"item": item, "answers": timeline}, nil196}197198// jobRoot names the thread a request attaches to: its e tag marked root.199func jobRoot(e event.Event) string {200 for _, tag := range e.Tags {201 if len(tag) > 3 && tag[0] == "e" && tag[3] == "root" && len(tag[1]) == 64 {202 return tag[1]203 }204 }205 return ""206}207208func jobItemFrom(e event.Event) jobItem {209 assignees := event.TagValues(e, "p")210 if assignees == nil {211 assignees = []string{}212 }213 return jobItem{ID: e.ID, Room: event.Tag(e, "h"), Requester: e.PubKey, Assignees: assignees, Subject: strings.TrimSpace(event.Tag(e, "subject")), Content: e.Content, CreatedAt: e.CreatedAt, Expires: event.Expiration(e), Root: jobRoot(e), State: "open", Status: "queued", Event: e}214}215216func jobAnswerFrom(row event.Event) jobAnswer {217 answer := jobAnswer{ID: row.ID, Kind: row.Kind, Provider: row.PubKey, Content: row.Content, CreatedAt: row.CreatedAt}218 if row.Kind != event.KIND_JOB_RESULT {219 return answer220 }221 request := false222 for _, tag := range row.Tags {223 if len(tag) < 2 || tag[1] == "" {224 continue225 }226 switch tag[0] {227 case "e":228 // The first e tag names the request; later ones are artifacts.229 if !request {230 request = true231 continue232 }233 answer.Artifacts = append(answer.Artifacts, jobArtifact{Type: "e", Value: tag[1]})234 case "a", "r":235 answer.Artifacts = append(answer.Artifacts, jobArtifact{Type: tag[0], Value: tag[1]})236 }237 }238 return answer239}240241// jobAnswers finds the answers naming the given requests in one query,242// newest first, keeping only what the session may see.243func (t *Tenant) jobAnswers(ctx context.Context, items []jobItem, session relay.Session, now int64) (map[string][]event.Event, error) {244 answers := make(map[string][]event.Event, len(items))245 if len(items) == 0 {246 return answers, nil247 }248 ids := make([]string, 0, len(items))249 wanted := make(map[string]bool, len(items))250 for _, item := range items {251 ids = append(ids, item.ID)252 wanted[item.ID] = true253 }254 rows, err := t.store.Query(ctx, event.Filter{Kinds: jobAnswerKinds, Tags: map[string][]string{"e": ids}}, storage.QueryOptions{Now: now, Access: storage.Access{PubKeys: session.PubKeys}, Limit: jobScan})255 if err != nil {256 return nil, err257 }258 sortEventsNewestFirst(rows.Events)259 for _, row := range rows.Events {260 if !t.gate.CanSee(ctx, row, session, nil) {261 continue262 }263 if id := gates.JobRequestID(row); wanted[id] {264 answers[id] = append(answers[id], row)265 }266 }267 return answers, nil268}269270// jobSettle derives one request's state from its answers, which arrive271// newest first. The newest terminal event wins: a result makes the task272// done, a cancel cancelled and an error failed. Otherwise the newest273// accepted or progress answer sets the status, and a request with no274// answers is queued.275func jobSettle(item *jobItem, rows []event.Event) {276 item.State, item.Status = "open", "queued"277 item.Progress, item.Result, item.Error, item.CancelledAt = nil, nil, nil, 0278 settled := false279 for _, row := range rows {280 answer := jobAnswerFrom(row)281 switch row.Kind {282 case event.KIND_JOB_RESULT:283 if item.Result == nil {284 item.Result = &answer285 }286 if !settled {287 item.State, item.Status, settled = "done", "done", true288 }289 case event.KIND_JOB_ERROR:290 if item.Error == nil {291 item.Error = &answer292 }293 if !settled {294 item.State, item.Status, settled = "failed", "failed", true295 }296 case event.KIND_JOB_CANCEL:297 if item.CancelledAt == 0 {298 item.CancelledAt = row.CreatedAt299 }300 if !settled {301 item.State, item.Status, settled = "cancelled", "cancelled", true302 }303 case event.KIND_JOB_ACCEPTED, event.KIND_JOB_PROGRESS:304 if item.Progress == nil {305 item.Progress = &answer306 if !settled {307 item.Status = "accepted"308 if row.Kind == event.KIND_JOB_PROGRESS {309 item.Status = "running"310 }311 }312 }313 }314 }315}316317func jobCursor(item jobItem) string {318 return strconv.FormatInt(item.CreatedAt, 10) + ":" + item.ID319}