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, true) } 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, false) } // 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, true) } // 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. // forced=true marks the resulting post_log row so it's excluded from the // global daily cap (manual overrides shouldn't steal the rotation budget). func (q *Queue) postItem(item QueueItem, forced bool) 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, forced) slog.Info("story posted", "guid", item.GUID, "channel", item.Channel, "event_id", eventID, ) return true }