File Coverage

blib/lib/Net/Async/WebSearch.pm

Criterion	Covered	Total	%
statement	258	280	92.1
branch	75	114	65.7
condition	56	109	51.3
subroutine	39	44	88.6
pod	12	12	100.0
total	440	559	78.7

package Net::Async::WebSearch;
# ABSTRACT: IO::Async multi-provider web search aggregator
use strict;
use warnings;
use parent 'IO::Async::Notifier';

use Carp qw( croak );
use Future ();
use Future::Utils qw( fmap_void );
use HTTP::Request::Common qw( GET );
use Net::Async::HTTP ();
use URI ();

use Net::Async::WebSearch::Provider ();
use Net::Async::WebSearch::Result ();

our $VERSION = '0.002';

# Reciprocal Rank Fusion constant (Cormack et al.)
our $RRF_K = 60;

sub _init {
  my ( $self, $args ) = @_;
  $self->SUPER::_init($args);
  $self->{providers} = [];
  $self->{http}      = delete $args->{http};
  $self->{rrf_k}     = delete $args->{rrf_k} // $RRF_K;
  $self->{default_limit} = delete $args->{default_limit} // 10;
  $self->{per_provider_limit} = delete $args->{per_provider_limit} // 10;
  $self->{fetch_concurrency} = delete $args->{fetch_concurrency} // 100;
  $self->{fetch_concurrency_per_target_ip}
    = delete $args->{fetch_concurrency_per_target_ip} // 5;
  $self->{fetch_timeout}     = delete $args->{fetch_timeout};
  $self->{fetch_max_bytes}   = delete $args->{fetch_max_bytes};
  $self->{fetch_user_agent}  = delete $args->{fetch_user_agent};
  if ( my $provs = delete $args->{providers} ) {
    $self->add_provider($_) for @$provs;
  }
  return;
}

sub configure_unknown {
  my ( $self, %args ) = @_;
  for my $k (qw(
    http rrf_k default_limit per_provider_limit providers
    fetch_concurrency fetch_concurrency_per_target_ip
    fetch_timeout fetch_max_bytes fetch_user_agent
  )) {
    delete $args{$k};
  }
  return unless %args;
  croak "Unknown configuration keys: ".join(',', sort keys %args);
}

sub providers         { @{ $_[0]->{providers} } }
sub rrf_k             { @_ > 1 ? ($_[0]->{rrf_k} = $_[1]) : $_[0]->{rrf_k} }
sub default_limit     { @_ > 1 ? ($_[0]->{default_limit} = $_[1]) : $_[0]->{default_limit} }
sub per_provider_limit{ @_ > 1 ? ($_[0]->{per_provider_limit} = $_[1]) : $_[0]->{per_provider_limit} }

sub add_provider {
  my ( $self, $p ) = @_;
  croak "provider must be a Net::Async::WebSearch::Provider"
    unless ref $p && $p->isa('Net::Async::WebSearch::Provider');
  my %taken = map { $_->name => 1 } @{ $self->{providers} };
  if ( $taken{ $p->name } ) {
    my $base = $p->name;
    my $n    = 2;
    $n++ while $taken{ "$base#$n" };
    $p->name( "$base#$n" );
  }
  push @{ $self->{providers} }, $p;
  return $p;
}

sub provider {
  my ( $self, $name ) = @_;
  for my $p ( @{ $self->{providers} } ) {
    return $p if $p->name eq $name;
  }
  return;
}

sub providers_matching {
  my ( $self, $sel ) = @_;
  return grep { $_->matches($sel) } @{ $self->{providers} };
}

sub http {
  my ( $self ) = @_;
  return $self->{http} if $self->{http};
  my $http = Net::Async::HTTP->new(
    user_agent => 'Net-Async-WebSearch/'.$VERSION,
    max_connections_per_host => $self->{fetch_concurrency_per_target_ip} // 5,
    max_in_flight => 0,
  );
  $self->add_child($http);
  return $self->{http} = $http;
}

sub _on_added_to_loop {
  my ( $self, $loop ) = @_;
  $self->SUPER::_on_added_to_loop($loop) if $self->can('SUPER::_on_added_to_loop');
  $self->http;
}

#----------------------------------------------------------------------
# Provider selection
#----------------------------------------------------------------------

sub _select_providers {
  my ( $self, %args ) = @_;
  my $only    = $args{only};
  my $exclude = $args{exclude};
  my @sel;
  for my $p ( @{ $self->{providers} } ) {
    next unless $p->enabled;
    if ( $only && @$only ) {
      next unless grep { $p->matches($_) } @$only;
    }
    if ( $exclude && @$exclude ) {
      next if grep { $p->matches($_) } @$exclude;
    }
    push @sel, $p;
  }
  return @sel;
}

#----------------------------------------------------------------------
# URL normalization for dedup
#----------------------------------------------------------------------

sub _normalize_url {
  my ( $self, $url ) = @_;
  return '' unless defined $url && length $url;
  my $u = eval { URI->new($url)->canonical } or return lc $url;
  my $s = $u->as_string;
  $s =~ s{#.*$}{};
  $s =~ s{/+$}{};
  return lc $s;
}

#----------------------------------------------------------------------
# Fetch (optional page-body retrieval after search)
#----------------------------------------------------------------------

sub _fetch_one {
  my ( $self, $result, %args ) = @_;
  my $url = $result->url;
  return Future->done unless defined $url && length $url;
  my $req = GET($url);
  $req->header( 'User-Agent' => $args{user_agent} // $self->{fetch_user_agent}
                                                  // 'Net-Async-WebSearch/'.$VERSION );
  $req->header( 'Accept'     => $args{accept}     // '*/*' );

  my %req_args = ( request => $req );
  if ( defined $args{timeout} ) {
    $req_args{timeout} = $args{timeout};
  }
  # NB: max_bytes is enforced on the decoded body below — Net::Async::HTTP
  # does not cap the on-the-wire body length itself.

  return $self->http->do_request(%req_args)->then(sub {
    my ( $resp ) = @_;
    my $ct      = $resp->header('Content-Type');
    my $charset = ( $ct && $ct =~ /charset=([^\s;]+)/ ) ? lc $1 : undef;
    my $body    = eval { $resp->decoded_content };
    $body = $resp->content if !defined $body;
    if ( defined $args{max_bytes} && defined $body && length($body) > $args{max_bytes} ) {
      $body = substr($body, 0, $args{max_bytes});
    }
    $result->fetched({
      ok           => $resp->is_success ? 1 : 0,
      status       => $resp->code,
      status_line  => $resp->status_line,
      final_url    => ( $resp->request ? $resp->request->uri.'' : $url ),
      content_type => $ct,
      charset      => $charset,
      body         => $body,
      error        => $resp->is_success ? undef : $resp->status_line,
    });
    Future->done;
  })->else(sub {
    my ( $err ) = @_;
    $result->fetched({
      ok           => 0,
      status       => undef,
      status_line  => undef,
      final_url    => $url,
      content_type => undef,
      charset      => undef,
      body         => undef,
      error        => "$err",
    });
    Future->done;
  });
}

sub _fetch_results {
  my ( $self, $results_ref, %args ) = @_;
  my $n = $args{fetch} or return Future->done($results_ref);
  return Future->done($results_ref) unless $results_ref && @$results_ref;
  my $cap = $n < @$results_ref ? $n : scalar @$results_ref;
  my @targets = @{$results_ref}[ 0 .. $cap - 1 ];
  my $conc = $args{fetch_concurrency} // $self->{fetch_concurrency} // 100;
  my %fargs = (
    timeout    => $args{fetch_timeout}    // $self->{fetch_timeout},
    max_bytes  => $args{fetch_max_bytes}  // $self->{fetch_max_bytes},
    user_agent => $args{fetch_user_agent} // $self->{fetch_user_agent},
    accept     => $args{fetch_accept},
  );
  my $on_fetch = $args{on_fetch};
  return fmap_void(
    sub {
      my $r = shift;
      $self->_fetch_one($r, %fargs)->on_done(sub {
        $on_fetch->($r) if $on_fetch;
      });
    },
    foreach     => \@targets,
    concurrent  => $conc,
  )->then(sub { Future->done($results_ref) });
}

#----------------------------------------------------------------------
# Core dispatch
#----------------------------------------------------------------------

sub _dispatch {
  my ( $self, %args ) = @_;
  my $query = $args{query};
  croak "search requires 'query'" unless defined $query && length $query;

  my @providers = $self->_select_providers( only => $args{only}, exclude => $args{exclude} );
  return ( [], [] ) unless @providers;

  my %base = (
    limit      => $args{per_provider_limit} // $self->per_provider_limit,
    language   => $args{language},
    region     => $args{region},
    safesearch => $args{safesearch},
  );

  my $popts = $args{provider_opts} || {};
  my @futures;
  for my $p (@providers) {
    my %merged = %base;
    # Apply provider_opts in insertion order: class-leaf / tag first,
    # exact name last so it wins.
    for my $sel ( sort { ($a eq $p->name) <=> ($b eq $p->name) } keys %$popts ) {
      next unless $p->matches($sel);
      %merged = ( %merged, %{ $popts->{$sel} } );
    }
    push @futures, {
      provider => $p,
      future   => $p->search( $self->http, $query, \%merged ),
    };
  }
  return ( \@providers, \@futures );
}

#----------------------------------------------------------------------
# Mode: collect
#----------------------------------------------------------------------

sub search {
  my ( $self, %args ) = @_;
  $args{mode} ||= 'collect';
  return $self->search_stream(%args) if $args{mode} eq 'stream';
  return $self->search_race(%args)   if $args{mode} eq 'race';

  my ( $provs, $futs ) = $self->_dispatch(%args);
  return Future->done({ results => [], errors => [], stats => { providers => 0 } })
    unless $futs && @$futs;

  my $limit = $args{limit} // $self->default_limit;
  my $k     = $self->rrf_k;

  my @wrapped = map {
    my $name = $_->{provider}->name;
    $_->{future}->else(sub {
      my @err = @_;
      Future->done({ __error => 1, provider => $name, error => $err[0] });
    });
  } @$futs;

  return Future->needs_all(@wrapped)->then(sub {
    my @per_provider = @_;
    my @errors;
    my %agg;       # normalized_url => { result => $first, score => $n, providers => {p=>rank,...} }
    for my $payload (@per_provider) {
      if ( ref $payload eq 'HASH' && $payload->{__error} ) {
        push @errors, { provider => $payload->{provider}, error => $payload->{error} };
        next;
      }
      for my $r ( @{ $payload || [] } ) {
        my $key = $self->_normalize_url( $r->url );
        next unless length $key;
        my $slot = $agg{$key} ||= { result => $r, score => 0, providers => {} };
        $slot->{providers}{ $r->provider } = $r->rank;
        $slot->{score} += 1 / ( $k + $r->rank );
        # Prefer a result that has a snippet, if the first had none.
        if ( !$slot->{result}->snippet && $r->snippet ) {
          $slot->{result} = $r;
        }
      }
    }
    my @merged =
      map {
        my $s = $_;
        $s->{result}->score( $s->{score} );
        $s->{result}->extra->{providers} = { %{ $s->{providers} } };
        $s->{result};
      }
      sort { $b->{score} <=> $a->{score} }
      values %agg;

    @merged = splice @merged, 0, $limit if @merged > $limit;

    my $final = {
      results => \@merged,
      errors  => \@errors,
      stats   => {
        providers       => scalar @$provs,
        providers_ok    => ( scalar @$provs ) - scalar @errors,
        providers_error => scalar @errors,
        merged          => scalar @merged,
      },
    };

    if ( $args{fetch} ) {
      return $self->_fetch_results(\@merged, %args)->then(sub {
        $final->{stats}{fetched} = scalar grep { $_->fetched } @merged;
        Future->done($final);
      });
    }
    return Future->done($final);
  });
}

#----------------------------------------------------------------------
# Mode: stream — fire on_result per result as soon as each provider arrives
#----------------------------------------------------------------------

sub search_stream {
  my ( $self, %args ) = @_;
  my $cb = $args{on_result} or croak "stream mode requires 'on_result' coderef";
  my $on_provider_done  = $args{on_provider_done};
  my $on_provider_error = $args{on_provider_error};
  my $on_fetch          = $args{on_fetch};

  my ( $provs, $futs ) = $self->_dispatch(%args);
  return Future->done({ results => [], errors => [], stats => { providers => 0 } })
    unless $futs && @$futs;

  my @all;
  my @errors;
  my $seen_key = {};

  my $fetch_cap      = $args{fetch} // 0;
  my $fetch_started  = 0;
  my @fetch_futures;
  my %fargs = (
    timeout    => $args{fetch_timeout}    // $self->{fetch_timeout},
    max_bytes  => $args{fetch_max_bytes}  // $self->{fetch_max_bytes},
    user_agent => $args{fetch_user_agent} // $self->{fetch_user_agent},
    accept     => $args{fetch_accept},
  );

  my @wrapped;
  for my $entry (@$futs) {
    my $name = $entry->{provider}->name;
    my $f = $entry->{future}->then(sub {
      my ( $results ) = @_;
      for my $r (@$results) {
        my $key = $self->_normalize_url( $r->url );
        next unless length $key;
        next if $seen_key->{$key}++;
        push @all, $r;
        $cb->($r);
        if ( $fetch_cap && $fetch_started < $fetch_cap ) {
          $fetch_started++;
          push @fetch_futures, $self->_fetch_one($r, %fargs)->on_done(sub {
            $on_fetch->($r) if $on_fetch;
          });
        }
      }
      $on_provider_done->( $name, $results ) if $on_provider_done;
      Future->done;
    })->else(sub {
      my ( $err ) = @_;
      push @errors, { provider => $name, error => $err };
      $on_provider_error->( $name, $err ) if $on_provider_error;
      Future->done;
    });
    push @wrapped, $f;
  }

  return Future->needs_all(@wrapped)->then(sub {
    # Wait for any pending fetches the stream kicked off.
    return @fetch_futures
      ? Future->needs_all(@fetch_futures)
      : Future->done;
  })->then(sub {
    Future->done({
      results => \@all,
      errors  => \@errors,
      stats   => {
        providers       => scalar @$provs,
        providers_ok    => ( scalar @$provs ) - scalar @errors,
        providers_error => scalar @errors,
        emitted         => scalar @all,
        ( $fetch_cap ? ( fetched => scalar grep { $_->fetched } @all ) : () ),
      },
    });
  });
}

#----------------------------------------------------------------------
# Mode: race — resolve with whichever provider comes back first (non-error)
#----------------------------------------------------------------------

sub search_race {
  my ( $self, %args ) = @_;
  my ( $provs, $futs ) = $self->_dispatch(%args);
  return Future->done({ results => [], errors => [], stats => { providers => 0 } })
    unless $futs && @$futs;

  my $limit = $args{limit} // $self->default_limit;

  # Wrap each so we can distinguish "first success" from "first completion".
  my $winner = Future->new;
  my @errors;
  my @remaining = map {
    my $name = $_->{provider}->name;
    my $f = $_->{future};
    $f->on_done(sub {
      my ( $results ) = @_;
      return if $winner->is_ready;
      my @top = @$results;
      @top = splice @top, 0, $limit if @top > $limit;
      $winner->done({
        provider => $name,
        results  => \@top,
        errors   => \@errors,
        stats    => {
          providers      => scalar @$provs,
          winning        => $name,
        },
      });
    });
    $f->on_fail(sub {
      my ( $err ) = @_;
      push @errors, { provider => $name, error => $err };
    });
    $f;
  } @$futs;

  # If everyone fails, resolve with errors.
  Future->wait_all(@remaining)->on_done(sub {
    return if $winner->is_ready;
    $winner->done({
      provider => undef,
      results  => [],
      errors   => \@errors,
      stats    => {
        providers       => scalar @$provs,
        providers_error => scalar @errors,
      },
    });
  });

  return $winner unless $args{fetch};
  return $winner->then(sub {
    my ( $out ) = @_;
    return Future->done($out) unless @{ $out->{results} };
    $self->_fetch_results( $out->{results}, %args )->then(sub {
      $out->{stats}{fetched} = scalar grep { $_->fetched } @{ $out->{results} };
      Future->done($out);
    });
  });
}

1;

__END__