[PATCH 2/2] lei: start implementing inotify Maildir support

[Eric Wong <e@80x24.org>]

This allows lei to automatically note keyword (message flag)
changes made to a Maildir and propagate it into lei/store:

	lei add-watch --state=tag-ro /path/to/Maildir

This doesn't persist across restarts, yet.  In the future,
it will be applied automatically to "lei q" output Maildirs
by default (with an option to disable it).

State values of tag-rw, index-<ro|rw>, import-<ro|rw> will all
be supported for Maildir.

This represents a fairly major internal change that's fairly
intrusive, but the whole daemon-oriented design was to
facilitate being able to automatically monitor (and propagate)
Maildir/IMAP flag changes.
---
 MANIFEST                         |   6 ++
 lib/PublicInbox/DirIdle.pm       |  17 +++++
 lib/PublicInbox/LEI.pm           | 116 ++++++++++++++++++++++++++---
 lib/PublicInbox/LeiAddWatch.pm   |  41 ++++++++++
 lib/PublicInbox/LeiInput.pm      |   1 +
 lib/PublicInbox/LeiLsWatch.pm    |  15 ++++
 lib/PublicInbox/LeiNoteEvent.pm  | 124 +++++++++++++++++++++++++++++++
 lib/PublicInbox/LeiSearch.pm     |   2 +-
 lib/PublicInbox/LeiSelfSocket.pm |  45 +++++++++++
 lib/PublicInbox/LeiWatch.pm      |  13 ++++
 t/lei-watch.t                    |  49 ++++++++++++
 11 files changed, 416 insertions(+), 13 deletions(-)
 create mode 100644 lib/PublicInbox/LeiAddWatch.pm
 create mode 100644 lib/PublicInbox/LeiLsWatch.pm
 create mode 100644 lib/PublicInbox/LeiNoteEvent.pm
 create mode 100644 lib/PublicInbox/LeiSelfSocket.pm
 create mode 100644 lib/PublicInbox/LeiWatch.pm
 create mode 100644 t/lei-watch.t

diff --git a/MANIFEST b/MANIFEST
index 146a32ab..1d79b7c9 100644
--- a/MANIFEST
+++ b/MANIFEST
@@ -198,6 +198,7 @@ lib/PublicInbox/Isearch.pm
 lib/PublicInbox/KQNotify.pm
 lib/PublicInbox/LEI.pm
 lib/PublicInbox/LeiALE.pm
+lib/PublicInbox/LeiAddWatch.pm
 lib/PublicInbox/LeiAuth.pm
 lib/PublicInbox/LeiBlob.pm
 lib/PublicInbox/LeiConvert.pm
@@ -220,8 +221,10 @@ lib/PublicInbox/LeiLsLabel.pm
 lib/PublicInbox/LeiLsMailSource.pm
 lib/PublicInbox/LeiLsMailSync.pm
 lib/PublicInbox/LeiLsSearch.pm
+lib/PublicInbox/LeiLsWatch.pm
 lib/PublicInbox/LeiMailSync.pm
 lib/PublicInbox/LeiMirror.pm
+lib/PublicInbox/LeiNoteEvent.pm
 lib/PublicInbox/LeiOverview.pm
 lib/PublicInbox/LeiP2q.pm
 lib/PublicInbox/LeiPmdir.pm
@@ -232,6 +235,7 @@ lib/PublicInbox/LeiRemote.pm
 lib/PublicInbox/LeiRm.pm
 lib/PublicInbox/LeiSavedSearch.pm
 lib/PublicInbox/LeiSearch.pm
+lib/PublicInbox/LeiSelfSocket.pm
 lib/PublicInbox/LeiStore.pm
 lib/PublicInbox/LeiStoreErr.pm
 lib/PublicInbox/LeiSucks.pm
@@ -239,6 +243,7 @@ lib/PublicInbox/LeiTag.pm
 lib/PublicInbox/LeiToMail.pm
 lib/PublicInbox/LeiUp.pm
 lib/PublicInbox/LeiViewText.pm
+lib/PublicInbox/LeiWatch.pm
 lib/PublicInbox/LeiXSearch.pm
 lib/PublicInbox/Linkify.pm
 lib/PublicInbox/Listener.pm
@@ -434,6 +439,7 @@ t/lei-q-save.t
 t/lei-q-thread.t
 t/lei-sigpipe.t
 t/lei-tag.t
+t/lei-watch.t
 t/lei.t
 t/lei_dedupe.t
 t/lei_external.t
diff --git a/lib/PublicInbox/DirIdle.pm b/lib/PublicInbox/DirIdle.pm
index 5142d005..7031e5fd 100644
--- a/lib/PublicInbox/DirIdle.pm
+++ b/lib/PublicInbox/DirIdle.pm
@@ -53,6 +53,23 @@ sub new {
 	$self;
 }
 
+sub add_watches {
+	my ($self, $dirs, $gone) = @_;
+	my $fl = $MAIL_IN | ($gone ? $MAIL_GONE : 0);
+	for my $d (@$dirs) {
+		$self->{inot}->watch($d, $fl);
+	}
+	PublicInbox::FakeInotify::poll_once($self) if !$ino_cls;
+}
+
+sub rm_watches {
+	my ($self, $dir) = @_;
+	my $inot = $self->{inot};
+	if (my $cb = $inot->can('rm_watches')) { # TODO for fake watchers
+		$cb->($inot, $dir);
+	}
+}
+
 sub event_step {
 	my ($self) = @_;
 	my $cb = $self->{cb};
diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index a9f5edae..b92d7512 100644
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -28,14 +28,15 @@ use Time::HiRes qw(stat); # ctime comparisons for config cache
 use File::Path qw(mkpath);
 use File::Spec;
 our $quit = \&CORE::exit;
-our ($current_lei, $errors_log, $listener, $oldset, $dir_idle);
-my ($recv_cmd, $send_cmd);
+our ($current_lei, $errors_log, $listener, $oldset, $dir_idle,
+	$recv_cmd, $send_cmd);
 my $GLP = Getopt::Long::Parser->new;
 $GLP->configure(qw(gnu_getopt no_ignore_case auto_abbrev));
 my $GLP_PASS = Getopt::Long::Parser->new;
 $GLP_PASS->configure(qw(gnu_getopt no_ignore_case auto_abbrev pass_through));
 
 our %PATH2CFG; # persistent for socket daemon
+our $MDIR2CFGPATH; # /path/to/maildir => { /path/to/config => undef }
 
 # TBD: this is a documentation mechanism to show a subcommand
 # (may) pass options through to another command:
@@ -232,10 +233,9 @@ our %CMD = ( # sorted in order of importance/use:
 	qw(exact! all jobs:i indexed), @c_opt ],
 
 'add-watch' => [ 'LOCATION', 'watch for new messages and flag changes',
-	qw(import! kw! interval=s recursive|r
-	exclude=s include=s), @c_opt ],
+	qw(poll-interval=s state=s recursive|r), @c_opt ],
 'ls-watch' => [ '[FILTER...]', 'list active watches with numbers and status',
-		qw(format|f=s z), @c_opt ],
+		qw(l z|0), @c_opt ],
 'pause-watch' => [ '[WATCH_NUMBER_OR_FILTER]', qw(all local remote), @c_opt ],
 'resume-watch' => [ '[WATCH_NUMBER_OR_FILTER]', qw(all local remote), @c_opt ],
 'forget-watch' => [ '{WATCH_NUMBER|--prune}', 'stop and forget a watch',
@@ -391,6 +391,7 @@ my %OPTDESC = (
 'format|f=s	ls-search' => ['OUT|json|jsonl|concatjson',
 			'listing output format' ],
 'l	ls-search' => 'long listing format',
+'l	ls-watch' => 'long listing format',
 'l	ls-mail-source' => 'long listing format',
 'url	ls-mail-source' => 'show full URL of newsgroup or IMAP folder',
 'format|f=s	ls-external' => $ls_format,
@@ -435,7 +436,7 @@ my %CONFIG_KEYS = (
 	'leistore.dir' => 'top-level storage location',
 );
 
-my @WQ_KEYS = qw(lxs l2m ikw pmd wq1); # internal workers
+my @WQ_KEYS = qw(lxs l2m ikw pmd wq1 lne); # internal workers
 
 sub _drop_wq {
 	my ($self) = @_;
@@ -538,7 +539,7 @@ sub _lei_atfork_child {
 		chdir '/' or die "chdir(/): $!";
 		close($_) for (grep(defined, delete @$self{qw(0 1 2 sock)}));
 		if (my $cfg = $self->{cfg}) {
-			delete $cfg->{-lei_store};
+			delete @$cfg{qw(-lei_store -watches -lei_note_event)};
 		}
 	} else { # worker, Net::NNTP (Net::Cmd) uses STDERR directly
 		open STDERR, '+>&='.fileno($self->{2}) or warn "open $!";
@@ -555,6 +556,8 @@ sub _lei_atfork_child {
 	undef $listener;
 	undef $dir_idle;
 	%PATH2CFG = ();
+	$MDIR2CFGPATH = {};
+	eval 'no warnings; undef $PublicInbox::LeiNoteEvent::to_flush';
 	undef $errors_log;
 	$quit = \&CORE::exit;
 	$self->{-eml_noisy} or # only "lei import" sets this atm
@@ -781,10 +784,12 @@ sub _lei_cfg ($;$) {
 	my $f = _config_path($self);
 	my @st = stat($f);
 	my $cur_st = @st ? pack('dd', $st[10], $st[7]) : ''; # 10:ctime, 7:size
-	my ($sto, $sto_dir);
+	my ($sto, $sto_dir, $watches, $lne);
 	if (my $cfg = $PATH2CFG{$f}) { # reuse existing object in common case
 		return ($self->{cfg} = $cfg) if $cur_st eq $cfg->{-st};
-		($sto, $sto_dir) = @$cfg{qw(-lei_store leistore.dir)};
+		($sto, $sto_dir, $watches, $lne) =
+				@$cfg{qw(-lei_store leistore.dir -watches
+					-lei_note_event)};
 	}
 	if (!@st) {
 		unless ($creat) {
@@ -805,6 +810,8 @@ sub _lei_cfg ($;$) {
 			eq File::Spec->canonpath($cfg->{'leistore.dir'} //
 						store_path($self))) {
 		$cfg->{-lei_store} = $sto;
+		$cfg->{-lei_note_event} = $lne;
+		$cfg->{-watches} = $watches if $watches;
 	}
 	if (scalar(keys %PATH2CFG) > 5) {
 		# FIXME: use inotify/EVFILT_VNODE to detect unlinked configs
@@ -817,7 +824,7 @@ sub _lei_cfg ($;$) {
 
 sub _lei_store ($;$) {
 	my ($self, $creat) = @_;
-	my $cfg = _lei_cfg($self, $creat);
+	my $cfg = _lei_cfg($self, $creat) // return;
 	$cfg->{-lei_store} //= do {
 		require PublicInbox::LeiStore;
 		my $dir = $cfg->{'leistore.dir'} // store_path($self);
@@ -1126,6 +1133,53 @@ sub dump_and_clear_log {
 	}
 }
 
+sub cfg2lei ($) {
+	my ($cfg) = @_;
+	my $lei = bless { env => { %{$cfg->{-env}} } }, __PACKAGE__;
+	open($lei->{0}, '<&', \*STDIN) or die "dup 0: $!";
+	open($lei->{1}, '>>&', \*STDOUT) or die "dup 1: $!";
+	open($lei->{2}, '>>&', \*STDERR) or die "dup 2: $!";
+	open($lei->{3}, '/') or die "open /: $!";
+	chdir($lei->{3}) or die "chdir /': $!";
+	my ($x, $y);
+	socketpair($x, $y, AF_UNIX, SOCK_SEQPACKET, 0) or die "socketpair: $!";
+	$lei->{sock} = $x;
+	require PublicInbox::LeiSelfSocket;
+	PublicInbox::LeiSelfSocket->new($y); # adds to event loop
+	$lei;
+}
+
+sub dir_idle_handler ($) { # PublicInbox::DirIdle callback
+	my ($ev) = @_; # Linux::Inotify2::Event or duck type
+	my $fn = $ev->fullname;
+	if ($fn =~ m!\A(.+)/(new|cur)/([^/]+)\z!) { # Maildir file
+		my ($mdir, $nc, $bn) = ($1, $2, $3);
+		$nc = '' if $ev->IN_DELETE;
+		for my $f (keys %{$MDIR2CFGPATH->{$mdir} // {}}) {
+			my $cfg = $PATH2CFG{$f} // next;
+			eval {
+				local %ENV = %{$cfg->{-env}};
+				my $lei = cfg2lei($cfg);
+				$lei->dispatch('note-event',
+						"maildir:$mdir", $nc, $bn, $fn);
+			};
+			warn "E note-event $f: $@\n" if $@;
+		}
+	}
+	if ($ev->can('cancel') && ($ev->IN_IGNORE || $ev->IN_UNMOUNT)) {
+		$ev->cancel;
+	}
+	if ($fn =~ m!\A(.+)/(?:new|cur)\z! && !-e $fn) {
+		delete $MDIR2CFGPATH->{$1};
+	}
+	if (!-e $fn) { # config file or Maildir gone
+		for my $cfgpaths (values %$MDIR2CFGPATH) {
+			delete $cfgpaths->{$fn};
+		}
+		delete $PATH2CFG{$fn};
+	}
+}
+
 # lei(1) calls this when it can't connect
 sub lazy_start {
 	my ($path, $errno, $narg) = @_;
@@ -1175,6 +1229,7 @@ sub lazy_start {
 	return if $pid;
 	$0 = "lei-daemon $path";
 	local %PATH2CFG;
+	local $MDIR2CFGPATH;
 	$listener->blocking(0);
 	my $exit_code;
 	my $pil = PublicInbox::Listener->new($listener, \&accept_dispatch);
@@ -1204,8 +1259,8 @@ sub lazy_start {
 	local $SIG{PIPE} = 'IGNORE';
 	require PublicInbox::DirIdle;
 	local $dir_idle = PublicInbox::DirIdle->new([$sock_dir], sub {
-		# just rely on wakeup ot hit PostLoopCallback set below
-		_dir_idle_handler(@_) if $_[0]->fullname ne $path;
+		# just rely on wakeup to hit PostLoopCallback set below
+		dir_idle_handler($_[0]) if $_[0]->fullname ne $path;
 	}, 1);
 	if ($sigfd) {
 		undef $sigfd; # unref, already in DS::DescriptorMap
@@ -1293,4 +1348,41 @@ sub wq_eof { # EOF callback for main daemon
 	$wq1->wq_wait_old(\&wq_done_wait, $lei);
 }
 
+sub watch_state_ok ($) {
+	my ($state) = $_[-1]; # $_[0] may be $self
+	$state =~ /\Apause|(?:import|index|tag)-(?:ro|rw)\z/;
+}
+
+sub refresh_watches {
+	my ($lei) = @_;
+	my $cfg = _lei_cfg($lei) or return;
+	$cfg->{-env} //= { %{$lei->{env}}, PWD => '/' }; # for cfg2lei
+	my $watches = $cfg->{-watches} //= {};
+	require PublicInbox::LeiWatch;
+	for my $w (grep(/\Awatch\..+\.state\z/, keys %$cfg)) {
+		my $url = substr($w, length('watch.'), -length('.state'));
+		my $lw = $watches->{$w} //= PublicInbox::LeiWatch->new($url);
+		my $state = $cfg->get_1("watch.$url", 'state');
+		if (!watch_state_ok($state)) {
+			$lei->err("watch.$url.state=$state not supported");
+			next;
+		}
+		my $f = $cfg->{'-f'};
+		if ($url =~ /\Amaildir:(.+)/i) {
+			my $d = File::Spec->canonpath($1);
+			if ($state eq 'pause') {
+				delete $MDIR2CFGPATH->{$d}->{$f};
+				scalar(keys %{$MDIR2CFGPATH->{$d}}) or
+					delete $MDIR2CFGPATH->{$d};
+			} elsif (!exists($MDIR2CFGPATH->{$d}->{$f})) {
+				$dir_idle->add_watches(["$d/cur", "$d/new"], 1);
+				$MDIR2CFGPATH->{$d}->{$f} = undef;
+			}
+		} else { # TODO: imap/nntp/jmap
+			$lei->child_error(1,
+				"E: watch $url not supported, yet");
+		}
+	}
+}
+
 1;
diff --git a/lib/PublicInbox/LeiAddWatch.pm b/lib/PublicInbox/LeiAddWatch.pm
new file mode 100644
index 00000000..671d54f9
--- /dev/null
+++ b/lib/PublicInbox/LeiAddWatch.pm
@@ -0,0 +1,41 @@
+# Copyright all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+
+# "lei add-watch" command
+package PublicInbox::LeiAddWatch;
+use strict;
+use v5.10.1;
+use parent qw(PublicInbox::LeiInput);
+
+sub lei_add_watch {
+	my ($lei, @argv) = @_;
+	my $cfg = $lei->_lei_cfg(1);
+	my $self = bless {}, __PACKAGE__;
+	$lei->{opt}->{'mail-sync'} = 1; # for prepare_inputs
+	my $state = $lei->{opt}->{'state'} // 'import-rw';
+	$lei->watch_state_ok($state) or
+		return $lei->fail("invalid state: $state");
+	my $vmd_mod = $self->vmd_mod_extract(\@argv);
+	return $lei->fail(join("\n", @{$vmd_mod->{err}})) if $vmd_mod->{err};
+	$self->prepare_inputs($lei, \@argv) or return;
+	my @vmd;
+	while (my ($type, $vals) = each %$vmd_mod) {
+		push @vmd, "$type:$_" for @$vals;
+	}
+	my $vmd0 = shift @vmd;
+	for my $w (@{$self->{inputs}}) {
+		# clobber existing, allow multiple
+		if (defined($vmd0)) {
+			$lei->_config("watch.$w.vmd", '--replace-all', $vmd0);
+			for my $v (@vmd) {
+				$lei->_config("watch.$w.vmd", $v);
+			}
+		}
+		next if defined $cfg->{"watch.$w.state"};
+		$lei->_config("watch.$w.state", $state);
+	}
+	delete $lei->{cfg}; # force reload
+	$lei->refresh_watches;
+}
+
+1;
diff --git a/lib/PublicInbox/LeiInput.pm b/lib/PublicInbox/LeiInput.pm
index de2a8ff1..fa330df5 100644
--- a/lib/PublicInbox/LeiInput.pm
+++ b/lib/PublicInbox/LeiInput.pm
@@ -372,6 +372,7 @@ sub input_only_net_merge_all_done {
 
 # like Getopt::Long, but for +kw:FOO and -kw:FOO to prepare
 # for update_xvmd -> update_vmd
+# returns something like { "+L" => [ @Labels ], ... }
 sub vmd_mod_extract {
 	my $argv = $_[-1];
 	my $vmd_mod = {};
diff --git a/lib/PublicInbox/LeiLsWatch.pm b/lib/PublicInbox/LeiLsWatch.pm
new file mode 100644
index 00000000..f96dc4ec
--- /dev/null
+++ b/lib/PublicInbox/LeiLsWatch.pm
@@ -0,0 +1,15 @@
+# Copyright all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+
+package PublicInbox::LeiLsWatch;
+use strict;
+use v5.10.1;
+
+sub lei_ls_watch {
+	my ($lei) = @_;
+	my $cfg = $lei->_lei_cfg or return;
+	my @w = (join("\n", keys %$cfg) =~ m/^watch\.(.+?)\.state$/sgm);
+	$lei->puts(join("\n", @w)) if @w;
+}
+
+1;
diff --git a/lib/PublicInbox/LeiNoteEvent.pm b/lib/PublicInbox/LeiNoteEvent.pm
new file mode 100644
index 00000000..bf15cd26
--- /dev/null
+++ b/lib/PublicInbox/LeiNoteEvent.pm
@@ -0,0 +1,124 @@
+# Copyright (C) all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+
+# internal command for dealing with inotify, kqueue vnodes, etc
+package PublicInbox::LeiNoteEvent;
+use strict;
+use v5.10.1;
+use parent qw(PublicInbox::IPC);
+
+my $flush_timer;
+our $to_flush; # { cfgpath => $lei }
+
+sub flush_lei ($) {
+	my ($lei) = @_;
+	if (my $lne = delete $lei->{cfg}->{-lei_note_event}) {
+		$lne->wq_close(1, undef, $lei); # runs _lei_wq_eof;
+	} else { # lms_clear_src calls only:
+		my $wait = $lei->{sto}->ipc_do('done');
+	}
+}
+
+# we batch up writes and flush every 5s (matching Linux default
+# writeback behavior) since MUAs can trigger a storm of inotify events
+sub flush_task { # PublicInbox::DS timer callback
+	undef $flush_timer;
+	my $todo = $to_flush // return;
+	$to_flush = undef;
+	for my $lei (values %$todo) { flush_lei($lei) }
+}
+
+# sets a timer to flush
+sub note_event_arm_done ($) {
+	my ($lei) = @_;
+	$flush_timer //= PublicInbox::DS::add_timer(5, \&flush_task);
+	$to_flush->{$lei->{cfg}->{'-f'}} //= $lei;
+}
+
+sub eml_event ($$$$) {
+	my ($self, $eml, $kw, $state) = @_;
+	my $sto = $self->{lei}->{sto};
+	my $lse = $self->{lse} //= $sto->search;
+	my $vmd = { kw => $kw };
+	if ($state =~ /\Aimport-(?:rw|ro)\z/) {
+		$sto->ipc_do('set_eml', $eml, $vmd);
+	} elsif ($state =~ /\Aindex-(?:rw|ro)\z/) {
+		my $xoids = $self->{lei}->ale->xoids_for($eml);
+		$sto->ipc_do('index_eml_only', $eml, $vmd, $xoids);
+	} elsif ($state =~ /\Atag-(?:rw|ro)\z/) {
+		my $c = $lse->kw_changed($eml, $kw, my $docids = []);
+		if (scalar @$docids) { # already in lei/store
+			$sto->ipc_do('set_eml_vmd', undef, $vmd, $docids) if $c;
+		} elsif (my $xoids = $self->{lei}->ale->xoids_for($eml)) {
+			# it's in an external, only set kw, here
+			$sto->ipc_do('set_xvmd', $xoids, $eml, $vmd);
+		} # else { totally unknown
+	} else {
+		warn "unknown state: $state (in $self->{lei}->{cfg}->{'-f'})\n";
+	}
+}
+
+sub maildir_event { # via wq_io_do
+	my ($self, $fn, $kw, $state) = @_;
+	my $eml = PublicInbox::InboxWritable::eml_from_path($fn) // return;
+	eml_event($self, $eml, $kw, $state);
+}
+
+sub lei_note_event {
+	my ($lei, $folder, $new_cur, $bn, $fn, @rest) = @_;
+	die "BUG: unexpected: @rest" if @rest;
+	my $cfg = $lei->_lei_cfg or return; # gone (race)
+	my $sto = $lei->_lei_store or return; # gone
+	return flush_lei($lei) if $folder eq 'done'; # special case
+	my $lms = $sto->search->lms or return;
+	my $err = $lms->arg2folder($lei, [ $folder ]);
+	return if $err->{fail};
+	undef $lms;
+	my $state = $cfg->get_1("watch.$folder", 'state') // 'pause';
+	return if $state eq 'pause';
+	$lei->ale; # prepare
+	$sto->write_prepare($lei);
+	if ($new_cur eq '') {
+		$sto->ipc_do('lms_clear_src', $folder, \$bn);
+		return note_event_arm_done($lei);
+	}
+	require PublicInbox::MdirReader;
+	my $self = $cfg->{-lei_note_event} //= do {
+		my $wq = bless {}, __PACKAGE__;
+		# MUAs such as mutt can trigger massive rename() storms so
+		# use all CPU power available:
+		my $jobs = $wq->detect_nproc // 1;
+		my ($op_c, $ops) = $lei->workers_start($wq, $jobs);
+		$lei->wait_wq_events($op_c, $ops);
+		note_event_arm_done($lei);
+		$lei->{lne} = $wq;
+	};
+	if ($folder =~ /\Amaildir:/i) {
+		my $fl = PublicInbox::MdirReader::maildir_basename_flags($bn)
+			// return;
+		return if index($fl, 'T') >= 0;
+		my $kw = PublicInbox::MdirReader::flags2kw($fl);
+		$self->wq_io_do('maildir_event', [], $fn, $kw, $state);
+	} # else: TODO: imap
+}
+
+sub ipc_atfork_child {
+	my ($self) = @_;
+	$self->{lei}->_lei_atfork_child(1); # persistent, for a while
+	$self->SUPER::ipc_atfork_child;
+}
+
+sub lne_done_wait {
+	my ($arg, $pid) = @_;
+	my ($self, $lei) = @$arg;
+	$lei->can('wq_done_wait')->($arg, $pid);
+}
+
+sub _lei_wq_eof { # EOF callback for main lei daemon
+	my ($lei) = @_;
+	my $lne = delete $lei->{lne} or return $lei->fail;
+	my $wait = $lei->{sto}->ipc_do('done');
+	$lne->wq_wait_old(\&lne_done_wait, $lei);
+}
+
+1;
diff --git a/lib/PublicInbox/LeiSearch.pm b/lib/PublicInbox/LeiSearch.pm
index 06ea6299..37bfc65e 100644
--- a/lib/PublicInbox/LeiSearch.pm
+++ b/lib/PublicInbox/LeiSearch.pm
@@ -5,7 +5,7 @@
 package PublicInbox::LeiSearch;
 use strict;
 use v5.10.1;
-use parent qw(PublicInbox::ExtSearch);
+use parent qw(PublicInbox::ExtSearch); # PublicInbox::Search->reopen
 use PublicInbox::Search qw(xap_terms);
 use PublicInbox::ContentHash qw(content_digest content_hash);
 use PublicInbox::MID qw(mids mids_for_index);
diff --git a/lib/PublicInbox/LeiSelfSocket.pm b/lib/PublicInbox/LeiSelfSocket.pm
new file mode 100644
index 00000000..3d847649
--- /dev/null
+++ b/lib/PublicInbox/LeiSelfSocket.pm
@@ -0,0 +1,45 @@
+# Copyright all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+
+# dummy placeholder socket for internal lei commands.
+# This receives what script/lei receives, but isn't connected
+# to an interactive terminal so I'm not sure what to do with it...
+package PublicInbox::LeiSelfSocket;
+use strict;
+use v5.10.1;
+use parent qw(PublicInbox::DS);
+use Data::Dumper;
+$Data::Dumper::Useqq = 1; # should've been the Perl default :P
+use PublicInbox::Syscall qw(EPOLLIN EPOLLET);
+use PublicInbox::Spawn;
+my $recv_cmd;
+
+sub new {
+	my ($cls, $r) = @_;
+	my $self = bless { sock => $r }, $cls;
+	$r->blocking(0);
+	no warnings 'once';
+	$recv_cmd = $PublicInbox::LEI::recv_cmd;
+	$self->SUPER::new($r, EPOLLIN|EPOLLET);
+}
+
+sub event_step {
+	my ($self) = @_;
+	while (1) {
+		my (@fds) = $recv_cmd->($self->{sock}, my $buf, 4096 * 33);
+		if (scalar(@fds) == 1 && !defined($fds[0])) {
+			return if $!{EAGAIN};
+			next if $!{EINTR};
+			die "recvmsg: $!";
+		}
+		# open so perl can auto-close them:
+		for my $fd (@fds) {
+			open(my $newfh, '+<&=', $fd) or die "open +<&=$fd: $!";
+		}
+		return $self->close if $buf eq '';
+		warn Dumper({ 'unexpected self msg' => $buf, fds => \@fds });
+		# TODO: figure out what to do with these messages...
+	}
+}
+
+1;
diff --git a/lib/PublicInbox/LeiWatch.pm b/lib/PublicInbox/LeiWatch.pm
new file mode 100644
index 00000000..35267b58
--- /dev/null
+++ b/lib/PublicInbox/LeiWatch.pm
@@ -0,0 +1,13 @@
+# Copyright all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+
+# represents a Maildir or IMAP "watch" item
+package PublicInbox::LeiWatch;
+use strict;
+use v5.10.1;
+use parent qw(PublicInbox::IPC);
+
+# "url" may be something like "maildir:/path/to/dir"
+sub new { bless { url => $_[1] }, $_[0] }
+
+1;
diff --git a/t/lei-watch.t b/t/lei-watch.t
new file mode 100644
index 00000000..3a2f9e64
--- /dev/null
+++ b/t/lei-watch.t
@@ -0,0 +1,49 @@
+#!perl -w
+# Copyright all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+use strict; use v5.10.1; use PublicInbox::TestCommon;
+use File::Path qw(make_path);
+require_mods('lei');
+my $have_fast_inotify = eval { require Linux::Inotify2 } ||
+	eval { require IO::KQueue };
+
+$have_fast_inotify or
+	diag("$0 IO::KQueue or Linux::Inotify2 missing, test will be slow");
+
+my ($ro_home, $cfg_path) = setup_public_inboxes;
+test_lei(sub {
+	my $md = "$ENV{HOME}/md";
+	my $md2 = $md.'2';
+	lei_ok 'ls-watch';
+	is($lei_out, '', 'nothing in ls-watch, yet');
+	if (0) { # TODO
+		my $url = 'imaps://example.com/foo.bar.0';
+		lei_ok([qw(add-watch --state=pause), $url], undef, {});
+		lei_ok 'ls-watch';
+		is($lei_out, "$url\n", 'ls-watch shows added watch');
+		ok(!lei(qw(add-watch --state=pause), 'bogus'.$url),
+			'bogus URL rejected');
+	}
+
+	# first, make sure tag-ro works
+	make_path("$md/new", "$md/cur", "$md/tmp");
+	lei_ok qw(add-watch --state=tag-ro), $md;
+	lei_ok 'ls-watch';
+	like($lei_out, qr/^\Qmaildir:$md\E$/sm, 'maildir shown');
+	lei_ok qw(q mid:testmessage@example.com -o), $md, '-I', "$ro_home/t1";
+	my @f = glob("$md/cur/*:2,");
+	is(scalar(@f), 1, 'got populated maildir with one result');
+	rename($f[0], "$f[0]S") or xbail "rename $!"; # set (S)een
+	$have_fast_inotify or tick(2);
+	lei_ok qw(note-event done); # flushes immediately (instead of 5s)
+
+	lei_ok qw(q mid:testmessage@example.com -o), $md2, '-I', "$ro_home/t1";
+	my @f2 = glob("$md2/*/*");
+	is(scalar(@f2), 1, 'got one result');
+	like($f2[0], qr/S\z/, 'seen set from rename');
+	my $e2 = eml_load($f2[0]);
+	my $e1 = eml_load("$f[0]S");
+	is_deeply($e2, $e1, 'results match');
+});
+
+done_testing;