Bulk.pm 30.8 KB
Newer Older
1
# -*- indent-tabs-mode: nil; -*-
2
# vim:ft=perl:et:sw=4
3
4
# $Id$

5
# Sympa - SYsteme de Multi-Postage Automatique
6
7
8
9
10
#
# Copyright (c) 1997, 1998, 1999 Institut Pasteur & Christophe Wolfhugel
# Copyright (c) 1997, 1998, 1999, 2000, 2001, 2002, 2003, 2004, 2005,
# 2006, 2007, 2008, 2009, 2010, 2011 Comite Reseau des Universites
# Copyright (c) 2011, 2012, 2013, 2014 GIP RENATER
11
12
13
14
15
16
17
18
19
20
21
#
# 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.
#
22
23
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
24
25
26
27

package Bulk;

use strict;
28
use warnings;
29
use Encode;
30
use Mail::Address;
31
use Time::HiRes qw(time);
32
use MIME::Parser;
33
use MIME::Base64;
34
use Term::ProgressBar;
35
use URI::Escape;
36
37
use constant MAX => 100_000;

38
39
use tools;
use tt2;
40
use Sympa::Language;
41
42
43
use Log;
use Conf;
use List;
44
use SDM;
45
46

## Database and SQL statement handlers
47
my $sth;
48
49
50
51
52
53
54

# fingerprint of last message stored in bulkspool
my $message_fingerprint;

# create an empty Bulk
#sub new {
#    my $pkg = shift;
sikeda's avatar
sikeda committed
55
#    my $packet = Bulk::next();;
56
57
58
#    bless \$packet, $pkg;
#    return $packet
#}
59
60
61
##
# get next packet to process, order is controled by priority_message, then by
# priority_packet, then by creation date.
62
# Packets marked as being sent with VERP will be treated last.
63
# Next lock the packetb to prevent multiple proccessing of a single packet
64
65

sub next {
66
    Log::do_log('debug', '');
67

68
    # lock next packet
sikeda's avatar
sikeda committed
69
    my $lock = tools::get_lockname();
70
71

    my $order;
72
73
    my $limit_oracle = '';
    my $limit_sybase = '';
74
    my $limit_other  = '';
75
    ## Only the first record found is locked, thanks to the "LIMIT 1" clause
76
77
78
79
80
81
    $order =
        'ORDER BY priority_message_bulkmailer ASC, priority_packet_bulkmailer ASC, reception_date_bulkmailer ASC, verp_bulkmailer ASC';
    if (   $Conf::Conf{'db_type'} eq 'mysql'
        or $Conf::Conf{'db_type'} eq 'Pg'
        or $Conf::Conf{'db_type'} eq 'SQLite') {
        $limit_other = 'LIMIT 1';
82
    } elsif ($Conf::Conf{'db_type'} eq 'Oracle') {
83
        $limit_oracle = 'AND rownum <= 1';
84
    } elsif ($Conf::Conf{'db_type'} eq 'Sybase') {
85
        $limit_sybase = 'TOP 1';
86
87
88
    }

    # Select the most prioritary packet to lock.
89
    unless (
90
91
92
        $sth = SDM::do_prepared_query(
            sprintf(
                q{SELECT %s messagekey_bulkmailer AS messagekey,
93
94
95
96
97
		         packetid_bulkmailer AS packetid
		  FROM bulkmailer_table
		  WHERE lock_bulkmailer IS NULL AND
		        delivery_date_bulkmailer <= ?
		  %s %s %s},
98
99
100
101
102
103
104
105
                $limit_sybase, $limit_oracle, $order, $limit_other
            ),
            int time
        )
        ) {
        Log::do_log('err',
            'Unable to get the most prioritary packet from database');
        return undef;
106
    }
107

108
    my $packet;
109
110
    unless ($packet = $sth->fetchrow_hashref('NAME_lc')) {
        return undef;
111
    }
112

113
    my $sth;
114
    # Lock the packet previously selected.
115
116
117
118
119
120
121
122
123
124
125
    unless (
        $sth = SDM::do_query(
            "UPDATE bulkmailer_table SET lock_bulkmailer=%s WHERE messagekey_bulkmailer='%s' AND packetid_bulkmailer='%s' AND lock_bulkmailer IS NULL",
            SDM::quote($lock),
            $packet->{'messagekey'},
            $packet->{'packetid'}
        )
        ) {
        Log::do_log('err', 'Unable to lock packet %s for message %s',
            $packet->{'packetid'}, $packet->{'messagekey'});
        return undef;
126
    }
127

sikeda's avatar
sikeda committed
128
    if ($sth->rows < 0) {
129
130
131
132
133
134
135
        Log::do_log(
            'err',
            'Unable to lock packet %s for message %s, though the query succeeded',
            $packet->{'packetid'},
            $packet->{'messagekey'}
        );
        return undef;
sikeda's avatar
sikeda committed
136
137
    }
    unless ($sth->rows) {
138
139
        Log::do_log('info', 'Bulk packet is already locked');
        return undef;
140
141
    }

142
    # select the packet that has been locked previously
143
    #FIXME: A column name is recEipients_bulkmailer.
144
145
146
147
148
149
150
151
152
153
154
    unless (
        $sth = SDM::do_query(
            "SELECT messagekey_bulkmailer AS messagekey, messageid_bulkmailer AS messageid, packetid_bulkmailer AS packetid, receipients_bulkmailer AS recipients, returnpath_bulkmailer AS returnpath, listname_bulkmailer AS listname, robot_bulkmailer AS robot, priority_message_bulkmailer AS priority_message, priority_packet_bulkmailer AS priority_packet, verp_bulkmailer AS verp, tracking_bulkmailer AS tracking, merge_bulkmailer as merge, reception_date_bulkmailer AS reception_date, delivery_date_bulkmailer AS delivery_date FROM bulkmailer_table WHERE lock_bulkmailer=%s %s",
            SDM::quote($lock),
            $order
        )
        ) {
        Log::do_log('err',
            'Unable to retrieve informations for packet %s of message %s',
            $packet->{'packetid'}, $packet->{'messagekey'});
        return undef;
155
    }
156

157
    my $result = $sth->fetchrow_hashref('NAME_lc');
158

159
160
161
162
    return $result;

}

163
164
# remove a packet from database by packet id. return undef if packet does not
# exist
165

166
sub remove {
167
    my $messagekey = shift;
168
    my $packetid   = shift;
169

170
    Log::do_log('debug', '(%s, %s)', $messagekey, $packetid);
171

172
173
174
175
176
177
178
179
180
181
    unless (
        $sth = SDM::do_query(
            "DELETE FROM bulkmailer_table WHERE packetid_bulkmailer = %s AND messagekey_bulkmailer = %s",
            SDM::quote($packetid),
            SDM::quote($messagekey)
        )
        ) {
        Log::do_log('err', 'Unable to delete packet %s of message %s',
            $packetid, $messagekey);
        return undef;
182
183
    }
    return $sth;
184
185
186
187
}

sub messageasstring {
    my $messagekey = shift;
188
    Log::do_log('debug', '(%s)', $messagekey);
189
190
191
192
193
194
195
196
197
198
199
200
201

    unless (
        $sth = SDM::do_query(
            "SELECT message_bulkspool AS message FROM bulkspool_table WHERE messagekey_bulkspool = %s",
            SDM::quote($messagekey)
        )
        ) {
        Log::do_log(
            'err',
            'Unable to retrieve message %s text representation from database',
            $messagekey
        );
        return undef;
202
    }
203

204
    my $messageasstring = $sth->fetchrow_hashref('NAME_lc');
205

206
    unless ($messageasstring) {
207
208
        Log::do_log('err', 'Could not fetch message %s from spool',
            $messagekey);
209
        return undef;
210
211
    }
    my $msg = MIME::Base64::decode($messageasstring->{'message'});
212
213
214
215
216
    unless ($msg) {
        Log::do_log('err',
            "could not decode message $messagekey extrated from spool (base64)"
        );
        return undef;
217
    }
218
    return $msg;
219
220
}
#################################"
221
# fetch message from bulkspool_table by key
222
223
224
#
sub message_from_spool {
    my $messagekey = shift;
225
    Log::do_log('debug', '(messagekey: %s)', $messagekey);
226
227
228
229
230
231
232
233
234
235
236

    unless (
        $sth = SDM::do_query(
            "SELECT message_bulkspool AS message, messageid_bulkspool AS messageid, dkim_d_bulkspool AS  dkim_d,  dkim_i_bulkspool AS  dkim_i, dkim_privatekey_bulkspool AS dkim_privatekey, dkim_selector_bulkspool AS dkim_selector FROM bulkspool_table WHERE messagekey_bulkspool = %s",
            SDM::quote($messagekey)
        )
        ) {
        Log::do_log('err',
            'Unable to retrieve message %s full data from database',
            $messagekey);
        return undef;
237
    }
238

239
    my $message_from_spool = $sth->fetchrow_hashref('NAME_lc');
240
241
    $sth->finish;

242
243
244
245
246
247
248
249
250
251
    return (
        {   'messageasstring' =>
                MIME::Base64::decode($message_from_spool->{'message'}),
            'messageid'       => $message_from_spool->{'messageid'},
            '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'},
        }
    );
252

253
}
254
255
256
257

############################################################
#  merge_msg                                               #
############################################################
258
#  Merge a message with custom attributes of a user.       #
259
260
#                                                          #
#                                                          #
261
262
#  IN : - MIME::Entity                                     #
#       - $rcpt : a recipient                              #
263
264
265
#       - $bulk : HASH                                     #
#       - $data : HASH with user's data                    #
#  OUT : 1 | undef                                         #
266
267
268
269
#                                                          #
############################################################
sub merge_msg {
    my $entity = shift;
270
271
272
    my $rcpt   = shift;
    my $bulk   = shift;
    my $data   = shift;
273

274
    unless (ref $entity eq 'MIME::Entity') {
275
        Log::do_log('err', 'False entity');
276
        return undef;
277
278
    }

279
    # Initialize parameters at first only once.
280
    $data->{'headers'} ||= {};
281
    my $headers = $entity->head;
282
283
284
285
286
287
288
289
    foreach my $key (
        qw/subject x-originating-ip message-id date x-original-to from to thread-topic content-type/
        ) {
        next unless $headers->count($key);
        my $value = $headers->get($key, 0);
        chomp $value;
        $value =~ s/(?:\r\n|\r|\n)(?=[ \t])//g;    # unfold
        $data->{'headers'}{$key} = $value;
290
    }
291
    $data->{'subject'} = tools::decode_header($headers, 'Subject');
292

293
294
295
296
297
    return _merge_msg($entity, $rcpt, $bulk, $data);
}

sub _merge_msg {
    my $entity = shift;
298
299
300
301
302
303
304
    my $rcpt   = shift;
    my $bulk   = shift;
    my $data   = shift;

    my $enc = $entity->head->mime_encoding;
    # Parts with nonstandard encodings aren't modified.
    if ($enc and $enc !~ /^(?:base64|quoted-printable|[78]bit|binary)$/i) {
305
        return $entity;
306
307
308
    }
    my $eff_type = $entity->effective_type || 'text/plain';
    # Signed or encrypted parts aren't modified.
309
310
    if ($eff_type =~ m{^multipart/(signed|encrypted)$}) {
        return $entity;
311
312
    }

313
    if ($entity->parts) {
314
315
316
317
318
319
        foreach my $part ($entity->parts) {
            unless (_merge_msg($part, $rcpt, $bulk, $data)) {
                Log::do_log('err', 'Failed to merge message part');
                return undef;
            }
        }
320
    } elsif ($eff_type =~ m{^(?:multipart|message)(?:/|\Z)}i) {
321
322
        # multipart or message types without subparts.
        return $entity;
323
    } elsif (MIME::Tools::textual_type($eff_type)) {
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
        my ($charset, $in_cset, $bodyh, $body, $utf8_body);

        $data->{'part'} = {
            description =>
                tools::decode_header($entity, 'Content-Description'),
            disposition =>
                lc($entity->head->mime_attr('Content-Disposition') || ''),
            encoding => $enc,
            type     => $eff_type,
        };

        $bodyh = $entity->bodyhandle;
        # Encoded body or null body won't be modified.
        if (!$bodyh or $bodyh->is_encoded) {
            return $entity;
        }

        $body = $bodyh->as_string;
        unless (defined $body and length $body) {
            return $entity;
        }

        ## Detect charset.  If charset is unknown, detect 7-bit charset.
        $charset = $entity->head->mime_attr('Content-Type.Charset');
        $in_cset = MIME::Charset->new($charset || 'NONE');
        unless ($in_cset->decoder) {
            $in_cset =
                MIME::Charset->new(MIME::Charset::detect_7bit_charset($body)
                    || 'NONE');
        }
        unless ($in_cset->decoder) {
            Log::do_log('err', 'Unknown charset "%s"', $charset);
            return undef;
        }
        $in_cset->encoder($in_cset);    # no charset conversion

        ## Only decodable bodies are allowed.
        eval { $utf8_body = Encode::encode_utf8($in_cset->decode($body, 1)); };
        if ($@) {
            Log::do_log('err', 'Cannot decode by charset "%s"', $charset);
            return undef;
        }

        ## PARSAGE ##

        my $message_output;
        unless (
            merge_data(
                'rcpt'           => $rcpt,
                'messageid'      => $bulk->{'messageid'},
                'listname'       => $bulk->{'listname'},
                'robot'          => $bulk->{'robot'},
                'data'           => $data,
                'body'           => $utf8_body,
                'message_output' => \$message_output,
            )
            ) {
381
            Log::do_log('err', 'Error merging message');
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
            return undef;
        }
        $utf8_body = $message_output;

        ## Data not encodable by original charset will fallback to UTF-8.
        my ($newcharset, $newenc);
        ($body, $newcharset, $newenc) =
            $in_cset->body_encode(Encode::decode_utf8($utf8_body),
            Replacement => 'FALLBACK');
        unless ($newcharset) {    # bug in MIME::Charset?
            Log::do_log('err', 'Can\'t determine output charset');
            return undef;
        } elsif ($newcharset ne $in_cset->as_string) {
            $entity->head->mime_attr('Content-Transfer-Encoding' => $newenc);
            $entity->head->mime_attr('Content-Type.Charset' => $newcharset);

            ## normalize newline to CRLF if transfer-encoding is BASE64.
            $body =~ s/\r\n|\r|\n/\r\n/g
                if $newenc and $newenc eq 'BASE64';
        } else {
            ## normalize newline to CRLF if transfer-encoding is BASE64.
            $body =~ s/\r\n|\r|\n/\r\n/g
                if $enc and uc $enc eq 'BASE64';
        }

        ## Save new body.
        my $io = $bodyh->open('w');
        unless ($io
            and $io->print($body)
            and $io->close) {
            Log::do_log('err', 'Can\'t write in Entity: %s', $!);
            return undef;
        }
        $entity->sync_headers(Length => 'COMPUTE')
            if $entity->head->get('Content-Length');

        return $entity;
419
420
421
    }

    return $entity;
422
423
424
425
426
427
428
429
}

############################################################
#  merge_data                                              #
############################################################
#  This function retrieves the customized data of the      #
#  users then parse the message. It returns the message    #
#  personalized to bulk.pl                                 #
sikeda's avatar
sikeda committed
430
431
432
#  It uses the method tt2::parse_tt2()                      #
#  It uses the method List::get_list_member_no_object()     #
#  It uses the method tools::get_fingerprint()              #
433
#                                                          #
434
# IN : - rcpt : the recipient email                        #
435
#      - listname : the name of the list                   #
436
#      - robot_id : the host                               #
437
438
439
440
441
442
443
#      - data : HASH with many data                        #
#      - body : message with the TT2                       #
#      - message_output : object, IO::Scalar               #
#                                                          #
# OUT : - message_output : customized message              #
#     | undef                                              #
#                                                          #
444
############################################################
445
sub merge_data {
446
447
448
449
450
451
452
453
454
    my %params             = @_;
    my $rcpt               = $params{'rcpt'},
        my $listname       = $params{'listname'},
        my $robot_id       = $params{'robot'},
        my $data           = $params{'data'},
        my $body           = $params{'body'},
        my $message_output = $params{'message_output'},

        my $options;
455
    $options->{'is_not_template'} = 1;
456

457
458
    # get_list_member_no_object() return the user's details with the custom
    # attributes
459
    my $user = List::get_list_member_no_object(
460
461
462
463
        {   'email'  => $rcpt,
            'name'   => $listname,
            'domain' => $robot_id,
        }
464
    );
465

sikeda's avatar
sikeda committed
466
    $user->{'escaped_email'} = URI::Escape::uri_escape($rcpt);
467
    my $language = Sympa::Language->instance;
468
469
    $user->{'friendly_date'} = $language->gettext_strftime("%d %b %Y  %H:%M",
        localtime($user->{'date'}));
470

471
472
    # this method as been removed because some users may forward
    # authentication link
sikeda's avatar
sikeda committed
473
    # $user->{'fingerprint'} = tools::get_fingerprint($rcpt);
474

475
476
    $data->{'user'}     = $user;
    $data->{'robot'}    = $robot_id;
477
    $data->{'listname'} = $listname;
478

479
480
481
    # Parse the TT2 in the message : replace the tags and the parameters by
    # the corresponding values
    unless (tt2::parse_tt2($data, \$body, $message_output, '', $options)) {
482
        Log::do_log('err', 'Unable to parse body: "%s"', \$body);
483
        return undef;
484
485
486
487
488
    }

    return 1;
}

489
490
##
sub store {
491
    my %data = @_;
492

serge.aumont's avatar
   
serge.aumont committed
493
    my $message = $data{'message'};
494
495
496
497
498
499
    my $msg_id  = $message->{'msg'}->head->get('Message-ID');
    chomp $msg_id;
    my $rcpts            = $data{'rcpts'};
    my $from             = $data{'from'};
    my $robot            = $data{'robot'};
    my $listname         = $data{'listname'};
500
    my $priority_message = $data{'priority_message'};
501
502
503
504
505
506
507
508
509
    my $priority_packet  = $data{'priority_packet'};
    my $delivery_date    = $data{'delivery_date'};
    my $verp             = $data{'verp'};
    my $tracking         = $data{'tracking'};
    $tracking = '' unless (($tracking eq 'dsn') || ($tracking eq 'mdn'));
    $verp = 0 unless ($verp);
    my $merge = $data{'merge'};
    $merge = 0 unless ($merge);
    my $dkim        = $data{'dkim'};
510
511
    my $tag_as_last = $data{'tag_as_last'};

512
513
    Log::do_log(
        'debug',
514
        '(<msg>, <rcpts>, from = %s, robot = %s, listname= %s, priority_message = %s, delivery_date= %s, verp = %s, tracking = %s, merge = %s, dkim: d= %s i=%s, last: %s)',
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
        $from,
        $robot,
        $listname,
        $priority_message,
        $delivery_date,
        $verp,
        $tracking,
        $merge,
        $dkim->{'d'},
        $dkim->{'i'},
        $tag_as_last
    );

    $priority_message = Conf::get_robot_conf($robot, 'sympa_priority')
        unless ($priority_message);
    $priority_packet = Conf::get_robot_conf($robot, 'sympa_packet_priority')
        unless ($priority_packet);
532

533
534
    #creation of a MIME entity to extract the real sender of a message
    my $parser = MIME::Parser->new();
535
    $parser->output_to_core(1);
536

serge.aumont's avatar
   
serge.aumont committed
537
538
    my $msg = $message->{'msg'}->as_string;
    if ($message->{'protected'}) {
539
        $msg = $message->{'msg_as_string'};
serge.aumont's avatar
   
serge.aumont committed
540
    }
541
542
    my @sender_hdr =
        Mail::Address->parse($message->{'msg'}->head->get('From'));
543
544
    my $message_sender = $sender_hdr[0]->address;

545
546
    $msg = MIME::Base64::encode($msg);

547
    ##-----------------------------##
548

sikeda's avatar
sikeda committed
549
    my $messagekey = tools::md5_fingerprint($msg);
550

551
    # first store the message in bulk_spool_table
552
553
554
555
556
    # because as soon as packet are created bulk.pl may distribute them
    # Compare the current message finger print to the fingerprint
    # of the last call to store() ($message_fingerprint is a global var)
    # If fingerprint is the same, then the message should not be stored
    # again in bulkspool_table
557

558
559
    my $message_already_on_spool;

560
561
562
    if (    defined $message_fingerprint
        and defined $messagekey
        and $messagekey eq $message_fingerprint) {
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
        $message_already_on_spool = 1;

    } else {

        ## search if this message is already in spool database : mailfile may
        ## perform multiple submission of exactly the same message
        unless (
            $sth = SDM::do_query(
                "SELECT count(*) FROM bulkspool_table WHERE ( messagekey_bulkspool = %s )",
                SDM::quote($messagekey)
            )
            ) {
            Log::do_log('err',
                'Unable to check whether message %s is in spool already',
                $messagekey);
            return undef;
        }

        $message_already_on_spool = $sth->fetchrow;
        $sth->finish();

        # if message is not found in bulkspool_table store it
        if ($message_already_on_spool == 0) {
            my $statement = q{INSERT INTO bulkspool_table
587
588
589
590
591
		  (messagekey_bulkspool, messageid_bulkspool,
		   message_bulkspool, lock_bulkspool,
		   dkim_d_bulkspool, dkim_i_bulkspool,
		   dkim_selector_bulkspool, dkim_privatekey_bulkspool)
		  VALUES (?, ?, ?, 1, ?, ?, ?, ?)};
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
            my $statementtrace = $statement;
            $statementtrace =~ s/\n\s*/ /g;
            $statementtrace =~ s/\?/\%s/g;

            unless (
                SDM::do_prepared_query(
                    $statement,        $messagekey,
                    $msg_id,           $msg,
                    $dkim->{d},        $dkim->{i},
                    $dkim->{selector}, $dkim->{private_key}
                )
                ) {
                Log::do_log(
                    'err',
                    'Unable to add message in bulkspool_table "%s"',
                    sprintf($statementtrace,
                        SDM::quote($messagekey),
                        SDM::quote($msg_id),
                        SDM::quote(substr($msg, 0, 100)),
                        SDM::quote($dkim->{d}),
                        SDM::quote($dkim->{i}),
                        SDM::quote($dkim->{selector}),
                        SDM::quote(substr($dkim->{private_key}, 0, 30)))
                );
                return undef;
            }

            #log in stat_table to make statistics...
            unless ($message_sender =~ /($robot)\@/) {
                # ignore messages sent by robot
                unless ($message_sender =~ /($listname)-request/) {
                    # ignore messages of requests
                    Log::db_stat_log(
                        {   'robot'     => $robot,
                            'list'      => $listname,
                            'operation' => 'send_mail',
                            'parameter' => length($msg),
                            'mail'      => $message_sender,
                            'client'    => '',
                            'daemon'    => 'sympa.pl'
                        }
                    );
                }
            }
            $message_fingerprint = $messagekey;
        }
638
639
    }

640
    my $current_date = int(time);
641

642
    # second : create each recipient packet in bulkmailer_table
643
644
645
    my $type = ref $rcpts;

    unless (ref $rcpts) {
646
647
648
649
        my @tab = ($rcpts);
        my @tabtab;
        push @tabtab, \@tab;
        $rcpts = \@tabtab;
650
    }
651

652
    my $priority_for_packet;
653
    my $already_tagged = 0;
654
655
    # Initialize counter used to check wether we are copying the last packet.
    my $packet_rank = 0;
656
    foreach my $packet (@{$rcpts}) {
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
        $priority_for_packet = $priority_packet;
        if ($tag_as_last && !$already_tagged) {
            $priority_for_packet = $priority_packet + 5;
            $already_tagged      = 1;
        }
        $type = ref $packet;
        my $rcptasstring;
        if (ref $packet eq 'ARRAY') {
            $rcptasstring = join ',', @{$packet};
        } else {
            $rcptasstring = $packet;
        }
        my $packetid = tools::md5_fingerprint($rcptasstring);
        my $packet_already_exist;
        if (ref($listname) =~ /List/i) {
            $listname = $listname->{'name'};
        }
        if ($message_already_on_spool) {
            ## search if this packet is already in spool database : mailfile
            ## may perform multiple submission of exactly the same message
            unless (
                $sth = SDM::do_query(
                    "SELECT count(*) FROM bulkmailer_table WHERE ( messagekey_bulkmailer = %s AND  packetid_bulkmailer = %s)",
                    SDM::quote($messagekey),
                    SDM::quote($packetid)
                )
                ) {
                Log::do_log(
                    'err',
                    'Unable to check presence of packet %s of message %s in database',
                    $packetid,
                    $messagekey
                );
                return undef;
            }
            $packet_already_exist = $sth->fetchrow;
            $sth->finish();
        }

        if ($packet_already_exist) {
            Log::do_log('err',
                'Duplicate message not stored in bulmailer_table');

        } else {
            unless (
                SDM::do_query(
                    "INSERT INTO bulkmailer_table (messagekey_bulkmailer,messageid_bulkmailer,packetid_bulkmailer,receipients_bulkmailer,returnpath_bulkmailer,robot_bulkmailer,listname_bulkmailer, verp_bulkmailer, tracking_bulkmailer, merge_bulkmailer, priority_message_bulkmailer, priority_packet_bulkmailer, reception_date_bulkmailer, delivery_date_bulkmailer) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)",
                    SDM::quote($messagekey),
                    SDM::quote($msg_id),
                    SDM::quote($packetid),
                    SDM::quote($rcptasstring),
                    SDM::quote($from),
                    SDM::quote($robot),
                    SDM::quote($listname),
                    $verp,
                    SDM::quote($tracking),
                    $merge,
                    $priority_message,
                    $priority_for_packet,
                    $current_date,
                    $delivery_date
                )
                ) {
                Log::do_log(
                    'err',
                    'Unable to add packet %s of message %s to database spool',
                    $packetid,
                    $msg_id
                );
                return undef;
            }
        }
        $packet_rank++;
730
    }
731
732
733
734
735
736
737
738
739
740
741
    # last : unlock message in bulkspool_table so it is now possible to remove
    # this message if no packet has a ref on it
    unless (
        SDM::do_query(
            "UPDATE bulkspool_table SET lock_bulkspool='0' WHERE messagekey_bulkspool = %s",
            SDM::quote($messagekey)
        )
        ) {
        Log::do_log('err', 'Unable to unlock packet %s in bulkmailer_table',
            $messagekey);
        return undef;
742
    }
743
    return 1;
744
745
746
747
}

## remove file that are not referenced by any packet
sub purge_bulkspool {
748
    Log::do_log('debug', '');
749

750
751
752
753
754
755
756
757
    unless (
        $sth = SDM::do_query(
            "SELECT messagekey_bulkspool AS messagekey FROM bulkspool_table LEFT JOIN bulkmailer_table ON messagekey_bulkspool = messagekey_bulkmailer WHERE messagekey_bulkmailer IS NULL AND lock_bulkspool = 0"
        )
        ) {
        Log::do_log('err',
            'Unable to check messages unreferenced by packets in database');
        return undef;
758
759
760
    }

    my $count = 0;
761
762
763
764
765
766
767
768
769
770
    while (my $key = $sth->fetchrow_hashref('NAME_lc')) {
        if (Bulk::remove_bulkspool_message('bulkspool', $key->{'messagekey'}))
        {
            $count++;
        } else {
            Log::do_log('err',
                'Unable to remove message (key = %s) from bulkspool_table',
                $key->{'messagekey'});
        }
    }
771
772
773
774
775
    $sth->finish;
    return $count;
}

sub remove_bulkspool_message {
776
    my $spool      = shift;
777
778
    my $messagekey = shift;

779
780
    my $table = $spool . '_table';
    my $key   = 'messagekey_' . $spool;
781

782
783
784
785
786
787
788
789
790
    unless (
        SDM::do_query(
            "DELETE FROM %s WHERE %s = %s", $table,
            $key,                           SDM::quote($messagekey)
        )
        ) {
        Log::do_log('err', 'Unable to delete %s %s from %s',
            $table, $key, $messagekey);
        return undef;
791
792
793
794
795
    }

    return 1;
}

796
# test the maximal message size the database will accept
797
798
799
800
801
802
803
804
805
806
807
808
sub store_test {
    my $value_test       = shift;
    my $divider          = 100;
    my $steps            = 50;
    my $maxtest          = $value_test / $divider;
    my $size_increment   = $divider * $maxtest / $steps;
    my $barmax           = $size_increment * $steps * ($steps + 1) / 2;
    my $even_part        = $barmax / $steps;
    my $rcpts            = 'nobody@cru.fr';
    my $from             = 'sympa-test@notadomain';
    my $robot            = 'notarobot';
    my $listname         = 'notalist';
809
    my $priority_message = 9;
810
811
812
813
814
815
    my $delivery_date    = time;
    my $verp             = 'on';
    my $merge            = 1;

    Log::do_log(
        'debug',
816
        '(<msg>, <rcpts>, from = %s, robot = %s, listname= %s, priority_message = %s, delivery_date= %s, verp = %s, merge = %s)',
817
818
819
820
821
822
823
824
        $from,
        $robot,
        $listname,
        $priority_message,
        $delivery_date,
        $verp,
        $merge
    );
825

826
827
    print "maxtest: $maxtest\n";
    print "barmax: $barmax\n";
828
829
830
831
832
833
    my $progress = Term::ProgressBar->new(
        {   name  => 'Total size transfered',
            count => $barmax,
            ETA   => 'linear',
        }
    );
834
    $priority_message = 9;
835

sikeda's avatar
sikeda committed
836
    my $messagekey = tools::md5_fingerprint(time());
837
    my $msg;
838
839
    $progress->max_update_rate(1);
    my $next_update = 0;
840
    my $total       = 0;
841
842

    my $result = 0;
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879

    for (my $z = 1; $z <= $steps; $z++) {
        $msg = MIME::Base64::decode($msg);
        for (my $i = 1; $i <= 1024 * $size_increment; $i++) {
            $msg .= 'a';
        }
        $msg = MIME::Base64::encode($msg);
        my $time = time();
        $progress->message(
            sprintf
                "Test storing and removing of a %5d kB message (step %s out of %s)",
            $z * $size_increment,
            $z, $steps
        );
        #
        unless (
            SDM::do_query(
                "INSERT INTO bulkspool_table (messagekey_bulkspool, message_bulkspool, lock_bulkspool) VALUES (%s, %s, '1')",
                SDM::quote($messagekey),
                SDM::quote($msg)
            )
            ) {
            return (($z - 1) * $size_increment);
        }
        unless (Bulk::remove_bulkspool_message('bulkspool', $messagekey)) {
            Log::do_log(
                'err',
                'Unable to remove test message (key = %s) from bulkspool_table',
                $messagekey
            );
        }
        $total += $z * $size_increment;
        $progress->message(sprintf ".........[OK. Done in %.2f sec]",
            time() - $time);
        $next_update = $progress->update($total + $even_part)
            if $total > $next_update && $total < $barmax;
        $result = $z * $size_increment;
880
    }
881
    $progress->update($barmax)
882
        if $barmax >= $next_update;
883
    return $result;
884
885
}

886
887
## Return the number of remaining packets in the bulkmailer table.
sub get_remaining_packets_count {
888
    Log::do_log('debug3', '');
889
890

    my $m_count = 0;
891

892
893
894
895
896
897
898
899
    unless (
        $sth = SDM::do_prepared_query(
            "SELECT COUNT(*) FROM bulkmailer_table WHERE lock_bulkmailer IS NULL"
        )
        ) {
        Log::do_log('err',
            'Unable to count remaining packets in bulkmailer_table');
        return undef;
900
901
902
    }

    my @result = $sth->fetchrow_array();
903

904
905
906
    return $result[0];
}

907
908
## Returns 1 if the number of remaining packets in the bulkmailer table
## exceeds
909
910
## the value of the 'bulk_fork_threshold' config parameter.
sub there_is_too_much_remaining_packets {
911
    Log::do_log('debug3', '');
sikeda's avatar
sikeda committed
912
    my $remaining_packets = get_remaining_packets_count();
913
914
915
916
917
    if ($remaining_packets > Conf::get_robot_conf('*', 'bulk_fork_threshold'))
    {
        return $remaining_packets;
    } else {
        return 0;
918
919
    }
}
920

921
1;