]> Sergey Matveev's repositories - public-inbox.git/blob - lib/PublicInbox/LeiOverview.pm
ipc: wq_do => wq_io_do
[public-inbox.git] / lib / PublicInbox / LeiOverview.pm
1 # Copyright (C) 2021 all contributors <meta@public-inbox.org>
2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
3
4 # per-mitem/smsg iterators for search results
5 # "ovv" => "Overview viewer"
6 package PublicInbox::LeiOverview;
7 use strict;
8 use v5.10.1;
9 use parent qw(PublicInbox::Lock);
10 use POSIX qw(strftime);
11 use Fcntl qw(F_GETFL O_APPEND);
12 use File::Spec;
13 use File::Temp ();
14 use PublicInbox::MID qw($MID_EXTRACT);
15 use PublicInbox::Address qw(pairs);
16 use PublicInbox::Config;
17 use PublicInbox::Search qw(get_pct);
18 use PublicInbox::LeiDedupe;
19 use PublicInbox::LeiToMail;
20
21 # cf. https://en.wikipedia.org/wiki/JSON_streaming
22 my $JSONL = 'ldjson|ndjson|jsonl'; # 3 names for the same thing
23
24 sub _iso8601 ($) { strftime('%Y-%m-%dT%H:%M:%SZ', gmtime($_[0])) }
25
26 # we open this in the parent process before ->wq_io_do handoff
27 sub ovv_out_lk_init ($) {
28         my ($self) = @_;
29         my $tmp = File::Temp->new("lei-ovv.dst.$$.lock-XXXXXX",
30                                         TMPDIR => 1, UNLINK => 0);
31         $self->{"lk_id.$self.$$"} = $self->{lock_path} = $tmp->filename;
32 }
33
34 sub ovv_out_lk_cancel ($) {
35         my ($self) = @_;
36         my $lock_path = delete $self->{"lk_id.$self.$$"} or return;
37         unlink($lock_path);
38 }
39
40 sub detect_fmt ($$) {
41         my ($lei, $dst) = @_;
42         if ($dst =~ m!\A([:/]+://)!) {
43                 $lei->fail("$1 support not implemented, yet\n");
44         } elsif (!-e $dst || -d _) {
45                 'maildir'; # the default TODO: MH?
46         } elsif (-f _ || -p _) {
47                 $lei->fail("unable to determine mbox family of $dst\n");
48         } else {
49                 $lei->fail("unable to determine format of $dst\n");
50         }
51 }
52
53 sub new {
54         my ($class, $lei) = @_;
55         my $opt = $lei->{opt};
56         my $dst = $opt->{output} // '-';
57         $dst = '/dev/stdout' if $dst eq '-';
58
59         my $fmt = $opt->{'format'};
60         $fmt = lc($fmt) if defined $fmt;
61         if ($dst =~ s/\A([a-z0-9]+)://is) { # e.g. Maildir:/home/user/Mail/
62                 my $ofmt = lc $1;
63                 $fmt //= $ofmt;
64                 return $lei->fail(<<"") if $fmt ne $ofmt;
65 --format=$fmt and --output=$ofmt conflict
66
67         }
68         $fmt //= 'json' if $dst eq '/dev/stdout';
69         $fmt //= detect_fmt($lei, $dst) or return;
70
71         if (index($dst, '://') < 0) { # not a URL, so assume path
72                  $dst = File::Spec->canonpath($dst);
73         } # else URL
74
75         my $self = bless { fmt => $fmt, dst => $dst }, $class;
76         $lei->{ovv} = $self;
77         my $json;
78         if ($fmt =~ /\A($JSONL|(?:concat)?json)\z/) {
79                 $json = $self->{json} = ref(PublicInbox::Config->json);
80         }
81         if ($dst eq '/dev/stdout') {
82                 my $isatty = $lei->{need_pager} = -t $lei->{1};
83                 $opt->{pretty} //= $isatty;
84                 if (!$isatty && -f _) {
85                         my $fl = fcntl($lei->{1}, F_GETFL, 0) //
86                                 return $lei->fail("fcntl(stdout): $!");
87                         ovv_out_lk_init($self) unless ($fl & O_APPEND);
88                 } else {
89                         ovv_out_lk_init($self);
90                 }
91         } elsif (!$opt->{quiet}) {
92                 $lei->{-progress} = 1;
93         }
94         if ($json) {
95                 $lei->{dedupe} //= PublicInbox::LeiDedupe->new($lei);
96         } else {
97                 # default to the cheapest sort since MUA usually resorts
98                 $lei->{opt}->{'sort'} //= 'docid' if $dst ne '/dev/stdout';
99                 $lei->{l2m} = eval { PublicInbox::LeiToMail->new($lei) };
100                 return $lei->fail($@) if $@;
101         }
102         $self;
103 }
104
105 # called once by parent
106 sub ovv_begin {
107         my ($self, $lei) = @_;
108         if ($self->{fmt} eq 'json') {
109                 $lei->out('[');
110         } # TODO HTML/Atom/...
111 }
112
113 # called once by parent (via PublicInbox::EOFpipe)
114 sub ovv_end {
115         my ($self, $lei) = @_;
116         if ($self->{fmt} eq 'json') {
117                 # JSON doesn't allow trailing commas, and preventing
118                 # trailing commas is a PITA when parallelizing outputs
119                 $lei->out("null]\n");
120         } elsif ($self->{fmt} eq 'concatjson') {
121                 $lei->out("\n");
122         }
123 }
124
125 # prepares an smsg for JSON
126 sub _unbless_smsg {
127         my ($smsg, $mitem) = @_;
128
129         delete @$smsg{qw(lines bytes num tid)};
130         $smsg->{rt} = _iso8601(delete $smsg->{ts}); # JMAP receivedAt
131         $smsg->{dt} = _iso8601(delete $smsg->{ds}); # JMAP UTCDate
132         $smsg->{pct} = get_pct($mitem) if $mitem;
133         if (my $r = delete $smsg->{references}) {
134                 $smsg->{refs} = [ map { "<$_>" } ($r =~ m/$MID_EXTRACT/go) ];
135         }
136         if (my $m = delete($smsg->{mid})) {
137                 $smsg->{'m'} = "<$m>";
138         }
139         for my $f (qw(from to cc)) {
140                 my $v = delete $smsg->{$f} or next;
141                 $smsg->{substr($f, 0, 1)} = pairs($v);
142         }
143         $smsg->{'s'} = delete $smsg->{subject};
144         # can we be bothered to parse From/To/Cc into arrays?
145         scalar { %$smsg }; # unbless
146 }
147
148 sub ovv_atexit_child {
149         my ($self, $lei) = @_;
150         if (my $bref = delete $lei->{ovv_buf}) {
151                 my $lk = $self->lock_for_scope;
152                 $lei->out($$bref);
153         }
154 }
155
156 # JSON module ->pretty output wastes too much vertical white space,
157 # this (IMHO) provides better use of screen real-estate while not
158 # being excessively compact:
159 sub _json_pretty {
160         my ($json, $k, $v) = @_;
161         if (ref $v eq 'ARRAY') {
162                 if (@$v) {
163                         my $sep = ",\n" . (' ' x (length($k) + 7));
164                         if (ref($v->[0])) { # f/t/c
165                                 $v = '[' . join($sep, map {
166                                         my $pair = $json->encode($_);
167                                         $pair =~ s/(null|"),"/$1, "/g;
168                                         $pair;
169                                 } @$v) . ']';
170                         } else { # references
171                                 $v = '[' . join($sep, map {
172                                         substr($json->encode([$_]), 1, -1);
173                                 } @$v) . ']';
174                         }
175                 } else {
176                         $v = '[]';
177                 }
178         }
179         qq{  "$k": }.$v;
180 }
181
182 sub ovv_each_smsg_cb { # runs in wq worker usually
183         my ($self, $lei, $ibxish) = @_;
184         my ($json, $dedupe);
185         if (my $pkg = $self->{json}) {
186                 $json = $pkg->new;
187                 $json->utf8->canonical;
188                 $json->ascii(1) if $lei->{opt}->{ascii};
189         }
190         my $l2m = $lei->{l2m};
191         if (!$l2m) {
192                 $dedupe = $lei->{dedupe} // die 'BUG: {dedupe} missing';
193                 $dedupe->prepare_dedupe;
194         }
195         $lei->{ovv_buf} = \(my $buf = '') if !$l2m;
196         if ($l2m && !$ibxish) { # remote https?:// mboxrd
197                 my $g2m = $l2m->can('git_to_mail');
198                 my $wcb = $l2m->write_cb($lei);
199                 sub {
200                         my ($smsg, undef, $eml) = @_; # no mitem in $_[1]
201                         $wcb->(undef, $smsg, $eml);
202                 };
203         } elsif ($l2m && $l2m->{-wq_s1}) {
204                 my $git_dir = $ibxish->git->{git_dir};
205                 sub {
206                         my ($smsg, $mitem) = @_;
207                         $smsg->{pct} = get_pct($mitem) if $mitem;
208                         $l2m->wq_io_do('write_mail', [], $git_dir, $smsg);
209                 }
210         } elsif ($self->{fmt} =~ /\A(concat)?json\z/ && $lei->{opt}->{pretty}) {
211                 my $EOR = ($1//'') eq 'concat' ? "\n}" : "\n},";
212                 sub { # DIY prettiness :P
213                         my ($smsg, $mitem) = @_;
214                         return if $dedupe->is_smsg_dup($smsg);
215                         $smsg = _unbless_smsg($smsg, $mitem);
216                         $buf .= "{\n";
217                         $buf .= join(",\n", map {
218                                 my $v = $smsg->{$_};
219                                 if (ref($v)) {
220                                         _json_pretty($json, $_, $v);
221                                 } else {
222                                         $v = $json->encode([$v]);
223                                         qq{  "$_": }.substr($v, 1, -1);
224                                 }
225                         } sort keys %$smsg);
226                         $buf .= $EOR;
227                         return if length($buf) < 65536;
228                         my $lk = $self->lock_for_scope;
229                         $lei->out($buf);
230                         $buf = '';
231                 }
232         } elsif ($json) {
233                 my $ORS = $self->{fmt} eq 'json' ? ",\n" : "\n"; # JSONL
234                 sub {
235                         my ($smsg, $mitem) = @_;
236                         return if $dedupe->is_smsg_dup($smsg);
237                         $buf .= $json->encode(_unbless_smsg(@_)) . $ORS;
238                         return if length($buf) < 65536;
239                         my $lk = $self->lock_for_scope;
240                         $lei->out($buf);
241                         $buf = '';
242                 }
243         } else {
244                 die "TODO: unhandled case $self->{fmt}"
245         }
246 }
247
248 no warnings 'once';
249 *DESTROY = \&ovv_out_lk_cancel;
250
251 1;