feat(cron): add in-process cron scheduler

Add robfig/cron-based task scheduler with per-UTC-date idempotency guard. Enables
scheduled jobs (e.g., daily price checks) without external scheduler dependency.
Module interface updated to support cron registration.
This commit is contained in:
tiennm99 committed 2026-06-28 09:58:12 +07:00
1 parent bfaecef86c
commit 2432c910dc
5 files changed
+293 -8

No files matched your search

+72 -4
View File
@@ -12,6 +12,7 @@ import (
"github.com/tiennm99/miti99bot/internal/log"
"github.com/tiennm99/miti99bot/internal/modules"
"github.com/tiennm99/miti99bot/internal/storage"
)
// terminalKind classifies a permanent send failure by blast radius.
@@ -76,15 +77,24 @@ func classifyTerminal(err error) terminalKind {
return terminalNone
}
// dailyPushCronName is the cron route segment + EventBridge schedule key.
// dailyPushCronName is the cron route segment + in-process scheduler key.
// Must match the regex in internal/server/router.go (^[a-z0-9_]{1,32}$).
const dailyPushCronName = "lolschedule_daily_push"
// dailyPushSchedule is documentation only — the real fire time lives in
// EventBridge Scheduler. Cron expression is UTC;
// 01:00 UTC == 08:00 ICT.
// dailyPushSchedule drives the in-process scheduler (internal/cron) on
// self-host; it was also the documented EventBridge time on AWS. Cron
// expression is UTC; 01:00 UTC == 08:00 ICT.
const dailyPushSchedule = "0 1 * * *"
// lastPushDateKey records the UTC date (YYYY-MM-DD) of the most recent
// completed daily push. The handler claims this key before fanning out and
// no-ops if it is already today's date, making the push idempotent per UTC
// date. This defends against every double-fire window — cutover overlap
// (EventBridge still live while the container's scheduler runs), rolling
// deploys that briefly run two containers, and operator misconfiguration —
// none of which a single trigger source can prevent.
const lastPushDateKey = "daily_push:last_date"
// telegramRateLimitThreshold is the subscriber count above which we throttle
// sends to stay under Telegram's global 30 msg/sec cap. Below it we send hot.
const telegramRateLimitThreshold = 30
@@ -119,6 +129,50 @@ func (s *state) dailyPushHandler(ctx context.Context, deps modules.Deps) error {
return runDailyPush(ctx, s, deps.Bot)
}
// claimDailyPush atomically records that today's UTC push is happening and
// reports whether THIS caller won the claim. It is the idempotency primitive
// for the daily push: a winner proceeds to fan out; a loser (another trigger
// already claimed today) returns false and sends nothing.
//
// The claim is a compare-and-swap on lastPushDateKey so two simultaneous
// triggers cannot both win. Every real KV backend implements
// CompareAndSwapStore; if a bare store without CAS is used (only in narrow
// tests), it falls back to a plain Put — losing the atomic guarantee but
// preserving the "no-op if already today" behaviour.
func claimDailyPush(ctx context.Context, kv storage.KVStore, today string) (bool, error) {
current, err := kv.Get(ctx, lastPushDateKey)
switch {
case err == nil:
if string(current) == today {
return false, nil // already pushed today
}
case errors.Is(err, storage.ErrNotFound):
current = nil // never pushed
default:
return false, err
}
cas, ok := kv.(storage.CompareAndSwapStore)
if !ok {
if err := kv.Put(ctx, lastPushDateKey, []byte(today)); err != nil {
return false, err
}
return true, nil
}
var expected []byte
if len(current) > 0 {
expected = current
}
if err := cas.CompareAndSwap(ctx, lastPushDateKey, expected, []byte(today)); err != nil {
if errors.Is(err, storage.ErrConflict) {
return false, nil // another trigger claimed today first
}
return false, err
}
return true, nil
}
// runDailyPush is the testable core: fetch subscribers, fetch today's matches,
// fan out to every subscriber. Per-chat send failures are logged but do not
// abort the batch — one bad chat does not deny the rest.
@@ -144,6 +198,20 @@ func runDailyPush(ctx context.Context, s *state, sender messageSender) error {
filtered := FilterMajor(events)
text := RenderToday(filtered, from)
// Idempotency gate: claim today's push before sending. A lost claim means
// another trigger already pushed (or is pushing) for this UTC date, so we
// send nothing. Placed after the fetch so a transient fetch failure does
// not consume the day's claim.
today := s.now().UTC().Format("2006-01-02")
won, err := claimDailyPush(ctx, s.kv, today)
if err != nil {
return fmt.Errorf("lolschedule daily push: claim date: %w", err)
}
if !won {
log.Info("lolschedule daily push: already pushed today, skipping", "date", today)
return nil
}
throttle := len(subs) > telegramRateLimitThreshold
var sent, failed int
deadChats := map[int64]struct{}{}
+27
View File
@@ -167,6 +167,33 @@ func TestRunDailyPush_ForwardsMessageThreadID(t *testing.T) {
}
}
// TestRunDailyPush_IdempotentPerDate locks in the double-fire guard: invoking
// the handler twice on the same UTC date sends each subscriber exactly one
// digest. Defends against cutover overlap, rolling-deploy overlap, and operator
// misconfiguration (all double-fire windows the daily push must survive).
func TestRunDailyPush_IdempotentPerDate(t *testing.T) {
s := newTestState(t)
seedFreshCache(t, s.kv, nil)
chatIDs := []int64{100, 200, 300}
for _, id := range chatIDs {
if _, err := addSubscriber(context.Background(), s.kv, id, 0); err != nil {
t.Fatalf("addSubscriber %d: %v", id, err)
}
}
sender := &fakeSender{}
for i := 0; i < 2; i++ {
if err := runDailyPush(context.Background(), s, sender); err != nil {
t.Fatalf("runDailyPush call %d: %v", i+1, err)
}
}
if len(sender.calls) != len(chatIDs) {
t.Errorf("two same-date pushes sent %d messages, want %d (one per subscriber)",
len(sender.calls), len(chatIDs))
}
}
func TestRunDailyPush_PartialFailureContinues(t *testing.T) {
s := newTestState(t)
seedFreshCache(t, s.kv, nil)
+6 -4
View File
@@ -28,9 +28,11 @@ const (
// purely for logging/metrics, not flow control.
type CommandHandler func(ctx context.Context, b *bot.Bot, update *models.Update) error
// CronHandler runs when EventBridge Scheduler hits /cron/{name}. Crons receive the
// per-module-prefixed Deps via the registry; handlers should not capture the
// base Deps from the factory closure or KV writes will collide across modules.
// CronHandler runs when a cron fires — driven by the in-process scheduler
// (internal/cron) on self-host, or by a POST to /cron/{name} for manual
// triggers. Crons receive the per-module-prefixed Deps via the registry;
// handlers should not capture the base Deps from the factory closure or KV
// writes will collide across modules.
type CronHandler func(ctx context.Context, deps Deps) error
// Command is a single Telegram bot command exposed by a module.
@@ -43,7 +45,7 @@ type Command struct {
// Cron is a single scheduled job exposed by a module.
type Cron struct {
Schedule string // documentation only; real schedule lives in EventBridge Scheduler
Schedule string // 5-field cron expr (UTC); the in-process scheduler fires the handler on it
Name string // unique within module
Handler CronHandler // required
}