184 lines
6.9 KiB
Go
184 lines
6.9 KiB
Go
package mysql
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"database/sql"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"time"
|
|
|
|
"git.yuntuo.hk/k2safe/orbit/internal/fault"
|
|
"git.yuntuo.hk/k2safe/orbit/internal/jsonvalue"
|
|
core "git.yuntuo.hk/k2safe/orbit/model"
|
|
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
|
"git.yuntuo.hk/k2safe/orbit/platform/events"
|
|
platform "git.yuntuo.hk/k2safe/orbit/platform/model"
|
|
)
|
|
|
|
type eventCursor struct {
|
|
Version int `json:"v"`
|
|
Kind string `json:"k"`
|
|
Namespace core.ID `json:"n"`
|
|
Credential core.ID `json:"c"`
|
|
Parent core.ID `json:"p"`
|
|
State platform.DeliveryState `json:"s"`
|
|
At time.Time `json:"t"`
|
|
ID core.ID `json:"i"`
|
|
}
|
|
|
|
func parseEventCursor(raw string, t auth.Token, parent core.ID, kind string, state platform.DeliveryState) (eventCursor, error) {
|
|
var c eventCursor
|
|
b, err := base64.RawURLEncoding.Strict().DecodeString(raw)
|
|
if err != nil || len(b) > 1024 {
|
|
return c, fault.Invalid("invalid event cursor")
|
|
}
|
|
if _, err = jsonvalue.Canonical(b); err != nil {
|
|
return c, fault.Invalid("invalid event cursor")
|
|
}
|
|
var fields map[string]json.RawMessage
|
|
if json.Unmarshal(b, &fields) != nil || len(fields) != 8 {
|
|
return c, fault.Invalid("invalid event cursor")
|
|
}
|
|
for _, k := range []string{"v", "k", "n", "c", "p", "s", "t", "i"} {
|
|
if _, ok := fields[k]; !ok {
|
|
return c, fault.Invalid("invalid event cursor")
|
|
}
|
|
}
|
|
d := json.NewDecoder(bytes.NewReader(b))
|
|
d.DisallowUnknownFields()
|
|
if d.Decode(&c) != nil || c.Version != 1 || c.Kind != kind || c.Namespace != t.NamespaceID || c.Credential != t.CredentialID || c.Parent != parent || c.State != state || c.ID.Validate() != nil || c.At.Year() < 2000 || c.At.Year() > 9998 || c.At.Nanosecond()%1000 != 0 {
|
|
return c, fault.Invalid("cursor does not match event query")
|
|
}
|
|
return c, nil
|
|
}
|
|
func makeEventCursor(t auth.Token, parent core.ID, kind string, state platform.DeliveryState, at time.Time, id core.ID) (string, error) {
|
|
b, err := json.Marshal(eventCursor{1, kind, t.NamespaceID, t.CredentialID, parent, state, at, id})
|
|
return base64.RawURLEncoding.EncodeToString(b), err
|
|
}
|
|
func (s *EventStore) AuthorizeEvents(ctx context.Context, t auth.Token) error {
|
|
return s.transaction(ctx, t, "events:consume", func(*sql.Tx, auth.Permissions) error { return nil })
|
|
}
|
|
func (s *EventStore) BindConsumer(ctx context.Context, t auth.Token, id core.ID) (events.ConsumerBinding, error) {
|
|
var out events.ConsumerBinding
|
|
err := s.transaction(ctx, t, "events:consume", func(tx *sql.Tx, p auth.Permissions) error {
|
|
sub, err := readSubscription(ctx, tx, t, p, id, false)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
out.SubscriptionID = sub.ID
|
|
out.QueueID = sub.QueueID
|
|
out.Enabled = sub.Enabled
|
|
return tx.QueryRowContext(ctx, `SELECT application_id,UTC_TIMESTAMP(6) FROM orbit_namespaces WHERE id=?`, t.NamespaceID).Scan(&out.ApplicationID, &out.ServerTime)
|
|
})
|
|
return out, err
|
|
}
|
|
func (s *EventStore) ListSubscriptions(ctx context.Context, t auth.Token, queue core.ID, in events.PageInput) (events.SubscriptionPage, error) {
|
|
out := events.SubscriptionPage{Items: []events.Subscription{}}
|
|
err := s.transaction(ctx, t, "events:read", func(tx *sql.Tx, p auth.Permissions) error {
|
|
if !p.AllowsQueue(queue) {
|
|
return fault.Forbidden
|
|
}
|
|
q := `SELECT ` + subscriptionColumns + ` FROM orbit_subscriptions WHERE namespace_id=? AND credential_id=? AND queue_id=?`
|
|
args := []any{t.NamespaceID, t.CredentialID, queue}
|
|
if t.Console != nil {
|
|
q = `SELECT ` + subscriptionColumns + ` FROM orbit_subscriptions WHERE namespace_id=? AND queue_id=?`
|
|
args = []any{t.NamespaceID, queue}
|
|
}
|
|
if in.Cursor != "" {
|
|
c, err := parseEventCursor(in.Cursor, t, queue, "subscriptions", "")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
q += ` AND (created_at<? OR (created_at=? AND id<?))`
|
|
args = append(args, c.At, c.At, c.ID)
|
|
}
|
|
q += ` ORDER BY created_at DESC,id DESC LIMIT ?`
|
|
args = append(args, in.Limit+1)
|
|
rows, err := tx.QueryContext(ctx, q, args...)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
v, err := scanSubscription(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 = makeEventCursor(t, queue, "subscriptions", "", last.CreatedAt, last.ID)
|
|
}
|
|
return err
|
|
})
|
|
return out, err
|
|
}
|
|
func (s *EventStore) ListDeliveries(ctx context.Context, t auth.Token, id core.ID, in events.PageInput) (events.DeliveryPage, error) {
|
|
out := events.DeliveryPage{Items: []events.Delivery{}}
|
|
err := s.transaction(ctx, t, "events:read", func(tx *sql.Tx, p auth.Permissions) error {
|
|
if _, err := readSubscription(ctx, tx, t, p, id, false); err != nil {
|
|
return err
|
|
}
|
|
q := `SELECT d.id,d.subscription_id,d.event_id,r.sequence,d.revision,d.state,d.available_at,d.lease_epoch,d.lease_until,d.attempts,d.attempt_budget,d.acknowledged_at,d.created_at FROM orbit_deliveries d LEFT JOIN orbit_result_events r ON r.namespace_id=d.namespace_id AND r.id=d.event_id WHERE d.namespace_id=? AND d.subscription_id=?`
|
|
args := []any{t.NamespaceID, id}
|
|
if in.State != "" {
|
|
q += ` AND d.state=?`
|
|
args = append(args, in.State)
|
|
}
|
|
if in.Cursor != "" {
|
|
c, err := parseEventCursor(in.Cursor, t, id, "deliveries", in.State)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
q += ` AND (d.created_at<? OR (d.created_at=? AND d.id<?))`
|
|
args = append(args, c.At, c.At, c.ID)
|
|
}
|
|
q += ` ORDER BY d.created_at DESC,d.id DESC LIMIT ?`
|
|
args = append(args, in.Limit+1)
|
|
rows, err := tx.QueryContext(ctx, q, args...)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var v events.Delivery
|
|
if err = rows.Scan(&v.ID, &v.SubscriptionID, &v.EventID, &v.Sequence, &v.Revision, &v.State, &v.AvailableAt, &v.LeaseEpoch, &v.LeaseUntil, &v.Attempts, &v.AttemptBudget, &v.AcknowledgedAt, &v.CreatedAt); 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 = makeEventCursor(t, id, "deliveries", in.State, last.CreatedAt, last.ID)
|
|
}
|
|
return err
|
|
})
|
|
return out, err
|
|
}
|
|
func (s *EventStore) GetDeliveryDetail(ctx context.Context, t auth.Token, sub, id core.ID) (events.DeliveryDetail, error) {
|
|
var out events.DeliveryDetail
|
|
err := s.transaction(ctx, t, "events:read", func(tx *sql.Tx, p auth.Permissions) error {
|
|
if _, err := readSubscription(ctx, tx, t, p, sub, false); err != nil {
|
|
return err
|
|
}
|
|
d, err := readDelivery(ctx, tx, t.NamespaceID, sub, id, false)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
out.Delivery = d.view
|
|
out.Event, err = readResultEvent(ctx, tx, t.NamespaceID, d.view.EventID)
|
|
return err
|
|
})
|
|
return out, err
|
|
}
|