404 lines
16 KiB
Go
404 lines
16 KiB
Go
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"database/sql"
|
||
"encoding/base64"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
|
||
"git.yuntuo.hk/k2safe/orbit/internal/fault"
|
||
"git.yuntuo.hk/k2safe/orbit/internal/retention"
|
||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
||
)
|
||
|
||
const archiveMaxBytes = 8 << 20
|
||
const archiveMaxRows = 10000
|
||
|
||
// 查询定义只来自固定目录；浏览器不能提交表名、列名或SQL条件。
|
||
type archiveSource struct {
|
||
kind, table, clock, predicate string
|
||
days uint32
|
||
}
|
||
|
||
func archiveSources(p retention.Policy) []archiveSource {
|
||
return []archiveSource{
|
||
{"tasks", "pulse_tasks", "completed_at", `state IN ('SUCCEEDED','FAILED','CANCELLED') AND effect_uncertain=FALSE AND lease_until IS NULL AND owner_kind IS NULL`, p.TaskHistoryDays},
|
||
{"workflows", "weave_instances", "completed_at", `state IN ('SUCCEEDED','FAILED','CANCELLED') AND NOT EXISTS (SELECT 1 FROM pulse_tasks t WHERE t.namespace_id=r.namespace_id AND t.gate_id=r.gate_id AND (t.state NOT IN ('SUCCEEDED','FAILED','CANCELLED') OR t.effect_uncertain=TRUE)) AND NOT EXISTS (SELECT 1 FROM weave_inbox i WHERE i.namespace_id=r.namespace_id AND i.workflow_id=r.id AND i.state<>'PROCESSED')`, p.WorkflowHistoryDays},
|
||
{"events", "orbit_events", "occurred_at", `NOT EXISTS (SELECT 1 FROM orbit_deliveries d WHERE d.namespace_id=r.namespace_id AND d.event_id=r.id AND d.state<>'ACKED') AND NOT EXISTS (SELECT 1 FROM weave_inbox i WHERE i.namespace_id=r.namespace_id AND i.event_id=r.id AND i.state<>'PROCESSED')`, p.EventLogDays},
|
||
{"audit", "orbit_audit_events", "created_at", "TRUE", p.AuditLogDays},
|
||
{"deliveries", "orbit_deliveries", "acknowledged_at", "state='ACKED'", p.DeliveryHistoryDays},
|
||
{"inbox", "weave_inbox", "processed_at", `state='PROCESSED' AND EXISTS (SELECT 1 FROM weave_instances w WHERE w.namespace_id=r.namespace_id AND w.id=r.workflow_id AND w.state IN ('SUCCEEDED','FAILED','CANCELLED'))`, p.InboxHistoryDays},
|
||
{"idempotency", "orbit_idempotency_keys", "created_at", "expires_at<UTC_TIMESTAMP(6)", p.IdempotencyDays},
|
||
}
|
||
}
|
||
|
||
type archiveCursor struct {
|
||
Namespace core.ID
|
||
Kind string
|
||
Revision uint64
|
||
Cutoff time.Time
|
||
Time time.Time
|
||
ID core.ID
|
||
}
|
||
type archiveRoot struct {
|
||
id core.ID
|
||
at time.Time
|
||
}
|
||
|
||
func archiveCandidates(ctx context.Context, tx *sql.Tx, ns core.ID, source archiveSource, cutoff time.Time, limit int, cursor *archiveCursor) ([]archiveRoot, error) {
|
||
q := `SELECT /*+ MAX_EXECUTION_TIME(1500) */ r.id,r.` + source.clock + ` FROM ` + source.table + ` r WHERE r.namespace_id=? AND r.` + source.clock + `<? AND (` + source.predicate + `)`
|
||
args := []any{ns, cutoff}
|
||
if cursor != nil {
|
||
q += ` AND (r.` + source.clock + `>? OR (r.` + source.clock + `=? AND r.id>?))`
|
||
args = append(args, cursor.Time, cursor.Time, cursor.ID)
|
||
}
|
||
q += ` ORDER BY r.` + source.clock + `,r.id LIMIT ?`
|
||
args = append(args, limit)
|
||
rows, err := tx.QueryContext(ctx, q, args...)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
var items []archiveRoot
|
||
for rows.Next() {
|
||
var r archiveRoot
|
||
if err := rows.Scan(&r.id, &r.at); err != nil {
|
||
return nil, err
|
||
}
|
||
items = append(items, r)
|
||
}
|
||
return items, rows.Err()
|
||
}
|
||
|
||
func (s *RetentionPolicyStore) PreviewRetention(ctx context.Context, t auth.Token) (retention.Preview, error) {
|
||
ctx, cancel := context.WithTimeout(ctx, 8*time.Second)
|
||
defer cancel()
|
||
out := retention.Preview{SampleLimit: 100, Items: []retention.CandidatePage{}}
|
||
err := WithTx(ctx, s.db, func(tx *sql.Tx) error {
|
||
if _, err := retentionActor(ctx, tx, t); err != nil {
|
||
return err
|
||
}
|
||
p, err := s.readPolicy(ctx, tx, t.NamespaceID, false)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
out.Items = nil // WithTx 发生提交前重试时丢弃前次样本。
|
||
out.Policy = p
|
||
now, err := dbNow(ctx, tx)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
for _, source := range archiveSources(p) {
|
||
cutoff := now.Add(-time.Duration(source.days) * 24 * time.Hour)
|
||
roots, err := archiveCandidates(ctx, tx, t.NamespaceID, source, cutoff, 101, nil)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
out.Items = append(out.Items, retention.CandidatePage{Kind: source.kind, Cutoff: cutoff, Count: min(len(roots), 100), More: len(roots) > 100})
|
||
}
|
||
return nil
|
||
})
|
||
return out, err
|
||
}
|
||
|
||
// archiveData 是独立历史快照，不是迁移/灾难恢复备份，也不是源数据可删除的证明。
|
||
type archiveData struct {
|
||
Policy retention.Policy `json:"policy"` // Policy 冻结的保留策略。
|
||
Kind string `json:"kind"` // Kind 当前批次根资源类别。
|
||
SnapshotAt time.Time `json:"snapshot_at"` // SnapshotAt 读取数据库时间。
|
||
Cutoff time.Time `json:"cutoff"` // Cutoff 当前批次截止时间。
|
||
SourceRetained bool `json:"source_retained"` // SourceRetained 表示源数据库全部记录仍然保留。
|
||
RootCount int `json:"root_count"` // RootCount 本批根记录数量。
|
||
NextCursor string `json:"next_cursor"` // NextCursor 后续批次位置，非提交水位。
|
||
Tables map[string][]map[string]any `json:"tables"` // Tables 固定表目录下的历史快照；二进制为十六进制，JSON及整数为文本以保留精度。
|
||
}
|
||
|
||
func (s *RetentionPolicyStore) ExportRetention(ctx context.Context, t auth.Token, in retention.ArchiveInput) (retention.ArchiveSnapshot, error) {
|
||
return s.exportRetention(ctx, t.NamespaceID, in, func(ctx context.Context, tx *sql.Tx) (string, error) {
|
||
actor, err := retentionActor(ctx, tx, t)
|
||
return "console:" + string(actor), err
|
||
})
|
||
}
|
||
|
||
// ExportRetentionForJob 仅供持有数据库管理权限的归档 CLI 使用，不注册为 HTTP 路由。
|
||
// 每批核验应用/命名空间启用状态和数据库策略版本；作业标识进入审计，不能冒充控制台用户。
|
||
func (s *RetentionPolicyStore) ExportRetentionForJob(ctx context.Context, namespace, job core.ID, in retention.ArchiveInput) (retention.ArchiveSnapshot, error) {
|
||
if namespace.Validate() != nil || job.Validate() != nil {
|
||
return retention.ArchiveSnapshot{}, fault.Invalid("invalid archive job identity")
|
||
}
|
||
return s.exportRetention(ctx, namespace, in, func(ctx context.Context, tx *sql.Tx) (string, error) {
|
||
var enabled bool
|
||
err := tx.QueryRowContext(ctx, `SELECT a.enabled AND n.enabled FROM orbit_namespaces n JOIN orbit_applications a ON a.id=n.application_id WHERE n.id=? FOR SHARE`, namespace).Scan(&enabled)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
if !enabled {
|
||
return "", fault.Forbidden
|
||
}
|
||
return "archive:" + string(job), nil
|
||
})
|
||
}
|
||
|
||
func (s *RetentionPolicyStore) exportRetention(ctx context.Context, namespace core.ID, in retention.ArchiveInput, authorize func(context.Context, *sql.Tx) (string, error)) (retention.ArchiveSnapshot, error) {
|
||
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||
defer cancel()
|
||
if in.Limit == 0 {
|
||
in.Limit = 20
|
||
}
|
||
if in.Limit < 1 || in.Limit > 100 {
|
||
return retention.ArchiveSnapshot{}, fault.Invalid("limit must be 1-100")
|
||
}
|
||
// 导出不自动重放事务；失败没有成功包，源数据始终保留。
|
||
tx, err := s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelRepeatableRead})
|
||
if err != nil {
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
defer tx.Rollback()
|
||
actor, err := authorize(ctx, tx)
|
||
if err != nil {
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
p, err := s.readPolicy(ctx, tx, namespace, true)
|
||
if err != nil {
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
if !p.Enabled || p.Revision != in.ExpectedRevision {
|
||
return retention.ArchiveSnapshot{}, fault.Conflict
|
||
}
|
||
var source archiveSource
|
||
for _, v := range archiveSources(p) {
|
||
if v.kind == in.Kind {
|
||
source = v
|
||
}
|
||
}
|
||
if source.kind == "" {
|
||
return retention.ArchiveSnapshot{}, fault.Invalid("unsupported archive kind")
|
||
}
|
||
now, err := dbNow(ctx, tx)
|
||
if err != nil {
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
cutoff := now.Add(-time.Duration(source.days) * 24 * time.Hour)
|
||
var cursor *archiveCursor
|
||
if in.Cursor != "" {
|
||
if len(in.Cursor) > 2048 {
|
||
return retention.ArchiveSnapshot{}, fault.Invalid("invalid archive cursor")
|
||
}
|
||
b, e := base64.RawURLEncoding.DecodeString(in.Cursor)
|
||
var c archiveCursor
|
||
if e != nil || json.Unmarshal(b, &c) != nil || c.Namespace != namespace || c.Kind != in.Kind || c.Revision != p.Revision || c.ID.Validate() != nil || c.Cutoff.After(cutoff) || c.Time.IsZero() || !c.Time.Before(c.Cutoff) {
|
||
return retention.ArchiveSnapshot{}, fault.Invalid("invalid archive cursor")
|
||
}
|
||
cursor = &c
|
||
cutoff = c.Cutoff
|
||
}
|
||
roots, err := archiveCandidates(ctx, tx, namespace, source, cutoff, in.Limit+1, cursor)
|
||
if err != nil {
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
data := archiveData{Policy: p, Kind: in.Kind, SnapshotAt: now, Cutoff: cutoff, SourceRetained: true, Tables: map[string][]map[string]any{}}
|
||
if len(roots) > in.Limit {
|
||
roots = roots[:in.Limit]
|
||
last := roots[len(roots)-1]
|
||
raw, _ := json.Marshal(archiveCursor{namespace, in.Kind, p.Revision, cutoff, last.at, last.id})
|
||
data.NextCursor = base64.RawURLEncoding.EncodeToString(raw)
|
||
}
|
||
data.RootCount = len(roots)
|
||
budget := archiveBudget{}
|
||
for _, root := range roots {
|
||
if err := archiveRows(ctx, tx, &data, &budget, source.table, "id=?", root.id); err != nil {
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
if source.kind == "tasks" {
|
||
if err := archiveTaskHistory(ctx, tx, &data, &budget, root.id); err != nil {
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
}
|
||
if source.kind == "events" {
|
||
if err := archiveRows(ctx, tx, &data, &budget, "orbit_result_events", "id=?", root.id); err != nil {
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
}
|
||
if source.kind == "workflows" {
|
||
for _, table := range []string{"weave_steps", "weave_activations", "weave_waits", "weave_signals", "weave_poll_states", "weave_poll_rounds", "weave_inbox"} {
|
||
if err := archiveRows(ctx, tx, &data, &budget, table, "workflow_id=?", root.id); err != nil {
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
}
|
||
// 流程固定版本与全部历史关联任务也一起导出，不能只保存实例头。
|
||
if err := archiveRows(ctx, tx, &data, &budget, "weave_definition_versions", "id IN (SELECT definition_version_id FROM weave_instances WHERE namespace_id=? AND id=?)", namespace, root.id); err != nil {
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
taskRows, err := tx.QueryContext(ctx, `SELECT id FROM pulse_tasks WHERE namespace_id=? AND gate_id=(SELECT gate_id FROM weave_instances WHERE namespace_id=? AND id=?) LIMIT 10001`, namespace, namespace, root.id)
|
||
if err != nil {
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
var taskIDs []core.ID
|
||
for taskRows.Next() {
|
||
var id core.ID
|
||
if err := taskRows.Scan(&id); err != nil {
|
||
taskRows.Close()
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
taskIDs = append(taskIDs, id)
|
||
}
|
||
err = taskRows.Err()
|
||
taskRows.Close()
|
||
if err != nil {
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
if len(taskIDs) > archiveMaxRows {
|
||
return retention.ArchiveSnapshot{}, fault.Invalid("archive too large; use a smaller batch")
|
||
}
|
||
for _, id := range taskIDs {
|
||
if err := archiveRows(ctx, tx, &data, &budget, "pulse_tasks", "id=?", id); err != nil {
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
if err := archiveTaskHistory(ctx, tx, &data, &budget, id); err != nil {
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
}
|
||
if err := archiveRows(ctx, tx, &data, &budget, "orbit_events", "workflow_id=?", root.id); err != nil {
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
if err := archiveRows(ctx, tx, &data, &budget, "orbit_audit_events", "resource_kind='workflow' AND resource_id=?", root.id); err != nil {
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
}
|
||
}
|
||
content, err := json.Marshal(data)
|
||
if err != nil {
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
if len(content) > archiveMaxBytes {
|
||
return retention.ArchiveSnapshot{}, fault.Invalid("archive too large; use a smaller batch")
|
||
}
|
||
sum := sha256.Sum256(content)
|
||
digest := hex.EncodeToString(sum[:])
|
||
// 只记录包摘要和数量，不将归档正文或秘密写到审计/日志。
|
||
op, err := core.NewID()
|
||
if err != nil {
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
audit, err := core.NewID()
|
||
if err != nil {
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
info, _ := json.Marshal(map[string]any{"kind": in.Kind, "policy_revision": p.Revision, "root_count": len(roots), "rows": budget.rows, "sha256": digest, "source_retained": true})
|
||
_, err = tx.ExecContext(ctx, "INSERT INTO orbit_audit_events(namespace_id,id,operation_id,actor_id,action,resource_kind,resource_id,reason,`after`) VALUES (?,?,?,?,'retention.archive.export','retention_policy',?,'export retained history snapshot',?)", namespace, audit, op, actor, namespace, string(info))
|
||
if err != nil {
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return retention.ArchiveSnapshot{}, err
|
||
}
|
||
return retention.ArchiveSnapshot{FormatVersion: 1, SHA256: digest, Content: string(content)}, nil
|
||
}
|
||
|
||
type archiveBudget struct{ rows, bytes int }
|
||
|
||
func archiveTaskHistory(ctx context.Context, tx *sql.Tx, d *archiveData, b *archiveBudget, id core.ID) error {
|
||
if err := archiveRows(ctx, tx, d, b, "pulse_attempts", "task_id=?", id); err != nil {
|
||
return err
|
||
}
|
||
rows, err := tx.QueryContext(ctx, "SELECT id FROM pulse_attempts WHERE namespace_id=? AND task_id=? ORDER BY id LIMIT ?", d.Policy.NamespaceID, id, archiveMaxRows+1)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer rows.Close()
|
||
var attemptIDs []core.ID
|
||
for rows.Next() {
|
||
var attemptID core.ID
|
||
if err := rows.Scan(&attemptID); err != nil {
|
||
return err
|
||
}
|
||
attemptIDs = append(attemptIDs, attemptID)
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return err
|
||
}
|
||
if len(attemptIDs) > archiveMaxRows {
|
||
return fault.Invalid("archive too large; use a smaller batch")
|
||
}
|
||
for _, attemptID := range attemptIDs {
|
||
for _, table := range []string{"pulse_attempt_receipts", "pulse_attempt_evidence"} {
|
||
if err := archiveRows(ctx, tx, d, b, table, "attempt_id=?", attemptID); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
if err := archiveRows(ctx, tx, d, b, "orbit_result_events", "id IN (SELECT id FROM orbit_events WHERE namespace_id=? AND aggregate_kind='task' AND aggregate_id=?)", d.Policy.NamespaceID, id); err != nil {
|
||
return err
|
||
}
|
||
if err := archiveRows(ctx, tx, d, b, "orbit_events", "aggregate_kind='task' AND aggregate_id=?", id); err != nil {
|
||
return err
|
||
}
|
||
return archiveRows(ctx, tx, d, b, "orbit_audit_events", "resource_kind='task' AND resource_id=?", id)
|
||
}
|
||
|
||
// archiveRows 仅由上方固定目录调用。密钥摘要不导出，业务JSON文本及大整数不经过浮点转换。
|
||
func archiveRows(ctx context.Context, tx *sql.Tx, d *archiveData, b *archiveBudget, table, where string, args ...any) error {
|
||
q := "SELECT /*+ MAX_EXECUTION_TIME(1500) */ * FROM " + table + " WHERE namespace_id=? AND " + where + " ORDER BY id LIMIT ?"
|
||
params := append([]any{d.Policy.NamespaceID}, args...)
|
||
params = append(params, archiveMaxRows-b.rows+1)
|
||
rows, err := tx.QueryContext(ctx, q, params...)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer rows.Close()
|
||
cols, err := rows.ColumnTypes()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
for rows.Next() {
|
||
values := make([]any, len(cols))
|
||
dest := make([]any, len(cols))
|
||
for i := range values {
|
||
dest[i] = &values[i]
|
||
}
|
||
if err := rows.Scan(dest...); err != nil {
|
||
return err
|
||
}
|
||
record := map[string]any{}
|
||
for i, col := range cols {
|
||
name := col.Name()
|
||
if strings.Contains(name, "secret") || name == "token_digest" {
|
||
continue
|
||
}
|
||
v := values[i]
|
||
switch x := v.(type) {
|
||
case []byte:
|
||
if col.DatabaseTypeName() == "BINARY" || col.DatabaseTypeName() == "VARBINARY" {
|
||
v = hex.EncodeToString(x)
|
||
} else {
|
||
v = string(x)
|
||
}
|
||
case time.Time:
|
||
v = x.UTC().Format(time.RFC3339Nano)
|
||
case int64, uint64, float64:
|
||
v = fmt.Sprint(x)
|
||
}
|
||
record[name] = v
|
||
}
|
||
raw, err := json.Marshal(record)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
b.rows++
|
||
b.bytes += len(raw)
|
||
if b.rows > archiveMaxRows || b.bytes > archiveMaxBytes {
|
||
return fault.Invalid("archive too large; use a smaller batch")
|
||
}
|
||
d.Tables[table] = append(d.Tables[table], record)
|
||
}
|
||
return rows.Err()
|
||
}
|
||
|
||
var _ retention.ArchiveStore = (*RetentionPolicyStore)(nil)
|