Thunder & Candidate Pipeline 仕様書

// パーティション分散によるスレッド処理
let num_partitions = args.tweet_events_num_partitions;
let kafka_num_threads = args.kafka_num_threads;
let partitions_per_thread = num_partitions.div_ceil(kafka_num_threads);

// バッチが一定サイズに達したら処理を実行
if message_buffer.len() >= batch_size {
    let messages = std::mem::take(&mut message_buffer);
    process_message_batch(messages, batch_num, producer_clone, post_retention_sec).await?;
    consumer.write().await.commit_offsets()?;
}

// Kafka catchup 完了を待機
for _ in 0..args.kafka_num_threads {
    rx.recv().await;
}
post_store.finalize_init().await?;

pub struct PostStore {
    /// 投稿ID から完全な投稿データへのマップ
    posts: Arc<DashMap<i64, LightPost>>,

    /// ユーザーID からオリジナル投稿参照リストへのマップ
    original_posts_by_user: Arc<DashMap<i64, VecDeque<TinyPost>>>,

    /// ユーザーID からリプライ/リツイート参照リストへのマップ
    secondary_posts_by_user: Arc<DashMap<i64, VecDeque<TinyPost>>>,

    /// ユーザーID から動画投稿参照リストへのマップ
    video_posts_by_user: Arc<DashMap<i64, VecDeque<TinyPost>>>,

    /// 削除された投稿のトラッキング
    deleted_posts: Arc<DashMap<i64, bool>>,

    /// 保持期間（秒）
    retention_seconds: u64,

    /// リクエストタイムアウト
    request_timeout: Duration,
}

// 投稿挿入時の分類
if is_original {
    original_posts_by_user.entry(author_id).or_default().push_back(tiny_post);
} else {
    secondary_posts_by_user.entry(author_id).or_default().push_back(tiny_post);
}

// 動画投稿の判定
let mut video_eligible = post.has_video;
if !video_eligible && post.is_retweet {
    if let Some(source_post) = self.posts.get(&source_post_id) {
        video_eligible = !source_post.is_reply && source_post.has_video;
    }
}
if post.is_reply {
    video_eligible = false;
}

// デフォルト: 2日間
impl Default for PostStore {
    fn default() -> Self {
        Self::new(2 * 24 * 60 * 60, 0)  // 2日 = 172,800秒
    }
}

// バックグラウンドで定期的に古い投稿を削除
pub fn start_auto_trim(self: Arc<Self>, interval_minutes: u64) {
    tokio::spawn(async move {
        let mut interval = tokio::time::interval(Duration::from_secs(interval_minutes * 60));
        loop {
            interval.tick().await;
            let trimmed = self.trim_old_posts().await;
            if trimmed > 0 {
                info!("Auto-trim: removed {} old posts", trimmed);
            }
        }
    });
}

#[tonic::async_trait]
impl InNetworkPostsService for ThunderServiceImpl {
    async fn get_in_network_posts(
        &self,
        request: Request<GetInNetworkPostsRequest>,
    ) -> Result<Response<GetInNetworkPostsResponse>, Status>
}

pub fn server(self) -> InNetworkPostsServiceServer<Self> {
    InNetworkPostsServiceServer::new(self)
        .accept_compressed(tonic::codec::CompressionEncoding::Zstd)
        .send_compressed(tonic::codec::CompressionEncoding::Zstd)
}

pub struct ThunderServiceImpl {
    post_store: Arc<PostStore>,
    strato_client: Arc<StratoClient>,
    /// 同時リクエスト数を制限するセマフォ
    request_semaphore: Arc<Semaphore>,
}

// ノンブロッキングでセマフォ取得を試行
let _permit = match self.request_semaphore.try_acquire() {
    Ok(permit) => {
        IN_FLIGHT_REQUESTS.inc();
        permit
    }
    Err(_) => {
        REJECTED_REQUESTS.inc();
        return Err(Status::resource_exhausted(
            "Server at capacity, please retry",
        ));
    }
};

async fn fetch_candidates(&self, query: &Q) -> Vec<C> {
    let sources: Vec<_> = self.sources()
        .iter()
        .filter(|s| s.enable(query))
        .collect();

    // 全ソースを並列実行
    let source_futures = sources.iter().map(|s| s.get_candidates(query));
    let results = join_all(source_futures).await;

    // 結果を収集
    let mut collected = Vec::new();
    for (source, result) in sources.iter().zip(results) {
        match result {
            Ok(mut candidates) => collected.append(&mut candidates),
            Err(err) => error!("Source {} failed: {}", source.name(), err),
        }
    }
    collected
}

// エラー時のフォールバック例
match filter.filter(query, candidates).await {
    Ok(result) => {
        candidates = result.kept;
        all_removed.extend(result.removed);
    }
    Err(err) => {
        error!("Filter {} failed: {}", filter.name(), err);
        candidates = backup;  // 元のリストを復元
    }
}

#[async_trait]
pub trait CandidatePipeline<Q, C>: Send + Sync
{
    fn query_hydrators(&self) -> &[Box<dyn QueryHydrator<Q>>];
    fn sources(&self) -> &[Box<dyn Source<Q, C>>];
    fn hydrators(&self) -> &[Box<dyn Hydrator<Q, C>>];
    fn filters(&self) -> &[Box<dyn Filter<Q, C>>];
    fn scorers(&self) -> &[Box<dyn Scorer<Q, C>>];
    fn selector(&self) -> &dyn Selector<Q, C>;
    fn post_selection_hydrators(&self) -> &[Box<dyn Hydrator<Q, C>>];
    fn post_selection_filters(&self) -> &[Box<dyn Filter<Q, C>>];
    fn side_effects(&self) -> Arc<Vec<Box<dyn SideEffect<Q, C>>>>;
    fn result_size(&self) -> usize;
}

async fn execute(&self, query: Q) -> PipelineResult<Q, C> {
    // 1. クエリ補完
    let hydrated_query = self.hydrate_query(query).await;

    // 2. 候補取得
    let candidates = self.fetch_candidates(&hydrated_query).await;

    // 3. データ補完
    let hydrated_candidates = self.hydrate(&hydrated_query, candidates).await;

    // 4. フィルタリング
    let (kept_candidates, filtered_candidates) =
        self.filter(&hydrated_query, hydrated_candidates).await;

    // 5. スコアリング
    let scored_candidates = self.score(&hydrated_query, kept_candidates).await;

    // 6. 選択
    let selected_candidates = self.select(&hydrated_query, scored_candidates);

    // 7. 選択後処理
    let post_hydrated = self.hydrate_post_selection(&hydrated_query, selected_candidates).await;
    let (final_candidates, _) = self.filter_post_selection(&hydrated_query, post_hydrated).await;

    // 8. 結果サイズ制限
    final_candidates.truncate(self.result_size());

    // 9. 副作用実行（非同期、結果待機なし）
    self.run_side_effects(input);

    PipelineResult { ... }
}

pub struct PipelineResult<Q, C> {
    pub retrieved_candidates: Vec<C>,   // 取得された全候補
    pub filtered_candidates: Vec<C>,     // 除外された候補
    pub selected_candidates: Vec<C>,     // 最終選択された候補
    pub query: Arc<Q>,                   // 補完後のクエリ
}