package app

import (
	"context"
	"errors"
	"fmt"
	"os"
	"strings"
	"time"

	"github.com/spf13/cobra"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/rpc"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
)

type cdcCLIOptions struct {
	address string
	output  string
	timeout time.Duration
}

func init() {
	opts := &cdcCLIOptions{}
	cdcCmd := &cobra.Command{
		Use:   "cdc",
		Short: "operate and migrate CDC changefeeds through the ops gRPC API",
	}
	cdcCmd.PersistentFlags().StringVar(
		&opts.address,
		"address",
		"127.0.0.1:10000",
		"serverless-service ops gRPC address",
	)
	cdcCmd.PersistentFlags().StringVar(&opts.output, "output", "table", "output format: table or json")
	cdcCmd.PersistentFlags().DurationVar(&opts.timeout, "timeout", 10*time.Minute, "command timeout")
	cdcCmd.AddCommand(
		newCDCListCommand(opts),
		newCDCGetCommand(opts),
		newCDCPrecheckCommand(opts),
		newCDCMigrateCommand(opts),
		newCDCCompareCommand(opts),
		newCDCPauseCommand(opts),
		newCDCResumeCommand(opts),
		newCDCDeleteCommand(opts),
	)
	rootCmd.AddCommand(cdcCmd)
}

func withCDCClient(
	cmd *cobra.Command,
	opts *cdcCLIOptions,
	fn func(context.Context, rpc.ChangefeedOpsServiceClient) (proto.Message, error),
) error {
	ctx, cancel := context.WithTimeout(cmd.Context(), opts.timeout)
	defer cancel()
	conn, err := grpc.DialContext(ctx, opts.address,
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithBlock(),
	)
	if err != nil {
		return fmt.Errorf("dial ops gRPC %s: %w", opts.address, err)
	}
	defer func() { _ = conn.Close() }()
	resp, err := fn(ctx, rpc.NewChangefeedOpsServiceClient(conn))
	if err != nil {
		return err
	}
	return printCDCResponse(opts.output, resp)
}

func newCDCListCommand(opts *cdcCLIOptions) *cobra.Command {
	var clusterID, sinkType string
	var limit int32
	cmd := &cobra.Command{
		Use:   "list",
		Short: "list legacy and newarch changefeeds of a cluster",
		RunE: func(cmd *cobra.Command, _ []string) error {
			call := func(ctx context.Context, client rpc.ChangefeedOpsServiceClient) (proto.Message, error) {
				return client.ListClusterChangefeeds(ctx, &rpc.ListClusterChangefeedsRequest{
					ClusterId: clusterID, SinkType: sinkType, Limit: limit,
				})
			}
			return withCDCClient(cmd, opts, call)
		},
	}
	cmd.Flags().StringVar(&clusterID, "cluster-id", "", "user cluster ID")
	cmd.Flags().StringVar(&sinkType, "sink-type", "", "optional sink type filter")
	cmd.Flags().Int32Var(&limit, "limit", 100, "maximum number of changefeeds")
	_ = cmd.MarkFlagRequired("cluster-id")
	return cmd
}

func newCDCGetCommand(opts *cdcCLIOptions) *cobra.Command {
	var clusterID, changefeedID string
	cmd := &cobra.Command{
		Use:   "get",
		Short: "get one changefeed",
		RunE: func(cmd *cobra.Command, _ []string) error {
			call := func(ctx context.Context, client rpc.ChangefeedOpsServiceClient) (proto.Message, error) {
				return client.GetChangefeed(ctx, &rpc.GetChangefeedRequest{
					ClusterId: clusterID, ChangefeedId: changefeedID,
				})
			}
			return withCDCClient(cmd, opts, call)
		},
	}
	bindChangefeedFlags(cmd, &clusterID, &changefeedID)
	return cmd
}

func newCDCPrecheckCommand(opts *cdcCLIOptions) *cobra.Command {
	var clusterID, changefeedID, displayName string
	var rcu uint32
	cmd := &cobra.Command{
		Use:   "precheck",
		Short: "validate a legacy-to-newarch migration without changing state",
		RunE: func(cmd *cobra.Command, _ []string) error {
			var passed bool
			call := func(ctx context.Context, client rpc.ChangefeedOpsServiceClient) (proto.Message, error) {
				resp, err := client.PrecheckMigrateChangefeed(ctx, &rpc.PrecheckMigrateChangefeedRequest{
					ClusterId: clusterID, ChangefeedId: changefeedID, Rcu: rcu, DisplayName: displayName,
				})
				if resp != nil {
					passed = resp.GetPassed()
				}
				return resp, err
			}
			err := withCDCClient(cmd, opts, call)
			if err != nil {
				return err
			}
			if !passed {
				return errors.New("migration precheck failed")
			}
			return nil
		},
	}
	bindMigrationFlags(cmd, &clusterID, &changefeedID, &displayName, &rcu)
	return cmd
}

func newCDCMigrateCommand(opts *cdcCLIOptions) *cobra.Command {
	var clusterID, changefeedID, displayName string
	var rcu uint32
	var wait bool
	cmd := &cobra.Command{
		Use:   "migrate",
		Short: "pause a healthy legacy changefeed and create its newarch replacement",
		RunE: func(cmd *cobra.Command, _ []string) error {
			call := func(ctx context.Context, client rpc.ChangefeedOpsServiceClient) (proto.Message, error) {
				resp, err := client.MigrateChangefeed(ctx, &rpc.MigrateChangefeedRequest{
					ClusterId: clusterID, ChangefeedId: changefeedID, Rcu: rcu, DisplayName: displayName,
				})
				if err != nil {
					return resp, err
				}
				if resp == nil || resp.GetTargetChangefeedId() == "" {
					return nil, errors.New(
						"migration returned no target changefeed ID; list cluster changefeeds before resuming the source",
					)
				}
				if !wait {
					return resp, nil
				}
				waited, waitErr := waitForChangefeed(ctx, client, clusterID, resp.GetTargetChangefeedId(), false)
				if info, ok := waited.(*rpc.ChangefeedInfo); ok && info != nil {
					resp.State = info.GetState()
				}
				return resp, waitErr
			}
			return withCDCClient(cmd, opts, call)
		},
	}
	bindMigrationFlags(cmd, &clusterID, &changefeedID, &displayName, &rcu)
	cmd.Flags().BoolVar(&wait, "wait", true, "wait until the target becomes running or failed")
	return cmd
}

func newCDCCompareCommand(opts *cdcCLIOptions) *cobra.Command {
	var clusterID, sourceID, targetID string
	cmd := &cobra.Command{
		Use:   "compare",
		Short: "compare legacy and newarch kernel state and configuration",
		RunE: func(cmd *cobra.Command, _ []string) error {
			var passed bool
			call := func(ctx context.Context, client rpc.ChangefeedOpsServiceClient) (proto.Message, error) {
				resp, err := client.CompareChangefeeds(ctx, &rpc.CompareChangefeedsRequest{
					ClusterId: clusterID, SourceChangefeedId: sourceID, TargetChangefeedId: targetID,
				})
				if resp != nil {
					passed = resp.GetPassed()
				}
				return resp, err
			}
			err := withCDCClient(cmd, opts, call)
			if err != nil {
				return err
			}
			if !passed {
				return errors.New("changefeed comparison failed")
			}
			return nil
		},
	}
	cmd.Flags().StringVar(&clusterID, "cluster-id", "", "user cluster ID")
	cmd.Flags().StringVar(&sourceID, "source-changefeed-id", "", "legacy source changefeed ID")
	cmd.Flags().StringVar(&targetID, "target-changefeed-id", "", "newarch target changefeed ID")
	_ = cmd.MarkFlagRequired("cluster-id")
	_ = cmd.MarkFlagRequired("source-changefeed-id")
	_ = cmd.MarkFlagRequired("target-changefeed-id")
	return cmd
}

func newCDCPauseCommand(opts *cdcCLIOptions) *cobra.Command {
	var clusterID, changefeedID, expectedArch string
	cmd := &cobra.Command{
		Use:   "pause",
		Short: "pause a changefeed",
		RunE: func(cmd *cobra.Command, _ []string) error {
			call := func(ctx context.Context, client rpc.ChangefeedOpsServiceClient) (proto.Message, error) {
				return client.PauseChangefeed(ctx, &rpc.PauseChangefeedRequest{
					ClusterId: clusterID, ChangefeedId: changefeedID, ExpectedArch: expectedArch,
				})
			}
			return withCDCClient(cmd, opts, call)
		},
	}
	bindChangefeedMutationFlags(cmd, &clusterID, &changefeedID, &expectedArch)
	return cmd
}

func newCDCResumeCommand(opts *cdcCLIOptions) *cobra.Command {
	var clusterID, changefeedID, expectedArch string
	cmd := &cobra.Command{
		Use:   "resume",
		Short: "resume a paused or failed changefeed",
		RunE: func(cmd *cobra.Command, _ []string) error {
			call := func(ctx context.Context, client rpc.ChangefeedOpsServiceClient) (proto.Message, error) {
				return client.ResumeChangefeed(ctx, &rpc.ResumeChangefeedRequest{
					ClusterId: clusterID, ChangefeedId: changefeedID, ExpectedArch: expectedArch,
				})
			}
			return withCDCClient(cmd, opts, call)
		},
	}
	bindChangefeedMutationFlags(cmd, &clusterID, &changefeedID, &expectedArch)
	return cmd
}

func newCDCDeleteCommand(opts *cdcCLIOptions) *cobra.Command {
	var clusterID, changefeedID, expectedArch string
	var wait, yes bool
	cmd := &cobra.Command{
		Use:   "delete",
		Short: "delete a legacy or newarch changefeed",
		RunE: func(cmd *cobra.Command, _ []string) error {
			if !yes {
				return errors.New("delete requires --yes")
			}
			call := func(ctx context.Context, client rpc.ChangefeedOpsServiceClient) (proto.Message, error) {
				resp, err := client.DeleteChangefeed(ctx, &rpc.DeleteChangefeedRequest{
					ClusterId: clusterID, ChangefeedId: changefeedID, ExpectedArch: expectedArch,
				})
				if err != nil || !wait {
					return resp, err
				}
				return waitForChangefeed(ctx, client, clusterID, changefeedID, true)
			}
			return withCDCClient(cmd, opts, call)
		},
	}
	bindChangefeedMutationFlags(cmd, &clusterID, &changefeedID, &expectedArch)
	cmd.Flags().BoolVar(&wait, "wait", true, "wait until the changefeed is deleted")
	cmd.Flags().BoolVar(&yes, "yes", false, "confirm the destructive operation")
	return cmd
}

func bindChangefeedFlags(cmd *cobra.Command, clusterID, changefeedID *string) {
	cmd.Flags().StringVar(clusterID, "cluster-id", "", "user cluster ID")
	cmd.Flags().StringVar(changefeedID, "changefeed-id", "", "changefeed ID")
	_ = cmd.MarkFlagRequired("cluster-id")
	_ = cmd.MarkFlagRequired("changefeed-id")
}

func bindChangefeedMutationFlags(cmd *cobra.Command, clusterID, changefeedID, expectedArch *string) {
	bindChangefeedFlags(cmd, clusterID, changefeedID)
	cmd.Flags().StringVar(expectedArch, "expected-arch", "", "safety guard: replication_worker or ticdc_newarch")
}

func bindMigrationFlags(
	cmd *cobra.Command,
	clusterID, changefeedID, displayName *string,
	rcu *uint32,
) {
	bindChangefeedFlags(cmd, clusterID, changefeedID)
	cmd.Flags().Uint32Var(rcu, "rcu", 0, "target RCU; defaults to the source RCU or 3")
	cmd.Flags().StringVar(displayName, "display-name", "", "optional target display name; defaults to the source name")
}

func waitForChangefeed(
	ctx context.Context,
	client rpc.ChangefeedOpsServiceClient,
	clusterID, changefeedID string,
	wantDeleted bool,
) (proto.Message, error) {
	ticker := time.NewTicker(3 * time.Second)
	defer ticker.Stop()
	for {
		resp, err := client.GetChangefeed(ctx, &rpc.GetChangefeedRequest{
			ClusterId: clusterID, ChangefeedId: changefeedID,
		})
		if err != nil {
			if wantDeleted && isChangefeedNotFound(err) {
				return &rpc.ChangefeedInfo{ClusterId: clusterID, ChangefeedId: changefeedID, State: "deleted"}, nil
			}
			return nil, err
		}
		if wantDeleted {
			if resp.GetState() == "deleted" {
				return resp, nil
			}
		} else {
			switch resp.GetState() {
			case "running":
				return resp, nil
			case "create_failed", "failed", "running_failed":
				return resp, fmt.Errorf("target changefeed entered state %s: %s", resp.GetState(), resp.GetError())
			}
		}
		select {
		case <-ctx.Done():
			return resp, ctx.Err()
		case <-ticker.C:
		}
	}
}

func isChangefeedNotFound(err error) bool {
	if status.Code(err) == codes.NotFound {
		return true
	}
	// Older ops servers returned bizerr.ErrCodeRecordNotFound as gRPC
	// Unknown. Keep delete --wait compatible with those deployed versions.
	return status.Code(err) == codes.Unknown && strings.Contains(err.Error(), "code=49900004")
}

func printCDCResponse(output string, message proto.Message) error {
	if message == nil {
		return nil
	}
	if strings.EqualFold(output, "json") {
		data, err := protojson.MarshalOptions{Indent: "  ", UseProtoNames: true}.Marshal(message)
		if err != nil {
			return err
		}
		_, err = fmt.Fprintln(os.Stdout, string(data))
		return err
	}
	if !strings.EqualFold(output, "table") {
		return fmt.Errorf("unsupported output format %q", output)
	}
	switch resp := message.(type) {
	case *rpc.PrecheckMigrateChangefeedResponse:
		if err := printChecks(resp.GetChecks()); err != nil {
			return err
		}
		return writeCDCOutput("RESULT\t%t\n", resp.GetPassed())
	case *rpc.CompareChangefeedsResponse:
		if err := printChecks(resp.GetChecks()); err != nil {
			return err
		}
		return writeCDCOutput(
			"SOURCE_CHECKPOINT_TSO\t%d\nTARGET_START_TSO\t%d\n"+
				"TARGET_CHECKPOINT_TSO\t%d\nRESULT\t%t\n",
			resp.GetSourceCheckpointTso(), resp.GetTargetStartTso(), resp.GetTargetCheckpointTso(), resp.GetPassed())
	case *rpc.ListClusterChangefeedsResponse:
		if err := writeCDCOutput("CHANGEFEED_ID\tARCH\tSTATE\tCHECKPOINT_TSO\tDISPLAY_NAME\n"); err != nil {
			return err
		}
		for _, item := range resp.GetItems() {
			if err := writeCDCOutput(
				"%s\t%s\t%s\t%d\t%s\n",
				item.GetChangefeedId(),
				item.GetArch(),
				item.GetState(),
				item.GetCheckpointTso(),
				item.GetDisplayName(),
			); err != nil {
				return err
			}
		}
		return nil
	case *rpc.MigrateChangefeedResponse:
		return writeCDCOutput(
			"SOURCE_CHANGEFEED_ID\t%s\nTARGET_CHANGEFEED_ID\t%s\n"+
				"SOURCE_CHECKPOINT_TSO\t%d\nTARGET_START_TSO\t%d\nTARGET_STATE\t%s\n",
			resp.GetSourceChangefeedId(), resp.GetTargetChangefeedId(), resp.GetSourceCheckpointTso(), resp.GetTargetStartTso(), resp.GetState())
	case *rpc.ChangefeedInfo:
		return writeCDCOutput(
			"CHANGEFEED_ID\t%s\nARCH\t%s\nSTATE\t%s\nSTART_TSO\t%d\n"+
				"CHECKPOINT_TSO\t%d\nERROR\t%s\n",
			resp.GetChangefeedId(), resp.GetArch(), resp.GetState(), resp.GetStartTso(), resp.GetCheckpointTso(), resp.GetError())
	default:
		data, err := protojson.MarshalOptions{Indent: "  ", UseProtoNames: true}.Marshal(message)
		if err != nil {
			return err
		}
		_, err = fmt.Fprintln(os.Stdout, string(data))
		return err
	}
}

func printChecks(checks []*rpc.MigrationCheckResult) error {
	if err := writeCDCOutput("CHECK\tRESULT\tMESSAGE\n"); err != nil {
		return err
	}
	for _, check := range checks {
		result := "PASS"
		if !check.GetPassed() {
			result = "FAIL"
		}
		if err := writeCDCOutput("%s\t%s\t%s\n", check.GetName(), result, check.GetMessage()); err != nil {
			return err
		}
	}
	return nil
}

func writeCDCOutput(format string, args ...any) error {
	_, err := fmt.Fprintf(os.Stdout, format, args...)
	return err
}