orbit/internal/storage/mysql/retention_policy.go

package mysql

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"strings"

	"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"
	driver "github.com/go-sql-driver/mysql"
)

const retentionPolicyOperation = "retention.policy.update"
const retentionPolicyColumns = `revision,enabled,idempotency_days,task_history_days,workflow_history_days,event_log_days,delivery_history_days,inbox_history_days,audit_log_days,updated_at`

// RetentionPolicyStore 保存命名空间级策略，数据库配置优先于部署默认值。
// 不缓存配置；节点重启和跨节点读取保持数据库中同一版本。
type RetentionPolicyStore struct {
	db       *sql.DB
	defaults retention.Policy
}

func NewRetentionPolicyStore(db *sql.DB, defaults ...retention.Policy) *RetentionPolicyStore {
	p := retention.Defaults("")
	if len(defaults) > 0 {
		p = defaults[0]
	}
	return &RetentionPolicyStore{db: db, defaults: p}
}

var _ retention.Store = (*RetentionPolicyStore)(nil)

func (s *RetentionPolicyStore) defaultPolicy(namespace core.ID) retention.Policy {
	p := s.defaults
	p.NamespaceID = namespace
	p.Source = "default"
	return p
}

type retentionQuery interface {
	QueryRowContext(context.Context, string, ...any) *sql.Row
}

func (s *RetentionPolicyStore) readPolicy(ctx context.Context, db retentionQuery, namespace core.ID, lock bool) (retention.Policy, error) {
	p := s.defaultPolicy(namespace)
	q := `SELECT ` + retentionPolicyColumns + ` FROM orbit_retention_policies WHERE namespace_id=?`
	if lock {
		q += " FOR UPDATE"
	}
	err := db.QueryRowContext(ctx, q, namespace).Scan(&p.Revision, &p.Enabled, &p.IdempotencyDays, &p.TaskHistoryDays, &p.WorkflowHistoryDays, &p.EventLogDays, &p.DeliveryHistoryDays, &p.InboxHistoryDays, &p.AuditLogDays, &p.UpdatedAt)
	if errors.Is(err, sql.ErrNoRows) {
		return p, p.Validate()
	}
	if err != nil {
		return retention.Policy{}, err
	}
	p.Source = "database"
	return p, p.Validate()
}

// GetRetentionPolicy 供经过会话认证的控制台和数据库管理工具读取配置。
func (s *RetentionPolicyStore) GetRetentionPolicy(ctx context.Context, namespace core.ID) (retention.Policy, error) {
	if namespace.Validate() != nil {
		return retention.Policy{}, fault.Invalid("invalid namespace")
	}
	return s.readPolicy(ctx, s.db, namespace, false)
}

// retentionActor 在策略事务内验证控制台会话。普通 SDK 凭据不能修改保留策略。
// 管理员必须具备当前命名空间全部队列的管理范围；不添加角色体系或跨命名空间权限。
func retentionActor(ctx context.Context, tx *sql.Tx, token auth.Token) (core.ID, error) {
	p, err := authenticate(ctx, tx, token)
	if err != nil {
		return "", err
	}
	all := false
	for _, q := range p.Queues {
		if q == "*" {
			all = true
		}
	}
	if !all || !p.Allows("queues:manage") {
		return "", fault.Forbidden
	}
	var user core.ID
	err = tx.QueryRowContext(ctx, `SELECT u.id FROM orbit_console_sessions s
 JOIN orbit_console_users u ON u.id=s.user_id AND u.namespace_id=s.namespace_id AND u.credential_id=s.credential_id
 WHERE s.namespace_id=? AND s.credential_id=? AND s.token_digest=? AND s.revoked_at IS NULL
 AND s.expires_at>UTC_TIMESTAMP(6) AND u.enabled=TRUE FOR SHARE`, token.NamespaceID, token.CredentialID, token.Digest).Scan(&user)
	if errors.Is(err, sql.ErrNoRows) {
		return "", fault.Unauthorized
	}
	return user, err
}

// UpdateRetentionPolicy 在同一事务中核验版本、保存策略、前后快照、账号身份和幂等回执。
func (s *RetentionPolicyStore) UpdateRetentionPolicy(ctx context.Context, token auth.Token, key string, in retention.UpdateInput) (retention.Policy, error) {
	if strings.TrimSpace(key) == "" || len(key) > 128 {
		return retention.Policy{}, fault.Invalid("idempotency key must be 1-128 bytes")
	}
	want, err := in.ToPolicy(token.NamespaceID)
	if err != nil {
		return retention.Policy{}, fault.Invalid(err.Error())
	}
	var out retention.Policy
	err = WithTx(ctx, s.db, func(tx *sql.Tx) error {
		user, err := retentionActor(ctx, tx, token)
		if err != nil {
			return err
		}
		identity, cached, err := reserveIdentity(ctx, tx, token.NamespaceID, retentionPolicyOperation, "", key, "retention_policy", token.NamespaceID, in)
		if err != nil {
			return err
		}
		if cached != nil {
			return json.Unmarshal(cached, &out)
		}
		current, err := s.readPolicy(ctx, tx, token.NamespaceID, true)
		if err != nil {
			return err
		}
		if current.Revision != in.ExpectedRevision {
			return fault.Conflict
		}
		if current.Revision == 0 {
			_, err = tx.ExecContext(ctx, `INSERT INTO orbit_retention_policies(namespace_id,enabled,idempotency_days,task_history_days,workflow_history_days,event_log_days,delivery_history_days,inbox_history_days,audit_log_days) VALUES (?,?,?,?,?,?,?,?,?)`, token.NamespaceID, want.Enabled, want.IdempotencyDays, want.TaskHistoryDays, want.WorkflowHistoryDays, want.EventLogDays, want.DeliveryHistoryDays, want.InboxHistoryDays, want.AuditLogDays)
			var me *driver.MySQLError
			if errors.As(err, &me) && me.Number == 1062 {
				return fault.Conflict
			}
		} else {
			_, err = tx.ExecContext(ctx, `UPDATE orbit_retention_policies SET revision=revision+1,enabled=?,idempotency_days=?,task_history_days=?,workflow_history_days=?,event_log_days=?,delivery_history_days=?,inbox_history_days=?,audit_log_days=?,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND revision=?`, want.Enabled, want.IdempotencyDays, want.TaskHistoryDays, want.WorkflowHistoryDays, want.EventLogDays, want.DeliveryHistoryDays, want.InboxHistoryDays, want.AuditLogDays, token.NamespaceID, in.ExpectedRevision)
		}
		if err != nil {
			return err
		}
		out, err = s.readPolicy(ctx, tx, token.NamespaceID, false)
		if err != nil {
			return err
		}
		auditID, err := core.NewID()
		if err != nil {
			return err
		}
		old, _ := json.Marshal(current)
		updated, _ := json.Marshal(out)
		_, err = tx.ExecContext(ctx, "INSERT INTO orbit_audit_events(namespace_id,id,operation_id,actor_id,action,resource_kind,resource_id,reason,`before`,`after`) VALUES (?,?,?,?,?,'retention_policy',?,?,?,?)", token.NamespaceID, auditID, identity, "console:"+string(user), retentionPolicyOperation, token.NamespaceID, in.Reason, string(old), string(updated))
		if err != nil {
			return err
		}
		return saveIdentity(ctx, tx, token.NamespaceID, identity, out)
	})
	return out, err
}