}
} elsif ($self->{fmt} =~ /\A(concat)?json\z/ && $lei->{opt}->{pretty}) {
my $EOR = ($1//'') eq 'concat' ? "\n}" : "\n},";
+ my $lse = $lei->{sto}->search;
sub { # DIY prettiness :P
my ($smsg, $mitem) = @_;
return if $dedupe->is_smsg_dup($smsg);
+ $lse->xsmsg_vmd($smsg);
$smsg = _unbless_smsg($smsg, $mitem);
$buf .= "{\n";
$buf .= join(",\n", map {
}
} elsif ($json) {
my $ORS = $self->{fmt} eq 'json' ? ",\n" : "\n"; # JSONL
+ my $lse = $lei->{sto}->search;
sub {
my ($smsg, $mitem) = @_;
return if $dedupe->is_smsg_dup($smsg);
+ $lse->xsmsg_vmd($smsg);
$buf .= $json->encode(_unbless_smsg(@_)) . $ORS;
return if length($buf) < 65536;
my $lk = $self->lock_for_scope;
wantarray ? sort(keys(%$kw)) : $kw;
}
+sub xsmsg_vmd {
+ my ($self, $smsg) = @_;
+ return if $smsg->{kw};
+ my $xdb = $self->xdb; # set {nshard};
+ my %kw;
+ $kw{flagged} = 1 if delete($smsg->{lei_q_tt_flagged});
+ my @num = $self->over->blob_exists($smsg->{blob});
+ for my $num (@num) { # there should only be one...
+ my $kw = xap_terms('K', $xdb, num2docid($self, $num));
+ %kw = (%kw, %$kw);
+ }
+ $smsg->{kw} = [ sort keys %kw ] if scalar(keys(%kw));
+}
+
# when a message has no Message-IDs at all, this is needed for
# unsent Draft messages, at least
sub content_key ($) {
}
sub _cmp_1st { # git->cat_async callback
- my ($bref, $oid, $type, $size, $cmp) = @_; # cmp: [chash, found, smsg]
- if (content_hash(PublicInbox::Eml->new($bref)) eq $cmp->[0]) {
+ my ($bref, $oid, $type, $size, $cmp) = @_; # cmp: [chash, xoids, smsg]
+ if ($bref && content_hash(PublicInbox::Eml->new($bref)) eq $cmp->[0]) {
$cmp->[1]->{$oid} = $cmp->[2]->{num};
}
}
-sub xids_for { # returns { OID => docid } mapping for $eml matches
+sub xoids_for { # returns { OID => docid } mapping for $eml matches
my ($self, $eml, $min) = @_;
my ($chash, $mids) = content_key($eml);
my @overs = ($self->over // $self->overs_all);
my $git = $self->git;
- my $found = {};
+ my $xoids = {};
for my $mid (@$mids) {
for my $o (@overs) {
my ($id, $prev);
while (my $cur = $o->next_by_mid($mid, \$id, \$prev)) {
- next if $found->{$cur->{blob}};
+ next if $cur->{bytes} == 0 ||
+ $xoids->{$cur->{blob}};
$git->cat_async($cur->{blob}, \&_cmp_1st,
- [ $chash, $found, $cur ]);
- if ($min && scalar(keys %$found) >= $min) {
+ [ $chash, $xoids, $cur ]);
+ if ($min && scalar(keys %$xoids) >= $min) {
$git->cat_async_wait;
- return $found;
+ return $xoids;
}
}
}
}
$git->cat_async_wait;
- scalar(keys %$found) ? $found : undef;
+ scalar(keys %$xoids) ? $xoids : undef;
}
# returns true if $eml is indexed by lei/store and keywords don't match
sub kw_changed {
my ($self, $eml, $new_kw_sorted) = @_;
- my $found = xids_for($self, $eml, 1) // return;
- my ($num) = values %$found;
+ my $xoids = xoids_for($self, $eml, 1) // return;
+ my ($num) = values %$xoids;
my @cur_kw = msg_keywords($self, $num);
join("\0", @$new_kw_sorted) eq join("\0", @cur_kw) ? 0 : 1;
}
for my $mid (@$mids) {
my ($id, $prev);
while (my $cur = $oidx->next_by_mid($mid, \$id, \$prev)) {
+ next if $cur->{bytes} == 0; # external-only message
my $oid = $cur->{blob};
my $docid = $cur->{num};
my $bref = $im ? $im->cat_blob($oid) : undef;
my ($self, $eml, $vmd) = @_;
my $im = $self->importer; # may create new epoch
my $eidx = eidx_init($self); # writes ALL.git/objects/info/alternates
- my $oidx = $eidx->{oidx};
+ my $oidx = $eidx->{oidx}; # PublicInbox::Import::add checks this
my $smsg = bless { -oidx => $oidx }, 'PublicInbox::Smsg';
$im->add($eml, undef, $smsg) or return; # duplicate returns undef
add_eml($self, $eml, $vmd) // set_eml_vmd($self, $eml, $vmd);
}
-sub add_eml_maybe {
- my ($self, $eml) = @_;
- my $lxs = $self->{lxs_all_local} // die 'BUG: no {lxs_all_local}';
- return if $lxs->xids_for($eml, 1);
- add_eml($self, $eml);
-}
-
# set or update keywords for external message, called via ipc_do
-sub set_xkw {
- my ($self, $eml, $kw) = @_;
- my $lxs = $self->{lxs_all_local} // die 'BUG: no {lxs_all_local}';
- if ($lxs->xids_for($eml, 1)) { # is it in a local external?
- # TODO: index keywords only
- } else {
- set_eml($self, $eml, { kw => $kw });
+sub set_xvmd {
+ my ($self, $xoids, $eml, $vmd) = @_;
+
+ my $eidx = eidx_init($self);
+ my $oidx = $eidx->{oidx};
+
+ # see if we can just update existing docs
+ for my $oid (keys %$xoids) {
+ my @docids = $oidx->blob_exists($oid) or next;
+ scalar(@docids) > 1 and
+ warn "W: $oid indexed as multiple docids: @docids\n";
+ for my $docid (@docids) {
+ my $idx = $eidx->idx_shard($docid);
+ $idx->ipc_do('set_vmd', $docid, $vmd);
+ }
+ delete $xoids->{$oid}; # all done with this oid
}
+ return unless scalar(keys(%$xoids));
+
+ # see if it was indexed, but with different OID(s)
+ if (my @docids = _docids_for($self, $eml)) {
+ for my $docid (@docids) {
+ for my $oid (keys %$xoids) {
+ $oidx->add_xref3($docid, -1, $oid, '.');
+ }
+ my $idx = $eidx->idx_shard($docid);
+ $idx->ipc_do('set_vmd', $docid, $vmd);
+ }
+ return;
+ }
+ # totally unseen
+ my $smsg = bless { blob => '' }, 'PublicInbox::Smsg';
+ $smsg->{num} = $oidx->adj_counter('eidx_docid', '+');
+ # save space for an externals-only message
+ my $hdr = $eml->header_obj;
+ $smsg->populate($hdr); # sets lines == 0
+ $smsg->{bytes} = 0;
+ delete @$smsg{qw(From Subject)};
+ $smsg->{to} = $smsg->{cc} = $smsg->{from} = '';
+ $oidx->add_overview($hdr, $smsg); # subject+references for threading
+ $smsg->{subject} = '';
+ for my $oid (keys %$xoids) {
+ $oidx->add_xref3($smsg->{num}, -1, $oid, '.');
+ }
+ my $idx = $eidx->idx_shard($smsg->{num});
+ $idx->index_eml(PublicInbox::Eml->new("\n\n"), $smsg);
+ $idx->ipc_do('add_vmd', $smsg->{num}, $vmd);
}
sub checkpoint {
$self->SUPER::ipc_atfork_child;
}
-sub refresh_local_externals {
- my ($self) = @_;
- my $cfg = $self->{lei}->_lei_cfg or return;
- my $cur_cfg = $self->{cur_cfg} // -1;
- my $lxs = $self->{lxs_all_local};
- if ($cfg != $cur_cfg || !$lxs) {
- $lxs = PublicInbox::LeiXSearch->new;
- my @loc = $self->{lei}->externals_each;
- for my $loc (@loc) { # locals only
- $lxs->prepare_external($loc) if -d $loc;
- }
- $self->{lei}->ale->refresh_externals($lxs);
- $lxs->{git} = $self->{lei}->ale->git;
- $self->{lxs_all_local} = $lxs;
- $self->{cur_cfg} = $cfg;
- }
-}
-
sub write_prepare {
my ($self, $lei) = @_;
unless ($self->{-ipc_req}) {
- require PublicInbox::LeiXSearch;
$self->ipc_lock_init($lei->store_path . '/ipc.lock');
# Mail we import into lei are private, so headers filtered out
# by -mda for public mail are not appropriate
$self->ipc_worker_spawn('lei_store', $lei->oldset,
{ lei => $lei });
}
- my $wait = $self->ipc_do('refresh_local_externals');
$lei->{sto} = $self;
}
use PublicInbox::ProcessPipe;
use PublicInbox::Spawn qw(which spawn popen_rd);
use PublicInbox::LeiDedupe;
+use PublicInbox::Git;
use PublicInbox::GitAsyncCat;
use PublicInbox::PktOp qw(pkt_do);
use Symbol qw(gensym);
my $atomic_append = !defined($ovv->{lock_path});
my $dedupe = $lei->{dedupe};
$dedupe->prepare_dedupe;
+ my $lse = $lei->{sto} ? $lei->{sto}->search : undef;
sub { # for git_to_mail
my ($buf, $smsg, $eml) = @_;
$eml //= PublicInbox::Eml->new($buf);
return if $dedupe->is_dup($eml, $smsg->{blob});
+ $lse->xsmsg_vmd($smsg) if $lse;
$buf = $eml2mbox->($eml, $smsg);
return atomic_append($lei, $buf) if $atomic_append;
my $lk = $ovv->lock_for_scope;
my ($lei, $lse, $eml, $kw) = @_;
return unless $lse;
my $x = $lse->kw_changed($eml, $kw);
+ my $vmd = { kw => $kw };
if ($x) {
- $lei->{sto}->ipc_do('set_eml', $eml, { kw => $kw });
+ $lei->{sto}->ipc_do('set_eml', $eml, $vmd);
} elsif (!defined($x)) {
- $lei->{sto}->ipc_do('set_xkw', $eml, $kw);
+ if (my $xoids = $lei->{ale}->xoids_for($eml)) {
+ $lei->{sto}->ipc_do('set_xvmd', $xoids, $eml, $vmd);
+ } else {
+ $lei->{sto}->ipc_do('set_eml', $eml, $vmd);
+ }
}
}
my $dedupe = $lei->{dedupe};
$dedupe->prepare_dedupe if $dedupe;
my $dst = $lei->{ovv}->{dst};
+ my $lse = $lei->{sto} ? $lei->{sto}->search : undef;
sub { # for git_to_mail
my ($buf, $smsg, $eml) = @_;
$dst // return $lei->fail; # dst may be undef-ed in last run
$buf //= \($eml->as_string);
+ $lse->xsmsg_vmd($smsg) if $lse;
return _buf2maildir($dst, $buf, $smsg) if !$dedupe;
$eml //= PublicInbox::Eml->new($$buf); # copy buf
return if $dedupe->is_dup($eml, $smsg->{blob});
my $imap_append = $lei->{net}->can('imap_append');
my $mic = $lei->{net}->mic_get($self->{uri});
my $folder = $self->{uri}->mailbox;
+ my $lse = $lei->{sto} ? $lei->{sto}->search : undef;
sub { # for git_to_mail
my ($bref, $smsg, $eml) = @_;
$mic // return $lei->fail; # dst may be undef-ed in last run
$eml //= PublicInbox::Eml->new($$bref); # copy bref
return if $dedupe->is_dup($eml, $smsg->{blob});
}
+ $lse->xsmsg_vmd($smsg) if $lse;
eval { $imap_append->($mic, $folder, $bref, $smsg, $eml) };
if (my $err = $@) {
undef $mic;
my $num = int(($docid - 1) / $nshard) + 1;
my $ibx = $self->{shard2ibx}->[$shard];
my $smsg = $ibx->over->get_art($num);
+ return if $smsg->{bytes} == 0;
mitem_kw($smsg, $mitem) if $ibx->can('msg_keywords');
$smsg->{docid} = $docid;
$smsg;
sub over {}
-sub overs_all { # for xids_for
- my ($self) = @_;
- grep(defined, map { $_->over } locals($self))
-}
-
sub _mset_more ($$) {
my ($mset, $mo) = @_;
my $size = $mset->size;
my $mset;
my $each_smsg = $lei->{ovv}->ovv_each_smsg_cb($lei, $ibxish);
my $can_kw = !!$ibxish->can('msg_keywords');
- my $fl = $lei->{opt}->{threads} > 1 ? [ 'flagged' ] : undef;
+ my $fl = $lei->{opt}->{threads} > 1 ? 1 : undef;
do {
$mset = $srch->mset($mo->{qstr}, $mo);
mset_progress($lei, $desc, $mset->size,
while ($over->expand_thread($ctx)) {
for my $n (@{$ctx->{xids}}) {
my $smsg = $over->get_art($n) or next;
- wait_startq($lei);
my $mitem = delete $n2item{$smsg->{num}};
+ next if $smsg->{bytes} == 0;
+ wait_startq($lei); # wait for keyword updates
if ($mitem) {
if ($can_kw) {
mitem_kw($smsg, $mitem, $fl);
} elsif ($fl) {
- $smsg->{kw} = $fl;
+ $smsg->{lei_q_tt_flagged} = 1;
}
}
$each_smsg->($smsg, $mitem);
sub each_remote_eml { # callback for MboxReader->mboxrd
my ($eml, $self, $lei, $each_smsg) = @_;
- if (my $sto = $self->{import_sto}) {
- $sto->ipc_do('add_eml_maybe', $eml);
+ if ($self->{import_sto} && !$lei->{ale}->xoids_for($eml, 1)) {
+ $self->{import_sto}->ipc_do('add_eml', $eml);
}
my $smsg = bless {}, 'PublicInbox::Smsg';
$smsg->populate($eml);
package PublicInbox::Over;
use strict;
use v5.10.1;
-use DBI;
+use DBI qw(:sql_types); # SQL_BLOB
use DBD::SQLite;
use PublicInbox::Smsg;
use Compress::Zlib qw(uncompress);
}
}
+sub blob_exists {
+ my ($self, $oidhex) = @_;
+ if (wantarray) {
+ my $sth = $self->dbh->prepare_cached(<<'', undef, 1);
+SELECT docid FROM xref3 WHERE oidbin = ?
+
+ $sth->bind_param(1, pack('H*', $oidhex), SQL_BLOB);
+ $sth->execute;
+ my $tmp = $sth->fetchall_arrayref;
+ map { $_->[0] } @$tmp;
+ } else {
+ my $sth = $self->dbh->prepare_cached(<<'', undef, 1);
+SELECT COUNT(*) FROM xref3 WHERE oidbin = ?
+
+ $sth->bind_param(1, pack('H*', $oidhex), SQL_BLOB);
+ $sth->execute;
+ $sth->fetchrow_array;
+ }
+}
+
1;
}
-sub blob_exists {
- my ($self, $oidhex) = @_;
- my $sth = $self->dbh->prepare_cached(<<'', undef, 1);
-SELECT COUNT(*) FROM xref3 WHERE oidbin = ?
-
- $sth->bind_param(1, pack('H*', $oidhex), SQL_BLOB);
- $sth->execute;
- $sth->fetchrow_array;
-}
-
1;
begin_txn_lazy($self);
my $doc = _get_doc($self, $docid) or return;
term_generator($self)->set_document($doc);
+
+ # '.' is special for lei_store
$doc->add_boolean_term('O'.$eidx_key) if $eidx_key ne '.';
+
index_list_id($self, $doc, $eml);
$self->{xdb}->replace_document($docid, $doc);
}
is($str, "hi\n", '->new modified body like Email::Simple');
is($eml->body, "hi\n", '->body works');
is($eml->as_string, "a: b\n\nhi\n", '->as_string');
+ my $empty = PublicInbox::Eml->new("\n\n");
+ is($empty->as_string, "\n\n", 'empty message');
}
for my $cls (@classes) {
PublicInbox::MdirReader::maildir_each_eml("$d/md", sub {
push @md, $_[2];
});
- is(scalar(@md), scalar(@mboxrd), 'got expected emails in Maildir');
+ is(scalar(@md), scalar(@mboxrd), 'got expected emails in Maildir') or
+ diag $lei_err;
@md = sort { ${$a->{bdy}} cmp ${$b->{bdy}} } @md;
@mboxrd = sort { ${$a->{bdy}} cmp ${$b->{bdy}} } @mboxrd;
my @rd_nostatus = map {
lei_ok(qw(q -o), "mboxrd:/dev/stdout", qw(m:qp@example.com)) or
diag $lei_err;
like($lei_out, qr/^Status: OR\n/sm, 'Status set by previous augment');
-}
+} # /mbox + mbox.gz tests
-});
+my ($ro_home, $cfg_path) = setup_public_inboxes;
+
+# import keywords-only for external messages:
+$o = "$ENV{HOME}/kwdir";
+my $m = 'alpine.DEB.2.20.1608131214070.4924@example';
+my @inc = ('-I', "$ro_home/t1");
+lei_ok(qw(q -o), $o, "m:$m", @inc);
+
+# emulate MUA marking a Maildir message as read:
+@fn = glob("$o/cur/*");
+scalar(@fn) == 1 or BAIL_OUT "wrote multiple or zero files: ".explain(\@fn);
+rename($fn[0], "$fn[0]S") or BAIL_OUT "rename $!";
+
+lei_ok(qw(q -o), $o, 'bogus', \'clobber output dir to import keywords');
+@fn = glob("$o/cur/*");
+is_deeply(\@fn, [], 'output dir actually clobbered');
+lei_ok('q', "m:$m", @inc);
+my $res = json_utf8->decode($lei_out);
+is_deeply($res->[0]->{kw}, ['seen'], 'seen flag set for external message')
+ or diag explain($res);
+lei_ok('q', "m:$m", '--no-external');
+is_deeply($res = json_utf8->decode($lei_out), [ undef ],
+ 'external message not imported') or diag explain($res);
+
+$o = "$ENV{HOME}/kwmboxrd";
+lei_ok(qw(q -o), "mboxrd:$o", "m:$m", @inc);
+
+# emulate MUA marking mboxrd message as unread
+open my $fh, '<', $o or BAIL_OUT;
+my $s = do { local $/; <$fh> };
+$s =~ s/^Status: OR\n/Status: O\nX-Status: A\n/sm or
+ fail "failed to clear R flag in $s";
+open $fh, '>', $o or BAIL_OUT;
+print $fh $s or BAIL_OUT;
+close $fh or BAIL_OUT;
+
+lei_ok(qw(q -o), "mboxrd:$o", 'm:bogus', @inc,
+ \'clobber mbox to import keywords');
+lei_ok(qw(q -o), "mboxrd:$o", "m:$m", @inc);
+open $fh, '<', $o or BAIL_OUT;
+$s = do { local $/; <$fh> };
+like($s, qr/^Status: O\n/ms, 'seen keyword gone in mbox');
+like($s, qr/^X-Status: A\n/ms, 'answered flag set');
+
+}); # test_lei
done_testing;
'flagged set in direct hit');
lei_ok qw(q -tt m:testmessage@example.com --only), "$ro_home/t2";
$res = json_utf8->decode($lei_out);
- is_deeply($res->[0]->{kw}, [ 'flagged' ],
- 'flagged set on external with -tt');
+ is_deeply($res->[0]->{kw}, [ qw(flagged seen) ],
+ 'flagged set on external with -tt') or diag explain($res);
lei_ok qw(q -t m:testmessage@example.com --only), "$ro_home/t2";
$res = json_utf8->decode($lei_out);
- ok(!exists($res->[0]->{kw}), 'flagged not set on external with 1 -t');
+ is_deeply($res->[0]->{kw}, [ 'seen' ],
+ 'flagged not set on external with 1 -t') or diag explain($res);
});
done_testing;