#!/usr/bin/perl

use strict;
use warnings;

use threads;
use threads::shared;

use Acme::Damn ();
use Data::Serializer ();
use File::Copy ();
use File::Temp ();
use Thread::Queue ();
use Time::HiRes ();

package API::Utils::WorkQueue;

use Carp;

INIT { _init(); };
END { _cleanup(); };

# public
our $NUM_THREADS = 1;
our $PERSISTENCE_FILENAME = "/tmp/.work_queue";
our $WORK = sub { return 1; };
our $WORK_ITEM_MATCHER;

# private
my $work_queue;
my $requires_save;
my %work_items_processing;

return 1;

###############################################################################

sub enqueue($) {
    my $self = shift;
    my $x = shift;

    # adding an item to the queue will make it available to our worker threads
    $work_queue->enqueue($x);
    $requires_save ++;

    return 1;
}

###############################################################################

sub status() {
    my $self = shift;

    # simple status report, can be added to later
    
    my $num_in_queue = scalar(@{$work_queue});
    return "$num_in_queue items in the queue\n";
}

###############################################################################

sub queuesize() {
    my $self = shift;
    return scalar(@{ $work_queue });
}

###############################################################################

sub quit() {
    my $self = shift;
    my $x = shift;

    if($x == 1) {
        exit;
    }

    return 0;
}

###############################################################################

# Checks if an item_type + item_id combo is currently queued or processing
#  by a worker
sub checkItemPending {
    my $class = shift;
    my ( $matchData ) = @_;

    return { 'queued' => 0 } unless $WORK_ITEM_MATCHER;

    my @foundItems = _findInQueue( $matchData );
    if ( @foundItems > 0 ) {
        return { 'queued' => 1 };
    }

    @foundItems = _findInProcessing( $matchData );
    if ( @foundItems > 0 ) {
        return { 'queued' => 1 };
    }

    return { 'queued' => 0 };
}

###############################################################################

sub _findInQueue {
    my ( $matchData ) = @_;

    croak "matchData required" unless $matchData;

    # nothing can be in the queue if the work queue isn't initialized
    return unless $work_queue;

    my @found;
    for(my $i = 0; $i < $work_queue->pending(); $i ++) {
        my $item = $work_queue->peek($i);
        if (
            defined $item
            and &$WORK_ITEM_MATCHER( $item, $matchData )
        ) {
            push(@found, $item);
        }
    }

    return @found;
}

###############################################################################

sub _findInProcessing {
    my ( $matchData ) = @_;

    croak "matchData required" unless $matchData;

    my @found;
    for my $item ( values %work_items_processing ) {
        if ( &$WORK_ITEM_MATCHER( $item, $matchData ) ) {
            push(@found, $item);
        }
    }

    return @found;
}
###############################################################################


sub _do_work {
    local $SIG{'TERM'} = sub { threads->exit(); };
    my $threadID = threads->tid();

    # keep trying to get work
    while(1) {
        my $item = $work_queue->dequeue_nb();
        if(not defined $item) {
            Time::HiRes::sleep(0.01);
            next;
        }

        $work_items_processing{ $threadID } = $item;

        # if the process tries to die, we'll just assume
        # that it's a permanent failure
        my $result = -1;
        eval {
            # if $item is blessed, get rid of it
            if(UNIVERSAL::can($item, 'isa')) {
                $item = Acme::Damn::damn($item);
            }
            
            # get rid of the thread gunk Thread::Queue stuck in there
            my $work_item = _clone($item);

            # WORK() should return:
            #   1 for success
            #   0 for temporary failure (will try again later)
            #   -1 for permanent failure (will not try again)
            $result = &$WORK($work_item);
        };
        if($@) {
            warn "Got $@\n";
        }

        # in case of a temporary failure, append to end of queue
        if(defined $result
            and $result == 0
            ) {
            $work_queue->enqueue($item);
        }

        delete $work_items_processing{ $threadID };
        $requires_save ++;
    }
}

###############################################################################

sub _load_queue {
    my $serializer = Data::Serializer->new();

    if(-f $PERSISTENCE_FILENAME) {
        # the serializer might die if the persistence file is corrupt
        eval {
            my $queue = $serializer->retrieve($PERSISTENCE_FILENAME);
            $work_queue->enqueue(@{$queue});
        };
        if($@) {
            warn "Problem loading saved queue: $@\n";
            unlink $PERSISTENCE_FILENAME;
        }
    }
    else {
        unlink $PERSISTENCE_FILENAME;
    }
}

###############################################################################

sub _save_queue {
    # the default serializer seems to work well enough. JSON
    # and Storable have their quirks in threaded code.
    my $serializer = Data::Serializer->new();

    # get all the items in the queue
    my @items = ();
    for(my $i = 0; $i < $work_queue->pending(); $i ++) {
        my $item = $work_queue->peek($i);
        if(defined $item) {
            push(@items, $item);
        }
    }

    # making sure that the persistence file doesn't get saved "halfway".
    # `move` will just rename the file, which is as close to
    # an atomic transaction as we're going to get.

    my ($of, $tempname) = File::Temp::tempfile();
    $serializer->store(\@items, $of);
    close($of);

    File::Copy::move($tempname, $PERSISTENCE_FILENAME);
}

###############################################################################

sub _init {
    $work_queue = Thread::Queue->new();
    _load_queue();
    $requires_save = 1;
    threads::shared::share($requires_save);
    threads::shared::share(%work_items_processing);

    # spawn some worker threads
    for(1 .. $NUM_THREADS) {
        # make sure to create these in scalar context
        my $thr = threads->create(\&_do_work)->detach();
    }

    # start a dedicated thread for saving the queue. this can get pretty slow
    # if there is a large queue. if your worker threads spend all their
    # time saving the queue, then they'll fall behind and the queue will
    # get bigger and bigger.
    my $thr = threads->create(sub {
            while(1) {
                if($requires_save) {
                    _save_queue();
                    $requires_save = 0;
                }
                sleep 1;
            }
        })->detach();
}

###############################################################################

sub _cleanup {
    # our threads would get cleaned up anyways, this just saves us a warning
    for my $thread (threads->list()) {
        $thread->kill("SIGTERM")->detach();
    }
}

###############################################################################
# recursively copy a data structure
sub _clone {
    my $ref = shift;
    my $type = ref $ref;

    if( $type eq 'HASH' ) {
        return { map _clone( $_ ), %{$ref} };
    }
    elsif( $type eq 'ARRAY' ) {
        return [ map _clone( $_ ), @{$ref} ];
    }
    elsif( $type eq 'REF' ) {
        return \ _clone( $$ref );
    }
    else {
        return $ref;
    }
}

###############################################################################
