Netpulse_SasS/server/internal/grpcapi/ncm_schedule.go

package grpcapi

import (
	"context"
	"time"

	"github.com/netpulse/netpulse/server/internal/cronx"
)

// scheduleLockKey — окреме advisory-блокування під планувальник бекапів.
const scheduleLockKey = 0x6e70_6263 // "npbc"

// ScheduleBackups ставить бекапи в чергу за розкладом.
//
// Раз на хвилину, бо cron дрібніший за хвилину не буває. Кілька
// екземплярів сервера безпечні: тік бере advisory-блокування, тож
// розклад розкручує рівно один — інакше кожен хост отримував би стільки
// завдань, скільки інстансів у кластері.
func (s *Service) ScheduleBackups(ctx context.Context) {
	t := time.NewTicker(time.Minute)
	defer t.Stop()

	s.log.Info("планувальник бекапів запущено")

	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			if n := s.scheduleTick(ctx); n > 0 {
				s.log.Info("бекапи поставлено в чергу за розкладом", "хостів", n)
			}
		}
	}
}

func (s *Service) scheduleTick(ctx context.Context) int {
	conn, err := s.store.WorkerPool().Acquire(ctx)
	if err != nil {
		s.log.Error("планувальник: з'єднання", "err", err)
		return 0
	}
	defer conn.Release()

	var got bool
	if err := conn.QueryRow(ctx, `SELECT pg_try_advisory_lock($1)`,
		int64(scheduleLockKey)).Scan(&got); err != nil {
		s.log.Error("планувальник: блокування", "err", err)
		return 0
	}
	if !got {
		// Розклад розкручує інший інстанс — штатний стан.
		return 0
	}
	defer func() {
		_, _ = conn.Exec(context.WithoutCancel(ctx),
			`SELECT pg_advisory_unlock($1)`, int64(scheduleLockKey))
	}()

	due, err := s.store.DuePolicies(ctx, 200)
	if err != nil {
		s.log.Error("планувальник: вибірка розкладів", "err", err)
		return 0
	}

	now := time.Now()
	queued := 0

	for _, d := range due {
		sched, err := cronx.Parse(d.Cron)
		if err != nil {
			// Некоректний розклад не має зупиняти решту й не має
			// повторюватись щохвилини у журналі: відсуваємо політику на
			// добу, щоб скарга лишилась помітною, але не набридала.
			s.log.Warn("некоректний розклад бекапу",
				"device", d.DeviceID, "cron", d.Cron, "err", err)
			_ = s.store.SetNextBackup(ctx, d.DeviceID, now.Add(24*time.Hour))
			continue
		}

		next, err := sched.Next(now)
		if err != nil {
			s.log.Warn("розклад без найближчого запуску",
				"device", d.DeviceID, "cron", d.Cron)
			_ = s.store.SetNextBackup(ctx, d.DeviceID, now.Add(24*time.Hour))
			continue
		}

		// Спершу переносимо позначку, потім ставимо завдання.
		//
		// Саме в такому порядку: якщо процес упаде між двома кроками,
		// хост пропустить один бекап. Зворотний порядок дав би
		// нескінченну чергу однакових завдань — а це гірше, бо кладе
		// й пристрій, і сервер.
		if err := s.store.SetNextBackup(ctx, d.DeviceID, next); err != nil {
			s.log.Error("планувальник: перенесення розкладу", "err", err)
			continue
		}

		if _, err := s.store.EnqueueConfigJob(ctx, d.TenantID, d.DeviceID, "schedule", ""); err != nil {
			s.log.Error("планувальник: постановка завдання",
				"device", d.DeviceID, "err", err)
			continue
		}
		queued++
	}
	return queued
}