<commit_before>package jobs

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"sync"
	"time"

	"github.com/cozy/cozy-stack/pkg/consts"
	"github.com/cozy/cozy-stack/pkg/couchdb"
	"github.com/cozy/cozy-stack/pkg/logger"
	"github.com/cozy/cozy-stack/pkg/prefixer"
	"github.com/sirupsen/logrus"
)

// memScheduler is a centralized scheduler of many triggers. It starts all of
// them and schedules jobs accordingly.
type memScheduler struct {
	broker Broker

	ts  map[string]Trigger
	mu  sync.RWMutex
	log *logrus.Entry
}

// NewMemScheduler creates a new in-memory scheduler that will load all
// registered triggers and schedule their work.
func NewMemScheduler() Scheduler {
	return newMemScheduler()
}

func newMemScheduler() *memScheduler {
	return &memScheduler{
		ts:  make(map[string]Trigger),
		log: logger.WithNamespace("mem-scheduler"),
	}
}

type inst struct {
	Domain string `json:"domain"`
	Prefix string `json:"prefix"`
}

func (i inst) DBPrefix() string {
	if i.Prefix != "" {
		return i.Prefix
	}
	return i.Domain
}

func (i inst) DomainName() string {
	return i.Domain
}

// StartScheduler will start the scheduler by actually loading all triggers
// from the scheduler's storage and associate for each of them a go routine in
// which they wait for the trigger send job requests.
func (s *memScheduler) StartScheduler(b Broker) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	var ts []*TriggerInfos
	err := couchdb.ForeachDocs(couchdb.GlobalDB, consts.Instances, func(_ string, data json.RawMessage) error {
		var db inst
		if err := json.Unmarshal(data, &db); err != nil {
			return err
		}
		err := couchdb.ForeachDocs(db, consts.Triggers, func(_ string, data json.RawMessage) error {
			var t *TriggerInfos
			if err := json.Unmarshal(data, &t); err != nil {
				return err
			}
			ts = append(ts, t)
			return nil
		})
		if err != nil && !couchdb.IsNoDatabaseError(err) {
			return err
		}
		return nil
	})
	if err != nil && !couchdb.IsNoDatabaseError(err) {
		return err
	}

	s.broker = b

	for _, infos := range ts {
		t, err := fromTriggerInfos(infos)
		if err != nil {
			return err
		}
		s.ts[t.DBPrefix()+"/"+infos.TID] = t
		go s.schedule(t)
	}

	return nil
}

// ShutdownScheduler shuts down the scheduling of triggers
func (s *memScheduler) ShutdownScheduler(ctx context.Context) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	fmt.Print("  shutting down in-memory scheduler...")
	for _, t := range s.ts {
		t.Unschedule()
	}
	fmt.Println("ok.")
	return nil
}

// AddTrigger will add a new trigger to the scheduler. The trigger is persisted
// in storage.
func (s *memScheduler) AddTrigger(t Trigger) error {
	s.mu.Lock()<commit_msg>Do not crash stack at start with a malformed trigger<commit_after>package jobs

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"sync"
	"time"

	"github.com/cozy/cozy-stack/pkg/consts"
	"github.com/cozy/cozy-stack/pkg/couchdb"
	"github.com/cozy/cozy-stack/pkg/logger"
	"github.com/cozy/cozy-stack/pkg/prefixer"
	"github.com/sirupsen/logrus"
)

// memScheduler is a centralized scheduler of many triggers. It starts all of
// them and schedules jobs accordingly.
type memScheduler struct {
	broker Broker

	ts  map[string]Trigger
	mu  sync.RWMutex
	log *logrus.Entry
}

// NewMemScheduler creates a new in-memory scheduler that will load all
// registered triggers and schedule their work.
func NewMemScheduler() Scheduler {
	return newMemScheduler()
}

func newMemScheduler() *memScheduler {
	return &memScheduler{
		ts:  make(map[string]Trigger),
		log: logger.WithNamespace("mem-scheduler"),
	}
}

type inst struct {
	Domain string `json:"domain"`
	Prefix string `json:"prefix"`
}

func (i inst) DBPrefix() string {
	if i.Prefix != "" {
		return i.Prefix
	}
	return i.Domain
}

func (i inst) DomainName() string {
	return i.Domain
}

// StartScheduler will start the scheduler by actually loading all triggers
// from the scheduler's storage and associate for each of them a go routine in
// which they wait for the trigger send job requests.
func (s *memScheduler) StartScheduler(b Broker) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	var ts []*TriggerInfos
	err := couchdb.ForeachDocs(couchdb.GlobalDB, consts.Instances, func(_ string, data json.RawMessage) error {
		var db inst
		if err := json.Unmarshal(data, &db); err != nil {
			return err
		}
		err := couchdb.ForeachDocs(db, consts.Triggers, func(_ string, data json.RawMessage) error {
			var t *TriggerInfos
			if err := json.Unmarshal(data, &t); err != nil {
				return err
			}
			ts = append(ts, t)
			return nil
		})
		if err != nil && !couchdb.IsNoDatabaseError(err) {
			return err
		}
		return nil
	})
	if err != nil && !couchdb.IsNoDatabaseError(err) {
		return err
	}

	s.broker = b

	for _, infos := range ts {
		t, err := fromTriggerInfos(infos)
		if err != nil {
			joblog.Errorf("Could not start trigger with ID %s: %s",
				infos.ID(), err.Error())
			continue
		}
		s.ts[t.DBPrefix()+"/"+infos.TID] = t
		go s.schedule(t)
	}

	return nil
}

// ShutdownScheduler shuts down the scheduling of triggers
func (s *memScheduler) ShutdownScheduler(ctx context.Context) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	fmt.Print("  shutting down in-memory scheduler...")
	for _, t := range s.ts {
		t.Unschedule()
	}
	fmt.Println("ok.")
	return nil
}

// AddTrigger will add a new trigger to the scheduler. The trigger is persisted
// in storage.
func (s *memScheduler) AddTrigger(t Trigger) error {
	s.mu.Lock()
{'fraction_non_alphanumeric': 0.12447373238147538, 'fraction_numerical': 0.0, 'mean_word_length': 3.6660973526900085, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2234397', 'n_tokens_mistral': 2041, 'n_tokens_neox': 1937, 'n_words': 563}