183 lines
6.3 KiB
Go
183 lines
6.3 KiB
Go
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"database/sql"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"math"
|
||
|
||
"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/platform/events"
|
||
driver "github.com/go-sql-driver/mysql"
|
||
)
|
||
|
||
// EventStore 管理独立结果订阅；消费者事务不锁任务、队列、应用额度或结果流行。
|
||
// 创建时按认证→请求身份→队列共享锁→结果流锁顺序确定提交边界。
|
||
type EventStore struct{ *TaskStore }
|
||
|
||
func NewEventStore(db *sql.DB) *EventStore { return &EventStore{NewTaskStore(db)} }
|
||
|
||
var _ events.Store = (*EventStore)(nil)
|
||
|
||
const subscriptionColumns = `id,queue_id,credential_id,name,revision,enabled,filter,cursor_sequence,pending_count,max_pending,max_attempts,created_at,updated_at`
|
||
|
||
func readSubscription(ctx context.Context, tx *sql.Tx, t auth.Token, p auth.Permissions, id core.ID, lock bool) (events.Subscription, error) {
|
||
q := `SELECT ` + subscriptionColumns + ` FROM orbit_subscriptions WHERE namespace_id=? AND id=? AND credential_id=? AND queue_id IS NOT NULL`
|
||
if lock {
|
||
q += ` FOR UPDATE`
|
||
}
|
||
out, err := scanSubscription(tx.QueryRowContext(ctx, q, t.NamespaceID, id, t.CredentialID))
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return out, fault.NotFound
|
||
}
|
||
if err != nil {
|
||
return out, err
|
||
}
|
||
if !p.AllowsQueue(out.QueueID) {
|
||
return events.Subscription{}, fault.NotFound
|
||
}
|
||
return out, nil
|
||
}
|
||
func scanSubscription(row scanner) (events.Subscription, error) {
|
||
var out events.Subscription
|
||
var raw []byte
|
||
err := row.Scan(&out.ID, &out.QueueID, &out.CredentialID, &out.Name, &out.Revision, &out.Enabled, &raw, &out.CursorSequence, &out.PendingCount, &out.MaxPending, &out.MaxAttempts, &out.CreatedAt, &out.UpdatedAt)
|
||
if err != nil {
|
||
return out, err
|
||
}
|
||
err = json.Unmarshal(raw, &out.Types)
|
||
return out, err
|
||
}
|
||
|
||
func (s *EventStore) CreateSubscription(ctx context.Context, t auth.Token, key string, in events.CreateInput) (events.SubscriptionResult, error) {
|
||
var out events.SubscriptionResult
|
||
err := s.transaction(ctx, t, "events:manage", func(tx *sql.Tx, p auth.Permissions) error {
|
||
if !p.AllowsQueue(in.QueueID) {
|
||
return fault.Forbidden
|
||
}
|
||
id, err := core.NewID()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, "subscription.create", string(t.CredentialID), key, "subscription", id, in)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if cached != nil {
|
||
if err = json.Unmarshal(cached, &out); err != nil {
|
||
return err
|
||
}
|
||
out.Deduplicated = true
|
||
return nil
|
||
}
|
||
var queue core.ID
|
||
if err = tx.QueryRowContext(ctx, `SELECT id FROM pulse_queues WHERE namespace_id=? AND id=? FOR SHARE`, t.NamespaceID, in.QueueID).Scan(&queue); errors.Is(err, sql.ErrNoRows) {
|
||
return fault.NotFound
|
||
} else if err != nil {
|
||
return err
|
||
}
|
||
head, retained, err := lockEventStream(ctx, tx, t.NamespaceID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
cursor := head
|
||
if in.FromSequence != nil {
|
||
cursor = *in.FromSequence
|
||
}
|
||
if cursor > head {
|
||
return fault.Invalid("from_sequence is ahead of committed history")
|
||
}
|
||
if cursor < retained-1 {
|
||
return events.HistoryExpired
|
||
}
|
||
var count int
|
||
if err = tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM orbit_subscriptions WHERE namespace_id=?`, t.NamespaceID).Scan(&count); err != nil {
|
||
return err
|
||
}
|
||
if count >= events.MaxSubscriptions {
|
||
return events.SubscriptionLimit
|
||
}
|
||
filter, err := json.Marshal(in.Types)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
_, err = tx.ExecContext(ctx, `INSERT INTO orbit_subscriptions(namespace_id,id,name,credential_id,filter,queue_id,cursor_sequence,max_pending,max_attempts) VALUES (?,?,?,?,?,?,?,?,?)`, t.NamespaceID, id, in.Name, t.CredentialID, string(filter), in.QueueID, cursor, in.MaxPending, in.MaxAttempts)
|
||
var me *driver.MySQLError
|
||
if errors.As(err, &me) && me.Number == 1062 {
|
||
return fault.Conflict
|
||
}
|
||
if err != nil {
|
||
return err
|
||
}
|
||
out.Subscription, err = readSubscription(ctx, tx, t, p, id, false)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if err = appendAudit(ctx, tx, t, identity, id, "subscription", "subscription.create", "create durable result subscription", nil, out.Subscription); err != nil {
|
||
return err
|
||
}
|
||
return saveIdentity(ctx, tx, t.NamespaceID, identity, out)
|
||
})
|
||
return out, err
|
||
}
|
||
func (s *EventStore) GetSubscription(ctx context.Context, t auth.Token, id core.ID) (events.Subscription, error) {
|
||
var out events.Subscription
|
||
err := s.transaction(ctx, t, "events:read", func(tx *sql.Tx, p auth.Permissions) error {
|
||
var err error
|
||
out, err = readSubscription(ctx, tx, t, p, id, false)
|
||
return err
|
||
})
|
||
return out, err
|
||
}
|
||
func (s *EventStore) ControlSubscription(ctx context.Context, t auth.Token, id core.ID, key string, in events.ControlInput) (events.SubscriptionResult, error) {
|
||
var out events.SubscriptionResult
|
||
err := s.transaction(ctx, t, "events:manage", func(tx *sql.Tx, p auth.Permissions) error {
|
||
identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, "subscription.control", eventScope(t.CredentialID, id), key, "subscription", id, in)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
before, err := readSubscription(ctx, tx, t, p, id, true)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if cached != nil {
|
||
if err = json.Unmarshal(cached, &out); err != nil {
|
||
return err
|
||
}
|
||
out.Deduplicated = true
|
||
return nil
|
||
}
|
||
if before.Revision != in.ExpectedRevision || before.Revision == math.MaxUint64 {
|
||
return fault.Conflict
|
||
}
|
||
if _, err = tx.ExecContext(ctx, `UPDATE orbit_subscriptions SET enabled=?,revision=revision+1,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, in.Enabled, t.NamespaceID, id); err != nil {
|
||
return err
|
||
}
|
||
out.Subscription, err = readSubscription(ctx, tx, t, p, id, false)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if err = appendAudit(ctx, tx, t, identity, id, "subscription", "subscription.control", in.Reason, before, out.Subscription); err != nil {
|
||
return err
|
||
}
|
||
return saveIdentity(ctx, tx, t.NamespaceID, identity, out)
|
||
})
|
||
return out, err
|
||
}
|
||
|
||
// eventScope 将带域隔离的复合资源身份压缩到已有64字符幂等作用域，避免截断或串联歧义。
|
||
func eventScope(ids ...core.ID) string {
|
||
h := sha256.New()
|
||
h.Write([]byte("orbit-event-scope-v1"))
|
||
for _, id := range ids {
|
||
h.Write([]byte{0})
|
||
h.Write([]byte(id))
|
||
}
|
||
return hex.EncodeToString(h.Sum(nil))
|
||
}
|