b617d403b7
Manual !post overrides were counted toward daily_cap_total, so a few forced posts could starve the round-robin rotation for the rest of the day. Tag forced rows in post_log and skip them in CountAllPostsInWindow so the cap only meters the auto-rotation.
278 lines
7.4 KiB
Go
278 lines
7.4 KiB
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, 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
|
|
}
|