# Copyright (C) all contributors # License: AGPL-3.0+ # for maintaining synchronization between lei/store <=> Maildir|MH|IMAP|JMAP package PublicInbox::LeiMailSync; use strict; use v5.10.1; use parent qw(PublicInbox::Lock); use PublicInbox::Compat qw(uniqstr); use DBI qw(:sql_types); # SQL_BLOB use PublicInbox::ContentHash qw(git_sha); use Carp (); use PublicInbox::Git qw(%HEXLEN2SHA); use PublicInbox::IO qw(read_all); sub dbh_new { my ($self) = @_; my $f = $self->{filename}; my $creat = !-s $f; if ($creat) { require PublicInbox::Syscall; 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, }); # no sqlite_unicode, here, all strings are binary create_tables($self, $dbh); $dbh->do('PRAGMA journal_mode = WAL') if $creat; $dbh->do('PRAGMA case_sensitive_like = ON'); $dbh; } sub new { my ($cls, $f) = @_; bless { filename => $f, fmap => {}, lock_path => "$f.flock", }, $cls; } sub lms_write_prepare { ($_[0]->{dbh} //= dbh_new($_[0])); $_[0] } sub lms_pause { my ($self) = @_; $self->{fmap} = {}; my $dbh = delete $self->{dbh}; eval { $dbh->do('PRAGMA optimize') } if $dbh; } sub create_tables { my ($self, $dbh) = @_; my $lk = $self->lock_for_scope; $dbh->do(<<''); CREATE TABLE IF NOT EXISTS folders ( fid INTEGER PRIMARY KEY, loc VARBINARY NOT NULL, /* URL;UIDVALIDITY=$N or $TYPE:/pathname */ UNIQUE (loc) ) $dbh->do(<<''); CREATE TABLE IF NOT EXISTS blob2num ( oidbin VARBINARY NOT NULL, fid INTEGER NOT NULL, /* folder ID */ uid INTEGER NOT NULL, /* NNTP article number, IMAP UID, MH number */ /* not UNIQUE(fid, uid), since we may have broken servers */ UNIQUE (oidbin, fid, uid) ) # speeds up LeiImport->ck_update_kw (for "lei import") by 5-6x: $dbh->do(<<''); CREATE INDEX IF NOT EXISTS idx_fid_uid ON blob2num(fid,uid) $dbh->do(<<''); CREATE TABLE IF NOT EXISTS blob2name ( oidbin VARBINARY NOT NULL, fid INTEGER NOT NULL, /* folder ID */ name VARBINARY NOT NULL, /* Maildir basename, JMAP blobId */ /* not UNIQUE(fid, name), since we may have broken software */ UNIQUE (oidbin, fid, name) ) # speeds up LeiImport->pmdir_cb (for "lei import") by ~6x: $dbh->do(<<''); CREATE INDEX IF NOT EXISTS idx_fid_name ON blob2name(fid,name) } # used to fixup pre-1.7.0 folders sub update_fid ($$$) { my ($dbh, $fid, $loc) = @_; my $sth = $dbh->prepare(<<''); UPDATE folders SET loc = ? WHERE fid = ? $sth->bind_param(1, $loc, SQL_BLOB); $sth->bind_param(2, $fid); $sth->execute; } sub get_fid ($$$) { my ($sth, $folder, $dbh) = @_; $sth->bind_param(1, $folder, SQL_BLOB); $sth->execute; my ($fid) = $sth->fetchrow_array; if (defined $fid) { # for downgrade+upgrade (1.8 -> 1.7 -> 1.8) my $del = $dbh->prepare_cached(<<''); DELETE FROM folders WHERE loc = ? AND fid != ? $del->execute($folder, $fid); } else { $sth->bind_param(1, $folder, SQL_VARCHAR); $sth->execute; # fixup old stuff ($fid) = $sth->fetchrow_array; update_fid($dbh, $fid, $folder) if defined($fid); } $fid; } sub fid_for { my ($self, $folder, $creat) = @_; my $dbh = $self->{dbh} //= dbh_new($self); my $sth = $dbh->prepare_cached(<<'', undef, 1); SELECT fid FROM folders WHERE loc = ? LIMIT 1 my $fid = get_fid($sth, $folder, $dbh); return $fid if defined($fid); # caller had trailing slash (LeiToMail) if ($folder =~ s!\A((?:maildir|mh):.*?)/+\z!$1!i) { $fid = get_fid($sth, $folder, $dbh); if (defined $fid) { update_fid($dbh, $fid, $folder); return $fid; } # sometimes we stored trailing slash.. } elsif ($folder =~ m!\A(?:maildir|mh):!i) { $fid = get_fid($sth, $folder, $dbh); if (defined $fid) { update_fid($dbh, $fid, $folder); return $fid; } } elsif ($creat && $folder =~ m!\Aimaps?://!i) { require PublicInbox::URIimap; my $uri = PublicInbox::URIimap->new($folder); $uri->uidvalidity // Carp::croak("BUG: $folder has no UIDVALIDITY"); defined($uri->uid) and Carp::confess("BUG: $folder has UID"); } return unless $creat; ($fid) = $dbh->selectrow_array('SELECT MAX(fid) FROM folders'); $fid += 1; # in case we're reusing, clobber existing stale refs: $dbh->do('DELETE FROM blob2name WHERE fid = ?', undef, $fid); $dbh->do('DELETE FROM blob2num WHERE fid = ?', undef, $fid); $sth = $dbh->prepare('INSERT INTO folders (fid, loc) VALUES (?, ?)'); $sth->bind_param(1, $fid); $sth->bind_param(2, $folder, SQL_BLOB); $sth->execute; $fid; } sub add_folders { my ($self, @folders) = @_; my $lk = $self->lock_for_scope; for my $f (@folders) { $self->{fmap}->{$f} //= fid_for($self, $f, 1) } } sub set_src { my ($self, $oidbin, $folder, $id) = @_; my $lk = $self->lock_for_scope; my $fid = $self->{fmap}->{$folder} //= fid_for($self, $folder, 1); my $dbh = $self->{dbh}; my ($sth, @param3, $del_old); if (ref($id)) { # scalar name @param3 = ($$id, SQL_BLOB); $sth = $dbh->prepare_cached(<<''); INSERT OR IGNORE INTO blob2name (oidbin, fid, name) VALUES (?, ?, ?) $del_old = $dbh->prepare_cached(<<''); DELETE FROM blob2name WHERE oidbin = ? AND fid = ? AND name = ? } else { # numeric ID (IMAP UID, MH number) @param3 = ($id); $sth = $dbh->prepare_cached(<<''); INSERT OR IGNORE INTO blob2num (oidbin, fid, uid) VALUES (?, ?, ?) $del_old = $dbh->prepare_cached(<<''); DELETE FROM blob2num WHERE oidbin = ? AND fid = ? AND uid = ? } $sth->bind_param(1, $oidbin, SQL_BLOB); $sth->bind_param(2, $fid); $sth->bind_param(3, @param3); my $ret = $sth->execute; $del_old->execute($oidbin, $fid, $param3[0]); $ret; } sub clear_src { my ($self, $folder, $id) = @_; my $lk = $self->lock_for_scope; my $fid = $self->{fmap}->{$folder} //= fid_for($self, $folder, 1); my ($sth, @param3); if (ref($id)) { # scalar name @param3 = ($$id, SQL_BLOB); $sth = $self->{dbh}->prepare_cached(<<''); DELETE FROM blob2name WHERE fid = ? AND name = ? } else { @param3 = ($id); $sth = $self->{dbh}->prepare_cached(<<''); DELETE FROM blob2num WHERE fid = ? AND uid = ? } $sth->bind_param(1, $fid); $sth->bind_param(2, @param3); my $ret = $sth->execute; # older versions may not have used SQL_BLOB: if (defined($ret) && $ret == 0 && scalar(@param3) == 2) { $sth->bind_param(1, $fid); $sth->bind_param(2, $param3[0]); $ret = $sth->execute; } $ret; } # Maildir-only sub mv_src { my ($self, $folder, $oidbin, $id, $newbn) = @_; my $lk = $self->lock_for_scope; my $fid = $self->{fmap}->{$folder} //= fid_for($self, $folder, 1); $self->{dbh}->begin_work; my $sth = $self->{dbh}->prepare_cached(<<''); UPDATE blob2name SET name = ? WHERE fid = ? AND oidbin = ? AND name = ? # eval since unique constraint may fail due to race $sth->bind_param(1, $newbn, SQL_BLOB); $sth->bind_param(2, $fid); $sth->bind_param(3, $oidbin, SQL_BLOB); $sth->bind_param(4, $$id, SQL_BLOB); my $nr = eval { $sth->execute }; if (!defined($nr) || $nr == 0) { # $nr may be `0E0' # delete from old, pre-SQL_BLOB rows: my $del_old = $self->{dbh}->prepare_cached(<<''); DELETE FROM blob2name WHERE fid = ? AND oidbin = ? AND name = ? $del_old->execute($fid, $oidbin, $$id); # missing-OK $del_old->execute($fid, $oidbin, $newbn); # ditto # may race with a clear_src, ensure new value exists $sth = $self->{dbh}->prepare_cached(<<''); INSERT OR IGNORE INTO blob2name (oidbin, fid, name) VALUES (?, ?, ?) $sth->bind_param(1, $oidbin, SQL_BLOB); $sth->bind_param(2, $fid); $sth->bind_param(3, $newbn, SQL_BLOB); $sth->execute; } $self->{dbh}->commit; } # read-only, iterates every oidbin + UID or name for a given folder sub each_src { my ($self, $folder, $cb, @args) = @_; my $dbh = $self->{dbh} //= dbh_new($self); my ($fid, @rng); my $and_ge_le = ''; if (ref($folder) eq 'HASH') { $fid = $folder->{fid} // die "BUG: no `fid'"; @rng = grep(defined, @$folder{qw(min max)}); $and_ge_le = 'AND uid >= ? AND uid <= ?' if @rng; } else { $fid = $self->{fmap}->{$folder} //= fid_for($self, $folder) // return; } # minimize implicit txn time to avoid blocking writers by # batching SELECTs. This looks wonky but is necessary since # $cb-> may access the DB on its own. my $ary = $dbh->selectall_arrayref(<<"", undef, $fid, @rng); SELECT _rowid_,oidbin,uid FROM blob2num WHERE fid = ? $and_ge_le ORDER BY _rowid_ ASC LIMIT 1000 my $min = @$ary ? $ary->[-1]->[0] : undef; while (defined $min) { for my $row (@$ary) { $cb->($row->[1], $row->[2], @args) } $ary = $dbh->selectall_arrayref(<<"", undef, $fid, @rng, $min); SELECT _rowid_,oidbin,uid FROM blob2num WHERE fid = ? $and_ge_le AND _rowid_ > ? ORDER BY _rowid_ ASC LIMIT 1000 $min = @$ary ? $ary->[-1]->[0] : undef; } $ary = $dbh->selectall_arrayref(<<'', undef, $fid); SELECT _rowid_,oidbin,name FROM blob2name WHERE fid = ? ORDER BY _rowid_ ASC LIMIT 1000 $min = @$ary ? $ary->[-1]->[0] : undef; while (defined $min) { for my $row (@$ary) { $cb->($row->[1], \($row->[2]), @args) } $ary = $dbh->selectall_arrayref(<<'', undef, $fid, $min); SELECT _rowid_,oidbin,name FROM blob2name WHERE fid = ? AND _rowid_ > ? ORDER BY _rowid_ ASC LIMIT 1000 $min = @$ary ? $ary->[-1]->[0] : undef; } } sub location_stats { my ($self, $folder) = @_; my $dbh = $self->{dbh} //= dbh_new($self); my $fid; my $ret = {}; $fid = $self->{fmap}->{$folder} //= fid_for($self, $folder) // return; my ($row) = $dbh->selectrow_array(<<"", undef, $fid); SELECT COUNT(name) FROM blob2name WHERE fid = ? $ret->{'name.count'} = $row if $row; my $ntype = ($folder =~ m!\A(?:nntps?|s?news)://!i) ? 'article' : (($folder =~ m!\Aimaps?://!i) ? 'uid' : "TODO<$folder>"); for my $op (qw(count min max)) { ($row) = $dbh->selectrow_array(<<"", undef, $fid); SELECT $op(uid) FROM blob2num WHERE fid = ? $row or last; $ret->{"$ntype.$op"} = $row; } $ret; } # must be called with lock sub _forget_fids ($;@) { my $dbh = shift; $dbh->begin_work; for my $t (qw(blob2name blob2num folders)) { my $sth = $dbh->prepare_cached("DELETE FROM $t WHERE fid = ?"); $sth->execute($_) for @_; } $dbh->commit; } # returns a { location => [ list-of-ids-or-names ] } mapping sub locations_for { my ($self, $oidbin) = @_; my ($fid, $sth, $id, %fid2id, %seen); my $dbh = $self->{dbh} //= dbh_new($self); $sth = $dbh->prepare('SELECT fid,uid FROM blob2num WHERE oidbin = ?'); $sth->bind_param(1, $oidbin, SQL_BLOB); $sth->execute; while (my ($fid, $uid) = $sth->fetchrow_array) { push @{$fid2id{$fid}}, $uid; $seen{"$uid.$fid"} = 1; } # deal with 1.7.0 DBs :< $sth->bind_param(1, $oidbin, SQL_VARCHAR); $sth->execute; while (my ($fid, $uid) = $sth->fetchrow_array) { next if $seen{"$uid.$fid"}; push @{$fid2id{$fid}}, $uid; } %seen = (); $sth = $dbh->prepare('SELECT fid,name FROM blob2name WHERE oidbin = ?'); $sth->bind_param(1, $oidbin, SQL_BLOB); $sth->execute; while (my ($fid, $name) = $sth->fetchrow_array) { push @{$fid2id{$fid}}, $name; $seen{"$fid.$name"} = 1; } # deal with 1.7.0 DBs :< $sth->bind_param(1, $oidbin, SQL_VARCHAR); $sth->execute; while (my ($fid, $name) = $sth->fetchrow_array) { next if $seen{"$fid.$name"}; push @{$fid2id{$fid}}, $name; } $sth = $dbh->prepare('SELECT loc FROM folders WHERE fid = ? LIMIT 1'); my $ret = {}; my $drop_fids = $dbh->{ReadOnly} ? undef : {}; while (my ($fid, $ids) = each %fid2id) { $sth->execute($fid); my ($loc) = $sth->fetchrow_array; unless (defined $loc) { my $del = ''; if ($drop_fids) { $del = ' (deleting)'; $drop_fids->{$fid} = $fid; } my $oidhex = unpack('H*', $oidbin); warn "E: fid=$fid for $oidhex stale/unknown:\n", map { 'E: '.(ref() ? $$_ : "#$_")."$del\n"; } @$ids; next; } $ret->{$loc} = $ids; } if ($drop_fids && scalar(values %$drop_fids)) { my $lk = $self->lock_for_scope; _forget_fids($self->{dbh}, values %$drop_fids); } scalar(keys %$ret) ? $ret : undef; } # returns a list of folders used for completion sub folders { my ($self, @pfx) = @_; my $sql = 'SELECT loc FROM folders'; my $re; if (defined($pfx[0])) { $sql .= ' WHERE loc REGEXP ?'; # DBD::SQLite uses perlre if (ref($pfx[0])) { # assume qr// "Regexp" $re = $pfx[0]; } else { $re = !!$pfx[1] ? '.*' : ''; $re .= quotemeta($pfx[0]); $re .= '.*'; } } my $sth = ($self->{dbh} //= dbh_new($self))->prepare($sql); $sth->bind_param(1, $re) if defined($re); $sth->execute; map { $_->[0] } @{$sth->fetchall_arrayref}; } sub blob_mismatch ($$$) { my ($f, $oidhex, $rawref) = @_; my $sha = $HEXLEN2SHA{length($oidhex)}; my $got = git_sha($sha, $rawref)->hexdigest; $got eq $oidhex ? undef : warn("$f changed $oidhex => $got\n"); } sub local_blob { my ($self, $oidhex, $vrfy) = @_; my $dbh = $self->{dbh} //= dbh_new($self); my $oidbin = pack('H*', $oidhex); my $b2n = $dbh->prepare(<<''); SELECT f.loc,b.name FROM blob2name b LEFT JOIN folders f ON b.fid = f.fid WHERE b.oidbin = ? $b2n->bind_param(1, $oidbin, SQL_BLOB); $b2n->execute; while (my ($d, $n) = $b2n->fetchrow_array) { substr($d, 0, length('maildir:')) = ''; # n.b. both mbsync and offlineimap use ":2," as a suffix # in "new/", despite (from what I understand of reading # ), the ":2," only # applies to files in "cur/". my @try = $n =~ /:2,[a-zA-Z]+\z/ ? qw(cur new) : qw(new cur); for my $x (@try) { my $f = "$d/$x/$n"; open my $fh, '<', $f or next; # some (buggy) Maildir writers are non-atomic: my $raw = read_all($fh, -s $fh // next); next if $vrfy && blob_mismatch $f, $oidhex, \$raw; return \$raw; } } # MH, except `uid' is not always unique (can be packed) $b2n = $dbh->prepare(<<''); SELECT f.loc,b.uid FROM blob2num b LEFT JOIN folders f ON b.fid = f.fid WHERE b.oidbin = ? AND f.loc REGEXP '^mh:/' $b2n->bind_param(1, $oidbin, SQL_BLOB); $b2n->execute; while (my ($f, $n) = $b2n->fetchrow_array) { $f =~ s/\Amh://s or die "BUG: not MH: $f"; $f .= "/$n"; open my $fh, '<', $f or next; my $raw = read_all($fh, -s $fh // next); next if blob_mismatch $f, $oidhex, \$raw; return \$raw; } undef; } sub match_imap_url { my ($self, $url, $all) = @_; # $all = [ $lms->folders ]; $all //= [ $self->folders ]; require PublicInbox::URIimap; my $want = PublicInbox::URIimap->new($url)->canonical; my ($s, $h, $mb) = ($want->scheme, $want->host, $want->mailbox); my @uri = map { PublicInbox::URIimap->new($_)->canonical } grep(m!\A\Q$s\E://.*?\Q$h\E\b.*?/\Q$mb\E\b!, @$all); my @match; for my $x (@uri) { next if $x->mailbox ne $want->mailbox; next if $x->host ne $want->host; next if $x->port != $want->port; my $x_uidval = $x->uidvalidity; next if ($want->uidvalidity // $x_uidval) != $x_uidval; # allow nothing in want to possibly match ";AUTH=ANONYMOUS" if (defined($x->auth) && !defined($want->auth) && !defined($want->user)) { push @match, $x; # or maybe user was forgotten on CLI: } elsif (defined($x->user) && !defined($want->user)) { push @match, $x; } elsif (($x->user//"\0") eq ($want->user//"\0")) { push @match, $x; } } return @match if wantarray; scalar(@match) <= 1 ? $match[0] : "E: `$url' is ambiguous:\n\t".join("\n\t", @match)."\n"; } sub match_nntp_url ($$$) { my ($self, $url, $all) = @_; # $all = [ $lms->folders ]; $all //= [ $self->folders ]; require PublicInbox::URInntps; my $want = PublicInbox::URInntps->new($url)->canonical; my ($s, $h, $p) = ($want->scheme, $want->host, $want->port); my $ng = $want->group; # force scalar (no article ranges) my @uri = map { PublicInbox::URInntps->new($_)->canonical } grep(m!\A\Q$s\E://.*?\Q$h\E\b.*?/\Q$ng\E\b!, @$all); my @match; for my $x (@uri) { next if $x->group ne $ng || $x->host ne $h || $x->port != $p; # maybe user was forgotten on CLI: if (defined($x->userinfo) && !defined($want->userinfo)) { push @match, $x; } elsif (($x->userinfo//"\0") eq ($want->userinfo//"\0")) { push @match, $x; } } return @match if wantarray; scalar(@match) <= 1 ? $match[0] : "E: `$url' is ambiguous:\n\t".join("\n\t", @match)."\n"; } # returns undef on failure, number on success sub group2folders { my ($self, $lei, $all, $folders) = @_; return $lei->fail(< $_ } split(/,/, $all); my @ok = grep(defined, delete(@x{qw(local remote), ''})); push(@ok, '') if $all eq ''; my @no = keys %x; if (@no) { @no = (join(',', @no)); return $lei->fail(<folders; for my $ok (@ok) { if ($ok eq 'local') { push @$folders, grep(!m!\A[a-z0-9\+]+://!i, @all); } elsif ($ok eq 'remote') { push @$folders, grep(m!\A[a-z0-9\+]+://!i, @all); } elsif ($ok ne '') { return $lei->fail("--all=$all not understood"); } else { push @$folders, @all; } } @$folders = uniqstr @$folders; scalar(@$folders) || $lei->fail(<folders; my %all = map { $_ => 1 } @all; my @no; for (@$folders) { next if $all{$_}; # ok if (m!\A(maildir|mh):(.+)!i) { my $type = lc $1; my $d = "$type:".$lei->abs_path($2); push(@no, $_) unless $all{$d}; $_ = $d; } elsif (-d "$_/new" && -d "$_/cur") { my $d = 'maildir:'.$lei->abs_path($_); push(@no, $_) unless $all{$d}; $_ = $d; } elsif (m!\Aimaps?://!i) { my $orig = $_; my $res = match_imap_url($self, $orig, \@all); if (ref $res) { $_ = $$res; $lei->qerr(<qerr(<lock_for_scope; _forget_fids($self->{dbh}, map { delete($self->{fmap}->{$_}) // fid_for($self, $_) // (); } @folders); } # only used for changing canonicalization errors sub rename_folder { my ($self, $old, $new) = @_; my $lk = $self->lock_for_scope; my $ofid = delete($self->{fmap}->{$old}) // fid_for($self, $old) // return; eval { $self->{dbh}->do(<{fmap}->{$new} // fid_for($self, $new); for my $t (qw(blob2name blob2num)) { $self->{dbh}->do(<{dbh}->do(<{fmap}->{$url} //= fid_for($self, $url) // return (); my $sth = $self->{dbh}->prepare_cached(<execute($fid, $uid); # for public-inbox <= 1.7.0: uniqstr(map { $_->[0] } @{$sth->fetchall_arrayref}); } sub name_oidbin ($$$) { my ($self, $mdir, $nm) = @_; my $fid = $self->{fmap}->{$mdir} //= fid_for($self, $mdir) // return; my $sth = $self->{dbh}->prepare_cached(<bind_param(1, $fid); $sth->bind_param(2, $nm, SQL_BLOB); $sth->execute; my @bin = map { $_->[0] } @{$sth->fetchall_arrayref}; $sth->bind_param(1, $fid); $sth->bind_param(2, $nm, SQL_VARCHAR); $sth->execute; my @old = map { $_->[0] } @{$sth->fetchall_arrayref}; uniqstr @bin, @old # for public-inbox <= 1.7.0 } sub imap_oidhex { my ($self, $lei, $uid_uri) = @_; my $mailbox_uri = $uid_uri->clone; $mailbox_uri->uid(undef); my $folders = [ $$mailbox_uri ]; eval { $self->arg2folder($lei, $folders) }; $lei->qerr("# no sync information for $mailbox_uri") if $@; map { unpack('H*',$_) } num_oidbin($self, $folders->[0], $uid_uri->uid) } 1;