101 lines
3.0 KiB
Go
101 lines
3.0 KiB
Go
package mysql
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"hyapp/pkg/appcode"
|
|
"hyapp/pkg/xerr"
|
|
)
|
|
|
|
// UserOutboxRecord is a durable user-service fact waiting for MQ fanout.
|
|
type UserOutboxRecord struct {
|
|
AppCode string
|
|
EventID string
|
|
EventType string
|
|
AggregateType string
|
|
AggregateID int64
|
|
PayloadJSON string
|
|
CreatedAtMS int64
|
|
}
|
|
|
|
// ClaimPendingUserOutbox locks pending user facts so one worker can publish them.
|
|
func (r *Repository) ClaimPendingUserOutbox(ctx context.Context, workerID string, batchSize int) ([]UserOutboxRecord, error) {
|
|
if r == nil || r.db == nil {
|
|
return nil, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
|
}
|
|
if batchSize <= 0 {
|
|
batchSize = 100
|
|
}
|
|
tx, err := r.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
rows, err := tx.QueryContext(ctx, `
|
|
SELECT app_code, event_id, event_type, aggregate_type, aggregate_id, CAST(payload_json AS CHAR), created_at_ms
|
|
FROM user_outbox
|
|
WHERE status = 'pending'
|
|
ORDER BY created_at_ms ASC, event_id ASC
|
|
LIMIT ? FOR UPDATE SKIP LOCKED
|
|
`, batchSize)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
records := make([]UserOutboxRecord, 0, batchSize)
|
|
for rows.Next() {
|
|
var record UserOutboxRecord
|
|
if err := rows.Scan(&record.AppCode, &record.EventID, &record.EventType, &record.AggregateType, &record.AggregateID, &record.PayloadJSON, &record.CreatedAtMS); err != nil {
|
|
_ = rows.Close()
|
|
return nil, err
|
|
}
|
|
records = append(records, record)
|
|
}
|
|
if err := rows.Close(); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
nowMS := time.Now().UTC().UnixMilli()
|
|
for _, record := range records {
|
|
if _, err := tx.ExecContext(ctx, `
|
|
UPDATE user_outbox
|
|
SET status = 'running', updated_at_ms = ?
|
|
WHERE app_code = ? AND event_id = ? AND status = 'pending'
|
|
`, nowMS, appcode.Normalize(record.AppCode), record.EventID); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return nil, err
|
|
}
|
|
return records, nil
|
|
}
|
|
|
|
// MarkUserOutboxDelivered marks a user fact as published after RocketMQ accepts it.
|
|
func (r *Repository) MarkUserOutboxDelivered(ctx context.Context, eventID string) error {
|
|
if r == nil || r.db == nil {
|
|
return xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
|
}
|
|
_, err := r.db.ExecContext(ctx, `
|
|
UPDATE user_outbox
|
|
SET status = 'delivered', updated_at_ms = ?
|
|
WHERE app_code = ? AND event_id = ?
|
|
`, time.Now().UTC().UnixMilli(), appcode.FromContext(ctx), eventID)
|
|
return err
|
|
}
|
|
|
|
// MarkUserOutboxPending returns a failed publish to pending so the next tick retries.
|
|
func (r *Repository) MarkUserOutboxPending(ctx context.Context, eventID string) error {
|
|
if r == nil || r.db == nil {
|
|
return xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
|
}
|
|
_, err := r.db.ExecContext(ctx, `
|
|
UPDATE user_outbox
|
|
SET status = 'pending', updated_at_ms = ?
|
|
WHERE app_code = ? AND event_id = ? AND status = 'running'
|
|
`, time.Now().UTC().UnixMilli(), appcode.FromContext(ctx), eventID)
|
|
return err
|
|
}
|