]> Sergey Matveev's repositories - public-inbox.git/blob - lib/PublicInbox/Import.pm
import: begin supporting this without ssoma.lock
[public-inbox.git] / lib / PublicInbox / Import.pm
1 # Copyright (C) 2016-2018 all contributors <meta@public-inbox.org>
2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
3 #
4 # git fast-import-based ssoma-mda MDA replacement
5 # This is only ever run by public-inbox-mda and public-inbox-learn,
6 # not the WWW or NNTP code which only requires read-only access.
7 package PublicInbox::Import;
8 use strict;
9 use warnings;
10 use Fcntl qw(:flock :DEFAULT);
11 use PublicInbox::Spawn qw(spawn);
12 use PublicInbox::MID qw(mid_mime mid2path);
13 use PublicInbox::Address;
14
15 sub new {
16         my ($class, $git, $name, $email, $ibx) = @_;
17         my $ref = 'refs/heads/master';
18         if ($ibx) {
19                 $ref = $ibx->{ref_head} || 'refs/heads/master';
20                 $name ||= $ibx->{name};
21                 $email ||= $ibx->{-primary_address};
22         }
23         bless {
24                 git => $git,
25                 ident => "$name <$email>",
26                 mark => 1,
27                 ref => $ref,
28                 inbox => $ibx,
29                 ssoma_lock => 1, # disable for v2
30         }, $class
31 }
32
33 # idempotent start function
34 sub gfi_start {
35         my ($self) = @_;
36
37         return ($self->{in}, $self->{out}) if $self->{pid};
38
39         my ($in_r, $in_w, $out_r, $out_w);
40         pipe($in_r, $in_w) or die "pipe failed: $!";
41         pipe($out_r, $out_w) or die "pipe failed: $!";
42         my $git = $self->{git};
43         my $git_dir = $git->{git_dir};
44
45         my $lockfh;
46         if ($self->{ssoma_lock}) {
47                 my $lockpath = "$git_dir/ssoma.lock";
48                 sysopen($lockfh, $lockpath, O_WRONLY|O_CREAT) or
49                         die "failed to open lock $lockpath: $!";
50                 # wait for other processes to be done
51                 flock($lockfh, LOCK_EX) or die "lock failed: $!\n";
52         }
53
54         local $/ = "\n";
55         chomp($self->{tip} = $git->qx(qw(rev-parse --revs-only), $self->{ref}));
56
57         my @cmd = ('git', "--git-dir=$git_dir", qw(fast-import
58                         --quiet --done --date-format=rfc2822));
59         my $rdr = { 0 => fileno($out_r), 1 => fileno($in_w) };
60         my $pid = spawn(\@cmd, undef, $rdr);
61         die "spawn fast-import failed: $!" unless defined $pid;
62         $out_w->autoflush(1);
63         $self->{in} = $in_r;
64         $self->{out} = $out_w;
65         $self->{lockfh} = $lockfh;
66         $self->{pid} = $pid;
67         $self->{nchg} = 0;
68         binmode $out_w, ':raw' or die "binmode :raw failed: $!";
69         binmode $in_r, ':raw' or die "binmode :raw failed: $!";
70         ($in_r, $out_w);
71 }
72
73 sub wfail () { die "write to fast-import failed: $!" }
74
75 sub now2822 () {
76         my @t = gmtime(time);
77         my $day = qw(Sun Mon Tue Wed Thu Fri Sat)[$t[6]];
78         my $mon = qw(Jan Feb Mar Apr May Jun Jul Aug Sep Oct Nov Dec)[$t[4]];
79
80         sprintf('%s, %2d %s %d %02d:%02d:%02d +0000',
81                 $day, $t[3], $mon, $t[5] + 1900, $t[2], $t[1], $t[0]);
82 }
83
84 sub norm_body ($) {
85         my ($mime) = @_;
86         my $b = $mime->body_raw;
87         $b =~ s/(\r?\n)+\z//s;
88         $b
89 }
90
91 sub _check_path ($$$$) {
92         my ($r, $w, $tip, $path) = @_;
93         return if $tip eq '';
94         print $w "ls $tip $path\n" or wfail;
95         local $/ = "\n";
96         defined(my $info = <$r>) or die "EOF from fast-import: $!";
97         $info =~ /\Amissing / ? undef : $info;
98 }
99
100 # returns undef on non-existent
101 # ('MISMATCH', msg) on mismatch
102 # (:MARK, msg) on success
103 sub remove {
104         my ($self, $mime, $msg) = @_; # mime = Email::MIME
105
106         my $mid = mid_mime($mime);
107         my $path = mid2path($mid);
108
109         my ($r, $w) = $self->gfi_start;
110         my $tip = $self->{tip};
111         my $info = _check_path($r, $w, $tip, $path) or return ('MISSING',undef);
112         $info =~ m!\A100644 blob ([a-f0-9]{40})\t!s or die "not blob: $info";
113         my $blob = $1;
114
115         print $w "cat-blob $blob\n" or wfail;
116         local $/ = "\n";
117         $info = <$r>;
118         defined $info or die "EOF from fast-import / cat-blob: $!";
119         $info =~ /\A[a-f0-9]{40} blob (\d+)\n\z/ or
120                                 die "unexpected cat-blob response: $info";
121         my $left = $1;
122         my $offset = 0;
123         my $buf = '';
124         my $n;
125         while ($left > 0) {
126                 $n = read($r, $buf, $left, $offset);
127                 defined($n) or die "read cat-blob failed: $!";
128                 $n == 0 and die 'fast-export (cat-blob) died';
129                 $left -= $n;
130                 $offset += $n;
131         }
132         $n = read($r, my $lf, 1);
133         defined($n) or die "read final byte of cat-blob failed: $!";
134         die "bad read on final byte: <$lf>" if $lf ne "\n";
135         my $cur = PublicInbox::MIME->new($buf);
136         my $cur_s = $cur->header('Subject');
137         $cur_s = '' unless defined $cur_s;
138         my $cur_m = $mime->header('Subject');
139         $cur_m = '' unless defined $cur_m;
140         if ($cur_s ne $cur_m || norm_body($cur) ne norm_body($mime)) {
141                 return ('MISMATCH', $cur);
142         }
143
144         my $ref = $self->{ref};
145         my $commit = $self->{mark}++;
146         my $parent = $tip =~ /\A:/ ? $tip : undef;
147         unless ($parent) {
148                 print $w "reset $ref\n" or wfail;
149         }
150         my $ident = $self->{ident};
151         my $now = now2822();
152         $msg ||= 'rm';
153         my $len = length($msg) + 1;
154         print $w "commit $ref\nmark :$commit\n",
155                 "author $ident $now\n",
156                 "committer $ident $now\n",
157                 "data $len\n$msg\n\n",
158                 'from ', ($parent ? $parent : $tip), "\n" or wfail;
159         print $w "D $path\n\n" or wfail;
160         $self->{nchg}++;
161         (($self->{tip} = ":$commit"), $cur);
162 }
163
164 # returns undef on duplicate
165 sub add {
166         my ($self, $mime, $check_cb) = @_; # mime = Email::MIME
167
168         my $from = $mime->header('From');
169         my ($email) = PublicInbox::Address::emails($from);
170         my ($name) = PublicInbox::Address::names($from);
171         # git gets confused with:
172         #  "'A U Thor <u@example.com>' via foo" <foo@example.com>
173         # ref:
174         # <CAD0k6qSUYANxbjjbE4jTW4EeVwOYgBD=bXkSu=akiYC_CB7Ffw@mail.gmail.com>
175         $name =~ tr/<>//d;
176
177         my $date = $mime->header('Date');
178         my $subject = $mime->header('Subject');
179         $subject = '(no subject)' unless defined $subject;
180         my $mid = mid_mime($mime);
181         my $path = mid2path($mid);
182
183         my ($r, $w) = $self->gfi_start;
184         my $tip = $self->{tip};
185         _check_path($r, $w, $tip, $path) and return;
186
187         # kill potentially confusing/misleading headers
188         $mime->header_set($_) for qw(bytes lines content-length status);
189         if ($check_cb) {
190                 $mime = $check_cb->($mime) or return;
191         }
192
193         $mime = $mime->as_string;
194         my $blob = $self->{mark}++;
195         print $w "blob\nmark :$blob\ndata ", length($mime), "\n" or wfail;
196         print $w $mime, "\n" or wfail;
197         my $ref = $self->{ref};
198         my $commit = $self->{mark}++;
199         my $parent = $tip =~ /\A:/ ? $tip : undef;
200
201         unless ($parent) {
202                 print $w "reset $ref\n" or wfail;
203         }
204
205         utf8::encode($email);
206         utf8::encode($name);
207         utf8::encode($subject);
208         # quiet down wide character warnings:
209         print $w "commit $ref\nmark :$commit\n",
210                 "author $name <$email> $date\n",
211                 "committer $self->{ident} ", now2822(), "\n" or wfail;
212         print $w "data ", (length($subject) + 1), "\n",
213                 $subject, "\n\n" or wfail;
214         if ($tip ne '') {
215                 print $w 'from ', ($parent ? $parent : $tip), "\n" or wfail;
216         }
217         print $w "M 100644 :$blob $path\n\n" or wfail;
218         $self->{nchg}++;
219         $self->{tip} = ":$commit";
220 }
221
222 sub run_die ($$) {
223         my ($cmd, $env) = @_;
224         my $pid = spawn($cmd, $env, undef);
225         defined $pid or die "spawning ".join(' ', @$cmd)." failed: $!";
226         waitpid($pid, 0) == $pid or die join(' ', @$cmd) .' did not finish';
227         $? == 0 or die join(' ', @$cmd) . " failed: $?\n";
228 }
229
230 sub done {
231         my ($self) = @_;
232         my $w = delete $self->{out} or return;
233         my $r = delete $self->{in} or die 'BUG: missing {in} when done';
234         print $w "done\n" or wfail;
235         my $pid = delete $self->{pid} or die 'BUG: missing {pid} when done';
236         waitpid($pid, 0) == $pid or die 'fast-import did not finish';
237         $? == 0 or die "fast-import failed: $?";
238         my $nchg = delete $self->{nchg};
239
240         # for compatibility with existing ssoma installations
241         # we can probably remove this entirely by 2020
242         my $git_dir = $self->{git}->{git_dir};
243         my @cmd = ('git', "--git-dir=$git_dir");
244         my $index = "$git_dir/ssoma.index";
245         if ($nchg && -e $index && !$ENV{FAST}) {
246                 my $env = { GIT_INDEX_FILE => $index };
247                 run_die([@cmd, qw(read-tree -m -v -i), $self->{ref}], $env);
248         }
249         if ($nchg) {
250                 run_die([@cmd, 'update-server-info'], undef);
251                 eval {
252                         require PublicInbox::SearchIdx;
253                         my $inbox = $self->{inbox} || $git_dir;
254                         my $s = PublicInbox::SearchIdx->new($inbox);
255                         $s->index_sync({ ref => $self->{ref} });
256                 };
257
258                 eval { run_die([@cmd, qw(gc --auto)], undef) };
259         }
260
261         $self->{ssoma_lock} or return;
262         my $lockfh = delete $self->{lockfh} or die "BUG: not locked: $!";
263         flock($lockfh, LOCK_UN) or die "unlock failed: $!";
264         close $lockfh or die "close lock failed: $!";
265 }
266
267 1;
268 __END__
269 =pod
270
271 =head1 NAME
272
273 PublicInbox::Import - message importer for public-inbox
274
275 =head1 VERSION
276
277 version 1.0
278
279 =head1 SYNOPSYS
280
281         use Email::MIME;
282         use PublicInbox::Git;
283         use PublicInbox::Import;
284
285         chomp(my $git_dir = `git rev-parse --git-dir`);
286         $git_dir or die "GIT_DIR= must be specified\n";
287         my $git = PublicInbox::Git->new($git_dir);
288         my @committer = ('inbox', 'inbox@example.org');
289         my $im = PublicInbox::Import->new($git, @committer);
290
291         # to add a message:
292         my $message = "From: <u\@example.org>\n".
293                 "Subject: test message \n" .
294                 "Date: Thu, 01 Jan 1970 00:00:00 +0000\n" .
295                 "Message-ID: <m\@example.org>\n".
296                 "\ntest message";
297         my $parsed = Email::MIME->new($message);
298         my $ret = $im->add($parsed);
299         if (!defined $ret) {
300                 warn "duplicate: ",
301                         $parsed->header_obj->header_raw('Message-ID'), "\n";
302         } else {
303                 print "imported at mark $ret\n";
304         }
305         $im->done;
306
307         # to remove a message
308         my $junk = Email::MIME->new($message);
309         my ($mark, $orig) = $im->remove($junk);
310         if ($mark eq 'MISSING') {
311                 print "not found\n";
312         } elsif ($mark eq 'MISMATCH') {
313                 print "Message exists but does not match\n\n",
314                         $orig->as_string, "\n",;
315         } else {
316                 print "removed at mark $mark\n\n",
317                         $orig->as_string, "\n";
318         }
319         $im->done;
320
321 =head1 DESCRIPTION
322
323 An importer and remover for public-inboxes which takes L<Email::MIME>
324 messages as input and stores them in a ssoma repository as
325 documented in L<https://ssoma.public-inbox.org/ssoma_repository.txt>,
326 except it does not allow duplicate Message-IDs.
327
328 It requires L<git(1)> and L<git-fast-import(1)> to be installed.
329
330 =head1 METHODS
331
332 =cut
333
334 =head2 new
335
336         my $im = PublicInbox::Import->new($git, @committer);
337
338 Initialize a new PublicInbox::Import object.
339
340 =head2 add
341
342         my $parsed = Email::MIME->new($message);
343         $im->add($parsed);
344
345 Adds a message to to the git repository.  This will acquire
346 C<$GIT_DIR/ssoma.lock> and start L<git-fast-import(1)> if necessary.
347
348 Messages added will not be visible to other processes until L</done>
349 is called, but L</remove> may be called on them.
350
351 =head2 remove
352
353         my $junk = Email::MIME->new($message);
354         my ($code, $orig) = $im->remove($junk);
355
356 Removes a message from the repository.  On success, it returns
357 a ':'-prefixed numeric code representing the git-fast-import
358 mark and the original messages as an Email::MIME object.
359 If the message could not be found, the code is "MISSING"
360 and the original message is undef.  If there is a mismatch where
361 the "Message-ID" is matched but the subject and body do not match,
362 the returned code is "MISMATCH" and the conflicting message
363 is returned as orig.
364
365 =head2 done
366
367 Finalizes the L<git-fast-import(1)> and unlocks the repository.
368 Calling this is required to finalize changes to a repository.
369
370 =head1 SEE ALSO
371
372 L<Email::MIME>
373
374 =head1 CONTACT
375
376 All feedback welcome via plain-text mail to L<mailto:meta@public-inbox.org>
377
378 The mail archives are hosted at L<https://public-inbox.org/meta/>
379
380 =head1 COPYRIGHT
381
382 Copyright (C) 2016 all contributors L<mailto:meta@public-inbox.org>
383
384 License: AGPL-3.0+ L<http://www.gnu.org/licenses/agpl-3.0.txt>
385
386 =cut