package serverless

import (
	"encoding/json"
	"errors"
	"time"
)

// JSONDuration used to wrap duration into json format
type JSONDuration struct {
	duration time.Duration
}

// MarshalJSON marshal duration to string
func (d JSONDuration) MarshalJSON() ([]byte, error) {
	return json.Marshal(d.duration.Nanoseconds())
}

// UnmarshalJSON unmarshal json value to wrapped duration
func (d *JSONDuration) UnmarshalJSON(b []byte) error {
	var v interface{}
	if err := json.Unmarshal(b, &v); err != nil {
		return err
	}
	switch value := v.(type) {
	case float64:
		d.duration = time.Duration(value)
		return nil
	case string:
		var err error
		d.duration, err = time.ParseDuration(value)
		if err != nil {
			return err
		}
		return nil
	default:
		return errors.New("invalid duration")
	}
}

// refer to https://github.com/pingcap/tiflow/blob/v7.5.6/cdc/api/v2/model.go
type ChangefeedConfig struct {
	Namespace     *string        `json:"namespace,omitempty"`
	ID            *string        `json:"changefeed_id,omitempty"`
	ReplicaConfig *ReplicaConfig `json:"replica_config,omitempty"`
	SinkURI       *string        `json:"sink_uri,omitempty"`
	StartTS       *uint64        `json:"start_ts,omitempty"`
	TargetTS      *uint64        `json:"target_ts,omitempty"`
}

type ReplicaConfig struct {
	MemoryQuota                  *uint64                    `json:"memory_quota,omitempty"`
	BDRMode                      *bool                      `json:"bdr_mode,omitempty"`
	CaseSensitive                *bool                      `json:"case_sensitive,omitempty"`
	CheckGCSafePoint             *bool                      `json:"check_gc_safe_point,omitempty"`
	Consistent                   *ConsistentConfig          `json:"consistent,omitempty"`
	EnableSyncPoint              *bool                      `json:"enable_sync_point,omitempty"`
	Filter                       *FilterConfig              `json:"filter,omitempty"`
	ForceReplicate               *bool                      `json:"force_replicate,omitempty"`
	IgnoreIneligibleTable        *bool                      `json:"ignore_ineligible_table,omitempty"`
	Mounter                      *MounterConfig             `json:"mounter,omitempty"`
	Sink                         *SinkConfig                `json:"sink,omitempty"`
	SyncPointInterval            *JSONDuration              `json:"sync_point_interval,omitempty"`
	SyncPointRetention           *JSONDuration              `json:"sync_point_retention,omitempty"`
	Scheduler                    *ChangefeedSchedulerConfig `json:"scheduler,omitempty"`
	Integrity                    *IntegrityConfig           `json:"integrity,omitempty"`
	SQLMode                      *string                    `json:"sql_mode,omitempty"`
	SyncedStatus                 *SyncedStatusConfig        `json:"synced_status,omitempty"`
	ChangefeedErrorStuckDuration *JSONDuration              `json:"changefeed_error_stuck_duration,omitempty"`
	// version >= v7.6.0
	EnableTableMonitor *bool `json:"enable_table_monitor,omitempty"`
}

type ChangefeedSchedulerConfig struct {
	EnableTableAcrossNodes *bool  `json:"enable_table_across_nodes,omitempty"`
	RegionThreshold        *int32 `json:"region_threshold,omitempty"`
	WriteKeyThreshold      *int32 `json:"write_key_threshold,omitempty"`
}

type IntegrityConfig struct {
	IntegrityCheckLevel   *string `json:"integrity_check_level,omitempty"`
	CorruptionHandleLevel *string `json:"corruption_handle_level,omitempty"`
}

type SyncedStatusConfig struct {
	SyncedCheckInterval *int64 `json:"synced_check_interval,omitempty"`
	CheckpointInterval  *int64 `json:"checkpoint_interval,omitempty"`
}
type ConsistentConfig struct {
	FlushInterval         *uint64                `json:"flush_interval,omitempty"`
	Level                 *string                `json:"level,omitempty"`
	MaxLogSize            *uint64                `json:"max_log_size,omitempty"`
	Storage               *string                `json:"storage,omitempty"`
	UseFileBackend        *bool                  `json:"use_file_backend,omitempty"`
	EncodingWorkerNum     *int                   `json:"encoding_worker_num,omitempty"`
	FlushWorkerNum        *int                   `json:"flush_worker_num,omitempty"`
	Compression           *string                `json:"compression,omitempty"`
	FlushConcurrency      *int                   `json:"flush_concurrency,omitempty"`
	MetaFlushIntervalInMs *int64                 `json:"meta_flush_interval"`
	MemoryUsage           *ConsistentMemoryUsage `json:"memory_usage,omitempty"`
}

type ConsistentMemoryUsage struct {
	MemoryQuotaPercentage *uint64 `json:"memory_quota_percentage,omitempty"`
}

type FilterConfig struct {
	EventFilters     []*EventFilter `json:"event_filters,omitempty"`
	IgnoreTxnStartTS []*uint64      `json:"ignore_txn_start_ts,omitempty"`
	Rules            []string       `json:"rules,omitempty"`
}

type EventFilter struct {
	IgnoreDeleteValueExpr    *string  `json:"ignore_delete_value_expr,omitempty"`
	IgnoreEvent              []string `json:"ignore_event,omitempty"`
	IgnoreInsertValueExpr    *string  `json:"ignore_insert_value_expr,omitempty"`
	IgnoreSQL                []string `json:"ignore_sql,omitempty"`
	IgnoreUpdateNewValueExpr *string  `json:"ignore_update_new_value_expr,omitempty"`
	IgnoreUpdateOldValueExpr *string  `json:"ignore_update_old_value_expr,omitempty"`
	Matcher                  []string `json:"matcher,omitempty"`
}

type MounterConfig struct {
	WorkerNum *int `json:"worker_num,omitempty"`
}

type SinkConfig struct {
	Protocol                         *string             `json:"protocol,omitempty"`
	SchemaRegistry                   *string             `json:"schema_registry,omitempty"`
	CSV                              *CSVConfig          `json:"csv,omitempty"`
	Dispatchers                      []*DispatcherConfig `json:"dispatchers,omitempty"`
	ColumnSelectors                  []*ColumnSelector   `json:"column_selectors,omitempty"`
	TransactionAtomicity             *string             `json:"transaction_atomicity,omitempty"`
	EncoderConcurrency               *int                `json:"encoder_concurrency,omitempty"`
	Terminator                       *string             `json:"terminator,omitempty"`
	DateSeparator                    *string             `json:"date_separator,omitempty"`
	EnablePartitionSeparator         *bool               `json:"enable_partition_separator,omitempty"`
	FileIndexWidth                   *int                `json:"file_index_width,omitempty"`
	EnableKafkaSinkV2                *bool               `json:"enable_kafka_sink_v2,omitempty"`
	OnlyOutputUpdatedColumns         *bool               `json:"only_output_updated_columns,omitempty"`
	DeleteOnlyOutputHandleKeyColumns *bool               `json:"delete_only_output_handle_key_columns,omitempty"`
	ContentCompatible                *bool               `json:"content_compatible,omitempty"`
	SafeMode                         *bool               `json:"safe_mode,omitempty"`
	KafkaConfig                      *KafkaConfig        `json:"kafka_config,omitempty"`
	PulsarConfig                     *PulsarConfig       `json:"pulsar_config,omitempty"`
	MySQLConfig                      *MySQLConfig        `json:"mysql_config,omitempty"`
	CloudStorageConfig               *CloudStorageConfig `json:"cloud_storage_config,omitempty"`
	AdvanceTimeoutInSec              *uint               `json:"advance_timeout,omitempty"`
	SendBootstrapIntervalInSec       *int64              `json:"send_bootstrap_interval_in_sec,omitempty"`
	SendBootstrapInMsgCount          *int32              `json:"send_bootstrap_in_msg_count,omitempty"`
	SendBootstrapToAllPartition      *bool               `json:"send_bootstrap_to_all_partition,omitempty"`
	SendAllBootstrapAtStart          *bool               `json:"send_all_bootstrap_at_start,omitempty"`
	OpenProtocolConfig               *OpenProtocolConfig `json:"open,omitempty"`
	// available since v8.0
	DebeziumDisableSchema *bool `json:"debezium_disable_schema,omitempty"`
	// available since v8.0
	DebeziumConfig *DebeziumConfig `json:"debezium,omitempty"`
}

type ColumnSelector struct {
	Columns []string `json:"columns,omitempty"`
	Matcher []string `json:"matcher,omitempty"`
}

type CSVConfig struct {
	Delimiter            *string `json:"delimiter,omitempty"`
	IncludeCommitTS      *bool   `json:"include_commit_ts,omitempty"`
	Null                 *string `json:"null,omitempty"`
	Quote                *string `json:"quote,omitempty"`
	BinaryEncodingMethod *string `json:"binary_encoding_method,omitempty"`
}

type DispatcherConfig struct {
	Matcher   []string `json:"matcher,omitempty"`
	Partition *string  `json:"partition,omitempty"`
	Topic     *string  `json:"topic,omitempty"`
	Index     *string  `json:"index,omitempty"`
	Columns   []string `json:"columns,omitempty"`
}

type CloudStorageConfig struct {
	WorkerCount          *int    `json:"worker_count,omitempty"`
	FlushInterval        *string `json:"flush_interval,omitempty"`
	FileSize             *int    `json:"file_size,omitempty"`
	OutputColumnID       *bool   `json:"output_column_id,omitempty"`
	FileExpirationDays   *int    `json:"file_expiration_days,omitempty"`
	FileCleanupCronSpec  *string `json:"file_cleanup_cron_spec,omitempty"`
	FlushConcurrency     *int    `json:"flush_concurrency,omitempty"`
	OutputRawChangeEvent *bool   `json:"output_raw_change_event,omitempty"`
}

type OpenProtocolConfig struct {
	OutputOldValue *bool `json:"output_old_value,omitempty"`
}

type DebeziumConfig struct {
	OutputOldValue *bool `json:"output_old_value,omitempty"`
}

type KafkaConfig struct {
	PartitionNum                 *int32                    `json:"partition_num,omitempty"`
	ReplicationFactor            *int16                    `json:"replication_factor,omitempty"`
	KafkaVersion                 *string                   `json:"kafka_version,omitempty"`
	MaxMessageBytes              *int                      `json:"max_message_bytes,omitempty"`
	Compression                  *string                   `json:"compression,omitempty"`
	KafkaClientID                *string                   `json:"kafka_client_id,omitempty"`
	AutoCreateTopic              *bool                     `json:"auto_create_topic,omitempty"`
	DialTimeout                  *string                   `json:"dial_timeout,omitempty"`
	WriteTimeout                 *string                   `json:"write_timeout,omitempty"`
	ReadTimeout                  *string                   `json:"read_timeout,omitempty"`
	RequiredAcks                 *int                      `json:"required_acks,omitempty"`
	SASLUser                     *string                   `json:"sasl_user,omitempty"`
	SASLPassword                 *string                   `json:"sasl_password,omitempty"`
	SASLMechanism                *string                   `json:"sasl_mechanism,omitempty"`
	SASLGssAPIAuthType           *string                   `json:"sasl_gssapi_auth_type,omitempty"`
	SASLGssAPIKeytabPath         *string                   `json:"sasl_gssapi_keytab_path,omitempty"`
	SASLGssAPIKerberosConfigPath *string                   `json:"sasl_gssapi_kerberos_config_path,omitempty"`
	SASLGssAPIServiceName        *string                   `json:"sasl_gssapi_service_name,omitempty"`
	SASLGssAPIUser               *string                   `json:"sasl_gssapi_user,omitempty"`
	SASLGssAPIPassword           *string                   `json:"sasl_gssapi_password,omitempty"`
	SASLGssAPIRealm              *string                   `json:"sasl_gssapi_realm,omitempty"`
	SASLGssAPIDisablePafxfast    *bool                     `json:"sasl_gssapi_disable_pafxfast,omitempty"`
	SASLOAuthClientID            *string                   `json:"sasl_oauth_client_id,omitempty"`
	SASLOAuthClientSecret        *string                   `json:"sasl_oauth_client_secret,omitempty"`
	SASLOAuthTokenURL            *string                   `json:"sasl_oauth_token_url,omitempty"`
	SASLOAuthScopes              []string                  `json:"sasl_oauth_scopes,omitempty"`
	SASLOAuthGrantType           *string                   `json:"sasl_oauth_grant_type,omitempty"`
	SASLOAuthAudience            *string                   `json:"sasl_oauth_audience,omitempty"`
	EnableTLS                    *bool                     `json:"enable_tls,omitempty"`
	CA                           *string                   `json:"ca,omitempty"`
	Cert                         *string                   `json:"cert,omitempty"`
	Key                          *string                   `json:"key,omitempty"`
	InsecureSkipVerify           *bool                     `json:"insecure_skip_verify,omitempty"`
	CodecConfig                  *CodecConfig              `json:"codec_config,omitempty"`
	LargeMessageHandle           *LargeMessageHandleConfig `json:"large_message_handle,omitempty"`
	GlueSchemaRegistryConfig     *GlueSchemaRegistryConfig `json:"glue_schema_registry_config,omitempty"`
	OutputRawChangeEvent         *bool                     `json:"output_raw_change_event,omitempty"`
}

type PulsarConfig struct {
	TLSKeyFilePath          *string       `json:"tls-certificate-path,omitempty"`
	TLSCertificateFile      *string       `json:"tls-private-key-path,omitempty"`
	TLSTrustCertsFilePath   *string       `json:"tls-trust-certs-file-path,omitempty"`
	PulsarProducerCacheSize *int32        `json:"pulsar-producer-cache-size,omitempty"`
	PulsarVersion           *string       `json:"pulsar-version,omitempty"`
	CompressionType         *string       `json:"compression-type,omitempty"`
	AuthenticationToken     *string       `json:"authentication-token,omitempty"`
	ConnectionTimeout       *int          `json:"connection-timeout,omitempty"`
	OperationTimeout        *int          `json:"operation-timeout,omitempty"`
	BatchingMaxMessages     *uint         `json:"batching-max-messages,omitempty"`
	BatchingMaxPublishDelay *int          `json:"batching-max-publish-delay,omitempty"`
	SendTimeout             *int          `json:"send-timeout,omitempty"`
	TokenFromFile           *string       `json:"token-from-file,omitempty"`
	BasicUserName           *string       `json:"basic-user-name,omitempty"`
	BasicPassword           *string       `json:"basic-password,omitempty"`
	AuthTLSCertificatePath  *string       `json:"auth-tls-certificate-path,omitempty"`
	AuthTLSPrivateKeyPath   *string       `json:"auth-tls-private-key-path,omitempty"`
	OAuth2                  *PulsarOAuth2 `json:"oauth2,omitempty"`
	OutputRawChangeEvent    *bool         `json:"output-raw-change-event,omitempty"`
}

type PulsarOAuth2 struct {
	OAuth2IssuerURL  string `json:"oauth2-issuer-url,omitempty"`
	OAuth2Audience   string `json:"oauth2-audience,omitempty"`
	OAuth2PrivateKey string `json:"oauth2-private-key,omitempty"`
	OAuth2ClientID   string `json:"oauth2-client-id,omitempty"`
	OAuth2Scope      string `json:"oauth2-scope,omitempty"`
}

type MySQLConfig struct {
	WorkerCount                  *int    `json:"worker_count,omitempty"`
	MaxTxnRow                    *int    `json:"max_txn_row,omitempty"`
	MaxMultiUpdateRowSize        *int    `json:"max_multi_update_row_size,omitempty"`
	MaxMultiUpdateRowCount       *int    `json:"max_multi_update_row_count,omitempty"`
	TiDBTxnMode                  *string `json:"tidb_txn_mode,omitempty"`
	SSLCa                        *string `json:"ssl_ca,omitempty"`
	SSLCert                      *string `json:"ssl_cert,omitempty"`
	SSLKey                       *string `json:"ssl_key,omitempty"`
	TimeZone                     *string `json:"time_zone,omitempty"`
	WriteTimeout                 *string `json:"write_timeout,omitempty"`
	ReadTimeout                  *string `json:"read_timeout,omitempty"`
	Timeout                      *string `json:"timeout,omitempty"`
	EnableBatchDML               *bool   `json:"enable_batch_dml,omitempty"`
	EnableMultiStatement         *bool   `json:"enable_multi_statement,omitempty"`
	EnableCachePreparedStatement *bool   `json:"enable_cache_prepared_statement,omitempty"`
}

type CodecConfig struct {
	EnableTidbExtension            *bool   `json:"enable_tidb_extension,omitempty"`
	MaxBatchSize                   *int32  `json:"max_batch_size,omitempty"`
	AvroEnableWatermark            *bool   `json:"avro_enable_watermark,omitempty"`
	AvroDecimalHandlingMode        *string `json:"avro_decimal_handling_mode,omitempty"`
	AvroBigintUnsignedHandlingMode *string `json:"avro_bigint_unsigned_handling_mode,omitempty"`
	EncodingFormat                 *string `json:"encoding_format,omitempty"`
}

type LargeMessageHandleConfig struct {
	LargeMessageHandleOption      *string `json:"large_message_handle_option,omitempty"`
	LargeMessageHandleCompression *string `json:"large_message_handle_compression,omitempty"`
	ClaimCheckStorageUri          *string `json:"claim_check_storage_uri,omitempty"`
	ClaimCheckRawValue            *bool   `json:"claim_check_raw_value,omitempty"`
}

type GlueSchemaRegistryConfig struct {
	RegistryName    *string `json:"registry_name,omitempty"`
	Region          *string `json:"region,omitempty"`
	AccessKey       *string `json:"access_key,omitempty"`
	SecretAccessKey *string `json:"secret_access_key,omitempty"`
	Token           *string `json:"token,omitempty"`
}

type AdminJobType int

type ChangefeedState string

const (
	StateNormal   ChangefeedState = "normal"
	StatePending  ChangefeedState = "pending"
	StateFailed   ChangefeedState = "failed"
	StateStopped  ChangefeedState = "stopped"
	StateRemoved  ChangefeedState = "removed"
	StateFinished ChangefeedState = "finished"
	StateWarning  ChangefeedState = "warning"
	// StateUnInitialized is used for the changefeed that has not been initialized
	// it only exists in memory for a short time and will not be persisted to storage
	StateUnInitialized ChangefeedState = ""
)

type Changefeed struct {
	UpstreamID     *uint64              `json:"upstream_id,omitempty"`
	Namespace      *string              `json:"namespace,omitempty"`
	ID             *string              `json:"id,omitempty"`
	SinkURI        *string              `json:"sink_uri,omitempty"`
	Config         *ReplicaConfig       `json:"config,omitempty"`
	CreatorVersion *string              `json:"creator_version,omitempty"`
	CreateTime     *string              `json:"create_time,omitempty"`
	StartTS        *uint64              `json:"start_ts,omitempty"`
	TargetTS       *uint64              `json:"target_ts,omitempty"`
	CheckpointTS   *uint64              `json:"checkpoint_ts,omitempty"`
	CheckpointTime *string              `json:"checkpoint_time,omitempty"`
	ResolvedTS     *uint64              `json:"resolved_ts,omitempty"`
	State          ChangefeedState      `json:"state,omitempty"`
	Error          *RunningError        `json:"error,omitempty"`
	AdminJobType   AdminJobType         `json:"admin_job_type,omitempty"`
	TaskStatus     []*CaptureTaskStatus `json:"task_status,omitempty"`
}

type RunningError struct {
	Time    *time.Time `json:"time"`
	Addr    *string    `json:"addr,omitempty"`
	Code    *string    `json:"code,omitempty"`
	Message *string    `json:"message,omitempty"`
}

type CaptureTaskStatus struct {
	CaptureID *string `json:"capture_id,omitempty"`
	// Table list, containing tables that processor should process
	Tables []int64 `json:"table_ids,omitempty"`
}

type ListChangefeedsResponse struct {
	Total *int                    `json:"total"`
	Items []*ChangefeedCommonInfo `json:"items"`
}

type ChangefeedCommonInfo struct {
	UpstreamID     *uint64         `json:"upstream_id,omitempty"`
	Namespace      *string         `json:"namespace,omitempty"`
	ID             *string         `json:"id"`
	State          ChangefeedState `json:"state"`
	CheckpointTSO  *uint64         `json:"checkpoint_tso"`
	CheckpointTime *string         `json:"checkpoint_time"`
	Error          *RunningError   `json:"error,omitempty"`
}

type ResumeChangefeedConfig struct {
	OverwriteCheckpointTs *uint64 `json:"overwrite_checkpoint_ts,omitempty"`
}