書き込みフック

/// `users` への書き込みで何が変わったか。users へのすべての書き込みクエリで共有する
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UserChanged {
    pub ids: Vec<i64>,
    pub action: &'static str,
}

impl WriteQuery for RenameUser {
    type Change = UserChanged;

    fn change(&self) -> UserChanged {
        UserChanged { ids: vec![self.id], action: "rename" }
    }
}

impl WriteQuery for DeactivateUsers {
    type Change = UserChanged;

    fn change(&self) -> UserChanged {
        UserChanged { ids: self.ids.clone(), action: "deactivate" }
    }
}

/// `audit_log` テーブル
pub struct AuditLog;
impl Entity for AuditLog {}

pub struct RecordAudit {
    pub subject: i64,
    pub action: &'static str,
}

impl Query for RecordAudit {
    type Effect = Write<AuditLog>;
}

impl Execute<PgSource> for RecordAudit {
    type Output = ();
    type Error = sqlx::Error;

    async fn execute(self, conn: &mut PgConnection) -> Result<(), sqlx::Error> {
        sqlx::query("INSERT INTO audit_log (subject, action, at) VALUES ($1, $2, now())")
            .bind(self.subject)
            .bind(self.action)
            .execute(conn)
            .await?;
        Ok(())
    }
}

/// 変更されたユーザーごとに監査行を 1 行書き込む
#[derive(Clone)]
pub struct AuditTrail {
    db: Db<PgSource, ReadWrite>,
}

impl Processor<UserChanged> for AuditTrail {
    type Output = ();
    type Error = sqlx::Error;

    async fn process(&self, change: UserChanged) -> Result<(), sqlx::Error> {
        for subject in change.ids {
            self.db.process(RecordAudit { subject, action: change.action }).await?;
        }
        Ok(())
    }
}

/// エラーハンドラ。フックの失敗はここで止まり、呼び出し元には届かない
#[derive(Clone)]
pub struct LogHookError;

impl<E: Display + Send + 'static> Processor<E> for LogHookError {
    type Output = ();
    type Error = std::convert::Infallible;

    async fn process(&self, error: E) -> Result<(), std::convert::Infallible> {
        eprintln!("write hook failed: {error}");
        Ok(())
    }
}

/// 監査付きのプライマリハンドル
pub type AuditedPrimary = AfterWrite<Db<PgSource, ReadWrite>, AuditTrail, LogHookError>;

pub fn audited(dbs: &Databases) -> AuditedPrimary {
    let audit = AuditTrail { db: dbs.primary.clone() };
    AfterWrite::new(dbs.primary.clone(), audit, LogHookError)
}

pub async fn use_it(dbs: &Databases) -> Result<(), sqlx::Error> {
    let users = audited(dbs);

    // 素のハンドルと同じ出力・同じエラー。そのまま置き換えられる
    let renamed: bool = rename_user(&users, 7, "Haruki".into()).await?;

    // 読み取りはフックに触れずに素通りする
    let user: Option<UserRow> = users.process(FindUser { id: 7 }).await?;
    println!("{renamed} {user:?}");
    Ok(())
}

/// これから実行される書き込みを数える。失敗しない
#[derive(Clone, Default)]
pub struct CountWrites(Arc<AtomicU64>);

impl Processor<UserChanged> for CountWrites {
    type Output = ();
    type Error = std::convert::Infallible;

    async fn process(&self, _: UserChanged) -> Result<(), std::convert::Infallible> {
        self.0.fetch_add(1, Ordering::Relaxed);
        Ok(())
    }
}

pub type Instrumented =
    AfterWrite<BeforeWrite<Db<PgSource, ReadWrite>, CountWrites, LogHookError>, AuditTrail, LogHookError>;

pub fn instrumented(dbs: &Databases, counter: CountWrites) -> Instrumented {
    let counted = BeforeWrite::new(dbs.primary.clone(), counter, LogHookError);
    AfterWrite::new(counted, AuditTrail { db: dbs.primary.clone() }, LogHookError)
}

users.process(DeletePost { id: 1 }).await

error[E0277]: the trait bound `DeletePost: WriteQuery` is not satisfied
   = note: required for `WriteKind` to implement `HookDispatch<DeletePost, Db<SqlxSource<Postgres>, ReadWrite>, AuditTrail, LogHookError>`
   = note: required for `AfterWrite<Db<SqlxSource<Postgres>, ReadWrite>, AuditTrail, LogHookError>` to implement `Processor<DeletePost>`

error[E0277]: the trait bound `AuditTrail: Processor<PostChanged>` is not satisfied

#[cfg(test)]
mod tests {
    use super::*;
    use std::convert::Infallible;
    use tokio::sync::Mutex;

    /// データベースの代役
    struct FakeUsers;

    impl Processor<RenameUser> for FakeUsers {
        type Output = u64;
        type Error = &'static str;

        async fn process(&self, query: RenameUser) -> Result<u64, &'static str> {
            if query.id == 0 { Err("no such user") } else { Ok(1) }
        }
    }

    /// 受け取った変更をすべて記録する
    #[derive(Default)]
    struct Recorder(Mutex<Vec<UserChanged>>);

    impl Processor<UserChanged> for Recorder {
        type Output = ();
        type Error = Infallible;

        async fn process(&self, change: UserChanged) -> Result<(), Infallible> {
            self.0.lock().await.push(change);
            Ok(())
        }
    }

    #[tokio::test]
    async fn hook_runs_only_after_successful_writes() {
        let recorder = Recorder::default();
        let users = AfterWrite::new(FakeUsers, &recorder, LogHookError);

        assert_eq!(users.process(RenameUser { id: 7, name: "Haruki".into() }).await, Ok(1));
        assert_eq!(users.process(RenameUser { id: 0, name: "Nobody".into() }).await, Err("no such user"));

        let seen = recorder.0.lock().await;
        assert_eq!(*seen, [UserChanged { ids: vec![7], action: "rename" }]);
    }
}