nostr-poster/internal/scheduler/scheduler.go

// internal/scheduler/scheduler.go
package scheduler

import (
	"context"
	"fmt"
	"path/filepath"
	"strings"
	"sync"
	"encoding/json"
	

	"github.com/robfig/cron/v3"
	"git.sovbit.dev/Enki/nostr-poster/internal/db"
	"git.sovbit.dev/Enki/nostr-poster/internal/models"
	"git.sovbit.dev/Enki/nostr-poster/internal/utils"
	"go.uber.org/zap"
)

// MediaUploader defines the interface for uploading media
type MediaUploader interface {
    UploadFile(filePath string, caption string, altText string) (string, string, error)
    DeleteFile(fileHash string) error
    WithCustomURL(customURL string) MediaUploader
}

// PostPublisher defines the interface for publishing posts
type PostPublisher interface {
	PublishEvent(ctx context.Context, event interface{}) ([]string, error)
}

// ContentPoster defines the function to post content
type ContentPoster func(
	pubkey string,
	contentPath string,
	contentType string,
	mediaURL string,
	mediaHash string,
	caption string,
	hashtags []string,
) error

// ContentPosterWithEncoding defines a function to post content and return encoded event
type ContentPosterWithEncoding func(
	pubkey string,
	contentPath string,
	contentType string,
	mediaURL string,
	mediaHash string,
	caption string,
	hashtags []string,
) (*models.EventResponse, error)

// Scheduler manages scheduled content posting
type Scheduler struct {
	db                 *db.DB
	cron               *cron.Cron
	logger             *zap.Logger
	contentDir         string
	archiveDir         string
	nip94Uploader      MediaUploader
	blossomUploader    MediaUploader
	postContent        ContentPoster
	postContentEncoded ContentPosterWithEncoding // New field for NIP-19 support
	botJobs            map[int64]cron.EntryID
	mu                 sync.RWMutex
}

// NewScheduler creates a new content scheduler
func NewScheduler(
	db *db.DB,
	logger *zap.Logger,
	contentDir string,
	archiveDir string,
	nip94Uploader MediaUploader,
	blossomUploader MediaUploader,
	postContent ContentPoster,
	postContentEncoded ContentPosterWithEncoding, // New parameter for NIP-19 support
) *Scheduler {
	if logger == nil {
		// Create a default logger
		var err error
		logger, err = zap.NewProduction()
		if err != nil {
			logger = zap.NewNop()
		}
	}
	
	// Create a new cron scheduler with seconds precision
	cronScheduler := cron.New(cron.WithSeconds())
	
	return &Scheduler{
		db:                 db,
		cron:               cronScheduler,
		logger:             logger,
		contentDir:         contentDir,
		archiveDir:         archiveDir,
		nip94Uploader:      nip94Uploader,
		blossomUploader:    blossomUploader,
		postContent:        postContent,
		postContentEncoded: postContentEncoded, // Initialize the new field
		botJobs:            make(map[int64]cron.EntryID),
	}
}

// Start starts the scheduler
func (s *Scheduler) Start() error {
	// Load all bots with enabled post configs
	query := `
	SELECT b.*, pc.*, mc.*
	FROM bots b
	JOIN post_config pc ON b.id = pc.bot_id
	JOIN media_config mc ON b.id = mc.bot_id
	WHERE pc.enabled = 1
	`
	
	rows, err := s.db.Queryx(query)
	if err != nil {
		return fmt.Errorf("failed to load bots: %w", err)
	}
	defer rows.Close()
	
	// Process each bot
	for rows.Next() {
		var bot models.Bot
		var postConfig models.PostConfig
		var mediaConfig models.MediaConfig
		
		// Map the results to our structs
		err := rows.Scan(
			&bot.ID, &bot.Pubkey, &bot.EncryptedPrivkey, &bot.Name, &bot.DisplayName, 
			&bot.Bio, &bot.Nip05, &bot.ZapAddress, &bot.ProfilePicture, &bot.Banner, 
			&bot.CreatedAt, &bot.OwnerPubkey,
			&postConfig.ID, &postConfig.BotID, &postConfig.Hashtags, &postConfig.IntervalMinutes, 
			&postConfig.PostTemplate, &postConfig.Enabled,
			&mediaConfig.ID, &mediaConfig.BotID, &mediaConfig.PrimaryService, 
			&mediaConfig.FallbackService, &mediaConfig.Nip94ServerURL, &mediaConfig.BlossomServerURL,
		)
		if err != nil {
			s.logger.Error("Failed to scan bot row", zap.Error(err))
			continue
		}
		
		// Set the associated config
		bot.PostConfig = &postConfig
		bot.MediaConfig = &mediaConfig
		
		// Schedule the bot
		if err := s.ScheduleBot(&bot); err != nil {
			s.logger.Error("Failed to schedule bot", 
				zap.String("name", bot.Name),
				zap.Error(err))
		}
	}
	
	// Start the cron scheduler
	s.cron.Start()
	
	return nil
}

// Stop stops the scheduler
func (s *Scheduler) Stop() {
	if s.cron != nil {
		s.cron.Stop()
	}
}

// ScheduleBot schedules posting for a specific bot
func (s *Scheduler) ScheduleBot(bot *models.Bot) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	
	// Unschedule if already scheduled
	if entryID, exists := s.botJobs[bot.ID]; exists {
		s.cron.Remove(entryID)
		delete(s.botJobs, bot.ID)
	}
	
	// Create bot-specific content directory if it doesn't exist
	botContentDir := filepath.Join(s.contentDir, fmt.Sprintf("bot_%d", bot.ID))
	if err := utils.EnsureDir(botContentDir); err != nil {
		return fmt.Errorf("failed to create bot content directory: %w", err)
	}
	
	// Create bot-specific archive directory if it doesn't exist
	botArchiveDir := filepath.Join(s.archiveDir, fmt.Sprintf("bot_%d", bot.ID))
	if err := utils.EnsureDir(botArchiveDir); err != nil {
		return fmt.Errorf("failed to create bot archive directory: %w", err)
	}
	
	// Parse hashtags
	var hashtags []string
	if bot.PostConfig.Hashtags != "" {
		if err := json.Unmarshal([]byte(bot.PostConfig.Hashtags), &hashtags); err != nil {
			s.logger.Warn("Failed to parse hashtags", 
				zap.String("hashtags", bot.PostConfig.Hashtags),
				zap.Error(err))
			hashtags = []string{}
		}
	}
	
	// Create the schedule expression
	// Format: "0 */X * * * *" where X is the interval in minutes
	// This runs at 0 seconds, every X minutes
	schedule := fmt.Sprintf("0 */%d * * * *", bot.PostConfig.IntervalMinutes)
	
	// Create a job function that captures the bot's information
	jobFunc := func() {
		s.logger.Info("Running scheduled post", 
			zap.Int64("bot_id", bot.ID),
			zap.String("name", bot.Name))
		
		// Get a random media file
		contentPath, err := utils.GetRandomFile(botContentDir, utils.GetAllSupportedMediaExtensions())
		if err != nil {
			s.logger.Error("Failed to get random file", 
				zap.Int64("bot_id", bot.ID),
				zap.Error(err))
			return
		}
		
		// Get content type
		contentType, err := utils.GetFileContentType(contentPath)
		if err != nil {
			s.logger.Error("Failed to determine content type",
				zap.Int64("bot_id", bot.ID),
				zap.String("file", contentPath),
				zap.Error(err))
			return
		}
		
		// Select the appropriate uploader
		var uploader MediaUploader
		if bot.MediaConfig.PrimaryService == "blossom" {
			uploader = s.blossomUploader
		} else {
			uploader = s.nip94Uploader
		}
		
		// Upload the file
		mediaURL, mediaHash, err := uploader.UploadFile(contentPath, "", "")
		if err != nil {
			// Try fallback if available
			if bot.MediaConfig.FallbackService != "" && bot.MediaConfig.FallbackService != bot.MediaConfig.PrimaryService {
				s.logger.Warn("Primary upload failed, trying fallback",
					zap.Int64("bot_id", bot.ID),
					zap.String("primary", bot.MediaConfig.PrimaryService),
					zap.String("fallback", bot.MediaConfig.FallbackService),
					zap.Error(err))
				
				if bot.MediaConfig.FallbackService == "blossom" {
					uploader = s.blossomUploader
				} else {
					uploader = s.nip94Uploader
				}
				
				mediaURL, mediaHash, err = uploader.UploadFile(contentPath, "", "")
				if err != nil {
					s.logger.Error("Fallback upload failed",
						zap.Int64("bot_id", bot.ID),
						zap.String("file", contentPath),
						zap.Error(err))
					return
				}
			} else {
				s.logger.Error("File upload failed",
					zap.Int64("bot_id", bot.ID),
					zap.String("file", contentPath),
					zap.Error(err))
				return
			}
		}
		
		// Get the base filename without extension
		filename := filepath.Base(contentPath)
		caption := strings.TrimSuffix(filename, filepath.Ext(filename))
		
		// Post the content - use encoded version if available, otherwise use the original
		var postErr error
		
		if s.postContentEncoded != nil {
			// Use the NIP-19 encoded version
			encodedEvent, err := s.postContentEncoded(
				bot.Pubkey,
				contentPath,
				contentType,
				mediaURL,
				mediaHash,
				caption,
				hashtags,
			)
			
			if err != nil {
				s.logger.Error("Failed to post content with encoding",
					zap.Int64("bot_id", bot.ID),
					zap.String("file", contentPath),
					zap.Error(err))
				postErr = err
			} else {
				// Success with encoded version
				s.logger.Info("Successfully posted content with NIP-19 encoding",
					zap.Int64("bot_id", bot.ID),
					zap.String("file", contentPath),
					zap.String("media_url", mediaURL),
					zap.String("event_id", encodedEvent.ID),
					zap.String("note", encodedEvent.Note),
					zap.String("nevent", encodedEvent.Nevent))
				postErr = nil
			}
		} else {
			// Fall back to original function
			postErr = s.postContent(
				bot.Pubkey,
				contentPath,
				contentType,
				mediaURL,
				mediaHash,
				caption,
				hashtags,
			)
			
			if postErr != nil {
				s.logger.Error("Failed to post content",
					zap.Int64("bot_id", bot.ID),
					zap.String("file", contentPath),
					zap.Error(postErr))
			} else {
				s.logger.Info("Successfully posted content",
					zap.Int64("bot_id", bot.ID),
					zap.String("file", contentPath),
					zap.String("media_url", mediaURL))
			}
		}
		
		// If posting failed, return without archiving the file
		if postErr != nil {
			return
		}
		
		// Move the file to the archive directory
		archivePath := filepath.Join(botArchiveDir, filepath.Base(contentPath))
		if err := utils.MoveFile(contentPath, archivePath); err != nil {
			s.logger.Error("Failed to archive file",
				zap.Int64("bot_id", bot.ID),
				zap.String("file", contentPath),
				zap.String("archive", archivePath),
				zap.Error(err))
			return
		}
	}
	
	// Schedule the job
	entryID, err := s.cron.AddFunc(schedule, jobFunc)
	if err != nil {
		return fmt.Errorf("failed to schedule bot: %w", err)
	}
	
	// Store the entry ID
	s.botJobs[bot.ID] = entryID
	
	s.logger.Info("Bot scheduled successfully",
		zap.Int64("bot_id", bot.ID),
		zap.String("name", bot.Name),
		zap.String("schedule", schedule))
	
	return nil
}

// UnscheduleBot removes a bot from the scheduler
func (s *Scheduler) UnscheduleBot(botID int64) {
	s.mu.Lock()
	defer s.mu.Unlock()
	
	if entryID, exists := s.botJobs[botID]; exists {
		s.cron.Remove(entryID)
		delete(s.botJobs, botID)
		s.logger.Info("Bot unscheduled", zap.Int64("bot_id", botID))
	}
}

// UpdateBotSchedule updates a bot's schedule
func (s *Scheduler) UpdateBotSchedule(bot *models.Bot) error {
	// Unschedule first
	s.UnscheduleBot(bot.ID)
	
	// Reschedule with new configuration
	return s.ScheduleBot(bot)
}

// GetNIP94Uploader returns the NIP-94 uploader
func (s *Scheduler) GetNIP94Uploader() MediaUploader {
    return s.nip94Uploader
}

// GetBlossomUploader returns the Blossom uploader
func (s *Scheduler) GetBlossomUploader() MediaUploader {
    return s.blossomUploader
}

func (s *Scheduler) GetContentDir() string {
    return s.contentDir
}

// RunNow triggers an immediate post for a bot
func (s *Scheduler) RunNow(botID int64) error {
	// Load the bot with its configurations
	var bot models.Bot
	err := s.db.Get(&bot, "SELECT * FROM bots WHERE id = ?", botID)
	if err != nil {
		return fmt.Errorf("failed to load bot: %w", err)
	}
	
	// Load post config
	var postConfig models.PostConfig
	err = s.db.Get(&postConfig, "SELECT * FROM post_config WHERE bot_id = ?", botID)
	if err != nil {
		return fmt.Errorf("failed to load post config: %w", err)
	}
	bot.PostConfig = &postConfig
	
	// Load media config
	var mediaConfig models.MediaConfig
	err = s.db.Get(&mediaConfig, "SELECT * FROM media_config WHERE bot_id = ?", botID)
	if err != nil {
		return fmt.Errorf("failed to load media config: %w", err)
	}
	bot.MediaConfig = &mediaConfig
	
	// Create job and run it
	s.ScheduleBot(&bot)
	
	s.mu.RLock()
	entryID, exists := s.botJobs[botID]
	s.mu.RUnlock()
	
	if !exists {
		return fmt.Errorf("bot not scheduled")
	}
	
	// Get the job
	entry := s.cron.Entry(entryID)
	if entry.Job == nil {
		return fmt.Errorf("job not found")
	}
	
	// Run the job
	entry.Job.Run()
	
	return nil
}