126 lines
3.7 KiB
Go
126 lines
3.7 KiB
Go
package notifier
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log"
|
|
"time"
|
|
|
|
"naviwatcher/internal/config"
|
|
"naviwatcher/internal/database"
|
|
"naviwatcher/internal/scanner"
|
|
)
|
|
|
|
// NotifyOnce queries for releases that have not yet been notified, builds a
|
|
// digest, sends it through the given Sender, and marks each release as sent.
|
|
// Releases already present in notifications_sent are excluded upstream by
|
|
// GetUnnotifiedReleases, so this is idempotent across runs.
|
|
//
|
|
// If there are no unnotified releases the digest reports "no new missing
|
|
// releases" and nothing is marked sent (there is nothing to mark).
|
|
//
|
|
// uiBaseURL is the externally-reachable base URL of the Web UI, appended to the
|
|
// digest so operators can jump to the dashboard.
|
|
func NotifyOnce(ctx context.Context, db *database.DB, sender Sender, cfg config.TelegramConfig, uiBaseURL string) (int, error) {
|
|
if sender == nil {
|
|
return 0, fmt.Errorf("notifier: sender must not be nil")
|
|
}
|
|
|
|
unnotified, err := database.GetUnnotifiedReleases(db)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("notifier: query unnotified releases: %w", err)
|
|
}
|
|
|
|
missing := make([]scanner.MissingRelease, 0, len(unnotified))
|
|
for _, r := range unnotified {
|
|
missing = append(missing, scanner.MissingRelease{
|
|
RGID: r.RGID,
|
|
ArtistID: r.ArtistID,
|
|
Title: r.Title,
|
|
Type: r.Type,
|
|
ReleaseDate: r.ReleaseDate,
|
|
})
|
|
}
|
|
|
|
message := FormatDigest(missing, uiBaseURL)
|
|
if err := sender.Send(ctx, message); err != nil {
|
|
return 0, fmt.Errorf("notifier: send digest: %w", err)
|
|
}
|
|
|
|
for _, r := range unnotified {
|
|
if err := database.MarkNotificationSent(db, r.RGID); err != nil {
|
|
return 0, fmt.Errorf("notifier: mark sent for %s: %w", r.RGID, err)
|
|
}
|
|
}
|
|
return len(unnotified), nil
|
|
}
|
|
|
|
// Schedule produces the next firing time strictly after the given time. It
|
|
// mirrors the robfig/cron Schedule interface so cron specs and simple
|
|
// interval-based schedules are interchangeable and testable.
|
|
type Schedule interface {
|
|
Next(time.Time) time.Time
|
|
}
|
|
|
|
// notifyFunc is the unit of work the scheduler runs on each firing. It mirrors
|
|
// the signature of NotifyOnce so the scheduler can be tested with a stub.
|
|
type notifyFunc func(ctx context.Context) error
|
|
|
|
// StartScheduler runs the notify function on a schedule until ctx is cancelled.
|
|
// It is no-op-safe: if enabled is false it returns immediately without starting
|
|
// a goroutine. Each firing runs in its own goroutine so a slow send does not
|
|
// delay the next scheduled tick; the scheduler still computes the next tick from
|
|
// the wall clock and does not drift.
|
|
//
|
|
// The schedule and notify function are injectable so tests can drive a fixed or
|
|
// frequent schedule without a real cron spec or Telegram server.
|
|
func StartScheduler(ctx context.Context, enabled bool, schedule Schedule, notify notifyFunc, now func() time.Time) {
|
|
if !enabled || schedule == nil || notify == nil {
|
|
log.Println("Notifier scheduler disabled or misconfigured; not starting.")
|
|
return
|
|
}
|
|
if now == nil {
|
|
now = time.Now
|
|
}
|
|
|
|
go func() {
|
|
timer := time.NewTimer(0)
|
|
defer timer.Stop()
|
|
// Fire immediately on start (startup digest), then schedule subsequent runs.
|
|
first := true
|
|
for {
|
|
var wait time.Duration
|
|
if first {
|
|
first = false
|
|
wait = 0
|
|
} else {
|
|
next := schedule.Next(now())
|
|
if next.IsZero() {
|
|
log.Println("Notifier schedule has no next fire; stopping scheduler.")
|
|
return
|
|
}
|
|
wait = time.Until(next)
|
|
if wait < 0 {
|
|
wait = 0
|
|
}
|
|
}
|
|
|
|
timer.Reset(wait)
|
|
select {
|
|
case <-ctx.Done():
|
|
log.Println("Notifier scheduler stopped.")
|
|
return
|
|
case <-timer.C:
|
|
go func() {
|
|
if err := notify(ctx); err != nil {
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
log.Printf("Notifier run failed: %v", err)
|
|
}
|
|
}()
|
|
}
|
|
}
|
|
}()
|
|
}
|