package imkafka

import (
	"context"
	"encoding/json"
	"hotgo/internal/consts"
	"hotgo/internal/library/imtenant"
	"hotgo/internal/library/queue"
	"time"

	"github.com/gogf/gf/v2/database/gdb"
	"github.com/gogf/gf/v2/errors/gerror"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/os/gctx"
	"github.com/gogf/gf/v2/os/gtime"
	"github.com/gogf/gf/v2/util/gconv"
	"github.com/gogf/gf/v2/util/guid"
)

func PublishAdminCommand(ctx context.Context, in AdminCommandInput) (res *AdminCommandModel, err error) {
	if in.TenantID <= 0 || in.TenantCode == "" {
		return nil, gerror.New("IM租户信息不完整")
	}
	if in.EventType == "" {
		return nil, gerror.New("IM命令类型不能为空")
	}
	if in.IdempotencyKey == "" {
		in.IdempotencyKey = "cmd_" + guid.S()
	}
	commandID := "cmd_" + guid.S()
	eventID := "evt_" + guid.S()
	now := time.Now().UTC().Round(time.Millisecond)

	payload := in.Payload
	if payload == nil {
		payload = map[string]any{}
	}
	payload["command_id"] = commandID

	payloadBytes, err := json.Marshal(payload)
	if err != nil {
		return nil, gerror.Wrap(err, "序列化IM命令Payload失败")
	}
	envelope := Envelope{
		EventID:        eventID,
		TraceID:        gctx.CtxId(ctx),
		TenantCode:     in.TenantCode,
		SourceService:  consts.ImKafkaSourceService,
		TargetService:  consts.ImKafkaTargetService,
		EventType:      in.EventType,
		BizType:        in.BizType,
		BizID:          in.BizID,
		IdempotencyKey: in.IdempotencyKey,
		Version:        in.Version,
		OccurredAt:     now,
		Actor:          in.Actor,
		Payload:        payloadBytes,
	}
	envelopeBytes, err := json.Marshal(envelope)
	if err != nil {
		return nil, gerror.Wrap(err, "序列化IM命令Envelope失败")
	}

	db := adminDB()
	err = db.Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
		nowTime := gtime.Now()
		if _, err := tx.Exec(`INSERT INTO im_admin_commands
				(command_id,tenant_id,tenant_code,event_id,event_type,biz_type,biz_id,idempotency_key,status,payload,envelope,created_by,created_at,updated_at)
				VALUES(?,?,?,?,?,?,?,?,?,CAST(? AS JSON),CAST(? AS JSON),?,?,?)`,
			commandID, in.TenantID, in.TenantCode, eventID, in.EventType, in.BizType, in.BizID,
			in.IdempotencyKey, consts.ImAdminCommandStatusPending, string(payloadBytes), string(envelopeBytes),
			in.CreatedBy, nowTime, nowTime); err != nil {
			return gerror.Wrap(err, "创建IM后台命令失败")
		}
		if _, err := tx.Exec(`INSERT INTO im_admin_outbox
				(event_id,command_id,topic,event_type,biz_type,biz_id,key_value,envelope,status,created_at,updated_at)
				VALUES(?,?,?,?,?,?,?,CAST(? AS JSON),?,?,?)`,
			eventID, commandID, consts.QueueImAdminCommandTopic, in.EventType, in.BizType, in.BizID,
			in.BizID, string(envelopeBytes), consts.ImAdminOutboxStatusPending, nowTime, nowTime); err != nil {
			return gerror.Wrap(err, "创建IM后台命令Outbox失败")
		}
		return nil
	})
	if err != nil {
		return nil, err
	}

	res = &AdminCommandModel{
		CommandID:      commandID,
		EventID:        eventID,
		Status:         consts.ImAdminCommandStatusPending,
		TenantID:       in.TenantID,
		TenantCode:     in.TenantCode,
		EventType:      in.EventType,
		IdempotencyKey: in.IdempotencyKey,
	}
	if err = DispatchOutboxByEvent(ctx, eventID); err != nil {
		return res, err
	}
	res.Status = consts.ImAdminCommandStatusRunning
	return res, nil
}

func DispatchOutboxByEvent(ctx context.Context, eventID string) error {
	row, err := adminDB().GetOne(ctx, `SELECT * FROM im_admin_outbox WHERE event_id=? AND status=? LIMIT 1`,
		eventID, consts.ImAdminOutboxStatusPending)
	if err != nil {
		return gerror.Wrap(err, "读取IM命令Outbox失败")
	}
	if row.IsEmpty() {
		return nil
	}
	return dispatchOutboxRecord(ctx, row)
}

func PublishConfigEvent(ctx context.Context, in ConfigEventInput) error {
	if in.EventType == "" {
		return gerror.New("IM配置事件类型不能为空")
	}
	eventID := "evt_" + guid.S()
	now := time.Now().UTC().Round(time.Millisecond)
	payload := map[string]any{
		"tenant_code": in.TenantCode,
		"tenant_id":   in.TenantID,
		"group":       in.Group,
		"keys":        in.Keys,
	}
	payloadBytes, err := json.Marshal(payload)
	if err != nil {
		return gerror.Wrap(err, "序列化IM配置事件Payload失败")
	}
	envelope := Envelope{
		EventID:       eventID,
		TraceID:       gctx.CtxId(ctx),
		TenantCode:    in.TenantCode,
		SourceService: consts.ImKafkaSourceService,
		TargetService: consts.ImKafkaTargetService,
		EventType:     in.EventType,
		BizType:       "config",
		BizID:         in.BizID,
		OccurredAt:    now,
		Actor:         in.Actor,
		Payload:       payloadBytes,
	}
	envelopeBytes, err := json.Marshal(envelope)
	if err != nil {
		return gerror.Wrap(err, "序列化IM配置事件Envelope失败")
	}
	nowTime := gtime.Now()
	_, err = adminDB().Exec(ctx, `INSERT INTO im_admin_outbox
		(event_id,command_id,topic,event_type,biz_type,biz_id,key_value,envelope,status,created_at,updated_at)
		VALUES(?,?,?,?,?,?,?,CAST(? AS JSON),?,?,?)`,
		eventID, "", consts.QueueImConfigEventTopic, in.EventType, "config", in.BizID, in.TenantCode,
		string(envelopeBytes), consts.ImAdminOutboxStatusPending, nowTime, nowTime)
	if err != nil {
		return gerror.Wrap(err, "创建IM配置事件Outbox失败")
	}
	return DispatchOutboxByEvent(ctx, eventID)
}

func PublishEvent(ctx context.Context, in EventInput) (eventID string, err error) {
	if in.Topic == "" {
		return "", gerror.New("IM Kafka Topic不能为空")
	}
	if in.EventType == "" {
		return "", gerror.New("IM Kafka事件类型不能为空")
	}
	eventID = "evt_" + guid.S()
	now := time.Now().UTC().Round(time.Millisecond)
	payload := in.Payload
	if payload == nil {
		payload = map[string]any{}
	}
	payloadBytes, err := json.Marshal(payload)
	if err != nil {
		return "", gerror.Wrap(err, "序列化IM Kafka事件Payload失败")
	}
	targetService := in.TargetService
	if targetService == "" {
		targetService = consts.ImKafkaTargetService
	}
	envelope := Envelope{
		EventID:        eventID,
		TraceID:        gctx.CtxId(ctx),
		TenantCode:     in.TenantCode,
		SourceService:  consts.ImKafkaSourceService,
		TargetService:  targetService,
		EventType:      in.EventType,
		BizType:        in.BizType,
		BizID:          in.BizID,
		IdempotencyKey: in.IdempotencyKey,
		Version:        in.Version,
		OccurredAt:     now,
		Actor:          in.Actor,
		Payload:        payloadBytes,
	}
	envelopeBytes, err := json.Marshal(envelope)
	if err != nil {
		return "", gerror.Wrap(err, "序列化IM Kafka事件Envelope失败")
	}
	nowTime := gtime.Now()
	_, err = adminDB().Exec(ctx, `INSERT INTO im_admin_outbox
		(event_id,command_id,topic,event_type,biz_type,biz_id,key_value,envelope,status,created_at,updated_at)
		VALUES(?,?,?,?,?,?,?,CAST(? AS JSON),?,?,?)`,
		eventID, "", in.Topic, in.EventType, in.BizType, in.BizID, in.BizID,
		string(envelopeBytes), consts.ImAdminOutboxStatusPending, nowTime, nowTime)
	if err != nil {
		return "", gerror.Wrap(err, "创建IM Kafka事件Outbox失败")
	}
	if err = DispatchOutboxByEvent(ctx, eventID); err != nil {
		return eventID, err
	}
	return eventID, nil
}

func DispatchPendingOutbox(ctx context.Context, limit int) (sent int, err error) {
	if limit <= 0 {
		limit = 100
	}
	rows, err := adminDB().GetAll(ctx, `SELECT * FROM im_admin_outbox
		WHERE status IN (?,?) AND (next_retry_at IS NULL OR next_retry_at <= NOW())
		ORDER BY id ASC LIMIT ?`, consts.ImAdminOutboxStatusPending, consts.ImAdminOutboxStatusFailed, limit)
	if err != nil {
		return 0, gerror.Wrap(err, "读取待投递IM命令Outbox失败")
	}
	for _, row := range rows {
		if err = dispatchOutboxRecord(ctx, row); err != nil {
			g.Log().Warningf(ctx, "IM命令Outbox投递失败 event:%s err:%+v", row["event_id"].String(), err)
			continue
		}
		sent++
	}
	return sent, nil
}

func dispatchOutboxRecord(ctx context.Context, row gdb.Record) error {
	eventID := row["event_id"].String()
	commandID := row["command_id"].String()
	topic := row["topic"].String()
	body := row["envelope"].String()
	if topic == "" || body == "" {
		return markOutboxFailed(ctx, eventID, "MALFORMED_OUTBOX", "topic or envelope is empty")
	}
	if g.Cfg().MustGet(ctx, "queue.driver").String() != "kafka" {
		err := gerror.New("IM Kafka未启用，请将queue.driver配置为kafka")
		_ = markOutboxFailed(ctx, eventID, "KAFKA_DISABLED", err.Error())
		return err
	}
	if err := queue.Push(topic, body); err != nil {
		_ = markOutboxFailed(ctx, eventID, "KAFKA_SEND_FAILED", err.Error())
		return err
	}
	now := gtime.Now()
	_, err := adminDB().Exec(ctx, `UPDATE im_admin_outbox
		SET status=?,attempts=attempts+1,sent_at=?,last_reason='',last_error='',updated_at=?
		WHERE event_id=?`, consts.ImAdminOutboxStatusSent, now, now, eventID)
	if err != nil {
		return gerror.Wrap(err, "更新IM命令Outbox投递状态失败")
	}
	if commandID != "" {
		_, err = adminDB().Exec(ctx, `UPDATE im_admin_commands
			SET status=?,sent_at=?,updated_at=? WHERE command_id=?`,
			consts.ImAdminCommandStatusRunning, now, now, commandID)
		if err != nil {
			return gerror.Wrap(err, "更新IM后台命令投递状态失败")
		}
	}
	return nil
}

func markOutboxFailed(ctx context.Context, eventID string, reason string, message string) error {
	now := gtime.Now()
	_, err := adminDB().Exec(ctx, `UPDATE im_admin_outbox
		SET status=?,attempts=attempts+1,next_retry_at=?,last_reason=?,last_error=?,updated_at=?
		WHERE event_id=?`, consts.ImAdminOutboxStatusFailed, now.Add(30*time.Second), reason, message, now, eventID)
	if err != nil {
		return gerror.Wrap(err, "更新IM命令Outbox失败状态失败")
	}
	return nil
}

func ApplyCommandResult(ctx context.Context, result CommandResult) error {
	commandID := result.CommandID
	if commandID == "" {
		commandID = result.CommandEventID
	}
	if commandID == "" && result.CommandEventID == "" {
		return gerror.New("IM命令结果缺少command_id")
	}
	status := gconv.String(result.Status)
	switch status {
	case consts.ImAdminCommandStatusSucceeded, consts.ImAdminCommandStatusFailed, consts.ImAdminCommandStatusDead:
	default:
		status = consts.ImAdminCommandStatusFailed
	}
	data := g.Map{
		"status":       status,
		"reason":       result.Reason,
		"message":      result.Message,
		"completed_at": gtime.Now(),
		"updated_at":   gtime.Now(),
	}
	if len(result.Result) > 0 {
		data["result"] = string(result.Result)
	}
	whereColumn := "event_id"
	whereValue := result.CommandEventID
	if result.CommandID != "" {
		whereColumn = "command_id"
		whereValue = result.CommandID
	}
	sql := `UPDATE im_admin_commands SET status=?,reason=?,message=?,completed_at=?,updated_at=?`
	args := []any{data["status"], data["reason"], data["message"], data["completed_at"], data["updated_at"]}
	if resultValue, ok := data["result"]; ok {
		sql += `,result=CAST(? AS JSON)`
		args = append(args, resultValue)
	}
	sql += ` WHERE ` + whereColumn + `=?`
	args = append(args, whereValue)
	_, err := adminDB().Exec(ctx, sql, args...)
	if err != nil {
		return gerror.Wrap(err, "更新IM后台命令结果失败")
	}
	return nil
}

func adminDB() gdb.DB {
	return imtenant.RegistryDB()
}