meli
/
mailpot
4
Fork 0
Code Issues 7 Pull Requests 2 Releases Wiki Activity

Add error queue for unhandled emails

pull/1/head
Manos Pitsidianakis 2022-06-05 14:14:00 +03:00
parent b6ac161fb3
commit 43e100917b
5 changed files with 181 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

89
cli/src/main.rs
View File

@ -99,6 +99,35 @@ enum Command {
#[structopt(long)]
dry_run: bool,
},
/// Mail that has not been handled properly end up in the error queue.
ErrorQueue {
#[structopt(subcommand)]
cmd: ErrorQueueCommand,
},
}
#[derive(Debug, StructOpt)]
enum ErrorQueueCommand {
/// List.
List,
/// Print entry in RFC5322 or JSON format.
Print {
/// index of entry.
#[structopt(long)]
index: Vec<i64>,
/// JSON format.
#[structopt(long)]
json: bool,
},
/// Delete entry and print it in stdout.
Delete {
/// index of entry.
#[structopt(long)]
index: Vec<i64>,
/// Do not print in stdout.
#[structopt(long)]
quiet: bool,
},
}
#[derive(Debug, StructOpt)]
@ -533,6 +562,66 @@ fn run_app(opt: Opt) -> Result<()> {
}
}
}
ErrorQueue { cmd } => match cmd {
ErrorQueueCommand::List => {
let db = Database::open_or_create_db()?;
let errors = db.error_queue()?;
if errors.is_empty() {
println!("Error queue is empty.");
} else {
for e in errors {
println!(
"- {} {} {} {} {}",
e["pk"],
e["datetime"],
e["from_address"],
e["to_address"],
e["subject"]
);
}
}
}
ErrorQueueCommand::Print { index, json } => {
let db = Database::open_or_create_db()?;
let mut errors = db.error_queue()?;
if !index.is_empty() {
errors.retain(|el| index.contains(&el.pk()));
}
if errors.is_empty() {
println!("Error queue is empty.");
} else {
for e in errors {
if json {
println!("{:#}", e);
} else {
println!("{}", e["message"]);
}
}
}
}
ErrorQueueCommand::Delete { index, quiet } => {
let mut db = Database::open_or_create_db()?;
let mut errors = db.error_queue()?;
if !index.is_empty() {
errors.retain(|el| index.contains(&el.pk()));
}
if errors.is_empty() {
if !quiet {
println!("Error queue is empty.");
}
} else {
if !quiet {
println!("Deleting error queue elements {:?}", &index);
}
db.delete_from_error_queue(index)?;
if !quiet {
for e in errors {
println!("{}", e["message"]);
}
}
}
}
},
}
Ok(())

3
core/src/db.rs
View File

@ -34,6 +34,9 @@ pub struct Database {
pub connection: DbConnection,
}
mod error_queue;
pub use error_queue::*;
impl Database {
pub fn db_path() -> Result<PathBuf> {
let mut config_path = None;

67
core/src/db/error_queue.rs 100644
View File

@ -0,0 +1,67 @@
/*
* This file is part of mailpot
*
* Copyright 2020 - Manos Pitsidianakis
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
use super::*;
use serde_json::{json, Value};
impl Database {
pub fn error_queue(&self) -> Result<Vec<DbVal<Value>>> {
let mut stmt = self.connection.prepare("SELECT * FROM error_queue;")?;
let error_iter = stmt.query_map([], |row| {
let pk = row.get::<_, i64>("pk")?;
Ok(DbVal(
json!({
"pk" : pk,
"to_address": row.get::<_, String>("to_address")?,
"from_address": row.get::<_, String>("from_address")?,
"subject": row.get::<_, String>("subject")?,
"message_id": row.get::<_, String>("message_id")?,
"message": row.get::<_, String>("message")?,
"timestamp": row.get::<_, String>("timestamp")?,
"datetime": row.get::<_, String>("datetime")?,
}),
pk,
))
})?;
let mut ret = vec![];
for error in error_iter {
let error = error?;
ret.push(error);
}
Ok(ret)
}
pub fn delete_from_error_queue(&mut self, index: Vec<i64>) -> Result<()> {
let tx = self.connection.transaction()?;
if index.is_empty() {
tx.execute("DELETE FROM error_queue;", [])?;
} else {
for i in index {
tx.execute(
"DELETE FROM error_queue WHERE pk = ?;",
rusqlite::params![i],
)?;
}
};
tx.commit()?;
Ok(())
}
}

11
core/src/schema.sql
View File

@ -82,6 +82,17 @@ CREATE TABLE IF NOT EXISTS post_event (
FOREIGN KEY (post) REFERENCES post(pk) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS error_queue (
pk INTEGER PRIMARY KEY NOT NULL,
to_address TEXT NOT NULL,
from_address TEXT NOT NULL,
subject TEXT NOT NULL,
message_id TEXT NOT NULL,
message BLOB NOT NULL,
timestamp INTEGER NOT NULL DEFAULT (unixepoch()),
datetime TEXT NOT NULL DEFAULT (datetime())
);
CREATE INDEX IF NOT EXISTS post_listpk_idx ON post(list);
CREATE INDEX IF NOT EXISTS post_msgid_idx ON post(message_id);
CREATE INDEX IF NOT EXISTS mailing_lists_idx ON mailing_lists(id);

11
core/src/schema.sql.m4
View File

@ -86,6 +86,17 @@ CREATE TABLE IF NOT EXISTS post_event (
FOREIGN KEY (post) REFERENCES post(pk) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS error_queue (
pk INTEGER PRIMARY KEY NOT NULL,
to_address TEXT NOT NULL,
from_address TEXT NOT NULL,
subject TEXT NOT NULL,
message_id TEXT NOT NULL,
message BLOB NOT NULL,
timestamp INTEGER NOT NULL DEFAULT (unixepoch()),
datetime TEXT NOT NULL DEFAULT (datetime())
);
CREATE INDEX IF NOT EXISTS post_listpk_idx ON post(list);
CREATE INDEX IF NOT EXISTS post_msgid_idx ON post(message_id);
CREATE INDEX IF NOT EXISTS mailing_lists_idx ON mailing_lists(id);