package mysql import ( "bytes" "context" "database/sql" "encoding/base64" "encoding/json" "time" "git.yuntuo.hk/k2safe/orbit/internal/fault" "git.yuntuo.hk/k2safe/orbit/internal/jsonvalue" core "git.yuntuo.hk/k2safe/orbit/model" "git.yuntuo.hk/k2safe/orbit/platform/auth" "git.yuntuo.hk/k2safe/orbit/pulse/schedule" ) type scheduleCursor struct { Version int `json:"v"` Kind string `json:"k"` Namespace core.ID `json:"n"` Parent core.ID `json:"p"` At time.Time `json:"t"` ID core.ID `json:"i"` } func parseScheduleCursor(raw string, ns, parent core.ID, kind string) (scheduleCursor, error) { var c scheduleCursor b, err := base64.RawURLEncoding.Strict().DecodeString(raw) if err != nil || len(b) > 768 { return c, fault.Invalid("invalid schedule cursor") } if _, err := jsonvalue.Canonical(b); err != nil { return c, fault.Invalid("invalid schedule cursor") } var fields map[string]json.RawMessage if json.Unmarshal(b, &fields) != nil || len(fields) != 6 { return c, fault.Invalid("invalid schedule cursor") } for _, key := range []string{"v", "k", "n", "p", "t", "i"} { if _, ok := fields[key]; !ok { return c, fault.Invalid("invalid schedule cursor") } } d := json.NewDecoder(bytes.NewReader(b)) d.DisallowUnknownFields() if d.Decode(&c) != nil || c.Version != 1 || c.Namespace != ns || c.Parent != parent || c.Kind != kind || c.ID.Validate() != nil || c.At.Year() < 2000 || c.At.Year() > 9998 || c.At.Nanosecond()%1000 != 0 { return c, fault.Invalid("cursor does not match schedule query") } return c, nil } func makeScheduleCursor(ns, parent core.ID, kind string, at time.Time, id core.ID) (string, error) { b, err := json.Marshal(scheduleCursor{1, kind, ns, parent, at, id}) return base64.RawURLEncoding.EncodeToString(b), err } func (s *ScheduleStore) List(ctx context.Context, t auth.Token, queue core.ID, in schedule.PageInput) (schedule.Page, error) { out := schedule.Page{Items: []schedule.Summary{}} err := s.transaction(ctx, t, "schedules:read", func(tx *sql.Tx, p auth.Permissions) error { if !p.AllowsQueue(queue) { return fault.Forbidden } q := `SELECT ` + scheduleColumns + ` FROM pulse_schedules WHERE namespace_id=? AND queue_id=?` args := []any{t.NamespaceID, queue} if in.Cursor != "" { c, err := parseScheduleCursor(in.Cursor, t.NamespaceID, queue, "schedules") if err != nil { return err } q += ` AND (created_at in.Limit { out.Items = out.Items[:in.Limit] last := out.Items[len(out.Items)-1] out.NextCursor, err = makeScheduleCursor(t.NamespaceID, queue, "schedules", last.CreatedAt, last.ID) } return err }) return out, err } func (s *ScheduleStore) Occurrences(ctx context.Context, t auth.Token, id core.ID, in schedule.PageInput) (schedule.OccurrencePage, error) { out := schedule.OccurrencePage{Items: []schedule.OccurrenceView{}} err := s.transaction(ctx, t, "schedules:read", func(tx *sql.Tx, p auth.Permissions) error { if _, err := authorizeSchedule(ctx, tx, t.NamespaceID, id, p); err != nil { return err } q := `SELECT id,schedule_id,schedule_version,scheduled_at,decision,task_id,covered_until,COALESCE(reason,''),created_at FROM pulse_schedule_occurrences WHERE namespace_id=? AND schedule_id=?` args := []any{t.NamespaceID, id} if in.Cursor != "" { c, err := parseScheduleCursor(in.Cursor, t.NamespaceID, id, "occurrences") if err != nil { return err } q += ` AND (scheduled_at in.Limit { out.Items = out.Items[:in.Limit] last := out.Items[len(out.Items)-1] out.NextCursor, err = makeScheduleCursor(t.NamespaceID, id, "occurrences", last.ScheduledAt, last.ID) } return err }) return out, err }