maq/sn0int
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Trigger events on database inserts and updates

Browse Source
This commit is contained in:
kpcyrd
2020-06-12 21:32:15 +02:00
parent 9825bad1fe
commit 983df4a12e
5 changed files with 93 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
modules/harness/notify-ohai.lua
View File

@@ -4,7 +4,5 @@
-- Source: notifications
function run(arg)
-- TODO your code here
sleep(1)
info('notication!: ' .. json_encode(arg))
end

10
modules/harness/notify-slow.lua Normal file
View File

@@ -0,0 +1,10 @@
-- Description: TODO your description here
-- Version: 0.1.0
-- License: GPL-3.0
-- Source: notifications
function run(arg)
-- TODO your code here
sleep(1)
info('notication!: ' .. json_encode(arg))
end

26
src/models/domain.rs
View File

@@ -122,24 +122,6 @@ impl Domain {
}
}
pub struct PrintableDomain {
value: String,
}
impl fmt::Display for PrintableDomain {
fn fmt(&self, w: &mut fmt::Formatter) -> fmt::Result {
write!(w, "{:?}", self.value)
}
}
impl Printable<PrintableDomain> for Domain {
fn printable(&self, _db: &Database) -> Result<PrintableDomain> {
Ok(PrintableDomain {
value: self.value.to_string(),
})
}
}
pub struct DetailedDomain {
id: i32,
value: String,
@@ -222,14 +204,6 @@ impl Upsertable<Domain> for NewDomain {
}
}
impl Printable<PrintableDomain> for NewDomain {
fn printable(&self, _db: &Database) -> Result<PrintableDomain> {
Ok(PrintableDomain {
value: self.value.to_string(),
})
}
}
#[derive(Debug, Serialize, Deserialize)]
pub struct InsertDomain {
pub value: String,

87
src/models/mod.rs
View File

@@ -1,9 +1,10 @@
use crate::db::{Database, Table, Filter, Family};
use crate::engine::ctx::State;
use crate::errors::*;
use crate::db::{Database, Table, Filter};
use crate::fmt;
use crate::schema::*;
use std::borrow::Cow;
use std::sync::Arc;
use crate::engine::ctx::State;
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -28,39 +29,39 @@ pub enum Insert {
}
impl Insert {
pub fn label(&self, db: &Database) -> Result<String> {
let label = match self {
Insert::Domain(x) => format!("{:?}", x.value),
Insert::Subdomain(x) => format!("{:?}", x.value),
Insert::IpAddr(x) => format!("{:?}", x.value),
pub fn value(&self, db: &Database) -> Result<Cow<String>> {
let value = match self {
Insert::Domain(x) => Cow::Borrowed(&x.value),
Insert::Subdomain(x) => Cow::Borrowed(&x.value),
Insert::IpAddr(x) => Cow::Borrowed(&x.value),
Insert::SubdomainIpAddr(x) => {
let subdomain = Subdomain::by_id(db, x.subdomain_id)?;
let ipaddr = IpAddr::by_id(db, x.ip_addr_id)?;
format!("{:?}+{:?}", subdomain.value, ipaddr.value)
Cow::Owned(format!("{}+{}", subdomain.value, ipaddr.value))
},
Insert::Url(x) => format!("{:?}", x.value),
Insert::Email(x) => format!("{:?}", x.value),
Insert::PhoneNumber(x) => format!("{:?}", x.value),
Insert::Device(x) => format!("{:?}", x.value),
Insert::Network(x) => format!("{:?}", x.value),
Insert::Url(x) => Cow::Borrowed(&x.value),
Insert::Email(x) => Cow::Borrowed(&x.value),
Insert::PhoneNumber(x) => Cow::Borrowed(&x.value),
Insert::Device(x) => Cow::Borrowed(&x.value),
Insert::Network(x) => Cow::Borrowed(&x.value),
Insert::NetworkDevice(x) => {
let network = Network::by_id(db, x.network_id)?;
let device = Device::by_id(db, x.device_id)?;
format!("{:?}+{:?}", network.value, device.value)
Cow::Owned(format!("{}+{}", network.value, device.value))
},
Insert::Account(x) => format!("{:?}", x.value),
Insert::Breach(x) => format!("{:?}", x.value),
Insert::Account(x) => Cow::Borrowed(&x.value),
Insert::Breach(x) => Cow::Borrowed(&x.value),
Insert::BreachEmail(x) => {
let breach = Breach::by_id(db, x.breach_id)?;
let email = Email::by_id(db, x.email_id)?;
format!("{:?}+{:?}", breach.value, email.value)
Cow::Owned(format!("{}+{}", breach.value, email.value))
}
Insert::Image(x) => format!("{:?}", x.value),
Insert::Port(x) => format!("{:?}", x.value),
Insert::Netblock(x) => format!("{:?}", x.value),
Insert::CryptoAddr(x) => format!("{:?}", x.value),
Insert::Image(x) => Cow::Borrowed(&x.value),
Insert::Port(x) => Cow::Borrowed(&x.value),
Insert::Netblock(x) => Cow::Borrowed(&x.value),
Insert::CryptoAddr(x) => Cow::Borrowed(&x.value),
};
Ok(label)
Ok(value)
}
#[inline]
@@ -68,26 +69,27 @@ impl Insert {
Table::from(self).into()
}
pub fn printable(&self, db: &Database) -> Result<String> {
Ok(match self {
Insert::Domain(x) => format!("Domain: {}", x.printable(db)?),
Insert::Subdomain(x) => format!("Subdomain: {}", x.printable(db)?),
Insert::IpAddr(x) => format!("IpAddr: {}", x.printable(db)?),
Insert::SubdomainIpAddr(x) => x.printable(db)?.to_string(),
Insert::Url(x) => format!("Url: {}", x.printable(db)?),
Insert::Email(x) => format!("Email: {}", x.printable(db)?),
Insert::PhoneNumber(x) => format!("PhoneNumber: {}", x.printable(db)?),
Insert::Device(x) => format!("Device: {}", x.printable(db)?),
Insert::Network(x) => format!("Network: {}", x.printable(db)?),
Insert::NetworkDevice(x) => x.printable(db)?.to_string(),
Insert::Account(x) => format!("Account: {}", x.printable(db)?),
Insert::Breach(x) => format!("Breach: {}", x.printable(db)?),
Insert::BreachEmail(x) => x.printable(db)?.to_string(),
Insert::Image(x) => format!("Image: {}", x.printable(db)?),
Insert::Port(x) => format!("Port: {}", x.printable(db)?),
Insert::Netblock(x) => format!("Netblock: {}", x.printable(db)?),
Insert::CryptoAddr(x) => format!("CryptoAddr: {}", x.printable(db)?),
})
#[inline]
pub fn family(&self) -> &str {
match self {
Insert::Domain(_) => Family::Domain.as_str(),
Insert::Subdomain(_) => Family::Subdomain.as_str(),
Insert::IpAddr(_) => Family::Ipaddr.as_str(),
Insert::SubdomainIpAddr(_) => Family::SubdomainIpaddr.as_str(),
Insert::Url(_) => Family::Url.as_str(),
Insert::Email(_) => Family::Email.as_str(),
Insert::PhoneNumber(_) => Family::Phonenumber.as_str(),
Insert::Device(_) => Family::Device.as_str(),
Insert::Network(_) => Family::Network.as_str(),
Insert::NetworkDevice(_) => Family::NetworkDevice.as_str(),
Insert::Account(_) => Family::Account.as_str(),
Insert::Breach(_) => Family::Breach.as_str(),
Insert::BreachEmail(_) => Family::BreachEmail.as_str(),
Insert::Image(_) => Family::Image.as_str(),
Insert::Port(_) => Family::Port.as_str(),
Insert::Netblock(_) => Family::Netblock.as_str(),
Insert::CryptoAddr(_) => Family::Cryptoaddr.as_str(),
}
}
}
@@ -320,6 +322,7 @@ pub trait Updateable<M> {
fn fmt(&self, updates: &mut Vec<String>);
}
// TODO: Printable could probably be dropped
pub trait Printable<T: Sized> {
fn printable(&self, db: &Database) -> Result<T>;
}

60
src/worker.rs
View File

@@ -3,7 +3,7 @@ use crate::errors::*;
use crate::blobs::Blob;
use crate::channel;
use crate::cmd::run_cmd::Params;
use crate::db::{Database, DbChange, Family};
use crate::db::{DbChange, Family};
use crate::db::ttl::Ttl;
use crate::engine::Module;
use crate::ipc;
@@ -169,7 +169,17 @@ impl EventWithCallback for DatabaseEvent {
}
impl DatabaseEvent {
pub fn insert<T: SpinLogger>(object: Insert, ttl: Option<i32>, tx: DbSender, spinner: &mut T, db: &Database, verbose: u64) {
pub fn notify<T: SpinLogger>(rl: &mut Shell, spinner: &mut T, topic: &str, subject: String) {
if let Err(err) = notify::trigger_notify_event(rl, spinner, topic, &Notification {
subject,
body: None,
}) {
spinner.error(&format!("Failed to send notifications: {}", err));
}
}
pub fn insert<T: SpinLogger>(rl: &mut Shell, object: Insert, ttl: Option<i32>, tx: DbSender, spinner: &mut T, verbose: u64) {
let db = rl.db();
if verbose >= 1 {
spinner.debug(&format!("Inserting: {:?}", object));
}
@@ -185,9 +195,14 @@ impl DatabaseEvent {
}
}
// TODO: replace id with actual object(?)
if let Ok(obj) = object.printable(db) {
spinner.log(&obj.to_string()); // TODO: also include fields here
if let Ok(value) = object.value(db) {
// TODO: also include fields, see update
let log = format!("Adding {} {:?}", object.family(), value); // TODO: also include fields here
spinner.log(&log);
let subject = format!("Added {} {:?}", object.family(), value);
let topic = format!("db:{}:{}:insert", object.family(), value);
Self::notify(rl, spinner, &topic, subject);
} else {
spinner.error(&format!("Failed to query necessary fields for {:?}", object));
}
@@ -201,9 +216,15 @@ impl DatabaseEvent {
}
// TODO: replace id with actual object(?)
match object.label(&db) {
Ok(label) => {
spinner.log(&format!("Updating {} ({})", label, update));
match object.value(&db) {
Ok(value) => {
spinner.log(&format!("Updating {} {:?} ({})", object.family(), value, update));
// TODO: in the future we could consider firing multiple events, one for each column
// TODO: this would be super noisy if a lot of fields change though
let subject = format!("Updated {} {:?} ({})", object.family(), value, update);
let topic = format!("db:{}:{}:update", object.family(), value);
Self::notify(rl, spinner, &topic, subject);
},
Err(err) => {
// TODO: this should be unreachable
@@ -262,22 +283,16 @@ impl DatabaseEvent {
let result = match result {
Ok(true) => {
Self::spinner_log_new_activity(&object, spinner, verbose);
// TODO: we don't want to copy the match arms everywhere
let topic = format!("activity:{}", object.topic);
let mut subject = format!("New activity: {:?}", object.topic);
if let Some(uniq) = &object.uniq {
subject += &format!(" ({:?})", uniq);
}
if let Err(err) = notify::trigger_notify_event(rl, spinner, &topic, &Notification {
subject,
body: None,
}) {
let err = err.to_string();
spinner.error(&format!("Failed to send notifications: {}", err));
Err(err)
} else {
Ok(DatabaseResponse::Inserted(0))
}
let topic = format!("activity:{}", object.topic);
Self::notify(rl, spinner, &topic, subject);
Ok(DatabaseResponse::Inserted(0))
},
Ok(false) => Ok(DatabaseResponse::NoChange(0)),
Err(err) => {
@@ -291,12 +306,12 @@ impl DatabaseEvent {
}
pub fn apply<T: SpinLogger>(self, rl: &mut Shell, tx: DbSender, spinner: &mut T, verbose: u64) {
let db = rl.db();
match self {
DatabaseEvent::Insert(object) => Self::insert(object, None, tx, spinner, db, verbose),
DatabaseEvent::InsertTtl((object, ttl)) => Self::insert(object, Some(ttl), tx, spinner, db, verbose),
DatabaseEvent::Insert(object) => Self::insert(rl, object, None, tx, spinner, verbose),
DatabaseEvent::InsertTtl((object, ttl)) => Self::insert(rl, object, Some(ttl), tx, spinner, verbose),
DatabaseEvent::Activity(object) => Self::activity(rl, object, tx, spinner, verbose),
DatabaseEvent::Select((family, value)) => {
let db = rl.db();
let result = match db.get_opt(&family, &value) {
Ok(Some(id)) => Ok(DatabaseResponse::Found(id)),
Ok(None) => Ok(DatabaseResponse::None),
@@ -306,6 +321,7 @@ impl DatabaseEvent {
tx.send(result).expect("Failed to send db result to channel");
},
DatabaseEvent::Update((object, update)) => {
let db = rl.db();
if verbose >= 1 {
spinner.debug(&format!("Updating: {:?}", update));
}