100 lines
3.2 KiB
Go
100 lines
3.2 KiB
Go
|
// Copyright (c) 2026 Petr Balvín <opensource@petrbalvin.org> (https://petrbalvin.org)
|
||
|
|
// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0
|
||
|
|
|
||
|
|
// Package scheduler publishes scheduled posts whose publish_at date
|
||
|
|
// has arrived, either from the in-app loop or the publish-due CLI.
|
||
|
|
package scheduler
|
||
|
|
|
||
|
|
import (
|
||
|
|
"log/slog"
|
||
|
|
"time"
|
||
|
|
|
||
|
|
"sourcedock.dev/petrbalvin/interpres/v2"
|
||
|
|
"sourcedock.dev/petrbalvin/volumen/internal/post"
|
||
|
|
)
|
||
|
|
|
||
|
|
// Store is the content the scheduler reads and writes: every post, and a
|
||
|
|
// save for the ones whose date has arrived.
|
||
|
|
type Store interface {
|
||
|
|
All() []*post.Post
|
||
|
|
Save(p *post.Post) (*post.Post, error)
|
||
|
|
}
|
||
|
|
|
||
|
|
// MinInterval is the shortest interval the in-app loop accepts.
|
||
|
|
// time.NewTicker panics on a non-positive duration, so a bad
|
||
|
|
// configuration is refused here instead of reaching the runtime.
|
||
|
|
const MinInterval = time.Second
|
||
|
|
|
||
|
|
// PublishDueWith publishes every post whose publish_at date is today or
|
||
|
|
// earlier and returns the slugs that changed, in store order, together with
|
||
|
|
// the number of posts whose save failed. Publishing removes the publish_at
|
||
|
|
// key and defaults date to it. The sink is called once per published post,
|
||
|
|
// so the caller can deliver the same post.published webhook the admin
|
||
|
|
// delivers; it may be nil. A post whose save fails is skipped, so one bad
|
||
|
|
// file cannot stop the sweep.
|
||
|
|
func PublishDueWith(st Store, onPublished func(*post.Post)) ([]string, int) {
|
||
|
|
today := post.TodayUTC()
|
||
|
|
var published []string
|
||
|
|
failures := 0
|
||
|
|
for _, cached := range st.All() {
|
||
|
|
// Cached posts are shared with request handlers; mutate a clone.
|
||
|
|
p := cached.Clone()
|
||
|
|
due, ok := p.DueAt()
|
||
|
|
if !ok || due.After(today) {
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
p.Metadata.Delete("publish_at")
|
||
|
|
if _, hasDate := p.Metadata.Get("date"); !hasDate {
|
||
|
|
p.Metadata.Set("date", interpres.LocalDate{Time: due})
|
||
|
|
}
|
||
|
|
if _, err := st.Save(p); err != nil {
|
||
|
|
failures++
|
||
|
|
slog.Warn("scheduler: cannot publish", "slug", p.Slug(), "error", err)
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
published = append(published, p.Slug())
|
||
|
|
slog.Info("scheduler published", "slug", p.Slug())
|
||
|
|
if onPublished != nil {
|
||
|
|
onPublished(p)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return published, failures
|
||
|
|
}
|
||
|
|
|
||
|
|
// DuePosts lists the posts PublishDue would publish.
|
||
|
|
func DuePosts(st Store) []*post.Post {
|
||
|
|
today := post.TodayUTC()
|
||
|
|
var due []*post.Post
|
||
|
|
for _, p := range st.All() {
|
||
|
|
if at, ok := p.DueAt(); ok && !at.After(today) {
|
||
|
|
due = append(due, p)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return due
|
||
|
|
}
|
||
|
|
|
||
|
|
// Run publishes due posts once at start-up and then every interval
|
||
|
|
// until stop is closed. An interval below MinInterval is reported and
|
||
|
|
// the loop does not start.
|
||
|
|
func Run(st Store, interval time.Duration, stop <-chan struct{}, onPublished func(*post.Post)) {
|
||
|
|
if interval < MinInterval {
|
||
|
|
slog.Error("scheduler: interval below the minimum, loop not started",
|
||
|
|
"interval", interval, "minimum", MinInterval)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
slog.Info("scheduler started", "interval", interval)
|
||
|
|
if published, failures := PublishDueWith(st, onPublished); len(published) > 0 || failures > 0 {
|
||
|
|
slog.Info("scheduler: start-up sweep", "published", len(published), "failed", failures)
|
||
|
|
}
|
||
|
|
ticker := time.NewTicker(interval)
|
||
|
|
defer ticker.Stop()
|
||
|
|
for {
|
||
|
|
select {
|
||
|
|
case <-stop:
|
||
|
|
return
|
||
|
|
case <-ticker.C:
|
||
|
|
PublishDueWith(st, onPublished)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|