84 lines
4.5 KiB
Go
84 lines
4.5 KiB
Go
package mysql
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"time"
|
|
)
|
|
|
|
// RetentionInventory 是指定截止时间前的只读保留清单。
|
|
// 它只提供核对数量，不代表记录已经满足归档条件，也不执行删除或搬迁。
|
|
type RetentionInventory struct {
|
|
// Cutoff 是清单使用的 UTC 截止时间，严格早于该时间的记录才会计入。
|
|
Cutoff time.Time `json:"cutoff"`
|
|
// ExpiredIdempotency 是已过幂等保留期的请求记录数量。
|
|
ExpiredIdempotency uint64 `json:"expired_idempotency"`
|
|
// TerminalTasks 是已进入确定终态且已完成的任务数量。
|
|
TerminalTasks uint64 `json:"terminal_tasks"`
|
|
// UncertainTasks 是仍带未知效果的任务数量，必须人工核对后才能处理。
|
|
UncertainTasks uint64 `json:"uncertain_tasks"`
|
|
// ActiveTasks 是尚未终结的任务数量，不能进入普通历史归档。
|
|
ActiveTasks uint64 `json:"active_tasks"`
|
|
// TerminalWorkflows 是已进入确定终态且已完成的流程数量。
|
|
TerminalWorkflows uint64 `json:"terminal_workflows"`
|
|
// ActiveWorkflows 是尚未终结的流程数量，不能进入普通历史归档。
|
|
ActiveWorkflows uint64 `json:"active_workflows"`
|
|
// AcknowledgedDeliveries 是已经确认的结果交付数量。
|
|
AcknowledgedDeliveries uint64 `json:"acknowledged_deliveries"`
|
|
// UnconfirmedDeliveries 是仍未确认的结果交付数量，不能无声清理。
|
|
UnconfirmedDeliveries uint64 `json:"unconfirmed_deliveries"`
|
|
// ProcessedInbox 是已经完成内部消费的流程事件数量。
|
|
ProcessedInbox uint64 `json:"processed_inbox"`
|
|
// UnprocessedInbox 是尚未完成内部消费的流程事件数量，不能清理。
|
|
UnprocessedInbox uint64 `json:"unprocessed_inbox"`
|
|
// Events 是截止时间前追加的不可变事件数量。
|
|
Events uint64 `json:"events"`
|
|
// AuditEvents 是截止时间前追加的管理审计数量，默认应单独保留。
|
|
AuditEvents uint64 `json:"audit_events"`
|
|
}
|
|
|
|
// ReadRetentionInventory performs bounded-by-time, read-only count queries.
|
|
// It deliberately does not decide policy or mutate any table; callers must
|
|
// review retention windows, backups, outstanding effects and audit rules first.
|
|
func ReadRetentionInventory(ctx context.Context, db *sql.DB, cutoff time.Time) (RetentionInventory, error) {
|
|
if db == nil {
|
|
return RetentionInventory{}, errors.New("database is required")
|
|
}
|
|
if cutoff.IsZero() {
|
|
return RetentionInventory{}, errors.New("cutoff is required")
|
|
}
|
|
cutoff = cutoff.UTC()
|
|
var now time.Time
|
|
if err := db.QueryRowContext(ctx, "SELECT UTC_TIMESTAMP(6)").Scan(&now); err != nil {
|
|
return RetentionInventory{}, errors.New("database time unavailable")
|
|
}
|
|
if cutoff.After(now.UTC()) {
|
|
return RetentionInventory{}, errors.New("cutoff cannot be in the future")
|
|
}
|
|
result := RetentionInventory{Cutoff: cutoff}
|
|
queries := []struct {
|
|
destination *uint64
|
|
query string
|
|
}{
|
|
{&result.ExpiredIdempotency, "SELECT COUNT(*) FROM orbit_idempotency_keys WHERE expires_at<?"},
|
|
{&result.TerminalTasks, "SELECT COUNT(*) FROM pulse_tasks WHERE state IN ('SUCCEEDED','FAILED','CANCELLED') AND completed_at IS NOT NULL AND completed_at<? AND effect_uncertain=FALSE"},
|
|
{&result.UncertainTasks, "SELECT COUNT(*) FROM pulse_tasks WHERE effect_uncertain=TRUE AND created_at<?"},
|
|
{&result.ActiveTasks, "SELECT COUNT(*) FROM pulse_tasks WHERE state NOT IN ('SUCCEEDED','FAILED','CANCELLED') AND created_at<?"},
|
|
{&result.TerminalWorkflows, "SELECT COUNT(*) FROM weave_instances WHERE state IN ('SUCCEEDED','FAILED','CANCELLED') AND completed_at IS NOT NULL AND completed_at<?"},
|
|
{&result.ActiveWorkflows, "SELECT COUNT(*) FROM weave_instances WHERE state NOT IN ('SUCCEEDED','FAILED','CANCELLED') AND created_at<?"},
|
|
{&result.AcknowledgedDeliveries, "SELECT COUNT(*) FROM orbit_deliveries WHERE state='ACKED' AND acknowledged_at IS NOT NULL AND acknowledged_at<?"},
|
|
{&result.UnconfirmedDeliveries, "SELECT COUNT(*) FROM orbit_deliveries WHERE state<>'ACKED' AND created_at<?"},
|
|
{&result.ProcessedInbox, "SELECT COUNT(*) FROM weave_inbox WHERE state='PROCESSED' AND processed_at IS NOT NULL AND processed_at<?"},
|
|
{&result.UnprocessedInbox, "SELECT COUNT(*) FROM weave_inbox WHERE state<>'PROCESSED' AND created_at<?"},
|
|
{&result.Events, "SELECT COUNT(*) FROM orbit_events WHERE occurred_at<?"},
|
|
{&result.AuditEvents, "SELECT COUNT(*) FROM orbit_audit_events WHERE created_at<?"},
|
|
}
|
|
for _, item := range queries {
|
|
if err := db.QueryRowContext(ctx, item.query, cutoff).Scan(item.destination); err != nil {
|
|
return RetentionInventory{}, errors.New("retention inventory unavailable")
|
|
}
|
|
}
|
|
return result, nil
|
|
}
|