81 lines
3.0 KiB
Go
81 lines
3.0 KiB
Go
package mysql
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
|
|
"git.yuntuo.hk/k2safe/orbit/internal/fault"
|
|
core "git.yuntuo.hk/k2safe/orbit/model"
|
|
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
|
"git.yuntuo.hk/k2safe/orbit/pulse/service"
|
|
"git.yuntuo.hk/k2safe/orbit/weave/instance"
|
|
)
|
|
|
|
const workflowOperationColumns = `id,resource_id,operation_id,actor_id,action,created_at`
|
|
|
|
func scanWorkflowOperation(row scanner) (instance.OperationSummary, error) {
|
|
var v instance.OperationSummary
|
|
err := row.Scan(&v.ID, &v.WorkflowID, &v.OperationID, &v.ActorID, &v.Action, &v.CreatedAt)
|
|
return v, err
|
|
}
|
|
func (s *WorkflowInstanceStore) ListOperations(ctx context.Context, t auth.Token, id core.ID, in service.HistoryPageInput) (instance.OperationPage, error) {
|
|
out := instance.OperationPage{Items: []instance.OperationSummary{}}
|
|
err := s.transaction(ctx, t, instance.AuditAction, func(tx *sql.Tx, p auth.Permissions) error {
|
|
if _, _, _, err := readWorkflowForQuery(ctx, tx, t.NamespaceID, id, p); err != nil {
|
|
return err
|
|
}
|
|
query := `SELECT ` + workflowOperationColumns + ` FROM orbit_audit_events WHERE namespace_id=? AND resource_kind='workflow' AND resource_id=?`
|
|
args := []any{t.NamespaceID, id}
|
|
if in.Cursor != "" {
|
|
c, err := decodeHistoryCursorKind(in.Cursor, t.NamespaceID, id, "", "workflow-operations")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
query += ` AND (created_at<? OR (created_at=? AND id<?))`
|
|
args = append(args, *c.Created, *c.Created, c.ID)
|
|
}
|
|
query += ` ORDER BY created_at DESC,id DESC LIMIT ?`
|
|
args = append(args, in.Limit+1)
|
|
rows, err := tx.QueryContext(ctx, query, args...)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
v, err := scanWorkflowOperation(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 = encodeHistoryCursor(historyCursor{Version: 1, Kind: "workflow-operations", Namespace: t.NamespaceID, Task: id, ID: last.ID, Created: &last.CreatedAt})
|
|
}
|
|
return err
|
|
})
|
|
return out, err
|
|
}
|
|
|
|
func (s *WorkflowInstanceStore) GetOperation(ctx context.Context, t auth.Token, id, audit core.ID) (instance.OperationView, error) {
|
|
var out instance.OperationView
|
|
err := s.transaction(ctx, t, instance.AuditAction, func(tx *sql.Tx, p auth.Permissions) error {
|
|
if _, _, _, err := readWorkflowForQuery(ctx, tx, t.NamespaceID, id, p); err != nil {
|
|
return err
|
|
}
|
|
var before, after []byte
|
|
err := tx.QueryRowContext(ctx, `SELECT `+workflowOperationColumns+",reason,COALESCE(evidence_ref,''),`before`,`after` FROM orbit_audit_events WHERE namespace_id=? AND resource_kind='workflow' AND resource_id=? AND id=?", t.NamespaceID, id, audit).Scan(&out.ID, &out.WorkflowID, &out.OperationID, &out.ActorID, &out.Action, &out.CreatedAt, &out.Reason, &out.EvidenceRef, &before, &after)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return fault.NotFound
|
|
}
|
|
out.Before, out.After = before, after
|
|
return err
|
|
})
|
|
return out, err
|
|
}
|