506 lines
17 KiB
Go
506 lines
17 KiB
Go
// Command replay-deliveries replays an explicitly selected set of expired
|
|
// notice delivery leases from immutable owner outbox facts.
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"flag"
|
|
"fmt"
|
|
"log"
|
|
"os"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
roomeventsv1 "hyapp.local/api/proto/events/room/v1"
|
|
"hyapp/pkg/appcode"
|
|
"hyapp/pkg/tencentim"
|
|
"hyapp/pkg/walletmq"
|
|
"hyapp/services/notice-service/internal/config"
|
|
"hyapp/services/notice-service/internal/modules/roomnotice"
|
|
"hyapp/services/notice-service/internal/modules/walletnotice"
|
|
mysqlplatform "hyapp/services/notice-service/internal/platform/mysql"
|
|
)
|
|
|
|
const (
|
|
walletSource = "wallet_outbox"
|
|
roomSource = "room_outbox"
|
|
c2cChannel = "tencent_im_c2c"
|
|
)
|
|
|
|
type repeatedFlag []string
|
|
|
|
func (values *repeatedFlag) String() string { return strings.Join(*values, ",") }
|
|
|
|
func (values *repeatedFlag) Set(value string) error {
|
|
value = strings.TrimSpace(value)
|
|
if value == "" {
|
|
return errors.New("flag value cannot be empty")
|
|
}
|
|
*values = append(*values, value)
|
|
return nil
|
|
}
|
|
|
|
type deliverySelector struct {
|
|
SourceName string
|
|
AppCode string
|
|
EventID string
|
|
}
|
|
|
|
func (selector deliverySelector) String() string {
|
|
return selector.SourceName + ":" + selector.AppCode + ":" + selector.EventID
|
|
}
|
|
|
|
type deliverySelectorFlag []deliverySelector
|
|
|
|
func (values *deliverySelectorFlag) String() string {
|
|
parts := make([]string, 0, len(*values))
|
|
for _, selector := range *values {
|
|
parts = append(parts, selector.String())
|
|
}
|
|
return strings.Join(parts, ",")
|
|
}
|
|
|
|
func (values *deliverySelectorFlag) Set(value string) error {
|
|
parts := strings.SplitN(strings.TrimSpace(value), ":", 3)
|
|
if len(parts) != 3 {
|
|
return errors.New("delivery must use source_name:app_code:event_id")
|
|
}
|
|
selector := deliverySelector{
|
|
SourceName: strings.TrimSpace(parts[0]),
|
|
AppCode: strings.TrimSpace(parts[1]),
|
|
EventID: strings.TrimSpace(parts[2]),
|
|
}
|
|
if selector.SourceName != walletSource && selector.SourceName != roomSource {
|
|
return fmt.Errorf("unsupported delivery source %q", selector.SourceName)
|
|
}
|
|
if selector.AppCode == "" || selector.EventID == "" {
|
|
return errors.New("delivery app_code and event_id are required")
|
|
}
|
|
selector.AppCode = appcode.Normalize(selector.AppCode)
|
|
*values = append(*values, selector)
|
|
return nil
|
|
}
|
|
|
|
type deliveryCandidate struct {
|
|
Selector string `json:"selector"`
|
|
SourceName string `json:"source_name"`
|
|
AppCode string `json:"app_code"`
|
|
EventID string `json:"event_id"`
|
|
Status string `json:"status"`
|
|
RetryCount int `json:"retry_count"`
|
|
UpdatedAtMS int64 `json:"updated_at_ms"`
|
|
}
|
|
|
|
type replayItem struct {
|
|
Candidate deliveryCandidate
|
|
WalletMessage *walletmq.WalletOutboxMessage
|
|
RoomEnvelope *roomeventsv1.EventEnvelope
|
|
}
|
|
|
|
func main() {
|
|
configPath := flag.String("config", "services/notice-service/configs/config.yaml", "path to notice-service config")
|
|
execute := flag.Bool("execute", false, "publish selected due deliveries; default is a read-only preview")
|
|
limit := flag.Int("limit", 100, "maximum candidate count, 1..500")
|
|
var eventIDs repeatedFlag
|
|
var appCodes repeatedFlag
|
|
var deliveries deliverySelectorFlag
|
|
flag.Var(&eventIDs, "event-id", "preview filter for a source event ID; repeat for multiple events")
|
|
flag.Var(&appCodes, "app-code", "required App partition for indexed preview; repeat for multiple Apps")
|
|
flag.Var(&deliveries, "delivery", "exact source_name:app_code:event_id selector copied from preview; repeat for multiple deliveries")
|
|
flag.Parse()
|
|
|
|
if *limit <= 0 || *limit > 500 {
|
|
log.Fatal("limit must be between 1 and 500")
|
|
}
|
|
if len(deliveries) > 0 && (len(eventIDs) > 0 || len(appCodes) > 0) {
|
|
log.Fatal("--delivery cannot be combined with --event-id or --app-code")
|
|
}
|
|
if *execute && len(deliveries) == 0 {
|
|
// A write run is pinned to full primary-key selectors copied from preview;
|
|
// separate App/event filters can form an accidental cross product.
|
|
log.Fatal("--execute requires at least one exact --delivery from a preview")
|
|
}
|
|
if len(deliveries) > *limit {
|
|
log.Fatal("exact delivery count exceeds --limit")
|
|
}
|
|
if len(deliveries) == 0 && len(appCodes) == 0 {
|
|
// notice_delivery_events indexes start with source_name,app_code. Refusing
|
|
// an unpartitioned preview prevents a LIMIT query from scanning a 20+ GiB table.
|
|
log.Fatal("preview requires at least one --app-code or exact --delivery")
|
|
}
|
|
|
|
cfg, err := config.Load(*configPath)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
|
defer cancel()
|
|
store, err := mysqlplatform.Open(ctx, cfg.MySQLDSN)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
defer func() { _ = store.Close() }()
|
|
|
|
nowMS := time.Now().UnixMilli()
|
|
var candidates []deliveryCandidate
|
|
if len(deliveries) > 0 {
|
|
candidates, err = loadExactDueDeliveries(ctx, store.DB, deliveries, nowMS)
|
|
} else {
|
|
candidates, err = listDueDeliveries(ctx, store.DB, eventIDs, appCodes, *limit, nowMS)
|
|
}
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
if !*execute {
|
|
encoder := json.NewEncoder(os.Stdout)
|
|
for _, candidate := range candidates {
|
|
if err := encoder.Encode(candidate); err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
}
|
|
fmt.Fprintf(os.Stderr, "previewed %d due delivery rows; no writes performed\n", len(candidates))
|
|
return
|
|
}
|
|
if !cfg.TencentIM.Enabled {
|
|
log.Fatal("tencent_im.enabled is required for replay")
|
|
}
|
|
walletRepo, err := walletnotice.NewMySQLRepository(store.DB, cfg.WalletDatabase)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
roomRepo, err := roomnotice.NewMySQLRepository(store.DB, cfg.RoomDatabase)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
items, err := loadReplayItems(ctx, candidates, walletRepo, roomRepo)
|
|
if err != nil {
|
|
// Preload every immutable owner fact before creating an IM client or
|
|
// publishing the first item. One missing/corrupt fact therefore aborts the
|
|
// exact set without producing a partial operator repair.
|
|
log.Fatal(err)
|
|
}
|
|
publisher, err := tencentim.NewRESTClient(cfg.TencentIM.RESTConfig())
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
walletService := walletnotice.New(walletnotice.Config{NodeID: cfg.NodeID}, walletRepo, publisher)
|
|
roomService := roomnotice.New(roomnotice.Config{NodeID: cfg.NodeID}, roomRepo, publisher)
|
|
walletOptions := walletWorkerOptions(cfg)
|
|
roomOptions := roomWorkerOptions(cfg)
|
|
|
|
failures := 0
|
|
for _, item := range items {
|
|
candidate := item.Candidate
|
|
candidateCtx := appcode.WithContext(ctx, candidate.AppCode)
|
|
var replayErr error
|
|
switch candidate.SourceName {
|
|
case walletSource:
|
|
if item.WalletMessage == nil {
|
|
replayErr = errors.New("wallet owner snapshot is missing")
|
|
} else {
|
|
_, replayErr = walletService.ProcessWalletOutboxMessage(candidateCtx, *item.WalletMessage, walletOptions)
|
|
}
|
|
case roomSource:
|
|
if item.RoomEnvelope == nil {
|
|
replayErr = errors.New("room owner snapshot is missing")
|
|
} else {
|
|
_, replayErr = roomService.ProcessRoomOutboxEnvelope(candidateCtx, item.RoomEnvelope, roomOptions)
|
|
}
|
|
default:
|
|
replayErr = fmt.Errorf("unsupported source %q", candidate.SourceName)
|
|
}
|
|
if replayErr == nil {
|
|
replayErr = requireDelivered(ctx, store.DB, candidate)
|
|
}
|
|
if replayErr != nil {
|
|
failures++
|
|
fmt.Fprintf(os.Stderr, "replay failed source=%s app_code=%s event_id=%s: %v\n", candidate.SourceName, candidate.AppCode, candidate.EventID, replayErr)
|
|
continue
|
|
}
|
|
fmt.Printf("delivered source=%s app_code=%s event_id=%s\n", candidate.SourceName, candidate.AppCode, candidate.EventID)
|
|
}
|
|
if failures > 0 {
|
|
log.Fatalf("replay finished with %d failures out of %d selected rows", failures, len(candidates))
|
|
}
|
|
fmt.Printf("replay delivered %d selected rows\n", len(candidates))
|
|
}
|
|
|
|
func loadReplayItems(ctx context.Context, candidates []deliveryCandidate, walletRepo *walletnotice.MySQLRepository, roomRepo *roomnotice.MySQLRepository) ([]replayItem, error) {
|
|
items := make([]replayItem, 0, len(candidates))
|
|
for _, candidate := range candidates {
|
|
candidateCtx := appcode.WithContext(ctx, candidate.AppCode)
|
|
item := replayItem{Candidate: candidate}
|
|
switch candidate.SourceName {
|
|
case walletSource:
|
|
if walletRepo == nil {
|
|
return nil, errors.New("wallet owner repository is not configured")
|
|
}
|
|
message, err := walletRepo.LoadWalletOutboxMessage(candidateCtx, candidate.AppCode, candidate.EventID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("load owner snapshot source=%s app_code=%s event_id=%s: %w", candidate.SourceName, candidate.AppCode, candidate.EventID, err)
|
|
}
|
|
item.WalletMessage = &message
|
|
case roomSource:
|
|
if roomRepo == nil {
|
|
return nil, errors.New("room owner repository is not configured")
|
|
}
|
|
envelope, err := roomRepo.LoadRoomOutboxEnvelope(candidateCtx, candidate.AppCode, candidate.EventID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("load owner snapshot source=%s app_code=%s event_id=%s: %w", candidate.SourceName, candidate.AppCode, candidate.EventID, err)
|
|
}
|
|
item.RoomEnvelope = envelope
|
|
default:
|
|
return nil, fmt.Errorf("unsupported source %q", candidate.SourceName)
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
func listDueDeliveries(ctx context.Context, db *sql.DB, eventIDs []string, appCodes []string, limit int, nowMS int64) ([]deliveryCandidate, error) {
|
|
if db == nil {
|
|
return nil, errors.New("notice database is not configured")
|
|
}
|
|
if limit <= 0 || limit > 500 {
|
|
return nil, errors.New("limit must be between 1 and 500")
|
|
}
|
|
apps := normalizeAppCodes(appCodes)
|
|
if len(apps) == 0 {
|
|
return nil, errors.New("at least one app_code is required for indexed preview")
|
|
}
|
|
if len(apps) > 32 {
|
|
return nil, errors.New("app_code filter count exceeds 32")
|
|
}
|
|
ids := normalizeEventIDs(eventIDs)
|
|
if len(ids) > 500 {
|
|
return nil, errors.New("event_id filter count exceeds 500")
|
|
}
|
|
candidates := make([]deliveryCandidate, 0, limit)
|
|
for _, sourceName := range []string{walletSource, roomSource} {
|
|
for _, appCodeValue := range apps {
|
|
branches := []struct {
|
|
indexName string
|
|
status string
|
|
dueColumn string
|
|
}{
|
|
{indexName: "idx_notice_delivery_retry", status: "retryable", dueColumn: "next_retry_at_ms"},
|
|
{indexName: "idx_notice_delivery_lock", status: "delivering", dueColumn: "lock_until_ms"},
|
|
}
|
|
for _, branch := range branches {
|
|
query, args := dueDeliveryBranch(sourceName, appCodeValue, branch.indexName, branch.status, branch.dueColumn, ids, limit, nowMS)
|
|
branchCandidates, err := queryDueDeliveryCandidates(ctx, db, query, args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
candidates = append(candidates, branchCandidates...)
|
|
}
|
|
}
|
|
}
|
|
sort.Slice(candidates, func(i, j int) bool {
|
|
if candidates[i].UpdatedAtMS != candidates[j].UpdatedAtMS {
|
|
return candidates[i].UpdatedAtMS < candidates[j].UpdatedAtMS
|
|
}
|
|
return candidates[i].Selector < candidates[j].Selector
|
|
})
|
|
if len(candidates) > limit {
|
|
candidates = candidates[:limit]
|
|
}
|
|
return candidates, nil
|
|
}
|
|
|
|
func loadExactDueDeliveries(ctx context.Context, db *sql.DB, selectors []deliverySelector, nowMS int64) ([]deliveryCandidate, error) {
|
|
if db == nil {
|
|
return nil, errors.New("notice database is not configured")
|
|
}
|
|
normalized, err := normalizeDeliverySelectors(selectors)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
candidates := make([]deliveryCandidate, 0, len(selectors))
|
|
for _, selector := range normalized {
|
|
key := selector.String()
|
|
var candidate deliveryCandidate
|
|
err = db.QueryRowContext(ctx, `
|
|
SELECT source_name, app_code, source_event_id, status, retry_count, updated_at_ms
|
|
FROM notice_delivery_events FORCE INDEX (PRIMARY)
|
|
WHERE source_name = ? AND app_code = ? AND source_event_id = ? AND channel = ?
|
|
AND ((status = 'retryable' AND next_retry_at_ms <= ?)
|
|
OR (status = 'delivering' AND lock_until_ms <= ?))`,
|
|
selector.SourceName,
|
|
selector.AppCode,
|
|
selector.EventID,
|
|
c2cChannel,
|
|
nowMS,
|
|
nowMS,
|
|
).Scan(&candidate.SourceName, &candidate.AppCode, &candidate.EventID, &candidate.Status, &candidate.RetryCount, &candidate.UpdatedAtMS)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return nil, fmt.Errorf("delivery %q does not exist or is no longer due", key)
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if candidate.SourceName != selector.SourceName || candidate.AppCode != selector.AppCode || candidate.EventID != selector.EventID {
|
|
return nil, fmt.Errorf("delivery %q returned a mismatched primary key", key)
|
|
}
|
|
candidate.Selector = key
|
|
candidates = append(candidates, candidate)
|
|
}
|
|
return candidates, nil
|
|
}
|
|
|
|
func normalizeDeliverySelectors(values []deliverySelector) ([]deliverySelector, error) {
|
|
normalized := make([]deliverySelector, 0, len(values))
|
|
seen := make(map[string]struct{}, len(values))
|
|
for _, value := range values {
|
|
selector := deliverySelector{
|
|
SourceName: strings.TrimSpace(value.SourceName),
|
|
AppCode: strings.TrimSpace(value.AppCode),
|
|
EventID: strings.TrimSpace(value.EventID),
|
|
}
|
|
if selector.SourceName != walletSource && selector.SourceName != roomSource {
|
|
return nil, fmt.Errorf("unsupported delivery source %q", selector.SourceName)
|
|
}
|
|
if selector.AppCode == "" || selector.EventID == "" {
|
|
return nil, errors.New("delivery app_code and event_id are required")
|
|
}
|
|
selector.AppCode = appcode.Normalize(selector.AppCode)
|
|
key := selector.String()
|
|
if _, exists := seen[key]; exists {
|
|
return nil, fmt.Errorf("duplicate delivery selector %q", key)
|
|
}
|
|
seen[key] = struct{}{}
|
|
normalized = append(normalized, selector)
|
|
}
|
|
return normalized, nil
|
|
}
|
|
|
|
func dueDeliveryBranch(sourceName string, appCodeValue string, indexName string, status string, dueColumn string, eventIDs []string, limit int, nowMS int64) (string, []any) {
|
|
// Callers pass only compile-time index/status/column tuples. Splitting status
|
|
// branches gives MySQL the complete (source,app,channel,status,due_time) prefix.
|
|
query := `
|
|
SELECT source_name, app_code, source_event_id, status, retry_count, updated_at_ms
|
|
FROM notice_delivery_events FORCE INDEX (` + indexName + `)
|
|
WHERE source_name = ? AND app_code = ? AND channel = ? AND status = ?
|
|
AND ` + dueColumn + ` <= ?`
|
|
args := []any{sourceName, appCodeValue, c2cChannel, status, nowMS}
|
|
if len(eventIDs) > 0 {
|
|
query += " AND source_event_id IN (" + placeholders(len(eventIDs)) + ")"
|
|
for _, eventID := range eventIDs {
|
|
args = append(args, eventID)
|
|
}
|
|
}
|
|
query += " ORDER BY updated_at_ms ASC, source_event_id ASC LIMIT ?"
|
|
args = append(args, limit)
|
|
return query, args
|
|
}
|
|
|
|
func queryDueDeliveryCandidates(ctx context.Context, db *sql.DB, query string, args ...any) ([]deliveryCandidate, error) {
|
|
rows, err := db.QueryContext(ctx, query, args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
candidates := make([]deliveryCandidate, 0)
|
|
for rows.Next() {
|
|
var candidate deliveryCandidate
|
|
if err := rows.Scan(&candidate.SourceName, &candidate.AppCode, &candidate.EventID, &candidate.Status, &candidate.RetryCount, &candidate.UpdatedAtMS); err != nil {
|
|
return nil, err
|
|
}
|
|
candidate.Selector = deliverySelector{SourceName: candidate.SourceName, AppCode: candidate.AppCode, EventID: candidate.EventID}.String()
|
|
candidates = append(candidates, candidate)
|
|
}
|
|
return candidates, rows.Err()
|
|
}
|
|
|
|
func normalizeAppCodes(values []string) []string {
|
|
normalized := make([]string, 0, len(values))
|
|
seen := make(map[string]struct{}, len(values))
|
|
for _, value := range values {
|
|
if strings.TrimSpace(value) == "" {
|
|
continue
|
|
}
|
|
value = appcode.Normalize(value)
|
|
if _, exists := seen[value]; exists {
|
|
continue
|
|
}
|
|
seen[value] = struct{}{}
|
|
normalized = append(normalized, value)
|
|
}
|
|
return normalized
|
|
}
|
|
|
|
func normalizeEventIDs(values []string) []string {
|
|
normalized := make([]string, 0, len(values))
|
|
seen := make(map[string]struct{}, len(values))
|
|
for _, value := range values {
|
|
value = strings.TrimSpace(value)
|
|
if value == "" {
|
|
continue
|
|
}
|
|
if _, exists := seen[value]; exists {
|
|
continue
|
|
}
|
|
seen[value] = struct{}{}
|
|
normalized = append(normalized, value)
|
|
}
|
|
return normalized
|
|
}
|
|
|
|
func requireDelivered(ctx context.Context, db *sql.DB, candidate deliveryCandidate) error {
|
|
var status string
|
|
err := db.QueryRowContext(ctx, `
|
|
SELECT status
|
|
FROM notice_delivery_events
|
|
WHERE source_name = ? AND app_code = ? AND source_event_id = ? AND channel = ?`,
|
|
candidate.SourceName,
|
|
candidate.AppCode,
|
|
candidate.EventID,
|
|
c2cChannel,
|
|
).Scan(&status)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if status != "delivered" {
|
|
return fmt.Errorf("delivery status is %q after replay", status)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func walletWorkerOptions(cfg config.Config) walletnotice.WalletNoticeWorkerOptions {
|
|
worker := cfg.WalletNoticeWorker
|
|
return walletnotice.WalletNoticeWorkerOptions{
|
|
WorkerID: "wallet-notice-replay-" + cfg.NodeID,
|
|
BatchSize: worker.BatchSize,
|
|
LockTTL: worker.LockTTL,
|
|
PublishTimeout: worker.PublishTimeout,
|
|
MaxRetryCount: worker.MaxRetryCount,
|
|
InitialBackoff: worker.InitialBackoff,
|
|
MaxBackoff: worker.MaxBackoff,
|
|
}
|
|
}
|
|
|
|
func roomWorkerOptions(cfg config.Config) roomnotice.RoomNoticeWorkerOptions {
|
|
worker := cfg.RoomNoticeWorker
|
|
return roomnotice.RoomNoticeWorkerOptions{
|
|
WorkerID: "room-notice-replay-" + cfg.NodeID,
|
|
BatchSize: worker.BatchSize,
|
|
LockTTL: worker.LockTTL,
|
|
PublishTimeout: worker.PublishTimeout,
|
|
MaxRetryCount: worker.MaxRetryCount,
|
|
InitialBackoff: worker.InitialBackoff,
|
|
MaxBackoff: worker.MaxBackoff,
|
|
}
|
|
}
|
|
|
|
func placeholders(count int) string {
|
|
values := make([]string, count)
|
|
for index := range values {
|
|
values[index] = "?"
|
|
}
|
|
return strings.Join(values, ",")
|
|
}
|