+# Copyright (C) 2018 all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+
+# for XOVER, OVER in NNTP, and feeds/homepage/threads in PSGI
+# Unlike Msgmap, this is an _UNSTABLE_ database which can be
+# tweaked/updated over time and rebuilt.
+package PublicInbox::OverIdx;
+use strict;
+use warnings;
+use base qw(PublicInbox::Over);
+use IO::Handle;
+use DBI qw(:sql_types); # SQL_BLOB
+
+sub dbh_new {
+ my ($self) = @_;
+ my $dbh = $self->SUPER::dbh_new;
+ $dbh->do('PRAGMA synchronous = OFF'); # commit_fsync instead
+ $dbh->do('PRAGMA journal_mode = TRUNCATE');
+ $dbh->do('PRAGMA cache_size = 80000');
+ create_tables($dbh);
+ $dbh;
+}
+
+sub commit_fsync {
+ my $fn = $_[0]->{filename};
+ if (open my $fh, '+<', $fn) {
+ $fh->sync;
+ close $fh;
+ }
+}
+
+sub get_counter ($$) {
+ my ($dbh, $key) = @_;
+ my $sth = $dbh->prepare_cached(<<'', undef, 1);
+SELECT val FROM counter WHERE key = ? LIMIT 1
+
+ $sth->execute($key);
+ $sth->fetchrow_array;
+}
+
+sub adj_counter ($$$) {
+ my ($self, $key, $op) = @_;
+ my $dbh = $self->{dbh};
+ my $sth = $dbh->prepare_cached(<<"");
+UPDATE counter SET val = val $op 1 WHERE key = ?
+
+ $sth->execute($key);
+
+ get_counter($dbh, $key);
+}
+
+sub next_tid { adj_counter($_[0], 'thread', '+') }
+sub next_ghost_num { adj_counter($_[0], 'ghost', '-') }
+
+sub id_for ($$$$$) {
+ my ($self, $tbl, $id_col, $val_col, $val) = @_;
+ my $dbh = $self->{dbh};
+ my $in = $dbh->prepare_cached(<<"")->execute($val);
+INSERT OR IGNORE INTO $tbl ($val_col) VALUES (?)
+
+ if ($in == 0) {
+ my $sth = $dbh->prepare_cached(<<"", undef, 1);
+SELECT $id_col FROM $tbl WHERE $val_col = ? LIMIT 1
+
+ $sth->execute($val);
+ $sth->fetchrow_array;
+ } else {
+ $dbh->last_insert_id(undef, undef, $tbl, $id_col);
+ }
+}
+
+sub sid {
+ my ($self, $path) = @_;
+ return unless defined $path && $path ne '';
+ id_for($self, 'subject', 'sid', 'path' => $path);
+}
+
+sub mid2id {
+ my ($self, $mid) = @_;
+ id_for($self, 'msgid', 'id', 'mid' => $mid);
+}
+
+sub delete_by_num {
+ my ($self, $num) = @_;
+ my $dbh = $self->{dbh};
+ foreach (qw(over id2num)) {
+ $dbh->prepare_cached(<<"")->execute($num);
+DELETE FROM $_ WHERE num = ?
+
+ }
+}
+
+# this includes ghosts
+sub each_by_mid {
+ my ($self, $mid, $cols, $cb) = @_;
+ my $dbh = $self->{dbh};
+
+=over
+ I originally wanted to stuff everything into a single query:
+
+ SELECT over.* FROM over
+ LEFT JOIN id2num ON over.num = id2num.num
+ LEFT JOIN msgid ON msgid.id = id2num.id
+ WHERE msgid.mid = ? AND over.num >= ?
+ ORDER BY over.num ASC
+ LIMIT 1000
+
+ But it's faster broken out (and we're always in a
+ transaction for subroutines in this file)
+=cut
+
+ my $sth = $dbh->prepare_cached(<<'', undef, 1);
+SELECT id FROM msgid WHERE mid = ? LIMIT 1
+
+ $sth->execute($mid);
+ my $id = $sth->fetchrow_array;
+ defined $id or return;
+
+ push(@$cols, 'num');
+ $cols = join(',', map { $_ } @$cols);
+ my $lim = 10;
+ my $prev = get_counter($dbh, 'ghost');
+ while (1) {
+ $sth = $dbh->prepare_cached(<<"", undef, 1);
+SELECT num FROM id2num WHERE id = ? AND num >= ?
+ORDER BY num ASC
+LIMIT $lim
+
+ $sth->execute($id, $prev);
+ my $nums = $sth->fetchall_arrayref;
+ my $nr = scalar(@$nums) or return;
+ $prev = $nums->[-1]->[0];
+
+ $sth = $dbh->prepare_cached(<<"", undef, 1);
+SELECT $cols FROM over WHERE over.num = ? LIMIT 1
+
+ foreach (@$nums) {
+ $sth->execute($_->[0]);
+ my $smsg = $sth->fetchrow_hashref;
+ $cb->(PublicInbox::Over::load_from_row($smsg)) or
+ return;
+ }
+ return if $nr != $lim;
+ }
+}
+
+# this will create a ghost as necessary
+sub resolve_mid_to_tid {
+ my ($self, $mid) = @_;
+ my $tid;
+ each_by_mid($self, $mid, ['tid'], sub {
+ my ($smsg) = @_;
+ my $cur_tid = $smsg->{tid};
+ if (defined $tid) {
+ merge_threads($self, $tid, $cur_tid);
+ } else {
+ $tid = $cur_tid;
+ }
+ 1;
+ });
+ defined $tid ? $tid : create_ghost($self, $mid);
+}
+
+sub create_ghost {
+ my ($self, $mid) = @_;
+ my $id = $self->mid2id($mid);
+ my $num = $self->next_ghost_num;
+ $num < 0 or die "ghost num is non-negative: $num\n";
+ my $tid = $self->next_tid;
+ my $dbh = $self->{dbh};
+ $dbh->prepare_cached(<<'')->execute($num, $tid);
+INSERT INTO over (num, tid) VALUES (?,?)
+
+ $dbh->prepare_cached(<<'')->execute($id, $num);
+INSERT INTO id2num (id, num) VALUES (?,?)
+
+ $tid;
+}
+
+sub merge_threads {
+ my ($self, $winner_tid, $loser_tid) = @_;
+ return if $winner_tid == $loser_tid;
+ my $dbh = $self->{dbh};
+ $dbh->prepare_cached(<<'')->execute($winner_tid, $loser_tid);
+UPDATE over SET tid = ? WHERE tid = ?
+
+}
+
+sub link_refs {
+ my ($self, $refs, $old_tid) = @_;
+ my $tid;
+
+ if (@$refs) {
+ # first ref *should* be the thread root,
+ # but we can never trust clients to do the right thing
+ my $ref = $refs->[0];
+ $tid = resolve_mid_to_tid($self, $ref);
+ merge_threads($self, $tid, $old_tid) if defined $old_tid;
+
+ # the rest of the refs should point to this tid:
+ foreach my $i (1..$#$refs) {
+ $ref = $refs->[$i];
+ my $ptid = resolve_mid_to_tid($self, $ref);
+ merge_threads($self, $tid, $ptid);
+ }
+ } else {
+ $tid = defined $old_tid ? $old_tid : $self->next_tid;
+ }
+ $tid;
+}
+
+sub add_over {
+ my ($self, $values) = @_;
+ my ($ts, $num, $mids, $refs, $xpath, $ddd) = @$values;
+ my $old_tid;
+ my $vivified = 0;
+
+ $self->begin_lazy;
+ $self->delete_by_num($num);
+ foreach my $mid (@$mids) {
+ my $v = 0;
+ each_by_mid($self, $mid, ['tid'], sub {
+ my ($cur) = @_;
+ my $cur_tid = $cur->{tid};
+ my $n = $cur->{num};
+ die "num must not be zero for $mid" if !$n;
+ $old_tid = $cur_tid unless defined $old_tid;
+ if ($n > 0) { # regular mail
+ merge_threads($self, $old_tid, $cur_tid);
+ } elsif ($n < 0) { # ghost
+ link_refs($self, $refs, $old_tid);
+ $self->delete_by_num($n);
+ $v++;
+ }
+ 1;
+ });
+ $v > 1 and warn "BUG: vivified multiple ($v) ghosts for $mid\n";
+ $vivified += $v;
+ }
+ my $tid = $vivified ? $old_tid : link_refs($self, $refs, $old_tid);
+ my $sid = $self->sid($xpath);
+ my $dbh = $self->{dbh};
+ my $sth = $dbh->prepare_cached(<<'');
+INSERT INTO over (num, tid, sid, ts, ddd)
+VALUES (?,?,?,?,?)
+
+ my $n = 0;
+ my @v = ($num, $tid, $sid, $ts);
+ foreach (@v) { $sth->bind_param(++$n, $_) }
+ $sth->bind_param(++$n, $ddd, SQL_BLOB);
+ $sth->execute;
+ $sth = $dbh->prepare_cached(<<'');
+INSERT INTO id2num (id, num) VALUES (?,?)
+
+ foreach my $mid (@$mids) {
+ my $id = $self->mid2id($mid);
+ $sth->execute($id, $num);
+ }
+}
+
+sub delete_articles {
+ my ($self, $nums) = @_;
+ my $dbh = $self->connect;
+ $self->delete_by_num($_) foreach @$nums;
+}
+
+sub remove_oid {
+ my ($self, $oid, $mid) = @_;
+ $self->begin_lazy;
+ each_by_mid($self, $mid, ['ddd'], sub {
+ my ($smsg) = @_;
+ $self->delete_by_num($smsg->{num}) if $smsg->{blob} eq $oid;
+ 1;
+ });
+}
+
+sub create_tables {
+ my ($dbh) = @_;
+
+ $dbh->do(<<'');
+CREATE TABLE IF NOT EXISTS over (
+ num INTEGER NOT NULL,
+ tid INTEGER NOT NULL,
+ sid INTEGER,
+ ts INTEGER,
+ ddd VARBINARY, /* doc-data-deflated */
+ UNIQUE (num)
+)
+
+ $dbh->do('CREATE INDEX IF NOT EXISTS idx_tid ON over (tid)');
+ $dbh->do('CREATE INDEX IF NOT EXISTS idx_sid ON over (sid)');
+ $dbh->do('CREATE INDEX IF NOT EXISTS idx_ts ON over (ts)');
+
+ $dbh->do(<<'');
+CREATE TABLE IF NOT EXISTS counter (
+ key VARCHAR(8) PRIMARY KEY NOT NULL,
+ val INTEGER DEFAULT 0,
+ UNIQUE (key)
+)
+
+ $dbh->do("INSERT OR IGNORE INTO counter (key) VALUES ('thread')");
+ $dbh->do("INSERT OR IGNORE INTO counter (key) VALUES ('ghost')");
+
+ $dbh->do(<<'');
+CREATE TABLE IF NOT EXISTS subject (
+ sid INTEGER PRIMARY KEY AUTOINCREMENT,
+ path VARCHAR(40) NOT NULL,
+ UNIQUE (path)
+)
+
+ $dbh->do(<<'');
+CREATE TABLE IF NOT EXISTS id2num (
+ id INTEGER NOT NULL,
+ num INTEGER NOT NULL,
+ UNIQUE (id, num)
+)
+
+ # performance critical:
+ $dbh->do('CREATE INDEX IF NOT EXISTS idx_inum ON id2num (num)');
+ $dbh->do('CREATE INDEX IF NOT EXISTS idx_id ON id2num (id)');
+
+ $dbh->do(<<'');
+CREATE TABLE IF NOT EXISTS msgid (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ mid VARCHAR(244) NOT NULL,
+ UNIQUE (mid)
+)
+
+}
+
+sub commit_lazy {
+ my ($self) = @_;
+ delete $self->{txn} or return;
+ $self->{dbh}->commit;
+}
+
+sub begin_lazy {
+ my ($self) = @_;
+ return if $self->{txn};
+ my $dbh = $self->connect or return;
+ $dbh->begin_work;
+ # $dbh->{Profile} = 2;
+ $self->{txn} = 1;
+}
+
+sub rollback_lazy {
+ my ($self) = @_;
+ delete $self->{txn} or return;
+ $self->{dbh}->rollback;
+}
+
+sub disconnect {
+ my ($self) = @_;
+ die "in transaction" if $self->{txn};
+ $self->{dbh} = undef;
+}
+
+sub create {
+ my ($self) = @_;
+ unless (-r $self->{filename}) {
+ require File::Path;
+ require File::Basename;
+ File::Path::mkpath(File::Basename::dirname($self->{filename}));
+ }
+ # create the DB:
+ PublicInbox::Over::connect($self);
+ $self->disconnect;
+}
+
+1;