// Copyright 2025 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package canal

import (
	"context"
	"crypto/sha256"
	"sort"
	"strconv"
	"time"

	"github.com/goccy/go-json"
	"github.com/mailru/easyjson/jwriter"
	"github.com/pingcap/log"
	commonType "github.com/pingcap/ticdc/pkg/common"
	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
	"github.com/pingcap/ticdc/pkg/errors"
	"github.com/pingcap/ticdc/pkg/sink/codec/common"
	"github.com/pingcap/ticdc/pkg/sink/kafka/claimcheck"
	"go.uber.org/zap"
	"golang.org/x/text/encoding/charmap"
)

var bytesDecoder = charmap.ISO8859_1.NewDecoder()

type checksumEntry struct {
	Name  string `json:"name"`
	Value any    `json:"value"`
}

// TODO: we need to reorg this code later, including use util.jsonWriter and other unreasonable code
func fillColumns(
	valueMap map[int64]optionalString,
	tableInfo *commonType.TableInfo,
	onlyHandleKeyColumn bool,
	out *jwriter.Writer,
	columnSelector commonEvent.Selector,
) error {
	if len(tableInfo.GetColumns()) == 0 {
		out.RawString("null")
		return nil
	}
	out.RawByte('[')
	out.RawByte('{')
	isFirst := true
	for _, col := range tableInfo.GetColumns() {
		if col != nil {
			if col.IsVirtualGenerated() || !columnSelector.Select(col) {
				continue
			}
			colID := col.ID
			if onlyHandleKeyColumn && !tableInfo.IsHandleKey(col.ID) {
				continue
			}
			if isFirst {
				isFirst = false
			} else {
				out.RawByte(',')
			}
			out.String(col.Name.O)
			out.RawByte(':')
			optionalStr := valueMap[colID]
			if optionalStr.isNull {
				out.RawString("null")
			} else {
				out.String(optionalStr.value)
			}
		}
	}
	out.RawByte('}')
	out.RawByte(']')
	return nil
}

func fillUpdateColumns(
	newValueMap, oldValueMap map[int64]optionalString,
	tableInfo *commonType.TableInfo,
	onlyHandleKeyColumn bool,
	onlyOutputUpdatedColumn bool,
	out *jwriter.Writer,
	columnSelector commonEvent.Selector,
) error {
	if len(tableInfo.GetColumns()) == 0 {
		out.RawString("null")
		return nil
	}
	out.RawByte('[')
	out.RawByte('{')
	isFirst := true
	for _, col := range tableInfo.GetColumns() {
		if col != nil {
			if col.IsVirtualGenerated() || !columnSelector.Select(col) {
				continue
			}
			colID := col.ID
			// column equal, do not output it
			if onlyOutputUpdatedColumn && newValueMap[colID] == oldValueMap[colID] {
				continue
			}
			if onlyHandleKeyColumn && !tableInfo.IsHandleKey(col.ID) {
				continue
			}
			if isFirst {
				isFirst = false
			} else {
				out.RawByte(',')
			}
			out.String(col.Name.O)
			out.RawByte(':')
			optionalStr := oldValueMap[colID]
			if optionalStr.isNull {
				out.RawString("null")
			} else {
				out.String(optionalStr.value)
			}
		}
	}
	out.RawByte('}')
	out.RawByte(']')
	return nil
}

func newJSONMessageForDML(
	e *commonEvent.RowEvent,
	config *common.Config,
	messageTooLarge bool,
	claimCheckFileName string,
) ([]byte, error) {
	isDelete := e.IsDelete()

	onlyHandleKey := messageTooLarge
	if isDelete && config.DeleteOnlyHandleKeyColumns {
		onlyHandleKey = true
	}
	onlyOutputPKColumns := onlyHandleKey || config.OnlyOutputPKColumns

	columnLen := 0
	for _, col := range e.TableInfo.GetColumns() {
		if col != nil && !col.IsVirtualGenerated() && e.ColumnSelector.Select(col) {
			columnLen += 1
		}
	}
	if columnLen == 0 {
		return nil, errors.ErrCanalEncodeFailed.GenWithStack("not found valid columns for the event")
	}

	mysqlTypeMap := make(map[string]string, columnLen)
	// TODO: use util.JsonWriter
	out := &jwriter.Writer{}
	out.RawByte('{')
	{
		const prefix string = ",\"id\":"
		out.RawString(prefix[1:])
		out.Int64(0) // ignored by both Canal Adapter and Flink
	}
	{
		const prefix string = ",\"database\":"
		out.RawString(prefix)
		out.String(e.TableInfo.GetSchemaName())
	}
	{
		const prefix string = ",\"table\":"
		out.RawString(prefix)
		out.String(e.TableInfo.GetTableName())
	}
	{
		const prefix string = ",\"pkNames\":"
		out.RawString(prefix)
		pkNames := e.PrimaryKeyColumnNames()
		if pkNames == nil {
			out.RawString("null")
		} else {
			out.RawByte('[')
			for v25, v26 := range pkNames {
				if v25 > 0 {
					out.RawByte(',')
				}
				out.String(v26)
			}
			out.RawByte(']')
		}
	}
	{
		const prefix string = ",\"isDdl\":"
		out.RawString(prefix)
		out.Bool(false)
	}
	{
		const prefix string = ",\"type\":"
		out.RawString(prefix)
		out.String(eventTypeString(e))
	}
	{
		const prefix string = ",\"es\":"
		out.RawString(prefix)
		out.Int64(convertToCanalTs(e.CommitTs))
	}
	{
		const prefix string = ",\"ts\":"
		out.RawString(prefix)
		out.Int64(time.Now().UnixMilli()) // ignored by both Canal Adapter and Flink
	}
	{
		const prefix string = ",\"sql\":"
		out.RawString(prefix)
		out.String("")
	}

	valueMap := make(map[int64]optionalString, columnLen)        // colId -> value
	javaTypeMap := make(map[int64]common.JavaSQLType, columnLen) // colId -> javaType

	row := e.GetRows()
	if e.IsDelete() {
		row = e.GetPreRows()
	}
	for idx, col := range e.TableInfo.GetColumns() {
		if col == nil || col.IsVirtualGenerated() || !e.ColumnSelector.Select(col) {
			continue
		}
		value, javaType := formatColumnValue(row, idx, col)
		valueMap[col.ID] = value
		javaTypeMap[col.ID] = javaType
	}
	{
		const prefix string = ",\"sqlType\":"
		out.RawString(prefix)
		emptyColumn := true
		tableInfo := e.TableInfo
		columnInfos := tableInfo.GetColumns()
		for _, col := range columnInfos {
			if col != nil && !col.IsVirtualGenerated() && e.ColumnSelector.Select(col) {
				colID := col.ID
				colName := col.Name.O
				if onlyOutputPKColumns && !tableInfo.IsHandleKey(colID) {
					continue
				}
				if emptyColumn {
					out.RawByte('{')
					emptyColumn = false
				} else {
					out.RawByte(',')
				}

				out.String(colName)
				out.RawByte(':')
				out.Int32(int32(javaTypeMap[colID]))
				mysqlTypeMap[colName] = common.GetMySQLType(col, config.ContentCompatible)
			}
		}
		if emptyColumn {
			out.RawString(`null`)
		} else {
			out.RawByte('}')
		}
	}
	{
		const prefix string = ",\"mysqlType\":"
		out.RawString(prefix)
		if len(mysqlTypeMap) == 0 {
			out.RawString(`null`)
		} else {
			out.RawByte('{')
			isFirst := true
			for typeKey, typeValue := range mysqlTypeMap {
				if isFirst {
					isFirst = false
				} else {
					out.RawByte(',')
				}
				out.String(typeKey)
				out.RawByte(':')
				out.String(typeValue)
			}
			out.RawByte('}')
		}
	}

	if e.IsDelete() {
		out.RawString(",\"old\":null")
		out.RawString(",\"data\":")
		if err := fillColumns(valueMap, e.TableInfo, onlyOutputPKColumns, out, e.ColumnSelector); err != nil {
			return nil, err
		}
	} else if e.IsInsert() {
		out.RawString(",\"old\":null")
		out.RawString(",\"data\":")
		if err := fillColumns(valueMap, e.TableInfo, onlyOutputPKColumns, out, e.ColumnSelector); err != nil {
			return nil, err
		}
	} else if e.IsUpdate() {
		out.RawString(",\"old\":")

		oldValueMap := make(map[int64]optionalString, 0) // colId -> value
		preRow := e.GetPreRows()
		for idx, col := range e.TableInfo.GetColumns() {
			if col == nil || col.IsVirtualGenerated() || !e.ColumnSelector.Select(col) {
				continue
			}
			value, _ := formatColumnValue(preRow, idx, col)
			oldValueMap[col.ID] = value
		}

		if err := fillUpdateColumns(valueMap, oldValueMap, e.TableInfo, onlyOutputPKColumns,
			config.OnlyOutputUpdatedColumns, out, e.ColumnSelector); err != nil {
			return nil, err
		}
		out.RawString(",\"data\":")
		if err := fillColumns(valueMap, e.TableInfo, onlyOutputPKColumns, out, e.ColumnSelector); err != nil {
			return nil, err
		}
	} else {
		log.Panic("unreachable event type", zap.Any("event", e))
	}

	if config.EnableTiDBExtension {
		const prefix string = ",\"_tidb\":"
		out.RawString(prefix)
		out.RawByte('{')
		out.RawString("\"commitTs\":")
		out.Uint64(e.CommitTs)

		// only send handle key may happen in 2 cases:
		// 1. delete event, and set only handle key config. no need to encode `onlyHandleKey` field
		// 2. event larger than the max message size, and enable large message handle to the `handleKeyOnly`, encode `onlyHandleKey` field
		if messageTooLarge {
			if config.LargeMessageHandle.HandleKeyOnly() {
				out.RawByte(',')
				out.RawString("\"onlyHandleKey\":true")
			}
			if config.LargeMessageHandle.EnableClaimCheck() {
				out.RawByte(',')
				out.RawString("\"claimCheckLocation\":")
				out.String(claimCheckFileName)
			}
		}

		// Only when both EnableTiDBExtension and OutputRowKey are true, the rowkey will be output.
		if config.OutputRowKey {
			out.RawByte(',')
			out.RawString("\"rowkey\":")
			out.Base64Bytes(e.Event.RowKey)
		}

		if config.OnlyOutputPKColumns {
			originTs, checksum, err := buildOriginTsAndChecksum(e, valueMap)
			if err != nil {
				return nil, err
			}
			out.RawString(",\"originTs\":")
			out.Uint64(originTs)
			out.RawString(",\"checksum\":")
			out.Base64Bytes(checksum)
		}

		out.RawByte('}')
	}
	out.RawByte('}')

	value, err := out.BuildBytes()
	if err != nil {
		return nil, errors.WrapError(errors.ErrCanalEncodeFailed, err)
	}
	return value, nil
}

func eventTypeString(e *commonEvent.RowEvent) string {
	if e.IsDelete() {
		return "DELETE"
	}
	if e.IsInsert() {
		return "INSERT"
	}
	return "UPDATE"
}

// buildOriginTsAndChecksum derives originTs from the _tidb_origin_ts column in valueMap (new row)
// and a stable checksum over other non-PK columns. For UPDATE, if _tidb_origin_ts is absent or null
// in the post-update row, originTs stays 0; the checksum still captures non-PK data in valueMap.
func buildOriginTsAndChecksum(
	e *commonEvent.RowEvent,
	valueMap map[int64]optionalString,
) (uint64, []byte, error) {
	var (
		originTs uint64
		entries  = make([]checksumEntry, 0, len(valueMap))
	)
	for _, col := range e.TableInfo.GetColumns() {
		if col == nil || col.IsVirtualGenerated() || !e.ColumnSelector.Select(col) {
			continue
		}
		value, ok := valueMap[col.ID]
		if !ok {
			continue
		}
		if col.Name.O == commonEvent.OriginTsColumn {
			if value.isNull {
				continue
			}
			parsedOriginTs, err := strconv.ParseUint(value.value, 10, 64)
			if err != nil {
				return 0, nil, errors.WrapError(errors.ErrCanalEncodeFailed, err)
			}
			originTs = parsedOriginTs
			continue
		}
		if e.TableInfo.IsHandleKey(col.ID) {
			continue
		}

		var checksumValue any
		if !value.isNull {
			checksumValue = value.value
		}
		entries = append(entries, checksumEntry{
			Name:  col.Name.O,
			Value: checksumValue,
		})
	}

	sort.Slice(entries, func(i, j int) bool {
		return entries[i].Name < entries[j].Name
	})
	payload, err := json.Marshal(entries)
	if err != nil {
		return 0, nil, errors.WrapError(errors.ErrCanalEncodeFailed, err)
	}
	sum := sha256.Sum256(payload)
	return originTs, sum[:], nil
}

// JSONRowEventEncoder encodes row event in JSON format
type JSONRowEventEncoder struct {
	messages []*common.Message

	claimCheck *claimcheck.ClaimCheck
	config     *common.Config
}

// NewJSONRowEventEncoder creates a new JSONRowEventEncoder
func NewJSONRowEventEncoder(ctx context.Context, config *common.Config) (common.EventEncoder, error) {
	claimCheck, err := claimcheck.New(ctx, config.LargeMessageHandle, config.ChangefeedID)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return &JSONRowEventEncoder{
		messages:   make([]*common.Message, 0, 1),
		config:     config,
		claimCheck: claimCheck,
	}, nil
}

func (c *JSONRowEventEncoder) newJSONMessageForDDL(e *commonEvent.DDLEvent) canalJSONMessageInterface {
	msg := &JSONMessage{
		ID:            0, // ignored by both Canal Adapter and Flink
		Schema:        e.GetSchemaName(),
		Table:         e.GetTableName(),
		IsDDL:         true,
		EventType:     convertDdlEventType(e.Type).String(),
		ExecutionTime: convertToCanalTs(e.GetCommitTs()),
		BuildTime:     time.Now().UnixMilli(), // timestamp
		Query:         e.Query,
	}

	if !c.config.EnableTiDBExtension {
		return msg
	}

	return &canalJSONMessageWithTiDBExtension{
		JSONMessage: msg,
		Extensions:  &tidbExtension{CommitTs: e.GetCommitTs()},
	}
}

func (c *JSONRowEventEncoder) newJSONMessage4CheckpointEvent(
	ts uint64,
) *canalJSONMessageWithTiDBExtension {
	return &canalJSONMessageWithTiDBExtension{
		JSONMessage: &JSONMessage{
			ID:            0,
			IsDDL:         false,
			EventType:     tidbWaterMarkType,
			ExecutionTime: convertToCanalTs(ts),
			BuildTime:     time.Now().UnixNano() / int64(time.Millisecond), // converts to milliseconds
		},
		Extensions: &tidbExtension{WatermarkTs: ts},
	}
}

// EncodeCheckpointEvent implements the RowEventEncoder interface
func (c *JSONRowEventEncoder) EncodeCheckpointEvent(ts uint64) (*common.Message, error) {
	if !c.config.EnableTiDBExtension {
		return nil, nil
	}

	msg := c.newJSONMessage4CheckpointEvent(ts)
	value, err := json.Marshal(msg)
	if err != nil {
		return nil, errors.WrapError(errors.ErrCanalEncodeFailed, err)
	}

	value, err = common.Compress(
		c.config.ChangefeedID, c.config.LargeMessageHandle.LargeMessageHandleCompression, value,
	)
	if err != nil {
		return nil, errors.Trace(err)
	}

	return common.NewMsg(nil, value), nil
}

// AppendRowChangedEvent implements the interface EventJSONBatchEncoder
func (c *JSONRowEventEncoder) AppendRowChangedEvent(
	ctx context.Context,
	_ string,
	e *commonEvent.RowEvent,
) error {
	value, err := newJSONMessageForDML(e, c.config, false, "")
	if err != nil {
		return errors.Trace(err)
	}

	value, err = common.Compress(
		c.config.ChangefeedID, c.config.LargeMessageHandle.LargeMessageHandleCompression, value,
	)
	if err != nil {
		return errors.Trace(err)
	}

	m := common.NewMsg(nil, value)
	m.Callback = e.Callback
	m.IncRowsCount()

	originLength := m.Length()
	if m.Length() > c.config.MaxMessageBytes {
		// for single message that is longer than max-message-bytes, do not send it.
		if c.config.LargeMessageHandle.Disabled() {
			log.Error("Single message is too large for canal-json",
				zap.Int("maxMessageBytes", c.config.MaxMessageBytes),
				zap.Int("length", originLength),
				zap.Any("table", e.TableInfo.TableName))
			return errors.ErrMessageTooLarge.GenWithStackByArgs(e.TableInfo.GetTableName(), originLength, c.config.MaxMessageBytes)
		}

		if c.config.LargeMessageHandle.HandleKeyOnly() {
			value, err = newJSONMessageForDML(e, c.config, true, "")
			if err != nil {
				return errors.Trace(err)
			}
			value, err = common.Compress(
				c.config.ChangefeedID, c.config.LargeMessageHandle.LargeMessageHandleCompression, value,
			)
			if err != nil {
				return errors.Trace(err)
			}

			m.Value = value
			length := m.Length()
			if length > c.config.MaxMessageBytes {
				log.Error("Single message is still too large for canal-json only encode handle-key columns",
					zap.Int("maxMessageBytes", c.config.MaxMessageBytes),
					zap.Int("originLength", originLength),
					zap.Int("length", length),
					zap.Any("table", e.TableInfo.TableName))
				return errors.ErrMessageTooLarge.GenWithStackByArgs(e.TableInfo.GetTableName(), length, c.config.MaxMessageBytes)
			}
			log.Warn("Single message is too large for canal-json, only encode handle-key columns",
				zap.Int("maxMessageBytes", c.config.MaxMessageBytes),
				zap.Int("originLength", originLength),
				zap.Int("length", length),
				zap.Any("table", e.TableInfo.TableName))
		}

		if c.config.LargeMessageHandle.EnableClaimCheck() {
			claimCheckFileName := claimcheck.NewFileName()
			if err = c.claimCheck.WriteMessage(ctx, m.Key, m.Value, claimCheckFileName); err != nil {
				return errors.Trace(err)
			}

			m, err = c.newClaimCheckLocationMessage(e, claimCheckFileName)
			if err != nil {
				return errors.Trace(err)
			}
		}
	}

	c.messages = append(c.messages, m)
	return nil
}

func (c *JSONRowEventEncoder) newClaimCheckLocationMessage(
	event *commonEvent.RowEvent, fileName string,
) (*common.Message, error) {
	claimCheckLocation := c.claimCheck.FileNameWithPrefix(fileName)
	value, err := newJSONMessageForDML(event, c.config, true, claimCheckLocation)
	if err != nil {
		return nil, errors.WrapError(errors.ErrCanalEncodeFailed, err)
	}

	value, err = common.Compress(
		c.config.ChangefeedID, c.config.LargeMessageHandle.LargeMessageHandleCompression, value,
	)
	if err != nil {
		return nil, errors.Trace(err)
	}

	result := common.NewMsg(nil, value)
	result.Callback = event.Callback
	result.IncRowsCount()

	length := result.Length()
	if length > c.config.MaxMessageBytes {
		log.Warn("Single message is too large for canal-json, when create the claim check location message",
			zap.Int("maxMessageBytes", c.config.MaxMessageBytes),
			zap.Int("length", length),
			zap.Any("table", event.TableInfo.TableName))
		return nil, errors.ErrMessageTooLarge.GenWithStackByArgs(event.TableInfo.GetTableName(), length, c.config.MaxMessageBytes)
	}
	return result, nil
}

// Build implements the RowEventEncoder interface
func (c *JSONRowEventEncoder) Build() []*common.Message {
	if len(c.messages) == 0 {
		return nil
	}

	result := c.messages
	c.messages = nil
	return result
}

// EncodeDDLEvent encodes DDL events
func (c *JSONRowEventEncoder) EncodeDDLEvent(e *commonEvent.DDLEvent) (*common.Message, error) {
	message := c.newJSONMessageForDDL(e)
	value, err := json.Marshal(message)
	if err != nil {
		return nil, errors.WrapError(errors.ErrCanalEncodeFailed, err)
	}
	value, err = common.Compress(
		c.config.ChangefeedID, c.config.LargeMessageHandle.LargeMessageHandleCompression, value,
	)
	if err != nil {
		return nil, errors.Trace(err)
	}

	return common.NewMsg(nil, value), nil
}

func (c *JSONRowEventEncoder) Clean() {
	if c.claimCheck != nil {
		c.claimCheck.CleanMetrics()
	}
}