]> Sergey Matveev's repositories - public-inbox.git/blob - lib/PublicInbox/SearchIdx.pm
imaptracker: preserve WAL journal_mode if set by user
[public-inbox.git] / lib / PublicInbox / SearchIdx.pm
1 # Copyright (C) 2015-2020 all contributors <meta@public-inbox.org>
2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
3 # based on notmuch, but with no concept of folders, files or flags
4 #
5 # Indexes mail with Xapian and our (SQLite-based) ::Msgmap for use
6 # with the web and NNTP interfaces.  This index maintains thread
7 # relationships for use by PublicInbox::SearchThread.
8 # This writes to the search index.
9 package PublicInbox::SearchIdx;
10 use strict;
11 use v5.10.1;
12 use parent qw(PublicInbox::Search PublicInbox::Lock Exporter);
13 use PublicInbox::Eml;
14 use PublicInbox::InboxWritable;
15 use PublicInbox::MID qw(mids_for_index mids);
16 use PublicInbox::MsgIter;
17 use PublicInbox::IdxStack;
18 use Carp qw(croak);
19 use POSIX qw(strftime);
20 use PublicInbox::OverIdx;
21 use PublicInbox::Spawn qw(spawn nodatacow_dir);
22 use PublicInbox::Git qw(git_unquote);
23 use PublicInbox::MsgTime qw(msg_timestamp msg_datestamp);
24 our @EXPORT_OK = qw(crlf_adjust log2stack is_ancestor check_size);
25 my $X = \%PublicInbox::Search::X;
26 my ($DB_CREATE_OR_OPEN, $DB_OPEN);
27 our $DB_NO_SYNC = 0;
28 our $BATCH_BYTES = $ENV{XAPIAN_FLUSH_THRESHOLD} ? 0x7fffffff : 1_000_000;
29 use constant DEBUG => !!$ENV{DEBUG};
30
31 my $xapianlevels = qr/\A(?:full|medium)\z/;
32 my $hex = '[a-f0-9]';
33 my $OID = $hex .'{40,}';
34
35 sub new {
36         my ($class, $ibx, $creat, $shard) = @_;
37         ref $ibx or die "BUG: expected PublicInbox::Inbox object: $ibx";
38         my $levels = qr/\A(?:full|medium|basic)\z/;
39         my $inboxdir = $ibx->{inboxdir};
40         my $version = $ibx->version;
41         my $indexlevel = 'full';
42         my $altid = $ibx->{altid};
43         if ($altid) {
44                 require PublicInbox::AltId;
45                 $altid = [ map { PublicInbox::AltId->new($ibx, $_); } @$altid ];
46         }
47         if ($ibx->{indexlevel}) {
48                 if ($ibx->{indexlevel} =~ $levels) {
49                         $indexlevel = $ibx->{indexlevel};
50                 } else {
51                         die("Invalid indexlevel $ibx->{indexlevel}\n");
52                 }
53         }
54         $ibx = PublicInbox::InboxWritable->new($ibx);
55         my $self = bless {
56                 ibx => $ibx,
57                 xpfx => $inboxdir, # for xpfx_init
58                 -altid => $altid,
59                 ibx_ver => $version,
60                 indexlevel => $indexlevel,
61         }, $class;
62         $self->xpfx_init;
63         $self->{-set_indexlevel_once} = 1 if $indexlevel eq 'medium';
64         if ($ibx->{-skip_docdata}) {
65                 $self->{-set_skip_docdata_once} = 1;
66                 $self->{-skip_docdata} = 1;
67         }
68         $ibx->umask_prepare;
69         if ($version == 1) {
70                 $self->{lock_path} = "$inboxdir/ssoma.lock";
71                 my $dir = $self->xdir;
72                 $self->{over} = PublicInbox::OverIdx->new("$dir/over.sqlite3");
73                 $self->{over}->{-no_fsync} = 1 if $ibx->{-no_fsync};
74         } elsif ($version == 2) {
75                 defined $shard or die "shard is required for v2\n";
76                 # shard is a number
77                 $self->{shard} = $shard;
78                 $self->{lock_path} = undef;
79         } else {
80                 die "unsupported inbox version=$version\n";
81         }
82         $self->{creat} = ($creat || 0) == 1;
83         $self;
84 }
85
86 sub need_xapian ($) { $_[0]->{indexlevel} =~ $xapianlevels }
87
88 sub idx_release {
89         my ($self, $wake) = @_;
90         if (need_xapian($self)) {
91                 my $xdb = delete $self->{xdb} or croak 'not acquired';
92                 $xdb->close;
93         }
94         $self->lock_release($wake) if $self->{creat};
95         undef;
96 }
97
98 sub load_xapian_writable () {
99         return 1 if $X->{WritableDatabase};
100         PublicInbox::Search::load_xapian() or return;
101         my $xap = $PublicInbox::Search::Xap;
102         for (qw(Document TermGenerator WritableDatabase)) {
103                 $X->{$_} = $xap.'::'.$_;
104         }
105         eval 'require '.$X->{WritableDatabase} or die;
106         *sortable_serialise = $xap.'::sortable_serialise';
107         $DB_CREATE_OR_OPEN = eval($xap.'::DB_CREATE_OR_OPEN()');
108         $DB_OPEN = eval($xap.'::DB_OPEN()');
109         my $ver = (eval($xap.'::major_version()') << 16) |
110                 (eval($xap.'::minor_version()') << 8);
111         $DB_NO_SYNC = 0x4 if $ver >= 0x10400;
112         1;
113 }
114
115 sub idx_acquire {
116         my ($self) = @_;
117         my $flag;
118         my $dir = $self->xdir;
119         if (need_xapian($self)) {
120                 croak 'already acquired' if $self->{xdb};
121                 load_xapian_writable();
122                 $flag = $self->{creat} ? $DB_CREATE_OR_OPEN : $DB_OPEN;
123         }
124         if ($self->{creat}) {
125                 require File::Path;
126                 $self->lock_acquire;
127
128                 # don't create empty Xapian directories if we don't need Xapian
129                 my $is_shard = defined($self->{shard});
130                 if (!-d $dir && (!$is_shard ||
131                                 ($is_shard && need_xapian($self)))) {
132                         File::Path::mkpath($dir);
133                         nodatacow_dir($dir);
134                         $self->{-set_has_threadid_once} = 1;
135                 }
136         }
137         return unless defined $flag;
138         $flag |= $DB_NO_SYNC if $self->{ibx}->{-no_fsync};
139         my $xdb = eval { ($X->{WritableDatabase})->new($dir, $flag) };
140         croak "Failed opening $dir: $@" if $@;
141         $self->{xdb} = $xdb;
142 }
143
144 sub add_val ($$$) {
145         my ($doc, $col, $num) = @_;
146         $num = sortable_serialise($num);
147         $doc->add_value($col, $num);
148 }
149
150 sub term_generator ($) { # write-only
151         my ($self) = @_;
152
153         $self->{term_generator} //= do {
154                 my $tg = $X->{TermGenerator}->new;
155                 $tg->set_stemmer($self->stemmer);
156                 $tg;
157         }
158 }
159
160 sub index_text ($$$$) {
161         my ($self, $text, $wdf_inc, $prefix) = @_;
162         my $tg = term_generator($self); # man Search::Xapian::TermGenerator
163
164         if ($self->{indexlevel} eq 'full') {
165                 $tg->index_text($text, $wdf_inc, $prefix);
166                 $tg->increase_termpos;
167         } else {
168                 $tg->index_text_without_positions($text, $wdf_inc, $prefix);
169         }
170 }
171
172 sub index_headers ($$) {
173         my ($self, $smsg) = @_;
174         my @x = (from => 'A', # Author
175                 subject => 'S', to => 'XTO', cc => 'XCC');
176         while (my ($field, $pfx) = splice(@x, 0, 2)) {
177                 my $val = $smsg->{$field};
178                 index_text($self, $val, 1, $pfx) if $val ne '';
179         }
180 }
181
182 sub index_diff_inc ($$$$) {
183         my ($self, $text, $pfx, $xnq) = @_;
184         if (@$xnq) {
185                 index_text($self, join("\n", @$xnq), 1, 'XNQ');
186                 @$xnq = ();
187         }
188         index_text($self, $text, 1, $pfx);
189 }
190
191 sub index_old_diff_fn {
192         my ($self, $seen, $fa, $fb, $xnq) = @_;
193
194         # no renames or space support for traditional diffs,
195         # find the number of leading common paths to strip:
196         my @fa = split('/', $fa);
197         my @fb = split('/', $fb);
198         while (scalar(@fa) && scalar(@fb)) {
199                 $fa = join('/', @fa);
200                 $fb = join('/', @fb);
201                 if ($fa eq $fb) {
202                         unless ($seen->{$fa}++) {
203                                 index_diff_inc($self, $fa, 'XDFN', $xnq);
204                         }
205                         return 1;
206                 }
207                 shift @fa;
208                 shift @fb;
209         }
210         0;
211 }
212
213 sub index_diff ($$$) {
214         my ($self, $txt, $doc) = @_;
215         my %seen;
216         my $in_diff;
217         my @xnq;
218         my $xnq = \@xnq;
219         foreach (split(/\n/, $txt)) {
220                 if ($in_diff && s/^ //) { # diff context
221                         index_diff_inc($self, $_, 'XDFCTX', $xnq);
222                 } elsif (/^-- $/) { # email signature begins
223                         $in_diff = undef;
224                 } elsif (m!^diff --git "?[^/]+/.+ "?[^/]+/.+\z!) {
225                         # wait until "---" and "+++" to capture filenames
226                         $in_diff = 1;
227                 # traditional diff:
228                 } elsif (m/^diff -(.+) (\S+) (\S+)$/) {
229                         my ($opt, $fa, $fb) = ($1, $2, $3);
230                         push @xnq, $_;
231                         # only support unified:
232                         next unless $opt =~ /[uU]/;
233                         $in_diff = index_old_diff_fn($self, \%seen, $fa, $fb,
234                                                         $xnq);
235                 } elsif (m!^--- ("?[^/]+/.+)!) {
236                         my $fn = $1;
237                         $fn = (split('/', git_unquote($fn), 2))[1];
238                         $seen{$fn}++ or index_diff_inc($self, $fn, 'XDFN', $xnq);
239                         $in_diff = 1;
240                 } elsif (m!^\+\+\+ ("?[^/]+/.+)!)  {
241                         my $fn = $1;
242                         $fn = (split('/', git_unquote($fn), 2))[1];
243                         $seen{$fn}++ or index_diff_inc($self, $fn, 'XDFN', $xnq);
244                         $in_diff = 1;
245                 } elsif (/^--- (\S+)/) {
246                         $in_diff = $1;
247                         push @xnq, $_;
248                 } elsif (defined $in_diff && /^\+\+\+ (\S+)/) {
249                         $in_diff = index_old_diff_fn($self, \%seen, $in_diff,
250                                                         $1, $xnq);
251                 } elsif ($in_diff && s/^\+//) { # diff added
252                         index_diff_inc($self, $_, 'XDFB', $xnq);
253                 } elsif ($in_diff && s/^-//) { # diff removed
254                         index_diff_inc($self, $_, 'XDFA', $xnq);
255                 } elsif (m!^index ([a-f0-9]+)\.\.([a-f0-9]+)!) {
256                         my ($ba, $bb) = ($1, $2);
257                         index_git_blob_id($doc, 'XDFPRE', $ba);
258                         index_git_blob_id($doc, 'XDFPOST', $bb);
259                         $in_diff = 1;
260                 } elsif (/^@@ (?:\S+) (?:\S+) @@\s*$/) {
261                         # traditional diff w/o -p
262                 } elsif (/^@@ (?:\S+) (?:\S+) @@\s*(\S+.*)$/) {
263                         # hunk header context
264                         index_diff_inc($self, $1, 'XDFHH', $xnq);
265                 # ignore the following lines:
266                 } elsif (/^(?:dis)similarity index/ ||
267                                 /^(?:old|new) mode/ ||
268                                 /^(?:deleted|new) file mode/ ||
269                                 /^(?:copy|rename) (?:from|to) / ||
270                                 /^(?:dis)?similarity index / ||
271                                 /^\\ No newline at end of file/ ||
272                                 /^Binary files .* differ/) {
273                         push @xnq, $_;
274                 } elsif ($_ eq '') {
275                         # possible to be in diff context, some mail may be
276                         # stripped by MUA or even GNU diff(1).  "git apply"
277                         # treats a bare "\n" as diff context, too
278                 } else {
279                         push @xnq, $_;
280                         warn "non-diff line: $_\n" if DEBUG && $_ ne '';
281                         $in_diff = undef;
282                 }
283         }
284
285         index_text($self, join("\n", @xnq), 1, 'XNQ');
286 }
287
288 sub index_xapian { # msg_iter callback
289         my $part = $_[0]->[0]; # ignore $depth and $idx
290         my ($self, $doc) = @{$_[1]};
291         my $ct = $part->content_type || 'text/plain';
292         my $fn = $part->filename;
293         if (defined $fn && $fn ne '') {
294                 index_text($self, $fn, 1, 'XFN');
295         }
296         if ($part->{is_submsg}) {
297                 my $mids = mids_for_index($part);
298                 index_ids($self, $doc, $part, $mids);
299                 my $smsg = bless {}, 'PublicInbox::Smsg';
300                 $smsg->populate($part);
301                 index_headers($self, $smsg);
302         }
303
304         my ($s, undef) = msg_part_text($part, $ct);
305         defined $s or return;
306         $_[0]->[0] = $part = undef; # free memory
307
308         # split off quoted and unquoted blocks:
309         my @sections = PublicInbox::MsgIter::split_quotes($s);
310         undef $s; # free memory
311         for my $txt (@sections) {
312                 if ($txt =~ /\A>/) {
313                         index_text($self, $txt, 0, 'XQUOT');
314                 } else {
315                         # does it look like a diff?
316                         if ($txt =~ /^(?:diff|---|\+\+\+) /ms) {
317                                 index_diff($self, $txt, $doc);
318                         } else {
319                                 index_text($self, $txt, 1, 'XNQ');
320                         }
321                 }
322                 undef $txt; # free memory
323         }
324 }
325
326 sub index_ids ($$$$) {
327         my ($self, $doc, $hdr, $mids) = @_;
328         for my $mid (@$mids) {
329                 index_text($self, $mid, 1, 'XM');
330
331                 # because too many Message-IDs are prefixed with
332                 # "Pine.LNX."...
333                 if ($mid =~ /\w{12,}/) {
334                         my @long = ($mid =~ /(\w{3,}+)/g);
335                         index_text($self, join(' ', @long), 1, 'XM');
336                 }
337         }
338         $doc->add_boolean_term('Q' . $_) for @$mids;
339         for my $l ($hdr->header_raw('List-Id')) {
340                 $l =~ /<([^>]+)>/ or next;
341                 my $lid = $1;
342                 $doc->add_boolean_term('G' . $lid);
343                 index_text($self, $lid, 1, 'XL'); # probabilistic
344         }
345 }
346
347 sub add_xapian ($$$$) {
348         my ($self, $eml, $smsg, $mids) = @_;
349         my $doc = $X->{Document}->new;
350         add_val($doc, PublicInbox::Search::TS(), $smsg->{ts});
351         my @ds = gmtime($smsg->{ds});
352         my $yyyymmdd = strftime('%Y%m%d', @ds);
353         add_val($doc, PublicInbox::Search::YYYYMMDD(), $yyyymmdd);
354         my $dt = strftime('%Y%m%d%H%M%S', @ds);
355         add_val($doc, PublicInbox::Search::DT(), $dt);
356         add_val($doc, PublicInbox::Search::BYTES(), $smsg->{bytes});
357         add_val($doc, PublicInbox::Search::UID(), $smsg->{num});
358         add_val($doc, PublicInbox::Search::THREADID, $smsg->{tid});
359
360         my $tg = term_generator($self);
361         $tg->set_document($doc);
362         index_headers($self, $smsg);
363
364         msg_iter($eml, \&index_xapian, [ $self, $doc ]);
365         index_ids($self, $doc, $eml, $mids);
366
367         # by default, we maintain compatibility with v1.5.0 and earlier
368         # by writing to docdata.glass, users who never exect to downgrade can
369         # use --skip-docdata
370         if (!$self->{-skip_docdata}) {
371                 # WWW doesn't need {to} or {cc}, only NNTP
372                 $smsg->{to} = $smsg->{cc} = '';
373                 PublicInbox::OverIdx::parse_references($smsg, $eml, $mids);
374                 my $data = $smsg->to_doc_data;
375                 $doc->set_data($data);
376         }
377
378         if (my $altid = $self->{-altid}) {
379                 foreach my $alt (@$altid) {
380                         my $pfx = $alt->{xprefix};
381                         foreach my $mid (@$mids) {
382                                 my $id = $alt->mid2alt($mid);
383                                 next unless defined $id;
384                                 $doc->add_boolean_term($pfx . $id);
385                         }
386                 }
387         }
388         $self->{xdb}->replace_document($smsg->{num}, $doc);
389 }
390
391 sub _msgmap_init ($) {
392         my ($self) = @_;
393         die "BUG: _msgmap_init is only for v1\n" if $self->{ibx_ver} != 1;
394         $self->{mm} //= eval {
395                 require PublicInbox::Msgmap;
396                 my $rw = $self->{ibx}->{-no_fsync} ? 2 : 1;
397                 PublicInbox::Msgmap->new($self->{ibx}->{inboxdir}, $rw);
398         };
399 }
400
401 sub add_message {
402         # mime = PublicInbox::Eml or Email::MIME object
403         my ($self, $mime, $smsg, $sync) = @_;
404         my $mids = mids_for_index($mime);
405         $smsg //= bless { blob => '' }, 'PublicInbox::Smsg'; # test-only compat
406         $smsg->{mid} //= $mids->[0]; # v1 compatibility
407         $smsg->{num} //= do { # v1
408                 _msgmap_init($self);
409                 index_mm($self, $mime, $smsg->{blob}, $sync);
410         };
411
412         # v1 and tests only:
413         $smsg->populate($mime, $sync);
414         $smsg->{bytes} //= length($mime->as_string);
415
416         eval {
417                 # order matters, overview stores every possible piece of
418                 # data in doc_data (deflated).  Xapian only stores a subset
419                 # of the fields which exist in over.sqlite3.  We may stop
420                 # storing doc_data in Xapian sometime after we get multi-inbox
421                 # search working.
422                 if (my $over = $self->{over}) { # v1 only
423                         $over->add_overview($mime, $smsg);
424                 }
425                 if (need_xapian($self)) {
426                         add_xapian($self, $mime, $smsg, $mids);
427                 }
428         };
429
430         if ($@) {
431                 warn "failed to index message <".join('> <',@$mids).">: $@\n";
432                 return undef;
433         }
434         $smsg->{num};
435 }
436
437 sub xdb_remove {
438         my ($self, $oid, @removed) = @_;
439         my $xdb = $self->{xdb} or return;
440         for my $num (@removed) {
441                 my $doc = eval { $xdb->get_document($num) };
442                 unless ($doc) {
443                         warn "E: $@\n" if $@;
444                         warn "E: #$num $oid missing in Xapian\n";
445                         next;
446                 }
447                 my $smsg = bless {}, 'PublicInbox::Smsg';
448                 $smsg->load_expand($doc);
449                 my $blob = $smsg->{blob} // '(unset)';
450                 if ($blob eq $oid) {
451                         $xdb->delete_document($num);
452                 } else {
453                         warn "E: #$num $oid != $blob in Xapian\n";
454                 }
455         }
456 }
457
458 sub remove_by_oid {
459         my ($self, $oid, $num) = @_;
460         die "BUG: remove_by_oid is v2-only\n" if $self->{over};
461         $self->begin_txn_lazy;
462         xdb_remove($self, $oid, $num) if need_xapian($self);
463 }
464
465 sub index_git_blob_id {
466         my ($doc, $pfx, $objid) = @_;
467
468         my $len = length($objid);
469         for (my $len = length($objid); $len >= 7; ) {
470                 $doc->add_term($pfx.$objid);
471                 $objid = substr($objid, 0, --$len);
472         }
473 }
474
475 # v1 only
476 sub unindex_eml {
477         my ($self, $oid, $eml) = @_;
478         my $mids = mids($eml);
479         my $nr = 0;
480         my %tmp;
481         for my $mid (@$mids) {
482                 my @removed = eval { $self->{over}->remove_oid($oid, $mid) };
483                 if ($@) {
484                         warn "E: failed to remove <$mid> from overview: $@\n";
485                 } else {
486                         $nr += scalar @removed;
487                         $tmp{$_}++ for @removed;
488                 }
489         }
490         if (!$nr) {
491                 $mids = join('> <', @$mids);
492                 warn "W: <$mids> missing for removal from overview\n";
493         }
494         while (my ($num, $nr) = each %tmp) {
495                 warn "BUG: $num appears >1 times ($nr) for $oid\n" if $nr != 1;
496         }
497         if ($nr) {
498                 $self->{mm}->num_delete($_) for (keys %tmp);
499         } else { # just in case msgmap and over.sqlite3 become desynched:
500                 $self->{mm}->mid_delete($mids->[0]);
501         }
502         xdb_remove($self, $oid, keys %tmp) if need_xapian($self);
503 }
504
505 sub index_mm {
506         my ($self, $mime, $oid, $sync) = @_;
507         my $mids = mids($mime);
508         my $mm = $self->{mm};
509         if ($sync->{reindex}) {
510                 my $over = $self->{over};
511                 for my $mid (@$mids) {
512                         my ($num, undef) = $over->num_mid0_for_oid($oid, $mid);
513                         return $num if defined $num;
514                 }
515                 $mm->num_for($mids->[0]) // $mm->mid_insert($mids->[0]);
516         } else {
517                 # fallback to num_for since filters like RubyLang set the number
518                 $mm->mid_insert($mids->[0]) // $mm->num_for($mids->[0]);
519         }
520 }
521
522 # returns the number of bytes to add if given a non-CRLF arg
523 sub crlf_adjust ($) {
524         if (index($_[0], "\r\n") < 0) {
525                 # common case is LF-only, every \n needs an \r;
526                 # so favor a cheap tr// over an expensive m//g
527                 $_[0] =~ tr/\n/\n/;
528         } else { # count number of '\n' w/o '\r', expensive:
529                 scalar(my @n = ($_[0] =~ m/(?<!\r)\n/g));
530         }
531 }
532
533 sub index_both { # git->cat_async callback
534         my ($bref, $oid, $type, $size, $sync) = @_;
535         my ($nr, $max) = @$sync{qw(nr max)};
536         ++$$nr;
537         $$max -= $size;
538         $size += crlf_adjust($$bref);
539         my $smsg = bless { bytes => $size, blob => $oid }, 'PublicInbox::Smsg';
540         my $self = $sync->{sidx};
541         my $eml = PublicInbox::Eml->new($bref);
542         $smsg->{num} = index_mm($self, $eml, $oid, $sync) or
543                 die "E: could not generate NNTP article number for $oid";
544         add_message($self, $eml, $smsg, $sync);
545 }
546
547 sub unindex_both { # git->cat_async callback
548         my ($bref, $oid, $type, $size, $self) = @_;
549         unindex_eml($self, $oid, PublicInbox::Eml->new($bref));
550 }
551
552 # called by public-inbox-index
553 sub index_sync {
554         my ($self, $opt) = @_;
555         delete $self->{lock_path} if $opt->{-skip_lock};
556         $self->{ibx}->with_umask(\&_index_sync, $self, $opt);
557         if ($opt->{reindex}) {
558                 my %again = %$opt;
559                 delete @again{qw(rethread reindex)};
560                 index_sync($self, \%again);
561         }
562 }
563
564 sub check_size { # check_async cb for -index --max-size=...
565         my ($oid, $type, $size, $arg, $git) = @_;
566         (($type // '') eq 'blob') or die "E: bad $oid in $git->{git_dir}";
567         if ($size <= $arg->{max_size}) {
568                 $git->cat_async($oid, $arg->{index_oid}, $arg);
569         } else {
570                 warn "W: skipping $oid ($size > $arg->{max_size})\n";
571         }
572 }
573
574 sub v1_checkpoint ($$;$) {
575         my ($self, $sync, $stk) = @_;
576         $self->{ibx}->git->check_async_wait;
577         $self->{ibx}->git->cat_async_wait;
578
579         # latest_cmt may be undef
580         my $newest = $stk ? $stk->{latest_cmt} : undef;
581         if ($newest) {
582                 my $cur = $self->{mm}->last_commit || '';
583                 if (need_update($self, $cur, $newest)) {
584                         $self->{mm}->last_commit($newest);
585                 }
586         } else {
587                 ${$sync->{max}} = $self->{batch_bytes};
588         }
589
590         $self->{mm}->{dbh}->commit;
591         if ($newest && need_xapian($self)) {
592                 my $xdb = $self->{xdb};
593                 my $cur = $xdb->get_metadata('last_commit');
594                 if (need_update($self, $cur, $newest)) {
595                         $xdb->set_metadata('last_commit', $newest);
596                 }
597
598                 # let SearchView know a full --reindex was done so it can
599                 # generate ->has_threadid-dependent links
600                 if ($sync->{reindex} && !ref($sync->{reindex})) {
601                         my $n = $xdb->get_metadata('has_threadid');
602                         $xdb->set_metadata('has_threadid', '1') if $n ne '1';
603                 }
604         }
605
606         $self->{over}->rethread_done($sync->{-opt}) if $newest; # all done
607         commit_txn_lazy($self);
608         $self->{ibx}->git->cleanup;
609         my $nr = ${$sync->{nr}};
610         idx_release($self, $nr);
611         # let another process do some work...
612         if (my $pr = $sync->{-opt}->{-progress}) {
613                 $pr->("indexed $nr/$sync->{ntodo}\n") if $nr;
614         }
615         if (!$stk) { # more to come
616                 begin_txn_lazy($self);
617                 $self->{mm}->{dbh}->begin_work;
618         }
619 }
620
621 # only for v1
622 sub process_stack {
623         my ($self, $sync, $stk) = @_;
624         my $git = $self->{ibx}->git;
625         my $max = $self->{batch_bytes};
626         my $nr = 0;
627         $sync->{nr} = \$nr;
628         $sync->{max} = \$max;
629         $sync->{sidx} = $self;
630
631         $self->{mm}->{dbh}->begin_work;
632         if (my @leftovers = keys %{delete($sync->{D}) // {}}) {
633                 warn('W: unindexing '.scalar(@leftovers)." leftovers\n");
634                 for my $oid (@leftovers) {
635                         $oid = unpack('H*', $oid);
636                         $git->cat_async($oid, \&unindex_both, $self);
637                 }
638         }
639         if ($sync->{max_size} = $sync->{-opt}->{max_size}) {
640                 $sync->{index_oid} = \&index_both;
641         }
642         while (my ($f, $at, $ct, $oid) = $stk->pop_rec) {
643                 if ($f eq 'm') {
644                         my $arg = { %$sync, autime => $at, cotime => $ct };
645                         if ($sync->{max_size}) {
646                                 $git->check_async($oid, \&check_size, $arg);
647                         } else {
648                                 $git->cat_async($oid, \&index_both, $arg);
649                         }
650                         v1_checkpoint($self, $sync) if $max <= 0;
651                 } elsif ($f eq 'd') {
652                         $git->cat_async($oid, \&unindex_both, $self);
653                 }
654         }
655         v1_checkpoint($self, $sync, $stk);
656 }
657
658 sub log2stack ($$$$) {
659         my ($sync, $git, $range, $ibx) = @_;
660         my $D = $sync->{D}; # OID_BIN => NR (if reindexing, undef otherwise)
661         my ($add, $del);
662         if ($ibx->version == 1) {
663                 my $path = $hex.'{2}/'.$hex.'{38}';
664                 $add = qr!\A:000000 100644 \S+ ($OID) A\t$path$!;
665                 $del = qr!\A:100644 000000 ($OID) \S+ D\t$path$!;
666         } else {
667                 $del = qr!\A:\d{6} 100644 $OID ($OID) [AM]\td$!;
668                 $add = qr!\A:\d{6} 100644 $OID ($OID) [AM]\tm$!;
669         }
670
671         # Count the new files so they can be added newest to oldest
672         # and still have numbers increasing from oldest to newest
673         my $fh = $git->popen(qw(log --raw -r --pretty=tformat:%at-%ct-%H
674                                 --no-notes --no-color --no-renames --no-abbrev),
675                                 $range);
676         my ($at, $ct, $stk);
677         while (<$fh>) {
678                 if (/\A([0-9]+)-([0-9]+)-($OID)$/o) {
679                         ($at, $ct) = ($1 + 0, $2 + 0);
680                         $stk //= PublicInbox::IdxStack->new($3);
681                 } elsif (/$del/) {
682                         my $oid = $1;
683                         if ($D) { # reindex case
684                                 $D->{pack('H*', $oid)}++;
685                         } else { # non-reindex case:
686                                 $stk->push_rec('d', $at, $ct, $oid);
687                         }
688                 } elsif (/$add/) {
689                         my $oid = $1;
690                         if ($D) {
691                                 my $oid_bin = pack('H*', $oid);
692                                 my $nr = --$D->{$oid_bin};
693                                 delete($D->{$oid_bin}) if $nr <= 0;
694
695                                 # nr < 0 (-1) means it never existed
696                                 $stk->push_rec('m', $at, $ct, $oid) if $nr < 0;
697                         } else {
698                                 $stk->push_rec('m', $at, $ct, $oid);
699                         }
700                 }
701         }
702         close $fh or die "git log failed: \$?=$?";
703         $stk //= PublicInbox::IdxStack->new;
704         $stk->read_prepare;
705 }
706
707 sub prepare_stack ($$$) {
708         my ($self, $sync, $range) = @_;
709         my $git = $self->{ibx}->git;
710
711         if (index($range, '..') < 0) {
712                 # don't show annoying git errors to users who run -index
713                 # on empty inboxes
714                 $git->qx(qw(rev-parse -q --verify), "$range^0");
715                 return PublicInbox::IdxStack->new->read_prepare if $?;
716         }
717         $sync->{D} = $sync->{reindex} ? {} : undef; # OID_BIN => NR
718         log2stack($sync, $git, $range, $self->{ibx});
719 }
720
721 # --is-ancestor requires git 1.8.0+
722 sub is_ancestor ($$$) {
723         my ($git, $cur, $tip) = @_;
724         return 0 unless $git->check($cur);
725         my $cmd = [ 'git', "--git-dir=$git->{git_dir}",
726                 qw(merge-base --is-ancestor), $cur, $tip ];
727         my $pid = spawn($cmd);
728         waitpid($pid, 0) == $pid or die join(' ', @$cmd) .' did not finish';
729         $? == 0;
730 }
731
732 sub need_update ($$$) {
733         my ($self, $cur, $new) = @_;
734         my $git = $self->{ibx}->git;
735         return 1 if $cur && !is_ancestor($git, $cur, $new);
736         my $range = $cur eq '' ? $new : "$cur..$new";
737         chomp(my $n = $git->qx(qw(rev-list --count), $range));
738         ($n eq '' || $n > 0);
739 }
740
741 # The last git commit we indexed with Xapian or SQLite (msgmap)
742 # This needs to account for cases where Xapian or SQLite is
743 # out-of-date with respect to the other.
744 sub _last_x_commit {
745         my ($self, $mm) = @_;
746         my $lm = $mm->last_commit || '';
747         my $lx = '';
748         if (need_xapian($self)) {
749                 $lx = $self->{xdb}->get_metadata('last_commit') || '';
750         } else {
751                 $lx = $lm;
752         }
753         # Use last_commit from msgmap if it is older or unset
754         if (!$lm || ($lx && $lm && is_ancestor($self->{ibx}->git, $lm, $lx))) {
755                 $lx = $lm;
756         }
757         $lx;
758 }
759
760 sub reindex_from ($$) {
761         my ($reindex, $last_commit) = @_;
762         return $last_commit unless $reindex;
763         ref($reindex) eq 'HASH' ? $reindex->{from} : '';
764 }
765
766 # indexes all unindexed messages (v1 only)
767 sub _index_sync {
768         my ($self, $opt) = @_;
769         my $tip = $opt->{ref} || 'HEAD';
770         my $git = $self->{ibx}->git;
771         $self->{batch_bytes} = $opt->{batch_size} // $BATCH_BYTES;
772         $git->batch_prepare;
773         my $pr = $opt->{-progress};
774         my $sync = { reindex => $opt->{reindex}, -opt => $opt };
775         my $xdb = $self->begin_txn_lazy;
776         $self->{over}->rethread_prepare($opt);
777         my $mm = _msgmap_init($self);
778         if ($sync->{reindex}) {
779                 my $last = $mm->last_commit;
780                 if ($last) {
781                         $tip = $last;
782                 } else {
783                         # somebody just blindly added --reindex when indexing
784                         # for the first time, allow it:
785                         undef $sync->{reindex};
786                 }
787         }
788         my $last_commit = _last_x_commit($self, $mm);
789         my $lx = reindex_from($sync->{reindex}, $last_commit);
790         my $range = $lx eq '' ? $tip : "$lx..$tip";
791         $pr->("counting changes\n\t$range ... ") if $pr;
792         my $stk = prepare_stack($self, $sync, $range);
793         $sync->{ntodo} = $stk ? $stk->num_records : 0;
794         $pr->("$sync->{ntodo}\n") if $pr; # continue previous line
795         process_stack($self, $sync, $stk);
796 }
797
798 sub DESTROY {
799         # order matters for unlocking
800         $_[0]->{xdb} = undef;
801         $_[0]->{lockfh} = undef;
802 }
803
804 sub _begin_txn {
805         my ($self) = @_;
806         my $xdb = $self->{xdb} || idx_acquire($self);
807         $self->{over}->begin_lazy if $self->{over};
808         $xdb->begin_transaction if $xdb;
809         $self->{txn} = 1;
810         $xdb;
811 }
812
813 sub begin_txn_lazy {
814         my ($self) = @_;
815         $self->{ibx}->with_umask(\&_begin_txn, $self) if !$self->{txn};
816 }
817
818 # store 'indexlevel=medium' in v2 shard=0 and v1 (only one shard)
819 # This metadata is read by Admin::detect_indexlevel:
820 sub set_metadata_once {
821         my ($self) = @_;
822
823         return if $self->{shard}; # only continue if undef or 0, not >0
824         my $xdb = $self->{xdb};
825
826         if (delete($self->{-set_has_threadid_once})) {
827                 $xdb->set_metadata('has_threadid', '1');
828         }
829         if (delete($self->{-set_indexlevel_once})) {
830                 my $level = $xdb->get_metadata('indexlevel');
831                 if (!$level || $level ne 'medium') {
832                         $xdb->set_metadata('indexlevel', 'medium');
833                 }
834         }
835         if (delete($self->{-set_skip_docdata_once})) {
836                 $xdb->get_metadata('skip_docdata') or
837                         $xdb->set_metadata('skip_docdata', '1');
838         }
839 }
840
841 sub _commit_txn {
842         my ($self) = @_;
843         if (my $xdb = $self->{xdb}) {
844                 set_metadata_once($self);
845                 $xdb->commit_transaction;
846         }
847         $self->{over}->commit_lazy if $self->{over};
848 }
849
850 sub commit_txn_lazy {
851         my ($self) = @_;
852         delete($self->{txn}) and
853                 $self->{ibx}->with_umask(\&_commit_txn, $self);
854 }
855
856 sub worker_done {
857         my ($self) = @_;
858         if (need_xapian($self)) {
859                 die "$$ $0 xdb not released\n" if $self->{xdb};
860         }
861         die "$$ $0 still in transaction\n" if $self->{txn};
862 }
863
864 1;