同期

[dependencies]
wakuwaku = { version = "1.0.0-rc3", features = ["meilisearch", "sqlx-pg"] }

use wakuwaku::deps::sqlx::{self, PgConnection};
use wakuwaku::entity::sqlx::backend::PgSource;
use wakuwaku::entity::{Entity, Execute, Query, Read};

/// `post` テーブル: 信頼できる唯一の情報源。
pub struct Post;
impl Entity for Post {}

/// `post` テーブルの行。
pub struct PostRow {
    pub id: i64,
    pub workspace_id: i64,
    pub author_id: i64,
    pub title: String,
    pub summary: String,
    pub tags: Vec<String>,
    /// 下書きの間は `None`。
    pub published_at: Option<i64>,
}

type PostTuple = (i64, i64, i64, String, String, Vec<String>, Option<i64>);

impl From<PostTuple> for PostRow {
    fn from(
        (id, workspace_id, author_id, title, summary, tags, published_at): PostTuple,
    ) -> Self {
        Self { id, workspace_id, author_id, title, summary, tags, published_at }
    }
}

/// 投稿を id で読み込む。
pub struct LoadPosts {
    ids: Vec<i64>,
}

impl Query for LoadPosts {
    type Effect = Read<Post>;
}

impl Execute<PgSource> for LoadPosts {
    type Output = Vec<PostRow>;
    type Error = sqlx::Error;

    async fn execute(self, conn: &mut PgConnection) -> Result<Vec<PostRow>, sqlx::Error> {
        let rows: Vec<PostTuple> = sqlx::query_as(
            "SELECT id, workspace_id, author_id, title, summary, tags, \
                    EXTRACT(EPOCH FROM published_at)::bigint \
             FROM post WHERE id = ANY($1)",
        )
        .bind(&self.ids)
        .fetch_all(conn)
        .await?;
        Ok(rows.into_iter().map(PostRow::from).collect())
    }
}

use wakuwaku::entity::meilisearch::sync::Projection;

/// `post` の行が `posts` のドキュメントになる方法。
pub struct PostProjection;

impl Projection for PostProjection {
    type Index = PostIndex;
    type Row = PostRow;
    type Load = LoadPosts;

    fn load(&self, keys: &[i64]) -> LoadPosts {
        LoadPosts { ids: keys.to_vec() }
    }

    fn key(&self, row: &PostRow) -> i64 {
        row.id
    }

    fn project(&self, row: PostRow) -> Option<PostDocument> {
        // 下書きは検索対象にしない: `None` はそのドキュメントを削除する。
        let published_at = row.published_at?;
        Some(PostDocument {
            id: row.id,
            workspace_id: row.workspace_id,
            author_id: row.author_id,
            title: row.title,
            summary: row.summary,
            tags: row.tags,
            published_at,
        })
    }
}

use std::convert::Infallible;
use wakuwaku::deps::kanau::processor::Processor;
use wakuwaku::entity::meilisearch::sync::{Indexer, Reindex, SyncError};
use wakuwaku::entity::meilisearch::MeiliSource;
use wakuwaku::entity::{AfterWrite, Db, ReadOnly, ReadWrite, Write, WriteQuery};

/// 下書きを公開する。
pub struct PublishPost {
    pub id: i64,
}

impl Query for PublishPost {
    type Effect = (Read<Post>, Write<Post>);
}

impl Execute<PgSource> for PublishPost {
    type Output = u64;
    type Error = sqlx::Error;

    async fn execute(self, conn: &mut PgConnection) -> Result<u64, sqlx::Error> {
        let done = sqlx::query("UPDATE post SET published_at = now() WHERE id = $1")
            .bind(self.id)
            .execute(conn)
            .await?;
        Ok(done.rows_affected())
    }
}

/// `post` に対するすべての書き込みクエリがフックに報告するもの。
pub struct PostChanged {
    pub id: i64,
}

impl WriteQuery for PublishPost {
    type Change = PostChanged;

    fn change(&self) -> PostChanged {
        PostChanged { id: self.id }
    }
}

impl From<PostChanged> for Reindex<PostIndex> {
    fn from(change: PostChanged) -> Self {
        Reindex::one(change.id)
    }
}

/// フックのエラーはここで終わる。リクエストはすでに成功している。
pub struct LogSyncError;

impl Processor<SyncError<sqlx::Error>> for LogSyncError {
    type Output = ();
    type Error = Infallible;

    async fn process(&self, error: SyncError<sqlx::Error>) -> Result<(), Infallible> {
        eprintln!("reindexing posts failed: {error}");
        Ok(())
    }
}

pub type PostIndexerFor = Indexer<Db<PgSource, ReadOnly>, Db<MeiliSource, PostIndexer>, PostProjection>;

pub fn wire(
    primary: PgSource,
    meili: Db<MeiliSource, PostIndexer>,
) -> AfterWrite<Db<PgSource, ReadWrite>, PostIndexerFor, LogSyncError> {
    // プライマリから読み込む: 遅延しているレプリカは古い行を返しうる。
    let loader: Db<PgSource, ReadOnly> = Db::new(primary.clone());
    let indexer = Indexer::new(loader, meili, PostProjection).without_waiting();
    AfterWrite::new(Db::new(primary), indexer, LogSyncError)
}

use wakuwaku::deps::sqlx::Postgres;
use wakuwaku::entity::sqlx::transaction::{Either, Tx, TxEnd, TxFlow, TxState};
use wakuwaku::entity::tx_effect;

/// 複数の下書きを 1 つのトランザクションで公開する: すべてか、何もしないか。
pub enum PublishAll {
    /// `pending` はこれから公開するもの、`done` は公開済みのもの。
    Running { pending: Vec<i64>, done: Vec<i64> },
    End(TxEnd<PublishAll>),
}

/// `PublishAll` がロールバックした理由。
pub enum PublishAllFailed {
    /// この id の投稿がない。
    Missing(i64),
    Database(sqlx::Error),
}

tx_effect! {
    /// `PublishAll` が触れるものすべて。
    pub enum PublishAllEffect { write Post }
}

impl From<TxEnd<Self>> for PublishAll {
    fn from(end: TxEnd<Self>) -> Self {
        Self::End(end)
    }
}

impl TxFlow for PublishAll {
    type Capability = PublishAllEffect;
    type Database = Postgres;
    type Commit = ();
    type Abort = PublishAllFailed;
    /// トランザクションが触れた投稿。フックのためのもの。
    type Report = Reindex<PostIndex>;

    async fn step(tx: Tx<'_, Self>, state: Self) -> TxState<'_, Self> {
        let Self::Running { mut pending, mut done } = state else {
            unreachable!("never stepped when terminal")
        };
        let Some(id) = pending.pop() else {
            // すべて公開した: ドライバーがコミットする。
            return tx.with(Self::End(TxEnd::Commit { output: (), report: Reindex::new(done) }));
        };
        let (tx, published) = tx.run(PublishPost { id }).await;
        let failed = match published {
            Ok(1) => {
                done.push(id);
                return tx.with(Self::Running { pending, done });
            }
            Ok(_) => PublishAllFailed::Missing(id),
            Err(error) => PublishAllFailed::Database(error),
        };
        tx.with(Self::End(TxEnd::Rollback { output: failed, report: Reindex::new(done) }))
    }

    fn finished(self) -> Either<Self, TxEnd<Self>> {
        match self {
            Self::End(end) => Either::Right(end),
            state => Either::Left(state),
        }
    }
}

use wakuwaku::entity::meilisearch::sync::Reindexed;
use wakuwaku::entity::sqlx::transaction::TxFinalState;
use wakuwaku::entity::sqlx::transaction::hook::AfterTransaction;

/// トランザクションフック: コミットされたら、トランザクションが触れたものを再インデックスする。
pub struct ReindexCommitted<I>(pub I);

impl<I> Processor<TxFinalState<Reindex<PostIndex>>> for ReindexCommitted<I>
where
    I: Processor<Reindex<PostIndex>, Output = Reindexed> + Sync,
    I::Error: Send,
{
    type Output = ();
    type Error = I::Error;

    async fn process(&self, state: TxFinalState<Reindex<PostIndex>>) -> Result<(), I::Error> {
        match state {
            // 読み直しは冪等なので、コミット不明でも再インデックスして安全。
            TxFinalState::Committed(keys) | TxFinalState::CommitUnknown(keys) => {
                self.0.process(keys).await?;
                Ok(())
            }
            TxFinalState::RolledBack(_) => Ok(()),
        }
    }
}

pub fn wire_transactions(
    primary: PgSource,
    meili: Db<MeiliSource, PostIndexer>,
) -> AfterTransaction<Db<PgSource, ReadWrite>, ReindexCommitted<PostIndexerFor>, LogSyncError> {
    let loader: Db<PgSource, ReadOnly> = Db::new(primary.clone());
    let indexer = Indexer::new(loader, meili, PostProjection).without_waiting();
    AfterTransaction::new(Db::new(primary), ReindexCommitted(indexer), LogSyncError)
}

use wakuwaku::entity::sqlx::transaction::{Transaction, TxError, TxOutput};

/// `ids` を公開する。トランザクションがコミットされたら、フックがそれらを再インデックスする。
pub async fn publish_all<D>(db: &D, ids: Vec<i64>) -> Result<TxOutput<(), PublishAllFailed>, TxError>
where
    D: Processor<Transaction<PublishAll>, Output = TxOutput<(), PublishAllFailed>, Error = TxError>,
{
    db.process(Transaction(PublishAll::Running { pending: ids, done: Vec::new() })).await
}

use wakuwaku::entity::meilisearch::sync::Scan;

/// `post` の 1 ページ (id 順)。
pub struct PostPage {
    after: Option<i64>,
    limit: i64,
}

impl Query for PostPage {
    type Effect = Read<Post>;
}

impl Execute<PgSource> for PostPage {
    type Output = Vec<PostRow>;
    type Error = sqlx::Error;

    async fn execute(self, conn: &mut PgConnection) -> Result<Vec<PostRow>, sqlx::Error> {
        let rows: Vec<PostTuple> = sqlx::query_as(
            "SELECT id, workspace_id, author_id, title, summary, tags, \
                    EXTRACT(EPOCH FROM published_at)::bigint \
             FROM post WHERE $1::bigint IS NULL OR id > $1 ORDER BY id LIMIT $2",
        )
            .bind(self.after)
            .bind(self.limit)
            .fetch_all(conn)
            .await?;
        Ok(rows.into_iter().map(PostRow::from).collect())
    }
}

impl Scan for PostProjection {
    type Page = PostPage;

    fn page(&self, after: Option<&i64>, limit: usize) -> PostPage {
        PostPage {
            after: after.copied(),
            limit: i64::try_from(limit).unwrap_or(i64::MAX),
        }
    }
}

use wakuwaku::entity::meilisearch::sync::{RebuildOptions, Rebuilt, rebuild};

/// `post` から `posts` を再構築する。
pub async fn rebuild_posts(
    source: &Db<PgSource, ReadOnly>,
    meili: &Db<MeiliSource, PostIndexer>,
) -> Result<Rebuilt, SyncError<sqlx::Error>> {
    let rebuilt = rebuild(&PostProjection, source, meili, &RebuildOptions::default()).await?;
    eprintln!("rebuilt posts with {} documents", rebuilt.documents);
    Ok(rebuilt)
}

use std::time::Duration;
use wakuwaku::{CronJob, CronJobInterval, CronTick};

/// 毎晩 `posts` を再構築する。
pub struct RebuildPosts {
    pub source: Db<PgSource, ReadOnly>,
    pub meili: Db<MeiliSource, PostIndexer>,
}

impl Processor<CronTick> for RebuildPosts {
    type Output = ();
    type Error = SyncError<sqlx::Error>;

    async fn process(&self, _tick: CronTick) -> Result<(), Self::Error> {
        rebuild_posts(&self.source, &self.meili).await?;
        Ok(())
    }
}

impl CronJob for RebuildPosts {
    // 再構築の終了から次の開始まで 1 日: 実行は決して重ならない。
    const INTERVAL: CronJobInterval = CronJobInterval::AfterFinish(Duration::from_secs(24 * 60 * 60));
}

use std::time::{SystemTime, UNIX_EPOCH};

/// `since` 以降に変更された投稿の id (論理削除されたものを含む)。
pub struct ChangedSince {
    since: SystemTime,
}

impl Query for ChangedSince {
    type Effect = Read<Post>;
}

impl Execute<PgSource> for ChangedSince {
    type Output = Vec<i64>;
    type Error = sqlx::Error;

    async fn execute(self, conn: &mut PgConnection) -> Result<Vec<i64>, sqlx::Error> {
        let since = self.since.duration_since(UNIX_EPOCH).unwrap_or_default().as_secs_f64();
        sqlx::query_scalar("SELECT id FROM post WHERE updated_at >= to_timestamp($1)")
            .bind(since)
            .fetch_all(conn)
            .await
    }
}

pub async fn rebuild_and_catch_up(
    source: &Db<PgSource, ReadOnly>,
    meili: &Db<MeiliSource, PostIndexer>,
) -> Result<Reindexed, SyncError<sqlx::Error>> {
    let rebuilt = rebuild(&PostProjection, source, meili, &RebuildOptions::default()).await?;
    let changed = source
        .process(ChangedSince { since: rebuilt.started_at })
        .await
        .map_err(SyncError::Source)?;
    let indexer = Indexer::new(source.clone(), meili.clone(), PostProjection);
    indexer.process(Reindex::<PostIndex>::new(changed)).await
}