public-inbox-pop3d - a mostly read-only POP3 server

Old account expiry has not been implemented, but it seems to
work well with both mpop(1) and getmail(1).  The strictness of
mpop was particularly helpful in ironing out bugs in our
implementation of (dreaded) message sequence numbers.

"EXPIRE 0" (RFC 2449) can theoretically save numerous "DELE"
commands, but that's untested by real-world clients.  mpop
supports PIPELINING which is effective in hiding latency,
and the core networking functionality is already well-tested
from our NNTP and IMAP implementations.

Configuration requires "publicinbox.pop3state" to point to
a directory writable by the otherwise read-only daemon.
See public-inbox-pop3d(1) manpage for more usage details.

5 files changed, 687 insertions, 10 deletions

diff --git a/lib/PublicInbox/Config.pm b/lib/PublicInbox/Config.pm
index 0f002e5e..a31b5b74 100644
--- a/lib/PublicInbox/Config.pm
+++ b/lib/PublicInbox/Config.pm
@@ -1,4 +1,4 @@
-# Copyright (C) 2014-2021 all contributors <meta@public-inbox.org>
+# Copyright (C) all contributors <meta@public-inbox.org>
 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
 #
 # Used throughout the project for reading configuration
@@ -434,7 +434,8 @@ sub _fill_ibx {
         # more things to encourage decentralization
         for my $k (qw(address altid nntpmirror imapmirror
                         coderepo hide listid url
-                        infourl watchheader nntpserver imapserver)) {
+                        infourl watchheader
+                        nntpserver imapserver pop3server)) {
                 my $v = $self->{"$pfx.$k"} // next;
                 $ibx->{$k} = _array($v);
         }
diff --git a/lib/PublicInbox/Daemon.pm b/lib/PublicInbox/Daemon.pm
index 75719c34..fbce9154 100644
--- a/lib/PublicInbox/Daemon.pm
+++ b/lib/PublicInbox/Daemon.pm
@@ -32,8 +32,8 @@ my %tls_opt; # scheme://sockname => args for IO::Socket::SSL->start_SSL
 my $reexec_pid;
 my ($uid, $gid);
 my ($default_cert, $default_key);
-my %KNOWN_TLS = ( 443 => 'https', 563 => 'nntps', 993 => 'imaps' );
-my %KNOWN_STARTTLS = ( 119 => 'nntp', 143 => 'imap' );
+my %KNOWN_TLS = (443 => 'https', 563 => 'nntps', 993 => 'imaps', 995 =>'pop3s');
+my %KNOWN_STARTTLS = (110 => 'pop3', 119 => 'nntp', 143 => 'imap');
 
 sub accept_tls_opt ($) {
         my ($opt_str) = @_;
@@ -155,7 +155,7 @@ EOF
                         $tls_opt{"$scheme://$l"} = accept_tls_opt($1);
                 } elsif (defined($default_cert)) {
                         $tls_opt{"$scheme://$l"} = accept_tls_opt('');
-                } elsif ($scheme =~ /\A(?:https|imaps|imaps)\z/) {
+                } elsif ($scheme =~ /\A(?:https|imaps|nntps|pop3s)\z/) {
                         die "$orig specified w/o cert=\n";
                 }
                 $scheme =~ /\A(http|imap|nntp|pop3)/ and
@@ -622,7 +622,7 @@ sub daemon_loop ($) {
                 $l =~ s!\A([^:]+)://!!;
                 my $scheme = $1 // '';
                 my $xn = $xnetd->{$l} // $xnetd->{''};
-                if ($scheme =~ m!\A(?:https|imaps|nntps)!) {
+                if ($scheme =~ m!\A(?:https|imaps|nntps|pop3s)!) {
                         $post_accept{$l} = tls_start_cb($v, $xn->{post_accept});
                 } elsif ($xn->{tlsd}) { # STARTTLS, $k eq '' is OK
                         $xn->{tlsd}->{accept_tls} = $v;
diff --git a/lib/PublicInbox/Inbox.pm b/lib/PublicInbox/Inbox.pm
index 1579d500..da81fb67 100644
--- a/lib/PublicInbox/Inbox.pm
+++ b/lib/PublicInbox/Inbox.pm
@@ -1,4 +1,4 @@
-# Copyright (C) 2016-2021 all contributors <meta@public-inbox.org>
+# Copyright (C) all contributors <meta@public-inbox.org>
 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
 #
 # Represents a public-inbox (which may have multiple mailing addresses)
@@ -230,8 +230,9 @@ sub base_url {
         $url;
 }
 
+# imapserver, nntpserver, and pop3server configs are used here:
 sub _x_url ($$$) {
-        my ($self, $x, $ctx) = @_; # $x is "nntp" or "imap"
+        my ($self, $x, $ctx) = @_; # $x is "imap", "nntp", or "pop3"
         # no checking for nntp_usable here, we can point entirely
         # to non-local servers or users run by a different user
         my $ns = $self->{"${x}server"} //
@@ -253,7 +254,7 @@ sub _x_url ($$$) {
                                 if ($group) {
                                         $u .= '/' if $u !~ m!/\z!;
                                         $u .= $group;
-                                } else { # n.b. IMAP uses "newsgroup"
+                                } else { # n.b. IMAP and POP3 use "newsgroup"
                                         warn <<EOM;
 publicinbox.$self->{name}.${x}mirror=$_ missing newsgroup name
 EOM
@@ -273,8 +274,9 @@ EOM
 }
 
 # my ($self, $ctx) = @_;
-sub nntp_url { $_[0]->{-nntp_url} //= _x_url($_[0], 'nntp', $_[1]) }
 sub imap_url { $_[0]->{-imap_url} //= _x_url($_[0], 'imap', $_[1]) }
+sub nntp_url { $_[0]->{-nntp_url} //= _x_url($_[0], 'nntp', $_[1]) }
+sub pop3_url { $_[0]->{-pop3_url} //= _x_url($_[0], 'pop3', $_[1]) }
 
 sub nntp_usable {
         my ($self) = @_;
diff --git a/lib/PublicInbox/POP3.pm b/lib/PublicInbox/POP3.pm
new file mode 100644
index 00000000..86123563
--- /dev/null
+++ b/lib/PublicInbox/POP3.pm
@@ -0,0 +1,443 @@
+# Copyright (C) all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+#
+# Each instance of this represents a POP3 client connected to
+# public-inbox-{netd,pop3d}.  Much of this was taken from IMAP.pm and NNTP.pm
+#
+# POP3 is one mailbox per-user, so the "USER" command is like the
+# format of -imapd and is mapped to $NEWSGROUP.$SLICE (large inboxes
+# are sliced into 50K mailboxes in both POP3 and IMAP to avoid overloading
+# clients)
+#
+# Unlike IMAP, the "$NEWSGROUP" mailbox (without $SLICE) is a rolling
+# window of the latest messages.  We can do this for POP3 since the
+# typical POP3 session is short-lived while long-lived IMAP sessions
+# would cause slices to grow on the server side without bounds.
+#
+# Like IMAP, POP3 also has per-session message sequence numbers (MSN),
+# which require mapping to UIDs.  The offset of an entry into our
+# per-client cache is: (MSN-1)
+#
+# fields:
+# - uuid - 16-byte (binary) UUID representation (before successful login)
+# - cache - one-dimentional arrayref of (UID, bytesize, oidhex)
+# - nr_dele - number of deleted messages
+# - expire - string of packed unsigned short offsets
+# - user_id - user-ID mapped to UUID (on successful login + lock)
+# - txn_max_uid - for storing max deleted UID persistently
+# - ibx - PublicInbox::Inbox object
+# - slice - unsigned integer slice number (0..Inf), -1 => latest
+# - salt - pre-auth for APOP
+# - uid_dele - maximum deleted from previous session at login (NNTP ARTICLE)
+# - uid_base - base UID for mailbox slice (0-based) (same as IMAP)
+package PublicInbox::POP3;
+use v5.12;
+use parent qw(PublicInbox::DS);
+use PublicInbox::Syscall qw(EPOLLIN EPOLLONESHOT);
+use PublicInbox::GitAsyncCat;
+use PublicInbox::DS qw(now);
+use Errno qw(EAGAIN);
+use Digest::MD5 qw(md5);
+use PublicInbox::IMAP; # for UID slice stuff
+
+use constant {
+        LINE_MAX => 512, # XXX unsure
+};
+
+# XXX FIXME: duplicated stuff from NNTP.pm and IMAP.pm
+
+sub err ($$;@) {
+        my ($self, $fmt, @args) = @_;
+        printf { $self->{pop3d}->{err} } $fmt."\n", @args;
+}
+
+sub out ($$;@) {
+        my ($self, $fmt, @args) = @_;
+        printf { $self->{pop3d}->{out} } $fmt."\n", @args;
+}
+
+sub zflush {} # noop
+
+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 long_step {
+        my ($self) = @_;
+        # wbuf is unset or empty, here; {long} may add to it
+        my ($fd, $cb, $t0, @args) = @{$self->{long_cb}};
+        my $more = eval { $cb->($self, @args) };
+        if ($@ || !$self->{sock}) { # something bad happened...
+                delete $self->{long_cb};
+                my $elapsed = now() - $t0;
+                if ($@) {
+                        err($self,
+                            "%s during long response[$fd] - %0.6f",
+                            $@, $elapsed);
+                }
+                out($self, " deferred[$fd] aborted - %0.6f", $elapsed);
+                $self->close;
+        } elsif ($more) { # $self->{wbuf}:
+                # control passed to ibx_async_cat if $more == \undef
+                requeue_once($self) if !ref($more);
+        } else { # all done!
+                delete $self->{long_cb};
+                my $elapsed = now() - $t0;
+                my $fd = fileno($self->{sock});
+                out($self, " deferred[$fd] done - %0.6f", $elapsed);
+                my $wbuf = $self->{wbuf}; # do NOT autovivify
+                $self->requeue unless $wbuf && @$wbuf;
+        }
+}
+
+sub long_response ($$;@) {
+        my ($self, $cb, @args) = @_; # cb returns true if more, false if done
+        my $sock = $self->{sock} or return;
+        # make sure we disable reading during a long response,
+        # clients should not be sending us stuff and making us do more
+        # work while we are stream a response to them
+        $self->{long_cb} = [ fileno($sock), $cb, now(), @args ];
+        long_step($self); # kick off!
+        undef;
+}
+
+sub _greet ($) {
+        my ($self) = @_;
+        my $s = $self->{salt} = sprintf('%x.%x', int(rand(0x7fffffff)), time);
+        $self->write("+OK POP3 server ready <$s\@public-inbox>\r\n");
+}
+
+sub new ($$$) {
+        my ($class, $sock, $pop3d) = @_;
+        my $self = bless { pop3d => $pop3d }, __PACKAGE__;
+        my $ev = EPOLLIN;
+        my $wbuf;
+        if ($sock->can('accept_SSL') && !$sock->accept_SSL) {
+                return CORE::close($sock) if $! != EAGAIN;
+                $ev = PublicInbox::TLS::epollbit() or return CORE::close($sock);
+                $wbuf = [ \&PublicInbox::DS::accept_tls_step, \&_greet ];
+        }
+        $self->SUPER::new($sock, $ev | EPOLLONESHOT);
+        if ($wbuf) {
+                $self->{wbuf} = $wbuf;
+        } else {
+                _greet($self);
+        }
+        $self;
+}
+
+# POP user is $UUID1@$NEWSGROUP.$SLICE
+sub cmd_user ($$) {
+        my ($self, $mailbox) = @_;
+        $self->{salt} // return \"-ERR already authed\r\n";
+        $mailbox =~ s/\A([a-f0-9\-]+)\@//i or
+                return \"-ERR no UUID@ in mailbox name\r\n";
+        my $user = $1;
+        $user =~ tr/-//d; # most have dashes, some (dbus-uuidgen) don't
+        $user =~ m!\A[a-f0-9]{32}\z!i or return \"-ERR user has no UUID\r\n";
+        my $slice;
+        $mailbox =~ s/\.([0-9]+)\z// and $slice = $1 + 0;
+        my $ibx = $self->{pop3d}->{pi_cfg}->lookup_newsgroup($mailbox) //
+                return \"-ERR $mailbox does not exist\r\n";
+        my $uidmax = $ibx->mm(1)->num_highwater // 0;
+        if (defined $slice) {
+                my $max = int($uidmax / PublicInbox::IMAP::UID_SLICE);
+                my $tip = "$mailbox.$max";
+                return \"-ERR $mailbox.$slice does not exist ($tip does)\r\n"
+                        if $slice > $max;
+                $self->{uid_base} = $slice * PublicInbox::IMAP::UID_SLICE;
+                $self->{slice} = $slice;
+        } else { # latest 50K messages
+                my $base = $uidmax - PublicInbox::IMAP::UID_SLICE;
+                $self->{uid_base} = $base < 0 ? 0 : $base;
+                $self->{slice} = -1;
+        }
+        $self->{ibx} = $ibx;
+        $self->{uuid} = pack('H*', $user); # deleted by _login_ok
+        $slice //= '(latest)';
+        \"+OK $ibx->{newsgroup} slice=$slice selected\r\n";
+}
+
+sub _login_ok ($) {
+        my ($self) = @_;
+        if ($self->{pop3d}->lock_mailbox($self)) {
+                $self->{uid_max} = $self->{ibx}->over(1)->max;
+                \"+OK logged in\r\n";
+        } else {
+                \"-ERR unable to lock maildrop\r\n";
+        }
+}
+
+sub cmd_apop {
+        my ($self, $mailbox, $hex) = @_;
+        my $res = cmd_user($self, $mailbox); # sets {uuid}
+        return $res if substr($$res, 0, 1) eq '-';
+        my $s = delete($self->{salt}) // die 'BUG: salt missing';
+        return _login_ok($self) if md5("<$s\@public-inbox>anonymous") eq
+                                pack('H*', $hex);
+        $self->{salt} = $s;
+        \"-ERR APOP password mismatch\r\n";
+}
+
+sub cmd_pass {
+        my ($self, $pass) = @_;
+        $self->{ibx} // return \"-ERR mailbox unspecified\r\n";
+        my $s = delete($self->{salt}) // return \"-ERR already authed\r\n";
+        return _login_ok($self) if $pass eq 'anonymous';
+        $self->{salt} = $s;
+        \"-ERR password is not `anonymous'\r\n";
+}
+
+sub cmd_stls {
+        my ($self) = @_;
+        my $sock = $self->{sock} or return;
+        return \"-ERR TLS already enabled\r\n" if $sock->can('stop_SSL');
+        my $opt = $self->{pop3d}->{accept_tls} or
+                return \"-ERR can't start TLS negotiation\r\n";
+        $self->write(\"+OK begin TLS negotiation now\r\n");
+        $self->{sock} = IO::Socket::SSL->start_SSL($sock, %$opt);
+        $self->requeue if PublicInbox::DS::accept_tls_step($self);
+        undef;
+}
+
+sub need_txn ($) {
+        exists($_[0]->{salt}) ? \"-ERR not in TRANSACTION\r\n" : undef;
+}
+
+sub _stat_cache ($) {
+        my ($self) = @_;
+        my ($beg, $end) = (($self->{uid_dele} // -1) + 1, $self->{uid_max});
+        PublicInbox::IMAP::uid_clamp($self, \$beg, \$end);
+        my $opt = { limit => PublicInbox::IMAP::UID_SLICE };
+        my $m = $self->{ibx}->over(1)->do_get(<<'', $opt, $beg, $end);
+SELECT num,ddd FROM over WHERE num >= ? AND num <= ?
+ORDER BY num ASC
+
+        [ map { ($_->{num}, $_->{bytes} + 0, $_->{blob}) } @$m ];
+}
+
+sub cmd_stat {
+        my ($self) = @_;
+        my $err; $err = need_txn($self) and return $err;
+        my $cache = $self->{cache} //= _stat_cache($self);
+        my $tot = 0;
+        for (my $i = 1; $i < scalar(@$cache); $i += 3) { $tot += $cache->[$i] }
+        my $nr = @$cache / 3 - ($self->{nr_dele} // 0);
+        "+OK $nr $tot\r\n";
+}
+
+# for LIST and UIDL
+sub _list {
+        my ($desc, $idx, $self, $msn) = @_;
+        my $err; $err = need_txn($self) and return $err;
+        my $cache = $self->{cache} //= _stat_cache($self);
+        if (defined $msn) {
+                my $base_off = ($msn - 1) * 3;
+                my $val = $cache->[$base_off + $idx] //
+                                return \"-ERR no such message\r\n";
+                "+OK $desc listing follows\r\n$msn $val\r\n.\r\n";
+        } else { # always +OK, even if no messages
+                my $res = "+OK $desc listing follows\r\n";
+                my $msn = 0;
+                for (my $i = 0; $i < scalar(@$cache); $i += 3) {
+                        ++$msn;
+                        defined($cache->[$i]) and
+                                $res .= "$msn $cache->[$i + $idx]\r\n";
+                }
+                $res .= ".\r\n";
+        }
+}
+
+sub cmd_list { _list('scan', 1, @_) }
+sub cmd_uidl { _list('unique-id', 2, @_) }
+
+sub mark_dele ($$) {
+        my ($self, $off) = @_;
+        my $base_off = $off * 3;
+        my $cache = $self->{cache};
+        my $uid = $cache->[$base_off] // return; # already deleted
+
+        my $old = $self->{txn_max_uid} //= $uid;
+        $self->{txn_max_uid} = $uid if $uid > $old;
+
+        $cache->[$base_off] = undef; # clobber UID
+        $cache->[$base_off + 1] = 0; # zero bytes (simplifies cmd_stat)
+        $cache->[$base_off + 2] = undef; # clobber oidhex
+        ++$self->{nr_dele};
+}
+
+sub retr_cb { # called by git->cat_async via ibx_async_cat
+        my ($bref, $oid, $type, $size, $args) = @_;
+        my ($self, $off, $top_nr) = @$args;
+        my $hex = $self->{cache}->[$off * 3 + 2] //
+                die "BUG: no hex (oid=$oid)";
+        if (!defined($oid)) {
+                # it's possible to have TOCTOU if an admin runs
+                # public-inbox-(edit|purge), just move onto the next message
+                warn "E: $hex missing in $self->{ibx}->{inboxdir}\n";
+                $self->write(\"-ERR no such message\r\n");
+                return $self->requeue;
+        } elsif ($hex ne $oid) {
+                $self->close;
+                die "BUG: $hex != $oid";
+        }
+        PublicInbox::IMAP::to_crlf_full($bref);
+        if (defined $top_nr) {
+                my ($hdr, $bdy) = split(/\r\n\r\n/, $$bref, 2);
+                $bref = \$hdr;
+                $hdr .= "\r\n\r\n";
+                my @tmp = split(/^/m, $bdy);
+                $hdr .= join('', splice(@tmp, 0, $top_nr));
+        }
+        $$bref =~ s/^\./../gms;
+        $$bref .= substr($$bref, -2, 2) eq "\r\n" ? ".\r\n" : "\r\n.\r\n";
+        $self->msg_more("+OK message follows\r\n");
+        $self->write($bref);
+        $self->{expire} .= pack('S', $off + 1) if exists $self->{expire};
+        $self->requeue;
+}
+
+sub cmd_retr {
+        my ($self, $msn, $top_nr) = @_;
+        return \"-ERR lines must be a non-negative number\r\n" if
+                        (defined($top_nr) && $top_nr !~ /\A[0-9]+\z/);
+        my $err; $err = need_txn($self) and return $err;
+        my $cache = $self->{cache} //= _stat_cache($self);
+        my $off = $msn - 1;
+        my $hex = $cache->[$off * 3 + 2] // return \"-ERR no such message\r\n";
+        ${ibx_async_cat($self->{ibx}, $hex, \&retr_cb,
+                        [ $self, $off, $top_nr ])};
+}
+
+sub cmd_noop { $_[0]->write(\"+OK\r\n") }
+
+sub cmd_rset {
+        my ($self) = @_;
+        my $err; $err = need_txn($self) and return $err;
+        delete $self->{cache};
+        delete $self->{txn_max_uid};
+        \"+OK\r\n";
+}
+
+sub cmd_dele {
+        my ($self, $msn) = @_;
+        my $err; $err = need_txn($self) and return $err;
+        $self->{cache} //= _stat_cache($self);
+        $msn =~ /\A[1-9][0-9]*\z/ or return \"-ERR no such message\r\n";
+        mark_dele($self, $msn - 1) ? \"+OK\r\n" : \"-ERR no such message\r\n";
+}
+
+# RFC 2449
+sub cmd_capa {
+        my ($self) = @_;
+        $self->{expire} = ''; # "EXPIRE 0" allows clients to avoid DELE commands
+        \<<EOM;
++OK Capability list follows\r
+TOP\r
+USER\r
+PIPELINING\r
+UIDL\r
+EXPIRE 0\r
+.\r
+EOM
+}
+
+sub close {
+        my ($self) = @_;
+        $self->{pop3d}->unlock_mailbox($self);
+        $self->SUPER::close;
+}
+
+sub cmd_quit {
+        my ($self) = @_;
+        if (defined(my $txn_id = $self->{txn_id})) {
+                my $user_id = $self->{user_id} // die 'BUG: no {user_id}';
+                if (my $exp = delete $self->{expire}) {
+                        mark_dele($self, $_) for unpack('S*', $exp);
+                }
+                my $dbh = $self->{pop3d}->{-state_dbh};
+                my $lk = $self->{pop3d}->lock_for_scope;
+                my $sth;
+                $dbh->begin_work;
+
+                if (defined $self->{txn_max_uid}) {
+                        $sth = $dbh->prepare_cached(<<'');
+UPDATE deletes SET uid_dele = ? WHERE txn_id = ? AND uid_dele < ?
+
+                        $sth->execute($self->{txn_max_uid}, $txn_id,
+                                        $self->{txn_max_uid});
+                }
+                $sth = $dbh->prepare_cached(<<'');
+UPDATE users SET last_seen = ? WHERE user_id = ?
+
+                $sth->execute(time, $user_id);
+                $dbh->commit;
+        }
+        $self->write(\"+OK public-inbox POP3 server signing off\r\n");
+        $self->close;
+        undef;
+}
+
+# returns 1 if we can continue, 0 if not due to buffered writes or disconnect
+sub process_line ($$) {
+        my ($self, $l) = @_;
+        my ($req, @args) = split(/[ \t]+/, $l);
+        return 1 unless defined($req); # skip blank line
+        $req = $self->can('cmd_'.lc($req));
+        my $res = $req ? eval { $req->($self, @args) } :
+                \"-ERR command not recognized\r\n";
+        my $err = $@;
+        if ($err && $self->{sock}) {
+                chomp($l);
+                err($self, 'error from: %s (%s)', $l, $err);
+                $res = \"-ERR program fault - command not performed\r\n";
+        }
+        defined($res) ? $self->write($res) : 0;
+}
+
+# callback used by PublicInbox::DS for any (e)poll (in/out/hup/err)
+sub event_step {
+        my ($self) = @_;
+        return unless $self->flush_write && $self->{sock} && !$self->{long_cb};
+
+        # only read more requests if we've drained the write buffer,
+        # otherwise we can be buffering infinitely w/o backpressure
+        my $rbuf = $self->{rbuf} // \(my $x = '');
+        my $line = index($$rbuf, "\n");
+        while ($line < 0) {
+                return $self->close if length($$rbuf) >= LINE_MAX;
+                $self->do_read($rbuf, LINE_MAX, length($$rbuf)) or return;
+                $line = index($$rbuf, "\n");
+        }
+        $line = substr($$rbuf, 0, $line + 1, '');
+        $line =~ s/\r?\n\z//s;
+        return $self->close if $line =~ /[[:cntrl:]]/s;
+        my $t0 = now();
+        my $fd = fileno($self->{sock}); # may become invalid after process_line
+        my $r = eval { process_line($self, $line) };
+        my $pending = $self->{wbuf} ? ' pending' : '';
+        out($self, "[$fd] %s - %0.6f$pending - $r", $line, now() - $t0);
+        return $self->close if $r < 0;
+        $self->rbuf_idle($rbuf);
+
+        # maybe there's more pipelined data, or we'll have
+        # to register it for socket-readiness notifications
+        $self->requeue unless $pending;
+}
+
+no warnings 'once';
+*cmd_top = \&cmd_retr;
+
+1;
diff --git a/lib/PublicInbox/POP3D.pm b/lib/PublicInbox/POP3D.pm
new file mode 100644
index 00000000..c7bf1755
--- /dev/null
+++ b/lib/PublicInbox/POP3D.pm
@@ -0,0 +1,231 @@
+# Copyright (C) all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+
+# represents an POP3D (currently a singleton)
+package PublicInbox::POP3D;
+use v5.12;
+use parent qw(PublicInbox::Lock);
+use DBI qw(:sql_types); # SQL_BLOB
+use Carp ();
+use File::Temp 0.19 (); # 0.19 for ->newdir
+use PublicInbox::Config;
+use PublicInbox::POP3;
+use PublicInbox::Syscall;
+use File::Temp 0.19 (); # 0.19 for ->newdir
+use File::FcntlLock;
+use Fcntl qw(F_SETLK F_UNLCK F_WRLCK SEEK_SET);
+
+sub new {
+        my ($cls, $pi_cfg) = @_;
+        $pi_cfg //= PublicInbox::Config->new;
+        my $d = $pi_cfg->{'publicinbox.pop3state'} //
+                die "publicinbox.pop3state undefined\n";
+        -d $d or do {
+                require File::Path;
+                File::Path::make_path($d, { mode => 0700 });
+                PublicInbox::Syscall::nodatacow_dir($d);
+        };
+        bless {
+                err => \*STDERR,
+                out => \*STDOUT,
+                pi_cfg => $pi_cfg,
+                lock_path => "$d/db.lock", # PublicInbox::Lock to protect SQLite
+                # interprocess lock is the $pop3state/txn.locks file
+                # txn_locks => {}, # intraworker locks
+                # accept_tls => { SSL_server => 1, ..., SSL_reuse_ctx => ... }
+        }, $cls;
+}
+
+sub refresh_groups { # PublicInbox::Daemon callback
+        my ($self, $sig) = @_;
+        # TODO share pi_cfg with nntpd/imapd inside -netd
+        my $new = PublicInbox::Config->new;
+        my $old = $self->{pi_cfg};
+        my $s = 'publicinbox.pop3state';
+        $new->{$s} //= $old->{$s};
+        if ($new->{$s} ne $old->{$s}) {
+                warn <<EOM;
+$s changed: `$old->{$s}' => `$new->{$s}', config reload ignored
+EOM
+        } else {
+                $self->{pi_cfg} = $new;
+        }
+}
+
+# persistent tables
+sub create_state_tables ($$) {
+        my ($self, $dbh) = @_;
+
+        $dbh->do(<<''); # map publicinbox.<name>.newsgroup to integers
+CREATE TABLE IF NOT EXISTS newsgroups (
+        newsgroup_id INTEGER PRIMARY KEY NOT NULL,
+        newsgroup VARBINARY NOT NULL,
+        UNIQUE (newsgroup) )
+
+        # the $NEWSGROUP_NAME.$SLICE_INDEX is part of the POP3 username;
+        # POP3 has no concept of folders/mailboxes like IMAP/JMAP
+        $dbh->do(<<'');
+CREATE TABLE IF NOT EXISTS mailboxes (
+        mailbox_id INTEGER PRIMARY KEY NOT NULL,
+        newsgroup_id INTEGER NOT NULL REFERENCES newsgroups,
+        slice INTEGER NOT NULL, /* -1 for most recent slice */
+        UNIQUE (newsgroup_id, slice) )
+
+        $dbh->do(<<''); # actual users are differentiated by their UUID
+CREATE TABLE IF NOT EXISTS users (
+        user_id INTEGER PRIMARY KEY NOT NULL,
+        uuid VARBINARY NOT NULL,
+        last_seen INTEGER NOT NULL, /* to expire idle accounts */
+        UNIQUE (uuid) )
+
+        # we only track the highest-numbered deleted message per-UUID@mailbox
+        $dbh->do(<<'');
+CREATE TABLE IF NOT EXISTS deletes (
+        txn_id INTEGER PRIMARY KEY NOT NULL, /* -1 == txn lock offset */
+        user_id INTEGER NOT NULL REFERENCES users,
+        mailbox_id INTEGER NOT NULL REFERENCES mailboxes,
+        uid_dele INTEGER NOT NULL DEFAULT -1, /* IMAP UID, NNTP article */
+        UNIQUE(user_id, mailbox_id) )
+
+}
+
+sub state_dbh_new {
+        my ($self) = @_;
+        my $f = "$self->{pi_cfg}->{'publicinbox.pop3state'}/db.sqlite3";
+        my $creat = !-s $f;
+        if ($creat) {
+                open my $fh, '+>>', $f or Carp::croak "open($f): $!";
+                PublicInbox::Syscall::nodatacow_fh($fh);
+        }
+
+        my $dbh = DBI->connect("dbi:SQLite:dbname=$f",'','', {
+                AutoCommit => 1,
+                RaiseError => 1,
+                PrintError => 0,
+                sqlite_use_immediate_transaction => 1,
+                sqlite_see_if_its_a_number => 1,
+        });
+        $dbh->do('PRAGMA journal_mode = WAL') if $creat;
+        $dbh->do('PRAGMA foreign_keys = ON'); # don't forget this
+
+        # ensure the interprocess fcntl lock file exists
+        $f = "$self->{pi_cfg}->{'publicinbox.pop3state'}/txn.locks";
+        open my $fh, '+>>', $f or Carp::croak("open($f): $!");
+        $self->{txn_fh} = $fh;
+
+        create_state_tables($self, $dbh);
+        $dbh;
+}
+
+sub lock_mailbox {
+        my ($self, $pop3) = @_; # pop3 - PublicInbox::POP3 client object
+        my $lk = $self->lock_for_scope; # lock the SQLite DB, only
+        my $dbh = $self->{-state_dbh} //= state_dbh_new($self);
+        my ($user_id, $ngid, $mbid, $txn_id);
+        my $uuid = delete $pop3->{uuid};
+        $dbh->begin_work;
+
+        # 1. make sure the user exists, update `last_seen'
+        my $sth = $dbh->prepare_cached(<<'');
+INSERT OR IGNORE INTO users (uuid, last_seen) VALUES (?,?)
+
+        $sth->bind_param(1, $uuid, SQL_BLOB);
+        $sth->bind_param(2, time);
+        if ($sth->execute == 0) { # existing user
+                $sth = $dbh->prepare_cached(<<'', undef, 1);
+SELECT user_id FROM users WHERE uuid = ?
+
+                $sth->bind_param(1, $uuid, SQL_BLOB);
+                $sth->execute;
+                $user_id = $sth->fetchrow_array //
+                        die 'BUG: user '.unpack('H*', $uuid).' not found';
+                $sth = $dbh->prepare_cached(<<'');
+UPDATE users SET last_seen = ? WHERE user_id = ?
+
+                $sth->execute(time, $user_id);
+        } else { # new user
+                $user_id = $dbh->last_insert_id(undef, undef,
+                                                'users', 'user_id')
+        }
+
+        # 2. make sure the newsgroup has an integer ID
+        $sth = $dbh->prepare_cached(<<'');
+INSERT OR IGNORE INTO newsgroups (newsgroup) VALUES (?)
+
+        my $ng = $pop3->{ibx}->{newsgroup};
+        $sth->bind_param(1, $ng, SQL_BLOB);
+        if ($sth->execute == 0) {
+                $sth = $dbh->prepare_cached(<<'', undef, 1);
+SELECT newsgroup_id FROM newsgroups WHERE newsgroup = ?
+
+                $sth->bind_param(1, $ng, SQL_BLOB);
+                $sth->execute;
+                $ngid = $sth->fetchrow_array // die "BUG: `$ng' not found";
+        } else {
+                $ngid = $dbh->last_insert_id(undef, undef,
+                                                'newsgroups', 'newsgroup_id');
+        }
+
+        # 3. ensure the mailbox exists
+        $sth = $dbh->prepare_cached(<<'');
+INSERT OR IGNORE INTO mailboxes (newsgroup_id, slice) VALUES (?,?)
+
+        if ($sth->execute($ngid, $pop3->{slice}) == 0) {
+                $sth = $dbh->prepare_cached(<<'', undef, 1);
+SELECT mailbox_id FROM mailboxes WHERE newsgroup_id = ? AND slice = ?
+
+                $sth->execute($ngid, $pop3->{slice});
+                $mbid = $sth->fetchrow_array //
+                        die "BUG: mailbox_id for $ng.$pop3->{slice} not found";
+        } else {
+                $mbid = $dbh->last_insert_id(undef, undef,
+                                                'mailboxes', 'mailbox_id');
+        }
+
+        # 4. ensure the (max) deletes row exists for locking
+        $sth = $dbh->prepare_cached(<<'');
+INSERT OR IGNORE INTO deletes (user_id,mailbox_id) VALUES (?,?)
+
+        if ($sth->execute($user_id, $mbid) == 0) {
+                $sth = $dbh->prepare_cached(<<'', undef, 1);
+SELECT txn_id,uid_dele FROM deletes WHERE user_id = ? AND mailbox_id = ?
+
+                $sth->execute($user_id, $mbid);
+                ($txn_id, $pop3->{uid_dele}) = $sth->fetchrow_array;
+        } else {
+                $txn_id = $dbh->last_insert_id(undef, undef,
+                                                'deletes', 'txn_id');
+        }
+        $dbh->commit;
+
+        # see if it's locked by the same worker:
+        return if $self->{txn_locks}->{$txn_id};
+
+        # see if it's locked by another worker:
+        my $fs = File::FcntlLock->new;
+        $fs->l_type(F_WRLCK);
+        $fs->l_whence(SEEK_SET);
+        $fs->l_start($txn_id - 1);
+        $fs->l_len(1);
+        $fs->lock($self->{txn_fh}, F_SETLK) or return;
+
+        $pop3->{user_id} = $user_id;
+        $pop3->{txn_id} = $txn_id;
+        $self->{txn_locks}->{$txn_id} = 1;
+}
+
+sub unlock_mailbox {
+        my ($self, $pop3) = @_;
+        my $txn_id = delete($pop3->{txn_id}) // return;
+        delete $self->{txn_locks}->{$txn_id}; # same worker
+
+        # other workers
+        my $fs = File::FcntlLock->new;
+        $fs->l_type(F_UNLCK);
+        $fs->l_whence(SEEK_SET);
+        $fs->l_start($txn_id - 1);
+        $fs->l_len(1);
+        $fs->lock($self->{txn_fh}, F_SETLK) or die "F_UNLCK: $!";
+}
+
+1;