]> Sergey Matveev's repositories - public-inbox.git/blobdiff - lib/PublicInbox/IMAP.pm
imap: use git-cat-file asynchronously
[public-inbox.git] / lib / PublicInbox / IMAP.pm
index 673e164621654a0201ba970af8185b730390412c..2282e3ce78c6be097b289d04f68ce9c9b0e0842b 100644 (file)
@@ -21,8 +21,10 @@ use PublicInbox::Eml;
 use PublicInbox::EmlContentFoo qw(parse_content_disposition);
 use PublicInbox::DS qw(now);
 use PublicInbox::Syscall qw(EPOLLIN EPOLLONESHOT);
+use PublicInbox::GitAsyncCat;
 use Text::ParseWords qw(parse_line);
 use Errno qw(EAGAIN);
+
 my $Address;
 for my $mod (qw(Email::Address::XS Mail::Address)) {
        eval "require $mod" or next;
@@ -33,9 +35,6 @@ die "neither Email::Address::XS nor Mail::Address loaded: $@" if !$Address;
 sub LINE_MAX () { 512 } # does RFC 3501 have a limit like RFC 977?
 
 my %FETCH_NEED_BLOB = ( # for future optimization
-       'BODY.PEEK[HEADER]' => 1,
-       'BODY.PEEK[TEXT]' => 1,
-       'BODY.PEEK[]' => 1,
        'BODY[HEADER]' => 1,
        'BODY[TEXT]' => 1,
        'BODY[]' => 1,
@@ -124,6 +123,12 @@ sub cmd_login ($$$$) {
        login_success($self, $tag);
 }
 
+sub cmd_close ($$) {
+       my ($self, $tag) = @_;
+       delete $self->{ibx} ? "$tag OK Close done\r\n"
+                               : "$tag BAD No mailbox\r\n";
+}
+
 sub cmd_logout ($$) {
        my ($self, $tag) = @_;
        delete $self->{logged_in};
@@ -348,11 +353,50 @@ sub fetch_body ($;$) {
        join('', @hold);
 }
 
-sub uid_fetch_cb { # called by git->cat_async
+sub dummy_message ($$) {
+       my ($seqno, $ibx) = @_;
+       my $ret = <<EOF;
+From: nobody\@localhost\r
+To: nobody\@localhost\r
+Date: Thu, 01 Jan 1970 00:00:00 +0000\r
+Message-ID: <dummy-$seqno\@$ibx->{newsgroup}>\r
+Subject: dummy message #$seqno\r
+\r
+You're seeing this message because your IMAP client didn't use UIDs.\r
+The message which used to use this sequence number was likely spam\r
+and removed by the administrator.\r
+EOF
+       \$ret;
+}
+
+sub requeue_once ($) {
+       my ($self) = @_;
+       # COMPRESS users all share the same DEFLATE context.
+       # Flush it here to ensure clients don't see
+       # each other's data
+       $self->zflush;
+
+       # no recursion, schedule another call ASAP,
+       # but only after all pending writes are done.
+       # autovivify wbuf:
+       my $new_size = push(@{$self->{wbuf}}, \&long_step);
+
+       # wbuf may be populated by $cb, no need to rearm if so:
+       $self->requeue if $new_size == 1;
+}
+
+sub uid_fetch_cb { # called by git->cat_async via git_async_msg
        my ($bref, $oid, $type, $size, $fetch_m_arg) = @_;
        my ($self, undef, $ibx, undef, undef, $msgs, $want) = @$fetch_m_arg;
        my $smsg = shift @$msgs or die 'BUG: no smsg';
-       $smsg->{blob} eq $oid or die "BUG: $smsg->{blob} != $oid";
+       if (!defined($oid)) {
+               # it's possible to have TOCTOU if an admin runs
+               # public-inbox-(edit|purge), just move onto the next message
+               return requeue_once($self) unless defined $want->{-seqno};
+               $bref = dummy_message($smsg->{num}, $ibx);
+       } else {
+               $smsg->{blob} eq $oid or die "BUG: $smsg->{blob} != $oid";
+       }
        $$bref =~ s/(?<!\r)\n/\r\n/sg; # make strict clients happy
 
        # fixup old bug from import (pre-a0c07cba0e5d8b6a)
@@ -365,8 +409,8 @@ sub uid_fetch_cb { # called by git->cat_async
        $want->{INTERNALDATE} and
                $self->msg_more(' INTERNALDATE "'.$smsg->internaldate.'"');
        $want->{FLAGS} and $self->msg_more(' FLAGS ()');
-       for ('RFC822', 'BODY[]', 'BODY.PEEK[]') {
-               next unless $want->{$_};
+       for ('RFC822', 'BODY[]') {
+               $want->{$_} or next;
                $self->msg_more(" $_ {".length($$bref)."}\r\n");
                $self->msg_more($$bref);
        }
@@ -376,14 +420,14 @@ sub uid_fetch_cb { # called by git->cat_async
        $want->{ENVELOPE} and
                $self->msg_more(' ENVELOPE '.eml_envelope($eml));
 
-       for my $f ('RFC822.HEADER', 'BODY[HEADER]', 'BODY.PEEK[HEADER]') {
-               next unless $want->{$f};
-               $self->msg_more(" $f {".length(${$eml->{hdr}})."}\r\n");
+       for ('RFC822.HEADER', 'BODY[HEADER]') {
+               $want->{$_} or next;
+               $self->msg_more(" $_ {".length(${$eml->{hdr}})."}\r\n");
                $self->msg_more(${$eml->{hdr}});
        }
-       for my $f ('RFC822.TEXT', 'BODY[TEXT]') {
-               next unless $want->{$f};
-               $self->msg_more(" $f {".length($$bref)."}\r\n");
+       for ('RFC822.TEXT', 'BODY[TEXT]') {
+               $want->{$_} or next;
+               $self->msg_more(" $_ {".length($$bref)."}\r\n");
                $self->msg_more($$bref);
        }
        $want->{BODYSTRUCTURE} and
@@ -394,6 +438,7 @@ sub uid_fetch_cb { # called by git->cat_async
                partial_emit($self, $partial, $eml);
        }
        $self->msg_more(")\r\n");
+       requeue_once($self);
 }
 
 sub uid_fetch_m { # long_response
@@ -406,11 +451,7 @@ sub uid_fetch_m { # long_response
                }
                $$beg = $msgs->[-1]->{num} + 1;
        }
-       my $git = $ibx->git;
-       $git->cat_async_begin; # TODO: actually make async
-       $git->cat_async($msgs->[0]->{blob}, \&uid_fetch_cb, \@_);
-       $git->cat_async_wait;
-       1;
+       git_async_msg($ibx, $msgs->[0], \&uid_fetch_cb, \@_);
 }
 
 sub cmd_status ($$$;@) {
@@ -518,25 +559,23 @@ sub hdrs_regexp ($) {
 
 # BODY[($SECTION_IDX.)?HEADER.FIELDS.NOT ($HDRS)]<$offset.$bytes>
 sub partial_hdr_not {
-       my ($eml, $section_idx, $hdrs) = @_;
+       my ($eml, $section_idx, $hdrs_re) = @_;
        if (defined $section_idx) {
                $eml = eml_body_idx($eml, $section_idx) or return;
        }
        my $str = $eml->header_obj->as_string;
-       my $re = hdrs_regexp($hdrs);
-       $str =~ s/$re//g;
+       $str =~ s/$hdrs_re//g;
        $str .= "\r\n";
 }
 
 # BODY[($SECTION_IDX.)?HEADER.FIELDS ($HDRS)]<$offset.$bytes>
 sub partial_hdr_get {
-       my ($eml, $section_idx, $hdrs) = @_;
+       my ($eml, $section_idx, $hdrs_re) = @_;
        if (defined $section_idx) {
                $eml = eml_body_idx($eml, $section_idx) or return;
        }
        my $str = $eml->header_obj->as_string;
-       my $re = hdrs_regexp($hdrs);
-       join('', ($str =~ m/($re)/g), "\r\n");
+       join('', ($str =~ m/($hdrs_re)/g), "\r\n");
 }
 
 sub partial_prepare ($$$) {
@@ -544,24 +583,23 @@ sub partial_prepare ($$$) {
 
        # recombine [ "BODY[1.HEADER.FIELDS", "(foo", "bar)]" ]
        # back to: "BODY[1.HEADER.FIELDS (foo bar)]"
-       return unless $att =~ /\ABODY(?:\.PEEK)?\[/s;
+       return unless $att =~ /\ABODY\[/s;
        until (rindex($att, ']') >= 0) {
                my $next = shift @$want or return;
                $att .= ' ' . uc($next);
        }
-       if ($att =~ /\ABODY(?:\.PEEK)?\[
-                               ([0-9]+(?:\.[0-9]+)*)? # 1 - section_idx
-                               (?:\.(HEADER|MIME|TEXT))? # 2 - section_name
+       if ($att =~ /\ABODY\[([0-9]+(?:\.[0-9]+)*)? # 1 - section_idx
+                       (?:\.(HEADER|MIME|TEXT))? # 2 - section_name
                        \](?:<([0-9]+)(?:\.([0-9]+))?>)?\z/sx) { # 3, 4
                $partial->{$att} = [ \&partial_body, $1, $2, $3, $4 ];
-       } elsif ($att =~ /\ABODY(?:\.PEEK)?\[
-                               (?:([0-9]+(?:\.[0-9]+)*)\.)? # 1 - section_idx
+       } elsif ($att =~ /\ABODY\[(?:([0-9]+(?:\.[0-9]+)*)\.)? # 1 - section_idx
                                (?:HEADER\.FIELDS(\.NOT)?)\x20 # 2
                                \(([A-Z0-9\-\x20]+)\) # 3 - hdrs
                        \](?:<([0-9]+)(?:\.([0-9]+))?>)?\z/sx) { # 4 5
-               $partial->{$att} = [ $2 ? \&partial_hdr_not
-                                       : \&partial_hdr_get,
-                                       $1, $3, $4, $5 ];
+               my $tmp = $partial->{$att} = [ $2 ? \&partial_hdr_not
+                                               : \&partial_hdr_get,
+                                               $1, undef, $4, $5 ];
+               $tmp->[2] = hdrs_regexp($3);
        } else {
                undef;
        }
@@ -591,26 +629,27 @@ sub partial_emit ($$$) {
        }
 }
 
-sub cmd_uid_fetch ($$$;@) {
-       my ($self, $tag, $range, @want) = @_;
+sub fetch_common ($$$$) {
+       my ($self, $tag, $range, $want) = @_;
        my $ibx = $self->{ibx} or return "$tag BAD No mailbox selected\r\n";
-       if ($want[0] =~ s/\A\(//s) {
-               $want[-1] =~ s/\)\z//s or return "$tag BAD no rparen\r\n";
+       if ($want->[0] =~ s/\A\(//s) {
+               $want->[-1] =~ s/\)\z//s or return "$tag BAD no rparen\r\n";
        }
        my (%partial, %want);
-       while (defined(my $att = shift @want)) {
+       while (defined(my $att = shift @$want)) {
                $att = uc($att);
+               $att =~ s/\ABODY\.PEEK\[/BODY\[/; # we're read-only
                my $x = $FETCH_ATT{$att};
                if ($x) {
                        %want = (%want, %$x);
-               } elsif (!partial_prepare(\%partial, \@want, $att)) {
+               } elsif (!partial_prepare(\%partial, $want, $att)) {
                        return "$tag BAD param: $att\r\n";
                }
        }
 
        # stabilize partial order for consistency and ease-of-debugging:
        if (scalar keys %partial) {
-               $want{-partial} = [ map {
+               $want{-partial} = [ map {;
                        [ $_, @{$partial{$_}} ]
                } sort keys %partial ];
        }
@@ -629,8 +668,51 @@ sub cmd_uid_fetch ($$$;@) {
        } else {
                return "$tag BAD fetch range\r\n";
        }
-       long_response($self, \&uid_fetch_m, $tag, $ibx,
-                               \$beg, $end, $msgs, \%want);
+       [ $tag, $ibx, \$beg, $end, $msgs, \%want ];
+}
+
+sub cmd_uid_fetch ($$$;@) {
+       my ($self, $tag, $range, @want) = @_;
+       my $args = fetch_common($self, $tag, $range, \@want);
+       ref($args) eq 'ARRAY' ?
+               long_response($self, \&uid_fetch_m, @$args) :
+               $args; # error
+}
+
+sub seq_fetch_m { # long_response
+       my ($self, $tag, $ibx, $beg, $end, $msgs, $want) = @_;
+       if (!@$msgs) { # refill
+               @$msgs = @{$ibx->over->query_xover($$beg, $end)};
+               if (!@$msgs) {
+                       $self->write(\"$tag OK Fetch done\r\n");
+                       return;
+               }
+               $$beg = $msgs->[-1]->{num} + 1;
+       }
+       my $seq = $want->{-seqno}++;
+       my $cur_num = $msgs->[0]->{num};
+       if ($cur_num == $seq) { # as expected
+               git_async_msg($ibx, $msgs->[0], \&uid_fetch_cb, \@_);
+       } elsif ($cur_num > $seq) {
+               # send dummy messages until $seq catches up to $cur_num
+               my $smsg = bless { num => $seq, ts => 0 }, 'PublicInbox::Smsg';
+               unshift @$msgs, $smsg;
+               my $bref = dummy_message($seq, $ibx);
+               uid_fetch_cb($bref, undef, undef, undef, \@_);
+               $smsg; # blessed response since uid_fetch_cb requeues
+       } else { # should not happen
+               die "BUG: cur_num=$cur_num < seq=$seq";
+       }
+}
+
+sub cmd_fetch ($$$;@) {
+       my ($self, $tag, $range, @want) = @_;
+       my $args = fetch_common($self, $tag, $range, \@want);
+       ref($args) eq 'ARRAY' ? do {
+               my $want = $args->[-1];
+               $want->{-seqno} = ${$args->[2]}; # $$beg
+               long_response($self, \&seq_fetch_m, @$args)
+       } : $args; # error
 }
 
 sub uid_search_all { # long_response
@@ -740,17 +822,8 @@ sub long_step {
        } elsif ($more) { # $self->{wbuf}:
                $self->update_idle_time;
 
-               # COMPRESS users all share the same DEFLATE context.
-               # Flush it here to ensure clients don't see
-               # each other's data
-               $self->zflush;
-
-               # no recursion, schedule another call ASAP, but only after
-               # all pending writes are done.  autovivify wbuf:
-               my $new_size = push(@{$self->{wbuf}}, \&long_step);
-
-               # wbuf may be populated by $cb, no need to rearm if so:
-               $self->requeue if $new_size == 1;
+               # control passed to $more may be a GitAsyncCat object
+               requeue_once($self) if !ref($more);
        } else { # all done!
                delete $self->{long_cb};
                my $elapsed = now() - $t0;