<commit_before>// Package creator contains the creator service that should create a new entry in the database for a github repo
// that is not already in there.
package creator

import (
	"fmt"
	"log"
	"sort"
	"strconv"
	"time"

	"github.com/evermax/stargraph/api"
	"github.com/evermax/stargraph/github"
	"github.com/evermax/stargraph/lib/store"
	"github.com/evermax/stargraph/service"
	"github.com/streadway/amqp"
)

// Creator contains the database to use, the type of service (creator)
// and the job queue to send job to workers. It implements the service.SWorker interface.
type Creator struct {
	t        string
	db       store.Store
	jobQueue chan service.Job
}

// NewCreator creates a new creator
func NewCreator(db store.Store) Creator {
	// connect to the AMQP server
	return Creator{
		t:  service.CreatorName,
		db: db,
	}
}

// Type will return the string "creator" to implement the interface.
func (c Creator) Type() string {
	return c.t
}

func (c Creator) JobQueue() chan service.Job {
	return c.jobQueue
}

// Run create a connection to the AMQP server and listen to incoming requests
// To create Github repository graphs.
func (c Creator) Run(amqpURL, addQueueN string) error {

	// Dial connection to the AMQP server
	conn, err := amqp.Dial(amqpURL)
	if err != nil {
		return fmt.Errorf("Failed to connect to AMQP with url %s: %v", amqpURL, err)
	}
	defer conn.Close()

	// Open a channel of communication through the connection
	ch, err := conn.Channel()
	if err != nil {
		return fmt.Errorf("Failed to open a channel: %v", err)
	}
	defer ch.Close()

	// Declare a queue
	q, err := ch.QueueDeclare(
		addQueueN, // name
		true,      // durable
		false,     // delete when unused
		false,     // exclusive
		false,     // no-wait
		nil,       // arguments
	)
	if err != nil {
		return fmt.Errorf("Failed to declare a queue: %v", err)
	}

	err = ch.Qos(
		1,     // prefetch count
		0,     // prefetch size
		false, // global
	)
	if err != nil {
		return fmt.Errorf("Failed to set QoS: %v", err)
	}

	// Register as a consumer of the queue and
	// return a channel that brings the incoming messages
	msgs, err := ch.Consume(
		q.Name, // queue
		"",     // consumer
		false,  // auto-ack
		false,  // exclusive
		false,  // no-local
		false,  // no-wait
		nil,    // args
	)

	if err != nil {
		return fmt.Errorf("Failed to register as consumer: %v", err)
	}

	forever := make(chan bool)
	go func() {
		for d := range msgs {
			log.Printf("Received a message: %s", d.Body)

			err := c.creatorWork(d.Body)
			if err == store.ErrAlreadyExist {
				d.Ack(false)
				log.Printf("WARN: Asked to recreate %s, aborting", d.Body)
				continue
			}
			if err != nil {
				// log it
				d.Nack(false, true)
				continue
			}

			d.Ack(false)
			log.Printf("Done")
		}
		forever <- true
	}()
	<-forever
	return nil
}

func (c Creator) creatorWork(body []byte) error {
	apiJob, err := api.Unmarshal(body)
	if err != nil {
		return fmt.Errorf("Umarshalling error with %s: %v", body, err)
	}

	repoInfo := github.RepoInfo{
		WorkedOn: true,
	}

	// Create the repository on the store, claim the work
	key, err := c.db.AddRepo(repoInfo)
	if err != nil {
		return fmt.Errorf("Adding to store error with %s: %v", body, err)
	}

	timestamps, err := GetAllTimestamps(c.jobQueue, 100, apiJob.Token, repoInfo)
	if err != nil {
		return fmt.Errorf("Error with %s: %v", body, err)<commit_msg>Make creator use mq.MessageQueue instead of amqp directly<commit_after>// Package creator contains the creator service that should create a new entry in the database for a github repo
// that is not already in there.
package creator

import (
	"fmt"
	"log"
	"sort"
	"strconv"
	"time"

	"github.com/evermax/stargraph/api"
	"github.com/evermax/stargraph/github"
	"github.com/evermax/stargraph/lib/mq"
	"github.com/evermax/stargraph/lib/store"
	"github.com/evermax/stargraph/service"
)

// Creator contains the database to use, the type of service (creator)
// and the job queue to send job to workers. It implements the service.SWorker interface.
type Creator struct {
	t         string
	db        store.Store
	messageQ  mq.MessageQueue
	jobQueue  chan service.Job
	queueName string
}

// NewCreator creates a new creator
func NewCreator(db store.Store, queue mq.MessageQueue) Creator {
	// connect to the AMQP server
	return Creator{
		t:        service.CreatorName,
		db:       db,
		messageQ: queue,
	}
}

// Type will return the string "creator" to implement the interface.
func (c Creator) Type() string {
	return c.t
}

func (c Creator) JobQueue() chan service.Job {
	return c.jobQueue
}

// Run create a connection to the AMQP server and listen to incoming requests
// To create Github repository graphs.
func (c Creator) Run() error {
	return c.messageQ.Consume(c.queueName, c.receiveMessage)
}

func (c Creator) receiveMessage(d mq.Delivery, forever chan bool) {
	log.Printf("Received a message: %s", d.Body)

	err := c.creatorWork(d.Body())
	if err == store.ErrAlreadyExist {
		d.Ack(false)
		log.Printf("WARN: Asked to recreate %s, aborting", d.Body)
		return
	}
	if err != nil {
		// log it
		d.Nack(false, true)
		return
	}

	d.Ack(false)
	log.Printf("Done")
}

func (c Creator) creatorWork(body []byte) error {
	apiJob, err := api.Unmarshal(body)
	if err != nil {
		return fmt.Errorf("Umarshalling error with %s: %v", body, err)
	}

	repoInfo := github.RepoInfo{
		WorkedOn: true,
	}

	// Create the repository on the store, claim the work
	key, err := c.db.AddRepo(repoInfo)
	if err != nil {
		return fmt.Errorf("Adding to store error with %s: %v", body, err)
	}

	timestamps, err := GetAllTimestamps(c.jobQueue, 100, apiJob.Token, repoInfo)
	if err != nil {
		return fmt.Errorf("Error with %s: %v", body, err)
{'fraction_non_alphanumeric': 0.12095221424820585, 'fraction_numerical': 0.0014003150708909505, 'mean_word_length': 3.1616897305171157, 'pattern_counts': {'":': 0, '<': 5, '<?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': '743733', 'n_tokens_mistral': 2093, 'n_tokens_neox': 1961, 'n_words': 657}