<commit_before>package sarah

import (
	"fmt"
	"github.com/oklahomer/go-sarah/log"
	"github.com/oklahomer/go-sarah/worker"
	"golang.org/x/net/context"
	"path/filepath"
	"strings"
	"sync"
	"time"
)

// Config contains some configuration variables for Runner and its underlying structs.
type Config struct {
	Worker           *worker.Config `json:"worker" yaml:"worker"`
	PluginConfigRoot string         `json:"plugin_config_root" yaml:"plugin_config_root"`
	TimeZone         string         `json:"timezone" yaml:"timezone"`
}

// NewConfig creates and returns new Config instance with default settings.
// Use json.Unmarshal, yaml.Unmarshal, or manual manipulation to overload default values.
func NewConfig() *Config {
	return &Config{
		Worker: worker.NewConfig(),
		// PluginConfigRoot defines the root directory to be used when searching for plugins' configuration files.
		PluginConfigRoot: "",
		TimeZone:         time.Now().Location().String(),
	}
}

// Runner is the core of sarah.
//
// This takes care of lifecycle of each Bot implementation and plugin execution;
// Bot is responsible for bot-specific implementation such as connection handling, message reception and sending.
//
// Developers can register desired number of Bot and Commands to create own bot experience.
type Runner struct {
	config         *Config
	bots           []Bot
	scheduledTasks map[BotType][]ScheduledTask
	alerters       *alerters
}

// NewRunner creates and return new Runner instance.
func NewRunner(config *Config) *Runner {
	return &Runner{
		config:         config,
		bots:           []Bot{},
		scheduledTasks: make(map[BotType][]ScheduledTask),
		alerters:       &alerters{},
	}
}

// RegisterBot register given Bot implementation to Runner instance
func (runner *Runner) RegisterBot(bot Bot) {
	runner.bots = append(runner.bots, bot)
}

// RegisterScheduledTask register given ScheduledTask implementation to Runner.
// Once Runner.Run is called, registered tasks are scheduled and will be executed as configured.
func (runner *Runner) RegisterScheduledTask(botType BotType, task ScheduledTask) {
	tasks, ok := runner.scheduledTasks[botType]
	if !ok {
		tasks = []ScheduledTask{}
	}

	runner.scheduledTasks[botType] = append(tasks, task)
}

// RegisterAlerter register given Alerter implementation to Runner instance.
// Developer can register as many Alerters as one wishes.
func (runner *Runner) RegisterAlerter(alerter Alerter) {
	runner.alerters.appendAlerter(alerter)
}

// Run starts Bot interaction.
// At this point Runner starts its internal workers and schedulers, runs each bot, and starts listening to incoming messages.
func (runner *Runner) Run(ctx context.Context) {
	workerJob := worker.Run(ctx, runner.config.Worker)

	loc, locErr := time.LoadLocation(runner.config.TimeZone)
	if locErr != nil {
		panic(fmt.Sprintf("Given timezone can't be converted to time.Location: %s.", locErr.Error()))
	}
	taskScheduler := runScheduler(ctx, loc)

	watcher, err := runConfigWatcher(ctx)
	if err != nil {
		panic(fmt.Sprintf("failed to run watcher: %s.", err.Error()))
	}

	var wg sync.WaitGroup
	for _, bot := range runner.bots {
		wg.Add(1)

		botType := bot.BotType()
		log.Infof("starting %s", botType.String())

		// each Bot has its own context propagating Runner's lifecycle
		botCtx, errNotifier := botSupervisor(ctx, botType, runner.alerters)

		// run Bot
		receiveInput := setupInputReceiver(botCtx, bot, workerJob)
		go bot.Run(botCtx, receiveInput, errNotifier)

		// setup config directory
		var configDir string
		if runner.config.PluginConfigRoot != "" {
			configDir = filepath.Join(runner.config.PluginConfigRoot, strings.ToLower(bot.BotType().String()))
		}

		// build commands with stashed builder settings
		commands := stashedCommandBuilders.build(botType, configDir)
		for _, command := range commands {
			bot.AppendCommand(command)
		}

		// Setup schedule registration function that tied to this particular bot type.
		updateSchedule := func(c context.Context, b Bot) func(ScheduledTask) error {
			return func(t ScheduledTask) error {
				log.Infof("registering task for %s: %s", b.BotType().String(), t.Identifier())
				return taskScheduler.update(b.BotType(), t, func() {
					executeScheduledTask(c, b, t)
				})
			}
		}(botCtx, bot) // Beware of closure...

		// set cron jobs with registered scheduled task
		if tasks, ok := runner.scheduledTasks[botType]; ok {
			for _, task := range tasks {
				if task.Schedule() == "" {
					log.Errorf("failed to schedule a task. id: %s. reason: %s.", task.Identifier(), "No schedule given.")
					continue
				}
				if err := updateSchedule(task); err != nil {
					log.Errorf("failed to schedule a task. id: %s. reason: %s.", task.Identifier(), err.Error())
				}
			}
		}

		// build scheduled task with stashed builder settings
		tasks := stashedScheduledTaskBuilders.build(botType, configDir)
		for _, task := range tasks {
			if err := updateSchedule(task); err != nil {
				log.Errorf("failed to schedule a task. id: %s. reason: %s.", task.Identifier(), err.Error())
			}
		}

		// supervise configuration files' directory
		if configDir != "" {
			err := watcher.watch(botCtx, botType, configDir, pluginUpdaterFunc(botCtx, bot, taskScheduler))
			if err != nil {
				log.Errorf("failed to watch %s: %s", configDir, err.Error())
			}
		}

		go func(c context.Context) {
			select {
			case <-c.Done():
				wg.Done()
			}
		}(botCtx)
	}

	wg.Wait()
}

func pluginUpdaterFunc(botCtx context.Context, bot Bot, taskScheduler scheduler) func(string) {
	return func(path string) {
		dir, filename := filepath.Split(path)
		id := strings.TrimSuffix(filename, filepath.Ext(filename)) // buzz.yaml to buzz

		if builder := stashedCommandBuilders.find(bot.BotType(), id); builder != nil {
			log.Infof("start rebuilding command due to config file change: %s.", id)
			command, err := builder.Build(dir)
			if err != nil {
				log.Errorf("can't configure command. %s. %#v", err.Error(), builder)
			} else {
				bot.AppendCommand(command) // replaces the old one.
			}
		}<commit_msg>Leave detailed log on Bot failure<commit_after>package sarah

import (
	"fmt"
	"github.com/oklahomer/go-sarah/log"
	"github.com/oklahomer/go-sarah/worker"
	"golang.org/x/net/context"
	"path/filepath"
	"runtime"
	"strings"
	"sync"
	"time"
)

// Config contains some configuration variables for Runner and its underlying structs.
type Config struct {
	Worker           *worker.Config `json:"worker" yaml:"worker"`
	PluginConfigRoot string         `json:"plugin_config_root" yaml:"plugin_config_root"`
	TimeZone         string         `json:"timezone" yaml:"timezone"`
}

// NewConfig creates and returns new Config instance with default settings.
// Use json.Unmarshal, yaml.Unmarshal, or manual manipulation to overload default values.
func NewConfig() *Config {
	return &Config{
		Worker: worker.NewConfig(),
		// PluginConfigRoot defines the root directory to be used when searching for plugins' configuration files.
		PluginConfigRoot: "",
		TimeZone:         time.Now().Location().String(),
	}
}

// Runner is the core of sarah.
//
// This takes care of lifecycle of each Bot implementation and plugin execution;
// Bot is responsible for bot-specific implementation such as connection handling, message reception and sending.
//
// Developers can register desired number of Bot and Commands to create own bot experience.
type Runner struct {
	config         *Config
	bots           []Bot
	scheduledTasks map[BotType][]ScheduledTask
	alerters       *alerters
}

// NewRunner creates and return new Runner instance.
func NewRunner(config *Config) *Runner {
	return &Runner{
		config:         config,
		bots:           []Bot{},
		scheduledTasks: make(map[BotType][]ScheduledTask),
		alerters:       &alerters{},
	}
}

// RegisterBot register given Bot implementation to Runner instance
func (runner *Runner) RegisterBot(bot Bot) {
	runner.bots = append(runner.bots, bot)
}

// RegisterScheduledTask register given ScheduledTask implementation to Runner.
// Once Runner.Run is called, registered tasks are scheduled and will be executed as configured.
func (runner *Runner) RegisterScheduledTask(botType BotType, task ScheduledTask) {
	tasks, ok := runner.scheduledTasks[botType]
	if !ok {
		tasks = []ScheduledTask{}
	}

	runner.scheduledTasks[botType] = append(tasks, task)
}

// RegisterAlerter register given Alerter implementation to Runner instance.
// Developer can register as many Alerters as one wishes.
func (runner *Runner) RegisterAlerter(alerter Alerter) {
	runner.alerters.appendAlerter(alerter)
}

// Run starts Bot interaction.
// At this point Runner starts its internal workers and schedulers, runs each bot, and starts listening to incoming messages.
func (runner *Runner) Run(ctx context.Context) {
	workerJob := worker.Run(ctx, runner.config.Worker)

	loc, locErr := time.LoadLocation(runner.config.TimeZone)
	if locErr != nil {
		panic(fmt.Sprintf("Given timezone can't be converted to time.Location: %s.", locErr.Error()))
	}
	taskScheduler := runScheduler(ctx, loc)

	watcher, err := runConfigWatcher(ctx)
	if err != nil {
		panic(fmt.Sprintf("failed to run watcher: %s.", err.Error()))
	}

	var wg sync.WaitGroup
	for _, bot := range runner.bots {
		wg.Add(1)

		botType := bot.BotType()
		log.Infof("starting %s", botType.String())

		// Each Bot has its own context propagating Runner's lifecycle.
		botCtx, errNotifier := botSupervisor(ctx, botType, runner.alerters)

		// Prepare function  that receives Input.
		receiveInput := setupInputReceiver(ctx, bot, workerJob)

		// run Bot
		go runBot(botCtx, bot, receiveInput, errNotifier)

		// Setup config directory.
		var configDir string
		if runner.config.PluginConfigRoot != "" {
			configDir = filepath.Join(runner.config.PluginConfigRoot, strings.ToLower(bot.BotType().String()))
		}

		// build commands with stashed builder settings
		commands := stashedCommandBuilders.build(botType, configDir)
		for _, command := range commands {
			bot.AppendCommand(command)
		}

		// Setup schedule registration function that is tied to this particular bot type.
		updateSchedule := func(c context.Context, b Bot) func(ScheduledTask) error {
			return func(t ScheduledTask) error {
				log.Infof("registering task for %s: %s", b.BotType().String(), t.Identifier())
				return taskScheduler.update(b.BotType(), t, func() {
					executeScheduledTask(c, b, t)
				})
			}
		}(botCtx, bot) // Beware of closure...

		// Set cron jobs. Jobs include...
		// - pre-registered scheduled task
		// - those built with stashed builders
		tasks := append(runner.scheduledTasks[botType], stashedScheduledTaskBuilders.build(botType, configDir)...)
		for _, task := range tasks {
			// Make sure schedule is given. Especially those pre-registered tasks.
			if task.Schedule() == "" {
				log.Errorf("failed to schedule a task. id: %s. reason: %s.", task.Identifier(), "No schedule given.")
				continue
			}

			if err := updateSchedule(task); err != nil {
				log.Errorf("failed to schedule a task. id: %s. reason: %s.", task.Identifier(), err.Error())
			}
		}

		// supervise configuration files' directory
		if configDir != "" {
			err := watcher.watch(botCtx, botType, configDir, pluginUpdaterFunc(botCtx, bot, taskScheduler))
			if err != nil {
				log.Errorf("failed to watch %s: %s", configDir, err.Error())
			}
		}

		go func(c context.Context) {
			select {
			case <-c.Done():
				wg.Done()
			}
		}(botCtx)
	}

	wg.Wait()
}

func runBot(ctx context.Context, bot Bot, receiveInput func(Input) error, errNotifier func(error)) {
	// When bot panics, recover and tell as much detailed information as possible via error notification channel.
	// Notified channel sends alert to notify administrator.
	defer func() {
		if r := recover(); r != nil {
			stack := []string{fmt.Sprintf("panic in bot: %s. %#v.", bot.BotType(), r)}

			// Inform stack trace
			for depth := 0; ; depth++ {
				_, src, line, ok := runtime.Caller(depth)
				if !ok {
					break
				}
				stack = append(stack, fmt.Sprintf(" -> depth:%d. file:%s. line:%d.", depth, src, line))
			}

			errNotifier(NewBotNonContinuableError(strings.Join(stack, "\n")))
		}
	}()

	bot.Run(ctx, receiveInput, errNotifier)
}

func pluginUpdaterFunc(botCtx context.Context, bot Bot, taskScheduler scheduler) func(string) {
	return func(path string) {
		dir, filename := filepath.Split(path)
		id := strings.TrimSuffix(filename, filepath.Ext(filename)) // buzz.yaml to buzz

		if builder := stashedCommandBuilders.find(bot.BotType(), id); builder != nil {
			log.Infof("start rebuilding command due to config file change: %s.", id)
			command, err := builder.Build(dir)
			if err != nil {
				log.Errorf("can't configure command. %s. %#v", err.Error(), builder)
			} else {
				bot.AppendCommand(command) // replaces the old one.
			}
		}
{'fraction_non_alphanumeric': 0.12215110833593507, 'fraction_numerical': 0.0002341554792382142, 'mean_word_length': 3.7720670391061453, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1711302', 'n_tokens_mistral': 4214, 'n_tokens_neox': 3849, 'n_words': 1203}