pigeon-cli 0.4.28

Pigeon: authenticate, sink, and transform personal data from external services.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
use std::path::PathBuf;

use clap::{Args, Subcommand};

use crate::core::observability::Observable;

#[derive(Args, Debug)]
pub struct JobArgs {
    #[command(subcommand)]
    pub command: JobCommands,
}

#[derive(Subcommand, Debug)]
pub enum JobCommands {
    /// Run a job (see subcommands for available job types)
    Run(RunArgs),
}

#[derive(Args, Debug)]
pub struct RunArgs {
    #[command(subcommand)]
    pub job_type: JobType,
}

#[derive(Subcommand, Debug)]
pub enum JobType {
    /// Email jobs addressed through the keyring. See subcommands for the
    /// available actions.
    Email(EmailArgs),

    /// Recursively scans a bucket, always inflates every zip found (the
    /// zip container itself is never uploaded, only its inflated
    /// contents), content-hashes every file bucket-wide to keep one
    /// byte-identical copy of each, writes a human-readable merge report,
    /// and optionally uploads the result unencrypted to a (possibly
    /// different) bucket-config (ADR-0082). Every file is always processed
    /// and every zip is always expanded -- there is no file-type or
    /// zip-expansion selection, and this job never offers encryption.
    Deduplicate {
        /// Alias of a configured bucket-config to pull from. Repeat the
        /// flag for more than one (`--source-bucket a --source-bucket b`)
        /// -- every listed bucket is downloaded into one shared local
        /// staging tree and deduplicated across the combined set, not
        /// per-bucket (ADR-0109). Interactively multi-selected from the
        /// configured bucket-configs when omitted and stdin is a terminal;
        /// at least one is required otherwise.
        #[arg(long)]
        source_bucket: Vec<String>,

        /// Local directory to stage and store output under. Defaults to a
        /// directory under the OS temp directory when omitted.
        #[arg(long)]
        local_output: Option<PathBuf>,

        /// Alias of a configured bucket-config to upload the deduped
        /// result to, once local processing is complete. Always uploaded
        /// unencrypted. Single bucket only (ADR-0109).
        #[arg(long)]
        destination_bucket: Option<String>,

        /// Maximum number of files to download/hash concurrently.
        #[arg(long)]
        concurrency: Option<usize>,

        /// rclone --transfers for the upload phase, now one bulk `rclone
        /// copy` rather than a per-file task queue (ADR-0136). Defaults to
        /// 8 when omitted. With `--upload-only`, this is the only transfer
        /// concurrency flag that has any effect.
        #[arg(long)]
        transfers: Option<usize>,

        /// rclone --checkers for the upload phase. Defaults to 16 when
        /// omitted.
        #[arg(long)]
        checkers: Option<usize>,

        /// rclone --tpslimit for the upload phase. Unset by default -- no
        /// rate cap.
        #[arg(long)]
        tpslimit: Option<usize>,

        /// Resumes uploading an already-completed local deduplicate run instead
        /// of starting a new one: skips the bucket listing/download/hash/
        /// placement phases entirely (and the source bucket credentials
        /// they'd otherwise need) and uploads straight from an existing
        /// `--local-output`'s `result/` tree, picking up where a prior
        /// run's upload phase left off via the same `.staging/.uploaded`
        /// index (ADR-0089). Requires a `--local-output` from a completed
        /// prior run (its `.staging/.processed` checkpoint must exist and
        /// its `result/` must be non-empty) and a mandatory
        /// `--remote-output`.
        #[arg(long)]
        upload_only: bool,

        /// Alias of a configured bucket-config this run's report, the
        /// shared observability log, and a transcript of its printed
        /// output are uploaded to, always unencrypted, under a
        /// `YYYY-MM-DD-job-name-{run-id}/` prefix (ADR-0100). Mandatory --
        /// interactively selected when omitted and stdin is a terminal;
        /// required otherwise.
        #[arg(long)]
        report_bucket: Option<String>,

        /// Suppress every wizard prompt (ADR-0114): each omitted input
        /// resolves via its non-interactive fallback (or errors if none
        /// exists) instead of asking, and the final "proceed?"
        /// confirmation is auto-answered yes.
        #[arg(long)]
        non_interactive: bool,
    },

    /// Shells out to the external `rclone` binary for copy/delete
    /// operations against pigeon keyring `BucketConfig` aliases. `rclone`
    /// is always invoked against a config file pigeon itself generates
    /// from the keyring (ADR-0136, reversing ADR-0101/ADR-0110's original
    /// raw-passthrough design) -- never an externally-provisioned or
    /// user-general-purpose `rclone.conf`. See subcommands for the
    /// available actions. Requires `rclone` on `PATH`.
    Rclone(RcloneArgs),

    /// Transcodes image files (`png`/`jpeg`/`heic`) into full-size,
    /// maximum-quality `.jpg`, video files (`mov`/`m4v`/`mp4`, ADR-0122)
    /// into space-optimized H.265/HEVC `.mp4` at a chosen `--video-quality`,
    /// or audio-only recordings carrying an `.mp4` extension (`mp4-audio`,
    /// ADR-0142) into Opus `.opus`
    /// (ADR-0112, pipeline reworked by ADR-0116): shells out to `rclone` to
    /// pull `--source` into a local staging tree, transcoding/copying each
    /// file locally and pushing it individually via `rclone copyto` the
    /// moment it's ready, rather than waiting for the whole batch.
    /// `--source`/`--destination` are pigeon keyring `BucketConfig`
    /// aliases, optionally composed with a prefix within that bucket
    /// (`alias:prefix/`, ADR-0138); an omitted prefix defaults to
    /// `--input-file-type`'s own extension (source) or output extension
    /// (destination). Deduplication is never involved -- every output file's
    /// name is derived deterministically and is unique by construction;
    /// duplicate source content is deliberately retained at separate
    /// destination names. A single file's transcode or push failure (e.g.
    /// a HEIC file the local `ffmpeg` build can't decode, retried before
    /// giving up) is recorded and reported but never stops any other file
    /// from being processed and pushed; every file that does succeed stays
    /// checkpointed for a resumed rerun. Requires both `rclone` and
    /// `ffmpeg`/`ffprobe` on `PATH`.
    Transform {
        /// Which input format this run transcodes: `png`, `jpeg`, `heic`,
        /// `mov`, `m4v`, `mp4`, or `mp4-audio`. Interactively selected when
        /// omitted and stdin is a terminal; required otherwise.
        /// `mp4-audio` is for audio-only recordings carrying an `.mp4`
        /// extension (voice notes), transcoded to `.opus`; a file that
        /// actually has a video stream is refused rather than silently
        /// losing its video.
        #[arg(long)]
        input_file_type: Option<String>,

        /// Compression level for a true video input kind (`mov`/`m4v`/`mp4`
        /// -- not `mp4-audio`, which has no video stream):
        /// `low`, `medium` (default), `high`, or `lossless` -- a small
        /// vetted `libx265` CRF menu (ADR-0122). Ignored for every other
        /// input kind. Interactively selected when omitted, stdin is a
        /// terminal, and the input kind is video; defaults to `medium`
        /// otherwise.
        #[arg(long)]
        video_quality: Option<String>,

        /// Where to pull from: a configured bucket-config alias, optionally
        /// composed with a prefix inside it -- `media` is the whole bucket,
        /// `media:mp4/` its `mp4/` directory (ADR-0138). An omitted prefix
        /// defaults to `--input-file-type`'s own extension, so
        /// `--source media --input-file-type mp4` means `media:mp4/`; an
        /// explicit prefix is always used verbatim. Interactively selected
        /// (bucket root, then the default prefix) when omitted and stdin is
        /// a terminal; required otherwise.
        #[arg(long)]
        source: Option<String>,

        /// Where to push to, same `alias[:prefix/]` form as `--source`
        /// (ADR-0138). An omitted prefix defaults to `--input-file-type`'s
        /// *output* extension (`jpg` for an image kind, `mp4` for a video
        /// one, `opus` for `mp4-audio`), so one bucket can serve as both
        /// source and destination when the prefixes differ.
        #[arg(long)]
        destination: Option<String>,

        /// Local directory this run stages pulled source files, transcoded
        /// results, and bookkeeping under. Defaults to a directory under
        /// the OS temp directory when omitted.
        #[arg(long)]
        local_output: Option<PathBuf>,

        /// Maximum number of files to transcode concurrently (the ffmpeg
        /// phase only -- `rclone` manages its own internal parallelism via
        /// `--download-transfers`/`--download-checkers`). Interactively
        /// prompted (a cores-based default) when omitted and stdin is a
        /// terminal; required otherwise.
        #[arg(long)]
        concurrency: Option<usize>,

        /// rclone --transfers for the download (pull) phase. Defaults to 8
        /// when omitted.
        #[arg(long)]
        download_transfers: Option<usize>,

        /// rclone --checkers for the download (pull) phase. Defaults to 16
        /// when omitted.
        #[arg(long)]
        download_checkers: Option<usize>,

        /// Maximum number of concurrent per-file push (`rclone copyto`)
        /// operations. Defaults to 8 when omitted. Separate from
        /// `--download-transfers` to allow independent tuning of pull vs.
        /// push phases (ADR-0129).
        #[arg(long)]
        upload_transfers: Option<usize>,

        /// Maximum number of downloaded but not-yet-processed files
        /// (pending transcode/push) before the pull phase is paused to allow
        /// the pipeline to catch up. Defaults to `concurrency * 4` when
        /// omitted. Enables automatic disk-pressure backpressure without
        /// requiring manual tuning (ADR-0129).
        #[arg(long)]
        max_pending_downloads: Option<usize>,

        /// rclone --tpslimit, used by the pull phase only. Unset by
        /// default -- no rate cap.
        #[arg(long)]
        tpslimit: Option<usize>,

        /// Alias of a configured bucket-config this run's report, the
        /// shared observability log, and a transcript of its printed
        /// output are uploaded to, always unencrypted, under a
        /// `YYYY-MM-DD-job-name-{run-id}/` prefix (ADR-0100). Mandatory --
        /// interactively selected when omitted and stdin is a terminal;
        /// required otherwise.
        #[arg(long)]
        report_bucket: Option<String>,

        /// Suppress every wizard prompt (ADR-0114): each omitted input
        /// resolves via its non-interactive fallback (or errors if none
        /// exists) instead of asking, and the final "proceed?"
        /// confirmation is auto-answered yes.
        #[arg(long)]
        non_interactive: bool,
    },
}

#[derive(Args, Debug)]
pub struct EmailArgs {
    #[command(subcommand)]
    pub action: EmailAction,
}

#[derive(Subcommand, Debug)]
pub enum EmailAction {
    /// Reads one or more authenticated IMAP identities and delivers each
    /// message's raw `.eml` plus its unpacked attachments to
    /// `--destination`, at a declared, rate-limit-respecting pace
    /// (ADR-0146). Deliberately minimal: no deduplication, no indexing, no
    /// renaming, no content transform, and no encryption -- it moves bytes
    /// and nothing else. Each message is pushed individually via `rclone
    /// copyto` the moment it is fetched and its local copy deleted
    /// immediately afterwards, so a run lasting days delivers
    /// incrementally and local disk stays bounded to the in-flight window
    /// regardless of how large the account is. A message that fails is
    /// simply not checkpointed, so a rerun refetches it; nothing is
    /// retained locally for diagnosis, because the server still has it.
    /// Requires `rclone` on `PATH`.
    Import {
        /// Aliases of the identities to import, comma-separated.
        /// Interactively multi-selected from the authenticated identities
        /// when omitted and stdin is a terminal; required otherwise.
        #[arg(long, value_delimiter = ',')]
        identities: Option<Vec<String>>,

        /// Where to deliver to: a configured bucket-config alias,
        /// optionally composed with a prefix inside it -- `archive` is the
        /// whole bucket, `archive:eml/` its `eml/` directory (ADR-0138).
        /// Unlike `transform`, an omitted prefix stays the bucket root
        /// (this job has no file-type concept to derive one from).
        /// Mandatory -- interactively selected when omitted and stdin is a
        /// terminal; required otherwise.
        #[arg(long)]
        destination: Option<String>,

        /// Local directory in-flight messages and this run's bookkeeping
        /// are staged under. Defaults to a directory under the OS temp
        /// directory when omitted. Unlike the other email jobs this is not
        /// an output tree: each message's local copy is deleted as soon as
        /// its push resolves (ADR-0146 §5), so it stays near-empty.
        #[arg(long)]
        local_output: Option<PathBuf>,

        /// Ceiling on how many messages per minute are fetched, across
        /// every selected identity -- the run's throughput governor
        /// (ADR-0146 §4). Deliberately has no "unlimited" value: the
        /// absence of a cap is what this job exists to fix. Interactively
        /// prompted when omitted and stdin is a terminal; defaults to 120
        /// otherwise, which puts a 200,000-message account at roughly 28
        /// hours.
        #[arg(long)]
        messages_per_minute: Option<usize>,

        /// How many UIDs one `UID FETCH` asks for at a time. Small enough
        /// that a failed batch loses little work, large enough to amortise
        /// the round trip. Interactively prompted when omitted and stdin is
        /// a terminal; defaults to 20 otherwise.
        #[arg(long)]
        batch_size: Option<usize>,

        /// Maximum simultaneous IMAP connections opened to any one
        /// identity. A per-identity cap set via `pigeon keyring
        /// add/modify email` still overrides this for that identity only
        /// (ADR-0080). Interactively prompted when omitted and stdin is a
        /// terminal; defaults to 2 otherwise -- well below the other email
        /// jobs' 6, since throughput here comes from
        /// `--messages-per-minute` rather than from parallelism.
        #[arg(long)]
        max_connections: Option<usize>,

        /// Maximum concurrent per-object pushes (`rclone copyto`).
        /// Interactively prompted when omitted and stdin is a terminal;
        /// defaults to 4 otherwise -- deliberately below the other jobs'
        /// upload concurrency, since these pushes interleave with a
        /// throttled fetch rather than racing to drain a finished queue.
        #[arg(long)]
        push_concurrency: Option<usize>,

        /// Alias of a configured bucket-config this run's report, the
        /// shared observability log, and a transcript of its printed
        /// output are uploaded to, always unencrypted, under a
        /// `YYYY-MM-DD-job-name-{run-id}/` prefix (ADR-0100). Mandatory --
        /// interactively selected when omitted and stdin is a terminal;
        /// required otherwise.
        #[arg(long)]
        report_bucket: Option<String>,

        /// Suppress every wizard prompt (ADR-0114): each omitted input
        /// resolves via its non-interactive fallback (or errors if none
        /// exists) instead of asking, and the final "proceed?"
        /// confirmation is auto-answered yes.
        #[arg(long)]
        non_interactive: bool,
    },
}

#[derive(Args, Debug)]
pub struct RcloneArgs {
    #[command(subcommand)]
    pub action: RcloneAction,
}

#[derive(Subcommand, Debug)]
pub enum RcloneAction {
    /// Copies data from one configured bucket-config to another by
    /// shelling out to `rclone copy`, against a config file pigeon itself
    /// generates from the keyring (ADR-0136, reversing ADR-0101's original
    /// raw-passthrough design). `--transfers`/`--checkers`/`--tpslimit` are
    /// overridable per-destination (ADR-0108); every other rclone
    /// performance/retry flag stays fixed.
    Copy {
        /// Where to pull from: a configured bucket-config alias, optionally
        /// composed with a prefix inside it -- `media` is the whole bucket,
        /// `media:mp4/` its `mp4/` directory (ADR-0138). Unlike
        /// `transform`, an omitted prefix stays the bucket root (this job
        /// has no file-type concept to derive one from). Interactively
        /// selected when omitted and stdin is a terminal; required
        /// otherwise.
        #[arg(long)]
        source_bucket: Option<String>,

        /// Where to push to, same `alias[:prefix/]` form as
        /// `--source-bucket` (ADR-0138). Interactively selected when
        /// omitted and stdin is a terminal; required otherwise.
        #[arg(long)]
        destination_bucket: Option<String>,

        /// Local directory this run's rclone log (also serving as this
        /// job's report) and transcript are written under. Defaults to a
        /// directory under the OS temp directory when omitted. Unlike
        /// every other job, this is not a staging area for transferred
        /// data -- rclone transfers directly source -> destination with no
        /// pigeon-side staging.
        #[arg(long)]
        local_output: Option<PathBuf>,

        /// Alias of a configured bucket-config this run's report (the
        /// rclone log itself, for this job), the shared observability log,
        /// and a transcript of its printed output are uploaded to, always
        /// unencrypted, under a `YYYY-MM-DD-job-name-{run-id}/` prefix
        /// (ADR-0100). Mandatory -- interactively selected when omitted
        /// and stdin is a terminal; required otherwise.
        #[arg(long)]
        report_bucket: Option<String>,

        /// rclone --transfers (concurrent file transfers). Defaults to 8
        /// (ADR-0106) when omitted.
        #[arg(long)]
        transfers: Option<usize>,

        /// rclone --checkers (concurrent list/compare operations).
        /// Defaults to 16 (ADR-0106) when omitted.
        #[arg(long)]
        checkers: Option<usize>,

        /// rclone --tpslimit (transactions/sec ceiling across transfers
        /// and checkers combined). Unset by default -- no rate cap --
        /// unless a specific destination needs one (ADR-0108).
        #[arg(long)]
        tpslimit: Option<usize>,

        /// Suppress every wizard prompt (ADR-0114): each omitted input
        /// resolves via its non-interactive fallback (or errors if none
        /// exists) instead of asking, and the final "proceed?"
        /// confirmation is auto-answered yes.
        #[arg(long)]
        non_interactive: bool,
    },

    /// Recursively and permanently deletes everything in a configured
    /// bucket-config by shelling out to `rclone purge` (ADR-0110).
    /// Irreversible -- there is no `--dry-run` and no undo beyond the
    /// existing `--non-interactive`/interactive confirmation. Unlike
    /// `copy`, there is no destination and no per-file transfer-rate
    /// tuning, since purge transfers no file content.
    Delete {
        /// What to recursively and permanently delete the contents of: a
        /// configured bucket-config alias, optionally composed with a
        /// prefix to scope the purge to one directory inside it
        /// (`media:mp4/`, ADR-0138). An omitted prefix purges the **whole
        /// bucket**. Interactively selected when omitted and stdin is a
        /// terminal; required otherwise.
        #[arg(long)]
        source_bucket: Option<String>,

        /// Local directory this run's rclone log (also serving as this
        /// job's report) and transcript are written under. Defaults to a
        /// directory under the OS temp directory when omitted.
        #[arg(long)]
        local_output: Option<PathBuf>,

        /// Alias of a configured bucket-config this run's report, the
        /// shared observability log, and a transcript of its printed
        /// output are uploaded to, always unencrypted, under a
        /// `YYYY-MM-DD-job-name-{run-id}/` prefix (ADR-0100). Mandatory --
        /// interactively selected when omitted and stdin is a terminal;
        /// required otherwise.
        #[arg(long)]
        report_bucket: Option<String>,

        /// rclone --checkers (concurrent list operations during
        /// enumeration). Defaults to 16 when omitted.
        #[arg(long)]
        checkers: Option<usize>,

        /// Suppress every wizard prompt (ADR-0114): each omitted input
        /// resolves via its non-interactive fallback (or errors if none
        /// exists) instead of asking, and the final "proceed?"
        /// confirmation is auto-answered yes.
        #[arg(long)]
        non_interactive: bool,
    },
}

impl Observable for JobType {
    fn command_name(&self) -> &'static str {
        match self {
            JobType::Email(args) => match args.action {
                EmailAction::Import { .. } => "job.email-import",
            },
            JobType::Deduplicate { .. } => "job.deduplicate",
            JobType::Rclone(args) => match args.action {
                RcloneAction::Copy { .. } => "job.rclone-copy",
                RcloneAction::Delete { .. } => "job.rclone-delete",
            },
            JobType::Transform { .. } => "job.transform",
        }
    }
}

impl JobType {
    /// The bare dash-form job name (e.g. `"deduplicate"`) used in the
    /// report-bucket upload prefix (ADR-0100) -- `command_name()`'s value
    /// with its `"job."` prefix stripped, rather than a 6th place in this
    /// codebase repeating the same job-name literals already duplicated
    /// across every job's `worker.rs` metrics call sites.
    pub(crate) fn job_name(&self) -> &'static str {
        self.command_name().trim_start_matches("job.")
    }
}