package API::WarehouseOS::Daemon;
use strict;
use warnings;
use threads;
use Thread::Queue;
use MIME::Base64;
use Data::Dumper;
use LWP::UserAgent;
use LWP::Protocol::PSGI;
use File::Copy;
use File::Temp;
use Data::Serializer;
use Data::Serializer::Raw;
use JSON::XS;
use Try::Tiny;
use Safe::Isa;
use Moose;
use ECS::Config;
use ECS::WarehouseOS::Client;
use ECS::WarehouseOS::Error;
use API::WarehouseOS;
use API::Dispatch::PSGI;
use Time::HiRes qw/sleep/;
no indirect;

use lib qw(libs);
use API::Config;
BEGIN { use lib $API::Config::Config->{'Website_libs'} }
# use API::WarehouseOS::Daemon;

our $RETRY_COUNT  = 10;
# maximum number of seconds to delay between retries
our $API_RETRY_MAX_SECONDS = 300;
# number of seconds by which to increment retry delay
our $API_RETRY_INCREMENT_SECONDS = 2;


our $EOF = [EOF => "false"];
# Message_Type == only the types we care about
use Type::Tiny::Enum Message_Type => [qw/
	OrderComplete
	PickComplete
	VerifyComplete
/];


has lwp => (
	is		=> 'ro',
	isa		=> 'LWP::UserAgent',
	lazy	=> 1,
	builder	=> '_build_lwp',
);

has json => (
	is		=> 'ro',
	isa		=> 'JSON::XS',
	lazy	=> 1,
	builder	=> '_build_json',
);

has apiClient => (
	is		=> 'ro',
	isa		=> 'ECS::WarehouseOS::Client',
	lazy	=> 1,
	builder	=> '_build_api_client',
);

has do_quit => (
	is		=> 'rw',
	isa		=> 'Bool',
	default	=> 0
);

has send_interval => (
	is		=> 'rw',
	isa		=> 'Int',
	default	=> 10,
);

has queue_file => (
	is		=> 'ro',
	isa		=> 'Str',
	lazy	=> 1,
	builder	=> '_build_queue_file',
);

has pid_file => qw/
	is ro
	isa Str
	lazy 1
	builder _build_pid_file
/;
	 

has queue => (
	is		=> 'ro',
	isa		=> 'Thread::Queue',
	lazy	=> 1,
	builder	=> 'load',
);

has serializer => (
	is		=> 'ro',
	isa		=> 'Data::Serializer::Raw',
	lazy	=> 1,
	builder	=> '_build_serializer',
);

has handler => (
	is		=> 'ro',
	isa		=> 'API::Dispatch::PSGI',
	lazy	=> 1,
	builder	=> '_build_handler',
);

sub _build_json {
	return JSON::XS->new->allow_blessed->convert_blessed->allow_nonref;
}

sub _build_lwp {
	return LWP::UserAgent->new;
}

sub _build_api_client {
	ECS::WarehouseOS::Client->new;
}

sub _build_serializer {
	return Data::Serializer::Raw->new
}

sub _build_queue_file {
	my $queue_file = join '-' => 
		'/tmp/.warehouseos', 
		$ENV{HOSTNAME},
		($ECS::Config::c{sandbox} ?
			$ENV{firstInit} :
			()
		);
}

sub _build_pid_file {
	my $pid_file = (join '-' => 
		'/tmp/.warehouseos', 
		$ENV{HOSTNAME},
		($ECS::Config::c{sandbox} ?
			$ENV{firstInit} :
			()
		)).'.pid';
}


sub _build_handler {
	my $handler = API::Dispatch::PSGI->new( dispatchTo => 'WarehouseOS' );	
	return $handler;
}

# sender
sub send {
	my $self = shift;
	my $queue = $self->queue;
	die "Need Thread::Queue object for sender\n" unless $queue;
	while (1) {
		last if $self->do_quit;
		my @messages = grep is_Message_Type( $_->{messageType} ), $self->get_messages();
		last if $self->do_quit;
		if (@messages) {
			@$_{ qw/type decodedMessage/ } = (
				$_->{messageType},
				MIME::Base64::decode_base64( $_->{message} ),
			) for @messages;
			last if $self->do_quit;
			$queue->enqueue(map( [@$_{qw/type decodedMessage/}], @messages )); 
			$self->save();
			$self->confirm_messages( map( $_->{messageId}, @messages ) );
		}
	} continue {
		threads->yield;
		sleep $self->send_interval;
		last if $self->do_quit;
	}
	return;
}

#receiver
sub receive {
	my $self = shift;
	my $queue = $self->queue;
	die "Need Thread::Queue object for receiver\n" unless $queue;
	my $app = $self->handler;
	my $guard = LWP::Protocol::PSGI->register( $app, host => 'api' );
	my $retry_count = 0;
	my $success = 0;
	my $retval;
	while (my $item = $queue->dequeue) {
		$DB::single = 1;
		my ($type, $message) = @$item;
		last if $type eq 'EOF'; # exit signal
		my $content = $message;
		my $url = sprintf 'http://api/%s/%s.html', $type, lc $type;
		my $req = HTTP::Request->new( POST => $url, ['Content-Type' => 'application/json'], $message );
		my $res = do {
			local $SIG{__WARN__} = sub {};
			$self->lwp->request( $req );
		};
		if ($res->code != 200) {
			$retval = $res->header( 'X-API-Error' ) || $res->status_line;
			$success = 0;
			next;
		}
		my $response = try {
			$self->json->decode( $res->decoded_content );
		} catch {
			$retval = $_;
			$success = 0;
		};
		next if $response == 0;
		if (ref $response ne 'HASH' || !defined $response->{errorCode}) {
			$retval = "Incorrect API response";
			next;
		}
		if ($response->{errorCode}) { # un-success
			$retval = $response->{message} || $res->header( 'X-API-Error' ) || $res->status_line; 
			next;
		}
		$success = 1;
		$retval = $response->{message} || $res->status_line;
	} continue {
		printf $success ? "OK " : "NOT OK ";
		print $retry_count&&!$success ? "$retval : Retry $retry_count\n" : "$retval\n";
		if (!$success && ++$retry_count < $RETRY_COUNT) {
			sleep 0.1 * $retry_count;
			redo;
		}
		$self->save();
		$retry_count = 0;
		
	}
	$self->save();
}
	
sub write_pid {
	(my $fh = IO::File->new(shift->pid_file, 'w'))->print($$);
}

sub eof {
	shift->queue->enqueue( $EOF );
}

my $once = 0;
sub get_messages {
	my $self = shift;
	#return ();
	my $messages = $self->callWOS( GET => getMessages => () );
	return unless $messages && ref( $messages ) eq 'ARRAY';
	return @{ $messages || [] };
}

sub confirm_messages {
	my $self = shift;
	my @messages = @_; # [0..2];
	my $result = $self->callWOS( POST => confirmMessages => \@messages );
	return unless $result && ref( $result ) eq 'HASH';
	print "# ".$result->{message}."\n";
	return 1;
}

sub callWOS {
	my $self = shift;

	# do-nothing if we're quitting
	return if $self->do_quit;

	my $api = $self->apiClient;

	my ($action, $method, $data) = @_;
	my $response = try {
		$api->$method( $data );
	} catch {
		warn "Unable to handle API response ($_)\n";
		$_;
	};
	# retry on invalid API response
	
	if ( $response->$_isa( "ECS::WarehouseOS::Error" ) ) {
		#  don't retry  if we're quitting
		return if $self->do_quit;

		$self->{RETRY} ||= 0;
		$self->{RETRY}++; 
		$self->{RETRY} = List::Util::min( $self->{RETRY}, int( $API_RETRY_MAX_SECONDS / $API_RETRY_INCREMENT_SECONDS ));

		sleep $self->{RETRY} * $API_RETRY_INCREMENT_SECONDS;
		warn "RETRY: ".$self->{RETRY}."\n";

		#  don't retry call if we're quitting
		return if $self->do_quit;

		# prepare @_ to call callWOS again
		unshift @_, $self;
		# avoid overflowing the stack by using goto &$code instead of recursion
		my $sub = $self->can('callWOS');
		goto &$sub;
	}
	$self->{RETRY} = 0;
	if (ref( $response ) eq 'HASH' && $response->{errorCode}) {
		die "Error from API ($method): $response->{message}\n";
	}
	return $response;
}

sub save {  
	my $self = shift;
	my $queue = $self->queue;
	die "Need queue to Save\n" unless $queue;
	my $serializer = $self->serializer;
    my $count = 0;
	lock $queue;
	my @items;
	while (my $item = $queue->peek( $count-- )) {
		unshift @items, $item;
	}
	my ($fh, $tempname) = File::Temp::tempfile();
	$serializer->store(\@items,$fh);
	close $fh;
	File::Copy::move( $tempname, $self->queue_file );
	return;
}

sub load {
	my $self = shift;
	my $serializer = $self->serializer;
	return Thread::Queue->new unless -f $self->queue_file;
	my $queue = try {
		my $items = (-f $self->queue_file ? $serializer->retrieve( $self->queue_file ) : [] );
		Thread::Queue->new( @$items );
	};
	$queue ||= Thread::Queue->new;
	return $queue;
}

__PACKAGE__->meta->make_immutable;
no Moose;

1;
