]> Sergey Matveev's repositories - public-inbox.git/blob - lib/PublicInbox/WatchMaildir.pm
watch: improve fairness during full rescans
[public-inbox.git] / lib / PublicInbox / WatchMaildir.pm
1 # Copyright (C) 2016 all contributors <meta@public-inbox.org>
2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
3 #
4 # ref: https://cr.yp.to/proto/maildir.html
5 #       http://wiki2.dovecot.org/MailboxFormat/Maildir
6 package PublicInbox::WatchMaildir;
7 use strict;
8 use warnings;
9 use PublicInbox::MIME;
10 use Email::MIME::ContentType;
11 $Email::MIME::ContentType::STRICT_PARAMS = 0; # user input is imperfect
12 use PublicInbox::Git;
13 use PublicInbox::Import;
14 use PublicInbox::MDA;
15 use PublicInbox::Spawn qw(spawn);
16 use File::Temp qw//;
17
18 sub new {
19         my ($class, $config) = @_;
20         my (%mdmap, @mdir, $spamc, $spamdir);
21
22         # "publicinboxwatch" is the documented namespace
23         # "publicinboxlearn" is legacy but may be supported
24         # indefinitely...
25         foreach my $pfx (qw(publicinboxwatch publicinboxlearn)) {
26                 my $k = "$pfx.watchspam";
27                 if (my $dir = $config->{$k}) {
28                         if ($dir =~ s/\Amaildir://) {
29                                 $dir =~ s!/+\z!!;
30                                 # skip "new", no MUA has seen it, yet.
31                                 my $cur = "$dir/cur";
32                                 $spamdir = $cur;
33                                 push @mdir, $cur;
34                                 $mdmap{$cur} = 'watchspam';
35                         } else {
36                                 warn "unsupported $k=$dir\n";
37                         }
38                 }
39         }
40
41         my $k = 'publicinboxwatch.spamcheck';
42         my $spamcheck = $config->{$k};
43         if ($spamcheck) {
44                 if ($spamcheck eq 'spamc') {
45                         $spamcheck = 'PublicInbox::Spamcheck::Spamc';
46                 }
47                 if ($spamcheck =~ /::/) {
48                         eval "require $spamcheck";
49                         $spamcheck = _spamcheck_cb($spamcheck->new);
50                 } else {
51                         warn "unsupported $k=$spamcheck\n";
52                         $spamcheck = undef;
53                 }
54         }
55         foreach $k (keys %$config) {
56                 $k =~ /\Apublicinbox\.([^\.]+)\.watch\z/ or next;
57                 my $name = $1;
58                 my $watch = $config->{$k};
59                 if ($watch =~ s/\Amaildir://) {
60                         $watch =~ s!/+\z!!;
61                         my $inbox = $config->lookup_name($name);
62                         if (my $wm = $inbox->{watchheader}) {
63                                 my ($k, $v) = split(/:/, $wm, 2);
64                                 $inbox->{-watchheader} = [ $k, qr/\Q$v\E/ ];
65                         }
66                         my $new = "$watch/new";
67                         my $cur = "$watch/cur";
68                         push @mdir, $new, $cur;
69                         die "$new already in use\n" if $mdmap{$new};
70                         die "$cur already in use\n" if $mdmap{$cur};
71                         $mdmap{$new} = $mdmap{$cur} = $inbox;
72                 } else {
73                         warn "watch unsupported: $k=$watch\n";
74                 }
75         }
76         return unless @mdir;
77
78         my $mdre = join('|', map { quotemeta($_) } @mdir);
79         $mdre = qr!\A($mdre)/!;
80         bless {
81                 spamcheck => $spamcheck,
82                 spamdir => $spamdir,
83                 mdmap => \%mdmap,
84                 mdir => \@mdir,
85                 mdre => $mdre,
86                 config => $config,
87                 importers => {},
88                 opendirs => {}, # dirname => dirhandle (in progress scans)
89         }, $class;
90 }
91
92 sub _done_for_now {
93         my ($self) = @_;
94         my $opendirs = $self->{opendirs};
95
96         # spamdir scanning means every importer remains open
97         my $spamdir = $self->{spamdir};
98         return if defined($spamdir) && $opendirs->{$spamdir};
99
100         foreach my $im (values %{$self->{importers}}) {
101                 # not done if we're scanning
102                 next if $opendirs->{$im->{git}->{git_dir}};
103                 $im->done;
104         }
105 }
106
107 sub _try_fsn_paths {
108         my ($self, $scan_re, $paths) = @_;
109         foreach (@$paths) {
110                 my $path = $_->{path};
111                 if ($path =~ $scan_re) {
112                         scan($self, $path);
113                 } else {
114                         _try_path($self, $path);
115                 }
116         }
117         _done_for_now($self);
118 }
119
120 sub _remove_spam {
121         my ($self, $path) = @_;
122         # path must be marked as (S)een
123         $path =~ /:2,[A-R]*S[T-Za-z]*\z/ or return;
124         my $mime = _path_to_mime($path) or return;
125         _force_mid($mime);
126         $self->{config}->each_inbox(sub {
127                 my ($ibx) = @_;
128                 eval {
129                         my $im = _importer_for($self, $ibx);
130                         $im->remove($mime);
131                         if (my $scrub = _scrubber_for($ibx)) {
132                                 my $scrubbed = $scrub->scrub($mime) or return;
133                                 $scrubbed == 100 and return;
134                                 $im->remove($scrubbed);
135                         }
136                 };
137                 if ($@) {
138                         warn "error removing spam at: ", $path,
139                                 " from ", $ibx->{name}, ': ', $@, "\n";
140                 }
141         })
142 }
143
144 # used to hash the relevant portions of a message when there are conflicts
145 sub _hash_mime2 {
146         my ($mime) = @_;
147         require Digest::SHA;
148         my $dig = Digest::SHA->new('SHA-1');
149         $dig->add($mime->header_obj->header_raw('Subject'));
150         $dig->add($mime->body_raw);
151         $dig->hexdigest;
152 }
153
154 sub _force_mid {
155         my ($mime) = @_;
156         # probably a bad idea, but we inject a Message-Id if
157         # one is missing, here..
158         my $mid = $mime->header_obj->header_raw('Message-Id');
159         if (!defined $mid || $mid =~ /\A\s*\z/) {
160                 $mid = '<' . _hash_mime2($mime) . '@generated>';
161                 $mime->header_set('Message-Id', $mid);
162         }
163 }
164
165 sub _try_path {
166         my ($self, $path) = @_;
167         my @p = split(m!/+!, $path);
168         return if $p[-1] !~ /\A[a-zA-Z0-9][\w:,=\.]+\z/;
169         if ($p[-1] =~ /:2,([A-Z]+)\z/i) {
170                 my $flags = $1;
171                 return if $flags =~ /[DT]/; # no [D]rafts or [T]rashed mail
172         }
173         return unless -f $path;
174         if ($path !~ $self->{mdre}) {
175                 warn "unrecognized path: $path\n";
176                 return;
177         }
178         my $inbox = $self->{mdmap}->{$1};
179         unless ($inbox) {
180                 warn "unmappable dir: $1\n";
181                 return;
182         }
183         if (!ref($inbox) && $inbox eq 'watchspam') {
184                 return _remove_spam($self, $path);
185         }
186         my $im = _importer_for($self, $inbox);
187         my $mime = _path_to_mime($path) or return;
188         $mime->header_set($_) foreach @PublicInbox::MDA::BAD_HEADERS;
189         my $wm = $inbox->{-watchheader};
190         if ($wm) {
191                 my $v = $mime->header_obj->header_raw($wm->[0]);
192                 return unless ($v && $v =~ $wm->[1]);
193         }
194         if (my $scrub = _scrubber_for($inbox)) {
195                 my $ret = $scrub->scrub($mime) or return;
196                 $ret == 100 and return;
197                 $mime = $ret;
198         }
199
200         _force_mid($mime);
201         $im->add($mime, $self->{spamcheck});
202 }
203
204 sub quit { $_[0]->{quit} = 1 }
205
206 sub watch {
207         my ($self) = @_;
208         my $scan = File::Temp->newdir("public-inbox-watch.$$.scan.XXXXXX",
209                                         TMPDIR => 1);
210         my $scandir = $self->{scandir} = $scan->dirname;
211         my $re = qr!\A$scandir/!;
212         my $cb = sub { _try_fsn_paths($self, $re, \@_) };
213
214         # lazy load here, we may support watching via IMAP IDLE
215         # in the future...
216         require Filesys::Notify::Simple;
217         my $fsn = Filesys::Notify::Simple->new([@{$self->{mdir}}, $scandir]);
218         $fsn->wait($cb) until ($self->{quit});
219 }
220
221 sub trigger_scan {
222         my ($self, $base) = @_;
223         my $dir = $self->{scandir} or die "not watch-ing, yet\n";
224         open my $fh, '>', "$dir/$base" or die "open $dir/$base failed: $!\n";
225         close $fh or die "close $dir/$base failed: $!\n";
226 }
227
228 sub scan {
229         my ($self, $path) = @_;
230         my $max = 10;
231         my $opendirs = $self->{opendirs};
232         my @dirnames = keys %$opendirs;
233         foreach my $dir (@dirnames) {
234                 my $dh = delete $opendirs->{$dir};
235                 my $n = $max;
236                 while (my $fn = readdir($dh)) {
237                         _try_path($self, "$dir/$fn");
238                         last if --$n < 0;
239                 }
240                 $opendirs->{$dir} = $dh if $n < 0;
241         }
242         if ($path =~ /full\z/) {
243                 foreach my $dir (@{$self->{mdir}}) {
244                         next if $opendirs->{$dir}; # already in progress
245                         my $ok = opendir(my $dh, $dir);
246                         unless ($ok) {
247                                 warn "failed to open $dir: $!\n";
248                                 next;
249                         }
250                         my $n = $max;
251                         while (my $fn = readdir($dh)) {
252                                 _try_path($self, "$dir/$fn");
253                                 last if --$n < 0;
254                         }
255                         $opendirs->{$dir} = $dh if $n < 0;
256                 }
257         }
258         if (keys %$opendirs) { # do we have more work to do?
259                 trigger_scan($self, 'cont');
260         } else {
261                 _done_for_now($self);
262         }
263 }
264
265 sub _path_to_mime {
266         my ($path) = @_;
267         if (open my $fh, '<', $path) {
268                 local $/;
269                 my $str = <$fh>;
270                 $str or return;
271                 return PublicInbox::MIME->new(\$str);
272         } elsif ($!{ENOENT}) {
273                 return;
274         } else {
275                 warn "failed to open $path: $!\n";
276                 return;
277         }
278 }
279
280 sub _importer_for {
281         my ($self, $inbox) = @_;
282         my $im = $inbox->{-import} ||= eval {
283                 my $git = $inbox->git;
284                 my $name = $inbox->{name};
285                 my $addr = $inbox->{-primary_address};
286                 PublicInbox::Import->new($git, $name, $addr, $inbox);
287         };
288
289         my $importers = $self->{importers};
290         if (scalar(keys(%$importers)) > 2) {
291                 delete $importers->{"$im"};
292                 _done_for_now($self);
293         }
294
295         $importers->{"$im"} = $im;
296 }
297
298 sub _scrubber_for {
299         my ($inbox) = @_;
300         my $f = $inbox->{filter};
301         if ($f && $f =~ /::/) {
302                 my @args = (-inbox => $inbox);
303                 # basic line splitting, only
304                 # Perhaps we can have proper quote splitting one day...
305                 ($f, @args) = split(/\s+/, $f) if $f =~ /\s+/;
306
307                 eval "require $f";
308                 if ($@) {
309                         warn $@;
310                 } else {
311                         # e.g: PublicInbox::Filter::Vger->new(@args)
312                         return $f->new(@args);
313                 }
314         }
315         undef;
316 }
317
318 sub _spamcheck_cb {
319         my ($sc) = @_;
320         sub {
321                 my ($mime) = @_;
322                 my $tmp = '';
323                 if ($sc->spamcheck($mime, \$tmp)) {
324                         return PublicInbox::MIME->new(\$tmp);
325                 }
326                 warn $mime->header('Message-ID')." failed spam check\n";
327                 undef;
328         }
329 }
330
331 1;