package main

import (
	"context"
	"encoding/json"
	"time"

	"github.com/segmentio/kafka-go"
	"github.com/tinode/chat/server/logs"
	"github.com/tinode/chat/server/store"
	"github.com/tinode/chat/server/store/types"
)

const (
	kafkaOutboxPollInterval = 2 * time.Second
	kafkaOutboxTenantLimit  = 1000
	kafkaOutboxBatchLimit   = 100
)

func startKafkaOutboxPoller() {
	if !kafkaEnabled() {
		return
	}
	kafkaBus.wg.Add(1)
	go func() {
		defer kafkaBus.wg.Done()
		ticker := time.NewTicker(kafkaOutboxPollInterval)
		defer ticker.Stop()
		for {
			select {
			case <-kafkaBus.ctx.Done():
				return
			default:
				drainKafkaOutbox(kafkaBus.ctx)
			}
			select {
			case <-kafkaBus.ctx.Done():
				return
			case <-ticker.C:
			}
		}
	}()
}

func enqueueKafkaOutbox(tenantID types.TenantID, topicName, key string, envelope kafkaEnvelope) error {
	if envelope.OccurredAt.IsZero() {
		envelope.OccurredAt = time.Now().UTC()
	}
	raw, err := json.Marshal(envelope)
	if err != nil {
		return err
	}
	return store.HTTPInfrastructure.CreateKafkaOutbox(tenantID, &types.KafkaOutboxRecord{
		EventID:     envelope.EventID,
		Topic:       topicName,
		EventType:   envelope.EventType,
		BizType:     envelope.BizType,
		BizID:       envelope.BizID,
		Key:         key,
		Envelope:    raw,
		Status:      types.KafkaOutboxPending,
		NextRetryAt: nil,
	})
}

func drainKafkaOutbox(ctx context.Context) {
	if store.Tenants == nil || store.HTTPInfrastructure == nil {
		return
	}
	tenants, err := store.Tenants.ListActive(kafkaOutboxTenantLimit)
	if err != nil {
		logs.Warn.Printf("kafka outbox: list active tenants: %v", err)
		return
	}
	now := time.Now().UTC().Round(time.Millisecond)
	for _, tenant := range tenants {
		if tenant.ID.IsZero() {
			continue
		}
		records, err := store.HTTPInfrastructure.ListPendingKafkaOutbox(tenant.ID, kafkaOutboxBatchLimit, now)
		if err != nil {
			logs.Warn.Printf("kafka outbox: list tenant=%d: %v", tenant.ID, err)
			continue
		}
		for _, record := range records {
			if err = publishKafkaOutboxRecord(ctx, record); err != nil {
				nextRetryAt := now.Add(kafkaOutboxRetryDelay(record.Attempts + 1))
				if markErr := store.HTTPInfrastructure.MarkKafkaOutboxFailed(tenant.ID, record.EventID,
					"KAFKA_PUBLISH_FAILED", err.Error(), nextRetryAt, now); markErr != nil {
					logs.Warn.Printf("kafka outbox: mark failed event=%s tenant=%d: %v", record.EventID, tenant.ID, markErr)
				}
				continue
			}
			if err = store.HTTPInfrastructure.MarkKafkaOutboxSent(tenant.ID, record.EventID, time.Now().UTC().Round(time.Millisecond)); err != nil {
				logs.Warn.Printf("kafka outbox: mark sent event=%s tenant=%d: %v", record.EventID, tenant.ID, err)
			}
		}
	}
}

func publishKafkaOutboxRecord(ctx context.Context, record types.KafkaOutboxRecord) error {
	topic := kafkaTopic(record.Topic)
	if topic == "" {
		topic = record.Topic
	}
	return kafkaBus.writer.WriteMessages(ctx, kafka.Message{
		Topic: topic,
		Key:   []byte(record.Key),
		Value: record.Envelope,
		Time:  time.Now().UTC(),
	})
}

func kafkaOutboxRetryDelay(attempt int) time.Duration {
	if attempt <= 0 {
		attempt = 1
	}
	if kafkaBus == nil || len(kafkaBus.config.RetryBackoffMS) == 0 {
		return time.Duration(attempt) * 10 * time.Second
	}
	index := attempt - 1
	if index >= len(kafkaBus.config.RetryBackoffMS) {
		index = len(kafkaBus.config.RetryBackoffMS) - 1
	}
	return time.Duration(kafkaBus.config.RetryBackoffMS[index]) * time.Millisecond
}