package MooseX::JobQueue::Manager::Source::DBIC;

use Moose::Role;
use MooseX::POE;
use MooseX::AttributeHelpers;

with 'MooseX::JobQueue::Manager';

use Sys::Hostname;

=head1 NAME

MooseX::JobQueue::Manager::Source::DBIC - DBIx::Class Job Source for 
MooseX::JobQueue::Manager

=head1 DESCRIPTION

Find jobs from resultsources in a L<DBIx::Class::Schema>. 

=head1 RESULT SOURCE STRUCTURE

 __PACKAGE__->load_components('TimeStamp', 'InflateColumn::DateTime', 'Core');
 
 __PACKAGE__->table('job_queue');
 __PACKAGE__->add_columns(
 	job_id => {
 		data_type         => 'int',
 		is_auto_increment => 1,
 		is_nullable       => 0,
 		extra => { unsigned => 1 },
 	},
 
 	status => {
 		data_type => 'enum',
 		is_nullable => 0,
 		default => 'available',
 		extra => { list => [qw/available in_progress done failed/] },
 	},
 
 	job_type => { 
 		data_type   => 'varchar',
 		is_nullable => 0,
 	},
 	args => { # Storable args to pass to the job
 		data_type   => 'blob',
 		is_nullable => 1,
 	},
 	not_before => { # Only run the job after the given time. 
 	                # Default value set in insert method
 		data_type => 'datetime'
 	},
 	coalesce => { # field use to gorup related jobs. For example Email would
 	              # put the domain in this column so that emails to the same
 	              # domain would get handled by the same worker (up to a limit)
 		data_type   => 'varchar',
 		is_nullable => 1,
 		size        => 255,
 	},
 	locked_by => { # Used to allow the job manager process to run on multiple 
 	               # machines talking to the same DB. Stores "${host}-{$pid} here
 		data_type   => 'varchar',
 		is_nullable => 1,
 		size        => 255,
 	},
 	cr_date => {
 		data_type => 'datetime',
 		set_on_create => 1
 	},
 );
 __PACKAGE__->set_primary_key('job_id');
 __PACKAGE__->inflate_column('args', {
 	inflate => sub { Storable::thaw(shift) },
 	deflate => sub { Storable::nfreeze(shift) },
 });
 

=head1 ATTRIBUTES

=cut

has 'schema' => (
	isa      => 'DBIx::Class::Schema',
	is       => 'ro',
	required => 1,
	documentation => <<'POD'

=head2 schema

The L<DBIx::Class::Schema> instance from which to find jobs.

=cut

POD

);

has 'resultsources' => (
	metaclass => 'Collection::Array',
	is => 'rw',
	isa => 'ArrayRef',
	required => 1,
	provides => {
		'push'  => 'add_resultsource',
		'count' => 'num_resultsources',
		'get'   => 'resultsource',
	},
  documentation => <<'POD'

=head2 resultsources

ArrayRef of result source names from which to find jobs.

=cut

POD
);

# Which resultsource to query next for jobs. Ensures fairness with multiple
# sources
has 'next_resultsource_idx' => (
	is => 'rw',
	isa => 'Int',
	default => 0,
);

# Should *not* be called form in a transaction. All operaitons are
# safe since we use a unique lockign string, and _lock_jobs only ever finds
# things with the explicit locking str.
#
# (Oh, plus we get deadlock when this is a transaction in some cases.)
sub _find_jobs {
	my ($self, $source) = @_;

	my $rs = $self->schema->resultset($source);
	my $job_ids = $self->_lock_jobs( $rs->search_rs( 
		{ status => 'available',
			not_before => { '<=' => $self->schema->storage->datetime_parser->format_datetime(DateTime->now) },
		},
		{ group_by => ['me.job_type', 'me.coalesce' ],
			order_by => 'me.job_id',
		}),
		$self->max_workers - $self->num_workers
	);

	return 0 unless @$job_ids;

	my $jobs = $rs->search({ job_id => { -in => $job_ids } });

	my $new_job_count = 0;
	while (my $job = $jobs->next) {
		my $handler = $self->job_handlers->{$job->job_type};

		unless (ref $handler eq 'CODE') {
			$self->logger->error( "No handler found for job type @{[$job->job_type]} on job id @{[$job->id]}. Status set to failed" );
			# TODO: Probably set the status to failed/no_handler
			$job->status('failed');
			$job->update;
			next;
		}

		my $coalesced_job_ids = $self->coalesce_jobs($job);

		# It shouldn't really be possible for coalesce_jobs. But better safe than sorry
		next unless @$coalesced_job_ids;

		$new_job_count++;
		my $job_rs = $rs->search({job_id => { -in => $coalesced_job_ids } });

		# Set status to in_progress, and unlock it, so that job handler doesn't 
		# have to worry/know about the locking.
		$job_rs->update({ status => 'in_progress', locked_by => undef });

		my $worker_state = {
			pid => undef, # Not know yet
			jobs_rs => $job_rs,
			job_ids => $coalesced_job_ids
		};

    # This needs to be handled in a generic fashion
		my ($wheel_id, $wheel_pid) = $self->spawn( \&run_code => [ $handler, $self->logger, $worker_state ] );

		unless ($wheel_id) {
			# Shouldn't happen anymore. We tried to create too many workers
			$self->logger->debug('spawn returned undef!');
			$job_rs->update({status => 'available'});
			last;
		}

		$worker_state->{pid} = $wheel_pid;
		$self->logger->notice("Wheel ID $wheel_id (pid: $wheel_pid) spawned to handle job ids: @{ $worker_state->{job_ids} }");

		$self->worker_states->{$wheel_id} = $worker_state;
	}
	return $new_job_count;
}

=head2 coalesce_jobs($job)

Return a ResultSet of jobs that should be handled by the same worker as 
C<$job>. 

TODO: Delegate this to somewhere in the handler classes.

=cut

sub coalesce_jobs {
	my ($self, $job) = @_;

	my $source = $job->result_source;
	my $rs = $source->resultset->search_rs({
			job_type => $job->job_type,
			coalesce => $job->coalesce,
			status	 => 'available',
			not_before => { '<=' => $self->schema->storage->datetime_parser->format_datetime(DateTime->now) },
		},
	);

	$self->logger->debug("Coalesing jobs for " . $job->id);
	return $self->_lock_jobs(
		$rs, 
		$self->coalesce_limit-1, # We already have one job, the one we are using as the condition
		$job->locked_by, # Use the same lock by string
	);
	
}

# Helper funciton that locks jobs ensures only this process get hold of them, 
# including other job_managers on different machines. Returns \@job_ids
my $lock_counter = 1;
sub _lock_jobs {
	my ($self, $rs, $number, $lock_by_str) = @_;

	# A string we use that must be uniquie across all hosts and processes 
	# running the job manager
	unless ($lock_by_str) {
		$lock_by_str = hostname . "-$$-$lock_counter";
		$lock_counter++;
	}

	# Maximum number of jobs do we want to find?
	$number ||= 1;

	# List of job ids we want to try and lock
	my @candidate_job_ids = $rs->search({ locked_by => undef }, { rows => $number } )
	                           ->get_column('job_id')->all;
	

	# Lock those ones that aren't locked already
	# There might not be any other jobs - could happen when coalescing
	$rs->search({locked_by => undef, job_id => { -in => \@candidate_job_ids } })
					->update({ locked_by => $lock_by_str })
	  if @candidate_job_ids;

	# Then find out which ones we actaully managed to lock
	return [ $rs->search({locked_by => $lock_by_str})->get_column('job_id')->all ];
}

sub next_resultsource {
	my ($self) = @_;

	my $n = $self->next_resultsource_idx;

	$n = 0 if $n >= $self->num_resultsources || $n < 0;

	my $rs = $self->resultsource($n);
	
	$self->next_resultsource_idx(++$n);

	return $rs;
}

event find_jobs => sub {
	my ($self, $kernel) = @_[ OBJECT, KERNEL ];

	return if $self->num_workers == $self->max_workers;
	eval {
		$self->logger->debug("Finding jobs");
		my $num_found = 0;
		# Check each RS once each time if we can
		for ($self->num_resultsources) {
			my $source = $self->next_resultsource;
			my $count = eval { $self->_find_jobs($source) } || 0;
			if ($@) {
				$self->logger->error("No table found for source $source. Removing it from list of resultsources. (error was: $@)");
				$self->resultsources( [ grep { $_ ne $source } @{$self->resultsources} ] );
				undef $@;
				$kernel->delay_set( check_resultsource => 5*60, $source, 5*60*2);
				$self->next_resultsource_idx($self->next_resultsource_idx-1);
				next;
			}
			$num_found += $count;

			# Dont try and find anymore.
			last if $self->num_workers == $self->max_workers;
		}
		$self->logger->debug("Found $num_found job(s)");
	};

	if ($@) {
		$self->logger->critical("Error whilst finding jobs: $@");
		die $@;
	}

	# If we have no workers left, and we've not bee told to terminate, delay an
	# event to find more jobs.
	return if $self->terminating || $self->num_workers;

	$self->sleep_alarm_id($kernel->delay_set('find_jobs', $self->sleep_interval));
};

=head2 check_resultsource

Check if a resultsource is alive again (can we select anything from it.) If it
appears to be alive, push it to the list of soruce to check, otherwise 
schedule another check with an exponential delay (i.e. check in 5mins, then 10
mins, then 20 mins, then 40 etc.) up to a maximum wait of 640mins, giving a 
total check time of 21:15 hours.

=cut

event check_resultsource => sub {
	my ($self, $kernel, $source_name, $next_delay) = @_[OBJECT,KERNEL,ARG0,ARG1];

	eval {
		$self->schema->resultset($source_name)->search( [ \"1 = 0" ])->all
	};

	if ($@) {

		if ($next_delay <= 640*60) {
			$self->logger->debug("Recheck of result source $source_name still failed: $@");
			my $delay = $next_delay;
			$next_delay *= 2;
			$kernel->delay_set(check_resultsource => $delay, $source_name, $next_delay);
		} else {
			$self->logger->warning("Re-checking result source $source_name has reached maximum delay without being found alive.");
		}
	} else {
		# It worked! Add it back to list of source to check
		$self->logger->info("Re-check of $source_name succeeded.");
		$self->add_resultsource($source_name);
	}
};

=head2 run_code

POE::Wheel::Run doesn't eval codeblocks. If they die they cause all kinds of 
warnings from POE. This is bad. We wrap the actualy job handler in an eval here
and any jobs that are still marked as 'in_progress' get set back to available.

=cut

augment run_code => sub {
	my ($codeblock, $logger, $state, @args) = @_;

	my ($job_rs, $job_ids) = @$state{qw/jobs_rs job_ids/};

	eval {
		$job_rs->result_source->storage->ensure_connected;
	};

	if ($@) {
		$logger->critical("Error getting valid database conection: $@\n The following job ids will need unlocking and to have their status manually set back to available: " .	join(', ', @$job_ids) );
		return;
	}

  inner();

	if ($@) {

		# Tidy up jobs;
		$job_rs->reset;
		while (my $job = $job_rs->next) {
			next unless $job->status eq 'in_progress';

			# TODO: Do we want to back it off exponentailly? Status of failed maybe?
			$logger->alert( "Job @{[$job->id]} set back to available due to failed handler" );
			$job->status('available');
			$job->not_before(DateTime->now->add(minutes => 5));
			$job->update;
		}
	}

}

1;


__END__

=head1 AUTHOR

Ash Berlin C<< <ash@cpan.org> >>

=cut

