package telegram

import (
	"context"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
)

// ReadMessages runs a blocking getUpdates long-poll loop, emitting each
// normalized inbound message to emit until ctx is canceled or the adapter is
// disconnected. When running in webhook mode it does not poll; it simply waits
// for ctx to be canceled, since messages are delivered through the message
// handler via ProcessUpdate instead.
//
// The update offset is advanced only after a batch has been fully processed, so
// a canceled read does not lose the batch; acknowledgment is batched so the
// getUpdates call does not re-deliver the same update_id flood.
func (a *Adapter) ReadMessages(ctx context.Context, emit func(adapter.UnifiedMessage)) {
	if emit == nil {
		return
	}
	if a.isWebhookMode() {
		<-ctx.Done()
		return
	}
	if err := a.requireConnected(); err != nil {
		a.log.Warn("read not started", "err", err)
		return
	}

	pollCtx, cancel := context.WithCancel(ctx)
	a.mu.Lock()
	a.runID++
	a.pollCancel = cancel
	runID := a.runID
	a.mu.Unlock()
	defer func() {
		a.mu.Lock()
		if a.runID == runID {
			a.pollCancel = nil
		}
		a.mu.Unlock()
		cancel()
	}()

	for {
		if err := pollCtx.Err(); err != nil {
			return
		}
		updates, err := a.getUpdates(pollCtx)
		if err != nil {
			a.log.Warn("long poll failed", "err", err)
			if !sleepWait(pollCtx, pollInterval) {
				return
			}
			continue
		}
		if len(updates) == 0 {
			if !sleepWait(pollCtx, pollInterval) {
				return
			}
			continue
		}
		for _, u := range updates {
			msg, ok := a.toUnifiedMessage(u)
			if !ok {
				continue
			}
			if err := msg.Validate(); err != nil {
				a.log.Warn("skip invalid message", "err", err)
				continue
			}
			a.inlineInboundMedia(pollCtx, &msg)
			emit(msg)
		}
		a.mu.Lock()
		a.offset = updates[len(updates)-1].UpdateID + 1
		a.mu.Unlock()
	}
}

// sleepWait blocks for d or until ctx is canceled, returning false when the
// context fires first.
func sleepWait(ctx context.Context, d time.Duration) bool {
	t := time.NewTimer(d)
	defer t.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-t.C:
		return true
	}
}