Pete/internal/poster/queue.go

package poster

import (
	"context"
	"log/slog"
	"sync"
	"time"

	"pete/internal/config"
	"pete/internal/dedup"
	"pete/internal/matrix"
	"pete/internal/storage"
)

const maxRetries = 3

// QueueItem represents a story ready to be posted.
type QueueItem struct {
	GUID       string
	Headline   string
	Lede       string
	ImageURL   string
	ArticleURL string
	Source     string
	Channel    string
	Platforms  []string
	retries    int
}

// Queue manages per-channel metered release of stories.
type Queue struct {
	mu       sync.Mutex
	queues   map[string][]QueueItem // channel -> items
	config   config.PostingConfig
	mx       *matrix.Client
	done     chan struct{}
}

// NewQueue creates a new metered release queue.
func NewQueue(cfg config.PostingConfig, mx *matrix.Client) *Queue {
	return &Queue{
		queues: make(map[string][]QueueItem),
		config: cfg,
		mx:     mx,
		done:   make(chan struct{}),
	}
}

// Enqueue adds a story to the appropriate channel queue.
// Stories for unconfigured channels are dropped with a warning.
func (q *Queue) Enqueue(item QueueItem) {
	if _, ok := q.mx.ChannelRoomID(item.Channel); !ok {
		slog.Warn("dropping story for unconfigured channel",
			"guid", item.GUID,
			"channel", item.Channel,
		)
		return
	}
	q.mu.Lock()
	defer q.mu.Unlock()
	q.queues[item.Channel] = append(q.queues[item.Channel], item)
	slog.Info("story queued for posting",
		"guid", item.GUID,
		"channel", item.Channel,
		"queue_depth", len(q.queues[item.Channel]),
	)
}

// Start runs the queue drain ticker. Blocks until ctx is cancelled, then
// drops any remaining queued items (no flush-posting on shutdown — we'd
// rather lose them than dump a flood into the channel).
func (q *Queue) Start(ctx context.Context) {
	defer close(q.done)
	ticker := time.NewTicker(10 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			q.dropOnShutdown()
			return
		case <-ticker.C:
			q.drain()
		}
	}
}

// Wait blocks until the queue goroutine has finished (including graceful drain).
func (q *Queue) Wait() {
	<-q.done
}

// ForcePost pops the next queued item for the given channel and posts it
// immediately, bypassing min-interval, burst cap, and daily cap. Dedup
// (canonical URL cooldown) still applies. Returns true only if a Matrix
// message was actually sent — a dedup-skip returns false so the caller
// can fall back (e.g., to a DB lookup) instead of silently consuming the
// queued item with no user-visible result.
func (q *Queue) ForcePost(channel string) bool {
	q.mu.Lock()
	items := q.queues[channel]
	if len(items) == 0 {
		q.mu.Unlock()
		return false
	}
	item := items[0]
	q.queues[channel] = items[1:]
	q.mu.Unlock()

	slog.Info("force-posting story on demand",
		"guid", item.GUID, "channel", channel)
	return q.postItem(item)
}

func (q *Queue) drain() {
	q.mu.Lock()
	channels := make([]string, 0, len(q.queues))
	for ch := range q.queues {
		channels = append(channels, ch)
	}
	q.mu.Unlock()

	for _, ch := range channels {
		q.drainChannel(ch)
	}
}

// dropOnShutdown clears any pending queues, logging the count. We deliberately
// do NOT post these — the daily cap exists for a reason, and a flush on Ctrl-C
// would dump everything pending into the rooms at once.
func (q *Queue) dropOnShutdown() {
	q.mu.Lock()
	defer q.mu.Unlock()
	total := 0
	for ch, items := range q.queues {
		if len(items) > 0 {
			slog.Info("dropping queued items on shutdown",
				"channel", ch, "count", len(items))
			total += len(items)
		}
		q.queues[ch] = nil
	}
	if total > 0 {
		slog.Info("shutdown: queue drop complete", "total_dropped", total)
	}
}

func (q *Queue) drainChannel(channel string) {
	q.mu.Lock()
	items := q.queues[channel]
	if len(items) == 0 {
		q.mu.Unlock()
		return
	}
	q.mu.Unlock()

	now := time.Now().Unix()
	minInterval := int64(q.config.MinIntervalSeconds)
	burstWindow := int64(q.config.BurstCapWindowSeconds)

	// Global daily cap: hard ceiling across ALL channels (rolling 24h).
	// Checked first so it short-circuits per-channel logic.
	if q.config.DailyCapTotal > 0 {
		dayStart := now - 24*3600
		todays := storage.CountAllPostsInWindow(dayStart)
		if todays >= q.config.DailyCapTotal {
			slog.Debug("global daily cap reached, holding queue",
				"channel", channel, "posts_24h", todays, "cap", q.config.DailyCapTotal)
			return
		}
	}

	// Check minimum interval since last post (per channel)
	lastPost := storage.GetLastPostTime(channel)
	if now-lastPost < minInterval {
		return
	}

	// Check burst cap (per channel)
	windowStart := now - burstWindow
	postsInWindow := storage.CountPostsInWindow(channel, windowStart)
	if postsInWindow >= q.config.BurstCapCount {
		return
	}

	// Dequeue one item
	q.mu.Lock()
	items = q.queues[channel]
	if len(items) == 0 {
		q.mu.Unlock()
		return
	}
	item := items[0]
	q.queues[channel] = items[1:]
	q.mu.Unlock()

	q.postItem(item)
}

// PostNow sends a story immediately, bypassing the in-memory queue and all
// pacing limits. Last-mile canonical-URL dedup still applies. Used by !post
// to satisfy on-demand requests with stories pulled directly from storage.
func (q *Queue) PostNow(item QueueItem) {
	q.postItem(item)
}

// postItem returns true when a Matrix event was actually sent, false on
// any non-success path (dedup-skip, transport failure with or without
// retry, dead-letter). ForcePost uses the return value to decide whether
// to acknowledge the user-initiated !post or fall back to a DB lookup.
func (q *Queue) postItem(item QueueItem) bool {
	// Last-mile dedup: if this canonical URL was already posted to this channel
	// within the cooldown window, drop silently. Catches "same article, different
	// GUID across feeds" and any race where two items slipped past ingest dedup.
	canonical := dedup.CanonicalURL(item.ArticleURL)
	cooldownSec := int64(q.config.DedupCooldownHours) * 3600
	if storage.WasCanonicalPostedRecently(canonical, item.Channel, cooldownSec) {
		slog.Info("dropping duplicate post (canonical URL posted within cooldown)",
			"guid", item.GUID,
			"channel", item.Channel,
			"url_canonical", canonical,
			"cooldown_hours", q.config.DedupCooldownHours,
		)
		return false
	}

	story := &matrix.PostableStory{
		ImageURL:   item.ImageURL,
		Headline:   item.Headline,
		ArticleURL: item.ArticleURL,
		Lede:       item.Lede,
		Source:     item.Source,
		Channel:    item.Channel,
		Platforms:  item.Platforms,
	}

	eventID, imageSent, err := q.mx.PostStory(item.Channel, story)
	if err != nil {
		item.retries++
		if item.retries >= maxRetries {
			slog.Error("story dead-lettered after max retries",
				"guid", item.GUID,
				"channel", item.Channel,
				"err", err,
			)
			return false
		}
		slog.Error("failed to post story, will retry",
			"guid", item.GUID,
			"channel", item.Channel,
			"attempt", item.retries,
			"err", err,
		)
		// If the image already went up before the text failed, clear it on the
		// retry so we don't send the m.image event a second time.
		if imageSent {
			item.ImageURL = ""
		}
		// Re-queue at front for retry
		q.mu.Lock()
		q.queues[item.Channel] = append([]QueueItem{item}, q.queues[item.Channel]...)
		q.mu.Unlock()
		return false
	}

	// Record in post log (INSERT OR IGNORE via unique index)
	storage.InsertPostLog(item.GUID, item.Channel, string(eventID), canonical)

	slog.Info("story posted",
		"guid", item.GUID,
		"channel", item.Channel,
		"event_id", eventID,
	)
	return true
}