<commit_before>
#
#  Multi::API  -  The public VNDB API
#

package Multi::API;

use strict;
use warnings;
use Multi::Core;
use AnyEvent::Socket;
use AnyEvent::Handle;
use POE::Filter::VNDBAPI 'encode_filters';
use Digest::SHA 'sha256';
use Encode 'encode_utf8', 'decode_utf8';
use Crypt::ScryptKDF 'scrypt_raw';;
use VNDBUtil 'normalize_query', 'norm_ip';
use JSON::XS;


# what our JSON encoder considers 'true' or 'false'
sub TRUE  () { JSON::XS::true }
sub FALSE () { JSON::XS::false }

my %O = (
  port => 19534,
  logfile => "$VNDB::M{log_dir}/api.log",
  conn_per_ip => 5,
  max_results => 25, # For get vn/release/producer/character
  max_results_lists => 100, # For get votelist/vnlist/wishlist
  default_results => 10,
  throttle_cmd => [ 6, 100 ], # interval between each command, allowed burst
  throttle_sql => [ 60, 1 ], # sql time multiplier, allowed burst (in sql time)
);


my %C;
my $connid = 0;


sub writelog {
  my $c = ref $_[0] && shift;
  my($msg, @args) = @_;
  if(open(my $F, '>>:utf8', $O{logfile})) {
    printf $F "[%s] %s: %s\n", scalar localtime,
      $c ? sprintf '%d %s:%d', $c->{id}, $c->{ip}, $c->{port} : 'global',
      @args ? sprintf $msg, @args : $msg;
    close $F;
  }
}


sub run {
  shift;
  %O = (%O, @_);

  push_watcher tcp_server undef, $O{port}, \&newconn;
  writelog 'API starting up on port %d', $O{port};
}


sub unload {
  $C{$_}{h}->destroy() for keys %C;
  %C = ();
}


sub newconn {
  my $c = {
    fh    => $_[0],
    ip    => $_[1],
    port  => $_[2],
    id    => ++$connid,
    cid   => norm_ip($_[1]),
    filt  => POE::Filter::VNDBAPI->new(),
  };

  if($O{conn_per_ip} <= grep $c->{ip} eq $C{$_}{ip}, keys %C) {
    writelog $c, 'Connection denied, limit of %d connections per IP reached', $O{conn_per_ip};
    close $c->{fh};
    return;
  }

  writelog $c, 'Connected';
  $C{$connid} = $c;

  $c->{h} = AnyEvent::Handle->new(
    rbuf_max =>     50*1024, # Commands aren't very huge, a 50k read buffer should suffice.
    wbuf_max => 5*1024*1024,
    fh       => $c->{fh},
    on_error => sub {
      writelog $c, 'IO error: %s', $_[2];
      $c->{h}->destroy;
      delete $C{$c->{id}};
    },
    on_eof => sub {
      writelog $c, 'Disconnected';
      $c->{h}->destroy;
      delete $C{$c->{id}};
    },
  );
  cmd_read($c);
}


sub cres {
  my($c, $msg, $log, @arg) = @_;
  $msg = $c->{filt}->put([$msg])->[0];
  $c->{h}->push_write($msg);
  writelog $c, '[%2d/%4.0fms %5.0f] %s',
    $c->{sqlq}, $c->{sqlt}*1000, length($msg),
    @arg ? sprintf $log, @arg : $log;
  cmd_read($c);
}


sub cerr {
  my($c, $id, $msg, %o) = @_;
  cres $c, [ error => { id => $id, msg => $msg, %o } ], "Error: %s, %s", $id, $msg;
  return undef;
}


# Wrapper around pg_cmd() that updates the SQL throttle for the client and
# sends an error response if the query error'ed. The callback is not called on
# error.
sub cpg {
  my($c, $q, $a, $cb) = @_;
  pg_cmd $q, $a, sub {
    my($res, $time) = @_;
    $c->{sqlq}++;
    $c->{sqlt} += $time;
    return cerr $c, internal => 'SQL error' if pg_expect $res;
    throttle $O{throttle_sql}, "api_sql_$c->{cid}", $time;
    $cb->($res);
  };
}


sub cmd_read {
  my $c = shift;

  # Prolly should make POE::Filter::VNDBAPI aware of AnyEvent::Handle stuff, so
  # this code wouldn't require a few protocol specific chunks.
  $c->{h}->push_read(line => "\x04", sub {
    my $cmd = $c->{filt}->get([$_[1], "\x04"]);
    die "No or too many commands in a single message" if @$cmd != 1;

    my @arg;
    ($cmd, @arg) = @{$cmd->[0]};

    # log raw message (except login command, which may include a password)
    (my $msg = $_[1]) =~ s/[\r\n]*/ /;
    $msg =~ s/^[\s\r\n\t]+//;
    $msg =~ s/[\s\r\n\t]+$//;
    writelog $c, decode_utf8 "< $msg" if $cmd && $cmd ne 'login';

    # Stats for the current cmd
    $c->{sqlt} = $c->{sqlq} = 0;

    # parse error
    return cerr $c, $arg[0]{id}, $arg[0]{msg} if !defined $cmd;

    # check for thottle rule violation
    for ('cmd', 'sql') {
      my $left = throttle $O{"throttle_$_"}, "api_${_}_$c->{cid}", 0;
      return cerr $c, throttled => 'Throttle limit reached.', type => $_,
          minwait  => int(10*($left))/10+1,
          fullwait => int(10*($left + $O{"throttle_$_"}[0] * $O{"throttle_$_"}[1]))/10+1
        if $left;
    }

    # update commands/second throttle
    throttle $O{throttle_cmd}, "api_cmd_$c->{cid}";
    cmd_handle($c, $cmd, @arg);
  });
}


sub cmd_handle {
  my($c, $cmd, @arg) = @_;<commit_msg>Multi::API: Throttle "throttled" error replies<commit_after>
#
#  Multi::API  -  The public VNDB API
#

package Multi::API;

use strict;
use warnings;
use Multi::Core;
use AnyEvent::Socket;
use AnyEvent::Handle;
use POE::Filter::VNDBAPI 'encode_filters';
use Digest::SHA 'sha256';
use Encode 'encode_utf8', 'decode_utf8';
use Crypt::ScryptKDF 'scrypt_raw';;
use VNDBUtil 'normalize_query', 'norm_ip';
use JSON::XS;


# what our JSON encoder considers 'true' or 'false'
sub TRUE  () { JSON::XS::true }
sub FALSE () { JSON::XS::false }

my %O = (
  port => 19534,
  logfile => "$VNDB::M{log_dir}/api.log",
  conn_per_ip => 5,
  max_results => 25, # For get vn/release/producer/character
  max_results_lists => 100, # For get votelist/vnlist/wishlist
  default_results => 10,
  throttle_cmd => [ 6, 100 ], # interval between each command, allowed burst
  throttle_sql => [ 60, 1 ],  # sql time multiplier, allowed burst (in sql time)
  throttle_thr => [ 2, 10 ],  # interval between "throttled" replies, allowed burst
);


my %C;
my $connid = 0;


sub writelog {
  my $c = ref $_[0] && shift;
  my($msg, @args) = @_;
  if(open(my $F, '>>:utf8', $O{logfile})) {
    printf $F "[%s] %s: %s\n", scalar localtime,
      $c ? sprintf '%d %s:%d', $c->{id}, $c->{ip}, $c->{port} : 'global',
      @args ? sprintf $msg, @args : $msg;
    close $F;
  }
}


sub run {
  shift;
  %O = (%O, @_);

  push_watcher tcp_server undef, $O{port}, \&newconn;
  writelog 'API starting up on port %d', $O{port};
}


sub unload {
  $C{$_}{h}->destroy() for keys %C;
  %C = ();
}


sub newconn {
  my $c = {
    fh    => $_[0],
    ip    => $_[1],
    port  => $_[2],
    id    => ++$connid,
    cid   => norm_ip($_[1]),
    filt  => POE::Filter::VNDBAPI->new(),
  };

  if($O{conn_per_ip} <= grep $c->{ip} eq $C{$_}{ip}, keys %C) {
    writelog $c, 'Connection denied, limit of %d connections per IP reached', $O{conn_per_ip};
    close $c->{fh};
    return;
  }

  writelog $c, 'Connected';
  $C{$connid} = $c;

  $c->{h} = AnyEvent::Handle->new(
    rbuf_max =>     50*1024, # Commands aren't very huge, a 50k read buffer should suffice.
    wbuf_max => 5*1024*1024,
    fh       => $c->{fh},
    on_error => sub {
      writelog $c, 'IO error: %s', $_[2];
      $c->{h}->destroy;
      delete $C{$c->{id}};
    },
    on_eof => sub {
      writelog $c, 'Disconnected';
      $c->{h}->destroy;
      delete $C{$c->{id}};
    },
  );
  cmd_read($c);
}


sub cres {
  my($c, $msg, $log, @arg) = @_;
  $msg = $c->{filt}->put([$msg])->[0];
  $c->{h}->push_write($msg);
  writelog $c, '[%2d/%4.0fms %5.0f] %s',
    $c->{sqlq}, $c->{sqlt}*1000, length($msg),
    @arg ? sprintf $log, @arg : $log;
  cmd_read($c);
}


sub cerr {
  my($c, $id, $msg, %o) = @_;
  cres $c, [ error => { id => $id, msg => $msg, %o } ], "Error: %s, %s", $id, $msg;
  return undef;
}


# Wrapper around pg_cmd() that updates the SQL throttle for the client and
# sends an error response if the query error'ed. The callback is not called on
# error.
sub cpg {
  my($c, $q, $a, $cb) = @_;
  pg_cmd $q, $a, sub {
    my($res, $time) = @_;
    $c->{sqlq}++;
    $c->{sqlt} += $time;
    return cerr $c, internal => 'SQL error' if pg_expect $res;
    throttle $O{throttle_sql}, "api_sql_$c->{cid}", $time;
    $cb->($res);
  };
}


sub cmd_read {
  my $c = shift;

  # Prolly should make POE::Filter::VNDBAPI aware of AnyEvent::Handle stuff, so
  # this code wouldn't require a few protocol specific chunks.
  $c->{h}->push_read(line => "\x04", sub {
    my $cmd = $c->{filt}->get([$_[1], "\x04"]);
    die "No or too many commands in a single message" if @$cmd != 1;

    my @arg;
    ($cmd, @arg) = @{$cmd->[0]};

    # log raw message (except login command, which may include a password)
    (my $msg = $_[1]) =~ s/[\r\n]*/ /;
    $msg =~ s/^[\s\r\n\t]+//;
    $msg =~ s/[\s\r\n\t]+$//;
    writelog $c, decode_utf8 "< $msg" if $cmd && $cmd ne 'login';

    # Stats for the current cmd
    $c->{sqlt} = $c->{sqlq} = 0;

    # parse error
    return cerr $c, $arg[0]{id}, $arg[0]{msg} if !defined $cmd;

    # check for thottle rule violation
    for ('cmd', 'sql') {
      my $left = throttle $O{"throttle_$_"}, "api_${_}_$c->{cid}", 0;
      next if !$left;

      # Too many throttle rule violations? Misbehaving client, disconnect.
      if(throttle $O{throttle_thr}, "api_thr_$c->{cid}") {
        writelog $c, 'Too many throttled replies, disconnecting.';
        $c->{h}->destroy;
        delete $C{$c->{id}};
        return;
      }

      return cerr $c, throttled => 'Throttle limit reached.', type => $_,
          minwait  => int(10*($left))/10+1,
          fullwait => int(10*($left + $O{"throttle_$_"}[0] * $O{"throttle_$_"}[1]))/10+1;
    }

    # update commands/second throttle
    throttle $O{throttle_cmd}, "api_cmd_$c->{cid}";
    cmd_handle($c, $cmd, @arg);
  });
}


sub cmd_handle {
  my($c, $cmd, @arg) = @_;
{'fraction_non_alphanumeric': 0.21583734617442482, 'fraction_numerical': 0.018512573568753344, 'mean_word_length': 2.990606319385141, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 140, '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': '6226095', 'n_tokens_mistral': 4033, 'n_tokens_neox': 3838, 'n_words': 986}