travelynx/lib/Travelynx/Command/work.pm

package Travelynx::Command::work;

# Copyright (C) 2020-2023 Birte Kristina Friesel
#
# SPDX-License-Identifier: AGPL-3.0-or-later
use Mojo::Base 'Mojolicious::Command';
use Mojo::Promise;

use DateTime;
use JSON;
use List::Util;

has description => 'Update real-time data of active journeys';

has usage => sub { shift->extract_usage };

sub run {
	my ($self) = @_;

	my $now              = DateTime->now( time_zone => 'Europe/Berlin' );
	my $checkin_deadline = $now->clone->subtract( hours => 48 );
	my $json             = JSON->new;

	if ( -e 'maintenance' ) {
		$self->app->log->debug('work: "maintenance" file found, aborting');
		return;
	}

	my $num_incomplete = $self->app->in_transit->delete_incomplete_checkins(
		earlier_than => $checkin_deadline );

	if ($num_incomplete) {
		$self->app->log->debug("Removed ${num_incomplete} incomplete checkins");
	}

	my $errors = 0;

	for my $entry ( $self->app->in_transit->get_all_active ) {

		if ( -e 'maintenance' ) {
			$self->app->log->debug('work: "maintenance" file found, aborting');
			return;
		}

		my $uid      = $entry->{user_id};
		my $dep      = $entry->{dep_eva};
		my $arr      = $entry->{arr_eva};
		my $train_id = $entry->{train_id};

		if ( $entry->{is_hafas} ) {

			eval {

				$self->app->hafas->get_journey_p(
					trip_id => $train_id,
					service => $entry->{backend_name}
				)->then(
					sub {
						my ($journey) = @_;

						my $found_dep;
						my $found_arr;
						for my $stop ( $journey->route ) {
							if ( $stop->loc->eva == $dep ) {
								$found_dep = $stop;
							}
							if ( $arr and $stop->loc->eva == $arr ) {
								$found_arr = $stop;
								last;
							}
						}
						if ( not $found_dep ) {
							$self->app->log->debug(
								"Did not find $dep within journey $train_id");
							return;
						}

						if ( $found_dep->rt_dep ) {
							$self->app->in_transit->update_departure_hafas(
								uid     => $uid,
								journey => $journey,
								stop    => $found_dep,
								dep_eva => $dep,
								arr_eva => $arr
							);
							if (    $entry->{backend_id} <= 1
								and $journey->class <= 16
								and $found_dep->rt_dep->epoch > $now->epoch )
							{
								$self->app->add_wagonorder(
									uid          => $uid,
									train_id     => $journey->id,
									is_departure => 1,
									eva          => $dep,
									datetime     => $found_dep->sched_dep,
									train_type   => $journey->type,
									train_no     => $journey->number,
								);
								$self->app->add_stationinfo( $uid, 1,
									$journey->id, $found_dep->loc->eva );
							}
						}

						if ( $found_arr and $found_arr->rt_arr ) {
							$self->app->in_transit->update_arrival_hafas(
								uid     => $uid,
								journey => $journey,
								stop    => $found_arr,
								dep_eva => $dep,
								arr_eva => $arr
							);
							if (    $entry->{backend_id} <= 1
								and $journey->class <= 16
								and $found_arr->rt_arr->epoch - $now->epoch
								< 600 )
							{
								$self->app->add_wagonorder(
									uid        => $uid,
									train_id   => $journey->id,
									is_arrival => 1,
									eva        => $arr,
									datetime   => $found_arr->sched_dep,
									train_type => $journey->type,
									train_no   => $journey->number,
								);
								$self->app->add_stationinfo( $uid, 0,
									$journey->id, $found_dep->loc->eva,
									$found_arr->loc->eva );
							}
						}
					}
				)->catch(
					sub {
						my ($err) = @_;
						if ( $err
							=~ m{svcResL\[0\][.]err is (?:FAIL|PARAMETER)$} )
						{
							# HAFAS do be weird. These are not actionable.
							$self->app->log->debug(
"work($uid) @ HAFAS $entry->{backend_name}: journey: $err"
							);
						}
						else {
							$self->app->log->error(
"work($uid) @ HAFAS $entry->{backend_name}: journey: $err"
							);
						}
					}
				)->wait;

				if (    $arr
					and $entry->{real_arr_ts}
					and $now->epoch - $entry->{real_arr_ts} > 600 )
				{
					$self->app->checkout_p(
						station => $arr,
						force   => 2,
						dep_eva => $dep,
						arr_eva => $arr,
						uid     => $uid
					)->wait;
				}
			};
			if ($@) {
				$errors += 1;
				$self->app->log->error(
					"work($uid) @ HAFAS $entry->{backend_name}: $@");
			}
			next;
		}

		# TODO irgendwo ist hier ne race condition wo ein neuer checkin (in HAFAS) mit IRIS-Daten überschrieben wird.
		# Die ganzen updates brauchen wirklich mal sanity checks mit train id ...

		# Note: IRIS data is not always updated in real-time. Both departure and
		# arrival delays may take several minutes to appear, especially in case
		# of large-scale disturbances. We work around this by continuing to
		# update departure data for up to 15 minutes after departure and
		# delaying automatic checkout by at least 10 minutes.

		eval {
			if ( $now->epoch - $entry->{real_dep_ts} < 900 ) {
				my $status = $self->app->iris->get_departures(
					station    => $dep,
					lookbehind => 30,
					lookahead  => 30
				);
				if ( $status->{errstr} ) {
					die("get_departures($dep): $status->{errstr}\n");
				}

				my ($train) = List::Util::first { $_->train_id eq $train_id }
				@{ $status->{results} };

				if ( not $train ) {
					$self->app->log->debug(
						"could not find train $train_id at $dep\n");
					return;
				}

				$self->app->in_transit->update_departure(
					uid     => $uid,
					train   => $train,
					dep_eva => $dep,
					arr_eva => $arr,
					route   => [ $self->app->iris->route_diff($train) ]
				);

				if ( $train->departure_is_cancelled and $arr ) {
					my $checked_in
					  = $self->app->in_transit->update_departure_cancelled(
						uid     => $uid,
						train   => $train,
						dep_eva => $dep,
						arr_eva => $arr,
					  );

					# depending on the amount of users in transit, some time may
					# have passed between fetching $entry from the database and
					# now. Only check out if the user is still checked into this
					# train.
					if ($checked_in) {

						# check out (adds a cancelled journey and resets journey state
						# to checkin
						$self->app->checkout_p(
							station => $arr,
							force   => 2,
							dep_eva => $dep,
							arr_eva => $arr,
							uid     => $uid
						)->wait;
					}
				}
				else {
					$self->app->add_route_timestamps( $uid, $train, 1 );
					$self->app->add_wagonorder(
						uid          => $uid,
						train_id     => $train->train_id,
						is_departure => 1,
						eva          => $dep,
						datetime     => $train->sched_departure,
						train_type   => $train->type,
						train_no     => $train->train_no
					);
					$self->app->add_stationinfo( $uid, 1, $train->train_id,
						$dep, $arr );
				}
			}
		};
		if ($@) {
			$errors += 1;
			$self->app->log->error("work($uid) @ IRIS: departure: $@");
		}

		eval {
			if (
				$arr
				and ( not $entry->{real_arr_ts}
					or $now->epoch - $entry->{real_arr_ts} < 600 )
			  )
			{
				my $status = $self->app->iris->get_departures(
					station    => $arr,
					lookbehind => 20,
					lookahead  => 220
				);
				if ( $status->{errstr} ) {
					die("get_departures($arr): $status->{errstr}\n");
				}

				# Note that a train may pass the same station several times.
				# Notable example: S41 / S42 ("Ringbahn") both starts and
				# terminates at Berlin Südkreuz
				my ($train) = List::Util::first {
					$_->train_id eq $train_id
					  and $_->sched_arrival
					  and $_->sched_arrival->epoch > $entry->{sched_dep_ts}
				}
				@{ $status->{results} };

				$train //= List::Util::first { $_->train_id eq $train_id }
				@{ $status->{results} };

				if ( not $train ) {

					# If we haven't seen the train yet, its arrival is probably
					# too far in the future. This is not critical.
					return;
				}

				my $checked_in = $self->app->in_transit->update_arrival(
					uid     => $uid,
					train   => $train,
					route   => [ $self->app->iris->route_diff($train) ],
					dep_eva => $dep,
					arr_eva => $arr,
				);

				if ( $checked_in and $train->arrival_is_cancelled ) {

					# check out (adds a cancelled journey and resets journey state
					# to destination selection)
					$self->app->checkout_p(
						station => $arr,
						force   => 0,
						dep_eva => $dep,
						arr_eva => $arr,
						uid     => $uid
					)->wait;
				}
				else {
					$self->app->add_route_timestamps(
						$uid, $train, 0,
						(
							defined $entry->{real_arr_ts}
							  and $now->epoch > $entry->{real_arr_ts}
						) ? 1 : 0
					);
					$self->app->add_wagonorder(
						uid        => $uid,
						train_id   => $train->train_id,
						is_arrival => 1,
						eva        => $arr,
						datetime   => $train->sched_departure,
						train_type => $train->type,
						train_no   => $train->train_no
					);
					$self->app->add_stationinfo( $uid, 0, $train->train_id,
						$dep, $arr );
				}
			}
			elsif ( $entry->{real_arr_ts} ) {
				my ( undef, $error ) = $self->app->checkout_p(
					station => $arr,
					force   => 2,
					dep_eva => $dep,
					arr_eva => $arr,
					uid     => $uid
				)->catch(
					sub {
						my ($error) = @_;
						$self->app->log->error(
							"work($uid) @ IRIS: arrival: $error");
						$errors += 1;
					}
				)->wait;
			}
		};
		if ($@) {
			$self->app->log->error("work($uid) @ IRIS: arrival: $@");
			$errors += 1;
		}

		eval { };
	}

	my $started_at       = $now;
	my $main_finished_at = DateTime->now( time_zone => 'Europe/Berlin' );
	my $worker_duration  = $main_finished_at->epoch - $started_at->epoch;

	if ( $self->app->config->{influxdb}->{url} ) {
		if ( $self->app->mode eq 'development' ) {
			$self->app->log->debug( 'POST '
				  . $self->app->config->{influxdb}->{url}
				  . " worker runtime_seconds=${worker_duration},errors=${errors}"
			);
		}
		else {
			$self->app->ua->post_p( $self->app->config->{influxdb}->{url},
				"worker runtime_seconds=${worker_duration},errors=${errors}" )
			  ->wait;
		}
	}

	if ( not $self->app->config->{traewelling}->{separate_worker} ) {
		$self->app->start('traewelling');
	}

	# add_wagonorder and add_stationinfo assume a permanently running IOLoop
	# and do not allow Mojolicious commands to wait until they have completed.
	# Hence, some add_wagonorder and add_stationinfo calls made here may not
	# complete before the work command exits, and thus have no effect.
	#
	# This is not ideal and will need fixing at some point.  Until then, here
	# is the pragmatic solution for 99% of the associated issues.
	Mojo::Promise->timer(5)->wait;
}

1;

__END__

=head1 SYNOPSIS

  Usage: index.pl work

  Work Work Work.

  Should be called from a cronjob every three minutes or so.