package mocks

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/apache/pulsar-client-go/pulsar"
)

// MockPulsarClient is a mock implementation of pulsar.Client.
type MockPulsarClient struct {
	mu        sync.RWMutex
	producers map[string]*MockProducer
	consumers map[string]*MockConsumer
	closed    bool
}

// NewMockPulsarClient creates a new mock Pulsar client.
func NewMockPulsarClient() *MockPulsarClient {
	return &MockPulsarClient{
		producers: make(map[string]*MockProducer),
		consumers: make(map[string]*MockConsumer),
	}
}

// CreateProducer creates a mock producer.
func (m *MockPulsarClient) CreateProducer(options pulsar.ProducerOptions) (pulsar.Producer, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.closed {
		return nil, errors.New("client is closed")
	}

	if m.producers == nil {
		m.producers = make(map[string]*MockProducer)
	}

	producer := NewMockProducer(options.Topic)
	m.producers[options.Topic] = producer
	return producer, nil
}

// Subscribe creates a mock consumer.
func (m *MockPulsarClient) Subscribe(options pulsar.ConsumerOptions) (pulsar.Consumer, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.closed {
		return nil, errors.New("client is closed")
	}

	if m.consumers == nil {
		m.consumers = make(map[string]*MockConsumer)
	}

	consumer := NewMockConsumer(options.Topic, options.Name)
	m.consumers[options.Name] = consumer
	return consumer, nil
}

// CreateReader is not implemented.
func (m *MockPulsarClient) CreateReader(options pulsar.ReaderOptions) (pulsar.Reader, error) {
	return nil, errors.New("CreateReader not implemented")
}

// CreateTableView is not implemented.
func (m *MockPulsarClient) CreateTableView(options pulsar.TableViewOptions) (pulsar.TableView, error) {
	return nil, errors.New("CreateTableView not implemented")
}

// NewTransaction creates a new transaction.
func (m *MockPulsarClient) NewTransaction(timeout time.Duration) (pulsar.Transaction, error) {
	return nil, errors.New("not implemented")
}

// TopicPartitions returns the partitions for a topic.
func (m *MockPulsarClient) TopicPartitions(topic string) ([]string, error) {
	return []string{topic}, nil
}

// Close closes the mock client.
func (m *MockPulsarClient) Close() {
	m.mu.Lock()
	defer m.mu.Unlock()

	m.closed = true
	for _, producer := range m.producers {
		producer.Close()
	}
	for _, consumer := range m.consumers {
		consumer.Close()
	}
}

// GetProducer returns a producer by topic (for testing).
func (m *MockPulsarClient) GetProducer(topic string) *MockProducer {
	m.mu.RLock()
	defer m.mu.RUnlock()

	return m.producers[topic]
}

// GetConsumer returns a consumer by name (for testing).
func (m *MockPulsarClient) GetConsumer(name string) *MockConsumer {
	m.mu.RLock()
	defer m.mu.RUnlock()

	return m.consumers[name]
}

// MockProducer is a mock implementation of pulsar.Producer.
type MockProducer struct {
	topic    string
	name     string
	messages []*pulsar.ProducerMessage
	mu       sync.RWMutex
	closed   bool
}

// NewMockProducer creates a new mock producer.
func NewMockProducer(topic string) *MockProducer {
	return &MockProducer{
		topic:    topic,
		name:     "mock-producer",
		messages: make([]*pulsar.ProducerMessage, 0),
	}
}

// Topic returns the topic name.
func (m *MockProducer) Topic() string {
	return m.topic
}

// Name returns the producer name.
func (m *MockProducer) Name() string {
	return m.name
}

// Send sends a message.
func (m *MockProducer) Send(ctx context.Context, msg *pulsar.ProducerMessage) (pulsar.MessageID, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.closed {
		return nil, errors.New("producer is closed")
	}

	m.messages = append(m.messages, msg)
	return &MockMessageID{id: len(m.messages)}, nil
}

// SendAsync sends a message asynchronously.
func (m *MockProducer) SendAsync(
	ctx context.Context,
	msg *pulsar.ProducerMessage,
	callback func(pulsar.MessageID, *pulsar.ProducerMessage, error),
) {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.closed {
		callback(nil, msg, errors.New("producer is closed"))
		return
	}

	m.messages = append(m.messages, msg)
	callback(&MockMessageID{id: len(m.messages)}, msg, nil)
}

// LastSequenceID returns the last sequence ID.
func (m *MockProducer) LastSequenceID() int64 {
	return 0
}

// Flush flushes pending messages.
func (m *MockProducer) Flush() error {
	return nil
}

// FlushWithCtx flushes pending messages with context.
func (m *MockProducer) FlushWithCtx(ctx context.Context) error {
	return nil
}

// Close closes the producer.
func (m *MockProducer) Close() {
	m.mu.Lock()
	defer m.mu.Unlock()

	m.closed = true
}

// GetMessages returns all sent messages (for testing).
func (m *MockProducer) GetMessages() []*pulsar.ProducerMessage {
	m.mu.RLock()
	defer m.mu.RUnlock()

	result := make([]*pulsar.ProducerMessage, len(m.messages))
	copy(result, m.messages)
	return result
}

// MockConsumer is a mock implementation of pulsar.Consumer.
type MockConsumer struct {
	topic       string
	name        string
	messageChan chan pulsar.ConsumerMessage
	mu          sync.RWMutex
	closed      bool
}

const (
	// defaultMessageChannelSize 定义消息通道的默认缓冲大小.
	defaultMessageChannelSize = 10
)

// NewMockConsumer creates a new mock consumer.
func NewMockConsumer(topic, name string) *MockConsumer {
	return &MockConsumer{
		topic:       topic,
		name:        name,
		messageChan: make(chan pulsar.ConsumerMessage, defaultMessageChannelSize),
	}
}

// Subscription returns the subscription name.
func (m *MockConsumer) Subscription() string {
	return m.name
}

// Topic returns the topic name.
func (m *MockConsumer) Topic() string {
	return m.topic
}

// Chan returns the message channel.
func (m *MockConsumer) Chan() <-chan pulsar.ConsumerMessage {
	return m.messageChan
}

// Ack acknowledges a message.
func (m *MockConsumer) Ack(msg pulsar.Message) error {
	m.mu.RLock()
	defer m.mu.RUnlock()

	if m.closed {
		return errors.New("consumer is closed")
	}
	return nil
}

// Nack negatively acknowledges a message.
func (m *MockConsumer) Nack(msg pulsar.Message) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	// Mock implementation: 实际不做任何操作
	_ = msg
}

// NackID negatively acknowledges a message by ID.
func (m *MockConsumer) NackID(msgID pulsar.MessageID) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	// Mock implementation: 实际不做任何操作
	_ = msgID
}

// Unsubscribe unsubscribes the consumer.
func (m *MockConsumer) Unsubscribe() error {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.closed {
		return errors.New("consumer is closed")
	}
	return nil
}

// UnsubscribeForce forcefully unsubscribes the consumer.
func (m *MockConsumer) UnsubscribeForce() error {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.closed {
		return errors.New("consumer is closed")
	}
	return nil
}

// Receive receives a single message.
func (m *MockConsumer) Receive(ctx context.Context) (pulsar.Message, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	if m.closed {
		return nil, errors.New("consumer is closed")
	}

	select {
	case msg := <-m.messageChan:
		return msg.Message, nil
	case <-ctx.Done():
		return nil, ctx.Err()
	}
}

// AckCumulative acknowledges all messages up to and including the provided message.
func (m *MockConsumer) AckCumulative(msg pulsar.Message) error {
	m.mu.RLock()
	defer m.mu.RUnlock()

	if m.closed {
		return errors.New("consumer is closed")
	}
	return nil
}

// AckID acknowledges a message by ID.
func (m *MockConsumer) AckID(msgID pulsar.MessageID) error {
	m.mu.RLock()
	defer m.mu.RUnlock()

	if m.closed {
		return errors.New("consumer is closed")
	}
	return nil
}

// AckIDCumulative acknowledges all messages up to and including the provided message ID.
func (m *MockConsumer) AckIDCumulative(msgID pulsar.MessageID) error {
	m.mu.RLock()
	defer m.mu.RUnlock()

	if m.closed {
		return errors.New("consumer is closed")
	}
	return nil
}

// AckIDList acknowledges a list of message IDs.
func (m *MockConsumer) AckIDList(msgIDs []pulsar.MessageID) error {
	m.mu.RLock()
	defer m.mu.RUnlock()

	if m.closed {
		return errors.New("consumer is closed")
	}
	return nil
}

// AckWithTxn acknowledges a message with transaction.
func (m *MockConsumer) AckWithTxn(msg pulsar.Message, txn pulsar.Transaction) error {
	m.mu.RLock()
	defer m.mu.RUnlock()

	if m.closed {
		return errors.New("consumer is closed")
	}
	return nil
}

// GetLastMessageIDs returns the last message IDs.
func (m *MockConsumer) GetLastMessageIDs() ([]pulsar.TopicMessageID, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	if m.closed {
		return nil, errors.New("consumer is closed")
	}
	return []pulsar.TopicMessageID{}, nil
}

// ReconsumeLater reconsumes a message later with delay.
func (m *MockConsumer) ReconsumeLater(msg pulsar.Message, delay time.Duration) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	// Mock implementation: 实际不做任何操作
	_, _ = msg, delay
}

// ReconsumeLaterWithCustomProperties reconsumes a message later with custom properties.
func (m *MockConsumer) ReconsumeLaterWithCustomProperties(
	msg pulsar.Message,
	customProperties map[string]string,
	delay time.Duration,
) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	// Mock implementation: 实际不做任何操作
	_, _, _ = msg, customProperties, delay
}

// Seek seeks to a message ID.
func (m *MockConsumer) Seek(msgID pulsar.MessageID) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.closed {
		return errors.New("consumer is closed")
	}
	return nil
}

// SeekByTime seeks to a time.
func (m *MockConsumer) SeekByTime(t time.Time) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.closed {
		return errors.New("consumer is closed")
	}
	return nil
}

// Name returns the consumer name.
func (m *MockConsumer) Name() string {
	m.mu.RLock()
	defer m.mu.RUnlock()

	return m.name
}

// Close closes the consumer.
func (m *MockConsumer) Close() {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.closed {
		return
	}
	m.closed = true
	close(m.messageChan)
}

// SendMessage sends a message to the consumer channel (for testing).
func (m *MockConsumer) SendMessage(msg pulsar.ConsumerMessage) error {
	m.mu.RLock()
	defer m.mu.RUnlock()

	if m.closed {
		return errors.New("consumer is closed")
	}

	select {
	case m.messageChan <- msg:
		return nil
	default:
		return errors.New("channel full")
	}
}

// MockMessageID is a mock implementation of pulsar.MessageID.
type MockMessageID struct {
	id int
}

// Serialize serializes the message ID.
func (m *MockMessageID) Serialize() []byte {
	return []byte{byte(m.id)}
}

// BatchIdx returns the batch index.
func (m *MockMessageID) BatchIdx() int32 {
	return 0
}

// BatchSize returns the batch size.
func (m *MockMessageID) BatchSize() int32 {
	return 1
}

// String returns the string representation of the message ID.
func (m *MockMessageID) String() string {
	return fmt.Sprintf("mock-message-id-%d", m.id)
}

// EntryID returns the entry ID.
func (m *MockMessageID) EntryID() int64 {
	return int64(m.id)
}

// LedgerID returns the ledger ID.
func (m *MockMessageID) LedgerID() int64 {
	return int64(m.id)
}

// PartitionIdx returns the partition index.
func (m *MockMessageID) PartitionIdx() int32 {
	return 0
}

// MockMessage is a mock implementation of pulsar.Message.
type MockMessage struct {
	key     string
	payload []byte
	id      pulsar.MessageID
}

// NewMockMessage creates a new mock message.
func NewMockMessage(key string, payload []byte) *MockMessage {
	return &MockMessage{
		key:     key,
		payload: payload,
		id:      &MockMessageID{id: 1},
	}
}

// Topic returns the topic name.
func (m *MockMessage) Topic() string {
	return "mock-topic"
}

// Properties returns message properties.
func (m *MockMessage) Properties() map[string]string {
	return make(map[string]string)
}

// Payload returns the message payload.
func (m *MockMessage) Payload() []byte {
	return m.payload
}

// ID returns the message ID.
func (m *MockMessage) ID() pulsar.MessageID {
	return m.id
}

// PublishTime returns the publish time.
func (m *MockMessage) PublishTime() time.Time {
	return time.Now()
}

// EventTime returns the event time.
func (m *MockMessage) EventTime() time.Time {
	return time.Time{}
}

// Key returns the message key.
func (m *MockMessage) Key() string {
	return m.key
}

// OrderingKey returns the ordering key.
func (m *MockMessage) OrderingKey() string {
	return ""
}

// RedeliveryCount returns the redelivery count.
func (m *MockMessage) RedeliveryCount() uint32 {
	return 0
}

// IsReplicated returns whether the message is replicated.
func (m *MockMessage) IsReplicated() bool {
	return false
}

// GetReplicatedFrom returns the replication source.
func (m *MockMessage) GetReplicatedFrom() string {
	return ""
}

// GetSchemaValue returns the schema value.
func (m *MockMessage) GetSchemaValue(v interface{}) error {
	return nil
}

// GetEncryptionContext returns the encryption context.
func (m *MockMessage) GetEncryptionContext() *pulsar.EncryptionContext {
	return nil
}

// Index returns the message index.
func (m *MockMessage) Index() *uint64 {
	return nil
}

// BrokerPublishTime returns the broker publish time.
func (m *MockMessage) BrokerPublishTime() *time.Time {
	return nil
}

// ProducerName returns the producer name.
func (m *MockMessage) ProducerName() string {
	return "mock-producer"
}

// SchemaVersion returns the schema version.
func (m *MockMessage) SchemaVersion() []byte {
	return nil
}

// ReplicatedFrom returns the replication source.
func (m *MockMessage) ReplicatedFrom() string {
	return ""
}

// NewMockConsumerMessage creates a new mock consumer message.
func NewMockConsumerMessage(key string, payload []byte) pulsar.ConsumerMessage {
	return pulsar.ConsumerMessage{
		Message:  NewMockMessage(key, payload),
		Consumer: nil,
	}
}