orbit/internal/storage/mysql/schedule_history.go

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<? OR (created_at=? AND id<?))`
			args = append(args, c.At, c.At, c.ID)
		}
		q += ` ORDER BY created_at DESC,id DESC LIMIT ?`
		args = append(args, in.Limit+1)
		rows, err := tx.QueryContext(ctx, q, args...)
		if err != nil {
			return err
		}
		defer rows.Close()
		for rows.Next() {
			v, err := scanSchedule(rows)
			if err != nil {
				return err
			}
			out.Items = append(out.Items, v)
		}
		if err := rows.Err(); err != nil {
			return err
		}
		if len(out.Items) > 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<? OR (scheduled_at=? AND id<?))`
			args = append(args, c.At, c.At, c.ID)
		}
		q += ` ORDER BY scheduled_at DESC,id DESC LIMIT ?`
		args = append(args, in.Limit+1)
		rows, err := tx.QueryContext(ctx, q, args...)
		if err != nil {
			return err
		}
		defer rows.Close()
		for rows.Next() {
			var v schedule.OccurrenceView
			if err := rows.Scan(&v.ID, &v.ScheduleID, &v.ScheduleVersion, &v.ScheduledAt, &v.Decision, &v.TaskID, &v.CoveredUntil, &v.Reason, &v.CreatedAt); err != nil {
				return err
			}
			out.Items = append(out.Items, v)
		}
		if err := rows.Err(); err != nil {
			return err
		}
		if len(out.Items) > 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
}