lib/PublicInbox/LeiToMail.pm

   1 # Copyright (C) 2020-2021 all contributors <meta@public-inbox.org>
   2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
   3
   4 # Writes PublicInbox::Eml objects atomically to a mbox variant or Maildir
   5 package PublicInbox::LeiToMail;
   6 use strict;
   7 use v5.10.1;
   8 use parent qw(PublicInbox::IPC);
   9 use PublicInbox::Eml;
  10 use PublicInbox::Lock;
  11 use PublicInbox::ProcessPipe;
  12 use PublicInbox::Spawn qw(which spawn popen_rd);
  13 use PublicInbox::LeiDedupe;
  14 use PublicInbox::OnDestroy;
  15 use Symbol qw(gensym);
  16 use IO::Handle; # ->autoflush
  17 use Fcntl qw(SEEK_SET SEEK_END O_CREAT O_EXCL O_WRONLY);
  18 use Errno qw(EEXIST ESPIPE ENOENT);
  19 use PublicInbox::Git;
  20
  21 my %kw2char = ( # Maildir characters
  22         draft => 'D',
  23         flagged => 'F',
  24         answered => 'R',
  25         seen => 'S'
  26 );
  27
  28 my %kw2status = (
  29         flagged => [ 'X-Status' => 'F' ],
  30         answered => [ 'X-Status' => 'A' ],
  31         seen => [ 'Status' => 'R' ],
  32         draft => [ 'X-Status' => 'T' ],
  33 );
  34
  35 sub _mbox_hdr_buf ($$$) {
  36         my ($eml, $type, $smsg) = @_;
  37         $eml->header_set($_) for (qw(Lines Bytes Content-Length));
  38
  39         # Messages are always 'O' (non-\Recent in IMAP), it saves
  40         # MUAs the trouble of rewriting the mbox if no other
  41         # changes are made
  42         my %hdr = (Status => [ 'O' ]); # set Status, X-Status
  43         for my $k (@{$smsg->{kw} // []}) {
  44                 if (my $ent = $kw2status{$k}) {
  45                         push @{$hdr{$ent->[0]}}, $ent->[1];
  46                 } else { # X-Label?
  47                         warn "TODO: keyword `$k' not supported for mbox\n";
  48                 }
  49         }
  50         while (my ($name, $chars) = each %hdr) {
  51                 $eml->header_set($name, join('', sort @$chars));
  52         }
  53         my $buf = delete $eml->{hdr};
  54
  55         # fixup old bug from import (pre-a0c07cba0e5d8b6a)
  56         $$buf =~ s/\A[\r\n]*From [^\r\n]*\r?\n//s;
  57         my $ident = $smsg->{blob} // 'lei';
  58         if (defined(my $pct = $smsg->{pct})) { $ident .= "=$pct" }
  59
  60         substr($$buf, 0, 0, # prepend From line
  61                 "From $ident\@$type Thu Jan  1 00:00:00 1970$eml->{crlf}");
  62         $buf;
  63 }
  64
  65 sub atomic_append { # for on-disk destinations (O_APPEND, or O_EXCL)
  66         my ($fh, $buf) = @_;
  67         defined(my $w = syswrite($fh, $$buf)) or die "write: $!";
  68         $w == length($$buf) or die "short write: $w != ".length($$buf);
  69 }
  70
  71 sub _print_full {
  72         my ($fh, $buf) = @_;
  73         print $fh $$buf or die "print: $!";
  74 }
  75
  76 sub eml2mboxrd ($;$) {
  77         my ($eml, $smsg) = @_;
  78         my $buf = _mbox_hdr_buf($eml, 'mboxrd', $smsg);
  79         if (my $bdy = delete $eml->{bdy}) {
  80                 $$bdy =~ s/^(>*From )/>$1/gm;
  81                 $$buf .= $eml->{crlf};
  82                 substr($$bdy, 0, 0, $$buf); # prepend header
  83                 $buf = $bdy;
  84         }
  85         $$buf .= $eml->{crlf};
  86         $buf;
  87 }
  88
  89 sub eml2mboxo {
  90         my ($eml, $smsg) = @_;
  91         my $buf = _mbox_hdr_buf($eml, 'mboxo', $smsg);
  92         if (my $bdy = delete $eml->{bdy}) {
  93                 $$bdy =~ s/^From />From /gm;
  94                 $$buf .= $eml->{crlf};
  95                 substr($$bdy, 0, 0, $$buf); # prepend header
  96                 $buf = $bdy;
  97         }
  98         $$buf .= $eml->{crlf};
  99         $buf;
 100 }
 101
 102 sub _mboxcl_common ($$$) {
 103         my ($buf, $bdy, $crlf) = @_;
 104         # add Lines: so mutt won't have to add it on MUA close
 105         my $lines = $$bdy =~ tr!\n!\n!;
 106         $$buf .= 'Content-Length: '.length($$bdy).$crlf.
 107                 'Lines: '.$lines.$crlf.$crlf;
 108         substr($$bdy, 0, 0, $$buf); # prepend header
 109         $_[0] = $bdy;
 110 }
 111
 112 # mboxcl still escapes "From " lines
 113 sub eml2mboxcl {
 114         my ($eml, $smsg) = @_;
 115         my $buf = _mbox_hdr_buf($eml, 'mboxcl', $smsg);
 116         my $crlf = $eml->{crlf};
 117         if (my $bdy = delete $eml->{bdy}) {
 118                 $$bdy =~ s/^From />From /gm;
 119                 _mboxcl_common($buf, $bdy, $crlf);
 120         }
 121         $$buf .= $crlf;
 122         $buf;
 123 }
 124
 125 # mboxcl2 has no "From " escaping
 126 sub eml2mboxcl2 {
 127         my ($eml, $smsg) = @_;
 128         my $buf = _mbox_hdr_buf($eml, 'mboxcl2', $smsg);
 129         my $crlf = $eml->{crlf};
 130         if (my $bdy = delete $eml->{bdy}) {
 131                 _mboxcl_common($buf, $bdy, $crlf);
 132         }
 133         $$buf .= $crlf;
 134         $buf;
 135 }
 136
 137 sub git_to_mail { # git->cat_async callback
 138         my ($bref, $oid, $type, $size, $arg) = @_;
 139         if ($type ne 'blob') {
 140                 if ($type eq 'missing') {
 141                         warn "missing $oid\n";
 142                 } else {
 143                         warn "unexpected type=$type for $oid\n";
 144                 }
 145         }
 146         my ($write_cb, $smsg) = @$arg;
 147         if ($smsg->{blob} ne $oid) {
 148                 die "BUG: expected=$smsg->{blob} got=$oid";
 149         }
 150         $write_cb->($bref, $smsg) if $size > 0;
 151 }
 152
 153 sub reap_compress { # dwaitpid callback
 154         my ($lei, $pid) = @_;
 155         my $cmd = delete $lei->{"pid.$pid"};
 156         return if $? == 0;
 157         $lei->fail("@$cmd failed", $? >> 8);
 158 }
 159
 160 # all of these support -c for stdout and -d for decompression,
 161 # mutt is commonly distributed with hooks for gz, bz2 and xz, at least
 162 # { foo => '' } means "--foo" is passed to the command-line,
 163 # otherwise { foo => '--bar' } passes "--bar"
 164 our %zsfx2cmd = (
 165         gz => [ qw(GZIP pigz gzip), { rsyncable => '', threads => '-p' } ],
 166         bz2 => [ 'bzip2', {} ],
 167         xz => [ 'xz', { threads => '-T' } ],
 168         # XXX does anybody care for these?  I prefer zstd on entire FSes,
 169         # so it's probably not necessary on a per-file basis
 170         # zst => [ 'zstd', { -default => [ qw(-q) ], # it's noisy by default
 171         #       rsyncable => '', threads => '-T' } ],
 172         # zz => [ 'pigz', { -default => [ '--zlib' ],
 173         #       rsyncable => '', threads => '-p' }],
 174         # lzo => [ 'lzop', {} ],
 175         # lzma => [ 'lzma', {} ],
 176 );
 177
 178 sub zsfx2cmd ($$$) {
 179         my ($zsfx, $decompress, $lei) = @_;
 180         my $x = $zsfx2cmd{$zsfx} // die "no support for suffix=.$zsfx";
 181         my @info = @$x;
 182         my $cmd_opt = pop @info;
 183         my @cmd = (undef, $decompress ? qw(-dc) : qw(-c));
 184         for my $exe (@info) {
 185                 # I think respecting client's ENV{GZIP} is OK, not sure
 186                 # about ENV overrides for other, less-common compressors
 187                 if ($exe eq uc($exe)) {
 188                         $exe = $lei->{env}->{$exe} or next;
 189                 }
 190                 $cmd[0] = which($exe) and last;
 191         }
 192         $cmd[0] // die join(' or ', @info)." missing for .$zsfx";
 193         # push @cmd, @{$cmd_opt->{-default}} if $cmd_opt->{-default};
 194         for my $bool (qw(rsyncable)) {
 195                 my $switch = $cmd_opt->{rsyncable} // next;
 196                 push @cmd, '--'.($switch || $bool);
 197         }
 198         for my $key (qw(threads)) { # support compression level?
 199                 my $switch = $cmd_opt->{$key} // next;
 200                 my $val = $lei->{opt}->{$key} // next;
 201                 push @cmd, $switch, $val;
 202         }
 203         \@cmd;
 204 }
 205
 206 sub _post_augment_mbox { # open a compressor process
 207         my ($self, $lei, $zpipe) = @_;
 208         my $zsfx = $self->{zsfx} or return;
 209         my $cmd = zsfx2cmd($zsfx, undef, $lei);
 210         my ($r, $w) = splice(@$zpipe, 0, 2);
 211         my $rdr = { 0 => $r, 1 => $lei->{1}, 2 => $lei->{2} };
 212         my $pid = spawn($cmd, $lei->{env}, $rdr);
 213         my $pp = gensym;
 214         my $dup = bless { "pid.$pid" => $cmd }, ref($lei);
 215         $dup->{$_} = $lei->{$_} for qw(2 sock);
 216         tie *$pp, 'PublicInbox::ProcessPipe', $pid, $w, \&reap_compress, $dup;
 217         $lei->{1} = $pp;
 218         die 'BUG: unexpected {ovv}->{lock_path}' if $lei->{ovv}->{lock_path};
 219         $lei->{ovv}->ovv_out_lk_init;
 220 }
 221
 222 sub decompress_src ($$$) {
 223         my ($in, $zsfx, $lei) = @_;
 224         my $cmd = zsfx2cmd($zsfx, 1, $lei);
 225         popen_rd($cmd, $lei->{env}, { 0 => $in, 2 => $lei->{2} });
 226 }
 227
 228 sub dup_src ($) {
 229         my ($in) = @_;
 230         open my $dup, '+>>&', $in or die "dup: $!";
 231         $dup;
 232 }
 233
 234 # --augment existing output destination, with deduplication
 235 sub _augment { # MboxReader eml_cb
 236         my ($eml, $lei) = @_;
 237         # ignore return value, just populate the skv
 238         $lei->{dedupe}->is_dup($eml);
 239 }
 240
 241 sub _mbox_write_cb ($$) {
 242         my ($self, $lei) = @_;
 243         my $ovv = $lei->{ovv};
 244         my $m = 'eml2'.$ovv->{fmt};
 245         my $eml2mbox = $self->can($m) or die "$self->$m missing";
 246         my $out = $lei->{1} // die "no stdout ($m, $ovv->{dst})"; # redirected earlier
 247         $out->autoflush(1);
 248         my $write = $ovv->{lock_path} ? \&_print_full : \&atomic_append;
 249         my $dedupe = $lei->{dedupe};
 250         $dedupe->prepare_dedupe;
 251         sub { # for git_to_mail
 252                 my ($buf, $smsg, $eml) = @_;
 253                 return unless $out;
 254                 $eml //= PublicInbox::Eml->new($buf);
 255                 if (!$dedupe->is_dup($eml, $smsg->{blob})) {
 256                         $buf = $eml2mbox->($eml, $smsg);
 257                         my $lk = $ovv->lock_for_scope;
 258                         eval { $write->($out, $buf) };
 259                         if ($@) {
 260                                 die $@ if ref($@) ne 'PublicInbox::SIGPIPE';
 261                                 undef $out
 262                         }
 263                 }
 264         }
 265 }
 266
 267 sub _maildir_each_file ($$;@) {
 268         my ($dir, $cb, @arg) = @_;
 269         for my $d (qw(new/ cur/)) {
 270                 my $pfx = $dir.$d;
 271                 opendir my $dh, $pfx or next;
 272                 while (defined(my $fn = readdir($dh))) {
 273                         $cb->($pfx.$fn, @arg) if $fn =~ /:2,[A-Za-z]*\z/;
 274                 }
 275         }
 276 }
 277
 278 sub _augment_file { # _maildir_each_file cb
 279         my ($f, $lei) = @_;
 280         my $eml = PublicInbox::InboxWritable::eml_from_path($f) or return;
 281         _augment($eml, $lei);
 282 }
 283
 284 # _maildir_each_file callback, \&CORE::unlink doesn't work with it
 285 sub _unlink { unlink($_[0]) }
 286
 287 sub _rand () {
 288         state $seq = 0;
 289         sprintf('%x,%x,%x,%x', rand(0xffffffff), time, $$, ++$seq);
 290 }
 291
 292 sub _buf2maildir {
 293         my ($dst, $buf, $smsg) = @_;
 294         my $kw = $smsg->{kw} // [];
 295         my $sfx = join('', sort(map { $kw2char{$_} // () } @$kw));
 296         my $rand = ''; # chosen by die roll :P
 297         my ($tmp, $fh, $final);
 298         my $common = $smsg->{blob} // _rand;
 299         if (defined(my $pct = $smsg->{pct})) { $common .= "=$pct" }
 300         do {
 301                 $tmp = $dst.'tmp/'.$rand.$common;
 302         } while (!sysopen($fh, $tmp, O_CREAT|O_EXCL|O_WRONLY) &&
 303                 $! == EEXIST && ($rand = _rand.','));
 304         if (print $fh $$buf and close($fh)) {
 305                 # ignore new/ and write only to cur/, otherwise MUAs
 306                 # with R/W access to the Maildir will end up doing
 307                 # a mass rename which can take a while with thousands
 308                 # of messages.
 309                 $dst .= 'cur/';
 310                 $rand = '';
 311                 do {
 312                         $final = $dst.$rand.$common.':2,'.$sfx;
 313                 } while (!link($tmp, $final) && $! == EEXIST &&
 314                         ($rand = _rand.','));
 315                 unlink($tmp) or warn "W: failed to unlink $tmp: $!\n";
 316         } else {
 317                 my $err = $!;
 318                 unlink($tmp);
 319                 die "Error writing $smsg->{blob} to $dst: $err";
 320         }
 321 }
 322
 323 sub _maildir_write_cb ($$) {
 324         my ($self, $lei) = @_;
 325         my $dedupe = $lei->{dedupe};
 326         $dedupe->prepare_dedupe;
 327         my $dst = $lei->{ovv}->{dst};
 328         sub { # for git_to_mail
 329                 my ($buf, $smsg, $eml) = @_;
 330                 $buf //= \($eml->as_string);
 331                 return _buf2maildir($dst, $buf, $smsg) if !$dedupe;
 332                 $eml //= PublicInbox::Eml->new($$buf); # copy buf
 333                 return if $dedupe->is_dup($eml, $smsg->{blob});
 334                 undef $eml;
 335                 _buf2maildir($dst, $buf, $smsg);
 336         }
 337 }
 338
 339 sub write_cb { # returns a callback for git_to_mail
 340         my ($self, $lei) = @_;
 341         # _mbox_write_cb or _maildir_write_cb
 342         my $m = "_$self->{base_type}_write_cb";
 343         $self->$m($lei);
 344 }
 345
 346 sub new {
 347         my ($cls, $lei) = @_;
 348         my $fmt = $lei->{ovv}->{fmt};
 349         my $dst = $lei->{ovv}->{dst};
 350         my $self = bless {}, $cls;
 351         if ($fmt eq 'maildir') {
 352                 $self->{base_type} = 'maildir';
 353                 -e $dst && !-d _ and die
 354                                 "$dst exists and is not a directory\n";
 355                 $lei->{ovv}->{dst} = $dst .= '/' if substr($dst, -1) ne '/';
 356         } elsif (substr($fmt, 0, 4) eq 'mbox') {
 357                 (-d $dst || (-e _ && !-w _)) and die
 358                         "$dst exists and is not a writable file\n";
 359                 $self->can("eml2$fmt") or die "bad mbox --format=$fmt\n";
 360                 $self->{base_type} = 'mbox';
 361         } else {
 362                 die "bad mail --format=$fmt\n";
 363         }
 364         $lei->{dedupe} = PublicInbox::LeiDedupe->new($lei);
 365         $self;
 366 }
 367
 368 sub _pre_augment_maildir {} # noop
 369
 370 sub _do_augment_maildir {
 371         my ($self, $lei) = @_;
 372         my $dst = $lei->{ovv}->{dst};
 373         if ($lei->{opt}->{augment}) {
 374                 my $dedupe = $lei->{dedupe};
 375                 if ($dedupe && $dedupe->prepare_dedupe) {
 376                         require PublicInbox::InboxWritable; # eml_from_path
 377                         _maildir_each_file($dst, \&_augment_file, $lei);
 378                         $dedupe->pause_dedupe;
 379                 }
 380         } else { # clobber existing Maildir
 381                 _maildir_each_file($dst, \&_unlink);
 382         }
 383 }
 384
 385 sub _post_augment_maildir {
 386         my ($self, $lei) = @_;
 387         my $dst = $lei->{ovv}->{dst};
 388         for my $x (qw(tmp new cur)) {
 389                 my $d = $dst.$x;
 390                 next if -d $d;
 391                 require File::Path;
 392                 File::Path::mkpath($d);
 393                 -d $d or die "$d is not a directory";
 394         }
 395 }
 396
 397 sub _pre_augment_mbox {
 398         my ($self, $lei) = @_;
 399         my $dst = $lei->{ovv}->{dst};
 400         if ($dst ne '/dev/stdout') {
 401                 my $mode = -p $dst ? '>' : '+>>';
 402                 if (-f _ && !$lei->{opt}->{augment} and !unlink($dst)) {
 403                         $! == ENOENT or die "unlink($dst): $!";
 404                 }
 405                 open my $out, $mode, $dst or die "open($dst): $!";
 406                 $lei->{old_1} = $lei->{1};
 407                 $lei->{1} = $out;
 408         }
 409         # Perl does SEEK_END even with O_APPEND :<
 410         $self->{seekable} = seek($lei->{1}, 0, SEEK_SET);
 411         if (!$self->{seekable} && $! != ESPIPE && $dst ne '/dev/stdout') {
 412                 die "seek($dst): $!\n";
 413         }
 414         state $zsfx_allow = join('|', keys %zsfx2cmd);
 415         ($self->{zsfx}) = ($dst =~ /\.($zsfx_allow)\z/) or return;
 416         pipe(my ($r, $w)) or die "pipe: $!";
 417         [ $r, $w ];
 418 }
 419
 420 sub _do_augment_mbox {
 421         my ($self, $lei) = @_;
 422         return if !$lei->{opt}->{augment};
 423         my $dedupe = $lei->{dedupe};
 424         my $dst = $lei->{ovv}->{dst};
 425         die "cannot augment $dst, not seekable\n" if !$self->{seekable};
 426         my $out = $lei->{1};
 427         if (-s $out && $dedupe && $dedupe->prepare_dedupe) {
 428                 my $zsfx = $self->{zsfx};
 429                 my $rd = $zsfx ? decompress_src($out, $zsfx, $lei) :
 430                                 dup_src($out);
 431                 my $fmt = $lei->{ovv}->{fmt};
 432                 require PublicInbox::MboxReader;
 433                 PublicInbox::MboxReader->$fmt($rd, \&_augment, $lei);
 434         }
 435         # maybe some systems don't honor O_APPEND, Perl does this:
 436         seek($out, 0, SEEK_END) or die "seek $dst: $!";
 437         $dedupe->pause_dedupe if $dedupe;
 438 }
 439
 440 sub pre_augment { # fast (1 disk seek), runs in main daemon
 441         my ($self, $lei) = @_;
 442         # _pre_augment_maildir, _pre_augment_mbox
 443         my $m = "_pre_augment_$self->{base_type}";
 444         $self->$m($lei);
 445 }
 446
 447 sub do_augment { # slow, runs in wq worker
 448         my ($self, $lei) = @_;
 449         # _do_augment_maildir, _do_augment_mbox
 450         my $m = "_do_augment_$self->{base_type}";
 451         $self->$m($lei);
 452 }
 453
 454 sub post_augment { # fast (spawn compressor or mkdir), runs in main daemon
 455         my ($self, $lei, @args) = @_;
 456         # _post_augment_maildir, _post_augment_mbox
 457         my $m = "_post_augment_$self->{base_type}";
 458         $self->$m($lei, @args);
 459 }
 460
 461 sub write_mail { # via ->wq_do
 462         my ($self, $git_dir, $smsg, $lei) = @_;
 463         my $not_done = delete $self->{$lei->{each_smsg_not_done}};
 464         my $wcb = $self->{wcb} //= do { # first message
 465                 my %sig = $lei->atfork_child_wq($self);
 466                 @SIG{keys %sig} = values %sig; # not local
 467                 $lei->{dedupe}->prepare_dedupe;
 468                 $self->write_cb($lei);
 469         };
 470         my $git = $self->{"$$\0$git_dir"} //= PublicInbox::Git->new($git_dir);
 471         $git->cat_async($smsg->{blob}, \&git_to_mail, [$wcb, $smsg, $not_done]);
 472 }
 473
 474 # We rely on OnDestroy to run this before ->DESTROY, since ->DESTROY
 475 # ordering is unstable at worker exit and may cause segfaults
 476 sub reap_gits {
 477         my ($self) = @_;
 478         delete $self->{wcb};
 479         for my $git (delete @$self{grep(/\A$$\0/, keys %$self)}) {
 480                 $git->async_wait_all;
 481         }
 482 }
 483
 484 sub DESTROY { delete $_[0]->{wcb} }
 485
 486 sub ipc_atfork_child { # runs after IPC::wq_worker_loop
 487         my ($self) = @_;
 488         $self->SUPER::ipc_atfork_child;
 489         # reap_gits needs to run before $self->DESTROY,
 490         # IPC.pm will ensure that.
 491         PublicInbox::OnDestroy->new($$, \&reap_gits, $self);
 492 }
 493
 494 1;