#!/usr/bin/perl

# bulk.pl - This script does the mail sending for Sympa
# RCS Identication ; $Revision: 4985 $ ; $Date: 2008-05-02 12:06:27 +0200 (Fri, 02 May 2008) $ 
#
# Sympa - SYsteme de Multi-Postage Automatique
# Copyright (c) 1997, 1998, 1999, 2000, 2001 Comite Reseau des Universites
# Copyright (c) 1997,1998, 1999 Institut Pasteur & Christophe Wolfhugel
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.

## Options :  F         -> do not detach TTY
##         :  d		-> debug -d is equiv to -dF
## Now, it is impossible to use -dF but you have to write it -d -F

=pod 

=head1 NAME 

I<bulk.pl> - Daemon for submitting to smtp engine bulkmailer_table content.

=head1 DESCRIPTION 

This script must be run along with sympa. It regularly checks the bulkmailer_table content and submit the messages it finds in it to the sendmail engine. Several deamon should be used on deferent server for hugue traffic.

=cut 

use lib '/usr/local/libexec/sympa';

use strict;

use Conf;
use Log;
use Commands;
use Getopt::Long;

use mail;
use Bulk;
use List;
use tools;
use Sympa::Constants;

my $daemon_name = &Log::set_daemon($0);
my $date_of_last_activity = time();
local $main::daemon_usage = 'DAEMON_MASTER'; ## Default is to launch bulk as master daemon.

## Check options
##  --debug : sets the debug mode
##  --foreground : prevents the script from beeing daemonized
##  --mail : logs every sendmail calls
my %options;
unless (&GetOptions(\%main::options, 'debug|d', 'foreground|F','mail|m')) {
    &fatal_err("Unknown options.");
}

if ($main::options{'debug'}) {
    $main::options{'log_level'} = 2 unless ($main::options{'log_level'});
}

$main::options{'foreground'} = 1 if ($main::options{'debug'});
$main::options{'log_to_stderr'} = 1 if ($main::options{'debug'} || $main::options{'foreground'});

my $sympa_conf_file = Sympa::Constants::CONFIG;

# Load sympa.conf
unless (Conf::load($sympa_conf_file)) {
    &fatal_err("Unable to load sympa configuration, file $sympa_conf_file has errors.");
}

## Check database connectivity
unless (&List::check_db_connect()) {
    &fatal_err('Database %s defined in sympa.conf has not the right structure or is unreachable.', $Conf::Conf{'db_name'});
}

do_openlog($Conf::Conf{'syslog'}, $Conf::Conf{'log_socket_type'}, 'bulk');

# setting log_level using conf unless it is set by calling option
if ($main::options{'log_level'}) {
    &Log::set_log_level($main::options{'log_level'});
    do_log('info', "Configuration file read, log level set using options : $main::options{'log_level'}"); 
}else{
    &Log::set_log_level($Conf::Conf{'log_level'});
    do_log('info', "Configuration file read, default log level $Conf::Conf{'log_level'}"); 
}

## Set the process as main bulk daemon by default.
my $is_main_bulk = 0;

## Put ourselves in background if not in debug mode. 
unless ($main::options{'debug'} || $main::options{'foreground'}) {
   open(STDERR, ">> /dev/null");
   open(STDOUT, ">> /dev/null");
   if (open(TTY, "/dev/tty")) {
#      ioctl(TTY, &TIOCNOTTY, 0);
      close(TTY);
   }
   setpgrp(0, 0);
   if ((my $child_pid = fork) != 0) {
       do_log('info',"Starting bulk master daemon, pid %s",$child_pid);
       exit(0);
   }
}
do_openlog($Conf::Conf{'syslog'}, $Conf::Conf{'log_socket_type'}, 'bulk');
## If process is running in foreground, don't write STDERR to a dedicated file
my $options;
$options->{'stderr_to_tty'} = 1 if ($main::options{'foreground'});
$options->{'multiple_process'} = 0;

# Saves the pid number
&tools::write_pid($Conf::Conf{'pidfile_bulk'}, $$, $options);

## Set the UserID & GroupID for the process
$( = $) = (getgrnam(Sympa::Constants::GROUP))[2];
$< = $> = (getpwnam(Sympa::Constants::USER))[2];

## Required on FreeBSD to change ALL IDs(effective UID + real UID + saved UID)
&POSIX::setuid((getpwnam(Sympa::Constants::USER))[2]);
&POSIX::setgid((getgrnam(Sympa::Constants::GROUP))[2]);

## Check if the UID has correctly been set (usefull on OS X)
unless (($( == (getgrnam(Sympa::Constants::GROUP))[2]) && ($< == (getpwnam(Sympa::Constants::USER))[2])) {
    &fatal_err("Failed to change process userID and groupID. Note that on some OS Perl scripts can't change their real UID. In such circumstances Sympa should be run via SUDO.");
}

## Sets the UMASK
umask(oct($Conf::Conf{'umask'}));

## Change to list root
unless (chdir($Conf::Conf{'home'})) {
    &do_log('err','unable to change directory');
    exit (-1);
}

my $pinfo = &List::_apply_defaults();

do_log('notice', "bulk.pl %s Started", Sympa::Constants::VERSION);


## Catch SIGTERM, in order to exit cleanly, whenever possible.
$SIG{'TERM'} = 'sigterm';
my $end = 0;

my $opensmtp = 0 ;
my $fh = 'fh0000000000';	## File handle for the stream.

my $messagekey;       # the key of the current message in the message_table   
my $messageasstring_init;  # the current message as a string
my $messageasstring;  # the transformed message.
my $last_check_date = 0;

my $timeout = $Conf::Conf{'bulk_wait_to_fork'};
my $last_check_date = time();

$options->{'multiple_process'} = 1;
$Conf::Conf{'maxsmtp'} = int($Conf::Conf{'maxsmtp'}/$Conf::Conf{'bulk_max_count'});

while (!$end) {
    &List::init_list_cache();
    my $bulk;
    my $message_from_spool ;
	
    unless ($main::options{'foreground'}) {
        ## 
	## Create child bulks if too much packets are waiting to be sent in the bulk_mailer table.
	## Only the main bulk process (DAEMON_MASTER) can create child processes
 	## Check if we need to run new child processes every 'bulk_wait_to_fork' (sympa.conf parameter) seconds 
	if (($main::daemon_usage eq 'DAEMON_MASTER') && (time() - $last_check_date > $timeout)){
	    
	    ## Start new processes if at least 'bulk_fork_threshold' (sympa.conf parameter) in the bulkmailer_table table
	    ## Only if no bulk child process is still running
	    my $number_of_pids = &tools::get_number_of_pids($Conf::Conf{'pidfile_bulk'});
	    unless (defined($number_of_pids)){
		&tools::write_pid($Conf::Conf{'pidfile_bulk'}, $$, $options);
		$number_of_pids = &tools::get_number_of_pids($Conf::Conf{'pidfile_bulk'});
	    }
	    if((my $r_packets = &Bulk::there_is_too_much_remaining_packets()) && defined($number_of_pids) && ($number_of_pids <= 1) ){
		
		## disconnect from database before fork
		## to prevent DB handlers to be shared by different processes
		my $dbh = &List::db_get_handler();
		$dbh->disconnect; # when loading conf in database disconnect because of sharing dbh may crash bulk.pl
		
		if($Conf::Conf{'bulk_max_count'} > 1) {
		    &do_log('info','Workload increased: %s packets to process. Creating %s child bulks to increase sending rate.', $r_packets, $Conf::Conf{'bulk_max_count'}-1);
		    for my $process_count(1..$Conf::Conf{'bulk_max_count'}-1){
			if ((my $child_pid = fork) != 0) {
			    do_log('info', "Starting bulk child daemon, pid %s", $child_pid);
			    # Saves the pid number
			    &tools::write_pid($Conf::Conf{'pidfile_bulk'}, $child_pid, $options);
			    sleep 1; 
			}else{
			    ## We're in a child bulk process
			    $date_of_last_activity = time();
			    $main::daemon_usage = 'DAEMON_SLAVE'; # automatic lists creation
			    do_openlog($Conf::Conf{'syslog'}, $Conf::Conf{'log_socket_type'}, 'bulk');
			    do_log('info', "Bulk slave daemon started with pid %s", $$);
			    last;
			}
		    }
		}
	    }
	    $last_check_date = time();
	}
    }
    ## If a child bulk process is running for long enough, stop it (if the number of remaining packets to send is reasonnable).
    if (($main::daemon_usage eq 'DAEMON_SLAVE') && 
		(time() - $date_of_last_activity > $Conf::Conf{'bulk_lazytime'}) && 
		!(my $r_packets = &Bulk::there_is_too_much_remaining_packets())){
		&do_log('info', "Process %s didn't send any message since %s seconds, exiting.", $$, $Conf::Conf{'bulk_lazytime'});
		last;
    }
	
	## Go through the bulk_mailer table and process messages
    if ($bulk = Bulk::next()) {
	if ($bulk->{'messagekey'} ne $messagekey) {
	    # current packet is no related to the same message as the previous packet
            # so it is needed to fetch the new message from message_table 
	    $message_from_spool = &Bulk::message_from_spool($bulk->{'messagekey'});

	    $messageasstring_init = $message_from_spool->{'messageasstring'};
	    
	    unless ( $messageasstring_init ) {
		&do_log('err',"internal error : current packet (messagekey= %s) contain a ref to a null message",$bulk->{'messagekey'});
		unless (&Bulk::remove($bulk->{'messagekey'},$bulk->{'packetid'})) {
		    &do_log('err',"failed to remove processed packet '%s', messagekey '%s'", $bulk->{'messagekey'},$bulk->{'packetid'});
		}
	    }
	    $messagekey = $bulk->{'messagekey'};
	}

	my $data; #HASH which will contain the attributes of the subscriber
	
	# Initialization of the HASH : $data. It will be used by parse_tt2 to personalized messages.
	$data->{'listname'} = $bulk->{'listname'};
	$data->{'robot'} = $bulk->{'robot'};
	$data->{'to'} = $bulk->{'receipients'};
	$data->{'wwsympa_url'} = $Conf{'wwsympa_url'};
	
	my $rcpt; # It is the email of a subscriber, use it in the foreach
	my @rcpts = split /,/,$bulk->{'receipients'}; # Contain all the subscribers
	## Use an intermediate handler to encode to filesystem_encoding
	my $user;
	
	# message transformation must be done in the folowing order 
        #  -1 headers modifications (done in sympa.pl)
        #  -2 merge
        #  -3 smime sign
        #  -4 smime encrypt
	#  -5 remove existing signature if altered 
        #  -6 dkim sign
        # 
 
	if (($bulk->{'verp'})||($bulk->{'merge'})){ # message needs personalization
	    
	    foreach $rcpt (@rcpts) {
		my $return_path = $bulk->{'returnpath'};
		if ($bulk->{'verp'}) {
		    my $return_path = $rcpt;
		    $return_path =~ s/\@/\=\=a\=\=/; 
		    $return_path = "$Conf::Conf{'bounce_email_prefix'}+$return_path\=\=$bulk->{'listname'}\@$bulk->{'robot'}"; 
		}
		if ($bulk->{'merge'}) { 
		    
		    ### Parse an MIME::Entity message
		    my $parser = new MIME::Parser;
		    $parser->output_to_core(1);
		    my $entity = $parser->parse_data($messageasstring_init); ## Retrouver la méthode qui correspond
		    
		    unless(&Bulk::merge_msg ($entity, $rcpt, $bulk, $data)){
			&do_log('err', 'Erreur d appel &Bulk::merge_msg');
		    }

		    $messageasstring = $entity->as_string;		    
		}else{
		    $messageasstring = $messageasstring_init;
		}
		if (&Conf::get_robot_conf($bulk->{'robot'}, 'dkim_feature') eq 'on'){
		    $messageasstring = &tools::remove_invalid_dkim_signature($messageasstring );
		}
		if ($message_from_spool->{'dkim_d'}){ # apply dkim signature AFTER any other message transformation.
		    $messageasstring = &tools::dkim_sign($messageasstring,{'dkim_d'=>           $message_from_spool->{'dkim_d'},
									   'dkim_i'=>           $message_from_spool->{'dkim_i'},
									   'dkim_selector'=>    $message_from_spool->{'dkim_selector'},
									   'dkim_privatekey' => $message_from_spool->{'dkim_privatekey'},
									   'dkim_header_list' =>$message_from_spool->{'dkim_header_list'} } );
		}		
		*SMTP = &mail::smtpto($return_path, \$rcpt, $bulk->{'robot'});
		# Message with customized data
		print SMTP $messageasstring;
		close SMTP;
	    }
	}else{ # message dont needs personalization, they can be sent by packet
	    if (&Conf::get_robot_conf($bulk->{'robot'}, 'dkim_feature') eq 'on'){
		$messageasstring_init = &tools::remove_invalid_dkim_signature($messageasstring_init );
	    }
	    # Initial message
	    if ($message_from_spool->{'dkim_d'}){
		$messageasstring_init = &tools::dkim_sign($messageasstring_init,{'dkim_d'=>           $message_from_spool->{'dkim_d'},
										 'dkim_i'=>           $message_from_spool->{'dkim_i'},
										 'dkim_selector'=>    $message_from_spool->{'dkim_selector'},
										 'dkim_privatekey' => $message_from_spool->{'dkim_privatekey'},
										 'dkim_header_list' =>$message_from_spool->{'dkim_header_list'} } );
	    }
	    *SMTP = &mail::smtpto($bulk->{'returnpath'}, \@rcpts, $bulk->{'robot'});		
	    print SMTP $messageasstring_init;
	    close SMTP;
	}
	
	## Remove packet once it has been processed
	unless (&Bulk::remove($bulk->{'messagekey'},$bulk->{'packetid'})) {
	    &do_log('err',"failed to remove processed packet '%s', messagekey '%s'", $bulk->{'messagekey'},$bulk->{'packetid'});
	}

	
	if($bulk->{'priority_packet'} == $Conf::Conf{'sympa_packet_priority'} + 5){
	    &do_log('notice','Done sending message %s to list %s@%s (priority %s) in %s seconds since scheduled expedition date.', 
		    $bulk->{'messageid'}, 
		    $bulk->{'listname'}, 
		    $bulk->{'robot'}, 
		    $bulk->{'priority_message'}, 
		    time() - $bulk->{'delivery_date'});
	}
	$date_of_last_activity = time();
    }else{
	## Sleep for a while if bulk_mailer DB table is empty
	sleep $Conf::Conf{'bulk_sleep'}; 
    }
    &mail::reaper;
}
do_log('notice', 'bulk.pl exited normally due to signal');
&tools::remove_pid($Conf::Conf{'pidfile_bulk'}, $$, $options);

exit(0);


## When we catch SIGTERM, just change the value of the loop
## variable.
sub sigterm {
    &do_log('notice', 'signal TERM received, still processing current task');
    $end = 1;
}

