logbot-consumer

#!/usr/bin/perl
# import event json files created by logbot-irc and inserts into network
# database

use local::lib;
use v5.10;
use strict;
use warnings;

use FindBin qw( $RealBin );
use lib "$RealBin/lib";

BEGIN { $ENV{TZ} = 'UTC' }

use Encode qw( encode );
use LogBot::Config qw( find_config load_config reload_config );
use LogBot::Database qw( dbh execute_with_retry );
use LogBot::JobQueue ();
use LogBot::Util qw( event_to_string file_for logbot_init timestamp touch );

my $config = load_config(find_config(shift)) // die "syntax: logbot-consumer <config file>\n";
logbot_init($config);

my $quit = 0;
$SIG{INT} = sub { $quit = 1 };

# init/cache database/queue
dbh($config, read_write => 1, cached => 1);
my $job_queue = LogBot::JobQueue->new($config);

my $last_status = '';
while (!$quit) {

    # wait for next job, blocking call
    my ($job, $event) = $job_queue->fetch_job();
    last unless defined $job;

    if ($ENV{DEBUG}) {
        say timestamp(), ' << ', encode('UTF-8', event_to_string($event));
    }

    $config = reload_config($config);

    # deal with lag between queue and config
    my $ignore = 0;
    if (my $channel = $config->{channels}->{ $event->{channel} }) {
        $ignore = 1 if $channel->{no_logs};
        $ignore = 1 if $channel->{disabled};
    } else {
        $ignore = 1;
    }
    if ($ignore) {
        say timestamp(), ' -- ignoring event' if $ENV{DEBUG};
        $job_queue->delete_job($job);
        next;
    }

    if ($event->{type} == 3) {

        # topic
        $quit = !execute_with_retry(
            $config,
            sub {
                my ($dbh) = @_;
                return 0 if $quit;

                my ($id, $time, $topic) =
                    $dbh->selectrow_array('SELECT id,time,topic FROM topics WHERE channel=?', undef, $event->{channel});
                $topic //= '';
                if (!$id) {
                    $dbh->do('INSERT INTO topics(time,channel,topic) VALUES (?,?,?)',
                        undef, $event->{time}, $event->{channel}, $event->{text});
                } elsif ($event->{time} > $time && $event->{text} ne $topic) {
                    $dbh->do('UPDATE topics SET time=?, topic=? WHERE id=?',
                        undef, $event->{time}, $event->{text}, $id);
                    touch(file_for($config, 'topics_lastmod'));
                }

                return 1;
            }
        );

    } else {

        # channel message, action, or notice
        $quit = !execute_with_retry(
            $config,
            sub {
                my ($dbh) = @_;
                return 0 if $quit;

                $dbh->do('INSERT INTO logs(time,channel,nick,type,text) VALUES (?,?,?,?,?)',
                    undef, $event->{time}, $event->{channel}, $event->{nick}, $event->{type}, $event->{text});

                return 1;
            }
        );
    }

    $job_queue->delete_job($job);
    undef $job;
}