[PATCH] lei q|up: support v2:/path/to/inboxdir destination

[PATCH] lei q|up: support v2:/path/to/inboxdir destination

[Eric Wong]

This allows "lei-managed pseudo mailing lists" as described
by Konstantin.

Alternates use is optional and can be enables via --shared.

This doesn't manage or edit ~/.public-inbox/config; presumably
there'll need to be some tweaking of search parameters before
finalizing and making the inbox publicly accessible via HTTP/NNTP.

Link: https://public-inbox.org/meta/20210426164454.5zd5kgugfhfwfkpo@nitro.local/T/
---
 lib/PublicInbox/LEI.pm            |  2 +-
 lib/PublicInbox/LeiOverview.pm    |  2 ++
 lib/PublicInbox/LeiSavedSearch.pm |  8 ++++-
 lib/PublicInbox/LeiToMail.pm      | 59 +++++++++++++++++++++++++++++--
 lib/PublicInbox/LeiXSearch.pm     |  1 +
 lib/PublicInbox/V2Writable.pm     | 11 +++++-
 t/lei-q-save.t                    | 30 ++++++++++++++++
 7 files changed, 107 insertions(+), 6 deletions(-)

diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index e5ff9e5d..f2dfc320 100644
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -163,7 +163,7 @@ our %CMD = ( # sorted in order of importance/use:
 	qw(save output|mfolder|o=s format|f=s dedupe|d=s threads|t+
 	sort|s=s reverse|r offset=i pretty jobs|j=s globoff|g augment|a
 	import-before! lock=s@ rsyncable alert=s@ mua=s verbose|v+
-	color! mail-sync!), @c_opt, opt_dash('limit|n=i', '[0-9]+') ],
+	shared color! mail-sync!), @c_opt, opt_dash('limit|n=i', '[0-9]+') ],
 
 'up' => [ 'OUTPUT|--all', 'update saved search',
 	qw(jobs|j=s lock=s@ alert=s@ mua=s verbose|v+ all:s), @c_opt ],
diff --git a/lib/PublicInbox/LeiOverview.pm b/lib/PublicInbox/LeiOverview.pm
index 28891460..e4242d9b 100644
--- a/lib/PublicInbox/LeiOverview.pm
+++ b/lib/PublicInbox/LeiOverview.pm
@@ -108,6 +108,8 @@ sub new {
 			$opt->{alert} //= [ ':WINCH,:bell' ] if -t $lei->{1};
 		}
 	}
+	return $lei->fail('--shared is only for v2 inbox output') if
+		$self->{fmt} ne 'v2' && $lei->{opt}->{shared};
 	$self;
 }
 
diff --git a/lib/PublicInbox/LeiSavedSearch.pm b/lib/PublicInbox/LeiSavedSearch.pm
index 48d252f1..929380ed 100644
--- a/lib/PublicInbox/LeiSavedSearch.pm
+++ b/lib/PublicInbox/LeiSavedSearch.pm
@@ -14,7 +14,7 @@ use PublicInbox::Spawn qw(run_die);
 use PublicInbox::ContentHash qw(git_sha);
 use PublicInbox::MID qw(mids_for_index);
 use Digest::SHA qw(sha256_hex);
-my $LOCAL_PFX = qr!\A(?:maildir|mh|mbox.+|mmdf):!i; # TODO: put in LeiToMail?
+my $LOCAL_PFX = qr!\A(?:maildir|mh|mbox.+|mmdf|v2):!i; # TODO: put in LeiToMail?
 
 # move this to PublicInbox::Config if other things use it:
 my %cquote = ("\n" => '\\n', "\t" => '\\t', "\b" => '\\b');
@@ -290,6 +290,12 @@ EOM
 	my $dir_old = lss_dir_for($lei, \$old_path, 1);
 	my $dir_new = lss_dir_for($lei, \$new_path);
 	return if $dir_new eq $dir_old; # no change, likely
+
+	($old_out =~ m!\Av2:!i || $new_out =~ m!\Av2:!) and
+		return $lei->fail(<<EOM);
+conversions from/to v2 inboxes not supported at this time
+EOM
+
 	return $lei->fail(<<EOM) if -e $dir_new;
 lei.q.output changed from `$old_out' to `$new_out'
 However, $dir_new exists
diff --git a/lib/PublicInbox/LeiToMail.pm b/lib/PublicInbox/LeiToMail.pm
index b3aec50b..a7382169 100644
--- a/lib/PublicInbox/LeiToMail.pm
+++ b/lib/PublicInbox/LeiToMail.pm
@@ -353,16 +353,28 @@ sub _text_write_cb ($$) {
 	sub { # for git_to_mail
 		my ($bref, $smsg, $eml) = @_;
 		$lse->xsmsg_vmd($smsg) if $lse;
-		$eml //= PublicInbox::Eml->new($bref); # copy bref
+		$eml //= PublicInbox::Eml->new($bref);
 		return if $dedupe && $dedupe->is_dup($eml, $smsg);
 		my $lk = $ovv->lock_for_scope;
 		$lei->out(${$lvt->eml_to_text($smsg, $eml)}, "\n");
 	}
 }
 
+sub _v2_write_cb ($$) {
+	my ($self, $lei) = @_;
+	my $dedupe = $lei->{dedupe};
+	$dedupe->prepare_dedupe if $dedupe;
+	sub { # for git_to_mail
+		my ($bref, $smsg, $eml) = @_;
+		$eml //= PublicInbox::Eml->new($bref);
+		return if $dedupe && $dedupe->is_dup($eml, $smsg);
+		$lei->{v2w}->ipc_do('add', $eml); # V2Writable->add
+	}
+}
+
 sub write_cb { # returns a callback for git_to_mail
 	my ($self, $lei) = @_;
-	# _mbox_write_cb, _maildir_write_cb or _imap_write_cb
+	# _mbox_write_cb, _maildir_write_cb, _imap_write_cb, _v2_write_cb
 	my $m = "_$self->{base_type}_write_cb";
 	$self->$m($lei);
 }
@@ -400,6 +412,13 @@ sub new {
 		require PublicInbox::LeiViewText;
 		$lei->{lvt} = PublicInbox::LeiViewText->new($lei);
 		$self->{base_type} = 'text';
+	} elsif ($fmt eq 'v2') {
+		die "--dedupe=oid and v2 are incompatible\n" if
+			($lei->{opt}->{dedupe}//'') eq 'oid';
+		$self->{base_type} = 'v2';
+		$lei->{opt}->{save} = \1;
+		die "--mua incompatible with v2\n" if $lei->{opt}->{mua};
+		$dst = $lei->{ovv}->{dst} = $lei->abs_path($dst);
 	} else {
 		die "bad mail --format=$fmt\n";
 	}
@@ -599,9 +618,43 @@ sub _do_augment_mbox {
 	$dedupe->pause_dedupe if $dedupe;
 }
 
+sub _pre_augment_v2 {
+	my ($self, $lei) = @_;
+	my $dir = $self->{dst};
+	require PublicInbox::InboxWritable;
+	my ($ibx, @creat);
+	if (-d $dir) {
+		my $opt = { -min_inbox_version => 2 };
+		require PublicInbox::Admin;
+		my @ibx = PublicInbox::Admin::resolve_inboxes([ $dir ], $opt);
+		$ibx = $ibx[0] or die "$dir is not a v2 inbox\n";
+	} else {
+		$creat[0] = {};
+		$ibx = PublicInbox::Inbox->new({
+			name => 'lei-result', # XXX configurable
+			inboxdir => $dir,
+			version => 2,
+			address => [ 'lei@example.com' ],
+		});
+	}
+	PublicInbox::InboxWritable->new($ibx, @creat);
+	$ibx->init_inbox if @creat;
+	my $v2w = $lei->{v2w} = $ibx->importer;
+	$v2w->ipc_lock_init("$dir/ipc.lock");
+	$v2w->ipc_worker_spawn("lei/v2w $dir", $lei->oldset, { lei => $lei });
+	return if !$lei->{opt}->{shared};
+	my $d = "$lei->{ale}->{git}->{git_dir}/objects";
+	my $al = "$dir/git/0.git/objects/info/alternates";
+	open my $fh, '+>>', $al or die "open($al): $!";
+	seek($fh, 0, SEEK_SET) or die "seek($al): $!";
+	grep(/\A\Q$d\E\n/, <$fh>) and return;
+	print $fh "$d\n" or die "print($al): $!";
+	close $fh or die "close($al): $!";
+}
+
 sub pre_augment { # fast (1 disk seek), runs in same process as post_augment
 	my ($self, $lei) = @_;
-	# _pre_augment_maildir, _pre_augment_mbox
+	# _pre_augment_maildir, _pre_augment_mbox, _pre_augment_v2
 	my $m = $self->can("_pre_augment_$self->{base_type}") or return;
 	$m->($self, $lei);
 }
diff --git a/lib/PublicInbox/LeiXSearch.pm b/lib/PublicInbox/LeiXSearch.pm
index 2e548a7a..d6d42a01 100644
--- a/lib/PublicInbox/LeiXSearch.pm
+++ b/lib/PublicInbox/LeiXSearch.pm
@@ -368,6 +368,7 @@ sub query_done { # EOF callback for main daemon
 		warn "BUG: {sto} missing with --mail-sync";
 	}
 	my $wait = $lei->{sto} ? $lei->{sto}->ipc_do('done') : undef;
+	$wait = $lei->{v2w} ? $lei->{v2w}->ipc_do('done') : undef;
 	$lei->{ovv}->ovv_end($lei);
 	my $start_mua;
 	if ($l2m) { # close() calls LeiToMail reap_compress
diff --git a/lib/PublicInbox/V2Writable.pm b/lib/PublicInbox/V2Writable.pm
index 0461257f..573d9c6f 100644
--- a/lib/PublicInbox/V2Writable.pm
+++ b/lib/PublicInbox/V2Writable.pm
@@ -6,7 +6,7 @@
 package PublicInbox::V2Writable;
 use strict;
 use v5.10.1;
-use parent qw(PublicInbox::Lock);
+use parent qw(PublicInbox::Lock PublicInbox::IPC);
 use PublicInbox::SearchIdxShard;
 use PublicInbox::IPC;
 use PublicInbox::Eml;
@@ -1431,4 +1431,13 @@ W: interrupted, --xapian-only --reindex required upon restart
 EOF
 }
 
+sub ipc_atfork_child {
+	my ($self) = @_;
+	if (my $lei = delete $self->{lei}) {
+		$lei->_lei_atfork_child;
+		close(delete $lei->{pkt_op_p});
+	}
+	$self->SUPER::ipc_atfork_child;
+}
+
 1;
diff --git a/t/lei-q-save.t b/t/lei-q-save.t
index bea65133..694b33b2 100644
--- a/t/lei-q-save.t
+++ b/t/lei-q-save.t
@@ -3,6 +3,8 @@
 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
 use strict; use v5.10.1; use PublicInbox::TestCommon;
 use PublicInbox::Smsg;
+use List::Util qw(sum);
+
 my $doc1 = eml_load('t/plack-qp.eml');
 $doc1->header_set('Date', PublicInbox::Smsg::date({ds => time - (86400 * 5)}));
 my $doc2 = eml_load('t/utf8.eml');
@@ -165,5 +167,33 @@ test_lei(sub {
 			skip "symlinks not supported in $home?: $!", 1;
 		lei_ok('up', "$home/ln -s");
 	};
+
+	my $v2 = "$home/v2"; # v2: as an output destination
+	my (@before, @after);
+	require PublicInbox::MboxReader;
+	lei_ok(qw(q z:0.. -o), "v2:$v2");
+	lei_ok(qw(q z:0.. -o), "mboxrd:$home/before", '--only', $v2, '-j1,1');
+	open my $fh, '<', "$home/before";
+	PublicInbox::MboxReader->mboxrd($fh, sub { push @before, $_[0] });
+	isnt(scalar(@before), 0, 'initial v2 written');
+	my $orig = sum(map { -f $_ ? -s _ : () } (
+			glob("$v2/git/0.git/objects/*/*")));
+	lei_ok(qw(import t/data/0001.patch));
+	lei_ok 'up', $v2;
+	lei_ok(qw(q z:0.. -o), "mboxrd:$home/after", '--only', $v2, '-j1,1');
+	open $fh, '<', "$home/after";
+	PublicInbox::MboxReader->mboxrd($fh, sub { push @after, $_[0] });
+
+	my $last = shift @after;
+	$last->header_set('Status');
+	is_deeply($last, eml_load('t/data/0001.patch'), 'lei up worked on v2');
+	is_deeply(\@before, \@after, 'got same results');
+
+	my $v2s = "$home/v2s";
+	lei_ok(qw(q --shared z:0.. -o), "v2:$v2s");
+	my $shared = sum(map { -f $_ ? -s _ : () } (
+			glob("$v2s/git/0.git/objects/*/*")));
+	ok($shared < $orig, 'fewer bytes stored with --shared') or
+		diag "shared=$shared orig=$orig";
 });
 done_testing;