File Coverage

blib/lib/Bot/Telegram.pm

Criterion	Covered	Total	%
statement	168	168	100.0
branch	49	50	98.0
condition	26	31	83.8
subroutine	35	35	100.0
pod	13	13	100.0
total	291	297	97.9

package Bot::Telegram;
# ABSTRACT: a micro^W nano framework for creating Telegram bots based on L
our $VERSION = '1.10'; # VERSION

use v5.16.3;

use Mojo::Base 'Mojo::EventEmitter';
use WWW::Telegram::BotAPI;
use Mojo::Promise;
use Mojo::Log;

use Mojo::JSON 'encode_json';
use Mojo::Transaction::HTTP;
use Mojo::Message::Response;

use Bot::Telegram::X::InvalidArgumentsError;
use Bot::Telegram::X::InvalidStateError;

use constant ERR_NODETAILS => 'no details available';

use constant DEFAULT_POLLING_TIMEOUT => 20;
use constant DEFAULT_POLLING_INTERVAL => 0.3;

use constant DEFAULT_POLLING_ERROR_CB => sub {
  my ($self, $tx, $type) = @_;

  my $message = sub {
    return $tx -> {error}{msg}
      unless $self -> is_async;

    for ($type) {
      /agent/ and return $tx -> error -> {message};
      /api/ and return ($tx -> res -> json // {}) -> {description};
    }
  } -> () || ERR_NODETAILS;

  $self -> log -> warn("Polling failed (error type: $type): $message");
};

use constant DEFAULT_CALLBACK_ERROR_CB => sub {
  my ($self, undef, $err) = @_;
  $self -> log -> warn("Update processing failed: $err");
};

has [qw/api current_update polling_config/];
has [qw/_polling
        _polling_timer
        _polling_interval
        _polling_request_id/
    ];

has callbacks => sub { {} };
has ioloop => sub { Mojo::IOLoop -> new };
has log => sub { Mojo::Log -> new -> level('info') };

sub new {
  my $self = shift -> SUPER::new(@_);

  $self -> on(polling_error  => DEFAULT_POLLING_ERROR_CB);
  $self -> on(callback_error => DEFAULT_CALLBACK_ERROR_CB);

  $self
}

################################################################################
# General
################################################################################

sub init_api {
  my ($self, %args) = @_;

  Bot::Telegram::X::InvalidArgumentsError
    -> throw('No token provided')
    unless exists $args{token};

  $args{async} //= 1;

  $self -> api(WWW::Telegram::BotAPI -> new(%args));

  return $self;
}

sub is_async {
  my $self = shift;

  Bot::Telegram::X::InvalidStateError
    -> throw('API is not initialized')
    unless $self -> api;

  $self -> api -> {async};
}

sub api_request { shift -> api -> api_request(@_) }
sub api_request_p {
  my ($self, @args) = @_;

  Mojo::Promise -> new(sub {
    my ($resolve, $reject) = @_;

    $self -> api -> api_request(@args, sub {
      my ($ua, $tx) = @_;
      my $response = $tx -> res -> json;

      ((!$tx -> error && ref $response && $$response{ok})
        ? $resolve
        : $reject) -> ($ua, $tx);
    })
  });
}

################################################################################
# Callbacks
################################################################################

sub set_callbacks {
  my ($self, %cbs) = @_;

  while ( my ($key, $val) = each %cbs) {
    $self -> callbacks -> {$key} = $val;
  }

  return $self;
}

sub remove_callbacks {
  my ($self, @events) = @_;

  foreach my $event (@events) {
    delete $self -> callbacks -> {$event};
  }

  return $self;
}

################################################################################
# Updates
################################################################################

sub shift_offset {
  my $self = shift;

  for (my $update = $self -> current_update) {
    $self -> polling_config -> {offset} = $$update{update_id} + 1
      if $$update{update_id} >= $self -> polling_config -> {offset};
  }

  $self
}

sub process_update {
  my ($self, $update) = @_;

  $self -> current_update($update);
  my $type = $self -> _get_update_type($update);

  eval {
    # If update type is recognized, call the appropriate callback
    if ($type) {
      $self -> callbacks
            -> {$type}
            -> ($self, $update);
    }

    # Otherwise report an unknown update
    else { $self -> emit(unknown_update => $update) }
  };

  # Report a callback error if we failed to handle the update
  $self -> emit(callback_error => $update, $@) if $@;

  return $self;
}

# Return the update type if we have a callback for it
# Or just return zero, if we don't
sub _get_update_type {
  my ($self, $update) = @_;

  exists $$update{$_}
    and return $_
    for keys %{ $self -> callbacks };

  return 0;
}

################################################################################
# Webhook
################################################################################

sub set_webhook {
  my ($self, $config, $cb) = @_;

  Bot::Telegram::X::InvalidStateError
    -> throw('Disable long polling first')
    if $self -> is_polling;

  Bot::Telegram::X::InvalidArgumentsError
    -> throw('No config provided')
    unless ref $config;

  $self -> api -> api_request(
    setWebhook => $config,
    ref $cb eq 'CODE' ? $cb : undef);

  return $self;
}

################################################################################
# Long polling
################################################################################

sub start_polling {
  my $self = shift;
  my $config = $self -> polling_config;
  if (ref $_[0] eq 'HASH') {
    $config = shift;
    $config -> {offset} //= 0; # make sure we won't get any uninitiailzed warnings in shift_offset
  }

  my (%opts) = @_;

  if ($opts{restart}) {
    $self -> stop_polling;
  } else {
    Bot::Telegram::X::InvalidStateError
      -> throw('Already running')
      if $self -> is_polling;
  }

  $self -> polling_config($config // { timeout => DEFAULT_POLLING_TIMEOUT, offset => 0 });

  $self -> _polling_interval($opts{interval} // DEFAULT_POLLING_INTERVAL);
  $self -> _polling(1);
  $self -> _poll;
}

sub stop_polling {
  my $self = shift;

  return $self unless $self -> is_polling;

  for (my $agent = $self -> api -> agent) {
    $self -> _polling(undef);

    # In synchronous mode, it's enough to simply clear state
    return $self -> _polling_interval(undef)
      unless $agent -> isa('Mojo::UserAgent')
      and $self -> is_async;

    # In asynchronous mode, we also need to cancel existing timers
    for (my $loop = Mojo::IOLoop -> singleton) {
      $loop -> remove($self -> _polling_request_id);
      $loop -> remove($self -> _polling_timer)
        if $self -> _polling_timer; # if another request is scheduled, cancel it
    }

    # Reset state
    $self -> _polling_request_id(undef)
          -> _polling_interval(undef)
          -> _polling_timer(undef);
  }

  $self
}

sub is_polling { !! shift -> _polling }

# In asynchronous mode: process getUpdates response or handle errors, if any
# In synchronous mode, WWW::Telegram::BotAPI::parse_error takes care of error handling for us.
sub _process_getUpdates_results {
  my $self = shift;
  my $async = $self -> is_async;
  my ($response, $error);

  $self -> log -> trace('processing getUpdates results');

  my $retry_after;

  if ($async) {
    my ($ua, $tx) = @_;
    $response = $tx -> res -> json // {};

    # Error
    if ($error = ($tx -> error or not $$response{ok})) {
      my $type = eval {
        return 'api' if $$response{error_code};
        return 'agent' if $tx -> error;
      } // 'unknown';

      $self -> emit(polling_error => $tx, $type);
    }
  } else {
    ($response, $error) = @_;
    # NOTE: $response and $error are mutually exclusive - only one is `defined` at a time

    if ($error) {
      $error = $self -> api -> parse_error;

      # no way to access the original $tx in synchronous mode
      # https://metacpan.org/dist/WWW-Telegram-BotAPI/source/lib/WWW/Telegram/BotAPI.pm#L228
      $self -> emit(polling_error => { error => $error }, $$error{type});
    }
  }

  # Handle rate limits
  if (exists $response -> {parameters}{retry_after}) {
    $retry_after = $response -> {parameters}{retry_after};
    $self -> log -> info("Rate limit exceeded, waiting ${retry_after}s before polling again");
  }

  # Process the updates we have retrieved (if any) and poll for more
  #  (unless someone or something has disabled the polling loop in the meantime)
  unless ($error) {
    for my $result ($$response{result}) {
      # last unless ref $result eq 'ARRAY'; # nothing to process

      $self -> process_update($_)
            -> shift_offset
        for @$result;
    }
  }

  return unless $self -> is_polling;
  $self -> log -> trace('still polling, scheduling another iteration...');

  if ($async) {
    my $tid = Mojo::IOLoop -> timer(
      $retry_after // $self -> _polling_interval,
      sub { $self -> tap(sub { $self -> log -> trace("it's polling time!") })
                  -> _poll });

    $self -> _polling_timer($tid);
  } else {
    my $d = $retry_after // $self -> _polling_interval;

    # Sleep
    $self -> ioloop -> timer($d, sub { $self -> ioloop -> stop });
    $self -> ioloop -> start;
    $self -> log -> trace("it's polling time!");

    $self -> _poll;
  }
}

sub _poll {
  my $self = shift;

  $self -> log -> trace('polling');

  if ($self -> is_async) {
    my $id = $self -> api -> api_request(
      getUpdates => $self -> polling_config,
      sub { $self -> _process_getUpdates_results(@_) }
    );

    # Assuming api_request always returns a valid ioloop connection ID when in asynchronous mode...
    $self -> _polling_request_id($id);
  } else {
    my $response = eval {
      $self -> api -> api_request(
        getUpdates => $self -> polling_config)
    };

    $self -> _process_getUpdates_results($response, $@);
  }
}

1

__END__