package upstream

import (
	"cmpp-platform/gateway/internal/queue"
	"context"
	"fmt"
	cmpp "github.com/bigwhite/gocmpp"
	cmpputils "github.com/bigwhite/gocmpp/utils"
	"log"
	"strings"
	"time"
)

// Segment callbacks are emitted in submission order. The aggregate result keeps
// the first sequence and Msg_Id while retaining every segment result for billing.

func (m *Manager) Submit(ctx context.Context, cmd queue.SubmitCommand) (queue.SubmitResult, error) {
	if err := validateSubmitCommand(cmd); err != nil {
		result := submitResult(cmd, 0, "", "rejected", "INVALID_COMMAND", err.Error())
		if postErr := m.post(ctx, "/gateway/events/submit-result", result); postErr != nil {
			return result, postErr
		}
		return result, err
	}

	pool, err := m.connectionFor(cmd)
	if err != nil {
		result := submitResult(cmd, 0, "", "rejected", "CONNECT_FAILED", err.Error())
		if postErr := m.post(ctx, "/gateway/events/submit-result", result); postErr != nil {
			return result, postErr
		}
		return result, err
	}

	result, err := pool.submit(ctx, cmd, func(segment queue.SubmitSegmentResult) {
		payload := struct {
			queue.Envelope
			SubmitID string `json:"submitId,omitempty"`
			queue.SubmitSegmentResult
		}{
			Envelope:            cmd.Envelope,
			SubmitID:            cmd.SubmitID,
			SubmitSegmentResult: segment,
		}
		callbackCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
		postErr := m.post(callbackCtx, "/gateway/events/submit-segment-result", payload)
		cancel()
		if postErr != nil {
			log.Printf(
				"protocol_event protocol=cmpp direction=gateway_to_api event=submit_segment_result status=forward_failed channel_id=%s message_id=%s segment=%d/%d error=%q",
				cmd.ChannelID, cmd.MessageID, segment.SegmentIndex, segment.SegmentTotal, postErr,
			)
		}
	})
	if err != nil {
		if postErr := m.post(ctx, "/gateway/events/submit-result", result); postErr != nil {
			return result, postErr
		}
		return result, err
	}
	if err := m.post(ctx, "/gateway/events/submit-result", result); err != nil {
		return result, err
	}
	return result, nil
}

func (p *connectionPool) submit(
	ctx context.Context,
	cmd queue.SubmitCommand,
	onSegment func(queue.SubmitSegmentResult),
) (queue.SubmitResult, error) {
	parts, err := splitSubmitContent(cmd.CMPP.MsgFmt, cmd.Content)
	if err != nil {
		result := submitResult(cmd, 0, "", "rejected", "ENCODE_FAILED", err.Error())
		return result, err
	}

	var firstSequence uint32
	var firstGatewayMessageID string
	segments := make([]queue.SubmitSegmentResult, 0, len(parts))
	for _, part := range parts {
		conn, release, err := p.acquireConnection(ctx)
		if err != nil {
			result := submitResult(cmd, 0, "", "timeout", "WINDOW_TIMEOUT", err.Error())
			result.Segments = segments
			return result, err
		}
		seq, gatewayMessageID, result, err := conn.submitPart(ctx, cmd, part)
		release()
		segment := submitSegmentResult(part, seq, gatewayMessageID, result)
		segments = append(segments, segment)
		if onSegment != nil {
			onSegment(segment)
		}
		if firstSequence == 0 {
			firstSequence = seq
		}
		if firstGatewayMessageID == "" {
			firstGatewayMessageID = gatewayMessageID
		}
		if err != nil {
			result.Segments = segments
			return result, err
		}
		if result.SubmitStatus != "accepted" {
			result.Segments = segments
			return result, nil
		}
	}

	result := submitResult(cmd, firstSequence, firstGatewayMessageID, "accepted", "", "")
	result.Segments = segments
	return result, nil
}

func (c *connection) submitPart(ctx context.Context, cmd queue.SubmitCommand, part submitPart) (uint32, string, queue.SubmitResult, error) {
	rspCh := make(chan submitPartResponse, 1)
	pkt := c.submitRequestPacket(cmd, part)

	c.mu.Lock()
	client := c.client
	closed := c.closed
	c.mu.Unlock()
	if closed || client == nil {
		err := fmt.Errorf("supplier connection is not available")
		result := submitResult(cmd, 0, "", "timeout", "CONNECTION_LOST", err.Error())
		return 0, "", result, err
	}
	c.sendMu.Lock()
	seq, err := client.SendReqPkt(pkt)
	c.sendMu.Unlock()
	if err != nil {
		c.emitProtocolLog(protocolLogEvent{
			Protocol:      "cmpp",
			Direction:     "platform_to_channel",
			EventType:     "submit",
			Status:        "failed",
			TenantID:      cmd.TenantID,
			ApplicationID: cmd.ApplicationID,
			ChannelID:     cmd.ChannelID,
			Account:       c.config.Account,
			MessageID:     cmd.MessageID,
			Phone:         cmd.PhoneNumber,
			ResultCode:    "SEND_FAILED",
			PayloadBytes:  len(part.MsgContent),
			Detail: map[string]any{
				"segmentTotal": part.PkTotal,
				"segmentIndex": part.PkNumber,
			},
		})
		c.close()
		result := submitResult(cmd, 0, "", "timeout", "SEND_FAILED", err.Error())
		return 0, "", result, err
	}
	c.emitProtocolLog(protocolLogEvent{
		Protocol:      "cmpp",
		Direction:     "platform_to_channel",
		EventType:     "submit",
		Status:        "success",
		TenantID:      cmd.TenantID,
		ApplicationID: cmd.ApplicationID,
		ChannelID:     cmd.ChannelID,
		Account:       c.config.Account,
		MessageID:     cmd.MessageID,
		Phone:         cmd.PhoneNumber,
		PayloadBytes:  len(part.MsgContent),
		Detail: map[string]any{
			"sequenceId":   seq,
			"segmentTotal": part.PkTotal,
			"segmentIndex": part.PkNumber,
		},
	})

	c.mu.Lock()
	c.pending[seq] = rspCh
	c.mu.Unlock()
	defer func() {
		c.mu.Lock()
		delete(c.pending, seq)
		c.mu.Unlock()
	}()

	waitCtx, cancel := context.WithTimeout(ctx, defaultSubmitTimeout)
	defer cancel()
	select {
	case <-waitCtx.Done():
		result := submitResult(cmd, seq, "", "timeout", "SUBMIT_TIMEOUT", waitCtx.Err().Error())
		return seq, "", result, waitCtx.Err()
	case rsp := <-rspCh:
		if rsp.err != nil {
			result := submitResult(cmd, seq, "", "timeout", "CONNECTION_LOST", rsp.err.Error())
			return seq, "", result, rsp.err
		}
		gatewayMessageID := fmt.Sprint(rsp.msgID)
		status := "accepted"
		errorCode := ""
		errorMessage := ""
		if rsp.result != 0 {
			status = "rejected"
			errorCode = fmt.Sprint(rsp.result)
			errorMessage = fmt.Sprintf("upstream submit rejected with result %d", rsp.result)
		}
		c.emitProtocolLog(protocolLogEvent{
			Protocol:         "cmpp",
			Direction:        "channel_to_platform",
			EventType:        "submit_resp",
			Status:           "success",
			TenantID:         cmd.TenantID,
			ApplicationID:    cmd.ApplicationID,
			ChannelID:        cmd.ChannelID,
			Account:          c.config.Account,
			MessageID:        cmd.MessageID,
			GatewayMessageID: gatewayMessageID,
			Phone:            cmd.PhoneNumber,
			ResultCode:       fmt.Sprint(rsp.result),
			Detail: map[string]any{
				"sequenceId":   rsp.seqID,
				"segmentTotal": part.PkTotal,
				"segmentIndex": part.PkNumber,
			},
		})
		if rsp.result == 0 {
			c.mu.Lock()
			c.tracker[rsp.msgID] = cmd
			c.mu.Unlock()
		}
		return seq, gatewayMessageID, submitResult(cmd, seq, gatewayMessageID, status, errorCode, errorMessage), nil
	}
}

func (c *connection) submitRequestPacket(cmd queue.SubmitCommand, part submitPart) cmpp.Packer {
	base := submitRequestFields{
		PkTotal:            part.PkTotal,
		PkNumber:           part.PkNumber,
		TpUdhi:             part.TpUdhi,
		RegisteredDelivery: uint8(cmd.CMPP.RegisteredDelivery),
		MsgLevel:           1,
		ServiceId:          cmd.CMPP.ServiceID,
		FeeUserType:        uint8(defaultInt(cmd.CMPP.FeeUserType, 2)),
		FeeTerminalId:      cmd.PhoneNumber,
		MsgFmt:             uint8(cmd.CMPP.MsgFmt),
		MsgSrc:             c.config.Account,
		FeeType:            defaultString(cmd.CMPP.FeeType, "02"),
		FeeCode:            defaultString(cmd.CMPP.FeeCode, "0"),
		SrcId:              cmd.CMPP.SrcID,
		DestUsrTl:          1,
		DestTerminalId:     []string{cmd.PhoneNumber},
		MsgLength:          uint8(len(part.MsgContent)),
		MsgContent:         part.MsgContent,
	}
	if protocolVersion(c.config.CMPPVersion) == cmpp.V20 {
		return &cmpp.Cmpp2SubmitReqPkt{
			PkTotal:            base.PkTotal,
			PkNumber:           base.PkNumber,
			RegisteredDelivery: base.RegisteredDelivery,
			MsgLevel:           base.MsgLevel,
			ServiceId:          base.ServiceId,
			FeeUserType:        base.FeeUserType,
			FeeTerminalId:      base.FeeTerminalId,
			TpUdhi:             base.TpUdhi,
			MsgFmt:             base.MsgFmt,
			MsgSrc:             base.MsgSrc,
			FeeType:            base.FeeType,
			FeeCode:            base.FeeCode,
			SrcId:              base.SrcId,
			DestUsrTl:          base.DestUsrTl,
			DestTerminalId:     base.DestTerminalId,
			MsgLength:          base.MsgLength,
			MsgContent:         base.MsgContent,
		}
	}
	return &cmpp.Cmpp3SubmitReqPkt{
		PkTotal:            base.PkTotal,
		PkNumber:           base.PkNumber,
		RegisteredDelivery: base.RegisteredDelivery,
		MsgLevel:           base.MsgLevel,
		ServiceId:          base.ServiceId,
		FeeUserType:        base.FeeUserType,
		FeeTerminalId:      base.FeeTerminalId,
		TpUdhi:             base.TpUdhi,
		MsgFmt:             base.MsgFmt,
		MsgSrc:             base.MsgSrc,
		FeeType:            base.FeeType,
		FeeCode:            base.FeeCode,
		SrcId:              base.SrcId,
		DestUsrTl:          base.DestUsrTl,
		DestTerminalId:     base.DestTerminalId,
		MsgLength:          base.MsgLength,
		MsgContent:         base.MsgContent,
	}
}

type submitRequestFields struct {
	PkTotal            uint8
	PkNumber           uint8
	RegisteredDelivery uint8
	MsgLevel           uint8
	ServiceId          string
	FeeUserType        uint8
	FeeTerminalId      string
	TpUdhi             uint8
	MsgFmt             uint8
	MsgSrc             string
	FeeType            string
	FeeCode            string
	SrcId              string
	DestUsrTl          uint8
	DestTerminalId     []string
	MsgLength          uint8
	MsgContent         string
}

func submitResult(cmd queue.SubmitCommand, sequenceID uint32, gatewayMessageID string, status string, code string, message string) queue.SubmitResult {
	if gatewayMessageID == "" {
		gatewayMessageID = fmt.Sprintf("GW-%s-%d", cmd.SubmitID, time.Now().UnixNano())
	}
	return queue.SubmitResult{
		Envelope: queue.Envelope{
			SchemaVersion: queue.SchemaVersion,
			MessageType:   queue.MessageTypeSubmitResult,
			TraceID:       cmd.TraceID,
			MessageID:     cmd.MessageID,
			ChannelID:     cmd.ChannelID,
			CreatedAt:     time.Now().UTC(),
		},
		SubmitID:         cmd.SubmitID,
		SequenceID:       sequenceID,
		GatewayMessageID: gatewayMessageID,
		SubmitStatus:     status,
		ErrorCode:        code,
		ErrorMessage:     message,
		SubmittedAt:      time.Now().UTC(),
	}
}

func submitSegmentResult(part submitPart, sequenceID uint32, gatewayMessageID string, result queue.SubmitResult) queue.SubmitSegmentResult {
	if gatewayMessageID == "" {
		gatewayMessageID = result.GatewayMessageID
	}
	return queue.SubmitSegmentResult{
		SegmentTotal:     int(part.PkTotal),
		SegmentIndex:     int(part.PkNumber),
		SequenceID:       sequenceID,
		GatewayMessageID: gatewayMessageID,
		SubmitStatus:     result.SubmitStatus,
		ErrorCode:        result.ErrorCode,
		ErrorMessage:     result.ErrorMessage,
		SubmittedAt:      result.SubmittedAt,
	}
}

func validateSubmitCommand(cmd queue.SubmitCommand) error {
	if cmd.MessageType != queue.MessageTypeSubmitCommand {
		return fmt.Errorf("unsupported messageType %q", cmd.MessageType)
	}
	if cmd.MessageID == "" || cmd.ChannelID == "" || cmd.SubmitID == "" {
		return fmt.Errorf("messageId, channelId and submitId are required")
	}
	if cmd.Upstream.GatewayHost == "" || cmd.Upstream.GatewayPort <= 0 {
		return fmt.Errorf("upstream gatewayHost and gatewayPort are required")
	}
	if cmd.Upstream.Account == "" || cmd.Upstream.PasswordCipher == "" {
		return fmt.Errorf("upstream account and passwordCipher are required")
	}
	if len(cmd.PhoneNumber) == 0 || len(cmd.Content) == 0 {
		return fmt.Errorf("phoneNumber and content are required")
	}
	return nil
}

func encodeContent(format int, content string) (string, error) {
	switch format {
	case 8:
		return cmpputils.Utf8ToUcs2(content)
	case 15:
		return cmpputils.Utf8ToGB18030(content)
	default:
		return content, nil
	}
}

func protocolVersion(version string) cmpp.Type {
	if strings.HasPrefix(version, "2") {
		return cmpp.V20
	}
	return cmpp.V30
}