// 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, ",")
}