lei import: help + completion for --new-only

[public-inbox.git] / lib / PublicInbox / LEI.pm

diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index 8adf70faf2d70dbcf629e103bdf9fd589b2673da..546fa773fc4c334cc83a27358ba74f016cf62dab 100644 (file)
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -116,6 +116,12 @@ sub cache_dir ($) {
                .'/lei');
 }
 
+sub url_folder_cache {
+       my ($self) = @_;
+       require PublicInbox::SharedKV; # URI => updated_at_sec_
+       PublicInbox::SharedKV->new(cache_dir($self).'/uri_folder');
+}
+
 sub ale {
        my ($self) = @_;
        $self->{ale} //= do {
@@ -196,6 +202,8 @@ our %CMD = ( # sorted in order of importance/use:
 'ls-label' => [ '', 'list labels', qw(z|0 stats:s), @c_opt ],
 'ls-mail-sync' => [ '[FILTER]', 'list mail sync folders',
                qw(z|0 globoff|g invert-match|v local remote), @c_opt ],
+'ls-mail-source' => [ 'URL', 'list IMAP or NNTP mail source folders',
+               qw(z|0 ascii l url), @c_opt ],
 'forget-external' => [ 'LOCATION...|--prune',
        'exclude further results from a publicinbox|extindex',
        qw(prune), @c_opt ],
@@ -240,11 +248,13 @@ our %CMD = ( # sorted in order of importance/use:
         @c_opt ],
 'import' => [ 'LOCATION...|--stdin',
        'one-time import/update from URL or filesystem',
-       qw(stdin| offset=i recursive|r exclude=s include|I=s
+       qw(stdin| offset=i recursive|r exclude=s include|I=s jobs=s new-only
        lock=s@ in-format|F=s kw! verbose|v+ incremental! mail-sync!),
        qw(no-torsocks torsocks=s), PublicInbox::LeiQuery::curl_opt(), @c_opt ],
 'forget-mail-sync' => [ 'LOCATION...',
        'forget sync information for a mail folder', @c_opt ],
+'prune-mail-sync' => [ 'LOCATION...|--all',
+       'prune dangling sync data for a mail folder', 'all:s', @c_opt ],
 'export-kw' => [ 'LOCATION...|--all',
        'one-time export of keywords of sync sources',
        qw(all:s mode=s), @c_opt ],
@@ -343,6 +353,7 @@ my %OPTDESC = (
        "or\xa0`-'\x{a0}for\x{a0}stdout)" ],
 'mua=s' => [ 'CMD',
        "MUA to run on --output Maildir or mbox (e.g.\xa0`mutt\xa0-f\xa0%f')" ],
+'new-only      import' => 'only import new messages from IMAP source',
 
 'inbox-version=i' => [ 'NUM|1|2',
                'force a public-inbox version with --mirror'],
@@ -379,6 +390,8 @@ my %OPTDESC = (
 'format|f=s    ls-search' => ['OUT|json|jsonl|concatjson',
                        'listing output format' ],
 'l     ls-search' => '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,
 
 'limit|n=i@' => ['NUM', 'limit on number of matches (default: 10000)' ],
@@ -421,7 +434,7 @@ my %CONFIG_KEYS = (
        'leistore.dir' => 'top-level storage location',
 );
 
-my @WQ_KEYS = qw(lxs l2m wq1 ikw); # internal workers
+my @WQ_KEYS = qw(lxs l2m ikw pmd wq1); # internal workers
 
 sub _drop_wq {
        my ($self) = @_;
@@ -441,6 +454,7 @@ sub x_it ($$) {
        # make sure client sees stdout before exit
        $self->{1}->autoflush(1) if $self->{1};
        dump_and_clear_log();
+       stop_pager($self);
        if ($self->{pkt_op_p}) { # to top lei-daemon
                $self->{pkt_op_p}->pkt_do('x_it', $code);
        } elsif ($self->{sock}) { # to lei(1) client
@@ -565,19 +579,26 @@ sub pkt_op_pair {
        $end;
 }
 
+sub incr {
+       my ($self, $field, $nr) = @_;
+       $self->{counters}->{$field} += $nr;
+}
+
 sub workers_start {
-       my ($lei, $wq, $jobs, $ops) = @_;
+       my ($lei, $wq, $jobs, $ops, $flds) = @_;
        $ops = {
                '!' => [ \&fail_handler, $lei ],
                '|' => [ \&sigpipe_handler, $lei ],
                'x_it' => [ \&x_it, $lei ],
                'child_error' => [ \&child_error, $lei ],
+               'incr' => [ \&incr, $lei ],
                ($ops ? %$ops : ()),
        };
        $ops->{''} //= [ $wq->can('_lei_wq_eof') || \&wq_eof, $lei ];
        my $end = $lei->pkt_op_pair;
        my $ident = $wq->{-wq_ident} // "lei-$lei->{cmd} worker";
-       $wq->wq_workers_start($ident, $jobs, $lei->oldset, { lei => $lei });
+       $flds->{lei} = $lei;
+       $wq->wq_workers_start($ident, $jobs, $lei->oldset, $flds);
        delete $lei->{pkt_op_p};
        my $op_c = delete $lei->{pkt_op_c};
        @$end = ();
@@ -585,6 +606,15 @@ sub workers_start {
        ($op_c, $ops);
 }
 
+# call this when we're ready to wait on events and yield to other clients
+sub wait_wq_events {
+       my ($lei, $op_c, $ops) = @_;
+       for my $wq (grep(defined, @$lei{qw(ikw pmd)})) { # auxiliary WQs
+               $wq->wq_close(1);
+       }
+       $op_c->{ops} = $ops;
+}
+
 sub _help {
        require PublicInbox::LeiHelp;
        PublicInbox::LeiHelp::call($_[0], $_[1], \%CMD, \%OPTDESC);
@@ -1078,10 +1108,11 @@ sub event_step {
 
 sub event_step_init {
        my ($self) = @_;
-       return if $self->{-event_init_done}++;
-       if (my $sock = $self->{sock}) { # using DS->EventLoop
+       my $sock = $self->{sock} or return;
+       $self->{-event_init_done} //= do { # persist til $ops done
                $self->SUPER::new($sock, EPOLLIN|EPOLLET);
-       }
+               $sock;
+       };
 }
 
 sub noop {}
@@ -1232,6 +1263,12 @@ sub busy { 1 } # prevent daemon-shutdown if client is connected
 # can immediately reread it
 sub DESTROY {
        my ($self) = @_;
+       if (my $counters = delete $self->{counters}) {
+               for my $k (sort keys %$counters) {
+                       my $nr = $counters->{$k};
+                       $self->child_error(1 << 8, "$nr $k messages");
+               }
+       }
        $self->{1}->autoflush(1) if $self->{1};
        stop_pager($self);
        # preserve $? for ->fail or ->x_it code