<commit_before>package workers

import (
	"fmt"
	"github.com/APTrust/exchange/constants"
	"github.com/APTrust/exchange/context"
	"github.com/APTrust/exchange/models"
	"github.com/APTrust/exchange/network"
	"github.com/nsqio/go-nsq"
	"sync"
	"time"
)

const (
	GENERIC_FILE_BATCH_SIZE = 50
)

// Records ingest data (objects, files and events) in Pharos
type APTRecorder struct {
	Context             *context.Context
	RecordChannel       chan *models.IngestState
	CleanupChannel      chan *models.IngestState
	WaitGroup           sync.WaitGroup
}

func NewAPTRecorder(_context *context.Context) (*APTRecorder) {
	recorder := &APTRecorder{
		Context: _context,
	}
	// Set up buffered channels
	workerBufferSize := _context.Config.RecordWorker.Workers * 10
	recorder.RecordChannel = make(chan *models.IngestState, workerBufferSize)
	recorder.CleanupChannel = make(chan *models.IngestState, workerBufferSize)
	// Set up a limited number of go routines
	for i := 0; i < _context.Config.RecordWorker.Workers; i++ {
		go recorder.record()
		go recorder.cleanup()
	}
	return recorder
}

// This is the callback that NSQ workers use to handle messages from NSQ.
func (recorder *APTRecorder) HandleMessage(message *nsq.Message) (error) {
	ingestState, err := GetIngestState(message, recorder.Context, false)
	if err != nil {
		recorder.Context.MessageLog.Error(err.Error())
		return err
	}

	// If this item was queued more than once, and this process or any
	// other is currently working on it, just finish the message and
	// assume that the in-progress worker will take care of the original.
	if ingestState.WorkItem.Node != "" && ingestState.WorkItem.Pid != 0 {
		recorder.Context.MessageLog.Info("Marking WorkItem %d (%s/%s) as finished " +
			"without doing any work, because this item is currently in process by " +
			"node %s, pid %s. WorkItem was last updated at %s.",
			ingestState.WorkItem.Id, ingestState.WorkItem.Bucket,
			ingestState.WorkItem.Name, ingestState.WorkItem.Node,
			ingestState.WorkItem.Pid, ingestState.WorkItem.UpdatedAt)
		message.Finish()
		return nil
	}

	// Disable auto response, so we can tell NSQ when we need to
	// that we're still working on this item.
	message.DisableAutoResponse()

	// Clear out any old errors, because we're going to retry
	// whatever may have failed on the last run.
	ingestState.IngestManifest.RecordResult.ClearErrors()

	// Tell Pharos that we've started to record this item.
	err = MarkWorkItemStarted(ingestState, recorder.Context,
		constants.StageRecord, "Recording object, file and event metadata in Pharos.")
	if err != nil {
		recorder.Context.MessageLog.Error(err.Error())
		return err
	}

	recorder.Context.MessageLog.Info("Putting %s/%s into record channel",
		ingestState.IngestManifest.S3Bucket, ingestState.IngestManifest.S3Key)

	recorder.RecordChannel <- ingestState

	// Return no error, so NSQ knows we're OK.
	return nil
}

// Step 1: Record data in Pharos
func (recorder *APTRecorder) record () {
	for ingestState := range recorder.RecordChannel {
		ingestState.IngestManifest.RecordResult.Start()
		ingestState.IngestManifest.RecordResult.Attempted = true
		ingestState.IngestManifest.RecordResult.AttemptNumber += 1
		recorder.buildEventsAndChecksums(ingestState)
		if !ingestState.IngestManifest.RecordResult.HasErrors() {
			recorder.saveAllPharosData(ingestState)
		}
		recorder.CleanupChannel <- ingestState
	}
}

// Step 2: Delete tar file from staging area and from receiving bucket.
func (recorder *APTRecorder) cleanup () {
	for ingestState := range recorder.CleanupChannel {
		// See if we have fatal errors, or too many recurring transient errors
		attemptNumber := ingestState.IngestManifest.RecordResult.AttemptNumber
		maxAttempts := int(recorder.Context.Config.StoreWorker.MaxAttempts)
		itsTimeToGiveUp := (ingestState.IngestManifest.HasFatalErrors() ||
			(ingestState.IngestManifest.HasErrors() && attemptNumber >= maxAttempts))

		if itsTimeToGiveUp {
			recorder.Context.MessageLog.Error("Failed to record %s/%s. Errors: %s.",
				ingestState.WorkItem.Bucket, ingestState.WorkItem.Name,
				ingestState.IngestManifest.AllErrorsAsString())
			ingestState.FinishNSQ()
			MarkWorkItemFailed(ingestState, recorder.Context)
		} else if ingestState.IngestManifest.RecordResult.HasErrors() {
			recorder.Context.MessageLog.Info("Requeueing WorkItem %d (%s/%s) due to transient errors. %s",
				ingestState.WorkItem.Id, ingestState.WorkItem.Bucket,
				ingestState.WorkItem.Name,
				ingestState.IngestManifest.AllErrorsAsString())
			ingestState.RequeueNSQ(1000)
			MarkWorkItemRequeued(ingestState, recorder.Context)
		} else {
			MarkWorkItemStarted(ingestState, recorder.Context, constants.StageCleanup,
				"Bag has been stored and recorded. Deleting files from receiving bucket " +
					"and staging area.")
			DeleteBagFromStaging(ingestState, recorder.Context, ingestState.IngestManifest.RecordResult)
			recorder.deleteBagFromReceivingBucket(ingestState)
			MarkWorkItemSucceeded(ingestState, recorder.Context, constants.StageCleanup)
			ingestState.FinishNSQ()
		}

		// Save our WorkItemState
		ingestState.IngestManifest.RecordResult.Finish()
		RecordWorkItemState(ingestState, recorder.Context, ingestState.IngestManifest.RecordResult)
	}
}

// Make sure the IntellectualObject and its component files have
// all of the checksums and PREMIS events we'll need to save.
// We build these now so that the PREMIS events will have UUIDs,
// and if we ever have to re-record this IntellectualObject after
// a partial save, we'll know which events are already recorded
// in Pharos and which were not. This was a problem in the old
// system, where record failured were common, and PREMIS events
// often wound up being recorded twice.
func (recorder *APTRecorder) buildEventsAndChecksums (ingestState *models.IngestState) {
	obj := ingestState.IngestManifest.Object
	err := obj.BuildIngestEvents()
	if err != nil {
		ingestState.IngestManifest.RecordResult.AddError(err.Error())
	}
	err = obj.BuildIngestChecksums()
	if err != nil {
		ingestState.IngestManifest.RecordResult.AddError(err.Error())
	}
}

func (recorder *APTRecorder) saveAllPharosData (ingestState *models.IngestState) {
	recorder.saveIntellectualObject(ingestState)
	if ingestState.IngestManifest.RecordResult.HasErrors() {
		recorder.Context.MessageLog.Error("Error saving IntellectualObject %s/%s: %v",
			ingestState.WorkItem.Bucket, ingestState.WorkItem.Name,
			ingestState.IngestManifest.RecordResult.AllErrorsAsString())
		return
	}
	recorder.saveGenericFiles(ingestState)
	if ingestState.IngestManifest.RecordResult.HasErrors() {
		recorder.Context.MessageLog.Error("Error saving one or more GenericFiles for " +
			"IntellectualObject %s/%s: %v",
			ingestState.WorkItem.Bucket, ingestState.WorkItem.Name,
			ingestState.IngestManifest.RecordResult.AllErrorsAsString())
		return
	}
}

func (recorder *APTRecorder) saveIntellectualObject (ingestState *models.IngestState) {
	obj := ingestState.IngestManifest.Object
	resp := recorder.Context.PharosClient.IntellectualObjectSave(obj)
	if resp.Error != nil {
		ingestState.IngestManifest.RecordResult.AddError(resp.Error.Error())
		return
	}
	savedObject := resp.IntellectualObject()
	if savedObject == nil {
		ingestState.IngestManifest.RecordResult.AddError(
			"Pharos returned nil IntellectualObject after save.")
		return
	}
	obj.Id = savedObject.Id
	obj.CreatedAt = savedObject.CreatedAt
	obj.UpdatedAt = savedObject.UpdatedAt
	obj.PropagateIdsToChildren()
}

func (recorder *APTRecorder) saveGenericFiles (ingestState *models.IngestState) {
	filesToCreate := make([]*models.GenericFile, 0)
	filesToUpdate := make([]*models.GenericFile, 0)
	for i, gf := range ingestState.IngestManifest.Object.GenericFiles {
		if i % GENERIC_FILE_BATCH_SIZE == 0 {
			recorder.createGenericFiles(ingestState, filesToCreate)
			if ingestState.IngestManifest.RecordResult.HasErrors() {
				break
			}
			recorder.updateGenericFiles(ingestState, filesToUpdate)
			if ingestState.IngestManifest.RecordResult.HasErrors() {
				break
			}
			filesToCreate = make([]*models.GenericFile, 0)
			filesToUpdate = make([]*models.GenericFile, 0)
		}
		if gf.IngestNeedsSave {
			if gf.IngestPreviousVersionExists {
				filesToUpdate = append(filesToUpdate, gf)
			} else {
				filesToCreate = append(filesToCreate, gf)<commit_msg>Don't re-save objects that were saved on first pass<commit_after>package workers

import (
	"fmt"
	"github.com/APTrust/exchange/constants"
	"github.com/APTrust/exchange/context"
	"github.com/APTrust/exchange/models"
	"github.com/APTrust/exchange/network"
	"github.com/nsqio/go-nsq"
	"sync"
	"time"
)

const (
	GENERIC_FILE_BATCH_SIZE = 50
)

// Records ingest data (objects, files and events) in Pharos
type APTRecorder struct {
	Context             *context.Context
	RecordChannel       chan *models.IngestState
	CleanupChannel      chan *models.IngestState
	WaitGroup           sync.WaitGroup
}

func NewAPTRecorder(_context *context.Context) (*APTRecorder) {
	recorder := &APTRecorder{
		Context: _context,
	}
	// Set up buffered channels
	workerBufferSize := _context.Config.RecordWorker.Workers * 10
	recorder.RecordChannel = make(chan *models.IngestState, workerBufferSize)
	recorder.CleanupChannel = make(chan *models.IngestState, workerBufferSize)
	// Set up a limited number of go routines
	for i := 0; i < _context.Config.RecordWorker.Workers; i++ {
		go recorder.record()
		go recorder.cleanup()
	}
	return recorder
}

// This is the callback that NSQ workers use to handle messages from NSQ.
func (recorder *APTRecorder) HandleMessage(message *nsq.Message) (error) {
	ingestState, err := GetIngestState(message, recorder.Context, false)
	if err != nil {
		recorder.Context.MessageLog.Error(err.Error())
		return err
	}

	// If this item was queued more than once, and this process or any
	// other is currently working on it, just finish the message and
	// assume that the in-progress worker will take care of the original.
	if ingestState.WorkItem.Node != "" && ingestState.WorkItem.Pid != 0 {
		recorder.Context.MessageLog.Info("Marking WorkItem %d (%s/%s) as finished " +
			"without doing any work, because this item is currently in process by " +
			"node %s, pid %s. WorkItem was last updated at %s.",
			ingestState.WorkItem.Id, ingestState.WorkItem.Bucket,
			ingestState.WorkItem.Name, ingestState.WorkItem.Node,
			ingestState.WorkItem.Pid, ingestState.WorkItem.UpdatedAt)
		message.Finish()
		return nil
	}

	// Disable auto response, so we can tell NSQ when we need to
	// that we're still working on this item.
	message.DisableAutoResponse()

	// Clear out any old errors, because we're going to retry
	// whatever may have failed on the last run.
	ingestState.IngestManifest.RecordResult.ClearErrors()

	// Tell Pharos that we've started to record this item.
	err = MarkWorkItemStarted(ingestState, recorder.Context,
		constants.StageRecord, "Recording object, file and event metadata in Pharos.")
	if err != nil {
		recorder.Context.MessageLog.Error(err.Error())
		return err
	}

	recorder.Context.MessageLog.Info("Putting %s/%s into record channel",
		ingestState.IngestManifest.S3Bucket, ingestState.IngestManifest.S3Key)

	recorder.RecordChannel <- ingestState

	// Return no error, so NSQ knows we're OK.
	return nil
}

// Step 1: Record data in Pharos
func (recorder *APTRecorder) record () {
	for ingestState := range recorder.RecordChannel {
		ingestState.IngestManifest.RecordResult.Start()
		ingestState.IngestManifest.RecordResult.Attempted = true
		ingestState.IngestManifest.RecordResult.AttemptNumber += 1
		recorder.buildEventsAndChecksums(ingestState)
		if !ingestState.IngestManifest.RecordResult.HasErrors() {
			recorder.saveAllPharosData(ingestState)
		}
		recorder.CleanupChannel <- ingestState
	}
}

// Step 2: Delete tar file from staging area and from receiving bucket.
func (recorder *APTRecorder) cleanup () {
	for ingestState := range recorder.CleanupChannel {
		// See if we have fatal errors, or too many recurring transient errors
		attemptNumber := ingestState.IngestManifest.RecordResult.AttemptNumber
		maxAttempts := int(recorder.Context.Config.StoreWorker.MaxAttempts)
		itsTimeToGiveUp := (ingestState.IngestManifest.HasFatalErrors() ||
			(ingestState.IngestManifest.HasErrors() && attemptNumber >= maxAttempts))

		if itsTimeToGiveUp {
			recorder.Context.MessageLog.Error("Failed to record %s/%s. Errors: %s.",
				ingestState.WorkItem.Bucket, ingestState.WorkItem.Name,
				ingestState.IngestManifest.AllErrorsAsString())
			ingestState.FinishNSQ()
			MarkWorkItemFailed(ingestState, recorder.Context)
		} else if ingestState.IngestManifest.RecordResult.HasErrors() {
			recorder.Context.MessageLog.Info("Requeueing WorkItem %d (%s/%s) due to transient errors. %s",
				ingestState.WorkItem.Id, ingestState.WorkItem.Bucket,
				ingestState.WorkItem.Name,
				ingestState.IngestManifest.AllErrorsAsString())
			ingestState.RequeueNSQ(1000)
			MarkWorkItemRequeued(ingestState, recorder.Context)
		} else {
			MarkWorkItemStarted(ingestState, recorder.Context, constants.StageCleanup,
				"Bag has been stored and recorded. Deleting files from receiving bucket " +
					"and staging area.")
			DeleteBagFromStaging(ingestState, recorder.Context, ingestState.IngestManifest.RecordResult)
			recorder.deleteBagFromReceivingBucket(ingestState)
			MarkWorkItemSucceeded(ingestState, recorder.Context, constants.StageCleanup)
			ingestState.FinishNSQ()
		}

		// Save our WorkItemState
		ingestState.IngestManifest.RecordResult.Finish()
		RecordWorkItemState(ingestState, recorder.Context, ingestState.IngestManifest.RecordResult)
	}
}

// Make sure the IntellectualObject and its component files have
// all of the checksums and PREMIS events we'll need to save.
// We build these now so that the PREMIS events will have UUIDs,
// and if we ever have to re-record this IntellectualObject after
// a partial save, we'll know which events are already recorded
// in Pharos and which were not. This was a problem in the old
// system, where record failured were common, and PREMIS events
// often wound up being recorded twice.
func (recorder *APTRecorder) buildEventsAndChecksums (ingestState *models.IngestState) {
	obj := ingestState.IngestManifest.Object
	err := obj.BuildIngestEvents()
	if err != nil {
		ingestState.IngestManifest.RecordResult.AddError(err.Error())
	}
	err = obj.BuildIngestChecksums()
	if err != nil {
		ingestState.IngestManifest.RecordResult.AddError(err.Error())
	}
}

func (recorder *APTRecorder) saveAllPharosData (ingestState *models.IngestState) {
	if (ingestState.IngestManifest.Object.Id == 0) {
		recorder.saveIntellectualObject(ingestState)
		if ingestState.IngestManifest.RecordResult.HasErrors() {
			recorder.Context.MessageLog.Error("Error saving IntellectualObject %s/%s: %v",
				ingestState.WorkItem.Bucket, ingestState.WorkItem.Name,
				ingestState.IngestManifest.RecordResult.AllErrorsAsString())
			return
		} else {
			recorder.Context.MessageLog.Info("Saved %s/%s with id %d",
				ingestState.WorkItem.Bucket, ingestState.WorkItem.Name,
				ingestState.IngestManifest.Object.Id)
		}
	} else {
			recorder.Context.MessageLog.Info(
				"No need to save %s/%s already has id %d",
				ingestState.WorkItem.Bucket, ingestState.WorkItem.Name,
				ingestState.IngestManifest.Object.Id)
	}
	recorder.saveGenericFiles(ingestState)
	if ingestState.IngestManifest.RecordResult.HasErrors() {
		recorder.Context.MessageLog.Error("Error saving one or more GenericFiles for " +
			"IntellectualObject %s/%s: %v",
			ingestState.WorkItem.Bucket, ingestState.WorkItem.Name,
			ingestState.IngestManifest.RecordResult.AllErrorsAsString())
		return
	}
}

func (recorder *APTRecorder) saveIntellectualObject (ingestState *models.IngestState) {
	obj := ingestState.IngestManifest.Object
	resp := recorder.Context.PharosClient.IntellectualObjectSave(obj)
	if resp.Error != nil {
		ingestState.IngestManifest.RecordResult.AddError(resp.Error.Error())
		return
	}
	savedObject := resp.IntellectualObject()
	if savedObject == nil {
		ingestState.IngestManifest.RecordResult.AddError(
			"Pharos returned nil IntellectualObject after save.")
		return
	}
	obj.Id = savedObject.Id
	obj.CreatedAt = savedObject.CreatedAt
	obj.UpdatedAt = savedObject.UpdatedAt
	obj.PropagateIdsToChildren()
}

func (recorder *APTRecorder) saveGenericFiles (ingestState *models.IngestState) {
	filesToCreate := make([]*models.GenericFile, 0)
	filesToUpdate := make([]*models.GenericFile, 0)
	for i, gf := range ingestState.IngestManifest.Object.GenericFiles {
		if i % GENERIC_FILE_BATCH_SIZE == 0 {
			recorder.createGenericFiles(ingestState, filesToCreate)
			if ingestState.IngestManifest.RecordResult.HasErrors() {
				break
			}
			recorder.updateGenericFiles(ingestState, filesToUpdate)
			if ingestState.IngestManifest.RecordResult.HasErrors() {
				break
			}
			filesToCreate = make([]*models.GenericFile, 0)
			filesToUpdate = make([]*models.GenericFile, 0)
		}
		if gf.IngestNeedsSave {
			if gf.IngestPreviousVersionExists {
				// Check for items with previous version and no Id?
				// That would be a problem.
				filesToUpdate = append(filesToUpdate, gf)
			} else if gf.IngestNeedsSave && gf.Id == 0 {
				filesToCreate = append(filesToCreate, gf)
{'fraction_non_alphanumeric': 0.09951892424505883, 'fraction_numerical': 0.002434359241870979, 'mean_word_length': 4.692510722533817, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 5, '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': '203845', 'n_tokens_mistral': 5936, 'n_tokens_neox': 5479, 'n_words': 1291}