diff options
Diffstat (limited to 'lib/PublicInbox')
| -rw-r--r-- | lib/PublicInbox/Config.pm | 5 | ||||
| -rw-r--r-- | lib/PublicInbox/Daemon.pm | 8 | ||||
| -rw-r--r-- | lib/PublicInbox/Inbox.pm | 10 | ||||
| -rw-r--r-- | lib/PublicInbox/POP3.pm | 443 | ||||
| -rw-r--r-- | lib/PublicInbox/POP3D.pm | 231 |
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; |