package lake

import (
	"bytes"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"time"
)

type orgScopedRequest interface {
	setOrgSlug(string)
}

type tokenExchangeRequest struct {
	TiDBOrgID      string `json:"tidbOrgId"`
	LakeOrgID      string `json:"lakeOrgId"`
	OrgSlug        string `json:"orgSlug,omitempty"`
	ContextType    string `json:"contextType,omitempty"`
	ContextID      string `json:"contextId,omitempty"`
	ContextVersion uint64 `json:"contextVersion,omitempty"`
	InitiatedBy    string `json:"initiatedBy,omitempty"`
}

type tokenExchangeResponse struct {
	Success bool `json:"success"`
	Data    struct {
		AccessToken string    `json:"accessToken"`
		TokenType   string    `json:"tokenType"`
		ExpiresAt   time.Time `json:"expiresAt"`
		OrgSlug     string    `json:"orgSlug"`
		TenantID    string    `json:"tenantId"`
	} `json:"data"`
}

type wireErrorResponse struct {
	Error   string `json:"error"`
	Message string `json:"message"`
}

type wireSuccessResponse struct {
	Success bool   `json:"success"`
	Message string `json:"message"`
}

type wireCreateResponse struct {
	Success bool   `json:"success"`
	ID      int32  `json:"id"`
	Message string `json:"message"`
}

type wireSetupInfoRequest struct {
	OrgSlug string `json:"orgSlug"`
}

type wireSetupInfoResponse struct {
	Success bool `json:"success"`
	Data    struct {
		ExternalID      string `json:"externalID"`
		TrustPrincipals []struct {
			Name     string `json:"name"`
			ARN      string `json:"arn"`
			Required bool   `json:"required"`
		} `json:"trustPrincipals"`
	} `json:"data"`
	Message string `json:"message"`
}

type wireS3Credentials struct {
	S3Bucket    string `json:"s3Bucket"`
	S3Region    string `json:"s3Region"`
	S3Endpoint  string `json:"s3Endpoint,omitempty"`
	S3AccessKey string `json:"s3AccessKey,omitempty"`
	S3SecretKey string `json:"s3SecretKey,omitempty"`
	SQSQueueURL string `json:"sqsQueueURL,omitempty"`
	SQSRegion   string `json:"sqsRegion,omitempty"`
	Database    string `json:"database,omitempty"`
	RoleARN     string `json:"roleARN,omitempty"`
	ExternalID  string `json:"externalID,omitempty"`
}

type wireDataSourceRequest struct {
	DataSourceID    int32             `json:"datasourceId,omitempty"`
	Name            string            `json:"name"`
	ServiceID       string            `json:"serviceId"`
	TiDBCredentials wireS3Credentials `json:"tidbCredentials"`
	OrgSlug         string            `json:"orgSlug"`
}

type wireGetDataSourceRequest struct {
	DataSourceID int32  `json:"datasourceId"`
	OrgSlug      string `json:"orgSlug"`
}

type wireDeleteDataSourceRequest struct {
	DataSourceIDs []int32 `json:"datasourceIds"`
	OrgSlug       string  `json:"orgSlug"`
}

type wireListDataSourcesRequest struct {
	Page      int32  `json:"page"`
	PerPage   int32  `json:"perPage"`
	ServiceID string `json:"serviceId"`
	OrgSlug   string `json:"orgSlug"`
}

type wireDataSourceInfoResponse struct {
	Success bool               `json:"success"`
	Data    wireDataSourceItem `json:"data"`
	Message string             `json:"message"`
}

type wireListDataSourcesResponse struct {
	Success bool `json:"success"`
	Data    struct {
		List  []wireDataSourceItem `json:"list"`
		Total int32                `json:"total"`
	} `json:"data"`
	Message string `json:"message"`
}

// wireDataSourceItem deliberately ignores the serviceId Lake echoes back.
// Requests name the service with the "TiDBCDC" alias, but responses carry
// Lake's internal numeric service key instead, so the two are not comparable
// and decoding it as a string fails outright.
type wireDataSourceItem struct {
	ID           int32           `json:"id"`
	DataSourceID int32           `json:"datasourceId"`
	Name         string          `json:"name"`
	ServiceArgs  json.RawMessage `json:"serviceArgs"`
	CreatedAt    string          `json:"createdAt"`
	UpdatedAt    string          `json:"updatedAt"`
}

type wireTiDBTask struct {
	SyncMode                wireSyncMode `json:"syncMode"`
	TableRules              []string     `json:"tableRules,omitempty"`
	PollIntervalSeconds     *int32       `json:"pollIntervalSeconds,omitempty"`
	BatchFileCount          *int32       `json:"batchFileCount,omitempty"`
	MergeIntervalSeconds    *int32       `json:"mergeIntervalSeconds,omitempty"`
	TableParallelism        *int32       `json:"tableParallelism,omitempty"`
	AllowDelete             bool         `json:"allowDelete,omitempty"`
	DumplingS3Prefix        string       `json:"dumplingS3Prefix,omitempty"`
	DumplingCSVSeparator    string       `json:"dumplingCsvSeparator,omitempty"`
	DumplingCSVNull         string       `json:"dumplingCsvNull,omitempty"`
	DumplingCSVSkipHeader   *int32       `json:"dumplingCsvSkipHeader,omitempty"`
	DumplingAutoCreateTable bool         `json:"dumplingAutoCreateTable,omitempty"`
	DumplingOnError         string       `json:"dumplingOnError,omitempty"`
	DumplingPurge           bool         `json:"dumplingPurge,omitempty"`
	ConflictKey             string       `json:"conflictKey,omitempty"`
	ChangefeedS3Prefix      string       `json:"changefeedS3Prefix,omitempty"`
	TargetDatabasePrefix    string       `json:"targetDatabasePrefix,omitempty"`
	TargetDatabaseSuffix    string       `json:"targetDatabaseSuffix,omitempty"`
	TargetTablePrefix       string       `json:"targetTablePrefix,omitempty"`
	TargetTableSuffix       string       `json:"targetTableSuffix,omitempty"`
}

type wireTaskRequest struct {
	TaskID              int32        `json:"taskId,omitempty"`
	Name                string       `json:"name"`
	DataSourceID        int32        `json:"datasourceId"`
	TiDBTask            wireTiDBTask `json:"tidbTask"`
	Warehouse           string       `json:"warehouse"`
	ContinuousIngestion int32        `json:"continuousIngestion,omitempty"`
	OrgSlug             string       `json:"orgSlug"`
}

type wireGetTaskRequest struct {
	TaskID  int32  `json:"taskId"`
	OrgSlug string `json:"orgSlug"`
}

type wireListTasksRequest struct {
	Page         int32  `json:"page"`
	PerPage      int32  `json:"perPage"`
	ServiceID    string `json:"serviceId"`
	DataSourceID int32  `json:"datasourceId,omitempty"`
	OrgSlug      string `json:"orgSlug"`
}

type wireOperateTaskRequest struct {
	TaskIDs []int32 `json:"taskIds"`
	Operate string  `json:"operate"`
	OrgSlug string  `json:"orgSlug"`
}

type wireTaskInfoResponse struct {
	Success bool         `json:"success"`
	Data    wireTaskItem `json:"data"`
	Message string       `json:"message"`
}

type wireListTasksResponse struct {
	Success bool `json:"success"`
	Data    struct {
		List  []wireTaskItem `json:"list"`
		Total int32          `json:"total"`
	} `json:"data"`
	Message string `json:"message"`
}

type wireTaskItem struct {
	ID                  int32            `json:"id"`
	TaskID              int32            `json:"taskId"`
	Name                string           `json:"name"`
	DataSourceID        int32            `json:"datasourceId"`
	Warehouse           string           `json:"warehouse"`
	ContinuousIngestion int32            `json:"continuousIngestion"`
	TaskStatus          wireTaskStatus   `json:"taskStatus"`
	Database            string           `json:"database"`
	TiDBTask            wireTiDBTask     `json:"tidbTask"`
	LastRunData         wireResponseData `json:"lastRunData"`
	CreatedAt           string           `json:"createdAt"`
	UpdatedAt           string           `json:"updatedAt"`
}

type wireListTaskRunsRequest struct {
	TaskID  int32  `json:"taskId"`
	Page    int32  `json:"page"`
	PerPage int32  `json:"perPage"`
	OrgSlug string `json:"orgSlug"`
}

type wireGetTaskRunRequest struct {
	ExecuteID string `json:"executeId"`
	OrgSlug   string `json:"orgSlug"`
}

func (r *wireSetupInfoRequest) setOrgSlug(orgSlug string)        { r.OrgSlug = orgSlug }
func (r *wireDataSourceRequest) setOrgSlug(orgSlug string)       { r.OrgSlug = orgSlug }
func (r *wireGetDataSourceRequest) setOrgSlug(orgSlug string)    { r.OrgSlug = orgSlug }
func (r *wireDeleteDataSourceRequest) setOrgSlug(orgSlug string) { r.OrgSlug = orgSlug }
func (r *wireListDataSourcesRequest) setOrgSlug(orgSlug string)  { r.OrgSlug = orgSlug }
func (r *wireTaskRequest) setOrgSlug(orgSlug string)             { r.OrgSlug = orgSlug }
func (r *wireGetTaskRequest) setOrgSlug(orgSlug string)          { r.OrgSlug = orgSlug }
func (r *wireListTasksRequest) setOrgSlug(orgSlug string)        { r.OrgSlug = orgSlug }
func (r *wireOperateTaskRequest) setOrgSlug(orgSlug string)      { r.OrgSlug = orgSlug }
func (r *wireListTaskRunsRequest) setOrgSlug(orgSlug string)     { r.OrgSlug = orgSlug }
func (r *wireGetTaskRunRequest) setOrgSlug(orgSlug string)       { r.OrgSlug = orgSlug }

type wireListTaskRunsResponse struct {
	Success bool `json:"success"`
	Data    struct {
		List  []wireTaskRunItem `json:"list"`
		Total int32             `json:"total"`
	} `json:"data"`
	Message string `json:"message"`
}

type wireTaskRunInfoResponse struct {
	Success bool            `json:"success"`
	Data    wireTaskRunItem `json:"data"`
	Message string          `json:"message"`
}

type wireTaskRunItem struct {
	ExecuteID    string            `json:"executeId"`
	TaskID       int32             `json:"taskId"`
	Status       wireTaskRunStatus `json:"status"`
	RequestData  json.RawMessage   `json:"requestData"`
	ResponseData wireResponseData  `json:"responseData"`
	CreatedAt    string            `json:"createdAt"`
	UpdatedAt    string            `json:"updatedAt"`
}

type wireResponseData struct {
	Kind        string           `json:"kind"`
	Status      json.RawMessage  `json:"status"`
	Error       string           `json:"error"`
	TiDBLastRun wireTiDBProgress `json:"tidbLastRun"`
}

type wireTiDBProgress struct {
	LastResolvedTS      wireUint64   `json:"lastResolvedTS"`
	RowsSynced          wireUint64   `json:"rowsSynced"`
	InsertCount         wireUint64   `json:"insertCount"`
	UpdateCount         wireUint64   `json:"updateCount"`
	DeleteCount         wireUint64   `json:"deleteCount"`
	TablesActive        wireInt32    `json:"tablesActive"`
	TablesError         wireInt32    `json:"tablesError"`
	DatabasesActive     wireInt32    `json:"databasesActive"`
	SyncMode            wireSyncMode `json:"syncMode"`
	FullSyncCompleted   bool         `json:"fullSyncCompleted"`
	FullSyncCompletedAt wireInt64    `json:"fullSyncCompletedAt"`
	BackupTSO           wireUint64   `json:"backupTSO"`
	// LastMessage carries why a run ended. Lake reports the actionable detail
	// only here - "warehouse ... not found: Bad warehouse" and the like - so
	// dropping it leaves a caller with a failure and no way to act on it.
	LastMessage string `json:"lastMessage"`
}

type wireUint64 uint64

func (v *wireUint64) UnmarshalJSON(data []byte) error {
	parsed, err := parseUintJSON(data, 64)
	if err != nil {
		return err
	}
	*v = wireUint64(parsed)
	return nil
}

type wireInt64 int64

func (v *wireInt64) UnmarshalJSON(data []byte) error {
	parsed, err := parseIntJSON(data, 64)
	if err != nil {
		return err
	}
	*v = wireInt64(parsed)
	return nil
}

type wireInt32 int32

func (v *wireInt32) UnmarshalJSON(data []byte) error {
	parsed, err := parseIntJSON(data, 32)
	if err != nil {
		return err
	}
	*v = wireInt32(parsed)
	return nil
}

func parseUintJSON(data []byte, bitSize int) (uint64, error) {
	value := strings.Trim(string(data), "\"")
	if value == "" || value == "null" {
		return 0, nil
	}
	return strconv.ParseUint(value, 10, bitSize)
}

func parseIntJSON(data []byte, bitSize int) (int64, error) {
	value := strings.Trim(string(data), "\"")
	if value == "" || value == "null" {
		return 0, nil
	}
	return strconv.ParseInt(value, 10, bitSize)
}

type wireSyncMode string

func (m *wireSyncMode) UnmarshalJSON(data []byte) error {
	value, err := parseEnumJSON(data)
	if err != nil {
		return err
	}
	switch value {
	case "1", string(SyncModeSnapshot):
		*m = wireSyncMode(SyncModeSnapshot)
	case "2", string(SyncModeCDC):
		*m = wireSyncMode(SyncModeCDC)
	case "3", string(SyncModeSnapshotAndCDC):
		*m = wireSyncMode(SyncModeSnapshotAndCDC)
	case "", "0":
		*m = ""
	default:
		return fmt.Errorf("unknown TiDB sync mode %q", value)
	}
	return nil
}

func (m wireSyncMode) MarshalJSON() ([]byte, error) {
	return json.Marshal(string(m))
}

type wireTaskStatus string

func (s *wireTaskStatus) UnmarshalJSON(data []byte) error {
	value, err := parseEnumJSON(data)
	if err != nil {
		return err
	}
	switch value {
	case "1", string(TaskStatusStopped):
		*s = wireTaskStatus(TaskStatusStopped)
	case "2", string(TaskStatusRunning):
		*s = wireTaskStatus(TaskStatusRunning)
	case "", "0":
		*s = wireTaskStatus(TaskStatusUnknown)
	default:
		return fmt.Errorf("unknown Task status %q", value)
	}
	return nil
}

type wireTaskRunStatus string

func (s *wireTaskRunStatus) UnmarshalJSON(data []byte) error {
	value, err := parseEnumJSON(data)
	if err != nil {
		return err
	}
	switch value {
	case "1", string(TaskRunStatusRunning):
		*s = wireTaskRunStatus(TaskRunStatusRunning)
	case "2", string(TaskRunStatusSuccess):
		*s = wireTaskRunStatus(TaskRunStatusSuccess)
	case "3", string(TaskRunStatusFail):
		*s = wireTaskRunStatus(TaskRunStatusFail)
	case "", "0":
		*s = wireTaskRunStatus(TaskRunStatusUnknown)
	default:
		return fmt.Errorf("unknown Task run status %q", value)
	}
	return nil
}

func parseEnumJSON(data []byte) (string, error) {
	data = bytes.TrimSpace(data)
	if len(data) == 0 || bytes.Equal(data, []byte("null")) {
		return "", nil
	}
	if data[0] == '"' {
		var value string
		if err := json.Unmarshal(data, &value); err != nil {
			return "", err
		}
		return value, nil
	}
	return string(data), nil
}

func parseTime(value string) time.Time {
	if value == "" {
		return time.Time{}
	}
	parsed, err := time.Parse(time.RFC3339Nano, value)
	if err != nil {
		return time.Time{}
	}
	return parsed
}

func progressFromWire(value wireTiDBProgress) TiDBProgress {
	progress := TiDBProgress{
		LastResolvedTS:    uint64(value.LastResolvedTS),
		RowsSynced:        uint64(value.RowsSynced),
		InsertCount:       uint64(value.InsertCount),
		UpdateCount:       uint64(value.UpdateCount),
		DeleteCount:       uint64(value.DeleteCount),
		TablesActive:      int32(value.TablesActive),
		TablesError:       int32(value.TablesError),
		DatabasesActive:   int32(value.DatabasesActive),
		SyncMode:          SyncMode(value.SyncMode),
		FullSyncCompleted: value.FullSyncCompleted,
		BackupTSO:         uint64(value.BackupTSO),
	}
	if value.FullSyncCompletedAt > 0 {
		// Seconds, not milliseconds. Confirmed against staging: Lake answered
		// "fullSyncCompletedAt":"1786395134" for a run that finished at
		// 2026-08-10T20:52:14Z. Read as milliseconds the same value lands in
		// January 1970, so a regression here is loud rather than subtle.
		progress.FullSyncCompletedAt = time.Unix(int64(value.FullSyncCompletedAt), 0).UTC()
	}
	return progress
}